feat(orca-front50): Adding scheduling agent to Disable unused Pipelin…

…es on Front50

orca-core/src/main/java/com/netflix/spinnaker/orca/pipeline/persistence/DualExecutionRepository.kt

@@ -230,6 +230,16 @@ class DualExecutionRepository(
       ).distinctBy { it.id }
   }
 
+  override fun retrievePipelineConfigIdsForApplicationWithCriteria(
+    @Nonnull application: String,
+    @Nonnull criteria: ExecutionCriteria
+  ): List<String> {
+    return (
+      primary.retrievePipelineConfigIdsForApplicationWithCriteria(application, criteria) +
+        previous.retrievePipelineConfigIdsForApplicationWithCriteria(application, criteria)
+      )
+  }
+
   override fun retrievePipelinesForPipelineConfigIdsBetweenBuildTimeBoundary(
     pipelineConfigIds: MutableList<String>,
     buildTimeStartBoundary: Long,

orca-core/src/main/java/com/netflix/spinnaker/orca/pipeline/persistence/ExecutionRepository.java

@@ -107,6 +107,9 @@ Collection<PipelineExecution> retrievePipelineExecutionDetailsForApplication(
       @Nonnull List<String> pipelineConfigIds,
       int queryTimeoutSeconds);
 
+  List<String> retrievePipelineConfigIdsForApplicationWithCriteria(
+      @Nonnull String application, @Nonnull ExecutionCriteria criteria);
+
   /**
    * Returns executions in the time boundary. Redis impl does not respect pageSize or offset params,
    * and returns all executions. Sql impl respects these params.

orca-core/src/main/java/com/netflix/spinnaker/orca/pipeline/persistence/InMemoryExecutionRepository.kt

@@ -305,6 +305,16 @@ class InMemoryExecutionRepository : ExecutionRepository {
       .distinctBy { it.id }
   }
 
+  override fun retrievePipelineConfigIdsForApplicationWithCriteria(
+    @Nonnull application: String,
+    @Nonnull criteria: ExecutionCriteria
+  ): List<String> {
+    return pipelines.values
+      .filter { it.application == application }
+      .applyCriteria(criteria)
+      .map { it.id }
+  }
+
   override fun retrieveOrchestrationForCorrelationId(correlationId: String): PipelineExecution {
     return retrieveByCorrelationId(ORCHESTRATION, correlationId)
   }

orca-front50/orca-front50.gradle

@@ -40,6 +40,9 @@ dependencies {
   testImplementation("com.github.ben-manes.caffeine:guava")
   testImplementation("org.apache.groovy:groovy-json")
   testRuntimeOnly("net.bytebuddy:byte-buddy")
+  testImplementation("org.junit.jupiter:junit-jupiter-api")
+  testImplementation("org.assertj:assertj-core")
+  testImplementation("org.mockito:mockito-junit-jupiter")
 }
 
 sourceSets {

orca-front50/src/main/java/com/netflix/spinnaker/orca/front50/scheduling/UnusedPipelineDisablePollingNotificationAgent.java

@@ -0,0 +1,169 @@
+/*
+ * Copyright 2024 Harness, Inc.
+ *
+ * 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.front50.scheduling;
+
+import static com.netflix.spinnaker.orca.api.pipeline.models.ExecutionType.PIPELINE;
+
+import com.netflix.spectator.api.Id;
+import com.netflix.spectator.api.LongTaskTimer;
+import com.netflix.spectator.api.Registry;
+import com.netflix.spinnaker.kork.retrofit.exceptions.SpinnakerHttpException;
+import com.netflix.spinnaker.orca.api.pipeline.models.ExecutionStatus;
+import com.netflix.spinnaker.orca.front50.Front50Service;
+import com.netflix.spinnaker.orca.notifications.AbstractPollingNotificationAgent;
+import com.netflix.spinnaker.orca.notifications.NotificationClusterLock;
+import com.netflix.spinnaker.orca.pipeline.persistence.ExecutionRepository;
+import java.time.Clock;
+import java.time.ZoneOffset;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression;
+import org.springframework.stereotype.Component;
+
+@Component
+@ConditionalOnExpression(
+    "${pollers.unused-pipelines-disable.enabled:false} && ${execution-repository.sql.enabled:false}")
+public class UnusedPipelineDisablePollingNotificationAgent
+    extends AbstractPollingNotificationAgent {
+
+  Front50Service front50service;
+
+  private static final List<String> COMPLETED_STATUSES =
+      ExecutionStatus.COMPLETED.stream().map(Enum::toString).collect(Collectors.toList());
+
+  private final Logger log =
+      LoggerFactory.getLogger(UnusedPipelineDisablePollingNotificationAgent.class);
+
+  private final Clock clock;
+  private final ExecutionRepository executionRepository;
+  private final Registry registry;
+
+  private final long pollingIntervalSec;
+  private final int thresholdDays;
+  private final boolean dryRun;
+
+  private final Id timerId;
+
+  @Autowired
+  public UnusedPipelineDisablePollingNotificationAgent(
+      NotificationClusterLock clusterLock,
+      ExecutionRepository executionRepository,
+      Front50Service front50Service,
+      Clock clock,
+      Registry registry,
+      @Value("${pollers.unused-pipelines-disable.interval-sec:3600}") long pollingIntervalSec,
+      @Value("${pollers.unused-pipelines-disable.threshold-days:365}") int thresholdDays,
+      @Value("${pollers.unused-pipelines-disable.dry-run:true}") boolean dryRun) {
+    super(clusterLock);
+    this.executionRepository = executionRepository;
+    this.clock = clock;
+    this.registry = registry;
+    this.pollingIntervalSec = pollingIntervalSec;
+    this.thresholdDays = thresholdDays;
+    this.dryRun = dryRun;
+    this.front50service = front50Service;
+
+    timerId = registry.createId("pollers.unusedPipelineDisable.timing");
+  }
+
+  @Override
+  protected long getPollingInterval() {
+    return pollingIntervalSec * 1000;
+  }
+
+  @Override
+  protected String getNotificationType() {
+    return UnusedPipelineDisablePollingNotificationAgent.class.getSimpleName();
+  }
+
+  @Override
+  protected void tick() {
+    LongTaskTimer timer = registry.longTaskTimer(timerId);
+    long timerId = timer.start();
+    try {
+      executionRepository
+          .retrieveAllApplicationNames(PIPELINE)
+          .forEach(
+              app -> {
+                log.info("Evaluating " + app + " for unused pipelines");
+                List<String> pipelineConfigIds =
+                    front50service.getPipelines(app, false, "enabled").stream()
+                        .map(p -> (String) p.get("id"))
+                        .collect(Collectors.toList());
+
+                ExecutionRepository.ExecutionCriteria criteria =
+                    new ExecutionRepository.ExecutionCriteria();
+                criteria.setStatuses(COMPLETED_STATUSES);
+                criteria.setStartTimeCutoff(
+                    clock.instant().atZone(ZoneOffset.UTC).minusDays(thresholdDays).toInstant());
+
+                List<String> orcaExecutionsCount =
+                    executionRepository.retrievePipelineConfigIdsForApplicationWithCriteria(
+                        app, criteria);
+
+                disableAppPipelines(app, orcaExecutionsCount, pipelineConfigIds);
+              });
+    } catch (Exception e) {
+      log.error("Disabling pipelines failed", e);
+    } finally {
+      timer.stop(timerId);
+    }
+  }
+
+  public void disableAppPipelines(
+      String app, List<String> orcaExecutionsCount, List<String> front50PipelineConfigIds) {
+
+    List<String> front50PipelineConfigIdsNotExecuted =
+        front50PipelineConfigIds.stream()
+            .filter(p -> !orcaExecutionsCount.contains(p))
+            .collect(Collectors.toList());
+
+    log.info(
+        "Found "
+            + front50PipelineConfigIdsNotExecuted.size()
+            + " pipelines to disable for Application "
+            + app);
+    front50PipelineConfigIdsNotExecuted.forEach(
+        p -> {
+          log.info("Disabling pipeline execution " + p);
+          if (!dryRun) {
+            disableFront50PipelineConfigId(p);
+          }
+        });
+  }
+
+  public void disableFront50PipelineConfigId(String pipelineConfigId) {
+    Map<String, Object> pipeline = front50service.getPipeline(pipelineConfigId);
+    if (pipeline.get("disabled") == null || !(boolean) pipeline.get("disabled")) {
+      pipeline.put("disabled", true);
+      try {
+        front50service.updatePipeline(pipelineConfigId, pipeline);
+      } catch (SpinnakerHttpException e) {
+        if (Arrays.asList(404, 403).contains(e.getResponseCode())) {
+          log.warn("Failed to disable pipeline " + pipelineConfigId + " due to " + e.getMessage());
+        } else {
+          throw e;
+        }
+      }
+    }
+  }
+}