diff --git a/.github/workflows/validate-build.yml b/.github/workflows/validate-build.yml new file mode 100644 index 00000000..8a6d8743 --- /dev/null +++ b/.github/workflows/validate-build.yml @@ -0,0 +1,29 @@ +name: Validate Build + +on: + push: + branches: [ main ] + paths-ignore: [ '**.md' ] + pull_request: + branches: [ main ] + paths-ignore: [ '**.md' ] + +jobs: + build: + + runs-on: ubuntu-latest + + steps: + - uses: actions/checkout@v2 + - name: Setup .NET + uses: actions/setup-dotnet@v1 + with: + dotnet-version: 6.0.x + - name: Enable App Service MyGet feed + run: dotnet nuget add source https://www.myget.org/F/azure-appservice/api/v3/index.json --name appservice-myget + - name: Restore dependencies + run: dotnet restore + - name: Build + run: dotnet build --no-restore -p:FileVersionRevision=$GITHUB_RUN_NUMBER + - name: Test + run: dotnet test --no-build --verbosity normal diff --git a/CHANGELOG.md b/CHANGELOG.md index 06736169..c3e61e97 100644 Binary files a/CHANGELOG.md and b/CHANGELOG.md differ diff --git a/DurableTask.Sdk.sln b/DurableTask.Sdk.sln index ae7fc47b..9af9be66 100644 --- a/DurableTask.Sdk.sln +++ b/DurableTask.Sdk.sln @@ -14,6 +14,7 @@ Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "misc", "misc", "{6E392DF7-D LICENSE = LICENSE nuget.config = nuget.config README.md = README.md + .github\workflows\validate-build.yml = .github\workflows\validate-build.yml EndProjectSection EndProject Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "ConsoleApp", "samples\ConsoleApp\ConsoleApp.csproj", "{F9812EB9-A3CD-4E80-8CEA-243AE2AF925F}" diff --git a/src/DurableTask/DurableTask.csproj b/src/DurableTask/DurableTask.csproj index a3b664dc..c177c3db 100644 --- a/src/DurableTask/DurableTask.csproj +++ b/src/DurableTask/DurableTask.csproj @@ -38,7 +38,7 @@ - + diff --git a/src/DurableTask/Grpc/DurableTaskGrpcClient.cs b/src/DurableTask/Grpc/DurableTaskGrpcClient.cs index 4ed49828..b47d77a9 100644 --- a/src/DurableTask/Grpc/DurableTaskGrpcClient.cs +++ b/src/DurableTask/Grpc/DurableTaskGrpcClient.cs @@ -25,6 +25,7 @@ public class DurableTaskGrpcClient : DurableTaskClient readonly IConfiguration? configuration; readonly GrpcChannel sidecarGrpcChannel; readonly TaskHubSidecarServiceClient sidecarClient; + readonly bool ownsChannel; bool isDisposed; @@ -35,8 +36,19 @@ public class DurableTaskGrpcClient : DurableTaskClient this.logger = SdkUtils.GetLogger(builder.loggerFactory ?? this.services.GetService() ?? NullLoggerFactory.Instance); this.configuration = builder.configuration ?? this.services.GetService(); - string sidecarAddress = builder.address ?? SdkUtils.GetSidecarAddress(this.configuration); - this.sidecarGrpcChannel = GrpcChannel.ForAddress(sidecarAddress); + if (builder.channel != null) + { + // Use the channel from the builder, which was given to us by the app (thus we don't own it and can't dispose it) + this.sidecarGrpcChannel = builder.channel; + this.ownsChannel = false; + } + else + { + // We have to create our own channel and are responsible for disposing it + this.sidecarGrpcChannel = GrpcChannel.ForAddress(builder.address ?? SdkUtils.GetSidecarAddress(this.configuration)); + this.ownsChannel = true; + } + this.sidecarClient = new TaskHubSidecarServiceClient(this.sidecarGrpcChannel); } @@ -48,8 +60,11 @@ public override async ValueTask DisposeAsync() { if (!this.isDisposed) { - await this.sidecarGrpcChannel.ShutdownAsync(); - this.sidecarGrpcChannel.Dispose(); + if (this.ownsChannel) + { + await this.sidecarGrpcChannel.ShutdownAsync(); + this.sidecarGrpcChannel.Dispose(); + } GC.SuppressFinalize(this); this.isDisposed = true; @@ -218,6 +233,7 @@ public sealed class Builder internal ILoggerFactory? loggerFactory; internal IDataConverter? dataConverter; internal IConfiguration? configuration; + internal GrpcChannel? channel; internal string? address; public Builder UseLoggerFactory(ILoggerFactory loggerFactory) @@ -238,6 +254,23 @@ public Builder UseAddress(string address) return this; } + /// + /// Configures a to use for communicating with the sidecar process. + /// + /// + /// This builder method allows you to provide your own gRPC channel for communicating with the Durable Task + /// sidecar service. Channels provided using this method won't be disposed when the client is disposed. + /// Rather, the caller remains responsible for shutting down the channel after disposing the client. + /// + /// The gRPC channel to use. + /// Returns this instance. + /// Thrown if is null. + public Builder UseGrpcChannel(GrpcChannel channel) + { + this.channel = channel ?? throw new ArgumentNullException(nameof(channel)); + return this; + } + public Builder UseDataConverter(IDataConverter dataConverter) { this.dataConverter = dataConverter ?? throw new ArgumentNullException(nameof(dataConverter)); diff --git a/src/DurableTask/Grpc/DurableTaskGrpcWorker.cs b/src/DurableTask/Grpc/DurableTaskGrpcWorker.cs index 4647855e..91d6b22d 100644 --- a/src/DurableTask/Grpc/DurableTaskGrpcWorker.cs +++ b/src/DurableTask/Grpc/DurableTaskGrpcWorker.cs @@ -37,6 +37,7 @@ public class DurableTaskGrpcWorker : IHostedService, IAsyncDisposable readonly ILogger logger; readonly IConfiguration? configuration; readonly GrpcChannel sidecarGrpcChannel; + readonly bool ownsChannel; readonly TaskHubSidecarServiceClient sidecarClient; readonly WorkerContext workerContext; @@ -62,8 +63,19 @@ public class DurableTaskGrpcWorker : IHostedService, IAsyncDisposable this.orchestrators = builder.taskProvider.orchestratorsBuilder.ToImmutable(); this.activities = builder.taskProvider.activitiesBuilder.ToImmutable(); - string sidecarAddress = builder.address ?? SdkUtils.GetSidecarAddress(this.configuration); - this.sidecarGrpcChannel = GrpcChannel.ForAddress(sidecarAddress); + if (builder.channel != null) + { + // Use the channel from the builder, which was given to us by the app (thus we don't own it and can't dispose it) + this.sidecarGrpcChannel = builder.channel; + this.ownsChannel = false; + } + else + { + // We have to create our own channel and are responsible for disposing it + this.sidecarGrpcChannel = GrpcChannel.ForAddress(builder.address ?? SdkUtils.GetSidecarAddress(this.configuration)); + this.ownsChannel = true; + } + this.sidecarClient = new TaskHubSidecarServiceClient(this.sidecarGrpcChannel); } @@ -156,6 +168,12 @@ async ValueTask IAsyncDisposable.DisposeAsync() { } + if (this.ownsChannel) + { + await this.sidecarGrpcChannel.ShutdownAsync(); + this.sidecarGrpcChannel.Dispose(); + } + GC.SuppressFinalize(this); } @@ -294,10 +312,34 @@ async Task OnRunOrchestratorAsync(P.OrchestratorRequest request) result.CustomStatus, result.Actions); - this.logger.SendingOrchestratorResponse(name, response.InstanceId, response.Actions.Count); + this.logger.SendingOrchestratorResponse( + name, + response.InstanceId, + response.Actions.Count, + GetActionsListForLogging(response.Actions)); + await this.sidecarClient.CompleteOrchestratorTaskAsync(response); } + static string GetActionsListForLogging(IReadOnlyList actions) + { + if (actions.Count == 0) + { + return string.Empty; + } + else if (actions.Count == 1) + { + return actions[0].OrchestratorActionTypeCase.ToString(); + } + else + { + // Returns something like "ScheduleTask x5, CreateTimer x1,..." + return string.Join(", ", actions + .GroupBy(a => a.OrchestratorActionTypeCase) + .Select(group => $"{group.Key} x{group.Count()}")); + } + } + OrchestratorExecutionResult CreateOrchestrationFailedActionResult(Exception e) { return this.CreateOrchestrationFailedActionResult( @@ -398,6 +440,7 @@ public sealed class Builder internal IServiceProvider? services; internal IConfiguration? configuration; internal string? address; + internal GrpcChannel? channel; internal Builder() { @@ -411,6 +454,23 @@ public Builder UseAddress(string address) return this; } + /// + /// Configures a to use for communicating with the sidecar process. + /// + /// + /// This builder method allows you to provide your own gRPC channel for communicating with the Durable Task + /// sidecar service. Channels provided using this method won't be disposed when the worker is disposed. + /// Rather, the caller remains responsible for shutting down the channel after disposing the worker. + /// + /// The gRPC channel to use. + /// Returns this instance. + /// Thrown if is null. + public Builder UseGrpcChannel(GrpcChannel channel) + { + this.channel = channel ?? throw new ArgumentNullException(nameof(channel)); + return this; + } + public Builder UseLoggerFactory(ILoggerFactory loggerFactory) { this.loggerFactory = loggerFactory ?? throw new ArgumentNullException(nameof(loggerFactory)); diff --git a/src/DurableTask/Logs.cs b/src/DurableTask/Logs.cs index 2a470dfc..0d3c646c 100644 --- a/src/DurableTask/Logs.cs +++ b/src/DurableTask/Logs.cs @@ -24,8 +24,8 @@ static partial class Logs [LoggerMessage(EventId = 10, Level = LogLevel.Debug, Message = "{instanceId}: Received request for '{name}' orchestrator.")] public static partial void ReceivedOrchestratorRequest(this ILogger logger, string name, string instanceId); - [LoggerMessage(EventId = 11, Level = LogLevel.Debug, Message = "{instanceId}: Sending {count} actions for '{name}' orchestrator.")] - public static partial void SendingOrchestratorResponse(this ILogger logger, string name, string instanceId, int count); + [LoggerMessage(EventId = 11, Level = LogLevel.Debug, Message = "{instanceId}: Sending {count} action(s) [{actionsList}] for '{name}' orchestrator.")] + public static partial void SendingOrchestratorResponse(this ILogger logger, string name, string instanceId, int count, string actionsList); [LoggerMessage(EventId = 12, Level = LogLevel.Warning, Message = "{instanceId}: '{name}' orchestrator failed with an unhandled exception: {details}.")] public static partial void OrchestratorFailed(this ILogger logger, string name, string instanceId, string details); diff --git a/test/DurableTask.Sdk.Tests/DurableTask.Sdk.Tests.csproj b/test/DurableTask.Sdk.Tests/DurableTask.Sdk.Tests.csproj index 57e83504..0b732b16 100644 --- a/test/DurableTask.Sdk.Tests/DurableTask.Sdk.Tests.csproj +++ b/test/DurableTask.Sdk.Tests/DurableTask.Sdk.Tests.csproj @@ -1,4 +1,4 @@ - + net6.0 @@ -20,6 +20,10 @@ + + + + diff --git a/test/DurableTask.Sdk.Tests/GrpcSidecarFixture.cs b/test/DurableTask.Sdk.Tests/GrpcSidecarFixture.cs new file mode 100644 index 00000000..961b7be4 --- /dev/null +++ b/test/DurableTask.Sdk.Tests/GrpcSidecarFixture.cs @@ -0,0 +1,80 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT License. + +using System; +using DurableTask.Core; +using DurableTask.Grpc; +using DurableTask.Sidecar; +using DurableTask.Sidecar.Grpc; +using Grpc.Net.Client; +using Microsoft.AspNetCore.Builder; +using Microsoft.AspNetCore.Hosting; +using Microsoft.AspNetCore.Server.Kestrel.Core; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Logging; + +namespace DurableTask.Sdk.Tests; + +public class GrpcSidecarFixture : IDisposable +{ + const string ListenAddress = "http://127.0.0.1:4002"; + + readonly IWebHost host; + readonly GrpcChannel sidecarChannel; + + public GrpcSidecarFixture() + { + InMemoryOrchestrationService service = new(); + + this.host = new WebHostBuilder() + .UseKestrel(options => + { + // Need to force Http2 in Kestrel in unencrypted scenarios + // https://docs.microsoft.com/en-us/aspnet/core/grpc/troubleshoot?view=aspnetcore-3.0 + options.ConfigureEndpointDefaults(listenOptions => listenOptions.Protocols = HttpProtocols.Http2); + }) + .UseUrls(ListenAddress) + .ConfigureServices(services => + { + services.AddGrpc(); + services.AddSingleton(service); + services.AddSingleton(service); + services.AddSingleton(); + }) + .Configure(app => + { + app.UseRouting(); + app.UseEndpoints(endpoints => + { + endpoints.MapGrpcService(); + }); + }) + .Build(); + + this.host.Start(); + + this.sidecarChannel = GrpcChannel.ForAddress(ListenAddress); + } + + public DurableTaskGrpcWorker.Builder GetWorkerBuilder() + { + // The gRPC channel is reused across tests to avoid the overhead of creating new connections (which is very slow) + return DurableTaskGrpcWorker.CreateBuilder().UseGrpcChannel(this.sidecarChannel); + + } + + public DurableTaskGrpcClient.Builder GetClientBuilder() + { + // The gRPC channel is reused across tests to avoid the overhead of creating new connections (which is very slow) + return DurableTaskGrpcClient.CreateBuilder().UseGrpcChannel(this.sidecarChannel); + } + + public void Dispose() + { + this.sidecarChannel.ShutdownAsync().GetAwaiter().GetResult(); + this.sidecarChannel.Dispose(); + + this.host.StopAsync().GetAwaiter().GetResult(); + this.host.Dispose(); + } +} diff --git a/test/DurableTask.Sdk.Tests/OrchestrationPatterns.cs b/test/DurableTask.Sdk.Tests/OrchestrationPatterns.cs index dac787fa..8922b6d4 100644 --- a/test/DurableTask.Sdk.Tests/OrchestrationPatterns.cs +++ b/test/DurableTask.Sdk.Tests/OrchestrationPatterns.cs @@ -9,6 +9,7 @@ using System.Threading; using System.Threading.Tasks; using DurableTask.Grpc; +using DurableTask.Sdk.Tests; using DurableTask.Sdk.Tests.Logging; using Microsoft.Extensions.Logging; using Xunit; @@ -16,12 +17,15 @@ namespace DurableTask.Tests; -public class OrchestrationPatterns : IDisposable +public class OrchestrationPatterns : IClassFixture, IDisposable { readonly CancellationTokenSource testTimeoutSource = new(Debugger.IsAttached ? TimeSpan.FromMinutes(5) : TimeSpan.FromSeconds(10)); readonly ILoggerFactory loggerFactory; - public OrchestrationPatterns(ITestOutputHelper output) + // Documentation on xunit test fixtures: https://xunit.net/docs/shared-context + readonly GrpcSidecarFixture sidecarFixture; + + public OrchestrationPatterns(ITestOutputHelper output, GrpcSidecarFixture sidecarFixture) { TestLogProvider logProvider = new(output); this.loggerFactory = LoggerFactory.Create(builder => @@ -29,6 +33,7 @@ public OrchestrationPatterns(ITestOutputHelper output) builder.AddProvider(logProvider); builder.SetMinimumLevel(LogLevel.Debug); }); + this.sidecarFixture = sidecarFixture; } /// @@ -48,7 +53,7 @@ void IDisposable.Dispose() /// DurableTaskGrpcWorker.Builder CreateWorkerBuilder() { - return DurableTaskGrpcWorker.CreateBuilder().UseLoggerFactory(this.loggerFactory); + return this.sidecarFixture.GetWorkerBuilder().UseLoggerFactory(this.loggerFactory); } /// @@ -56,7 +61,7 @@ DurableTaskGrpcWorker.Builder CreateWorkerBuilder() /// DurableTaskClient CreateDurableTaskClient() { - return DurableTaskGrpcClient.CreateBuilder().UseLoggerFactory(this.loggerFactory).Build(); + return this.sidecarFixture.GetClientBuilder().UseLoggerFactory(this.loggerFactory).Build(); } [Fact]