Write script pod events to task log (#992)

* Write script pod events to task log

* Fix only returning events once

* Write since times to disk in case of tentacle restart

* Handle file not found

* Fix path issue

* Change to use Series values

* Handle possible null series

* Change back to helper class

* Write warning events with warning messages

* Change how we do wrapped messages

* Don't return WrappedProcessOutput

* Change back to Count

* Add env var to disable pod events

* Log pulling & pulled as wait style

* Constrain the pulled message to the same field path as the pulling

* Added test and removed Task.WhenAll

* Removed test as it was too flaky

* Add pod event check in simple test

Author: APErebus

source/.run/Build and Push Kubernetes Tentacle docker image (powershell, amd64).run.xml

@@ -3,7 +3,7 @@
     <option name="SCRIPT_TEXT" value=".\build." />
     <option name="INDEPENDENT_SCRIPT_PATH" value="false" />
     <option name="SCRIPT_PATH" value="$PROJECT_DIR$/../build.ps1" />
-    <option name="SCRIPT_OPTIONS" value="-Target BuildAndLoadLocallyKubernetesTentacleContainerImage -DockerPlatform &quot;linux/amd64&quot;" />
+    <option name="SCRIPT_OPTIONS" value="-Target BuildAndLoadLocallyKubernetesTentacleContainerImage -DockerPlatform &quot;linux/amd64&quot; -Runtime &quot;linux-x64&quot;" />
     <option name="INDEPENDENT_SCRIPT_WORKING_DIRECTORY" value="true" />
     <option name="SCRIPT_WORKING_DIRECTORY" value="$PROJECT_DIR$/../" />
     <option name="INDEPENDENT_INTERPRETER_PATH" value="true" />

source/Octopus.Tentacle.Contracts/ScriptStatusResponse.cs

@@ -1,5 +1,6 @@
 using System;
 using System.Collections.Generic;
+using System.Diagnostics;
 using Newtonsoft.Json;
 
 namespace Octopus.Tentacle.Contracts
@@ -18,6 +19,7 @@ public enum ProcessOutputSource
         Debug
     }
 
