/*
Copyright 2016 The Kubernetes Authors.

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

    http://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 cri

import (
	"context"
	"errors"
	"fmt"
	"io"
	"strings"
	"sync/atomic"
	"time"

	"go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc"
	"go.opentelemetry.io/otel/propagation"
	"go.opentelemetry.io/otel/trace"
	"go.opentelemetry.io/otel/trace/noop"
	"google.golang.org/grpc"
	"google.golang.org/grpc/backoff"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/credentials/insecure"
	"google.golang.org/grpc/status"

	"k8s.io/component-base/logs/logreduction"
	internalapi "k8s.io/cri-api/pkg/apis"
	runtimeapi "k8s.io/cri-api/pkg/apis/runtime/v1"
	"k8s.io/klog/v2"
	utilexec "k8s.io/utils/exec"

	"k8s.io/cri-client/pkg/util"
)

// remoteRuntimeService is a gRPC implementation of internalapi.RuntimeService.
type remoteRuntimeService struct {
	timeout       time.Duration
	runtimeClient runtimeapi.RuntimeServiceClient
	// Cache last per-container error message to reduce log spam
	logReduction *logreduction.LogReduction
	conn         *grpc.ClientConn
	// useStreaming indicates whether to use streaming RPCs for list operations
	// when the CRIListStreaming feature gate is enabled. It falls back to false
	// if a streaming RPC returns Unimplemented. Multiple goroutines may
	// concurrently observe Unimplemented and store false, but that is harmless
	// because the store is idempotent.
	useStreaming atomic.Bool
}

const (
	// How frequently to report identical errors
	identicalErrorDelay = 1 * time.Minute

	// connection parameters
	maxBackoffDelay      = 3 * time.Second
	baseBackoffDelay     = 100 * time.Millisecond
	minConnectionTimeout = 5 * time.Second
)

// CRIVersion is the type for valid Container Runtime Interface (CRI) API
// versions.
type CRIVersion string

var (
	// ErrContainerStatusNil indicates that the returned container status is nil.
	ErrContainerStatusNil = errors.New("container status is nil")

	// ErrCommandTimedOut indicates that the exec sync command timed.
	ErrCommandTimedOut = errors.New("command timed out")
)

const (
	// CRIVersionV1 references the v1 CRI API.
	CRIVersionV1 CRIVersion = "v1"
)

// RemoteRuntimeServiceBuilder builds a new internalapi.RuntimeService.
//
// Construct a builder with NewRemoteRuntimeServiceBuilder, then chain With*
// methods to set non-default options before calling Build. Adding new options
// here is preferred over adding parameters to NewRemoteRuntimeService so that
// default values can be changed in a single place without having to update
// every callsite.
type RemoteRuntimeServiceBuilder struct {
	endpoint          string
	connectionTimeout time.Duration
	tracerProvider    trace.TracerProvider
	// tracerProviderSet tracks whether WithTracerProvider was called, so that
	// an explicit nil can be distinguished from the unset default and used to
	// opt out of installing the otelgrpc stats handler entirely.
	tracerProviderSet bool
	// useStreaming indicates whether to use streaming RPCs for list operations
	// when the CRIListStreaming feature gate is enabled. It is expected to
	// default to true once the feature graduates to GA.
	useStreaming bool
}

// NewRemoteRuntimeServiceBuilder returns a builder with default options for
// constructing a remote runtime service.
func NewRemoteRuntimeServiceBuilder() *RemoteRuntimeServiceBuilder {
	return &RemoteRuntimeServiceBuilder{}
}

// WithEndpoint sets the gRPC endpoint of the remote runtime service.
func (b *RemoteRuntimeServiceBuilder) WithEndpoint(endpoint string) *RemoteRuntimeServiceBuilder {
	b.endpoint = endpoint
	return b
}

// WithConnectionTimeout sets the timeout used when connecting to the remote
// runtime service.
func (b *RemoteRuntimeServiceBuilder) WithConnectionTimeout(connectionTimeout time.Duration) *RemoteRuntimeServiceBuilder {
	b.connectionTimeout = connectionTimeout
	return b
}

// WithTracerProvider sets the OpenTelemetry tracer provider used to
// instrument the gRPC client. If WithTracerProvider is not called, the
// otelgrpc stats handler is installed with a noop tracer provider, so no
// traces are produced but gRPC context propagation still works. Passing a nil
// provider explicitly opts out of installing the stats handler entirely,
// disabling both tracing and context propagation.
func (b *RemoteRuntimeServiceBuilder) WithTracerProvider(tp trace.TracerProvider) *RemoteRuntimeServiceBuilder {
	b.tracerProvider = tp
	b.tracerProviderSet = true
	return b
}

// WithUseStreaming controls whether streaming RPCs are used for list
// operations. If the runtime returns an Unimplemented error, the client
// automatically falls back to the corresponding unary RPC.
// NOTE: useStreaming is supposed to be gated by the CRIListStreaming feature
// gate and is expected to default to true once the feature graduates to GA,
// at which point this method may be removed.
func (b *RemoteRuntimeServiceBuilder) WithUseStreaming(useStreaming bool) *RemoteRuntimeServiceBuilder {
	b.useStreaming = useStreaming
	return b
}

// Build creates a new internalapi.RuntimeService using the configured
// options.
func (b *RemoteRuntimeServiceBuilder) Build(ctx context.Context) (internalapi.RuntimeService, error) {
	if b.endpoint == "" {
		return nil, errors.New("endpoint is required")
	}
	if b.connectionTimeout <= 0 {
		return nil, errors.New("connectionTimeout must be positive")
	}
	logger := klog.FromContext(ctx)
	logger.V(3).Info("Connecting to runtime service", "endpoint", b.endpoint)
	addr, dialer, err := util.GetAddressAndDialer(b.endpoint)
	if err != nil {
		return nil, err
	}
	ctx, cancel := context.WithTimeout(ctx, b.connectionTimeout)
	defer cancel()

	var dialOpts []grpc.DialOption
	dialOpts = append(dialOpts,
		grpc.WithTransportCredentials(insecure.NewCredentials()),
		grpc.WithAuthority("localhost"),
		grpc.WithContextDialer(dialer),
		grpc.WithDefaultCallOptions(grpc.MaxCallRecvMsgSize(maxMsgSize)))
	// When no tracer provider was configured, fall back to a noop provider
	// so context propagation still works without producing real traces.
	// See https://github.com/open-telemetry/opentelemetry-go-contrib/tree/main/examples/passthrough
	if !b.tracerProviderSet {
		b.tracerProvider = noop.NewTracerProvider()
	}
	// Install the otelgrpc stats handler unless the caller explicitly opted
	// out by calling WithTracerProvider(nil).
	if b.tracerProvider != nil {
		tracingOpts := []otelgrpc.Option{
			otelgrpc.WithMessageEvents(otelgrpc.ReceivedEvents, otelgrpc.SentEvents),
			otelgrpc.WithPropagators(propagation.NewCompositeTextMapPropagator(propagation.TraceContext{}, propagation.Baggage{})),
			otelgrpc.WithTracerProvider(b.tracerProvider),
		}
		dialOpts = append(dialOpts,
			grpc.WithStatsHandler(otelgrpc.NewClientHandler(tracingOpts...)))
	}

	connParams := grpc.ConnectParams{
		Backoff: backoff.DefaultConfig,
	}
	connParams.MinConnectTimeout = minConnectionTimeout
	connParams.Backoff.BaseDelay = baseBackoffDelay
	connParams.Backoff.MaxDelay = maxBackoffDelay
	dialOpts = append(dialOpts,
		grpc.WithConnectParams(connParams),
	)

	conn, err := grpc.NewClient(clientTargetForAddress(addr), dialOpts...)
	if err != nil {
		logger.Error(err, "Connect remote runtime service failed", "address", addr)
		return nil, err
	}

	service := &remoteRuntimeService{
		timeout:      b.connectionTimeout,
		logReduction: logreduction.NewLogReduction(identicalErrorDelay),
		conn:         conn,
	}
	service.useStreaming.Store(b.useStreaming)

	if err := service.validateServiceConnection(ctx, conn, b.endpoint); err != nil {
		return nil, fmt.Errorf("validate service connection: %w", err)
	}

	return service, nil
}

// NewRemoteRuntimeService creates a new internalapi.RuntimeService.
// If useStreaming is true, streaming RPCs will be used for list operations
// instead of unary RPCs. If the runtime returns an Unimplemented error,
// the client automatically falls back to unary RPCs.
//
// Deprecated: Use NewRemoteRuntimeServiceBuilder so that default values
// (e.g. useStreaming) can be changed in a single place without updating
// every callsite.
func NewRemoteRuntimeService(ctx context.Context, endpoint string, connectionTimeout time.Duration, tp trace.TracerProvider, useStreaming bool) (internalapi.RuntimeService, error) {
	return NewRemoteRuntimeServiceBuilder().
		WithEndpoint(endpoint).
		WithConnectionTimeout(connectionTimeout).
		WithTracerProvider(tp).
		WithUseStreaming(useStreaming).
		Build(ctx)
}

// Close will shutdown the internal gRPC client connection.
func (r *remoteRuntimeService) Close(ctx context.Context) error {
	logger := klog.FromContext(ctx)
	logger.V(3).Info("Closing runtime service connection")
	return r.conn.Close()
}

// validateServiceConnection tries to connect to the remote runtime service by
// using the CRI v1 API version and fails if that's not possible.
func (r *remoteRuntimeService) validateServiceConnection(ctx context.Context, conn *grpc.ClientConn, endpoint string) error {
	logger := klog.FromContext(ctx)
	logger.V(4).Info("Validating the CRI v1 API runtime version")
	r.runtimeClient = runtimeapi.NewRuntimeServiceClient(conn)

	if _, err := r.runtimeClient.Version(ctx, &runtimeapi.VersionRequest{}); err != nil {
		return fmt.Errorf("validate CRI v1 runtime API for endpoint %q: %w", endpoint, err)
	}

	logger.V(2).Info("Validated CRI v1 runtime API")
	return nil
}

// Version returns the runtime name, runtime version and runtime API version.
func (r *remoteRuntimeService) Version(ctx context.Context, apiVersion string) (*runtimeapi.VersionResponse, error) {
	logger := klog.FromContext(ctx)
	logger.V(10).Info("[RemoteRuntimeService] Version", "apiVersion", apiVersion, "timeout", r.timeout)

	ctx, cancel := context.WithTimeout(ctx, r.timeout)
	defer cancel()

	return r.versionV1(ctx, apiVersion)
}

func (r *remoteRuntimeService) versionV1(ctx context.Context, apiVersion string) (*runtimeapi.VersionResponse, error) {
	typedVersion, err := r.runtimeClient.Version(ctx, &runtimeapi.VersionRequest{
		Version: apiVersion,
	})
	logger := klog.FromContext(ctx)
	if err != nil {
		logger.Error(err, "Version from runtime service failed")
		return nil, err
	}

	logger.V(10).Info("[RemoteRuntimeService] Version Response", "apiVersion", typedVersion)

	var missingFields []string
	if typedVersion.Version == "" {
		missingFields = append(missingFields, "Version")
	}
	if typedVersion.RuntimeName == "" {
		missingFields = append(missingFields, "RuntimeName")
	}
	if typedVersion.RuntimeApiVersion == "" {
		missingFields = append(missingFields, "RuntimeApiVersion")
	}
	if typedVersion.RuntimeVersion == "" {
		missingFields = append(missingFields, "RuntimeVersion")
	}
	if len(missingFields) > 0 {
		return nil, fmt.Errorf("not all fields are set in VersionResponse (missing %s)", strings.Join(missingFields, ", "))
	}

	return typedVersion, err
}

// RunPodSandbox creates and starts a pod-level sandbox. Runtimes should ensure
// the sandbox is in ready state.
func (r *remoteRuntimeService) RunPodSandbox(ctx context.Context, config *runtimeapi.PodSandboxConfig, runtimeHandler string) (string, error) {
	// Use 2 times longer timeout for sandbox operation (4 mins by default)
	// TODO: Make the pod sandbox timeout configurable.
	timeout := r.timeout * 2

	logger := klog.FromContext(ctx)
	logger.V(10).Info("[RemoteRuntimeService] RunPodSandbox", "config", config, "runtimeHandler", runtimeHandler, "timeout", timeout)

	ctx, cancel := context.WithTimeout(ctx, timeout)
	defer cancel()

	resp, err := r.runtimeClient.RunPodSandbox(ctx, &runtimeapi.RunPodSandboxRequest{
		Config:         config,
		RuntimeHandler: runtimeHandler,
	})

	if err != nil {
		logger.Error(err, "RunPodSandbox from runtime service failed")
		return "", err
	}

	podSandboxID := resp.PodSandboxId

	if podSandboxID == "" {
		errorMessage := fmt.Sprintf("PodSandboxId is not set for sandbox %q", config.Metadata)
		err := errors.New(errorMessage)
		logger.Error(err, "RunPodSandbox failed")
		return "", err
	}

	logger.V(10).Info("[RemoteRuntimeService] RunPodSandbox Response", "podSandboxID", podSandboxID)

	return podSandboxID, nil
}

// StopPodSandbox stops the sandbox. If there are any running containers in the
// sandbox, they should be forced to termination.
func (r *remoteRuntimeService) StopPodSandbox(ctx context.Context, podSandBoxID string) (err error) {
	logger := klog.FromContext(ctx)
	logger.V(10).Info("[RemoteRuntimeService] StopPodSandbox", "podSandboxID", podSandBoxID, "timeout", r.timeout)

	ctx, cancel := context.WithTimeout(ctx, r.timeout)
	defer cancel()

	if _, err := r.runtimeClient.StopPodSandbox(ctx, &runtimeapi.StopPodSandboxRequest{
		PodSandboxId: podSandBoxID,
	}); err != nil {
		logger.Error(err, "StopPodSandbox from runtime service failed", "podSandboxID", podSandBoxID)
		return err
	}

	logger.V(10).Info("[RemoteRuntimeService] StopPodSandbox Response", "podSandboxID", podSandBoxID)

	return nil
}

// RemovePodSandbox removes the sandbox. If there are any containers in the
// sandbox, they should be forcibly removed.
func (r *remoteRuntimeService) RemovePodSandbox(ctx context.Context, podSandBoxID string) (err error) {
	logger := klog.FromContext(ctx)
	logger.V(10).Info("[RemoteRuntimeService] RemovePodSandbox", "podSandboxID", podSandBoxID, "timeout", r.timeout)
	ctx, cancel := context.WithTimeout(ctx, r.timeout)
	defer cancel()

	if _, err := r.runtimeClient.RemovePodSandbox(ctx, &runtimeapi.RemovePodSandboxRequest{
		PodSandboxId: podSandBoxID,
	}); err != nil {
		logger.Error(err, "RemovePodSandbox from runtime service failed", "podSandboxID", podSandBoxID)
		return err
	}

	logger.V(10).Info("[RemoteRuntimeService] RemovePodSandbox Response", "podSandboxID", podSandBoxID)

	return nil
}

// PodSandboxStatus returns the status of the PodSandbox.
func (r *remoteRuntimeService) PodSandboxStatus(ctx context.Context, podSandBoxID string, verbose bool) (*runtimeapi.PodSandboxStatusResponse, error) {
	logger := klog.FromContext(ctx)
	logger.V(10).Info("[RemoteRuntimeService] PodSandboxStatus", "podSandboxID", podSandBoxID, "timeout", r.timeout)
	ctx, cancel := context.WithTimeout(ctx, r.timeout)
	defer cancel()

	return r.podSandboxStatusV1(ctx, podSandBoxID, verbose)
}

func (r *remoteRuntimeService) podSandboxStatusV1(ctx context.Context, podSandBoxID string, verbose bool) (*runtimeapi.PodSandboxStatusResponse, error) {
	resp, err := r.runtimeClient.PodSandboxStatus(ctx, &runtimeapi.PodSandboxStatusRequest{
		PodSandboxId: podSandBoxID,
		Verbose:      verbose,
	})
	if err != nil {
		return nil, err
	}

	logger := klog.FromContext(ctx)
	logger.V(10).Info("[RemoteRuntimeService] PodSandboxStatus Response", "podSandboxID", podSandBoxID, "status", resp.Status)

	status := resp.Status
	if resp.Status != nil {
		if err := verifySandboxStatus(status); err != nil {
			return nil, err
		}
	}

	return resp, nil
}

// ListPodSandbox returns a list of PodSandboxes.
func (r *remoteRuntimeService) ListPodSandbox(ctx context.Context, filter *runtimeapi.PodSandboxFilter) ([]*runtimeapi.PodSandbox, error) {
	logger := klog.FromContext(ctx)
	logger.V(10).Info("[RemoteRuntimeService] ListPodSandbox", "filter", filter, "timeout", r.timeout)
	ctx, cancel := context.WithTimeout(ctx, r.timeout)
	defer cancel()

	if r.useStreaming.Load() {
		return r.streamPodSandboxesV1(ctx, filter)
	}
	return r.listPodSandboxV1(ctx, filter)
}

func (r *remoteRuntimeService) listPodSandboxV1(ctx context.Context, filter *runtimeapi.PodSandboxFilter) ([]*runtimeapi.PodSandbox, error) {
	resp, err := r.runtimeClient.ListPodSandbox(ctx, &runtimeapi.ListPodSandboxRequest{
		Filter: filter,
	})
	logger := klog.FromContext(ctx)
	if err != nil {
		logger.Error(err, "ListPodSandbox with filter from runtime service failed", "filter", filter)
		return nil, err
	}

	logger.V(10).Info("[RemoteRuntimeService] ListPodSandbox Response", "filter", filter, "items", resp.Items)

	return resp.Items, nil
}

func (r *remoteRuntimeService) streamPodSandboxesV1(ctx context.Context, filter *runtimeapi.PodSandboxFilter) ([]*runtimeapi.PodSandbox, error) {
	logger := klog.FromContext(ctx)
	stream, err := r.runtimeClient.StreamPodSandboxes(ctx, &runtimeapi.StreamPodSandboxesRequest{
		Filter: filter,
	})
	if err != nil {
		logger.Error(err, "StreamPodSandboxes from runtime service failed", "filter", filter)
		return nil, err
	}

	var items []*runtimeapi.PodSandbox
	for {
		resp, err := stream.Recv()
		if errors.Is(err, io.EOF) {
			break
		}
		if err != nil {
			// If the RPC is unimplemented, disable streaming and fall back to the unary RPC.
			// The Unimplemented status is not returned when creating the stream,
			// but when calling Recv() on the stream.
			if status.Code(err) == codes.Unimplemented {
				logger.Info("StreamPodSandboxes not implemented, falling back to ListPodSandbox", "filter", filter)
				r.useStreaming.Store(false)
				return r.listPodSandboxV1(ctx, filter)
			}
			logger.Error(err, "StreamPodSandboxes recv failed", "filter", filter, "itemsReceived", len(items))
			return nil, err
		}
		items = append(items, resp.PodSandboxes...)
	}

	logger.V(10).Info("[RemoteRuntimeService] StreamPodSandboxes Response", "filter", filter, "items", items)

	return items, nil
}

// CreateContainer creates a new container in the specified PodSandbox.
func (r *remoteRuntimeService) CreateContainer(ctx context.Context, podSandBoxID string, config *runtimeapi.ContainerConfig, sandboxConfig *runtimeapi.PodSandboxConfig) (string, error) {
	logger := klog.FromContext(ctx)
	logger.V(10).Info("[RemoteRuntimeService] CreateContainer", "podSandboxID", podSandBoxID, "timeout", r.timeout)
	ctx, cancel := context.WithTimeout(ctx, r.timeout)
	defer cancel()

	return r.createContainerV1(ctx, podSandBoxID, config, sandboxConfig)
}

func (r *remoteRuntimeService) createContainerV1(ctx context.Context, podSandBoxID string, config *runtimeapi.ContainerConfig, sandboxConfig *runtimeapi.PodSandboxConfig) (string, error) {
	logger := klog.FromContext(ctx)
	resp, err := r.runtimeClient.CreateContainer(ctx, &runtimeapi.CreateContainerRequest{
		PodSandboxId:  podSandBoxID,
		Config:        config,
		SandboxConfig: sandboxConfig,
	})
	if err != nil {
		logger.Error(err, "CreateContainer in sandbox from runtime service failed", "podSandboxID", podSandBoxID)
		return "", err
	}

	logger.V(10).Info("[RemoteRuntimeService] CreateContainer", "podSandboxID", podSandBoxID, "containerID", resp.ContainerId)
	if resp.ContainerId == "" {
		errorMessage := fmt.Sprintf("ContainerId is not set for container %q", config.Metadata)
		err := errors.New(errorMessage)
		logger.Error(err, "CreateContainer failed")
		return "", err
	}

	return resp.ContainerId, nil
}

// StartContainer starts the container.
func (r *remoteRuntimeService) StartContainer(ctx context.Context, containerID string) (err error) {
	logger := klog.FromContext(ctx)
	logger.V(10).Info("[RemoteRuntimeService] StartContainer", "containerID", containerID, "timeout", r.timeout)
	ctx, cancel := context.WithTimeout(ctx, r.timeout)
	defer cancel()

	if _, err := r.runtimeClient.StartContainer(ctx, &runtimeapi.StartContainerRequest{
		ContainerId: containerID,
	}); err != nil {
		logger.Error(err, "StartContainer from runtime service failed", "containerID", containerID)
		return err
	}
	logger.V(10).Info("[RemoteRuntimeService] StartContainer Response", "containerID", containerID)

	return nil
}

// StopContainer stops a running container with a grace period (i.e., timeout).
func (r *remoteRuntimeService) StopContainer(ctx context.Context, containerID string, timeout int64) (err error) {
	logger := klog.FromContext(ctx)
	logger.V(10).Info("[RemoteRuntimeService] StopContainer", "containerID", containerID, "timeout", timeout)
	// Use timeout + default timeout (2 minutes) as timeout to leave extra time
	// for SIGKILL container and request latency.
	t := r.timeout + time.Duration(timeout)*time.Second
	ctx, cancel := context.WithTimeout(ctx, t)
	defer cancel()

	r.logReduction.ClearID(containerID)

	if _, err := r.runtimeClient.StopContainer(ctx, &runtimeapi.StopContainerRequest{
		ContainerId: containerID,
		Timeout:     timeout,
	}); err != nil {
		logger.Error(err, "StopContainer from runtime service failed", "containerID", containerID)
		return err
	}
	logger.V(10).Info("[RemoteRuntimeService] StopContainer Response", "containerID", containerID)

	return nil
}

// RemoveContainer removes the container. If the container is running, the container
// should be forced to removal.
func (r *remoteRuntimeService) RemoveContainer(ctx context.Context, containerID string) (err error) {
	logger := klog.FromContext(ctx)
	logger.V(10).Info("[RemoteRuntimeService] RemoveContainer", "containerID", containerID, "timeout", r.timeout)
	ctx, cancel := context.WithTimeout(ctx, r.timeout)
	defer cancel()

	r.logReduction.ClearID(containerID)
	if _, err := r.runtimeClient.RemoveContainer(ctx, &runtimeapi.RemoveContainerRequest{
		ContainerId: containerID,
	}); err != nil {
		logger.Error(err, "RemoveContainer from runtime service failed", "containerID", containerID)
		return err
	}
	logger.V(10).Info("[RemoteRuntimeService] RemoveContainer Response", "containerID", containerID)

	return nil
}

// ListContainers lists containers by filters.
func (r *remoteRuntimeService) ListContainers(ctx context.Context, filter *runtimeapi.ContainerFilter) ([]*runtimeapi.Container, error) {
	logger := klog.FromContext(ctx)
	logger.V(10).Info("[RemoteRuntimeService] ListContainers", "filter", filter, "timeout", r.timeout)
	ctx, cancel := context.WithTimeout(ctx, r.timeout)
	defer cancel()

	if r.useStreaming.Load() {
		return r.streamContainersV1(ctx, filter)
	}
	return r.listContainersV1(ctx, filter)
}

func (r *remoteRuntimeService) listContainersV1(ctx context.Context, filter *runtimeapi.ContainerFilter) ([]*runtimeapi.Container, error) {
	resp, err := r.runtimeClient.ListContainers(ctx, &runtimeapi.ListContainersRequest{
		Filter: filter,
	})
	logger := klog.FromContext(ctx)
	if err != nil {
		logger.Error(err, "ListContainers with filter from runtime service failed", "filter", filter)
		return nil, err
	}
	logger.V(10).Info("[RemoteRuntimeService] ListContainers Response", "filter", filter, "containers", resp.Containers)

	return resp.Containers, nil
}

func (r *remoteRuntimeService) streamContainersV1(ctx context.Context, filter *runtimeapi.ContainerFilter) ([]*runtimeapi.Container, error) {
	logger := klog.FromContext(ctx)
	stream, err := r.runtimeClient.StreamContainers(ctx, &runtimeapi.StreamContainersRequest{
		Filter: filter,
	})
	if err != nil {
		logger.Error(err, "StreamContainers from runtime service failed", "filter", filter)
		return nil, err
	}

	var containers []*runtimeapi.Container
	for {
		resp, err := stream.Recv()
		if errors.Is(err, io.EOF) {
			break
		}
		if err != nil {
			// If the RPC is unimplemented, disable streaming and fall back to the unary RPC.
			// The Unimplemented status is not returned when creating the stream,
			// but when calling Recv() on the stream.
			if status.Code(err) == codes.Unimplemented {
				logger.Info("StreamContainers not implemented, falling back to ListContainers", "filter", filter)
				r.useStreaming.Store(false)
				return r.listContainersV1(ctx, filter)
			}
			logger.Error(err, "StreamContainers recv failed", "filter", filter, "itemsReceived", len(containers))
			return nil, err
		}
		containers = append(containers, resp.Containers...)
	}

	logger.V(10).Info("[RemoteRuntimeService] StreamContainers Response", "filter", filter, "containers", containers)

	return containers, nil
}

// ContainerStatus returns the container status.
func (r *remoteRuntimeService) ContainerStatus(ctx context.Context, containerID string, verbose bool) (*runtimeapi.ContainerStatusResponse, error) {
	logger := klog.FromContext(ctx)
	logger.V(10).Info("[RemoteRuntimeService] ContainerStatus", "containerID", containerID, "timeout", r.timeout)
	ctx, cancel := context.WithTimeout(ctx, r.timeout)
	defer cancel()

	return r.containerStatusV1(ctx, containerID, verbose)
}

func (r *remoteRuntimeService) containerStatusV1(ctx context.Context, containerID string, verbose bool) (*runtimeapi.ContainerStatusResponse, error) {
	resp, err := r.runtimeClient.ContainerStatus(ctx, &runtimeapi.ContainerStatusRequest{
		ContainerId: containerID,
		Verbose:     verbose,
	})
	logger := klog.FromContext(ctx)
	if err != nil {
		// Don't spam the log with endless messages about the same failure.
		if r.logReduction.ShouldMessageBePrinted(err.Error(), containerID) {
			logger.Error(err, "ContainerStatus from runtime service failed", "containerID", containerID)
		}
		return nil, err
	}
	r.logReduction.ClearID(containerID)
	logger.V(10).Info("[RemoteRuntimeService] ContainerStatus Response", "containerID", containerID, "status", resp.Status)

	status := resp.Status
	if resp.Status != nil {
		if err := verifyContainerStatus(status); err != nil {
			logger.Error(err, "verify ContainerStatus failed", "containerID", containerID)
			return nil, err
		}
	}

	return resp, nil
}

// UpdateContainerResources updates a containers resource config
func (r *remoteRuntimeService) UpdateContainerResources(ctx context.Context, containerID string, resources *runtimeapi.ContainerResources) (err error) {
	logger := klog.FromContext(ctx)
	logger.V(10).Info("[RemoteRuntimeService] UpdateContainerResources", "containerID", containerID, "timeout", r.timeout)
	ctx, cancel := context.WithTimeout(ctx, r.timeout)
	defer cancel()

	if _, err := r.runtimeClient.UpdateContainerResources(ctx, &runtimeapi.UpdateContainerResourcesRequest{
		ContainerId: containerID,
		Linux:       resources.GetLinux(),
		Windows:     resources.GetWindows(),
	}); err != nil {
		logger.Error(err, "UpdateContainerResources from runtime service failed", "containerID", containerID)
		return err
	}
	logger.V(10).Info("[RemoteRuntimeService] UpdateContainerResources Response", "containerID", containerID)

	return nil
}

// ExecSync executes a command in the container, and returns the stdout output.
// If command exits with a non-zero exit code, an error is returned.
func (r *remoteRuntimeService) ExecSync(ctx context.Context, containerID string, cmd []string, timeout time.Duration) (stdout []byte, stderr []byte, err error) {
	logger := klog.FromContext(ctx)
	logger.V(10).Info("[RemoteRuntimeService] ExecSync", "containerID", containerID, "timeout", timeout)
	// Do not set timeout when timeout is 0.
	var cancel context.CancelFunc
	if timeout != 0 {
		// Use timeout + default timeout (2 minutes) as timeout to leave some time for
		// the runtime to do cleanup.
		ctx, cancel = context.WithTimeout(ctx, r.timeout+timeout)
	} else {
		ctx, cancel = context.WithCancel(ctx)
	}
	defer cancel()

	return r.execSyncV1(ctx, containerID, cmd, timeout)
}

func (r *remoteRuntimeService) execSyncV1(ctx context.Context, containerID string, cmd []string, timeout time.Duration) (stdout []byte, stderr []byte, err error) {
	timeoutSeconds := int64(timeout.Seconds())
	req := &runtimeapi.ExecSyncRequest{
		ContainerId: containerID,
		Cmd:         cmd,
		Timeout:     timeoutSeconds,
	}
	resp, err := r.runtimeClient.ExecSync(ctx, req)
	logger := klog.FromContext(ctx)
	if err != nil {
		logger.Error(err, "ExecSync cmd from runtime service failed", "containerID", containerID, "cmd", cmd)

		// interpret DeadlineExceeded gRPC errors as timedout errors
		if status.Code(err) == codes.DeadlineExceeded {
			err = fmt.Errorf("%w: %q timed out after %s", ErrCommandTimedOut, strings.Join(cmd, " "), timeout)
		}

		return nil, nil, err
	}

	logger.V(10).Info("[RemoteRuntimeService] ExecSync Response", "containerID", containerID, "exitCode", resp.ExitCode)
	err = nil
	if resp.ExitCode != 0 {
		err = utilexec.CodeExitError{
			Err:  fmt.Errorf("command '%s' exited with %d: %s", strings.Join(cmd, " "), resp.ExitCode, resp.Stderr),
			Code: int(resp.ExitCode),
		}
	}

	return resp.Stdout, resp.Stderr, err
}

// Exec prepares a streaming endpoint to execute a command in the container, and returns the address.
func (r *remoteRuntimeService) Exec(ctx context.Context, req *runtimeapi.ExecRequest) (*runtimeapi.ExecResponse, error) {
	logger := klog.FromContext(ctx)
	logger.V(10).Info("[RemoteRuntimeService] Exec", "timeout", r.timeout)
	ctx, cancel := context.WithTimeout(ctx, r.timeout)
	defer cancel()

	return r.execV1(ctx, req)
}

func (r *remoteRuntimeService) execV1(ctx context.Context, req *runtimeapi.ExecRequest) (*runtimeapi.ExecResponse, error) {
	resp, err := r.runtimeClient.Exec(ctx, req)
	logger := klog.FromContext(ctx)
	if err != nil {
		logger.Error(err, "Exec cmd from runtime service failed", "containerID", req.ContainerId, "cmd", req.Cmd)
		return nil, err
	}
	logger.V(10).Info("[RemoteRuntimeService] Exec Response")

	if resp.Url == "" {
		errorMessage := "URL is not set"
		err := errors.New(errorMessage)
		logger.Error(err, "Exec failed")
		return nil, err
	}

	return resp, nil
}

// Attach prepares a streaming endpoint to attach to a running container, and returns the address.
func (r *remoteRuntimeService) Attach(ctx context.Context, req *runtimeapi.AttachRequest) (*runtimeapi.AttachResponse, error) {
	logger := klog.FromContext(ctx)
	logger.V(10).Info("[RemoteRuntimeService] Attach", "containerID", req.ContainerId, "timeout", r.timeout)
	ctx, cancel := context.WithTimeout(ctx, r.timeout)
	defer cancel()

	return r.attachV1(ctx, req)
}

func (r *remoteRuntimeService) attachV1(ctx context.Context, req *runtimeapi.AttachRequest) (*runtimeapi.AttachResponse, error) {
	logger := klog.FromContext(ctx)
	resp, err := r.runtimeClient.Attach(ctx, req)
	if err != nil {
		logger.Error(err, "Attach container from runtime service failed", "containerID", req.ContainerId)
		return nil, err
	}
	logger.V(10).Info("[RemoteRuntimeService] Attach Response", "containerID", req.ContainerId)

	if resp.Url == "" {
		errorMessage := "URL is not set"
		err := errors.New(errorMessage)
		logger.Error(err, "Attach failed")
		return nil, err
	}
	return resp, nil
}

// PortForward prepares a streaming endpoint to forward ports from a PodSandbox, and returns the address.
func (r *remoteRuntimeService) PortForward(ctx context.Context, req *runtimeapi.PortForwardRequest) (*runtimeapi.PortForwardResponse, error) {
	logger := klog.FromContext(ctx)
	logger.V(10).Info("[RemoteRuntimeService] PortForward", "podSandboxID", req.PodSandboxId, "port", req.Port, "timeout", r.timeout)
	ctx, cancel := context.WithTimeout(ctx, r.timeout)
	defer cancel()

	return r.portForwardV1(ctx, req)
}

func (r *remoteRuntimeService) portForwardV1(ctx context.Context, req *runtimeapi.PortForwardRequest) (*runtimeapi.PortForwardResponse, error) {
	resp, err := r.runtimeClient.PortForward(ctx, req)

	logger := klog.FromContext(ctx)
	if err != nil {
		logger.Error(err, "PortForward from runtime service failed", "podSandboxID", req.PodSandboxId)
		return nil, err
	}
	logger.V(10).Info("[RemoteRuntimeService] PortForward Response", "podSandboxID", req.PodSandboxId)

	if resp.Url == "" {
		errorMessage := "URL is not set"
		err := errors.New(errorMessage)
		logger.Error(err, "PortForward failed")
		return nil, err
	}

	return resp, nil
}

// UpdatePodSandboxResources synchronously updates the PodSandboxConfig with
// the pod-level resource configuration.
func (r *remoteRuntimeService) UpdatePodSandboxResources(ctx context.Context, req *runtimeapi.UpdatePodSandboxResourcesRequest) (*runtimeapi.UpdatePodSandboxResourcesResponse, error) {
	logger := klog.FromContext(ctx)
	logger.V(10).Info("[RemoteRuntimeService] UpdatePodSandboxResources", "PodSandboxId", req.PodSandboxId, "timeout", r.timeout)
	ctx, cancel := context.WithTimeout(ctx, r.timeout)
	defer cancel()

	resp, err := r.runtimeClient.UpdatePodSandboxResources(ctx, req)
	if err != nil {
		logger.Error(err, "UpdatePodSandboxResources from runtime service failed", "podSandboxID", req.PodSandboxId)
		return nil, err
	}
	logger.V(10).Info("[RemoteRuntimeService] UpdatePodSandboxResources Response", "podSandboxID", req.PodSandboxId)

	return resp, nil
}

// UpdateRuntimeConfig updates the config of a runtime service. The only
// update payload currently supported is the pod CIDR assigned to a node,
// and the runtime service just proxies it down to the network plugin.
func (r *remoteRuntimeService) UpdateRuntimeConfig(ctx context.Context, runtimeConfig *runtimeapi.RuntimeConfig) (err error) {
	logger := klog.FromContext(ctx)
	logger.V(10).Info("[RemoteRuntimeService] UpdateRuntimeConfig", "runtimeConfig", runtimeConfig, "timeout", r.timeout)
	ctx, cancel := context.WithTimeout(ctx, r.timeout)
	defer cancel()

	// Response doesn't contain anything of interest. This translates to an
	// Event notification to the network plugin, which can't fail, so we're
	// really looking to surface destination unreachable.
	if _, err := r.runtimeClient.UpdateRuntimeConfig(ctx, &runtimeapi.UpdateRuntimeConfigRequest{
		RuntimeConfig: runtimeConfig,
	}); err != nil {
		return err
	}
	logger.V(10).Info("[RemoteRuntimeService] UpdateRuntimeConfig Response", "runtimeConfig", runtimeConfig)

	return nil
}

// Status returns the status of the runtime.
func (r *remoteRuntimeService) Status(ctx context.Context, verbose bool) (*runtimeapi.StatusResponse, error) {
	logger := klog.FromContext(ctx)
	logger.V(10).Info("[RemoteRuntimeService] Status", "timeout", r.timeout)
	ctx, cancel := context.WithTimeout(ctx, r.timeout)
	defer cancel()

	return r.statusV1(ctx, verbose)
}

func (r *remoteRuntimeService) statusV1(ctx context.Context, verbose bool) (*runtimeapi.StatusResponse, error) {
	logger := klog.FromContext(ctx)
	resp, err := r.runtimeClient.Status(ctx, &runtimeapi.StatusRequest{
		Verbose: verbose,
	})
	if err != nil {
		logger.Error(err, "Status from runtime service failed")
		return nil, err
	}

	logger.V(10).Info("[RemoteRuntimeService] Status Response", "status", resp.Status)

	if resp.Status == nil || len(resp.Status.Conditions) < 2 {
		errorMessage := "RuntimeReady or NetworkReady condition are not set"
		err := errors.New(errorMessage)
		logger.Error(err, "Status failed")
		return nil, err
	}

	return resp, nil
}

// ContainerStats returns the stats of the container.
func (r *remoteRuntimeService) ContainerStats(ctx context.Context, containerID string) (*runtimeapi.ContainerStats, error) {
	logger := klog.FromContext(ctx)
	logger.V(10).Info("[RemoteRuntimeService] ContainerStats", "containerID", containerID, "timeout", r.timeout)
	ctx, cancel := context.WithTimeout(ctx, r.timeout)
	defer cancel()

	return r.containerStatsV1(ctx, containerID)
}

func (r *remoteRuntimeService) containerStatsV1(ctx context.Context, containerID string) (*runtimeapi.ContainerStats, error) {
	logger := klog.FromContext(ctx)
	resp, err := r.runtimeClient.ContainerStats(ctx, &runtimeapi.ContainerStatsRequest{
		ContainerId: containerID,
	})
	if err != nil {
		if r.logReduction.ShouldMessageBePrinted(err.Error(), containerID) {
			logger.Error(err, "ContainerStats from runtime service failed", "containerID", containerID)
		}
		return nil, err
	}
	r.logReduction.ClearID(containerID)
	logger.V(10).Info("[RemoteRuntimeService] ContainerStats Response", "containerID", containerID, "stats", resp.GetStats())

	return resp.GetStats(), nil
}

// ListContainerStats returns the list of ContainerStats given the filter.
func (r *remoteRuntimeService) ListContainerStats(ctx context.Context, filter *runtimeapi.ContainerStatsFilter) ([]*runtimeapi.ContainerStats, error) {
	logger := klog.FromContext(ctx)
	logger.V(10).Info("[RemoteRuntimeService] ListContainerStats", "filter", filter)
	ctx, cancel := context.WithTimeout(ctx, r.timeout)
	defer cancel()

	if r.useStreaming.Load() {
		return r.streamContainerStatsV1(ctx, filter)
	}
	return r.listContainerStatsV1(ctx, filter)
}

func (r *remoteRuntimeService) listContainerStatsV1(ctx context.Context, filter *runtimeapi.ContainerStatsFilter) ([]*runtimeapi.ContainerStats, error) {
	resp, err := r.runtimeClient.ListContainerStats(ctx, &runtimeapi.ListContainerStatsRequest{
		Filter: filter,
	})
	logger := klog.FromContext(ctx)
	if err != nil {
		logger.Error(err, "ListContainerStats with filter from runtime service failed", "filter", filter)
		return nil, err
	}
	logger.V(10).Info("[RemoteRuntimeService] ListContainerStats Response", "filter", filter, "stats", resp.GetStats())

	return resp.GetStats(), nil
}

func (r *remoteRuntimeService) streamContainerStatsV1(ctx context.Context, filter *runtimeapi.ContainerStatsFilter) ([]*runtimeapi.ContainerStats, error) {
	logger := klog.FromContext(ctx)
	stream, err := r.runtimeClient.StreamContainerStats(ctx, &runtimeapi.StreamContainerStatsRequest{
		Filter: filter,
	})
	if err != nil {
		logger.Error(err, "StreamContainerStats from runtime service failed", "filter", filter)
		return nil, err
	}

	var stats []*runtimeapi.ContainerStats
	for {
		resp, err := stream.Recv()
		if errors.Is(err, io.EOF) {
			break
		}
		if err != nil {
			// If the RPC is unimplemented, disable streaming and fall back to the unary RPC.
			// The Unimplemented status is not returned when creating the stream,
			// but when calling Recv() on the stream.
			if status.Code(err) == codes.Unimplemented {
				logger.Info("StreamContainerStats not implemented, falling back to ListContainerStats", "filter", filter)
				r.useStreaming.Store(false)
				return r.listContainerStatsV1(ctx, filter)
			}
			logger.Error(err, "StreamContainerStats recv failed", "filter", filter, "itemsReceived", len(stats))
			return nil, err
		}
		stats = append(stats, resp.ContainerStats...)
	}

	logger.V(10).Info("[RemoteRuntimeService] StreamContainerStats Response", "filter", filter, "stats", stats)

	return stats, nil
}

// PodSandboxStats returns the stats of the pod.
func (r *remoteRuntimeService) PodSandboxStats(ctx context.Context, podSandboxID string) (*runtimeapi.PodSandboxStats, error) {
	logger := klog.FromContext(ctx)
	logger.V(10).Info("[RemoteRuntimeService] PodSandboxStats", "podSandboxID", podSandboxID, "timeout", r.timeout)
	ctx, cancel := context.WithTimeout(ctx, r.timeout)
	defer cancel()

	return r.podSandboxStatsV1(ctx, podSandboxID)
}

func (r *remoteRuntimeService) podSandboxStatsV1(ctx context.Context, podSandboxID string) (*runtimeapi.PodSandboxStats, error) {
	logger := klog.FromContext(ctx)
	resp, err := r.runtimeClient.PodSandboxStats(ctx, &runtimeapi.PodSandboxStatsRequest{
		PodSandboxId: podSandboxID,
	})
	if err != nil {
		if r.logReduction.ShouldMessageBePrinted(err.Error(), podSandboxID) {
			logger.Error(err, "PodSandbox from runtime service failed", "podSandboxID", podSandboxID)
		}
		return nil, err
	}
	r.logReduction.ClearID(podSandboxID)
	logger.V(10).Info("[RemoteRuntimeService] PodSandbox Response", "podSandboxID", podSandboxID, "stats", resp.GetStats())

	return resp.GetStats(), nil
}

// ListPodSandboxStats returns the list of pod sandbox stats given the filter
func (r *remoteRuntimeService) ListPodSandboxStats(ctx context.Context, filter *runtimeapi.PodSandboxStatsFilter) ([]*runtimeapi.PodSandboxStats, error) {
	logger := klog.FromContext(ctx)
	logger.V(10).Info("[RemoteRuntimeService] ListPodSandboxStats", "filter", filter)
	// Set timeout, because runtimes are able to cache disk stats results
	ctx, cancel := context.WithTimeout(ctx, r.timeout)
	defer cancel()

	if r.useStreaming.Load() {
		return r.streamPodSandboxStatsV1(ctx, filter)
	}
	return r.listPodSandboxStatsV1(ctx, filter)
}

func (r *remoteRuntimeService) listPodSandboxStatsV1(ctx context.Context, filter *runtimeapi.PodSandboxStatsFilter) ([]*runtimeapi.PodSandboxStats, error) {
	resp, err := r.runtimeClient.ListPodSandboxStats(ctx, &runtimeapi.ListPodSandboxStatsRequest{
		Filter: filter,
	})
	logger := klog.FromContext(ctx)
	if err != nil {
		logger.Error(err, "ListPodSandboxStats with filter from runtime service failed", "filter", filter)
		return nil, err
	}
	logger.V(10).Info("[RemoteRuntimeService] ListPodSandboxStats Response", "filter", filter, "stats", resp.GetStats())

	return resp.GetStats(), nil
}

func (r *remoteRuntimeService) streamPodSandboxStatsV1(ctx context.Context, filter *runtimeapi.PodSandboxStatsFilter) ([]*runtimeapi.PodSandboxStats, error) {
	logger := klog.FromContext(ctx)
	stream, err := r.runtimeClient.StreamPodSandboxStats(ctx, &runtimeapi.StreamPodSandboxStatsRequest{
		Filter: filter,
	})
	if err != nil {
		logger.Error(err, "StreamPodSandboxStats from runtime service failed", "filter", filter)
		return nil, err
	}

	var stats []*runtimeapi.PodSandboxStats
	for {
		resp, err := stream.Recv()
		if errors.Is(err, io.EOF) {
			break
		}
		if err != nil {
			// If the RPC is unimplemented, disable streaming and fall back to the unary RPC.
			// The Unimplemented status is not returned when creating the stream,
			// but when calling Recv() on the stream.
			if status.Code(err) == codes.Unimplemented {
				logger.Info("StreamPodSandboxStats not implemented, falling back to ListPodSandboxStats", "filter", filter)
				r.useStreaming.Store(false)
				return r.listPodSandboxStatsV1(ctx, filter)
			}
			logger.Error(err, "StreamPodSandboxStats recv failed", "filter", filter, "itemsReceived", len(stats))
			return nil, err
		}
		stats = append(stats, resp.PodSandboxStats...)
	}

	logger.V(10).Info("[RemoteRuntimeService] StreamPodSandboxStats Response", "filter", filter, "stats", stats)

	return stats, nil
}

// ReopenContainerLog reopens the container log file.
func (r *remoteRuntimeService) ReopenContainerLog(ctx context.Context, containerID string) (err error) {
	logger := klog.FromContext(ctx)
	logger.V(10).Info("[RemoteRuntimeService] ReopenContainerLog", "containerID", containerID, "timeout", r.timeout)
	ctx, cancel := context.WithTimeout(ctx, r.timeout)
	defer cancel()

	if _, err := r.runtimeClient.ReopenContainerLog(ctx, &runtimeapi.ReopenContainerLogRequest{ContainerId: containerID}); err != nil {
		logger.Error(err, "ReopenContainerLog from runtime service failed", "containerID", containerID)
		return err
	}

	logger.V(10).Info("[RemoteRuntimeService] ReopenContainerLog Response", "containerID", containerID)
	return nil
}

// CheckpointContainer triggers a checkpoint of the given CheckpointContainerRequest
func (r *remoteRuntimeService) CheckpointContainer(ctx context.Context, options *runtimeapi.CheckpointContainerRequest) error {
	logger := klog.FromContext(ctx)
	logger.V(10).Info(
		"[RemoteRuntimeService] CheckpointContainer",
		"options",
		options,
	)
	if options == nil {
		return errors.New("CheckpointContainer requires non-nil CheckpointRestoreOptions parameter")
	}
	if options.Timeout < 0 {
		return errors.New("CheckpointContainer requires the timeout value to be > 0")
	}

	ctx, cancel := func(ctx context.Context) (context.Context, context.CancelFunc) {
		defaultTimeout := int64(r.timeout / time.Second)
		if options.Timeout > defaultTimeout {
			// The user requested a specific timeout, let's use that if it
			// is larger than the CRI default.
			return context.WithTimeout(ctx, time.Duration(options.Timeout)*time.Second)
		}
		// If the user requested a timeout less than the
		// CRI default, let's use the CRI default.
		options.Timeout = defaultTimeout
		return context.WithTimeout(ctx, r.timeout)
	}(ctx)
	defer cancel()

	_, err := r.runtimeClient.CheckpointContainer(
		ctx,
		options,
	)

	if err != nil {
		logger.Error(
			err,
			"CheckpointContainer from runtime service failed",
			"containerID",
			options.ContainerId,
		)
		return err
	}
	logger.V(10).Info(
		"[RemoteRuntimeService] CheckpointContainer Response",
		"containerID",
		options.ContainerId,
	)

	return nil
}

// CheckpointPod triggers a checkpoint of the given CheckpointPodRequest
func (r *remoteRuntimeService) CheckpointPod(ctx context.Context, options *runtimeapi.CheckpointPodRequest) error {
	logger := klog.FromContext(ctx)
	logger.V(10).Info(
		"[RemoteRuntimeService] CheckpointPod",
		"options",
		options,
	)
	if options == nil {
		return errors.New("CheckpointPod requires non-nil CheckpointPodRequest parameter")
	}

	// The checkpoint is bounded by the caller's context deadline (the kubelet
	// sets it from PodCheckpoint.spec.timeoutSeconds, clamped to its configured
	// ceiling) rather than a request field (KEP-5823). The default runtime
	// request timeout applies only when the caller did not set a deadline.
	if _, ok := ctx.Deadline(); !ok {
		var cancel context.CancelFunc
		ctx, cancel = context.WithTimeout(ctx, r.timeout)
		defer cancel()
	}

	_, err := r.runtimeClient.CheckpointPod(
		ctx,
		options,
	)

	if err != nil {
		logger.Error(
			err,
			"CheckpointPod from runtime service failed",
			"podSandboxID",
			options.PodSandboxId,
		)
		return err
	}
	logger.V(10).Info(
		"[RemoteRuntimeService] CheckpointPod Response",
		"podSandboxID",
		options.PodSandboxId,
	)

	return nil
}

// RestorePod restores a pod sandbox from a checkpoint
func (r *remoteRuntimeService) RestorePod(ctx context.Context, options *runtimeapi.RestorePodRequest) (*runtimeapi.RestorePodResponse, error) {
	logger := klog.FromContext(ctx)
	logger.V(10).Info(
		"[RemoteRuntimeService] RestorePod",
		"options",
		options,
	)
	if options == nil {
		return nil, errors.New("RestorePod requires non-nil RestorePodRequest parameter")
	}

	// As with CheckpointPod, the restore is bounded by the caller's context
	// deadline rather than a request field (KEP-5823). The default runtime
	// request timeout applies only when the caller did not set a deadline.
	if _, ok := ctx.Deadline(); !ok {
		var cancel context.CancelFunc
		ctx, cancel = context.WithTimeout(ctx, r.timeout)
		defer cancel()
	}

	resp, err := r.runtimeClient.RestorePod(
		ctx,
		options,
	)

	if err != nil {
		logger.Error(
			err,
			"RestorePod from runtime service failed",
			"checkpointPath",
			options.CheckpointPath,
		)
		return nil, err
	}
	if resp == nil {
		return nil, errors.New("RestorePod runtime service returned a nil response")
	}
	logger.V(10).Info(
		"[RemoteRuntimeService] RestorePod Response",
		"podSandboxID",
		resp.PodSandboxId,
		"restoredContainerCount",
		len(resp.RestoredContainers),
	)

	return resp, nil
}

func (r *remoteRuntimeService) GetContainerEvents(ctx context.Context, containerEventsCh chan *runtimeapi.ContainerEventResponse, connectionEstablishedCallback func(runtimeapi.RuntimeService_GetContainerEventsClient)) error {
	containerEventsStreamingClient, err := r.runtimeClient.GetContainerEvents(ctx, &runtimeapi.GetEventsRequest{})
	logger := klog.FromContext(ctx)
	if err != nil {
		logger.Error(err, "GetContainerEvents failed to get streaming client")
		return err
	}

	if connectionEstablishedCallback != nil {
		// The connection is successfully established and we have a streaming client ready for use.
		connectionEstablishedCallback(containerEventsStreamingClient)
	}

	for {
		resp, err := containerEventsStreamingClient.Recv()
		if errors.Is(err, io.EOF) {
			logger.Error(err, "container events stream is closed")
			return err
		}
		if err != nil {
			logger.Error(err, "failed to receive streaming container event")
			return err
		}
		if resp != nil {
			containerEventsCh <- resp
			logger.V(4).Info("container event received", "resp", resp)
		}
	}
}

// ListMetricDescriptors gets the descriptors for the metrics that will be returned in ListPodSandboxMetrics.
func (r *remoteRuntimeService) ListMetricDescriptors(ctx context.Context) ([]*runtimeapi.MetricDescriptor, error) {
	ctx, cancel := context.WithTimeout(ctx, r.timeout)
	defer cancel()

	resp, err := r.runtimeClient.ListMetricDescriptors(ctx, &runtimeapi.ListMetricDescriptorsRequest{})
	logger := klog.FromContext(ctx)
	if err != nil {
		logger.Error(err, "ListMetricDescriptors from runtime service failed")
		return nil, err
	}
	logger.V(10).Info("[RemoteRuntimeService] ListMetricDescriptors Response", "stats", resp.GetDescriptors())

	return resp.GetDescriptors(), nil
}

// ListPodSandboxMetrics retrieves the metrics for all pod sandboxes.
func (r *remoteRuntimeService) ListPodSandboxMetrics(ctx context.Context) ([]*runtimeapi.PodSandboxMetrics, error) {
	ctx, cancel := context.WithTimeout(ctx, r.timeout)
	defer cancel()

	if r.useStreaming.Load() {
		return r.streamPodSandboxMetricsV1(ctx)
	}
	return r.listPodSandboxMetricsV1(ctx)
}

func (r *remoteRuntimeService) listPodSandboxMetricsV1(ctx context.Context) ([]*runtimeapi.PodSandboxMetrics, error) {
	resp, err := r.runtimeClient.ListPodSandboxMetrics(ctx, &runtimeapi.ListPodSandboxMetricsRequest{})
	logger := klog.FromContext(ctx)
	if err != nil {
		logger.Error(err, "ListPodSandboxMetrics from runtime service failed")
		return nil, err
	}
	logger.V(10).Info("[RemoteRuntimeService] ListPodSandboxMetrics Response", "stats", resp.GetPodMetrics())

	return resp.GetPodMetrics(), nil
}

func (r *remoteRuntimeService) streamPodSandboxMetricsV1(ctx context.Context) ([]*runtimeapi.PodSandboxMetrics, error) {
	logger := klog.FromContext(ctx)
	stream, err := r.runtimeClient.StreamPodSandboxMetrics(ctx, &runtimeapi.StreamPodSandboxMetricsRequest{})
	if err != nil {
		logger.Error(err, "StreamPodSandboxMetrics from runtime service failed")
		return nil, err
	}

	var metrics []*runtimeapi.PodSandboxMetrics
	for {
		resp, err := stream.Recv()
		if errors.Is(err, io.EOF) {
			break
		}
		if err != nil {
			// If the RPC is unimplemented, disable streaming and fall back to the unary RPC.
			// The Unimplemented status is not returned when creating the stream,
			// but when calling Recv() on the stream.
			if status.Code(err) == codes.Unimplemented {
				logger.Info("StreamPodSandboxMetrics not implemented, falling back to ListPodSandboxMetrics")
				r.useStreaming.Store(false)
				return r.listPodSandboxMetricsV1(ctx)
			}
			logger.Error(err, "StreamPodSandboxMetrics recv failed", "itemsReceived", len(metrics))
			return nil, err
		}
		metrics = append(metrics, resp.PodSandboxMetrics...)
	}

	logger.V(10).Info("[RemoteRuntimeService] StreamPodSandboxMetrics Response", "metrics", metrics)

	return metrics, nil
}

// RuntimeConfig returns the configuration information of the runtime.
func (r *remoteRuntimeService) RuntimeConfig(ctx context.Context) (*runtimeapi.RuntimeConfigResponse, error) {
	ctx, cancel := context.WithTimeout(ctx, r.timeout)
	defer cancel()

	resp, err := r.runtimeClient.RuntimeConfig(ctx, &runtimeapi.RuntimeConfigRequest{})
	logger := klog.FromContext(ctx)
	if err != nil {
		logger.Error(err, "RuntimeConfig from runtime service failed")
		return nil, err
	}
	logger.V(10).Info("[RemoteRuntimeService] RuntimeConfigResponse", "linuxConfig", resp.GetLinux())

	return resp, nil
}
