First draft of deployment orchestration state graph

7 files changed, 318 insertions, 26 deletions

diff --git a/controller-server/src/main/java/com/yahoo/vespa/hosted/controller/deployment/DeploymentStatus.java b/controller-server/src/main/java/com/yahoo/vespa/hosted/controller/deployment/DeploymentStatus.java
index 092677255f2..5cc0d5fa809 100644
--- a/controller-server/src/main/java/com/yahoo/vespa/hosted/controller/deployment/DeploymentStatus.java
+++ b/controller-server/src/main/java/com/yahoo/vespa/hosted/controller/deployment/DeploymentStatus.java
@@ -1,18 +1,40 @@
 package com.yahoo.vespa.hosted.controller.deployment;
 
+import com.google.common.collect.ImmutableMap;
+import com.yahoo.component.Version;
+import com.yahoo.config.application.api.DeploymentInstanceSpec;
+import com.yahoo.config.application.api.DeploymentSpec;
+import com.yahoo.config.application.api.DeploymentSpec.DeclaredTest;
+import com.yahoo.config.application.api.DeploymentSpec.DeclaredZone;
 import com.yahoo.config.provision.ApplicationId;
+import com.yahoo.config.provision.Environment;
 import com.yahoo.config.provision.InstanceName;
+import com.yahoo.config.provision.RegionName;
+import com.yahoo.config.provision.SystemName;
+import com.yahoo.config.provision.zone.ZoneId;
 import com.yahoo.vespa.hosted.controller.Application;
 import com.yahoo.vespa.hosted.controller.api.integration.deployment.JobId;
 import com.yahoo.vespa.hosted.controller.api.integration.deployment.JobType;
+import com.yahoo.vespa.hosted.controller.application.Change;
+import com.yahoo.vespa.hosted.controller.application.Deployment;
 
+import java.time.Instant;
+import java.util.ArrayList;
+import java.util.LinkedHashMap;
+import java.util.List;
 import java.util.Map;
-import java.util.Objects;
+import java.util.Optional;
 import java.util.stream.Collectors;
 
+import static com.yahoo.config.provision.Environment.prod;
+import static com.yahoo.config.provision.Environment.staging;
+import static com.yahoo.config.provision.Environment.test;
+import static java.util.Comparator.naturalOrder;
+import static java.util.Objects.requireNonNull;
 import static java.util.stream.Collectors.collectingAndThen;
 import static java.util.stream.Collectors.groupingBy;
 import static java.util.stream.Collectors.toUnmodifiableList;
