2 * ========================LICENSE_START=================================
5 * Copyright (C) 2023 Nordix Foundation
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
11 * http://www.apache.org/licenses/LICENSE-2.0
13 * Unless required by applicable law or agreed to in writing, software
14 * distributed under the License is distributed on an "AS IS" BASIS,
15 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
16 * See the License for the specific language governing permissions and
17 * limitations under the License.
18 * ========================LICENSE_END===================================
21 package org.oran.pmproducer.tasks;
23 import java.io.ByteArrayOutputStream;
24 import java.io.IOException;
25 import java.time.OffsetDateTime;
26 import java.time.format.DateTimeFormatter;
27 import java.time.format.DateTimeFormatterBuilder;
28 import java.util.HashMap;
30 import java.util.zip.GZIPOutputStream;
34 import org.apache.commons.lang3.StringUtils;
35 import org.apache.kafka.clients.producer.ProducerConfig;
36 import org.apache.kafka.clients.producer.ProducerRecord;
37 import org.apache.kafka.common.serialization.ByteArraySerializer;
38 import org.oran.pmproducer.configuration.ApplicationConfig;
39 import org.oran.pmproducer.datastore.DataStore;
40 import org.oran.pmproducer.filter.FilteredData;
41 import org.oran.pmproducer.filter.PmReportFilter;
42 import org.oran.pmproducer.repository.Job.Parameters.KafkaDeliveryInfo;
43 import org.oran.pmproducer.repository.Jobs.JobGroup;
44 import org.oran.pmproducer.tasks.TopicListener.DataFromTopic;
45 import org.slf4j.Logger;
46 import org.slf4j.LoggerFactory;
48 import reactor.core.Disposable;
49 import reactor.core.publisher.Flux;
50 import reactor.core.publisher.Mono;
51 import reactor.kafka.sender.KafkaSender;
52 import reactor.kafka.sender.SenderOptions;
53 import reactor.kafka.sender.SenderRecord;
56 * The class streams data from a multi cast sink and sends the data to the Job
57 * owner via REST calls.
59 @SuppressWarnings("squid:S2629") // Invoke method(s) only conditionally
60 public class JobDataDistributor {
61 private static final Logger logger = LoggerFactory.getLogger(JobDataDistributor.class);
64 private final JobGroup jobGroup;
65 private Disposable subscription;
66 private final ErrorStats errorStats = new ErrorStats();
68 private final DataStore dataStore;
69 private static com.google.gson.Gson gson = new com.google.gson.GsonBuilder().disableHtmlEscaping().create();
70 private final ApplicationConfig applConfig;
72 private KafkaSender<byte[], byte[]> sender;
74 private class ErrorStats {
76 private int consumerFaultCounter = 0;
78 public void handleOkFromConsumer() {
79 this.consumerFaultCounter = 0;
82 @SuppressWarnings("java:S1172")
83 public void handleException(Throwable t) {
84 ++this.consumerFaultCounter;
88 public JobDataDistributor(JobGroup jobGroup, ApplicationConfig applConfig) {
89 this.applConfig = applConfig;
90 this.jobGroup = jobGroup;
91 this.dataStore = DataStore.create(applConfig);
92 this.dataStore.create(DataStore.Bucket.FILES).subscribe();
93 this.dataStore.create(DataStore.Bucket.LOCKS).subscribe();
95 SenderOptions<byte[], byte[]> senderOptions = senderOptions(applConfig, jobGroup.getDeliveryInfo());
96 this.sender = KafkaSender.create(senderOptions);
100 public void start(Flux<TopicListener.DataFromTopic> input) {
101 logger.debug("Starting distribution, to topic: {}", jobGroup.getId());
102 PmReportFilter filter = jobGroup.getFilter();
103 if (filter == null || filter.getFilterData().getPmRopEndTime() == null) {
104 this.subscription = filter(input, this.jobGroup) //
105 .flatMap(this::sendToClient) //
106 .onErrorResume(this::handleError) //
107 .subscribe(this::handleSentOk, //
108 this::handleExceptionInStream, //
109 () -> logger.warn("JobDataDistributor stopped jobId: {}", jobGroup.getId()));
112 if (filter != null && filter.getFilterData().getPmRopStartTime() != null) {
113 this.dataStore.createLock(collectHistoricalDataLockName()) //
114 .doOnNext(isLockGranted -> {
115 if (isLockGranted.booleanValue()) {
116 logger.debug("Checking historical PM ROP files, jobId: {}", this.jobGroup.getId());
118 logger.debug("Skipping check of historical PM ROP files, already done. jobId: {}",
119 this.jobGroup.getId());
122 .filter(isLockGranted -> isLockGranted) //
123 .flatMapMany(b -> Flux.fromIterable(filter.getFilterData().getSourceNames())) //
124 .doOnNext(sourceName -> logger.debug("Checking source name: {}, jobId: {}", sourceName,
125 this.jobGroup.getId())) //
126 .flatMap(sourceName -> dataStore.listObjects(DataStore.Bucket.FILES, sourceName), 1) //
127 .filter(this::isRopFile) //
128 .filter(fileName -> filterStartTime(filter.getFilterData(), fileName)) //
129 .filter(fileName -> filterEndTime(filter.getFilterData(), fileName)) //
130 .map(this::createFakeEvent) //
131 .flatMap(data -> TopicListener.getDataFromFileIfNewPmFileEvent(data, this.jobGroup.getType(),
133 .map(jobGroup::filter) //
135 .flatMap(this::sendToClient, 1) //
136 .onErrorResume(this::handleCollectHistoricalDataError) //
137 .doFinally(sig -> sendLastStoredRecord()) //
142 private static SenderOptions<byte[], byte[]> senderOptions(ApplicationConfig config,
143 KafkaDeliveryInfo deliveryInfo) {
145 String bootstrapServers = deliveryInfo.getBootStrapServers();
146 if (bootstrapServers == null || bootstrapServers.isEmpty()) {
147 bootstrapServers = config.getKafkaBootStrapServers();
150 Map<String, Object> props = new HashMap<>();
151 props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
152 props.put(ProducerConfig.ACKS_CONFIG, "all");
153 props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, ByteArraySerializer.class);
154 props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, ByteArraySerializer.class);
156 config.addKafkaSecurityProps(props);
158 return SenderOptions.create(props);
161 private void sendLastStoredRecord() {
163 FilteredData output = new FilteredData("", this.jobGroup.getType().getId(), null, data.getBytes());
165 sendToClient(output).subscribe();
168 private FilteredData gzip(FilteredData data) {
169 if (this.applConfig.isZipOutput()) {
171 ByteArrayOutputStream out = new ByteArrayOutputStream();
172 GZIPOutputStream gzip = new GZIPOutputStream(out);
173 gzip.write(data.value);
176 byte[] zipped = out.toByteArray();
177 return new FilteredData(data.getSourceName(), data.infoTypeId, data.key, zipped, true);
178 } catch (IOException e) {
179 logger.error("Unexpected exception when zipping: {}", e.getMessage());
187 private Mono<String> handleCollectHistoricalDataError(Throwable t) {
188 logger.error("Exception: {} job: {}", t.getMessage(), jobGroup.getId());
189 return tryDeleteLockFile() //
193 private String collectHistoricalDataLockName() {
194 return "collectHistoricalDataLock" + this.jobGroup.getId();
197 private DataFromTopic createFakeEvent(String fileName) {
198 NewFileEvent ev = new NewFileEvent(fileName);
199 return new DataFromTopic(this.jobGroup.getType().getId(), null, null, gson.toJson(ev).getBytes());
202 private static String fileTimePartFromRopFileName(String fileName) {
203 // "O-DU-1122/A20000626.2315+0200-2330+0200_HTTPS-6-73.json"
204 return fileName.substring(fileName.lastIndexOf("/") + 2);
207 private static boolean filterStartTime(PmReportFilter.FilterData filter, String fileName) {
209 OffsetDateTime fileStartTime = getStartTimeFromFileName(fileName);
210 OffsetDateTime startTime = OffsetDateTime.parse(filter.getPmRopStartTime());
211 boolean isMatch = fileStartTime.isAfter(startTime);
212 logger.debug("Checking file: {}, fileStartTime: {}, filterStartTime: {}, isAfter: {}", fileName,
213 fileStartTime, startTime, isMatch);
215 } catch (Exception e) {
216 logger.warn("Time parsing exception: {}", e.getMessage());
221 private boolean isRopFile(String fileName) {
222 return fileName.endsWith(".json") || fileName.endsWith(".json.gz");
225 private static boolean filterEndTime(PmReportFilter.FilterData filter, String fileName) {
226 if (filter.getPmRopEndTime() == null) {
230 OffsetDateTime fileEndTime = getEndTimeFromFileName(fileName);
231 OffsetDateTime endTime = OffsetDateTime.parse(filter.getPmRopEndTime());
232 boolean isMatch = fileEndTime.isBefore(endTime);
233 logger.debug("Checking file: {}, fileEndTime: {}, endTime: {}, isBefore: {}", fileName, fileEndTime,
237 } catch (Exception e) {
238 logger.warn("Time parsing exception: {}", e.getMessage());
243 private static OffsetDateTime getStartTimeFromFileName(String fileName) {
244 String fileTimePart = fileTimePartFromRopFileName(fileName);
245 // A20000626.2315+0200-2330+0200_HTTPS-6-73.json
246 fileTimePart = fileTimePart.substring(0, 18);
247 return parseFileDate(fileTimePart);
250 private static OffsetDateTime getEndTimeFromFileName(String fileName) {
251 String fileTimePart = fileTimePartFromRopFileName(fileName);
252 // A20000626.2315+0200-2330+0200_HTTPS-6-73.json
253 fileTimePart = fileTimePart.substring(0, 9) + fileTimePart.substring(19, 28);
254 return parseFileDate(fileTimePart);
257 private static OffsetDateTime parseFileDate(String timeStr) {
258 DateTimeFormatter startTimeFormatter =
259 new DateTimeFormatterBuilder().appendPattern("yyyyMMdd.HHmmZ").toFormatter();
260 return OffsetDateTime.parse(timeStr, startTimeFormatter);
263 private void handleExceptionInStream(Throwable t) {
264 logger.warn("JobDataDistributor exception: {}, jobId: {}", t.getMessage(), jobGroup.getId());
267 public Mono<String> sendToClient(FilteredData data) {
269 SenderRecord<byte[], byte[], Integer> senderRecord = senderRecord(data, this.getJobGroup().getDeliveryInfo());
271 logger.trace("Sending data '{}' to Kafka topic: {}", StringUtils.truncate(data.getValueAString(), 10),
272 this.getJobGroup().getDeliveryInfo());
274 return this.sender.send(Mono.just(senderRecord)) //
275 .doOnNext(n -> logger.debug("Sent data to Kafka topic: {}", this.getJobGroup().getDeliveryInfo())) //
276 .doOnError(t -> logger.warn("Failed to send to Kafka, job: {}, reason: {}", this.getJobGroup().getId(),
278 .onErrorResume(t -> Mono.empty()) //
284 public synchronized void stop() {
285 if (this.subscription != null) {
286 logger.debug("Stopped, job: {}", jobGroup.getId());
287 this.subscription.dispose();
288 this.subscription = null;
290 if (sender != null) {
295 tryDeleteLockFile().subscribe();
298 private Mono<Boolean> tryDeleteLockFile() {
299 return dataStore.deleteLock(collectHistoricalDataLockName()) //
300 .doOnNext(res -> logger.debug("Removed lockfile {} {}", collectHistoricalDataLockName(), res))
301 .onErrorResume(t -> Mono.just(false));
304 public synchronized boolean isRunning() {
305 return this.subscription != null;
308 private SenderRecord<byte[], byte[], Integer> senderRecord(FilteredData output, KafkaDeliveryInfo deliveryInfo) {
309 int correlationMetadata = 2;
311 new ProducerRecord<>(deliveryInfo.getTopic(), null, null, output.key, output.value, output.headers());
312 return SenderRecord.create(producerRecord, correlationMetadata);
315 private Flux<FilteredData> filter(Flux<DataFromTopic> inputFlux, JobGroup jobGroup) {
316 return inputFlux.doOnNext(data -> logger.trace("Received data, job {}", jobGroup.getId())) //
317 .doOnNext(data -> jobGroup.getJobs().forEach(job -> job.getStatistics().received(data.value))) //
318 .map(jobGroup::filter) //
319 .filter(f -> !f.isEmpty()) //
321 .doOnNext(f -> jobGroup.getJobs().forEach(job -> job.getStatistics().filtered(f.value))) //
322 .doOnNext(data -> logger.trace("Filtered data, job {}", jobGroup.getId())) //
326 private Mono<String> handleError(Throwable t) {
327 logger.warn("exception: {} job: {}", t.getMessage(), jobGroup.getId());
328 this.errorStats.handleException(t);
329 return Mono.empty(); // Ignore
332 private void handleSentOk(String data) {
333 this.errorStats.handleOkFromConsumer();