Skip to content

Commit 6491802

Browse files
k8s-ci-robotk8s-ci-robot
authored
Merge pull request #1886 from k8s-infra-cherrypick-robot/cherry-pick-1876-to-release-1.15
[release-1.15] Refactor metric defer() statements to gRPC metric interceptor
2 parents 9b98537 + 984b61c commit 6491802

File tree

12 files changed

+379
-123
lines changed

12 files changed

+379
-123
lines changed

‎cmd/gce-pd-csi-driver/main.go‎

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -143,6 +143,7 @@ func handle() {
143143
}()
144144
}
145145

146+
var metricsManager *metrics.MetricsManager = nil
146147
if *runControllerService && *httpEndpoint != "" {
147148
mm := metrics.NewMetricsManager()
148149
mm.InitializeHttpHandler(*httpEndpoint, *metricsPath)
@@ -151,6 +152,7 @@ func handle() {
151152
if metrics.IsGKEComponentVersionAvailable() {
152153
mm.EmitGKEComponentVersion()
153154
}
155+
metricsManager = &mm
154156
}
155157

156158
if len(*extraVolumeLabelsStr) > 0 && !*runControllerService {
@@ -261,7 +263,7 @@ func handle() {
261263
gce.WaitForOpBackoff.Steps = *waitForOpBackoffSteps
262264
gce.WaitForOpBackoff.Cap = *waitForOpBackoffCap
263265

264-
gceDriver.Run(*endpoint, *grpcLogCharCap, *enableOtelTracing)
266+
gceDriver.Run(*endpoint, *grpcLogCharCap, *enableOtelTracing, metricsManager)
265267
}
266268

267269
func notEmpty(v string) bool {

‎pkg/gce-pd-csi-driver/controller.go‎

Lines changed: 17 additions & 85 deletions
Original file line numberDiff line numberDiff line change
@@ -21,7 +21,6 @@ import (
2121
"math/rand"
2222
neturl "net/url"
2323
"sort"
24-
"strconv"
2524
"strings"
2625
"time"
2726

@@ -303,12 +302,14 @@ func (gceCS *GCEControllerServer) CreateVolume(ctx context.Context, req *csi.Cre
303302

304303
func (gceCS *GCEControllerServer) createVolumeInternal(ctx context.Context, req *csi.CreateVolumeRequest) (*csi.CreateVolumeResponse, error) {
305304
var err error
306-
diskTypeForMetric := metrics.DefaultDiskTypeForMetric
307-
enableConfidentialCompute := metrics.DefaultEnableConfidentialCompute
308-
enableStoragePools := metrics.DefaultEnableStoragePools
309-
defer func() {
310-
gceCS.Metrics.RecordOperationErrorMetrics("CreateVolume", err, diskTypeForMetric, enableConfidentialCompute, enableStoragePools)
311-
}()
305+
// Apply Parameters (case-insensitive). We leave validation of
306+
// the values to the cloud provider.
307+
params, err := gceCS.parameterProcessor().ExtractAndDefaultParameters(req.GetParameters(), gceCS.Driver.extraVolumeLabels, gceCS.Driver.extraTags)
308+
metrics.UpdateRequestMetadataFromParams(ctx, params)
309+
if err != nil {
310+
return nil, status.Errorf(codes.InvalidArgument, "failed to extract parameters: %v", err.Error())
311+
}
312+
312313
// Validate arguments
313314
volumeCapabilities := req.GetVolumeCapabilities()
314315
capacityRange := req.GetCapacityRange()
@@ -328,17 +329,6 @@ func (gceCS *GCEControllerServer) createVolumeInternal(ctx context.Context, req
328329
if err != nil {
329330
return nil, status.Errorf(codes.InvalidArgument, "VolumeCapabilities is invalid: %v", err.Error())
330331
}
331-
332-
// Apply Parameters (case-insensitive). We leave validation of
333-
// the values to the cloud provider.
334-
params, err := gceCS.parameterProcessor().ExtractAndDefaultParameters(req.GetParameters(), gceCS.Driver.extraVolumeLabels, gceCS.Driver.extraTags)
335-
diskTypeForMetric = params.DiskType
336-
enableConfidentialCompute = strconv.FormatBool(params.EnableConfidentialCompute)
337-
hasStoragePools := len(params.StoragePools) > 0
338-
enableStoragePools = strconv.FormatBool(hasStoragePools)
339-
if err != nil {
340-
return nil, status.Errorf(codes.InvalidArgument, "failed to extract parameters: %v", err.Error())
341-
}
342332
// https://github.com/container-storage-interface/spec/blob/master/spec.md#createvolume
343333
// mutable_parameters MUST take precedence over the values from parameters.
344334
mutableParams := req.GetMutableParameters()
@@ -782,14 +772,6 @@ func (gceCS *GCEControllerServer) ControllerModifyVolume(ctx context.Context, re
782772
return nil, status.Error(codes.InvalidArgument, "volume ID must be provided")
783773
}
784774

785-
diskType := metrics.DefaultDiskTypeForMetric
786-
enableConfidentialCompute := metrics.DefaultEnableConfidentialCompute
787-
enableStoragePools := metrics.DefaultEnableStoragePools
788-
789-
defer func() {
790-
gceCS.Metrics.RecordOperationErrorMetrics("ControllerModifyVolume", err, diskType, enableConfidentialCompute, enableStoragePools)
791-
}()
792-
793775
project, volKey, err := common.VolumeIDToKey(volumeID)
794776
if err != nil {
795777
// Cannot find volume associated with this ID because VolumeID is not in the correct format
@@ -806,6 +788,7 @@ func (gceCS *GCEControllerServer) ControllerModifyVolume(ctx context.Context, re
806788
klog.V(4).Infof("Modify Volume Parameters for %s: %v", volumeID, volumeModifyParams)
807789

808790
existingDisk, err := gceCS.CloudProvider.GetDisk(ctx, project, volKey, gce.GCEAPIVersionBeta)
791+
metrics.UpdateRequestMetadataFromDisk(ctx, existingDisk)
809792

810793
if err != nil {
811794
err = fmt.Errorf("Failed to get volume: %w", err)
@@ -816,9 +799,9 @@ func (gceCS *GCEControllerServer) ControllerModifyVolume(ctx context.Context, re
816799
err = status.Errorf(codes.Internal, "failed to get volume : %s", volumeID)
817800
return nil, err
818801
}
819-
diskType = existingDisk.GetPDType()
820802

821803
// Check if the disk supports dynamic IOPS/Throughput provisioning
804+
diskType := existingDisk.GetPDType()
822805
supportsIopsChange := gceCS.diskSupportsIopsChange(diskType)
823806
supportsThroughputChange := gceCS.diskSupportsThroughputChange(diskType)
824807
if !supportsIopsChange && !supportsThroughputChange {
@@ -834,8 +817,6 @@ func (gceCS *GCEControllerServer) ControllerModifyVolume(ctx context.Context, re
834817
return nil, err
835818
}
836819

837-
enableStoragePools = strconv.FormatBool(existingDisk.GetEnableStoragePools())
838-
839820
err = gceCS.CloudProvider.UpdateDisk(ctx, project, volKey, existingDisk, volumeModifyParams)
840821
if err != nil {
841822
klog.Errorf("Failed to modify volume %s: %v", volumeID, err)
@@ -883,12 +864,6 @@ func getGCEApiVersion(multiWriter bool) gce.GCEAPIVersion {
883864
func (gceCS *GCEControllerServer) deleteMultiZoneDisk(ctx context.Context, req *csi.DeleteVolumeRequest, project string, volKey *meta.Key) (*csi.DeleteVolumeResponse, error) {
884865
// List disks with same name
885866
var err error
886-
diskTypeForMetric := metrics.DefaultDiskTypeForMetric
887-
enableConfidentialCompute := metrics.DefaultEnableConfidentialCompute
888-
enableStoragePools := metrics.DefaultEnableStoragePools
889-
defer func() {
890-
gceCS.Metrics.RecordOperationErrorMetrics("DeleteVolume", err, diskTypeForMetric, enableConfidentialCompute, enableStoragePools)
891-
}()
892867
existingZones := []string{gceCS.CloudProvider.GetDefaultZone()}
893868
zones, err := getDefaultZonesInRegion(ctx, gceCS, existingZones)
894869
if err != nil {
@@ -910,7 +885,7 @@ func (gceCS *GCEControllerServer) deleteMultiZoneDisk(ctx context.Context, req *
910885
}
911886
disk, _ := gceCS.CloudProvider.GetDisk(ctx, project, zonalVolKey, gce.GCEAPIVersionV1)
912887
// TODO: Consolidate the parameters here, rather than taking the last.
913-
diskTypeForMetric, enableConfidentialCompute, enableStoragePools = metrics.GetMetricParameters(disk)
888+
metrics.UpdateRequestMetadataFromDisk(ctx, disk)
914889
err := gceCS.CloudProvider.DeleteDisk(ctx, project, zonalVolKey)
915890
if err != nil {
916891
deleteDiskErrs = append(deleteDiskErrs, gceCS.CloudProvider.DeleteDisk(ctx, project, volKey))
@@ -927,12 +902,6 @@ func (gceCS *GCEControllerServer) deleteMultiZoneDisk(ctx context.Context, req *
927902

928903
func (gceCS *GCEControllerServer) deleteSingleDeviceDisk(ctx context.Context, req *csi.DeleteVolumeRequest, project string, volKey *meta.Key) (*csi.DeleteVolumeResponse, error) {
929904
var err error
930-
diskTypeForMetric := metrics.DefaultDiskTypeForMetric
931-
enableConfidentialCompute := metrics.DefaultEnableConfidentialCompute
932-
enableStoragePools := metrics.DefaultEnableStoragePools
933-
defer func() {
934-
gceCS.Metrics.RecordOperationErrorMetrics("DeleteVolume", err, diskTypeForMetric, enableConfidentialCompute, enableStoragePools)
935-
}()
936905
volumeID := req.GetVolumeId()
937906
project, volKey, err = gceCS.CloudProvider.RepairUnderspecifiedVolumeKey(ctx, project, volKey)
938907
if err != nil {
@@ -948,7 +917,7 @@ func (gceCS *GCEControllerServer) deleteSingleDeviceDisk(ctx context.Context, re
948917
}
949918
defer gceCS.volumeLocks.Release(volumeID)
950919
disk, _ := gceCS.CloudProvider.GetDisk(ctx, project, volKey, gce.GCEAPIVersionV1)
951-
diskTypeForMetric, enableConfidentialCompute, enableStoragePools = metrics.GetMetricParameters(disk)
920+
metrics.UpdateRequestMetadataFromDisk(ctx, disk)
952921
err = gceCS.CloudProvider.DeleteDisk(ctx, project, volKey)
953922
if err != nil {
954923
return nil, common.LoggedError("Failed to delete disk: ", err)
@@ -960,12 +929,6 @@ func (gceCS *GCEControllerServer) deleteSingleDeviceDisk(ctx context.Context, re
960929

961930
func (gceCS *GCEControllerServer) ControllerPublishVolume(ctx context.Context, req *csi.ControllerPublishVolumeRequest) (*csi.ControllerPublishVolumeResponse, error) {
962931
var err error
963-
diskTypeForMetric := metrics.DefaultDiskTypeForMetric
964-
enableConfidentialCompute := metrics.DefaultEnableConfidentialCompute
965-
enableStoragePools := metrics.DefaultEnableStoragePools
966-
defer func() {
967-
gceCS.Metrics.RecordOperationErrorMetrics("ControllerPublishVolume", err, diskTypeForMetric, enableConfidentialCompute, enableStoragePools)
968-
}()
969932
// Only valid requests will be accepted
970933
_, _, _, err = gceCS.validateControllerPublishVolumeRequest(ctx, req)
971934
if err != nil {
@@ -978,7 +941,7 @@ func (gceCS *GCEControllerServer) ControllerPublishVolume(ctx context.Context, r
978941
}
979942

980943
resp, err, disk := gceCS.executeControllerPublishVolume(ctx, req)
981-
diskTypeForMetric, enableConfidentialCompute, enableStoragePools = metrics.GetMetricParameters(disk)
944+
metrics.UpdateRequestMetadataFromDisk(ctx, disk)
982945
if err != nil {
983946
klog.Infof("For node %s adding backoff due to error for volume %s: %v", req.NodeId, req.VolumeId, err)
984947
gceCS.errorBackoff.next(backoffId, common.CodeForError(err))
@@ -1192,12 +1155,6 @@ func (gceCS *GCEControllerServer) executeControllerPublishVolume(ctx context.Con
11921155

11931156
func (gceCS *GCEControllerServer) ControllerUnpublishVolume(ctx context.Context, req *csi.ControllerUnpublishVolumeRequest) (*csi.ControllerUnpublishVolumeResponse, error) {
11941157
var err error
1195-
diskTypeForMetric := metrics.DefaultDiskTypeForMetric
1196-
enableConfidentialCompute := metrics.DefaultEnableConfidentialCompute
1197-
enableStoragePools := metrics.DefaultEnableStoragePools
1198-
defer func() {
1199-
gceCS.Metrics.RecordOperationErrorMetrics("ControllerUnpublishVolume", err, diskTypeForMetric, enableConfidentialCompute, enableStoragePools)
1200-
}()
12011158
_, _, err = gceCS.validateControllerUnpublishVolumeRequest(ctx, req)
12021159
if err != nil {
12031160
return nil, err
@@ -1209,7 +1166,7 @@ func (gceCS *GCEControllerServer) ControllerUnpublishVolume(ctx context.Context,
12091166
return nil, status.Errorf(gceCS.errorBackoff.code(backoffId), "ControllerUnpublish not permitted on node %q due to backoff condition", req.NodeId)
12101167
}
12111168
resp, err, disk := gceCS.executeControllerUnpublishVolume(ctx, req)
1212-
diskTypeForMetric, enableConfidentialCompute, enableStoragePools = metrics.GetMetricParameters(disk)
1169+
metrics.UpdateRequestMetadataFromDisk(ctx, disk)
12131170
if err != nil {
12141171
klog.Infof("For node %s adding backoff due to error for volume %s: %v", req.NodeId, req.VolumeId, err)
12151172
gceCS.errorBackoff.next(backoffId, common.CodeForError(err))
@@ -1316,12 +1273,6 @@ func (gceCS *GCEControllerServer) parameterProcessor() *common.ParameterProcesso
13161273

13171274
func (gceCS *GCEControllerServer) ValidateVolumeCapabilities(ctx context.Context, req *csi.ValidateVolumeCapabilitiesRequest) (*csi.ValidateVolumeCapabilitiesResponse, error) {
13181275
var err error
1319-
diskTypeForMetric := metrics.DefaultDiskTypeForMetric
1320-
enableConfidentialCompute := metrics.DefaultEnableConfidentialCompute
1321-
enableStoragePools := metrics.DefaultEnableStoragePools
1322-
defer func() {
1323-
gceCS.Metrics.RecordOperationErrorMetrics("ValidateVolumeCapabilities", err, diskTypeForMetric, enableConfidentialCompute, enableStoragePools)
1324-
}()
13251276
if req.GetVolumeCapabilities() == nil || len(req.GetVolumeCapabilities()) == 0 {
13261277
return nil, status.Error(codes.InvalidArgument, "Volume Capabilities must be provided")
13271278
}
@@ -1348,7 +1299,7 @@ func (gceCS *GCEControllerServer) ValidateVolumeCapabilities(ctx context.Context
13481299
defer gceCS.volumeLocks.Release(volumeID)
13491300

13501301
disk, err := gceCS.CloudProvider.GetDisk(ctx, project, volKey, gce.GCEAPIVersionV1)
1351-
diskTypeForMetric, enableConfidentialCompute, enableStoragePools = metrics.GetMetricParameters(disk)
1302+
metrics.UpdateRequestMetadataFromDisk(ctx, disk)
13521303
if err != nil {
13531304
if gce.IsGCENotFoundError(err) {
13541305
return nil, status.Errorf(codes.NotFound, "Could not find disk %v: %v", volKey.Name, err.Error())
@@ -1564,12 +1515,6 @@ func (gceCS *GCEControllerServer) ControllerGetCapabilities(ctx context.Context,
15641515

15651516
func (gceCS *GCEControllerServer) CreateSnapshot(ctx context.Context, req *csi.CreateSnapshotRequest) (*csi.CreateSnapshotResponse, error) {
15661517
var err error
1567-
diskTypeForMetric := metrics.DefaultDiskTypeForMetric
1568-
enableConfidentialCompute := metrics.DefaultEnableConfidentialCompute
1569-
enableStoragePools := metrics.DefaultEnableStoragePools
1570-
defer func() {
1571-
gceCS.Metrics.RecordOperationErrorMetrics("CreateSnapshot", err, diskTypeForMetric, enableConfidentialCompute, enableStoragePools)
1572-
}()
15731518
// Validate arguments
15741519
volumeID := req.GetSourceVolumeId()
15751520
if len(req.Name) == 0 {
@@ -1595,7 +1540,7 @@ func (gceCS *GCEControllerServer) CreateSnapshot(ctx context.Context, req *csi.C
15951540

15961541
// Check if volume exists
15971542
disk, err := gceCS.CloudProvider.GetDisk(ctx, project, volKey, gce.GCEAPIVersionV1)
1598-
diskTypeForMetric, enableConfidentialCompute, enableStoragePools = metrics.GetMetricParameters(disk)
1543+
metrics.UpdateRequestMetadataFromDisk(ctx, disk)
15991544
if err != nil {
16001545
if gce.IsGCENotFoundError(err) {
16011546
return nil, status.Errorf(codes.NotFound, "CreateSnapshot could not find disk %v: %v", volKey.String(), err.Error())
@@ -1823,12 +1768,6 @@ func isCSISnapshotReady(status string) (bool, error) {
18231768

18241769
func (gceCS *GCEControllerServer) DeleteSnapshot(ctx context.Context, req *csi.DeleteSnapshotRequest) (*csi.DeleteSnapshotResponse, error) {
18251770
var err error
1826-
diskTypeForMetric := metrics.DefaultDiskTypeForMetric
1827-
enableConfidentialCompute := metrics.DefaultEnableConfidentialCompute
1828-
enableStoragePools := metrics.DefaultEnableStoragePools
1829-
defer func() {
1830-
gceCS.Metrics.RecordOperationErrorMetrics("DeleteSnapshot", err, diskTypeForMetric, enableConfidentialCompute, enableStoragePools)
1831-
}()
18321771
// Validate arguments
18331772
snapshotID := req.GetSnapshotId()
18341773
if len(snapshotID) == 0 {
@@ -1913,14 +1852,7 @@ func (gceCS *GCEControllerServer) ListSnapshots(ctx context.Context, req *csi.Li
19131852
}
19141853

19151854
func (gceCS *GCEControllerServer) ControllerExpandVolume(ctx context.Context, req *csi.ControllerExpandVolumeRequest) (*csi.ControllerExpandVolumeResponse, error) {
1916-
19171855
var err error
1918-
diskTypeForMetric := metrics.DefaultDiskTypeForMetric
1919-
enableConfidentialCompute := metrics.DefaultEnableConfidentialCompute
1920-
enableStoragePools := metrics.DefaultEnableStoragePools
1921-
defer func() {
1922-
gceCS.Metrics.RecordOperationErrorMetrics("ControllerExpandVolume", err, diskTypeForMetric, enableConfidentialCompute, enableStoragePools)
1923-
}()
19241856
volumeID := req.GetVolumeId()
19251857
if len(volumeID) == 0 {
19261858
return nil, status.Error(codes.InvalidArgument, "ControllerExpandVolume volume ID must be provided")
@@ -1950,7 +1882,7 @@ func (gceCS *GCEControllerServer) ControllerExpandVolume(ctx context.Context, re
19501882
}
19511883

19521884
sourceDisk, err := gceCS.CloudProvider.GetDisk(ctx, project, volKey, gce.GCEAPIVersionV1)
1953-
diskTypeForMetric, enableConfidentialCompute, enableStoragePools = metrics.GetMetricParameters(sourceDisk)
1885+
metrics.UpdateRequestMetadataFromDisk(ctx, sourceDisk)
19541886
resizedGb, err := gceCS.CloudProvider.ResizeDisk(ctx, project, volKey, reqBytes)
19551887

19561888
if err != nil {

‎pkg/gce-pd-csi-driver/gce-pd-driver.go‎

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -27,6 +27,7 @@ import (
2727
"sigs.k8s.io/gcp-compute-persistent-disk-csi-driver/pkg/deviceutils"
2828
gce "sigs.k8s.io/gcp-compute-persistent-disk-csi-driver/pkg/gce-cloud-provider/compute"
2929
metadataservice "sigs.k8s.io/gcp-compute-persistent-disk-csi-driver/pkg/gce-cloud-provider/metadata"
30+
"sigs.k8s.io/gcp-compute-persistent-disk-csi-driver/pkg/metrics"
3031
mountmanager "sigs.k8s.io/gcp-compute-persistent-disk-csi-driver/pkg/mount-manager"
3132
)
3233

@@ -170,12 +171,12 @@ func NewControllerServer(gceDriver *GCEDriver, cloudProvider gce.GCECompute, err
170171
}
171172
}
172173

173-
func (gceDriver *GCEDriver) Run(endpoint string, grpcLogCharCap int, enableOtelTracing bool) {
174+
func (gceDriver *GCEDriver) Run(endpoint string, grpcLogCharCap int, enableOtelTracing bool, metricsManager *metrics.MetricsManager) {
174175
maxLogChar = grpcLogCharCap
175176

176177
klog.V(4).Infof("Driver: %v", gceDriver.name)
177178
//Start the nonblocking GRPC
178-
s := NewNonBlockingGRPCServer(enableOtelTracing)
179+
s := NewNonBlockingGRPCServer(enableOtelTracing, metricsManager)
179180
// TODO(#34): Only start specific servers based on a flag.
180181
// In the future have this only run specific combinations of servers depending on which version this is.
181182
// The schema for that was in util. basically it was just s.start but with some nil servers.

‎pkg/gce-pd-csi-driver/gce-pd-driver_test.go‎

Lines changed: 8 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -41,8 +41,7 @@ func initBlockingGCEDriver(t *testing.T, cloudDisks []*gce.CloudDisk, readyToExe
4141
return initGCEDriverWithCloudProvider(t, fakeBlockingBlockProvider)
4242
}
4343

44-
func initGCEDriverWithCloudProvider(t *testing.T, cloudProvider gce.GCECompute) *GCEDriver {
45-
vendorVersion := "test-vendor"
44+
func controllerServerForTest(cloudProvider gce.GCECompute) *GCEControllerServer {
4645
gceDriver := GetGCEDriver()
4746
errorBackoffInitialDuration := 200 * time.Millisecond
4847
errorBackoffMaxDuration := 5 * time.Minute
@@ -55,7 +54,13 @@ func initGCEDriverWithCloudProvider(t *testing.T, cloudProvider gce.GCECompute)
5554
SupportsThroughputChange: []string{"hyperdisk-balanced", "hyperdisk-throughput", "hyperdisk-ml"},
5655
}
5756

58-
controllerServer := NewControllerServer(gceDriver, cloudProvider, errorBackoffInitialDuration, errorBackoffMaxDuration, fallbackRequisiteZones, enableStoragePools, multiZoneVolumeHandleConfig, listVolumesConfig, provisionableDisksConfig)
57+
return NewControllerServer(gceDriver, cloudProvider, errorBackoffInitialDuration, errorBackoffMaxDuration, fallbackRequisiteZones, enableStoragePools, multiZoneVolumeHandleConfig, listVolumesConfig, provisionableDisksConfig)
58+
}
59+
60+
func initGCEDriverWithCloudProvider(t *testing.T, cloudProvider gce.GCECompute) *GCEDriver {
61+
vendorVersion := "test-vendor"
62+
gceDriver := GetGCEDriver()
63+
controllerServer := controllerServerForTest(cloudProvider)
5964
err := gceDriver.SetupGCEDriver(driver, vendorVersion, nil, nil, nil, controllerServer, nil)
6065
if err != nil {
6166
t.Fatalf("Failed to setup GCE Driver: %v", err)

‎pkg/gce-pd-csi-driver/server.go‎

Lines changed: 16 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -24,6 +24,7 @@ import (
2424
"go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc"
2525
"google.golang.org/grpc"
2626
"k8s.io/klog/v2"
27+
"sigs.k8s.io/gcp-compute-persistent-disk-csi-driver/pkg/metrics"
2728

2829
csi "github.com/container-storage-interface/spec/lib/go/csi"
2930
)
@@ -40,15 +41,16 @@ type NonBlockingGRPCServer interface {
4041
ForceStop()
4142
}
4243

43-
func NewNonBlockingGRPCServer(enableOtelTracing bool) NonBlockingGRPCServer {
44-
return &nonBlockingGRPCServer{otelTracing: enableOtelTracing}
44+
func NewNonBlockingGRPCServer(enableOtelTracing bool, metricsManager *metrics.MetricsManager) NonBlockingGRPCServer {
45+
return &nonBlockingGRPCServer{otelTracing: enableOtelTracing, metricsManager: metricsManager}
4546
}
4647

4748
// NonBlocking server
4849
type nonBlockingGRPCServer struct {
49-
wg sync.WaitGroup
50-
server *grpc.Server
51-
otelTracing bool
50+
wg sync.WaitGroup
51+
server *grpc.Server
52+
otelTracing bool
53+
metricsManager *metrics.MetricsManager
5254
}
5355

5456
func (s *nonBlockingGRPCServer) Start(endpoint string, ids csi.IdentityServer, cs csi.ControllerServer, ns csi.NodeServer) {
@@ -73,10 +75,17 @@ func (s *nonBlockingGRPCServer) ForceStop() {
7375
}
7476

7577
func (s *nonBlockingGRPCServer) serve(endpoint string, ids csi.IdentityServer, cs csi.ControllerServer, ns csi.NodeServer) {
76-
grpcInterceptor := grpc.UnaryInterceptor(logGRPC)
78+
interceptors := []grpc.UnaryServerInterceptor{logGRPC}
79+
if s.metricsManager != nil {
80+
metricsInterceptor := metrics.MetricInterceptor{
81+
MetricsManager: s.metricsManager,
82+
}
83+
interceptors = append(interceptors, metricsInterceptor.UnaryInterceptor())
84+
}
7785
if s.otelTracing {
78-
grpcInterceptor = grpc.ChainUnaryInterceptor(logGRPC, otelgrpc.UnaryServerInterceptor())
86+
interceptors = append(interceptors, otelgrpc.UnaryServerInterceptor())
7987
}
88+
grpcInterceptor := grpc.ChainUnaryInterceptor(interceptors...)
8089

8190
opts := []grpc.ServerOption{
8291
grpcInterceptor,

0 commit comments

Comments
 (0)