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

   1 /*-
   2  * ============LICENSE_START=======================================================
   3  * SDC
   4  * ================================================================================
   5  * Copyright (C) 2019 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 com.att.aft.dme2.api.DME2Exception;
  24 import com.att.aft.dme2.iterator.DME2EndpointIterator;
  25 import com.att.aft.dme2.iterator.domain.DME2EndpointReference;
  26 import com.att.aft.dme2.manager.registry.DME2Endpoint;
  27 import com.att.nsa.apiClient.credentials.ApiCredential;
  28 import com.google.common.annotations.VisibleForTesting;
  29 import com.google.gson.Gson;
  30 import com.google.gson.GsonBuilder;
  31 import fj.data.Either;
  32 import org.apache.commons.lang3.StringUtils;
  33 import org.apache.http.HttpStatus;
  34 import org.openecomp.sdc.be.components.distribution.engine.IDmaapNotificationData.DmaapActionEnum;
  35 import org.openecomp.sdc.be.components.distribution.engine.IDmaapNotificationData.OperationaEnvironmentTypeEnum;
  36 import org.openecomp.sdc.be.components.distribution.engine.report.DistributionCompleteReporter;
  37 import org.openecomp.sdc.be.config.ConfigurationManager;
  38 import org.openecomp.sdc.be.config.DistributionEngineConfiguration;
  39 import org.openecomp.sdc.be.config.DmaapConsumerConfiguration;
  40 import org.openecomp.sdc.be.config.DmeConfiguration;
  41 import org.openecomp.sdc.be.dao.cassandra.CassandraOperationStatus;
  42 import org.openecomp.sdc.be.dao.cassandra.OperationalEnvironmentDao;
  43 import org.openecomp.sdc.be.datatypes.enums.EnvironmentStatusEnum;
  44 import org.openecomp.sdc.be.impl.ComponentsUtils;
  45 import org.openecomp.sdc.be.info.OperationalEnvInfo;
  46 import org.openecomp.sdc.be.resources.data.OperationalEnvironmentEntry;
  47 import org.openecomp.sdc.be.resources.data.auditing.AuditingActionEnum;
  48 import org.openecomp.sdc.common.datastructure.Wrapper;
  49 import org.openecomp.sdc.common.http.client.api.HttpResponse;
  50 import org.openecomp.sdc.common.log.wrappers.Logger;
  51 import org.springframework.stereotype.Service;
  52
  53 import javax.annotation.PostConstruct;
  54 import java.util.*;
  55 import java.util.concurrent.atomic.AtomicBoolean;
  56 import java.util.function.Function;
  57 import java.util.function.Supplier;
  58 import java.util.stream.Collectors;
  59
  60 import static org.apache.commons.lang3.StringUtils.isEmpty;
  61 import static org.openecomp.sdc.common.datastructure.FunctionalInterfaces.runMethodWithTimeOut;
  62
  63 /**
  64  * Allows to consume DMAAP topic and handle received notifications
  65  */
  66 @Service
  67 public class EnvironmentsEngine implements INotificationHandler {
  68
  69     private static final String MESSAGE_BUS = "MessageBus";
  70     private static final String UNKNOWN = "Unknown";
  71     private static final Logger log = Logger.getLogger(EnvironmentsEngine.class.getName());
  72     private ConfigurationManager configurationManager = ConfigurationManager.getConfigurationManager();
  73
  74     private Map<String, OperationalEnvironmentEntry> environments = new HashMap<>();
  75     private Map<String, AtomicBoolean> envNamePerStatus = new HashMap<>();
  76     private Map<String, DistributionEnginePollingTask> envNamePerPollingTask = new HashMap<>();
  77     private Map<String, DistributionEngineInitTask> envNamePerInitTask = new HashMap<>();
  78
  79     private final DmaapConsumer dmaapConsumer;
  80     private final OperationalEnvironmentDao operationalEnvironmentDao;
  81     private final DME2EndpointIteratorCreator epIterCreator;
  82     private final AaiRequestHandler aaiRequestHandler;
  83     private final ComponentsUtils componentUtils;
  84     private final CambriaHandler cambriaHandler;
  85     private final DistributionEngineClusterHealth distributionEngineClusterHealth;
  86     private final DistributionCompleteReporter distributionCompleteReporter;
  87
  88     public EnvironmentsEngine(DmaapConsumer dmaapConsumer, OperationalEnvironmentDao operationalEnvironmentDao, DME2EndpointIteratorCreator epIterCreator, AaiRequestHandler aaiRequestHandler, ComponentsUtils componentUtils, CambriaHandler cambriaHandler, DistributionEngineClusterHealth distributionEngineClusterHealth, DistributionCompleteReporter distributionCompleteReporter) {
  89         this.dmaapConsumer = dmaapConsumer;
  90         this.operationalEnvironmentDao = operationalEnvironmentDao;
  91         this.epIterCreator = epIterCreator;
  92         this.aaiRequestHandler = aaiRequestHandler;
  93         this.componentUtils = componentUtils;
  94         this.cambriaHandler = cambriaHandler;
  95         this.distributionEngineClusterHealth = distributionEngineClusterHealth;
  96         this.distributionCompleteReporter = distributionCompleteReporter;
  97     }
  98
  99     @VisibleForTesting
 100     @PostConstruct
 101     void init() {
 102         try {
 103             environments = populateEnvironments();
 104             createUebTopicsForEnvironments();
 105             initDmeGlobalConfig();
 106             if(!configurationManager.getConfiguration().getDmaapConsumerConfiguration().isActive()){
 107                 log.info("Environments engine is disabled");
 108                 return;
 109             }
 110             dmaapConsumer.consumeDmaapTopic(this::handleMessage,
 111                 (t, e) -> log.error("An error occurred upon consuming topic by Dmaap consumer client: ", e));
 112             log.info("Environments engine has been initialized.");
 113         } catch (Exception e) {
 114             log.error("An error occurred upon consuming topic by Dmaap consumer client.", e);
 115         }
 116     }
 117
 118     private void initDmeGlobalConfig() {
 119         DmaapConsumerConfiguration dmaapConsumerParams = ConfigurationManager.getConfigurationManager().getConfiguration().getDmaapConsumerConfiguration();
 120         if (dmaapConsumerParams == null) {
 121             log.warn("cannot read dmaap configuration file,DME might not be initialized properly");
 122             return;
 123         }
 124         System.setProperty("AFT_ENVIRONMENT", dmaapConsumerParams.getEnvironment()); // AFTPRD for production
 125         System.setProperty("AFT_LATITUDE", dmaapConsumerParams.getLatitude()!=null ? dmaapConsumerParams.getLatitude().toString() : "1.0"); // Replace with actual latitude
 126         System.setProperty("AFT_LONGITUDE", dmaapConsumerParams.getLongitude()!=null ? dmaapConsumerParams.getLongitude().toString() : "1.0"); // Replace with actual longitude
 127     }
 128
 129     public void connectUebTopicTenantIsolation(OperationalEnvironmentEntry opEnvEntry,
 130                                                AtomicBoolean status,
 131                                                Map<String, DistributionEngineInitTask> envNamePerInitTask, Map<String, DistributionEnginePollingTask> envNamePerPollingTask) {
 132         connectUebTopic(opEnvEntry, status, envNamePerInitTask, envNamePerPollingTask);
 133
 134     }
 135
 136     public void connectUebTopicForDistributionConfTopic(String envName,
 137                                                         AtomicBoolean status,
 138                                                         Map<String, DistributionEngineInitTask> envNamePerInitTask, Map<String, DistributionEnginePollingTask> envNamePerPollingTask) {
 139         connectUebTopic(environments.get(envName), status, envNamePerInitTask, envNamePerPollingTask);
 140
 141     }
 142
 143     /**
 144      * Allows to create and run UEB initializing and polling tasks
 145      *
 146      * @param status
 147      * @param envNamePerInitTask
 148      * @param envNamePerPollingTask
 149      * @param opEnvEntry
 150      */
 151     private void connectUebTopic(OperationalEnvironmentEntry opEnvEntry, AtomicBoolean status,
 152                                  Map<String, DistributionEngineInitTask> envNamePerInitTask,
 153                                  Map<String, DistributionEnginePollingTask> envNamePerPollingTask) {
 154
 155         String envId = opEnvEntry.getEnvironmentId();
 156
 157         DistributionEngineConfiguration distributionEngineConfiguration = ConfigurationManager.getConfigurationManager()
 158                 .getDistributionEngineConfiguration();
 159         DistributionEnginePollingTask distributionEnginePollingTask = new DistributionEnginePollingTask(
 160                 distributionEngineConfiguration, distributionCompleteReporter, componentUtils, distributionEngineClusterHealth,
 161                 opEnvEntry);
 162         String envName = configurationManager.getDistributionEngineConfiguration().getEnvironments().get(0);
 163         DistributionEngineInitTask distributionEngineInitTask = new DistributionEngineInitTask(0l,
 164                 distributionEngineConfiguration, envName, status, componentUtils, distributionEnginePollingTask,
 165                 opEnvEntry);
 166         distributionEngineInitTask.startTask();
 167         envNamePerInitTask.put(envId, distributionEngineInitTask);
 168         envNamePerPollingTask.put(envId, distributionEnginePollingTask);
 169
 170         log.debug("Environment envId = {} has been connected to the UEB topic", envId);
 171     }
 172
 173     @Override
 174     public boolean handleMessage(String notification) {
 175         DmaapConsumerConfiguration dmaapConsumerParams = ConfigurationManager.getConfigurationManager()
 176                 .getConfiguration().getDmaapConsumerConfiguration();
 177         Supplier<Boolean> supplier = () -> handleMessageLogic(notification);
 178         Either<Boolean, Boolean> eitherTimeOut = runMethodWithTimeOut(supplier,
 179                 dmaapConsumerParams.getTimeLimitForNotificationHandleMs());
 180
 181         boolean result;
 182         if (eitherTimeOut.isRight()) {
 183             result = false;
 184         } else {
 185             result = eitherTimeOut.left().value();
 186         }
 187         return result;
 188     }
 189
 190     public boolean handleMessageLogic(String notification) {
 191         Wrapper<Boolean> errorWrapper = new Wrapper<>();
 192         Wrapper<OperationalEnvironmentEntry> opEnvEntryWrapper = new Wrapper<>();
 193         try {
 194
 195             log.debug("handle message - for operational environment notification received: {}", notification);
 196             Gson gsonObj = new GsonBuilder().create();
 197
 198             IDmaapNotificationData notificationData = gsonObj.fromJson(notification,
 199                     DmaapNotificationDataImpl.class);
 200             IDmaapAuditNotificationData auditNotificationData = gsonObj.fromJson(notification,
 201                     DmaapNotificationDataImpl.class);
 202
 203             AuditingActionEnum actionEnum;
 204             switch (notificationData.getAction()) {
 205                 case CREATE:
 206                     actionEnum = AuditingActionEnum.CREATE_ENVIRONMENT;
 207                     break;
 208                 case UPDATE:
 209                     actionEnum = AuditingActionEnum.UPDATE_ENVIRONMENT;
 210                     break;
 211                 case DELETE:
 212                     actionEnum = AuditingActionEnum.DELETE_ENVIRONMENT;
 213                     break;
 214                 default:
 215                     actionEnum = AuditingActionEnum.UNKNOWN_ENVIRONMENT_NOTIFICATION;
 216                     break;
 217             }
 218             componentUtils.auditEnvironmentEngine(actionEnum,
 219                     notificationData.getOperationalEnvironmentId(), notificationData.getOperationalEnvironmentType().getEventTypenName(),
 220                     notificationData.getAction().getActionName(), auditNotificationData.getOperationalEnvironmentName(),
 221                     auditNotificationData.getTenantContext());
 222
 223             if (errorWrapper.isEmpty()) {
 224                 validateNotification(errorWrapper, notificationData, auditNotificationData);
 225             }
 226             // Perform Save In-Progress Dao
 227             if (errorWrapper.isEmpty()) {
 228                 saveEntryWithInProgressStatus(errorWrapper, opEnvEntryWrapper, notificationData);
 229             }
 230
 231             if (errorWrapper.isEmpty()) {
 232                 buildOpEnv(errorWrapper, opEnvEntryWrapper.getInnerElement());
 233             }
 234
 235         } catch (Exception e) {
 236             log.debug("handle message for operational environmet failed for notification: {} with error :{}",
 237                     notification, e.getMessage(), e);
 238             errorWrapper.setInnerElement(false);
 239
 240         }
 241         return errorWrapper.isEmpty();
 242     }
 243
 244     private void validateNotification(Wrapper<Boolean> errorWrapper, IDmaapNotificationData notificationData,
 245                                       IDmaapAuditNotificationData auditNotificationData) {
 246         // Check OperationaEnvironmentType
 247         if (errorWrapper.isEmpty()) {
 248             validateEnvironmentType(errorWrapper, notificationData, auditNotificationData);
 249         }
 250         // Check Action Type
 251         if (errorWrapper.isEmpty()) {
 252             validateActionType(errorWrapper, notificationData);
 253         }
 254         // Check is valid for create/update (not In-Progress state)
 255         if (errorWrapper.isEmpty()) {
 256             validateState(errorWrapper, notificationData);
 257         }
 258     }
 259
 260     public void buildOpEnv(Wrapper<Boolean> errorWrapper, OperationalEnvironmentEntry opEnvEntry) {
 261         // Get Env Info From A&AI
 262         if (errorWrapper.isEmpty()) {
 263             retrieveOpEnvInfoFromAAI(errorWrapper, opEnvEntry);
 264         }
 265
 266         if (errorWrapper.isEmpty()) {
 267             // Get List Of UEB Addresses From AFT_DME
 268             retrieveUebAddressesFromAftDme(errorWrapper, opEnvEntry);
 269         }
 270
 271         // Create UEB keys and set them on EnvEntry
 272         if (errorWrapper.isEmpty()) {
 273             createUebKeys(errorWrapper, opEnvEntry);
 274         }
 275
 276         // Create Topics
 277         if (errorWrapper.isEmpty()) {
 278             log.debug("handle message - Create Topics");
 279             createUebTopicsForEnvironment(opEnvEntry);
 280         }
 281
 282         // Save Status Complete and Add to Map
 283         if (errorWrapper.isEmpty()) {
 284             saveEntryWithCompleteStatus(errorWrapper, opEnvEntry);
 285         }
 286
 287         // Update Environments Map
 288         if (errorWrapper.isEmpty()) {
 289             environments.put(opEnvEntry.getEnvironmentId(), opEnvEntry);
 290         } else {
 291             saveEntryWithFailedStatus(errorWrapper, opEnvEntry);
 292         }
 293     }
 294
 295     private void saveEntryWithFailedStatus(Wrapper<Boolean> errorWrapper, OperationalEnvironmentEntry opEnvEntry) {
 296         log.debug("handle message - save OperationalEnvironment Failed Status");
 297         opEnvEntry.setStatus(EnvironmentStatusEnum.FAILED);
 298         saveOpEnvEntry(errorWrapper, opEnvEntry);
 299     }
 300
 301     void saveEntryWithCompleteStatus(Wrapper<Boolean> errorWrapper, OperationalEnvironmentEntry opEnvEntry) {
 302         log.debug("handle message - save OperationalEnvironment Complete Dao");
 303         opEnvEntry.setStatus(EnvironmentStatusEnum.COMPLETED);
 304         saveOpEnvEntry(errorWrapper, opEnvEntry);
 305
 306     }
 307
 308     void retrieveUebAddressesFromAftDme(Wrapper<Boolean> errorWrapper, OperationalEnvironmentEntry opEnvEntry) {
 309         log.debug("handle message - Get List Of UEB Addresses From AFT_DME");
 310         try {
 311             boolean isKeyFieldsValid = !isEmpty(opEnvEntry.getTenant()) && !isEmpty(opEnvEntry.getEcompWorkloadContext());
 312             if (isKeyFieldsValid) {
 313                 String opEnvKey = map2OpEnvKey(opEnvEntry);
 314                 String environmentId = opEnvEntry.getEnvironmentId();
 315                 List<String> uebHosts = discoverUebHosts(opEnvKey, environmentId);
 316                 opEnvEntry.setDmaapUebAddress(uebHosts.stream().collect(Collectors.toSet()));
 317             } else {
 318                 errorWrapper.setInnerElement(false);
 319                 log.debug("Can Not Build AFT DME Key from workLoad & Tenant Fields.");
 320             }
 321
 322         } catch (DME2Exception e) {
 323             errorWrapper.setInnerElement(false);
 324             log.error("Failed to retrieve Ueb Addresses From DME. ", e);
 325         }
 326     }
 327
 328     void createUebKeys(Wrapper<Boolean> errorWrapper, OperationalEnvironmentEntry opEnvEntry) {
 329         log.debug("handle message - Create UEB keys");
 330         List<String> discoverEndPoints = opEnvEntry.getDmaapUebAddress().stream()
 331                 .collect(Collectors.toList());
 332         Either<ApiCredential, CambriaErrorResponse> eitherCreateUebKeys = cambriaHandler
 333                 .createUebKeys(discoverEndPoints);
 334         if (eitherCreateUebKeys.isRight()) {
 335             errorWrapper.setInnerElement(false);
 336             log.debug("handle message - failed to create UEB Keys");
 337         } else {
 338             ApiCredential apiCredential = eitherCreateUebKeys.left().value();
 339             opEnvEntry.setUebApikey(apiCredential.getApiKey());
 340             opEnvEntry.setUebSecretKey(apiCredential.getApiSecret());
 341         }
 342     }
 343
 344     void retrieveOpEnvInfoFromAAI(Wrapper<Boolean> errorWrapper, OperationalEnvironmentEntry opEnvEntry) {
 345         log.debug("handle message - Get Env Info From A&AI");
 346         Either<OperationalEnvInfo, Integer> eitherOperationalEnvInfo = getOperationalEnvById(
 347                 opEnvEntry.getEnvironmentId());
 348         if (eitherOperationalEnvInfo.isRight()) {
 349             errorWrapper.setInnerElement(false);
 350             log.debug("handle message - failed to retrieve details from A&AI");
 351         } else {
 352             OperationalEnvInfo operationalEnvInfo = eitherOperationalEnvInfo.left().value();
 353             opEnvEntry.setEcompWorkloadContext(operationalEnvInfo.getWorkloadContext());
 354             opEnvEntry.setTenant(operationalEnvInfo.getTenantContext());
 355         }
 356     }
 357
 358     void saveEntryWithInProgressStatus(Wrapper<Boolean> errorWrapper, Wrapper<OperationalEnvironmentEntry> opEnvEntryWrapper, IDmaapNotificationData notificationData) {
 359         log.debug("handle message - save OperationalEnvironment In-Progress Dao");
 360         OperationalEnvironmentEntry opEnvEntry = new OperationalEnvironmentEntry();
 361         // Entry Environment ID holds actually the environment NAME
 362         opEnvEntry.setEnvironmentId(notificationData.getOperationalEnvironmentId());
 363         opEnvEntry.setStatus(EnvironmentStatusEnum.IN_PROGRESS);
 364         opEnvEntry.setIsProduction(false);
 365         saveOpEnvEntry(errorWrapper, opEnvEntry);
 366         opEnvEntryWrapper.setInnerElement(opEnvEntry);
 367
 368     }
 369
 370
 371     void validateState(Wrapper<Boolean> errorWrapper, IDmaapNotificationData notificationData) {
 372         log.debug("handle message - verify OperationalEnvironment not In-Progress");
 373         String opEnvId = notificationData.getOperationalEnvironmentId();
 374
 375         Either<OperationalEnvironmentEntry, CassandraOperationStatus> eitherOpEnv = operationalEnvironmentDao
 376                 .get(opEnvId);
 377         if (eitherOpEnv.isLeft()) {
 378             final OperationalEnvironmentEntry opEnvEntry = eitherOpEnv.left().value();
 379             if (StringUtils.equals(opEnvEntry.getStatus(), EnvironmentStatusEnum.IN_PROGRESS.getName())) {
 380                 errorWrapper.setInnerElement(false);
 381                 log.debug("handle message - validate State Failed Record Found With Status : {} Flow Stopped!", opEnvEntry.getStatus());
 382             }
 383         } else {
 384             CassandraOperationStatus operationStatus = eitherOpEnv.right().value();
 385             if (operationStatus != CassandraOperationStatus.NOT_FOUND) {
 386                 errorWrapper.setInnerElement(false);
 387                 log.debug("failed to retrieve operationa environment with id:{} cassandra error was :{}", opEnvId,
 388                         operationStatus);
 389             }
 390         }
 391
 392     }
 393
 394     void validateActionType(Wrapper<Boolean> errorWrapper, IDmaapNotificationData notificationData) {
 395         log.debug("handle message - verify Action Type");
 396         DmaapActionEnum action = notificationData.getAction();
 397         if (action == DmaapActionEnum.DELETE) {
 398             errorWrapper.setInnerElement(false);
 399             log.debug("handle message - validate Action Type Failed With Action Type: {} Flow Stopped!", action);
 400         }
 401     }
 402
 403     void validateEnvironmentType(Wrapper<Boolean> errorWrapper, IDmaapNotificationData notificationData,
 404                                  IDmaapAuditNotificationData auditNotificationData) {
 405         log.debug("handle message - verify OperationaEnvironmentType");
 406         OperationaEnvironmentTypeEnum envType = notificationData.getOperationalEnvironmentType();
 407         if (envType != OperationaEnvironmentTypeEnum.ECOMP) {
 408             errorWrapper.setInnerElement(false);
 409             log.debug("handle message - validate Environment Type Failed With Environment Type: {} Flow Stopped!", envType);
 410             componentUtils.auditEnvironmentEngine(AuditingActionEnum.UNSUPPORTED_ENVIRONMENT_TYPE,
 411                     notificationData.getOperationalEnvironmentId(), notificationData.getOperationalEnvironmentType().getEventTypenName(),
 412                     notificationData.getAction().getActionName(), auditNotificationData.getOperationalEnvironmentName(),
 413                     auditNotificationData.getTenantContext());
 414         }
 415     }
 416
 417
 418     private void saveOpEnvEntry(Wrapper<Boolean> errorWrapper, OperationalEnvironmentEntry entry) {
 419         entry.setLastModified(new Date(System.currentTimeMillis()));
 420         CassandraOperationStatus saveStaus = operationalEnvironmentDao.save(entry);
 421         if (saveStaus != CassandraOperationStatus.OK) {
 422             errorWrapper.setInnerElement(false);
 423             log.debug("handle message saving  operational environmet failed for id :{} with error : {}",
 424                     entry.getEnvironmentId(), saveStaus);
 425         }
 426     }
 427
 428     public List<String> discoverUebHosts(String opEnvKey, String env) throws DME2Exception {
 429         DmeConfiguration dmeConfiguration = configurationManager.getConfiguration().getDmeConfiguration();
 430         List<String> uebHosts = new LinkedList<>();
 431
 432         String lookupURI = String.format("http://%s/service=%s/version=1.0.0/envContext=%s/partner=*", dmeConfiguration.getDme2Search(), opEnvKey,
 433                 env);
 434         DME2EndpointIterator iterator = epIterCreator.create(lookupURI);
 435
 436         // Beginning iteration
 437         while (iterator.hasNext()) {
 438             DME2EndpointReference ref = iterator.next();
 439             DME2Endpoint dmeEndpoint = ref.getEndpoint();
 440             log.debug("DME returns EP with UEB host {}, UEB port: {}", dmeEndpoint.getHost(), dmeEndpoint.getPort());
 441             uebHosts.add(dmeEndpoint.getHost());
 442         }
 443
 444         return uebHosts;
 445     }
 446
 447     private String map2OpEnvKey(OperationalEnvironmentEntry entry) {
 448         return String.format("%s.%s.%s", entry.getTenant(), entry.getEcompWorkloadContext(), MESSAGE_BUS);
 449     }
 450
 451     private Map<String, OperationalEnvironmentEntry> populateEnvironments() {
 452         Map<String, OperationalEnvironmentEntry> envs = getEnvironmentsFromDb();
 453         OperationalEnvironmentEntry confEntry = readEnvFromConfig();
 454         envs.put(confEntry.getEnvironmentId(), confEntry);
 455         return envs;
 456     }
 457
 458     private OperationalEnvironmentEntry readEnvFromConfig() {
 459         OperationalEnvironmentEntry entry = new OperationalEnvironmentEntry();
 460         DistributionEngineConfiguration distributionEngineConfiguration = configurationManager
 461                 .getDistributionEngineConfiguration();
 462         entry.setUebApikey(distributionEngineConfiguration.getUebPublicKey());
 463         entry.setUebSecretKey(distributionEngineConfiguration.getUebSecretKey());
 464
 465         Set<String> puebEndpoints = new HashSet<>();
 466         puebEndpoints.addAll(distributionEngineConfiguration.getUebServers());
 467         entry.setDmaapUebAddress(puebEndpoints);
 468
 469         String envName = distributionEngineConfiguration.getEnvironments().size() == 1
 470                 ? distributionEngineConfiguration.getEnvironments().get(0) : UNKNOWN;
 471         entry.setEnvironmentId(envName);
 472
 473         if (log.isDebugEnabled()) {
 474             log.debug("Enviroment read from configuration: {}", entry);
 475         }
 476
 477         return entry;
 478     }
 479
 480     private Map<String, OperationalEnvironmentEntry> getEnvironmentsFromDb() {
 481         Either<List<OperationalEnvironmentEntry>, CassandraOperationStatus> opEnvResult = operationalEnvironmentDao
 482                 .getByEnvironmentsStatus(EnvironmentStatusEnum.COMPLETED);
 483
 484         if (opEnvResult.isLeft()) {
 485             Map<String, OperationalEnvironmentEntry> resultMap = opEnvResult.left().value().stream()
 486                     .collect(Collectors.toMap(OperationalEnvironmentEntry::getEnvironmentId, Function.identity()));
 487             resultMap.forEach((key, value) -> log.debug("Enviroment loaded from DB: {}", value));
 488             return resultMap;
 489         } else {
 490             CassandraOperationStatus status = opEnvResult.right().value();
 491             log.debug("Failed to populate Operation Envirenments Map from Cassandra, DB status: {}", status);
 492             return new HashMap<>();
 493         }
 494     }
 495
 496     void createUebTopicsForEnvironments() {
 497         environments.values().forEach(this::createUebTopicsForEnvironment);
 498     }
 499
 500     public void createUebTopicsForEnvironment(OperationalEnvironmentEntry opEnvEntry) {
 501         String envId = opEnvEntry.getEnvironmentId();
 502         log.debug("Create Environment {} on UEB Topic.", envId);
 503         AtomicBoolean status = new AtomicBoolean(false);
 504         envNamePerStatus.put(envId, status);
 505
 506         connectUebTopicTenantIsolation(opEnvEntry, status, envNamePerInitTask, envNamePerPollingTask);
 507     }
 508
 509     @VisibleForTesting
 510     void setConfigurationManager(ConfigurationManager configurationManager) {
 511         this.configurationManager = configurationManager;
 512     }
 513
 514     public Map<String, OperationalEnvironmentEntry> getEnvironments() {
 515         return environments;
 516     }
 517
 518
 519     public Either<OperationalEnvInfo, Integer> getOperationalEnvById(String id) {
 520         HttpResponse<String> resp = aaiRequestHandler.getOperationalEnvById(id);
 521         if (resp.getStatusCode() == HttpStatus.SC_OK) {
 522             try {
 523                 OperationalEnvInfo operationalEnvInfo = OperationalEnvInfo.createFromJson(resp.getResponse());
 524
 525                 log.debug("Get \"{}\" operational environment. {}", id, operationalEnvInfo);
 526                 return Either.left(operationalEnvInfo);
 527             } catch (Exception e) {
 528                 log.debug("Json convert to OperationalEnvInfo failed with exception ", e);
 529                 return Either.right(HttpStatus.SC_INTERNAL_SERVER_ERROR);
 530             }
 531         } else {
 532             log.debug("Get \"{}\" operational environment failed with statusCode: {}, description: {}", id,
 533                     resp.getStatusCode(), resp.getDescription());
 534             return Either.right(resp.getStatusCode());
 535         }
 536     }
 537
 538     public OperationalEnvironmentEntry getEnvironmentById(String envId) {
 539         return environments.get(envId);
 540     }
 541
 542     public boolean isInMap(OperationalEnvironmentEntry env) {
 543         return isInMap(env.getEnvironmentId());
 544     }
 545
 546     public boolean isInMap(String envId) {
 547         return environments.containsKey(envId);
 548     }
 549
 550     public void addToMap(OperationalEnvironmentEntry opEnvEntry) {
 551         environments.put(opEnvEntry.getEnvironmentId(), opEnvEntry);
 552
 553     }
 554 }