gerrit.onap Code Review - policy/clamp.git/blob

   1 /*-
   2  * ============LICENSE_START=======================================================
   3  *  Copyright (C) 2023 Nordix Foundation.
   4  * ================================================================================
   5  * Licensed under the Apache License, Version 2.0 (the "License");
   6  * you may not use this file except in compliance with the License.
   7  * You may obtain a copy of the License at
   8  *
   9  *      http://www.apache.org/licenses/LICENSE-2.0
  10  *
  11  * Unless required by applicable law or agreed to in writing, software
  12  * distributed under the License is distributed on an "AS IS" BASIS,
  13  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  14  * See the License for the specific language governing permissions and
  15  * limitations under the License.
  16  *
  17  * SPDX-License-Identifier: Apache-2.0
  18  * ============LICENSE_END=========================================================
  19  */
  20
  21 package org.onap.policy.clamp.acm.participant.intermediary.handler;
  22
  23 import java.io.Closeable;
  24 import java.io.IOException;
  25 import java.util.HashMap;
  26 import java.util.List;
  27 import java.util.Map;
  28 import java.util.UUID;
  29 import java.util.concurrent.ConcurrentHashMap;
  30 import java.util.concurrent.ExecutorService;
  31 import java.util.concurrent.Executors;
  32 import java.util.concurrent.Future;
  33 import lombok.RequiredArgsConstructor;
  34 import org.onap.policy.clamp.acm.participant.intermediary.api.AutomationCompositionElementListener;
  35 import org.onap.policy.clamp.acm.participant.intermediary.api.ParticipantIntermediaryApi;
  36 import org.onap.policy.clamp.models.acm.concepts.AcElementDeploy;
  37 import org.onap.policy.clamp.models.acm.concepts.AcElementRestart;
  38 import org.onap.policy.clamp.models.acm.concepts.AcTypeState;
  39 import org.onap.policy.clamp.models.acm.concepts.AutomationCompositionElementDefinition;
  40 import org.onap.policy.clamp.models.acm.concepts.DeployState;
  41 import org.onap.policy.clamp.models.acm.concepts.LockState;
  42 import org.onap.policy.clamp.models.acm.concepts.ParticipantRestartAc;
  43 import org.onap.policy.clamp.models.acm.concepts.StateChangeResult;
  44 import org.onap.policy.models.base.PfModelException;
  45 import org.slf4j.Logger;
  46 import org.slf4j.LoggerFactory;
  47 import org.springframework.stereotype.Component;
  48
  49 @Component
  50 @RequiredArgsConstructor
  51 public class ThreadHandler implements Closeable {
  52     private static final Logger LOGGER = LoggerFactory.getLogger(ThreadHandler.class);
  53
  54     private final AutomationCompositionElementListener listener;
  55     private final ParticipantIntermediaryApi intermediaryApi;
  56     private final CacheProvider cacheProvider;
  57
  58     private final Map<UUID, Future> executionMap = new ConcurrentHashMap<>();
  59
  60     private final ExecutorService executor = Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors());
  61
  62     /**
  63      * Handle an update on a automation composition element.
  64      *
  65      * @param messageId the messageId
  66      * @param instanceId the automationComposition Id
  67      * @param element the information on the automation composition element
  68      * @param properties properties Map
  69      */
  70     public void deploy(UUID messageId, UUID instanceId, AcElementDeploy element, Map<String, Object> properties) {
  71         cleanExecution(element.getId(), messageId);
  72         var result = executor.submit(() -> this.deployProcess(instanceId, element, properties));
  73         executionMap.put(element.getId(), result);
  74     }
  75
  76     private void deployProcess(UUID instanceId, AcElementDeploy element, Map<String, Object> properties) {
  77         try {
  78             listener.deploy(instanceId, element, properties);
  79         } catch (PfModelException e) {
  80             LOGGER.error("Automation composition element deploy failed {} {}", instanceId, e.getMessage());
  81             intermediaryApi.updateAutomationCompositionElementState(instanceId, element.getId(), DeployState.UNDEPLOYED,
  82                     null, StateChangeResult.FAILED, "Automation composition element deploy failed");
  83         }
  84         executionMap.remove(element.getId());
  85     }
  86
  87     /**
  88      * Handle a automation composition element state change.
  89      *
  90      * @param messageId the messageId
  91      * @param instanceId the automationComposition Id
  92      * @param elementId the ID of the automation composition element
  93      */
  94     public void undeploy(UUID messageId, UUID instanceId, UUID elementId) {
  95         cleanExecution(elementId, messageId);
  96         var result = executor.submit(() -> this.undeployProcess(instanceId, elementId));
  97         executionMap.put(elementId, result);
  98     }
  99
 100     private void undeployProcess(UUID instanceId, UUID elementId) {
 101         try {
 102             listener.undeploy(instanceId, elementId);
 103         } catch (PfModelException e) {
 104             LOGGER.error("Automation composition element undeploy failed {} {}", instanceId, e.getMessage());
 105             intermediaryApi.updateAutomationCompositionElementState(instanceId, elementId, DeployState.DEPLOYED, null,
 106                     StateChangeResult.FAILED, "Automation composition element undeploy failed");
 107         }
 108         executionMap.remove(elementId);
 109     }
 110
 111     /**
 112      * Handle a automation composition element lock.
 113      *
 114      * @param messageId the messageId
 115      * @param instanceId the automationComposition Id
 116      * @param elementId the ID of the automation composition element
 117      */
 118     public void lock(UUID messageId, UUID instanceId, UUID elementId) {
 119         cleanExecution(elementId, messageId);
 120         var result = executor.submit(() -> this.lockProcess(instanceId, elementId));
 121         executionMap.put(elementId, result);
 122     }
 123
 124     private void lockProcess(UUID instanceId, UUID elementId) {
 125         try {
 126             listener.lock(instanceId, elementId);
 127         } catch (PfModelException e) {
 128             LOGGER.error("Automation composition element lock failed {} {}", instanceId, e.getMessage());
 129             intermediaryApi.updateAutomationCompositionElementState(instanceId, elementId, null, LockState.UNLOCKED,
 130                     StateChangeResult.FAILED, "Automation composition element lock failed");
 131         }
 132         executionMap.remove(elementId);
 133     }
 134
 135     /**
 136      * Handle a automation composition element unlock.
 137      *
 138      * @param messageId the messageId
 139      * @param instanceId the automationComposition Id
 140      * @param elementId the ID of the automation composition element
 141      */
 142     public void unlock(UUID messageId, UUID instanceId, UUID elementId) {
 143         cleanExecution(elementId, messageId);
 144         var result = executor.submit(() -> this.unlockProcess(instanceId, elementId));
 145         executionMap.put(elementId, result);
 146     }
 147
 148     private void unlockProcess(UUID instanceId, UUID elementId) {
 149         try {
 150             listener.unlock(instanceId, elementId);
 151         } catch (PfModelException e) {
 152             LOGGER.error("Automation composition element unlock failed {} {}", instanceId, e.getMessage());
 153             intermediaryApi.updateAutomationCompositionElementState(instanceId, elementId, null, LockState.LOCKED,
 154                     StateChangeResult.FAILED, "Automation composition element unlock failed");
 155         }
 156         executionMap.remove(elementId);
 157     }
 158
 159     /**
 160      * Handle a automation composition element delete.
 161      *
 162      * @param messageId the messageId
 163      * @param instanceId the automationComposition Id
 164      * @param elementId the ID of the automation composition element
 165      */
 166     public void delete(UUID messageId, UUID instanceId, UUID elementId) {
 167         cleanExecution(elementId, messageId);
 168         var result = executor.submit(() -> this.deleteProcess(instanceId, elementId));
 169         executionMap.put(elementId, result);
 170     }
 171
 172     private void deleteProcess(UUID instanceId, UUID elementId) {
 173         try {
 174             listener.delete(instanceId, elementId);
 175         } catch (PfModelException e) {
 176             LOGGER.error("Automation composition element delete failed {} {}", instanceId, e.getMessage());
 177             intermediaryApi.updateAutomationCompositionElementState(instanceId, elementId, DeployState.UNDEPLOYED, null,
 178                     StateChangeResult.FAILED, "Automation composition element delete failed");
 179         }
 180         executionMap.remove(elementId);
 181     }
 182
 183     /**
 184      * Handle a automation composition element properties update.
 185      *
 186      * @param messageId the messageId
 187      * @param instanceId the automationComposition Id
 188      * @param element the information on the automation composition element
 189      * @param properties properties Map
 190      */
 191     public void update(UUID messageId, UUID instanceId, AcElementDeploy element, Map<String, Object> properties) {
 192         cleanExecution(element.getId(), messageId);
 193         var result = executor.submit(() -> this.updateProcess(instanceId, element, properties));
 194         executionMap.put(element.getId(), result);
 195     }
 196
 197     private void updateProcess(UUID instanceId, AcElementDeploy element, Map<String, Object> properties) {
 198         try {
 199             listener.update(instanceId, element, properties);
 200         } catch (PfModelException e) {
 201             LOGGER.error("Automation composition element update failed {} {}", instanceId, e.getMessage());
 202             intermediaryApi.updateAutomationCompositionElementState(instanceId, element.getId(), DeployState.DEPLOYED,
 203                     null, StateChangeResult.FAILED, "Automation composition element update failed");
 204         }
 205         executionMap.remove(element.getId());
 206     }
 207
 208     private void cleanExecution(UUID execIdentificationId, UUID messageId) {
 209         var process = executionMap.get(execIdentificationId);
 210         if (process != null) {
 211             if (!process.isDone()) {
 212                 process.cancel(true);
 213             }
 214             executionMap.remove(execIdentificationId);
 215         }
 216         cacheProvider.getMsgIdentification().put(execIdentificationId, messageId);
 217     }
 218
 219     /**
 220      * Handles prime a Composition Definition.
 221      *
 222      * @param messageId the messageId
 223      * @param compositionId the compositionId
 224      * @param list the list of AutomationCompositionElementDefinition
 225      */
 226     public void prime(UUID messageId, UUID compositionId, List<AutomationCompositionElementDefinition> list) {
 227         cleanExecution(compositionId, messageId);
 228         var result = executor.submit(() -> this.primeProcess(compositionId, list));
 229         executionMap.put(compositionId, result);
 230     }
 231
 232     private void primeProcess(UUID compositionId, List<AutomationCompositionElementDefinition> list) {
 233         try {
 234             listener.prime(compositionId, list);
 235             executionMap.remove(compositionId);
 236         } catch (PfModelException e) {
 237             LOGGER.error("Composition Defintion prime failed {} {}", compositionId, e.getMessage());
 238             intermediaryApi.updateCompositionState(compositionId, AcTypeState.COMMISSIONED, StateChangeResult.FAILED,
 239                     "Composition Defintion prime failed");
 240         }
 241     }
 242
 243     /**
 244      * Handles deprime a Composition Definition.
 245      *
 246      * @param messageId the messageId
 247      * @param compositionId the compositionId
 248      */
 249     public void deprime(UUID messageId, UUID compositionId) {
 250         cleanExecution(compositionId, messageId);
 251         var result = executor.submit(() -> this.deprimeProcess(compositionId));
 252         executionMap.put(compositionId, result);
 253     }
 254
 255     private void deprimeProcess(UUID compositionId) {
 256         try {
 257             listener.deprime(compositionId);
 258             executionMap.remove(compositionId);
 259         } catch (PfModelException e) {
 260             LOGGER.error("Composition Defintion deprime failed {} {}", compositionId, e.getMessage());
 261             intermediaryApi.updateCompositionState(compositionId, AcTypeState.PRIMED, StateChangeResult.FAILED,
 262                     "Composition Defintion deprime failed");
 263         }
 264     }
 265
 266     /**
 267      * Handles restarted scenario.
 268      *
 269      * @param messageId the messageId
 270      * @param compositionId the compositionId
 271      * @param list the list of AutomationCompositionElementDefinition
 272      * @param state the state of the composition
 273      * @param automationCompositionList list of ParticipantRestartAc
 274      */
 275     public void restarted(UUID messageId, UUID compositionId, List<AutomationCompositionElementDefinition> list,
 276             AcTypeState state, List<ParticipantRestartAc> automationCompositionList) {
 277         try {
 278             listener.handleRestartComposition(compositionId, list, state);
 279         } catch (PfModelException e) {
 280             LOGGER.error("Composition Defintion restarted failed {} {}", compositionId, e.getMessage());
 281             intermediaryApi.updateCompositionState(compositionId, state, StateChangeResult.FAILED,
 282                     "Composition Defintion restarted failed");
 283         }
 284
 285         for (var automationComposition : automationCompositionList) {
 286             for (var element : automationComposition.getAcElementList()) {
 287                 cleanExecution(element.getId(), messageId);
 288                 var result = executor.submit(() -> this
 289                         .restartedInstanceProcess(automationComposition.getAutomationCompositionId(), element));
 290                 executionMap.put(element.getId(), result);
 291             }
 292         }
 293     }
 294
 295     private void restartedInstanceProcess(UUID instanceId, AcElementRestart element) {
 296         try {
 297             var map = new HashMap<>(cacheProvider.getCommonProperties(instanceId, element.getId()));
 298             map.putAll(element.getProperties());
 299
 300             listener.handleRestartInstance(instanceId, getAcElementDeploy(element), map, element.getDeployState(),
 301                     element.getLockState());
 302             executionMap.remove(element.getId());
 303         } catch (PfModelException e) {
 304             LOGGER.error("Automation composition element deploy failed {} {}", instanceId, e.getMessage());
 305             intermediaryApi.updateAutomationCompositionElementState(instanceId, element.getId(),
 306                     element.getDeployState(), element.getLockState(), StateChangeResult.FAILED,
 307                     "Automation composition element restart failed");
 308         }
 309     }
 310
 311     private AcElementDeploy getAcElementDeploy(AcElementRestart element) {
 312         var acElementDeploy = new AcElementDeploy();
 313         acElementDeploy.setId(element.getId());
 314         acElementDeploy.setDefinition(element.getDefinition());
 315         acElementDeploy.setProperties(element.getProperties());
 316         acElementDeploy.setToscaServiceTemplateFragment(element.getToscaServiceTemplateFragment());
 317         return acElementDeploy;
 318     }
 319
 320     /**
 321      * Closes this stream and releases any system resources associated
 322      * with it. If the stream is already closed then invoking this
 323      * method has no effect.
 324      *
 325      * @throws IOException if an I/O error occurs
 326      */
 327     @Override
 328     public void close() throws IOException {
 329         executor.shutdown();
 330     }
 331
 332     /**
 333      * Handles AutomationComposition Migration.
 334      *
 335      * @param messageId the messageId
 336      * @param instanceId the automationComposition Id
 337      * @param element the information on the automation composition element
 338      * @param compositionTargetId the composition to migrate
 339      */
 340     public void migrate(UUID messageId, UUID instanceId, AcElementDeploy element, UUID compositionTargetId,
 341             Map<String, Object> properties) {
 342         cleanExecution(element.getId(), messageId);
 343         var result = executor.submit(() -> this.migrateProcess(instanceId, element, compositionTargetId, properties));
 344         executionMap.put(element.getId(), result);
 345     }
 346
 347     private void migrateProcess(UUID instanceId, AcElementDeploy element, UUID compositionTargetId,
 348             Map<String, Object> properties) {
 349         try {
 350             listener.migrate(instanceId, element, compositionTargetId, properties);
 351         } catch (PfModelException e) {
 352             LOGGER.error("Automation composition element migrate failed {} {}", instanceId, e.getMessage());
 353             intermediaryApi.updateAutomationCompositionElementState(instanceId, element.getId(), DeployState.DEPLOYED,
 354                     null, StateChangeResult.FAILED, "Automation composition element migrate failed");
 355         }
 356         executionMap.remove(element.getId());
 357     }
 358 }