Merge pull request #133211 from saschagrunert/grpc-close

Add remote runtime/image `Close()` API

pkg/kubelet/kuberuntime/instrumented_services.go

@@ -335,6 +335,15 @@ func (in instrumentedImageManagerService) ImageFsInfo(ctx context.Context) (*run
 	return fsInfo, nil
 }
 
+func (in instrumentedImageManagerService) Close() error {
+	const operation = "close"
+	defer recordOperation(operation, time.Now())
+
+	err := in.service.Close()
+	recordError(operation, err)
+	return err
+}
+
 func (in instrumentedRuntimeService) CheckpointContainer(ctx context.Context, options *runtimeapi.CheckpointContainerRequest) error {
 	const operation = "checkpoint_container"
 	defer recordOperation(operation, time.Now())
@@ -379,3 +388,12 @@ func (in instrumentedRuntimeService) RuntimeConfig(ctx context.Context) (*runtim
 	recordError(operation, err)
 	return out, err
 }
+
+func (in instrumentedRuntimeService) Close() error {
+	const operation = "close"
+	defer recordOperation(operation, time.Now())
+
+	err := in.service.Close()
+	recordError(operation, err)
+	return err
+}

staging/src/k8s.io/cri-api/pkg/apis/services.go

@@ -123,6 +123,8 @@ type RuntimeService interface {
 	Status(ctx context.Context, verbose bool) (*runtimeapi.StatusResponse, error)
 	// RuntimeConfig returns the configuration information of the runtime.
 	RuntimeConfig(ctx context.Context) (*runtimeapi.RuntimeConfigResponse, error)
+	// Close will shutdown the internal gRPC client connection.
+	Close() error
 }
 
 // ImageManagerService interface should be implemented by a container image
@@ -139,4 +141,6 @@ type ImageManagerService interface {
 	RemoveImage(ctx context.Context, image *runtimeapi.ImageSpec) error
 	// ImageFsInfo returns information of the filesystem(s) used to store the read-only layers and the writeable layer.
 	ImageFsInfo(ctx context.Context) (*runtimeapi.ImageFsInfoResponse, error)
+	// Close will shutdown the internal gRPC client connection.
+	Close() error
 }

staging/src/k8s.io/cri-api/pkg/apis/testing/fake_image_service.go

@@ -254,3 +254,16 @@ type pulledImage struct {
 	imageSpec  *runtimeapi.ImageSpec
 	authConfig *runtimeapi.AuthConfig
 }
+
+// Close will shutdown the internal gRPC client connection.
+func (r *FakeImageService) Close() error {
+	r.Lock()
+	defer r.Unlock()
+
+	r.Called = append(r.Called, "Close")
+	if err := r.popError("Close"); err != nil {
+		return err
+	}
+
+	return nil
+}

staging/src/k8s.io/cri-api/pkg/apis/testing/fake_runtime_service.go

@@ -808,3 +808,16 @@ func (r *FakeRuntimeService) UpdatePodSandboxResources(context.Context, *runtime
 
 	return &runtimeapi.UpdatePodSandboxResourcesResponse{}, nil
 }
+
+// Close will shutdown the internal gRPC client connection.
+func (r *FakeRuntimeService) Close() error {
+	r.Lock()
+	defer r.Unlock()
+
+	r.Called = append(r.Called, "Close")
+	if err := r.popError("Close"); err != nil {
+		return err
+	}
+
+	return nil
+}

staging/src/k8s.io/cri-client/pkg/fake/fake_runtime.go

@@ -373,3 +373,8 @@ func (f *RemoteRuntime) RuntimeConfig(ctx context.Context, req *kubeapi.RuntimeC
 func (f *RemoteRuntime) UpdatePodSandboxResources(ctx context.Context, req *kubeapi.UpdatePodSandboxResourcesRequest) (*kubeapi.UpdatePodSandboxResourcesResponse, error) {
 	return f.RuntimeService.UpdatePodSandboxResources(ctx, req)
 }
+
+// Close will shutdown the internal gRPC client connection.
+func (f *RemoteRuntime) Close() error {
+	return f.RuntimeService.Close()
+}

staging/src/k8s.io/cri-client/pkg/remote_image.go

@@ -44,6 +44,7 @@ type remoteImageService struct {
 	timeout     time.Duration
 	imageClient runtimeapi.ImageServiceClient
 	logger      *klog.Logger
+	conn        *grpc.ClientConn
 }
 
 // NewRemoteImageService creates a new internalapi.ImageManagerService.
@@ -94,6 +95,7 @@ func NewRemoteImageService(endpoint string, connectionTimeout time.Duration, tp
 	service := &remoteImageService{
 		timeout: connectionTimeout,
 		logger:  logger,
+		conn:    conn,
 	}
 	if err := service.validateServiceConnection(ctx, conn, endpoint); err != nil {
 		return nil, fmt.Errorf("validate service connection: %w", err)
@@ -103,6 +105,12 @@ func NewRemoteImageService(endpoint string, connectionTimeout time.Duration, tp
 
 }
 
+// Close will shutdown the internal gRPC client connection.
+func (r *remoteImageService) Close() error {
+	r.log(3, "Closing image service connection")
+	return r.conn.Close()
+}
+
 func (r *remoteImageService) log(level int, msg string, keyAndValues ...any) {
 	internal.Log(r.logger, level, msg, keyAndValues...)
 }

staging/src/k8s.io/cri-client/pkg/remote_runtime.go

@@ -50,6 +50,7 @@ type remoteRuntimeService struct {
 	// Cache last per-container error message to reduce log spam
 	logReduction *logreduction.LogReduction
 	logger       *klog.Logger
+	conn         *grpc.ClientConn
 }
 
 const (
@@ -127,6 +128,7 @@ func NewRemoteRuntimeService(endpoint string, connectionTimeout time.Duration, t
 		timeout:      connectionTimeout,
 		logReduction: logreduction.NewLogReduction(identicalErrorDelay),
 		logger:       logger,
+		conn:         conn,
 	}
 
 	if err := service.validateServiceConnection(ctx, conn, endpoint); err != nil {
@@ -136,6 +138,12 @@ func NewRemoteRuntimeService(endpoint string, connectionTimeout time.Duration, t
 	return service, nil
 }
 
+// Close will shutdown the internal gRPC client connection.
+func (r *remoteRuntimeService) Close() error {
+	r.log(3, "Closing runtime service connection")
+	return r.conn.Close()
+}
+
 func (r *remoteRuntimeService) log(level int, msg string, keyAndValues ...any) {
 	internal.Log(r.logger, level, msg, keyAndValues...)
 }