models-sim/policy-models-simulators/src/main/java/org/onap/policy/models/simulators/Main.java

   1 /*-
   2  * ============LICENSE_START=======================================================
   3  * Copyright (C) 2020-2021 AT&T Intellectual Property. All rights reserved.
   4  * Modifications Copyright (C) 2020-2021 Bell Canada. All rights reserved.
   5  * Modifications Copyright 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  *
  19  * SPDX-License-Identifier: Apache-2.0
  20  * ============LICENSE_END=========================================================
  21  */
  22
  23 package org.onap.policy.models.simulators;
  24
  25 import java.io.FileNotFoundException;
  26 import java.io.IOException;
  27 import java.lang.reflect.InvocationTargetException;
  28 import java.util.HashMap;
  29 import java.util.List;
  30 import java.util.Map;
  31 import java.util.Properties;
  32 import java.util.concurrent.atomic.AtomicReference;
  33 import lombok.AccessLevel;
  34 import lombok.Getter;
  35 import org.apache.commons.lang3.StringUtils;
  36 import org.onap.policy.common.endpoints.event.comm.TopicEndpointManager;
  37 import org.onap.policy.common.endpoints.event.comm.TopicSink;
  38 import org.onap.policy.common.endpoints.event.comm.TopicSource;
  39 import org.onap.policy.common.endpoints.http.server.HttpServletServer;
  40 import org.onap.policy.common.endpoints.http.server.HttpServletServerFactoryInstance;
  41 import org.onap.policy.common.endpoints.parameters.TopicParameters;
  42 import org.onap.policy.common.endpoints.properties.PolicyEndPointProperties;
  43 import org.onap.policy.common.gson.GsonMessageBodyHandler;
  44 import org.onap.policy.common.parameters.BeanValidationResult;
  45 import org.onap.policy.common.utils.coder.Coder;
  46 import org.onap.policy.common.utils.coder.CoderException;
  47 import org.onap.policy.common.utils.coder.StandardCoder;
  48 import org.onap.policy.common.utils.network.NetworkUtil;
  49 import org.onap.policy.common.utils.resources.ResourceUtils;
  50 import org.onap.policy.common.utils.services.Registry;
  51 import org.onap.policy.common.utils.services.ServiceManagerContainer;
  52 import org.onap.policy.models.sim.dmaap.parameters.DmaapSimParameterGroup;
  53 import org.onap.policy.models.sim.dmaap.provider.DmaapSimProvider;
  54 import org.onap.policy.models.sim.dmaap.rest.CambriaMessageBodyHandler;
  55 import org.onap.policy.models.sim.dmaap.rest.TextMessageBodyHandler;
  56 import org.onap.policy.simulators.CdsSimulator;
  57 import org.onap.policy.simulators.TopicServer;
  58 import org.slf4j.Logger;
  59 import org.slf4j.LoggerFactory;
  60
  61 /**
  62  * This class runs all simulators specified in the parameter file.
  63  */
  64 public class Main extends ServiceManagerContainer {
  65     private static final Logger logger = LoggerFactory.getLogger(Main.class);
  66
  67     private static final String CANNOT_CONNECT = "cannot connect to port ";
  68
  69     @Getter(AccessLevel.PROTECTED)
  70     private static Main instance;
  71
  72
  73     /**
  74      * Runs the simulators.
  75      *
  76      * @param paramFile parameter file name
  77      */
  78     public Main(String paramFile) {
  79         super(Main.class.getPackage().getName());
  80
  81         SimulatorParameters params = readParameters(paramFile);
  82         BeanValidationResult result = params.validate("simulators");
  83         if (!result.isValid()) {
  84             logger.error("invalid parameters:\n{}", result.getResult());
  85             throw new IllegalArgumentException("invalid simulator parameters");
  86         }
  87
  88         DmaapSimParameterGroup dmaapProv = params.getDmaapProvider();
  89         String dmaapName = (dmaapProv != null ? dmaapProv.getName() : null);
  90
  91         // dmaap provider
  92         if (dmaapProv != null) {
  93             String provName = dmaapName.replace("simulator", "provider");
  94             AtomicReference<DmaapSimProvider> provRef = new AtomicReference<>();
  95             addAction(provName, () -> provRef.set(buildDmaapProvider(dmaapProv)), () -> provRef.get().shutdown());
  96         }
  97
  98         CdsServerParameters cdsServer = params.getGrpcServer();
  99
 100         // Cds Simulator
 101         if (cdsServer != null) {
 102             AtomicReference<CdsSimulator> cdsSim = new AtomicReference<>();
 103             addAction(cdsServer.getName(), () -> cdsSim.set(buildCdsSimulator(cdsServer)), () -> cdsSim.get().stop());
 104         }
 105
 106         // REST server simulators
 107         // @formatter:off
 108         for (ClassRestServerParameters restsim : params.getRestServers()) {
 109             AtomicReference<HttpServletServer> ref = new AtomicReference<>();
 110             if (StringUtils.isNotBlank(restsim.getResourceLocation())) {
 111                 String resourceLocationId = restsim.getProviderClass() + "_RESOURCE_LOCATION";
 112                 addAction(resourceLocationId,
 113                     () -> Registry.register(resourceLocationId, restsim.getResourceLocation()),
 114                     () -> Registry.unregister(resourceLocationId));
 115             }
 116             addAction(restsim.getName(),
 117                 () -> ref.set(buildRestServer(dmaapName, restsim)),
 118                 () -> ref.get().shutdown());
 119         }
 120
 121         // NOTE: topics must be started AFTER the (dmaap) rest servers
 122
 123         // topic sinks
 124         Map<String, TopicSink> sinks = new HashMap<>();
 125         for (TopicParameters topicParams : params.getTopicSinks()) {
 126             String topic = topicParams.getTopic();
 127             addAction("Sink " + topic,
 128                 () -> sinks.put(topic, startSink(topicParams)),
 129                 () -> sinks.get(topic).shutdown());
 130         }
 131
 132         // topic sources
 133         Map<String, TopicSource> sources = new HashMap<>();
 134         for (TopicParameters topicParams : params.getTopicSources()) {
 135             String topic = topicParams.getTopic();
 136             addAction("Source " + topic,
 137                 () -> sources.put(topic, startSource(topicParams)),
 138                 () -> sources.get(topic).shutdown());
 139         }
 140
 141         // topic server simulators
 142         for (TopicServerParameters topicsim : params.getTopicServers()) {
 143             AtomicReference<TopicServer<?>> ref = new AtomicReference<>();
 144             addAction(topicsim.getName(),
 145                 () -> ref.set(buildTopicServer(topicsim, sinks, sources)),
 146                 () -> ref.get().shutdown());
 147         }
 148         // @formatter:on
 149     }
 150
 151     /**
 152      * The main method. The arguments are validated, thus adding the NOSONAR.
 153      *
 154      * @param args the arguments, the first of which is the name of the parameter file
 155      */
 156     public static void main(final String[] args) { // NOSONAR
 157         /*
 158          * Only one argument is used and is validated implicitly by the constructor (i.e.,
 159          * file-not-found), thus sonar is disabled.
 160          */
 161
 162         try {
 163             if (args.length != 1) {
 164                 throw new IllegalArgumentException("arg(s): parameter-file-name");
 165             }
 166
 167             instance = new Main(args[0]);
 168             instance.start();
 169
 170         } catch (RuntimeException e) {
 171             logger.error("failed to start simulators", e);
 172         }
 173     }
 174
 175     private SimulatorParameters readParameters(String paramFile) {
 176         try {
 177             var paramsJson = getResourceAsString(paramFile);
 178             if (paramsJson == null) {
 179                 throw new IllegalArgumentException(new FileNotFoundException(paramFile));
 180             }
 181
 182             String hostName = NetworkUtil.getHostname();
 183             logger.info("replacing 'HOST_NAME' with {} in {}", hostName, paramFile);
 184
 185             paramsJson = paramsJson.replace("${HOST_NAME}", hostName);
 186
 187             return makeCoder().decode(paramsJson, SimulatorParameters.class);
 188
 189         } catch (CoderException e) {
 190             throw new IllegalArgumentException("cannot decode " + paramFile, e);
 191         }
 192     }
 193
 194     private DmaapSimProvider buildDmaapProvider(DmaapSimParameterGroup params) {
 195         var prov = new DmaapSimProvider(params);
 196         DmaapSimProvider.setInstance(prov);
 197         prov.start();
 198         return prov;
 199     }
 200
 201     private CdsSimulator buildCdsSimulator(CdsServerParameters params) throws IOException {
 202         var cdsSimulator = new CdsSimulator(params.getHost(), params.getPort(), params.getResourceLocation(),
 203             params.getSuccessRepeatCount(), params.getRequestedResponseDelayMs());
 204         cdsSimulator.start();
 205         return cdsSimulator;
 206     }
 207
 208
 209     private TopicSink startSink(TopicParameters params) {
 210         TopicSink sink = TopicEndpointManager.getManager().addTopicSinks(List.of(params)).get(0);
 211         sink.start();
 212         return sink;
 213     }
 214
 215     private TopicSource startSource(TopicParameters params) {
 216         TopicSource source = TopicEndpointManager.getManager().addTopicSources(List.of(params)).get(0);
 217         source.start();
 218         return source;
 219     }
 220
 221     private HttpServletServer buildRestServer(String dmaapName, ClassRestServerParameters params) {
 222         try {
 223             var props = getServerProperties(dmaapName, params);
 224             HttpServletServer testServer = makeServer(props);
 225             testServer.waitedStart(5000);
 226
 227             String svcpfx = PolicyEndPointProperties.PROPERTY_HTTP_SERVER_SERVICES + "." + params.getName();
 228             String hostName = props.getProperty(svcpfx + PolicyEndPointProperties.PROPERTY_HTTP_HOST_SUFFIX);
 229
 230             if (!isTcpPortOpen(hostName, testServer.getPort())) {
 231                 throw new IllegalStateException(CANNOT_CONNECT + testServer.getPort());
 232             }
 233
 234             return testServer;
 235
 236         } catch (InterruptedException e) {
 237             Thread.currentThread().interrupt();
 238             throw new IllegalStateException("interrupted while building " + params.getName(), e);
 239         }
 240     }
 241
 242     private TopicServer<?> buildTopicServer(TopicServerParameters params, Map<String, TopicSink> sinks,
 243                     Map<String, TopicSource> sources) {
 244         try {
 245             // find the desired sink
 246             TopicSink sink = sinks.get(params.getSink());
 247             if (sink == null) {
 248                 throw new IllegalArgumentException("invalid sink topic " + params.getSink());
 249             }
 250
 251             // find the desired source
 252             TopicSource source = sources.get(params.getSource());
 253             if (source == null) {
 254                 throw new IllegalArgumentException("invalid source topic " + params.getSource());
 255             }
 256
 257             // create the topic server
 258             return (TopicServer<?>) Class.forName(params.getProviderClass())
 259                             .getDeclaredConstructor(TopicSink.class, TopicSource.class).newInstance(sink, source);
 260
 261         } catch (InstantiationException | IllegalAccessException | InvocationTargetException | NoSuchMethodException
 262                         | SecurityException | ClassNotFoundException e) {
 263             throw new IllegalArgumentException("cannot create TopicServer: " + params.getName(), e);
 264         }
 265     }
 266
 267     /**
 268      * Creates a set of properties, suitable for building a REST server, from the
 269      * parameters.
 270      *
 271      * @param params parameters from which to build the properties
 272      * @return a set of properties representing the given parameters
 273      */
 274     private static Properties getServerProperties(String dmaapName, ClassRestServerParameters params) {
 275         final var props = new Properties();
 276         props.setProperty(PolicyEndPointProperties.PROPERTY_HTTP_SERVER_SERVICES, params.getName());
 277
 278         final String svcpfx = PolicyEndPointProperties.PROPERTY_HTTP_SERVER_SERVICES + "." + params.getName();
 279
 280         props.setProperty(PolicyEndPointProperties.PROPERTY_HTTP_SERVER_SERVICES, params.getName());
 281         props.setProperty(svcpfx + PolicyEndPointProperties.PROPERTY_HTTP_HOST_SUFFIX, params.getHost());
 282         props.setProperty(svcpfx + PolicyEndPointProperties.PROPERTY_HTTP_PORT_SUFFIX,
 283                         Integer.toString(params.getPort()));
 284         props.setProperty(svcpfx + PolicyEndPointProperties.PROPERTY_HTTP_HTTPS_SUFFIX,
 285                         Boolean.toString(params.isHttps()));
 286         props.setProperty(svcpfx + PolicyEndPointProperties.PROPERTY_HTTP_REST_CLASSES_SUFFIX,
 287                         params.getProviderClass());
 288         props.setProperty(svcpfx + PolicyEndPointProperties.PROPERTY_MANAGED_SUFFIX, "false");
 289         props.setProperty(svcpfx + PolicyEndPointProperties.PROPERTY_HTTP_SWAGGER_SUFFIX, "false");
 290         props.setProperty(svcpfx + PolicyEndPointProperties.PROPERTY_HTTP_SNI_HOST_CHECK_SUFFIX, "false");
 291         props.setProperty(svcpfx + PolicyEndPointProperties.PROPERTY_MANAGED_SUFFIX, "true");
 292
 293         if (dmaapName != null && dmaapName.equals(params.getName())) {
 294             props.setProperty(svcpfx + PolicyEndPointProperties.PROPERTY_HTTP_SERIALIZATION_PROVIDER,
 295                             String.join(",", CambriaMessageBodyHandler.class.getName(),
 296                                             GsonMessageBodyHandler.class.getName(),
 297                                             TextMessageBodyHandler.class.getName()));
 298         } else {
 299             props.setProperty(svcpfx + PolicyEndPointProperties.PROPERTY_HTTP_SERIALIZATION_PROVIDER, String.join(",",
 300                             GsonMessageBodyHandler.class.getName(), TextMessageBodyHandler.class.getName()));
 301         }
 302
 303         return props;
 304     }
 305
 306     // the following methods may be overridden by junit tests
 307
 308     protected String getResourceAsString(String resourceName) {
 309         return ResourceUtils.getResourceAsString(resourceName);
 310     }
 311
 312     protected Coder makeCoder() {
 313         return new StandardCoder();
 314     }
 315
 316     protected HttpServletServer makeServer(Properties props) {
 317         return HttpServletServerFactoryInstance.getServerFactory().build(props).get(0);
 318     }
 319
 320     protected boolean isTcpPortOpen(String hostName, int port) throws InterruptedException {
 321         return NetworkUtil.isTcpPortOpen(hostName, port, 100, 200L);
 322     }
 323 }