YT-23979: Report job's operation incarnation and add corresponding filter to API

commit_hash:e84d2a750dc219e544df03b6d84a154aa312ef1b

8 files changed, 29 insertions, 0 deletions

diff --git a/yt/cpp/mapreduce/http_client/rpc_parameters_serialization.cpp b/yt/cpp/mapreduce/http_client/rpc_parameters_serialization.cpp
index c986e4c025b..413a09ab978 100644
--- a/yt/cpp/mapreduce/http_client/rpc_parameters_serialization.cpp
+++ b/yt/cpp/mapreduce/http_client/rpc_parameters_serialization.cpp
@@ -555,6 +555,9 @@ TNode SerializeParamsForListJobs(
     if (options.WithMonitoringDescriptor_) {
         result["with_monitoring_descriptor"] = *options.WithMonitoringDescriptor_;
     }
+    if (options.OperationIncarnation_) {
+        result["operation_incarnation"] = *options.OperationIncarnation_;
+    }
     if (options.FromTime_) {
         result["from_time"] = ToString(options.FromTime_);
     }
diff --git a/yt/cpp/mapreduce/interface/operation.h b/yt/cpp/mapreduce/interface/operation.h
index aa109575a8a..218ead0572d 100644
--- a/yt/cpp/mapreduce/interface/operation.h
+++ b/yt/cpp/mapreduce/interface/operation.h
@@ -2880,6 +2880,10 @@ struct TListJobsOptions
     FLUENT_FIELD_OPTION(bool, WithMonitoringDescriptor);
 
     ///
+    /// @brief Return only jobs with given operation incarnation.
+    FLUENT_FIELD_OPTION(TString, OperationIncarnation);
+
+    ///
     /// @brief Search for jobs with start time >= `FromTime`.
     FLUENT_FIELD_OPTION(TInstant, FromTime);
 
diff --git a/yt/yt/client/api/operation_client.cpp b/yt/yt/client/api/operation_client.cpp
index 391740c7eb3..625b55fe611 100644
--- a/yt/yt/client/api/operation_client.cpp
+++ b/yt/yt/client/api/operation_client.cpp
@@ -311,6 +311,7 @@ void Serialize(const TJob& job, NYson::IYsonConsumer* consumer, TStringBuf idKey
             .OptionalItem("is_stale", job.IsStale)
             .OptionalItem("job_cookie", job.JobCookie)
             .OptionalItem("archive_features", job.ArchiveFeatures)
+            .OptionalItem("operation_incarnation", job.OperationIncarnation)
         .EndMap();
 }
 
diff --git a/yt/yt/client/api/operation_client.h b/yt/yt/client/api/operation_client.h
index 92f963c13d0..22ddbeb856e 100644
--- a/yt/yt/client/api/operation_client.h
+++ b/yt/yt/client/api/operation_client.h
@@ -209,6 +209,7 @@ struct TListJobsOptions
     std::optional<bool> WithCompetitors;
     std::optional<bool> WithMonitoringDescriptor;
     std::optional<TString> TaskName;
+    std::optional<std::string> OperationIncarnation;
 
     std::optional<TInstant> FromTime;
     std::optional<TInstant> ToTime;
@@ -386,6 +387,7 @@ struct TJob
     std::optional<TString> MonitoringDescriptor;
     std::optional<ui64> JobCookie;
     NYson::TYsonString ArchiveFeatures;
+    std::optional<std::string> OperationIncarnation;
 
     std::optional<bool> IsStale;
 
diff --git a/yt/yt/client/api/rpc_proxy/client_impl.cpp b/yt/yt/client/api/rpc_proxy/client_impl.cpp
index 09b7f158795..bc032c68d77 100644
--- a/yt/yt/client/api/rpc_proxy/client_impl.cpp
+++ b/yt/yt/client/api/rpc_proxy/client_impl.cpp
@@ -1523,6 +1523,9 @@ TFuture<TListJobsResult> TClient::ListJobs(
     if (options.TaskName) {
         req->set_task_name(*options.TaskName);
     }
+    if (options.OperationIncarnation) {
+        req->set_operation_incarnation(*options.OperationIncarnation);
+    }
     if (options.FromTime) {
         req->set_from_time(NYT::ToProto(*options.FromTime));
     }
diff --git a/yt/yt/client/api/rpc_proxy/helpers.cpp b/yt/yt/client/api/rpc_proxy/helpers.cpp
index 41f88cbcbf7..7bbdc3a3800 100644
--- a/yt/yt/client/api/rpc_proxy/helpers.cpp
+++ b/yt/yt/client/api/rpc_proxy/helpers.cpp
@@ -1014,6 +1014,9 @@ void ToProto(NProto::TJob* protoJob, const NApi::TJob& job)
     if (job.MonitoringDescriptor) {
         protoJob->set_monitoring_descriptor(*job.MonitoringDescriptor);
     }
+    if (job.OperationIncarnation) {
+        protoJob->set_operation_incarnation(*job.OperationIncarnation);
+    }
 }
 
 void FromProto(NApi::TJob* job, const NProto::TJob& protoJob)
@@ -1163,6 +1166,11 @@ void FromProto(NApi::TJob* job, const NProto::TJob& protoJob)
     } else {
         job->MonitoringDescriptor.reset();
     }
+    if (protoJob.has_operation_incarnation()) {
+        job->OperationIncarnation = protoJob.operation_incarnation();
+    } else {
+        job->OperationIncarnation.reset();
+    }
 }
 
 void ToProto(
diff --git a/yt/yt/client/driver/scheduler_commands.cpp b/yt/yt/client/driver/scheduler_commands.cpp
index 5002ccfb8ec..4fb6a48d759 100644
--- a/yt/yt/client/driver/scheduler_commands.cpp
+++ b/yt/yt/client/driver/scheduler_commands.cpp
@@ -490,6 +490,11 @@ void TListJobsCommand::Register(TRegistrar registrar)
         [] (TThis* command) -> auto& { return command->Options.WithMonitoringDescriptor; })
         .Optional(/*init*/ false);
 
+    registrar.ParameterWithUniversalAccessor<std::optional<std::string>>(
+        "operation_incarnation",
+        [] (TThis* command) -> auto& { return command->Options.OperationIncarnation; })
+        .Optional(/*init*/ false);
+
     registrar.ParameterWithUniversalAccessor<std::optional<TInstant>>(
         "from_time",
         [] (TThis* command) -> auto& { return command->Options.FromTime; })
diff --git a/yt/yt_proto/yt/client/api/rpc_proxy/proto/api_service.proto b/yt/yt_proto/yt/client/api/rpc_proxy/proto/api_service.proto
index 82e2d17d48c..f0c94b33ef3 100644
--- a/yt/yt_proto/yt/client/api/rpc_proxy/proto/api_service.proto
+++ b/yt/yt_proto/yt/client/api/rpc_proxy/proto/api_service.proto
@@ -2435,6 +2435,8 @@ message TReqListJobs
 
     optional bool with_monitoring_descriptor = 21;
 
+    optional string operation_incarnation = 25;
+
     optional uint64 from_time = 22; // TInstant
     optional uint64 to_time = 23; // TInstant
 
@@ -3172,6 +3174,7 @@ message TJob
     optional uint64 job_cookie = 27;
     optional bytes archive_features = 29; // YSON
     optional string monitoring_descriptor = 30;
+    optional string operation_incarnation = 31;
 }
 
 message TListJobsStatistics