fix(Operator): Fixed the WorkflowInstanceController not to claim instances of workflows explicitly assigned to other operators

Signed-off-by: Charles d'Avernas <[email protected]>

Author: cdavernas

src/operator/Synapse.Operator/Program.cs

@@ -62,6 +62,7 @@
 
         services.AddScoped<WorkflowController>();
         services.AddScoped<IResourceController<Workflow>>(provider => provider.GetRequiredService<WorkflowController>());
+        services.AddScoped<IWorkflowController>(provider => provider.GetRequiredService<WorkflowController>());
 
         services.AddScoped<WorkflowInstanceController>();
         services.AddScoped<IResourceController<WorkflowInstance>>(provider => provider.GetRequiredService<WorkflowInstanceController>());

src/operator/Synapse.Operator/Services/Interfaces/IOperatorController.cs

@@ -14,7 +14,7 @@
 namespace Synapse.Operator.Services;
 
 /// <summary>
-/// Defines the fundamentals of the service used to access the current Synapse Operator
+/// Defines the fundamentals of a service used to access the current Synapse Operator
 /// </summary>
 public interface IOperatorController
     : IHostedService
@@ -25,4 +25,4 @@ public interface IOperatorController
     /// </summary>
     IResourceMonitor<Resources.Operator> Operator { get; }
 
-}
+}

src/operator/Synapse.Operator/Services/Interfaces/IWorkflowController.cs

@@ -0,0 +1,27 @@
+// 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.Operator.Services;
+
+/// <summary>
+/// Defines the fundamentals of a service used to access all monitored workflows
+/// </summary>
+public interface IWorkflowController
+{
+
+    /// <summary>
+    /// Gets a dictionary containing all monitored workflows
+    /// </summary>
+    IReadOnlyDictionary<string, Workflow> Workflows { get; }
+
+}

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

@@ -25,7 +25,7 @@ namespace Synapse.Operator.Services;
 /// <param name="operatorOptions">The current <see cref="Configuration.OperatorOptions"/></param>
 /// <param name="operatorAccessor">The service used to access the current <see cref="Resources.Operator"/></param>
 public class WorkflowController(IServiceProvider serviceProvider, ILoggerFactory loggerFactory, IOptions<ResourceControllerOptions<Workflow>> controllerOptions, IResourceRepository resources, IOptions<OperatorOptions> operatorOptions, IOperatorController operatorAccessor)
-    : ResourceController<Workflow>(loggerFactory, controllerOptions, resources)
+    : ResourceController<Workflow>(loggerFactory, controllerOptions, resources), IWorkflowController
 {
 
     /// <summary>
@@ -48,6 +48,9 @@ public class WorkflowController(IServiceProvider serviceProvider, ILoggerFactory
     /// </summary>
     protected ConcurrentDictionary<string, WorkflowScheduler> Schedulers { get; } = [];
 
+    /// <inheritdoc/>
+    public IReadOnlyDictionary<string, Workflow> Workflows => this.Resources.AsReadOnly();
+
     /// <inheritdoc/>
     public override async Task StartAsync(CancellationToken cancellationToken)
     {

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

@@ -23,8 +23,9 @@ namespace Synapse.Operator.Services;
 /// <param name="controllerOptions">The service used to access the current <see cref="IOptions{TOptions}"/></param>
 /// <param name="repository">The service used to manage <see cref="IResource"/>s</param>
 /// <param name="operatorController">The service used to access the current <see cref="Resources.Operator"/></param>
+/// <param name="workflowController">The service used to access all monitored <see cref="Workflow"/>s</param>
 /// <param name="documents">The <see cref="IRepository"/> used to manage <see cref="Document"/>s</param>
-public class WorkflowInstanceController(IServiceProvider serviceProvider, ILoggerFactory loggerFactory, IOptions<ResourceControllerOptions<WorkflowInstance>> controllerOptions, IResourceRepository repository, IOperatorController operatorController, IRepository<Document, string> documents)
+public class WorkflowInstanceController(IServiceProvider serviceProvider, ILoggerFactory loggerFactory, IOptions<ResourceControllerOptions<WorkflowInstance>> controllerOptions, IResourceRepository repository, IOperatorController operatorController, IWorkflowController workflowController, IRepository<Document, string> documents)
     : ResourceController<WorkflowInstance>(loggerFactory, controllerOptions, repository)
 {
 
@@ -38,6 +39,11 @@ public class WorkflowInstanceController(IServiceProvider serviceProvider, ILogge
     /// </summary>
     protected IResourceMonitor<Resources.Operator> Operator => operatorController.Operator;
 
+    /// <summary>
+    /// Gets a dictionary containing all monitored <see cref="Workflow"/>s
+    /// </summary>
+    protected IReadOnlyDictionary<string, Workflow> Workflows => workflowController.Workflows;
+
     /// <summary>
     /// Gets the <see cref="IRepository"/> used to manage <see cref="Document"/>s
     /// </summary>
@@ -88,6 +94,8 @@ protected virtual async Task<bool> TryClaimAsync(WorkflowInstance resource, Canc
     {
         ArgumentNullException.ThrowIfNull(resource);
         if (resource.Metadata.Labels != null && resource.Metadata.Labels.TryGetValue(SynapseDefaults.Resources.Labels.Operator, out var operatorQualifiedName)) return operatorQualifiedName == this.Operator.Resource.GetQualifiedName();
+        if (this.Workflows.TryGetValue(this.GetResourceCacheKey(resource.Spec.Definition.Name, resource.Spec.Definition.Namespace), out var workflow) && workflow != null 
+            && workflow.Metadata.Labels != null && workflow.Metadata.Labels.TryGetValue(SynapseDefaults.Resources.Labels.Operator, out operatorQualifiedName)) return operatorQualifiedName == this.Operator.Resource.GetQualifiedName();
         try
         {
             var originalResource = resource.Clone();
@@ -113,6 +121,8 @@ protected virtual async Task<bool> TryReleaseAsync(WorkflowInstance resource, Ca
     {
         ArgumentNullException.ThrowIfNull(resource);
         if (resource.Metadata.Labels != null && resource.Metadata.Labels.TryGetValue(SynapseDefaults.Resources.Labels.Operator, out var operatorQualifiedName)) return operatorQualifiedName == this.Operator.Resource.GetQualifiedName();
+        if (this.Workflows.TryGetValue(this.GetResourceCacheKey(resource.Spec.Definition.Name, resource.Spec.Definition.Namespace), out var workflow) && workflow != null
+           && workflow.Metadata.Labels != null && workflow.Metadata.Labels.TryGetValue(SynapseDefaults.Resources.Labels.Operator, out operatorQualifiedName)) return operatorQualifiedName == this.Operator.Resource.GetQualifiedName();
         try
         {
             var originalResource = resource.Clone();