feat(operations): Replace KatoService with OperationsRunner in server…

… group tasks

(cherry picked from commit 6dd8ec125f1c4ac9be040fd2c6cb2963692bb208)

orca-api/src/main/java/com/netflix/spinnaker/orca/api/operations/OperationsInput.java

@@ -32,4 +32,20 @@ public class OperationsInput {
   public boolean hasCloudProvider() {
     return this.cloudProvider != null && !this.cloudProvider.isEmpty();
   }
+
+  public static OperationsInput of(
+      String cloudProvider,
+      Collection<? extends Map<String, Map>> operations,
+      StageExecution stageExecution) {
+    return builder()
+        .cloudProvider(cloudProvider)
+        .operations(operations)
+        .stageExecution(stageExecution)
+        .build();
+  }
+
+  public static OperationsInput of(
+      Collection<? extends Map<String, Map>> operations, StageExecution stageExecution) {
+    return of(null, operations, stageExecution);
+  }
 }

orca-clouddriver/src/main/groovy/com/netflix/spinnaker/orca/clouddriver/tasks/servergroup/AbstractServerGroupTask.groovy

@@ -17,12 +17,14 @@
 package com.netflix.spinnaker.orca.clouddriver.tasks.servergroup
 
 import com.netflix.spinnaker.moniker.Moniker
+import com.netflix.spinnaker.orca.api.operations.OperationsContext
+import com.netflix.spinnaker.orca.api.operations.OperationsInput
+import com.netflix.spinnaker.orca.api.operations.OperationsRunner
 import com.netflix.spinnaker.orca.api.pipeline.models.ExecutionStatus
 import com.netflix.spinnaker.kork.core.RetrySupport
 import com.netflix.spinnaker.orca.api.pipeline.RetryableTask
 import com.netflix.spinnaker.orca.api.pipeline.models.StageExecution
 import com.netflix.spinnaker.orca.api.pipeline.TaskResult
-import com.netflix.spinnaker.orca.clouddriver.KatoService
 import com.netflix.spinnaker.orca.clouddriver.pipeline.servergroup.support.Location
 import com.netflix.spinnaker.orca.clouddriver.pipeline.servergroup.support.Location.Type
 import com.netflix.spinnaker.orca.clouddriver.pipeline.servergroup.support.TargetServerGroup
@@ -45,7 +47,7 @@ abstract class AbstractServerGroupTask implements CloudProviderAware, RetryableT
   }
 
   @Autowired
-  KatoService kato
+  OperationsRunner operationsRunner
 
   @Autowired
   RetrySupport retrySupport
@@ -79,15 +81,16 @@ abstract class AbstractServerGroupTask implements CloudProviderAware, RetryableT
       return TaskResult.ofStatus(ExecutionStatus.SUCCEEDED)
     }
 
