Kato clouddriver rebase

kato/kato-aws/src/main/groovy/com/netflix/spinnaker/kato/aws/deploy/ops/AbstractInstanceLoadBalancerRegistrationAtomicOperation.groovy

@@ -67,8 +67,7 @@ abstract class AbstractInstanceLoadBalancerRegistrationAtomicOperation implement
       def loadBalancerNames = asg.loadBalancerNames
       if (!loadBalancerNames) {
         // instances exist in this ASG but there is no load balancer to act against
-        task.updateStatus phaseName, "No load balancers associated with ASG (${asg.autoScalingGroupName})"
-        task.fail()
+        task.updateStatus phaseName, "${performingAction} instances not required for ASG ${asg.autoScalingGroupName}, no load balancers are associated with this ASG"
         return
       }
 

kato/kato-aws/src/test/groovy/com/netflix/spinnaker/kato/aws/deploy/ops/DeregisterInstancesFromLoadBalancerAtomicOperationUnitSpec.groovy

@@ -54,11 +54,11 @@ class DeregisterInstancesFromLoadBalancerAtomicOperationUnitSpec extends Instanc
     }
   }
 
-  void 'should fail task if no load balancers found'() {
+  void 'should noop task if no load balancers found'() {
     setup:
     TaskRepository.threadLocalTask.set(Mock(Task) {
       _ * updateStatus(_,_)
-      1 * fail()
+      0 * fail()
     })
 
     def asg = Mock(AutoScalingGroup) {

kato/kato-aws/src/test/groovy/com/netflix/spinnaker/kato/aws/deploy/ops/RegisterInstancesWithLoadBalancerAtomicOperationUnitSpec.groovy

@@ -54,11 +54,11 @@ class RegisterInstancesWithLoadBalancerAtomicOperationUnitSpec extends InstanceL
     }
   }
 
-  void 'should fail task if no load balancers found'() {
+  void 'should noop task if no load balancers found'() {
     setup:
     TaskRepository.threadLocalTask.set(Mock(Task) {
       _ * updateStatus(_,_)
-      1 * fail()
+      0 * fail()
     })
 
     def asg = Mock(AutoScalingGroup) {

kato/kato-gce/src/main/groovy/com/netflix/spinnaker/kato/gce/deploy/GCEUtil.groovy

@@ -33,6 +33,7 @@ import com.netflix.spinnaker.kato.config.GceConfig
 import com.netflix.spinnaker.kato.data.task.Task
 import com.netflix.spinnaker.kato.gce.deploy.description.CreateGoogleHttpLoadBalancerDescription
 import com.netflix.spinnaker.kato.gce.deploy.description.UpsertGoogleNetworkLoadBalancerDescription
+import com.netflix.spinnaker.kato.gce.deploy.exception.GoogleResourceNotFoundException
 import com.netflix.spinnaker.kato.gce.deploy.ops.ReplicaPoolBuilder
 
 class GCEUtil {
@@ -55,7 +56,7 @@ class GCEUtil {
     if (machineType) {
       return machineType
     } else {
-      updateStatusAndThrowException("Machine type $machineTypeName not found.", task, phase)
+      updateStatusAndThrowNotFoundException("Machine type $machineTypeName not found.", task, phase)
     }
   }
 
@@ -69,7 +70,7 @@ class GCEUtil {
     def imageListCallback = new JsonBatchCallback<ImageList>() {
       @Override
       void onFailure(GoogleJsonError e, HttpHeaders responseHeaders) throws IOException {
-        updateStatusAndThrowException("Error locating $sourceImageName in these projects: $imageProjects: $e.message", task, phase)
+        updateStatusAndThrowNotFoundException("Error locating $sourceImageName in these projects: $imageProjects: $e.message", task, phase)
       }
 
       @Override
@@ -91,7 +92,7 @@ class GCEUtil {
     if (sourceImage) {
       return sourceImage
     } else {
-      updateStatusAndThrowException("Source image $sourceImageName not found in any of these projects: $imageProjects.", task, phase)
+      updateStatusAndThrowNotFoundException("Source image $sourceImageName not found in any of these projects: $imageProjects.", task, phase)
     }
   }
 
@@ -115,7 +116,7 @@ class GCEUtil {
     if (network) {
       return network
     } else {
-      updateStatusAndThrowException("Network $networkName not found.", task, phase)
+      updateStatusAndThrowNotFoundException("Network $networkName not found.", task, phase)
     }
   }
 
@@ -159,7 +160,7 @@ class GCEUtil {
     } else {
       def foundNames = foundForwardingRules.collect { it.name }
 
-      updateStatusAndThrowException("Network load balancers ${forwardingRuleNames - foundNames} not found.", task, phase)
+      updateStatusAndThrowNotFoundException("Network load balancers ${forwardingRuleNames - foundNames} not found.", task, phase)
     }
   }
 
@@ -191,7 +192,7 @@ class GCEUtil {
     } else {
       def foundNames = foundInstances.collect { it.name }
 
-      updateStatusAndThrowException("Instances ${instanceLocalNames - foundNames} not found.", task, phase)
+      updateStatusAndThrowNotFoundException("Instances ${instanceLocalNames - foundNames} not found.", task, phase)
     }
   }
 
@@ -265,7 +266,7 @@ class GCEUtil {
     }
 
     if (useDiskTypeUrl) {
-      diskType = GCEUtil.buildDiskTypeUrl(projectName, zone, diskType)
+      diskType = buildDiskTypeUrl(projectName, zone, diskType)
     }
 
     def attachedDiskInitializeParams = new AttachedDiskInitializeParams(sourceImage: sourceImage.selfLink,
@@ -323,11 +324,11 @@ class GCEUtil {
                              GoogleCredentials credentials,
                              ReplicaPoolBuilder replicaPoolBuilder) {
     def maxSeqNumber = -1
-    def managedInstanceGroups = GCEUtil.queryManagedInstanceGroups(project,
-                                                                   region,
-                                                                   credentials,
-                                                                   replicaPoolBuilder,
-                                                                   APPLICATION_NAME)
+    def managedInstanceGroups = queryManagedInstanceGroups(project,
+                                                           region,
+                                                           credentials,
+                                                           replicaPoolBuilder,
+                                                           APPLICATION_NAME)
 
     for (def managedInstanceGroup : managedInstanceGroups) {
       def names = Names.parseName(managedInstanceGroup.getName())
@@ -357,9 +358,9 @@ class GCEUtil {
     return appName;
   }
 
-  private static void updateStatusAndThrowException(String errorMsg, Task task, String phase) {
+  private static void updateStatusAndThrowNotFoundException(String errorMsg, Task task, String phase) {
     task.updateStatus phase, errorMsg
-    throw new GCEResourceNotFoundException(errorMsg)
+    throw new GoogleResourceNotFoundException(errorMsg)
   }
 
   public static String getLocalName(String fullUrl) {

kato/kato-gce/src/main/groovy/com/netflix/spinnaker/kato/gce/deploy/GoogleOperationPoller.groovy

@@ -0,0 +1,129 @@
+/*
+ * Copyright 2015 Google, 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.kato.gce.deploy
+
+import com.google.api.services.compute.Compute
+import com.google.api.services.compute.model.Operation
+import com.google.api.services.replicapool.Replicapool
+import com.netflix.spinnaker.kato.data.task.Task
+import com.netflix.spinnaker.kato.gce.deploy.config.GoogleConfig
+import com.netflix.spinnaker.kato.gce.deploy.exception.GoogleOperationException
+import com.netflix.spinnaker.kato.gce.deploy.exception.GoogleOperationTimedOutException
+import com.netflix.spinnaker.kato.gce.deploy.exception.GoogleResourceNotFoundException
+import org.springframework.beans.factory.annotation.Autowired
+
+class GoogleOperationPoller {
+
+  // This only exists to facilitate testing.
+  static class ThreadSleeper {
+    void sleep(long seconds) {
+      Thread.currentThread().sleep(seconds * 1000)
+    }
+  }
+
+  @Autowired
+  GoogleConfig.GoogleConfigurationProperties googleConfigurationProperties
+
+  private ThreadSleeper threadSleeper = new ThreadSleeper()
+
+  // The methods below are used to wait on the operation specified in |operationName|. This is used in practice to
+  // turn the asynchronous GCE client operations into synchronous calls. Will poll the state of the operation until
+  // either state is DONE or |timeoutSeconds| is reached.
+  Operation waitForRegionalOperation(Compute compute, String projectName, String region, String operationName,
+                                     Long timeoutSeconds, Task task, String resourceString, String basePhase) {
+    return handleFinishedAsyncOperation(
+        waitForOperation({compute.regionOperations().get(projectName, region, operationName).execute()},
+                         getTimeout(timeoutSeconds)), task, resourceString, basePhase)
+  }
+
+  Operation waitForGlobalOperation(Compute compute, String projectName, String operationName,
+                                   Long timeoutSeconds, Task task, String resourceString, String basePhase) {
+    return handleFinishedAsyncOperation(
+        waitForOperation({compute.globalOperations().get(projectName, operationName).execute()},
+                         getTimeout(timeoutSeconds)), task, resourceString, basePhase)
+  }
+
+  // This method is like the two above except that it operates using a Replicapool object (rather than Compute), which
+  // is the base class for operations relating to managed instance groups.
+  void waitForZoneOperation(Replicapool replicapool, String projectName, String zone, String operationName,
+                            Long timeoutSeconds, Task task, String resourceString, String basePhase) {
+    handleFinishedAsyncOperation(
+        waitForOperation({replicapool.zoneOperations().get(projectName, zone, operationName).execute()},
+                         getTimeout(timeoutSeconds)), task, resourceString, basePhase)
+  }
+
+  private long getTimeout(Long timeoutSeconds) {
+    // Note that we cannot use an Elvis operator here because we might have a timeoutSeconds value of
+    // zero. In that case, we still want to pass that value. So we use null comparison here instead.
+    Math.max(timeoutSeconds != null ? timeoutSeconds : googleConfigurationProperties.asyncOperationTimeoutSecondsDefault, 0)
+  }
+
+  private static handleFinishedAsyncOperation(Operation operation, Task task, String resourceString, String basePhase) {
+    if (!operation) {
+      String errorMsg = "Operation on $resourceString timed out."
+      task.updateStatus basePhase, errorMsg
+      throw new GoogleOperationTimedOutException(errorMsg)
+    }
+
+    if (operation.getError()) {
+      def error = operation?.getError()?.getErrors()?.get(0)
+      String errorMsg = "Failed to complete operation on $resourceString with error: $error"
+      task.updateStatus basePhase, errorMsg
+      throw new GoogleOperationException(errorMsg)
+    }
+
+    task.updateStatus basePhase, "Done operating on $resourceString."
+  }
+
+  /*
+    This method does not correct for potential drift at each interval (we trade some precision for readability).
+    The timeoutSeconds parameter is really treated as a lower-bound. We will poll until the operation reaches a DONE
+    state or until <em>at least</em> that many seconds have passed.
+   */
+  private Operation waitForOperation(Closure getOperation, long timeoutSeconds) {
+    int totalTimePollingSeconds = 0
+    boolean timeoutExceeded = false
+
+    // Fibonacci backoff in seconds, up to googleConfigurationProperties.asyncOperationMaxPollingIntervalSeconds interval.
+    int pollInterval = 1
+    int pollIncrement = 0
+
+    while (!timeoutExceeded) {
+      threadSleeper.sleep(pollInterval)
+
+      totalTimePollingSeconds += pollInterval
+
+      Operation operation = getOperation()
+
+      if (operation.getStatus() == "DONE") {
+        return operation
+      }
+
+      if (totalTimePollingSeconds > timeoutSeconds) {
+        timeoutExceeded = true
+      } else {
+        // Update polling interval.
+        int oldIncrement = pollIncrement
+        pollIncrement = pollInterval
+        pollInterval += oldIncrement
+        pollInterval = Math.min(pollInterval, googleConfigurationProperties.asyncOperationMaxPollingIntervalSeconds)
+      }
+    }
+
+    return null
+  }
+}