+    [DebuggerDisplay("{Occurred} | {Source} | {Text}")]
     public class ProcessOutput
     {
         public ProcessOutput(ProcessOutputSource source, string text) : this(source, text, DateTimeOffset.UtcNow)

source/Octopus.Tentacle.Kubernetes.Tests.Integration/KubernetesScriptServiceV1IntegrationTest.cs

@@ -1,5 +1,6 @@
 using FluentAssertions;
 using FluentAssertions.Execution;
+using Octopus.Client.Model.Endpoints;
 using Octopus.Tentacle.Client.Scripts.Models;
 using Octopus.Tentacle.Client.Scripts.Models.Builders;
 using Octopus.Tentacle.CommonTestUtils;
@@ -53,6 +54,7 @@ public async Task RunSimpleScript(ScriptType scriptType)
         var result = await TentacleClient.ExecuteScript(command, StatusReceived, ScriptCompleted, new InMemoryLog(), CancellationToken);
 
         //Assert
+        logs.Should().Contain(po => po.Text.StartsWith("[POD EVENT]")); // Verify that we are receiving some pod events
         logs.Should().Contain(po => po.Source == ProcessOutputSource.StdOut && po.Text == "Hello World");
         scriptCompleted.Should().BeTrue();
         result.ExitCode.Should().Be(0);

source/Octopus.Tentacle/Kubernetes/IKubernetesPodLogService.cs

@@ -6,6 +6,7 @@
 using System.Threading;
 using System.Threading.Tasks;
 using k8s.Autorest;
+using k8s.Models;
 using Octopus.Diagnostics;
 using Octopus.Tentacle.Contracts;
 
@@ -21,39 +22,65 @@ class KubernetesPodLogService : KubernetesService, IKubernetesPodLogService
         readonly IKubernetesPodMonitor podMonitor;
         readonly ITentacleScriptLogProvider scriptLogProvider;
         readonly IScriptPodSinceTimeStore scriptPodSinceTimeStore;
+        readonly IKubernetesEventService eventService;
 
-        public KubernetesPodLogService(IKubernetesClientConfigProvider configProvider, IKubernetesPodMonitor podMonitor, ITentacleScriptLogProvider scriptLogProvider, IScriptPodSinceTimeStore scriptPodSinceTimeStore, ISystemLog log) 
+        public KubernetesPodLogService(
+            IKubernetesClientConfigProvider configProvider,
+            IKubernetesPodMonitor podMonitor,
+            ITentacleScriptLogProvider scriptLogProvider,
+            IScriptPodSinceTimeStore scriptPodSinceTimeStore,
+            IKubernetesEventService eventService,
+            ISystemLog log)
             : base(configProvider, log)
         {
             this.podMonitor = podMonitor;
             this.scriptLogProvider = scriptLogProvider;
             this.scriptPodSinceTimeStore = scriptPodSinceTimeStore;
+            this.eventService = eventService;
         }
 
         public async Task<(IReadOnlyCollection<ProcessOutput> Outputs, long NextSequenceNumber)> GetLogs(ScriptTicket scriptTicket, long lastLogSequence, CancellationToken cancellationToken)
         {
             var tentacleScriptLog = scriptLogProvider.GetOrCreate(scriptTicket);
             var podName = scriptTicket.ToKubernetesScriptPodName();
 
-            var podLogs = await GetPodLogs();
+            //we start both tasks now so we can overlap the API calls
+            var podLogsTask = GetPodLogs();
+            var podEventsTask = GetPodEvents(scriptTicket, podName, cancellationToken);
+
+            var podLogs = await podLogsTask;
             if (podLogs.Outputs.Any())
             {
                 var nextSinceTime = podLogs.Outputs.Max(o => o.Occurred);
-                scriptPodSinceTimeStore.UpdateSinceTime(scriptTicket, nextSinceTime);
-                
+                scriptPodSinceTimeStore.UpdatePodLogsSinceTime(scriptTicket, nextSinceTime);
+
                 //We can use our EOS marker to detect completion quicker than the Pod status
                 if (podLogs.ExitCode != null)
                     podMonitor.MarkAsCompleted(scriptTicket, podLogs.ExitCode.Value);
             }
 
             //We are making the assumption that the clock on the Tentacle Pod is in sync with API Server
             var tentacleLogs = tentacleScriptLog.PopLogs();
-            var combinedLogs = podLogs.Outputs.Concat(tentacleLogs).OrderBy(o => o.Occurred).ToList();
+            var podEventLogs = await podEventsTask;
+
+            var combinedLogs = podLogs
+                .Outputs
+                .Concat(tentacleLogs)
+                .Concat(podEventLogs)
+                .OrderBy(o => o.Occurred)
+                .SelectMany(o => o switch
+                {
+                    //if this is a wrapped output, expand it in place
+                    WrappedProcessOutput wrappedOutput => wrappedOutput.Expand(),
+                    _ => new[] { o }
+                })
+                .ToList();
+
             return (combinedLogs, podLogs.NextSequenceNumber);
 
             async Task<(IReadOnlyCollection<ProcessOutput> Outputs, long NextSequenceNumber, int? ExitCode)> GetPodLogs()
             {
-                var sinceTime = scriptPodSinceTimeStore.GetSinceTime(scriptTicket);
+                var sinceTime = scriptPodSinceTimeStore.GetPodLogsSinceTime(scriptTicket);
                 try
                 {
                     return await GetPodLogsWithSinceTime(sinceTime);
@@ -63,7 +90,7 @@ public KubernetesPodLogService(IKubernetesClientConfigProvider configProvider, I
                     var message = $"Unexpected Pod log line numbers found with sinceTime='{sinceTime}', loading all logs";
                     tentacleScriptLog.Verbose(message);
                     Log.Warn(ex, message);
-                    
+
                     //If we somehow come across weird/missing line numbers, try load the whole Pod logs to see if that helps
                     return await GetPodLogsWithSinceTime(null);
                 }
@@ -77,13 +104,64 @@ public KubernetesPodLogService(IKubernetesClientConfigProvider configProvider, I
 
             async Task<(IReadOnlyCollection<ProcessOutput> Outputs, long NextSequenceNumber, int? ExitCode)> ReadPodLogsFromStream(Stream stream)
             {
-                using (var reader = new StreamReader(stream))
-                {
-                    return await PodLogReader.ReadPodLogs(lastLogSequence, reader);
-                }
+                using var reader = new StreamReader(stream);
+                return await PodLogReader.ReadPodLogs(lastLogSequence, reader);
             }
         }
 
+        async Task<IEnumerable<ProcessOutput>> GetPodEvents(ScriptTicket scriptTicket, string podName, CancellationToken cancellationToken)
+        {
+            //if we don't want to write pod events to the task log, don't do anything
+            if (KubernetesConfig.DisablePodEventsInTaskLog)
+            {
+                return Array.Empty<ProcessOutput>();
+            }
+            
+            
+            var sinceTime = scriptPodSinceTimeStore.GetPodEventsSinceTime(scriptTicket);
+
+            var allEvents = await eventService.FetchAllEventsAsync(KubernetesConfig.Namespace, podName, cancellationToken);
+            if (allEvents is null)
+            {
+                return Array.Empty<ProcessOutput>();
+            }
+
+            var relevantEvents = allEvents.Items
+                .Select(e => (Event: e, Occurred: EventHelpers.GetLatestTimestampInEvent(e)))
+                .Where(x => x.Occurred.HasValue)
+                .Select(x => (x.Event, Occurred: new DateTimeOffset(x.Occurred!.Value, TimeSpan.Zero)))
+                .OrderBy(x => x.Occurred)
+                .SkipWhile(e => e.Occurred <= sinceTime);
+
+            var events = relevantEvents.Select((x) =>
+                {
+                    var (ev, occurred) = x;
+
+                    var formattedMessage = $"[POD EVENT] {ev.Reason} | {ev.Message} (Count: {ev.Count ?? 1})";
+
+                    if (ev.IsWarning())
+                        return new WrappedProcessOutput(ProcessOutputSource.StdOut, formattedMessage, occurred, "warning");
+
+                    //if we are pulling a container, show it as a "wait"
+                    if (ev.IsPullingReason())
+                        return new WrappedProcessOutput(ProcessOutputSource.StdOut, formattedMessage, occurred, "wait");
+
+                    //if this is a Pulled event, and we had a Pulling event with the same path (i.e. the same container), then show this as "wait"
+                    if (ev.IsPulledReason() && allEvents.Items.Any(e => e.IsPullingReason() && ev.InvolvedObject.FieldPath == e.InvolvedObject.FieldPath))
+                        return new WrappedProcessOutput(ProcessOutputSource.StdOut, formattedMessage, occurred, "wait");
+
+                    return new ProcessOutput(ProcessOutputSource.Debug, formattedMessage, occurred);
+                })
+                .ToArray();
+
+            if (events.Any())
+            {
+                //update the events since time, so we don't get duplicate events
+                scriptPodSinceTimeStore.UpdatePodEventsSinceTime(scriptTicket, events.Max(o => o.Occurred));
+            }
+
+            return events;
+        }
 
         async Task<Stream?> GetLogStream(string podName, DateTimeOffset? sinceTime, CancellationToken cancellationToken)
         {
@@ -107,5 +185,36 @@ public KubernetesPodLogService(IKubernetesClientConfigProvider configProvider, I
                 }
             }
         }
+
+        class WrappedProcessOutput : ProcessOutput
+        {
+            static readonly TimeSpan OneTick = TimeSpan.FromTicks(1);
+            readonly string wrapper;
+
+            public WrappedProcessOutput(ProcessOutputSource source, string text, DateTimeOffset occurred, string wrapper)
+                : base(source, text, occurred)
+            {
+                this.wrapper = wrapper;
+            }
+
+            public IEnumerable<ProcessOutput> Expand()
+            {
+                return new[]
+                {
+                    //we add the service messages one tick before and after so they are correctly ordered
+                    new ProcessOutput(ProcessOutputSource.StdOut, $"##octopus[stdout-{wrapper}]", Occurred.Subtract(OneTick)),
+                    new ProcessOutput(Source, Text, Occurred),
+                    new ProcessOutput(ProcessOutputSource.StdOut, "##octopus[stdout-default]", Occurred.Add(OneTick)),
+                };
+            }
+        }
+    }
+
+    public static class EventExtensions
+    {
+        public static bool IsPullingReason(this Corev1Event @event) => @event.Reason.Equals("Pulling", StringComparison.OrdinalIgnoreCase);
+        public static bool IsPulledReason(this Corev1Event @event) => @event.Reason.Equals("Pulled", StringComparison.OrdinalIgnoreCase);
+        public static bool IsWarning(this Corev1Event @event) => @event.Type.Equals("Warning", StringComparison.OrdinalIgnoreCase);
+
     }
 }

source/Octopus.Tentacle/Kubernetes/KubernetesConfig.cs

@@ -21,6 +21,7 @@ public static class KubernetesConfig
 
         public static TimeSpan PodsConsideredOrphanedAfterTimeSpan => TimeSpan.FromMinutes(int.TryParse(Environment.GetEnvironmentVariable($"{EnvVarPrefix}__PODSCONSIDEREDORPHANEDAFTERMINUTES"), out var podsConsideredOrphanedAfterTimeSpan) ? podsConsideredOrphanedAfterTimeSpan : 10);
         public static bool DisableAutomaticPodCleanup => bool.TryParse(Environment.GetEnvironmentVariable($"{EnvVarPrefix}__DISABLEAUTOPODCLEANUP"), out var disableAutoCleanup) && disableAutoCleanup;
+        public static bool DisablePodEventsInTaskLog => bool.TryParse(Environment.GetEnvironmentVariable($"{EnvVarPrefix}__DISABLEPODEVENTSINTASKLOG"), out var disable) && disable;
 
         public static string HelmReleaseNameVariableName => $"{EnvVarPrefix}__HELMRELEASENAME";
         public static string HelmReleaseName => GetRequiredEnvVar(HelmReleaseNameVariableName, "Unable to determine Helm release name.");

source/Octopus.Tentacle/Kubernetes/KubernetesEventService.cs

@@ -6,12 +6,14 @@
 using k8s.Autorest;
 using k8s.Models;
 using Octopus.Diagnostics;
+using Octopus.Tentacle.Contracts;
 
 namespace Octopus.Tentacle.Kubernetes
 {
     public interface IKubernetesEventService
     {
         Task<Corev1EventList?> FetchAllEventsAsync(string kubernetesNamespace, CancellationToken cancellationToken);
+        Task<Corev1EventList?> FetchAllEventsAsync(string kubernetesNamespace, string podName, CancellationToken cancellationToken);
     }
 
     public class KubernetesEventService : KubernetesService, IKubernetesEventService
@@ -35,5 +37,25 @@ public KubernetesEventService(IKubernetesClientConfigProvider configProvider, IS
                 }
             });
         }
