Implement client metrics interceptor for continuous benchmark job

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

@@ -35,10 +35,14 @@ const (
 	RestartPolicyAlways    RestartPolicy = "Always"
 	RestartPolicyOnFailure RestartPolicy = "OnFailure"
 	RestartPolicyNever     RestartPolicy = "Never"
+
+	volumeName    = "vald-benchmark-job-config"
+	configMapName = "vald-benchmark-operator-config"
+	svcAccount    = "vald-benchmark-operator"
 )
 
-const (
-	svcAccount = "vald-benchmark-operator"
+var (
+	mode = int32(420)
 )
 
 type BenchmarkJobTpl interface {
@@ -138,6 +142,52 @@ 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{
+						// FIXME: get benchmark operator configmap name
+						Name: configMapName,
+					},
+					DefaultMode: &mode,
+				},
 			},
 		},
 	}

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

@@ -0,0 +1,79 @@
+// 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"
+
+	"github.com/vdaas/vald/internal/errors"
+	"github.com/vdaas/vald/internal/log"
+	"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...))
+	}
+	// FIXME: implement sending metric data
+	log.Debug(record)
+	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