feat(pubsub): support constraints (#202)

echo-pipelinetriggers/src/main/java/com/netflix/spinnaker/echo/pipelinetriggers/artifacts/ArtifactMatcher.java

@@ -23,7 +23,9 @@
 import lombok.extern.slf4j.Slf4j;
 
 import java.util.List;
+import java.util.Map;
 import java.util.function.Predicate;
+import java.util.regex.Pattern;
 import java.util.stream.Collectors;
 
 @Slf4j
@@ -48,4 +50,30 @@ public static Boolean anyArtifactsMatchExpected(List<Artifact> messageArtifacts,
         .anyMatch(e -> e.matches(a));
     return messageArtifacts.stream().anyMatch(expectedArtifactMatch);
   }
+
+  /**
+   * Check that there is a key in the payload for each constraint declared in a Trigger.
+   * Also check that if there is a value for a given key, that the value matches the value in the payload.
+   * @param constraints A map of constraints configured in the Trigger (eg, created in Deck).
+   * @param payload A map of the payload contents POST'd in the Webhook.
+   * @return Whether every key (and value if applicable) in the constraints map is represented in the payload.
+   */
+  public static boolean isConstraintInPayload(final Map constraints, final Map payload) {
+    for (Object key : constraints.keySet()) {
+      if (!payload.containsKey(key) || payload.get(key) == null) {
+        log.info("Webhook trigger ignored. Item " + key.toString() + " was not found in payload");
+        return false;
+      }
+
+      if (constraints.get(key) != null && !matches(constraints.get(key).toString(), payload.get(key).toString()) ) {
+        log.info("Webhook trigger ignored. Value of item " + key.toString() + " (" + payload.get(key) + ") in payload does not match constraint " + constraints.get(key));
+        return false;
+      }
+    }
+    return true;
+  }
+
+  private static boolean matches(String us, String other) {
+    return Pattern.compile(us).asPredicate().test(other);
+  }
 }

echo-pipelinetriggers/src/main/java/com/netflix/spinnaker/echo/pipelinetriggers/monitor/PubsubEventMonitor.java

@@ -38,6 +38,7 @@
 import java.util.function.Predicate;
 
 import static com.netflix.spinnaker.echo.pipelinetriggers.artifacts.ArtifactMatcher.anyArtifactsMatchExpected;
+import static com.netflix.spinnaker.echo.pipelinetriggers.artifacts.ArtifactMatcher.isConstraintInPayload;
 
 /**
  * Triggers pipelines in _Orca_ when a trigger-enabled pubsub message arrives.
@@ -103,6 +104,7 @@ protected Predicate<Trigger> matchTriggerFor(final TriggerEvent event, final Pip
     return trigger -> trigger.getType().equalsIgnoreCase(PUBSUB_TRIGGER_TYPE)
         && trigger.getPubsubSystem().equalsIgnoreCase(description.getPubsubSystem().toString())
         && trigger.getSubscriptionName().equalsIgnoreCase(description.getSubscriptionName())
+        && (trigger.getConstraints() == null || isConstraintInPayload(trigger.getConstraints(), event.getPayload()))
         && anyArtifactsMatchExpected(description.getArtifacts(), trigger, pipeline);
   }
 

echo-pipelinetriggers/src/main/java/com/netflix/spinnaker/echo/pipelinetriggers/monitor/WebhookEventMonitor.java

@@ -38,6 +38,8 @@
 import java.util.function.Function;
 import java.util.function.Predicate;
 
+import static com.netflix.spinnaker.echo.pipelinetriggers.artifacts.ArtifactMatcher.isConstraintInPayload;
+
 @Component @Slf4j
 public class WebhookEventMonitor extends TriggerMonitor {
 
@@ -118,27 +120,6 @@ protected Predicate<Trigger> matchTriggerFor(final TriggerEvent event, final Pip
           );
   }
 
-  /**
-   * Check that there is a key in the payload for each constraint declared in a Trigger.
-   * Also check that if there is a value for a given key, that the value matches the value in the payload.
-   * @param constraints A map of constraints configured in the Trigger (eg, created in Deck).
-   * @param payload A map of the payload contents POST'd in the Webhook.
-   * @return Whether every key (and value if applicable) in the constraints map is represented in the payload.
-     */
-  protected boolean isConstraintInPayload(final Map constraints, final Map payload) {
-    for (Object key : constraints.keySet()) {
-      if (!payload.containsKey(key) || payload.get(key) == null) {
-        log.info("Webhook trigger ignored. Item " + key.toString() + " was not found in payload");
-        return false;
-      }
-      if (!constraints.get(key).equals("") && (!constraints.get(key).equals(payload.get(key)))){
-        log.info("Webhook trigger ignored. Value of item " + key.toString() + " in payload does not match constraint");
-        return false;
-      }
-    }
-    return true;
-  }
-
   protected void onMatchingPipeline(Pipeline pipeline) {
     super.onMatchingPipeline(pipeline);
     val id = registry.createId("pipelines.triggered")

echo-pubsub/src/main/java/com/netflix/spinnaker/echo/pubsub/PubsubMessageHandler.java

@@ -16,6 +16,7 @@
 
 package com.netflix.spinnaker.echo.pubsub;
 
+import com.fasterxml.jackson.databind.ObjectMapper;
 import com.netflix.spinnaker.echo.model.Event;
 import com.netflix.spinnaker.echo.model.Metadata;
 import com.netflix.spinnaker.echo.model.pubsub.MessageDescription;
@@ -28,6 +29,7 @@
 import redis.clients.jedis.Jedis;
 import redis.clients.jedis.JedisPool;
 
+import java.io.IOException;
 import java.util.HashMap;
 import java.util.Map;
 
@@ -45,6 +47,9 @@ public class PubsubMessageHandler {
   @Autowired
   private PubsubEventMonitor pubsubEventMonitor;
 
+  @Autowired
+  ObjectMapper objectMapper;
+
   private static final String SET_IF_NOT_EXIST = "NX";
   private static final String SET_EXPIRE_TIME_MILLIS = "PX";
   private static final String SUCCESS = "OK";
@@ -108,6 +113,12 @@ private void processEvent(MessageDescription description) {
     Map<String, Object> content = new HashMap<>();
     Metadata details = new Metadata();
 
+    try {
+      event.setPayload(objectMapper.readValue(description.getMessagePayload(), Map.class));
+    } catch (IOException e) {
+      log.warn("Could not parse message payload as JSON", e);
+    }
+
     content.put("messageDescription", description);
     details.setType(PubsubEventMonitor.PUBSUB_TRIGGER_TYPE);
 

echo-pubsub/src/test/groovy/com/netflix/spinnaker/echo/pubsub/PubsubMessageHandlerSpec.groovy

@@ -16,6 +16,7 @@
 
 package com.netflix.spinnaker.echo.pubsub
 
+import com.fasterxml.jackson.databind.ObjectMapper
 import com.netflix.spinnaker.echo.model.pubsub.MessageDescription
 import com.netflix.spinnaker.echo.model.pubsub.PubsubSystem
 import com.netflix.spinnaker.echo.pipelinetriggers.monitor.PubsubEventMonitor
@@ -49,6 +50,7 @@ class PubsubMessageHandlerSpec extends Specification {
     pubsubMessageHandler = new PubsubMessageHandler()
     pubsubMessageHandler.setJedisPool(embeddedRedis.getPool())
     pubsubMessageHandler.setPubsubEventMonitor(pubsubEventMonitor)
+    pubsubMessageHandler.setObjectMapper(new ObjectMapper())
   }
 
   def cleanup() {