datarouter-node/src/main/java/org/onap/dmaap/datarouter/node/NodeConfigManager.java

   1 /*******************************************************************************
   2  * ============LICENSE_START==================================================
   3  * * org.onap.dmaap
   4  * * ===========================================================================
   5  * * Copyright © 2017 AT&T Intellectual Property. All rights reserved.
   6  * * ===========================================================================
   7  * * Licensed under the Apache License, Version 2.0 (the "License");
   8  * * you may not use this file except in compliance with the License.
   9  * * You may obtain a copy of the License at
  10  * *
  11  *  *      http://www.apache.org/licenses/LICENSE-2.0
  12  * *
  13  *  * Unless required by applicable law or agreed to in writing, software
  14  * * distributed under the License is distributed on an "AS IS" BASIS,
  15  * * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  16  * * See the License for the specific language governing permissions and
  17  * * limitations under the License.
  18  * * ============LICENSE_END====================================================
  19  * *
  20  * * ECOMP is a trademark and service mark of AT&T Intellectual Property.
  21  * *
  22  ******************************************************************************/
  23
  24
  25 package org.onap.dmaap.datarouter.node;
  26
  27 import com.att.eelf.configuration.EELFLogger;
  28 import com.att.eelf.configuration.EELFManager;
  29 import java.io.File;
  30 import java.io.FileInputStream;
  31 import java.io.InputStreamReader;
  32 import java.io.Reader;
  33 import java.net.URL;
  34 import java.util.Properties;
  35 import java.util.Timer;
  36 import org.apache.log4j.Logger;
  37 import org.onap.dmaap.datarouter.node.eelf.EelfMsgs;
  38
  39
  40 /**
  41  * Maintain the configuration of a Data Router node
  42  * <p>
  43  * The NodeConfigManager is the single point of contact for servlet, delivery, event logging, and log retention
  44  * subsystems to access configuration information.  (Log4J has its own configuration mechanism).
  45  * <p>
  46  * There are two basic sets of configuration data.  The static local configuration data, stored in a local configuration
  47  * file (created as part of installation by SWM), and the dynamic global configuration data fetched from the data router
  48  * provisioning server.
  49  */
  50 public class NodeConfigManager implements DeliveryQueueHelper {
  51
  52     private static EELFLogger eelflogger = EELFManager.getInstance()
  53             .getLogger(NodeConfigManager.class);
  54     private static Logger logger = Logger.getLogger("org.onap.dmaap.datarouter.node.NodeConfigManager");
  55     private static NodeConfigManager base = new NodeConfigManager();
  56
  57     private Timer timer = new Timer("Node Configuration Timer", true);
  58     private long maxfailuretimer;
  59     private long initfailuretimer;
  60     private long expirationtimer;
  61     private double failurebackoff;
  62     private long fairtimelimit;
  63     private int fairfilelimit;
  64     private double fdpstart;
  65     private double fdpstop;
  66     private int deliverythreads;
  67     private String provurl;
  68     private String provhost;
  69     private IsFrom provcheck;
  70     private int gfport;
  71     private int svcport;
  72     private int port;
  73     private String spooldir;
  74     private String logdir;
  75     private long logretention;
  76     private String redirfile;
  77     private String kstype;
  78     private String ksfile;
  79     private String kspass;
  80     private String kpass;
  81     private String tstype;
  82     private String tsfile;
  83     private String tspass;
  84     private String myname;
  85     private RedirManager rdmgr;
  86     private RateLimitedOperation pfetcher;
  87     private NodeConfig config;
  88     private File quiesce;
  89     private PublishId pid;
  90     private String nak;
  91     private TaskList configtasks = new TaskList();
  92     private String eventlogurl;
  93     private String eventlogprefix;
  94     private String eventlogsuffix;
  95     private String eventloginterval;
  96     private boolean followredirects;
  97
  98
  99     /**
 100      * Get the default node configuration manager
 101      */
 102     public static NodeConfigManager getInstance() {
 103         return (base);
 104     }
 105
 106     /**
 107      * Initialize the configuration of a Data Router node
 108      */
 109     private NodeConfigManager() {
 110         Properties p = new Properties();
 111         try {
 112             p.load(new FileInputStream(System
 113                     .getProperty("org.onap.dmaap.datarouter.node.properties", "/opt/app/datartr/etc/node.properties")));
 114         } catch (Exception e) {
 115
 116             NodeUtils.setIpAndFqdnForEelf("NodeConfigManager");
 117             eelflogger.error(EelfMsgs.MESSAGE_PROPERTIES_LOAD_ERROR);
 118             logger.error("NODE0301 Unable to load local configuration file " + System
 119                             .getProperty("org.onap.dmaap.datarouter.node.properties", "/opt/app/datartr/etc/node.properties"),
 120                     e);
 121         }
 122         provurl = p.getProperty("ProvisioningURL", "https://feeds-drtr.web.att.com/internal/prov");
 123         try {
 124             provhost = (new URL(provurl)).getHost();
 125         } catch (Exception e) {
 126             NodeUtils.setIpAndFqdnForEelf("NodeConfigManager");
 127             eelflogger.error(EelfMsgs.MESSAGE_BAD_PROV_URL, provurl);
 128             logger.error("NODE0302 Bad provisioning server URL " + provurl);
 129             System.exit(1);
 130         }
 131         logger.info("NODE0303 Provisioning server is " + provhost);
 132         eventlogurl = p.getProperty("LogUploadURL", "https://feeds-drtr.web.att.com/internal/logs");
 133         provcheck = new IsFrom(provhost);
 134         gfport = Integer.parseInt(p.getProperty("IntHttpPort", "8080"));
 135         svcport = Integer.parseInt(p.getProperty("IntHttpsPort", "8443"));
 136         port = Integer.parseInt(p.getProperty("ExtHttpsPort", "443"));
 137         long minpfinterval = Long.parseLong(p.getProperty("MinProvFetchInterval", "10000"));
 138         long minrsinterval = Long.parseLong(p.getProperty("MinRedirSaveInterval", "10000"));
 139         spooldir = p.getProperty("SpoolDir", "spool");
 140         File fdir = new File(spooldir + "/f");
 141         fdir.mkdirs();
 142         for (File junk : fdir.listFiles()) {
 143             if (junk.isFile()) {
 144                 junk.delete();
 145             }
 146         }
 147         logdir = p.getProperty("LogDir", "logs");
 148         (new File(logdir)).mkdirs();
 149         logretention = Long.parseLong(p.getProperty("LogRetention", "30")) * 86400000L;
 150         eventlogprefix = logdir + "/events";
 151         eventlogsuffix = ".log";
 152         String redirfile = p.getProperty("RedirectionFile", "etc/redirections.dat");
 153         kstype = p.getProperty("KeyStoreType", "jks");
 154         ksfile = p.getProperty("KeyStoreFile", "etc/keystore");
 155         kspass = p.getProperty("KeyStorePassword", "changeme");
 156         kpass = p.getProperty("KeyPassword", "changeme");
 157         tstype = p.getProperty("TrustStoreType", "jks");
 158         tsfile = p.getProperty("TrustStoreFile");
 159         tspass = p.getProperty("TrustStorePassword", "changeme");
 160         if (tsfile != null && tsfile.length() > 0) {
 161             System.setProperty("javax.net.ssl.trustStoreType", tstype);
 162             System.setProperty("javax.net.ssl.trustStore", tsfile);
 163             System.setProperty("javax.net.ssl.trustStorePassword", tspass);
 164         }
 165         nak = p.getProperty("NodeAuthKey", "Node123!");
 166         quiesce = new File(p.getProperty("QuiesceFile", "etc/SHUTDOWN"));
 167         myname = NodeUtils.getCanonicalName(kstype, ksfile, kspass);
 168         if (myname == null) {
 169             NodeUtils.setIpAndFqdnForEelf("NodeConfigManager");
 170             eelflogger.error(EelfMsgs.MESSAGE_KEYSTORE_FETCH_ERROR, ksfile);
 171             logger.error("NODE0309 Unable to fetch canonical name from keystore file " + ksfile);
 172             System.exit(1);
 173         }
 174         logger.info("NODE0304 My certificate says my name is " + myname);
 175         pid = new PublishId(myname);
 176         rdmgr = new RedirManager(redirfile, minrsinterval, timer);
 177         pfetcher = new RateLimitedOperation(minpfinterval, timer) {
 178             public void run() {
 179                 fetchconfig();
 180             }
 181         };
 182         logger.info("NODE0305 Attempting to fetch configuration at " + provurl);
 183         pfetcher.request();
 184     }
 185
 186     private void localconfig() {
 187         followredirects = Boolean.parseBoolean(getProvParam("FOLLOW_REDIRECTS", "false"));
 188         eventloginterval = getProvParam("LOGROLL_INTERVAL", "30s");
 189         initfailuretimer = 10000;
 190         maxfailuretimer = 3600000;
 191         expirationtimer = 86400000;
 192         failurebackoff = 2.0;
 193         deliverythreads = 40;
 194         fairfilelimit = 100;
 195         fairtimelimit = 60000;
 196         fdpstart = 0.05;
 197         fdpstop = 0.2;
 198         try {
 199             initfailuretimer = (long) (Double.parseDouble(getProvParam("DELIVERY_INIT_RETRY_INTERVAL")) * 1000);
 200         } catch (Exception e) {
 201         }
 202         try {
 203             maxfailuretimer = (long) (Double.parseDouble(getProvParam("DELIVERY_MAX_RETRY_INTERVAL")) * 1000);
 204         } catch (Exception e) {
 205         }
 206         try {
 207             expirationtimer = (long) (Double.parseDouble(getProvParam("DELIVERY_MAX_AGE")) * 1000);
 208         } catch (Exception e) {
 209         }
 210         try {
 211             failurebackoff = Double.parseDouble(getProvParam("DELIVERY_RETRY_RATIO"));
 212         } catch (Exception e) {
 213         }
 214         try {
 215             deliverythreads = Integer.parseInt(getProvParam("DELIVERY_THREADS"));
 216         } catch (Exception e) {
 217         }
 218         try {
 219             fairfilelimit = Integer.parseInt(getProvParam("FAIR_FILE_LIMIT"));
 220         } catch (Exception e) {
 221         }
 222         try {
 223             fairtimelimit = (long) (Double.parseDouble(getProvParam("FAIR_TIME_LIMIT")) * 1000);
 224         } catch (Exception e) {
 225         }
 226         try {
 227             fdpstart = Double.parseDouble(getProvParam("FREE_DISK_RED_PERCENT")) / 100.0;
 228         } catch (Exception e) {
 229         }
 230         try {
 231             fdpstop = Double.parseDouble(getProvParam("FREE_DISK_YELLOW_PERCENT")) / 100.0;
 232         } catch (Exception e) {
 233         }
 234         if (fdpstart < 0.01) {
 235             fdpstart = 0.01;
 236         }
 237         if (fdpstart > 0.5) {
 238             fdpstart = 0.5;
 239         }
 240         if (fdpstop < fdpstart) {
 241             fdpstop = fdpstart;
 242         }
 243         if (fdpstop > 0.5) {
 244             fdpstop = 0.5;
 245         }
 246     }
 247
 248     private void fetchconfig() {
 249         try {
 250             System.out.println("provurl:: " + provurl);
 251             Reader r = new InputStreamReader((new URL(provurl)).openStream());
 252             config = new NodeConfig(new ProvData(r), myname, spooldir, port, nak);
 253             localconfig();
 254             configtasks.startRun();
 255             Runnable rr;
 256             while ((rr = configtasks.next()) != null) {
 257                 try {
 258                     rr.run();
 259                 } catch (Exception e) {
 260                 }
 261             }
 262         } catch (Exception e) {
 263             NodeUtils.setIpAndFqdnForEelf("fetchconfigs");
 264             eelflogger.error(EelfMsgs.MESSAGE_CONF_FAILED, e.toString());
 265             logger.error("NODE0306 Configuration failed " + e.toString() + " - try again later", e);
 266             pfetcher.request();
 267         }
 268     }
 269
 270     /**
 271      * Process a gofetch request from a particular IP address.  If the IP address is not an IP address we would go to to
 272      * fetch the provisioning data, ignore the request.  If the data has been fetched very recently (default 10
 273      * seconds), wait a while before fetching again.
 274      */
 275     public synchronized void gofetch(String remoteaddr) {
 276         if (provcheck.isFrom(remoteaddr)) {
 277             logger.info("NODE0307 Received configuration fetch request from provisioning server " + remoteaddr);
 278             pfetcher.request();
 279         } else {
 280             logger.info("NODE0308 Received configuration fetch request from unexpected server " + remoteaddr);
 281         }
 282     }
 283
 284     /**
 285      * Am I configured?
 286      */
 287     public boolean isConfigured() {
 288         return (config != null);
 289     }
 290
 291     /**
 292      * Am I shut down?
 293      */
 294     public boolean isShutdown() {
 295         return (quiesce.exists());
 296     }
 297
 298     /**
 299      * Given a routing string, get the targets.
 300      *
 301      * @param routing Target string
 302      * @return array of targets
 303      */
 304     public Target[] parseRouting(String routing) {
 305         return (config.parseRouting(routing));
 306     }
 307
 308     /**
 309      * Given a set of credentials and an IP address, is this request from another node?
 310      *
 311      * @param credentials Credentials offered by the supposed node
 312      * @param ip IP address the request came from
 313      * @return If the credentials and IP address are recognized, true, otherwise false.
 314      */
 315     public boolean isAnotherNode(String credentials, String ip) {
 316         return (config.isAnotherNode(credentials, ip));
 317     }
 318
 319     /**
 320      * Check whether publication is allowed.
 321      *
 322      * @param feedid The ID of the feed being requested
 323      * @param credentials The offered credentials
 324      * @param ip The requesting IP address
 325      * @return True if the IP and credentials are valid for the specified feed.
 326      */
 327     public String isPublishPermitted(String feedid, String credentials, String ip) {
 328         return (config.isPublishPermitted(feedid, credentials, ip));
 329     }
 330
 331     /**
 332      * Check who the user is given the feed ID and the offered credentials.
 333      *
 334      * @param feedid The ID of the feed specified
 335      * @param credentials The offered credentials
 336      * @return Null if the credentials are invalid or the user if they are valid.
 337      */
 338     public String getAuthUser(String feedid, String credentials) {
 339         return (config.getAuthUser(feedid, credentials));
 340     }
 341
 342     /**
 343      * Check if the publish request should be sent to another node based on the feedid, user, and source IP address.
 344      *
 345      * @param feedid The ID of the feed specified
 346      * @param user The publishing user
 347      * @param ip The IP address of the publish endpoint
 348      * @return Null if the request should be accepted or the correct hostname if it should be sent to another node.
 349      */
 350     public String getIngressNode(String feedid, String user, String ip) {
 351         return (config.getIngressNode(feedid, user, ip));
 352     }
 353
 354     /**
 355      * Get a provisioned configuration parameter (from the provisioning server configuration)
 356      *
 357      * @param name The name of the parameter
 358      * @return The value of the parameter or null if it is not defined.
 359      */
 360     public String getProvParam(String name) {
 361         return (config.getProvParam(name));
 362     }
 363
 364     /**
 365      * Get a provisioned configuration parameter (from the provisioning server configuration)
 366      *
 367      * @param name The name of the parameter
 368      * @param defaultValue The value to use if the parameter is not defined
 369      * @return The value of the parameter or deflt if it is not defined.
 370      */
 371     public String getProvParam(String name, String defaultValue) {
 372         name = config.getProvParam(name);
 373         if (name == null) {
 374             name = defaultValue;
 375         }
 376         return (name);
 377     }
 378
 379     /**
 380      * Generate a publish ID
 381      */
 382     public String getPublishId() {
 383         return (pid.next());
 384     }
 385
 386     /**
 387      * Get all the outbound spooling destinations. This will include both subscriptions and nodes.
 388      */
 389     public DestInfo[] getAllDests() {
 390         return (config.getAllDests());
 391     }
 392
 393     /**
 394      * Register a task to run whenever the configuration changes
 395      */
 396     public void registerConfigTask(Runnable task) {
 397         configtasks.addTask(task);
 398     }
 399
 400     /**
 401      * Deregister a task to run whenever the configuration changes
 402      */
 403     public void deregisterConfigTask(Runnable task) {
 404         configtasks.removeTask(task);
 405     }
 406
 407     /**
 408      * Get the URL to deliver a message to.
 409      *
 410      * @param destinfo The destination information
 411      * @param fileid The file ID
 412      * @return The URL to deliver to
 413      */
 414     public String getDestURL(DestInfo destinfo, String fileid) {
 415         String subid = destinfo.getSubId();
 416         String purl = destinfo.getURL();
 417         if (followredirects && subid != null) {
 418             purl = rdmgr.lookup(subid, purl);
 419         }
 420         return (purl + "/" + fileid);
 421     }
 422
 423     /**
 424      * Is a destination redirected?
 425      */
 426     public boolean isDestRedirected(DestInfo destinfo) {
 427         return (followredirects && rdmgr.isRedirected(destinfo.getSubId()));
 428     }
 429
 430     /**
 431      * Set up redirection on receipt of a 3XX from a target URL
 432      */
 433     public boolean handleRedirection(DestInfo destinfo, String redirto, String fileid) {
 434         fileid = "/" + fileid;
 435         String subid = destinfo.getSubId();
 436         String purl = destinfo.getURL();
 437         if (followredirects && subid != null && redirto.endsWith(fileid)) {
 438             redirto = redirto.substring(0, redirto.length() - fileid.length());
 439             if (!redirto.equals(purl)) {
 440                 rdmgr.redirect(subid, purl, redirto);
 441                 return (true);
 442             }
 443         }
 444         return (false);
 445     }
 446
 447     /**
 448      * Handle unreachable target URL
 449      */
 450     public void handleUnreachable(DestInfo destinfo) {
 451         String subid = destinfo.getSubId();
 452         if (followredirects && subid != null) {
 453             rdmgr.forget(subid);
 454         }
 455     }
 456
 457     /**
 458      * Get the timeout before retrying after an initial delivery failure
 459      */
 460     public long getInitFailureTimer() {
 461         return (initfailuretimer);
 462     }
 463
 464     /**
 465      * Get the maximum timeout between delivery attempts
 466      */
 467     public long getMaxFailureTimer() {
 468         return (maxfailuretimer);
 469     }
 470
 471     /**
 472      * Get the ratio between consecutive delivery attempts
 473      */
 474     public double getFailureBackoff() {
 475         return (failurebackoff);
 476     }
 477
 478     /**
 479      * Get the expiration timer for deliveries
 480      */
 481     public long getExpirationTimer() {
 482         return (expirationtimer);
 483     }
 484
 485     /**
 486      * Get the maximum number of file delivery attempts before checking if another queue has work to be performed.
 487      */
 488     public int getFairFileLimit() {
 489         return (fairfilelimit);
 490     }
 491
 492     /**
 493      * Get the maximum amount of time spent delivering files before checking if another queue has work to be performed.
 494      */
 495     public long getFairTimeLimit() {
 496         return (fairtimelimit);
 497     }
 498
 499     /**
 500      * Get the targets for a feed
 501      *
 502      * @param feedid The feed ID
 503      * @return The targets this feed should be delivered to
 504      */
 505     public Target[] getTargets(String feedid) {
 506         return (config.getTargets(feedid));
 507     }
 508
 509     /**
 510      * Get the spool directory for temporary files
 511      */
 512     public String getSpoolDir() {
 513         return (spooldir + "/f");
 514     }
 515
 516     /**
 517      * Get the base directory for spool directories
 518      */
 519     public String getSpoolBase() {
 520         return (spooldir);
 521     }
 522
 523     /**
 524      * Get the key store type
 525      */
 526     public String getKSType() {
 527         return (kstype);
 528     }
 529
 530     /**
 531      * Get the key store file
 532      */
 533     public String getKSFile() {
 534         return (ksfile);
 535     }
 536
 537     /**
 538      * Get the key store password
 539      */
 540     public String getKSPass() {
 541         return (kspass);
 542     }
 543
 544     /**
 545      * Get the key password
 546      */
 547     public String getKPass() {
 548         return (kpass);
 549     }
 550
 551     /**
 552      * Get the http port
 553      */
 554     public int getHttpPort() {
 555         return (gfport);
 556     }
 557
 558     /**
 559      * Get the https port
 560      */
 561     public int getHttpsPort() {
 562         return (svcport);
 563     }
 564
 565     /**
 566      * Get the externally visible https port
 567      */
 568     public int getExtHttpsPort() {
 569         return (port);
 570     }
 571
 572     /**
 573      * Get the external name of this machine
 574      */
 575     public String getMyName() {
 576         return (myname);
 577     }
 578
 579     /**
 580      * Get the number of threads to use for delivery
 581      */
 582     public int getDeliveryThreads() {
 583         return (deliverythreads);
 584     }
 585
 586     /**
 587      * Get the URL for uploading the event log data
 588      */
 589     public String getEventLogUrl() {
 590         return (eventlogurl);
 591     }
 592
 593     /**
 594      * Get the prefix for the names of event log files
 595      */
 596     public String getEventLogPrefix() {
 597         return (eventlogprefix);
 598     }
 599
 600     /**
 601      * Get the suffix for the names of the event log files
 602      */
 603     public String getEventLogSuffix() {
 604         return (eventlogsuffix);
 605     }
 606
 607     /**
 608      * Get the interval between event log file rollovers
 609      */
 610     public String getEventLogInterval() {
 611         return (eventloginterval);
 612     }
 613
 614     /**
 615      * Should I follow redirects from subscribers?
 616      */
 617     public boolean isFollowRedirects() {
 618         return (followredirects);
 619     }
 620
 621     /**
 622      * Get the directory where the event and node log files live
 623      */
 624     public String getLogDir() {
 625         return (logdir);
 626     }
 627
 628     /**
 629      * How long do I keep log files (in milliseconds)
 630      */
 631     public long getLogRetention() {
 632         return (logretention);
 633     }
 634
 635     /**
 636      * Get the timer
 637      */
 638     public Timer getTimer() {
 639         return (timer);
 640     }
 641
 642     /**
 643      * Get the feed ID for a subscription
 644      *
 645      * @param subid The subscription ID
 646      * @return The feed ID
 647      */
 648     public String getFeedId(String subid) {
 649         return (config.getFeedId(subid));
 650     }
 651
 652     /**
 653      * Get the authorization string this node uses
 654      *
 655      * @return The Authorization string for this node
 656      */
 657     public String getMyAuth() {
 658         return (config.getMyAuth());
 659     }
 660
 661     /**
 662      * Get the fraction of free spool disk space where we start throwing away undelivered files.  This is
 663      * FREE_DISK_RED_PERCENT / 100.0.  Default is 0.05.  Limited by 0.01 <= FreeDiskStart <= 0.5.
 664      */
 665     public double getFreeDiskStart() {
 666         return (fdpstart);
 667     }
 668
 669     /**
 670      * Get the fraction of free spool disk space where we stop throwing away undelivered files.  This is
 671      * FREE_DISK_YELLOW_PERCENT / 100.0.  Default is 0.2.  Limited by FreeDiskStart <= FreeDiskStop <= 0.5.
 672      */
 673     public double getFreeDiskStop() {
 674         return (fdpstop);
 675     }
 676
 677     /**
 678      * Get the spool directory for a subscription
 679      */
 680     public String getSpoolDir(String subid, String remoteaddr) {
 681         if (provcheck.isFrom(remoteaddr)) {
 682             String sdir = config.getSpoolDir(subid);
 683             if (sdir != null) {
 684                 logger.info("NODE0310 Received subscription reset request for subscription " + subid
 685                         + " from provisioning server " + remoteaddr);
 686             } else {
 687                 logger.info("NODE0311 Received subscription reset request for unknown subscription " + subid
 688                         + " from provisioning server " + remoteaddr);
 689             }
 690             return (sdir);
 691         } else {
 692             logger.info("NODE0312 Received subscription reset request from unexpected server " + remoteaddr);
 693             return (null);
 694         }
 695     }
 696 }