gerrit.onap Code Review - policy/apex-pdp.git/blob

   1 /*-
   2  * ============LICENSE_START=======================================================
   3  *  Copyright (C) 2019-2021, 2023 Nordix Foundation.
   4  *  Modifications Copyright (C) 2019, 2021 AT&T Intellectual Property. All rights reserved.
   5  *  Modifications Copyright (C) 2021 Bell Canada. All rights reserved.
   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.apex.services.onappf;
  24
  25 import java.util.List;
  26 import lombok.Getter;
  27 import lombok.Setter;
  28 import org.onap.policy.apex.services.onappf.comm.PdpStateChangeListener;
  29 import org.onap.policy.apex.services.onappf.comm.PdpStatusPublisher;
  30 import org.onap.policy.apex.services.onappf.comm.PdpUpdateListener;
  31 import org.onap.policy.apex.services.onappf.exception.ApexStarterException;
  32 import org.onap.policy.apex.services.onappf.exception.ApexStarterRunTimeException;
  33 import org.onap.policy.apex.services.onappf.handler.PdpMessageHandler;
  34 import org.onap.policy.apex.services.onappf.parameters.ApexStarterParameterGroup;
  35 import org.onap.policy.apex.services.onappf.rest.HealthCheckRestControllerV1;
  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.RestServer;
  40 import org.onap.policy.common.endpoints.listeners.MessageTypeDispatcher;
  41 import org.onap.policy.common.utils.network.NetworkUtil;
  42 import org.onap.policy.common.utils.services.Registry;
  43 import org.onap.policy.common.utils.services.ServiceManager;
  44 import org.onap.policy.common.utils.services.ServiceManagerException;
  45 import org.onap.policy.models.pdp.enums.PdpMessageType;
  46 import org.onap.policy.models.tosca.authorative.concepts.ToscaConceptIdentifier;
  47 import org.slf4j.Logger;
  48 import org.slf4j.LoggerFactory;
  49
  50 /**
  51  * This class activates the ApexStarter as a complete service together with all its handlers.
  52  *
  53  * @author Ajith Sreekumar (ajith.sreekumar@est.tech)
  54  */
  55 public class ApexStarterActivator {
  56
  57     private static final Logger LOGGER = LoggerFactory.getLogger(ApexStarterActivator.class);
  58     @Getter
  59     private final ApexStarterParameterGroup parameterGroup;
  60     private final List<TopicSink> topicSinks; // topics to which apex-pdp sends pdp status
  61     private final List<TopicSource> topicSources; // topics to which apex-pdp listens to for messages from pap.
  62     private static final String[] MSG_TYPE_NAMES = { "messageName" };
  63
  64     /**
  65      * Listens for messages on the topic, decodes them into a message, and then dispatches them.
  66      */
  67     private final MessageTypeDispatcher msgDispatcher;
  68
  69     /**
  70      * Used to manage the services.
  71      */
  72     private final ServiceManager manager;
  73
  74     /**
  75      * The ApexStarter REST API server.
  76      */
  77     private RestServer restServer;
  78
  79     @Getter
  80     @Setter(lombok.AccessLevel.PRIVATE)
  81     private volatile boolean alive = false;
  82
  83     @Getter
  84     private List<ToscaConceptIdentifier> supportedPolicyTypes;
  85
  86     @Getter
  87     private final String instanceId;
  88
  89     /**
  90      * Instantiate the activator for onappf PDP-A.
  91      *
  92      * @param apexStarterParameterGroup the parameters for the onappf PDP-A service
  93      */
  94     public ApexStarterActivator(final ApexStarterParameterGroup apexStarterParameterGroup) {
  95
  96         topicSinks = TopicEndpointManager.getManager()
  97                         .addTopicSinks(apexStarterParameterGroup.getTopicParameterGroup().getTopicSinks());
  98
  99         topicSources = TopicEndpointManager.getManager()
 100                         .addTopicSources(apexStarterParameterGroup.getTopicParameterGroup().getTopicSources());
 101
 102         instanceId = NetworkUtil.genUniqueName("apex");
 103         LOGGER.debug("ApexStarterActivator initializing with instance id: {}", instanceId);
 104         try {
 105             this.parameterGroup = apexStarterParameterGroup;
 106             this.msgDispatcher = new MessageTypeDispatcher(MSG_TYPE_NAMES);
 107         } catch (final RuntimeException e) {
 108             throw new ApexStarterRunTimeException(e);
 109         }
 110
 111         final var pdpUpdateListener = new PdpUpdateListener();
 112         final var pdpStateChangeListener = new PdpStateChangeListener();
 113         final var pdpMessageHandler = new PdpMessageHandler();
 114         supportedPolicyTypes =
 115             pdpMessageHandler.getSupportedPolicyTypesFromParameters(apexStarterParameterGroup.getPdpStatusParameters());
 116
 117         // @formatter:off
 118         this.manager = new ServiceManager()
 119                 .addAction("topics",
 120                     () -> TopicEndpointManager.getManager().start(),
 121                     () -> TopicEndpointManager.getManager().shutdown())
 122                 .addAction("set alive",
 123                     () -> setAlive(true),
 124                     () -> setAlive(false))
 125                 .addAction("register pdp status context object",
 126                     () -> Registry.register(ApexStarterConstants.REG_PDP_STATUS_OBJECT,
 127                                 pdpMessageHandler.createPdpStatusFromParameters(instanceId,
 128                                         apexStarterParameterGroup.getPdpStatusParameters())),
 129                     () -> Registry.unregister(ApexStarterConstants.REG_PDP_STATUS_OBJECT))
 130                 .addAction("topic sinks",
 131                     () -> Registry.register(ApexStarterConstants.REG_APEX_PDP_TOPIC_SINKS, topicSinks),
 132                     () -> Registry.unregister(ApexStarterConstants.REG_APEX_PDP_TOPIC_SINKS))
 133                 .addAction("Pdp Status publisher",
 134                     () -> Registry.register(ApexStarterConstants.REG_PDP_STATUS_PUBLISHER,
 135                                 new PdpStatusPublisher(topicSinks,
 136                                         apexStarterParameterGroup.getPdpStatusParameters().getTimeIntervalMs())),
 137                     this::stopAndRemovePdpStatusPublisher)
 138                 .addAction("Register pdp update listener",
 139                     () -> msgDispatcher.register(PdpMessageType.PDP_UPDATE.name(), pdpUpdateListener),
 140                     () -> msgDispatcher.unregister(PdpMessageType.PDP_UPDATE.name()))
 141                 .addAction("Register pdp state change request dispatcher",
 142                     () -> msgDispatcher.register(PdpMessageType.PDP_STATE_CHANGE.name(), pdpStateChangeListener),
 143                     () -> msgDispatcher.unregister(PdpMessageType.PDP_STATE_CHANGE.name()))
 144                 .addAction("Message Dispatcher",
 145                     this::registerMsgDispatcher,
 146                     this::unregisterMsgDispatcher)
 147                 .addAction("Create REST server",
 148                     () -> restServer = new RestServer(apexStarterParameterGroup.getRestServerParameters(),
 149                                 List.of(), List.of(HealthCheckRestControllerV1.class)),
 150                     () -> restServer = null)
 151                 .addAction("Rest Server",
 152                     () -> restServer.start(),
 153                     () -> restServer.stop());
 154
 155         // @formatter:on
 156     }
 157
 158     /**
 159      * Method to stop and unregister the pdp status publisher.
 160      */
 161     private void stopAndRemovePdpStatusPublisher() {
 162         final var pdpStatusPublisher =
 163                 Registry.get(ApexStarterConstants.REG_PDP_STATUS_PUBLISHER, PdpStatusPublisher.class);
 164         pdpStatusPublisher.terminate();
 165         Registry.unregister(ApexStarterConstants.REG_PDP_STATUS_PUBLISHER);
 166     }
 167
 168     /**
 169      * Initialize ApexStarter service.
 170      *
 171      * @throws ApexStarterException on errors in initializing the service
 172      */
 173     public void initialize() throws ApexStarterException {
 174         if (isAlive()) {
 175             throw new IllegalStateException("activator already initialized");
 176         }
 177
 178         try {
 179             LOGGER.debug("ApexStarter starting as a service . . .");
 180             manager.start();
 181             LOGGER.debug("ApexStarter started as a service");
 182         } catch (final ServiceManagerException exp) {
 183             LOGGER.error("ApexStarter service startup failed");
 184             throw new ApexStarterException(exp.getMessage(), exp);
 185         }
 186     }
 187
 188     /**
 189      * Terminate ApexStarter.
 190      *
 191      * @throws ApexStarterException on errors in terminating the service
 192      */
 193     public void terminate() throws ApexStarterException {
 194         if (!isAlive()) {
 195             throw new IllegalStateException("activator is not running");
 196         }
 197         try {
 198             final var pdpStatusPublisher =
 199                     Registry.get(ApexStarterConstants.REG_PDP_STATUS_PUBLISHER, PdpStatusPublisher.class);
 200             // send a final heartbeat with terminated status
 201             pdpStatusPublisher.send(new PdpMessageHandler().getTerminatedPdpStatus());
 202             manager.stop();
 203             Registry.unregister(ApexStarterConstants.REG_APEX_STARTER_ACTIVATOR);
 204         } catch (final ServiceManagerException exp) {
 205             LOGGER.error("ApexStarter termination failed");
 206             throw new ApexStarterException(exp.getMessage(), exp);
 207         }
 208     }
 209
 210     /**
 211      * Registers the dispatcher with the topic source(s).
 212      */
 213     private void registerMsgDispatcher() {
 214         for (final TopicSource source : topicSources) {
 215             source.register(msgDispatcher);
 216         }
 217     }
 218
 219     /**
 220      * Unregisters the dispatcher from the topic source(s).
 221      */
 222     private void unregisterMsgDispatcher() {
 223         for (final TopicSource source : topicSources) {
 224             source.unregister(msgDispatcher);
 225         }
 226     }
 227 }