Fq proxy timeout fix merge (#11348)

evanevanevanevannnn · web-flow · commit 202eb87229a8 · 2024-11-07T15:30:17.000+03:00
diff --git a/ydb/core/fq/libs/actors/pending_fetcher.cpp b/ydb/core/fq/libs/actors/pending_fetcher.cpp
@@ -435,6 +435,7 @@ class TPendingFetcher : public NActors::TActorBootstrapped<TPendingFetcher> {
             TenantName,
             task.result_limit(),
             NProtoInterop::CastFromProto(task.execution_limit()),
+            NProtoInterop::CastFromProto(task.request_submitted_at()),
             NProtoInterop::CastFromProto(task.request_started_at()),
             task.restart_count(),
             task.job_id().value(),
diff --git a/ydb/core/fq/libs/compute/common/run_actor_params.cpp b/ydb/core/fq/libs/compute/common/run_actor_params.cpp
@@ -49,6 +49,7 @@ TRunActorParams::TRunActorParams(
     const TString& tenantName,
     uint64_t resultBytesLimit,
     TDuration executionTtl,
+    TInstant requestSubmittedAt,
     TInstant requestStartedAt,
     ui32 restartCount,
     const TString& jobId,
@@ -103,6 +104,7 @@ TRunActorParams::TRunActorParams(
     , TenantName(tenantName)
     , ResultBytesLimit(resultBytesLimit)
     , ExecutionTtl(executionTtl)
+    , RequestSubmittedAt(requestSubmittedAt)
     , RequestStartedAt(requestStartedAt)
     , RestartCount(restartCount)
     , JobId(jobId)
diff --git a/ydb/core/fq/libs/compute/common/run_actor_params.h b/ydb/core/fq/libs/compute/common/run_actor_params.h
@@ -68,6 +68,7 @@ struct TRunActorParams { // TODO2 : Change name
         const TString& tenantName,
         uint64_t resultBytesLimit,
         TDuration executionTtl,
+        TInstant requestSubmittedAt,
         TInstant requestStartedAt,
         ui32 restartCount,
         const TString& jobId,
@@ -131,6 +132,7 @@ struct TRunActorParams { // TODO2 : Change name
     const TString TenantName;
     const uint64_t ResultBytesLimit;
     const TDuration ExecutionTtl;
+    TInstant RequestSubmittedAt;
     TInstant RequestStartedAt;
     const ui32 RestartCount;
     const TString JobId;
diff --git a/ydb/core/fq/libs/compute/ydb/events/events.h b/ydb/core/fq/libs/compute/ydb/events/events.h
@@ -71,11 +71,11 @@ struct TEvYdbCompute {
 
     // Events
     struct TEvExecuteScriptRequest : public NActors::TEventLocal<TEvExecuteScriptRequest, EvExecuteScriptRequest> {
-        TEvExecuteScriptRequest(TString sql, TString idempotencyKey, const TDuration& resultTtl, const TDuration& operationTimeout, Ydb::Query::Syntax syntax, Ydb::Query::ExecMode execMode, Ydb::Query::StatsMode statsMode, const TString& traceId, const std::map<TString, Ydb::TypedValue>& queryParameters)
+        TEvExecuteScriptRequest(TString sql, TString idempotencyKey, const TDuration& resultTtl, const TInstant& operationDeadline, Ydb::Query::Syntax syntax, Ydb::Query::ExecMode execMode, Ydb::Query::StatsMode statsMode, const TString& traceId, const std::map<TString, Ydb::TypedValue>& queryParameters)
             : Sql(std::move(sql))
             , IdempotencyKey(std::move(idempotencyKey))
             , ResultTtl(resultTtl)
-            , OperationTimeout(operationTimeout)
+            , OperationDeadline(operationDeadline)
             , Syntax(syntax)
             , ExecMode(execMode)
             , StatsMode(statsMode)
@@ -86,7 +86,7 @@ struct TEvYdbCompute {
         TString Sql;
         TString IdempotencyKey;
         TDuration ResultTtl;
-        TDuration OperationTimeout;
+        TInstant OperationDeadline;
         Ydb::Query::Syntax Syntax = Ydb::Query::SYNTAX_YQL_V1;
         Ydb::Query::ExecMode ExecMode = Ydb::Query::EXEC_MODE_EXECUTE;
         Ydb::Query::StatsMode StatsMode = Ydb::Query::StatsMode::STATS_MODE_FULL;
diff --git a/ydb/core/fq/libs/compute/ydb/executer_actor.cpp b/ydb/core/fq/libs/compute/ydb/executer_actor.cpp
@@ -115,7 +115,7 @@ class TExecuterActor : public TBaseComputeActor<TExecuterActor> {
     }
 
     void SendExecuteScript() {
-        Register(new TRetryActor<TEvYdbCompute::TEvExecuteScriptRequest, TEvYdbCompute::TEvExecuteScriptResponse, TString, TString, TDuration, TDuration, Ydb::Query::Syntax, Ydb::Query::ExecMode, Ydb::Query::StatsMode, TString, std::map<TString, Ydb::TypedValue>>(Counters.GetCounters(ERequestType::RT_EXECUTE_SCRIPT), SelfId(), Connector, Params.Sql, Params.JobId, Params.ResultTtl, Params.ExecutionTtl, GetSyntax(), GetExecuteMode(), StatsMode, Params.JobId + "_" + ToString(Params.RestartCount), Params.QueryParameters));
+        Register(new TRetryActor<TEvYdbCompute::TEvExecuteScriptRequest, TEvYdbCompute::TEvExecuteScriptResponse, TString, TString, TDuration, TInstant, Ydb::Query::Syntax, Ydb::Query::ExecMode, Ydb::Query::StatsMode, TString, std::map<TString, Ydb::TypedValue>>(Counters.GetCounters(ERequestType::RT_EXECUTE_SCRIPT), SelfId(), Connector, Params.Sql, Params.JobId, Params.ResultTtl, Params.RequestSubmittedAt + Params.ExecutionTtl, GetSyntax(), GetExecuteMode(), StatsMode, Params.JobId + "_" + ToString(Params.RestartCount), Params.QueryParameters));
     }
 
     Ydb::Query::Syntax GetSyntax() const {
diff --git a/ydb/core/fq/libs/compute/ydb/ydb_connector_actor.cpp b/ydb/core/fq/libs/compute/ydb/ydb_connector_actor.cpp
@@ -47,7 +47,7 @@ class TYdbConnectorActor : public NActors::TActorBootstrapped<TYdbConnectorActor
         const auto& event = *ev->Get();
         NYdb::NQuery::TExecuteScriptSettings settings;
         settings.ResultsTtl(event.ResultTtl);
-        settings.OperationTimeout(event.OperationTimeout);
+        settings.OperationTimeout(event.OperationDeadline - TInstant::Now());
         settings.Syntax(event.Syntax);
         settings.ExecMode(event.ExecMode);
         settings.StatsMode(event.StatsMode);
diff --git a/ydb/core/fq/libs/control_plane_storage/internal/task_get.cpp b/ydb/core/fq/libs/control_plane_storage/internal/task_get.cpp
@@ -448,6 +448,7 @@ void TYdbControlPlaneStorageActor::Handle(TEvControlPlaneStorage::TEvGetTaskRequ
             newTask->set_result_limit(task.Internal.result_limit());
             *newTask->mutable_execution_limit() = NProtoInterop::CastToProto(ExtractLimit(task));
             *newTask->mutable_request_started_at() = task.Query.meta().started_at();
+            *newTask->mutable_request_submitted_at() = task.Query.meta().submitted_at();
 
             newTask->set_restart_count(task.RetryCount);
             auto* jobId = newTask->mutable_job_id();
diff --git a/ydb/core/fq/libs/control_plane_storage/internal/task_ping.cpp b/ydb/core/fq/libs/control_plane_storage/internal/task_ping.cpp
@@ -180,7 +180,16 @@ TPingTaskParams ConstructHardPingTask(
                 policy = it->second;
             }
 
-            if (retryLimiter.UpdateOnRetry(Now(), policy)) {
+            auto now = TInstant::Now();
+            auto executionDeadline = TInstant::Max();
+
+            auto submittedAt = NProtoInterop::CastFromProto(query.meta().submitted_at());
+            auto executionTtl = NProtoInterop::CastFromProto(internal.execution_ttl());
+            if (submittedAt && executionTtl) {
+                executionDeadline = submittedAt + executionTtl;
+            }
+
+            if (retryLimiter.UpdateOnRetry(now, policy) && now < executionDeadline) {
                 queryStatus.Clear();
                 // failing query is throttled for backoff period
                 backoff = policy.BackoffPeriod * (retryLimiter.RetryRate + 1);
@@ -191,7 +200,7 @@ TPingTaskParams ConstructHardPingTask(
                 TStringBuilder builder;
                 builder << "Query failed with code " << NYql::NDqProto::StatusIds_StatusCode_Name(request.status_code())
                     << " and will be restarted (RetryCount: " << retryLimiter.RetryCount << ")"
-                    << " at " << Now();
+                    << " at " << now;
                 transientIssues->AddIssue(NYql::TIssue(builder));
             } else {
                 // failure query should be processed instantly
@@ -202,7 +211,7 @@ TPingTaskParams ConstructHardPingTask(
                 if (policy.RetryCount) {
                     builder << " (failure rate " << retryLimiter.RetryRate << " exceeds limit of "  << policy.RetryCount << ")";
                 }
-                builder << " at " << Now();
+                builder << " at " << now;
 
                 // in case of problems with finalization, do not change the issues
                 if (query.meta().status() == FederatedQuery::QueryMeta::FAILING || query.meta().status() == FederatedQuery::QueryMeta::ABORTING_BY_SYSTEM || query.meta().status() == FederatedQuery::QueryMeta::ABORTING_BY_USER) {
diff --git a/ydb/core/fq/libs/protos/fq_private.proto b/ydb/core/fq/libs/protos/fq_private.proto
@@ -120,6 +120,8 @@ message GetTaskResult {
         NFq.NConfig.TYdbStorageConfig compute_connection = 37;
         google.protobuf.Duration result_ttl = 38;
         map<string, Ydb.TypedValue> parameters = 39;
+
+        google.protobuf.Timestamp request_submitted_at = 40;
     }
     repeated Task tasks = 1;
 }

Original file line number	Diff line number	Diff line change
`@@ -115,7 +115,7 @@ class TExecuterActor : public TBaseComputeActor<TExecuterActor> {`
`115`	`115`	`}`
`116`	`116`
`117`	`117`	`void SendExecuteScript() {`
`118`		- Register(new TRetryActor<TEvYdbCompute::TEvExecuteScriptRequest, TEvYdbCompute::TEvExecuteScriptResponse, TString, TString, TDuration, TDuration, Ydb::Query::Syntax, Ydb::Query::ExecMode, Ydb::Query::StatsMode, TString, std::map<TString, Ydb::TypedValue>>(Counters.GetCounters(ERequestType::RT_EXECUTE_SCRIPT), SelfId(), Connector, Params.Sql, Params.JobId, Params.ResultTtl, Params.ExecutionTtl, GetSyntax(), GetExecuteMode(), StatsMode, Params.JobId + "_" + ToString(Params.RestartCount), Params.QueryParameters));
	`118`	+ Register(new TRetryActor<TEvYdbCompute::TEvExecuteScriptRequest, TEvYdbCompute::TEvExecuteScriptResponse, TString, TString, TDuration, TInstant, Ydb::Query::Syntax, Ydb::Query::ExecMode, Ydb::Query::StatsMode, TString, std::map<TString, Ydb::TypedValue>>(Counters.GetCounters(ERequestType::RT_EXECUTE_SCRIPT), SelfId(), Connector, Params.Sql, Params.JobId, Params.ResultTtl, Params.RequestSubmittedAt + Params.ExecutionTtl, GetSyntax(), GetExecuteMode(), StatsMode, Params.JobId + "_" + ToString(Params.RestartCount), Params.QueryParameters));
`119`	`119`	`}`
`120`	`120`
`121`	`121`	`Ydb::Query::Syntax GetSyntax() const {`
Original file line number	Diff line number	Diff line change
`@@ -120,6 +120,8 @@ message GetTaskResult {`
`120`	`120`	`NFq.NConfig.TYdbStorageConfig compute_connection = 37;`
`121`	`121`	`google.protobuf.Duration result_ttl = 38;`
`122`	`122`	`map<string, Ydb.TypedValue> parameters = 39;`
	`123`	`+`
	`124`	`+ google.protobuf.Timestamp request_submitted_at = 40;`
`123`	`125`	`}`
`124`	`126`	`repeated Task tasks = 1;`
`125`	`127`	`}`