main/src/main/java/org/onap/policy/pap/main/startstop/PapActivator.java

   1 /*-
   2  * ============LICENSE_START=======================================================
   3  *  Copyright (C) 2019 Nordix Foundation.
   4  *  Modifications Copyright (C) 2019-2021 AT&T Intellectual Property.
   5  * ================================================================================
   6  * Licensed under the Apache License, Version 2.0 (the "License");
   7  * you may not use this file except in compliance with the License.
   8  * You may obtain a copy of the License at
   9  *
  10  *      http://www.apache.org/licenses/LICENSE-2.0
  11  *
  12  * Unless required by applicable law or agreed to in writing, software
  13  * distributed under the License is distributed on an "AS IS" BASIS,
  14  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  15  * See the License for the specific language governing permissions and
  16  * limitations under the License.
  17  *
  18  * SPDX-License-Identifier: Apache-2.0
  19  * ============LICENSE_END=========================================================
  20  */
  21
  22 package org.onap.policy.pap.main.startstop;
  23
  24 import java.util.Arrays;
  25 import java.util.concurrent.Executors;
  26 import java.util.concurrent.ScheduledExecutorService;
  27 import java.util.concurrent.TimeUnit;
  28 import java.util.concurrent.atomic.AtomicReference;
  29 import org.onap.policy.common.endpoints.event.comm.TopicEndpointManager;
  30 import org.onap.policy.common.endpoints.event.comm.TopicSource;
  31 import org.onap.policy.common.endpoints.http.client.HttpClientFactoryInstance;
  32 import org.onap.policy.common.endpoints.http.server.RestServer;
  33 import org.onap.policy.common.endpoints.listeners.MessageTypeDispatcher;
  34 import org.onap.policy.common.endpoints.listeners.RequestIdDispatcher;
  35 import org.onap.policy.common.parameters.ParameterService;
  36 import org.onap.policy.common.utils.services.Registry;
  37 import org.onap.policy.common.utils.services.ServiceManagerContainer;
  38 import org.onap.policy.models.pap.concepts.PolicyNotification;
  39 import org.onap.policy.models.pdp.concepts.PdpMessage;
  40 import org.onap.policy.models.pdp.concepts.PdpStatus;
  41 import org.onap.policy.models.pdp.enums.PdpMessageType;
  42 import org.onap.policy.pap.main.PapConstants;
  43 import org.onap.policy.pap.main.PolicyModelsProviderFactoryWrapper;
  44 import org.onap.policy.pap.main.PolicyPapRuntimeException;
  45 import org.onap.policy.pap.main.comm.PdpHeartbeatListener;
  46 import org.onap.policy.pap.main.comm.PdpModifyRequestMap;
  47 import org.onap.policy.pap.main.comm.Publisher;
  48 import org.onap.policy.pap.main.comm.TimerManager;
  49 import org.onap.policy.pap.main.notification.PolicyNotifier;
  50 import org.onap.policy.pap.main.parameters.PapParameterGroup;
  51 import org.onap.policy.pap.main.parameters.PdpModifyRequestMapParams;
  52 import org.onap.policy.pap.main.rest.HealthCheckRestControllerV1;
  53 import org.onap.policy.pap.main.rest.PapAafFilter;
  54 import org.onap.policy.pap.main.rest.PapStatisticsManager;
  55 import org.onap.policy.pap.main.rest.PdpGroupCreateOrUpdateControllerV1;
  56 import org.onap.policy.pap.main.rest.PdpGroupDeleteControllerV1;
  57 import org.onap.policy.pap.main.rest.PdpGroupDeployControllerV1;
  58 import org.onap.policy.pap.main.rest.PdpGroupHealthCheckControllerV1;
  59 import org.onap.policy.pap.main.rest.PdpGroupQueryControllerV1;
  60 import org.onap.policy.pap.main.rest.PdpGroupStateChangeControllerV1;
  61 import org.onap.policy.pap.main.rest.PolicyComponentsHealthCheckControllerV1;
  62 import org.onap.policy.pap.main.rest.PolicyComponentsHealthCheckProvider;
  63 import org.onap.policy.pap.main.rest.PolicyStatusControllerV1;
  64 import org.onap.policy.pap.main.rest.PolicyUndeployerImpl;
  65 import org.onap.policy.pap.main.rest.StatisticsRestControllerV1;
  66
  67 /**
  68  * This class activates Policy Administration (PAP) as a complete service together with all its controllers, listeners &
  69  * handlers.
  70  *
  71  * @author Ram Krishna Verma (ram.krishna.verma@est.tech)
  72  */
  73 public class PapActivator extends ServiceManagerContainer {
  74     private static final String[] MSG_TYPE_NAMES = { "messageName" };
  75     private static final String[] REQ_ID_NAMES = { "response", "responseTo" };
  76
  77     /**
  78      * Max number of heat beats that can be missed before PAP removes a PDP.
  79      */
  80     private static final int MAX_MISSED_HEARTBEATS = 3;
  81
  82     private final PapParameterGroup papParameterGroup;
  83
  84     /**
  85      * Listens for messages on the topic, decodes them into a {@link PdpStatus} message, and then dispatches them to
  86      * {@link #reqIdDispatcher}.
  87      */
  88     private final MessageTypeDispatcher msgDispatcher;
  89
  90     /**
  91      * Listens for {@link PdpStatus} messages and then routes them to the listener associated with the ID of the
  92      * originating request.
  93      */
  94     private final RequestIdDispatcher<PdpStatus> reqIdDispatcher;
  95
  96     /**
  97      * Listener for anonymous {@link PdpStatus} messages either for registration or heartbeat.
  98      */
  99     private final PdpHeartbeatListener pdpHeartbeatListener;
 100
 101     /**
 102      * Instantiate the activator for policy pap as a complete service.
 103      *
 104      * @param papParameterGroup the parameters for the pap service
 105      */
 106     public PapActivator(final PapParameterGroup papParameterGroup) {
 107         super("Policy PAP");
 108
 109         TopicEndpointManager.getManager().addTopics(papParameterGroup.getTopicParameterGroup());
 110
 111         try {
 112             this.papParameterGroup = papParameterGroup;
 113             this.msgDispatcher = new MessageTypeDispatcher(MSG_TYPE_NAMES);
 114             this.reqIdDispatcher = new RequestIdDispatcher<>(PdpStatus.class, REQ_ID_NAMES);
 115             this.pdpHeartbeatListener = new PdpHeartbeatListener(papParameterGroup.getPdpParameters());
 116
 117         } catch (final RuntimeException e) {
 118             throw new PolicyPapRuntimeException(e);
 119         }
 120
 121         papParameterGroup.getRestServerParameters().setName(papParameterGroup.getName());
 122
 123         final var pdpUpdateLock = new Object();
 124         final var pdpParams = papParameterGroup.getPdpParameters();
 125         final AtomicReference<Publisher<PdpMessage>> pdpPub = new AtomicReference<>();
 126         final AtomicReference<Publisher<PolicyNotification>> notifyPub = new AtomicReference<>();
 127         final AtomicReference<TimerManager> pdpUpdTimers = new AtomicReference<>();
 128         final AtomicReference<TimerManager> pdpStChgTimers = new AtomicReference<>();
 129         final AtomicReference<ScheduledExecutorService> pdpExpirationTimer = new AtomicReference<>();
 130         final AtomicReference<PolicyModelsProviderFactoryWrapper> daoFactory = new AtomicReference<>();
 131         final AtomicReference<PdpModifyRequestMap> requestMap = new AtomicReference<>();
 132         final AtomicReference<RestServer> restServer = new AtomicReference<>();
 133         final AtomicReference<PolicyNotifier> notifier = new AtomicReference<>();
 134
 135         // @formatter:off
 136         addAction("PAP parameters",
 137             () -> ParameterService.register(papParameterGroup),
 138             () -> ParameterService.deregister(papParameterGroup.getName()));
 139
 140         addAction("DAO Factory",
 141             () -> daoFactory.set(new PolicyModelsProviderFactoryWrapper(
 142                                     papParameterGroup.getDatabaseProviderParameters())),
 143             () -> daoFactory.get().close());
 144
 145         addAction("DAO Factory registration",
 146             () -> Registry.register(PapConstants.REG_PAP_DAO_FACTORY, daoFactory.get()),
 147             () -> Registry.unregister(PapConstants.REG_PAP_DAO_FACTORY));
 148
 149         addAction("Pdp Heartbeat Listener",
 150             () -> reqIdDispatcher.register(pdpHeartbeatListener),
 151             () -> reqIdDispatcher.unregister(pdpHeartbeatListener));
 152
 153         addAction("Request ID Dispatcher",
 154             () -> msgDispatcher.register(PdpMessageType.PDP_STATUS.name(), this.reqIdDispatcher),
 155             () -> msgDispatcher.unregister(PdpMessageType.PDP_STATUS.name()));
 156
 157         addAction("Message Dispatcher",
 158             this::registerMsgDispatcher,
 159             this::unregisterMsgDispatcher);
 160
 161         addAction("topics",
 162             TopicEndpointManager.getManager()::start,
 163             TopicEndpointManager.getManager()::shutdown);
 164
 165         addAction("PAP statistics",
 166             () -> Registry.register(PapConstants.REG_STATISTICS_MANAGER, new PapStatisticsManager()),
 167             () -> Registry.unregister(PapConstants.REG_STATISTICS_MANAGER));
 168
 169         addAction("PDP publisher",
 170             () -> {
 171                 pdpPub.set(new Publisher<>(PapConstants.TOPIC_POLICY_PDP_PAP));
 172                 startThread(pdpPub.get());
 173             },
 174             () -> pdpPub.get().stop());
 175
 176         addAction("Policy Notification publisher",
 177             () -> {
 178                 notifyPub.set(new Publisher<>(PapConstants.TOPIC_POLICY_NOTIFICATION));
 179                 startThread(notifyPub.get());
 180                 notifier.set(new PolicyNotifier(notifyPub.get(), daoFactory.get()));
 181             },
 182             () -> notifyPub.get().stop());
 183
 184         addAction("Policy Notifier",
 185             () -> Registry.register(PapConstants.REG_POLICY_NOTIFIER, notifier.get()),
 186             () -> Registry.unregister(PapConstants.REG_POLICY_NOTIFIER));
 187
 188         addAction("PDP update timers",
 189             () -> {
 190                 pdpUpdTimers.set(new TimerManager("update", pdpParams.getUpdateParameters().getMaxWaitMs()));
 191                 startThread(pdpUpdTimers.get());
 192             },
 193             () -> pdpUpdTimers.get().stop());
 194
 195         addAction("PDP state-change timers",
 196             () -> {
 197                 pdpStChgTimers.set(new TimerManager("state-change", pdpParams.getUpdateParameters().getMaxWaitMs()));
 198                 startThread(pdpStChgTimers.get());
 199             },
 200             () -> pdpStChgTimers.get().stop());
 201
 202         addAction("PDP modification lock",
 203             () -> Registry.register(PapConstants.REG_PDP_MODIFY_LOCK, pdpUpdateLock),
 204             () -> Registry.unregister(PapConstants.REG_PDP_MODIFY_LOCK));
 205
 206         addAction("PDP modification requests",
 207             () -> {
 208                 requestMap.set(new PdpModifyRequestMap(
 209                             PdpModifyRequestMapParams.builder()
 210                                     .maxPdpAgeMs(MAX_MISSED_HEARTBEATS * pdpParams.getHeartBeatMs())
 211                                     .daoFactory(daoFactory.get())
 212                                     .modifyLock(pdpUpdateLock)
 213                                     .params(pdpParams)
 214                                     .policyNotifier(notifier.get())
 215                                     .pdpPublisher(pdpPub.get())
 216                                     .responseDispatcher(reqIdDispatcher)
 217                                     .stateChangeTimers(pdpStChgTimers.get())
 218                                     .updateTimers(pdpUpdTimers.get())
 219                                     .build()));
 220                 Registry.register(PapConstants.REG_PDP_MODIFY_MAP, requestMap.get());
 221
 222                 // now that it's registered, we can attach a "policy undeploy" provider
 223                 requestMap.get().setPolicyUndeployer(new PolicyUndeployerImpl());
 224             },
 225             () -> Registry.unregister(PapConstants.REG_PDP_MODIFY_MAP));
 226
 227         addAction("PDP expiration timer",
 228             () -> {
 229                 long frequencyMs = pdpParams.getHeartBeatMs();
 230                 pdpExpirationTimer.set(Executors.newScheduledThreadPool(1));
 231                 pdpExpirationTimer.get().scheduleWithFixedDelay(
 232                     requestMap.get()::removeExpiredPdps,
 233                     frequencyMs,
 234                     frequencyMs,
 235                     TimeUnit.MILLISECONDS);
 236             },
 237             () -> pdpExpirationTimer.get().shutdownNow());
 238
 239         addAction("PAP client executor",
 240             () ->
 241                 PolicyComponentsHealthCheckProvider.initializeClientHealthCheckExecutorService(papParameterGroup,
 242                                 HttpClientFactoryInstance.getClientFactory()),
 243                 PolicyComponentsHealthCheckProvider::cleanup);
 244
 245         addAction("REST server",
 246             () -> {
 247                 var server = new RestServer(papParameterGroup.getRestServerParameters(), PapAafFilter.class,
 248                                 HealthCheckRestControllerV1.class,
 249                                 StatisticsRestControllerV1.class,
 250                                 PdpGroupCreateOrUpdateControllerV1.class,
 251                                 PdpGroupDeployControllerV1.class,
 252                                 PdpGroupDeleteControllerV1.class,
 253                                 PdpGroupStateChangeControllerV1.class,
 254                                 PdpGroupQueryControllerV1.class,
 255                                 PdpGroupHealthCheckControllerV1.class,
 256                                 PolicyStatusControllerV1.class,
 257                                 PolicyComponentsHealthCheckControllerV1.class);
 258                 restServer.set(server);
 259                 restServer.get().start();
 260             },
 261             () -> restServer.get().stop());
 262         // @formatter:on
 263     }
 264
 265     /**
 266      * Starts a background thread.
 267      *
 268      * @param runner function to run in the background
 269      */
 270     private void startThread(final Runnable runner) {
 271         final var thread = new Thread(runner);
 272         thread.setDaemon(true);
 273
 274         thread.start();
 275     }
 276
 277     /**
 278      * Get the parameters used by the activator.
 279      *
 280      * @return the parameters of the activator
 281      */
 282     public PapParameterGroup getParameterGroup() {
 283         return papParameterGroup;
 284     }
 285
 286     /**
 287      * Registers the dispatcher with the topic source(s).
 288      */
 289     private void registerMsgDispatcher() {
 290         for (final TopicSource source : TopicEndpointManager.getManager()
 291                 .getTopicSources(Arrays.asList(PapConstants.TOPIC_POLICY_PDP_PAP))) {
 292             source.register(msgDispatcher);
 293         }
 294     }
 295
 296     /**
 297      * Unregisters the dispatcher from the topic source(s).
 298      */
 299     private void unregisterMsgDispatcher() {
 300         for (final TopicSource source : TopicEndpointManager.getManager()
 301                 .getTopicSources(Arrays.asList(PapConstants.TOPIC_POLICY_PDP_PAP))) {
 302             source.unregister(msgDispatcher);
 303         }
 304     }
 305 }