Feature/workflow http/grpc record interceptor

.github/workflows/build_and_it.yml

@@ -51,7 +51,8 @@ jobs:
           tar -xvf dtm_1.18.0_linux_amd64.tar.gz
           pwd
           mkdir /home/runner/work/client-csharp/client-csharp/logs
-          nohup ./dtm > /home/runner/work/client-csharp/client-csharp/logs/dtm.log 2>&1 &
+          echo "UpdateBranchSync: 1" > ./config.yml
+          nohup ./dtm -c ./config.yml > /home/runner/work/client-csharp/client-csharp/logs/dtm.log 2>&1 &
           sleep 5
           curl "127.0.0.1:36789/api/dtmsvr/newGid"
       - name: Setup Busi Service

src/DtmCommon/Constant.cs

@@ -67,8 +67,9 @@ public class Barrier
 
             public static readonly Dictionary<string, string> OpDict = new Dictionary<string, string>()
             {
-                { "cancel", "try" },
-                { "compensate", "action" },
+                { "cancel", "try" }, // tcc
+                { "compensate", "action" }, // saga
+                { "rollback", "action" }, // workflow
             };
             public static readonly string REDIS_LUA_CheckAdjustAmount = @" -- RedisCheckAdjustAmount
 local v = redis.call('GET', KEYS[1])

src/Dtmcli/Constant.cs

@@ -4,7 +4,8 @@ internal static class Constant
     {
         internal const string DtmClientHttpName = "dtmClient";
         internal const string BranchClientHttpName = "branchClient";
-
+        internal const string WorkflowBranchClientHttpName = "WF";
+
         internal static class Request
         {
             internal const string CONTENT_TYPE = "application/json";

src/Dtmcli/DtmClient.cs

@@ -148,7 +148,7 @@ public async Task<TransGlobal> Query(string gid, CancellationToken cancellationT
             var client = _httpClientFactory.CreateClient(Constant.DtmClientHttpName);
             var response = await client.GetAsync(url, cancellationToken).ConfigureAwait(false);
             var dtmContent = await response.Content.ReadAsStringAsync().ConfigureAwait(false);
-            DtmImp.Utils.CheckStatus(response.StatusCode, dtmContent);
+            DtmImp.Utils.CheckStatusCode(response.StatusCode);
             return JsonSerializer.Deserialize<TransGlobal>(dtmContent, _jsonOptions);
         }
 
@@ -167,7 +167,7 @@ public async Task<string> QueryStatus(string gid, CancellationToken cancellation
             var client = _httpClientFactory.CreateClient(Constant.DtmClientHttpName);
             var response = await client.GetAsync(url, cancellationToken).ConfigureAwait(false);
             var dtmContent = await response.Content.ReadAsStringAsync().ConfigureAwait(false);
-            DtmImp.Utils.CheckStatus(response.StatusCode, dtmContent);
+            DtmImp.Utils.CheckStatusCode(response.StatusCode);
             var graph = JsonSerializer.Deserialize<TransGlobalForStatus>(dtmContent, _jsonOptions);
             return graph.Transaction == null
                 ? string.Empty

src/Dtmcli/DtmImp/Utils.cs

@@ -40,6 +40,14 @@ public static void CheckStatus(HttpStatusCode status, string dtmResult)
             }
         }
 
+        public static void CheckStatusCode(HttpStatusCode status)
+        {
+            if (status != HttpStatusCode.OK)
+            {
+                throw new DtmException(string.Format(CheckStatusMsgFormat, status.ToString(), string.Empty));
+            }
+        }
+
         /// <summary>
         /// OrString return the first not null or not empty string
         /// </summary>

src/Dtmcli/TransGlobal.cs

@@ -28,7 +28,7 @@ public class DtmTransaction
     {
         [JsonPropertyName("id")] public int Id { get; set; }
 
-        [JsonPropertyName("create_time")] public DateTimeOffset CreateTime { get; set; }
+        [JsonPropertyName("create_time")] public DateTimeOffset? CreateTime { get; set; }
 
         [JsonPropertyName("update_time")] public DateTimeOffset UpdateTime { get; set; }
 
@@ -64,9 +64,9 @@ public class DtmBranch
     {
         [JsonPropertyName("id")] public int Id { get; set; }
 
-        [JsonPropertyName("create_time")] public DateTimeOffset CreateTime { get; set; }
+        [JsonPropertyName("create_time")] public DateTimeOffset? CreateTime { get; set; }
 
-        [JsonPropertyName("update_time")] public DateTimeOffset UpdateTime { get; set; }
+        [JsonPropertyName("update_time")] public DateTimeOffset? UpdateTime { get; set; }
 
         [JsonPropertyName("gid")] public string Gid { get; set; }
 

src/Dtmworkflow/ServiceCollectionExtensions.cs

@@ -23,6 +23,8 @@ public static IServiceCollection AddDtmWorkflow(this IServiceCollection services
             services.TryAddSingleton<IWorkflowFactory, WorkflowFactory>();
             services.TryAddSingleton<WorkflowGlobalTransaction>();
 
+            // AddHttpClient(services);
+
             return services;
         }
 
@@ -33,8 +35,22 @@ public static IServiceCollection AddDtmWorkflow(this IServiceCollection services
 
             services.TryAddSingleton<IWorkflowFactory, WorkflowFactory>();
             services.TryAddSingleton<WorkflowGlobalTransaction>();
+
+            // AddHttpClient(services);
 
             return services;
         }
+
+        // private static void AddHttpClient(IServiceCollection services /*, DtmOptions options*/)
+        // {
+        //     services.AddHttpClient(Dtmcli.Constant.WorkflowBranchClientHttpName, client =>
+        //     {
+        //         // TODO DtmOptions
+        //         // client.Timeout = TimeSpan.FromMilliseconds(options.BranchTimeout);                
+        //     }).AddHttpMessageHandler<WorkflowHttpInterceptor>();
+        //
+        //     // TODO how to inject workflow instance?
+        //     services.AddTransient<WorkflowHttpInterceptor>();
+        // }
     }
-}
+}

