2 * ============LICENSE_START=======================================================
4 * ================================================================================
5 * Copyright © 2021 AT&T Intellectual Property. All rights reserved.
6 * ================================================================================
7 * Licensed under the Apache License, Version 2.0 (the "License");
8 * you may not use this file except in compliance with the License.
9 * You may obtain a copy of the License at
10 * http://www.apache.org/licenses/LICENSE-2.0
12 * Unless required by applicable law or agreed to in writing, software
13 * distributed under the License is distributed on an "AS IS" BASIS,
14 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
15 * See the License for the specific language governing permissions and
16 * limitations under the License.
17 * ============LICENSE_END=========================================================
20 package org.o.ran.oam.nf.oam.adopter.pm.rest.manager.mapper;
22 import com.fasterxml.jackson.core.JsonParser;
23 import com.fasterxml.jackson.dataformat.csv.CsvMapper;
24 import com.fasterxml.jackson.dataformat.csv.CsvSchema;
25 import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
26 import io.reactivex.rxjava3.core.Single;
27 import java.io.IOException;
28 import java.util.ArrayList;
29 import java.util.Iterator;
30 import java.util.List;
32 import java.util.zip.ZipEntry;
33 import java.util.zip.ZipInputStream;
34 import org.o.ran.oam.nf.oam.adopter.api.CommonEventFormat302ONAP;
35 import org.o.ran.oam.nf.oam.adopter.api.Event;
36 import org.o.ran.oam.nf.oam.adopter.pm.rest.manager.PerformanceManagementMapperConfigProvider;
37 import org.o.ran.oam.nf.oam.adopter.pm.rest.manager.pojos.CsvConfiguration;
38 import org.o.ran.oam.nf.oam.adopter.pm.rest.manager.pojos.VesMappingConfiguration;
39 import org.slf4j.Logger;
40 import org.slf4j.LoggerFactory;
41 import org.springframework.beans.factory.annotation.Autowired;
42 import org.springframework.stereotype.Service;
45 public class PerformanceManagementFile2VesMapper {
46 private static final Logger LOG = LoggerFactory.getLogger(PerformanceManagementFile2VesMapper.class);
48 private static final String CSV_EXTENSION = ".csv";
49 private final PerformanceManagementMapperConfigProvider pmConfigProvider;
50 private static final int THRESHOLD_SIZE = 1000000000; // 1 GB
51 private static final int THRESHOLD_RATIO = 10;
52 private static final int THRESHOLD_ENTRIES = 10000;
55 public PerformanceManagementFile2VesMapper(final PerformanceManagementMapperConfigProvider pmConfigProvider) {
56 this.pmConfigProvider = pmConfigProvider;
60 * Translate CSV in ZipInputStream format to list of CommonEventFormat302ONAP events.
62 * @param zipInputStream csv
63 * @param hostIp source Ip Address
64 * @return CommonEventFormat302ONAP events
66 @SuppressFBWarnings("REC_CATCH_EXCEPTION")
67 public Single<List<CommonEventFormat302ONAP>> map(final ZipInputStream zipInputStream, final String hostIp) {
68 LOG.info("Converting ZIP files to VES Message started");
69 final List<CommonEventFormat302ONAP> listOfNotifications = new ArrayList<>();
70 final CsvSchema schema = CsvSchema.emptySchema().withHeader();
71 final var mapper = new CsvMapper();
72 mapper.configure(JsonParser.Feature.AUTO_CLOSE_SOURCE, false);
75 final var mappingConfiguration = pmConfigProvider.getVesMappingConfiguration();
76 var totalSizeEntry = 0;
77 var totalEntryArchive = 0;
78 while ((entry = zipInputStream.getNextEntry()) != null) {
79 final var size = entry.getSize();
81 totalSizeEntry += size;
82 if (totalSizeEntry > THRESHOLD_SIZE || size == -1) {
83 throw new IllegalStateException("File to be unzipped too big.");
86 final long compressionRatio = totalSizeEntry / entry.getCompressedSize();
87 if (compressionRatio > THRESHOLD_RATIO) {
88 return Single.error(new Exception("Wrong file type, threshold to high."));
91 if (totalEntryArchive > THRESHOLD_ENTRIES) {
92 // too much entries in this archive, can lead to inodes exhaustion of the system
93 return Single.error(new Exception("Too many files"));
96 final String entryName = entry.getName();
97 if (!entryName.endsWith(CSV_EXTENSION)) {
98 return Single.error(new Exception("Wrong file type :" + entryName));
101 final Iterator<Map<String, String>> iterator =
102 mapper.readerFor(Map.class).with(schema).readValues(zipInputStream);
103 final List<List<Event>> mappedEvents = toEvent(mappingConfiguration, hostIp, iterator);
105 mappedEvents.forEach(mapped -> {
106 final var eventFormat = new CommonEventFormat302ONAP();
107 eventFormat.setEventList(mapped);
108 listOfNotifications.add(eventFormat);
111 } catch (final Exception e) {
112 return Single.error(new Exception("Failed to process file", e));
115 zipInputStream.closeEntry();
116 } catch (final IOException e) {
117 LOG.warn("Failed to close zip stream", e);
120 LOG.info("Converting ZIP files to VES Message finished");
121 return Single.just(listOfNotifications);
124 private static List<List<Event>> toEvent(final VesMappingConfiguration mappingConfiguration, final String hostIp,
125 final Iterator<Map<String, String>> iterator) {
126 final List<List<Event>> globalList = new ArrayList<>();
127 final int batchSize = mappingConfiguration.getBatchSize();
129 List<Event> events = new ArrayList<>();
130 final CsvConfiguration csv = mappingConfiguration.getCsv();
131 while (iterator.hasNext()) {
132 final var event = new Event();
133 final Map<String, String> recordMap = iterator.next();
134 event.setCommonEventHeader(CommonEventHeaderHandler.toCommonEventHeader(mappingConfiguration, hostIp, csv,
135 recordMap, sequence));
136 event.setMeasurementFields(MeasurementFieldsHandler.toMeasurementFields(mappingConfiguration, recordMap));
139 if (sequence % batchSize == 0) {
140 globalList.add(events);
141 events = new ArrayList<>();
144 if (!events.isEmpty()) {
145 globalList.add(events);