gerrit.onap Code Review - sdc.git/blob

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