-    def taskId = kato.requestOperations(cloudProvider, [[(serverGroupAction): operation]])
+    OperationsInput operationsInput = OperationsInput.of(cloudProvider, [[(serverGroupAction): operation]], stage)
+    OperationsContext operationsContext = operationsRunner.run(operationsInput)
 
     def stageOutputs = [
-        "notification.type"   : serverGroupAction.toLowerCase(),
-        "kato.last.task.id"   : taskId,
-        "deploy.account.name" : account,
-        "asgName"             : operation.serverGroupName,
-        "serverGroupName"     : operation.serverGroupName,
-        "deploy.server.groups": deployServerGroups(operation)
+        "notification.type"             : serverGroupAction.toLowerCase(),
+        (operationsContext.contextKey()): operationsContext.contextValue(),
+        "deploy.account.name"           : account,
+        "asgName"                       : operation.serverGroupName,
+        "serverGroupName"               : operation.serverGroupName,
+        "deploy.server.groups"          : deployServerGroups(operation)
     ]
     if (addTargetOpOutputs) {
       stageOutputs = stageOutputs + [

orca-clouddriver/src/main/groovy/com/netflix/spinnaker/orca/clouddriver/tasks/servergroup/AbstractUpdateLaunchSettingsTask.groovy

@@ -16,23 +16,25 @@
 
 package com.netflix.spinnaker.orca.clouddriver.tasks.servergroup
 
+import com.netflix.spinnaker.orca.api.operations.OperationsContext
+import com.netflix.spinnaker.orca.api.operations.OperationsInput
+import com.netflix.spinnaker.orca.api.operations.OperationsRunner
 import com.netflix.spinnaker.orca.api.pipeline.models.ExecutionStatus
 import com.netflix.spinnaker.orca.api.pipeline.Task
 import com.netflix.spinnaker.orca.api.pipeline.models.StageExecution
 import com.netflix.spinnaker.orca.api.pipeline.TaskResult
-import com.netflix.spinnaker.orca.clouddriver.KatoService
-import com.netflix.spinnaker.orca.clouddriver.model.TaskId
 import com.netflix.spinnaker.orca.clouddriver.utils.CloudProviderAware
 import com.netflix.spinnaker.orca.kato.tasks.DeploymentDetailsAware
 import groovy.util.logging.Slf4j
 import javax.annotation.Nonnull
 
 @Slf4j
 abstract class AbstractUpdateLaunchSettingsTask implements Task, DeploymentDetailsAware, CloudProviderAware {
-  KatoService kato
+  OperationsRunner operationsRunner
   String defaultBakeAccount
-  AbstractUpdateLaunchSettingsTask(KatoService kato, String defaultBakeAccount) {
-    this.kato = kato
+
+  AbstractUpdateLaunchSettingsTask(OperationsRunner operationsRunner, String defaultBakeAccount) {
+    this.operationsRunner = operationsRunner
     this.defaultBakeAccount = defaultBakeAccount
   }
 
@@ -49,10 +51,11 @@ abstract class AbstractUpdateLaunchSettingsTask implements Task, DeploymentDetai
       ops = [[(getOperation()): stage.context]]
     }
 
-    def taskId = kato.requestOperations(cloudProvider, ops)
+    OperationsInput operationsInput = OperationsInput.of(cloudProvider, ops, stage)
+    OperationsContext operationsContext = operationsRunner.run(operationsInput)
 
     return TaskResult.builder(ExecutionStatus.SUCCEEDED)
-      .context(getContext(stage, taskId))
+      .context(getContext(stage, operationsContext))
       .build()
   }
 
@@ -92,5 +95,5 @@ abstract class AbstractUpdateLaunchSettingsTask implements Task, DeploymentDetai
   }
 
   abstract String getOperation()
-  abstract Map<String, Object> getContext(StageExecution stage, TaskId taskId)
+  abstract Map<String, Object> getContext(StageExecution stage, OperationsContext operationsContext)
 }

orca-clouddriver/src/main/groovy/com/netflix/spinnaker/orca/clouddriver/tasks/servergroup/AddServerGroupEntityTagsTask.groovy

@@ -16,15 +16,16 @@
 
 package com.netflix.spinnaker.orca.clouddriver.tasks.servergroup
 
+import com.netflix.spinnaker.orca.api.operations.OperationsContext
+import com.netflix.spinnaker.orca.api.operations.OperationsInput
+import com.netflix.spinnaker.orca.api.operations.OperationsRunner
 import com.netflix.spinnaker.orca.api.pipeline.models.StageExecution
 
 import java.util.concurrent.TimeUnit
 import com.fasterxml.jackson.annotation.JsonProperty
 import com.netflix.spinnaker.orca.api.pipeline.models.ExecutionStatus
 import com.netflix.spinnaker.orca.api.pipeline.RetryableTask
 import com.netflix.spinnaker.orca.api.pipeline.TaskResult
-import com.netflix.spinnaker.orca.clouddriver.KatoService
-import com.netflix.spinnaker.orca.clouddriver.model.TaskId
 import com.netflix.spinnaker.orca.clouddriver.utils.CloudProviderAware
 
 import groovy.util.logging.Slf4j
@@ -38,7 +39,7 @@ class AddServerGroupEntityTagsTask implements CloudProviderAware, RetryableTask
   long timeout = TimeUnit.MINUTES.toMillis(15)
 
   @Autowired
-  KatoService kato
+  OperationsRunner operationsRunner
 
   @Autowired
   Collection<ServerGroupEntityTagGenerator> tagGenerators
@@ -50,11 +51,14 @@ class AddServerGroupEntityTagsTask implements CloudProviderAware, RetryableTask
       if (!tagOperations) {
         return TaskResult.ofStatus(ExecutionStatus.SUCCEEDED)
       }