src/Dtmworkflow/Workflow.Imp.cs

@@ -64,7 +64,7 @@
             }
 
             err = Utils.GrpcError2DtmError(err);
-
+            
             if (err != null && err is not DtmCommon.DtmFailureException) throw err;
 
             try
@@ -161,23 +161,23 @@
             };
         }
 
-        private Exception StepResultToGrpc(StepResult r, IMessage reply)
+        internal Exception StepResultToGrpc(StepResult r, IMessage reply)
         {
             if (r.Error == null && r.Status == DtmCommon.Constant.StatusSucceed)
             {
-                // Check 
-
+                // TODO Check 
+                // dtmgimp.MustProtoUnmarshal(s.Data, reply.(protoreflect.ProtoMessage));
             }
 
             return r.Error;
         }
 
-        private StepResult StepResultFromGrpc(IMessage reply, Exception err)
+        internal StepResult StepResultFromGrpc(IMessage reply, Exception err)
         {
             var sr = new StepResult
             {
-                // GRPCError2DtmError
-                Error = null,
+                // TODO GRPCError2DtmError
+                Error = Utils.GrpcError2DtmError(err),
             };
 
             sr.Status = WfErrorToStatus(sr.Error);
@@ -193,7 +193,7 @@
             return sr;
         }
 
