From adfb0693ec18025abcd8b9036bafe20a25f2e496 Mon Sep 17 00:00:00 2001 From: "halil.cakal" Date: Mon, 3 Jul 2023 15:24:02 +0100 Subject: Subscription Creation: NCMP to DMI CloudEvent transformation - Add mapper to convert client event into ncmp event - Add sample json object of ncmp version - Change subscription event consumer to consume CloudEvents - Change subscription event forwarder to publish CloudEvents - Change test producer config to support CloudEvents - Change sample subscription event json to comply with new schema - Add more test for missing branches - Change packages of the mappers into relevant directory Issue-ID: CPS-1737 Change-Id: I8c9e6e7bf713a8fb530a0586dfb2bce796a462f5 Signed-off-by: halil.cakal --- .../ClientSubscriptionEventMapper.java | 34 ++++++++++ .../avcsubscription/SubscriptionEventConsumer.java | 28 ++++---- .../SubscriptionEventForwarder.java | 72 +++++++++++--------- .../avcsubscription/SubscriptionEventMapper.java | 15 ++-- .../impl/utils/SubscriptionEventCloudMapper.java | 79 ++++++++++++++++++++++ 5 files changed, 174 insertions(+), 54 deletions(-) create mode 100644 cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/events/avcsubscription/ClientSubscriptionEventMapper.java create mode 100644 cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/utils/SubscriptionEventCloudMapper.java (limited to 'cps-ncmp-service/src/main') diff --git a/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/events/avcsubscription/ClientSubscriptionEventMapper.java b/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/events/avcsubscription/ClientSubscriptionEventMapper.java new file mode 100644 index 000000000..59b1d09c7 --- /dev/null +++ b/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/events/avcsubscription/ClientSubscriptionEventMapper.java @@ -0,0 +1,34 @@ +/* + * ============LICENSE_START======================================================= + * Copyright (C) 2023 Nordix Foundation + * ================================================================================ + * 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. + * + * SPDX-License-Identifier: Apache-2.0 + * ============LICENSE_END========================================================= + */ + +package org.onap.cps.ncmp.api.impl.events.avcsubscription; + +import org.mapstruct.Mapper; +import org.mapstruct.Mapping; +import org.onap.cps.ncmp.events.avcsubscription1_0_0.client_to_ncmp.SubscriptionEvent; + +@Mapper(componentModel = "spring") +public interface ClientSubscriptionEventMapper { + + @Mapping(target = "data.predicates.targets", ignore = true) + org.onap.cps.ncmp.events.avcsubscription1_0_0.ncmp_to_dmi.SubscriptionEvent toNcmpSubscriptionEvent( + SubscriptionEvent subscriptionEvent); + +} diff --git a/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/events/avcsubscription/SubscriptionEventConsumer.java b/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/events/avcsubscription/SubscriptionEventConsumer.java index 88b41d007..f511965c7 100644 --- a/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/events/avcsubscription/SubscriptionEventConsumer.java +++ b/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/events/avcsubscription/SubscriptionEventConsumer.java @@ -20,13 +20,14 @@ package org.onap.cps.ncmp.api.impl.events.avcsubscription; +import io.cloudevents.CloudEvent; import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; import org.apache.kafka.clients.consumer.ConsumerRecord; import org.onap.cps.ncmp.api.impl.subscriptions.SubscriptionPersistence; +import org.onap.cps.ncmp.api.impl.utils.SubscriptionEventCloudMapper; import org.onap.cps.ncmp.api.impl.yangmodels.YangModelSubscriptionEvent; -import org.onap.cps.ncmp.event.model.InnerSubscriptionEvent; -import org.onap.cps.ncmp.event.model.SubscriptionEvent; +import org.onap.cps.ncmp.events.avcsubscription1_0_0.client_to_ncmp.SubscriptionEvent; import org.onap.cps.spi.exceptions.OperationNotYetSupportedException; import org.springframework.beans.factory.annotation.Value; import org.springframework.kafka.annotation.KafkaListener; @@ -54,28 +55,25 @@ public class SubscriptionEventConsumer { * @param subscriptionEventConsumerRecord the event to be consumed */ @KafkaListener(topics = "${app.ncmp.avc.subscription-topic}", - properties = {"spring.json.value.default.type=org.onap.cps.ncmp.event.model.SubscriptionEvent"}) - public void consumeSubscriptionEvent( - final ConsumerRecord subscriptionEventConsumerRecord) { - final SubscriptionEvent subscriptionEvent = subscriptionEventConsumerRecord.value(); - final InnerSubscriptionEvent event = subscriptionEvent.getEvent(); - final String eventDatastore = event.getPredicates().getDatastore(); + containerFactory = "cloudEventConcurrentKafkaListenerContainerFactory") + public void consumeSubscriptionEvent(final ConsumerRecord subscriptionEventConsumerRecord) { + final CloudEvent cloudEvent = subscriptionEventConsumerRecord.value(); + final SubscriptionEvent subscriptionEvent = SubscriptionEventCloudMapper.toSubscriptionEvent(cloudEvent); + final String eventDatastore = subscriptionEvent.getData().getPredicates().getDatastore(); if (!(eventDatastore.equals("passthrough-running") || eventDatastore.equals("passthrough-operational"))) { throw new OperationNotYetSupportedException( "passthrough datastores are currently only supported for event subscriptions"); } - if ("CM".equals(event.getDataType().getDataCategory())) { - log.debug("Consuming event {} ...", subscriptionEvent); + if ("CM".equals(subscriptionEvent.getData().getDataType().getDataCategory())) { if (subscriptionModelLoaderEnabled) { persistSubscriptionEvent(subscriptionEvent); } - if ("CREATE".equals(subscriptionEvent.getEventType().value())) { + if ("CREATE".equals(cloudEvent.getType())) { log.info("Subscription for ClientID {} with name {} ...", - event.getSubscription().getClientID(), - event.getSubscription().getName()); + subscriptionEvent.getData().getSubscription().getClientID(), + subscriptionEvent.getData().getSubscription().getName()); if (notificationFeatureEnabled) { - subscriptionEventForwarder.forwardCreateSubscriptionEvent(subscriptionEvent, - subscriptionEventConsumerRecord.headers()); + subscriptionEventForwarder.forwardCreateSubscriptionEvent(subscriptionEvent); } } } else { diff --git a/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/events/avcsubscription/SubscriptionEventForwarder.java b/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/events/avcsubscription/SubscriptionEventForwarder.java index 1d87a057a..1fe963a27 100644 --- a/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/events/avcsubscription/SubscriptionEventForwarder.java +++ b/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/events/avcsubscription/SubscriptionEventForwarder.java @@ -21,13 +21,12 @@ package org.onap.cps.ncmp.api.impl.events.avcsubscription; import com.hazelcast.map.IMap; +import io.cloudevents.CloudEvent; import java.util.ArrayList; import java.util.Collection; -import java.util.Collections; import java.util.HashSet; import java.util.List; import java.util.Map; -import java.util.Objects; import java.util.Set; import java.util.concurrent.Executors; import java.util.concurrent.ScheduledExecutorService; @@ -35,16 +34,17 @@ import java.util.concurrent.TimeUnit; import java.util.stream.Collectors; import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; -import org.apache.kafka.common.header.Headers; import org.onap.cps.ncmp.api.impl.config.embeddedcache.ForwardedSubscriptionEventCacheConfig; import org.onap.cps.ncmp.api.impl.events.EventsPublisher; import org.onap.cps.ncmp.api.impl.subscriptions.SubscriptionPersistence; import org.onap.cps.ncmp.api.impl.subscriptions.SubscriptionStatus; import org.onap.cps.ncmp.api.impl.utils.DmiServiceNameOrganizer; +import org.onap.cps.ncmp.api.impl.utils.SubscriptionEventCloudMapper; import org.onap.cps.ncmp.api.impl.yangmodels.YangModelCmHandle; import org.onap.cps.ncmp.api.impl.yangmodels.YangModelSubscriptionEvent; import org.onap.cps.ncmp.api.inventory.InventoryPersistence; -import org.onap.cps.ncmp.event.model.SubscriptionEvent; +import org.onap.cps.ncmp.events.avcsubscription1_0_0.client_to_ncmp.SubscriptionEvent; +import org.onap.cps.ncmp.events.avcsubscription1_0_0.ncmp_to_dmi.CmHandle; import org.onap.cps.spi.exceptions.OperationNotYetSupportedException; import org.springframework.beans.factory.annotation.Value; import org.springframework.stereotype.Component; @@ -56,10 +56,11 @@ import org.springframework.stereotype.Component; public class SubscriptionEventForwarder { private final InventoryPersistence inventoryPersistence; - private final EventsPublisher eventsPublisher; + private final EventsPublisher eventsPublisher; private final IMap> forwardedSubscriptionEventCache; private final SubscriptionEventResponseOutcome subscriptionEventResponseOutcome; private final SubscriptionEventMapper subscriptionEventMapper; + private final ClientSubscriptionEventMapper clientSubscriptionEventMapper; private final SubscriptionPersistence subscriptionPersistence; private final ScheduledExecutorService executorService = Executors.newSingleThreadScheduledExecutor(); @Value("${app.ncmp.avc.subscription-forward-topic-prefix}") @@ -73,28 +74,22 @@ public class SubscriptionEventForwarder { * * @param subscriptionEvent the event to be forwarded */ - public void forwardCreateSubscriptionEvent(final SubscriptionEvent subscriptionEvent, - final Headers eventHeaders) { - final List cmHandleTargets = subscriptionEvent.getEvent().getPredicates().getTargets(); + public void forwardCreateSubscriptionEvent(final SubscriptionEvent subscriptionEvent) { + final List cmHandleTargets = subscriptionEvent.getData().getPredicates().getTargets(); if (cmHandleTargets == null || cmHandleTargets.isEmpty() - || cmHandleTargets.stream().anyMatch(id -> ((String) id).contains("*"))) { + || cmHandleTargets.stream().anyMatch(id -> (id).contains("*"))) { throw new OperationNotYetSupportedException( "CMHandle targets are required. \"Wildcard\" operations are not yet supported"); } - final List cmHandleTargetsAsStrings = cmHandleTargets.stream().map( - Objects::toString).collect(Collectors.toList()); final Collection yangModelCmHandles = - inventoryPersistence.getYangModelCmHandles(cmHandleTargetsAsStrings); - + inventoryPersistence.getYangModelCmHandles(cmHandleTargets); final Map>> dmiPropertiesPerCmHandleIdPerServiceName = DmiServiceNameOrganizer.getDmiPropertiesPerCmHandleIdPerServiceName(yangModelCmHandles); - - findDmisAndRespond(subscriptionEvent, eventHeaders, cmHandleTargetsAsStrings, - dmiPropertiesPerCmHandleIdPerServiceName); + findDmisAndRespond(subscriptionEvent, cmHandleTargets, dmiPropertiesPerCmHandleIdPerServiceName); } - private void findDmisAndRespond(final SubscriptionEvent subscriptionEvent, final Headers eventHeaders, - final List cmHandleTargetsAsStrings, + private void findDmisAndRespond(final SubscriptionEvent subscriptionEvent, + final List cmHandleTargetsAsStrings, final Map>> dmiPropertiesPerCmHandleIdPerServiceName) { final List cmHandlesThatExistsInDb = dmiPropertiesPerCmHandleIdPerServiceName.entrySet().stream() @@ -109,18 +104,20 @@ public class SubscriptionEventForwarder { updatesCmHandlesToRejectedAndPersistSubscriptionEvent(subscriptionEvent, targetCmHandlesDoesNotExistInDb); } if (dmisToRespond.isEmpty()) { - final String clientID = subscriptionEvent.getEvent().getSubscription().getClientID(); - final String subscriptionName = subscriptionEvent.getEvent().getSubscription().getName(); + final String clientID = subscriptionEvent.getData().getSubscription().getClientID(); + final String subscriptionName = subscriptionEvent.getData().getSubscription().getName(); subscriptionEventResponseOutcome.sendResponse(clientID, subscriptionName); } else { startResponseTimeout(subscriptionEvent, dmisToRespond); - forwardEventToDmis(dmiPropertiesPerCmHandleIdPerServiceName, subscriptionEvent, eventHeaders); + final org.onap.cps.ncmp.events.avcsubscription1_0_0.ncmp_to_dmi.SubscriptionEvent ncmpSubscriptionEvent = + clientSubscriptionEventMapper.toNcmpSubscriptionEvent(subscriptionEvent); + forwardEventToDmis(dmiPropertiesPerCmHandleIdPerServiceName, ncmpSubscriptionEvent); } } private void startResponseTimeout(final SubscriptionEvent subscriptionEvent, final Set dmisToRespond) { - final String subscriptionClientId = subscriptionEvent.getEvent().getSubscription().getClientID(); - final String subscriptionName = subscriptionEvent.getEvent().getSubscription().getName(); + final String subscriptionClientId = subscriptionEvent.getData().getSubscription().getClientID(); + final String subscriptionName = subscriptionEvent.getData().getSubscription().getName(); final String subscriptionEventId = subscriptionClientId + subscriptionName; forwardedSubscriptionEventCache.put(subscriptionEventId, dmisToRespond, @@ -137,20 +134,33 @@ public class SubscriptionEventForwarder { } private void forwardEventToDmis(final Map>> dmiNameCmHandleMap, - final SubscriptionEvent subscriptionEvent, - final Headers eventHeaders) { + final org.onap.cps.ncmp.events.avcsubscription1_0_0.ncmp_to_dmi.SubscriptionEvent + ncmpSubscriptionEvent) { dmiNameCmHandleMap.forEach((dmiName, cmHandlePropertiesMap) -> { - subscriptionEvent.getEvent().getPredicates().setTargets(Collections.singletonList(cmHandlePropertiesMap)); - final String eventKey = createEventKey(subscriptionEvent, dmiName); + final List cmHandleTargets = cmHandlePropertiesMap.entrySet().stream().map( + cmHandleAndProperties -> { + final CmHandle cmHandle = new CmHandle(); + cmHandle.setId(cmHandleAndProperties.getKey()); + cmHandle.setAdditionalProperties(cmHandleAndProperties.getValue()); + return cmHandle; + }).collect(Collectors.toList()); + + ncmpSubscriptionEvent.getData().getPredicates().setTargets(cmHandleTargets); + final String eventKey = createEventKey(ncmpSubscriptionEvent, dmiName); final String dmiAvcSubscriptionTopic = dmiAvcSubscriptionTopicPrefix + dmiName; - eventsPublisher.publishEvent(dmiAvcSubscriptionTopic, eventKey, eventHeaders, subscriptionEvent); + + final CloudEvent ncmpSubscriptionCloudEvent = + SubscriptionEventCloudMapper.toCloudEvent(ncmpSubscriptionEvent, eventKey); + eventsPublisher.publishCloudEvent(dmiAvcSubscriptionTopic, eventKey, ncmpSubscriptionCloudEvent); }); } - private String createEventKey(final SubscriptionEvent subscriptionEvent, final String dmiName) { - return subscriptionEvent.getEvent().getSubscription().getClientID() + private String createEventKey( + final org.onap.cps.ncmp.events.avcsubscription1_0_0.ncmp_to_dmi.SubscriptionEvent subscriptionEvent, + final String dmiName) { + return subscriptionEvent.getData().getSubscription().getClientID() + "-" - + subscriptionEvent.getEvent().getSubscription().getName() + + subscriptionEvent.getData().getSubscription().getName() + "-" + dmiName; } diff --git a/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/events/avcsubscription/SubscriptionEventMapper.java b/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/events/avcsubscription/SubscriptionEventMapper.java index dcbdcf339..bf9ceb1c3 100644 --- a/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/events/avcsubscription/SubscriptionEventMapper.java +++ b/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/events/avcsubscription/SubscriptionEventMapper.java @@ -27,17 +27,16 @@ import org.mapstruct.Mapping; import org.mapstruct.Named; import org.onap.cps.ncmp.api.impl.subscriptions.SubscriptionStatus; import org.onap.cps.ncmp.api.impl.yangmodels.YangModelSubscriptionEvent; -import org.onap.cps.ncmp.event.model.SubscriptionEvent; +import org.onap.cps.ncmp.events.avcsubscription1_0_0.client_to_ncmp.SubscriptionEvent; @Mapper(componentModel = "spring") public interface SubscriptionEventMapper { - @Mapping(source = "event.subscription.clientID", target = "clientId") - @Mapping(source = "event.subscription.name", target = "subscriptionName") - @Mapping(source = "event.subscription.isTagged", target = "tagged") - @Mapping(source = "event.predicates.targets", target = "predicates.targetCmHandles", + @Mapping(source = "data.subscription.clientID", target = "clientId") + @Mapping(source = "data.subscription.name", target = "subscriptionName") + @Mapping(source = "data.predicates.targets", target = "predicates.targetCmHandles", qualifiedByName = "mapTargetsToCmHandleTargets") - @Mapping(source = "event.predicates.datastore", target = "predicates.datastore") + @Mapping(source = "data.predicates.datastore", target = "predicates.datastore") YangModelSubscriptionEvent toYangModelSubscriptionEvent(SubscriptionEvent subscriptionEvent); /** @@ -47,8 +46,8 @@ public interface SubscriptionEventMapper { * @return TargetCmHandle list */ @Named("mapTargetsToCmHandleTargets") - default List mapTargetsToCmHandleTargets(List targets) { - return targets.stream().map(target -> new YangModelSubscriptionEvent.TargetCmHandle(target.toString(), + default List mapTargetsToCmHandleTargets(List targets) { + return targets.stream().map(target -> new YangModelSubscriptionEvent.TargetCmHandle(target, SubscriptionStatus.PENDING)) .collect(Collectors.toList()); } diff --git a/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/utils/SubscriptionEventCloudMapper.java b/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/utils/SubscriptionEventCloudMapper.java new file mode 100644 index 000000000..a7de47904 --- /dev/null +++ b/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/utils/SubscriptionEventCloudMapper.java @@ -0,0 +1,79 @@ +/* + * ============LICENSE_START======================================================= + * Copyright (C) 2023 Nordix Foundation + * ================================================================================ + * 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. + * + * SPDX-License-Identifier: Apache-2.0 + * ============LICENSE_END========================================================= + */ + +package org.onap.cps.ncmp.api.impl.utils; + +import com.fasterxml.jackson.databind.ObjectMapper; +import io.cloudevents.CloudEvent; +import io.cloudevents.core.CloudEventUtils; +import io.cloudevents.core.builder.CloudEventBuilder; +import io.cloudevents.core.data.PojoCloudEventData; +import io.cloudevents.jackson.PojoCloudEventDataMapper; +import java.net.URI; +import lombok.AccessLevel; +import lombok.NoArgsConstructor; +import lombok.extern.slf4j.Slf4j; +import org.onap.cps.ncmp.events.avcsubscription1_0_0.client_to_ncmp.SubscriptionEvent; + +@NoArgsConstructor(access = AccessLevel.PRIVATE) +@Slf4j +public class SubscriptionEventCloudMapper { + + private static final ObjectMapper objectMapper = new ObjectMapper(); + + /** + * Maps CloudEvent object to SubscriptionEvent. + * + * @param cloudEvent object. + * @return SubscriptionEvent deserialized. + */ + public static SubscriptionEvent toSubscriptionEvent(final CloudEvent cloudEvent) { + final PojoCloudEventData deserializedCloudEvent = CloudEventUtils + .mapData(cloudEvent, PojoCloudEventDataMapper.from(objectMapper, SubscriptionEvent.class)); + if (deserializedCloudEvent == null) { + log.debug("No data found in the consumed event"); + return null; + } else { + final SubscriptionEvent subscriptionEvent = deserializedCloudEvent.getValue(); + log.debug("Consuming event {}", subscriptionEvent); + return subscriptionEvent; + } + } + + /** + * Maps SubscriptionEvent to a CloudEvent. + * + * @param ncmpSubscriptionEvent object. + * @param eventKey as String. + * @return CloudEvent builded. + */ + public static CloudEvent toCloudEvent( + final org.onap.cps.ncmp.events.avcsubscription1_0_0.ncmp_to_dmi.SubscriptionEvent ncmpSubscriptionEvent, + final String eventKey) { + try { + return CloudEventBuilder.v1() + .withData(objectMapper.writeValueAsBytes(ncmpSubscriptionEvent)) + .withId(eventKey).withType("CREATE").withSource( + URI.create(ncmpSubscriptionEvent.getData().getSubscription().getClientID())).build(); + } catch (final Exception ex) { + throw new RuntimeException("The Cloud Event could not be constructed.", ex); + } + } +} -- cgit 1.2.3-korg