fix(pipelines): Generalize fix for pipeline race condition (#275)

spinnaker · Jun 18, 2018 · 816e304 · 816e304
1 parent 87177e7
commit 816e304
Show file tree

Hide file tree

Showing 17 changed files with 69 additions and 74 deletions.
diff --git a/...linetriggers/src/main/java/com/netflix/spinnaker/echo/pipelinetriggers/PipelineCache.java b/...linetriggers/src/main/java/com/netflix/spinnaker/echo/pipelinetriggers/PipelineCache.java
@@ -53,7 +53,6 @@ public class PipelineCache implements MonitoredPoller {
   private transient Subscription subscription;
 
   private transient SerializedSubject<List<Pipeline>, List<Pipeline>> pipelineSubject = ReplaySubject.<List<Pipeline>>createWithSize(1).toSerialized();
-  private transient AtomicReference<List<Pipeline>> pipelines = new AtomicReference<>(Collections.emptyList());
 
   @Autowired
   public PipelineCache(@NonNull Scheduler scheduler,
@@ -75,7 +74,7 @@ public void start() {
         .doOnError(this::onFront50Error)
         .retry()
         .map(PipelineCache::decorateTriggers)
-        .doOnNext(this::cachePipelines)
+        .doOnNext(this::logRefresh)
         .subscribe(pipelineSubject);
     }
   }
@@ -109,26 +108,12 @@ public int getPollingIntervalSeconds() {
    *
    * @return An observable emitting the pipelines as of the most recent polling cycle
    */
-  public Observable<List<Pipeline>> getPipelinesAsync() {
+  public Observable<List<Pipeline>> getPipelines() {
     return pipelineSubject.take(1);
   }
 
-  /**
-   * Returns the pipelines as of the most recent polling cycle.  If no polling cycles have been
-   * completed, returns an empty list.
-   *
-   * See {@link #getPipelinesAsync()} for an alternate way of getting pipelines that will wait at
-   * at least one polling cycle before returning a value.
-   *
-   * @return The pipelines as of the most recent polling cycle
-   */
-  public List<Pipeline> getPipelines() {
-    return pipelines.get();
-  }
-
-  private void cachePipelines(final List<Pipeline> pipelines) {
+  private void logRefresh(final List<Pipeline> pipelines) {
     log.info("Refreshing pipelines");
-    this.pipelines.set(pipelines);
   }
 
   private void onFront50Request(final long tick) {

diff --git a/.../src/main/java/com/netflix/spinnaker/echo/pipelinetriggers/monitor/BuildEventMonitor.java b/.../src/main/java/com/netflix/spinnaker/echo/pipelinetriggers/monitor/BuildEventMonitor.java
@@ -71,7 +71,8 @@ public void processEvent(Event event) {
     BuildEvent buildEvent = objectMapper.convertValue(event, BuildEvent.class);
     Observable.just(buildEvent)
       .doOnNext(this::onEchoResponse)
-      .subscribe(triggerEachMatchFrom(pipelineCache.getPipelines()));
+      .zipWith(pipelineCache.getPipelines(), TriggerMatchParameters::new)
+      .subscribe(triggerEachMatch());
   }
 
   @Override

diff --git a/...src/main/java/com/netflix/spinnaker/echo/pipelinetriggers/monitor/DockerEventMonitor.java b/...src/main/java/com/netflix/spinnaker/echo/pipelinetriggers/monitor/DockerEventMonitor.java
@@ -68,7 +68,8 @@ public void processEvent(Event event) {
     DockerEvent dockerEvent = objectMapper.convertValue(event, DockerEvent.class);
     Observable.just(dockerEvent)
       .doOnNext(this::onEchoResponse)
-      .subscribe(triggerEachMatchFrom(pipelineCache.getPipelines()));
+      .zipWith(pipelineCache.getPipelines(), TriggerMatchParameters::new)
+      .subscribe(triggerEachMatch());
   }
 
   @Override

diff --git a/...rs/src/main/java/com/netflix/spinnaker/echo/pipelinetriggers/monitor/GitEventMonitor.java b/...rs/src/main/java/com/netflix/spinnaker/echo/pipelinetriggers/monitor/GitEventMonitor.java
@@ -75,7 +75,8 @@ public void processEvent(Event event) {
     GitEvent buildEvent = objectMapper.convertValue(event, GitEvent.class);
     Observable.just(buildEvent)
       .doOnNext(this::onEchoResponse)
-      .subscribe(triggerEachMatchFrom(pipelineCache.getPipelines()));
+      .zipWith(pipelineCache.getPipelines(), TriggerMatchParameters::new)
+      .subscribe(triggerEachMatch());
   }
 
   @Override

diff --git a/...src/main/java/com/netflix/spinnaker/echo/pipelinetriggers/monitor/PubsubEventMonitor.java b/...src/main/java/com/netflix/spinnaker/echo/pipelinetriggers/monitor/PubsubEventMonitor.java
@@ -69,7 +69,7 @@ public void processEvent(Event event) {
 
     Observable.just(pubsubEvent)
       .doOnNext(this::onEchoResponse)
-      .zipWith(pipelineCache.getPipelinesAsync(), TriggerMatchParameters::new)
+      .zipWith(pipelineCache.getPipelines(), TriggerMatchParameters::new)
       .subscribe(triggerEachMatch());
   }
 

diff --git a/...rc/main/java/com/netflix/spinnaker/echo/pipelinetriggers/monitor/WebhookEventMonitor.java b/...rc/main/java/com/netflix/spinnaker/echo/pipelinetriggers/monitor/WebhookEventMonitor.java
@@ -74,7 +74,8 @@ public void processEvent(Event event) {
 
     Observable.just(webhookEvent)
       .doOnNext(this::onEchoResponse)
-      .subscribe(triggerEachMatchFrom(pipelineCache.getPipelines()));
+      .zipWith(pipelineCache.getPipelines(), TriggerMatchParameters::new)
+      .subscribe(triggerEachMatch());
   }
 
   @Override

diff --git a/...gers/src/test/groovy/com/netflix/spinnaker/echo/pipelinetriggers/PipelineCacheSpec.groovy b/...gers/src/test/groovy/com/netflix/spinnaker/echo/pipelinetriggers/PipelineCacheSpec.groovy
@@ -24,6 +24,7 @@ import com.netflix.spinnaker.echo.model.Pipeline
 import com.netflix.spinnaker.echo.model.Trigger
 import com.netflix.spinnaker.echo.services.Front50Service
 import com.netflix.spinnaker.echo.test.RetrofitStubs
+import rx.observers.TestSubscriber
 import rx.schedulers.Schedulers
 import spock.lang.Shared
 import spock.lang.Specification
@@ -136,15 +137,17 @@ class PipelineCacheSpec extends Specification implements RetrofitStubs {
 
   def "keeps polling if Front50 returns an error"() {
     given:
+    TestSubscriber<List<Pipeline>> testSubscriber = new TestSubscriber<>();
     def pipeline = Pipeline.builder().application('application').name('Pipeline').id('P1').build()
     pipelineCache.start()
 
     when:
     waitForTicks(3)
+    pipelineCache.getPipelines().subscribe(testSubscriber)
 
     then:
     front50.getPipelines() >> just([]) >> { throw unavailable() } >> just([pipeline])
-    pipelineCache.pipelines == [pipeline]
+    testSubscriber.assertValue([pipeline])
   }
 
   def "we can serialize pipelines with triggers that have a parent"() {

diff --git a/...t/groovy/com/netflix/spinnaker/echo/pipelinetriggers/monitor/BuildEventMonitorSpec.groovy b/...t/groovy/com/netflix/spinnaker/echo/pipelinetriggers/monitor/BuildEventMonitorSpec.groovy
@@ -5,8 +5,8 @@ import com.netflix.spectator.api.NoopRegistry
 import com.netflix.spinnaker.echo.model.Event
 import com.netflix.spinnaker.echo.model.Pipeline
 import com.netflix.spinnaker.echo.pipelinetriggers.PipelineCache
-import com.netflix.spinnaker.echo.pipelinetriggers.monitor.BuildEventMonitor
 import com.netflix.spinnaker.echo.test.RetrofitStubs
+import rx.Observable
 import rx.functions.Action1
 import spock.lang.Specification
 import spock.lang.Subject
@@ -27,7 +27,7 @@ class BuildEventMonitorSpec extends Specification implements RetrofitStubs {
   def "triggers pipelines for successful builds for #triggerType"() {
     given:
     def pipeline = createPipelineWith(trigger)
-    pipelineCache.getPipelines() >> [pipeline]
+    pipelineCache.getPipelines() >> Observable.just([pipeline])
 
     when:
     monitor.processEvent(objectMapper.convertValue(event, Event))
@@ -46,7 +46,7 @@ class BuildEventMonitorSpec extends Specification implements RetrofitStubs {
   @Unroll
   def "attaches #triggerType trigger to the pipeline"() {
     given:
-    pipelineCache.getPipelines() >> [pipeline]
+    pipelineCache.getPipelines() >> Observable.just([pipeline])
 
     when:
     monitor.processEvent(objectMapper.convertValue(event, Event))
@@ -67,7 +67,7 @@ class BuildEventMonitorSpec extends Specification implements RetrofitStubs {
 
   def "an event can trigger multiple pipelines"() {
     given:
-    pipelineCache.getPipelines() >> pipelines
+    pipelineCache.getPipelines() >> Observable.just(pipelines)
 
     when:
     monitor.processEvent(objectMapper.convertValue(event, Event))
@@ -90,7 +90,7 @@ class BuildEventMonitorSpec extends Specification implements RetrofitStubs {
   @Unroll
   def "does not trigger pipelines for #description builds"() {
     given:
-    pipelineCache.getPipelines() >> [pipeline]
+    pipelineCache.getPipelines() >> Observable.just([pipeline])
 
     when:
     monitor.processEvent(objectMapper.convertValue(event, Event))
@@ -113,7 +113,7 @@ class BuildEventMonitorSpec extends Specification implements RetrofitStubs {
   @Unroll
   def "does not trigger #description pipelines"() {
     given:
-    pipelineCache.getPipelines() >> [pipeline]
+    pipelineCache.getPipelines() >> Observable.just([pipeline])
 
     when:
     monitor.processEvent(objectMapper.convertValue(event, Event))
@@ -138,7 +138,7 @@ class BuildEventMonitorSpec extends Specification implements RetrofitStubs {
   @Unroll
   def "does not trigger a pipeline that has an enabled #triggerType trigger with missing #field"() {
     given:
-    pipelineCache.getPipelines() >> [badPipeline, goodPipeline]
+    pipelineCache.getPipelines() >> Observable.just([badPipeline, goodPipeline])
     println objectMapper.writeValueAsString(createBuildEventWith(SUCCESS))
 
     when:

diff --git a/.../groovy/com/netflix/spinnaker/echo/pipelinetriggers/monitor/DockerEventMonitorSpec.groovy b/.../groovy/com/netflix/spinnaker/echo/pipelinetriggers/monitor/DockerEventMonitorSpec.groovy
@@ -21,9 +21,9 @@ import com.netflix.spectator.api.NoopRegistry
 import com.netflix.spinnaker.echo.model.Event
 import com.netflix.spinnaker.echo.model.Pipeline
 import com.netflix.spinnaker.echo.pipelinetriggers.PipelineCache
-import com.netflix.spinnaker.echo.pipelinetriggers.monitor.DockerEventMonitor
 import com.netflix.spinnaker.echo.test.RetrofitStubs
 import com.netflix.spinnaker.kork.artifacts.model.Artifact
+import rx.Observable
 import rx.functions.Action1
 import spock.lang.Specification
 import spock.lang.Subject
@@ -42,7 +42,7 @@ class DockerEventMonitorSpec extends Specification implements RetrofitStubs {
   def "triggers pipelines for successful builds for #triggerType"() {
     given:
     def pipeline = createPipelineWith(trigger)
-    pipelineCache.getPipelines() >> [pipeline]
+    pipelineCache.getPipelines() >> Observable.just([pipeline])
 
     when:
     monitor.processEvent(objectMapper.convertValue(event, Event))
@@ -59,7 +59,7 @@ class DockerEventMonitorSpec extends Specification implements RetrofitStubs {
 
   def "attaches docker trigger to the pipeline"() {
     given:
-    pipelineCache.getPipelines() >> [pipeline]
+    pipelineCache.getPipelines() >> Observable.just([pipeline])
 
     when:
     monitor.processEvent(objectMapper.convertValue(event, Event))
@@ -87,7 +87,7 @@ class DockerEventMonitorSpec extends Specification implements RetrofitStubs {
 
   def "an event can trigger multiple pipelines"() {
     given:
-    pipelineCache.getPipelines() >> pipelines
+    pipelineCache.getPipelines() >> Observable.just(pipelines)
 
     when:
     monitor.processEvent(objectMapper.convertValue(event, Event))
@@ -110,7 +110,7 @@ class DockerEventMonitorSpec extends Specification implements RetrofitStubs {
   @Unroll
   def "does not trigger #description pipelines"() {
     given:
-    pipelineCache.getPipelines() >> [pipeline]
+    pipelineCache.getPipelines() >> Observable.just([pipeline])
 
     when:
     monitor.processEvent(objectMapper.convertValue(event, Event))
@@ -130,7 +130,7 @@ class DockerEventMonitorSpec extends Specification implements RetrofitStubs {
   @Unroll
   def "does not trigger #description pipelines for docker"() {
     given:
-    pipelineCache.getPipelines() >> [pipeline]
+    pipelineCache.getPipelines() >> Observable.just([pipeline])
 
     when:
     monitor.processEvent(objectMapper.convertValue(event, Event))
@@ -152,7 +152,7 @@ class DockerEventMonitorSpec extends Specification implements RetrofitStubs {
   @Unroll
   def "does not trigger a pipeline that has an enabled docker trigger with missing #field"() {
     given:
-    pipelineCache.getPipelines() >> [badPipeline, goodPipeline]
+    pipelineCache.getPipelines() >> Observable.just([badPipeline, goodPipeline])
 
     when:
     monitor.processEvent(objectMapper.convertValue(event, Event))
@@ -175,7 +175,7 @@ class DockerEventMonitorSpec extends Specification implements RetrofitStubs {
   def "triggers a pipeline that has an enabled docker trigger with regex"() {
     given:
     def pipeline = createPipelineWith(trigger)
-    pipelineCache.getPipelines() >> [pipeline]
+    pipelineCache.getPipelines() >> Observable.just([pipeline])
 
     when:
     monitor.processEvent(objectMapper.convertValue(event, Event))
@@ -194,7 +194,7 @@ class DockerEventMonitorSpec extends Specification implements RetrofitStubs {
   def "triggers a pipeline that has an enabled docker trigger with empty string for regex"() {
     given:
     def pipeline = createPipelineWith(trigger)
-    pipelineCache.getPipelines() >> [pipeline]
+    pipelineCache.getPipelines() >> Observable.just([pipeline])
 
     when:
     monitor.processEvent(objectMapper.convertValue(event, Event))
@@ -213,7 +213,7 @@ class DockerEventMonitorSpec extends Specification implements RetrofitStubs {
   def "triggers a pipeline that has an enabled docker trigger with only whitespace for regex"() {
     given:
     def pipeline = createPipelineWith(trigger)
-    pipelineCache.getPipelines() >> [pipeline]
+    pipelineCache.getPipelines() >> Observable.just([pipeline])
 
     when:
     monitor.processEvent(objectMapper.convertValue(event, Event))
@@ -232,7 +232,7 @@ class DockerEventMonitorSpec extends Specification implements RetrofitStubs {
   def "does not trigger a pipeline that has an enabled docker trigger with regex"() {
     given:
     def pipeline = createPipelineWith(trigger)
-    pipelineCache.getPipelines() >> [pipeline]
+    pipelineCache.getPipelines() >> Observable.just([pipeline])
 
     when:
     monitor.processEvent(objectMapper.convertValue(event, Event))