2 * ===============================LICENSE_START======================================
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
11 * http://www.apache.org/licenses/LICENSE-2.0
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===========================================
21 package org.openecomp.dcae.apod.analytics.cdap.plugins;
23 import co.cask.cdap.api.data.schema.Schema;
24 import co.cask.cdap.etl.api.StageMetrics;
25 import com.fasterxml.jackson.databind.ObjectMapper;
26 import com.google.common.base.Suppliers;
27 import org.apache.hadoop.conf.Configuration;
28 import org.openecomp.dcae.apod.analytics.cdap.common.CDAPPluginConstants;
29 import org.openecomp.dcae.apod.analytics.cdap.plugins.domain.config.dmaap.TestDMaaPMRSinkPluginConfig;
30 import org.openecomp.dcae.apod.analytics.cdap.plugins.domain.config.dmaap.TestDMaaPMRSourcePluginConfig;
31 import org.openecomp.dcae.apod.analytics.cdap.plugins.domain.config.filter.TestJsonPathFilterPluginConfig;
32 import org.openecomp.dcae.apod.analytics.cdap.plugins.domain.config.tca.TestSimpleTCAPluginConfig;
33 import org.openecomp.dcae.apod.analytics.model.util.json.AnalyticsModelObjectMapperSupplier;
34 import org.openecomp.dcae.apod.analytics.test.BaseDCAEAnalyticsUnitTest;
36 import java.io.IOException;
37 import java.io.Serializable;
38 import java.util.LinkedHashMap;
42 * @author Rajiv Singla . Creation Date: 1/23/2017.
44 public abstract class BaseAnalyticsCDAPPluginsUnitTest extends BaseDCAEAnalyticsUnitTest {
46 protected static final ObjectMapper ANALYTICS_MODEL_OBJECT_MAPPER =
47 Suppliers.memoize(new AnalyticsModelObjectMapperSupplier()).get();
49 protected static final String TCA_POLICY_JSON_FILE_LOCATION = "data/json/policy/tca_policy.json";
50 protected static final String CEF_MESSAGE_JSON_FILE_LOCATION = "data/json/cef/cef_message.json";
51 protected static final String CEF_NON_COMPLIANT_MESSAGE_JSON_FILE_LOCATION =
52 "data/json/cef/non_compliant_cef_message.json";
55 protected static final String DMAAP_MR_SOURCE_PLUGIN_REFERENCE_NAME = "testDMaaPMRSource";
56 protected static final String DMAAP_MR_SOURCE_PLUGIN_HOST_NAME = "dcae-msrt-mtl1-ftl.homer.com";
57 protected static final Integer DMAAP_MR_SOURCE_PLUGIN_PORT_NUMBER = 3905;
58 protected static final String DMAAP_MR_SOURCE_PLUGIN_TOPIC_NAME = "com.dcae.dmaap.FTL.DcaeTestVESSub";
59 protected static final Integer DMAAP_MR_SOURCE_PLUGIN_POLLING_INTERVAL = 1000;
60 protected static final String DMAAP_MR_SOURCE_PLUGIN_PROTOCOL = "https";
61 protected static final String DMAAP_MR_SOURCE_PLUGIN_USERNAME = "username";
62 protected static final String DMAAP_MR_SOURCE_PLUGIN_PASSWORD = "password";
63 protected static final String DMAAP_MR_SOURCE_PLUGIN_CONTENT_TYPE = "application/json";
64 protected static final String DMAAP_MR_SOURCE_PLUGIN_CONSUMER_GROUP = "G1";
65 protected static final String DMAAP_MR_SOURCE_PLUGIN_CONSUMER_ID = "C1";
66 protected static final Integer DMAAP_MR_SOURCE_PLUGIN_MESSAGE_LIMIT = 100;
67 protected static final Integer DMAAP_MR_SOURCE_PLUGIN_TIMEOUT = 10000;
70 protected static final String DMAAP_MR_SINK_PLUGIN_REFERENCE_NAME = "testDMaaPMRSINK";
71 protected static final String DMAAP_MR_SINK_PLUGIN_HOST_NAME = "dcae-msrt-mtl1-ftl.homer.com";
72 protected static final Integer DMAAP_MR_SINK_PLUGIN_PORT_NUMBER = 3905;
73 protected static final String DMAAP_MR_SINK_PLUGIN_TOPIC_NAME = "com.dcae.dmaap.FTL.DcaeTestVESPub";
74 protected static final String DMAAP_MR_SINK_PLUGIN_PROTOCOL = "https";
75 protected static final String DMAAP_MR_SINK_PLUGIN_USERNAME = "username";
76 protected static final String DMAAP_MR_SINK_PLUGIN_PASSWORD = "password";
77 protected static final String DMAAP_MR_SINK_PLUGIN_CONTENT_TYPE = "application/json";
78 protected static final String DMAAP_MR_SINK_MESSAGE_COLUMN_NAME = "message";
79 protected static final Integer DMAAP_MR_SINK_PLUGIN_MAX_BATCH_SIZE = 10;
80 protected static final Integer DMAAP_MR_SINK_PLUGIN_MAX_RECOVERY_QUEUE_SIZE = 100;
82 protected static final String VES_MESSAGE_FIELD_NAME = "message";
83 protected static final String TCA_PLUGIN_ALERT_FIELD_NAME = "alert";
84 protected static final String TCA_PLUGIN_MESSAGE_TYPE_FIELD_NAME = "tcaMessageType";
87 protected static final String JSON_PATH_FILTER_PLUGIN_REFERENCE_NAME = "JsonPathFilter";
88 protected static final String JSON_PATH_FILTER_PLUGIN_INCOMING_JSON_FIELD_NAME = "message";
89 protected static final String JSON_PATH_FILTER_PLUGIN_OUTPUT_SCHEMA_FILED_NAME = "filterMatched";
90 protected static final String JSON_PATH_FILTER_PLUGIN_JSON_FILTER_MAPPINGS =
91 "$.event.commonEventHeader.domain:measurementsForVfScaling," +
92 "$.event.commonEventHeader.functionalRole:vLoadBalancer;vFirewall";
93 protected static final String JSON_PATH_FILTER_PLUGIN_JSON_FILTER_OUTPUT_SCHEMA =
94 "{\"type\":\"record\"," +
95 "\"name\":\"etlSchemaBody\",\"fields\":" +
97 "{\"name\":\"ts\",\"type\":\"long\"}," +
98 "{\"name\":\"filterMatched\",\"type\":[\"boolean\",\"null\"]}," +
99 "{\"name\":\"responseCode\",\"type\":\"int\"}," +
100 "{\"name\":\"responseMessage\",\"type\":\"string\"}," +
101 "{\"name\":\"message\",\"type\":\"string\"}" +
105 protected static class MockStageMetrics implements StageMetrics, Serializable {
108 public void count(String metricName, int delta) {
109 LOG.debug("Mocking metric count, MetricName: {}, Delta: {}", metricName, delta);
113 public void gauge(String metricName, long value) {
114 LOG.debug("Mocking metric guage, MetricName: {}, Value: {}", metricName, value);
118 public void pipelineCount(String metricName, int delta) {
119 LOG.debug("Mocking metric pipelineCount, MetricName: {}, Delta: {}", metricName, delta);
123 public void pipelineGauge(String metricName, long value) {
124 LOG.debug("Mocking metric guage, pipelineGauge: {}, Value: {}", metricName, value);
128 protected static TestDMaaPMRSourcePluginConfig getTestDMaaPMRSourcePluginConfig() {
129 final TestDMaaPMRSourcePluginConfig sourcePluginConfig = new TestDMaaPMRSourcePluginConfig();
130 sourcePluginConfig.setReferenceName(DMAAP_MR_SOURCE_PLUGIN_REFERENCE_NAME);
131 sourcePluginConfig.setHostName(DMAAP_MR_SOURCE_PLUGIN_HOST_NAME);
132 sourcePluginConfig.setPortNumber(DMAAP_MR_SOURCE_PLUGIN_PORT_NUMBER);
133 sourcePluginConfig.setTopicName(DMAAP_MR_SOURCE_PLUGIN_TOPIC_NAME);
134 sourcePluginConfig.setPollingInterval(DMAAP_MR_SOURCE_PLUGIN_POLLING_INTERVAL);
135 sourcePluginConfig.setProtocol(DMAAP_MR_SOURCE_PLUGIN_PROTOCOL);
136 sourcePluginConfig.setUserName(DMAAP_MR_SOURCE_PLUGIN_USERNAME);
137 sourcePluginConfig.setUserPassword(DMAAP_MR_SOURCE_PLUGIN_PASSWORD);
138 sourcePluginConfig.setContentType(DMAAP_MR_SOURCE_PLUGIN_CONTENT_TYPE);
139 sourcePluginConfig.setConsumerGroup(DMAAP_MR_SOURCE_PLUGIN_CONSUMER_GROUP);
140 sourcePluginConfig.setConsumerId(DMAAP_MR_SOURCE_PLUGIN_CONSUMER_ID);
141 sourcePluginConfig.setMessageLimit(DMAAP_MR_SOURCE_PLUGIN_MESSAGE_LIMIT);
142 sourcePluginConfig.setTimeoutMS(DMAAP_MR_SOURCE_PLUGIN_TIMEOUT);
143 return sourcePluginConfig;
146 protected static TestDMaaPMRSinkPluginConfig getTestDMaaPMRSinkPluginConfig() {
147 final TestDMaaPMRSinkPluginConfig sinkPluginConfig = new TestDMaaPMRSinkPluginConfig();
148 sinkPluginConfig.setReferenceName(DMAAP_MR_SINK_PLUGIN_REFERENCE_NAME);
149 sinkPluginConfig.setHostName(DMAAP_MR_SINK_PLUGIN_HOST_NAME);
150 sinkPluginConfig.setPortNumber(DMAAP_MR_SINK_PLUGIN_PORT_NUMBER);
151 sinkPluginConfig.setTopicName(DMAAP_MR_SINK_PLUGIN_TOPIC_NAME);
152 sinkPluginConfig.setProtocol(DMAAP_MR_SINK_PLUGIN_PROTOCOL);
153 sinkPluginConfig.setUserName(DMAAP_MR_SINK_PLUGIN_USERNAME);
154 sinkPluginConfig.setUserPassword(DMAAP_MR_SINK_PLUGIN_PASSWORD);
155 sinkPluginConfig.setContentType(DMAAP_MR_SINK_PLUGIN_CONTENT_TYPE);
156 sinkPluginConfig.setMessageColumnName(DMAAP_MR_SINK_MESSAGE_COLUMN_NAME);
157 sinkPluginConfig.setMaxBatchSize(DMAAP_MR_SINK_PLUGIN_MAX_BATCH_SIZE);
158 sinkPluginConfig.setMaxRecoveryQueueSize(DMAAP_MR_SINK_PLUGIN_MAX_RECOVERY_QUEUE_SIZE);
159 return sinkPluginConfig;
163 protected static Configuration getTestConfiguration() {
164 final Configuration configuration = new Configuration();
165 final Map<String, String> sinkConfigurationMap = createSinkConfigurationMap();
166 for (Map.Entry<String, String> property : sinkConfigurationMap.entrySet()) {
167 configuration.set(property.getKey(), property.getValue());
169 return configuration;
172 protected static Map<String, String> createSinkConfigurationMap() {
174 Map<String, String> sinkConfig = new LinkedHashMap<>();
175 sinkConfig.put(CDAPPluginConstants.DMaaPMRSinkHadoopConfigFields.HOST_NAME, DMAAP_MR_SINK_PLUGIN_HOST_NAME);
176 sinkConfig.put(CDAPPluginConstants.DMaaPMRSinkHadoopConfigFields.TOPIC_NAME, DMAAP_MR_SINK_PLUGIN_TOPIC_NAME);
177 sinkConfig.put(CDAPPluginConstants.DMaaPMRSinkHadoopConfigFields.PORT_NUMBER,
178 DMAAP_MR_SINK_PLUGIN_PORT_NUMBER.toString());
179 sinkConfig.put(CDAPPluginConstants.DMaaPMRSinkHadoopConfigFields.PROTOCOL, DMAAP_MR_SINK_PLUGIN_PROTOCOL);
180 sinkConfig.put(CDAPPluginConstants.DMaaPMRSinkHadoopConfigFields.USER_NAME, DMAAP_MR_SINK_PLUGIN_USERNAME);
181 sinkConfig.put(CDAPPluginConstants.DMaaPMRSinkHadoopConfigFields.USER_PASS, DMAAP_MR_SINK_PLUGIN_PASSWORD);
182 sinkConfig.put(CDAPPluginConstants.DMaaPMRSinkHadoopConfigFields.CONTENT_TYPE,
183 DMAAP_MR_SINK_PLUGIN_CONTENT_TYPE);
184 sinkConfig.put(CDAPPluginConstants.DMaaPMRSinkHadoopConfigFields.MAX_BATCH_SIZE,
185 DMAAP_MR_SINK_PLUGIN_MAX_BATCH_SIZE.toString());
186 sinkConfig.put(CDAPPluginConstants.DMaaPMRSinkHadoopConfigFields.MAX_RECOVER_QUEUE_SIZE,
187 DMAAP_MR_SINK_PLUGIN_MAX_RECOVERY_QUEUE_SIZE.toString());
191 protected static Schema getDMaaPMRSinkTestSchema() {
192 return Schema.recordOf(
193 "DMaaPMRSinkTestSchema",
194 Schema.Field.of("message", Schema.of(Schema.Type.STRING)),
195 Schema.Field.of("field1", Schema.of(Schema.Type.STRING))
200 protected static TestSimpleTCAPluginConfig getTestSimpleTCAPluginConfig() {
201 final String policyJson;
203 policyJson = fromStream(TCA_POLICY_JSON_FILE_LOCATION);
204 } catch (IOException e) {
205 throw new RuntimeException("Error while parsing policy", e);
207 return new TestSimpleTCAPluginConfig(VES_MESSAGE_FIELD_NAME, policyJson, TCA_PLUGIN_ALERT_FIELD_NAME,
208 TCA_PLUGIN_MESSAGE_TYPE_FIELD_NAME, getSimpleTCAPluginInputSchema().toString(), false);
211 protected static Schema getSimpleTCAPluginInputSchema() {
212 return Schema.recordOf(
213 "TestSimpleTCAPluginInputSchema",
214 Schema.Field.of("message", Schema.of(Schema.Type.STRING)),
215 Schema.Field.of("inputField1", Schema.nullableOf(Schema.of(Schema.Type.STRING))),
216 Schema.Field.of("inputField2", Schema.nullableOf(Schema.of(Schema.Type.STRING)))
220 protected static Schema getJsonFilterPluginInputSchema() {
221 return Schema.recordOf(
222 "TestJsonFilterInputSchema",
223 Schema.Field.of("ts", Schema.of(Schema.Type.LONG)),
224 Schema.Field.of("responseCode", Schema.of(Schema.Type.INT)),
225 Schema.Field.of("responseMessage", Schema.of(Schema.Type.STRING)),
226 Schema.Field.of("message", Schema.of(Schema.Type.STRING))
230 protected static TestJsonPathFilterPluginConfig getJsonPathFilterPluginConfig() {
231 return new TestJsonPathFilterPluginConfig(JSON_PATH_FILTER_PLUGIN_REFERENCE_NAME,
232 JSON_PATH_FILTER_PLUGIN_INCOMING_JSON_FIELD_NAME,
233 JSON_PATH_FILTER_PLUGIN_OUTPUT_SCHEMA_FILED_NAME,
234 JSON_PATH_FILTER_PLUGIN_JSON_FILTER_MAPPINGS,
235 JSON_PATH_FILTER_PLUGIN_JSON_FILTER_OUTPUT_SCHEMA);