diff options
Diffstat (limited to 'dcae-analytics-cdap-plugins/src/main/java/org/onap/dcae/apod/analytics/cdap/plugins/streaming/dmaap')
4 files changed, 393 insertions, 0 deletions
diff --git a/dcae-analytics-cdap-plugins/src/main/java/org/onap/dcae/apod/analytics/cdap/plugins/streaming/dmaap/DMaaPMRReceiver.java b/dcae-analytics-cdap-plugins/src/main/java/org/onap/dcae/apod/analytics/cdap/plugins/streaming/dmaap/DMaaPMRReceiver.java new file mode 100644 index 0000000..593d5d1 --- /dev/null +++ b/dcae-analytics-cdap-plugins/src/main/java/org/onap/dcae/apod/analytics/cdap/plugins/streaming/dmaap/DMaaPMRReceiver.java @@ -0,0 +1,118 @@ +/* + * ===============================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.onap.dcae.apod.analytics.cdap.plugins.streaming.dmaap; + +import co.cask.cdap.api.data.format.StructuredRecord; +import co.cask.cdap.api.metrics.Metrics; +import com.google.common.base.Optional; +import org.apache.spark.storage.StorageLevel; +import org.apache.spark.streaming.receiver.Receiver; +import org.onap.dcae.apod.analytics.cdap.common.utils.DMaaPMRUtils; +import org.onap.dcae.apod.analytics.cdap.plugins.domain.config.dmaap.DMaaPMRSourcePluginConfig; +import org.onap.dcae.apod.analytics.cdap.plugins.utils.CDAPPluginUtils; +import org.onap.dcae.apod.analytics.cdap.plugins.utils.DMaaPSourceConfigMapper; +import org.onap.dcae.apod.analytics.common.exception.DCAEAnalyticsRuntimeException; +import org.onap.dcae.apod.analytics.dmaap.DMaaPMRFactory; +import org.onap.dcae.apod.analytics.dmaap.service.subscriber.DMaaPMRSubscriber; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.List; +import java.util.concurrent.TimeUnit; + +/** + * DMaaP MR Receiver which calls DMaaP MR Topic and stores structured records + * <p> + * @author Rajiv Singla . Creation Date: 1/19/2017. + */ +public class DMaaPMRReceiver extends Receiver<StructuredRecord> { + + private static final Logger LOG = LoggerFactory.getLogger(DMaaPMRReceiver.class); + private static final long serialVersionUID = 1L; + + private final DMaaPMRSourcePluginConfig pluginConfig; + private final transient Metrics metrics; + + public DMaaPMRReceiver(final StorageLevel storageLevel, final DMaaPMRSourcePluginConfig pluginConfig, + final Metrics metrics) { + super(storageLevel); + this.pluginConfig = pluginConfig; + this.metrics = metrics; + LOG.debug("Created DMaaP MR Receiver instance with plugin Config: {}", pluginConfig); + } + + @Override + public void onStart() { + + // create DMaaP MR Subscriber + final DMaaPMRSubscriber subscriber = + DMaaPMRFactory.create().createSubscriber(DMaaPSourceConfigMapper.map(pluginConfig)); + + // Start a new thread with indefinite loop until receiver is stopped + new Thread() { + @Override + public void run() { + while (!isStopped()) { + storeStructuredRecords(subscriber); + try { + final Integer pollingInterval = pluginConfig.getPollingInterval(); + LOG.debug("DMaaP MR Receiver sleeping for polling interval: {}", pollingInterval); + TimeUnit.MILLISECONDS.sleep(pollingInterval); + } catch (InterruptedException e) { + final String errorMessage = String.format( + "Interrupted Exception while DMaaP MR Receiver sleeping polling interval: %s", e); + throw new DCAEAnalyticsRuntimeException(errorMessage, LOG, e); + } + } + } + }.start(); + + } + + @Override + public void onStop() { + LOG.debug("Stopping DMaaP MR Receiver with plugin config: {}", pluginConfig); + } + + /** + * Fetches records from DMaaP MR Subscriber and store them as structured records + * + * @param subscriber DMaaP MR Subscriber Instance + */ + public void storeStructuredRecords(final DMaaPMRSubscriber subscriber) { + + LOG.debug("DMaaP MR Receiver start fetching messages from DMaaP MR Topic"); + + // Fetch messages from DMaaP MR Topic + final Optional<List<String>> subscriberMessagesOptional = + DMaaPMRUtils.getSubscriberMessages(subscriber, metrics); + + // store records + if (subscriberMessagesOptional.isPresent()) { + final List<String> messages = subscriberMessagesOptional.get(); + for (final String message : messages) { + store(CDAPPluginUtils.createDMaaPMRResponseStructuredRecord(message)); + } + LOG.debug("Stored DMaaP Subscriber messages as Structured Records. Message count {}", messages.size()); + } + } + +} diff --git a/dcae-analytics-cdap-plugins/src/main/java/org/onap/dcae/apod/analytics/cdap/plugins/streaming/dmaap/DMaaPMRSource.java b/dcae-analytics-cdap-plugins/src/main/java/org/onap/dcae/apod/analytics/cdap/plugins/streaming/dmaap/DMaaPMRSource.java new file mode 100644 index 0000000..b85f653 --- /dev/null +++ b/dcae-analytics-cdap-plugins/src/main/java/org/onap/dcae/apod/analytics/cdap/plugins/streaming/dmaap/DMaaPMRSource.java @@ -0,0 +1,70 @@ +/* + * ===============================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.onap.dcae.apod.analytics.cdap.plugins.streaming.dmaap; + +import co.cask.cdap.api.annotation.Description; +import co.cask.cdap.api.annotation.Name; +import co.cask.cdap.api.annotation.Plugin; +import co.cask.cdap.api.data.format.StructuredRecord; +import co.cask.cdap.etl.api.PipelineConfigurer; +import co.cask.cdap.etl.api.streaming.StreamingContext; +import co.cask.cdap.etl.api.streaming.StreamingSource; +import org.apache.spark.storage.StorageLevel; +import org.apache.spark.streaming.api.java.JavaDStream; +import org.onap.dcae.apod.analytics.cdap.common.utils.ValidationUtils; +import org.onap.dcae.apod.analytics.cdap.plugins.domain.config.dmaap.DMaaPMRSourcePluginConfig; +import org.onap.dcae.apod.analytics.cdap.plugins.domain.schema.dmaap.DMaaPSourceOutputSchema; +import org.onap.dcae.apod.analytics.cdap.plugins.validator.DMaaPMRSourcePluginConfigValidator; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * DMaaP MR Source Plugin which polls DMaaP MR topic at frequent intervals + * <p> + * @author Rajiv Singla . Creation Date: 1/18/2017. + */ +@Plugin(type = StreamingSource.PLUGIN_TYPE) +@Name("DMaaPMRSource") +@Description("Fetches DMaaP MR Messages at regular intervals") +public class DMaaPMRSource extends StreamingSource<StructuredRecord> { + + private static final Logger LOG = LoggerFactory.getLogger(DMaaPMRSource.class); + private static final long serialVersionUID = 1L; + + private final DMaaPMRSourcePluginConfig pluginConfig; + + public DMaaPMRSource(final DMaaPMRSourcePluginConfig pluginConfig) { + LOG.debug("Creating DMaaP MR Source plugin with plugin Config: {}", pluginConfig); + this.pluginConfig = pluginConfig; + } + + @Override + public void configurePipeline(PipelineConfigurer pipelineConfigurer) { + ValidationUtils.validateSettings(pluginConfig, new DMaaPMRSourcePluginConfigValidator()); + pipelineConfigurer.getStageConfigurer().setOutputSchema(DMaaPSourceOutputSchema.getSchema()); + } + + @Override + public JavaDStream<StructuredRecord> getStream(final StreamingContext streamingContext) throws Exception { + return streamingContext.getSparkStreamingContext().receiverStream( + new DMaaPMRReceiver(StorageLevel.MEMORY_ONLY(), pluginConfig, streamingContext.getMetrics())); + } +} diff --git a/dcae-analytics-cdap-plugins/src/main/java/org/onap/dcae/apod/analytics/cdap/plugins/streaming/dmaap/MockDMaaPMRReceiver.java b/dcae-analytics-cdap-plugins/src/main/java/org/onap/dcae/apod/analytics/cdap/plugins/streaming/dmaap/MockDMaaPMRReceiver.java new file mode 100644 index 0000000..f4f781a --- /dev/null +++ b/dcae-analytics-cdap-plugins/src/main/java/org/onap/dcae/apod/analytics/cdap/plugins/streaming/dmaap/MockDMaaPMRReceiver.java @@ -0,0 +1,132 @@ +/* + * ===============================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.onap.dcae.apod.analytics.cdap.plugins.streaming.dmaap; + +import co.cask.cdap.api.data.format.StructuredRecord; +import com.fasterxml.jackson.core.type.TypeReference; +import org.apache.spark.storage.StorageLevel; +import org.apache.spark.streaming.receiver.Receiver; +import org.onap.dcae.apod.analytics.cdap.plugins.domain.config.dmaap.DMaaPMRSourcePluginConfig; +import org.onap.dcae.apod.analytics.cdap.plugins.utils.CDAPPluginUtils; +import org.onap.dcae.apod.analytics.cdap.plugins.utils.DMaaPSourceConfigMapper; +import org.onap.dcae.apod.analytics.common.exception.DCAEAnalyticsRuntimeException; +import org.onap.dcae.apod.analytics.dmaap.DMaaPMRFactory; +import org.onap.dcae.apod.analytics.dmaap.service.subscriber.DMaaPMRSubscriber; +import org.onap.dcae.apod.analytics.model.domain.cef.EventListener; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.FileNotFoundException; +import java.io.IOException; +import java.io.InputStream; +import java.util.List; +import java.util.concurrent.TimeUnit; + +import static org.onap.dcae.apod.analytics.model.util.AnalyticsModelJsonUtils.readValue; +import static org.onap.dcae.apod.analytics.model.util.AnalyticsModelJsonUtils.writeValueAsString; + +/** + * DMaaP MR Receiver which calls DMaaP MR Topic and stores structured records + * <p> + * @author Rajiv Singla . Creation Date: 1/19/2017. + */ +public class MockDMaaPMRReceiver extends Receiver<StructuredRecord> { + + private static final Logger LOG = LoggerFactory.getLogger(MockDMaaPMRReceiver.class); + private static final long serialVersionUID = 1L; + + private static final String MOCK_MESSAGE_FILE_LOCATION = "ves_mock_messages.json"; + private static final TypeReference<List<EventListener>> EVENT_LISTENER_TYPE_REFERENCE = + new TypeReference<List<EventListener>>() { + }; + + private final DMaaPMRSourcePluginConfig pluginConfig; + + public MockDMaaPMRReceiver(final StorageLevel storageLevel, final DMaaPMRSourcePluginConfig pluginConfig) { + super(storageLevel); + this.pluginConfig = pluginConfig; + LOG.debug("Created DMaaP MR Receiver instance with plugin Config: {}", pluginConfig); + } + + @Override + public void onStart() { + + // create DMaaP MR Subscriber + final DMaaPMRSubscriber subscriber = + DMaaPMRFactory.create().createSubscriber(DMaaPSourceConfigMapper.map(pluginConfig)); + storeStructuredRecords(subscriber); + + } + + @Override + public void onStop() { + LOG.debug("Stopping DMaaP MR Receiver with plugin config: {}", pluginConfig); + } + + /** + * Fetches records from DMaaP MR Subscriber and store them as structured records + * + * @param subscriber DMaaP MR Subscriber Instance + */ + public void storeStructuredRecords(final DMaaPMRSubscriber subscriber) { + + LOG.debug("DMaaP MR Receiver start fetching messages from DMaaP MR Topic"); + + try (InputStream resourceAsStream = + Thread.currentThread().getContextClassLoader().getResourceAsStream(MOCK_MESSAGE_FILE_LOCATION)) { + + if (resourceAsStream == null) { + LOG.error("Unable to find file at location: {}", MOCK_MESSAGE_FILE_LOCATION); + throw new DCAEAnalyticsRuntimeException("Unable to find file", LOG, new FileNotFoundException()); + } + + List<EventListener> eventListeners = readValue(resourceAsStream, EVENT_LISTENER_TYPE_REFERENCE); + + final int totalMessageCount = eventListeners.size(); + LOG.debug("Mock message count to be written to cdap stream: ()", totalMessageCount); + + int i = 1; + for (EventListener eventListener : eventListeners) { + if (isStopped()) { + return; + } + final String eventListenerString = writeValueAsString(eventListener); + LOG.debug("=======>> Writing message to cdap stream no: {} of {}", i, totalMessageCount); + store(CDAPPluginUtils.createDMaaPMRResponseStructuredRecord(eventListenerString)); + i++; + try { + TimeUnit.MILLISECONDS.sleep(pluginConfig.getPollingInterval()); + } catch (InterruptedException e) { + LOG.error("Error while sleeping"); + throw new DCAEAnalyticsRuntimeException("Error while sleeping", LOG, e); + } + + } + + LOG.debug("Finished writing mock messages to CDAP Stream"); + + } catch (IOException e) { + LOG.error("Error while parsing json file"); + throw new DCAEAnalyticsRuntimeException("Error while parsing mock json file", LOG, e); + } + } + +} diff --git a/dcae-analytics-cdap-plugins/src/main/java/org/onap/dcae/apod/analytics/cdap/plugins/streaming/dmaap/MockDMaaPMRSource.java b/dcae-analytics-cdap-plugins/src/main/java/org/onap/dcae/apod/analytics/cdap/plugins/streaming/dmaap/MockDMaaPMRSource.java new file mode 100644 index 0000000..53b63e0 --- /dev/null +++ b/dcae-analytics-cdap-plugins/src/main/java/org/onap/dcae/apod/analytics/cdap/plugins/streaming/dmaap/MockDMaaPMRSource.java @@ -0,0 +1,73 @@ +/* + * ===============================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.onap.dcae.apod.analytics.cdap.plugins.streaming.dmaap; + +import co.cask.cdap.api.annotation.Description; +import co.cask.cdap.api.annotation.Name; +import co.cask.cdap.api.annotation.Plugin; +import co.cask.cdap.api.data.format.StructuredRecord; +import co.cask.cdap.etl.api.PipelineConfigurer; +import co.cask.cdap.etl.api.streaming.StreamingContext; +import co.cask.cdap.etl.api.streaming.StreamingSource; +import org.apache.spark.storage.StorageLevel; +import org.apache.spark.streaming.api.java.JavaDStream; +import org.onap.dcae.apod.analytics.cdap.common.exception.CDAPSettingsException; +import org.onap.dcae.apod.analytics.cdap.plugins.domain.config.dmaap.DMaaPMRSourcePluginConfig; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * A mock implementation of DMaaP MR Receiver which sends mock ves messages + * <p> + * @author Rajiv Singla . Creation Date: 2/15/2017. + */ +@Plugin(type = StreamingSource.PLUGIN_TYPE) +@Name("MockDMaaPMRSource") +@Description("Fetches DMaaP MR Messages at regular intervals") +public class MockDMaaPMRSource extends StreamingSource<StructuredRecord> { + + private static final Logger LOG = LoggerFactory.getLogger(MockDMaaPMRSource.class); + private static final long serialVersionUID = 1L; + + private final DMaaPMRSourcePluginConfig pluginConfig; + + public MockDMaaPMRSource(final DMaaPMRSourcePluginConfig pluginConfig) { + LOG.debug("Creating DMaaP MR Source plugin with plugin Config: {}", pluginConfig); + this.pluginConfig = pluginConfig; + } + + @Override + public void configurePipeline(PipelineConfigurer pipelineConfigurer) { + final Integer pollingInterval = pluginConfig.getPollingInterval(); + if (pollingInterval == null) { + final String errorMessage = "Polling Interval field must be present"; + throw new CDAPSettingsException(errorMessage, LOG, new IllegalArgumentException(errorMessage)); + } else { + LOG.info("Mock Message will be send every ms: {}", pollingInterval); + } + } + + @Override + public JavaDStream<StructuredRecord> getStream(final StreamingContext streamingContext) throws Exception { + return streamingContext.getSparkStreamingContext().receiverStream( + new MockDMaaPMRReceiver(StorageLevel.MEMORY_ONLY(), pluginConfig)); + } +} |