+import static java.util.stream.Collectors.toUnmodifiableMap;
 
 /**
  * Status of the deployment jobs of an {@link Application}.
@@ -23,10 +45,14 @@ public class DeploymentStatus {
 
     private final Application application;
     private final JobList jobs;
+    private final Map<JobId, List<StepStatus>> steps;
+    private final SystemName system = null; // TODO jonmv: Fix.
+    private final Version systemVersion = null; // TODO jonmv: Fix.
 
     public DeploymentStatus(Application application, Map<JobId, JobStatus> jobs) {
-        this.application = Objects.requireNonNull(application);
+        this.application = requireNonNull(application);
         this.jobs = JobList.from(jobs.values());
+        this.steps = null;//jobDependencies(application.deploymentSpec());
     }
 
     public Application application() {
@@ -56,4 +82,275 @@ public class DeploymentStatus {
                                        collectingAndThen(toUnmodifiableList(), JobList::from)));
     }
 
+    Optional<JobId> jobId(DeploymentSpec.Step step, InstanceName instance) {
+        if ( ! step.steps().isEmpty())
+            throw new IllegalArgumentException(step + " is not a primitive");
+
+        if ( ! step.delay().isZero())
+            return Optional.empty();
+
+        Optional<JobType> job;
+        if (step.concerns(prod) && step.isTest())
+            job = JobType.from(system, ((DeclaredTest) step).region());
+        else {
+            DeclaredZone zone = (DeclaredZone) step;
+            job = JobType.from(system, zone.environment(), zone.region().orElse(null));
+        }
+        if (job.isEmpty())
+            throw new IllegalArgumentException("No job is known for " + step + " in " + system);
+
+        return job.map(type -> new JobId(application.id().instance(instance), type));
+    }
+
+    /** Returns a DAG of the dependencies between the primitive steps in the spec, with iteration order equal to declaration order. */
+    Map<JobId, List<StepStatus>> jobDependencies(DeploymentSpec spec) {
+        Map<JobId, List<StepStatus>> dependencies = new LinkedHashMap<>();
+        List<StepStatus> previous = List.of();
+        for (DeploymentSpec.Step step : spec.steps())
+            previous = fillStep(dependencies, step, previous, spec.instanceNames().get(0));
+
+        return ImmutableMap.copyOf(dependencies);
+    }
+
+    /** Adds the primitive steps contained in the given step, which depend on the given previous primitives, to the dependency graph. */
+    List<StepStatus> fillStep(Map<JobId, List<StepStatus>> dependencies, DeploymentSpec.Step step,
+                              List<StepStatus> previous, InstanceName instance) {
+        if (step.steps().isEmpty()) {
+            if ( ! step.delay().isZero())
+                return List.of(new DelayStatus((DeploymentSpec.Delay) step, previous));
+
+            JobType jobType;
+            StepStatus stepStatus;
+            if (step.concerns(test) || step.concerns(staging)) { // SKIP?
+                jobType = JobType.from(system, ((DeclaredZone) step).environment(), ((DeclaredZone) step).region().get())
+                                         .orElseThrow(() -> new IllegalStateException("No job is known for " + step + " in " + system));
+                previous = new ArrayList<>(previous);
+                stepStatus = JobStepStatus.ofTestDeployment((DeclaredZone) step, List.of(), this, instance, jobType);
+                previous.add(stepStatus);
+                return previous;
+            }
+            else if (step.isTest()) {
+                jobType = JobType.from(system, ((DeclaredTest) step).region())
+                                          .orElseThrow(() -> new IllegalStateException("No job is known for " + step + " in " + system));
+                JobType preType = JobType.from(system, prod, ((DeclaredTest) step).region())
+                                         .orElseThrow(() -> new IllegalStateException("No job is known for " + step + " in " + system));
+                stepStatus = JobStepStatus.ofProductionTest((DeclaredTest) step, previous, this, instance, jobType, preType);
+                return List.of(stepStatus);
+            }
+            else {
+                jobType = JobType.from(system, ((DeclaredZone) step).environment(), ((DeclaredZone) step).region().get())
+                                         .orElseThrow(() -> new IllegalStateException("No job is known for " + step + " in " + system));
+                stepStatus = JobStepStatus.ofProductionDeployment((DeclaredZone) step, previous, this, instance, jobType);
+                return List.of(stepStatus);
+            }
+
+        }
+
+        Optional<InstanceName> stepInstance = Optional.of(step)
+                                                      .filter(DeploymentInstanceSpec.class::isInstance)
+                                                      .map(DeploymentInstanceSpec.class::cast)
+                                                      .map(DeploymentInstanceSpec::name);
+        if (step.isOrdered()) {
+            for (DeploymentSpec.Step nested : step.steps())
+                previous = fillStep(dependencies, nested, previous, stepInstance.orElse(instance));
+
+            return previous;
+        }
+
+        List<StepStatus> parallel = new ArrayList<>();
+        for (DeploymentSpec.Step nested : step.steps())
+            parallel.addAll(fillStep(dependencies, nested, previous, stepInstance.orElse(instance)));
+
+        return List.copyOf(parallel);
+    }
+
+    // Used to represent the system and staging tests that are implicitly required when no explicit tests are listed.
+    private static final DeploymentSpec.Step implicitTests = new DeploymentSpec.Step() {
+        @Override public boolean concerns(Environment environment, Optional<RegionName> region) { return false; }
+    };
+
+
+    /**
+     * Used to represent all steps — explicit and implicit — that may run in order to complete deployment of a change.
+     *
+     * Each node contains a step describing the node,
+     * a list of steps which need to be complete before the step may start,
+     * a list of jobs from which completion of the step is computed, and
+     * optionally, an instance name used to identify a job type for the step,
+     *
+     * The completion criterion for each type of step is implemented in subclasses of this.
+     */
+    public static abstract class StepStatus {
+
+        private final DeploymentSpec.Step step;
+        private final List<StepStatus> dependencies;
+        private final Optional<InstanceName> instance;
+
+        protected StepStatus(DeploymentSpec.Step step, List<StepStatus> dependencies) {
+            this(step, dependencies, null);
+        }
+
+        protected StepStatus(DeploymentSpec.Step step, List<StepStatus> dependencies, InstanceName instance) {
+            this.step = requireNonNull(step);
+            this.dependencies = List.copyOf(dependencies);
+            this.instance = Optional.ofNullable(instance);
+        }
+
+        /** The step defining this. */
+        public final DeploymentSpec.Step step() { return step; }
+
+        /** The list of steps that need to be complete before this may start. */
+        public final List<StepStatus> dependencies() { return dependencies; }
+
+        /** The instance of this, if any. */
+        public final Optional<InstanceName> instance() { return instance; }
+
+        /** The time at which this is complete on the given versions. */
+        public abstract Optional<Instant> completedAt(Change change);
+
+        /** The time at which all dependencies completed on the given version. */
+        public final Optional<Instant> readyAt(Change change) {
+            return dependencies.stream().allMatch(step -> step.completedAt(change).isPresent())
+                   ? dependencies.stream().map(step -> step.completedAt(change).get())
+                                 .max(naturalOrder())
+                                 .or(() -> Optional.of(Instant.EPOCH))
+                   : Optional.empty();
+        }
+
+    }
+
+
+    public static class DelayStatus extends StepStatus {
+
+        public DelayStatus(DeploymentSpec.Delay step, List<StepStatus> dependencies) {
+            super(step, dependencies);
+        }
+
+        @Override
+        public Optional<Instant> completedAt(Change change) {
+            return readyAt(change).map(completion -> completion.plus(step().delay()));
+        }
+
+    }
+
+
+    public static abstract class JobStepStatus extends StepStatus {
+
+        private final JobStatus job;
+
+        protected JobStepStatus(DeploymentSpec.Step step, List<StepStatus> dependencies, JobStatus job) {
+            super(step, dependencies, job.id().application().instance());
+            this.job = requireNonNull(job);
+        }
+
+        public static JobStepStatus ofProductionDeployment(DeclaredZone step, List<StepStatus> dependencies,
+                                                           DeploymentStatus status, InstanceName instance, JobType jobType) {
+            ZoneId zone = ZoneId.from(step.environment(), step.region().get());
+            JobStatus job = status.instanceJobs(instance).get(jobType);
+            Optional<Deployment> existingDeployment = Optional.ofNullable(status.application().require(instance)
+                                                                                .deployments().get(zone));
+
+            return new JobStepStatus(step, dependencies, job) {
+                /** Complete if deployment is on pinned version, and last successful deployment, or if given versions is strictly a downgrade, and this isn't forced by a pin. */
+                @Override
+                public Optional<Instant> completedAt(Change change) {
+                    if (     change.isPinned()
+                        &&   change.platform().isPresent()
+                        && ! existingDeployment.map(Deployment::version).equals(change.platform()))
+                        return Optional.empty();
+
+                    Change fullChange = status.application().change();
+                    if (existingDeployment.map(deployment ->    ! (change.upgrades(deployment.version()) || change.upgrades(deployment.applicationVersion()))
+                                                             &&   (fullChange.downgrades(deployment.version()) || fullChange.downgrades(deployment.applicationVersion())))
+                                          .orElse(false))
+                        return job.lastCompleted().flatMap(Run::end);
+
+                    return job.lastSuccess()
+                              .filter(run ->    change.platform().map(run.versions().targetPlatform()::equals).orElse(true)
+                                             && change.application().map(run.versions().targetApplication()::equals).orElse(true))
+                              .flatMap(Run::end);
+                }
+            };
+        }
+
+        public static JobStepStatus ofProductionTest(DeclaredTest step, List<StepStatus> dependencies,
+                                                     DeploymentStatus status, InstanceName instance, JobType testType, JobType jobType) {
+            JobStatus job = status.instanceJobs(instance).get(testType);
+            return new JobStepStatus(step, dependencies, job) {
+                @Override
+                public Optional<Instant> completedAt(Change change) {
+                    Versions toVerify = Versions.from(change, status.application.require(instance).deployments().get(ZoneId.from(prod, step.region())));
+                    return job.lastSuccess()
+                              .filter(run -> toVerify.targetsMatch(run.versions()))
+                              .filter(run -> status.instanceJobs(instance).get(jobType).lastCompleted()
+                                                   .map(last -> last.end().get().isBefore(run.start())).orElse(false))
+                              .map(run -> run.end().get());
+                }
+            };
+        }
+
+        public static JobStepStatus ofTestDeployment(DeclaredZone step, List<StepStatus> dependencies,
+                                                     DeploymentStatus status, InstanceName instance, JobType jobType) {
+            Versions versions = Versions.from(status.application, status.systemVersion);
+            JobStatus job = status.instanceJobs(instance).get(jobType);
+            return new JobStepStatus(step, dependencies, job) {
+                @Override
+                public Optional<Instant> completedAt(Change change) {
+                    return RunList.from(job)
+                                  .on(versions)
+                                  .status(RunStatus.success)
+                                  .asList().stream()
+                                  .map(run -> run.end().get())
+                                  .max(naturalOrder());
+                }
+            };
+        }
+
+    }
+
+    /*
+     *  Compute all JobIds to run: test and staging for first instance, unless declared in a parallel instance.
+     *  Create StepStatus for the first two, then
+     *
+     *
+     *
+     *
+     *
+     *
+     *
+     *
+     *
+     *
+     * Prod:        completedAt: change, fullChange
+     * Test:        completedAt: versions?
+     * Delay:       completedAt: change, fullChange
+     * Any:         readyAt: change -> versions
+     * Any:         testedAt: versions
+     *
+     * Start with system and staging for first instance as dependencies.
+     * Declared test jobs replace these for intra-instance dependents.
+     * Test and staging are implicitly parallel. (Well, they already are, if they don't depend on each other.)
+     *
+     * Map JobId to StepStatus:
+     * For each prod JobId: Compute Optional<Versions> to run for current change to be done.
+     * Prod jobs may wait for other prod jobs' to-do runs, but ignores their versions.
+     * Prod jobs may wait for test jobs, considering their versions.
+     * For each prod job, if job already triggered on desired versions, ignore the below.
+     * For each prod job, add other prod job dependencies.
+     * For each prod job, add explicit tests in same instance.
+     * For each prod versions to run, find all prod jobs for which those versions aren't tested (before the job), then
+     *      for each such set of jobs, find the last common dependency instance, and add the test for that, or
+     *      for each such set of jobs, add tests for those versions with the first declared instance; in any case
+     *      add all implicit tests to some structure for tracking.
+     *
+     * Eliminate already running jobs.
+     * Keep set of JobId x Versions for each StepStatus, in topological order, for starting jobs and for display.
+     * DepTri: Needs all jobs to run that are also ready. Test jobs are always ready.
+     * API: Needs all jobs to run, and what they are waiting for, like, delay (until), or other jobs, or pause.
+     *      To find dependency jobs, DFS and sort by topological order.
+     *
+     * anySysTest && anyStaTest || triggeredProd
+     *
+     */
+
 }
