Implement client metrics interceptor for continuous benchmark job

cmd/tools/benchmark/job/sample.yaml

@@ -48,6 +48,8 @@ server_config:
         header_table_size: 0
         interceptors:
           - "RecoverInterceptor"
+          - "TraceInterceptor"
+          - "MetricInterceptor"
         enable_reflection: true
       socket_option:
         reuse_port: true
@@ -136,8 +138,8 @@ server_config:
         ip_recover_destination_addr: false
   startup_strategy:
     - liveness
-    - readiness
     - grpc
+    - readiness
   full_shutdown_duration: 30s
   tls:
     ca: /path/to/ca
@@ -147,7 +149,7 @@ server_config:
 observability:
   enabled: false
   otlp:
-    collector_endpoint: ""
+    collector_endpoint: "opentelemetry-collector-collector.default.svc.cluster.local:4317"
     attribute:
       namespace: _MY_POD_NAMESPACE_
       pod_name: _MY_POD_NAME_
@@ -157,6 +159,8 @@ observability:
     trace_export_timeout: "1m"
     trace_max_export_batch_size: 1024
     trace_max_queue_size: 256
+    metrics_export_interval: "1s"
+    metrics_export_timeout: "1m"
   metrics:
     enable_cgo: true
     enable_goroutine: true
@@ -172,7 +176,7 @@ observability:
       - go_arch
       - algorithm_info
   trace:
-    enabled: false
+    enabled: true
 job:
   replica: 1
   repetition: 1
@@ -181,7 +185,7 @@ job:
   rps: 200
   concurrency_limit: 200
   client_config:
-    health_check_duration: "1s"
+    health_check_duration: ""
     connection_pool:
       enable_dns_resolver: true
       enable_rebalance: true
@@ -221,7 +225,9 @@ job:
       enable_backoff: false
       insecure: true
       timeout: ""
-      interceptors: []
+      interceptors:
+        - MetricInterceptor
+        - TraceInterceptor
       net:
         dns:
           cache_enabled: true

internal/k8s/vald/benchmark/job/job_template.go

@@ -35,19 +35,21 @@ const (
 	RestartPolicyAlways    RestartPolicy = "Always"
 	RestartPolicyOnFailure RestartPolicy = "OnFailure"
 	RestartPolicyNever     RestartPolicy = "Never"
-)
 
-const (
+	volumeName = "vald-benchmark-job-config"
 	svcAccount = "vald-benchmark-operator"
 )
 
+var mode = int32(420)
+
 type BenchmarkJobTpl interface {
 	CreateJobTpl(opts ...BenchmarkJobOption) (k8s.Job, error)
 }
 
 type benchmarkJobTpl struct {
 	containerName      string
 	containerImageName string
+	configMapName      string
 	imagePullPolicy    ImagePullPolicy
 	jobTpl             k8s.Job
 }
@@ -138,6 +140,51 @@ func (b *benchmarkJobTpl) CreateJobTpl(opts ...BenchmarkJobOption) (k8s.Job, err
 						},
 					},
 				},
+				{
+					Name: "MY_NODE_NAME",
+					ValueFrom: &corev1.EnvVarSource{
+						FieldRef: &corev1.ObjectFieldSelector{
+							FieldPath: "spec.nodeName",
+						},
+					},
+				},
+				{
+					Name: "MY_POD_NAMESPACE",
+					ValueFrom: &corev1.EnvVarSource{
+						FieldRef: &corev1.ObjectFieldSelector{
+							FieldPath: "metadata.namespace",
+						},
+					},
+				},
+				{
+					Name: "MY_POD_NAME",
+					ValueFrom: &corev1.EnvVarSource{
+						FieldRef: &corev1.ObjectFieldSelector{
+							FieldPath: "metadata.name",
+						},
+					},
+				},
+			},
+			VolumeMounts: []corev1.VolumeMount{
+				{
+					Name:      volumeName,
+					MountPath: "/etc/server",
+				},
+			},
+		},
+	}
+	// mount benchmark operator config map.
+	// It is used for bind only observability config for each benchmark job
+	b.jobTpl.Spec.Template.Spec.Volumes = []corev1.Volume{
+		{
+			Name: volumeName,
+			VolumeSource: corev1.VolumeSource{
+				ConfigMap: &corev1.ConfigMapVolumeSource{
+					LocalObjectReference: corev1.LocalObjectReference{
+						Name: b.configMapName,
+					},
+					DefaultMode: &mode,
+				},
 			},
 		},
 	}

internal/k8s/vald/benchmark/job/job_template_option.go

