run service: WatchGroups()

flyteidl2/workflow/run_definition.proto

@@ -10,6 +10,7 @@ import "flyteidl2/core/execution.proto";
 import "flyteidl2/task/common.proto";
 import "flyteidl2/task/run.proto";
 import "flyteidl2/task/task_definition.proto";
+import "google/protobuf/duration.proto";
 import "google/protobuf/timestamp.proto";
 
 option go_package = "github.com/flyteorg/flyte/v2/gen/go/flyteidl2/workflow";
@@ -312,3 +313,33 @@ message ActionEvent {
   // Timestamp when the event was observed and reported by the executor
   google.protobuf.Timestamp reported_time = 15;
 }
+
+// TaskGroup represents a group of runs for a specific task.
+message TaskGroup {
+  // Task name.
+  string task_name = 1;
+
+  // Environment name.
+  string environment_name = 2;
+
+  // Total number of runs for this task.
+  int64 total_runs = 3;
+
+  // Timestamp of the most recent run.
+  google.protobuf.Timestamp latest_run_time = 4;
+
+  // Recent run statuses, ordered from newest to oldest. Number of statuses determined by request.
+  repeated Phase recent_statuses = 5;
+
+  // Average failure rate of runs in this group (0.0 to 1.0).
+  // Computed as number of root actions with phase FAILED divided by total root actions.
+  double average_failure_rate = 6;
+
+  // Average duration of runs in this group.
+  google.protobuf.Duration average_duration = 7;
+
+  // Timestamp of the most recent finished run (terminal phase).
+  google.protobuf.Timestamp latest_finished_time = 8;
+
+  bool should_delete = 9;
+}

flyteidl2/workflow/run_service.proto

@@ -9,6 +9,7 @@ import "flyteidl2/task/common.proto";
 import "flyteidl2/task/run.proto";
 import "flyteidl2/task/task_definition.proto";
 import "flyteidl2/workflow/run_definition.proto";
+import "google/protobuf/timestamp.proto";
 
 option go_package = "github.com/flyteorg/flyte/v2/gen/go/flyteidl2/workflow";
 
@@ -64,6 +65,9 @@ service RunService {
 
   // AbortAction aborts a single action that was previously created or is currently being processed by a worker.
   rpc AbortAction(AbortActionRequest) returns (AbortActionResponse) {}
+
+  // Stream updates for task groups based on the provided filter criteria.
+  rpc WatchGroups(WatchGroupsRequest) returns (stream WatchGroupsResponse) {}
 }
 
 // Request message for creating a run.
@@ -292,3 +296,34 @@ message AbortActionRequest {
 }
 
 message AbortActionResponse {}
+
+// Request message for watching task groups.
+message WatchGroupsRequest {
+  oneof scope_by {
+    option (buf.validate.oneof).required = true;
+    // Watch groups within a project.
+    common.ProjectIdentifier project_id = 1;
+  }
+
+  // Filter groups after this date (inclusive). Required.
+  google.protobuf.Timestamp start_date = 2 [(buf.validate.field).required = true];
+
+  // Filter groups before this date (inclusive).
+  // If null, will stream updates up to current time.
+  // If not null, will return groups once and close stream.
+  google.protobuf.Timestamp end_date = 3;
+
+  // Common list request parameters.
+  common.ListRequest request = 4;
+}
+
+// Response message for watching task groups.
+message WatchGroupsResponse {
+  // List of task groups matching the filter criteria.
+  // For the initial response, this contains all groups.
+  // For subsequent updates, this contains only changed groups.
+  repeated TaskGroup task_groups = 1;
+
+  // Indicates when the initial List call is complete, and subsequent messages are updates.
+  bool sentinel = 2;
+}