diff --git a/controller-server/src/main/java/com/yahoo/vespa/hosted/controller/deployment/DeploymentSteps.java b/controller-server/src/main/java/com/yahoo/vespa/hosted/controller/deployment/DeploymentSteps.java
index 43e9476146f..61135613e98 100644
--- a/controller-server/src/main/java/com/yahoo/vespa/hosted/controller/deployment/DeploymentSteps.java
+++ b/controller-server/src/main/java/com/yahoo/vespa/hosted/controller/deployment/DeploymentSteps.java
@@ -73,7 +73,7 @@ public class DeploymentSteps {
 
     /** Returns test jobs to run for this spec */
     public List<JobType> testJobs() {
-        return jobs().stream().filter(JobType::isTest).collect(Collectors.toUnmodifiableList());
+        return jobs().stream().filter(type -> type.environment().isTest()).collect(Collectors.toUnmodifiableList());
     }
 
     /** Returns declared production jobs in this */
diff --git a/controller-server/src/main/java/com/yahoo/vespa/hosted/controller/deployment/DeploymentTrigger.java b/controller-server/src/main/java/com/yahoo/vespa/hosted/controller/deployment/DeploymentTrigger.java
index a5bed840fd6..0b7473aa109 100644
--- a/controller-server/src/main/java/com/yahoo/vespa/hosted/controller/deployment/DeploymentTrigger.java
+++ b/controller-server/src/main/java/com/yahoo/vespa/hosted/controller/deployment/DeploymentTrigger.java
@@ -27,7 +27,6 @@ import java.util.ArrayList;
 import java.util.Collection;
 import java.util.Collections;
 import java.util.List;
-import java.util.Map;
 import java.util.NoSuchElementException;
 import java.util.Objects;
 import java.util.Optional;
@@ -121,7 +120,7 @@ public class DeploymentTrigger {
      */
     public long triggerReadyJobs() {
         return computeReadyJobs().stream()
-                                 .collect(partitioningBy(job -> job.jobType().isTest()))
+                                 .collect(partitioningBy(job -> job.jobType().environment().isTest()))
                                  .entrySet().stream()
                                  .flatMap(entry -> (entry.getKey()
                                          // True for capacity constrained zones -- sort by priority and make a task for each job type.
@@ -234,10 +233,6 @@ public class DeploymentTrigger {
         return controller.applications();
     }
 
-    private Optional<Run> successOn(JobStatus status, Versions versions) {
-        return status.lastSuccess().filter(run -> versions.targetsMatch(run.versions()));
-    }
-
     private Optional<Deployment> deploymentFor(Instance instance, JobType jobType) {
         return Optional.ofNullable(instance.deployments().get(jobType.zone(controller.system())));
     }
@@ -376,7 +371,7 @@ public class DeploymentTrigger {
         if (firstFailing.isAfter(instant.minus(Duration.ofMinutes(1)))) return true;
 
         // Retry out of capacity errors in test environments every minute
-        if (job.isTest() && jobStatus.isOutOfCapacity()) {
+        if (job.environment().isTest() && jobStatus.isOutOfCapacity()) {
             return lastCompleted.isBefore(instant.minus(Duration.ofMinutes(1)));
         }
 
@@ -387,16 +382,6 @@ public class DeploymentTrigger {
         return lastCompleted.isBefore(instant.minus(Duration.ofHours(2))); // Retry at most every 2 hours
     }
 
-    // ---------- Job state helpers ----------
-
-    private List<JobType> runningProductionJobs(Map<JobType, JobStatus> status) {
-        return status.values().parallelStream()
-                     .filter(job -> job.isRunning())
-                     .map(job -> job.id().type())
-                     .filter(JobType::isProduction)
-                     .collect(toList());
-    }
-
     // ---------- Completion logic ----------
 
     /**
diff --git a/controller-server/src/main/java/com/yahoo/vespa/hosted/controller/deployment/InternalStepRunner.java b/controller-server/src/main/java/com/yahoo/vespa/hosted/controller/deployment/InternalStepRunner.java
index 995c257687f..d6de0b06ccc 100644
--- a/controller-server/src/main/java/com/yahoo/vespa/hosted/controller/deployment/InternalStepRunner.java
+++ b/controller-server/src/main/java/com/yahoo/vespa/hosted/controller/deployment/InternalStepRunner.java
@@ -654,7 +654,7 @@ public class InternalStepRunner implements StepRunner {
         DeploymentSpec spec = controller.applications().requireApplication(TenantAndApplicationId.from(id.application())).deploymentSpec();
 
         ZoneId zone = id.type().zone(controller.system());
-        boolean useTesterCertificate = controller.system().isPublic() && id.type().isTest();
+        boolean useTesterCertificate = controller.system().isPublic() && id.type().environment().isTest();
 
         byte[] servicesXml = servicesXml(controller.zoneRegistry().accessControlDomain(),
                                          ! controller.system().isPublic(),
diff --git a/controller-server/src/main/java/com/yahoo/vespa/hosted/controller/deployment/JobList.java b/controller-server/src/main/java/com/yahoo/vespa/hosted/controller/deployment/JobList.java
index 92bf1119020..fdcd11ffb39 100644
--- a/controller-server/src/main/java/com/yahoo/vespa/hosted/controller/deployment/JobList.java
+++ b/controller-server/src/main/java/com/yahoo/vespa/hosted/controller/deployment/JobList.java
@@ -4,6 +4,7 @@ package com.yahoo.vespa.hosted.controller.deployment;
 import com.yahoo.collections.AbstractFilteringList;
 import com.yahoo.component.Version;
 import com.yahoo.config.provision.ApplicationId;
+import com.yahoo.config.provision.InstanceName;
 import com.yahoo.vespa.hosted.controller.api.integration.deployment.ApplicationVersion;
 import com.yahoo.vespa.hosted.controller.api.integration.deployment.JobType;
 
@@ -56,22 +57,22 @@ public class JobList extends AbstractFilteringList<JobStatus, JobList> {
         return matching(JobList::failingApplicationChange);
     }
 
-    /** Returns the subset of jobs which are failing with the given run status */
+    /** Returns the subset of jobs which are failing with the given run status. */
     public JobList withStatus(RunStatus status) {
         return matching(job -> job.lastStatus().map(status::equals).orElse(false));
     }
 
-    /** Returns the subset of jobs of the given type -- most useful when negated */
+    /** Returns the subset of jobs of the given type -- most useful when negated. */
     public JobList type(Collection<? extends JobType> types) {
         return matching(job -> types.contains(job.id().type()));
     }
 
-    /** Returns the subset of jobs of the given type -- most useful when negated */
+    /** Returns the subset of jobs of the given type -- most useful when negated. */
     public JobList type(JobType... types) {
         return type(List.of(types));
     }
 
-    /** Returns the subset of jobs of which are production jobs */
+    /** Returns the subset of jobs of which are production jobs. */
     public JobList production() {
         return matching(job -> job.id().type().isProduction());
     }
diff --git a/controller-server/src/main/java/com/yahoo/vespa/hosted/controller/deployment/Versions.java b/controller-server/src/main/java/com/yahoo/vespa/hosted/controller/deployment/Versions.java
index af6f9af9d1a..87d4f3f0b9a 100644
--- a/controller-server/src/main/java/com/yahoo/vespa/hosted/controller/deployment/Versions.java
+++ b/controller-server/src/main/java/com/yahoo/vespa/hosted/controller/deployment/Versions.java
@@ -110,6 +110,15 @@ public class Versions {
                             deployment.map(Deployment::applicationVersion));
     }
 
+    public static Versions from(Change change, Deployment deployment) {
+        return new Versions(change.platform().filter(version -> change.isPinned() || deployment.version().isBefore(version))
+                                  .orElse(deployment.version()),
+                            change.application().filter(version -> deployment.applicationVersion().compareTo(version) < 0)
+                                  .orElse(deployment.applicationVersion()),
+                            Optional.of(deployment.version()),
+                            Optional.of(deployment.applicationVersion()));
+    }
+
     private static Version targetPlatform(Application application, Change change, Optional<Deployment> deployment,
                                           Version defaultVersion) {
         if (change.isPinned() && change.platform().isPresent())
diff --git a/controller-server/src/main/java/com/yahoo/vespa/hosted/controller/restapi/application/JobControllerApiHandlerHelper.java b/controller-server/src/main/java/com/yahoo/vespa/hosted/controller/restapi/application/JobControllerApiHandlerHelper.java
index 777a4bf264a..d10bad23b8a 100644
--- a/controller-server/src/main/java/com/yahoo/vespa/hosted/controller/restapi/application/JobControllerApiHandlerHelper.java
+++ b/controller-server/src/main/java/com/yahoo/vespa/hosted/controller/restapi/application/JobControllerApiHandlerHelper.java
@@ -248,7 +248,7 @@ class JobControllerApiHandlerHelper {
         int runs = 0;
         Cursor runArray = jobObject.setArray("runs");
         JobList jobList = JobList.from(status.values());
-        if (type.isTest()) {
+        if (type.environment().isTest()) {
             Deque<List<JobType>> pending = new ArrayDeque<>();
             pendingProduction.entrySet().stream()
                              .filter(typeVersions -> jobList.type(type).successOn(typeVersions.getValue()).isEmpty())