feat grpc: add absolute timepoint propagation on the client side

commit_hash:313054dc2c8ba0b5d9e18757c8ced60f58ecd183

Author: oigolovanova

core/functional_tests/basic_chaos/tests-deadline/test_client.py

@@ -401,11 +401,12 @@ async def test_dp_timeout_not_retried(
 
 async def test_absolute_deadline_propagated_as_is(call, mockserver):
     epoch_us_deadline = _make_deadline_epoch_us(10.0)
-    captured = {}
+    captured_headers = None
 
     @mockserver.handler('/test')
     async def mock(request):
-        captured['headers'] = dict(request.headers)
+        nonlocal captured_headers
+        captured_headers = dict(request.headers)
         return mockserver.make_response('OK!')
 
     response = await call(
@@ -415,4 +416,5 @@ async def mock(request):
         },
     )
     assert response.status == 200
-    assert captured['headers'].get(DP_ABSOLUTE_DEADLINE) == epoch_us_deadline
+    assert captured_headers is not None
+    assert captured_headers.get(DP_ABSOLUTE_DEADLINE) == epoch_us_deadline

grpc/functional_tests/basic_chaos/service.hpp

@@ -20,6 +20,8 @@
 #include <samples/greeter_client.usrv.pb.hpp>
 #include <samples/greeter_service.usrv.pb.hpp>
 
+#include "client.hpp"
+
 namespace samples {
 
 class GreeterServiceComponent final : public api::GreeterServiceBase::Component {
@@ -28,7 +30,8 @@ class GreeterServiceComponent final : public api::GreeterServiceBase::Component
 
     GreeterServiceComponent(const components::ComponentConfig& config, const components::ComponentContext& context)
         : api::GreeterServiceBase::Component(config, context),
-          prefix_(config["greeting-prefix"].As<std::string>())
+          prefix_(config["greeting-prefix"].As<std::string>()),
+          greeter_client_(context.FindComponent<GreeterClient>())
     {}
 
     inline SayHelloResult SayHello(CallContext& context, api::GreetingRequest&& request) final;
@@ -53,6 +56,7 @@ class GreeterServiceComponent final : public api::GreeterServiceBase::Component
 
 private:
     const std::string prefix_;
+    GreeterClient& greeter_client_;
 };
 
 GreeterServiceComponent::SayHelloResult GreeterServiceComponent::SayHello(
@@ -67,6 +71,10 @@ GreeterServiceComponent::SayHelloResult GreeterServiceComponent::SayHello(
             const engine::TaskCancellationBlocker block_cancel;
             TESTPOINT("testpoint_cancel", {});
         }
+    } else if (request.name() == "test_deadline_propagation") {
+        const auto greeting = greeter_client_.SayHello(request.name(), /*is_small_timeout=*/false);
+        response.set_greeting(greeting);
+        return response;
     }
 
     response.set_greeting(fmt::format("{}, {}!", prefix_, request.name()));

grpc/functional_tests/basic_chaos/static_config.yaml

@@ -83,7 +83,7 @@ components_manager:
         grpc-client-factory:
             middlewares:
                 grpc-client-deadline-propagation:
-                    enabled: false
+                    enabled: true
             channel-args: {}
         greeter-client:
             endpoint: '[::]:8081'

grpc/functional_tests/basic_chaos/tests-nonchaos/conftest.py

@@ -6,6 +6,17 @@
 
 pytest_plugins = ['pytest_userver.plugins.grpc']
 
+USERVER_CONFIG_HOOKS = ['_prepare_service_config']
+
+
+@pytest.fixture(scope='session')
+def _prepare_service_config(grpc_mockserver_endpoint):
+    def patch_config(config, config_vars):
+        components = config['components_manager']['components']
+        components['greeter-client']['endpoint'] = grpc_mockserver_endpoint
+
+    return patch_config
+
 
 @pytest.fixture(scope='session')
 def service_env():

grpc/functional_tests/basic_chaos/tests-nonchaos/test_deadline_absolute.py

@@ -4,6 +4,7 @@
 import pytest
 
 import samples.greeter_pb2 as greeter_pb2
+import samples.greeter_pb2_grpc as greeter_pb2_grpc
 
 DP_ABSOLUTE_DEADLINE = 'x-request-deadline'
 
@@ -84,3 +85,48 @@ async def test_absolute_deadline_invalid_format(grpc_client):
         timeout=5.0,
     )
     assert response.greeting == 'Hello, Python!'
+
+
+async def test_absolute_deadline_propagated_as_is(grpc_client, grpc_mockserver):
+    epoch_us_deadline = _make_deadline_epoch_us(120.0)
+    captured_metadata = None
+
+    @grpc_mockserver(greeter_pb2_grpc.GreeterServiceServicer.SayHello)
+    async def mock_say_hello(request, context: grpc.aio.ServicerContext):
+        nonlocal captured_metadata
+        captured_metadata = dict(context.invocation_metadata())
+        return greeter_pb2.GreetingResponse(
+            greeting='Hello, Python!',
+        )
+
+    response = await grpc_client.SayHello(
+        greeter_pb2.GreetingRequest(name='test_deadline_propagation'),
+        metadata=[(DP_ABSOLUTE_DEADLINE, epoch_us_deadline)],
+        timeout=5.0,
+    )
+    assert response.greeting == 'Hello, Python!'
+    assert captured_metadata is not None
+    assert captured_metadata.get(DP_ABSOLUTE_DEADLINE) == epoch_us_deadline
+
+
+async def test_absolute_deadline_synthesized_when_only_duration_propagated(
+    grpc_client,
+    grpc_mockserver,
+):
+    captured_metadata = None
+
+    @grpc_mockserver(greeter_pb2_grpc.GreeterServiceServicer.SayHello)
+    async def mock_say_hello(request, context: grpc.aio.ServicerContext):
+        nonlocal captured_metadata
+        captured_metadata = dict(context.invocation_metadata())
+        return greeter_pb2.GreetingResponse(
+            greeting='Hello, Python!',
+        )
+
+    response = await grpc_client.SayHello(
+        greeter_pb2.GreetingRequest(name='test_deadline_propagation'),
+        timeout=5.0,
+    )
+    assert response.greeting == 'Hello, Python!'
+    assert captured_metadata is not None
+    assert DP_ABSOLUTE_DEADLINE in captured_metadata

