YT-27710: Add RPC to TransactionPinger

* Changelog entry
  Type: fix
  Component: cpp-sdk

Allow TransactionPinger to ping via RPC if is being used in RPC Client.
commit_hash:9b38ce40b5667f107d6f23e40bf07dcf80ac11c7

1 files changed, 101 insertions, 8 deletions

diff --git a/yt/cpp/mapreduce/http_client/raw_client.cpp b/yt/cpp/mapreduce/http_client/raw_client.cpp
index 635ac09c979..4ccafb1a987 100644
--- a/yt/cpp/mapreduce/http_client/raw_client.cpp
+++ b/yt/cpp/mapreduce/http_client/raw_client.cpp
@@ -3,6 +3,7 @@
 #include "raw_requests.h"
 #include "rpc_parameters_serialization.h"
 
+#include <yt/cpp/mapreduce/common/expected_error_guard.h>
 #include <yt/cpp/mapreduce/common/helpers.h>
 #include <yt/cpp/mapreduce/common/retry_lib.h>
 
@@ -13,11 +14,20 @@
 
 #include <yt/cpp/mapreduce/interface/fluent.h>
 #include <yt/cpp/mapreduce/interface/fwd.h>
+#include <yt/cpp/mapreduce/interface/logging/yt_log.h>
 #include <yt/cpp/mapreduce/interface/operation.h>
 #include <yt/cpp/mapreduce/interface/tvm.h>
 
 #include <yt/cpp/mapreduce/io/helpers.h>
 
+#include <yt/yt/core/concurrency/thread_pool_poller.h>
+
+#include <yt/yt/core/http/client.h>
+#include <yt/yt/core/http/config.h>
+#include <yt/yt/core/http/http.h>
+#include <yt/yt/core/https/client.h>
+#include <yt/yt/core/https/config.h>
+
 #include <library/cpp/yson/node/node_io.h>
 
 #include <library/cpp/yt/yson_string/string.h>
@@ -26,6 +36,39 @@ namespace NYT::NDetail {
 
 ////////////////////////////////////////////////////////////////////////////////
 
+namespace {
+
+void CheckError(const TString& requestId, NHttp::IResponsePtr response)
+{
+    if (const auto* ytError = response->GetHeaders()->Find("X-YT-Error")) {
+        TYtError error;
+        error.ParseFrom(*ytError);
+
+        TErrorResponse errorResponse(std::move(error), requestId);
+        if (errorResponse.IsOk()) {
+            return;
+        }
+
+        if (TExpectedErrorGuard::IsErrorExpected(errorResponse)) {
+            YT_LOG_INFO("Received expected error, RSP %v - HTTP %v - %v",
+                requestId,
+                response->GetStatusCode(),
+                errorResponse.AsStrBuf());
+        } else {
+            YT_LOG_ERROR("RSP %v - HTTP %v - %v",
+                requestId,
+                response->GetStatusCode(),
+                errorResponse.AsStrBuf());
+        }
+
+        ythrow errorResponse;
+    }
+}
+
+} // anonymous namespace
+
+////////////////////////////////////////////////////////////////////////////////
+
 THttpRawClient::THttpRawClient(const TClientContext& context)
     : Context_(context)
 { }
@@ -265,14 +308,48 @@ TTransactionId THttpRawClient::StartTransaction(
 
 void THttpRawClient::PingTransaction(const TTransactionId& transactionId)
 {
-    TMutationId mutationId;
-    THttpHeader header("POST", "ping_tx");
-    header.MergeParameters(NRawClient::SerializeParamsForPingTx(transactionId));
-    TRequestConfig requestConfig;
-    requestConfig.HttpConfig = NHttpClient::THttpConfig{
-        .SocketTimeout = Context_.Config->PingTimeout
-    };
-    RequestWithoutRetry(Context_, mutationId, header)->GetResponse();
+    std::call_once(PingClientInitOnceFlag_, [this] () {
+        InitPingClient();
+    });
+
+    auto url = TString::Join(Context_.UseTLS ? "https://" : "http://", Context_.ServerName, "/api/", Context_.Config->ApiVersion, "/ping_tx");
+    auto headers = New<NHttp::THeaders>();
+    auto requestId = CreateGuidAsString();
+
+    headers->Add("Host", url);
+    headers->Add("User-Agent", TProcessState::Get()->ClientVersion);
+
+    if (const auto& serviceTicketAuth = Context_.ServiceTicketAuth) {
+        const auto serviceTicket = serviceTicketAuth->Ptr->IssueServiceTicket();
+        headers->Add("X-Ya-Service-Ticket", serviceTicket);
+    } else if (const auto& token = Context_.Token; !token.empty()) {
+        headers->Add("Authorization", "OAuth " + token);
+    }
+
+    headers->Add("Transfer-Encoding", "chunked");
+    headers->Add("X-YT-Correlation-Id", requestId);
+    headers->Add("X-YT-Header-Format", "<format=text>yson");
+    headers->Add("Content-Encoding", "identity");
+    headers->Add("Accept-Encoding", "identity");
+
+    TNode node;
+    node["transaction_id"] = GetGuidAsString(transactionId);
+    auto strParams = NodeToYsonString(node);
+
+    YT_LOG_DEBUG("REQ %v - sending request (HostName: %v; Method POST %v; X-YT-Parameters (sent in body): %v)",
+        requestId,
+        Context_.ServerName,
+        url,
+        strParams);
+
+    auto response = NConcurrency::WaitFor(PingHttpClient_->Post(url, TSharedRef::FromString(strParams), headers))
+        .ValueOrThrow();
+    CheckError(requestId, response);
+
+    YT_LOG_DEBUG("RSP %v - received response %v bytes. (%v)",
+        requestId,
+        response->ReadAll().size(),
+        strParams);
 }
 
 void THttpRawClient::AbortTransaction(
@@ -1170,6 +1247,22 @@ IRawClientPtr THttpRawClient::Clone(const TClientContext& context)
     return ::MakeIntrusive<THttpRawClient>(context);
 }
 
+void THttpRawClient::InitPingClient() {
+    auto httpPoller = NConcurrency::CreateThreadPoolPoller(
+        Context_.Config->AsyncHttpClientThreads,
+        "tx_http_client_poller");
+
+    if (Context_.UseTLS) {
+        auto httpsClientConfig = NYT::New<NHttps::TClientConfig>();
+        httpsClientConfig->MaxIdleConnections = 16;
+        PingHttpClient_ = NHttps::CreateClient(std::move(httpsClientConfig), std::move(httpPoller));
+    } else {
+        auto httpClientConfig = NYT::New<NHttp::TClientConfig>();
+        httpClientConfig->MaxIdleConnections = 16;
+        PingHttpClient_ = NHttp::CreateClient(std::move(httpClientConfig), std::move(httpPoller));
+    }
+}
+
 ////////////////////////////////////////////////////////////////////////////////
 
 } // namespace NYT::NDetail