Record run start time (#150)

microsoft · Nov 19, 2024 · 235f2fb · 235f2fb
1 parent 9586806
commit 235f2fb
Show file tree

Hide file tree

Showing 10 changed files with 68 additions and 14 deletions.
diff --git a/cli/integrationtest/controlplane_test.go b/cli/integrationtest/controlplane_test.go
@@ -103,11 +103,16 @@ func TestEndToEnd(t *testing.T) {
 		"-b", fmt.Sprintf("input=%s", inputBufferId),
 		"-b", fmt.Sprintf("output=%s", outputBufferId))
 
-	waitForRunSuccess(t, runId)
+	run := waitForRunSuccess(t, runId)
 
 	output := runCommandSucceeds(t, "sh", "-c", fmt.Sprintf(`tyger buffer read "%s"`, outputSasUri))
 
 	require.Equal("Hello: Bonjour", output)
+
+	require.NotNil(run.StartedAt)
+	require.GreaterOrEqual(*run.StartedAt, run.CreatedAt)
+	require.NotNil(run.FinishedAt)
+	require.GreaterOrEqual(*run.FinishedAt, *run.StartedAt)
 }
 
 func TestEndToEndWithAutomaticallyCreatedBuffers(t *testing.T) {

diff --git a/cli/integrationtest/expected_openapi_spec.yaml b/cli/integrationtest/expected_openapi_spec.yaml
@@ -747,6 +747,11 @@ components:
           description: The time the run was created. Populated by the system.
           format: date-time
           nullable: true
+        startedAt:
+          type: string
+          description: The time the run's job started. Populated by the system.
+          format: date-time
+          nullable: true
         finishedAt:
           type: string
           description: The time the run finished. Populated by the system.

diff --git a/cli/internal/controlplane/model/model.go b/cli/internal/controlplane/model/model.go
@@ -207,6 +207,7 @@ type RunMetadata struct {
 	StatusReason string     `json:"statusReason,omitempty"`
 	RunningCount *int       `json:"runningCount,omitempty"`
 	CreatedAt    time.Time  `json:"createdAt,omitempty"`
+	StartedAt    *time.Time `json:"startedAt,omitempty"`
 	FinishedAt   *time.Time `json:"finishedAt,omitempty"`
 }
 

diff --git a/cli/internal/dataplane/relayserver.go b/cli/internal/dataplane/relayserver.go
@@ -58,6 +58,14 @@ func RelayInputServer(
 			}
 
 			defer complete()
+
+			defer func() {
+				flusher, ok := w.(http.Flusher)
+				if ok {
+					flusher.Flush()
+				}
+			}()
+
 			if outputWriter == io.Discard {
 				log.Warn().Msg("Discarding input data")
 				w.WriteHeader(http.StatusAccepted)
@@ -120,6 +128,13 @@ func RelayOutputServer(
 			}
 			defer complete()
 
+			defer func() {
+				flusher, ok := w.(http.Flusher)
+				if ok {
+					flusher.Flush()
+				}
+			}()
+
 			_, err := io.Copy(w, inputReader)
 			if err != nil {
 				log.Error().Err(err).Msg("transfer failed")

diff --git a/server/ControlPlane/Compute/Docker/DockerRunCreator.cs b/server/ControlPlane/Compute/Docker/DockerRunCreator.cs
@@ -120,8 +120,7 @@ public async Task<Run> CreateRun(Run run, string? idempotencyKey, CancellationTo
         run = await Repository.CreateRun(run, idempotencyKey, cancellationToken);
 
         var bufferMap = await GetBufferMap(jobCodespec.Buffers, run.Job.Buffers!, cancellationToken);
-
-        string mainContainerName = $"tyger-run-{run.Id}-main";
+        string mainContainerName = MainContainerName(run.Id!.Value);
 
         if (run.Job.Buffers != null)
         {
@@ -471,6 +470,11 @@ void WriteTombstone()
         return run with { Status = RunStatus.Running };
     }
 
+    internal static string MainContainerName(long runId)
+    {
+        return $"tyger-run-{runId}-main";
+    }
+
     private async Task CreateAndStartContainer(CreateContainerParameters sidecarContainerParameters, CancellationToken cancellationToken)
     {
         var createResponse = await _client.Containers.CreateContainerAsync(sidecarContainerParameters, cancellationToken);

diff --git a/server/ControlPlane/Compute/Docker/DockerRunReader.cs b/server/ControlPlane/Compute/Docker/DockerRunReader.cs
@@ -141,9 +141,17 @@ async Task ScheduleFirstUpdate()
 
     public static Run UpdateRunFromContainers(Run run, IReadOnlyList<ContainerInspectResponse> containers)
     {
+        var mainContainerName = $"/{DockerRunCreator.MainContainerName(run.Id!.Value)}"; // inspect always returns names with a leading slash
+
+        var mainContainer = containers.FirstOrDefault(c => c.Name == mainContainerName);
+        if (mainContainer != null)
+        {
+            run = run with { StartedAt = mainContainer?.State.StartedAt == null ? null : DateTimeOffset.Parse(mainContainer.State.StartedAt) };
+        }
+
         if (run.Status is RunStatus.Canceled)
         {
-            return run;
+            return run with { Status = RunStatus.Canceled };
         }
 
         var socketCount = containers.Aggregate(
@@ -159,14 +167,15 @@ public static Run UpdateRunFromContainers(Run run, IReadOnlyList<ContainerInspec
             return run with { Status = RunStatus.Failed };
         }
 
-        if (containers.Any(c => c.State.Status == "exited" && c.State.ExitCode != 0))
+        ContainerInspectResponse? exited;
+        if ((exited = containers.FirstOrDefault(c => c.State.Status == "exited" && c.State.ExitCode != 0)) != null)
         {
-            return run with { Status = RunStatus.Failed };
+            return run with { Status = RunStatus.Failed, FinishedAt = DateTimeOffset.Parse(exited.State.FinishedAt) };
         }
 
         if (containers.All(c => c.State.Status == "exited" && c.State.ExitCode == 0))
         {
-            return run with { Status = RunStatus.Succeeded };
+            return run with { Status = RunStatus.Succeeded, FinishedAt = DateTimeOffset.Parse(containers.Max(c => c.State.FinishedAt)!) };
         }
 
         if (socketCount > 0)
@@ -175,7 +184,7 @@ public static Run UpdateRunFromContainers(Run run, IReadOnlyList<ContainerInspec
             // If the main container has opened a socket, we consider the run successful if all other containers have exited successfully
             if (mainSocketContainer != null && containers.Where(c => c.ID != mainSocketContainer.ID).All(c => c.State.Status == "exited" && c.State.ExitCode == 0))
             {
-                return run with { Status = RunStatus.Succeeded };
+                return run with { Status = RunStatus.Succeeded, FinishedAt = DateTimeOffset.Parse(containers.Max(c => c.State.FinishedAt)!) };
             }
         }
 

diff --git a/server/ControlPlane/Compute/Kubernetes/RunObjects.cs b/server/ControlPlane/Compute/Kubernetes/RunObjects.cs
@@ -56,11 +56,14 @@ public void ClearCachedMetadata()
 
     private ObservedRunState GetStatus()
     {
+        var startedAt = GetStartedTime();
+
         if (GetFailureTimeAndReason() is (var failureTime, var reason))
         {
             return new(Id, RunStatus.Failed, JobPods.Length, WorkerPods.Length)
             {
                 StatusReason = reason,
+                StartedAt = startedAt,
                 FinishedAt = failureTime,
             };
         }
@@ -69,6 +72,7 @@ private ObservedRunState GetStatus()
         {
             return new(Id, RunStatus.Succeeded, JobPods.Length, WorkerPods.Length)
             {
+                StartedAt = startedAt,
                 FinishedAt = successTime,
             };
         }
@@ -77,24 +81,25 @@ private ObservedRunState GetStatus()
 
         if (runningCount > 0)
         {
-            return new(Id, RunStatus.Running, JobPods.Length, WorkerPods.Length) { RunningCount = runningCount };
+            return new(Id, RunStatus.Running, JobPods.Length, WorkerPods.Length) { StartedAt = startedAt, RunningCount = runningCount };
         }
 
-        return new(Id, RunStatus.Pending, JobPods.Length, WorkerPods.Length);
+        return new(Id, RunStatus.Pending, JobPods.Length, WorkerPods.Length) { StartedAt = startedAt };
     }
 
     private (DateTimeOffset, string)? GetFailureTimeAndReason()
     {
+        var fallbackTime = DateTimeOffset.UtcNow;
         var containerStatus = JobPods
             .Where(p => p?.Status?.ContainerStatuses != null)
             .SelectMany(p => p!.Status.ContainerStatuses)
             .Where(cs => cs.State?.Terminated?.ExitCode is not null and not 0)
-            .MinBy(cs => cs.State.Terminated.FinishedAt!.Value);
+            .MinBy(cs => cs.State.Terminated.FinishedAt ?? fallbackTime); // sometimes FinishedAt is null https://github.com/kubernetes/kubernetes/issues/104107
 
         if (containerStatus != null)
         {
             var reason = $"{(containerStatus.Name == "main" ? "Main" : "Sidecar")} exited with code {containerStatus.State.Terminated.ExitCode}";
-            return (containerStatus.State.Terminated.FinishedAt!.Value, reason);
+            return (containerStatus.State.Terminated.FinishedAt ?? fallbackTime, reason);
         }
 
         return null;
@@ -105,7 +110,7 @@ private ObservedRunState GetStatus()
         DateTimeOffset? startedTime = null;
         foreach (var pod in JobPods)
         {
-            if (pod == null)
+            if (pod?.Status?.ContainerStatuses == null)
             {
                 continue;
             }

diff --git a/server/ControlPlane/Database/Repository.cs b/server/ControlPlane/Database/Repository.cs
@@ -523,7 +523,7 @@ FOR UPDATE
 
             var runJson = reader.GetString(0);
             var run = JsonSerializer.Deserialize<Run>(runJson, _serializerOptions) ?? throw new InvalidOperationException("Failed to deserialize run.");
-            updatedRun = run with { Status = state.Status, StatusReason = state.StatusReason, RunningCount = state.RunningCount, FinishedAt = state.FinishedAt, Job = run.Job with { NodePool = state.JobNodePool }, Worker = run.Worker == null ? null : run.Worker with { NodePool = state.WorkerNodePool } };
+            updatedRun = state.ApplyToRun(run);
             if (updatedRun.Equals(run))
             {
                 return;

diff --git a/server/ControlPlane/Model/Model.cs b/server/ControlPlane/Model/Model.cs
@@ -451,6 +451,11 @@ public record Run : ModelBase
     /// </summary>
     public DateTimeOffset? CreatedAt { get; init; }
 
+    /// <summary>
+    /// The time the run's job started. Populated by the system.
+    /// </summary>
+    public DateTimeOffset? StartedAt { get; init; }
+
     /// <summary>
     /// The time the run finished. Populated by the system.
     /// </summary>
@@ -486,6 +491,7 @@ public Run WithoutSystemProperties()
             StatusReason = null,
             RunningCount = null,
             CreatedAt = default,
+            StartedAt = default,
             FinishedAt = null,
         };
     }

diff --git a/server/ControlPlane/Model/ObservedRunState.cs b/server/ControlPlane/Model/ObservedRunState.cs
@@ -25,6 +25,7 @@ public ObservedRunState(Run run, DateTimeOffset? databaseUpdatedAt)
     {
         StatusReason = run.StatusReason;
         RunningCount = run.RunningCount;
+        StartedAt = run.StartedAt;
         FinishedAt = run.FinishedAt;
         JobNodePool = run.Job.NodePool;
         WorkerNodePool = run.Worker?.NodePool;
@@ -35,6 +36,8 @@ public ObservedRunState(Run run, DateTimeOffset? databaseUpdatedAt)
 
     public int? RunningCount { get; init; }
 
+    public DateTimeOffset? StartedAt { get; init; }
+
     public DateTimeOffset? FinishedAt { get; init; }
 
     public string? JobNodePool { get; init; }
@@ -51,6 +54,7 @@ public readonly Run ApplyToRun(Run run)
             Status = Status,
             StatusReason = StatusReason,
             RunningCount = RunningCount,
+            StartedAt = StartedAt,
             FinishedAt = FinishedAt,
             Job = run.Job with { NodePool = JobNodePool },
             Worker = run.Worker == null ? null : run.Worker with { NodePool = WorkerNodePool }