diff options
author | 2021-07-09 09:41:08 +0200 | |
---|---|---|
committer | 2021-07-28 13:47:46 +0200 | |
commit | 0b06291c6c0c6f8f1e8240c8c6b1175648a177aa (patch) | |
tree | 6a0a50a7836727dbe96259282550679876186219 /src/main/java/org | |
parent | de818385611ad0afb133996a4afe5b1b00ab1c57 (diff) |
O1 PM Bulk support - collecting events, basic structure for PM Bulk File creation
Issue-ID: INT-1945
Signed-off-by: Stanislav Marszalek <s.marszalek2@partner.samsung.com>
Change-Id: If08908035719798d8d7b129ddcdb6ef62f1787fe
Diffstat (limited to 'src/main/java/org')
36 files changed, 1086 insertions, 189 deletions
diff --git a/src/main/java/org/onap/a1pesimulator/controller/RanCellController.java b/src/main/java/org/onap/a1pesimulator/controller/RanCellController.java index d454116..ae20f1c 100644 --- a/src/main/java/org/onap/a1pesimulator/controller/RanCellController.java +++ b/src/main/java/org/onap/a1pesimulator/controller/RanCellController.java @@ -14,10 +14,12 @@ package org.onap.a1pesimulator.controller; import java.util.Optional; + +import org.onap.a1pesimulator.data.ReportingMethodEnum; import org.onap.a1pesimulator.data.cell.CellDetails; import org.onap.a1pesimulator.data.cell.RanCell; -import org.onap.a1pesimulator.data.ves.Event; -import org.onap.a1pesimulator.data.ves.RanPeriodicVesEvent; +import org.onap.a1pesimulator.data.fileready.RanPeriodicEvent; +import org.onap.a1pesimulator.data.ves.VesEvent; import org.onap.a1pesimulator.service.cell.RanCellService; import org.onap.a1pesimulator.service.cell.RanCellStateService; import org.onap.a1pesimulator.service.ves.RanVesBrokerService; @@ -32,6 +34,9 @@ import org.springframework.web.bind.annotation.RequestMapping; import org.springframework.web.bind.annotation.RequestParam; import org.springframework.web.bind.annotation.RestController; +import io.swagger.annotations.ApiOperation; +import io.swagger.annotations.ApiParam; + @RestController @RequestMapping({"${restapi.version}/ran/cells"}) public class RanCellController { @@ -62,22 +67,25 @@ public class RanCellController { return ResponseEntity.ok(ranCellService.getCellById(identifier)); } + @ApiOperation("Start sending failure VES events for specific cell") @PostMapping(value = "/{identifier}/startFailure") - public ResponseEntity<String> startSendingFailureVesEvents(final @PathVariable String identifier) { + public ResponseEntity<String> startSendingFailureVesEvents(@ApiParam(value = "Cell Id") final @PathVariable String identifier, + @ApiParam(value = "Reporting Method", defaultValue = "FILE_READY", required = true) final @RequestParam() ReportingMethodEnum reportingMethod) { ranCellService.failure(identifier); - ranVesBrokerService.startSendingFailureVesEvents(identifier); + ranVesBrokerService.startSendingFailureVesEvents(identifier, reportingMethod); ranCellStateService.failingState(identifier); return ResponseEntity.accepted().body("Failure VES Event sending started"); } + @ApiOperation("Stop sending failure VES events for specific cell") @PostMapping(value = "/{identifier}/stopFailure") - public ResponseEntity<Void> stopSendingFailureVesEvents(final @PathVariable String identifier) { + public ResponseEntity<Void> stopSendingFailureVesEvents(@ApiParam(value = "Cell Id") final @PathVariable String identifier) { ranCellService.recoverFromFailure(identifier); - Optional<RanPeriodicVesEvent> vesEvent = ranVesBrokerService.stopSendingVesEvents(identifier); + Optional<RanPeriodicEvent> vesEvent = ranVesBrokerService.stopSendingVesEvents(identifier); if (!vesEvent.isPresent()) { return ResponseEntity.notFound().build(); @@ -87,15 +95,18 @@ public class RanCellController { return ResponseEntity.accepted().build(); } + @ApiOperation("Start sending normal VES events for specific cell and in specific granularity period") @PostMapping(value = "/{identifier}/start") - public ResponseEntity<String> startSendingVesEvents(final @RequestBody Optional<Event> vesEventOpt, - final @PathVariable String identifier, final @RequestParam(required = false) Integer interval) { + public ResponseEntity<String> startSendingVesEvents(@ApiParam(value = "Standard Measurement Event JSON") final @RequestBody Optional<VesEvent> vesEventOpt, + @ApiParam(value = "Cell Id") final @PathVariable String identifier, + @ApiParam(value = "Granularity period in seconds", example = "60") final @RequestParam(required = false) Integer interval, + @ApiParam(value = "Reporting Method", defaultValue = "FILE_READY", required = true) final @RequestParam() ReportingMethodEnum reportingMethod) { log.info("Start sending ves events every {} seconds for {} ", getInterval(interval), identifier); - Event vesEvent = vesEventOpt.orElse(ranVesBrokerService.getGlobalPmVesStructure()); + VesEvent vesEvent = vesEventOpt.orElse(ranVesBrokerService.getGlobalPmVesStructure()); ResponseEntity<String> responseEntity = - ranVesBrokerService.startSendingVesEvents(identifier, vesEvent, getInterval(interval)); + ranVesBrokerService.startSendingVesEvents(identifier, vesEvent, getInterval(interval), reportingMethod); if (!responseEntity.getStatusCode().is2xxSuccessful()) { return responseEntity; } @@ -104,10 +115,11 @@ public class RanCellController { return responseEntity; } + @ApiOperation("Stop sending normal VES events for specific cell") @PostMapping(value = "/{identifier}/stop") - public ResponseEntity<Void> stopSendingVesEvents(final @PathVariable String identifier) { + public ResponseEntity<Void> stopSendingVesEvents(@ApiParam(value = "Cell Id") final @PathVariable String identifier) { log.info("Stop sending custom ves events for {}", identifier); - Optional<RanPeriodicVesEvent> vesEvent = ranVesBrokerService.stopSendingVesEvents(identifier); + Optional<RanPeriodicEvent> vesEvent = ranVesBrokerService.stopSendingVesEvents(identifier); if (!vesEvent.isPresent()) { return ResponseEntity.notFound().build(); } @@ -117,7 +129,7 @@ public class RanCellController { } @GetMapping(value = "/{identifier}/eventStructure") - public ResponseEntity<Event> getVesEventStructure(final @PathVariable String identifier) { + public ResponseEntity<VesEvent> getVesEventStructure(final @PathVariable String identifier) { if (!ranVesBrokerService.getEnabledEventElementIdentifiers().contains(identifier)) { return ResponseEntity.notFound().build(); } diff --git a/src/main/java/org/onap/a1pesimulator/data/Event.java b/src/main/java/org/onap/a1pesimulator/data/Event.java new file mode 100644 index 0000000..2850931 --- /dev/null +++ b/src/main/java/org/onap/a1pesimulator/data/Event.java @@ -0,0 +1,26 @@ +package org.onap.a1pesimulator.data; + +import org.onap.a1pesimulator.data.ves.CommonEventHeader; + +import com.fasterxml.jackson.annotation.JsonInclude; +import com.fasterxml.jackson.annotation.JsonTypeInfo; +import com.fasterxml.jackson.annotation.JsonTypeInfo.As; +import com.fasterxml.jackson.annotation.JsonTypeInfo.Id; +import com.fasterxml.jackson.annotation.JsonTypeName; + +import lombok.AllArgsConstructor; +import lombok.Builder; +import lombok.Data; +import lombok.NoArgsConstructor; + +@Data +@Builder +@AllArgsConstructor +@NoArgsConstructor +@JsonTypeName("event") +@JsonTypeInfo(include = As.WRAPPER_OBJECT, use = Id.NAME) +@JsonInclude(JsonInclude.Include.NON_NULL) +public class Event { + + private CommonEventHeader commonEventHeader; +} diff --git a/src/main/java/org/onap/a1pesimulator/data/ReportingMethodEnum.java b/src/main/java/org/onap/a1pesimulator/data/ReportingMethodEnum.java new file mode 100644 index 0000000..ee2a47f --- /dev/null +++ b/src/main/java/org/onap/a1pesimulator/data/ReportingMethodEnum.java @@ -0,0 +1,16 @@ +package org.onap.a1pesimulator.data; + +import lombok.Getter; + +@Getter +public enum ReportingMethodEnum { + FILE_READY("File ready"), + VES("VES"); + + public final String value; + + ReportingMethodEnum(String stateName) { + this.value = stateName; + } + +} diff --git a/src/main/java/org/onap/a1pesimulator/data/RequestParameters.java b/src/main/java/org/onap/a1pesimulator/data/RequestParameters.java new file mode 100644 index 0000000..6486f46 --- /dev/null +++ b/src/main/java/org/onap/a1pesimulator/data/RequestParameters.java @@ -0,0 +1,19 @@ +package org.onap.a1pesimulator.data; + +import org.onap.a1pesimulator.data.ves.VesEvent; + +import lombok.Builder; +import lombok.Data; + +/** + * Request parameters object to pass multiply param values to methods + */ +@Data +@Builder +public class RequestParameters { + + String identifier; + VesEvent vesEvent; + Integer interval; + ReportingMethodEnum reportingMethod; +} diff --git a/src/main/java/org/onap/a1pesimulator/data/VnfConfig.java b/src/main/java/org/onap/a1pesimulator/data/VnfConfig.java index bd0e1d0..9e59e9e 100644 --- a/src/main/java/org/onap/a1pesimulator/data/VnfConfig.java +++ b/src/main/java/org/onap/a1pesimulator/data/VnfConfig.java @@ -13,8 +13,12 @@ package org.onap.a1pesimulator.data; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + import com.fasterxml.jackson.annotation.JsonIgnoreProperties; import com.fasterxml.jackson.annotation.JsonProperty; + import lombok.Getter; @@ -22,6 +26,8 @@ import lombok.Getter; @JsonIgnoreProperties(ignoreUnknown = true) public class VnfConfig { + private static final Logger log = LoggerFactory.getLogger(VnfConfig.class); + @JsonProperty("vesHost") private String vesHost; @JsonProperty("vesPort") @@ -34,4 +40,14 @@ public class VnfConfig { private String vnfId; @JsonProperty("vnfName") private String vnfName; + @JsonProperty("repPeriod") + private int repPeriod; + + public int getRepPeriod() { + if (repPeriod == 0) { + log.info("repPeriod is not set or is 0, defaulting to 60s."); + return 60; + } + return repPeriod; + } } diff --git a/src/main/java/org/onap/a1pesimulator/data/fileready/EventMemoryHolder.java b/src/main/java/org/onap/a1pesimulator/data/fileready/EventMemoryHolder.java new file mode 100644 index 0000000..95b4e3e --- /dev/null +++ b/src/main/java/org/onap/a1pesimulator/data/fileready/EventMemoryHolder.java @@ -0,0 +1,37 @@ +/* + * Copyright (C) 2021 Samsung Electronics + * 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 + */ + +package org.onap.a1pesimulator.data.fileready; + +import java.time.ZonedDateTime; + +import org.onap.a1pesimulator.data.ves.VesEvent; + +import lombok.AllArgsConstructor; +import lombok.Getter; +import lombok.Setter; + +/** + * EventMemoryHolder object stores event's information which are collacted for a specific reporting period + */ +@Getter +@Setter +@AllArgsConstructor +public class EventMemoryHolder { + + private String cellId; + private String jobId; + private Integer granPeriod; + private ZonedDateTime eventDate; + private VesEvent event; +} diff --git a/src/main/java/org/onap/a1pesimulator/data/fileready/FileData.java b/src/main/java/org/onap/a1pesimulator/data/fileready/FileData.java new file mode 100644 index 0000000..7e47897 --- /dev/null +++ b/src/main/java/org/onap/a1pesimulator/data/fileready/FileData.java @@ -0,0 +1,18 @@ +package org.onap.a1pesimulator.data.fileready; + +import java.io.File; + +import lombok.Builder; +import lombok.Data; + +/** + * File data object to stored File Ready Event, PM Bulk File and its archive + */ +@Data +@Builder +public class FileData { + + File pmBulkFile; + File archivedPmBulkFile; + FileReadyEvent fileReadyEvent; +} diff --git a/src/main/java/org/onap/a1pesimulator/data/fileready/FileReadyEvent.java b/src/main/java/org/onap/a1pesimulator/data/fileready/FileReadyEvent.java new file mode 100644 index 0000000..49ca713 --- /dev/null +++ b/src/main/java/org/onap/a1pesimulator/data/fileready/FileReadyEvent.java @@ -0,0 +1,24 @@ +/* + * Copyright (C) 2021 Samsung Electronics + * 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 + */ + +package org.onap.a1pesimulator.data.fileready; + +import org.onap.a1pesimulator.data.Event; + +import lombok.Data; + +@Data +public class FileReadyEvent extends Event { + + private NotificationFields notificationFields; +} diff --git a/src/main/java/org/onap/a1pesimulator/data/ves/Event.java b/src/main/java/org/onap/a1pesimulator/data/fileready/NotificationFields.java index ded1848..5ac0ced 100644 --- a/src/main/java/org/onap/a1pesimulator/data/ves/Event.java +++ b/src/main/java/org/onap/a1pesimulator/data/fileready/NotificationFields.java @@ -11,13 +11,13 @@ * limitations under the License */ -package org.onap.a1pesimulator.data.ves; +package org.onap.a1pesimulator.data.fileready; + +import java.util.List; +import java.util.Map; import com.fasterxml.jackson.annotation.JsonInclude; -import com.fasterxml.jackson.annotation.JsonTypeInfo; -import com.fasterxml.jackson.annotation.JsonTypeInfo.As; -import com.fasterxml.jackson.annotation.JsonTypeInfo.Id; -import com.fasterxml.jackson.annotation.JsonTypeName; + import lombok.AllArgsConstructor; import lombok.Builder; import lombok.Data; @@ -27,13 +27,20 @@ import lombok.NoArgsConstructor; @Builder @NoArgsConstructor @AllArgsConstructor -@JsonTypeName("event") -@JsonTypeInfo(include = As.WRAPPER_OBJECT, use = Id.NAME) @JsonInclude(JsonInclude.Include.NON_NULL) -public class Event { +public class NotificationFields { + + private String changeIdentifier; + private String changeType; + private String notificationFieldsVersion; + + private List<ArrayOfNamedHashMap> arrayOfNamedHashMap; - private CommonEventHeader commonEventHeader; - private FaultFields faultFields; + @Data + @JsonInclude(JsonInclude.Include.NON_NULL) + public static class ArrayOfNamedHashMap { - private MeasurementFields measurementFields; + private String name; + private Map<String, String> hashMap; + } } diff --git a/src/main/java/org/onap/a1pesimulator/data/fileready/RanPeriodicEvent.java b/src/main/java/org/onap/a1pesimulator/data/fileready/RanPeriodicEvent.java new file mode 100644 index 0000000..56d375e --- /dev/null +++ b/src/main/java/org/onap/a1pesimulator/data/fileready/RanPeriodicEvent.java @@ -0,0 +1,32 @@ +/* + * Copyright (C) 2021 Samsung Electronics + * 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 + */ + +package org.onap.a1pesimulator.data.fileready; + +import java.util.concurrent.ScheduledFuture; + +import org.onap.a1pesimulator.data.ves.VesEvent; +import org.onap.a1pesimulator.service.common.AbstractRanRunnable; + +import lombok.Builder; +import lombok.Data; + +@Data +@Builder +public class RanPeriodicEvent { + + private VesEvent event; + private Integer interval; + private ScheduledFuture<?> scheduledFuture; + private AbstractRanRunnable ranRunnable; +} diff --git a/src/main/java/org/onap/a1pesimulator/data/ves/RanPeriodicVesEvent.java b/src/main/java/org/onap/a1pesimulator/data/fileready/RanPeriodicSendReport.java index 908f66b..ca812e7 100644 --- a/src/main/java/org/onap/a1pesimulator/data/ves/RanPeriodicVesEvent.java +++ b/src/main/java/org/onap/a1pesimulator/data/fileready/RanPeriodicSendReport.java @@ -11,19 +11,20 @@ * limitations under the License */ -package org.onap.a1pesimulator.data.ves; +package org.onap.a1pesimulator.data.fileready; import java.util.concurrent.ScheduledFuture; + +import org.onap.a1pesimulator.service.fileready.RanSendReportsRunnable; + import lombok.Builder; import lombok.Data; -import org.onap.a1pesimulator.service.ves.RanSendVesRunnable; @Data @Builder -public class RanPeriodicVesEvent { +public class RanPeriodicSendReport { - private Event event; private Integer interval; private ScheduledFuture<?> scheduledFuture; - private RanSendVesRunnable sendVesRunnable; + private RanSendReportsRunnable ranSendReportsRunnable; } diff --git a/src/main/java/org/onap/a1pesimulator/data/ves/GlobalVesConfiguration.java b/src/main/java/org/onap/a1pesimulator/data/ves/GlobalVesConfiguration.java index 7532573..2d51a50 100644 --- a/src/main/java/org/onap/a1pesimulator/data/ves/GlobalVesConfiguration.java +++ b/src/main/java/org/onap/a1pesimulator/data/ves/GlobalVesConfiguration.java @@ -21,5 +21,5 @@ import lombok.Getter; public class GlobalVesConfiguration { private Integer interval; - private Event event; + private VesEvent event; } diff --git a/src/main/java/org/onap/a1pesimulator/data/ves/VesEvent.java b/src/main/java/org/onap/a1pesimulator/data/ves/VesEvent.java new file mode 100644 index 0000000..3655ede --- /dev/null +++ b/src/main/java/org/onap/a1pesimulator/data/ves/VesEvent.java @@ -0,0 +1,25 @@ +/* + * Copyright (C) 2021 Samsung Electronics + * 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 + */ + +package org.onap.a1pesimulator.data.ves; + +import org.onap.a1pesimulator.data.Event; + +import lombok.Data; + +@Data +public class VesEvent extends Event { + + private FaultFields faultFields; + private MeasurementFields measurementFields; +} diff --git a/src/main/java/org/onap/a1pesimulator/service/a1/SetLowRangeValuesOnPolicyAction.java b/src/main/java/org/onap/a1pesimulator/service/a1/SetLowRangeValuesOnPolicyAction.java index e608aa5..29c55de 100644 --- a/src/main/java/org/onap/a1pesimulator/service/a1/SetLowRangeValuesOnPolicyAction.java +++ b/src/main/java/org/onap/a1pesimulator/service/a1/SetLowRangeValuesOnPolicyAction.java @@ -14,9 +14,10 @@ package org.onap.a1pesimulator.service.a1; import java.util.List; -import org.onap.a1pesimulator.data.ves.Event; + +import org.onap.a1pesimulator.data.fileready.RanPeriodicEvent; +import org.onap.a1pesimulator.data.ves.VesEvent; import org.onap.a1pesimulator.data.ves.MeasurementFields.AdditionalMeasurement; -import org.onap.a1pesimulator.data.ves.RanPeriodicVesEvent; import org.onap.a1pesimulator.service.ves.RanVesBrokerService; import org.onap.a1pesimulator.util.JsonUtils; import org.onap.a1pesimulator.util.RanVesUtils; @@ -42,12 +43,12 @@ public class SetLowRangeValuesOnPolicyAction implements OnPolicyAction { vesBrokerService.getPeriodicEventsCache().values().forEach(this::updateEvent); } - private void updateEvent(RanPeriodicVesEvent periodicEvent) { + private void updateEvent(RanPeriodicEvent periodicEvent) { List<AdditionalMeasurement> lowRangeValues = RanVesUtils.setLowRangeValues( periodicEvent.getEvent().getMeasurementFields().getAdditionalMeasurements()); - Event clonedEvent = JsonUtils.INSTANCE.clone(periodicEvent.getEvent()); + VesEvent clonedEvent = JsonUtils.INSTANCE.clone(periodicEvent.getEvent()); clonedEvent.getMeasurementFields().setAdditionalMeasurements(lowRangeValues); - periodicEvent.getSendVesRunnable().updateEvent(clonedEvent); + periodicEvent.getRanRunnable().updateEvent(clonedEvent); } } diff --git a/src/main/java/org/onap/a1pesimulator/service/common/AbstractRanRunnable.java b/src/main/java/org/onap/a1pesimulator/service/common/AbstractRanRunnable.java new file mode 100644 index 0000000..cfb1390 --- /dev/null +++ b/src/main/java/org/onap/a1pesimulator/service/common/AbstractRanRunnable.java @@ -0,0 +1,30 @@ +package org.onap.a1pesimulator.service.common; + +import java.util.Collection; + +import org.onap.a1pesimulator.data.ves.VesEvent; +import org.onap.a1pesimulator.service.ves.OnEventAction; + +public abstract class AbstractRanRunnable implements Runnable { + + protected VesEvent event; + protected final EventCustomizer eventCustomizer; + protected final Collection<OnEventAction> onEventAction; + + + protected AbstractRanRunnable(VesEvent event, EventCustomizer eventCustomizer, + Collection<OnEventAction> onEventActions) { + this.event = event; + this.eventCustomizer = eventCustomizer; + this.onEventAction = onEventActions; + } + + public void updateEvent(VesEvent event) { + this.event = event; + } + + @Override + public void run() { + //will be implemented in subclasses + } +} diff --git a/src/main/java/org/onap/a1pesimulator/service/common/EventCustomizer.java b/src/main/java/org/onap/a1pesimulator/service/common/EventCustomizer.java new file mode 100644 index 0000000..23096bd --- /dev/null +++ b/src/main/java/org/onap/a1pesimulator/service/common/EventCustomizer.java @@ -0,0 +1,10 @@ +package org.onap.a1pesimulator.service.common; + +import java.util.function.Function; + +import org.onap.a1pesimulator.data.ves.VesEvent; + +@FunctionalInterface +public interface EventCustomizer extends Function<VesEvent, VesEvent> { + +} diff --git a/src/main/java/org/onap/a1pesimulator/service/fileready/FileReadyEventService.java b/src/main/java/org/onap/a1pesimulator/service/fileready/FileReadyEventService.java new file mode 100644 index 0000000..47b338d --- /dev/null +++ b/src/main/java/org/onap/a1pesimulator/service/fileready/FileReadyEventService.java @@ -0,0 +1,48 @@ +package org.onap.a1pesimulator.service.fileready; + +import static org.onap.a1pesimulator.service.fileready.FtpServerService.deletePMBulkFile; + +import java.io.File; + +import org.onap.a1pesimulator.data.fileready.FileData; +import org.springframework.stereotype.Service; + +import reactor.core.publisher.Mono; + +/** + * Service for PM Bulk File creation and its handling + */ + +@Service +public class FileReadyEventService { + + /** + * It will create FileReadyEvent.json which will go to VES Collector + * + * @return created FileReadyEvent + */ + protected Mono<FileData> createFileReadyEventAndDeleteTmpFile(Mono<FileData> fileMono) { + return fileMono + .map(this::createFileReadyEvent) + .doOnNext(file -> deleteTempArchivedBulkFile(file.getArchivedPmBulkFile())); + } + + /** + * Creates File Ready Event + * + * @param fileData information about PM Bulk Files created in previous steps + * @return added newly created FileReadyEvent to FileData + */ + protected FileData createFileReadyEvent(FileData fileData) { + return fileData; + } + + /** + * Deletes temporary archived PM Bulk File + * + * @param fileMono temporary archived PM Bulk File + */ + private void deleteTempArchivedBulkFile(File fileMono) { + deletePMBulkFile(fileMono); + } +} diff --git a/src/main/java/org/onap/a1pesimulator/service/fileready/FtpServerService.java b/src/main/java/org/onap/a1pesimulator/service/fileready/FtpServerService.java new file mode 100644 index 0000000..56735fd --- /dev/null +++ b/src/main/java/org/onap/a1pesimulator/service/fileready/FtpServerService.java @@ -0,0 +1,103 @@ +package org.onap.a1pesimulator.service.fileready; + +import java.io.File; +import java.io.IOException; +import java.nio.file.Files; + +import org.onap.a1pesimulator.data.fileready.FileData; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.beans.factory.annotation.Value; +import org.springframework.context.annotation.PropertySource; +import org.springframework.stereotype.Service; + +import reactor.core.publisher.Mono; + +@Service +@PropertySource("classpath:application.properties") +public class FtpServerService { + + private static final Logger log = LoggerFactory.getLogger(FtpServerService.class); + + @Value("${ftp.server.url}") + private String ftpServerUrl; + + @Value("${ftp.server.protocol}") + private String ftpServerProtocol; + + @Value("${ftp.server.port}") + private String ftpServerPort; + + @Value("${ftp.server.filepath}") + private String ftpServerFilepath; + + @Value("${ftp.server.username}") + private String ftpServerUsername; + + @Value("${ftp.server.password}") + private String ftpServerPassword; + + public Mono<FileData> uploadFileToFtp(FileData fileData) { + return Mono.just(fileData) + .flatMap(this::tryToCompressFile) + .flatMap(this::tryToUploadFileToFtp) + .onErrorResume(throwable -> resumeError(throwable, fileData)) + .doOnNext(file -> deletePMBulkFile(file.getPmBulkFile())); + } + + /** + * Trying to compress file into .gz + * + * @param fileData file to be archived + * @return archived file + */ + private Mono<FileData> tryToCompressFile(FileData fileData) { + fileData.setArchivedPmBulkFile(fileData.getPmBulkFile()); + return Mono.just(fileData); + } + + /** + * Upload file to FTP + * + * @param fileData archived file in Mono + * @return archived file for fileReadyEvent + */ + private Mono<FileData> tryToUploadFileToFtp(FileData fileData) { + return Mono.just(fileData); + } + + + /** + * Deletes created PM Bulk File xml from temp storage after successful upload to FTP + * + * @param file file which we gonna delete + */ + public static void deletePMBulkFile(File file) { + try { + log.trace("Deleting file: {}", file.getAbsoluteFile()); + Files.delete(file.toPath()); + } catch (IOException e) { + log.warn("Could not delete file: {}", file.getName(), e); + } + } + + /** + * Get path to FTP server + * + * @return for example: "sftp://foo:pass@106.120.119.170:2222/upload/" + */ + public String getFtpPath() { + return ftpServerProtocol + "://" + ftpServerUsername + ":" + ftpServerPassword + "@" + ftpServerUrl + ":" + ftpServerPort + "/" + ftpServerFilepath + + "/"; + } + + /** + * Try to clean up things after an exception + */ + private Mono<FileData> resumeError(Throwable throwable, FileData fileData) { + log.error("Error occurs while uploading file to FTP server", throwable); + deletePMBulkFile(fileData.getPmBulkFile()); + deletePMBulkFile(fileData.getArchivedPmBulkFile()); + return Mono.empty(); + } +} diff --git a/src/main/java/org/onap/a1pesimulator/service/fileready/PMBulkFileService.java b/src/main/java/org/onap/a1pesimulator/service/fileready/PMBulkFileService.java new file mode 100644 index 0000000..980a653 --- /dev/null +++ b/src/main/java/org/onap/a1pesimulator/service/fileready/PMBulkFileService.java @@ -0,0 +1,110 @@ +package org.onap.a1pesimulator.service.fileready; + +import static java.util.Comparator.comparing; +import static java.util.Objects.isNull; +import static org.onap.a1pesimulator.util.Constants.EMPTY_STRING; +import static org.onap.a1pesimulator.util.Constants.TEMP_DIR; +import static org.onap.a1pesimulator.util.Convertors.YYYYMMDD_PATTERN; +import static org.onap.a1pesimulator.util.Convertors.truncateToSpecifiedMinutes; +import static org.onap.a1pesimulator.util.Convertors.zonedDateTimeToString; + +import java.io.File; +import java.time.ZonedDateTime; +import java.util.Collections; +import java.util.Comparator; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.concurrent.atomic.AtomicInteger; + +import org.onap.a1pesimulator.data.fileready.EventMemoryHolder; +import org.onap.a1pesimulator.data.fileready.FileData; +import org.springframework.stereotype.Service; + +import reactor.core.publisher.Mono; + +/** + * Service for PM Bulk File creation and handling + */ + +@Service +public class PMBulkFileService { + + private static Map<String, AtomicInteger> uniqueFileNamesWithCount; + + /** + * Generate PM Bulk File xml from stored events + * + * @param collectedEvents list of stored events + * @return generated file in Mono object + */ + public Mono<FileData> generatePMBulkFileXml(List<EventMemoryHolder> collectedEvents) { + return Mono.just(FileData.builder().pmBulkFile(getXmlFile(collectedEvents)).build()); + } + + /** + * Generate PM Bulk File and its name. Example: D20050907.1030+0000-20050909.1500+0000_DomainId_-_2 + * + * @param collectedEvents list of stored events + * @return newly created File + */ + private static File getXmlFile(List<EventMemoryHolder> collectedEvents) { + StringBuilder fileNameBuilder = new StringBuilder("D"); + ZonedDateTime firstEventTime = earliestEventTime(collectedEvents); + ZonedDateTime lastEventTime = latestEventTime(collectedEvents); + fileNameBuilder.append(zonedDateTimeToString(firstEventTime, YYYYMMDD_PATTERN)).append("."); + fileNameBuilder.append(zonedDateTimeToString(truncateToSpecifiedMinutes(firstEventTime, 5), "HHmmZ")).append("-"); + fileNameBuilder.append(zonedDateTimeToString(lastEventTime, YYYYMMDD_PATTERN)).append("."); + fileNameBuilder.append(zonedDateTimeToString(truncateToSpecifiedMinutes(lastEventTime, 5), "HHmmZ")); + fileNameBuilder.append(appendRcIfNecessary(fileNameBuilder)); + fileNameBuilder.append(".xml"); + + return new File(TEMP_DIR, fileNameBuilder.toString()); + } + + /** + * The RC parameter is a running count and shall be appended only if the filename is otherwise not unique, i.e. more than one file is generated and all + * other parameters of the file name are identical. + * + * @param fileNameBuilder stringBuilder which contains currently generated file name + * @return sequence number or empty string + */ + private static String appendRcIfNecessary(StringBuilder fileNameBuilder) { + String fileName = fileNameBuilder.toString(); + int sequence = 0; + if (isNull(uniqueFileNamesWithCount)) { + uniqueFileNamesWithCount = Collections.synchronizedMap(new HashMap<>()); + } + if (uniqueFileNamesWithCount.containsKey(fileName)) { + sequence = uniqueFileNamesWithCount.get(fileName).incrementAndGet(); + } else { + uniqueFileNamesWithCount.put(fileName, new AtomicInteger(0)); + } + return sequence > 0 ? "_-_" + sequence : EMPTY_STRING; + } + + /** + * Get ZonedDateTime of the earliest event in that reporting period + * + * @param collectedEvents list of compared events + * @return the earliest ZonedDateTime + */ + private static ZonedDateTime earliestEventTime(List<EventMemoryHolder> collectedEvents) { + return collectedEvents.stream() + .map(EventMemoryHolder::getEventDate) + .min(comparing(ZonedDateTime::toEpochSecond, Comparator.nullsLast(Comparator.naturalOrder()))) + .orElse(ZonedDateTime.now()); + } + + /** + * Get ZonedDateTime of the latest event in that reporting period + * + * @param collectedEvents list of compared events + * @return the latest ZonedDateTime + */ + private static ZonedDateTime latestEventTime(List<EventMemoryHolder> collectedEvents) { + return collectedEvents.stream().map(EventMemoryHolder::getEventDate) + .max(comparing(ZonedDateTime::toEpochSecond, Comparator.nullsLast(Comparator.naturalOrder()))) + .orElse(ZonedDateTime.now()); + } +} diff --git a/src/main/java/org/onap/a1pesimulator/service/fileready/RanFileReadyHolder.java b/src/main/java/org/onap/a1pesimulator/service/fileready/RanFileReadyHolder.java new file mode 100644 index 0000000..148c059 --- /dev/null +++ b/src/main/java/org/onap/a1pesimulator/service/fileready/RanFileReadyHolder.java @@ -0,0 +1,135 @@ +/* + * Copyright (C) 2021 Samsung Electronics + * 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 + */ + +package org.onap.a1pesimulator.service.fileready; + +import static java.util.Objects.isNull; + +import java.time.ZonedDateTime; +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; + +import org.onap.a1pesimulator.data.fileready.EventMemoryHolder; +import org.onap.a1pesimulator.data.fileready.FileData; +import org.onap.a1pesimulator.data.ves.VesEvent; +import org.onap.a1pesimulator.exception.VesBrokerException; +import org.onap.a1pesimulator.service.ves.RanVesSender; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.stereotype.Service; +import org.springframework.util.CollectionUtils; + +import reactor.core.publisher.Mono; + +/** + * Entry point for PM Bulk File event + */ +@Service +public class RanFileReadyHolder { + + private static final Logger log = LoggerFactory.getLogger(RanFileReadyHolder.class); + + private List<EventMemoryHolder> collectedEvents; + private final RanVesSender ranVesSender; + private final FtpServerService ftpServerService; + private final PMBulkFileService xmlFileService; + private final FileReadyEventService fileReadyEventService; + + public RanFileReadyHolder(RanVesSender ranVesSender, FtpServerService ftpServerService, PMBulkFileService xmlFileService, + FileReadyEventService fileReadyEventService) { + this.ranVesSender = ranVesSender; + this.ftpServerService = ftpServerService; + this.xmlFileService = xmlFileService; + this.fileReadyEventService = fileReadyEventService; + } + + /** + * Run entire process: PM Bulk File creation-> upload to FTP -> delete temp PM Bulk File -> create File Ready Event - > send it to VES Collector + * collectedEvents are synchronized to not be updated by other threads during PM Bulk File creation + */ + public void createPMBulkFileAndSendFileReadyMessage() { + synchronized (getCollectedEvents()) { + Mono.justOrEmpty(getCollectedEvents()) + .filter(this::areSomeEventsStored) + .flatMap(xmlFileService::generatePMBulkFileXml) + .map(ftpServerService::uploadFileToFtp) + .flatMap(fileReadyEventService::createFileReadyEventAndDeleteTmpFile) + .doOnNext(this::sendEventToVesCollector) + .subscribe(fileData -> informAboutSuccess(), this::informAboutError); + } + } + + /** + * Adds current event to the memory, which is List<EventMemoryHolder> + * + * @param vesEvent event from specific cell + * @throws VesBrokerException in case of any problem with adding to List, it throws an exception + */ + public void saveEventToMemory(VesEvent vesEvent, String cellId, String jobId, Integer granPeriod) throws VesBrokerException { + try { + getCollectedEvents().add(new EventMemoryHolder(cellId, jobId, granPeriod, ZonedDateTime.now(), vesEvent)); + log.trace("Saving VES event for cell {} with granularity period {} and sequence number {}", cellId, granPeriod, getCollectedEvents().size()); + } catch (Exception e) { + String errorMsg = "Failed to save VES event to memory with exception:" + e; + throw new VesBrokerException(errorMsg); + } + } + + /** + * Sends FileReadyEvent to VES Collector + * + * @param fileData object with FileReadyEvent file + */ + private void sendEventToVesCollector(FileData fileData) { + ranVesSender.send(fileData.getFileReadyEvent()); + } + + /** + * Log about successful operation + */ + private void informAboutSuccess() { + log.info("PM Bulk file was generated, uploaded to FTP and File ready event was send to VES Collector"); + } + + /** + * Log an error if occurs during the process + * + * @param throwable - error raised in some of the steps + */ + private void informAboutError(Throwable throwable) { + log.info("File ready event was unsuccessful: {}", throwable.getMessage()); + } + + /** + * Check if there are any Events stored in the memory. Used before creating PM Bulk File xml + * + * @param collectedEvents list of stored events + * @return true there is at least one event / false - no event at all + */ + private boolean areSomeEventsStored(List<EventMemoryHolder> collectedEvents) { + return !CollectionUtils.isEmpty(collectedEvents); + } + + /** + * Factory to get List<EventMemoryHolder> + * + * @return existing or newly created List<EventMemoryHolder> + */ + public List<EventMemoryHolder> getCollectedEvents() { + if (isNull(collectedEvents)) { + collectedEvents = Collections.synchronizedList(new ArrayList<>()); + } + return collectedEvents; + } +} diff --git a/src/main/java/org/onap/a1pesimulator/service/fileready/RanSaveFileReadyRunnable.java b/src/main/java/org/onap/a1pesimulator/service/fileready/RanSaveFileReadyRunnable.java new file mode 100644 index 0000000..53d4600 --- /dev/null +++ b/src/main/java/org/onap/a1pesimulator/service/fileready/RanSaveFileReadyRunnable.java @@ -0,0 +1,54 @@ +/* + * Copyright (C) 2021 Samsung Electronics + * 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 + */ + +package org.onap.a1pesimulator.service.fileready; + +import java.util.Collection; +import java.util.UUID; + +import org.onap.a1pesimulator.data.ves.VesEvent; +import org.onap.a1pesimulator.exception.VesBrokerException; +import org.onap.a1pesimulator.service.common.AbstractRanRunnable; +import org.onap.a1pesimulator.service.common.EventCustomizer; +import org.onap.a1pesimulator.service.ves.OnEventAction; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +public class RanSaveFileReadyRunnable extends AbstractRanRunnable { + + private static final Logger log = LoggerFactory.getLogger(RanSaveFileReadyRunnable.class); + private final Integer granPeriod; + private final String cellId; + private final String jobId; + private final RanFileReadyHolder ranFileReadyHolder; + + public RanSaveFileReadyRunnable(RanFileReadyHolder ranFileReadyHolder, String cellId, VesEvent event, EventCustomizer eventCustomizer, Integer interval, + Collection<OnEventAction> onEventActions) { + super(event, eventCustomizer, onEventActions); + this.ranFileReadyHolder = ranFileReadyHolder; + this.granPeriod = interval; + this.cellId = cellId; + this.jobId = UUID.randomUUID() + "-" + cellId; + } + + @Override + public void run() { + try { + VesEvent customizedEvent = eventCustomizer.apply(event); + onEventAction.forEach(action -> action.onEvent(customizedEvent)); + ranFileReadyHolder.saveEventToMemory(customizedEvent, cellId, jobId, granPeriod); + } catch (VesBrokerException e) { + log.error("Saving file ready event failed: {}", e.getMessage()); + } + } +} diff --git a/src/main/java/org/onap/a1pesimulator/service/fileready/RanSendReportsRunnable.java b/src/main/java/org/onap/a1pesimulator/service/fileready/RanSendReportsRunnable.java new file mode 100644 index 0000000..f92d479 --- /dev/null +++ b/src/main/java/org/onap/a1pesimulator/service/fileready/RanSendReportsRunnable.java @@ -0,0 +1,28 @@ +/* + * Copyright (C) 2021 Samsung Electronics + * 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 + */ + +package org.onap.a1pesimulator.service.fileready; + +public class RanSendReportsRunnable implements Runnable { + + protected final RanFileReadyHolder ranFileReadyHolder; + + public RanSendReportsRunnable(RanFileReadyHolder ranFileReadyHolder) { + this.ranFileReadyHolder = ranFileReadyHolder; + } + + @Override + public void run() { + ranFileReadyHolder.createPMBulkFileAndSendFileReadyMessage(); + } +} diff --git a/src/main/java/org/onap/a1pesimulator/service/ves/OnEventAction.java b/src/main/java/org/onap/a1pesimulator/service/ves/OnEventAction.java index b34594a..ba1ddab 100644 --- a/src/main/java/org/onap/a1pesimulator/service/ves/OnEventAction.java +++ b/src/main/java/org/onap/a1pesimulator/service/ves/OnEventAction.java @@ -13,10 +13,10 @@ package org.onap.a1pesimulator.service.ves; -import org.onap.a1pesimulator.data.ves.Event; +import org.onap.a1pesimulator.data.ves.VesEvent; @FunctionalInterface public interface OnEventAction { - void onEvent(Event event); + void onEvent(VesEvent event); } diff --git a/src/main/java/org/onap/a1pesimulator/service/ves/RanCellEventCustomizer.java b/src/main/java/org/onap/a1pesimulator/service/ves/RanCellEventCustomizer.java index 9922329..35c9215 100644 --- a/src/main/java/org/onap/a1pesimulator/service/ves/RanCellEventCustomizer.java +++ b/src/main/java/org/onap/a1pesimulator/service/ves/RanCellEventCustomizer.java @@ -15,10 +15,11 @@ package org.onap.a1pesimulator.service.ves; import java.util.List; import java.util.Optional; -import org.onap.a1pesimulator.data.ves.Event; + +import org.onap.a1pesimulator.data.ves.VesEvent; import org.onap.a1pesimulator.data.ves.MeasurementFields.AdditionalMeasurement; +import org.onap.a1pesimulator.service.common.EventCustomizer; import org.onap.a1pesimulator.service.ue.RanUeHolder; -import org.onap.a1pesimulator.service.ves.RanSendVesRunnable.EventCustomizer; import org.onap.a1pesimulator.util.Constants; import org.onap.a1pesimulator.util.JsonUtils; import org.onap.a1pesimulator.util.RanVesUtils; @@ -35,36 +36,36 @@ public class RanCellEventCustomizer implements EventCustomizer { } @Override - public Event apply(Event t) { - Event event = JsonUtils.INSTANCE.clone(t); + public VesEvent apply(VesEvent t) { + VesEvent event = JsonUtils.INSTANCE.clone(t); return customizeEvent(event); } - private Event customizeEvent(Event event) { + private VesEvent customizeEvent(VesEvent event) { RanVesUtils.updateHeader(event); enrichWithUeData(event); randomizeEvent(event); return event; } - private void randomizeEvent(Event event) { + private void randomizeEvent(VesEvent event) { List<AdditionalMeasurement> additionalMeasurementsToRandomize = event.getMeasurementFields().getAdditionalMeasurements(); event.getMeasurementFields().setAdditionalMeasurements( RanVesUtils.randomizeAdditionalMeasurements(additionalMeasurementsToRandomize)); } - private void enrichWithUeData(Event event) { + private void enrichWithUeData(VesEvent event) { Optional<AdditionalMeasurement> identity = event.getMeasurementFields().getAdditionalMeasurements().stream() - .filter(msrmnt -> Constants.MEASUREMENT_FIELD_IDENTIFIER - .equalsIgnoreCase( - msrmnt.getName())) - .findAny(); + .filter(msrmnt -> Constants.MEASUREMENT_FIELD_IDENTIFIER + .equalsIgnoreCase( + msrmnt.getName())) + .findAny(); identity.ifPresent(m -> addTrafficModelMeasurement(event, m)); } - private void addTrafficModelMeasurement(Event event, AdditionalMeasurement identity) { + private void addTrafficModelMeasurement(VesEvent event, AdditionalMeasurement identity) { AdditionalMeasurement trafficModelMeasurement = RanVesUtils.buildTrafficModelMeasurement(identity, ranUeHolder, UE_PARAM_TRAFFIC_MODEL_RANGE); event.getMeasurementFields().getAdditionalMeasurements().add(trafficModelMeasurement); diff --git a/src/main/java/org/onap/a1pesimulator/service/ves/RanCellFailureEventCustomizer.java b/src/main/java/org/onap/a1pesimulator/service/ves/RanCellFailureEventCustomizer.java index ac2c4fc..c45c717 100644 --- a/src/main/java/org/onap/a1pesimulator/service/ves/RanCellFailureEventCustomizer.java +++ b/src/main/java/org/onap/a1pesimulator/service/ves/RanCellFailureEventCustomizer.java @@ -20,26 +20,28 @@ import java.util.List; import java.util.Map; import java.util.Map.Entry; import java.util.Optional; -import lombok.AllArgsConstructor; -import lombok.EqualsAndHashCode; -import org.onap.a1pesimulator.data.ves.Event; + +import org.onap.a1pesimulator.data.ves.VesEvent; import org.onap.a1pesimulator.data.ves.MeasurementFields.AdditionalMeasurement; +import org.onap.a1pesimulator.service.common.EventCustomizer; import org.onap.a1pesimulator.service.ue.RanUeHolder; -import org.onap.a1pesimulator.service.ves.RanSendVesRunnable.EventCustomizer; import org.onap.a1pesimulator.util.Constants; import org.onap.a1pesimulator.util.JsonUtils; import org.onap.a1pesimulator.util.RanVesUtils; +import lombok.AllArgsConstructor; +import lombok.EqualsAndHashCode; + public class RanCellFailureEventCustomizer implements EventCustomizer { private static final String UE_PARAM_TRAFFIC_MODEL_RANGE = "[[50->10]]"; private final RanUeHolder ranUeHolder; - private final Event event; + private final VesEvent event; private final Map<Key, Value> additionalMeasurementsValues = new HashMap<>(); private final ValueFactory valueFactory; - public RanCellFailureEventCustomizer(Event event, RanUeHolder ranUeHolder) { + public RanCellFailureEventCustomizer(VesEvent event, RanUeHolder ranUeHolder) { this.ranUeHolder = ranUeHolder; this.event = event; valueFactory = new ValueFactory(); @@ -47,11 +49,11 @@ public class RanCellFailureEventCustomizer implements EventCustomizer { } @Override - public Event apply(Event t) { + public VesEvent apply(VesEvent t) { return customizeEvent(JsonUtils.INSTANCE.clone(this.event)); } - private void collectAdditionalMeasurementValues(Event event) { + private void collectAdditionalMeasurementValues(VesEvent event) { Collection<AdditionalMeasurement> additionalMeasurementsToResolve = event.getMeasurementFields().getAdditionalMeasurements(); additionalMeasurementsToResolve.forEach(this::collectAdditionalMeasurementValue); @@ -67,14 +69,14 @@ public class RanCellFailureEventCustomizer implements EventCustomizer { } } - private Event customizeEvent(Event event) { + private VesEvent customizeEvent(VesEvent event) { RanVesUtils.updateHeader(event); enrichWithUeData(event); resolveRanges(event); return event; } - private void resolveRanges(Event event) { + private void resolveRanges(VesEvent event) { List<AdditionalMeasurement> additionalMeasurementsToResolve = event.getMeasurementFields().getAdditionalMeasurements(); @@ -94,17 +96,17 @@ public class RanCellFailureEventCustomizer implements EventCustomizer { } } - private void enrichWithUeData(Event event) { + private void enrichWithUeData(VesEvent event) { Optional<AdditionalMeasurement> identity = event.getMeasurementFields().getAdditionalMeasurements().stream() - .filter(msrmnt -> Constants.MEASUREMENT_FIELD_IDENTIFIER - .equalsIgnoreCase( - msrmnt.getName())) - .findAny(); + .filter(msrmnt -> Constants.MEASUREMENT_FIELD_IDENTIFIER + .equalsIgnoreCase( + msrmnt.getName())) + .findAny(); identity.ifPresent(m -> addTrafficModelMeasurement(event, m)); } - private void addTrafficModelMeasurement(Event event, AdditionalMeasurement identity) { + private void addTrafficModelMeasurement(VesEvent event, AdditionalMeasurement identity) { AdditionalMeasurement trafficModelMeasurement = RanVesUtils.buildTrafficModelMeasurement(identity, ranUeHolder, UE_PARAM_TRAFFIC_MODEL_RANGE); event.getMeasurementFields().getAdditionalMeasurements().add(trafficModelMeasurement); diff --git a/src/main/java/org/onap/a1pesimulator/service/ves/RanCheckCellIsDeadOnEvent.java b/src/main/java/org/onap/a1pesimulator/service/ves/RanCheckCellIsDeadOnEvent.java index 1330e04..3d0c400 100644 --- a/src/main/java/org/onap/a1pesimulator/service/ves/RanCheckCellIsDeadOnEvent.java +++ b/src/main/java/org/onap/a1pesimulator/service/ves/RanCheckCellIsDeadOnEvent.java @@ -18,7 +18,7 @@ import static org.onap.a1pesimulator.service.cell.RanCellStateService.TOPIC_CELL import java.util.Optional; import org.onap.a1pesimulator.data.cell.CellDetails; import org.onap.a1pesimulator.data.cell.state.CellStateEnum; -import org.onap.a1pesimulator.data.ves.Event; +import org.onap.a1pesimulator.data.ves.VesEvent; import org.onap.a1pesimulator.data.ves.MeasurementFields; import org.onap.a1pesimulator.service.cell.RanCellsHolder; import org.slf4j.Logger; @@ -54,7 +54,7 @@ public class RanCheckCellIsDeadOnEvent implements OnEventAction { } @Override - public void onEvent(Event event) { + public void onEvent(VesEvent event) { Optional<String> cellId = getCellIdentifier(event); Optional<String> throughput = getCellThroughput(event); Optional<String> latency = getCellLatency(event); @@ -93,19 +93,19 @@ public class RanCheckCellIsDeadOnEvent implements OnEventAction { } } - private Optional<String> getCellIdentifier(Event event) { + private Optional<String> getCellIdentifier(VesEvent event) { return getValueFromAdditionalMeasurement(event, "identifier"); } - private Optional<String> getCellThroughput(Event event) { + private Optional<String> getCellThroughput(VesEvent event) { return getValueFromAdditionalMeasurement(event, "throughput"); } - private Optional<String> getCellLatency(Event event) { + private Optional<String> getCellLatency(VesEvent event) { return getValueFromAdditionalMeasurement(event, "latency"); } - private Optional<String> getValueFromAdditionalMeasurement(Event event, String key) { + private Optional<String> getValueFromAdditionalMeasurement(VesEvent event, String key) { Optional<MeasurementFields.AdditionalMeasurement> measurement = getAdditionalMeasurement(event, key); return measurement.map(this::getValueFromAdditionalMeasurement); } @@ -114,10 +114,10 @@ public class RanCheckCellIsDeadOnEvent implements OnEventAction { return measurement.getHashMap().get("value"); } - private Optional<MeasurementFields.AdditionalMeasurement> getAdditionalMeasurement(Event event, + private Optional<MeasurementFields.AdditionalMeasurement> getAdditionalMeasurement(VesEvent event, String additionalMeasurement) { return event.getMeasurementFields().getAdditionalMeasurements().stream() - .filter(e -> e.getName().equals(additionalMeasurement)).findFirst(); + .filter(e -> e.getName().equals(additionalMeasurement)).findFirst(); } private long addDelayTime(long epoch) { diff --git a/src/main/java/org/onap/a1pesimulator/service/ves/RanEventCustomizerFactory.java b/src/main/java/org/onap/a1pesimulator/service/ves/RanEventCustomizerFactory.java index 3fbeda9..c7c2ee9 100644 --- a/src/main/java/org/onap/a1pesimulator/service/ves/RanEventCustomizerFactory.java +++ b/src/main/java/org/onap/a1pesimulator/service/ves/RanEventCustomizerFactory.java @@ -14,9 +14,10 @@ package org.onap.a1pesimulator.service.ves; import java.text.MessageFormat; -import org.onap.a1pesimulator.data.ves.Event; + +import org.onap.a1pesimulator.data.ves.VesEvent; +import org.onap.a1pesimulator.service.common.EventCustomizer; import org.onap.a1pesimulator.service.ue.RanUeHolder; -import org.onap.a1pesimulator.service.ves.RanSendVesRunnable.EventCustomizer; import org.springframework.stereotype.Component; @Component @@ -30,7 +31,7 @@ public class RanEventCustomizerFactory { this.regularEventCustomizer = regularEventCustomizer; } - public EventCustomizer getEventCustomizer(Event event, Mode mode) { + public EventCustomizer getEventCustomizer(VesEvent event, Mode mode) { switch (mode) { case REGULAR: return regularEventCustomizer; diff --git a/src/main/java/org/onap/a1pesimulator/service/ves/RanSendVesRunnable.java b/src/main/java/org/onap/a1pesimulator/service/ves/RanSendVesRunnable.java index 7378bc0..c537a5f 100644 --- a/src/main/java/org/onap/a1pesimulator/service/ves/RanSendVesRunnable.java +++ b/src/main/java/org/onap/a1pesimulator/service/ves/RanSendVesRunnable.java @@ -14,44 +14,30 @@ package org.onap.a1pesimulator.service.ves; import java.util.Collection; -import java.util.function.Function; -import org.onap.a1pesimulator.data.ves.Event; -import org.onap.a1pesimulator.exception.VesBrokerException; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; -public class RanSendVesRunnable implements Runnable { +import org.onap.a1pesimulator.data.ves.VesEvent; +import org.onap.a1pesimulator.service.common.AbstractRanRunnable; +import org.onap.a1pesimulator.service.common.EventCustomizer; - private static final Logger log = LoggerFactory.getLogger(RanSendVesRunnable.class); +public class RanSendVesRunnable extends AbstractRanRunnable { private final RanVesSender vesSender; - private Event event; - private final EventCustomizer eventCustomizer; - private final Collection<OnEventAction> onEventAction; - public RanSendVesRunnable(RanVesSender vesSender, Event event, EventCustomizer eventCustomizer, + public RanSendVesRunnable(RanVesSender vesSender, VesEvent event, EventCustomizer eventCustomizer, Collection<OnEventAction> onEventActions) { + super(event, eventCustomizer, onEventActions); this.vesSender = vesSender; - this.event = event; - this.eventCustomizer = eventCustomizer; - this.onEventAction = onEventActions; } @Override public void run() { - try { - Event customizedEvent = eventCustomizer.apply(event); - onEventAction.forEach(action -> action.onEvent(customizedEvent)); - vesSender.send(customizedEvent); - } catch (VesBrokerException e) { - log.error("Sending scheduled event failed: {}", e.getMessage()); - } + VesEvent customizedEvent = eventCustomizer.apply(event); + onEventAction.forEach(action -> action.onEvent(customizedEvent)); + vesSender.send(customizedEvent); } - public void updateEvent(Event event) { + @Override + public void updateEvent(VesEvent event) { this.event = event; } - - @FunctionalInterface - public interface EventCustomizer extends Function<Event, Event> { } } diff --git a/src/main/java/org/onap/a1pesimulator/service/ves/RanVesBrokerService.java b/src/main/java/org/onap/a1pesimulator/service/ves/RanVesBrokerService.java index 8a90d46..8767251 100644 --- a/src/main/java/org/onap/a1pesimulator/service/ves/RanVesBrokerService.java +++ b/src/main/java/org/onap/a1pesimulator/service/ves/RanVesBrokerService.java @@ -16,27 +16,29 @@ package org.onap.a1pesimulator.service.ves; import java.util.Collection; import java.util.Map; import java.util.Optional; -import org.onap.a1pesimulator.data.ves.Event; -import org.onap.a1pesimulator.data.ves.RanPeriodicVesEvent; + +import org.onap.a1pesimulator.data.ReportingMethodEnum; +import org.onap.a1pesimulator.data.fileready.RanPeriodicEvent; +import org.onap.a1pesimulator.data.ves.VesEvent; import org.springframework.http.ResponseEntity; public interface RanVesBrokerService { - ResponseEntity<String> startSendingVesEvents(String identifier, Event vesEvent, Integer interval); + ResponseEntity<String> startSendingVesEvents(String identifier, VesEvent vesEvent, Integer interval, ReportingMethodEnum reportingMethods); - Optional<RanPeriodicVesEvent> stopSendingVesEvents(String identifier); + Optional<RanPeriodicEvent> stopSendingVesEvents(String identifier); - Map<String, RanPeriodicVesEvent> getPeriodicEventsCache(); + Map<String, RanPeriodicEvent> getPeriodicEventsCache(); Collection<String> getEnabledEventElementIdentifiers(); - Event getEventStructure(String identifier); + VesEvent getEventStructure(String identifier); - Event startSendingFailureVesEvents(String identifier); + VesEvent startSendingFailureVesEvents(String identifier, ReportingMethodEnum reportingMethods); - Event getGlobalPmVesStructure(); + VesEvent getGlobalPmVesStructure(); - void setGlobalPmVesStructure(Event event); + void setGlobalPmVesStructure(VesEvent event); Integer getGlobalVesInterval(); diff --git a/src/main/java/org/onap/a1pesimulator/service/ves/RanVesBrokerServiceImpl.java b/src/main/java/org/onap/a1pesimulator/service/ves/RanVesBrokerServiceImpl.java index 861bd36..4417212 100644 --- a/src/main/java/org/onap/a1pesimulator/service/ves/RanVesBrokerServiceImpl.java +++ b/src/main/java/org/onap/a1pesimulator/service/ves/RanVesBrokerServiceImpl.java @@ -17,9 +17,11 @@ import java.util.Collection; import java.util.Collections; import java.util.Map; import java.util.Optional; -import org.onap.a1pesimulator.data.ves.Event; + +import org.onap.a1pesimulator.data.ReportingMethodEnum; +import org.onap.a1pesimulator.data.fileready.RanPeriodicEvent; +import org.onap.a1pesimulator.data.ves.VesEvent; import org.onap.a1pesimulator.data.ves.MeasurementFields.AdditionalMeasurement; -import org.onap.a1pesimulator.data.ves.RanPeriodicVesEvent; import org.onap.a1pesimulator.util.Constants; import org.springframework.http.ResponseEntity; import org.springframework.stereotype.Service; @@ -37,31 +39,29 @@ public class RanVesBrokerServiceImpl implements RanVesBrokerService { } @Override - public Map<String, RanPeriodicVesEvent> getPeriodicEventsCache() { + public Map<String, RanPeriodicEvent> getPeriodicEventsCache() { return vesHolder.getPeriodicEventsCache(); } @Override - public ResponseEntity<String> startSendingVesEvents(String identifier, Event vesEvent, Integer interval) { - + public ResponseEntity<String> startSendingVesEvents(String identifier, VesEvent vesEvent, Integer interval, ReportingMethodEnum reportingMethod) { enrichWithIdentifier(identifier, vesEvent); - vesHolder.startSendingVesEvents(identifier, vesEvent, interval); - - return ResponseEntity.accepted().body("VES Event sending started"); + ResponseEntity<String> response = vesHolder.startSendingVesEvents(identifier, vesEvent, interval, reportingMethod); + return ResponseEntity.accepted().body(response.getBody()); } @Override - public Event startSendingFailureVesEvents(String identifier) { + public VesEvent startSendingFailureVesEvents(String identifier, ReportingMethodEnum reportingMethod) { - Event vesEvent = vesDataProvider.getFailurePmVesEvent(); + var vesEvent = vesDataProvider.getFailurePmVesEvent(); enrichWithIdentifier(identifier, vesEvent); - vesHolder.startSendingFailureVesEvents(identifier, vesEvent); + vesHolder.startSendingFailureVesEvents(identifier, vesEvent, reportingMethod); return vesEvent; } @Override - public Optional<RanPeriodicVesEvent> stopSendingVesEvents(String identifier) { + public Optional<RanPeriodicEvent> stopSendingVesEvents(String identifier) { return vesHolder.stopSendingVesEvents(identifier); } @@ -71,17 +71,17 @@ public class RanVesBrokerServiceImpl implements RanVesBrokerService { } @Override - public Event getEventStructure(String identifier) { + public VesEvent getEventStructure(String identifier) { return vesHolder.getEventStructure(identifier); } @Override - public Event getGlobalPmVesStructure() { + public VesEvent getGlobalPmVesStructure() { return vesDataProvider.getPmVesEvent(); } @Override - public void setGlobalPmVesStructure(Event event) { + public void setGlobalPmVesStructure(VesEvent event) { vesDataProvider.setPmVesEvent(event); } @@ -95,20 +95,20 @@ public class RanVesBrokerServiceImpl implements RanVesBrokerService { vesDataProvider.setInterval(interval); } - private void enrichWithIdentifier(String identifier, Event event) { + private void enrichWithIdentifier(String identifier, VesEvent event) { if (event.getMeasurementFields() == null || event.getMeasurementFields().getAdditionalMeasurements() == null) { return; } Collection<AdditionalMeasurement> additionalMeasurements = event.getMeasurementFields().getAdditionalMeasurements(); Optional<AdditionalMeasurement> identityOpt = additionalMeasurements.stream() - .filter(m -> Constants.MEASUREMENT_FIELD_IDENTIFIER - .equalsIgnoreCase(m.getName())) - .findAny(); + .filter(m -> Constants.MEASUREMENT_FIELD_IDENTIFIER + .equalsIgnoreCase(m.getName())) + .findAny(); if (identityOpt.isPresent()) { identityOpt.get().getHashMap().put(Constants.MEASUREMENT_FIELD_IDENTIFIER, identifier); } else { - AdditionalMeasurement measurement = new AdditionalMeasurement(); + var measurement = new AdditionalMeasurement(); measurement.setName(Constants.MEASUREMENT_FIELD_IDENTIFIER); measurement.setHashMap(Collections.singletonMap(Constants.MEASUREMENT_FIELD_VALUE, identifier)); additionalMeasurements.add(measurement); diff --git a/src/main/java/org/onap/a1pesimulator/service/ves/RanVesDataProvider.java b/src/main/java/org/onap/a1pesimulator/service/ves/RanVesDataProvider.java index 95743f3..9a9a2f6 100644 --- a/src/main/java/org/onap/a1pesimulator/service/ves/RanVesDataProvider.java +++ b/src/main/java/org/onap/a1pesimulator/service/ves/RanVesDataProvider.java @@ -15,8 +15,8 @@ package org.onap.a1pesimulator.service.ves; import java.io.IOException; import java.net.URL; -import lombok.Setter; -import org.onap.a1pesimulator.data.ves.Event; + +import org.onap.a1pesimulator.data.ves.VesEvent; import org.onap.a1pesimulator.util.JsonUtils; import org.springframework.beans.factory.annotation.Value; import org.springframework.cache.annotation.Cacheable; @@ -24,6 +24,8 @@ import org.springframework.core.io.Resource; import org.springframework.core.io.ResourceLoader; import org.springframework.stereotype.Service; +import lombok.Setter; + @Service public class RanVesDataProvider { @@ -31,9 +33,9 @@ public class RanVesDataProvider { private static final String PM_FAILURE_VES_LOCATION = "classpath:failurePmVes.json"; @Setter - private Event pmVesEvent; + private VesEvent pmVesEvent; @Setter - private Event failurePmVesEvent; + private VesEvent failurePmVesEvent; @Setter private Integer interval; @@ -46,15 +48,15 @@ public class RanVesDataProvider { } @Cacheable("pmVes") - public Event loadPmVesEvent() { + public VesEvent loadPmVesEvent() { URL resourceUrl = getResourceURL(resourceLoader.getResource(PM_VES_LOCATION)); - return JsonUtils.INSTANCE.deserializeFromFileUrl(resourceUrl, Event.class); + return JsonUtils.INSTANCE.deserializeFromFileUrl(resourceUrl, VesEvent.class); } @Cacheable("failurePmVes") - public Event loadFailurePmVesEvent() { + public VesEvent loadFailurePmVesEvent() { URL resourceUrl = getResourceURL(resourceLoader.getResource(PM_FAILURE_VES_LOCATION)); - return JsonUtils.INSTANCE.deserializeFromFileUrl(resourceUrl, Event.class); + return JsonUtils.INSTANCE.deserializeFromFileUrl(resourceUrl, VesEvent.class); } public Integer getRegularVesInterval() { @@ -68,14 +70,14 @@ public class RanVesDataProvider { return defaultInterval; } - public Event getPmVesEvent() { + public VesEvent getPmVesEvent() { if (pmVesEvent == null) { return loadPmVesEvent(); } return pmVesEvent; } - public Event getFailurePmVesEvent() { + public VesEvent getFailurePmVesEvent() { if (failurePmVesEvent == null) { return loadFailurePmVesEvent(); } diff --git a/src/main/java/org/onap/a1pesimulator/service/ves/RanVesHolder.java b/src/main/java/org/onap/a1pesimulator/service/ves/RanVesHolder.java index d53d8dd..f711347 100644 --- a/src/main/java/org/onap/a1pesimulator/service/ves/RanVesHolder.java +++ b/src/main/java/org/onap/a1pesimulator/service/ves/RanVesHolder.java @@ -13,6 +13,9 @@ package org.onap.a1pesimulator.service.ves; +import static java.util.Objects.isNull; +import static java.util.Objects.nonNull; + import java.text.MessageFormat; import java.util.Collection; import java.util.Map; @@ -20,61 +23,122 @@ import java.util.Optional; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ScheduledFuture; import java.util.function.BiFunction; -import org.onap.a1pesimulator.data.ves.Event; -import org.onap.a1pesimulator.data.ves.RanPeriodicVesEvent; + +import org.onap.a1pesimulator.data.ReportingMethodEnum; +import org.onap.a1pesimulator.data.RequestParameters; +import org.onap.a1pesimulator.data.fileready.RanPeriodicEvent; +import org.onap.a1pesimulator.data.fileready.RanPeriodicSendReport; +import org.onap.a1pesimulator.data.ves.VesEvent; +import org.onap.a1pesimulator.service.common.AbstractRanRunnable; +import org.onap.a1pesimulator.service.common.EventCustomizer; +import org.onap.a1pesimulator.service.fileready.RanFileReadyHolder; +import org.onap.a1pesimulator.service.fileready.RanSaveFileReadyRunnable; +import org.onap.a1pesimulator.service.fileready.RanSendReportsRunnable; import org.onap.a1pesimulator.service.ves.RanEventCustomizerFactory.Mode; -import org.onap.a1pesimulator.service.ves.RanSendVesRunnable.EventCustomizer; +import org.onap.a1pesimulator.util.VnfConfigReader; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; import org.springframework.http.ResponseEntity; import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler; import org.springframework.stereotype.Service; +import lombok.Getter; + @Service public class RanVesHolder { - private final Map<String, RanPeriodicVesEvent> periodicEventsCache = new ConcurrentHashMap<>(); + private static final Logger log = LoggerFactory.getLogger(RanVesHolder.class); + private final Map<String, RanPeriodicEvent> periodicEventsCache = new ConcurrentHashMap<>(); private final RanVesDataProvider vesDataProvider; private final RanEventCustomizerFactory eventCustomizerFactory; private final ThreadPoolTaskScheduler vesPmThreadPoolTaskScheduler; private final Collection<OnEventAction> onEventActions; + private final RanFileReadyHolder ranFileReadyHolder; private final RanVesSender vesSender; + private final VnfConfigReader vnfConfigReader; + private ThreadSendReportFunction threadSendReportFunction; - public RanVesHolder(ThreadPoolTaskScheduler vesPmThreadPoolTaskScheduler, RanVesSender vesSender, + public RanVesHolder(ThreadPoolTaskScheduler vesPmThreadPoolTaskScheduler, RanFileReadyHolder ranFileReadyHolder, RanVesSender vesSender, + VnfConfigReader vnfConfigReader, RanEventCustomizerFactory eventCustomizerFactory, RanVesDataProvider vesDataProvider, Collection<OnEventAction> onEventActions) { this.vesPmThreadPoolTaskScheduler = vesPmThreadPoolTaskScheduler; + this.ranFileReadyHolder = ranFileReadyHolder; this.vesSender = vesSender; + this.vnfConfigReader = vnfConfigReader; this.eventCustomizerFactory = eventCustomizerFactory; this.vesDataProvider = vesDataProvider; this.onEventActions = onEventActions; } - Map<String, RanPeriodicVesEvent> getPeriodicEventsCache() { + /** + * Thread for periodical sending of PM Bulk Files and fileReady Events + */ + private void startSendingReports() { + if (isNull(threadSendReportFunction) || !threadSendReportFunction.isProcessRunning()) { + int repPeriod = vnfConfigReader.getVnfConfig().getRepPeriod(); + threadSendReportFunction = new ThreadSendReportFunction(vesPmThreadPoolTaskScheduler, repPeriod, ranFileReadyHolder); + threadSendReportFunction.startEvent(); + log.info("Start sending reports every {} seconds", repPeriod); + } + } + + /** + * Stops sending the report after the last cell was stopped. It send the last report before stop completely + */ + private void stopSendingReports() { + if (nonNull(threadSendReportFunction) && !isAnyEventRunning()) { + threadSendReportFunction.getRanPeriodicVesEvent().getScheduledFuture().cancel(false); + sendLastReportAfterCancel(); + log.info("Stop sending reports every {} seconds", vnfConfigReader.getVnfConfig().getRepPeriod()); + } + } + + /** + * Sends the last report after all threads were stopped + */ + private void sendLastReportAfterCancel() { + log.trace("Send last report after report thread was canceled"); + ranFileReadyHolder.createPMBulkFileAndSendFileReadyMessage(); + } + + Map<String, RanPeriodicEvent> getPeriodicEventsCache() { return periodicEventsCache; } - ResponseEntity<String> startSendingVesEvents(String identifier, Event vesEvent, Integer interval) { + ResponseEntity<String> startSendingVesEvents(String identifier, VesEvent vesEvent, Integer interval, ReportingMethodEnum reportingMethod) { periodicEventsCache.compute(identifier, - new ThreadCacheUpdateFunction(vesPmThreadPoolTaskScheduler, vesEvent, interval, - eventCustomizerFactory.getEventCustomizer(vesEvent, Mode.REGULAR), onEventActions, vesSender)); + new ThreadCacheUpdateFunction(vesPmThreadPoolTaskScheduler, eventCustomizerFactory.getEventCustomizer(vesEvent, Mode.REGULAR), onEventActions, + ranFileReadyHolder, vesSender, RequestParameters.builder() + .vesEvent(vesEvent).identifier(identifier).reportingMethod(reportingMethod).interval(interval).build())); + if (ReportingMethodEnum.FILE_READY.equals(reportingMethod)) { + startSendingReports(); + } return ResponseEntity.accepted().body("VES Event sending started"); } - ResponseEntity<String> startSendingFailureVesEvents(String identifier, Event vesEvent) { + ResponseEntity<String> startSendingFailureVesEvents(String identifier, VesEvent vesEvent, ReportingMethodEnum reportingMethod) { - periodicEventsCache.compute(identifier, new ThreadCacheUpdateFunction(vesPmThreadPoolTaskScheduler, vesEvent, - vesDataProvider.getFailureVesInterval(), - eventCustomizerFactory.getEventCustomizer(vesEvent, Mode.FAILURE), onEventActions, vesSender)); + periodicEventsCache.compute(identifier, + new ThreadCacheUpdateFunction(vesPmThreadPoolTaskScheduler, eventCustomizerFactory.getEventCustomizer(vesEvent, Mode.FAILURE), onEventActions, + ranFileReadyHolder, + vesSender, RequestParameters.builder().vesEvent(vesEvent).identifier(identifier).interval(vesDataProvider.getFailureVesInterval()) + .reportingMethod(reportingMethod).build())); + if (ReportingMethodEnum.FILE_READY.equals(reportingMethod)) { + startSendingReports(); + } return ResponseEntity.accepted().body("Failure VES Event sending started"); } - Optional<RanPeriodicVesEvent> stopSendingVesEvents(String identifier) { - RanPeriodicVesEvent periodicEvent = periodicEventsCache.remove(identifier); + Optional<RanPeriodicEvent> stopSendingVesEvents(String identifier) { + RanPeriodicEvent periodicEvent = periodicEventsCache.remove(identifier); if (periodicEvent == null) { return Optional.empty(); } periodicEvent.getScheduledFuture().cancel(false); + stopSendingReports(); return Optional.of(periodicEvent); } @@ -86,7 +150,11 @@ public class RanVesHolder { return periodicEventsCache.containsKey(identifier); } - Event getEventStructure(String identifier) { + public boolean isAnyEventRunning() { + return !periodicEventsCache.isEmpty(); + } + + VesEvent getEventStructure(String identifier) { if (!periodicEventsCache.containsKey(identifier)) { throw new IllegalArgumentException( MessageFormat.format("Cannot find event for given source {0}", identifier)); @@ -95,39 +163,76 @@ public class RanVesHolder { } private static class ThreadCacheUpdateFunction - implements BiFunction<String, RanPeriodicVesEvent, RanPeriodicVesEvent> { + implements BiFunction<String, RanPeriodicEvent, RanPeriodicEvent> { private final Integer interval; private final ThreadPoolTaskScheduler vesPmThreadPoolTaskScheduler; - private final Event vesEvent; + private final VesEvent vesEvent; private final EventCustomizer eventCustomizer; private final Collection<OnEventAction> onEventActions; + private final RanFileReadyHolder fileReadyHolder; private final RanVesSender vesSender; + private final String cellId; + private final ReportingMethodEnum reportingMethod; - public ThreadCacheUpdateFunction(ThreadPoolTaskScheduler vesPmThreadPoolTaskScheduler, Event vesEvent, - Integer interval, EventCustomizer eventCustomizer, Collection<OnEventAction> onEventActions, - RanVesSender vesSender) { + public ThreadCacheUpdateFunction(ThreadPoolTaskScheduler vesPmThreadPoolTaskScheduler, EventCustomizer eventCustomizer, + Collection<OnEventAction> onEventActions, + RanFileReadyHolder fileReadyHolder, RanVesSender vesSender, RequestParameters requestParameters) { this.vesPmThreadPoolTaskScheduler = vesPmThreadPoolTaskScheduler; - this.vesEvent = vesEvent; - this.interval = interval; + this.vesEvent = requestParameters.getVesEvent(); + this.interval = requestParameters.getInterval(); this.eventCustomizer = eventCustomizer; this.onEventActions = onEventActions; + this.fileReadyHolder = fileReadyHolder; this.vesSender = vesSender; + this.cellId = requestParameters.getIdentifier(); + this.reportingMethod = requestParameters.getReportingMethod(); } @Override - public RanPeriodicVesEvent apply(String key, RanPeriodicVesEvent value) { + public RanPeriodicEvent apply(String key, RanPeriodicEvent value) { if (value != null) { // if thread is registered then cancel it and schedule a new one value.getScheduledFuture().cancel(false); } - RanSendVesRunnable sendVesRunnable = + AbstractRanRunnable ranRunnable = (ReportingMethodEnum.FILE_READY.equals(reportingMethod)) ? + new RanSaveFileReadyRunnable(fileReadyHolder, cellId, vesEvent, eventCustomizer, interval, onEventActions) : new RanSendVesRunnable(vesSender, vesEvent, eventCustomizer, onEventActions); + ScheduledFuture<?> scheduledFuture = - vesPmThreadPoolTaskScheduler.scheduleAtFixedRate(sendVesRunnable, interval * 1000L); - return RanPeriodicVesEvent.builder().event(vesEvent).interval(interval).scheduledFuture(scheduledFuture) - .sendVesRunnable(sendVesRunnable).build(); + vesPmThreadPoolTaskScheduler.scheduleAtFixedRate(ranRunnable, interval * 1000L); + return RanPeriodicEvent.builder().event(vesEvent).interval(interval).scheduledFuture(scheduledFuture) + .ranRunnable(ranRunnable).build(); } } + + @Getter + private static class ThreadSendReportFunction { + + protected final Integer interval; + protected final ThreadPoolTaskScheduler vesPmThreadPoolTaskScheduler; + protected RanPeriodicSendReport ranPeriodicVesEvent; + protected ScheduledFuture<?> scheduledFuture; + protected final RanFileReadyHolder ranFileReadyHolder; + + public ThreadSendReportFunction(ThreadPoolTaskScheduler vesPmThreadPoolTaskScheduler, Integer interval, RanFileReadyHolder ranFileReadyHolder) { + this.vesPmThreadPoolTaskScheduler = vesPmThreadPoolTaskScheduler; + this.interval = interval; + this.ranFileReadyHolder = ranFileReadyHolder; + } + + public void startEvent() { + RanSendReportsRunnable ranSendReportsRunnable = + new RanSendReportsRunnable(ranFileReadyHolder); + scheduledFuture = vesPmThreadPoolTaskScheduler.scheduleAtFixedRate(ranSendReportsRunnable, interval * 1000L); + this.ranPeriodicVesEvent = RanPeriodicSendReport.builder().interval(interval).scheduledFuture(scheduledFuture) + .ranSendReportsRunnable(ranSendReportsRunnable).build(); + } + + public boolean isProcessRunning() { + return (nonNull(scheduledFuture) && !(scheduledFuture.isCancelled() || scheduledFuture.isDone())); + } + } + } diff --git a/src/main/java/org/onap/a1pesimulator/service/ves/RanVesSender.java b/src/main/java/org/onap/a1pesimulator/service/ves/RanVesSender.java index 9c50197..85bccbb 100644 --- a/src/main/java/org/onap/a1pesimulator/service/ves/RanVesSender.java +++ b/src/main/java/org/onap/a1pesimulator/service/ves/RanVesSender.java @@ -13,9 +13,11 @@ package org.onap.a1pesimulator.service.ves; +import static java.util.Objects.nonNull; + +import org.onap.a1pesimulator.data.Event; import org.onap.a1pesimulator.data.VnfConfig; import org.onap.a1pesimulator.data.ves.CommonEventHeader; -import org.onap.a1pesimulator.data.ves.Event; import org.onap.a1pesimulator.exception.VesBrokerException; import org.onap.a1pesimulator.util.JsonUtils; import org.onap.a1pesimulator.util.VnfConfigReader; @@ -31,6 +33,8 @@ import org.springframework.http.ResponseEntity; import org.springframework.stereotype.Service; import org.springframework.web.client.RestTemplate; +import reactor.core.publisher.Mono; + @Service public class RanVesSender { @@ -53,30 +57,33 @@ public class RanVesSender { this.vesCollectorPath = vesCollectorPath; } - public ResponseEntity<String> send(Event vesEvent) throws VesBrokerException { - VnfConfig vnfConfig = vnfConfigReader.getVnfConfig(); - String url = getVesCollectorUrl(vnfConfig); - HttpHeaders headers = new HttpHeaders(); - headers.setContentType(MediaType.APPLICATION_JSON); - headers.setBasicAuth(vnfConfig.getVesUser(), vnfConfig.getVesPassword()); + public Mono<HttpStatus> send(Event event) { + if (nonNull(event)) { + VnfConfig vnfConfig = vnfConfigReader.getVnfConfig(); + String url = getVesCollectorUrl(vnfConfig); + HttpHeaders headers = new HttpHeaders(); + headers.setContentType(MediaType.APPLICATION_JSON); + headers.setBasicAuth(vnfConfig.getVesUser(), vnfConfig.getVesPassword()); - setVnfInfo(vesEvent, vnfConfig); - String event = JsonUtils.INSTANCE.objectToPrettyString(vesEvent); + setVnfInfo(event, vnfConfig); + String eventInJson = JsonUtils.INSTANCE.objectToPrettyString(event); - log.info("Sending following VES event: {}", event); + log.trace("Sending following event: {} ", eventInJson); - HttpEntity<String> entity = new HttpEntity<>(event, headers); - ResponseEntity<String> response = restTemplate.exchange(url, HttpMethod.POST, entity, String.class); + HttpEntity<String> entity = new HttpEntity<>(eventInJson, headers); + ResponseEntity<String> response = restTemplate.exchange(url, HttpMethod.POST, entity, String.class); - log.debug("Response received: {}", response); + log.debug("Response received: {}", response); - if (response.getStatusCode() == HttpStatus.OK || response.getStatusCode() == HttpStatus.ACCEPTED) { - return response; - } else { - String errorMsg = - "Failed to send VES event to the collector with response status code:" + response.getStatusCode(); - throw new VesBrokerException(errorMsg); + if (response.getStatusCode() == HttpStatus.OK || response.getStatusCode() == HttpStatus.ACCEPTED) { + return Mono.just(response.getStatusCode()); + } else { + String errorMsg = + "Failed to send VES event to the collector with response status code:" + response.getStatusCode(); + return Mono.error(new VesBrokerException(errorMsg)); + } } + return Mono.error(new VesBrokerException("There is no event to send to the collector.")); } private String getVesCollectorUrl(VnfConfig vnfConfig) { @@ -89,4 +96,5 @@ public class RanVesSender { header.setSourceName(vnfConfig.getVnfName()); vesEvent.setCommonEventHeader(header); } + } diff --git a/src/main/java/org/onap/a1pesimulator/util/Constants.java b/src/main/java/org/onap/a1pesimulator/util/Constants.java index 1964efd..2af00b6 100644 --- a/src/main/java/org/onap/a1pesimulator/util/Constants.java +++ b/src/main/java/org/onap/a1pesimulator/util/Constants.java @@ -20,4 +20,8 @@ public class Constants { public static final String MEASUREMENT_FIELD_IDENTIFIER = "identifier"; public static final String MEASUREMENT_FIELD_VALUE = "value"; + public static final String FILE_READY_CHANGE_IDENTIFIER = "PM_MEAS_FILES"; + public static final String FILE_READY_CHANGE_TYPE = "FileReady"; + public static final String EMPTY_STRING = ""; + public static final String TEMP_DIR = System.getProperty("java.io.tmpdir"); } diff --git a/src/main/java/org/onap/a1pesimulator/util/Convertors.java b/src/main/java/org/onap/a1pesimulator/util/Convertors.java new file mode 100644 index 0000000..11e703f --- /dev/null +++ b/src/main/java/org/onap/a1pesimulator/util/Convertors.java @@ -0,0 +1,33 @@ +package org.onap.a1pesimulator.util; + +import java.time.ZonedDateTime; +import java.time.format.DateTimeFormatter; + +import lombok.experimental.UtilityClass; + +@UtilityClass +public class Convertors { + + public static final String YYYYMMDD_PATTERN = "yyyyMMdd"; + public static final String ISO_8601_DATE = "yyyy-MM-dd'T'HH:mm:ssZ"; + + public static String zonedDateTimeToString(ZonedDateTime localDateTime) { + DateTimeFormatter formatter = DateTimeFormatter.ofPattern(YYYYMMDD_PATTERN); + return zonedDateTimeToString(localDateTime, formatter); + } + + public static String zonedDateTimeToString(ZonedDateTime localDateTime, String pattern) { + DateTimeFormatter formatter = DateTimeFormatter.ofPattern(pattern); + return zonedDateTimeToString(localDateTime, formatter); + } + + public static String zonedDateTimeToString(ZonedDateTime localDateTime, DateTimeFormatter formatter) { + return localDateTime.format(formatter); + } + + public static ZonedDateTime truncateToSpecifiedMinutes(ZonedDateTime zonedDateTime, Integer minutes) { + int minute = zonedDateTime.getMinute(); + int remainder = minute % minutes; + return remainder != 0 ? zonedDateTime.withMinute(minute - remainder) : zonedDateTime; + } +} diff --git a/src/main/java/org/onap/a1pesimulator/util/RanVesUtils.java b/src/main/java/org/onap/a1pesimulator/util/RanVesUtils.java index d5f3e67..a0f4991 100644 --- a/src/main/java/org/onap/a1pesimulator/util/RanVesUtils.java +++ b/src/main/java/org/onap/a1pesimulator/util/RanVesUtils.java @@ -23,9 +23,10 @@ import java.util.Map; import java.util.Random; import java.util.function.UnaryOperator; import java.util.stream.Collectors; + +import org.onap.a1pesimulator.data.Event; import org.onap.a1pesimulator.data.ue.UserEquipment; import org.onap.a1pesimulator.data.ves.CommonEventHeader; -import org.onap.a1pesimulator.data.ves.Event; import org.onap.a1pesimulator.data.ves.MeasurementFields.AdditionalMeasurement; import org.onap.a1pesimulator.service.ue.RanUeHolder; @@ -39,7 +40,7 @@ public class RanVesUtils { private static final String PATTERN_MARKER_START = "\\[\\["; private static final String PATTERN_MARKER_END = "\\]\\]"; - private static final String UE_PARAM_TRAFFIC_MODEL = "trafficModel"; + public static final String UE_PARAM_TRAFFIC_MODEL = "trafficModel"; private static final int TEN_MINUTES_MICROSECONDS = 10 * 60 * 1000_000; private static final Random random = new Random(); |