From aaf2df8b908fcb48043d2cd51803d8fd99f18b43 Mon Sep 17 00:00:00 2001
From: sg481n
+This package provides an implementation of the authorization-related interfaces
+defined by the
+In order to perform the authorization, this package needs access to provisioning data
+about feeds and subscriptions. This package defines an interface
+(
+A software component that wishes to use this implementation must:
+AuthorizationResponse
interface gives the caller access to information about an authorization
+ * decision. This information includes the permit/deny decision itself, along with supplementary information in the form of
+ * advice and obligations. (The advice and obligations will not be used in Data Router R1.)
+ *
+ * @author J. F. Lucas
+ *
+ */
+public interface AuthorizationResponse {
+ /**
+ * Indicates whether the request is authorized or not.
+ *
+ * @return a boolean flag that is true
if the request is permitted, and false
otherwise.
+ */
+ public boolean isAuthorized();
+
+ /**
+ * Returns any advice elements that were included in the authorization response.
+ *
+ * @return A list of objects implementing the AuthorizationResponseSupplement
interface, with each object representing an
+ * advice element from the authorization response.
+ */
+ public ListAuthorizationResponseSupplement
interface, with each object representing an
+ * obligation element from the authorization response.
+ */
+ public ListAuthorizationResponseSupplement
interface carries supplementary
+ * information for an authorization response. In a XACML-based system, a response to an authorization request
+ * carries not just the permit/deny decision but, optionally, supplemental information in the form of advice and
+ * obligation elements. The structure of a XACML advice element and a XACML obligation element are similar: each has an identifier and
+ * a set of attributes (name-value) pairs. (The difference between a XACML advice element and a XACML obligation element is in
+ * how the recipient of the response--the Policy Enforcement Point, in XACML terminology--handles the element.)
+ *
+ * @author J. F. Lucas
+ *
+ */
+public interface AuthorizationResponseSupplement {
+ /** Return the identifier for the supplementary information element.
+ *
+ * @return a String
containing the identifier.
+ */
+ public String getId();
+
+ /** Return the attributes for the supplementary information element, as a Map
in which
+ * keys represent attribute identifiers and values represent attribute values.
+ *
+ * @return attributes for the supplementary information element.
+ */
+ public MapAuthorizer
interface. The class implements all of the logic necessary to determine if an API
+ * request is permitted. In Data Router R1, the classes that implement the Authorizer
interface will have
+ * local logic that makes the authorization decision. After R1, these classes will instead have logic that creates XACML
+ * authorization requests, sends these requests to a Policy Decision Point (PDP), and parses the XACML responses.
+ *
+ * @author J. F. Lucas
+ *
+ */
+public interface Authorizer {
+ /**
+ * Determine if the API request carried in the request
parameter is permitted.
+ *
+ * @param request the HTTP request for which an authorization decision is needed
+ * @return an object implementing the AuthorizationResponse
interface. This object includes the
+ * permit/deny decision for the request and (after R1) supplemental information related to the response in the form
+ * of advice and obligations.
+ */
+ public AuthorizationResponse decide(HttpServletRequest request);
+
+ /**
+ * Determine if the API request carried in the request
parameter, with additional attributes provided in
+ * the additionalAttrs
parameter, is permitted.
+ *
+ * @param request the HTTP request for which an authorization decision is needed
+ * @param additionalAttrs additional attributes that the Authorizer
can in making an authorization decision
+ * @return an object implementing the AuthorizationResponse
interface. This object includes the
+ * permit/deny decision for the request and (after R1) supplemental information related to the response in the form
+ * of advice and obligations.
+ */
+ public AuthorizationResponse decide(HttpServletRequest request, Maptrue
)
+ * or something else (false
).
+ * @param advice list of advice elements returned in the response.
+ * @param obligations list of obligation elements returned in the response.
+ */
+ public AuthRespImpl(boolean authorized, Listtrue
) or something else (false
).
+ */
+ public AuthRespImpl(boolean authorized) {
+ this(authorized, null, null);
+ }
+
+ /**
+ * Indicates whether the request is authorized or not.
+ *
+ * @return a boolean flag that is true
if the request is permitted, and false
otherwise.
+ */
+ @Override
+ public boolean isAuthorized() {
+ return authorized;
+ }
+
+ /**
+ * Returns any advice elements that were included in the authorization response.
+ *
+ * @return A list of objects implementing the AuthorizationResponseSupplement
interface, with each object representing an
+ * advice element from the authorization response.
+ */
+ @Override
+ public ListAuthorizationResponseSupplement
interface, with each object representing an
+ * obligation element from the authorization response.
+ */
+ @Override
+ public ListString
containing the identifier.
+ */
+ @Override
+ public String getId() {
+ return id;
+ }
+
+ /** Return the attributes for the supplementary information element, as a Map
in which
+ * keys represent attribute identifiers and values represent attribute values.
+ *
+ * @return attributes for the supplementary information element.
+ */
+ @Override
+ public Map
+ *
request
parameter is permitted.
+ *
+ * @param request the HTTP request for which an authorization decision is needed
+ * @return an object implementing the AuthorizationResponse
interface. This object includes the
+ * permit/deny decision for the request and (after R1) supplemental information related to the response in the form
+ * of advice and obligations.
+ */
+ @Override
+ public AuthorizationResponse decide(HttpServletRequest request) {
+ return this.decide(request, null);
+ }
+
+ /**
+ * Determine if the API request carried in the request
parameter, with additional attributes provided in
+ * the additionalAttrs
parameter, is permitted. additionalAttrs
isn't used in R1.
+ *
+ * @param request the HTTP request for which an authorization decision is needed
+ * @param additionalAttrs additional attributes that the Authorizer
can in making an authorization decision
+ * @return an object implementing the AuthorizationResponse
interface. This object includes the
+ * permit/deny decision for the request and (after R1) supplemental information related to the response in the form
+ * of advice and obligations.
+ */
+ @Override
+ public AuthorizationResponse decide(HttpServletRequest request,
+ MapProvAuthorizer
needs to supply an implementation of this interface.
+ * @author J. F. Lucas
+ *
+ */
+public interface ProvDataProvider {
+
+ /** Get the identity of the owner of a feed.
+ *
+ * @param feedId the feed ID of the feed whose owner is being looked up.
+ * @return the feed owner's identity
+ */
+ public String getFeedOwner(String feedId);
+
+ /** Get the security classification of a feed.
+ *
+ * @param feedId the ID of the feed whose classification is being looked up.
+ * @return the classification of the feed.
+ */
+ public String getFeedClassification(String feedId);
+
+ /** Get the identity of the owner of a feed
+ *
+ * @param subId the ID of the subscripition whose owner is being looked up.
+ * @return the subscription owner's identity.
+ */
+ public String getSubscriptionOwner(String subId);
+
+ /** Get the identity of the owner of a feed by group id - Rally : US708115
+ *
+ * @param feedid, user the ID of the feed whose owner is being looked up.
+ * @return the feed owner's identity by group.
+ */
+ public String getGroupByFeedGroupId(String owner, String feedId);
+
+ /** Get the identity of the owner of a sub by group id Rally : US708115
+ *
+ * @param subid, user the ID of the feed whose owner is being looked up.
+ * @return the feed owner's identity by group.
+ */
+ public String getGroupBySubGroupId(String owner, String subId);
+}
diff --git a/datarouter-prov/src/main/java/com/att/research/datarouter/authz/impl/package.html b/datarouter-prov/src/main/java/com/att/research/datarouter/authz/impl/package.html
new file mode 100644
index 00000000..fae27ee0
--- /dev/null
+++ b/datarouter-prov/src/main/java/com/att/research/datarouter/authz/impl/package.html
@@ -0,0 +1,68 @@
+#-------------------------------------------------------------------------------
+# ============LICENSE_START==================================================
+# * org.onap.dmaap
+# * ===========================================================================
+# * Copyright © 2017 AT&T Intellectual Property. All rights reserved.
+# * ===========================================================================
+# * Licensed under the Apache License, Version 2.0 (the "License");
+# * you may not use this file except in compliance with the License.
+# * You may obtain a copy of the License at
+# *
+# * http://www.apache.org/licenses/LICENSE-2.0
+# *
+# * Unless required by applicable law or agreed to in writing, software
+# * distributed under the License is distributed on an "AS IS" BASIS,
+# * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# * See the License for the specific language governing permissions and
+# * limitations under the License.
+# * ============LICENSE_END====================================================
+# *
+# * ECOMP is a trademark and service mark of AT&T Intellectual Property.
+# *
+#-------------------------------------------------------------------------------
+
+
+
+
+
+com.att.research.datarouter.authz
package, intended for
+use with the provisioning server for Data Router Release 1. In DR R1, we do not
+have an external policy engine, so this implementation performs the authorization
+locally.
+com.att.research.datarouter.authz.impl.ProvDataProvider
) through which it
+expects to get this data. The provisioning server code must provide an implementation
+of this interface.
+
+
+com.att.research.datarouter.authz.impl.ProvDataProvider
+interface.
+ProvDataProvider
implementation.
+com.att.research.datarouter.authz.impl.ProvAuthorizer
+class defined in this package, passing it an instance of the ProvDataProvider
+implementation.
+
+Example: +
+
+ProvDataProvider dataProv = new MyDataProvider();
+Authorizer authz = new ProvAuthorizer(dataProv);
+
+
+
+
diff --git a/datarouter-prov/src/main/java/com/att/research/datarouter/authz/package.html b/datarouter-prov/src/main/java/com/att/research/datarouter/authz/package.html
new file mode 100644
index 00000000..7628ae82
--- /dev/null
+++ b/datarouter-prov/src/main/java/com/att/research/datarouter/authz/package.html
@@ -0,0 +1,38 @@
+#-------------------------------------------------------------------------------
+# ============LICENSE_START==================================================
+# * org.onap.dmaap
+# * ===========================================================================
+# * Copyright © 2017 AT&T Intellectual Property. All rights reserved.
+# * ===========================================================================
+# * Licensed under the Apache License, Version 2.0 (the "License");
+# * you may not use this file except in compliance with the License.
+# * You may obtain a copy of the License at
+# *
+# * http://www.apache.org/licenses/LICENSE-2.0
+# *
+# * Unless required by applicable law or agreed to in writing, software
+# * distributed under the License is distributed on an "AS IS" BASIS,
+# * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# * See the License for the specific language governing permissions and
+# * limitations under the License.
+# * ============LICENSE_END====================================================
+# *
+# * ECOMP is a trademark and service mark of AT&T Intellectual Property.
+# *
+#-------------------------------------------------------------------------------
+
+
+
+
+
++This package defines an interface that can be used by servlet-based HTTP APIs to +make authorization requests and receive authorization responses from an external +authorization entity such as a XACML Policy Decision Point (PDP). +
++In Data Router Release 1, there is no external authorization system. The provisioning server +will use an implementation of this interface for local authorization of provisioning requests. +
+ + diff --git a/datarouter-prov/src/main/java/com/att/research/datarouter/provisioning/BaseServlet.java b/datarouter-prov/src/main/java/com/att/research/datarouter/provisioning/BaseServlet.java new file mode 100644 index 00000000..a4784936 --- /dev/null +++ b/datarouter-prov/src/main/java/com/att/research/datarouter/provisioning/BaseServlet.java @@ -0,0 +1,869 @@ +/******************************************************************************* + * ============LICENSE_START================================================== + * * org.onap.dmaap + * * =========================================================================== + * * Copyright © 2017 AT&T Intellectual Property. All rights reserved. + * * =========================================================================== + * * Licensed under the Apache License, Version 2.0 (the "License"); + * * you may not use this file except in compliance with the License. + * * You may obtain a copy of the License at + * * + * * http://www.apache.org/licenses/LICENSE-2.0 + * * + * * Unless required by applicable law or agreed to in writing, software + * * distributed under the License is distributed on an "AS IS" BASIS, + * * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * * See the License for the specific language governing permissions and + * * limitations under the License. + * * ============LICENSE_END==================================================== + * * + * * ECOMP is a trademark and service mark of AT&T Intellectual Property. + * * + ******************************************************************************/ + + +package com.att.research.datarouter.provisioning; + +import static com.att.eelf.configuration.Configuration.MDC_SERVER_FQDN; + +import static com.att.eelf.configuration.Configuration.MDC_SERVER_IP_ADDRESS; +import static com.att.eelf.configuration.Configuration.MDC_SERVICE_NAME; + +import java.io.IOException; +import java.io.InputStream; +import java.net.InetAddress; +import java.net.UnknownHostException; +import java.security.cert.X509Certificate; +import java.sql.Connection; +import java.sql.SQLException; +import java.util.HashMap; +import java.util.HashSet; +import java.util.Map; +import java.util.Set; +import java.util.List; +import java.util.ArrayList; + +import javax.servlet.ServletConfig; +import javax.servlet.ServletException; +import javax.servlet.http.HttpServlet; +import javax.servlet.http.HttpServletRequest; + +import org.apache.log4j.Logger; +import org.json.JSONObject; +import org.json.JSONTokener; +import org.json.JSONException; +import org.slf4j.MDC; + +import com.att.research.datarouter.authz.Authorizer; +import com.att.research.datarouter.authz.impl.ProvAuthorizer; +import com.att.research.datarouter.authz.impl.ProvDataProvider; +import com.att.research.datarouter.provisioning.beans.Deleteable; +import com.att.research.datarouter.provisioning.beans.Feed; +import com.att.research.datarouter.provisioning.beans.Insertable; +import com.att.research.datarouter.provisioning.beans.NodeClass; +import com.att.research.datarouter.provisioning.beans.Parameters; +import com.att.research.datarouter.provisioning.beans.Subscription; +import com.att.research.datarouter.provisioning.beans.Updateable; +import com.att.research.datarouter.provisioning.utils.DB; +import com.att.research.datarouter.provisioning.utils.ThrottleFilter; +import com.att.research.datarouter.provisioning.beans.Group; //Groups feature Rally:US708115 - 1610 + +import java.util.Properties; +import java.util.regex.Pattern; +import javax.mail.Message; +import javax.mail.MessagingException; +import javax.mail.Multipart; +import javax.mail.Session; +import javax.mail.Transport; +import javax.mail.internet.AddressException; +import javax.mail.internet.InternetAddress; +import javax.mail.internet.MimeBodyPart; +import javax.mail.internet.MimeMessage; +import javax.mail.internet.MimeMultipart; +/** + * This is the base class for all Servlets in the provisioning code. + * It provides standard constants and some common methods. + * + * @author Robert Eby + * @version $Id: BaseServlet.java,v 1.16 2014/03/12 19:45:40 eby Exp $ + */ +@SuppressWarnings("serial") +public class BaseServlet extends HttpServlet implements ProvDataProvider { + public static final String BEHALF_HEADER = "X-ATT-DR-ON-BEHALF-OF"; + public static final String FEED_BASECONTENT_TYPE = "application/vnd.att-dr.feed"; + public static final String FEED_CONTENT_TYPE = "application/vnd.att-dr.feed; version=2.0"; + public static final String FEEDFULL_CONTENT_TYPE = "application/vnd.att-dr.feed-full; version=2.0"; + public static final String FEEDLIST_CONTENT_TYPE = "application/vnd.att-dr.feed-list; version=1.0"; + public static final String SUB_BASECONTENT_TYPE = "application/vnd.att-dr.subscription"; + public static final String SUB_CONTENT_TYPE = "application/vnd.att-dr.subscription; version=2.0"; + public static final String SUBFULL_CONTENT_TYPE = "application/vnd.att-dr.subscription-full; version=2.0"; + public static final String SUBLIST_CONTENT_TYPE = "application/vnd.att-dr.subscription-list; version=1.0"; + + + //Adding groups functionality, ...1610 + public static final String GROUP_BASECONTENT_TYPE = "application/vnd.att-dr.group"; + public static final String GROUP_CONTENT_TYPE = "application/vnd.att-dr.group; version=2.0"; + public static final String GROUPFULL_CONTENT_TYPE = "application/vnd.att-dr.group-full; version=2.0"; + public static final String GROUPLIST_CONTENT_TYPE = "application/vnd.att-dr.fegrouped-list; version=1.0"; + + + public static final String LOGLIST_CONTENT_TYPE = "application/vnd.att-dr.log-list; version=1.0"; + public static final String PROVFULL_CONTENT_TYPE1 = "application/vnd.att-dr.provfeed-full; version=1.0"; + public static final String PROVFULL_CONTENT_TYPE2 = "application/vnd.att-dr.provfeed-full; version=2.0"; + public static final String CERT_ATTRIBUTE = "javax.servlet.request.X509Certificate"; + + public static final String DB_PROBLEM_MSG = "There has been a problem with the DB. It is suggested you try the operation again."; + + public static final int DEFAULT_MAX_FEEDS = 10000; + public static final int DEFAULT_MAX_SUBS = 100000; + public static final int DEFAULT_POKETIMER1 = 5; + public static final int DEFAULT_POKETIMER2 = 30; + public static final String DEFAULT_DOMAIN = "web.att.com"; + public static final String DEFAULT_PROVSRVR_NAME = "feeds-drtr.web.att.com"; + public static final String RESEARCH_SUBNET = "135.207.136.128/25"; + public static final String STATIC_ROUTING_NODES = ""; //Adding new param for static Routing - Rally:US664862-1610 + + /** A boolean to trigger one time "provisioning changed" event on startup */ + private static boolean startmsg_flag = true; + /** This POD should require SSL connections from clients; pulled from the DB (PROV_REQUIRE_SECURE) */ + private static boolean require_secure = true; + /** This POD should require signed, recognized certificates from clients; pulled from the DB (PROV_REQUIRE_CERT) */ + private static boolean require_cert = true; + /** The set of authorized addresses and networks; pulled from the DB (PROV_AUTH_ADDRESSES) */ + private static Set+ * This servlet handles requests to URLs under /internal on the provisioning server. + * These include: + *
+ *URL Path | + *Method | + *Purpose | + *
---|---|---|
/internal/prov | + *GET | + *used to GET a full JSON copy of the provisioning data. | + *
/internal/fetchProv | + *GET | + *used to signal to a standby POD that the provisioning data should be fetched from the active POD. | + *
/internal/logs | + *GET | + *used to GET an index of log files and individual logs for this provisioning server. | + *
POST | + *used to POST log files from the individual nodes to this provisioning server. | + *|
/internal/api | + *GET | + *used to GET an individual parameter value. The parameter name is specified by the path after /api/. | + *
PUT | + *used to set an individual parameter value. The parameter name is specified by the path after /api/. | + *|
DELETE | + *used to remove an individual parameter value. The parameter name is specified by the path after /api/. | + *|
POST | + *used to create a new individual parameter value. The parameter name is specified by the path after /api/. | + *|
/internal/halt | + *GET | + *used to halt the server (must be accessed from 127.0.0.1). | + *
/internal/drlogs | + *GET | + *used to get a list of DR log entries available for retrieval. + * Note: these are the actual data router log entries sent to the provisioning server + * by the nodes, not the provisioning server's internal logs (access via /internal/logs above). + * The range is returned as a list of record sequence numbers. | + *
POST | + *used to retrieve specific log entries. + * The sequence numbers of the records to fetch are POST-ed; the records matching the sequence numbers are returned. | + *|
/internal/route/* | + ** | + *URLs under this path are handled via the {@link com.att.research.datarouter.provisioning.RouteServlet} | + *
+ * Authorization to use these URLs is a little different than for other URLs on the provisioning server. + * For the most part, the IP address that the request comes from should be either: + *
+ *+ * In addition, requests to /internal/halt can ONLY come from localhost (127.0.0.1) on the HTTP port. + *
+ *+ * All DELETE/GET/PUT/POST requests made to /internal/api on this servlet on the standby server are + * proxied to the active server (using the {@link ProxyServlet}) if it is up and reachable. + *
+ * + * @author Robert Eby + * @version $Id: InternalServlet.java,v 1.23 2014/03/24 18:47:10 eby Exp $ + */ +@SuppressWarnings("serial") +public class InternalServlet extends ProxyServlet { + private static Integer logseq = new Integer(0); // another piece of info to make log spool file names unique + //Adding EELF Logger Rally:US664892 + private static EELFLogger eelflogger = EELFManager.getInstance().getLogger("com.att.research.datarouter.provisioning.InternalServlet"); + + /** + * Delete a parameter at the address /internal/api/<parameter>. + * See the Internal API document for details on how this method should be invoked. + */ + @Override + public void doDelete(HttpServletRequest req, HttpServletResponse resp) throws IOException { + setIpAndFqdnForEelf("doDelete"); + eelflogger.info(EelfMsgs.MESSAGE_WITH_BEHALF_AND_FEEDID, req.getHeader(BEHALF_HEADER),getIdFromPath(req)+""); + EventLogRecord elr = new EventLogRecord(req); + if (!isAuthorizedForInternal(req)) { + elr.setMessage("Unauthorized."); + elr.setResult(HttpServletResponse.SC_FORBIDDEN); + eventlogger.info(elr); + resp.sendError(HttpServletResponse.SC_FORBIDDEN, "Unauthorized."); + return; + } + + String path = req.getPathInfo(); + if (path.startsWith("/api/")) { + if (isProxyOK(req) && isProxyServer()) { + super.doDelete(req, resp); + return; + } + String key = path.substring(5); + if (key.length() > 0) { + Parameters param = Parameters.getParameter(key); + if (param != null) { + if (doDelete(param)) { + elr.setResult(HttpServletResponse.SC_OK); + eventlogger.info(elr); + resp.setStatus(HttpServletResponse.SC_OK); + provisioningDataChanged(); + provisioningParametersChanged(); + } else { + // Something went wrong with the DELETE + elr.setResult(HttpServletResponse.SC_INTERNAL_SERVER_ERROR); + eventlogger.info(elr); + resp.sendError(HttpServletResponse.SC_INTERNAL_SERVER_ERROR, DB_PROBLEM_MSG); + } + return; + } + } + } + resp.sendError(HttpServletResponse.SC_NOT_FOUND, "Bad URL."); + } + /** + * Get some information (such as a parameter) underneath the /internal/ namespace. + * See the Internal API document for details on how this method should be invoked. + */ + @Override + public void doGet(HttpServletRequest req, HttpServletResponse resp) throws IOException { + setIpAndFqdnForEelf("doGet"); + eelflogger.info(EelfMsgs.MESSAGE_WITH_BEHALF_AND_FEEDID, req.getHeader(BEHALF_HEADER),getIdFromPath(req)+""); + String path = req.getPathInfo(); + if (path.equals("/halt") && !req.isSecure()) { + // request to halt the server - can ONLY come from localhost + String remote = req.getRemoteAddr(); + if (remote.equals("127.0.0.1")) { + intlogger.info("PROV0009 Request to HALT received."); + resp.setStatus(HttpServletResponse.SC_OK); + Main.shutdown(); + } else { + intlogger.info("PROV0010 Disallowed request to HALT received from "+remote); + resp.setStatus(HttpServletResponse.SC_FORBIDDEN); + } + return; + } + + EventLogRecord elr = new EventLogRecord(req); + if (!isAuthorizedForInternal(req)) { + elr.setMessage("Unauthorized."); + elr.setResult(HttpServletResponse.SC_FORBIDDEN); + eventlogger.info(elr); + resp.sendError(HttpServletResponse.SC_FORBIDDEN, "Unauthorized."); + return; + } + if (path.equals("/fetchProv") && !req.isSecure()) { + // if request came from active_pod or standby_pod and it is not us, reload prov data + SynchronizerTask s = SynchronizerTask.getSynchronizer(); + s.doFetch(); + resp.setStatus(HttpServletResponse.SC_OK); + return; + } + if (path.equals("/prov")) { + if (isProxyOK(req) && isProxyServer()) { + if (super.doGetWithFallback(req, resp)) + return; + // fall back to returning the local data if the remote is unreachable + intlogger.info("Active server unavailable; falling back to local copy."); + } + Poker p = Poker.getPoker(); + resp.setStatus(HttpServletResponse.SC_OK); + resp.setContentType(PROVFULL_CONTENT_TYPE2); + resp.getOutputStream().print(p.getProvisioningString()); + return; + } + if (path.equals("/logs") || path.equals("/logs/")) { + resp.setStatus(HttpServletResponse.SC_OK); + resp.setContentType("application/json"); + resp.getOutputStream().print(generateLogfileList().toString()); + return; + } + if (path.startsWith("/logs/")) { + Properties p = (new DB()).getProperties(); + String logdir = p.getProperty("com.att.research.datarouter.provserver.accesslog.dir"); + String logfile = path.substring(6); + if (logdir != null && logfile != null && logfile.indexOf('/') < 0) { + File log = new File(logdir + "/" + logfile); + if (log.exists() && log.isFile()) { + resp.setStatus(HttpServletResponse.SC_OK); + resp.setContentType("text/plain"); + Path logpath = Paths.get(log.getAbsolutePath()); + Files.copy(logpath, resp.getOutputStream()); + return; + } + } + resp.sendError(HttpServletResponse.SC_NO_CONTENT, "No file."); + return; + } + if (path.startsWith("/api/")) { + if (isProxyOK(req) && isProxyServer()) { + super.doGet(req, resp); + return; + } + String key = path.substring(5); + if (key.length() > 0) { + Parameters param = Parameters.getParameter(key); + if (param != null) { + resp.setStatus(HttpServletResponse.SC_OK); + resp.setContentType("text/plain"); + resp.getOutputStream().print(param.getValue() + "\n"); + return; + } + } + } + if (path.equals("/drlogs") || path.equals("/drlogs/")) { + // Special POD <=> POD API to determine what log file records are loaded here + LogfileLoader lfl = LogfileLoader.getLoader(); + resp.setStatus(HttpServletResponse.SC_OK); + resp.setContentType("text/plain"); + resp.getOutputStream().print(lfl.getBitSet().toString()); + return; + } + resp.sendError(HttpServletResponse.SC_NOT_FOUND, "Bad URL."); + } + /** + * Modify a parameter at the address /internal/api/<parameter>. + * See the Internal API document for details on how this method should be invoked. + */ + @Override + public void doPut(HttpServletRequest req, HttpServletResponse resp) throws IOException { + setIpAndFqdnForEelf("doPut"); + eelflogger.info(EelfMsgs.MESSAGE_WITH_BEHALF_AND_FEEDID, req.getHeader(BEHALF_HEADER),getIdFromPath(req)+""); + EventLogRecord elr = new EventLogRecord(req); + if (!isAuthorizedForInternal(req)) { + elr.setMessage("Unauthorized."); + elr.setResult(HttpServletResponse.SC_FORBIDDEN); + eventlogger.info(elr); + resp.sendError(HttpServletResponse.SC_FORBIDDEN, "Unauthorized."); + return; + } + String path = req.getPathInfo(); + if (path.startsWith("/api/")) { + if (isProxyOK(req) && isProxyServer()) { + super.doPut(req, resp); + return; + } + String key = path.substring(5); + if (key.length() > 0) { + Parameters param = Parameters.getParameter(key); + if (param != null) { + String t = catValues(req.getParameterValues("val")); + param.setValue(t); + if (doUpdate(param)) { + elr.setResult(HttpServletResponse.SC_OK); + eventlogger.info(elr); + resp.setStatus(HttpServletResponse.SC_OK); + provisioningDataChanged(); + provisioningParametersChanged(); + } else { + // Something went wrong with the UPDATE + elr.setResult(HttpServletResponse.SC_INTERNAL_SERVER_ERROR); + eventlogger.info(elr); + resp.sendError(HttpServletResponse.SC_INTERNAL_SERVER_ERROR, DB_PROBLEM_MSG); + } + return; + } + } + } + resp.sendError(HttpServletResponse.SC_NOT_FOUND, "Bad URL."); + } + /** + * Create some new information (such as a parameter or log entries) underneath the /internal/ namespace. + * See the Internal API document for details on how this method should be invoked. + */ + @SuppressWarnings("resource") + @Override + public void doPost(HttpServletRequest req, HttpServletResponse resp) throws IOException { + setIpAndFqdnForEelf("doPost"); + eelflogger.info(EelfMsgs.MESSAGE_WITH_BEHALF, req.getHeader(BEHALF_HEADER)); + EventLogRecord elr = new EventLogRecord(req); + if (!isAuthorizedForInternal(req)) { + elr.setMessage("Unauthorized."); + elr.setResult(HttpServletResponse.SC_FORBIDDEN); + eventlogger.info(elr); + resp.sendError(HttpServletResponse.SC_FORBIDDEN, "Unauthorized."); + return; + } + + String path = req.getPathInfo(); + if (path.startsWith("/api/")) { + if (isProxyOK(req) && isProxyServer()) { + super.doPost(req, resp); + return; + } + String key = path.substring(5); + if (key.length() > 0) { + Parameters param = Parameters.getParameter(key); + if (param == null) { + String t = catValues(req.getParameterValues("val")); + param = new Parameters(key, t); + if (doInsert(param)) { + elr.setResult(HttpServletResponse.SC_OK); + eventlogger.info(elr); + resp.setStatus(HttpServletResponse.SC_OK); + provisioningDataChanged(); + provisioningParametersChanged(); + } else { + // Something went wrong with the INSERT + elr.setResult(HttpServletResponse.SC_INTERNAL_SERVER_ERROR); + eventlogger.info(elr); + resp.sendError(HttpServletResponse.SC_INTERNAL_SERVER_ERROR, DB_PROBLEM_MSG); + } + return; + } + } + } + + if (path.equals("/logs") || path.equals("/logs/")) { + String ctype = req.getHeader("Content-Type"); + if (ctype == null || !ctype.equals("text/plain")) { + elr.setResult(HttpServletResponse.SC_UNSUPPORTED_MEDIA_TYPE); + elr.setMessage("Bad media type: "+ctype); + resp.setStatus(HttpServletResponse.SC_UNSUPPORTED_MEDIA_TYPE); + eventlogger.info(elr); + return; + } + String spooldir = (new DB()).getProperties().getProperty("com.att.research.datarouter.provserver.spooldir"); + String spoolname = String.format("%d-%d-", System.currentTimeMillis(), Thread.currentThread().getId()); + synchronized (logseq) { + // perhaps unnecessary, but it helps make the name unique + spoolname += logseq.toString(); + logseq++; + } + String encoding = req.getHeader("Content-Encoding"); + if (encoding != null) { + if (encoding.trim().equals("gzip")) { + spoolname += ".gz"; + } else { + elr.setResult(HttpServletResponse.SC_UNSUPPORTED_MEDIA_TYPE); + resp.setStatus(HttpServletResponse.SC_UNSUPPORTED_MEDIA_TYPE); + eventlogger.info(elr); + return; + } + } + // Determine space available -- available space must be at least 5% + FileSystem fs = (Paths.get(spooldir)).getFileSystem(); + long total = 0; + long avail = 0; + for (FileStore store: fs.getFileStores()) { + total += store.getTotalSpace(); + avail += store.getUsableSpace(); + } + try { fs.close(); } catch (Exception e) { } + if (((avail * 100) / total) < 5) { + elr.setResult(HttpServletResponse.SC_SERVICE_UNAVAILABLE); + resp.setStatus(HttpServletResponse.SC_SERVICE_UNAVAILABLE); + eventlogger.info(elr); + return; + } + Path tmppath = Paths.get(spooldir, spoolname); + Path donepath = Paths.get(spooldir, "IN."+spoolname); + Files.copy(req.getInputStream(), Paths.get(spooldir, spoolname), StandardCopyOption.REPLACE_EXISTING); + Files.move(tmppath, donepath, StandardCopyOption.REPLACE_EXISTING); + elr.setResult(HttpServletResponse.SC_CREATED); + resp.setStatus(HttpServletResponse.SC_CREATED); + eventlogger.info(elr); + LogfileLoader.getLoader(); // This starts the logfile loader "task" + return; + } + + if (path.equals("/drlogs") || path.equals("/drlogs/")) { + // Receive post request and generate log entries + String ctype = req.getHeader("Content-Type"); + if (ctype == null || !ctype.equals("text/plain")) { + elr.setResult(HttpServletResponse.SC_UNSUPPORTED_MEDIA_TYPE); + elr.setMessage("Bad media type: "+ctype); + resp.setStatus(HttpServletResponse.SC_UNSUPPORTED_MEDIA_TYPE); + eventlogger.info(elr); + return; + } + InputStream is = req.getInputStream(); + ByteArrayOutputStream bos = new ByteArrayOutputStream(); + int ch = 0; + while ((ch = is.read()) >= 0) + bos.write(ch); + RLEBitSet bs = new RLEBitSet(bos.toString()); // The set of records to retrieve + elr.setResult(HttpServletResponse.SC_OK); + resp.setStatus(HttpServletResponse.SC_OK); + resp.setContentType("text/plain"); + LogRecord.printLogRecords(resp.getOutputStream(), bs); + eventlogger.info(elr); + return; + } + + elr.setResult(HttpServletResponse.SC_NOT_FOUND); + resp.sendError(HttpServletResponse.SC_NOT_FOUND, "Bad URL."); + eventlogger.info(elr); + } + + private String catValues(String[] v) { + StringBuilder sb = new StringBuilder(); + if (v != null) { + String pfx = ""; + for (String s : v) { + sb.append(pfx); + sb.append(s); + pfx = "|"; + } + } + return sb.toString(); + } + private JSONArray generateLogfileList() { + JSONArray ja = new JSONArray(); + Properties p = (new DB()).getProperties(); + String s = p.getProperty("com.att.research.datarouter.provserver.accesslog.dir"); + if (s != null) { + String[] dirs = s.split(","); + for (String dir : dirs) { + File f = new File(dir); + String[] list = f.list(); + if (list != null) { + for (String s2 : list) { + if (!s2.startsWith(".")) + ja.put(s2); + } + } + } + } + return ja; + } +} diff --git a/datarouter-prov/src/main/java/com/att/research/datarouter/provisioning/LogServlet.java b/datarouter-prov/src/main/java/com/att/research/datarouter/provisioning/LogServlet.java new file mode 100644 index 00000000..7ef74d12 --- /dev/null +++ b/datarouter-prov/src/main/java/com/att/research/datarouter/provisioning/LogServlet.java @@ -0,0 +1,433 @@ +/******************************************************************************* + * ============LICENSE_START================================================== + * * org.onap.dmaap + * * =========================================================================== + * * Copyright © 2017 AT&T Intellectual Property. All rights reserved. + * * =========================================================================== + * * Licensed under the Apache License, Version 2.0 (the "License"); + * * you may not use this file except in compliance with the License. + * * You may obtain a copy of the License at + * * + * * http://www.apache.org/licenses/LICENSE-2.0 + * * + * * Unless required by applicable law or agreed to in writing, software + * * distributed under the License is distributed on an "AS IS" BASIS, + * * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * * See the License for the specific language governing permissions and + * * limitations under the License. + * * ============LICENSE_END==================================================== + * * + * * ECOMP is a trademark and service mark of AT&T Intellectual Property. + * * + ******************************************************************************/ + + +package com.att.research.datarouter.provisioning; + +import java.io.IOException; +import java.sql.Connection; +import java.sql.ResultSet; +import java.sql.SQLException; +import java.sql.Statement; +import java.text.ParseException; +import java.text.SimpleDateFormat; +import java.util.Date; +import java.util.HashMap; +import java.util.Map; + +import javax.servlet.ServletOutputStream; +import javax.servlet.http.HttpServletRequest; +import javax.servlet.http.HttpServletResponse; + +import org.json.LOGJSONObject; + +import com.att.eelf.configuration.EELFLogger; +import com.att.eelf.configuration.EELFManager; +import com.att.research.datarouter.provisioning.beans.DeliveryRecord; +import com.att.research.datarouter.provisioning.beans.EventLogRecord; +import com.att.research.datarouter.provisioning.beans.ExpiryRecord; +import com.att.research.datarouter.provisioning.beans.LOGJSONable; +import com.att.research.datarouter.provisioning.beans.PublishRecord; +import com.att.research.datarouter.provisioning.beans.Subscription; +import com.att.research.datarouter.provisioning.eelf.EelfMsgs; +import com.att.research.datarouter.provisioning.utils.DB; + +/** + * This servlet handles requests to the <feedLogURL> and <subLogURL>, + * which are generated by the provisioning server to handle the log query API. + * + * @author Robert Eby + * @version $Id: LogServlet.java,v 1.11 2014/03/28 17:27:02 eby Exp $ + */ +@SuppressWarnings("serial") +public class LogServlet extends BaseServlet { + //Adding EELF Logger Rally:US664892 + private static EELFLogger eelflogger = EELFManager.getInstance().getLogger("com.att.research.datarouter.provisioning.LogServlet"); + + private static final long TWENTYFOUR_HOURS = (24 * 60 * 60 * 1000L); + private static final String fmt1 = "yyyy-MM-dd'T'HH:mm:ss'Z'"; + private static final String fmt2 = "yyyy-MM-dd'T'HH:mm:ss.SSS'Z'"; + + private boolean isfeedlog; + + public abstract class RowHandler { + private final ServletOutputStream out; + private final String[] fields; + public boolean firstrow; + + public RowHandler(ServletOutputStream out, String fieldparam, boolean b) { + this.out = out; + this.firstrow = b; + this.fields = (fieldparam != null) ? fieldparam.split(":") : null; + } + public void handleRow(ResultSet rs) { + try { + LOGJSONable js = buildJSONable(rs); + LOGJSONObject jo = js.asJSONObject(); + if (fields != null) { + // filter out unwanted fields + LOGJSONObject j2 = new LOGJSONObject(); + for (String key : fields) { + Object v = jo.opt(key); + if (v != null) + j2.put(key, v); + } + jo = j2; + } + String t = firstrow ? "\n" : ",\n"; + t += jo.toString(); + out.print(t); + firstrow = false; + } catch (Exception e) { + // ignore + } + } + public abstract LOGJSONable buildJSONable(ResultSet rs) throws SQLException; + } + public class PublishRecordRowHandler extends RowHandler { + public PublishRecordRowHandler(ServletOutputStream out, String fields, boolean b) { + super(out, fields, b); + } + @Override + public LOGJSONable buildJSONable(ResultSet rs) throws SQLException { + return new PublishRecord(rs); + } + } + public class DeliveryRecordRowHandler extends RowHandler { + public DeliveryRecordRowHandler(ServletOutputStream out, String fields, boolean b) { + super(out, fields, b); + } + @Override + public LOGJSONable buildJSONable(ResultSet rs) throws SQLException { + return new DeliveryRecord(rs); + } + } + public class ExpiryRecordRowHandler extends RowHandler { + public ExpiryRecordRowHandler(ServletOutputStream out, String fields, boolean b) { + super(out, fields, b); + } + @Override + public LOGJSONable buildJSONable(ResultSet rs) throws SQLException { + return new ExpiryRecord(rs); + } + } + + /** + * This class must be created from either a {@link FeedLogServlet} or a {@link SubLogServlet}. + * @param isFeedLog boolean to handle those places where a feedlog request is different from + * a sublog request + */ + protected LogServlet(boolean isFeedLog) { + this.isfeedlog = isFeedLog; + } + + /** + * DELETE a logging URL -- not supported. + */ + @Override + public void doDelete(HttpServletRequest req, HttpServletResponse resp) throws IOException { + setIpAndFqdnForEelf("doDelete"); + eelflogger.info(EelfMsgs.MESSAGE_WITH_BEHALF_AND_FEEDID, req.getHeader(BEHALF_HEADER),getIdFromPath(req)+""); + String message = "DELETE not allowed for the logURL."; + EventLogRecord elr = new EventLogRecord(req); + elr.setMessage(message); + elr.setResult(HttpServletResponse.SC_METHOD_NOT_ALLOWED); + eventlogger.info(elr); + resp.sendError(HttpServletResponse.SC_METHOD_NOT_ALLOWED, message); + } + /** + * GET a logging URL -- retrieve logging data for a feed or subscription. + * See the Logging API document for details on how this method should be invoked. + */ + @Override + public void doGet(HttpServletRequest req, HttpServletResponse resp) throws IOException { + setIpAndFqdnForEelf("doGet"); + eelflogger.info(EelfMsgs.MESSAGE_WITH_BEHALF_AND_FEEDID, req.getHeader(BEHALF_HEADER),getIdFromPath(req)+""); + int id = getIdFromPath(req); + if (id < 0) { + resp.sendError(HttpServletResponse.SC_BAD_REQUEST, "Missing or bad feed/subscription number."); + return; + } + Map+ * A main class which may be used to start the provisioning server with an "embedded" Jetty server. + * Configuration is done via the properties file provserver.properties, which should be in the CLASSPATH. + * The provisioning server may also be packaged with a web.xml and started as a traditional webapp. + *
+ *+ * Most of the work of the provisioning server is carried out within the eight servlets (configured below) + * that are used to handle each of the eight types of requests the server may receive. + * In addition, there are background threads started to perform other tasks: + *
+ *+ * The provisioning server is stopped by issuing a GET to the URL http://127.0.0.1/internal/halt + * using curl or some other such tool. + *
+ * + * @author Robert Eby + * @version $Id: Main.java,v 1.12 2014/03/12 19:45:41 eby Exp $ + */ +public class Main { + /** The truststore to use if none is specified */ + public static final String DEFAULT_TRUSTSTORE = "/opt/java/jdk/jdk180/jre/lib/security/cacerts"; + public static final String KEYSTORE_TYPE_PROPERTY = "com.att.research.datarouter.provserver.keystore.type"; + public static final String KEYSTORE_PATH_PROPERTY = "com.att.research.datarouter.provserver.keystore.path"; + public static final String KEYSTORE_PASSWORD_PROPERTY = "com.att.research.datarouter.provserver.keystore.password"; + public static final String TRUSTSTORE_PATH_PROPERTY = "com.att.research.datarouter.provserver.truststore.path"; + public static final String TRUSTSTORE_PASSWORD_PROPERTY = "com.att.research.datarouter.provserver.truststore.password"; + + /** The one and only {@link Server} instance in this JVM */ + private static Server server; + + /** + * Starts the Data Router Provisioning server. + * @param args not used + * @throws Exception if Jetty has a problem starting + */ + public static void main(String[] args) throws Exception { + Security.setProperty("networkaddress.cache.ttl", "4"); + Logger logger = Logger.getLogger("com.att.research.datarouter.provisioning.internal"); + + // Check DB is accessible and contains the expected tables + if (! checkDatabase(logger)) + System.exit(1); + + logger.info("PROV0000 **** AT&T Data Router Provisioning Server starting...."); + + // Get properties + Properties p = (new DB()).getProperties(); + int http_port = Integer.parseInt(p.getProperty("com.att.research.datarouter.provserver.http.port", "8080")); + int https_port = Integer.parseInt(p.getProperty("com.att.research.datarouter.provserver.https.port", "8443")); + + // HTTP connector + SelectChannelConnector http = new SelectChannelConnector(); + http.setPort(http_port); + http.setMaxIdleTime(300000); + http.setRequestHeaderSize(2048); + http.setAcceptors(2); + http.setConfidentialPort(https_port); + http.setLowResourcesConnections(20000); + + // HTTPS connector + SslSelectChannelConnector https = new SslSelectChannelConnector(); + https.setPort(https_port); + https.setMaxIdleTime(30000); + https.setRequestHeaderSize(8192); + https.setAcceptors(2); + + // SSL stuff + SslContextFactory cf = https.getSslContextFactory(); + + /**Skip SSLv3 Fixes*/ + cf.addExcludeProtocols("SSLv3"); + logger.info("Excluded protocols prov-"+cf.getExcludeProtocols()); + /**End of SSLv3 Fixes*/ + + cf.setKeyStoreType(p.getProperty(KEYSTORE_TYPE_PROPERTY, "jks")); + cf.setKeyStorePath(p.getProperty(KEYSTORE_PATH_PROPERTY)); + cf.setKeyStorePassword(p.getProperty(KEYSTORE_PASSWORD_PROPERTY)); + cf.setKeyManagerPassword(p.getProperty("com.att.research.datarouter.provserver.keymanager.password")); + String ts = p.getProperty(TRUSTSTORE_PATH_PROPERTY); + if (ts != null && ts.length() > 0) { + System.out.println("@@ TS -> "+ts); + cf.setTrustStore(ts); + cf.setTrustStorePassword(p.getProperty(TRUSTSTORE_PASSWORD_PROPERTY)); + } else { + cf.setTrustStore(DEFAULT_TRUSTSTORE); + cf.setTrustStorePassword("changeit"); + } + cf.setTrustStore("/opt/app/datartr/self_signed/cacerts.jks"); + cf.setTrustStorePassword("changeit"); + cf.setWantClientAuth(true); + + // Servlet and Filter configuration + ServletContextHandler ctxt = new ServletContextHandler(0); + ctxt.setContextPath("/"); + ctxt.addServlet(new ServletHolder(new FeedServlet()), "/feed/*"); + ctxt.addServlet(new ServletHolder(new FeedLogServlet()), "/feedlog/*"); + ctxt.addServlet(new ServletHolder(new PublishServlet()), "/publish/*"); + ctxt.addServlet(new ServletHolder(new SubscribeServlet()), "/subscribe/*"); + ctxt.addServlet(new ServletHolder(new StatisticsServlet()), "/statistics/*"); + ctxt.addServlet(new ServletHolder(new SubLogServlet()), "/sublog/*"); + ctxt.addServlet(new ServletHolder(new GroupServlet()), "/group/*"); //Provision groups - Rally US708115 -1610 + ctxt.addServlet(new ServletHolder(new SubscriptionServlet()), "/subs/*"); + ctxt.addServlet(new ServletHolder(new InternalServlet()), "/internal/*"); + ctxt.addServlet(new ServletHolder(new RouteServlet()), "/internal/route/*"); + ctxt.addServlet(new ServletHolder(new DRFeedsServlet()), "/"); + ctxt.addFilter (new FilterHolder (new ThrottleFilter()), "/publish/*", FilterMapping.REQUEST); + + ContextHandlerCollection contexts = new ContextHandlerCollection(); + contexts.addHandler(ctxt); + + // Request log configuration + NCSARequestLog nrl = new NCSARequestLog(); + nrl.setFilename(p.getProperty("com.att.research.datarouter.provserver.accesslog.dir") + "/request.log.yyyy_mm_dd"); + nrl.setFilenameDateFormat("yyyyMMdd"); + nrl.setRetainDays(90); + nrl.setAppend(true); + nrl.setExtended(false); + nrl.setLogCookies(false); + nrl.setLogTimeZone("GMT"); + + RequestLogHandler reqlog = new RequestLogHandler(); + reqlog.setRequestLog(nrl); + + // Server's Handler collection + HandlerCollection hc = new HandlerCollection(); + hc.setHandlers(new Handler[] { contexts, new DefaultHandler() }); + hc.addHandler(reqlog); + + // Server's thread pool + QueuedThreadPool pool = new QueuedThreadPool(); + pool.setMinThreads(10); + pool.setMaxThreads(200); + pool.setDetailedDump(false); + + // Daemon to clean up the log directory on a daily basis + Timer rolex = new Timer(); + rolex.scheduleAtFixedRate(new PurgeLogDirTask(), 0, 86400000L); // run once per day + + // Start LogfileLoader + LogfileLoader.getLoader(); + + // The server itself + server = new Server(); + server.setThreadPool(pool); + server.setConnectors(new Connector[] { http, https }); + server.setHandler(hc); + server.setStopAtShutdown(true); + server.setSendServerVersion(true); + server.setSendDateHeader(true); + server.setGracefulShutdown(5000); // allow 5 seconds for servlets to wrap up + server.setDumpAfterStart(false); + server.setDumpBeforeStop(false); + + server.start(); + server.join(); + logger.info("PROV0001 **** AT&T Data Router Provisioning Server halted."); + } + + private static boolean checkDatabase(Logger logger) { + DB db = new DB(); + return db.runRetroFits(); + } + + /** + * Stop the Jetty server. + */ + public static void shutdown() { + new Thread() { + @Override + public void run() { + try { + server.stop(); + Thread.sleep(5000L); + System.exit(0); + } catch (Exception e) { + // ignore + } + } + }.start(); + } +} diff --git a/datarouter-prov/src/main/java/com/att/research/datarouter/provisioning/Poker.java b/datarouter-prov/src/main/java/com/att/research/datarouter/provisioning/Poker.java new file mode 100644 index 00000000..13350df0 --- /dev/null +++ b/datarouter-prov/src/main/java/com/att/research/datarouter/provisioning/Poker.java @@ -0,0 +1,318 @@ +/******************************************************************************* + * ============LICENSE_START================================================== + * * org.onap.dmaap + * * =========================================================================== + * * Copyright © 2017 AT&T Intellectual Property. All rights reserved. + * * =========================================================================== + * * Licensed under the Apache License, Version 2.0 (the "License"); + * * you may not use this file except in compliance with the License. + * * You may obtain a copy of the License at + * * + * * http://www.apache.org/licenses/LICENSE-2.0 + * * + * * Unless required by applicable law or agreed to in writing, software + * * distributed under the License is distributed on an "AS IS" BASIS, + * * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * * See the License for the specific language governing permissions and + * * limitations under the License. + * * ============LICENSE_END==================================================== + * * + * * ECOMP is a trademark and service mark of AT&T Intellectual Property. + * * + ******************************************************************************/ + +package com.att.research.datarouter.provisioning; + +import java.io.FileInputStream; +import java.io.IOException; +import java.io.InputStream; +import java.net.HttpURLConnection; +import java.net.InetAddress; +import java.net.MalformedURLException; +import java.net.URL; +import java.net.UnknownHostException; +import java.util.Arrays; +import java.util.HashSet; +import java.util.Map; +import java.util.Properties; +import java.util.Set; +import java.util.Timer; +import java.util.TimerTask; +import java.util.TreeSet; + +import javax.servlet.ServletException; + +import org.apache.log4j.Logger; +import org.json.JSONException; +import org.json.JSONObject; +import org.json.JSONTokener; + +import com.att.research.datarouter.provisioning.beans.EgressRoute; +import com.att.research.datarouter.provisioning.beans.Feed; +import com.att.research.datarouter.provisioning.beans.IngressRoute; +import com.att.research.datarouter.provisioning.beans.NetworkRoute; +import com.att.research.datarouter.provisioning.beans.Parameters; +import com.att.research.datarouter.provisioning.beans.Subscription; +import com.att.research.datarouter.provisioning.beans.Group; //Groups feature Rally:US708115 - 1610 +import com.att.research.datarouter.provisioning.utils.*; + +/** + * This class handles the two timers (described in R1 Design Notes), and takes care of issuing + * the GET to each node of the URL to "poke". + * + * @author Robert Eby + * @version $Id: Poker.java,v 1.11 2014/01/08 16:13:47 eby Exp $ + */ +public class Poker extends TimerTask { + /** Template used to generate the URL to issue the GET against */ + public static final String POKE_URL_TEMPLATE = "http://%s/internal/fetchProv"; + + + + + /** This is a singleton -- there is only one Poker object in the server */ + private static Poker p; + + /** + * Get the singleton Poker object. + * @return the Poker + */ + public static synchronized Poker getPoker() { + if (p == null) + p = new Poker(); + return p; + } + + private long timer1; + private long timer2; + private Timer rolex; + private String this_pod; // DNS name of this machine + private Logger logger; + private String provstring; + + private Poker() { + timer1 = timer2 = 0; + rolex = new Timer(); + logger = Logger.getLogger("com.att.research.datarouter.provisioning.internal"); + try { + this_pod = InetAddress.getLocalHost().getHostName(); + } catch (UnknownHostException e) { + this_pod = "*UNKNOWN*"; // not a major problem + } + provstring = buildProvisioningString(); + + rolex.scheduleAtFixedRate(this, 0L, 1000L); // Run once a second to check the timers + } + + /** + * This method sets the two timers described in the design notes. + * @param t1 the first timer controls how long to wait after a provisioning request before poking each node + * This timer can be reset if it has not "gone off". + * @param t2 the second timer set the outer bound on how long to wait. It cannot be reset. + */ + public void setTimers(long t1, long t2) { + synchronized (this_pod) { + if (timer1 == 0 || t1 > timer1) + timer1 = t1; + if (timer2 == 0) + timer2 = t2; + } + if (logger.isDebugEnabled()) + logger.debug("Poker timers set to " + timer1 + " and " + timer2); + + + } + + /** + * Return the last provisioning string built. + * @return the last provisioning string built. + */ + public String getProvisioningString() { + return provstring; + } + + /** + * The method to run at the predefined interval (once per second). This method checks + * to see if either of the two timers has expired, and if so, will rebuild the provisioning + * string, and poke all the nodes and other PODs. The timers are then reset to 0. + */ + @Override + public void run() { + try { + if (timer1 > 0) { + long now = System.currentTimeMillis(); + boolean fire = false; + synchronized (this_pod) { + if (now > timer1 || now > timer2) { + timer1 = timer2 = 0; + fire = true; + } + } + if (fire) { + // Rebuild the prov string + provstring = buildProvisioningString(); + + // Only the active POD should poke nodes, etc. + boolean active = SynchronizerTask.getSynchronizer().isActive(); + if (active) { + // Poke all the DR nodes + for (String n : BaseServlet.getNodes()) { + pokeNode(n); + } + // Poke the pod that is not us + for (String n : BaseServlet.getPods()) { + if (n.length() > 0 && !n.equals(this_pod)) + pokeNode(n); + } + } + } + } + } catch (Exception e) { + logger.warn("PROV0020: Caught exception in Poker: "+e); + e.printStackTrace(); + } + } + private void pokeNode(final String nodename) { + logger.debug("PROV0012 Poking node " + nodename + " ..."); + Runnable r = new Runnable() { + @Override + public void run() { + + try { + String u = String.format(POKE_URL_TEMPLATE, nodename+":"+DB.HTTP_PORT); + URL url = new URL(u); + HttpURLConnection conn = (HttpURLConnection) url.openConnection(); + conn.setConnectTimeout(60000); //Fixes for Itrack DATARTR-3, poke timeout + conn.connect(); + conn.getContentLength(); // Force the GET through + conn.disconnect(); + } catch (MalformedURLException e) { + logger.warn("PROV0013 MalformedURLException Error poking node "+nodename+": " + e.getMessage()); + } catch (IOException e) { + logger.warn("PROV0013 IOException Error poking node "+nodename+": " + e.getMessage()); + } + } + }; +// Thread t = new Thread(r); +// t.start(); + r.run(); + } + @SuppressWarnings("unused") + private String buildProvisioningString() { + StringBuilder sb = new StringBuilder("{\n"); + + // Append Feeds to the string + String pfx = "\n"; + sb.append("\"feeds\": ["); + for (Feed f : Feed.getAllFeeds()) { + sb.append(pfx); + sb.append(f.asJSONObject().toString()); + pfx = ",\n"; + } + sb.append("\n],\n"); + + //Append groups to the string - Rally:US708115 - 1610 + pfx = "\n"; + sb.append("\"groups\": ["); + for (Group s : Group.getAllgroups()) { + sb.append(pfx); + sb.append(s.asJSONObject().toString()); + pfx = ",\n"; + } + sb.append("\n],\n"); + + + // Append Subscriptions to the string + pfx = "\n"; + sb.append("\"subscriptions\": ["); + for (Subscription s : Subscription.getAllSubscriptions()) { + sb.append(pfx); + if(s!=null) + sb.append(s.asJSONObject().toString()); + pfx = ",\n"; + } + sb.append("\n],\n"); + + // Append Parameters to the string + pfx = "\n"; + sb.append("\"parameters\": {"); + Map+ * This servlet handles requests to URLs under /internal/route/ on the provisioning server. + * This part of the URL tree is used to manipulate the Data Router routing tables. + * These include: + *
+ *URL Path | + *Method | + *Purpose | + *
---|---|---|
/internal/route/ | + *GET | + *used to GET a full JSON copy of all three routing tables. | + *
/internal/route/ingress/ | + *GET | + *used to GET a full JSON copy of the ingress routing table (IRT). | + *
POST | + *used to create a new entry in the ingress routing table (IRT). | |
/internal/route/egress/ | + *GET | + *used to GET a full JSON copy of the egress routing table (ERT). | + *
POST | + *used to create a new entry in the egress routing table (ERT). | |
/internal/route/network/ | + *GET | + *used to GET a full JSON copy of the network routing table (NRT). | + *
POST | + *used to create a new entry in the network routing table (NRT). | + *|
/internal/route/ingress/<feed>/<user>/<subnet> | + *DELETE | + *used to DELETE the ingress route corresponding to feed, user and subnet. + * The / in the subnet specified should be replaced with a !, since / cannot be used in a URL. | + *
/internal/route/ingress/<seq> | + *DELETE | + *used to DELETE all ingress routes with the matching seq sequence number. | + *
/internal/route/egress/<sub> | + *DELETE | + *used to DELETE the egress route the matching sub subscriber number. | + *
/internal/route/network/<fromnode>/<tonode> | + *DELETE | + *used to DELETE the network route corresponding to fromnode + * and tonode. | + *
+ * Authorization to use these URLs is a little different than for other URLs on the provisioning server. + * For the most part, the IP address that the request comes from should be either: + *
+ *+ * All DELETE/GET/POST requests made to this servlet on the standby server are proxied to the + * active server (using the {@link ProxyServlet}) if it is up and reachable. + *
+ * + * @author Robert Eby + * @version $Id$ + */ +@SuppressWarnings("serial") +public class RouteServlet extends ProxyServlet { + /** + * DELETE route table entries by deleting part of the route table tree. + */ + @Override + public void doDelete(HttpServletRequest req, HttpServletResponse resp) throws IOException { + EventLogRecord elr = new EventLogRecord(req); + if (!isAuthorizedForInternal(req)) { + elr.setMessage("Unauthorized."); + elr.setResult(HttpServletResponse.SC_FORBIDDEN); + eventlogger.info(elr); + resp.sendError(HttpServletResponse.SC_FORBIDDEN, "Unauthorized."); + return; + } + if (isProxyOK(req) && isProxyServer()) { + super.doDelete(req, resp); + return; + } + + String path = req.getPathInfo(); + String[] parts = path.substring(1).split("/"); + Deleteable[] d = null; + if (parts[0].equals("ingress")) { + if (parts.length == 4) { + // /internal/route/ingress/For this to work correctly, the following code needs to be placed at the beginning of main().
+ *
+ * Security.setProperty("networkaddress.cache.ttl", "10");
+ *
+ *
+ * @author Robert Eby
+ * @version $Id: SynchronizerTask.java,v 1.10 2014/03/21 13:50:10 eby Exp $
+ */
+public class SynchronizerTask extends TimerTask {
+ /** This is a singleton -- there is only one SynchronizerTask object in the server */
+ private static SynchronizerTask synctask;
+
+ /** This POD is unknown -- not on the list of PODs */
+ public static final int UNKNOWN = 0;
+ /** This POD is active -- on the list of PODs, and the DNS CNAME points to us */
+ public static final int ACTIVE = 1;
+ /** This POD is standby -- on the list of PODs, and the DNS CNAME does not point to us */
+ public static final int STANDBY = 2;
+ private static final String[] stnames = { "UNKNOWN", "ACTIVE", "STANDBY" };
+ private static final long ONE_HOUR = 60 * 60 * 1000L;
+
+ private final Logger logger;
+ private final Timer rolex;
+ private final String spooldir;
+ private int state;
+ private boolean doFetch;
+ private long nextsynctime;
+ private AbstractHttpClient httpclient = null;
+
+ /**
+ * Get the singleton SynchronizerTask object.
+ * @return the SynchronizerTask
+ */
+ public static synchronized SynchronizerTask getSynchronizer() {
+ if (synctask == null)
+ synctask = new SynchronizerTask();
+ return synctask;
+ }
+
+ @SuppressWarnings("deprecation")
+ private SynchronizerTask() {
+ logger = Logger.getLogger("com.att.research.datarouter.provisioning.internal");
+ rolex = new Timer();
+ spooldir = (new DB()).getProperties().getProperty("com.att.research.datarouter.provserver.spooldir");
+ state = UNKNOWN;
+ doFetch = true; // start off with a fetch
+ nextsynctime = 0;
+
+ logger.info("PROV5000: Sync task starting, server state is UNKNOWN");
+ try {
+ Properties props = (new DB()).getProperties();
+ String type = props.getProperty(Main.KEYSTORE_TYPE_PROPERTY, "jks");
+ String store = props.getProperty(Main.KEYSTORE_PATH_PROPERTY);
+ String pass = props.getProperty(Main.KEYSTORE_PASSWORD_PROPERTY);
+ KeyStore keyStore = KeyStore.getInstance(type);
+ FileInputStream instream = new FileInputStream(new File(store));
+ keyStore.load(instream, pass.toCharArray());
+ instream.close();
+
+ store = props.getProperty(Main.TRUSTSTORE_PATH_PROPERTY);
+ pass = props.getProperty(Main.TRUSTSTORE_PASSWORD_PROPERTY);
+ KeyStore trustStore = null;
+ if (store != null && store.length() > 0) {
+ trustStore = KeyStore.getInstance(KeyStore.getDefaultType());
+ instream = new FileInputStream(new File(store));
+ trustStore.load(instream, pass.toCharArray());
+ instream.close();
+ }
+
+ // We are connecting with the node name, but the certificate will have the CNAME
+ // So we need to accept a non-matching certificate name
+ String keystorepass = props.getProperty(Main.KEYSTORE_PASSWORD_PROPERTY); //itrack.web.att.com/browse/DATARTR-6 for changing hard coded passphase ref
+ AbstractHttpClient hc = new DefaultHttpClient();
+ SSLSocketFactory socketFactory =
+ (trustStore == null)
+ ? new SSLSocketFactory(keyStore, keystorepass)
+ : new SSLSocketFactory(keyStore, keystorepass, trustStore);
+ socketFactory.setHostnameVerifier(SSLSocketFactory.ALLOW_ALL_HOSTNAME_VERIFIER);
+ Scheme sch = new Scheme("https", 443, socketFactory);
+ hc.getConnectionManager().getSchemeRegistry().register(sch);
+ httpclient = hc;
+
+ // Run once every 5 seconds to check DNS, etc.
+ long interval = 0;
+ try {
+ String s = props.getProperty("com.att.research.datarouter.provserver.sync_interval", "5000");
+ interval = Long.parseLong(s);
+ } catch (NumberFormatException e) {
+ interval = 5000L;
+ }
+ rolex.scheduleAtFixedRate(this, 0L, interval);
+ } catch (Exception e) {
+ logger.warn("PROV5005: Problem starting the synchronizer: "+e);
+ }
+ }
+
+ /**
+ * What is the state of this POD?
+ * @return one of ACTIVE, STANDBY, UNKNOWN
+ */
+ public int getState() {
+ return state;
+ }
+
+ /**
+ * Is this the active POD?
+ * @return true if we are active (the master), false otherwise
+ */
+ public boolean isActive() {
+ return state == ACTIVE;
+ }
+
+ /**
+ * This method is used to signal that another POD (the active POD) has sent us a /fetchProv request,
+ * and that we should re-synchronize with the master.
+ */
+ public void doFetch() {
+ doFetch = true;
+ }
+
+ /**
+ * Runs once a minute in order to +This package provides beans to represent the basic provisioning objects of the Data Router application. +These objects are defined by the document Data Router Release 1 Provisioning API Version 1.2. +
+ + diff --git a/datarouter-prov/src/main/java/com/att/research/datarouter/provisioning/eelf/EelfMsgs.java b/datarouter-prov/src/main/java/com/att/research/datarouter/provisioning/eelf/EelfMsgs.java new file mode 100644 index 00000000..3a230416 --- /dev/null +++ b/datarouter-prov/src/main/java/com/att/research/datarouter/provisioning/eelf/EelfMsgs.java @@ -0,0 +1,56 @@ +/******************************************************************************* + * ============LICENSE_START================================================== + * * org.onap.dmaap + * * =========================================================================== + * * Copyright © 2017 AT&T Intellectual Property. All rights reserved. + * * =========================================================================== + * * Licensed under the Apache License, Version 2.0 (the "License"); + * * you may not use this file except in compliance with the License. + * * You may obtain a copy of the License at + * * + * * http://www.apache.org/licenses/LICENSE-2.0 + * * + * * Unless required by applicable law or agreed to in writing, software + * * distributed under the License is distributed on an "AS IS" BASIS, + * * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * * See the License for the specific language governing permissions and + * * limitations under the License. + * * ============LICENSE_END==================================================== + * * + * * ECOMP is a trademark and service mark of AT&T Intellectual Property. + * * + ******************************************************************************/ +package com.att.research.datarouter.provisioning.eelf; + +import com.att.eelf.i18n.EELFResolvableErrorEnum; +import com.att.eelf.i18n.EELFResourceManager; + +public enum EelfMsgs implements EELFResolvableErrorEnum { + + /** + * Application message prints user (accepts one argument) + */ + MESSAGE_WITH_BEHALF, + + /** + * Application message prints user and FeedID (accepts two arguments) + */ + + MESSAGE_WITH_BEHALF_AND_FEEDID, + + /** + * Application message prints user and SUBID (accepts two arguments) + */ + + MESSAGE_WITH_BEHALF_AND_SUBID; + + + + /** + * Static initializer to ensure the resource bundles for this class are loaded... + * Here this application loads messages from three bundles + */ + static { + EELFResourceManager.loadMessageBundle("EelfMessages"); + } +} diff --git a/datarouter-prov/src/main/java/com/att/research/datarouter/provisioning/eelf/JettyFilter.java b/datarouter-prov/src/main/java/com/att/research/datarouter/provisioning/eelf/JettyFilter.java new file mode 100644 index 00000000..cfef910c --- /dev/null +++ b/datarouter-prov/src/main/java/com/att/research/datarouter/provisioning/eelf/JettyFilter.java @@ -0,0 +1,38 @@ +/******************************************************************************* + * ============LICENSE_START================================================== + * * org.onap.dmaap + * * =========================================================================== + * * Copyright © 2017 AT&T Intellectual Property. All rights reserved. + * * =========================================================================== + * * Licensed under the Apache License, Version 2.0 (the "License"); + * * you may not use this file except in compliance with the License. + * * You may obtain a copy of the License at + * * + * * http://www.apache.org/licenses/LICENSE-2.0 + * * + * * Unless required by applicable law or agreed to in writing, software + * * distributed under the License is distributed on an "AS IS" BASIS, + * * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * * See the License for the specific language governing permissions and + * * limitations under the License. + * * ============LICENSE_END==================================================== + * * + * * ECOMP is a trademark and service mark of AT&T Intellectual Property. + * * + ******************************************************************************/ +package com.att.research.datarouter.provisioning.eelf; + +import ch.qos.logback.classic.spi.ILoggingEvent; +import ch.qos.logback.core.filter.Filter; +import ch.qos.logback.core.spi.FilterReply; + +public class JettyFilter extends Filter+This package provides the servlets used by the provisioning server for the Data Router application. +URLs are from the document URLs for DR Release 1 Version 1.2. +
+URL Path | +Symbolic Name | +Servlet Name | +Allowed Methods | +|||
---|---|---|---|---|---|---|
/ | +<drFeedsUrl> | +{@link com.att.research.datarouter.provisioning.DRFeedsServlet} | +DELETE | +GET | +POST | +PUT | +
/feed/feedid | +<feedUrl> | +{@link com.att.research.datarouter.provisioning.FeedServlet} | +DELETE | +GET | +POST | +PUT | +
/publish/feedid | +<publishUrl> | +{@link com.att.research.datarouter.provisioning.PublishServlet} | +DELETE | +GET | +POST | +PUT | +
/subscribe/feedid | +<subscribeUrl> | +{@link com.att.research.datarouter.provisioning.SubscribeServlet} | +DELETE | +GET | +POST | +PUT | +
/feedlog/feedid | +<feedLogUrl> | +{@link com.att.research.datarouter.provisioning.FeedLogServlet} | +DELETE | +GET | +POST | +PUT | +
/subs/subid | +<subscriptionUrl> | +{@link com.att.research.datarouter.provisioning.SubscriptionServlet} | +DELETE | +GET | +POST | +PUT | +
/sublog/subid | +<subLogUrl> | +{@link com.att.research.datarouter.provisioning.SubLogServlet} | +DELETE | +GET | +POST | +PUT | +
/internal/* | +<internalUrl> | +{@link com.att.research.datarouter.provisioning.InternalServlet} | +DELETE | +GET | +POST | +PUT | +
/internal/route/* | +<routeUrl> | +{@link com.att.research.datarouter.provisioning.RouteServlet} | +DELETE | +GET | +POST | +PUT | +
+ * java com.att.research.datarouter.provisioning.utils.DRRouteCLI [ -s server ] [ command ] + *+ * A full description of the arguments to this command are + * here. + * + * @param args command line arguments + * @throws Exception for any unrecoverable problem + */ + public static void main(String[] args) throws Exception { + String server = System.getenv(ENV_VAR); + if (args.length >= 2 && args[0].equals("-s")) { + server = args[1]; + String[] t = new String[args.length-2]; + if (t.length > 0) + System.arraycopy(args, 2, t, 0, t.length); + args = t; + } + if (server == null || server.equals("")) { + System.err.println("dr-route: you need to specify a server, either via $PROVSRVR or the '-s' option."); + System.exit(1); + } + DRRouteCLI cli = new DRRouteCLI(server); + if (args.length > 0) { + boolean b = cli.runCommand(args); + System.exit(b ? 0 : 1); + } else { + cli.interactive(); + System.exit(0); + } + } + + public static final String ENV_VAR = "PROVSRVR"; + public static final String PROMPT = "dr-route> "; + public static final String DEFAULT_TRUSTSTORE_PATH = /* $JAVA_HOME + */ "/jre/lib/security/cacerts"; + + private final String server; + private int width = 120; // screen width (for list) + private AbstractHttpClient httpclient; + + /** + * Create a DRRouteCLI object connecting to the specified server. + * @param server the server to send command to + * @throws Exception + */ + public DRRouteCLI(String server) throws Exception { + this.server = server; + this.width = 120; + this.httpclient = new DefaultHttpClient(); + + Properties p = (new DB()).getProperties(); + String truststore_file = p.getProperty("com.att.research.datarouter.provserver.truststore.path"); + String truststore_pw = p.getProperty("com.att.research.datarouter.provserver.truststore.password"); + + KeyStore trustStore = KeyStore.getInstance(KeyStore.getDefaultType()); + if (truststore_file == null || truststore_file.equals("")) { + String jhome = System.getenv("JAVA_HOME"); + if (jhome == null || jhome.equals("")) + jhome = "/opt/java/jdk/jdk180"; + truststore_file = jhome + DEFAULT_TRUSTSTORE_PATH; + } + File f = new File(truststore_file); + if (f.exists()) { + FileInputStream instream = new FileInputStream(f); + try { + trustStore.load(instream, truststore_pw.toCharArray()); + } catch (Exception x) { + System.err.println("Problem reading truststore: "+x); + throw x; + } finally { + try { instream.close(); } catch (Exception ignore) {} + } + } + + SSLSocketFactory socketFactory = new SSLSocketFactory(trustStore); + Scheme sch = new Scheme("https", 443, socketFactory); + httpclient.getConnectionManager().getSchemeRegistry().register(sch); + } + + private void interactive() throws IOException { + LineNumberReader in = new LineNumberReader(new InputStreamReader(System.in)); + while (true) { + System.out.print(PROMPT); + String line = in.readLine(); + if (line == null) + return; + line = line.trim(); + if (line.equalsIgnoreCase("exit")) // "exit" may only be used in interactive mode + return; + if (line.equalsIgnoreCase("quit")) // "quit" may only be used in interactive mode + return; + String[] args = line.split("[ \t]+"); + if (args.length > 0) + runCommand(args); + } + } + + /** + * Run the command specified by the arguments. + * @param args The command line arguments. + * @return true if the command was valid and succeeded + */ + public boolean runCommand(String[] args) { + String cmd = args[0].trim().toLowerCase(); + if (cmd.equals("add")) { + if (args.length > 2) { + if (args[1].startsWith("in") && args.length >= 6) { + return addIngress(args); + } + if (args[1].startsWith("eg") && args.length == 4) { + return addEgress(args); + } + if (args[1].startsWith("ne") && args.length == 5) { + return addRoute(args); + } + } + System.err.println("Add command should be one of:"); + System.err.println(" add in[gress] feedid user subnet nodepatt [ seq ]"); + System.err.println(" add eg[ress] subid node"); + System.err.println(" add ne[twork] fromnode tonode vianode"); + } else if (cmd.startsWith("del")) { + if (args.length > 2) { + if (args[1].startsWith("in") && args.length == 5) { + return delIngress(args); + } + if (args[1].startsWith("in") && args.length == 3) { + return delIngress(args); + } + if (args[1].startsWith("eg") && args.length == 3) { + return delEgress(args); + } + if (args[1].startsWith("ne") && args.length == 4) { + return delRoute(args); + } + } + System.err.println("Delete command should be one of:"); + System.err.println(" del in[gress] feedid user subnet"); + System.err.println(" del in[gress] seq"); + System.err.println(" del eg[ress] subid"); + System.err.println(" del ne[twork] fromnode tonode"); + } else if (cmd.startsWith("lis")) { + return list(args); + } else if (cmd.startsWith("wid") && args.length > 1) { + width = Integer.parseInt(args[1]); + return true; + } else if (cmd.startsWith("?") || cmd.startsWith("hel") || cmd.startsWith("usa")) { + usage(); + } else if (cmd.startsWith("#")) { + // comment -- ignore + } else { + System.err.println("Command should be one of add, del, list, exit, quit"); + } + return false; + } + + private void usage() { + System.out.println("Enter one of the following commands:"); + System.out.println(" add in[gress] feedid user subnet nodepatt [ seq ]"); + System.out.println(" add eg[ress] subid node"); + System.out.println(" add ne[twork] fromnode tonode vianode"); + System.out.println(" del in[gress] feedid user subnet"); + System.out.println(" del in[gress] seq"); + System.out.println(" del eg[ress] subid"); + System.out.println(" del ne[twork] fromnode tonode"); + System.out.println(" list [ all | ingress | egress | network ]"); + System.out.println(" exit"); + System.out.println(" quit"); + } + + private boolean addIngress(String[] args) { + String url = String.format("https://%s/internal/route/ingress/?feed=%s&user=%s&subnet=%s&nodepatt=%s", server, args[2], args[3], args[4], args[5]); + if (args.length > 6) + url += "&seq=" + args[6]; + return doPost(url); + } + + private boolean addEgress(String[] args) { + String url = String.format("https://%s/internal/route/egress/?sub=%s&node=%s", server, args[2], args[3]); + return doPost(url); + } + + private boolean addRoute(String[] args) { + String url = String.format("https://%s/internal/route/network/?from=%s&to=%s&via=%s", server, args[2], args[3], args[4]); + return doPost(url); + } + + private boolean delIngress(String[] args) { + String url; + if (args.length == 5) { + String subnet = args[4].replaceAll("/", "!"); // replace the / with a ! + url = String.format("https://%s/internal/route/ingress/%s/%s/%s", server, args[2], args[3], subnet); + } else { + url = String.format("https://%s/internal/route/ingress/%s", server, args[2]); + } + return doDelete(url); + } + + private boolean delEgress(String[] args) { + String url = String.format("https://%s/internal/route/egress/%s", server, args[2]); + return doDelete(url); + } + + private boolean delRoute(String[] args) { + String url = String.format("https://%s/internal/route/network/%s/%s", server, args[2], args[3]); + return doDelete(url); + } + + private boolean list(String[] args) { + String tbl = (args.length == 1) ? "all" : args[1].toLowerCase(); + JSONObject jo = doGet("https://"+server+"/internal/route/"); // Returns all 3 tables + StringBuilder sb = new StringBuilder(); + if (tbl.startsWith("al") || tbl.startsWith("in")) { + // Display the IRT + JSONArray irt = jo.optJSONArray("ingress"); + int cw1 = 6, cw2 = 6, cw3 = 6, cw4 = 6; // determine column widths for first 4 cols + for (int i = 0; irt != null && i < irt.length(); i++) { + JSONObject e = irt.getJSONObject(i); + cw1 = Math.max(cw1, (""+ e.getInt("seq")).length()); + cw2 = Math.max(cw2, (""+e.getInt("feedid")).length()); + String t = e.optString("user"); + cw3 = Math.max(cw3, (t == null) ? 1 : t.length()); + t = e.optString("subnet"); + cw4 = Math.max(cw4, (t == null) ? 1 : t.length()); + } + + int nblank = cw1 + cw2 + cw3 + cw4 + 8; + sb.append("Ingress Routing Table\n"); + sb.append(String.format("%s %s %s %s Nodes\n", ext("Seq", cw1), ext("FeedID", cw2), ext("User", cw3), ext("Subnet", cw4))); + for (int i = 0; irt != null && i < irt.length(); i++) { + JSONObject e = irt.getJSONObject(i); + String seq = ""+e.getInt("seq"); + String feedid = ""+e.getInt("feedid"); + String user = e.optString("user"); + String subnet = e.optString("subnet"); + if (user.equals("")) user = "-"; + if (subnet.equals("")) subnet = "-"; + JSONArray nodes = e.getJSONArray("node"); + int sol = sb.length(); + sb.append(String.format("%s %s %s %s ", ext(seq, cw1), ext(feedid, cw2), ext(user, cw3), ext(subnet, cw4))); + for (int j = 0; j < nodes.length(); j++) { + String nd = nodes.getString(j); + int cursor = sb.length() - sol; + if (j > 0 && (cursor + nd.length() > width)) { + sb.append("\n"); + sol = sb.length(); + sb.append(ext(" ", nblank)); + } + sb.append(nd); + if ((j+1) < nodes.length()) { + sb.append(", "); + } + } + sb.append("\n"); + } + } + if (tbl.startsWith("al") || tbl.startsWith("eg")) { + // Display the ERT + JSONObject ert = jo.optJSONObject("egress"); + String[] subs = (ert == null) ? new String[0] : JSONObject.getNames(ert); + if (subs == null) + subs = new String[0]; + Arrays.sort(subs); + int cw1 = 5; + for (int i = 0; i < subs.length; i++) { + cw1 = Math.max(cw1, subs[i].length()); + } + + if (sb.length() > 0) + sb.append("\n"); + sb.append("Egress Routing Table\n"); + sb.append(String.format("%s Node\n", ext("SubID", cw1))); + for (int i = 0; i < subs.length; i++) { + String node = ert.getString(subs[i]); + sb.append(String.format("%s %s\n", ext(subs[i], cw1), node)); + } + } + if (tbl.startsWith("al") || tbl.startsWith("ne")) { + // Display the NRT + JSONArray nrt = jo.optJSONArray("routing"); + int cw1 = 4, cw2 = 4; + for (int i = 0; nrt != null && i < nrt.length(); i++) { + JSONObject e = nrt.getJSONObject(i); + String from = e.getString("from"); + String to = e.getString("to"); + cw1 = Math.max(cw1, from.length()); + cw2 = Math.max(cw2, to.length()); + } + + if (sb.length() > 0) + sb.append("\n"); + sb.append("Network Routing Table\n"); + sb.append(String.format("%s %s Via\n", ext("From", cw1), ext("To", cw2))); + for (int i = 0; nrt != null && i < nrt.length(); i++) { + JSONObject e = nrt.getJSONObject(i); + String from = e.getString("from"); + String to = e.getString("to"); + String via = e.getString("via"); + sb.append(String.format("%s %s %s\n", ext(from, cw1), ext(to, cw2), via)); + } + } + System.out.print(sb.toString()); + return true; + } + private String ext(String s, int n) { + if (s == null) + s = "-"; + while (s.length() < n) + s += " "; + return s; + } + + private boolean doDelete(String url) { + boolean rv = false; + HttpDelete meth = new HttpDelete(url); + try { + HttpResponse response = httpclient.execute(meth); + HttpEntity entity = response.getEntity(); + StatusLine sl = response.getStatusLine(); + rv = (sl.getStatusCode() == HttpServletResponse.SC_OK); + if (rv) { + System.out.println("Routing entry deleted."); + EntityUtils.consume(entity); + } else { + printErrorText(entity); + } + } catch (Exception e) { + } finally { + meth.releaseConnection(); + } + return rv; + } + + private JSONObject doGet(String url) { + JSONObject rv = new JSONObject(); + HttpGet meth = new HttpGet(url); + try { + HttpResponse response = httpclient.execute(meth); + HttpEntity entity = response.getEntity(); + StatusLine sl = response.getStatusLine(); + if (sl.getStatusCode() == HttpServletResponse.SC_OK) { + rv = new JSONObject(new JSONTokener(entity.getContent())); + } else { + printErrorText(entity); + } + } catch (Exception e) { + System.err.println(e); + } finally { + meth.releaseConnection(); + } + return rv; + } + + private boolean doPost(String url) { + boolean rv = false; + HttpPost meth = new HttpPost(url); + try { + HttpResponse response = httpclient.execute(meth); + HttpEntity entity = response.getEntity(); + StatusLine sl = response.getStatusLine(); + rv = (sl.getStatusCode() == HttpServletResponse.SC_OK); + if (rv) { + System.out.println("Routing entry added."); + EntityUtils.consume(entity); + } else { + printErrorText(entity); + } + } catch (Exception e) { + } finally { + meth.releaseConnection(); + } + return rv; + } + + private void printErrorText(HttpEntity entity) throws IllegalStateException, IOException { + // Look for and print only the part of the output between
...+ InputStream is = entity.getContent(); + StringBuilder sb = new StringBuilder(); + byte[] b = new byte[512]; + int n = 0; + while ((n = is.read(b)) > 0) { + sb.append(new String(b, 0, n)); + } + is.close(); + int ix = sb.indexOf("
"); + if (ix > 0) + sb.delete(0, ix+5); + ix = sb.indexOf(""); + if (ix > 0) + sb.delete(ix, sb.length()); + System.err.println(sb.toString()); + } +} diff --git a/datarouter-prov/src/main/java/com/att/research/datarouter/provisioning/utils/JSONUtilities.java b/datarouter-prov/src/main/java/com/att/research/datarouter/provisioning/utils/JSONUtilities.java new file mode 100644 index 00000000..e1676588 --- /dev/null +++ b/datarouter-prov/src/main/java/com/att/research/datarouter/provisioning/utils/JSONUtilities.java @@ -0,0 +1,76 @@ +/******************************************************************************* + * ============LICENSE_START================================================== + * * org.onap.dmaap + * * =========================================================================== + * * Copyright © 2017 AT&T Intellectual Property. All rights reserved. + * * =========================================================================== + * * Licensed under the Apache License, Version 2.0 (the "License"); + * * you may not use this file except in compliance with the License. + * * You may obtain a copy of the License at + * * + * * http://www.apache.org/licenses/LICENSE-2.0 + * * + * * Unless required by applicable law or agreed to in writing, software + * * distributed under the License is distributed on an "AS IS" BASIS, + * * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * * See the License for the specific language governing permissions and + * * limitations under the License. + * * ============LICENSE_END==================================================== + * * + * * ECOMP is a trademark and service mark of AT&T Intellectual Property. + * * + ******************************************************************************/ + + +package com.att.research.datarouter.provisioning.utils; + +import java.net.InetAddress; +import java.net.UnknownHostException; +import java.util.Collection; + +/** + * Some utility functions used when creating/validating JSON. + * + * @author Robert Eby + * @version $Id: JSONUtilities.java,v 1.1 2013/04/26 21:00:26 eby Exp $ + */ +public class JSONUtilities { + /** + * Does the String v represent a valid Internet address (with or without a + * mask length appended). + * @param v the string to check + * @return true if valid, false otherwise + */ + public static boolean validIPAddrOrSubnet(String v) { + String[] pp = { v, "" }; + if (v.indexOf('/') > 0) + pp = v.split("/"); + try { + InetAddress addr = InetAddress.getByName(pp[0]); + if (pp[1].length() > 0) { + // check subnet mask + int mask = Integer.parseInt(pp[1]); + if (mask > (addr.getAddress().length * 8)) + return false; + } + return true; + } catch (UnknownHostException e) { + return false; + } + } + /** + * Build a JSON array from a collection of Strings. + * @param coll the collection + * @return a String containing a JSON array + */ + public static String createJSONArray(Collection
This method maintains an {@link RLEBitSet} which can be used to easily see what records are presently in the + * database. + * This bit set is used to synchronize between provisioning servers.
+ * + * @author Robert Eby + * @version $Id: LogfileLoader.java,v 1.22 2014/03/12 19:45:41 eby Exp $ + */ +public class LogfileLoader extends Thread { + /** Default number of log records to keep when pruning. Keep 10M by default. */ + public static final long DEFAULT_LOG_RETENTION = 10000000L; + /** NOT USED: Percentage of free space required before old records are removed. */ + public static final int REQUIRED_FREE_PCT = 20; + + /** This is a singleton -- there is only one LogfileLoader object in the server */ + private static LogfileLoader p; + + /** + * Get the singleton LogfileLoader object, and start it if it is not running. + * @return the LogfileLoader + */ + public static synchronized LogfileLoader getLoader() { + if (p == null) + p = new LogfileLoader(); + if (!p.isAlive()) + p.start(); + return p; + } + + /** The PreparedStatement which is loaded by a Loadable. */ + public static final String INSERT_SQL = "insert into LOG_RECORDS values (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)"; + /** Each server can assign this many IDs */ + private static final long SET_SIZE = (1L << 56); + + private final Logger logger; + private final DB db; + private final String spooldir; + private final long set_start; + private final long set_end; + private RLEBitSet seq_set; + private long nextid; + private boolean idle; + + private LogfileLoader() { + this.logger = Logger.getLogger("com.att.research.datarouter.provisioning.internal"); + this.db = new DB(); + this.spooldir = db.getProperties().getProperty("com.att.research.datarouter.provserver.spooldir"); + this.set_start = getIdRange(); + this.set_end = set_start + SET_SIZE - 1; + this.seq_set = new RLEBitSet(); + this.nextid = 0; + this.idle = false; + + // This is a potentially lengthy operation, so has been moved to run() + //initializeNextid(); + this.setDaemon(true); + this.setName("LogfileLoader"); + } + + private long getIdRange() { + long n; + if (BaseServlet.isInitialActivePOD()) + n = 0; + else if (BaseServlet.isInitialStandbyPOD()) + n = SET_SIZE; + else + n = SET_SIZE * 2; + String r = String.format("[%X .. %X]", n, n+SET_SIZE-1); + logger.debug("This server shall assign RECORD_IDs in the range "+r); + return n; + } + /** + * Return the bit set representing the record ID's that are loaded in this database. + * @return the bit set + */ + public RLEBitSet getBitSet() { + return seq_set; + } + /** + * True if the LogfileLoader is currently waiting for work. + * @return true if idle + */ + public boolean isIdle() { + return idle; + } + /** + * Run continuously to look for new logfiles in the spool directory and import them into the DB. + * The spool is checked once per second. If free space on the MySQL filesystem falls below + * REQUIRED_FREE_PCT (normally 20%) then the oldest logfile entries are removed and the LOG_RECORDS + * table is compacted until free space rises above the threshold. + */ + @Override + public void run() { + initializeNextid(); // moved from the constructor + while (true) { + try { + File dirfile = new File(spooldir); + while (true) { + // process IN files + File[] infiles = dirfile.listFiles(new FilenameFilter() { + @Override + public boolean accept(File dir, String name) { + return name.startsWith("IN."); + } + }); + + if (infiles.length == 0) { + idle = true; + try { + Thread.sleep(1000L); + } catch (InterruptedException e) { + } + idle = false; + } else { + // Remove old rows + if (pruneRecords()) { + // Removed at least some entries, recompute the bit map + initializeNextid(); + } + + // Process incoming logfiles + for (File f : infiles) { + if (logger.isDebugEnabled()) + logger.debug("PROV8001 Starting " + f + " ..."); + long time = System.currentTimeMillis(); + int[] n = process(f); + time = System.currentTimeMillis() - time; + logger.info(String + .format("PROV8000 Processed %s in %d ms; %d of %d records.", + f.toString(), time, n[0], n[1])); + f.delete(); + } + } + } + } catch (Exception e) { + logger.warn("PROV0020: Caught exception in LogfileLoader: " + e); + e.printStackTrace(); + } + } + } + private boolean pruneRecords() { + boolean did1 = false; + long count = countRecords(); + long threshold = DEFAULT_LOG_RETENTION; + Parameters param = Parameters.getParameter(Parameters.PROV_LOG_RETENTION); + if (param != null) { + try { + long n = Long.parseLong(param.getValue()); + // This check is to prevent inadvertent errors from wiping the table out + if (n > 1000000L) + threshold = n; + } catch (NumberFormatException e) { + // ignore + } + } + logger.debug("Pruning LOG_RECORD table: records in DB="+count+", threshold="+threshold); + if (count > threshold) { + count -= threshold; // we need to remove this many records; + Maplong
s rather
+ * than int
s to represent the indices of the bits.
+ *
+ * @author Robert Eby
+ * @version $Id$
+ */
+public class RLEBitSet {
+ /**
+ * Used to represent a continues set of nbits 1 bits starting at start.
+ */
+ private class RLE implements Comparablefalse
.
+ */
+ public RLEBitSet() {
+ bitsets = new TreeSets
.
+ * @param s the initialization String
+ */
+ public RLEBitSet(String s) {
+ bitsets = new TreeSets
.
+ * @param s the initialization String, which consists of a comma or space separated list of
+ * non-negative numbers and ranges. An individual number represents the bit index to set.
+ * A range (two numbers separated by a dash) causes all bit indexes between the two numbers
+ * (inclusive) to be set.
+ * @exception NumberFormatException - if a number is incorrectly formatted
+ * @exception IndexOutOfBoundsException - if an index is negative
+ */
+ public void set(String s) throws NumberFormatException {
+ s = s.trim();
+ if (!s.isEmpty()) {
+ for (String s2 : s.split("[, \n]+")) {
+ if (s2.indexOf('-') >= 0) {
+ String[] pp = s2.split("-");
+ long f = Long.parseLong(pp[0]);
+ long t = Long.parseLong(pp[1]);
+ set(f, t+1);
+ } else
+ set(Long.parseLong(s2));
+ }
+ }
+ }
+ /**
+ * Sets the bit at the specified index to {@code true}.
+ * @param bit a bit index
+ */
+ public void set(long bit) {
+ set(bit, bit+1);
+ }
+ /**
+ * Sets the bits from the specified {@code from} (inclusive) to the
+ * specified {@code to} (exclusive) to {@code true}.
+ * @param from index of the first bit to be set
+ * @param to index after the last bit to be set
+ * @throws IndexOutOfBoundsException if {@code from} is negative,
+ * or {@code to} is negative,
+ * or {@code from} is larger than {@code to}
+ */
+ public void set(long from, long to) {
+ checkRange(from, to);
+ RLE newbits = new RLE(from, to-from);
+ synchronized (bitsets) {
+ for (RLE bs : bitsets) {
+ if (bs.intersects(newbits)) {
+ if (!newbits.isSubset(bs)) {
+ bitsets.remove(bs);
+ bitsets.add(newbits.union(bs));
+ coalesce();
+ }
+ return;
+ }
+ }
+ bitsets.add(newbits);
+ }
+ coalesce();
+ }
+ /**
+ * Sets all of the bits in this BitSet to {@code false}.
+ */
+ public void clear() {
+ synchronized (bitsets) {
+ bitsets.clear();
+ }
+ }
+ /**
+ * Sets the bit specified by the index to {@code false}.
+ * @param bit the index of the bit to be cleared
+ */
+ public void clear(long bit) {
+ clear(bit, bit+1);
+ }
+ /**
+ * Sets the bits from the specified {@code from} (inclusive) to the
+ * specified {@code to} (exclusive) to {@code false}.
+ * @param from index of the first bit to be cleared
+ * @param to index after the last bit to be cleared
+ * @throws IndexOutOfBoundsException if {@code from} is negative,
+ * or {@code to} is negative,
+ * or {@code from} is larger than {@code to}
+ */
+ public void clear(long from, long to) {
+ checkRange(from, to);
+ RLE newbits = new RLE(from, to-from);
+ List(no value) | filter disabled |
off | filter disabled |
N[,M[,action]] | set N, M, and action (used in the algorithm below). + * Action is drop or throttle. + * If M is missing, it defaults to 5 minutes. + * If the action is missing, it defaults to drop. + * |
+ * The action is triggered iff: + *
+This package provide various helper classes used by the provisioning server. +
+ + diff --git a/datarouter-prov/src/main/java/com/att/research/datarouter/reports/DailyLatencyReport.java b/datarouter-prov/src/main/java/com/att/research/datarouter/reports/DailyLatencyReport.java new file mode 100644 index 00000000..63d612f9 --- /dev/null +++ b/datarouter-prov/src/main/java/com/att/research/datarouter/reports/DailyLatencyReport.java @@ -0,0 +1,194 @@ +/******************************************************************************* + * ============LICENSE_START================================================== + * * org.onap.dmaap + * * =========================================================================== + * * Copyright © 2017 AT&T Intellectual Property. All rights reserved. + * * =========================================================================== + * * Licensed under the Apache License, Version 2.0 (the "License"); + * * you may not use this file except in compliance with the License. + * * You may obtain a copy of the License at + * * + * * http://www.apache.org/licenses/LICENSE-2.0 + * * + * * Unless required by applicable law or agreed to in writing, software + * * distributed under the License is distributed on an "AS IS" BASIS, + * * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * * See the License for the specific language governing permissions and + * * limitations under the License. + * * ============LICENSE_END==================================================== + * * + * * ECOMP is a trademark and service mark of AT&T Intellectual Property. + * * + ******************************************************************************/ + + +package com.att.research.datarouter.reports; + +import java.io.FileNotFoundException; +import java.io.PrintWriter; +import java.sql.Connection; +import java.sql.PreparedStatement; +import java.sql.ResultSet; +import java.sql.SQLException; +import java.text.SimpleDateFormat; +import java.util.ArrayList; +import java.util.Date; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.TreeSet; + +import com.att.research.datarouter.provisioning.utils.DB; + +/** + * Generate a daily per feed latency report. The report is a .csv file containing the following columns: + *date | the date for this record |
feedid | the Feed ID for this record |
minsize | the minimum size of all files published on this feed and date |
maxsize | the maximum size of all files published on this feed and date |
avgsize | the average size of all files published on this feed and date |
minlat | the minimum latency in delivering this feed to all subscribers (in ms) |
maxlat | the maximum latency in delivering this feed to all subscribers (in ms) |
avglat | the average latency in delivering this feed to all subscribers (in ms) |
fanout | the average number of subscribers this feed was delivered to |
+ * In the context of this report, latency is defined as the value + * (De - Ps) + * where: + *
+ *Ps is the time that the publication of the file to the node starts.
+ *De is the time that the delivery of the file to the subscriber ends.
+ * + * @author Robert P. Eby + * @version $Id: DailyLatencyReport.java,v 1.2 2013/11/06 16:23:54 eby Exp $ + */ +public class DailyLatencyReport extends ReportBase { + private static final String SELECT_SQL = + "select EVENT_TIME, TYPE, PUBLISH_ID, FEED_FILEID, FEEDID, CONTENT_LENGTH from LOG_RECORDS" + + " where EVENT_TIME >= ? and EVENT_TIME <= ?"; + + private class Job { + public long pubtime = 0; + public long clen = 0; + public ListDate | Feeds | ||||||||||
---|---|---|---|---|---|---|---|---|---|---|---|
"+date+" | ");
+ s.append(feeds.length).append(feeds.length > 1 ? " Feeds\n" : " Feed\n");
+ s.append("
|
Date | Feeds | Feed ID | Publish Count | Subs | Sub ID | Delivery Count |
---|---|---|---|---|---|---|
") + .append(date) + .append(" | "); + s.append("") + .append(feeds.length) + .append(" | "); + String px1 = ""; + for (String feed : feeds) { + JSONObject j3 = j2.getJSONObject(feed); + int pubcount = j3.getInt("pubcount"); + int subcnt = j3.length()-1; + int rc2 = (subcnt < 1) ? 1 : subcnt; + String[] subs = JSONObject.getNames(j3); + Arrays.sort(subs); + s.append(px1) + .append("") + .append(feed) + .append(" | "); + s.append("") + .append(pubcount) + .append(" | "); + s.append("") + .append(subcnt) + .append(" | "); + String px2 = ""; + for (String sub : subs) { + if (!sub.equals("pubcount")) { + s.append(px2); + s.append(""+sub+" | "); + s.append(""+j3.getInt(sub)+" | "); + s.append("