diff --git a/src/Worker/Core/DependencyInjection/DurableTaskWorkerBuilderExtensions.cs b/src/Worker/Core/DependencyInjection/DurableTaskWorkerBuilderExtensions.cs index ee065f52..3f349b71 100644 --- a/src/Worker/Core/DependencyInjection/DurableTaskWorkerBuilderExtensions.cs +++ b/src/Worker/Core/DependencyInjection/DurableTaskWorkerBuilderExtensions.cs @@ -109,4 +109,32 @@ public static IDurableTaskWorkerBuilder UseVersioning(this IDurableTaskWorkerBui }); return builder; } + + /// + /// Adds an orchestration filter to the specified . + /// + /// 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); + 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. + [Obsolete("Experimental")] + 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 b2bdca04..703bbbd4 100644 --- a/src/Worker/Core/DurableTaskWorkerOptions.cs +++ b/src/Worker/Core/DurableTaskWorkerOptions.cs @@ -145,6 +145,12 @@ 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. + /// + [Obsolete("Experimental")] + public IOrchestrationFilter? OrchestrationFilter { get; set; } + /// /// Gets a value indicating whether was explicitly set or not. /// @@ -156,6 +162,7 @@ public DataConverter DataConverter /// internal bool DataConverterExplicitlySet { get; private set; } + /// /// Applies these option values to another. /// @@ -169,6 +176,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/Core/IOrchestrationFilter.cs b/src/Worker/Core/IOrchestrationFilter.cs new file mode 100644 index 00000000..99520d24 --- /dev/null +++ b/src/Worker/Core/IOrchestrationFilter.cs @@ -0,0 +1,35 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT License. + +namespace Microsoft.DurableTask.Worker; + +/// +/// Defines a filter for validating orchestrations. +/// +[Obsolete("Experimental")] +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. + ValueTask IsOrchestrationValidAsync(OrchestrationFilterParameters info, CancellationToken cancellationToken = default); +} + +/// +/// Struct representation of orchestration information. +/// +public struct OrchestrationFilterParameters +{ + /// + /// Gets the name of the orchestration. + /// + public string? Name { get; init; } + + /// + /// Gets the tags associated with the orchestration. + /// + public IReadOnlyDictionary? Tags { get; init; } +} diff --git a/src/Worker/Grpc/GrpcDurableTaskWorker.Processor.cs b/src/Worker/Grpc/GrpcDurableTaskWorker.Processor.cs index 327bb0cc..df2e7ea8 100644 --- a/src/Worker/Grpc/GrpcDurableTaskWorker.Processor.cs +++ b/src/Worker/Grpc/GrpcDurableTaskWorker.Processor.cs @@ -30,13 +30,16 @@ class Processor readonly TaskHubSidecarServiceClient client; readonly DurableTaskShimFactory shimFactory; readonly GrpcDurableTaskWorkerOptions.InternalOptions internalOptions; + [Obsolete("Experimental")] + readonly IOrchestrationFilter? orchestrationFilter; - 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; @@ -374,6 +377,31 @@ async Task OnRunOrchestratorAsync( entityConversionState, cancellationToken); + bool filterPassed = true; + if (this.orchestrationFilter != null) + { + filterPassed = await this.orchestrationFilter.IsOrchestrationValidAsync( + new OrchestrationFilterParameters + { + Name = runtimeState.Name, + Tags = runtimeState.Tags != null ? new Dictionary(runtimeState.Tags) : null, + }, + cancellationToken); + } + + 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/GrpcDurableTaskWorker.cs b/src/Worker/Grpc/GrpcDurableTaskWorker.cs index cb11054d..463af441 100644 --- a/src/Worker/Grpc/GrpcDurableTaskWorker.cs +++ b/src/Worker/Grpc/GrpcDurableTaskWorker.cs @@ -17,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. @@ -27,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); @@ -41,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; } /// @@ -48,7 +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); - await new Processor(this, new(callInvoker)).ExecuteAsync(stoppingToken); + await new Processor(this, new(callInvoker), this.orchestrationFilter).ExecuteAsync(stoppingToken); } #if NET6_0_OR_GREATER 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..b03ff679 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,182 @@ 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()); } + [Obsolete("Experimental")] + [Fact] + public async Task FilterOrchestrationsByName() + { + // 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. We should see the orchestration denied. + orchestrationFilter.NameDenySet.Add(orchestratorName); + + // This should throw as the work is denied. + instanceId = await server.Client.ScheduleNewOrchestrationInstanceAsync(orchestratorName); + 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")] + [Fact] + public async Task FilterOrchestrationsByTag() + { + // 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. We should see the orchestration denied. + orchestrationFilter.TagDenyDict.Add("test", "true"); + + // This should throw as the work is denied. + instanceId = await server.Client.ScheduleNewOrchestrationInstanceAsync(orchestratorName, new StartOrchestrationOptions + { + Tags = orchestratorTags, + }); + 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 ValueTask IsOrchestrationValidAsync(OrchestrationFilterParameters info, CancellationToken cancellationToken = default) + { + return ValueTask.FromResult( + !this.NameDenySet.Contains(info.Name) + && !this.TagDenyDict.Any(kvp => info.Tags != null && 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