grpc/src/ugrpc/client/middlewares/deadline_propagation/middleware.cpp

@@ -2,11 +2,16 @@
 
 #include <userver/server/request/task_inherited_data.hpp>
 #include <userver/tracing/tags.hpp>
+#include <userver/utils/datetime.hpp>
 #include <userver/utils/impl/internal_tag.hpp>
 
+#include <ugrpc/impl/rpc_metadata.hpp>
 #include <userver/ugrpc/client/impl/call_state.hpp>
+#include <userver/ugrpc/impl/to_string.hpp>
 #include <userver/ugrpc/time_utils.hpp>
 
+#include <chrono>
+
 USERVER_NAMESPACE_BEGIN
 
 namespace ugrpc::client::middlewares::deadline_propagation {
@@ -29,8 +34,21 @@ void UpdateDeadline(MiddlewareCallContext& context) {
 
     auto& client_context = context.GetClientContext();
 
-    const auto context_time_left = ugrpc::TimespecToDuration(client_context.raw_deadline());
     const engine::Deadline task_deadline = USERVER_NAMESPACE::server::request::GetTaskInheritedDeadline();
+    const auto original_absolute_deadline = USERVER_NAMESPACE::server::request::GetTaskInheritedOriginalDeadline();
+
+    if (original_absolute_deadline.has_value()) {
+        const auto deadline_header = std::to_string(original_absolute_deadline->time_since_epoch().count());
+        client_context.AddMetadata(ugrpc::impl::kXRequestDeadline, ugrpc::impl::ToGrpcString(deadline_header));
+    } else if (task_deadline.IsReachable()) {
+        const auto absolute_deadline =
+            std::chrono::time_point_cast<std::chrono::microseconds>(std::chrono::system_clock::now()) +
+            std::chrono::duration_cast<std::chrono::microseconds>(task_deadline.TimeLeft());
+        const auto deadline_header = std::to_string(absolute_deadline.time_since_epoch().count());
+        client_context.AddMetadata(ugrpc::impl::kXRequestDeadline, ugrpc::impl::ToGrpcString(deadline_header));
+    }
+
+    const auto context_time_left = ugrpc::TimespecToDuration(client_context.raw_deadline());
 
     const auto client_deadline_reachable = (context_time_left != engine::Deadline::Duration::max());
     if (!task_deadline.IsReachable() && !client_deadline_reachable) {

grpc/src/ugrpc/impl/rpc_metadata.cpp

@@ -31,6 +31,8 @@ const grpc::string kXTestsuiteClientName = "x-testsuite-client-name";
 const grpc::string kTestsuiteNetworkCode = "network";
 const grpc::string kTestsuiteTimeoutCode = "timeout";
 
+const grpc::string kXRequestDeadline = "x-request-deadline";
+
 }  // namespace ugrpc::impl
 
 USERVER_NAMESPACE_END

grpc/src/ugrpc/impl/rpc_metadata.hpp

@@ -31,6 +31,8 @@ extern const grpc::string kXTestsuiteClientName;
 extern const grpc::string kTestsuiteNetworkCode;
 extern const grpc::string kTestsuiteTimeoutCode;
 
+extern const grpc::string kXRequestDeadline;
+
 }  // namespace ugrpc::impl
 
 USERVER_NAMESPACE_END

grpc/src/ugrpc/server/middlewares/deadline_propagation/middleware.cpp

@@ -10,6 +10,7 @@
 #include <userver/server/request/task_inherited_data.hpp>
 #include <userver/utils/impl/internal_tag.hpp>
 
+#include <ugrpc/impl/rpc_metadata.hpp>
 #include <userver/ugrpc/impl/statistics_scope.hpp>
 #include <userver/ugrpc/impl/to_string.hpp>
 #include <userver/ugrpc/status_codes.hpp>
@@ -42,7 +43,7 @@ bool CheckAndSetupDeadline(
 
     std::optional<USERVER_NAMESPACE::server::request::TaskInheritedOriginalDeadline> absolute_original_deadline;
     const auto& client_metadata = server_context.client_metadata();
-    const auto absolute_deadline_it = client_metadata.find("x-request-deadline");
+    const auto absolute_deadline_it = client_metadata.find(ugrpc::impl::kXRequestDeadline);
     if (absolute_deadline_it != client_metadata.end()) {
         absolute_original_deadline = USERVER_NAMESPACE::server::request::impl::ParseXRequestDeadlineString(
             ugrpc::impl::ToString(absolute_deadline_it->second)