-/*
- * ===============================LICENSE_START======================================
- * dcae-analytics
- * ================================================================================
- * 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===========================================
- */
-
-package org.openecomp.dcae.apod.analytics.cdap.tca.worker;
-
-import co.cask.cdap.api.TxRunnable;
-import co.cask.cdap.api.common.Bytes;
-import co.cask.cdap.api.data.DatasetContext;
-import co.cask.cdap.api.dataset.lib.CloseableIterator;
-import co.cask.cdap.api.dataset.lib.KeyValue;
-import co.cask.cdap.api.dataset.lib.ObjectMappedTable;
-import co.cask.cdap.api.metrics.Metrics;
-import co.cask.cdap.api.worker.WorkerContext;
-import com.google.common.base.Joiner;
-import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
-import org.apache.tephra.TransactionFailureException;
-import org.openecomp.dcae.apod.analytics.cdap.common.CDAPMetricsConstants;
-import org.openecomp.dcae.apod.analytics.cdap.common.persistance.tca.TCAVESAlertEntity;
-import org.openecomp.dcae.apod.analytics.cdap.common.persistance.tca.TCAVESAlertsPersister;
-import org.openecomp.dcae.apod.analytics.cdap.tca.utils.CDAPTCAUtils;
-import org.openecomp.dcae.apod.analytics.common.exception.DCAEAnalyticsRuntimeException;
-import org.openecomp.dcae.apod.analytics.common.utils.HTTPUtils;
-import org.openecomp.dcae.apod.analytics.dmaap.domain.response.DMaaPMRPublisherResponse;
-import org.openecomp.dcae.apod.analytics.dmaap.service.publisher.DMaaPMRPublisher;
-import org.quartz.DisallowConcurrentExecution;
-import org.quartz.Job;
-import org.quartz.JobDataMap;
-import org.quartz.JobExecutionContext;
-import org.quartz.JobExecutionException;
-import org.quartz.PersistJobDataAfterExecution;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.util.Date;
-import java.util.LinkedHashMap;
-import java.util.List;
-import java.util.Map;
-import java.util.Set;
-
-import static org.openecomp.dcae.apod.analytics.common.AnalyticsConstants.CDAP_ALERTS_TABLE_VARIABLE_NAME;
-import static org.openecomp.dcae.apod.analytics.common.AnalyticsConstants.DMAAP_METRICS_VARIABLE_NAME;
-import static org.openecomp.dcae.apod.analytics.common.AnalyticsConstants.DMAAP_PUBLISHER_VARIABLE_NAME;
-import static org.openecomp.dcae.apod.analytics.common.AnalyticsConstants.WORKER_CONTEXT_VARIABLE_NAME;
-
-/**
- * Quartz Job that will monitor any new alert messages in given TCA Alerts table and if any found publish them to
- * DMaaP MR topic
- *<p>
- * @author Rajiv Singla . Creation Date: 11/17/2016.
- */
-@DisallowConcurrentExecution
-@PersistJobDataAfterExecution
-@SuppressFBWarnings("SIC_INNER_SHOULD_BE_STATIC_ANON")
-public class TCADMaaPMRPublisherJob implements Job {
-
- private static final Logger LOG = LoggerFactory.getLogger(TCADMaaPMRPublisherJob.class);
-
- @Override
- public void execute(JobExecutionContext jobExecutionContext) throws JobExecutionException {
-
- LOG.debug("Starting DMaaP MR Topic Publisher fetch Job. Next firing time will be: {}",
- jobExecutionContext.getNextFireTime());
-
- // Get Job Data Map
- final JobDataMap jobDataMap = jobExecutionContext.getMergedJobDataMap();
-
- // Fetch all Job Params from Job Data Map
- final String cdapAlertsTableName = jobDataMap.getString(CDAP_ALERTS_TABLE_VARIABLE_NAME);
- final WorkerContext workerContext = (WorkerContext) jobDataMap.get(WORKER_CONTEXT_VARIABLE_NAME);
- final DMaaPMRPublisher publisher = (DMaaPMRPublisher) jobDataMap.get(DMAAP_PUBLISHER_VARIABLE_NAME);
- final Metrics metrics = (Metrics) jobDataMap.get(DMAAP_METRICS_VARIABLE_NAME);
-
- LOG.debug("Start looking for new message in Alerts Table: {}", cdapAlertsTableName);
-
- // Get new alerts from alerts table
- final Map<String, TCAVESAlertEntity> newAlertsMap = getNewAlertsMap(cdapAlertsTableName, workerContext);
-
- // If no new alerts are found - nothing to publish
- if (newAlertsMap.isEmpty()) {
- LOG.debug("No new alerts found in Alerts Table name: {}. Nothing to Publisher....", cdapAlertsTableName);
- metrics.count(CDAPMetricsConstants.TCA_PUBLISHER_NO_NEW_ALERTS_LOOKUP_METRIC, 1);
- return;
- }
-
- final int newAlertsCount = newAlertsMap.size();
- LOG.debug("Found new alerts in Alerts Table name: {}. No of new alerts: {}", cdapAlertsTableName,
- newAlertsCount);
- metrics.count(CDAPMetricsConstants.TCA_PUBLISHER_NEW_ALERTS_METRIC, newAlertsCount);
-
- // Get alert message strings from alert Entities
- final List<String> newAlertsMessages = CDAPTCAUtils.extractAlertFromAlertEntities(newAlertsMap.values());
-
- // Publish messages to DMaaP MR Topic
- try {
-
- final DMaaPMRPublisherResponse publisherResponse = publisher.publish(newAlertsMessages);
-
- final Integer responseCode = publisherResponse.getResponseCode();
- final String responseMessage = publisherResponse.getResponseMessage();
- final int pendingMessagesCount = publisherResponse.getPendingMessagesCount();
-
- LOG.debug("Publisher Response Code: {}, Publisher message: {}, Pending Messages Count: {}", responseCode,
- responseMessage, pendingMessagesCount);
-
- if (HTTPUtils.isSuccessfulResponseCode(responseCode)) {
- LOG.debug("Successfully Published alerts to DMaaP MR Topic.");
- metrics.count(CDAPMetricsConstants.TCA_PUBLISHER_SUCCESSFUL_DMAAP_RESPONSE_METRIC, 1);
- } else {
- LOG.warn("Unable to publish alerts to DMaaP MR Topic. Publisher will try to send it later....");
- metrics.count(CDAPMetricsConstants.TCA_PUBLISHER_UNSUCCESSFUL_DMAAP_RESPONSE_METRIC, 1);
- }
-
- } catch (DCAEAnalyticsRuntimeException e) {
- LOG.error("Exception while publishing messages to DMaaP MR Topic: {}", e);
- } finally {
- // delete send message from alerts table
- deleteAlertsByKey(cdapAlertsTableName, workerContext, newAlertsMap.keySet(), metrics);
- }
-
- LOG.debug("Finished DMaaP MR Topic Publisher fetch Job.");
-
- }
-
- /**
- * Gets New Messages from alerts table as Map with row keys as keys and {@link TCAVESAlertEntity} as values
- *
- * @param cdapAlertsTableName alerts table name
- * @param workerContext worker context
- * @return Map with row keys as keys and {@link TCAVESAlertEntity} as values
- */
- protected Map<String, TCAVESAlertEntity> getNewAlertsMap(final String cdapAlertsTableName,
- final WorkerContext workerContext) {
- final Map<String, TCAVESAlertEntity> newAlertsMap = new LinkedHashMap<>();
- try {
- workerContext.execute(new TxRunnable() {
- @Override
- public void run(DatasetContext context) throws Exception {
- final ObjectMappedTable<TCAVESAlertEntity> alertsTable = context.getDataset(cdapAlertsTableName);
- final Date currentTime = new Date();
- final String rowKey = TCAVESAlertsPersister.createRowKey(currentTime);
- final CloseableIterator<KeyValue<byte[], TCAVESAlertEntity>> scan = alertsTable.scan(null, rowKey);
- while (scan.hasNext()) {
- final KeyValue<byte[], TCAVESAlertEntity> alertEntityKeyValue = scan.next();
- newAlertsMap.put(Bytes.toString(alertEntityKeyValue.getKey()), alertEntityKeyValue.getValue());
- }
- }
- });
- } catch (TransactionFailureException e) {
- final String errorMessage = "Transaction Error while getting new alerts from alerts table: " + e.toString();
- throw new DCAEAnalyticsRuntimeException(errorMessage, LOG, e);
- }
- return newAlertsMap;
- }
-
- /**
- * Deletes rows in Alerts table for give rowKeys
- *
- * @param cdapAlertsTableName CDAP Alerts Table Name
- * @param workerContext Worker Context
- * @param rowKeys Row Key Set
- * @param metrics CDAP metrics
- */
- protected void deleteAlertsByKey(final String cdapAlertsTableName, final WorkerContext workerContext,
- final Set<String> rowKeys, final Metrics metrics) {
- LOG.debug("Deleting Published Alerts from alerts table with rowKeys: {}", Joiner.on(",").join(rowKeys));
- try {
- workerContext.execute(new TxRunnable() {
- @Override
- public void run(DatasetContext context) throws Exception {
- final ObjectMappedTable<TCAVESAlertEntity> alertsTable = context.getDataset(cdapAlertsTableName);
- for (String rowKey : rowKeys) {
- alertsTable.delete(rowKey);
- metrics.count(CDAPMetricsConstants.TCA_PUBLISHER_DELETED_ALERTS_METRIC, 1);
- }
- }
- });
- } catch (TransactionFailureException e) {
- final String errorMessage =
- "Transaction Error while deleting published alerts in alerts table: " + e.toString();
- throw new DCAEAnalyticsRuntimeException(errorMessage, LOG, e);
- }
- }
-}
+/*\r
+ * ===============================LICENSE_START======================================\r
+ * dcae-analytics\r
+ * ================================================================================\r
+ * Copyright © 2017 AT&T Intellectual Property. All rights reserved.\r
+ * ================================================================================\r
+ * Licensed under the Apache License, Version 2.0 (the "License");\r
+ * you may not use this file except in compliance with the License.\r
+ * You may obtain a copy of the License at\r
+ *\r
+ * http://www.apache.org/licenses/LICENSE-2.0\r
+ *\r
+ * Unless required by applicable law or agreed to in writing, software\r
+ * distributed under the License is distributed on an "AS IS" BASIS,\r
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\r
+ * See the License for the specific language governing permissions and\r
+ * limitations under the License.\r
+ * ============================LICENSE_END===========================================\r
+ */\r
+\r
+package org.openecomp.dcae.apod.analytics.cdap.tca.worker;\r
+\r
+import co.cask.cdap.api.TxRunnable;\r
+import co.cask.cdap.api.common.Bytes;\r
+import co.cask.cdap.api.data.DatasetContext;\r
+import co.cask.cdap.api.dataset.lib.CloseableIterator;\r
+import co.cask.cdap.api.dataset.lib.KeyValue;\r
+import co.cask.cdap.api.dataset.lib.ObjectMappedTable;\r
+import co.cask.cdap.api.metrics.Metrics;\r
+import co.cask.cdap.api.worker.WorkerContext;\r
+import com.google.common.base.Joiner;\r
+import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;\r
+import org.apache.tephra.TransactionFailureException;\r
+import org.openecomp.dcae.apod.analytics.cdap.common.CDAPMetricsConstants;\r
+import org.openecomp.dcae.apod.analytics.cdap.common.persistance.tca.TCAVESAlertEntity;\r
+import org.openecomp.dcae.apod.analytics.cdap.common.persistance.tca.TCAVESAlertsPersister;\r
+import org.openecomp.dcae.apod.analytics.cdap.tca.utils.CDAPTCAUtils;\r
+import org.openecomp.dcae.apod.analytics.common.exception.DCAEAnalyticsRuntimeException;\r
+import org.openecomp.dcae.apod.analytics.common.utils.HTTPUtils;\r
+import org.openecomp.dcae.apod.analytics.dmaap.domain.response.DMaaPMRPublisherResponse;\r
+import org.openecomp.dcae.apod.analytics.dmaap.service.publisher.DMaaPMRPublisher;\r
+import org.quartz.DisallowConcurrentExecution;\r
+import org.quartz.Job;\r
+import org.quartz.JobDataMap;\r
+import org.quartz.JobExecutionContext;\r
+import org.quartz.JobExecutionException;\r
+import org.quartz.PersistJobDataAfterExecution;\r
+import org.slf4j.Logger;\r
+import org.slf4j.LoggerFactory;\r
+\r
+import java.util.Date;\r
+import java.util.LinkedHashMap;\r
+import java.util.List;\r
+import java.util.Map;\r
+import java.util.Set;\r
+\r
+import static org.openecomp.dcae.apod.analytics.common.AnalyticsConstants.CDAP_ALERTS_TABLE_VARIABLE_NAME;\r
+import static org.openecomp.dcae.apod.analytics.common.AnalyticsConstants.DMAAP_METRICS_VARIABLE_NAME;\r
+import static org.openecomp.dcae.apod.analytics.common.AnalyticsConstants.DMAAP_PUBLISHER_VARIABLE_NAME;\r
+import static org.openecomp.dcae.apod.analytics.common.AnalyticsConstants.WORKER_CONTEXT_VARIABLE_NAME;\r
+\r
+/**\r
+ * Quartz Job that will monitor any new alert messages in given TCA Alerts table and if any found publish them to\r
+ * DMaaP MR topic\r
+ *<p>\r
+ * @author Rajiv Singla . Creation Date: 11/17/2016.\r
+ */\r
+@DisallowConcurrentExecution\r
+@PersistJobDataAfterExecution\r
+@SuppressFBWarnings("SIC_INNER_SHOULD_BE_STATIC_ANON")\r
+public class TCADMaaPMRPublisherJob implements Job {\r
+\r
+ private static final Logger LOG = LoggerFactory.getLogger(TCADMaaPMRPublisherJob.class);\r
+\r
+ @Override\r
+ public void execute(JobExecutionContext jobExecutionContext) throws JobExecutionException {\r
+\r
+ LOG.debug("Starting DMaaP MR Topic Publisher fetch Job. Next firing time will be: {}",\r
+ jobExecutionContext.getNextFireTime());\r
+\r
+ // Get Job Data Map\r
+ final JobDataMap jobDataMap = jobExecutionContext.getMergedJobDataMap();\r
+\r
+ // Fetch all Job Params from Job Data Map\r
+ final String cdapAlertsTableName = jobDataMap.getString(CDAP_ALERTS_TABLE_VARIABLE_NAME);\r
+ final WorkerContext workerContext = (WorkerContext) jobDataMap.get(WORKER_CONTEXT_VARIABLE_NAME);\r
+ final DMaaPMRPublisher publisher = (DMaaPMRPublisher) jobDataMap.get(DMAAP_PUBLISHER_VARIABLE_NAME);\r
+ final Metrics metrics = (Metrics) jobDataMap.get(DMAAP_METRICS_VARIABLE_NAME);\r
+\r
+ LOG.debug("Start looking for new message in Alerts Table: {}", cdapAlertsTableName);\r
+\r
+ // Get new alerts from alerts table\r
+ final Map<String, TCAVESAlertEntity> newAlertsMap = getNewAlertsMap(cdapAlertsTableName, workerContext);\r
+\r
+ // If no new alerts are found - nothing to publish\r
+ if (newAlertsMap.isEmpty()) {\r
+ LOG.debug("No new alerts found in Alerts Table name: {}. Nothing to Publisher....", cdapAlertsTableName);\r
+ metrics.count(CDAPMetricsConstants.TCA_PUBLISHER_NO_NEW_ALERTS_LOOKUP_METRIC, 1);\r
+ return;\r
+ }\r
+\r
+ final int newAlertsCount = newAlertsMap.size();\r
+ LOG.debug("Found new alerts in Alerts Table name: {}. No of new alerts: {}", cdapAlertsTableName,\r
+ newAlertsCount);\r
+ metrics.count(CDAPMetricsConstants.TCA_PUBLISHER_NEW_ALERTS_METRIC, newAlertsCount);\r
+\r
+ // Get alert message strings from alert Entities\r
+ final List<String> newAlertsMessages = CDAPTCAUtils.extractAlertFromAlertEntities(newAlertsMap.values());\r
+\r
+ // Publish messages to DMaaP MR Topic\r
+ try {\r
+\r
+ final DMaaPMRPublisherResponse publisherResponse = publisher.publish(newAlertsMessages);\r
+\r
+ final Integer responseCode = publisherResponse.getResponseCode();\r
+ final String responseMessage = publisherResponse.getResponseMessage();\r
+ final int pendingMessagesCount = publisherResponse.getPendingMessagesCount();\r
+\r
+ LOG.debug("Publisher Response Code: {}, Publisher message: {}, Pending Messages Count: {}", responseCode,\r
+ responseMessage, pendingMessagesCount);\r
+\r
+ if (HTTPUtils.isSuccessfulResponseCode(responseCode)) {\r
+ LOG.debug("Successfully Published alerts to DMaaP MR Topic.");\r
+ metrics.count(CDAPMetricsConstants.TCA_PUBLISHER_SUCCESSFUL_DMAAP_RESPONSE_METRIC, 1);\r
+ } else {\r
+ LOG.warn("Unable to publish alerts to DMaaP MR Topic. Publisher will try to send it later....");\r
+ metrics.count(CDAPMetricsConstants.TCA_PUBLISHER_UNSUCCESSFUL_DMAAP_RESPONSE_METRIC, 1);\r
+ }\r
+\r
+ } catch (DCAEAnalyticsRuntimeException e) {\r
+ LOG.error("Exception while publishing messages to DMaaP MR Topic: {}", e);\r
+ } finally {\r
+ // delete send message from alerts table\r
+ deleteAlertsByKey(cdapAlertsTableName, workerContext, newAlertsMap.keySet(), metrics);\r
+ }\r
+\r
+ LOG.debug("Finished DMaaP MR Topic Publisher fetch Job.");\r
+\r
+ }\r
+\r
+ /**\r
+ * Gets New Messages from alerts table as Map with row keys as keys and {@link TCAVESAlertEntity} as values\r
+ *\r
+ * @param cdapAlertsTableName alerts table name\r
+ * @param workerContext worker context\r
+ * @return Map with row keys as keys and {@link TCAVESAlertEntity} as values\r
+ */\r
+ protected Map<String, TCAVESAlertEntity> getNewAlertsMap(final String cdapAlertsTableName,\r
+ final WorkerContext workerContext) {\r
+ final Map<String, TCAVESAlertEntity> newAlertsMap = new LinkedHashMap<>();\r
+ try {\r
+ workerContext.execute(new TxRunnable() {\r
+ @Override\r
+ public void run(DatasetContext context) throws Exception {\r
+ final ObjectMappedTable<TCAVESAlertEntity> alertsTable = context.getDataset(cdapAlertsTableName);\r
+ final Date currentTime = new Date();\r
+ final String rowKey = TCAVESAlertsPersister.createRowKey(currentTime);\r
+ final CloseableIterator<KeyValue<byte[], TCAVESAlertEntity>> scan = alertsTable.scan(null, rowKey);\r
+ while (scan.hasNext()) {\r
+ final KeyValue<byte[], TCAVESAlertEntity> alertEntityKeyValue = scan.next();\r
+ newAlertsMap.put(Bytes.toString(alertEntityKeyValue.getKey()), alertEntityKeyValue.getValue());\r
+ }\r
+ }\r
+ });\r
+ } catch (TransactionFailureException e) {\r
+ final String errorMessage = "Transaction Error while getting new alerts from alerts table: " + e.toString();\r
+ throw new DCAEAnalyticsRuntimeException(errorMessage, LOG, e);\r
+ }\r
+ return newAlertsMap;\r
+ }\r
+\r
+ /**\r
+ * Deletes rows in Alerts table for give rowKeys\r
+ *\r
+ * @param cdapAlertsTableName CDAP Alerts Table Name\r
+ * @param workerContext Worker Context\r
+ * @param rowKeys Row Key Set\r
+ * @param metrics CDAP metrics\r
+ */\r
+ protected void deleteAlertsByKey(final String cdapAlertsTableName, final WorkerContext workerContext,\r
+ final Set<String> rowKeys, final Metrics metrics) {\r
+ LOG.debug("Deleting Published Alerts from alerts table with rowKeys: {}", Joiner.on(",").join(rowKeys));\r
+ try {\r
+ workerContext.execute(new TxRunnable() {\r
+ @Override\r
+ public void run(DatasetContext context) throws Exception {\r
+ final ObjectMappedTable<TCAVESAlertEntity> alertsTable = context.getDataset(cdapAlertsTableName);\r
+ for (String rowKey : rowKeys) {\r
+ alertsTable.delete(rowKey);\r
+ metrics.count(CDAPMetricsConstants.TCA_PUBLISHER_DELETED_ALERTS_METRIC, 1);\r
+ }\r
+ }\r
+ });\r
+ } catch (TransactionFailureException e) {\r
+ final String errorMessage =\r
+ "Transaction Error while deleting published alerts in alerts table: " + e.toString();\r
+ throw new DCAEAnalyticsRuntimeException(errorMessage, LOG, e);\r
+ }\r
+ }\r
+}\r
Code Review / dcaegen2 / analytics / tca.git / blobdiff