fix(logging): add fabulous logging for when pipelines fail to trigger (…

…#555)

* fix(logging): add fabulous logging for when pipelines fail to trigger

The impetus for this change is to be able to better alert on and understand pipeline trigger failures.
We've been bitten by trigger failures that went undiscovered (sometimes for days) until our users complained.
With this change, we can reliably alert on `orca.trigger.errors` metric which is incremented when kicking
of a pipeline fails (unlike the previous metric `orca.error` which was sometimes triggered erroneously).

When triggering does fail, this change will log, in much more detail, why the trigger failed as well as
log the full payload submitted to `orca` for quick and easy debugging/reproducability of the issue.

Bonus 1: This change (mostly) removes the reliance on Rx and replaces it with an executor to
trigger pipelines.

Bonus 2: Fixup calls to `front50` to allow them to be anonymous

echo-notifications/echo-notifications.gradle

@@ -27,7 +27,6 @@ dependencies {
     implementation "com.netflix.spinnaker.kork:kork-core"
     implementation "com.netflix.spinnaker.kork:kork-artifacts"
     implementation "com.netflix.spinnaker.kork:kork-web"
-    implementation "io.reactivex:rxjava"
     implementation "javax.mail:mail:1.4.1"
     implementation "org.springframework.boot:spring-boot-starter-mail"
     implementation "org.springframework.boot:spring-boot-starter-freemarker"

echo-notifications/src/main/groovy/com/netflix/spinnaker/echo/notification/DryRunNotificationAgent.java

@@ -76,14 +76,15 @@ public void sendNotifications(
               .type(Trigger.Type.DRYRUN.toString())
               .lastSuccessfulExecution(execution)
               .build();
-      orca.trigger(
+      OrcaService.TriggerResponse response =
+          orca.trigger(
               pipeline
                   .withName(format("%s (dry run)", pipeline.getName()))
                   .withId(null)
                   .withTrigger(trigger)
                   .withNotifications(
-                      mapper.convertValue(properties.getNotifications(), List.class)))
-          .subscribe(response -> log.info("Pipeline triggered: {}", response));
+                      mapper.convertValue(properties.getNotifications(), List.class)));
+      log.info("Pipeline triggered: {}", response);
     } catch (Exception ex) {
       log.error("Error triggering dry run of {}", pipelineConfigId, ex);
     }

echo-pipelinetriggers/echo-pipelinetriggers.gradle

@@ -41,7 +41,6 @@ dependencies {
 
   implementation "org.springframework.boot:spring-boot-starter-web"
   implementation "com.fasterxml.jackson.core:jackson-databind"
-  implementation "io.reactivex:rxjava"
   implementation "org.springframework:spring-context"
   implementation "com.netflix.spectator:spectator-api"
   implementation "org.apache.commons:commons-lang3"

echo-pipelinetriggers/src/main/java/com/netflix/spinnaker/echo/config/PipelineTriggerConfiguration.java

@@ -9,6 +9,8 @@
 import com.netflix.spinnaker.kork.dynamicconfig.DynamicConfigService;
 import com.netflix.spinnaker.retrofit.Slf4jRetrofitLogger;
 import com.squareup.okhttp.OkHttpClient;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Value;
@@ -73,6 +75,12 @@ public QuietPeriodIndicatorConfigurationProperties quietPeriodIndicatorConfigura
     return new QuietPeriodIndicatorConfigurationProperties();
   }
 
+  @Bean
+  public ExecutorService executorService(
+      @Value("${orca.pipeline-initiator-threadpool-size:16}") int threadPoolSize) {
+    return Executors.newFixedThreadPool(threadPoolSize);
+  }
+
   private <T> T bindRetrofitService(final Class<T> type, final String endpoint) {
     log.info("Connecting {} to {}", type.getSimpleName(), endpoint);
 

echo-pipelinetriggers/src/main/java/com/netflix/spinnaker/echo/pipelinetriggers/PipelineCache.java

@@ -25,6 +25,7 @@
 import com.netflix.spinnaker.echo.model.Trigger;
 import com.netflix.spinnaker.echo.pipelinetriggers.orca.OrcaService;
 import com.netflix.spinnaker.echo.services.Front50Service;
+import com.netflix.spinnaker.security.AuthenticatedRequest;
 import java.time.Duration;
 import java.time.Instant;
 import java.util.*;
@@ -182,7 +183,8 @@ private Optional<Pipeline> process(Map<String, Object> rawPipeline) {
   }
 
   private List<Map<String, Object>> fetchRawPipelines() {
-    List<Map<String, Object>> rawPipelines = front50.getPipelines();
+    List<Map<String, Object>> rawPipelines =
+        AuthenticatedRequest.allowAnonymous(() -> front50.getPipelines());
     return (rawPipelines == null) ? Collections.emptyList() : rawPipelines;
   }
 

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

@@ -78,7 +78,7 @@ private void triggerMatchingPipelines(T event) {
           .forEach(
               p -> {
                 recordMatchingPipeline(p);
-                pipelineInitiator.startPipeline(p);
+                pipelineInitiator.startPipeline(p, PipelineInitiator.TriggerSource.EVENT);
               });
     } catch (TimeoutException e) {
       log.error("Failed to get pipeline configs", e);

echo-pipelinetriggers/src/main/java/com/netflix/spinnaker/echo/pipelinetriggers/orca/OrcaService.java

@@ -23,41 +23,24 @@
 import java.util.Map;
 import retrofit.http.Body;
 import retrofit.http.GET;
-import retrofit.http.Header;
 import retrofit.http.POST;
 import retrofit.http.Query;
-import rx.Observable;
 
 public interface OrcaService {
 
   @POST("/orchestrate")
-  Observable<TriggerResponse> trigger(@Body Pipeline pipeline);
+  TriggerResponse trigger(@Body Pipeline pipeline);
 
   @POST("/plan")
   Map plan(@Body Map pipelineConfig, @Query("resolveArtifacts") boolean resolveArtifacts);
 
   @POST("/v2/pipelineTemplates/plan")
   Map<String, Object> v2Plan(@Body Map pipelineConfig);
 
-  @POST("/orchestrate")
-  Observable<TriggerResponse> trigger(
-      @Body Pipeline pipeline, @Header(TriggerResponse.X_SPINNAKER_USER) String runAsUser);
-
-  @GET("/pipelines")
-  Observable<Collection<PipelineResponse>> getLatestPipelineExecutions(
-      @Query("pipelineConfigIds") Collection<String> pipelineIds);
-
   @GET("/pipelines")
   Collection<PipelineResponse> getLatestPipelineExecutions(
       @Query("pipelineConfigIds") Collection<String> pipelineIds, @Query("limit") Integer limit);
 
-  // GET /pipelines accepts extra query params, which is used for echo extensions.
-  @GET("/pipelines")
-  Observable<Collection<PipelineResponse>> getLatestPipelineExecutions(
-      @Query("pipelineConfigIds") Collection<String> pipelineIds,
-      @Query("statuses") Collection<String> statuses,
-      @Query("limit") Integer limit);
-
   class TriggerResponse {
     // workaround for not having a constant value for reference via annotation:
     static final String X_SPINNAKER_USER = "X-SPINNAKER-USER";