catalog-be/src/main/java/org/openecomp/sdc/be/components/distribution/engine/DistributionEngineInitTask.java

   1 /*-
   2  * ============LICENSE_START=======================================================
   3  * SDC
   4  * ================================================================================
   5  * Copyright (C) 2017 AT&T Intellectual Property. 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  * ============LICENSE_END=========================================================
  19  */
  20
  21 package org.openecomp.sdc.be.components.distribution.engine;
  22
  23 import fj.data.Either;
  24 import org.openecomp.sdc.be.config.BeEcompErrorManager;
  25 import org.openecomp.sdc.be.config.DistributionEngineConfiguration;
  26 import org.openecomp.sdc.be.distribution.api.client.CambriaOperationStatus;
  27 import org.openecomp.sdc.be.impl.ComponentsUtils;
  28 import org.openecomp.sdc.be.resources.data.OperationalEnvironmentEntry;
  29 import org.openecomp.sdc.be.resources.data.auditing.AuditingActionEnum;
  30 import org.slf4j.Logger;
  31 import org.slf4j.LoggerFactory;
  32
  33 import java.util.HashSet;
  34 import java.util.Set;
  35 import java.util.concurrent.Executors;
  36 import java.util.concurrent.ScheduledExecutorService;
  37 import java.util.concurrent.ScheduledFuture;
  38 import java.util.concurrent.TimeUnit;
  39 import java.util.concurrent.atomic.AtomicBoolean;
  40 import java.util.stream.Collectors;
  41
  42 public class DistributionEngineInitTask implements Runnable {
  43
  44     public static final String INIT_DISTRIBUTION_ENGINE_FLOW = "initDistributionEngine";
  45     public static final String ALREADY_EXISTS = "ALREADY_EXISTS";
  46     public static final String CONSUMER = "CONSUMER";
  47     public static final String PRODUCER = "PRODUCER";
  48     public static final String CREATED = "CREATED";
  49     public static final String FAILED = "FAILED";
  50     public static final Integer HTTP_OK = 200;
  51
  52     private Long delayBeforeStartFlow = 0l;
  53     private DistributionEngineConfiguration deConfiguration;
  54     private String envName;
  55     private long retryInterval;
  56     private long currentRetryInterval;
  57     private long maxInterval;
  58     boolean maximumRetryInterval = false;
  59     private AtomicBoolean status = null;
  60     ComponentsUtils componentsUtils = null;
  61     DistributionEnginePollingTask distributionEnginePollingTask = null;
  62     private OperationalEnvironmentEntry environmentEntry;
  63
  64     private CambriaHandler cambriaHandler = new CambriaHandler();
  65
  66     public DistributionEngineInitTask(Long delayBeforeStartFlow, DistributionEngineConfiguration deConfiguration, String envName, AtomicBoolean status, ComponentsUtils componentsUtils, DistributionEnginePollingTask distributionEnginePollingTask, OperationalEnvironmentEntry environmentEntry) {
  67         super();
  68         this.delayBeforeStartFlow = delayBeforeStartFlow;
  69         this.deConfiguration = deConfiguration;
  70         this.envName = envName;
  71         this.retryInterval = deConfiguration.getInitRetryIntervalSec();
  72         this.currentRetryInterval = retryInterval;
  73         this.maxInterval = deConfiguration.getInitMaxIntervalSec();
  74         this.status = status;
  75         this.componentsUtils = componentsUtils;
  76         this.distributionEnginePollingTask = distributionEnginePollingTask;
  77         this.environmentEntry = environmentEntry;
  78     }
  79
  80     private ScheduledExecutorService scheduledExecutorService = Executors.newScheduledThreadPool(1);
  81
  82     private static final Logger logger = LoggerFactory.getLogger(DistributionEngineInitTask.class);
  83
  84     ScheduledFuture<?> scheduledFuture = null;
  85
  86     public void startTask() {
  87         if (scheduledExecutorService != null) {
  88             Integer retryInterval = deConfiguration.getInitRetryIntervalSec();
  89             logger.debug("Start Distribution Engine init task. retry interval {} seconds, delay before first run {} seconds", retryInterval, delayBeforeStartFlow);
  90             this.scheduledFuture = scheduledExecutorService.scheduleAtFixedRate(this, delayBeforeStartFlow, retryInterval, TimeUnit.SECONDS);
  91
  92         }
  93     }
  94
  95     public void restartTask() {
  96
  97         this.stopTask();
  98
  99         logger.debug("Start Distribution Engine init task. next run in {} seconds", this.currentRetryInterval);
 100
 101         long lastCurrentInterval = currentRetryInterval;
 102         incrementRetryInterval();
 103
 104         this.scheduledFuture = scheduledExecutorService.scheduleAtFixedRate(this, lastCurrentInterval, this.currentRetryInterval, TimeUnit.SECONDS);
 105
 106     }
 107
 108     protected void incrementRetryInterval() {
 109         if (currentRetryInterval < maxInterval) {
 110             currentRetryInterval *= 2;
 111             if (currentRetryInterval > maxInterval) {
 112                 setMaxRetryInterval();
 113             }
 114         } else {
 115             setMaxRetryInterval();
 116         }
 117     }
 118
 119     private void setMaxRetryInterval() {
 120         currentRetryInterval = maxInterval;
 121         maximumRetryInterval = true;
 122         logger.debug("Set next retry init interval to {}", maxInterval);
 123     }
 124
 125     public void stopTask() {
 126         if (scheduledFuture != null) {
 127             boolean result = scheduledFuture.cancel(true);
 128             logger.debug("Stop reinit task. result = {}", result);
 129             if (false == result) {
 130                 BeEcompErrorManager.getInstance().logBeUebSystemError(INIT_DISTRIBUTION_ENGINE_FLOW, "try to stop the reinit task");
 131             }
 132             scheduledFuture = null;
 133         }
 134     }
 135
 136     public void destroy() {
 137         this.stopTask();
 138         if (scheduledExecutorService != null) {
 139             scheduledExecutorService.shutdown();
 140         }
 141     }
 142
 143     @Override
 144     public void run() {
 145
 146         boolean result = false;
 147         result = initFlow();
 148
 149         if (true == result) {
 150             this.stopTask();
 151             this.status.set(true);
 152             if (this.distributionEnginePollingTask != null) {
 153                 String topicName = buildTopicName(deConfiguration.getDistributionStatusTopicName(), envName);
 154                 logger.debug("start polling distribution status topic {}", topicName);
 155                 this.distributionEnginePollingTask.startTask(topicName);
 156             }
 157         } else {
 158             if (false == maximumRetryInterval) {
 159                 this.restartTask();
 160             }
 161         }
 162     }
 163
 164     /**
 165      * run initialization flow
 166      *
 167      * @return
 168      */
 169     public boolean initFlow() {
 170
 171         logger.trace("Start init flow for environment {}", this.envName);
 172
 173         Set<String> topicsList = null;
 174         Either<Set<String>, CambriaErrorResponse> getTopicsRes = null;
 175
 176         getTopicsRes = cambriaHandler.getTopics(environmentEntry.getDmaapUebAddress().stream().collect(Collectors.toList()));
 177         if (getTopicsRes.isRight()) {
 178             CambriaErrorResponse status = getTopicsRes.right().value();
 179             if (status.getOperationStatus() == CambriaOperationStatus.NOT_FOUND) {
 180                 topicsList = new HashSet<>();
 181             } else {
 182                 BeEcompErrorManager.getInstance().logBeUebSystemError(INIT_DISTRIBUTION_ENGINE_FLOW, "try retrieve list of topics from U-EB server");
 183                 return false;
 184             }
 185         } else {
 186             topicsList = getTopicsRes.left().value();
 187         }
 188
 189         String notificationTopic = buildTopicName(deConfiguration.getDistributionNotifTopicName(), this.envName);
 190         logger.debug("Going to handle topic {}", notificationTopic);
 191
 192         boolean status = createTopicIfNotExists(topicsList, notificationTopic);
 193         if (false == status) {
 194             return false;
 195         }
 196
 197         CambriaErrorResponse registerProducerStatus = registerToTopic(notificationTopic, SubscriberTypeEnum.PRODUCER);
 198
 199         CambriaOperationStatus createStatus = registerProducerStatus.getOperationStatus();
 200
 201         if (createStatus != CambriaOperationStatus.OK) {
 202             return false;
 203         }
 204
 205         String statusTopic = buildTopicName(deConfiguration.getDistributionStatusTopicName(), this.envName);
 206         logger.debug("Going to handle topic {}", statusTopic);
 207         status = createTopicIfNotExists(topicsList, statusTopic);
 208         if (false == status) {
 209             return false;
 210         }
 211
 212         CambriaErrorResponse registerConcumerStatus = registerToTopic(statusTopic, SubscriberTypeEnum.CONSUMER);
 213
 214         if (registerConcumerStatus.getOperationStatus() != CambriaOperationStatus.OK) {
 215             return false;
 216         }
 217
 218         return true;
 219     }
 220
 221     private CambriaErrorResponse registerToTopic(String topicName, SubscriberTypeEnum subscriberType) {
 222         CambriaErrorResponse registerStatus = cambriaHandler.registerToTopic(environmentEntry.getDmaapUebAddress(), environmentEntry.getUebApikey(), environmentEntry.getUebSecretKey(), environmentEntry.getUebApikey(), subscriberType, topicName);
 223
 224         String role = CONSUMER;
 225         if (subscriberType == SubscriberTypeEnum.PRODUCER) {
 226             role = PRODUCER;
 227         }
 228         auditRegistration(topicName, registerStatus, role);
 229         return registerStatus;
 230     }
 231
 232     private void auditRegistration(String notificationTopic, CambriaErrorResponse registerProducerStatus, String role) {
 233         if (componentsUtils != null) {
 234             Integer httpCode = registerProducerStatus.getHttpCode();
 235             String httpCodeStr = String.valueOf(httpCode);
 236             this.componentsUtils.auditDistributionEngine(AuditingActionEnum.ADD_KEY_TO_TOPIC_ACL, this.envName, notificationTopic, role, environmentEntry.getUebApikey(), httpCodeStr);
 237         }
 238     }
 239
 240     private boolean createTopicIfNotExists(Set<String> topicsList, String topicName) {
 241
 242         if (topicsList.contains(topicName)) {
 243             if (componentsUtils != null) {
 244                 this.componentsUtils.auditDistributionEngine(AuditingActionEnum.CREATE_DISTRIBUTION_TOPIC, this.envName, topicName, null, null, ALREADY_EXISTS);
 245             }
 246             return true;
 247         }
 248
 249         CambriaErrorResponse createDistribTopicStatus = cambriaHandler.createTopic(environmentEntry.getDmaapUebAddress(), environmentEntry.getUebApikey(), environmentEntry.getUebSecretKey(), topicName, deConfiguration.getCreateTopic().getPartitionCount(),
 250                 deConfiguration.getCreateTopic().getReplicationCount());
 251
 252         CambriaOperationStatus status = createDistribTopicStatus.getOperationStatus();
 253         if (status == CambriaOperationStatus.TOPIC_ALREADY_EXIST) {
 254             if (componentsUtils != null) {
 255                 this.componentsUtils.auditDistributionEngine(AuditingActionEnum.CREATE_DISTRIBUTION_TOPIC, this.envName, topicName, null, null, ALREADY_EXISTS);
 256             }
 257         } else if (status == CambriaOperationStatus.OK) {
 258             if (componentsUtils != null) {
 259                 this.componentsUtils.auditDistributionEngine(AuditingActionEnum.CREATE_DISTRIBUTION_TOPIC, this.envName, topicName, null, null, CREATED);
 260             }
 261         } else {
 262             if (componentsUtils != null) {
 263                 this.componentsUtils.auditDistributionEngine(AuditingActionEnum.CREATE_DISTRIBUTION_TOPIC, this.envName, topicName, null, null, FAILED);
 264             }
 265             BeEcompErrorManager.getInstance().logBeUebSystemError(INIT_DISTRIBUTION_ENGINE_FLOW, "try to create topic " + topicName);
 266             return false;
 267         }
 268
 269         return true;
 270     }
 271
 272     public static String buildTopicName(String topicName, String environment) {
 273         return topicName + "-" + environment.toUpperCase();
 274     }
 275
 276     public boolean isActive() {
 277         return this.status.get();
 278     }
 279
 280     public long getCurrentRetryInterval() {
 281         return currentRetryInterval;
 282     }
 283
 284     protected void setCambriaHandler(CambriaHandler cambriaHandler) {
 285         this.cambriaHandler = cambriaHandler;
 286     }
 287 }