Process all metrics defined in the canary config. (#36)

spinnaker · Jun 15, 2017 · f492a09 · f492a09
1 parent 63000f4
commit f492a09
Show file tree

Hide file tree

Showing 11 changed files with 72 additions and 49 deletions.
diff --git a/kayenta-atlas/src/main/java/com/netflix/kayenta/atlas/controllers/AtlasFetchController.java b/kayenta-atlas/src/main/java/com/netflix/kayenta/atlas/controllers/AtlasFetchController.java
@@ -32,6 +32,7 @@
 import org.springframework.web.bind.annotation.RestController;
 
 import java.io.IOException;
+import java.util.Collections;
 
 @RestController
 @RequestMapping("/fetch/atlas")
@@ -82,7 +83,7 @@ public String queryMetrics(@RequestParam(required = false) final String metricsA
 
     return synchronousQueryProcessor.processQuery(resolvedMetricsAccountName,
                                                   resolvedStorageAccountName,
-                                                  canaryMetricConfig,
-                                                  atlasCanaryScope);
+                                                  Collections.singletonList(canaryMetricConfig),
+                                                  atlasCanaryScope).get(0);
   }
 }
diff --git a/kayenta-atlas/src/main/java/com/netflix/kayenta/atlas/orca/AtlasFetchTask.java b/kayenta-atlas/src/main/java/com/netflix/kayenta/atlas/orca/AtlasFetchTask.java
@@ -18,8 +18,8 @@
 
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.netflix.kayenta.atlas.canary.AtlasCanaryScope;
-import com.netflix.kayenta.metrics.SynchronousQueryProcessor;
 import com.netflix.kayenta.canary.CanaryConfig;
+import com.netflix.kayenta.metrics.SynchronousQueryProcessor;
 import com.netflix.kayenta.security.AccountCredentials;
 import com.netflix.kayenta.security.AccountCredentialsRepository;
 import com.netflix.kayenta.security.CredentialsHelper;
@@ -36,8 +36,8 @@
 import java.io.IOException;
 import java.time.Duration;
 import java.util.Collections;
+import java.util.List;
 import java.util.Map;
