diff options
Diffstat (limited to 'prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/tasks/commit')
5 files changed, 561 insertions, 0 deletions
diff --git a/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/tasks/commit/EpochDateTimeConversion.java b/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/tasks/commit/EpochDateTimeConversion.java new file mode 100644 index 00000000..4bf49208 --- /dev/null +++ b/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/tasks/commit/EpochDateTimeConversion.java @@ -0,0 +1,95 @@ +/* + * ============LICENSE_START======================================================= + * PNF-REGISTRATION-HANDLER + * ================================================================================ + * Copyright (C) 2023 Deutsche Telekom 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.dcaegen2.services.prh.tasks.commit; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.stereotype.Component; + +import java.time.LocalDateTime; +import java.time.LocalTime; +import java.time.ZoneId; +import java.time.ZonedDateTime; +import java.time.format.DateTimeFormatter; +import java.util.Date; + +/** + * This class will return start date time of the day and end date time of the day in epoch format. + * @author <a href="mailto:mohd.khan@t-systems.com">Mohd Usman Khan</a> on 3/13/23 + */ + +@Component +public class EpochDateTimeConversion { + + private static final Logger LOGGER = LoggerFactory.getLogger(EpochDateTimeConversion.class); + + private String daysForRecords = System.getenv("number_of_days"); + + public Long getStartDateOfTheDay(){ + return getEpochDateTime(atStartOfDay(getCurrentDate())); + } + + public Long getEndDateOfTheDay(){ + return getEpochDateTime(atEndOfDay(getCurrentDate())); + } + + private Long getEpochDateTime(Date date) + { + DateTimeFormatter dtf = DateTimeFormatter.ofPattern("E MMM dd HH:mm:ss zzz yyyy"); + ZonedDateTime zdt = ZonedDateTime.parse( date.toString(),dtf); + return zdt.toInstant().toEpochMilli(); + } + + private Date getCurrentDate() + { + return new java.util.Date(System.currentTimeMillis()); + } + + public Date atStartOfDay(Date date) { + LocalDateTime localDateTime = dateToLocalDateTime(date); + if(daysForRecords==null) + daysForRecords="1"; + LocalDateTime previousDay = localDateTime.minusDays(Integer.parseInt(daysForRecords) - 1l); + LocalDateTime previousStartTime = previousDay.with(LocalTime.MIN); + return localDateTimeToDate(previousStartTime); + } + + private Date atEndOfDay(Date date) { + LocalDateTime localDateTime = dateToLocalDateTime(date); + LocalDateTime endOfDay = localDateTime.with(LocalTime.MAX); + return localDateTimeToDate(endOfDay); + } + + private LocalDateTime dateToLocalDateTime(Date date) { + return LocalDateTime.ofInstant(date.toInstant(), ZoneId.systemDefault()); + } + + private Date localDateTimeToDate(LocalDateTime localDateTime) { + return Date.from(localDateTime.atZone(ZoneId.systemDefault()).toInstant()); + } + + public String getDaysForRecords() { + return daysForRecords; + } + + public void setDaysForRecords(String daysForRecords) { + this.daysForRecords = daysForRecords; + } +} diff --git a/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/tasks/commit/KafkaConsumerTask.java b/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/tasks/commit/KafkaConsumerTask.java new file mode 100644 index 00000000..4c70c713 --- /dev/null +++ b/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/tasks/commit/KafkaConsumerTask.java @@ -0,0 +1,35 @@ +/* + * ============LICENSE_START======================================================= + * PNF-REGISTRATION-HANDLER + * ================================================================================ + * Copyright (C) 2023 Deutsche Telekom 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.dcaegen2.services.prh.tasks.commit; + +import org.onap.dcaegen2.services.prh.adapter.aai.api.ConsumerDmaapModel; +import org.springframework.boot.configurationprocessor.json.JSONException; +import reactor.core.publisher.Flux; + +/** + * @author <a href="mailto:ajinkya-patil@t-systems.com">Ajinkya Patil</a> on 3/13/23 + */ + +public interface KafkaConsumerTask { + Flux<ConsumerDmaapModel> execute() throws JSONException; + + void commitOffset(); +} diff --git a/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/tasks/commit/KafkaConsumerTaskImpl.java b/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/tasks/commit/KafkaConsumerTaskImpl.java new file mode 100644 index 00000000..6b289f1c --- /dev/null +++ b/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/tasks/commit/KafkaConsumerTaskImpl.java @@ -0,0 +1,130 @@ +/* + * ============LICENSE_START======================================================= + * PNF-REGISTRATION-HANDLER + * ================================================================================ + * Copyright (C) 2023 Deutsche Telekom 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.dcaegen2.services.prh.tasks.commit; + +import org.apache.kafka.clients.consumer.ConsumerRecord; +import org.onap.dcaegen2.services.prh.adapter.aai.api.ConsumerDmaapModel; +import org.onap.dcaegen2.services.prh.configuration.CbsConfigurationForAutoCommitDisabledMode; +import org.onap.dcaegen2.services.prh.service.DmaapConsumerJsonParser; +import org.springframework.boot.configurationprocessor.json.JSONException; +import org.springframework.context.annotation.Profile; +import org.springframework.kafka.annotation.KafkaListener; +import org.springframework.kafka.listener.BatchAcknowledgingMessageListener; +import org.springframework.kafka.support.Acknowledgment; +import org.springframework.stereotype.Component; +import reactor.core.publisher.Flux; +import java.util.ArrayList; +import java.util.List; + +/** + * @author <a href="mailto:ajinkya-patil@t-systems.com">Ajinkya Patil</a> on + * 3/13/23 + */ + +@Profile("autoCommitDisabled") +@Component +public class KafkaConsumerTaskImpl implements KafkaConsumerTask, BatchAcknowledgingMessageListener<String, String> { + + + private DmaapConsumerJsonParser dmaapConsumerJsonParser; + + private EpochDateTimeConversion epochDateTimeConversion; + + private CbsConfigurationForAutoCommitDisabledMode cbsConfigurationForAutoCommitDisabledMode; + + private List<String> jsonEvent = new ArrayList<>(); + + public List<String> getJsonEvent() { + return jsonEvent; + } + + private Acknowledgment offset; + + public Acknowledgment getOffset() { + return offset; + } + + static String commonInURL = "/events/"; + + String kafkaTopic; + + String groupIdConfig; + + + public KafkaConsumerTaskImpl(CbsConfigurationForAutoCommitDisabledMode cbsConfigurationForAutoCommitDisabledMode + ,DmaapConsumerJsonParser dmaapConsumerJsonParser,EpochDateTimeConversion epochDateTimeConversion) { + this.cbsConfigurationForAutoCommitDisabledMode = cbsConfigurationForAutoCommitDisabledMode; + this.dmaapConsumerJsonParser = dmaapConsumerJsonParser; + this.epochDateTimeConversion = epochDateTimeConversion; + String kafkaTopicURL = this.cbsConfigurationForAutoCommitDisabledMode.getMessageRouterSubscribeRequest() + .sourceDefinition().topicUrl(); + kafkaTopic = getTopicFromTopicUrl(kafkaTopicURL); + groupIdConfig = cbsConfigurationForAutoCommitDisabledMode.getMessageRouterSubscribeRequest().consumerGroup(); + + System.setProperty("kafkaTopic", kafkaTopic); + System.setProperty("groupIdConfig", groupIdConfig); + + } + + @Override + @KafkaListener(topics = "${kafkaTopic}", groupId = "${groupIdConfig}") + public void onMessage(List<ConsumerRecord<String, String>> list, Acknowledgment acknowledgment) { + + if (list != null && !list.isEmpty()) { + list.stream().filter( + consumerRecord -> consumerRecord.timestamp() >= epochDateTimeConversion.getStartDateOfTheDay() + && consumerRecord.timestamp() <= epochDateTimeConversion.getEndDateOfTheDay()) + .map(ConsumerRecord::value).forEach(value -> { + jsonEvent.add(value); + }); + + } + + offset = acknowledgment; + } + + @Override + public Flux<ConsumerDmaapModel> execute() throws JSONException { + return dmaapConsumerJsonParser.getConsumerDmaapModelFromKafkaConsumerRecord(jsonEvent); + } + + public void setJsonEvent(List<String> jsonEvent) { + this.jsonEvent = jsonEvent; + } + + @Override + public void commitOffset() { + if (!jsonEvent.isEmpty()) { + jsonEvent.clear(); + } + if (offset != null) { + offset.acknowledge(); + } + } + + public String getTopicFromTopicUrl(String topicUrl) { + if (topicUrl.endsWith("/")) { + return topicUrl.substring(topicUrl.indexOf(commonInURL) + commonInURL.length(), topicUrl.lastIndexOf("/")); + } + return topicUrl.substring(topicUrl.indexOf(commonInURL) + commonInURL.length()); + } + +} diff --git a/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/tasks/commit/ScheduledTasksRunnerWithCommit.java b/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/tasks/commit/ScheduledTasksRunnerWithCommit.java new file mode 100644 index 00000000..91cdd122 --- /dev/null +++ b/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/tasks/commit/ScheduledTasksRunnerWithCommit.java @@ -0,0 +1,98 @@ +/* + * ============LICENSE_START======================================================= + * PNF-REGISTRATION-HANDLER + * ================================================================================ + * Copyright (C) 2023 Deutsche Telekom 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.dcaegen2.services.prh.tasks.commit; + +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; +import java.util.concurrent.ScheduledFuture; +import javax.annotation.PreDestroy; +import org.onap.dcaegen2.services.prh.configuration.PrhProperties; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.slf4j.Marker; +import org.slf4j.MarkerFactory; +import org.springframework.boot.context.event.ApplicationStartedEvent; +import org.springframework.context.annotation.Configuration; +import org.springframework.context.annotation.Profile; +import org.springframework.context.event.EventListener; +import org.springframework.scheduling.TaskScheduler; +import org.springframework.scheduling.annotation.EnableScheduling; + +/** + * @author <a href="mailto:pravin.kokane@t-systems.com">Pravin Kokane</a> on 3/13/23 + */ + +@Profile("autoCommitDisabled") +@Configuration +@EnableScheduling +public class ScheduledTasksRunnerWithCommit { + private static final Logger LOGGER = LoggerFactory.getLogger(ScheduledTasksRunnerWithCommit.class); + private static final Marker ENTRY = MarkerFactory.getMarker("ENTRY"); + private static List<ScheduledFuture> scheduledPrhTaskFutureList = new ArrayList<>(); + + private final TaskScheduler taskScheduler; + private final PrhProperties prhProperties; + + private ScheduledTasksWithCommit scheduledTasksWithCommit; + + public ScheduledTasksRunnerWithCommit(TaskScheduler taskScheduler, ScheduledTasksWithCommit scheduledTasksWithCommit, + PrhProperties prhProperties) { + this.taskScheduler = taskScheduler; + this.scheduledTasksWithCommit = scheduledTasksWithCommit; + this.prhProperties = prhProperties; + } + + @EventListener + public void onApplicationStartedEvent(ApplicationStartedEvent applicationStartedEvent) { + LOGGER.info(ENTRY,"### in onApplicationStartedEvent"); + LOGGER.info(ENTRY,"###tryToStartTaskWithCommit="+tryToStartTaskWithCommit()); + } + + /** + * Function which have to stop tasks execution. + */ + @PreDestroy + public synchronized void cancelTasks() { + LOGGER.info(ENTRY,"###In cancelTasks"); + scheduledPrhTaskFutureList.forEach(x -> x.cancel(false)); + scheduledPrhTaskFutureList.clear(); + } + + /** + * Function for starting scheduling PRH workflow. + * + * @return status of operation execution: true - started, false - not started + */ + + public synchronized boolean tryToStartTaskWithCommit() { + LOGGER.info(ENTRY, "Start scheduling PRH workflow with Commit Tasks Runner"); + if (scheduledPrhTaskFutureList.isEmpty()) { + Collections.synchronizedList(scheduledPrhTaskFutureList); + scheduledPrhTaskFutureList.add(taskScheduler + .scheduleWithFixedDelay(scheduledTasksWithCommit::scheduleKafkaPrhEventTask, + prhProperties.getWorkflowSchedulingInterval())); + return true; + } else { + return false; + } + } + +} diff --git a/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/tasks/commit/ScheduledTasksWithCommit.java b/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/tasks/commit/ScheduledTasksWithCommit.java new file mode 100644 index 00000000..352c0bbc --- /dev/null +++ b/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/tasks/commit/ScheduledTasksWithCommit.java @@ -0,0 +1,203 @@ +/* + * ============LICENSE_START======================================================= + * PNF-REGISTRATION-HANDLER + * ================================================================================ + * Copyright (C) 2023 Deutsche Telekom 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.dcaegen2.services.prh.tasks.commit; + +import static org.onap.dcaegen2.services.sdk.rest.services.model.logging.MdcVariables.RESPONSE_CODE; + +import java.util.Map; +import java.util.concurrent.CountDownLatch; +import org.onap.dcaegen2.services.prh.exceptions.DmaapEmptyResponseException; +import org.onap.dcaegen2.services.prh.exceptions.PrhTaskException; +import org.onap.dcaegen2.services.prh.tasks.AaiProducerTask; +import org.onap.dcaegen2.services.prh.tasks.AaiQueryTask; +import org.onap.dcaegen2.services.prh.tasks.BbsActionsTask; +import org.onap.dcaegen2.services.prh.tasks.DmaapPublisherTask; +import org.onap.dcaegen2.services.prh.adapter.aai.api.ConsumerDmaapModel; +import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.model.MessageRouterPublishResponse; +import org.onap.dcaegen2.services.sdk.rest.services.model.logging.MdcVariables; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.slf4j.MDC; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.beans.factory.annotation.Qualifier; +import org.springframework.boot.configurationprocessor.json.JSONException; +import org.springframework.context.annotation.Profile; +import org.springframework.http.HttpStatus; +import org.springframework.stereotype.Component; +import reactor.core.publisher.Flux; +import reactor.core.publisher.Mono; + +/** + * @author <a href="mailto:sangeeta.bellara@t-systems.com">Sangeeta Bellara</a> + * on 3/13/23 + */ +@Profile("autoCommitDisabled") +@Component +public class ScheduledTasksWithCommit { + + private static final Logger LOGGER = LoggerFactory.getLogger(ScheduledTasksWithCommit.class); + private static Boolean pnfFound = true; + private KafkaConsumerTask kafkaConsumerTask; + private DmaapPublisherTask dmaapReadyProducerTask; + private DmaapPublisherTask dmaapUpdateProducerTask; + public AaiQueryTask aaiQueryTask; + private AaiProducerTask aaiProducerTask; + private BbsActionsTask bbsActionsTask; + private Map<String, String> mdcContextMap; + + /** + * Constructor for tasks registration in PRHWorkflow. + * + * @param kafkaConsumerTask - fist task + * @param dmaapReadyPublisherTask - third task + * @param dmaapUpdatePublisherTask - fourth task + * @param aaiPublisherTask - second task + */ + @Autowired + public ScheduledTasksWithCommit(final KafkaConsumerTask kafkaConsumerTask, + @Qualifier("ReadyPublisherTask") final DmaapPublisherTask dmaapReadyPublisherTask, + @Qualifier("UpdatePublisherTask") final DmaapPublisherTask dmaapUpdatePublisherTask, + final AaiQueryTask aaiQueryTask, final AaiProducerTask aaiPublisherTask, + final BbsActionsTask bbsActionsTask, final Map<String, String> mdcContextMap) + + { + this.dmaapReadyProducerTask = dmaapReadyPublisherTask; + this.dmaapUpdateProducerTask = dmaapUpdatePublisherTask; + this.kafkaConsumerTask = kafkaConsumerTask; + this.aaiQueryTask = aaiQueryTask; + this.aaiProducerTask = aaiPublisherTask; + this.bbsActionsTask = bbsActionsTask; + this.mdcContextMap = mdcContextMap; + } + + static class State { + public ConsumerDmaapModel dmaapModel; + public Boolean activationStatus; + + public State(ConsumerDmaapModel dmaapModel, final Boolean activationStatus) { + this.dmaapModel = dmaapModel; + this.activationStatus = activationStatus; + } + } + + public void scheduleKafkaPrhEventTask() { + MdcVariables.setMdcContextMap(mdcContextMap); + try { + + LOGGER.info("Execution of tasks was registered with commit"); + CountDownLatch mainCountDownLatch = new CountDownLatch(1); + consumeFromKafkaMessage() + .flatMap(model -> queryAaiForPnf(model).doOnError(e -> { + LOGGER.info("PNF Not Found in AAI --> {}" + e); + LOGGER.info("PNF Not Found in AAI With description of exception --> {}" + e.getMessage()); + disableCommit(); + }).onErrorResume(e -> Mono.empty()) + + ) + .flatMap(this::queryAaiForConfiguration) + .flatMap(this::publishToAaiConfiguration) + .flatMap(this::processAdditionalFields).flatMap(this::publishToDmaapConfiguration) + + .onErrorResume(e -> Mono.empty()) + + .doOnTerminate(mainCountDownLatch::countDown) + .subscribe(this::onSuccess, this::onError, this::onCompleteKafka); + mainCountDownLatch.await(); + } catch (InterruptedException | JSONException e) { + LOGGER.warn("Interruption problem on countDownLatch {}", e); + Thread.currentThread().interrupt(); + } + } + + private static void disableCommit() { + pnfFound = false; + } + + private void onCompleteKafka() { + LOGGER.info("PRH tasks have been completed"); + if (pnfFound) { + kafkaConsumerTask.commitOffset(); + LOGGER.info("Committed the Offset"); + } else { + LOGGER.info("Offset not Committed"); + pnfFound = true; + } + } + + private void onSuccess(MessageRouterPublishResponse response) { + if (response.successful()) { + String statusCodeOk = HttpStatus.OK.name(); + MDC.put(RESPONSE_CODE, statusCodeOk); + LOGGER.info("Prh consumed tasks successfully. HTTP Response code from DMaaPProducer {}", statusCodeOk); + MDC.remove(RESPONSE_CODE); + } + } + + private void onError(Throwable throwable) { + if (!(throwable instanceof DmaapEmptyResponseException)) { + LOGGER.warn("Chain of tasks have been aborted due to errors in PRH workflow {}", throwable); + } + } + + private Flux<ConsumerDmaapModel> consumeFromKafkaMessage() throws JSONException { + return kafkaConsumerTask.execute(); + } + + private Mono<State> queryAaiForConfiguration(final ConsumerDmaapModel monoDMaaPModel) { + return aaiQueryTask.execute(monoDMaaPModel).map(x -> new State(monoDMaaPModel, x)); + } + + private Mono<ConsumerDmaapModel> queryAaiForPnf(final ConsumerDmaapModel monoDMaaPModel) { + + LOGGER.info("Find PNF --> " + monoDMaaPModel.getCorrelationId()); + return aaiQueryTask.findPnfinAAI(monoDMaaPModel); + } + + private Mono<State> publishToAaiConfiguration(final State state) { + try { + return aaiProducerTask.execute(state.dmaapModel).map(x -> state); + } catch (PrhTaskException e) { + LOGGER.warn("AAIProducerTask exception has been registered: {}", e); + return Mono.error(e); + } + } + + private Mono<State> processAdditionalFields(final State state) { + if (state.activationStatus) { + LOGGER.debug("Re-registration - Logical links won't be updated."); + return Mono.just(state); + } + return bbsActionsTask.execute(state.dmaapModel).map(x -> state); + } + + private Flux<MessageRouterPublishResponse> publishToDmaapConfiguration(final State state) { + try { + if (state.activationStatus) { + LOGGER.debug("Re-registration - Using PNF_UPDATE DMaaP topic."); + return dmaapUpdateProducerTask.execute(state.dmaapModel); + } + return dmaapReadyProducerTask.execute(state.dmaapModel); + } catch (PrhTaskException e) { + LOGGER.warn("DMaaPProducerTask exception has been registered: ", e); + return Flux.error(e); + } + } +} |