@@ -52,10 +52,19 @@
 // WithImagePullPolicy sets the docker image pull policy for benchmark job.
 func WithImagePullPolicy(p ImagePullPolicy) BenchmarkJobTplOption {
 	return func(b *benchmarkJobTpl) error {
-		if len(p) == 0 {
-			return nil
+		if len(p) > 0 {
+			b.imagePullPolicy = p
+		}
+		return nil
+	}
+}
+
+// WithOperatorConfigMap sets the configMapName for mounting Job Pod
+func WithOperatorConfigMap(cm string) BenchmarkJobTplOption {
+	return func(b *benchmarkJobTpl) error {
+		if len(cm) > 0 {
+			b.configMapName = cm
 		}
-		b.imagePullPolicy = p
 		return nil
 	}
 }

internal/net/grpc/interceptor/client/metric/metric.go

@@ -0,0 +1,89 @@
+// Copyright (C) 2019-2024 vdaas.org vald team <vald@vdaas.org>
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// You may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//	https://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+package metric
+
+import (
+	"context"
+	"time"
+
+	"github.com/vdaas/vald/internal/errors"
+	"github.com/vdaas/vald/internal/net/grpc/codes"
+	"github.com/vdaas/vald/internal/net/grpc/status"
+	"github.com/vdaas/vald/internal/observability/attribute"
+	"github.com/vdaas/vald/internal/observability/metrics"
+	"google.golang.org/grpc"
+)
+
+const (
+	latencyMetricsName       = "client_latency"
+	completedRPCsMetricsName = "client_completed_rpcs"
+
+	gRPCMethodKeyName = "grpc_client_method"
+	gRPCStatus        = "grpc_client_status"
+)
+
+func ClientMetricInterceptors() (grpc.UnaryClientInterceptor, grpc.StreamClientInterceptor, error) {
+	meter := metrics.GetMeter()
+
+	latencyHistgram, err := meter.Float64Histogram(
+		latencyMetricsName,
+		metrics.WithDescription("Client latency in milliseconds, by method"),
+		metrics.WithUnit(metrics.Milliseconds),
+	)
+	if err != nil {
+		return nil, nil, errors.Wrap(err, "failed to create latency metric")
+	}
+
+	completedRPCCnt, err := meter.Int64Counter(
+		completedRPCsMetricsName,
+		metrics.WithDescription("Count of RPCs by method and status"),
+		metrics.WithUnit(metrics.Milliseconds),
+	)
+	if err != nil {
+		return nil, nil, errors.Wrap(err, "failed to create completedRPCs metric")
+	}
+
+	record := func(ctx context.Context, method string, err error, latency float64) {
+		attrs := attributesFromError(method, err)
+		latencyHistgram.Record(ctx, latency, metrics.WithAttributes(attrs...))
+		completedRPCCnt.Add(ctx, 1, metrics.WithAttributes(attrs...))
+	}
+	return func(ctx context.Context, method string, req, reply any, cc *grpc.ClientConn, invoker grpc.UnaryInvoker, opts ...grpc.CallOption) error {
+			now := time.Now()
+			err := invoker(ctx, method, req, reply, cc, opts...)
+			elapsedTime := time.Since(now)
+			record(ctx, method, err, float64(elapsedTime)/float64(time.Millisecond))
+			return err
+		}, func(ctx context.Context, desc *grpc.StreamDesc, cc *grpc.ClientConn, method string, streamer grpc.Streamer, opts ...grpc.CallOption) (grpc.ClientStream, error) {
+			now := time.Now()
+			_, err := streamer(ctx, desc, cc, method, opts...)
+			elapsedTime := time.Since(now)
+			record(ctx, method, err, float64(elapsedTime)/float64(time.Millisecond))
+			return nil, nil
+		}, nil
+}
+
+func attributesFromError(method string, err error) []attribute.KeyValue {
+	code := codes.OK // default error is success when error is nil
+	if err != nil {
+		st, _ := status.FromError(err)
+		if st != nil {
+			code = st.Code()
+		}
+	}
+	return []attribute.KeyValue{
+		attribute.String(gRPCMethodKeyName, method),
+		attribute.String(gRPCStatus, code.String()),
+	}
+}

internal/net/grpc/option.go

@@ -24,8 +24,10 @@
 
 	"github.com/vdaas/vald/internal/backoff"
 	"github.com/vdaas/vald/internal/circuitbreaker"
+	"github.com/vdaas/vald/internal/errors"
 	"github.com/vdaas/vald/internal/log"
 	"github.com/vdaas/vald/internal/net"
+	"github.com/vdaas/vald/internal/net/grpc/interceptor/client/metric"
 	"github.com/vdaas/vald/internal/net/grpc/interceptor/client/trace"
 	"github.com/vdaas/vald/internal/strings"
 	"github.com/vdaas/vald/internal/sync/errgroup"
@@ -353,6 +355,16 @@
 					grpc.WithUnaryInterceptor(trace.UnaryClientInterceptor()),
 					grpc.WithStreamInterceptor(trace.StreamClientInterceptor()),
 				)