-      TaskId taskId = kato.requestOperations(tagOperations)
+
+      OperationsInput operationsInput = OperationsInput.of(tagOperations, stage)
+      OperationsContext operationsContext = operationsRunner.run(operationsInput)
+
       return TaskResult.builder(ExecutionStatus.SUCCEEDED).context(new HashMap<String, Object>() {
         {
           put("notification.type", "upsertentitytags")
-          put("kato.last.task.id", taskId)
+          put(operationsContext.contextKey(), operationsContext.contextValue())
         }
       }).build()
     } catch (Exception e) {

orca-clouddriver/src/main/groovy/com/netflix/spinnaker/orca/clouddriver/tasks/servergroup/CloneServerGroupTask.groovy

@@ -17,11 +17,13 @@
 package com.netflix.spinnaker.orca.clouddriver.tasks.servergroup
 
 import com.fasterxml.jackson.databind.ObjectMapper
+import com.netflix.spinnaker.orca.api.operations.OperationsContext
+import com.netflix.spinnaker.orca.api.operations.OperationsInput
+import com.netflix.spinnaker.orca.api.operations.OperationsRunner
 import com.netflix.spinnaker.orca.api.pipeline.models.ExecutionStatus
 import com.netflix.spinnaker.orca.api.pipeline.Task
 import com.netflix.spinnaker.orca.api.pipeline.models.StageExecution
 import com.netflix.spinnaker.orca.api.pipeline.TaskResult
-import com.netflix.spinnaker.orca.clouddriver.KatoService
 import com.netflix.spinnaker.orca.clouddriver.utils.CloudProviderAware
 
 import com.netflix.spinnaker.orca.clouddriver.tasks.servergroup.clone.CloneDescriptionDecorator
@@ -40,7 +42,7 @@ class CloneServerGroupTask implements CloudProviderAware, Task, DeploymentDetail
   Collection<CloneDescriptionDecorator> cloneDescriptionDecorators = []
 
   @Autowired
-  KatoService kato
+  OperationsRunner operationsRunner
 
   @Autowired
   ObjectMapper mapper
@@ -65,13 +67,15 @@ class CloneServerGroupTask implements CloudProviderAware, Task, DeploymentDetail
     }
 
     String credentials = getCredentials(stage)
