pmproducer/src/main/java/org/oran/pmproducer/repository/Job.java

   1 /*-
   2  * ========================LICENSE_START=================================
   3  * O-RAN-SC
   4  * %%
   5  * Copyright (C) 2023 Nordix Foundation
   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  *
  11  *      http://www.apache.org/licenses/LICENSE-2.0
  12  *
  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===================================
  19  */
  20
  21 package org.oran.pmproducer.repository;
  22
  23 import com.fasterxml.jackson.annotation.JsonProperty;
  24
  25 import io.swagger.v3.oas.annotations.media.Schema;
  26
  27 import java.lang.invoke.MethodHandles;
  28
  29 import lombok.Builder;
  30 import lombok.EqualsAndHashCode;
  31 import lombok.Getter;
  32 import lombok.ToString;
  33
  34 import org.oran.pmproducer.configuration.ApplicationConfig;
  35 import org.oran.pmproducer.filter.FilterFactory;
  36 import org.oran.pmproducer.filter.FilteredData;
  37 import org.oran.pmproducer.filter.PmReportFilter;
  38 import org.oran.pmproducer.tasks.TopicListener.DataFromTopic;
  39 import org.slf4j.Logger;
  40 import org.slf4j.LoggerFactory;
  41
  42 @ToString
  43 public class Job {
  44     private static final Logger logger = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
  45
  46     @Builder
  47     @Getter
  48     @Schema(name = "job_statistics", description = "Statistics information for one job")
  49     public static class Statistics {
  50
  51         @JsonProperty(value = "jobId", required = true)
  52         String jobId;
  53
  54         @JsonProperty(value = "typeId", required = true)
  55         String typeId;
  56
  57         @JsonProperty(value = "inputTopic", required = false)
  58         String inputTopic;
  59
  60         @JsonProperty(value = "outputTopic", required = false)
  61         String outputTopic;
  62
  63         @JsonProperty(value = "groupId", required = false)
  64         String groupId;
  65
  66         @JsonProperty(value = "clientId", required = false)
  67         String clientId;
  68
  69         @JsonProperty(value = "noOfReceivedObjects", required = true)
  70         @Builder.Default
  71         long noOfReceivedObjects = 0;
  72
  73         @JsonProperty(value = "noOfReceivedBytes", required = true)
  74         @Builder.Default
  75         long noOfReceivedBytes = 0;
  76
  77         @JsonProperty(value = "noOfSentObjects", required = true)
  78         @Builder.Default
  79         long noOfSentObjects = 0;
  80
  81         @JsonProperty(value = "noOfSentBytes", required = true)
  82         @Builder.Default
  83         long noOfSentBytes = 0;
  84
  85         public void received(byte[] bytes) {
  86             noOfReceivedBytes += bytes.length;
  87             noOfReceivedObjects += 1;
  88
  89         }
  90
  91         public void filtered(byte[] bytes) {
  92             noOfSentBytes += bytes.length;
  93             noOfSentObjects += 1;
  94         }
  95
  96     }
  97
  98     @Builder
  99     public static class Parameters {
 100
 101         @Getter
 102         private PmReportFilter.FilterData filter;
 103
 104         @Builder
 105         @EqualsAndHashCode
 106         public static class KafkaDeliveryInfo {
 107             @Getter
 108             private String topic;
 109
 110             @Getter
 111             private String bootStrapServers;
 112         }
 113
 114         @Getter
 115         private KafkaDeliveryInfo deliveryInfo;
 116     }
 117
 118     @Getter
 119     private final String id;
 120
 121     @Getter
 122     private final InfoType type;
 123
 124     @Getter
 125     private final String owner;
 126
 127     @Getter
 128     private final Parameters parameters;
 129
 130     @Getter
 131     private final String lastUpdated;
 132
 133     @Getter
 134     private final PmReportFilter filter;
 135
 136     @Getter
 137     private final Statistics statistics;
 138
 139     public Job(String id, InfoType type, String owner, String lastUpdated, Parameters parameters,
 140             ApplicationConfig appConfig) {
 141         this.id = id;
 142         this.type = type;
 143         this.owner = owner;
 144         this.lastUpdated = lastUpdated;
 145         this.parameters = parameters;
 146         filter = parameters.filter == null ? null : FilterFactory.create(parameters.getFilter());
 147
 148         statistics = Statistics.builder() //
 149                 .groupId(type.getKafkaGroupId()) //
 150                 .inputTopic(type.getKafkaInputTopic()) //
 151                 .jobId(id) //
 152                 .outputTopic(parameters.getDeliveryInfo() == null ? "" : parameters.getDeliveryInfo().topic) //
 153                 .typeId(type.getId()) //
 154                 .clientId(type.getKafkaClientId(appConfig)) //
 155                 .build();
 156
 157     }
 158
 159     public FilteredData filter(DataFromTopic data) {
 160         if (filter == null) {
 161             logger.debug("No filter used");
 162             return new FilteredData(data.infoTypeId, data.key, data.value);
 163         }
 164         return filter.filter(data);
 165     }
 166 }