+			case "metricinterceptor", "metric":
+				uci, sci, err := metric.ClientMetricInterceptors()
+				if err != nil {
+					lerr := errors.NewErrCriticalOption("gRPCInterceptors", "metric", errors.Wrap(err, "failed to create interceptor"))
+					log.Warn(lerr.Error())
+				}
+				g.dopts = append(g.dopts,
+					grpc.WithUnaryInterceptor(uci),
+					grpc.WithStreamInterceptor(sci),
+				)
 			default:
 			}
 		}

pkg/tools/benchmark/job/config/config.go

@@ -54,6 +54,7 @@
 	NAME                    = os.Getenv("CRD_NAME")
 	JOBNAME_ANNOTATION      = "before-job-name"
 	JOBNAMESPACE_ANNOTATION = "before-job-namespace"
+	SERVICE_NAME            = "vald-benchmark-job"
 )
 
 // NewConfig represents the set config from the given setting file path.
@@ -74,6 +75,7 @@
 
 	if cfg.Observability != nil {
 		cfg.Observability = cfg.Observability.Bind()
+		cfg.Observability.OTLP.Attribute.ServiceName = SERVICE_NAME
 	} else {
 		cfg.Observability = new(config.Observability)
 	}

pkg/tools/benchmark/job/usecase/benchmarkd.go

@@ -64,12 +64,34 @@
 		}
 	}
 
+	// bind metrics interceptor
+	var clientInterceptors []string
+	var obs observability.Observability
+	if cfg.Observability.Enabled {
+		obs, err = observability.NewWithConfig(
+			cfg.Observability,
+			infometrics.New("vald_benchmark_job_info", "Benchmark Job info", *cfg.Job),
+		)
+		if err != nil {
+			return nil, err
+		}
+		var str []string
+		str = append(str, "metric")
+		if cfg.Observability.Trace.Enabled {
+			str = append(str, "trace")
+		}
+		clientInterceptors = str
+	}
+
 	copts, err := cfg.Job.ClientConfig.Opts()
 	if err != nil {
 		return nil, err
 	}
 	if cfg.Job.ClientConfig.DialOption == nil {
-		copts = append(copts, grpc.WithInsecure(true))
+		copts = append(copts,
+			grpc.WithInsecure(true),
+			grpc.WithClientInterceptors(clientInterceptors...),
+		)
 	}
 	gcli := grpc.New(copts...)
 	vcli, err := vald.New(
@@ -131,17 +153,6 @@
 		}),
 	}
 
-	var obs observability.Observability
-	if cfg.Observability.Enabled {
-		obs, err = observability.NewWithConfig(
-			cfg.Observability,
-			infometrics.New("vald_benchmark_job_info", "Benchmark Job info", *cfg.Job),
-		)
-		if err != nil {
-			return nil, err
-		}
-	}
-
 	srv, err := starter.New(
 		starter.WithConfig(cfg.Server),
 		starter.WithREST(func(sc *iconf.Server) []server.Option {
@@ -197,6 +208,7 @@
 		if r.observability != nil {
 			oech = r.observability.Start(ctx)
 		}
+
 		dech, err = r.job.Start(ctx)
 		if err != nil {
 			ech <- err

pkg/tools/benchmark/operator/service/operator.go

@@ -59,6 +59,7 @@ type operator struct {
 	jobNamespace       string
 	jobImage           string
 	jobImagePullPolicy string
+	configMapName      string
 	scenarios          *atomic.Pointer[map[string]*scenario]
 	benchjobs          *atomic.Pointer[map[string]*v1.ValdBenchmarkJob]
 	jobs               *atomic.Pointer[map[string]string]
@@ -455,7 +456,6 @@ func (o *operator) createBenchmarkJob(ctx context.Context, scenario v1.ValdBench
 		}
 		// set status
 		bj.Status = v1.BenchmarkJobNotReady
-		// TODO: set metrics
 		// create benchmark job resource
 		c := o.ctrl.GetManager().GetClient()
 		if err := c.Create(ctx, bj); err != nil {
@@ -476,6 +476,7 @@ func (o *operator) createJob(ctx context.Context, bjr v1.ValdBenchmarkJob) error
 		benchjob.WithContainerName(bjr.GetName()),
 		benchjob.WithContainerImage(o.jobImage),
 		benchjob.WithImagePullPolicy(benchjob.ImagePullPolicy(o.jobImagePullPolicy)),
+		benchjob.WithOperatorConfigMap(o.configMapName),
 	)
 	if err != nil {
 		return err