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

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