pmproducer/src/main/java/org/oran/pmproducer/tasks/ProducerRegstrationTask.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.tasks;
  22
  23 import com.google.common.io.CharStreams;
  24 import com.google.gson.JsonParser;
  25
  26 import java.io.IOException;
  27 import java.io.InputStream;
  28 import java.io.InputStreamReader;
  29 import java.nio.charset.StandardCharsets;
  30
  31 import lombok.Getter;
  32
  33 import org.oran.pmproducer.clients.AsyncRestClient;
  34 import org.oran.pmproducer.clients.AsyncRestClientFactory;
  35 import org.oran.pmproducer.clients.SecurityContext;
  36 import org.oran.pmproducer.configuration.ApplicationConfig;
  37 import org.oran.pmproducer.controllers.ProducerCallbacksController;
  38 import org.oran.pmproducer.exceptions.ServiceException;
  39 import org.oran.pmproducer.r1.ConsumerJobInfo;
  40 import org.oran.pmproducer.r1.ProducerInfoTypeInfo;
  41 import org.oran.pmproducer.r1.ProducerRegistrationInfo;
  42 import org.oran.pmproducer.repository.InfoType;
  43 import org.oran.pmproducer.repository.InfoTypes;
  44 import org.slf4j.Logger;
  45 import org.slf4j.LoggerFactory;
  46 import org.springframework.beans.factory.annotation.Autowired;
  47 import org.springframework.http.HttpStatus;
  48 import org.springframework.scheduling.annotation.EnableScheduling;
  49 import org.springframework.scheduling.annotation.Scheduled;
  50 import org.springframework.stereotype.Component;
  51
  52 import reactor.core.publisher.Flux;
  53 import reactor.core.publisher.Mono;
  54
  55 /**
  56  * Registers the types and this producer in Innformation Coordinator Service.
  57  * This is done when needed.
  58  */
  59 @Component
  60 @EnableScheduling
  61 @SuppressWarnings("squid:S2629") // Invoke method(s) only conditionally
  62 public class ProducerRegstrationTask {
  63
  64     private static final Logger logger = LoggerFactory.getLogger(ProducerRegstrationTask.class);
  65     private final AsyncRestClient restClient;
  66     private final ApplicationConfig applicationConfig;
  67     private final InfoTypes types;
  68     private static com.google.gson.Gson gson = new com.google.gson.GsonBuilder().disableHtmlEscaping().create();
  69
  70     @Getter
  71     private boolean isRegisteredInIcs = false;
  72     private static final int REGISTRATION_SUPERVISION_INTERVAL_MS = 1000 * 10;
  73
  74     public ProducerRegstrationTask(@Autowired ApplicationConfig applicationConfig, @Autowired InfoTypes types,
  75             @Autowired SecurityContext securityContext) {
  76         AsyncRestClientFactory restClientFactory =
  77                 new AsyncRestClientFactory(applicationConfig.getWebClientConfig(), securityContext);
  78         this.restClient = restClientFactory.createRestClientNoHttpProxy("");
  79         this.applicationConfig = applicationConfig;
  80         this.types = types;
  81     }
  82
  83     @Scheduled(fixedRate = REGISTRATION_SUPERVISION_INTERVAL_MS)
  84     public void runSupervisionTask() {
  85         supervisionTask().subscribe( //
  86                 null, //
  87                 this::handleRegistrationFailure, //
  88                 this::handleRegistrationCompleted);
  89     }
  90
  91     public Mono<String> supervisionTask() {
  92         return checkRegistration() //
  93                 .filter(isRegistrationOk -> !isRegistrationOk || !this.isRegisteredInIcs) //
  94                 .flatMap(isRegisterred -> registerTypesAndProducer());
  95     }
  96
  97     private void handleRegistrationCompleted() {
  98         isRegisteredInIcs = true;
  99     }
 100
 101     private void handleRegistrationFailure(Throwable t) {
 102         logger.warn("Registration of producer failed {}", t.getMessage());
 103     }
 104
 105     private String producerRegistrationUrl() {
 106         final String producerId = this.applicationConfig.getSelfUrl().replace("/", "_");
 107         return applicationConfig.getIcsBaseUrl() + "/data-producer/v1/info-producers/" + producerId;
 108     }
 109
 110     // Returns TRUE if registration is correct
 111     private Mono<Boolean> checkRegistration() {
 112         return restClient.get(producerRegistrationUrl()) //
 113                 .flatMap(this::isRegisterredInfoCorrect) //
 114                 .onErrorResume(t -> Mono.just(Boolean.FALSE));
 115     }
 116
 117     private Mono<Boolean> isRegisterredInfoCorrect(String registerredInfoStr) {
 118         ProducerRegistrationInfo registerredInfo = gson.fromJson(registerredInfoStr, ProducerRegistrationInfo.class);
 119         if (isEqual(producerRegistrationInfo(), registerredInfo)) {
 120             logger.trace("Already registered in ICS");
 121             return Mono.just(Boolean.TRUE);
 122         } else {
 123             return Mono.just(Boolean.FALSE);
 124         }
 125     }
 126
 127     private String registerTypeUrl(InfoType type) {
 128         return applicationConfig.getIcsBaseUrl() + "/data-producer/v1/info-types/" + type.getId();
 129     }
 130
 131     private Mono<String> registerTypesAndProducer() {
 132         final int CONCURRENCY = 1;
 133
 134         return Flux.fromIterable(this.types.getAll()) //
 135                 .doOnNext(type -> logger.info("Registering type {}", type.getId())) //
 136                 .flatMap(this::createInputDataJob, CONCURRENCY)
 137                 .flatMap(type -> restClient.put(registerTypeUrl(type), gson.toJson(typeRegistrationInfo(type))),
 138                         CONCURRENCY) //
 139                 .collectList() //
 140                 .doOnNext(type -> logger.info("Registering producer")) //
 141                 .flatMap(resp -> restClient.put(producerRegistrationUrl(), gson.toJson(producerRegistrationInfo())));
 142     }
 143
 144     private Mono<InfoType> createInputDataJob(InfoType type) {
 145         if (type.getInputJobType() == null) {
 146             return Mono.just(type);
 147         }
 148
 149         ConsumerJobInfo info =
 150                 new ConsumerJobInfo(type.getInputJobType(), type.getInputJobDefinition(), "pmproducer", "");
 151
 152         final String JOB_ID = type.getId() + "_5b3f4db6-3d9e-11ed-b878-0242ac120002";
 153         String body = gson.toJson(info);
 154
 155         return restClient.put(consumerJobUrl(JOB_ID), body)
 156                 .doOnError(t -> logger.error("Could not create job of type {}, reason: {}", type.getInputJobType(),
 157                         t.getMessage()))
 158                 .onErrorResume(t -> Mono.just("")) //
 159                 .doOnNext(n -> logger.info("Created input job: {}, type: {}", JOB_ID, type.getInputJobType())) //
 160                 .map(x -> type);
 161     }
 162
 163     private String consumerJobUrl(String jobId) {
 164         return applicationConfig.getIcsBaseUrl() + "/data-consumer/v1/info-jobs/" + jobId;
 165     }
 166
 167     private Object typeSpecifcInfoObject() {
 168         return jsonObject("{}");
 169     }
 170
 171     private ProducerInfoTypeInfo typeRegistrationInfo(InfoType type) {
 172         try {
 173             return new ProducerInfoTypeInfo(jsonSchemaObject(type), typeSpecifcInfoObject());
 174         } catch (Exception e) {
 175             logger.error("Fatal error {}", e.getMessage());
 176             return null;
 177         }
 178     }
 179
 180     private Object jsonSchemaObject(InfoType type) throws IOException, ServiceException {
 181         final String schemaFile = "/typeSchemaPmData.json";
 182         return jsonObject(readSchemaFile(schemaFile));
 183     }
 184
 185     private String readSchemaFile(String filePath) throws IOException, ServiceException {
 186         InputStream in = getClass().getResourceAsStream(filePath);
 187         logger.debug("Reading application schema file from: {} with: {}", filePath, in);
 188         if (in == null) {
 189             throw new ServiceException("Could not readfile: " + filePath, HttpStatus.INTERNAL_SERVER_ERROR);
 190         }
 191         return CharStreams.toString(new InputStreamReader(in, StandardCharsets.UTF_8));
 192     }
 193
 194     @SuppressWarnings("java:S2139") // Log exception
 195     private Object jsonObject(String json) {
 196         try {
 197             return JsonParser.parseString(json).getAsJsonObject();
 198         } catch (Exception e) {
 199             logger.error("Bug, error in JSON: {} {}", json, e.getMessage());
 200             throw new NullPointerException(e.getMessage());
 201         }
 202     }
 203
 204     private boolean isEqual(ProducerRegistrationInfo a, ProducerRegistrationInfo b) {
 205         return a.jobCallbackUrl.equals(b.jobCallbackUrl) //
 206                 && a.producerSupervisionCallbackUrl.equals(b.producerSupervisionCallbackUrl) //
 207                 && a.supportedTypeIds.size() == b.supportedTypeIds.size();
 208     }
 209
 210     private ProducerRegistrationInfo producerRegistrationInfo() {
 211         return ProducerRegistrationInfo.builder() //
 212                 .jobCallbackUrl(baseUrl() + ProducerCallbacksController.JOB_URL) //
 213                 .producerSupervisionCallbackUrl(baseUrl() + ProducerCallbacksController.SUPERVISION_URL) //
 214                 .supportedTypeIds(this.types.typeIds()) //
 215                 .build();
 216     }
 217
 218     private String baseUrl() {
 219         return this.applicationConfig.getSelfUrl();
 220     }
 221 }