dcae-analytics-cdap-tca/src/main/java/org/openecomp/dcae/apod/analytics/cdap/tca/worker/TCADMaaPPublisherWorker.java

   1 /*\r
   2  * ===============================LICENSE_START======================================\r
   3  *  dcae-analytics\r
   4  * ================================================================================\r
   5  *    Copyright © 2017 AT&T Intellectual Property. All rights reserved.\r
   6  * ================================================================================\r
   7  *  Licensed under the Apache License, Version 2.0 (the "License");\r
   8  *  you may not use this file except in compliance with the License.\r
   9  *   You may obtain a copy of the License at\r
  10  *\r
  11  *          http://www.apache.org/licenses/LICENSE-2.0\r
  12  *\r
  13  *  Unless required by applicable law or agreed to in writing, software\r
  14  *  distributed under the License is distributed on an "AS IS" BASIS,\r
  15  *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\r
  16  *  See the License for the specific language governing permissions and\r
  17  *  limitations under the License.\r
  18  *  ============================LICENSE_END===========================================\r
  19  */\r
  20 \r
  21 package org.openecomp.dcae.apod.analytics.cdap.tca.worker;\r
  22 \r
  23 import co.cask.cdap.api.annotation.Property;\r
  24 import co.cask.cdap.api.metrics.Metrics;\r
  25 import co.cask.cdap.api.worker.WorkerContext;\r
  26 import org.openecomp.dcae.apod.analytics.cdap.common.CDAPComponentsConstants;\r
  27 import org.openecomp.dcae.apod.analytics.cdap.tca.settings.TCAAppPreferences;\r
  28 import org.openecomp.dcae.apod.analytics.cdap.tca.utils.AppPreferencesToPublisherConfigMapper;\r
  29 import org.openecomp.dcae.apod.analytics.cdap.tca.utils.CDAPTCAUtils;\r
  30 import org.openecomp.dcae.apod.analytics.common.AnalyticsConstants;\r
  31 import org.openecomp.dcae.apod.analytics.common.exception.DCAEAnalyticsRuntimeException;\r
  32 import org.openecomp.dcae.apod.analytics.dmaap.DMaaPMRFactory;\r
  33 import org.openecomp.dcae.apod.analytics.dmaap.domain.config.DMaaPMRPublisherConfig;\r
  34 import org.openecomp.dcae.apod.analytics.dmaap.service.publisher.DMaaPMRPublisher;\r
  35 import org.openecomp.dcae.apod.analytics.tca.utils.TCAUtils;\r
  36 import org.quartz.JobDataMap;\r
  37 import org.quartz.SchedulerException;\r
  38 import org.quartz.impl.StdSchedulerFactory;\r
  39 import org.slf4j.Logger;\r
  40 import org.slf4j.LoggerFactory;\r
  41 \r
  42 import java.util.concurrent.atomic.AtomicBoolean;\r
  43 \r
  44 import static java.lang.String.format;\r
  45 \r
  46 /**\r
  47  * TCA DMaaP Publisher will monitor alerts table at regular intervals and publish any alerts to DMaaP MR Publishing\r
  48  * Topic\r
  49  * <p>\r
  50  * @author Rajiv Singla . Creation Date: 11/16/2016.\r
  51  */\r
  52 public class TCADMaaPPublisherWorker extends BaseTCADMaaPMRWorker {\r
  53 \r
  54     private static final Logger LOG = LoggerFactory.getLogger(TCADMaaPPublisherWorker.class);\r
  55 \r
  56     private DMaaPMRPublisher publisher;\r
  57     private Metrics metrics;\r
  58     @Property\r
  59     private final String tcaVESAlertsTableName;\r
  60 \r
  61     public TCADMaaPPublisherWorker(final String tcaVESAlertsTableName) {\r
  62         this.tcaVESAlertsTableName = tcaVESAlertsTableName;\r
  63     }\r
  64 \r
  65     @Override\r
  66     public void configure() {\r
  67         setName(CDAPComponentsConstants.TCA_FIXED_DMAAP_PUBLISHER_WORKER);\r
  68         setDescription(CDAPComponentsConstants.TCA_FIXED_DMAAP_PUBLISHER_DESCRIPTION_WORKER);\r
  69         LOG.debug("Configuring TCA MR DMaaP Publisher worker with name: {}",\r
  70                 CDAPComponentsConstants.TCA_FIXED_DMAAP_PUBLISHER_WORKER);\r
  71     }\r
  72 \r
  73 \r
  74     @Override\r
  75     public void initialize(WorkerContext context) throws Exception {\r
  76         super.initialize(context);\r
  77 \r
  78         // Parse runtime arguments\r
  79         final TCAAppPreferences tcaAppPreferences = CDAPTCAUtils.getValidatedTCAAppPreferences(context);\r
  80 \r
  81         LOG.info("Initializing TCA MR DMaaP Publisher worker with preferences: {}", tcaAppPreferences);\r
  82 \r
  83         //  Map TCA App Preferences to DMaaP MR Publisher Config\r
  84         final DMaaPMRPublisherConfig publisherConfig = AppPreferencesToPublisherConfigMapper.map(tcaAppPreferences);\r
  85 \r
  86         LOG.info("TCA DMaaP MR Publisher worker will be polling TCA Alerts Table Name: {}", tcaVESAlertsTableName);\r
  87 \r
  88         // Create an instance of DMaaP MR Publisher\r
  89         LOG.debug("Creating an instance of DMaaP Publisher");\r
  90         publisher = DMaaPMRFactory.create().createPublisher(publisherConfig);\r
  91 \r
  92         // initialize a new Quartz scheduler\r
  93         initializeScheduler(tcaAppPreferences, new StdSchedulerFactory());\r
  94 \r
  95         // initialize scheduler state\r
  96         isSchedulerShutdown = new AtomicBoolean(true);\r
  97     }\r
  98 \r
  99 \r
 100     /**\r
 101      * Stop DMaaP Publisher\r
 102      */\r
 103     @Override\r
 104     public void stop() {\r
 105         // Close Publisher - which will flush any batch messages if present in batch queue\r
 106         if (publisher != null) {\r
 107             try {\r
 108                 publisher.close();\r
 109             } catch (Exception e) {\r
 110                 final String errorMessage = format("Error while shutting down DMaaP MR Publisher: %s", e);\r
 111                 throw new DCAEAnalyticsRuntimeException(errorMessage, LOG, e);\r
 112             }\r
 113         }\r
 114         // Shut down scheduler\r
 115         super.stop();\r
 116     }\r
 117 \r
 118 \r
 119     /**\r
 120      * Initializes a scheduler instance for DMaaP MR Publisher Job\r
 121      *\r
 122      * @throws SchedulerException SchedulerException\r
 123      */\r
 124     private void initializeScheduler(final TCAAppPreferences tcaAnalyticsAppConfig,\r
 125                                      final StdSchedulerFactory stdSchedulerFactory) throws SchedulerException {\r
 126 \r
 127         // Get Publisher polling interval\r
 128         final Integer publisherPollingInterval = tcaAnalyticsAppConfig.getPublisherPollingInterval();\r
 129 \r
 130         // Publisher Quartz Properties file\r
 131         final String quartzPublisherPropertiesFileName = AnalyticsConstants.TCA_QUARTZ_PUBLISHER_PROPERTIES_FILE_NAME;\r
 132 \r
 133         // Create a new JobDataMap containing information required by TCA DMaaP Publisher Job\r
 134         final JobDataMap jobDataMap = new JobDataMap();\r
 135         jobDataMap.put(AnalyticsConstants.CDAP_ALERTS_TABLE_VARIABLE_NAME, tcaVESAlertsTableName);\r
 136         jobDataMap.put(AnalyticsConstants.WORKER_CONTEXT_VARIABLE_NAME, getContext());\r
 137         jobDataMap.put(AnalyticsConstants.DMAAP_PUBLISHER_VARIABLE_NAME, publisher);\r
 138         jobDataMap.put(AnalyticsConstants.DMAAP_METRICS_VARIABLE_NAME, metrics);\r
 139 \r
 140         // Create new publisher scheduler\r
 141         scheduler = TCAUtils.createQuartzScheduler(publisherPollingInterval, stdSchedulerFactory,\r
 142                 quartzPublisherPropertiesFileName, jobDataMap, TCADMaaPMRPublisherJob.class,\r
 143                 AnalyticsConstants.TCA_DMAAP_PUBLISHER_QUARTZ_JOB_NAME,\r
 144                 AnalyticsConstants.TCA_DMAAP_PUBLISHER_QUARTZ_TRIGGER_NAME);\r
 145     }\r
 146 }\r