-import java.util.Optional;
 
 @Component
 public class AtlasFetchTask implements RetryableTask {
@@ -89,12 +89,12 @@ public TaskResult execute(Stage stage) {
       CanaryConfig canaryConfig =
         storageService.loadObject(resolvedStorageAccountName, ObjectType.CANARY_CONFIG, canaryConfigId.toLowerCase());
 
-      // TODO(duftler): Fetch _all_ metric sets specified in canaryConfig.getMetrics(), not just the first.
-      String metricSetListId = synchronousQueryProcessor.processQuery(resolvedMetricsAccountName,
-                                                                      resolvedStorageAccountName,
-                                                                      canaryConfig.getMetrics().get(0),
-                                                                      atlasCanaryScope);
-      Map outputs = Collections.singletonMap("metricSetListId", metricSetListId);
+      List<String> metricSetListIds = synchronousQueryProcessor.processQuery(resolvedMetricsAccountName,
+                                                                             resolvedStorageAccountName,
+                                                                             canaryConfig.getMetrics(),
+                                                                             atlasCanaryScope);
+
+      Map outputs = Collections.singletonMap("metricSetListIds", metricSetListIds);
 
       return new TaskResult(ExecutionStatus.SUCCEEDED, outputs);
     } catch (IOException e) {

diff --git a/kayenta-core/src/main/java/com/netflix/kayenta/canary/orca/CanaryJudgeTask.java b/kayenta-core/src/main/java/com/netflix/kayenta/canary/orca/CanaryJudgeTask.java
@@ -36,7 +36,6 @@
 import java.util.Collections;
 import java.util.List;
 import java.util.Map;
-import java.util.Optional;
 
 @Component
 public class CanaryJudgeTask implements RetryableTask {

diff --git a/kayenta-core/src/main/java/com/netflix/kayenta/metrics/SynchronousQueryProcessor.java b/kayenta-core/src/main/java/com/netflix/kayenta/metrics/SynchronousQueryProcessor.java
@@ -26,9 +26,8 @@
 import org.springframework.stereotype.Component;
 
 import java.io.IOException;
-import java.util.Collections;
+import java.util.ArrayList;
 import java.util.List;
-import java.util.Optional;
 import java.util.UUID;
 
 @Component
@@ -41,25 +40,30 @@ public class SynchronousQueryProcessor {
   @Autowired
   StorageServiceRepository storageServiceRepository;
 
-  // TODO(duftler): Take List<CanaryMetricConfig> instead, and return List<String>?
-  public String processQuery(String metricsAccountName,
-                             String storageAccountName,
-                             CanaryMetricConfig canaryMetricConfig,
-                             CanaryScope canaryScope) throws IOException {
+  public List<String> processQuery(String metricsAccountName,
+                                   String storageAccountName,
+                                   List<CanaryMetricConfig> canaryMetricConfigs,
+                                   CanaryScope canaryScope) throws IOException {
     MetricsService metricsService =
       metricsServiceRepository
         .getOne(metricsAccountName)
         .orElseThrow(() -> new IllegalArgumentException("No metrics service was configured; unable to read from metrics store."));
-    List<MetricSet> metricSetList = metricsService.queryMetrics(metricsAccountName, canaryMetricConfig, canaryScope);
 
     StorageService storageService =
       storageServiceRepository
         .getOne(storageAccountName)
         .orElseThrow(() -> new IllegalArgumentException("No storage service was configured; unable to write metric set list."));
-    String metricSetListId = UUID.randomUUID() + "";
 
-    storageService.storeObject(storageAccountName, ObjectType.METRIC_SET_LIST, metricSetListId, metricSetList);
+    List<String> metricSetListIds = new ArrayList<>();
 
-    return metricSetListId;
+    for (CanaryMetricConfig canaryMetricConfig : canaryMetricConfigs) {
+      List<MetricSet> metricSetList = metricsService.queryMetrics(metricsAccountName, canaryMetricConfig, canaryScope);
+      String metricSetListId = UUID.randomUUID() + "";
+
+      storageService.storeObject(storageAccountName, ObjectType.METRIC_SET_LIST, metricSetListId, metricSetList);
+      metricSetListIds.add(metricSetListId);
+    }
+
+    return metricSetListIds;
   }
 }
diff --git a/kayenta-core/src/main/java/com/netflix/kayenta/metrics/orca/MetricSetMixerServiceTask.java b/kayenta-core/src/main/java/com/netflix/kayenta/metrics/orca/MetricSetMixerServiceTask.java
@@ -33,10 +33,10 @@
 import org.springframework.stereotype.Component;
 
 import java.time.Duration;
+import java.util.ArrayList;
 import java.util.Collections;
 import java.util.List;
 import java.util.Map;
-import java.util.Optional;
 import java.util.UUID;
 
 @Component
@@ -67,8 +67,8 @@ public long getTimeout() {
   public TaskResult execute(Stage stage) {
     Map<String, Object> context = stage.getContext();
     String storageAccountName = (String)context.get("storageAccountName");
-    String controlMetricSetListId = (String)context.get("controlMetricSetListId");
-    String experimentMetricSetListId = (String)context.get("experimentMetricSetListId");
+    List<String> controlMetricSetListIds = (List<String>)context.get("controlMetricSetListIds");
+    List<String> experimentMetricSetListIds = (List<String>)context.get("experimentMetricSetListIds");
     String resolvedAccountName = CredentialsHelper.resolveAccountByNameOrType(storageAccountName,
                                                                               AccountCredentials.Type.OBJECT_STORE,
                                                                               accountCredentialsRepository);
@@ -77,17 +77,32 @@ public TaskResult execute(Stage stage) {
         .getOne(resolvedAccountName)
         .orElseThrow(() -> new IllegalArgumentException("No storage service was configured; unable to load metric set lists."));
 
-    List<MetricSet> controlMetricSetList =
-      storageService.loadObject(resolvedAccountName, ObjectType.METRIC_SET_LIST, controlMetricSetListId);
-    List<MetricSet> experimentMetricSetList =
-      storageService.loadObject(resolvedAccountName, ObjectType.METRIC_SET_LIST, experimentMetricSetListId);
-    List<MetricSetPair> metricSetPairList =
-      metricSetMixerService.mixAll(controlMetricSetList, experimentMetricSetList);
-    String metricSetPairListId = UUID.randomUUID() + "";
+    int controlMetricSetListIdsSize = controlMetricSetListIds.size();
+    int experimentMetricSetListIdsSize = experimentMetricSetListIds.size();
 
-    storageService.storeObject(resolvedAccountName, ObjectType.METRIC_SET_PAIR_LIST, metricSetPairListId, metricSetPairList);
+    if (controlMetricSetListIdsSize != experimentMetricSetListIdsSize) {
+      throw new IllegalArgumentException("Size of controlMetricSetListIds (" + controlMetricSetListIdsSize + ") does not " +
+                                         "match size of experimentMetricSetListIds (" + experimentMetricSetListIdsSize + ").");
+    }
 
-    Map outputs = Collections.singletonMap("metricSetPairListId", metricSetPairListId);
+    List<MetricSetPair> aggregatedMetricSetPairList = new ArrayList<>();
+
+    for (int i = 0; i < controlMetricSetListIdsSize; i++) {
+      List<MetricSet> controlMetricSetList =
+        storageService.loadObject(resolvedAccountName, ObjectType.METRIC_SET_LIST, controlMetricSetListIds.get(i));
+      List<MetricSet> experimentMetricSetList =
+        storageService.loadObject(resolvedAccountName, ObjectType.METRIC_SET_LIST, experimentMetricSetListIds.get(i));
+      List<MetricSetPair> metricSetPairList =
+        metricSetMixerService.mixAll(controlMetricSetList, experimentMetricSetList);
+
+      aggregatedMetricSetPairList.addAll(metricSetPairList);
+    }
+
+    String aggregatedMetricSetPairListId = UUID.randomUUID() + "";
+
+    storageService.storeObject(resolvedAccountName, ObjectType.METRIC_SET_PAIR_LIST, aggregatedMetricSetPairListId, aggregatedMetricSetPairList);
+
+    Map outputs = Collections.singletonMap("metricSetPairListId", aggregatedMetricSetPairListId);
 
     return new TaskResult(ExecutionStatus.SUCCEEDED, outputs);
   }

diff --git a/...src/main/java/com/netflix/kayenta/stackdriver/controllers/StackdriverFetchController.java b/...src/main/java/com/netflix/kayenta/stackdriver/controllers/StackdriverFetchController.java
@@ -33,6 +33,7 @@
 
 import java.io.IOException;
 import java.time.Instant;
+import java.util.Collections;
 import java.util.List;
 
 @RestController
@@ -94,7 +95,7 @@ public String queryMetrics(@RequestParam(required = false) final String metricsA
 
     return synchronousQueryProcessor.processQuery(resolvedMetricsAccountName,
                                                   resolvedStorageAccountName,
-                                                  canaryMetricConfig,
-                                                  stackdriverCanaryScope);
+                                                  Collections.singletonList(canaryMetricConfig),
+                                                  stackdriverCanaryScope).get(0);
   }
 }
diff --git a/...-stackdriver/src/main/java/com/netflix/kayenta/stackdriver/orca/StackdriverFetchTask.java b/...-stackdriver/src/main/java/com/netflix/kayenta/stackdriver/orca/StackdriverFetchTask.java
@@ -37,8 +37,8 @@
 import java.time.Duration;
 import java.time.Instant;
 import java.util.Collections;
+import java.util.List;
 import java.util.Map;
-import java.util.Optional;
 
 @Component
 public class StackdriverFetchTask implements RetryableTask {
@@ -97,12 +97,12 @@ public TaskResult execute(Stage stage) {
       stackdriverCanaryScope.setStart(startTimeMillis + "");
       stackdriverCanaryScope.setEnd(endTimeMillis + "");
 
-      // TODO(duftler): Fetch _all_ metric sets specified in canaryConfig.getMetrics(), not just the first.
-      String metricSetListId = synchronousQueryProcessor.processQuery(resolvedMetricsAccountName,
-                                                                      resolvedStorageAccountName,
-                                                                      canaryConfig.getMetrics().get(0),
-                                                                      stackdriverCanaryScope);
-      Map outputs = Collections.singletonMap("metricSetListId", metricSetListId);
+      List<String> metricSetListIds = synchronousQueryProcessor.processQuery(resolvedMetricsAccountName,
+                                                                             resolvedStorageAccountName,
+                                                                             canaryConfig.getMetrics(),
+                                                                             stackdriverCanaryScope);
+
+      Map outputs = Collections.singletonMap("metricSetListIds", metricSetListIds);
 
       return new TaskResult(ExecutionStatus.SUCCEEDED, outputs);
     } catch (IOException e) {

diff --git a/kayenta-web/src/main/java/com/netflix/kayenta/controllers/CanaryConfigController.java b/kayenta-web/src/main/java/com/netflix/kayenta/controllers/CanaryConfigController.java
@@ -97,6 +97,9 @@ public String storeCanaryConfig(@RequestParam(required = false) final String acc
     canaryConfig.setCreatedTimestampIso(Instant.ofEpochMilli(canaryConfig.getCreatedTimestamp()).toString());
     canaryConfig.setUpdatedTimestampIso(Instant.ofEpochMilli(canaryConfig.getUpdatedTimestamp()).toString());
 
+    // TODO(duftler): Since we use a provided canary config name as the unique identifier, we will lose historical
+    // versions of a canary config if a name is reused (will require an explicit DELETE, but there is still a path).
+    // Maybe we should consider serializing the canary config within a canary run?
     if (StringUtils.isEmpty(canaryConfig.getName())) {
       canaryConfig.setName(UUID.randomUUID() + "");
     }

diff --git a/kayenta-web/src/main/java/com/netflix/kayenta/controllers/CanaryController.java b/kayenta-web/src/main/java/com/netflix/kayenta/controllers/CanaryController.java
@@ -158,8 +158,8 @@ public Map<String, String> initiateCanary(@RequestParam(required = false) final
           .put("requisiteStageRefIds", new ImmutableList.Builder().add("1").add("2").build())
           .put("user", "[anonymous]")
           .put("storageAccountName", resolvedStorageAccountName)
-          .put("controlMetricSetListId", "${ #stage('Fetch Baseline from " + serviceType + "')['context']['metricSetListId']}")
-          .put("experimentMetricSetListId", "${ #stage('Fetch Canary from " + serviceType + "')['context']['metricSetListId']}")
+          .put("controlMetricSetListIds", "${ #stage('Fetch Baseline from " + serviceType + "')['context']['metricSetListIds']}")
+          .put("experimentMetricSetListIds", "${ #stage('Fetch Canary from " + serviceType + "')['context']['metricSetListIds']}")
           .build());
 
     Map<String, Object> canaryJudgeContext =

diff --git a/scratch/atlas_pipeline.json b/scratch/atlas_pipeline.json
@@ -55,8 +55,8 @@
       "type": "metricSetMixer",
       "user": "[anonymous]",
       "storageAccountName": "my-account-name-1",
-      "controlMetricSetListId": "${ #stage('Fetch Baseline from Atlas')['context']['metricSetListId']}",
-      "experimentMetricSetListId": "${ #stage('Fetch Canary from Atlas')['context']['metricSetListId']}"
+      "controlMetricSetListIds": "${ #stage('Fetch Baseline from Atlas')['context']['metricSetListIds']}",
+      "experimentMetricSetListIds": "${ #stage('Fetch Canary from Atlas')['context']['metricSetListIds']}"
     },
     {
       "name": "Perform Analysis",

diff --git a/scratch/stackdriver_pipeline.json b/scratch/stackdriver_pipeline.json
@@ -53,8 +53,8 @@
       "type": "metricSetMixer",
       "user": "[anonymous]",
       "storageAccountName": "my-account-name-1",
-      "controlMetricSetListId": "${ #stage('Fetch Baseline from Stackdriver')['context']['metricSetListId']}",
-      "experimentMetricSetListId": "${ #stage('Fetch Canary from Stackdriver')['context']['metricSetListId']}"
+      "controlMetricSetListIds": "${ #stage('Fetch Baseline from Stackdriver')['context']['metricSetListIds']}",
+      "experimentMetricSetListIds": "${ #stage('Fetch Canary from Stackdriver')['context']['metricSetListIds']}"
     },
     {
       "name": "Perform Analysis",