From b3fadcff9494a3270220f5f2871fd5af466d5d6d Mon Sep 17 00:00:00 2001 From: Hal Spang Date: Tue, 1 Jul 2025 15:28:23 -0700 Subject: [PATCH 1/4] Add ability to filter orchestrations at worker This change allows a user to set a lambda that acts a filter for orchestrations. If the lambda returns false, the orchestration will be abandoned, leaving it to be attempted again later. In this initial commit, the filter can work off of the orchestration name and the orchestration tags. Signed-off-by: Hal Spang --- .../DurableTaskWorkerBuilderExtensions.cs | 17 ++++ src/Worker/Core/DurableTaskWorkerOptions.cs | 22 ++++++ .../Grpc/GrpcDurableTaskWorker.Processor.cs | 21 +++++ src/Worker/Grpc/Logs.cs | 3 + .../GrpcSidecar/Grpc/TaskHubGrpcServer.cs | 12 ++- .../OrchestrationPatterns.cs | 79 ++++++++++++++++++- 6 files changed, 150 insertions(+), 4 deletions(-) diff --git a/src/Worker/Core/DependencyInjection/DurableTaskWorkerBuilderExtensions.cs b/src/Worker/Core/DependencyInjection/DurableTaskWorkerBuilderExtensions.cs index ee065f52..9300deea 100644 --- a/src/Worker/Core/DependencyInjection/DurableTaskWorkerBuilderExtensions.cs +++ b/src/Worker/Core/DependencyInjection/DurableTaskWorkerBuilderExtensions.cs @@ -109,4 +109,21 @@ public static IDurableTaskWorkerBuilder UseVersioning(this IDurableTaskWorkerBui }); return builder; } + + /// + /// Adds an orchestration filter to the specified . + /// + /// The builder to set the builder target for. + /// The filter function that determines whether an orchestration should be processed. + /// The same instance, allowing for method chaining. + public static IDurableTaskWorkerBuilder UseOrchestrationFilter(this IDurableTaskWorkerBuilder builder, Func orchestrationFilter) + { + Check.NotNull(builder); + Check.NotNull(orchestrationFilter); + builder.Services.Configure(builder.Name, options => + { + options.OrchestrationFilter = orchestrationFilter; + }); + return builder; + } } diff --git a/src/Worker/Core/DurableTaskWorkerOptions.cs b/src/Worker/Core/DurableTaskWorkerOptions.cs index b2bdca04..35086770 100644 --- a/src/Worker/Core/DurableTaskWorkerOptions.cs +++ b/src/Worker/Core/DurableTaskWorkerOptions.cs @@ -49,6 +49,22 @@ public enum VersionFailureStrategy Fail = 1, } + /// + /// Struct representation of orchestration information. + /// + public struct OrchestrationInfo + { + /// + /// Gets the name of the orchestration. + /// + public string Name { get; init; } + + /// + /// Gets the tags associated with the orchestration. + /// + public Dictionary Tags { get; init; } + } + /// /// Gets or sets the data converter. Default value is . /// @@ -156,6 +172,11 @@ public DataConverter DataConverter /// internal bool DataConverterExplicitlySet { get; private set; } + /// + /// Gets or sets a callback function that determines whether an orchestration should be accepted for work. + /// + public Func? OrchestrationFilter { get; set; } + /// /// Applies these option values to another. /// @@ -169,6 +190,7 @@ internal void ApplyTo(DurableTaskWorkerOptions other) other.MaximumTimerInterval = this.MaximumTimerInterval; other.EnableEntitySupport = this.EnableEntitySupport; other.Versioning = this.Versioning; + other.OrchestrationFilter = this.OrchestrationFilter; } } diff --git a/src/Worker/Grpc/GrpcDurableTaskWorker.Processor.cs b/src/Worker/Grpc/GrpcDurableTaskWorker.Processor.cs index 327bb0cc..eb6427fa 100644 --- a/src/Worker/Grpc/GrpcDurableTaskWorker.Processor.cs +++ b/src/Worker/Grpc/GrpcDurableTaskWorker.Processor.cs @@ -1,6 +1,7 @@ // Copyright (c) Microsoft Corporation. // Licensed under the MIT License. +using System.Collections.Immutable; using System.Text; using DurableTask.Core; using DurableTask.Core.Entities; @@ -12,6 +13,7 @@ using Microsoft.Extensions.DependencyInjection; using Microsoft.Extensions.Logging; using static Microsoft.DurableTask.Protobuf.TaskHubSidecarService; +using static Microsoft.DurableTask.Worker.DurableTaskWorkerOptions; using DTCore = DurableTask.Core; using P = Microsoft.DurableTask.Protobuf; @@ -374,6 +376,25 @@ async Task OnRunOrchestratorAsync( entityConversionState, cancellationToken); + bool filterPassed = this.worker.workerOptions.OrchestrationFilter?.Invoke(new OrchestrationInfo + { + Name = runtimeState.Name, + Tags = new(runtimeState.Tags ?? ImmutableDictionary.Empty), + }) ?? true; + + if (!filterPassed) + { + this.Logger.AbandoningOrchestrationDueToOrchestrationFilter(request.InstanceId, completionToken); + await this.client.AbandonTaskOrchestratorWorkItemAsync( + new P.AbandonOrchestrationTaskRequest + { + CompletionToken = completionToken, + }, + cancellationToken: cancellationToken); + + return; + } + // If versioning has been explicitly set, we attempt to follow that pattern. If it is not set, we don't compare versions here. failureDetails = EvaluateOrchestrationVersioning(versioning, runtimeState.Version, out versionFailure); diff --git a/src/Worker/Grpc/Logs.cs b/src/Worker/Grpc/Logs.cs index 240a4063..b5cc0d9c 100644 --- a/src/Worker/Grpc/Logs.cs +++ b/src/Worker/Grpc/Logs.cs @@ -57,5 +57,8 @@ static partial class Logs [LoggerMessage(EventId = 58, Level = LogLevel.Information, Message = "Abandoning orchestration. InstanceId = '{instanceId}'. Completion token = '{completionToken}'")] public static partial void AbandoningOrchestrationDueToVersioning(this ILogger logger, string instanceId, string completionToken); + + [LoggerMessage(EventId = 59, Level = LogLevel.Information, Message = "Abandoning orchestration due to filtering. InstanceId = '{instanceId}'. Completion token = '{completionToken}'")] + public static partial void AbandoningOrchestrationDueToOrchestrationFilter(this ILogger logger, string instanceId, string completionToken); } } diff --git a/test/Grpc.IntegrationTests/GrpcSidecar/Grpc/TaskHubGrpcServer.cs b/test/Grpc.IntegrationTests/GrpcSidecar/Grpc/TaskHubGrpcServer.cs index 167a437f..192a5a77 100644 --- a/test/Grpc.IntegrationTests/GrpcSidecar/Grpc/TaskHubGrpcServer.cs +++ b/test/Grpc.IntegrationTests/GrpcSidecar/Grpc/TaskHubGrpcServer.cs @@ -6,9 +6,9 @@ using DurableTask.Core; using DurableTask.Core.History; using DurableTask.Core.Query; -using Microsoft.DurableTask.Sidecar.Dispatcher; using Google.Protobuf.WellKnownTypes; using Grpc.Core; +using Microsoft.DurableTask.Sidecar.Dispatcher; using Microsoft.Extensions.Hosting; using Microsoft.Extensions.Logging; using Microsoft.Extensions.Options; @@ -589,6 +589,16 @@ static string GetTaskIdKey(string instanceId, int taskId) return string.Concat(instanceId, "__", taskId.ToString()); } + public override Task AbandonTaskActivityWorkItem(P.AbandonActivityTaskRequest request, ServerCallContext context) + { + return Task.FromResult(new()); + } + + public override Task AbandonTaskOrchestratorWorkItem(P.AbandonOrchestrationTaskRequest request, ServerCallContext context) + { + return Task.FromResult(new()); + } + /// /// A implementation that is used to control whether the task hub /// dispatcher can fetch new work-items, based on whether a client is currently connected. diff --git a/test/Grpc.IntegrationTests/OrchestrationPatterns.cs b/test/Grpc.IntegrationTests/OrchestrationPatterns.cs index 2634935e..40e6779e 100644 --- a/test/Grpc.IntegrationTests/OrchestrationPatterns.cs +++ b/test/Grpc.IntegrationTests/OrchestrationPatterns.cs @@ -3,7 +3,6 @@ using System.Text.Json; using System.Text.Json.Nodes; -using Microsoft.DurableTask.Abstractions; using Microsoft.DurableTask.Client; using Microsoft.DurableTask.Tests.Logging; using Microsoft.DurableTask.Worker; @@ -970,16 +969,90 @@ public async Task RunActivityWithTags() string instanceId = await server.Client.ScheduleNewOrchestrationInstanceAsync( orchestratorName, input: "World", options); - + OrchestrationMetadata metadata = await server.Client.WaitForInstanceCompletionAsync( instanceId, getInputsAndOutputs: true, this.TimeoutToken); - + Assert.NotNull(metadata); Assert.Equal(instanceId, metadata.InstanceId); Assert.Equal(OrchestrationRuntimeStatus.Completed, metadata.RuntimeStatus); Assert.Equal("Hello from tagged activity, World!", metadata.ReadOutputAs()); } + [Fact] + public async Task FilterOrchestrationsByName() + { + // Setup a worker with an Orchestration Filter. + TaskName orchestratorName = nameof(EmptyOrchestration); + ISet orchestratorNames = new HashSet(); + await using HostTestLifetime server = await this.StartWorkerAsync(b => + { + b.AddTasks(tasks => tasks.AddOrchestratorFunc(orchestratorName, ctx => Task.FromResult(null))); + b.UseOrchestrationFilter((info) => + { + return !orchestratorNames.Contains(info.Name); + }); + }); + + // Nothing in the filter set, the orchestration should complete. + string instanceId = await server.Client.ScheduleNewOrchestrationInstanceAsync(orchestratorName); + OrchestrationMetadata metadata = await server.Client.WaitForInstanceCompletionAsync( + instanceId, this.TimeoutToken); + + Assert.NotNull(metadata); + Assert.Equal(instanceId, metadata.InstanceId); + Assert.Equal(OrchestrationRuntimeStatus.Completed, metadata.RuntimeStatus); + + // Update the filter and re-enqueue. We should see the orchestration denied. + orchestratorNames.Add(orchestratorName); + + // This should throw as the work is denied. + instanceId = await server.Client.ScheduleNewOrchestrationInstanceAsync(orchestratorName); + await Assert.ThrowsAsync(async () => await server.Client.WaitForInstanceCompletionAsync(instanceId, new CancellationTokenSource(TimeSpan.FromSeconds(5)).Token)); + } + + [Fact] + public async Task FilterOrchestrationsByTag() + { + // Setup a worker with an Orchestration Filter. + TaskName orchestratorName = nameof(EmptyOrchestration); + IReadOnlyDictionary orchestratorTags = new Dictionary + { + { "test", "true" } + }; + IDictionary orchestratorTagFilter = new Dictionary(); + await using HostTestLifetime server = await this.StartWorkerAsync(b => + { + b.AddTasks(tasks => tasks.AddOrchestratorFunc(orchestratorName, ctx => Task.FromResult(null))); + b.UseOrchestrationFilter((info) => + { + return !orchestratorTagFilter.Any(kvp => info.Tags.ContainsKey(kvp.Key) && info.Tags[kvp.Key] == kvp.Value); + }); + }); + + // Nothing in the filter set, the orchestration should complete. + string instanceId = await server.Client.ScheduleNewOrchestrationInstanceAsync(orchestratorName, new StartOrchestrationOptions + { + Tags = orchestratorTags, + }); + OrchestrationMetadata metadata = await server.Client.WaitForInstanceCompletionAsync( + instanceId, this.TimeoutToken); + + Assert.NotNull(metadata); + Assert.Equal(instanceId, metadata.InstanceId); + Assert.Equal(OrchestrationRuntimeStatus.Completed, metadata.RuntimeStatus); + + // Update the filter and re-enqueue. We should see the orchestration denied. + orchestratorTagFilter.Add("test", "true"); + + // This should throw as the work is denied. + instanceId = await server.Client.ScheduleNewOrchestrationInstanceAsync(orchestratorName, new StartOrchestrationOptions + { + Tags = orchestratorTags, + }); + await Assert.ThrowsAsync(async () => await server.Client.WaitForInstanceCompletionAsync(instanceId, new CancellationTokenSource(TimeSpan.FromSeconds(5)).Token)); + } + // TODO: Test for multiple external events with the same name // TODO: Test for ContinueAsNew with external events that carry over // TODO: Test for catching activity exceptions of specific types From 954b4a76555ba98081cd39fefc08cfb504f35800 Mon Sep 17 00:00:00 2001 From: Hal Spang Date: Tue, 1 Jul 2025 18:32:26 -0700 Subject: [PATCH 2/4] Update to use interface for DI Signed-off-by: Hal Spang --- .../DurableTaskWorkerBuilderExtensions.cs | 23 +++++++++---- src/Worker/Core/DurableTaskWorkerOptions.cs | 25 +++----------- src/Worker/Core/IOrchestrationFilter.cs | 34 +++++++++++++++++++ .../Grpc/GrpcDurableTaskWorker.Processor.cs | 19 +++++++---- src/Worker/Grpc/GrpcDurableTaskWorker.cs | 4 ++- .../OrchestrationPatterns.cs | 31 ++++++++++------- 6 files changed, 90 insertions(+), 46 deletions(-) create mode 100644 src/Worker/Core/IOrchestrationFilter.cs diff --git a/src/Worker/Core/DependencyInjection/DurableTaskWorkerBuilderExtensions.cs b/src/Worker/Core/DependencyInjection/DurableTaskWorkerBuilderExtensions.cs index 9300deea..fea8054c 100644 --- a/src/Worker/Core/DependencyInjection/DurableTaskWorkerBuilderExtensions.cs +++ b/src/Worker/Core/DependencyInjection/DurableTaskWorkerBuilderExtensions.cs @@ -114,16 +114,25 @@ public static IDurableTaskWorkerBuilder UseVersioning(this IDurableTaskWorkerBui /// Adds an orchestration filter to the specified . /// /// The builder to set the builder target for. - /// The filter function that determines whether an orchestration should be processed. + /// The implementation of a that will be bound. /// The same instance, allowing for method chaining. - public static IDurableTaskWorkerBuilder UseOrchestrationFilter(this IDurableTaskWorkerBuilder builder, Func orchestrationFilter) + public static IDurableTaskWorkerBuilder UseOrchestrationFilter(this IDurableTaskWorkerBuilder builder) where TOrchestrationFilter : class, IOrchestrationFilter { Check.NotNull(builder); - Check.NotNull(orchestrationFilter); - builder.Services.Configure(builder.Name, options => - { - options.OrchestrationFilter = orchestrationFilter; - }); + builder.Services.AddSingleton(); + return builder; + } + + /// + /// Adds an orchestration filter to the specified . + /// + /// The builder to set the builder target for. + /// The instance of an to use. + /// The same instance, allowing for method chaining. + public static IDurableTaskWorkerBuilder UseOrchestrationFilter(this IDurableTaskWorkerBuilder builder, IOrchestrationFilter filter) + { + Check.NotNull(builder); + builder.Services.AddSingleton(filter); return builder; } } diff --git a/src/Worker/Core/DurableTaskWorkerOptions.cs b/src/Worker/Core/DurableTaskWorkerOptions.cs index 35086770..85efce58 100644 --- a/src/Worker/Core/DurableTaskWorkerOptions.cs +++ b/src/Worker/Core/DurableTaskWorkerOptions.cs @@ -49,22 +49,6 @@ public enum VersionFailureStrategy Fail = 1, } - /// - /// Struct representation of orchestration information. - /// - public struct OrchestrationInfo - { - /// - /// Gets the name of the orchestration. - /// - public string Name { get; init; } - - /// - /// Gets the tags associated with the orchestration. - /// - public Dictionary Tags { get; init; } - } - /// /// Gets or sets the data converter. Default value is . /// @@ -161,6 +145,11 @@ public DataConverter DataConverter /// public bool IsVersioningSet { get; internal set; } + /// + /// Gets or sets a callback function that determines whether an orchestration should be accepted for work. + /// + public IOrchestrationFilter? OrchestrationFilter { get; set; } + /// /// Gets a value indicating whether was explicitly set or not. /// @@ -172,10 +161,6 @@ public DataConverter DataConverter /// internal bool DataConverterExplicitlySet { get; private set; } - /// - /// Gets or sets a callback function that determines whether an orchestration should be accepted for work. - /// - public Func? OrchestrationFilter { get; set; } /// /// Applies these option values to another. diff --git a/src/Worker/Core/IOrchestrationFilter.cs b/src/Worker/Core/IOrchestrationFilter.cs new file mode 100644 index 00000000..09239169 --- /dev/null +++ b/src/Worker/Core/IOrchestrationFilter.cs @@ -0,0 +1,34 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT License. + +namespace Microsoft.DurableTask.Worker; + +/// +/// Defines a filter for validating orchestrations. +/// +public interface IOrchestrationFilter +{ + /// + /// Validate the orchestration against the filter represented by this interface. + /// + /// The information on the orchestration to validate. + /// The cancellation token for the request to timeout. + /// true if the orchestration is valid false otherwise. + Task IsOrchestrationValidAsync(OrchestrationInfo info, CancellationToken cancellationToken = default); +} + +/// +/// Struct representation of orchestration information. +/// +public struct OrchestrationInfo +{ + /// + /// Gets the name of the orchestration. + /// + public string Name { get; init; } + + /// + /// Gets the tags associated with the orchestration. + /// + public Dictionary Tags { get; init; } +} \ No newline at end of file diff --git a/src/Worker/Grpc/GrpcDurableTaskWorker.Processor.cs b/src/Worker/Grpc/GrpcDurableTaskWorker.Processor.cs index eb6427fa..1b4d51e7 100644 --- a/src/Worker/Grpc/GrpcDurableTaskWorker.Processor.cs +++ b/src/Worker/Grpc/GrpcDurableTaskWorker.Processor.cs @@ -13,7 +13,6 @@ using Microsoft.Extensions.DependencyInjection; using Microsoft.Extensions.Logging; using static Microsoft.DurableTask.Protobuf.TaskHubSidecarService; -using static Microsoft.DurableTask.Worker.DurableTaskWorkerOptions; using DTCore = DurableTask.Core; using P = Microsoft.DurableTask.Protobuf; @@ -32,13 +31,15 @@ class Processor readonly TaskHubSidecarServiceClient client; readonly DurableTaskShimFactory shimFactory; readonly GrpcDurableTaskWorkerOptions.InternalOptions internalOptions; + readonly IOrchestrationFilter? orchestrationFilter = null; - public Processor(GrpcDurableTaskWorker worker, TaskHubSidecarServiceClient client) + public Processor(GrpcDurableTaskWorker worker, TaskHubSidecarServiceClient client, IOrchestrationFilter? orchestrationFilter = null) { this.worker = worker; this.client = client; this.shimFactory = new DurableTaskShimFactory(this.worker.grpcOptions, this.worker.loggerFactory); this.internalOptions = this.worker.grpcOptions.Internal; + this.orchestrationFilter = orchestrationFilter; } ILogger Logger => this.worker.logger; @@ -376,11 +377,17 @@ async Task OnRunOrchestratorAsync( entityConversionState, cancellationToken); - bool filterPassed = this.worker.workerOptions.OrchestrationFilter?.Invoke(new OrchestrationInfo + bool filterPassed = true; + if (this.orchestrationFilter != null) { - Name = runtimeState.Name, - Tags = new(runtimeState.Tags ?? ImmutableDictionary.Empty), - }) ?? true; + filterPassed = await this.orchestrationFilter.IsOrchestrationValidAsync( + new OrchestrationInfo + { + Name = runtimeState.Name, + Tags = new Dictionary(runtimeState.Tags ?? ImmutableDictionary.Empty), + }, + cancellationToken); + } if (!filterPassed) { diff --git a/src/Worker/Grpc/GrpcDurableTaskWorker.cs b/src/Worker/Grpc/GrpcDurableTaskWorker.cs index cb11054d..301c41ba 100644 --- a/src/Worker/Grpc/GrpcDurableTaskWorker.cs +++ b/src/Worker/Grpc/GrpcDurableTaskWorker.cs @@ -2,6 +2,7 @@ // Licensed under the MIT License. using Microsoft.DurableTask.Worker.Hosting; +using Microsoft.Extensions.DependencyInjection; using Microsoft.Extensions.Logging; using Microsoft.Extensions.Options; @@ -48,7 +49,8 @@ protected override async Task ExecuteAsync(CancellationToken stoppingToken) { await using AsyncDisposable disposable = this.GetCallInvoker(out CallInvoker callInvoker, out string address); this.logger.StartingTaskHubWorker(address); - await new Processor(this, new(callInvoker)).ExecuteAsync(stoppingToken); + IOrchestrationFilter? filter = this.services.GetService(); + await new Processor(this, new(callInvoker), filter).ExecuteAsync(stoppingToken); } #if NET6_0_OR_GREATER diff --git a/test/Grpc.IntegrationTests/OrchestrationPatterns.cs b/test/Grpc.IntegrationTests/OrchestrationPatterns.cs index 40e6779e..28eff797 100644 --- a/test/Grpc.IntegrationTests/OrchestrationPatterns.cs +++ b/test/Grpc.IntegrationTests/OrchestrationPatterns.cs @@ -984,14 +984,11 @@ public async Task FilterOrchestrationsByName() { // Setup a worker with an Orchestration Filter. TaskName orchestratorName = nameof(EmptyOrchestration); - ISet orchestratorNames = new HashSet(); + var orchestrationFilter = new OrchestrationFilter(); await using HostTestLifetime server = await this.StartWorkerAsync(b => { b.AddTasks(tasks => tasks.AddOrchestratorFunc(orchestratorName, ctx => Task.FromResult(null))); - b.UseOrchestrationFilter((info) => - { - return !orchestratorNames.Contains(info.Name); - }); + b.UseOrchestrationFilter(orchestrationFilter); }); // Nothing in the filter set, the orchestration should complete. @@ -1004,7 +1001,7 @@ public async Task FilterOrchestrationsByName() Assert.Equal(OrchestrationRuntimeStatus.Completed, metadata.RuntimeStatus); // Update the filter and re-enqueue. We should see the orchestration denied. - orchestratorNames.Add(orchestratorName); + orchestrationFilter.NameDenySet.Add(orchestratorName); // This should throw as the work is denied. instanceId = await server.Client.ScheduleNewOrchestrationInstanceAsync(orchestratorName); @@ -1020,14 +1017,11 @@ public async Task FilterOrchestrationsByTag() { { "test", "true" } }; - IDictionary orchestratorTagFilter = new Dictionary(); + var orchestrationFilter = new OrchestrationFilter(); await using HostTestLifetime server = await this.StartWorkerAsync(b => { b.AddTasks(tasks => tasks.AddOrchestratorFunc(orchestratorName, ctx => Task.FromResult(null))); - b.UseOrchestrationFilter((info) => - { - return !orchestratorTagFilter.Any(kvp => info.Tags.ContainsKey(kvp.Key) && info.Tags[kvp.Key] == kvp.Value); - }); + b.UseOrchestrationFilter(orchestrationFilter); }); // Nothing in the filter set, the orchestration should complete. @@ -1043,7 +1037,7 @@ public async Task FilterOrchestrationsByTag() Assert.Equal(OrchestrationRuntimeStatus.Completed, metadata.RuntimeStatus); // Update the filter and re-enqueue. We should see the orchestration denied. - orchestratorTagFilter.Add("test", "true"); + orchestrationFilter.TagDenyDict.Add("test", "true"); // This should throw as the work is denied. instanceId = await server.Client.ScheduleNewOrchestrationInstanceAsync(orchestratorName, new StartOrchestrationOptions @@ -1053,6 +1047,19 @@ public async Task FilterOrchestrationsByTag() await Assert.ThrowsAsync(async () => await server.Client.WaitForInstanceCompletionAsync(instanceId, new CancellationTokenSource(TimeSpan.FromSeconds(5)).Token)); } + class OrchestrationFilter : IOrchestrationFilter + { + public ISet NameDenySet { get; set; } = new HashSet(); + public IDictionary TagDenyDict = new Dictionary(); + + public Task IsOrchestrationValidAsync(OrchestrationInfo info, CancellationToken cancellationToken = default) + { + return Task.FromResult( + !this.NameDenySet.Contains(info.Name) + && !this.TagDenyDict.Any(kvp => info.Tags.ContainsKey(kvp.Key) && info.Tags[kvp.Key] == kvp.Value)); + } + } + // TODO: Test for multiple external events with the same name // TODO: Test for ContinueAsNew with external events that carry over // TODO: Test for catching activity exceptions of specific types From a27a0f0889ea7ec893830f4ddd7d0a87ba19e60f Mon Sep 17 00:00:00 2001 From: Hal Spang Date: Mon, 7 Jul 2025 15:11:08 -0700 Subject: [PATCH 3/4] Add Obsolete attribute to orchestration filter This commit adds an Obsolete attribute to the new orchestration filtering. This is being done because we want to provide the functionality but also leave the implementation more open. This marks the API as Obsolete("Experimental") to show that. Signed-off-by: Hal Spang --- .../DependencyInjection/DurableTaskWorkerBuilderExtensions.cs | 3 +++ test/Grpc.IntegrationTests/OrchestrationPatterns.cs | 2 ++ 2 files changed, 5 insertions(+) diff --git a/src/Worker/Core/DependencyInjection/DurableTaskWorkerBuilderExtensions.cs b/src/Worker/Core/DependencyInjection/DurableTaskWorkerBuilderExtensions.cs index fea8054c..db0e8705 100644 --- a/src/Worker/Core/DependencyInjection/DurableTaskWorkerBuilderExtensions.cs +++ b/src/Worker/Core/DependencyInjection/DurableTaskWorkerBuilderExtensions.cs @@ -116,6 +116,8 @@ public static IDurableTaskWorkerBuilder UseVersioning(this IDurableTaskWorkerBui /// The builder to set the builder target for. /// The implementation of a that will be bound. /// The same instance, allowing for method chaining. + + [Obsolete("Experimental")] public static IDurableTaskWorkerBuilder UseOrchestrationFilter(this IDurableTaskWorkerBuilder builder) where TOrchestrationFilter : class, IOrchestrationFilter { Check.NotNull(builder); @@ -129,6 +131,7 @@ public static IDurableTaskWorkerBuilder UseOrchestrationFilterThe builder to set the builder target for. /// The instance of an to use. /// The same instance, allowing for method chaining. + [Obsolete("Experimental")] public static IDurableTaskWorkerBuilder UseOrchestrationFilter(this IDurableTaskWorkerBuilder builder, IOrchestrationFilter filter) { Check.NotNull(builder); diff --git a/test/Grpc.IntegrationTests/OrchestrationPatterns.cs b/test/Grpc.IntegrationTests/OrchestrationPatterns.cs index 28eff797..c8137674 100644 --- a/test/Grpc.IntegrationTests/OrchestrationPatterns.cs +++ b/test/Grpc.IntegrationTests/OrchestrationPatterns.cs @@ -979,6 +979,7 @@ public async Task RunActivityWithTags() Assert.Equal("Hello from tagged activity, World!", metadata.ReadOutputAs()); } + [Obsolete("Experimental")] [Fact] public async Task FilterOrchestrationsByName() { @@ -1008,6 +1009,7 @@ public async Task FilterOrchestrationsByName() await Assert.ThrowsAsync(async () => await server.Client.WaitForInstanceCompletionAsync(instanceId, new CancellationTokenSource(TimeSpan.FromSeconds(5)).Token)); } + [Obsolete("Experimental")] [Fact] public async Task FilterOrchestrationsByTag() { From 790a293ea94cb5adc10b1f7f70d3d8c2bc0eb20c Mon Sep 17 00:00:00 2001 From: Hal Spang Date: Mon, 7 Jul 2025 23:09:58 -0700 Subject: [PATCH 4/4] Expand obsolete usage add tests Signed-off-by: Hal Spang --- .../DurableTaskWorkerBuilderExtensions.cs | 1 - src/Worker/Core/DurableTaskWorkerOptions.cs | 1 + src/Worker/Core/IOrchestrationFilter.cs | 11 ++- .../Grpc/GrpcDurableTaskWorker.Processor.cs | 8 +- src/Worker/Grpc/GrpcDurableTaskWorker.cs | 10 +- .../OrchestrationPatterns.cs | 93 ++++++++++++++++++- 6 files changed, 105 insertions(+), 19 deletions(-) diff --git a/src/Worker/Core/DependencyInjection/DurableTaskWorkerBuilderExtensions.cs b/src/Worker/Core/DependencyInjection/DurableTaskWorkerBuilderExtensions.cs index db0e8705..3f349b71 100644 --- a/src/Worker/Core/DependencyInjection/DurableTaskWorkerBuilderExtensions.cs +++ b/src/Worker/Core/DependencyInjection/DurableTaskWorkerBuilderExtensions.cs @@ -116,7 +116,6 @@ public static IDurableTaskWorkerBuilder UseVersioning(this IDurableTaskWorkerBui /// The builder to set the builder target for. /// The implementation of a that will be bound. /// The same instance, allowing for method chaining. - [Obsolete("Experimental")] public static IDurableTaskWorkerBuilder UseOrchestrationFilter(this IDurableTaskWorkerBuilder builder) where TOrchestrationFilter : class, IOrchestrationFilter { diff --git a/src/Worker/Core/DurableTaskWorkerOptions.cs b/src/Worker/Core/DurableTaskWorkerOptions.cs index 85efce58..703bbbd4 100644 --- a/src/Worker/Core/DurableTaskWorkerOptions.cs +++ b/src/Worker/Core/DurableTaskWorkerOptions.cs @@ -148,6 +148,7 @@ public DataConverter DataConverter /// /// Gets or sets a callback function that determines whether an orchestration should be accepted for work. /// + [Obsolete("Experimental")] public IOrchestrationFilter? OrchestrationFilter { get; set; } /// diff --git a/src/Worker/Core/IOrchestrationFilter.cs b/src/Worker/Core/IOrchestrationFilter.cs index 09239169..99520d24 100644 --- a/src/Worker/Core/IOrchestrationFilter.cs +++ b/src/Worker/Core/IOrchestrationFilter.cs @@ -6,6 +6,7 @@ namespace Microsoft.DurableTask.Worker; /// /// Defines a filter for validating orchestrations. /// +[Obsolete("Experimental")] public interface IOrchestrationFilter { /// @@ -14,21 +15,21 @@ public interface IOrchestrationFilter /// The information on the orchestration to validate. /// The cancellation token for the request to timeout. /// true if the orchestration is valid false otherwise. - Task IsOrchestrationValidAsync(OrchestrationInfo info, CancellationToken cancellationToken = default); + ValueTask IsOrchestrationValidAsync(OrchestrationFilterParameters info, CancellationToken cancellationToken = default); } /// /// Struct representation of orchestration information. /// -public struct OrchestrationInfo +public struct OrchestrationFilterParameters { /// /// Gets the name of the orchestration. /// - public string Name { get; init; } + public string? Name { get; init; } /// /// Gets the tags associated with the orchestration. /// - public Dictionary Tags { get; init; } -} \ No newline at end of file + public IReadOnlyDictionary? Tags { get; init; } +} diff --git a/src/Worker/Grpc/GrpcDurableTaskWorker.Processor.cs b/src/Worker/Grpc/GrpcDurableTaskWorker.Processor.cs index 1b4d51e7..df2e7ea8 100644 --- a/src/Worker/Grpc/GrpcDurableTaskWorker.Processor.cs +++ b/src/Worker/Grpc/GrpcDurableTaskWorker.Processor.cs @@ -1,7 +1,6 @@ // Copyright (c) Microsoft Corporation. // Licensed under the MIT License. -using System.Collections.Immutable; using System.Text; using DurableTask.Core; using DurableTask.Core.Entities; @@ -31,7 +30,8 @@ class Processor readonly TaskHubSidecarServiceClient client; readonly DurableTaskShimFactory shimFactory; readonly GrpcDurableTaskWorkerOptions.InternalOptions internalOptions; - readonly IOrchestrationFilter? orchestrationFilter = null; + [Obsolete("Experimental")] + readonly IOrchestrationFilter? orchestrationFilter; public Processor(GrpcDurableTaskWorker worker, TaskHubSidecarServiceClient client, IOrchestrationFilter? orchestrationFilter = null) { @@ -381,10 +381,10 @@ async Task OnRunOrchestratorAsync( if (this.orchestrationFilter != null) { filterPassed = await this.orchestrationFilter.IsOrchestrationValidAsync( - new OrchestrationInfo + new OrchestrationFilterParameters { Name = runtimeState.Name, - Tags = new Dictionary(runtimeState.Tags ?? ImmutableDictionary.Empty), + Tags = runtimeState.Tags != null ? new Dictionary(runtimeState.Tags) : null, }, cancellationToken); } diff --git a/src/Worker/Grpc/GrpcDurableTaskWorker.cs b/src/Worker/Grpc/GrpcDurableTaskWorker.cs index 301c41ba..463af441 100644 --- a/src/Worker/Grpc/GrpcDurableTaskWorker.cs +++ b/src/Worker/Grpc/GrpcDurableTaskWorker.cs @@ -2,7 +2,6 @@ // Licensed under the MIT License. using Microsoft.DurableTask.Worker.Hosting; -using Microsoft.Extensions.DependencyInjection; using Microsoft.Extensions.Logging; using Microsoft.Extensions.Options; @@ -18,6 +17,7 @@ sealed partial class GrpcDurableTaskWorker : DurableTaskWorker readonly IServiceProvider services; readonly ILoggerFactory loggerFactory; readonly ILogger logger; + readonly IOrchestrationFilter? orchestrationFilter; /// /// Initializes a new instance of the class. @@ -28,13 +28,15 @@ sealed partial class GrpcDurableTaskWorker : DurableTaskWorker /// The generic worker options. /// The service provider. /// The logger. + /// The optional used to filter orchestration execution. public GrpcDurableTaskWorker( string name, IDurableTaskFactory factory, IOptionsMonitor grpcOptions, IOptionsMonitor workerOptions, IServiceProvider services, - ILoggerFactory loggerFactory) + ILoggerFactory loggerFactory, + IOrchestrationFilter? orchestrationFilter = null) : base(name, factory) { this.grpcOptions = Check.NotNull(grpcOptions).Get(name); @@ -42,6 +44,7 @@ public GrpcDurableTaskWorker( this.services = Check.NotNull(services); this.loggerFactory = Check.NotNull(loggerFactory); this.logger = loggerFactory.CreateLogger("Microsoft.DurableTask"); // TODO: use better category name. + this.orchestrationFilter = orchestrationFilter; } /// @@ -49,8 +52,7 @@ protected override async Task ExecuteAsync(CancellationToken stoppingToken) { await using AsyncDisposable disposable = this.GetCallInvoker(out CallInvoker callInvoker, out string address); this.logger.StartingTaskHubWorker(address); - IOrchestrationFilter? filter = this.services.GetService(); - await new Processor(this, new(callInvoker), filter).ExecuteAsync(stoppingToken); + await new Processor(this, new(callInvoker), this.orchestrationFilter).ExecuteAsync(stoppingToken); } #if NET6_0_OR_GREATER diff --git a/test/Grpc.IntegrationTests/OrchestrationPatterns.cs b/test/Grpc.IntegrationTests/OrchestrationPatterns.cs index c8137674..b03ff679 100644 --- a/test/Grpc.IntegrationTests/OrchestrationPatterns.cs +++ b/test/Grpc.IntegrationTests/OrchestrationPatterns.cs @@ -1006,7 +1006,43 @@ public async Task FilterOrchestrationsByName() // This should throw as the work is denied. instanceId = await server.Client.ScheduleNewOrchestrationInstanceAsync(orchestratorName); - await Assert.ThrowsAsync(async () => await server.Client.WaitForInstanceCompletionAsync(instanceId, new CancellationTokenSource(TimeSpan.FromSeconds(5)).Token)); + using CancellationTokenSource cts = new CancellationTokenSource(TimeSpan.FromSeconds(5)); + await Assert.ThrowsAsync(async () => await server.Client.WaitForInstanceCompletionAsync(instanceId, cts.Token)); + } + + [Obsolete("Experimental")] + [Fact] + public async Task FilterOrchestrationsByNamePassesWhenNotMatching() + { + // Setup a worker with an Orchestration Filter. + TaskName orchestratorName = nameof(EmptyOrchestration); + var orchestrationFilter = new OrchestrationFilter(); + await using HostTestLifetime server = await this.StartWorkerAsync(b => + { + b.AddTasks(tasks => tasks.AddOrchestratorFunc(orchestratorName, ctx => Task.FromResult(null))); + b.UseOrchestrationFilter(orchestrationFilter); + }); + + // Nothing in the filter set, the orchestration should complete. + string instanceId = await server.Client.ScheduleNewOrchestrationInstanceAsync(orchestratorName); + OrchestrationMetadata metadata = await server.Client.WaitForInstanceCompletionAsync( + instanceId, this.TimeoutToken); + + Assert.NotNull(metadata); + Assert.Equal(instanceId, metadata.InstanceId); + Assert.Equal(OrchestrationRuntimeStatus.Completed, metadata.RuntimeStatus); + + // Update the filter and re-enqueue. The name doesn't match so the filter should be OK. + orchestrationFilter.NameDenySet.Add($"not-{orchestratorName}"); + + // This should throw as the work is denied. + instanceId = await server.Client.ScheduleNewOrchestrationInstanceAsync(orchestratorName); + metadata = await server.Client.WaitForInstanceCompletionAsync( + instanceId, this.TimeoutToken); + + Assert.NotNull(metadata); + Assert.Equal(instanceId, metadata.InstanceId); + Assert.Equal(OrchestrationRuntimeStatus.Completed, metadata.RuntimeStatus); } [Obsolete("Experimental")] @@ -1046,19 +1082,66 @@ public async Task FilterOrchestrationsByTag() { Tags = orchestratorTags, }); - await Assert.ThrowsAsync(async () => await server.Client.WaitForInstanceCompletionAsync(instanceId, new CancellationTokenSource(TimeSpan.FromSeconds(5)).Token)); + using CancellationTokenSource cts = new CancellationTokenSource(TimeSpan.FromSeconds(5)); + await Assert.ThrowsAsync(async () => await server.Client.WaitForInstanceCompletionAsync(instanceId, cts.Token)); } + [Obsolete("Experimental")] + [Fact] + public async Task FilterOrchestrationsByTagPassesWithNoMatch() + { + // Setup a worker with an Orchestration Filter. + TaskName orchestratorName = nameof(EmptyOrchestration); + IReadOnlyDictionary orchestratorTags = new Dictionary + { + { "test", "true" } + }; + var orchestrationFilter = new OrchestrationFilter(); + await using HostTestLifetime server = await this.StartWorkerAsync(b => + { + b.AddTasks(tasks => tasks.AddOrchestratorFunc(orchestratorName, ctx => Task.FromResult(null))); + b.UseOrchestrationFilter(orchestrationFilter); + }); + + // Nothing in the filter set, the orchestration should complete. + string instanceId = await server.Client.ScheduleNewOrchestrationInstanceAsync(orchestratorName, new StartOrchestrationOptions + { + Tags = orchestratorTags, + }); + OrchestrationMetadata metadata = await server.Client.WaitForInstanceCompletionAsync( + instanceId, this.TimeoutToken); + + Assert.NotNull(metadata); + Assert.Equal(instanceId, metadata.InstanceId); + Assert.Equal(OrchestrationRuntimeStatus.Completed, metadata.RuntimeStatus); + + // Update the filter and re-enqueue. The tags don't match so the orchestration should be OK. + orchestrationFilter.TagDenyDict.Add("test", "false"); + + // This should throw as the work is denied. + instanceId = await server.Client.ScheduleNewOrchestrationInstanceAsync(orchestratorName, new StartOrchestrationOptions + { + Tags = orchestratorTags, + }); + metadata = await server.Client.WaitForInstanceCompletionAsync( + instanceId, this.TimeoutToken); + + Assert.NotNull(metadata); + Assert.Equal(instanceId, metadata.InstanceId); + Assert.Equal(OrchestrationRuntimeStatus.Completed, metadata.RuntimeStatus); + } + + [Obsolete("Experimental")] class OrchestrationFilter : IOrchestrationFilter { public ISet NameDenySet { get; set; } = new HashSet(); public IDictionary TagDenyDict = new Dictionary(); - public Task IsOrchestrationValidAsync(OrchestrationInfo info, CancellationToken cancellationToken = default) + public ValueTask IsOrchestrationValidAsync(OrchestrationFilterParameters info, CancellationToken cancellationToken = default) { - return Task.FromResult( + return ValueTask.FromResult( !this.NameDenySet.Contains(info.Name) - && !this.TagDenyDict.Any(kvp => info.Tags.ContainsKey(kvp.Key) && info.Tags[kvp.Key] == kvp.Value)); + && !this.TagDenyDict.Any(kvp => info.Tags != null && info.Tags.ContainsKey(kvp.Key) && info.Tags[kvp.Key] == kvp.Value)); } }