Merge pull request #537 from serverlessworkflow/feat-workflow-instance-resource-deletion

Updated the `WorkflowInstanceController` to run a background loop to clean-up outlived workflow instances

Author: cdavernas

src/core/Synapse.Core/Resources/OperatorCleanupOptions.cs

@@ -0,0 +1,35 @@
+// Copyright © 2024-Present The Synapse Authors
+//
+// 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.
+
+namespace Synapse.Resources;
+
+/// <summary>
+/// Represents the options used to configure the cleanup behavior of a Synapse Operator
+/// </summary>
+[DataContract]
+public record OperatorCleanupOptions
+{
+
+    /// <summary>
+    /// Gets or sets the time to live for completed workflow instances. Defaults to 7 days. If null, the operator will not delete completed workflow instances.
+    /// </summary>
+    [DataMember(Order = 1, Name = "ttl"), JsonPropertyOrder(1), JsonPropertyName("ttl"), YamlMember(Order = 1, Alias = "ttl")]
+    public TimeSpan? Ttl { get; set; } = TimeSpan.FromDays(7);
+
+    /// <summary>
+    /// Gets or sets the interval at which the operator sweeps for completed workflow instances to delete. Defaults to 5 minutes.
+    /// </summary>
+    [DataMember(Order = 2, Name = "interval"), JsonPropertyOrder(2), JsonPropertyName("interval"), YamlMember(Order = 2, Alias = "interval")]
+    public TimeSpan Interval { get; set; } = TimeSpan.FromMinutes(5);
+
+}

src/core/Synapse.Core/Resources/OperatorSpec.cs

@@ -36,4 +36,10 @@ public record OperatorSpec
     [DataMember(Order = 2, Name = "selector"), JsonPropertyOrder(2), JsonPropertyName("selector"), YamlMember(Order = 2, Alias = "selector")]
     public virtual IDictionary<string, string>? Selector { get; set; }
 
+    /// <summary>
+    /// Gets/sets options controlling retention of completed workflow instances and cleanup sweep behavior.
+    /// </summary>
+    [DataMember(Order = 3, Name = "cleanup"), JsonPropertyOrder(3), JsonPropertyName("cleanup"), YamlMember(Order = 3, Alias = "cleanup")]
+    public virtual OperatorCleanupOptions? Cleanup { get; set; } = new();
+
 }

src/core/Synapse.Core/Resources/RunnerConfiguration.cs

@@ -55,4 +55,5 @@ public record RunnerConfiguration
     /// </summary>
     [DataMember(Order = 5, Name = "publishLifecycleEvents"), JsonPropertyOrder(5), JsonPropertyName("publishLifecycleEvents"), YamlMember(Order = 5, Alias = "publishLifecycleEvents")]
     public virtual bool? PublishLifecycleEvents { get; set; } = true;
+
 }

src/operator/Synapse.Operator/Services/WorkflowInstanceController.cs

@@ -11,7 +11,6 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-using Neuroglia.Data.Infrastructure.ResourceOriented;
 using Neuroglia.Data.Infrastructure.Services;
 
 namespace Synapse.Operator.Services;
@@ -61,6 +60,7 @@ public override async Task StartAsync(CancellationToken cancellationToken)
         await base.StartAsync(cancellationToken).ConfigureAwait(false);
         this.Operator!.Select(b => b.Resource.Spec.Selector).SubscribeAsync(this.OnResourceSelectorChangedAsync, cancellationToken: cancellationToken);
         await this.OnResourceSelectorChangedAsync(this.Operator!.Resource.Spec.Selector).ConfigureAwait(false);
+        if (this.Operator?.Resource?.Spec?.Cleanup != null)_ = Task.Run(this.CleanupAsync, CancellationTokenSource.Token);
     }
 
     /// <inheritdoc/>
@@ -70,6 +70,59 @@ protected override Task ReconcileAsync(CancellationToken cancellationToken = def
         return base.ReconcileAsync(cancellationToken);
     }
 
+    /// <inheritdoc/>
+    protected virtual async Task CleanupAsync()
+    {
+        while (!CancellationTokenSource.IsCancellationRequested)
+        {
+            try
+            {
+                if (this.Operator?.Resource?.Spec?.Cleanup == null) break;
+                var cutoff = DateTimeOffset.UtcNow - this.Operator?.Resource?.Spec?.Cleanup.Ttl;
+                var deleted = 0;
+                var selectors = this.Options.LabelSelectors;
+
+                await foreach (var instance in this.Repository.GetAllAsync<WorkflowInstance>(labelSelectors: selectors, cancellationToken: CancellationTokenSource.Token))
+                {
+                    if (Handlers.ContainsKey(instance.GetQualifiedName())) continue;
+                    if (instance.IsOperative) continue;
+                    var finishedAt = instance.Status?.EndedAt ?? instance.Metadata.CreationTimestamp;
+                    if (finishedAt <= cutoff && !this.Handlers.ContainsKey(instance.GetQualifiedName()))
+                    {
+                        try { await this.TryReleaseAsync(instance, CancellationTokenSource.Token).ConfigureAwait(false); } catch { }
+                        try
+                        {
+                            await this.Repository.RemoveAsync<WorkflowInstance>(instance.GetName(), instance.GetNamespace(), false, CancellationTokenSource.Token).ConfigureAwait(false);
+                            deleted++;
+                        }
+                        catch (Exception ex)
+                        {
+                            Logger.LogWarning(ex, "Failed to delete expired workflow instance {instance}", instance.GetQualifiedName());
+                        }
+                    }
+                }
+            }
+            catch (OperationCanceledException) when (CancellationTokenSource.Token.IsCancellationRequested)
+            {
+                break;
+            }
+            catch (Exception ex)
+            {
+                this.Logger.LogError(ex, "Instance cleanup sweep failed");
+                try { await Task.Delay(TimeSpan.FromSeconds(5), CancellationTokenSource.Token).ConfigureAwait(false); } catch { }
+            }
+            try
+            {
+                var delay = this.Operator?.Resource?.Spec?.Cleanup?.Interval ?? TimeSpan.FromMinutes(5);
+                await Task.Delay(delay, CancellationTokenSource.Token).ConfigureAwait(false);
+            }
+            catch (OperationCanceledException) when (CancellationTokenSource.IsCancellationRequested)
+            {
+                break;
+            }
+        }
+    }
+
     /// <summary>
     /// Creates a new <see cref="WorkflowInstanceHandler"/> for the specified workflow
     /// </summary>

src/operator/Synapse.Operator/Services/WorkflowInstanceHandler.cs

@@ -12,7 +12,6 @@
 // limitations under the License.
 
 using Neuroglia.Data.Infrastructure.Services;
-using Neuroglia.Mediation;
 using System.Text;
 
 namespace Synapse.Operator.Services;