feat(backend): add gRPC metrics to api-server (RPS/latency), optimize execution spec reporting

backend/src/apiserver/api-server-metrics.md

@@ -0,0 +1,32 @@
+## Metrics
+
+Standard gRPC metrics are available via https://github.com/grpc-ecosystem/go-grpc-middleware/tree/main/providers/prometheus
+
+### Sample
+
+gRPC RPMs by grpc_method and grpc_code for the ml-pipeline server:
+```yaml
+sum by (grpc_service, grpc_method, grpc_code) (
+rate(grpc_server_handled_total{app=~"ml-pipeline", kubernetes_namespace="kubeflow"}[1m])
+) * 60
+```
+
+95th percentile gRPC latency by grpc_service and grpc_method for the ml-pipeline server:
+```yaml
+histogram_quantile(
+  0.95,
+  sum by (grpc_service, grpc_method, le) (
+    rate(grpc_server_handling_seconds_bucket{
+      app="ml-pipeline",
+      kubernetes_namespace="kubeflow"
+    }[1m])
+  )
+)
+```
+
+Gap in seconds between creating an execution spec (Argo or other backend) for a recurring run and reporting it via the persistence agent
+```yaml
+histogram_quantile(0.95, sum(rate(resource_manager_recurring_run_report_gap_bucket[1h])) by (le))
+```
+
+

backend/src/apiserver/main.go