+
+        public async Task<Corev1EventList?> FetchAllEventsAsync(string kubernetesNamespace, string podName, CancellationToken cancellationToken)
+        {
+            return await RetryPolicy.ExecuteAsync(async () =>
+            { 
+                try
+                {
+                    //get all the events for a specific script pod
+                    return await Client.CoreV1.ListNamespacedEventAsync(
+                        kubernetesNamespace,
+                        fieldSelector: $"involvedObject.name={podName}",
+                        cancellationToken: cancellationToken);
+                }
+                catch (HttpOperationException opException)
+                    when (opException.Response.StatusCode == HttpStatusCode.NotFound)
+                {
+                    return null;
+                }
+            });
+        }
     }
 }

source/Octopus.Tentacle/Kubernetes/ScriptPodSinceTimeStore.cs

@@ -1,33 +1,87 @@
 using System;
 using System.Collections.Concurrent;
 using Octopus.Tentacle.Contracts;
+using Octopus.Tentacle.Scripts;
 
 namespace Octopus.Tentacle.Kubernetes
 {
     public interface IScriptPodSinceTimeStore
     {
-        DateTimeOffset? GetSinceTime(ScriptTicket scriptTicket);
-        void UpdateSinceTime(ScriptTicket scriptTicket, DateTimeOffset nextSinceTime);
+        DateTimeOffset? GetPodLogsSinceTime(ScriptTicket scriptTicket);
+        void UpdatePodLogsSinceTime(ScriptTicket scriptTicket, DateTimeOffset nextSinceTime);
+        DateTimeOffset? GetPodEventsSinceTime(ScriptTicket scriptTicket);
+        void UpdatePodEventsSinceTime(ScriptTicket scriptTicket, DateTimeOffset nextSinceTime);
         void Delete(ScriptTicket scriptTicket);
     }
 
     public class ScriptPodSinceTimeStore : IScriptPodSinceTimeStore
     {
-        readonly ConcurrentDictionary<ScriptTicket, DateTimeOffset?> sinceTimes = new();
-        
-        public DateTimeOffset? GetSinceTime(ScriptTicket scriptTicket)
+        const string PodLogsSinceTimeFilename = "last-pod-log-timestamp.txt";
+        const string PodEventsSinceTimeFilename = "last-pod-event-timestamp.txt";
+
+        readonly IScriptWorkspaceFactory workspaceFactory;
+
+        readonly ConcurrentDictionary<ScriptTicket, Lazy<IScriptWorkspace>> workspaces = new();
+
+        readonly ConcurrentDictionary<ScriptTicket, DateTimeOffset> logsSinceTimes = new();
+        readonly ConcurrentDictionary<ScriptTicket, DateTimeOffset> eventsSinceTimes = new();
+
+        public ScriptPodSinceTimeStore(IScriptWorkspaceFactory workspaceFactory)
         {
-            return sinceTimes.GetOrAdd(scriptTicket, _ => null);
+            this.workspaceFactory = workspaceFactory;
         }
 
-        public void UpdateSinceTime(ScriptTicket scriptTicket, DateTimeOffset nextSinceTime)
+        public DateTimeOffset? GetPodLogsSinceTime(ScriptTicket scriptTicket) => GetTimestampFromMemoryOrDisk(scriptTicket, logsSinceTimes, PodLogsSinceTimeFilename);
+
+        public void UpdatePodLogsSinceTime(ScriptTicket scriptTicket, DateTimeOffset nextSinceTime) => SaveTimestampInMemoryAndDisk(scriptTicket, nextSinceTime, logsSinceTimes, PodLogsSinceTimeFilename);
+
+        public DateTimeOffset? GetPodEventsSinceTime(ScriptTicket scriptTicket) => GetTimestampFromMemoryOrDisk(scriptTicket, eventsSinceTimes, PodEventsSinceTimeFilename);
+
+        public void UpdatePodEventsSinceTime(ScriptTicket scriptTicket, DateTimeOffset nextSinceTime) => SaveTimestampInMemoryAndDisk(scriptTicket, nextSinceTime, eventsSinceTimes, PodEventsSinceTimeFilename);
+
+        IScriptWorkspace GetWorkspace(ScriptTicket scriptTicket) => workspaces.GetOrAdd(scriptTicket, new Lazy<IScriptWorkspace>(() => workspaceFactory.GetWorkspace(scriptTicket))).Value;
+
+        DateTimeOffset? GetTimestampFromMemoryOrDisk(ScriptTicket scriptTicket, ConcurrentDictionary<ScriptTicket, DateTimeOffset> memoryCache, string filename)
+        {
+            //if we have it in memory, all good
+            if (memoryCache.TryGetValue(scriptTicket, out var log))
+            {
+                return log;
+            }
+
+            var workspace = GetWorkspace(scriptTicket);
+            lock (workspace)
+            {
+                //otherwise try and load it from disk
+                var sinceTimeStr = workspace.TryReadFile(filename);
+                var sinceTime = sinceTimeStr is not null && DateTimeOffset.TryParse(sinceTimeStr, out var dto) ? dto : (DateTimeOffset?)null;
+
+                //if we have a value on disk, save it in memory
+                if (sinceTime.HasValue)
+                {
+                    memoryCache[scriptTicket] = sinceTime.Value;
+                }
+
+                return sinceTime;
+            }
+        }
+        
+        void SaveTimestampInMemoryAndDisk(ScriptTicket scriptTicket, DateTimeOffset nextSinceTime, ConcurrentDictionary<ScriptTicket, DateTimeOffset> memoryCache, string podLogsSinceTimeFilename)
         {
-            sinceTimes[scriptTicket] = nextSinceTime;
+            memoryCache[scriptTicket] = nextSinceTime;
+            
+            var workspace = GetWorkspace(scriptTicket);
+            lock (workspace)
+            {
+                workspace.WriteFile(podLogsSinceTimeFilename, nextSinceTime.ToString("O"));
+            }
         }
 
         public void Delete(ScriptTicket scriptTicket)
         {
-            sinceTimes.TryRemove(scriptTicket, out _);
+            workspaces.TryRemove(scriptTicket, out _);
+            logsSinceTimes.TryRemove(scriptTicket, out _);
+            eventsSinceTimes.TryRemove(scriptTicket, out _);
         }
     }
 }

source/Octopus.Tentacle/Scripts/IScriptWorkspace.cs

@@ -22,6 +22,8 @@ public interface IScriptWorkspace
         string LogFilePath { get; }
         void WriteFile(string filename, string contents);
         void CopyFile(string sourceFilePath, string destFileName, bool overwrite);
+        string ReadFile(string filename);
         void CheckReadiness();
+        string? TryReadFile(string filename);
     }
 }