-    def taskId = kato.requestOperations(cloudProvider, getDescriptions(stage, operation))
+
+    OperationsInput operationsInput = OperationsInput.of(cloudProvider, getDescriptions(stage, operation), stage)
+    OperationsContext operationsContext = operationsRunner.run(operationsInput)
 
     def outputs = [
-      "notification.type"   : "createcopylastasg",
-      "kato.result.expected": true,
-      "kato.last.task.id"   : taskId,
-      "deploy.account.name" : credentials,
+      "notification.type"             : "createcopylastasg",
+      "kato.result.expected"          : true,
+      (operationsContext.contextKey()): operationsContext.contextValue(),
+      "deploy.account.name"           : credentials,
     ]
 
     if (stage.context.suspendedProcesses) {

orca-clouddriver/src/main/groovy/com/netflix/spinnaker/orca/clouddriver/tasks/servergroup/CreateServerGroupTask.groovy

@@ -16,11 +16,13 @@
 
 package com.netflix.spinnaker.orca.clouddriver.tasks.servergroup
 
+import com.netflix.spinnaker.orca.api.operations.OperationsContext
+import com.netflix.spinnaker.orca.api.operations.OperationsInput
+import com.netflix.spinnaker.orca.api.operations.OperationsRunner
 import com.netflix.spinnaker.orca.api.pipeline.models.ExecutionStatus
 import com.netflix.spinnaker.orca.api.pipeline.RetryableTask
 import com.netflix.spinnaker.orca.api.pipeline.models.StageExecution
 import com.netflix.spinnaker.orca.api.pipeline.TaskResult
-import com.netflix.spinnaker.orca.clouddriver.KatoService
 import com.netflix.spinnaker.orca.clouddriver.utils.CloudProviderAware
 
 import com.netflix.spinnaker.orca.clouddriver.utils.HealthHelper
@@ -33,7 +35,7 @@ import org.springframework.stereotype.Component
 class CreateServerGroupTask implements CloudProviderAware, RetryableTask {
 
   @Autowired
-  KatoService kato
+  OperationsRunner operationsRunner
 
   @Autowired
   List<ServerGroupCreator> serverGroupCreators
@@ -51,13 +53,14 @@ class CreateServerGroupTask implements CloudProviderAware, RetryableTask {
     }
 
     def ops = creator.getOperations(stage)
-    def taskId = kato.requestOperations(cloudProvider, ops)
+    OperationsInput operationsInput = OperationsInput.of(cloudProvider, ops, stage)
+    OperationsContext operationsContext = operationsRunner.run(operationsInput)
 
     Map outputs = [
-        "notification.type"   : "createdeploy",
-        "kato.result.expected": creator.katoResultExpected,
-        "kato.last.task.id"   : taskId,
-        "deploy.account.name" : credentials
+        "notification.type"             : "createdeploy",
+        "kato.result.expected"          : creator.katoResultExpected,
+        (operationsContext.contextKey()): operationsContext.contextValue(),
+        "deploy.account.name"           : credentials
     ]
 
     if (stage.context.suspendedProcesses?.contains("AddToLoadBalancer")) {

orca-clouddriver/src/main/groovy/com/netflix/spinnaker/orca/clouddriver/tasks/servergroup/UpsertServerGroupTagsTask.groovy

@@ -16,11 +16,13 @@
 
 package com.netflix.spinnaker.orca.clouddriver.tasks.servergroup
 
+import com.netflix.spinnaker.orca.api.operations.OperationsContext
+import com.netflix.spinnaker.orca.api.operations.OperationsInput
+import com.netflix.spinnaker.orca.api.operations.OperationsRunner
 import com.netflix.spinnaker.orca.api.pipeline.models.ExecutionStatus
 import com.netflix.spinnaker.orca.api.pipeline.Task
 import com.netflix.spinnaker.orca.api.pipeline.models.StageExecution
 import com.netflix.spinnaker.orca.api.pipeline.TaskResult
-import com.netflix.spinnaker.orca.clouddriver.KatoService
 import com.netflix.spinnaker.orca.clouddriver.utils.CloudProviderAware
 
 import org.springframework.beans.factory.annotation.Autowired
@@ -32,12 +34,13 @@ import javax.annotation.Nonnull
 class UpsertServerGroupTagsTask implements CloudProviderAware, Task {
 
   @Autowired
-  KatoService kato
+  OperationsRunner operationsRunner
 
   @Nonnull
   @Override
   TaskResult execute(@Nonnull StageExecution stage) {
-    def taskId = kato.requestOperations(getCloudProvider(stage), [[upsertServerGroupTags: stage.context]])
+    OperationsInput operationsInput = OperationsInput.of(getCloudProvider(stage), [[upsertServerGroupTags: stage.context]], stage)
+    OperationsContext operationsContext = operationsRunner.run(operationsInput)
 
     def deployServerGroups = []
     if (stage.context.regions && (stage.context.serverGroupName || stage.context.asgName)) {
@@ -51,10 +54,10 @@ class UpsertServerGroupTagsTask implements CloudProviderAware, Task {
     }
 
     TaskResult.builder(ExecutionStatus.SUCCEEDED).context([
-        "notification.type"   : "upsertservergrouptags",
-        "deploy.account.name" : getCredentials(stage),
-        "kato.last.task.id"   : taskId,
-        "deploy.server.groups": deployServerGroups,
+        "notification.type"             : "upsertservergrouptags",
+        "deploy.account.name"           : getCredentials(stage),
+        (operationsContext.contextKey()): operationsContext.contextValue(),
+        "deploy.server.groups"          : deployServerGroups,
     ]).build()
   }
 }

orca-clouddriver/src/main/java/com/netflix/spinnaker/orca/clouddriver/tasks/servergroup/AbstractBulkServerGroupTask.java

@@ -17,12 +17,13 @@
 package com.netflix.spinnaker.orca.clouddriver.tasks.servergroup;
 
 import com.netflix.spinnaker.moniker.Moniker;
+import com.netflix.spinnaker.orca.api.operations.OperationsContext;
+import com.netflix.spinnaker.orca.api.operations.OperationsInput;
+import com.netflix.spinnaker.orca.api.operations.OperationsRunner;
 import com.netflix.spinnaker.orca.api.pipeline.RetryableTask;
 import com.netflix.spinnaker.orca.api.pipeline.TaskResult;
 import com.netflix.spinnaker.orca.api.pipeline.models.ExecutionStatus;
 import com.netflix.spinnaker.orca.api.pipeline.models.StageExecution;
-import com.netflix.spinnaker.orca.clouddriver.KatoService;
-import com.netflix.spinnaker.orca.clouddriver.model.TaskId;
 import com.netflix.spinnaker.orca.clouddriver.pipeline.servergroup.support.Location;
 import com.netflix.spinnaker.orca.clouddriver.pipeline.servergroup.support.TargetServerGroup;
 import com.netflix.spinnaker.orca.clouddriver.utils.CloudProviderAware;
@@ -41,7 +42,7 @@ public abstract class AbstractBulkServerGroupTask implements CloudProviderAware,
 
   @Autowired protected MonikerHelper monikerHelper;
 
-  @Autowired protected KatoService katoService;
+  @Autowired protected OperationsRunner operationsRunner;
 
   abstract void validateClusterStatus(Map<String, Object> operation, Moniker moniker);
 
@@ -118,11 +119,12 @@ public TaskResult execute(StageExecution stage) {
           operations.add(tmp);
         });
 
-    TaskId taskId = katoService.requestOperations(request.cloudProvider, operations);
+    OperationsInput operationsInput = OperationsInput.of(request.cloudProvider, operations, stage);
+    OperationsContext operationsContext = operationsRunner.run(operationsInput);
 
     Map<String, Object> result = new HashMap<>();
     result.put("deploy.account.name", request.getCredentials());
-    result.put("kato.last.task.id", taskId);
+    result.put(operationsContext.contextKey(), operationsContext.contextValue());
     Map<String, List<String>> regionToServerGroupNames = new HashMap<>();
     regionToServerGroupNames.put(
         request.getRegion(),

orca-clouddriver/src/main/java/com/netflix/spinnaker/orca/clouddriver/tasks/servergroup/UpdateLaunchConfigTask.java

@@ -16,9 +16,9 @@
 
 package com.netflix.spinnaker.orca.clouddriver.tasks.servergroup;
 
+import com.netflix.spinnaker.orca.api.operations.OperationsContext;
+import com.netflix.spinnaker.orca.api.operations.OperationsRunner;
 import com.netflix.spinnaker.orca.api.pipeline.models.StageExecution;
-import com.netflix.spinnaker.orca.clouddriver.KatoService;
-import com.netflix.spinnaker.orca.clouddriver.model.TaskId;
 import groovy.util.logging.Slf4j;
 import java.util.Collections;
 import java.util.HashMap;
@@ -34,12 +34,13 @@ public class UpdateLaunchConfigTask extends AbstractUpdateLaunchSettingsTask {
 
   @Autowired
   public UpdateLaunchConfigTask(
-      KatoService kato, @Value("${default.bake.account:default}") String defaultBakeAccount) {
-    super(kato, defaultBakeAccount);
+      OperationsRunner operationsRunner,
+      @Value("${default.bake.account:default}") String defaultBakeAccount) {
+    super(operationsRunner, defaultBakeAccount);
   }
 
   @Override
-  public Map<String, Object> getContext(StageExecution stage, TaskId taskId) {
+  public Map<String, Object> getContext(StageExecution stage, OperationsContext operationsContext) {
     final Map<String, Object> ctx = new HashMap<>();
     final String region = (String) stage.getContext().get("region");
     final String serverGroupName =
@@ -52,7 +53,7 @@ public Map<String, Object> getContext(StageExecution stage, TaskId taskId) {
     ctx.put("notification.type", "modifyasglaunchconfiguration");
     ctx.put("modifyasglaunchconfiguration.account.name", getCredentials(stage));
     ctx.put("modifyasglaunchconfiguration.region", region);
-    ctx.put("kato.last.task.id", taskId);
+    ctx.put(operationsContext.contextKey(), operationsContext.contextValue());
     ctx.put(
         "deploy.server.groups",
         Collections.singletonMap(region, Collections.singletonList(serverGroupName)));