GRPC和Polly- .NET Core 6

Question

我正在尝试将Polly用作我的.NET Core 6项目中GRPC的重试策略处理程序。我注意到retryfunc永远不会被调用。我从这个项目开始 grpc ＆amp; ASP.NET Core 3.1：具有Polly的弹性

class Program
{
    static async Task Main(string[] args)
    {
        // DI
        var services = new ServiceCollection();

        var loggerFactory = LoggerFactory.Create(logging =>
        {
            logging.AddConsole();
            logging.SetMinimumLevel(LogLevel.Debug);
        });

        var serverErrors = new HttpStatusCode[] { 
            HttpStatusCode.BadGateway, 
            HttpStatusCode.GatewayTimeout, 
            HttpStatusCode.ServiceUnavailable, 
            HttpStatusCode.InternalServerError, 
            HttpStatusCode.TooManyRequests, 
            HttpStatusCode.RequestTimeout 
        };

        var gRpcErrors = new StatusCode[] {
            StatusCode.DeadlineExceeded,
            StatusCode.Internal,
            StatusCode.NotFound,
            StatusCode.ResourceExhausted,
            StatusCode.Unavailable,
            StatusCode.Unknown
        };

        Func<HttpRequestMessage, IAsyncPolicy<HttpResponseMessage>> retryFunc = (request) =>
        {
            return Policy.HandleResult<HttpResponseMessage>(r => {
                
                var grpcStatus = StatusManager.GetStatusCode(r);
                var httpStatusCode = r.StatusCode;

                return (grpcStatus == null && serverErrors.Contains(httpStatusCode)) || // if the server send an error before gRPC pipeline
                       (httpStatusCode == HttpStatusCode.OK && gRpcErrors.Contains(grpcStatus.Value)); // if gRPC pipeline handled the request (gRPC always answers OK)
            })
            .WaitAndRetryAsync(3, (input) => TimeSpan.FromSeconds(3 + input), (result, timeSpan, retryCount, context) =>
                                {
                                    var grpcStatus = StatusManager.GetStatusCode(result.Result);
                                    Console.WriteLine($"Request failed with {grpcStatus}. Retry");
                                });
        };

        services.AddGrpcClient<CountryServiceClient>(o =>
        {
            o.Address = new Uri("https://localhost:5001");
        }).AddPolicyHandler(retryFunc);

        var provider = services.BuildServiceProvider();
        var client = provider.GetRequiredService<CountryServiceClient>();

        try
        {
            var countries = (await client.GetAllAsync(new EmptyRequest())).Countries.Select(x => new Country
            {
                CountryId = x.Id,
                Description = x.Description,
                CountryName = x.Name
            }).ToList();

            Console.WriteLine("Found countries");
            countries.ForEach(x => Console.WriteLine($"Found country {x.CountryName} ({x.CountryId}) {x.Description}"));

        }
        catch (RpcException e)
        {
            Console.WriteLine(e.Message);
        }
    }
}

，但最终WaitandRetryAsync从未被调用。

我在

我的测试很简单。我在没有聆听后端的情况下启动客户端，期望从console.writeline读取输出的3倍（$“请求”在控制台上失败了{grpcstatus}。retry”）; 。但是政策处理程序从未解雇。我有以下例外，而

Status(StatusCode="Unavailable", Detail="Error connecting to
subchannel.", DebugException="System.Net.Sockets.SocketException
(10061): No connection could be made because the target machine
actively refused it.

没有任何重试。

Answer 1

这对您不起作用，因为重试现在已内置在GRPC中。为了进行此工作，请按以下方式注册您的服务：

var defaultMethodConfig = new MethodConfig
{
   Names = { MethodName.Default },
   RetryPolicy = new RetryPolicy
   {
       MaxAttempts = 3,
       InitialBackoff = TimeSpan.FromSeconds(3),
       MaxBackoff = TimeSpan.FromSeconds(3),
       BackoffMultiplier = 1,
       RetryableStatusCodes =
       {
           // Whatever status codes you want to look for
           StatusCode.Unauthenticated, StatusCode.NotFound, StatusCode.Unavailable,
       }
   }
  };
  var services = new ServiceCollection();
  services.AddGrpcClient<TestServiceClient>(o => { 
      o.Address = new Uri("https://localhost:5001");
      o.ChannelOptionsActions.Add(options =>
      {
          options.ServiceConfig = new ServiceConfig {MethodConfigs = {defaultMethodConfig}};
      });
  });

将重试策略添加到您的客户。您可能会遇到的另一件事。当时我没有意识到这一点，但是在我的服务实施中，我设置了这样的错误：

var response = new MyServiceResponse()
// something bad happens
context.Status = new Status(StatusCode.Internal, "Something went wrong");
return response;

如果您这样实施您的服务，重试逻辑将不会启动，那么您实际上必须做更多这样的事情：

// something bad happens
throw new RpcException(new Status(StatusCode.Internal, "Something went wrong"));

您在注册客户时配置的重试逻辑将工作。希望有帮助。

Answer 2

在@petercsala的帮助下，我尝试了一些修复程序。

作为第一次尝试，我在没有依赖性注射的情况下尝试了，以下方式注册了策略

var policy = Policy
    .Handle<Exception>()
    .RetryAsync(3, (exception, count) =>
    {
        Console.WriteLine($"Request {count}, {exception.Message}. Retry");
    });

var channel = GrpcChannel.ForAddress("https://localhost:5001");
TestServiceClient client = new TestServiceClient(channel);

await policy.ExecuteAsync(async () => await client.TestAsync(new Empty()));

。

然后，我回到DI，并以下记录了该策略

IAsyncPolicy<HttpResponseMessage> policy = 
Policy<HttpResponseMessage>.Handle<Exception>().RetryAsync(3, (exception, count) =>
{
    Console.WriteLine($"Request {count}, {exception.Exception.Message}. Retry");
});

var services = new ServiceCollection();
services.AddGrpcClient<TestServiceClient>(o => { 
    o.Address = new Uri("https://localhost:5001");
}).AddPolicyHandler(policy);

var provider = services.BuildServiceProvider();
var client = provider.GetRequiredService<TestServiceClient>();
    
var testClient = (await client.TestAsync(new Empty()));

，但仍然无法正常工作。

最后，AddPolicyHandler似乎不适合GRPC客户？