<artifactId>logback-classic</artifactId>
</dependency>
+ <!-- DISTRIBUTED CACHING -->
+ <dependency>
+ <groupId>redis.clients</groupId>
+ <artifactId>jedis</artifactId>
+ </dependency>
+
<!-- FIND BUGS -->
<dependency>
<groupId>com.google.code.findbugs</groupId>
import org.onap.dcae.apod.analytics.model.domain.policy.tca.MetricsPerEventName;
import org.onap.dcae.apod.analytics.model.domain.policy.tca.Threshold;
import org.onap.dcae.apod.analytics.model.facade.tca.TCAVESResponse;
+import org.onap.dcae.apod.analytics.model.util.AnalyticsModelJsonUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import redis.clients.jedis.HostAndPort;
+import redis.clients.jedis.JedisCluster;
+import java.io.IOException;
import java.util.Date;
import java.util.List;
+import java.util.Set;
import static org.onap.dcae.apod.analytics.common.utils.PersistenceUtils.TABLE_ROW_KEY_COLUMN_NAME;
final MetricsPerEventName violatedMetricsPerEventName,
final TCAVESResponse tcavesResponse,
final String abatementTS,
- final ObjectMappedTable<TCAAlertsAbatementEntity> tcaAlertsAbatementTable) {
+ final ObjectMappedTable<TCAAlertsAbatementEntity> tcaAlertsAbatementTable,
+ final Set<HostAndPort> redisHostAndPorts) {
final String abatementTableKey = createKey(eventListener, violatedMetricsPerEventName);
final long currentTimestamp = new Date().getTime();
final String requestID = tcavesResponse.getRequestID();
final TCAAlertsAbatementEntity tcaAlertsAbatementEntity = new TCAAlertsAbatementEntity(currentTimestamp,
requestID, abatementTS);
- tcaAlertsAbatementTable.write(abatementTableKey, tcaAlertsAbatementEntity);
+
+ // if redis is enabled save entity in redis cluster
+ if (redisHostAndPorts != null) {
+ persistAlertAbatementEntityInRedis(redisHostAndPorts, abatementTableKey, tcaAlertsAbatementEntity);
+ } else {
+ tcaAlertsAbatementTable.write(abatementTableKey, tcaAlertsAbatementEntity);
+ }
LOG.debug("Persisted AlertsAbatementEntity: {} with Key: {}", tcaAlertsAbatementEntity, abatementTableKey);
public static TCAAlertsAbatementEntity lookUpByKey(final EventListener eventListener,
final MetricsPerEventName violatedMetricsPerEventName,
final ObjectMappedTable<TCAAlertsAbatementEntity>
- tcaAlertsAbatementTable) {
+ tcaAlertsAbatementTable,
+ final Set<HostAndPort> redisHostAndPorts) {
final String abatementTableKey = createKey(eventListener, violatedMetricsPerEventName);
+
+ // if redis is enabled get entity from redis cluster
+ if (redisHostAndPorts != null) {
+ return getAlertAbatementEntityFromRedis(redisHostAndPorts, abatementTableKey);
+ }
+
return tcaAlertsAbatementTable.read(abatementTableKey);
}
-
public static String createKey(final EventListener eventListener,
final MetricsPerEventName violatedMetricsPerEventName) {
// no null check required as all are required fields
return KEY_JOINER.join(abatementKeyList);
}
+ private static TCAAlertsAbatementEntity getAlertAbatementEntityFromRedis(final Set<HostAndPort> redisHostAndPorts,
+ final String abatementTableKey) {
+ try (final JedisCluster jedisCluster = new JedisCluster(redisHostAndPorts)) {
+ if (jedisCluster.exists(abatementTableKey)) {
+ return AnalyticsModelJsonUtils.readValue(jedisCluster.get(abatementTableKey),
+ TCAAlertsAbatementEntity.class);
+ } else {
+ return null;
+ }
+ } catch (IOException e) {
+ final String errorMessage = String.format("Unable to look up key: %s in redis cluster: %s",
+ abatementTableKey, redisHostAndPorts);
+ throw new DCAEAnalyticsRuntimeException(errorMessage, LOG, e);
+ }
+ }
+
+ private static void persistAlertAbatementEntityInRedis(final Set<HostAndPort> redisHostAndPorts,
+ final String abatementTableKey,
+ final TCAAlertsAbatementEntity tcaAlertsAbatementEntity) {
+ try (final JedisCluster jedisCluster = new JedisCluster(redisHostAndPorts)) {
+ jedisCluster.set(abatementTableKey, AnalyticsModelJsonUtils.writeValueAsString(tcaAlertsAbatementEntity));
+ } catch (IOException e) {
+ final String errorMessage = String.format("Unable to store key:value - %s:%s in redis cluster: %s",
+ abatementTableKey, tcaAlertsAbatementEntity, redisHostAndPorts);
+ throw new DCAEAnalyticsRuntimeException(errorMessage, LOG, e);
+ }
+ }
+
}
public void testPersist() throws Exception {
TCAAlertsAbatementPersister.persist(eventListener, violatedMetricsPerEventName, tcavesResponse,
- abatementTS, alertsAbatementTable);
+ abatementTS, alertsAbatementTable, null);
verify(alertsAbatementTable, times(1)).write(anyString(),
any(TCAAlertsAbatementEntity.class));
@Test
public void testLookUpByKey() throws Exception {
- TCAAlertsAbatementPersister.lookUpByKey(eventListener, violatedMetricsPerEventName, alertsAbatementTable);
+ TCAAlertsAbatementPersister.lookUpByKey(eventListener, violatedMetricsPerEventName, alertsAbatementTable, null);
verify(alertsAbatementTable, times(1)).read(eq(EXPECTED_LOOKUP_KEY));
}
import org.onap.dcae.apod.analytics.cdap.common.exception.CDAPSettingsException;
import org.onap.dcae.apod.analytics.cdap.common.persistance.tca.TCAAlertsAbatementEntity;
import org.onap.dcae.apod.analytics.cdap.common.persistance.tca.TCAAlertsAbatementPersister;
+import org.onap.dcae.apod.analytics.cdap.tca.settings.TCAAppPreferences;
+import org.onap.dcae.apod.analytics.cdap.tca.utils.CDAPTCAUtils;
+import org.onap.dcae.apod.analytics.common.exception.DCAEAnalyticsRuntimeException;
import org.onap.dcae.apod.analytics.model.domain.cef.EventListener;
import org.onap.dcae.apod.analytics.model.domain.policy.tca.ClosedLoopEventStatus;
import org.onap.dcae.apod.analytics.model.domain.policy.tca.MetricsPerEventName;
import org.onap.dcae.apod.analytics.tca.utils.TCAUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import redis.clients.jedis.HostAndPort;
+import redis.clients.jedis.JedisCluster;
+import redis.clients.jedis.JedisPool;
+import redis.clients.jedis.exceptions.JedisConnectionException;
import java.io.IOException;
import java.util.Date;
+import java.util.LinkedHashSet;
+import java.util.Map;
+import java.util.Set;
/**
* Flowlet responsible to sending out abatement alerts
@Property
private final String tcaAlertsAbatementTableName;
+ private Set<HostAndPort> redisHostAndPorts = null;
+
@Output(CDAPComponentsConstants.TCA_FIXED_VES_ALERTS_ABATEMENT_NAME_OUTPUT)
protected OutputEmitter<String> alertsAbatementOutputEmitter;
public void initialize(FlowletContext flowletContext) throws Exception {
super.initialize(flowletContext);
tcaAlertsAbatementTable = getContext().getDataset(tcaAlertsAbatementTableName);
+ // Parse runtime arguments
+ final TCAAppPreferences tcaAppPreferences = CDAPTCAUtils.getValidatedTCAAppPreferences(flowletContext);
+ if(tcaAppPreferences.getEnableRedisCaching()) {
+ final String redisHosts = tcaAppPreferences.getRedisHosts();
+ LOG.info("Redis Distributed Caching is enabled for abated alerts with Redis Hosts: {}", redisHosts);
+ redisHostAndPorts = getRedisHostsAndPorts(redisHosts);
+ checkRedisConnection(redisHostAndPorts);
+ } else {
+ LOG.info("Redis Distributed caching is disabled for abated alerts");
+ }
}
@ProcessInput(CDAPComponentsConstants.TCA_FIXED_VES_TCA_CALCULATOR_NAME_OUTPUT)
LOG.debug("Saving information for ONSET event for cefMessage: {}", cefMessage);
TCAAlertsAbatementPersister.persist(eventListener, violatedMetricsPerEventName, tcavesResponse,
- null, tcaAlertsAbatementTable);
+ null, tcaAlertsAbatementTable, redisHostAndPorts);
LOG.debug("Emitting ONSET alert: {}", alertMessageString);
alertsAbatementOutputEmitter.emit(alertMessageString);
break;
LOG.debug("Looking up previous sent alert for abated threshold: {}", violatedThreshold);
final TCAAlertsAbatementEntity previousAlertsAbatementEntry =
TCAAlertsAbatementPersister.lookUpByKey(eventListener, violatedMetricsPerEventName,
- tcaAlertsAbatementTable);
+ tcaAlertsAbatementTable, redisHostAndPorts);
if (previousAlertsAbatementEntry != null) {
// save new Abatement alert sent timestamp in table
TCAAlertsAbatementPersister.persist(eventListener, violatedMetricsPerEventName, tcavesResponse,
- Long.toString(newAbatementSentTS), tcaAlertsAbatementTable);
+ Long.toString(newAbatementSentTS), tcaAlertsAbatementTable, redisHostAndPorts);
// Set request id to be same as previous ONSET event request ID
tcavesResponse.setRequestID(previousAlertsAbatementEntry.getRequestId());
}
+ private static Set<HostAndPort> getRedisHostsAndPorts(final String redisHosts) {
+ final LinkedHashSet<HostAndPort> hostAndPorts = new LinkedHashSet<>();
+ final String[] redisHostsString = redisHosts.split(",");
+ for (String redisHostString : redisHostsString) {
+ hostAndPorts.add(HostAndPort.parseString(redisHostString.trim()));
+ }
+ return hostAndPorts;
+ }
+
+ private static void checkRedisConnection(final Set<HostAndPort> redisHostAndPorts) {
+ LOG.info("Checking Redis Connection for Redis Hosts: {}", redisHostAndPorts);
+ try (final JedisCluster jedisCluster = new JedisCluster(redisHostAndPorts)) {
+ final Map<String, JedisPool> clusterNodes = jedisCluster.getClusterNodes();
+ jedisCluster.get("testKey");
+ LOG.info("Confirmed redis cluster Nodes: {}", clusterNodes.keySet());
+ } catch (JedisConnectionException | IOException e) {
+ LOG.error("Unable to make Redis connection for given redisHosts: {}", redisHostAndPorts);
+ throw new DCAEAnalyticsRuntimeException("No Redis Connection", LOG, e);
+ }
+ }
+
}
protected String aaiEnrichmentProxyURL;
+
+ // Redis Distributed caching
+ protected Boolean enableRedisCaching;
+
+ protected String redisHosts;
+
/**
* Default constructor to setup default values for TCA App Preferences
*/
enableAlertCEFFormat = AnalyticsConstants.TCA_DEFAULT_ENABLE_CEF_FORMATTED_ALERT;
+ // aai
enableAAIEnrichment = AnalyticsConstants.TCA_DEFAULT_ENABLE_AAI_ENRICHMENT;
aaiEnrichmentIgnoreSSLCertificateErrors =
AnalyticsConstants.TCA_DEFAULT_AAI_ENRICHMENT_IGNORE_SSL_CERTIFICATE_ERRORS;
aaiEnrichmentProxyURL = AnalyticsConstants.TCA_DEFAULT_AAI_ENRICHMENT_PROXY_URL;
+ // redis
+ enableRedisCaching = AnalyticsConstants.TCA_DEFAULT_ENABLE_REDIS_CACHING;
+
}
public String getSubscriberHostName() {
return aaiEnrichmentProxyURL;
}
+ public Boolean getEnableRedisCaching() {
+ return enableRedisCaching;
+ }
+
+ public String getRedisHosts() {
+ return redisHosts;
+ }
+
@Override
public String toString() {
return Objects.toStringHelper(this)
.add("aaiVNFEnrichmentAPIPath", aaiVNFEnrichmentAPIPath)
.add("aaiVMEnrichmentAPIPath", aaiVMEnrichmentAPIPath)
.add("aaiEnrichmentProxyEnabled", aaiEnrichmentProxyURL == null ? "false" : "true")
+ .add("enableRedisCaching", enableRedisCaching)
+ .add("redisHosts", redisHosts)
.toString();
}
}
}
}
+ final Boolean enableRedisCaching = appPreferences.getEnableRedisCaching();
+
+ // if redis distributed caching is enabled then redis Hosts must be provided
+ if(enableRedisCaching) {
+ final String redisHosts = appPreferences.getRedisHosts();
+ if(isEmpty(redisHosts)) {
+ validationResponse.addErrorMessage("redisHosts",
+ "Redis Caching is enabled but no redis hosts are provided");
+ }
+ }
+
return validationResponse;
}
}
tcaTestAppPreferences.setAaiEnrichmentIgnoreSSLCertificateErrors(true);
tcaTestAppPreferences.setAaiVMEnrichmentAPIPath("VM_ENRICHMENT_PATH");
tcaTestAppPreferences.setAaiVNFEnrichmentAPIPath("VNF_ENRICHMENT_PATH");
+
+ tcaTestAppPreferences.setEnableRedisCaching(false);
+ tcaTestAppPreferences.setRedisHosts("127.0.0.1:6379");
return tcaTestAppPreferences;
}
preference.put("publisherMaxBatchSize", "1000");
preference.put("publisherMaxRecoveryQueueSize", "100");
preference.put("publisherPollingInterval", "6000");
+
+ preference.put("enableRedisCaching", "false");
+ preference.put("redisHosts", "127.0.0.1:6379");
+
return preference;
}
package org.onap.dcae.apod.analytics.cdap.tca.flowlet;
+import co.cask.cdap.api.app.ApplicationSpecification;
import co.cask.cdap.api.dataset.lib.ObjectMappedTable;
import co.cask.cdap.api.flow.flowlet.FlowletContext;
import co.cask.cdap.api.flow.flowlet.OutputEmitter;
import com.google.common.collect.ImmutableList;
import org.junit.Test;
+import org.mockito.Mockito;
import org.onap.dcae.apod.analytics.cdap.common.CDAPComponentsConstants;
import org.onap.dcae.apod.analytics.cdap.common.domain.tca.ThresholdCalculatorOutput;
import org.onap.dcae.apod.analytics.cdap.common.exception.CDAPSettingsException;
import org.onap.dcae.apod.analytics.model.domain.policy.tca.Threshold;
import org.onap.dcae.apod.analytics.tca.utils.TCAUtils;
+import java.io.IOException;
import java.util.Date;
import java.util.List;
final FlowletContext mockFlowletContext = mock(FlowletContext.class);
final ObjectMappedTable<TCAAlertsAbatementEntity> mockObjectMappedTable = mock(ObjectMappedTable.class);
when(mockFlowletContext.getDataset(eq(testTCAAlertsAbatementTableName))).thenReturn(mockObjectMappedTable);
+ configureFlowletContext(mockFlowletContext);
tcaAlertsAbatementFlowlet.initialize(mockFlowletContext);
doNothing().when(mockObjectMappedTable).write(any(String.class), any(TCAAlertsAbatementEntity.class));
final FlowletContext mockFlowletContext = mock(FlowletContext.class);
final ObjectMappedTable<TCAAlertsAbatementEntity> mockObjectMappedTable = mock(ObjectMappedTable.class);
when(mockFlowletContext.getDataset(eq(testTCAAlertsAbatementTableName))).thenReturn(mockObjectMappedTable);
+ configureFlowletContext(mockFlowletContext);
tcaAlertsAbatementFlowlet.initialize(mockFlowletContext);
doNothing().when(mockObjectMappedTable).write(any(String.class), any(TCAAlertsAbatementEntity.class));
final FlowletContext mockFlowletContext = mock(FlowletContext.class);
final ObjectMappedTable<TCAAlertsAbatementEntity> mockObjectMappedTable = mock(ObjectMappedTable.class);
when(mockFlowletContext.getDataset(eq(testTCAAlertsAbatementTableName))).thenReturn(mockObjectMappedTable);
+ configureFlowletContext(mockFlowletContext);
tcaAlertsAbatementFlowlet.initialize(mockFlowletContext);
doNothing().when(mockObjectMappedTable).write(any(String.class), any(TCAAlertsAbatementEntity.class));
final FlowletContext mockFlowletContext = mock(FlowletContext.class);
final ObjectMappedTable<TCAAlertsAbatementEntity> mockObjectMappedTable = mock(ObjectMappedTable.class);
when(mockFlowletContext.getDataset(eq(testTCAAlertsAbatementTableName))).thenReturn(mockObjectMappedTable);
+ configureFlowletContext(mockFlowletContext);
tcaAlertsAbatementFlowlet.initialize(mockFlowletContext);
-
doNothing().when(mockObjectMappedTable).write(any(String.class), any(TCAAlertsAbatementEntity.class));
when(mockObjectMappedTable.read(any(String.class))).thenReturn(null);
return thresholdCalculatorOutput;
}
+ private void configureFlowletContext(final FlowletContext mockFlowletContext) throws IOException {
+ when(mockFlowletContext.getRuntimeArguments()).thenReturn(getPreferenceMap());
+ ApplicationSpecification mockApplicationSpecification = Mockito.mock(ApplicationSpecification.class);
+ when(mockApplicationSpecification.getConfiguration()).thenReturn(fromStream(TCA_APP_CONFIG_FILE_LOCATION));
+ when(mockFlowletContext.getApplicationSpecification()).thenReturn(mockApplicationSpecification);
+ }
+
}
public void setAaiVMEnrichmentAPIPath(String aaiVMEnrichmentAPIPath) {
this.aaiVMEnrichmentAPIPath = aaiVMEnrichmentAPIPath;
}
+
+ public void setEnableRedisCaching(final Boolean enableRedisCaching) {
+ this.enableRedisCaching = enableRedisCaching;
+ }
+
+ public void setRedisHosts(final String redisHosts) {
+ this.redisHosts = redisHosts;
+ }
+
+
}
public static final String TCA_DEFAULT_AAI_ENRICHMENT_PROXY_URL = null;
+ // TCA Redis Distributed Caching default
+ public static final Boolean TCA_DEFAULT_ENABLE_REDIS_CACHING = true;
+
private AnalyticsConstants() {
\r
domain=measurementsForVfScaling\r
\r
+enableRedisCaching=false\r
+redisHosts=127.0.0.1:7000\r
+\r
configuration.metricsPerEventName.Mfvs_eNodeB_RANKPI.policy.eventName=Mfvs_eNodeB_RANKPI\r
configuration.metricsPerEventName.Mfvs_eNodeB_RANKPI.policyName=configuration.dcae.microservice.tca.xml\r
configuration.metricsPerEventName.Mfvs_eNodeB_RANKPI.policyVersion=v0.0.1\r
\r
tca_policy={\"domain\":\"measurementsForVfScaling\",\"metricsPerEventName\":[{\"eventName\":\"Mfvs_eNodeB_RANKPI\",\"controlLoopSchemaType\":\"VNF\",\"policyScope\":\"resource=vFirewall;type=configuration\",\"policyName\":\"configuration.dcae.microservice.tca.xml\",\"policyVersion\":\"v0.0.1\",\"thresholds\":[{\"closedLoopControlName\":\"CL-FRWL-LOW-TRAFFIC-SIG-d925ed73-8231-4d02-9545-db4e101f88f8\",\"closedLoopEventStatus\":\"ONSET\",\"version\":\"1.0.2\",\"fieldPath\":\"$.event.measurementsForVfScalingFields.vNicPerformanceArray[*].receivedBroadcastPacketsAccumulated\",\"thresholdValue\":4000,\"direction\":\"LESS_OR_EQUAL\",\"severity\":\"MAJOR\"},{\"closedLoopControlName\":\"CL-FRWL-HIGH-TRAFFIC-SIG-EA36FE84-9342-5E13-A656-EC5F21309A09\",\"closedLoopEventStatus\":\"ONSET\",\"version\":\"1.0.2\",\"fieldPath\":\"$.event.measurementsForVfScalingFields.vNicPerformanceArray[*].receivedBroadcastPacketsAccumulated\",\"thresholdValue\":20000,\"direction\":\"GREATER_OR_EQUAL\",\"severity\":\"CRITICAL\"}]},{\"eventName\":\"vLoadBalancer\",\"controlLoopSchemaType\":\"VNF\",\"policyScope\":\"resource=vLoadBalancer;type=configuration\",\"policyName\":\"configuration.dcae.microservice.tca.xml\",\"policyVersion\":\"v0.0.1\",\"thresholds\":[{\"closedLoopControlName\":\"CL-LBAL-LOW-TRAFFIC-SIG-FB480F95-A453-6F24-B767-FD703241AB1A\",\"closedLoopEventStatus\":\"ONSET\",\"version\":\"1.0.2\",\"fieldPath\":\"$.event.measurementsForVfScalingFields.vNicPerformanceArray[*].receivedBroadcastPacketsAccumulated\",\"thresholdValue\":500,\"direction\":\"LESS_OR_EQUAL\",\"severity\":\"MAJOR\"},{\"closedLoopControlName\":\"CL-LBAL-LOW-TRAFFIC-SIG-0C5920A6-B564-8035-C878-0E814352BC2B\",\"closedLoopEventStatus\":\"ONSET\",\"version\":\"1.0.2\",\"fieldPath\":\"$.event.measurementsForVfScalingFields.vNicPerformanceArray[*].receivedBroadcastPacketsAccumulated\",\"thresholdValue\":5000,\"direction\":\"GREATER_OR_EQUAL\",\"severity\":\"CRITICAL\"}]}]}\r
\r
+enableRedisCaching=false\r
+redisHosts=127.0.0.1:7000\r
<findbugs.jsr305.version>2.0.1</findbugs.jsr305.version>
<findbugs.annotations.version>3.0.0</findbugs.annotations.version>
<lombok.version>1.16.10</lombok.version>
-
+ <jedis.version>2.9.0</jedis.version>
<json.path.version>2.2.0</json.path.version>
<quartz.version>2.2.0</quartz.version>
<httpclient.version>4.5.2</httpclient.version>
<scope>provided</scope>
</dependency>
+ <!-- DISTRIBUTED CACHING -->
+ <dependency>
+ <groupId>redis.clients</groupId>
+ <artifactId>jedis</artifactId>
+ <version>${jedis.version}</version>
+ </dependency>
<!-- DEPENDENCY INJECTION -->
<dependency>