feat(upsertScalingPolicyTask): make upsertScalingPolicyTask retryable (…

…#2703) * feat(upsertScalingPolicyTask): make upsertScalingPolicyTask retryable * feat(upsertScalingPolicyTask): make upsertScalingPolicyTask retryable * feat(upsertScalingPolicyTask): make upsertScalingPolicyTask retryable
spinnaker · Mar 7, 2019 · 6eafe07 · 6eafe07
1 parent 6b907f3
commit 6eafe07
Show file tree

Hide file tree

Showing 2 changed files with 93 additions and 6 deletions.
diff --git a/...com/netflix/spinnaker/orca/clouddriver/tasks/scalingpolicy/UpsertScalingPolicyTask.groovy b/...com/netflix/spinnaker/orca/clouddriver/tasks/scalingpolicy/UpsertScalingPolicyTask.groovy
@@ -16,31 +16,42 @@
 
 package com.netflix.spinnaker.orca.clouddriver.tasks.scalingpolicy
 
+import java.util.concurrent.TimeUnit
 import com.netflix.spinnaker.orca.ExecutionStatus
-import com.netflix.spinnaker.orca.Task
+import com.netflix.spinnaker.orca.RetryableTask
 import com.netflix.spinnaker.orca.TaskResult
 import com.netflix.spinnaker.orca.clouddriver.KatoService
 import com.netflix.spinnaker.orca.clouddriver.tasks.AbstractCloudProviderAwareTask
 import com.netflix.spinnaker.orca.pipeline.model.Stage
 import org.springframework.beans.factory.annotation.Autowired
 import org.springframework.stereotype.Component
+import groovy.util.logging.Slf4j
 
 @Component
-class UpsertScalingPolicyTask extends AbstractCloudProviderAwareTask implements Task {
+@Slf4j
+class UpsertScalingPolicyTask extends AbstractCloudProviderAwareTask implements RetryableTask {
 
   @Autowired
   KatoService kato
 
+  long backoffPeriod = TimeUnit.SECONDS.toMillis(5)
+  long timeout = TimeUnit.SECONDS.toMillis(100)
+
   @Override
   TaskResult execute(Stage stage) {
-    def taskId = kato.requestOperations(getCloudProvider(stage), [[upsertScalingPolicy: stage.context]])
+    try {
+      def taskId = kato.requestOperations(getCloudProvider(stage), [[upsertScalingPolicy: stage.context]])
         .toBlocking()
         .first()
-
-    new TaskResult(ExecutionStatus.SUCCEEDED, [
+      return new TaskResult(ExecutionStatus.SUCCEEDED, [
         "deploy.account.name" : stage.context.credentials,
         "kato.last.task.id"   : taskId,
         "deploy.server.groups": [(stage.context.region): [stage.context.serverGroupName]]
-    ])
+      ])
+    }
+    catch (Exception e) {
+      log.error("Failed upsertScalingPolicy task (stageId: ${stage.id}, executionId: ${stage.execution.id})", e)
+      return new TaskResult(ExecutionStatus.RUNNING)
+    }
   }
 }
diff --git a/...netflix/spinnaker/orca/clouddriver/tasks/scalingpolicy/UpsertScalingPolicyTaskSpec.groovy b/...netflix/spinnaker/orca/clouddriver/tasks/scalingpolicy/UpsertScalingPolicyTaskSpec.groovy
@@ -0,0 +1,76 @@
+/*
+ * Copyright 2019 Netflix, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.netflix.spinnaker.orca.clouddriver.tasks.scalingpolicy
+
+import com.netflix.spinnaker.orca.clouddriver.KatoService
+import com.netflix.spinnaker.orca.clouddriver.model.TaskId
+import com.netflix.spinnaker.orca.pipeline.model.Execution
+import com.netflix.spinnaker.orca.pipeline.model.Stage
+import spock.lang.Shared
+import spock.lang.Specification
+import spock.lang.Unroll
+import rx.Observable
+
+class UpsertScalingPolicyTaskSpec extends Specification {
+
+  @Shared
+  def taskId = new TaskId(UUID.randomUUID().toString())
+
+  @Unroll
+  def "should retry task on exception"() {
+
+    given:
+    KatoService katoService = Mock(KatoService)
+    def task = new UpsertScalingPolicyTask(kato: katoService)
+    def stage = new Stage(Execution.newPipeline("orca"), "upsertScalingPolicy",
+      [credentials                : "abc", cloudProvider: "aCloud",
+       estimatedInstanceWarmup    : "300",
+       targetValue                : "75",
+       targetTrackingConfiguration:
+         [predefinedMetricSpecification:
+            [predefinedMetricType: "ASGAverageCPUUtilization"]]])
+
+    when:
+    def result = task.execute(stage)
+
+    then:
+    1 * katoService.requestOperations(_, _) >> { throw new Exception() }
+    result.status.toString() == "RUNNING"
+
+  }
+
+  @Unroll
+  def "should set the task status to SUCCEEDED for successful execution"() {
+
+    given:
+    KatoService katoService = Mock(KatoService)
+    def task = new UpsertScalingPolicyTask(kato: katoService)
+    def stage = new Stage(Execution.newPipeline("orca"), "upsertScalingPolicy",
+      [credentials                : "abc", cloudProvider: "aCloud",
+       estimatedInstanceWarmup    : "300",
+       targetValue                : "75",
+       targetTrackingConfiguration:
+         [predefinedMetricSpecification:
+            [predefinedMetricType: "ASGAverageCPUUtilization"]]])
+
+    when:
+    def result = task.execute(stage)
+
+    then:
+    1 * katoService.requestOperations(_, _) >> { Observable.from(taskId) }
+    result.status.toString() == "SUCCEEDED"
+  }
+}