feat(stages/webhook): Webhoook stage (#1257)

This PR is first implementation of the webhook stage discussed in issue spinnaker/spinnaker#1512.
Corresponds with Deck PR spinnaker/deck#3447

orca-core/src/main/groovy/com/netflix/spinnaker/orca/config/OrcaConfiguration.java

@@ -40,6 +40,7 @@
 @Configuration
 @ComponentScan({
   "com.netflix.spinnaker.orca.pipeline",
+  "com.netflix.spinnaker.orca.webhook",
   "com.netflix.spinnaker.orca.notifications.scheduling",
   "com.netflix.spinnaker.orca.restart",
   "com.netflix.spinnaker.orca.deprecation"
@@ -99,8 +100,8 @@ public StageNavigator stageNavigator(ApplicationContext applicationContext) {
 
   // TODO: this is a weird place to have this, feels like it should be a bean configurer or something
   public static ThreadPoolTaskExecutor applyThreadPoolMetrics(Registry registry,
-                                                       ThreadPoolTaskExecutor executor,
-                                                       String threadPoolName) {
+                                                              ThreadPoolTaskExecutor executor,
+                                                              String threadPoolName) {
     BiConsumer<String, Function<ThreadPoolExecutor, Integer>> createGuage =
       (name, valueCallback) -> {
         Id id = registry

orca-web/orca-web.gradle

@@ -42,6 +42,7 @@ dependencies {
   compile project(":orca-mine")
   compile project(":orca-mahe")
   compile project(":orca-igor")
+  compile project(":orca-webhook")
   compile project(":orca-eureka")
   compile project(":orca-spring-batch")
   compile project(":orca-pipelinetemplate")

orca-webhook/orca-webhook.gradle

@@ -0,0 +1,24 @@
+/*
+ * Copyright 2017 Schibsted ASA.
+ *
+ * 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.
+ */
+
+dependencies {
+  compile project(':orca-core')
+  compile spinnaker.dependency('kork')
+  compile spinnaker.dependency('bootAutoConfigure')
+  compile('com.jayway.jsonpath:json-path:2.2.0')
+  testCompile project(':orca-test')
+}

orca-webhook/src/main/groovy/com/netflix/spinnaker/orca/webhook/WebhookService.groovy

@@ -0,0 +1,42 @@
+/*
+ * Copyright 2017 Schibsted ASA.
+ *
+ * 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.webhook
+
+import groovy.transform.CompileStatic
+import org.springframework.beans.factory.annotation.Autowired
+import org.springframework.http.HttpEntity
+import org.springframework.http.HttpMethod
+import org.springframework.http.ResponseEntity
+import org.springframework.stereotype.Service
+import org.springframework.web.client.RestTemplate
+
+@Service
+class WebhookService {
+
+  @Autowired
+  RestTemplate restTemplate
+
+  ResponseEntity<Object> exchange(HttpMethod httpMethod, String url, Object payload) {
+    HttpEntity<Object> payloadEntity = new HttpEntity<>(payload)
+    return restTemplate.exchange(url, httpMethod, payloadEntity, Object)
+  }
+
+  ResponseEntity<Object> getStatus(String url) {
+    return restTemplate.getForEntity(url, Object)
+  }
+}

orca-webhook/src/main/groovy/com/netflix/spinnaker/orca/webhook/config/WebhookConfiguration.groovy

@@ -0,0 +1,39 @@
+/*
+ * Copyright 2017 Schibsted ASA.
+ *
+ * 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.webhook.config
+
+import com.netflix.spinnaker.orca.webhook.WebhookService
+import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty
+import org.springframework.context.annotation.Bean
+import org.springframework.context.annotation.ComponentScan
+import org.springframework.context.annotation.Configuration
+import org.springframework.web.client.RestTemplate;
+
+@Configuration
+@ConditionalOnProperty(prefix = "webhook.stage", value = "enabled", matchIfMissing = true)
+@ComponentScan(basePackageClasses = WebhookService)
+class WebhookConfiguration {
+
+  @Bean
+  @ConditionalOnMissingBean(RestTemplate)
+  RestTemplate restTemplate() {
+    new RestTemplate()
+  }
+
+}

orca-webhook/src/main/groovy/com/netflix/spinnaker/orca/webhook/pipeline/WebhookStage.groovy

@@ -0,0 +1,53 @@
+/*
+ * Copyright 2017 Schibsted ASA.
+ *
+ * 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.webhook.pipeline
+
+import com.netflix.spinnaker.orca.CancellableStage
+import com.netflix.spinnaker.orca.batch.RestartableStage
+import com.netflix.spinnaker.orca.pipeline.StageDefinitionBuilder
+import com.netflix.spinnaker.orca.pipeline.TaskNode
+import com.netflix.spinnaker.orca.pipeline.model.Execution
+import com.netflix.spinnaker.orca.pipeline.model.Stage
+
+import com.netflix.spinnaker.orca.webhook.tasks.CreateWebhookTask
+import com.netflix.spinnaker.orca.webhook.tasks.MonitorWebhookTask
+import groovy.transform.CompileStatic
+import groovy.util.logging.Slf4j
+import org.springframework.stereotype.Component
+
+@Slf4j
+@Component
+class WebhookStage implements StageDefinitionBuilder, RestartableStage, CancellableStage {
+
+  @Override
+  <T extends Execution<T>> void taskGraph(Stage<T> stage, TaskNode.Builder builder) {
+    String waitForCompletion = stage.context.waitForCompletion
+
+    builder.withTask("createWebhook", CreateWebhookTask)
+    if (waitForCompletion?.toBoolean()) {
+      builder
+        .withTask("monitorWebhook", MonitorWebhookTask)
+    }
+  }
+
+  @Override
+  CancellableStage.Result cancel(Stage stage) {
+    log.info("Cancelling stage (stageId: ${stage.id}, executionId: ${stage.execution.id}, context: ${stage.context as Map})")
+    return new CancellableStage.Result(stage, [:])
+  }
+}

orca-webhook/src/main/groovy/com/netflix/spinnaker/orca/webhook/tasks/CreateWebhookTask.groovy

@@ -0,0 +1,91 @@
+/*
+ * Copyright 2017 Schibsted ASA.
+ *
+ * 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.webhook.tasks
+
+import com.jayway.jsonpath.JsonPath
+import com.jayway.jsonpath.PathNotFoundException
+import com.netflix.spinnaker.orca.ExecutionStatus
+import com.netflix.spinnaker.orca.RetryableTask
+import com.netflix.spinnaker.orca.TaskResult
+import com.netflix.spinnaker.orca.pipeline.model.Stage
+import com.netflix.spinnaker.orca.webhook.WebhookService
+import groovy.transform.CompileStatic
+import org.apache.http.HttpHeaders
+import org.springframework.beans.factory.annotation.Autowired
+import org.springframework.http.HttpMethod
+import org.springframework.stereotype.Component
+
+@Component
+class CreateWebhookTask implements RetryableTask {
+
+  long backoffPeriod = 30000
+  long timeout = 300000
+
+  @Autowired
+  WebhookService webhookService;
+
+  @Override
+  TaskResult execute(Stage stage) {
+    String url = stage.context.url
+    def method = stage.context.method ? HttpMethod.valueOf(stage.context.method.toString().toUpperCase()) : HttpMethod.POST
+    def payload = stage.context.payload
+    boolean waitForCompletion = (stage.context.waitForCompletion as String)?.toBoolean()
+
+    def response = webhookService.exchange(method, url, payload)
+
+    def outputs = [:]
+    outputs << [statusCode: response.statusCode]
+    if (response.body) {
+      outputs << [buildInfo: response.body]
+    }
+    if (response.statusCode.is2xxSuccessful()) {
+      if (waitForCompletion) {
+        def statusUrl = null
+        def statusUrlResolution = stage.context.statusUrlResolution
+        switch (statusUrlResolution) {
+          case "getMethod":
+            statusUrl = url
+            break
+          case "locationHeader":
+            statusUrl = response.headers.getFirst(HttpHeaders.LOCATION)
+            break
+          case "webhookResponse":
+            try {
+              statusUrl = JsonPath.read(response.body, stage.context.statusUrlJsonPath as String)
+            } catch (PathNotFoundException e) {
+              return new TaskResult(ExecutionStatus.TERMINAL,
+                [error: [reason: e.message, response: response.body]])
+            }
+        }
+        if (!statusUrl || !(statusUrl instanceof String)) {
+          return new TaskResult(ExecutionStatus.TERMINAL,
+            outputs + [
+              error: "The status URL couldn't be resolved, but 'Wait for completion' was checked",
+              statusUrlValue: statusUrl
+            ])
+        }
+        stage.context.statusEndpoint = statusUrl
+        return new TaskResult(ExecutionStatus.SUCCEEDED, outputs + [statusEndpoint: statusUrl])
+      }
+      return new TaskResult(ExecutionStatus.SUCCEEDED, outputs)
+    } else {
+      return new TaskResult(ExecutionStatus.TERMINAL, outputs + [error: "The request did not return a 2xx status"])
+    }
+  }
+
+}

orca-webhook/src/main/groovy/com/netflix/spinnaker/orca/webhook/tasks/MonitorWebhookTask.groovy

@@ -0,0 +1,112 @@
+/*
+ * Copyright 2017 Schibsted ASA.
+ *
+ * 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.webhook.tasks
+
+import com.jayway.jsonpath.JsonPath
+import com.jayway.jsonpath.PathNotFoundException
+import com.netflix.spinnaker.orca.ExecutionStatus
+import com.netflix.spinnaker.orca.Task
+import com.netflix.spinnaker.orca.TaskResult
+import com.netflix.spinnaker.orca.pipeline.model.Stage
+import com.netflix.spinnaker.orca.webhook.WebhookService
+import groovy.transform.CompileStatic
+import org.springframework.beans.factory.annotation.Autowired
+import org.springframework.stereotype.Component
+
+@Component
+class MonitorWebhookTask implements Task {
+
+  @Autowired
+  WebhookService webhookService
+
+  static requiredParameters = ["statusEndpoint", "statusJsonPath"]
+
+  @Override
+  TaskResult execute(Stage stage) {
+    def missing = requiredParameters.findAll { !stage.context.get(it) }
+    if (!missing.empty) {
+      throw new IllegalStateException("Missing required parameter${missing.size() > 1 ? 's' : ''} '${missing.join('\', \'')}'")
+    }
+
+    String statusEndpoint = stage.context.statusEndpoint
+    String statusJsonPath = stage.context.statusJsonPath
+    String progressJsonPath = stage.context.progressJsonPath
+    String successStatuses = stage.context.successStatuses
+    String canceledStatuses = stage.context.canceledStatuses
+    String terminalStatuses = stage.context.terminalStatuses
+
+    def response = webhookService.getStatus(statusEndpoint)
+    def result
+    try {
+      result = JsonPath.read(response.body, statusJsonPath)
+    } catch (PathNotFoundException e) {
+      return new TaskResult(ExecutionStatus.TERMINAL,
+        [error: [reason: e.message, response: response.body]])
+    }
+    if (!(result instanceof String || result instanceof Number || result instanceof Boolean)) {
+      return new TaskResult(ExecutionStatus.TERMINAL,
+        [error: [reason: "The json path '${statusJsonPath}' did not resolve to a single value", value: result]])
+    }
+
+    def responsePayload = [buildInfo: response.body]
+
+    if (progressJsonPath) {
+      def progress
+      try {
+        progress = JsonPath.read(response.body, progressJsonPath)
+      } catch (PathNotFoundException e) {
+        return new TaskResult(ExecutionStatus.TERMINAL,
+          [error: [reason: e.message, response: response.body]])
+      }
+      if (!(progress instanceof String)) {
+        return new TaskResult(ExecutionStatus.TERMINAL,
+          [error: [reason: "The json path '${progressJsonPath}' did not resolve to a String value", value: progress]])
+      }
+      if (progress) {
+        responsePayload += [progressMessage: progress]
+      }
+    }
+
+    def statusMap = createStatusMap(successStatuses, canceledStatuses, terminalStatuses)
+
+    if (result instanceof Number) {
+      def status = result == 100 ? ExecutionStatus.SUCCEEDED : ExecutionStatus.RUNNING
+      return new TaskResult(status, responsePayload + [percentComplete: result])
+    } else if (statusMap.containsKey(result.toString().toUpperCase())) {
+      return new TaskResult(statusMap[result.toString().toUpperCase()], responsePayload)
+    }
+
+    return new TaskResult(ExecutionStatus.RUNNING, response ? responsePayload : [:])
+  }
+
+  private static Map<String, ?> createStatusMap(String successStatuses, String canceledStatuses, String terminalStatuses) {
+    Map statusMap = [:]
+    statusMap << mapStatuses(successStatuses, ExecutionStatus.SUCCEEDED)
+    if (canceledStatuses) {
+      statusMap << mapStatuses(canceledStatuses, ExecutionStatus.CANCELED)
+    }
+    if (terminalStatuses) {
+      statusMap << mapStatuses(terminalStatuses, ExecutionStatus.TERMINAL)
+    }
+    return statusMap
+  }
+
+  private static Map mapStatuses(String statuses, ExecutionStatus status) {
+    statuses.split(",").collectEntries { [(it.trim().toUpperCase()): status] }
+  }
+}