diff options
Diffstat (limited to 'src/main/java/org/onap/dcae/common/publishing')
7 files changed, 593 insertions, 0 deletions
diff --git a/src/main/java/org/onap/dcae/common/publishing/DMaaPConfigurationParser.java b/src/main/java/org/onap/dcae/common/publishing/DMaaPConfigurationParser.java new file mode 100644 index 00000000..274e4490 --- /dev/null +++ b/src/main/java/org/onap/dcae/common/publishing/DMaaPConfigurationParser.java @@ -0,0 +1,112 @@ +/*- + * ============LICENSE_START======================================================= + * org.onap.dcaegen2.collectors.ves + * ================================================================================ + * Copyright (C) 2018 Nokia. 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.common.publishing; + +import io.vavr.collection.List; +import io.vavr.collection.Map; +import io.vavr.control.Option; +import io.vavr.control.Try; +import org.onap.dcae.common.AnyNode; + +import java.net.URL; +import java.nio.file.Files; +import java.nio.file.Path; + +import org.json.JSONObject; + +import static io.vavr.API.*; +import static org.onap.dcae.common.publishing.VavrUtils.enhanceError; +import static org.onap.dcae.common.publishing.VavrUtils.f; + +/** + * @author Pawel Szalapski (pawel.szalapski@nokia.com) + */ +@SuppressWarnings("mapFailure takes a generic varargs, unchecked because of Javas type system limitation, actually safe to do") +public final class DMaaPConfigurationParser { + + public static Try<Map<String, PublisherConfig>> parseToDomainMapping(Path configLocation) { + return readFromFile(configLocation) + .flatMap(DMaaPConfigurationParser::toJSON) + .flatMap(DMaaPConfigurationParser::toConfigMap); + } + + public static Try<Map<String, PublisherConfig>> parseToDomainMapping(JSONObject config) { + return toJSON(config.toString()) + .flatMap(DMaaPConfigurationParser::toConfigMap); + } + + private static Try<String> readFromFile(Path configLocation) { + return Try(() -> new String(Files.readAllBytes(configLocation))) + .mapFailure(enhanceError(f("Could not read DMaaP configuration from location: '%s'", configLocation))); + } + + private static Try<AnyNode> toJSON(String config) { + return Try(() -> AnyNode.fromString(config)) + .mapFailure(enhanceError(f("DMaaP configuration '%s' is not a valid JSON document", config))); + } + + private static Try<Map<String, PublisherConfig>> toConfigMap(AnyNode config) { + return Try(() -> usesLegacyFormat(config) ? parseLegacyFormat(config) : parseNewFormat(config)) + .mapFailure(enhanceError(f("Parsing DMaaP configuration: '%s' failed, probably it is in unexpected format", config))); + } + + private static boolean usesLegacyFormat(AnyNode dMaaPConfig) { + return dMaaPConfig.has("channels"); + } + + private static Map<String, PublisherConfig> parseLegacyFormat(AnyNode root) { + return root.get("channels").toList().toMap( + channel -> channel.get("name").toString(), + channel -> { + String destinationsStr = channel.getAsOption("cambria.url") + .getOrElse(channel.getAsOption("cambria.hosts").get()) + .toString(); + String topic = channel.get("cambria.topic").toString(); + Option<String> maybeUser = channel.getAsOption("basicAuthUsername").map(AnyNode::toString); + Option<String> maybePassword = channel.getAsOption("basicAuthPassword").map(AnyNode::toString); + List<String> destinations = List(destinationsStr.split(",")); + return buildBasedOnAuth(maybeUser, maybePassword, topic, destinations); + }); + } + + private static Map<String, PublisherConfig> parseNewFormat(AnyNode root) { + return root.keys().toMap( + channelName -> channelName, + channelName -> { + AnyNode channelConfig = root.get(channelName); + Option<String> maybeUser = channelConfig.getAsOption("aaf_username").map(AnyNode::toString); + Option<String> maybePassword = channelConfig.getAsOption("aaf_password").map(AnyNode::toString); + URL topicURL = unchecked( + () -> new URL(channelConfig.get("dmaap_info").get("topic_url").toString())).apply(); + String[] pathSegments = topicURL.getPath().substring(1).split("/"); + String topic = pathSegments[1]; + String destination = "events".equals(pathSegments[0]) ? topicURL.getAuthority() : topicURL.getHost(); + List<String> destinations = List(destination); + return buildBasedOnAuth(maybeUser, maybePassword, topic, destinations); + }); + } + + private static PublisherConfig buildBasedOnAuth(Option<String> maybeUser, Option<String> maybePassword, + String topic, List<String> destinations) { + return maybeUser.flatMap(user -> maybePassword.map(password -> Tuple(user, password))) + .map(credentials -> new PublisherConfig(destinations, topic, credentials._1, credentials._2)) + .getOrElse(new PublisherConfig(destinations, topic)); + } +} diff --git a/src/main/java/org/onap/dcae/common/publishing/DMaaPEventPublisher.java b/src/main/java/org/onap/dcae/common/publishing/DMaaPEventPublisher.java new file mode 100644 index 00000000..aa3dc7a3 --- /dev/null +++ b/src/main/java/org/onap/dcae/common/publishing/DMaaPEventPublisher.java @@ -0,0 +1,100 @@ +/*- + * ============LICENSE_START======================================================= + * org.onap.dcaegen2.collectors.ves + * ================================================================================ + * Copyright (C) 2017 AT&T Intellectual Property. All rights reserved. + * Copyright (C) 2018 Nokia. 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.common.publishing; + +import com.att.nsa.cambria.client.CambriaBatchingPublisher; +import com.att.nsa.clock.SaClock; +import com.att.nsa.logging.LoggingContext; +import com.att.nsa.logging.log4j.EcompFields; +import io.vavr.collection.Map; +import io.vavr.control.Try; +import org.json.JSONObject; +import org.onap.dcae.common.VESLogger; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.IOException; + +import static org.onap.dcae.common.publishing.VavrUtils.f; + +/** + * @author Pawel Szalapski (pawel.szalapski@nokia.com) + */ +class DMaaPEventPublisher implements EventPublisher { + private static final int PENDING_MESSAGE_LOG_THRESHOLD = 100; + private static final String VES_UNIQUE_ID = "VESuniqueId"; + private static final Logger log = LoggerFactory.getLogger(DMaaPEventPublisher.class); + private final DMaaPPublishersCache publishersCache; + private final Logger outputLogger; + + DMaaPEventPublisher(DMaaPPublishersCache DMaaPPublishersCache, + Logger outputLogger) { + this.publishersCache = DMaaPPublishersCache; + this.outputLogger = outputLogger; + } + + @Override + public void sendEvent(JSONObject event, String domain) { + clearVesUniqueIdFromEvent(event); + publishersCache.getPublisher(domain) + .onEmpty(() -> + log.warn(f("Could not find event publisher for domain: '%s', dropping message: '%s'", domain, event))) + .forEach(publisher -> sendEvent(event, domain, publisher)); + } + + @Override + public void reconfigure(Map<String, PublisherConfig> dMaaPConfig) { + publishersCache.reconfigure(dMaaPConfig); + } + + private void sendEvent(JSONObject event, String domain, CambriaBatchingPublisher publisher) { + Try.run(() -> uncheckedSendEvent(event, domain, publisher)) + .onFailure(exc -> closePublisher(event, domain, exc)); + } + + private void uncheckedSendEvent(JSONObject event, String domain, CambriaBatchingPublisher publisher) + throws IOException { + int pendingMsgs = publisher.send("MyPartitionKey", event.toString()); + if (pendingMsgs > PENDING_MESSAGE_LOG_THRESHOLD) { + log.info("Pending messages count: " + pendingMsgs); + } + String infoMsg = f("Event: '%s' scheduled to be send asynchronously on domain: '%s'", event, domain); + log.info(infoMsg); + outputLogger.info(infoMsg); + } + + private void closePublisher(JSONObject event, String domain, Throwable e) { + log.error(f("Unable to schedule event: '%s' on domain: '%s'. Closing publisher and dropping message.", + event, domain), e); + publishersCache.closePublisherFor(domain); + } + + private void clearVesUniqueIdFromEvent(JSONObject event) { + if (event.has(VES_UNIQUE_ID)) { + String uuid = event.get(VES_UNIQUE_ID).toString(); + LoggingContext localLC = VESLogger.getLoggingContextForThread(uuid); + localLC.put(EcompFields.kBeginTimestampMs, SaClock.now()); + log.debug("Removing VESuniqueid object from event"); + event.remove(VES_UNIQUE_ID); + } + } +} diff --git a/src/main/java/org/onap/dcae/common/publishing/DMaaPPublishersBuilder.java b/src/main/java/org/onap/dcae/common/publishing/DMaaPPublishersBuilder.java new file mode 100644 index 00000000..a93073bf --- /dev/null +++ b/src/main/java/org/onap/dcae/common/publishing/DMaaPPublishersBuilder.java @@ -0,0 +1,61 @@ +/*- + * ============LICENSE_START======================================================= + * org.onap.dcaegen2.collectors.ves + * ================================================================================ + * Copyright (C) 2018 Nokia. 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.common.publishing; + +import com.att.nsa.cambria.client.CambriaBatchingPublisher; +import com.att.nsa.cambria.client.CambriaClientBuilders; +import com.att.nsa.cambria.client.CambriaClientBuilders.PublisherBuilder; +import io.vavr.control.Try; + +import static io.vavr.API.Try; +import static org.onap.dcae.common.publishing.VavrUtils.enhanceError; +import static org.onap.dcae.common.publishing.VavrUtils.f; + +/** + * @author Pawel Szalapski (pawel.szalapski@nokia.com) + */ +final class DMaaPPublishersBuilder { + + static Try<CambriaBatchingPublisher> buildPublisher(PublisherConfig config) { + return Try(() -> builder(config).build()) + .mapFailure(enhanceError(f("DMaaP client builder throws exception for this configuration: '%s'", config))); + } + + private static PublisherBuilder builder(PublisherConfig config) { + if (config.isSecured()) { + return authenticatedBuilder(config); + } else { + return unAuthenticatedBuilder(config); + } + } + + private static PublisherBuilder authenticatedBuilder(PublisherConfig config) { + return unAuthenticatedBuilder(config) + .usingHttps() + .authenticatedByHttp(config.userName().get(), config.password().get()); + } + + private static PublisherBuilder unAuthenticatedBuilder(PublisherConfig config) { + return new CambriaClientBuilders.PublisherBuilder() + .usingHosts(config.destinations().mkString(",")) + .onTopic(config.topic()) + .logSendFailuresAfter(5); + } +} diff --git a/src/main/java/org/onap/dcae/common/publishing/DMaaPPublishersCache.java b/src/main/java/org/onap/dcae/common/publishing/DMaaPPublishersCache.java new file mode 100644 index 00000000..b7997ef9 --- /dev/null +++ b/src/main/java/org/onap/dcae/common/publishing/DMaaPPublishersCache.java @@ -0,0 +1,121 @@ +/*- + * ============LICENSE_START======================================================= + * org.onap.dcaegen2.collectors.ves + * ================================================================================ + * Copyright (C) 2017-2018 AT&T Intellectual Property. All rights reserved. + * Copyright (C) 2018 Nokia. 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.common.publishing; + +import com.att.nsa.cambria.client.CambriaBatchingPublisher; +import com.google.common.cache.*; +import io.vavr.collection.Map; +import io.vavr.control.Option; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import javax.annotation.Nonnull; +import java.io.IOException; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicReference; + +import static io.vavr.API.Option; +import static org.onap.dcae.common.publishing.VavrUtils.f; + +/** + * @author Pawel Szalapski (pawel.szalapski@nokia.com) + */ +class DMaaPPublishersCache { + + private static final Logger log = LoggerFactory.getLogger(DMaaPPublishersCache.class); + private final LoadingCache<String, CambriaBatchingPublisher> publishersCache; + private AtomicReference<Map<String, PublisherConfig>> dMaaPConfiguration; + + DMaaPPublishersCache(Map<String, PublisherConfig> dMaaPConfiguration) { + this.dMaaPConfiguration = new AtomicReference<>(dMaaPConfiguration); + this.publishersCache = CacheBuilder.newBuilder() + .removalListener(new OnPublisherRemovalListener()) + .build(new CambriaPublishersCacheLoader()); + } + + DMaaPPublishersCache(CambriaPublishersCacheLoader dMaaPPublishersCacheLoader, + OnPublisherRemovalListener onPublisherRemovalListener, + Map<String, PublisherConfig> dMaaPConfiguration) { + this.dMaaPConfiguration = new AtomicReference<>(dMaaPConfiguration); + this.publishersCache = CacheBuilder.newBuilder() + .removalListener(onPublisherRemovalListener) + .build(dMaaPPublishersCacheLoader); + } + + Option<CambriaBatchingPublisher> getPublisher(String streamID) { + try { + return Option(publishersCache.getUnchecked(streamID)); + } catch (Exception e) { + log.warn("Could not create / load Cambria Publisher for streamID", e); + return Option.none(); + } + } + + void closePublisherFor(String streamId) { + publishersCache.invalidate(streamId); + } + + synchronized void reconfigure(Map<String, PublisherConfig> newConfig) { + Map<String, PublisherConfig> currentConfig = dMaaPConfiguration.get(); + Map<String, PublisherConfig> removedConfigurations = currentConfig + .filterKeys(domain -> !newConfig.containsKey(domain)); + Map<String, PublisherConfig> changedConfigurations = newConfig + .filterKeys(e -> currentConfig.containsKey(e) && !currentConfig.get(e).equals(newConfig.get(e))); + dMaaPConfiguration.set(newConfig); + removedConfigurations.merge(changedConfigurations).forEach(e -> publishersCache.invalidate(e._1)); + } + + static class OnPublisherRemovalListener implements RemovalListener<String, CambriaBatchingPublisher> { + + @Override + public void onRemoval(@Nonnull RemovalNotification<String, CambriaBatchingPublisher> notification) { + CambriaBatchingPublisher publisher = notification.getValue(); + if (publisher != null) { // The value might get Garbage Collected at this moment, regardless of @Nonnull + try { + int timeout = 20; + TimeUnit unit = TimeUnit.SECONDS; + java.util.List<?> stuck = publisher.close(timeout, unit); + if (!stuck.isEmpty()) { + log.error(f("Publisher got stuck and did not manage to close in '%s' '%s', " + + "%s messages were dropped", stuck.size(), timeout, unit)); + } + } catch (InterruptedException | IOException e) { + log.error("Could not close Cambria publisher, some messages might have been dropped", e); + Thread.currentThread().interrupt(); + } + } + } + } + + class CambriaPublishersCacheLoader extends CacheLoader<String, CambriaBatchingPublisher> { + + @Override + public CambriaBatchingPublisher load(@Nonnull String domain) { + return dMaaPConfiguration.get() + .get(domain) + .toTry(() -> new RuntimeException( + f("DMaaP configuration contains no configuration for domain: '%s'", domain))) + .flatMap(DMaaPPublishersBuilder::buildPublisher) + .get(); + } + } + +} diff --git a/src/main/java/org/onap/dcae/common/publishing/EventPublisher.java b/src/main/java/org/onap/dcae/common/publishing/EventPublisher.java new file mode 100644 index 00000000..42e721a8 --- /dev/null +++ b/src/main/java/org/onap/dcae/common/publishing/EventPublisher.java @@ -0,0 +1,38 @@ +/*- + * ============LICENSE_START======================================================= + * org.onap.dcaegen2.collectors.ves + * ================================================================================ + * Copyright (C) 2018 Nokia. 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.common.publishing; + +import io.vavr.collection.Map; +import org.json.JSONObject; +import org.slf4j.Logger; + +/** + * @author Pawel Szalapski (pawel.szalapski@nokia.com) + */ +public interface EventPublisher { + + static EventPublisher createPublisher(Logger outputLogger, Map<String, PublisherConfig> dMaaPConfig) { + return new DMaaPEventPublisher(new DMaaPPublishersCache(dMaaPConfig), outputLogger); + } + + void sendEvent(JSONObject event, String domain); + + void reconfigure(Map<String, PublisherConfig> dMaaPConfig); +} diff --git a/src/main/java/org/onap/dcae/common/publishing/PublisherConfig.java b/src/main/java/org/onap/dcae/common/publishing/PublisherConfig.java new file mode 100644 index 00000000..1fd0d316 --- /dev/null +++ b/src/main/java/org/onap/dcae/common/publishing/PublisherConfig.java @@ -0,0 +1,99 @@ +/*- + * ============LICENSE_START======================================================= + * org.onap.dcaegen2.collectors.ves + * ================================================================================ + * Copyright (C) 2018 Nokia. 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.common.publishing; + +import io.vavr.collection.List; +import io.vavr.control.Option; + +import java.util.Objects; + +/** + * @author Pawel Szalapski (pawel.szalapski@nokia.com) + */ +public final class PublisherConfig { + + private final List<String> destinations; + private final String topic; + private String userName; + private String password; + + PublisherConfig(List<String> destinations, String topic) { + this.destinations = destinations; + this.topic = topic; + } + + PublisherConfig(List<String> destinations, String topic, String userName, String password) { + this.destinations = destinations; + this.topic = topic; + this.userName = userName; + this.password = password; + } + + List<String> destinations() { + return destinations; + } + + String topic() { + return topic; + } + + Option<String> userName() { + return Option.of(userName); + } + + Option<String> password() { + return Option.of(password); + } + + boolean isSecured() { + return userName().isDefined() && password().isDefined(); + } + + + @Override + public boolean equals(Object o) { + if (this == o) { + return true; + } + if (o == null || getClass() != o.getClass()) { + return false; + } + PublisherConfig that = (PublisherConfig) o; + return Objects.equals(destinations, that.destinations) && + Objects.equals(topic, that.topic) && + Objects.equals(userName, that.userName) && + Objects.equals(password, that.password); + } + + @Override + public int hashCode() { + return Objects.hash(destinations, topic, userName, password); + } + + @Override + public String toString() { + return "PublisherConfig{" + + "destinations=" + destinations + + ", topic='" + topic + '\'' + + ", userName='" + userName + '\'' + + ", password='" + password + '\'' + + '}'; + } +} diff --git a/src/main/java/org/onap/dcae/common/publishing/VavrUtils.java b/src/main/java/org/onap/dcae/common/publishing/VavrUtils.java new file mode 100644 index 00000000..e1eae247 --- /dev/null +++ b/src/main/java/org/onap/dcae/common/publishing/VavrUtils.java @@ -0,0 +1,62 @@ +/*- + * ============LICENSE_START======================================================= + * org.onap.dcaegen2.collectors.ves + * ================================================================================ + * Copyright (C) 2018 Nokia. 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.common.publishing; + +import io.vavr.API; +import io.vavr.API.Match.Case; +import java.util.function.Consumer; +import org.slf4j.Logger; + +import static io.vavr.API.$; + +/** + * @author Pawel Szalapski (pawel.szalapski@nokia.com) + */ +public final class VavrUtils { + + private VavrUtils() { + // utils aggregator + } + + /** + * Shortcut for 'string interpolation' + */ + public static String f(String msg, Object... args) { + return String.format(msg, args); + } + + /** + * Wrap failure with a more descriptive message of what has failed and chain original cause. Used to provide a + * context for errors instead of raw exception. + */ + public static Case<Throwable, Throwable> enhanceError(String msg) { + return API.Case($(), e -> new RuntimeException(msg, e)); + } + + public static Case<Throwable, Throwable> enhanceError(String pattern, Object... arguments) { + return API.Case($(), e -> new RuntimeException(f(pattern, arguments), e)); + } + + public static Consumer<Throwable> logError(Logger withLogger) { + return e -> withLogger.error(e.getMessage(), e); + } + + +} |