-        private HttpResponseMessage StepResultToHttp(StepResult r)
+        internal HttpResponseMessage StepResultToHttp(StepResult r)
         {
             if (r.Error != null)
             {
@@ -203,7 +203,7 @@
             return Utils.NewJSONResponse(HttpStatusCode.OK, r.Data);
         }
 
-        private StepResult StepResultFromHTTP(HttpResponseMessage resp, Exception err)
+        internal StepResult StepResultFromHTTP(HttpResponseMessage resp, Exception err)
         {
             var sr = new StepResult
             {
@@ -212,7 +212,7 @@
 
             if (err == null)
             {
-                // HTTPResp2DtmError
+                (sr.Data, sr.Error) = Utils.HTTPResp2DtmError(resp); // TODO go 使用了 this.Options.HTTPResp2DtmError(resp), 方便定制
                 sr.Status = WfErrorToStatus(sr.Error);
             }
 
@@ -234,9 +234,9 @@
         }
 
 
-        private async Task<StepResult> RecordedDo(Func<DtmCommon.BranchBarrier, Task<StepResult>> fn)
+        internal async Task<StepResult> RecordedDo(Func<DtmCommon.BranchBarrier, Task<StepResult>> fn)
         {
-            var sr = await this.RecordedDoInner(fn);
+            StepResult sr = await this.RecordedDoInner(fn);
 
             // do not compensate the failed branch if !CompensateErrorBranch
             if (this.Options.CompensateErrorBranch && sr.Status == DtmCommon.Constant.StatusFailed)

src/Dtmworkflow/Workflow.cs

@@ -3,7 +3,9 @@
 using Microsoft.Extensions.Logging;
 using System;
 using System.Collections.Generic;
+using System.Net.Http;
 using System.Threading.Tasks;
+using Microsoft.Extensions.DependencyInjection;
 
 namespace Dtmworkflow
 {
@@ -32,7 +34,8 @@ public Workflow(IDtmClient httpClient, IDtmgRPCClient grpcClient, Dtmcli.IBranch
 
         public System.Net.Http.HttpClient NewRequest()
         {
-            return _httpClient.GetHttpClient("WF");
+            // return _httpClient.GetHttpClient("WF");
+            return new HttpClient(new WorkflowHttpInterceptor(this));
         }
 
         /// <summary>

src/Dtmworkflow/WorkflowGrpcInterceptor.cs

@@ -0,0 +1,103 @@
+using System;
+using System.Threading.Tasks;
+using Google.Protobuf;
+using Grpc.Core;
+using Grpc.Core.Interceptors;
+using Microsoft.Extensions.Logging;
+
+namespace Dtmworkflow;
+
+public class WorkflowGrpcInterceptor(Workflow wf, ILogger<WorkflowGrpcInterceptor> logger) : Interceptor
+{
+    public WorkflowGrpcInterceptor(Workflow wf) : this(wf, null)
+    {
+    }
+
+    public override AsyncUnaryCall<TResponse> AsyncUnaryCall<TRequest, TResponse>(
+        TRequest request,
+        ClientInterceptorContext<TRequest, TResponse> context,
+        AsyncUnaryCallContinuation<TRequest, TResponse> continuation)
+    {
+        logger?.LogDebug($"grpc client calling: {context.Host}{context.Method.FullName}");
+
+        if (wf == null)
+        {
+            return base.AsyncUnaryCall(request, context, continuation);
+        }
+
+        async Task<(AsyncUnaryCall<TResponse>, TResponse, Status)> Origin()
+        {
+            var newContext = Dtmgimp.TransInfo2Ctx(context, wf.TransBase.Gid, wf.TransBase.TransType, wf.WorkflowImp.CurrentBranch, wf.WorkflowImp.CurrentOp, wf.TransBase.Dtm);
+
+            var call = continuation(request, newContext);
+            TResponse response;
+            try
+            {
+                response = await call.ResponseAsync;
+            }
+            catch (Exception e)
+            {
+                logger?.LogDebug($"grpc client: {context.Host}{context.Method.FullName} ex: {e}");
+                response = null;
+            }
+
+            Status status = call.GetStatus();
+            return (
+                new AsyncUnaryCall<TResponse>(
+                    call.ResponseAsync,
+                    call.ResponseHeadersAsync,
+                    call.GetStatus,
+                    call.GetTrailers,
+                    call.Dispose),
+                response,
+                status
+            );
+        }
+
+        if (wf.WorkflowImp.CurrentOp != DtmCommon.Constant.OpAction)
+        {
+            var (newCall, _, _) = Origin().GetAwaiter().GetResult();
+            return newCall;
+        }
+
+        AsyncUnaryCall<TResponse> call = null;
+        StepResult sr = wf.RecordedDo(bb =>
+        {
+            (call, TResponse data, Status status) = Origin().GetAwaiter().GetResult();
+            RpcException err = status.StatusCode != StatusCode.OK ? new RpcException(status) : null;
+            return Task.FromResult(wf.StepResultFromGrpc(data as IMessage, err));
+        }).GetAwaiter().GetResult();
+        wf.StepResultToGrpc(sr, null);
+
+        return call;
+    }
+
+    private class Dtmgimp
+    {
+        public static ClientInterceptorContext<TRequest, TResponse> TransInfo2Ctx<TRequest, TResponse>(
+            ClientInterceptorContext<TRequest, TResponse> ctx,
+            string gid,
+            string transType,
+            string branchID,
+            string op,
+            string dtm) where TRequest : class where TResponse : class
+        {
+            // 创建一个新的元数据对象
+            var headers = new Metadata();
+            // 添加自定义元数据
+            const string dtmpre = "dtm-";
+            headers.Add(dtmpre + "gid", gid);
+            headers.Add(dtmpre + "trans_type", transType);
+            headers.Add(dtmpre + "branch_id", branchID);
+            headers.Add(dtmpre + "op", op);
+            headers.Add(dtmpre + "dtm", dtm);
+            // 修改上下文的元数据
+            var nctx = new ClientInterceptorContext<TRequest, TResponse>(
+                ctx.Method,
+                ctx.Host,
+                new CallOptions(headers: headers, deadline: ctx.Options.Deadline, cancellationToken: ctx.Options.CancellationToken));
+
+            return nctx;
+        }
+    }
+}

src/Dtmworkflow/WorkflowHttpInterceptor.cs

@@ -0,0 +1,39 @@
+using System;
+using System.Net.Http;
+using System.Threading;
+using System.Threading.Tasks;
+
+namespace Dtmworkflow;
+
+internal class WorkflowHttpInterceptor : DelegatingHandler
+{
+    private readonly Workflow _wf;
+
+    public WorkflowHttpInterceptor(Workflow wf)
+    {
+        this._wf = wf;
+        InnerHandler = new HttpClientHandler();
+    }
+
+    protected override async Task<HttpResponseMessage> SendAsync(HttpRequestMessage request, CancellationToken cancellationToken)
+    {
+        Func<DtmCommon.BranchBarrier, Task<StepResult>> origin = async (barrier) =>
+        {
+            var response = await base.SendAsync(request, cancellationToken);
+            return _wf.StepResultFromHTTP(response, null);
+        };
+
+        StepResult sr;
+        // in phase 2, do not save, because it is saved outer
+        if (_wf.WorkflowImp.CurrentOp != DtmCommon.Constant.OpAction)
+        {
+            sr = await origin(null);
+        }
+        else
+        {
+            sr = await _wf.RecordedDo(origin);
+        }
+
+        return _wf.StepResultToHttp(sr);
+    }
+}

tests/BusiGrpcService/Program.cs

@@ -8,6 +8,8 @@
 {
     // Setup a HTTP/2 endpoint without TLS.
     options.ListenLocalhost(5005, o => o.Protocols = HttpProtocols.Http2);
+    // test for workflow http branch
+    options.ListenLocalhost(5006, o => o.Protocols = HttpProtocols.Http1);
 });
 
 builder.Services.AddGrpc();
@@ -20,6 +22,15 @@
 
 // Configure the HTTP request pipeline.
 app.MapGrpcService<BusiApiService>();
+
+// test for workflow http branch
+app.MapGet("/test-http-ok1", () => "SUCCESS");
+app.MapGet("/test-http-ok2", () => "SUCCESS");
+app.MapGet("/409", context =>
+{    
+    context.Response.StatusCode = 409;
+    return context.Response.WriteAsync("i am body, the http branch is 409"); // FAILURE
+});
 app.MapGet("/", () => "Communication with gRPC endpoints must be made through a gRPC client. To learn how to create a client, visit: https://go.microsoft.com/fwlink/?linkid=2086909");
 
 app.Run();