@@ -27,6 +27,10 @@ import (
 	"strings"
 	"sync"
 
+	"github.com/prometheus/client_golang/prometheus"
+
+	grpc_prometheus "github.com/grpc-ecosystem/go-grpc-middleware/providers/prometheus"
+
 	"github.com/fsnotify/fsnotify"
 	"github.com/golang/glog"
 	"github.com/gorilla/mux"
@@ -208,7 +212,26 @@ func startRpcServer(resourceManager *resource.ResourceManager) {
 	if err != nil {
 		glog.Fatalf("Failed to start RPC server: %v", err)
 	}
-	s := grpc.NewServer(grpc.UnaryInterceptor(apiServerInterceptor), grpc.MaxRecvMsgSize(math.MaxInt32))
+
+	metrics := grpc_prometheus.NewServerMetrics(
+		grpc_prometheus.WithServerHandlingTimeHistogram(
+			grpc_prometheus.WithHistogramBuckets([]float64{
+				0.01, 0.03, 0.1, 0.3, 1, 3, 10, 15, 30, 60, 120, 300,
+			}),
+		),
+	)
+
+	prometheus.MustRegister(metrics)
+
+	s := grpc.NewServer(
+		grpc.ChainUnaryInterceptor(
+			apiServerInterceptor,
+			metrics.UnaryServerInterceptor(),
+		),
+		grpc.MaxRecvMsgSize(math.MaxInt32),
+	)
+
+	metrics.InitializeMetrics(s)
 
 	sharedExperimentServer := server.NewExperimentServer(resourceManager, &server.ExperimentServerOptions{CollectMetrics: *collectMetricsFlag})
 	sharedPipelineServer := server.NewPipelineServer(

backend/src/apiserver/resource/resource_manager.go

@@ -22,6 +22,7 @@ import (
 	"net"
 	"reflect"
 	"strconv"
+	"time"
 
 	scheduledworkflow "github.com/kubeflow/pipelines/backend/src/crd/pkg/apis/scheduledworkflow/v1beta1"
 
@@ -77,6 +78,13 @@ var (
 		Name: "resource_manager_workflow_runs_failed",
 		Help: "The current number of failed workflow runs",
 	}, extraLabels)
+
+	// Gap in seconds between creating an execution spec (Argo or other backend) for a recurring run and reporting it via the persistence agent.
+	recurringPipelineRunReportGap = promauto.NewHistogram(prometheus.HistogramOpts{
+		Name:    "resource_manager_recurring_run_report_gap",
+		Help:    "Delay of recurring pipeline run report",
+		Buckets: prometheus.ExponentialBuckets(1, 2, 15),
+	})
 )
 
 type ClientManagerInterface interface {
@@ -1238,8 +1246,8 @@ func (r *ResourceManager) DeleteJob(ctx context.Context, jobId string) error {
 
 // Creates new tasks or updates existing ones.
 // This is not a part of internal API exposed to persistence agent only.
-func (r *ResourceManager) CreateOrUpdateTasks(t []*model.Task) ([]*model.Task, error) {
-	tasks, err := r.taskStore.CreateOrUpdateTasks(t)
+func (r *ResourceManager) CreateOrUpdateTasks(t []*model.Task, runId string) ([]*model.Task, error) {
+	tasks, err := r.taskStore.CreateOrUpdateTasks(t, runId)
 	if err != nil {
 		return nil, util.Wrap(err, "Failed to create or update tasks")
 	}
@@ -1388,6 +1396,10 @@ func (r *ResourceManager) ReportWorkflowResource(ctx context.Context, execSpec u
 			},
 		}
 		run, err = r.runStore.CreateRun(run)
+		if r.options.CollectMetrics && !execStatus.StartedAtTime().Time.IsZero() {
+			reportGap := time.Since(execStatus.StartedAtTime().Time).Seconds()
+			recurringPipelineRunReportGap.Observe(reportGap)
+		}
 		if err != nil {
 			return nil, util.Wrapf(err, "Failed to report a workflow due to error creating run %s", runId)
 		} else {

backend/src/apiserver/server/report_server.go

@@ -43,7 +43,7 @@ func (s ReportServer) reportTasksFromExecution(execSpec util.ExecutionSpec, runI
 	if err != nil {
 		return nil, util.Wrap(err, "Failed to report tasks of an execution")
 	}
-	return s.resourceManager.CreateOrUpdateTasks(tasks)
+	return s.resourceManager.CreateOrUpdateTasks(tasks, runId)
 }
 
 // Reports a workflow.

backend/src/apiserver/storage/task_store.go

@@ -61,7 +61,7 @@ type TaskStoreInterface interface {
 	ListTasks(filterContext *model.FilterContext, opts *list.Options) ([]*model.Task, int, string, error)
 
 	// Creates new tasks or updates the existing ones.
-	CreateOrUpdateTasks(tasks []*model.Task) ([]*model.Task, error)
+	CreateOrUpdateTasks(tasks []*model.Task, runId string) ([]*model.Task, error)
 }
 
 type TaskStore struct {
@@ -324,15 +324,15 @@ func (s *TaskStore) GetTask(id string) (*model.Task, error) {
 }
 
 // Updates missing fields with existing data entries.
-func (s *TaskStore) patchWithExistingTasks(tasks []*model.Task) error {
+func (s *TaskStore) patchWithExistingTasks(tasks []*model.Task, runId string) error {
 	var podNames []string
 	for _, task := range tasks {
 		podNames = append(podNames, task.PodName)
 	}
 	sql, args, err := sq.
 		Select(taskColumns...).
 		From("tasks").
-		Where(sq.Eq{"PodName": podNames}).
+		Where(sq.Eq{"PodName": podNames, "RunUUID": runId}).
 		ToSql()
 	if err != nil {
 		return util.NewInternalServerError(err, "Failed to create query to check existing tasks")
@@ -359,7 +359,7 @@ func (s *TaskStore) patchWithExistingTasks(tasks []*model.Task) error {
 }
 
 // Creates new entries or updates existing ones.
-func (s *TaskStore) CreateOrUpdateTasks(tasks []*model.Task) ([]*model.Task, error) {
+func (s *TaskStore) CreateOrUpdateTasks(tasks []*model.Task, runId string) ([]*model.Task, error) {
 	buildQuery := func(ts []*model.Task) (string, []interface{}, error) {
 		sqlInsert := sq.Insert("tasks").Columns(taskColumnsWithPayload...)
 		for _, t := range ts {
@@ -405,7 +405,7 @@ func (s *TaskStore) CreateOrUpdateTasks(tasks []*model.Task) ([]*model.Task, err
 
 	// Check for existing tasks and fill empty field with existing data.
 	// Assumes that PodName column is a unique key.
-	if err := s.patchWithExistingTasks(tasks); err != nil {
+	if err := s.patchWithExistingTasks(tasks, runId); err != nil {
 		return nil, util.NewInternalServerError(err, "Failed to check for existing tasks")
 	}
 	for _, task := range tasks {

backend/src/apiserver/storage/task_store_test.go

@@ -504,7 +504,7 @@ func TestTaskStore_patchWithExistingTasks(t *testing.T) {
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			err := taskStore.patchWithExistingTasks(tt.tasks)
+			err := taskStore.patchWithExistingTasks(tt.tasks, defaultFakeRunIdTwo)
 			if tt.wantErr {
 				assert.NotNil(t, err)
 				assert.Contains(t, err.Error(), tt.errMsg)
@@ -611,7 +611,7 @@ func TestTaskStore_UpdateOrCreateTasks(t *testing.T) {
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			got, err := taskStore.CreateOrUpdateTasks(tt.tasks)
+			got, err := taskStore.CreateOrUpdateTasks(tt.tasks, defaultFakeRunIdTwo)
 			assert.Nil(t, err)
 			assert.Equal(t, tt.want, got)
 		})

go.mod

@@ -25,6 +25,7 @@ require (
 	github.com/google/go-cmp v0.6.0
 	github.com/google/uuid v1.6.0
 	github.com/gorilla/mux v1.8.0
+	github.com/grpc-ecosystem/go-grpc-middleware/providers/prometheus v1.1.0
 	github.com/grpc-ecosystem/go-grpc-middleware/v2 v2.3.2
 	github.com/grpc-ecosystem/grpc-gateway v1.16.0
 	github.com/jackc/pgx/v5 v5.5.4