diff --git a/cmd/kubemark/app/hollow_node.go b/cmd/kubemark/app/hollow_node.go index 7bf4e9ef392..9a545767f5a 100644 --- a/cmd/kubemark/app/hollow_node.go +++ b/cmd/kubemark/app/hollow_node.go @@ -24,7 +24,7 @@ import ( "github.com/spf13/cobra" "github.com/spf13/pflag" - oteltrace "go.opentelemetry.io/otel/trace" + "go.opentelemetry.io/otel/trace/noop" internalapi "k8s.io/cri-api/pkg/apis" "k8s.io/klog/v2" @@ -247,14 +247,15 @@ func run(ctx context.Context, config *hollowNodeConfig) error { return fmt.Errorf("Failed to start fake runtime, error: %w", err) } defer fakeRemoteRuntime.Stop() - runtimeService, err := remote.NewRemoteRuntimeService(endpoint, 15*time.Second, oteltrace.NewNoopTracerProvider()) + logger := klog.Background() + runtimeService, err := remote.NewRemoteRuntimeService(endpoint, 15*time.Second, noop.NewTracerProvider(), &logger) if err != nil { return fmt.Errorf("Failed to init runtime service, error: %w", err) } var imageService internalapi.ImageManagerService = fakeRemoteRuntime.ImageService if config.UseHostImageService { - imageService, err = remote.NewRemoteImageService(c.ImageServiceEndpoint, 15*time.Second, oteltrace.NewNoopTracerProvider()) + imageService, err = remote.NewRemoteImageService(c.ImageServiceEndpoint, 15*time.Second, noop.NewTracerProvider(), &logger) if err != nil { return fmt.Errorf("Failed to init image service, error: %w", err) } diff --git a/pkg/kubelet/cri/remote/remote_image.go b/pkg/kubelet/cri/remote/remote_image.go index ecd7c31a2f3..bef463c444d 100644 --- a/pkg/kubelet/cri/remote/remote_image.go +++ b/pkg/kubelet/cri/remote/remote_image.go @@ -41,11 +41,12 @@ import ( type remoteImageService struct { timeout time.Duration imageClient runtimeapi.ImageServiceClient + logger *klog.Logger } // NewRemoteImageService creates a new internalapi.ImageManagerService. -func NewRemoteImageService(endpoint string, connectionTimeout time.Duration, tp trace.TracerProvider) (internalapi.ImageManagerService, error) { - klog.V(3).InfoS("Connecting to image service", "endpoint", endpoint) +func NewRemoteImageService(endpoint string, connectionTimeout time.Duration, tp trace.TracerProvider, logger *klog.Logger) (internalapi.ImageManagerService, error) { + log(logger, 3, "Connecting to image service", "endpoint", endpoint) addr, dialer, err := util.GetAddressAndDialer(endpoint) if err != nil { return nil, err @@ -85,11 +86,14 @@ func NewRemoteImageService(endpoint string, connectionTimeout time.Duration, tp conn, err := grpc.DialContext(ctx, addr, dialOpts...) if err != nil { - klog.ErrorS(err, "Connect remote image service failed", "address", addr) + logErr(logger, err, "Connect remote image service failed", "address", addr) return nil, err } - service := &remoteImageService{timeout: connectionTimeout} + service := &remoteImageService{ + timeout: connectionTimeout, + logger: logger, + } if err := service.validateServiceConnection(ctx, conn, endpoint); err != nil { return nil, fmt.Errorf("validate service connection: %w", err) } @@ -98,17 +102,25 @@ func NewRemoteImageService(endpoint string, connectionTimeout time.Duration, tp } +func (r *remoteImageService) log(level int, msg string, keyAndValues ...any) { + log(r.logger, level, msg, keyAndValues...) +} + +func (r *remoteImageService) logErr(err error, msg string, keyAndValues ...any) { + logErr(r.logger, err, msg, keyAndValues...) +} + // validateServiceConnection tries to connect to the remote image service by // using the CRI v1 API version and fails if that's not possible. func (r *remoteImageService) validateServiceConnection(ctx context.Context, conn *grpc.ClientConn, endpoint string) error { - klog.V(4).InfoS("Validating the CRI v1 API image version") + r.log(4, "Validating the CRI v1 API image version") r.imageClient = runtimeapi.NewImageServiceClient(conn) if _, err := r.imageClient.ImageFsInfo(ctx, &runtimeapi.ImageFsInfoRequest{}); err != nil { return fmt.Errorf("validate CRI v1 image API for endpoint %q: %w", endpoint, err) } - klog.V(2).InfoS("Validated CRI v1 image API") + r.log(2, "Validated CRI v1 image API") return nil } @@ -125,7 +137,7 @@ func (r *remoteImageService) listImagesV1(ctx context.Context, filter *runtimeap Filter: filter, }) if err != nil { - klog.ErrorS(err, "ListImages with filter from image service failed", "filter", filter) + r.logErr(err, "ListImages with filter from image service failed", "filter", filter) return nil, err } @@ -146,7 +158,7 @@ func (r *remoteImageService) imageStatusV1(ctx context.Context, image *runtimeap Verbose: verbose, }) if err != nil { - klog.ErrorS(err, "Get ImageStatus from image service failed", "image", image.Image) + r.logErr(err, "Get ImageStatus from image service failed", "image", image.Image) return nil, err } @@ -154,7 +166,7 @@ func (r *remoteImageService) imageStatusV1(ctx context.Context, image *runtimeap if resp.Image.Id == "" || resp.Image.Size_ == 0 { errorMessage := fmt.Sprintf("Id or size of image %q is not set", image.Image) err := errors.New(errorMessage) - klog.ErrorS(err, "ImageStatus failed", "image", image.Image) + r.logErr(err, "ImageStatus failed", "image", image.Image) return nil, err } } @@ -177,7 +189,7 @@ func (r *remoteImageService) pullImageV1(ctx context.Context, image *runtimeapi. SandboxConfig: podSandboxConfig, }) if err != nil { - klog.ErrorS(err, "PullImage from image service failed", "image", image.Image) + r.logErr(err, "PullImage from image service failed", "image", image.Image) // We can strip the code from unknown status errors since they add no value // and will make them easier to read in the logs/events. @@ -193,7 +205,7 @@ func (r *remoteImageService) pullImageV1(ctx context.Context, image *runtimeapi. } if resp.ImageRef == "" { - klog.ErrorS(errors.New("PullImage failed"), "ImageRef of image is not set", "image", image.Image) + r.logErr(errors.New("PullImage failed"), "ImageRef of image is not set", "image", image.Image) errorMessage := fmt.Sprintf("imageRef of image %q is not set", image.Image) return "", errors.New(errorMessage) } @@ -209,7 +221,7 @@ func (r *remoteImageService) RemoveImage(ctx context.Context, image *runtimeapi. if _, err = r.imageClient.RemoveImage(ctx, &runtimeapi.RemoveImageRequest{ Image: image, }); err != nil { - klog.ErrorS(err, "RemoveImage from image service failed", "image", image.Image) + r.logErr(err, "RemoveImage from image service failed", "image", image.Image) return err } @@ -229,7 +241,7 @@ func (r *remoteImageService) ImageFsInfo(ctx context.Context) (*runtimeapi.Image func (r *remoteImageService) imageFsInfoV1(ctx context.Context) (*runtimeapi.ImageFsInfoResponse, error) { resp, err := r.imageClient.ImageFsInfo(ctx, &runtimeapi.ImageFsInfoRequest{}) if err != nil { - klog.ErrorS(err, "ImageFsInfo from image service failed") + r.logErr(err, "ImageFsInfo from image service failed") return nil, err } return resp, nil diff --git a/pkg/kubelet/cri/remote/remote_image_test.go b/pkg/kubelet/cri/remote/remote_image_test.go index 264b68e5abb..96ddba7d60b 100644 --- a/pkg/kubelet/cri/remote/remote_image_test.go +++ b/pkg/kubelet/cri/remote/remote_image_test.go @@ -26,20 +26,24 @@ import ( sdktrace "go.opentelemetry.io/otel/sdk/trace" "go.opentelemetry.io/otel/sdk/trace/tracetest" oteltrace "go.opentelemetry.io/otel/trace" + "go.opentelemetry.io/otel/trace/noop" internalapi "k8s.io/cri-api/pkg/apis" runtimeapi "k8s.io/cri-api/pkg/apis/runtime/v1" + "k8s.io/klog/v2" "k8s.io/kubernetes/pkg/kubelet/util" ) func createRemoteImageServiceWithTracerProvider(endpoint string, tp oteltrace.TracerProvider, t *testing.T) internalapi.ImageManagerService { - runtimeService, err := NewRemoteImageService(endpoint, defaultConnectionTimeout, tp) + logger := klog.Background() + runtimeService, err := NewRemoteImageService(endpoint, defaultConnectionTimeout, tp, &logger) require.NoError(t, err) return runtimeService } func createRemoteImageServiceWithoutTracerProvider(endpoint string, t *testing.T) internalapi.ImageManagerService { - runtimeService, err := NewRemoteImageService(endpoint, defaultConnectionTimeout, oteltrace.NewNoopTracerProvider()) + logger := klog.Background() + runtimeService, err := NewRemoteImageService(endpoint, defaultConnectionTimeout, noop.NewTracerProvider(), &logger) require.NoError(t, err) return runtimeService diff --git a/pkg/kubelet/cri/remote/remote_runtime.go b/pkg/kubelet/cri/remote/remote_runtime.go index a56a7bd1edc..d81ccd33133 100644 --- a/pkg/kubelet/cri/remote/remote_runtime.go +++ b/pkg/kubelet/cri/remote/remote_runtime.go @@ -47,6 +47,7 @@ type remoteRuntimeService struct { runtimeClient runtimeapi.RuntimeServiceClient // Cache last per-container error message to reduce log spam logReduction *logreduction.LogReduction + logger *klog.Logger } const ( @@ -77,8 +78,8 @@ const ( ) // NewRemoteRuntimeService creates a new internalapi.RuntimeService. -func NewRemoteRuntimeService(endpoint string, connectionTimeout time.Duration, tp trace.TracerProvider) (internalapi.RuntimeService, error) { - klog.V(3).InfoS("Connecting to runtime service", "endpoint", endpoint) +func NewRemoteRuntimeService(endpoint string, connectionTimeout time.Duration, tp trace.TracerProvider, logger *klog.Logger) (internalapi.RuntimeService, error) { + log(logger, 3, "Connecting to runtime service", "endpoint", endpoint) addr, dialer, err := util.GetAddressAndDialer(endpoint) if err != nil { return nil, err @@ -117,13 +118,14 @@ func NewRemoteRuntimeService(endpoint string, connectionTimeout time.Duration, t conn, err := grpc.DialContext(ctx, addr, dialOpts...) if err != nil { - klog.ErrorS(err, "Connect remote runtime failed", "address", addr) + logErr(logger, err, "Connect remote runtime failed", "address", addr) return nil, err } service := &remoteRuntimeService{ timeout: connectionTimeout, logReduction: logreduction.NewLogReduction(identicalErrorDelay), + logger: logger, } if err := service.validateServiceConnection(ctx, conn, endpoint); err != nil { @@ -133,23 +135,31 @@ func NewRemoteRuntimeService(endpoint string, connectionTimeout time.Duration, t return service, nil } +func (r *remoteRuntimeService) log(level int, msg string, keyAndValues ...any) { + log(r.logger, level, msg, keyAndValues...) +} + +func (r *remoteRuntimeService) logErr(err error, msg string, keyAndValues ...any) { + logErr(r.logger, err, msg, keyAndValues...) +} + // 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 { - klog.V(4).InfoS("Validating the CRI v1 API runtime version") + r.log(4, "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) } - klog.V(2).InfoS("Validated CRI v1 runtime API") + r.log(2, "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) { - klog.V(10).InfoS("[RemoteRuntimeService] Version", "apiVersion", apiVersion, "timeout", r.timeout) + r.log(10, "[RemoteRuntimeService] Version", "apiVersion", apiVersion, "timeout", r.timeout) ctx, cancel := context.WithTimeout(ctx, r.timeout) defer cancel() @@ -162,11 +172,11 @@ func (r *remoteRuntimeService) versionV1(ctx context.Context, apiVersion string) Version: apiVersion, }) if err != nil { - klog.ErrorS(err, "Version from runtime service failed") + r.logErr(err, "Version from runtime service failed") return nil, err } - klog.V(10).InfoS("[RemoteRuntimeService] Version Response", "apiVersion", typedVersion) + r.log(10, "[RemoteRuntimeService] Version Response", "apiVersion", typedVersion) if typedVersion.Version == "" || typedVersion.RuntimeName == "" || typedVersion.RuntimeApiVersion == "" || typedVersion.RuntimeVersion == "" { return nil, fmt.Errorf("not all fields are set in VersionResponse (%q)", *typedVersion) @@ -182,7 +192,7 @@ func (r *remoteRuntimeService) RunPodSandbox(ctx context.Context, config *runtim // TODO: Make the pod sandbox timeout configurable. timeout := r.timeout * 2 - klog.V(10).InfoS("[RemoteRuntimeService] RunPodSandbox", "config", config, "runtimeHandler", runtimeHandler, "timeout", timeout) + r.log(10, "[RemoteRuntimeService] RunPodSandbox", "config", config, "runtimeHandler", runtimeHandler, "timeout", timeout) ctx, cancel := context.WithTimeout(ctx, timeout) defer cancel() @@ -193,7 +203,7 @@ func (r *remoteRuntimeService) RunPodSandbox(ctx context.Context, config *runtim }) if err != nil { - klog.ErrorS(err, "RunPodSandbox from runtime service failed") + r.logErr(err, "RunPodSandbox from runtime service failed") return "", err } @@ -202,11 +212,11 @@ func (r *remoteRuntimeService) RunPodSandbox(ctx context.Context, config *runtim if podSandboxID == "" { errorMessage := fmt.Sprintf("PodSandboxId is not set for sandbox %q", config.Metadata) err := errors.New(errorMessage) - klog.ErrorS(err, "RunPodSandbox failed") + r.logErr(err, "RunPodSandbox failed") return "", err } - klog.V(10).InfoS("[RemoteRuntimeService] RunPodSandbox Response", "podSandboxID", podSandboxID) + r.log(10, "[RemoteRuntimeService] RunPodSandbox Response", "podSandboxID", podSandboxID) return podSandboxID, nil } @@ -214,7 +224,7 @@ func (r *remoteRuntimeService) RunPodSandbox(ctx context.Context, config *runtim // 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) { - klog.V(10).InfoS("[RemoteRuntimeService] StopPodSandbox", "podSandboxID", podSandBoxID, "timeout", r.timeout) + r.log(10, "[RemoteRuntimeService] StopPodSandbox", "podSandboxID", podSandBoxID, "timeout", r.timeout) ctx, cancel := context.WithTimeout(ctx, r.timeout) defer cancel() @@ -222,11 +232,11 @@ func (r *remoteRuntimeService) StopPodSandbox(ctx context.Context, podSandBoxID if _, err := r.runtimeClient.StopPodSandbox(ctx, &runtimeapi.StopPodSandboxRequest{ PodSandboxId: podSandBoxID, }); err != nil { - klog.ErrorS(err, "StopPodSandbox from runtime service failed", "podSandboxID", podSandBoxID) + r.logErr(err, "StopPodSandbox from runtime service failed", "podSandboxID", podSandBoxID) return err } - klog.V(10).InfoS("[RemoteRuntimeService] StopPodSandbox Response", "podSandboxID", podSandBoxID) + r.log(10, "[RemoteRuntimeService] StopPodSandbox Response", "podSandboxID", podSandBoxID) return nil } @@ -234,25 +244,25 @@ func (r *remoteRuntimeService) StopPodSandbox(ctx context.Context, podSandBoxID // 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) { - klog.V(10).InfoS("[RemoteRuntimeService] RemovePodSandbox", "podSandboxID", podSandBoxID, "timeout", r.timeout) + r.log(10, "[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 { - klog.ErrorS(err, "RemovePodSandbox from runtime service failed", "podSandboxID", podSandBoxID) + r.logErr(err, "RemovePodSandbox from runtime service failed", "podSandboxID", podSandBoxID) return err } - klog.V(10).InfoS("[RemoteRuntimeService] RemovePodSandbox Response", "podSandboxID", podSandBoxID) + r.log(10, "[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) { - klog.V(10).InfoS("[RemoteRuntimeService] PodSandboxStatus", "podSandboxID", podSandBoxID, "timeout", r.timeout) + r.log(10, "[RemoteRuntimeService] PodSandboxStatus", "podSandboxID", podSandBoxID, "timeout", r.timeout) ctx, cancel := context.WithTimeout(ctx, r.timeout) defer cancel() @@ -268,7 +278,7 @@ func (r *remoteRuntimeService) podSandboxStatusV1(ctx context.Context, podSandBo return nil, err } - klog.V(10).InfoS("[RemoteRuntimeService] PodSandboxStatus Response", "podSandboxID", podSandBoxID, "status", resp.Status) + r.log(10, "[RemoteRuntimeService] PodSandboxStatus Response", "podSandboxID", podSandBoxID, "status", resp.Status) status := resp.Status if resp.Status != nil { @@ -282,7 +292,7 @@ func (r *remoteRuntimeService) podSandboxStatusV1(ctx context.Context, podSandBo // ListPodSandbox returns a list of PodSandboxes. func (r *remoteRuntimeService) ListPodSandbox(ctx context.Context, filter *runtimeapi.PodSandboxFilter) ([]*runtimeapi.PodSandbox, error) { - klog.V(10).InfoS("[RemoteRuntimeService] ListPodSandbox", "filter", filter, "timeout", r.timeout) + r.log(10, "[RemoteRuntimeService] ListPodSandbox", "filter", filter, "timeout", r.timeout) ctx, cancel := context.WithTimeout(ctx, r.timeout) defer cancel() @@ -294,18 +304,18 @@ func (r *remoteRuntimeService) listPodSandboxV1(ctx context.Context, filter *run Filter: filter, }) if err != nil { - klog.ErrorS(err, "ListPodSandbox with filter from runtime service failed", "filter", filter) + r.logErr(err, "ListPodSandbox with filter from runtime service failed", "filter", filter) return nil, err } - klog.V(10).InfoS("[RemoteRuntimeService] ListPodSandbox Response", "filter", filter, "items", resp.Items) + r.log(10, "[RemoteRuntimeService] ListPodSandbox Response", "filter", filter, "items", resp.Items) return resp.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) { - klog.V(10).InfoS("[RemoteRuntimeService] CreateContainer", "podSandboxID", podSandBoxID, "timeout", r.timeout) + r.log(10, "[RemoteRuntimeService] CreateContainer", "podSandboxID", podSandBoxID, "timeout", r.timeout) ctx, cancel := context.WithTimeout(ctx, r.timeout) defer cancel() @@ -319,15 +329,15 @@ func (r *remoteRuntimeService) createContainerV1(ctx context.Context, podSandBox SandboxConfig: sandboxConfig, }) if err != nil { - klog.ErrorS(err, "CreateContainer in sandbox from runtime service failed", "podSandboxID", podSandBoxID) + r.logErr(err, "CreateContainer in sandbox from runtime service failed", "podSandboxID", podSandBoxID) return "", err } - klog.V(10).InfoS("[RemoteRuntimeService] CreateContainer", "podSandboxID", podSandBoxID, "containerID", resp.ContainerId) + r.log(10, "[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) - klog.ErrorS(err, "CreateContainer failed") + r.logErr(err, "CreateContainer failed") return "", err } @@ -336,24 +346,24 @@ func (r *remoteRuntimeService) createContainerV1(ctx context.Context, podSandBox // StartContainer starts the container. func (r *remoteRuntimeService) StartContainer(ctx context.Context, containerID string) (err error) { - klog.V(10).InfoS("[RemoteRuntimeService] StartContainer", "containerID", containerID, "timeout", r.timeout) + r.log(10, "[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 { - klog.ErrorS(err, "StartContainer from runtime service failed", "containerID", containerID) + r.logErr(err, "StartContainer from runtime service failed", "containerID", containerID) return err } - klog.V(10).InfoS("[RemoteRuntimeService] StartContainer Response", "containerID", containerID) + r.log(10, "[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) { - klog.V(10).InfoS("[RemoteRuntimeService] StopContainer", "containerID", containerID, "timeout", timeout) + r.log(10, "[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 @@ -366,10 +376,10 @@ func (r *remoteRuntimeService) StopContainer(ctx context.Context, containerID st ContainerId: containerID, Timeout: timeout, }); err != nil { - klog.ErrorS(err, "StopContainer from runtime service failed", "containerID", containerID) + r.logErr(err, "StopContainer from runtime service failed", "containerID", containerID) return err } - klog.V(10).InfoS("[RemoteRuntimeService] StopContainer Response", "containerID", containerID) + r.log(10, "[RemoteRuntimeService] StopContainer Response", "containerID", containerID) return nil } @@ -377,7 +387,7 @@ func (r *remoteRuntimeService) StopContainer(ctx context.Context, containerID st // 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) { - klog.V(10).InfoS("[RemoteRuntimeService] RemoveContainer", "containerID", containerID, "timeout", r.timeout) + r.log(10, "[RemoteRuntimeService] RemoveContainer", "containerID", containerID, "timeout", r.timeout) ctx, cancel := context.WithTimeout(ctx, r.timeout) defer cancel() @@ -385,17 +395,17 @@ func (r *remoteRuntimeService) RemoveContainer(ctx context.Context, containerID if _, err := r.runtimeClient.RemoveContainer(ctx, &runtimeapi.RemoveContainerRequest{ ContainerId: containerID, }); err != nil { - klog.ErrorS(err, "RemoveContainer from runtime service failed", "containerID", containerID) + r.logErr(err, "RemoveContainer from runtime service failed", "containerID", containerID) return err } - klog.V(10).InfoS("[RemoteRuntimeService] RemoveContainer Response", "containerID", containerID) + r.log(10, "[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) { - klog.V(10).InfoS("[RemoteRuntimeService] ListContainers", "filter", filter, "timeout", r.timeout) + r.log(10, "[RemoteRuntimeService] ListContainers", "filter", filter, "timeout", r.timeout) ctx, cancel := context.WithTimeout(ctx, r.timeout) defer cancel() @@ -407,17 +417,17 @@ func (r *remoteRuntimeService) listContainersV1(ctx context.Context, filter *run Filter: filter, }) if err != nil { - klog.ErrorS(err, "ListContainers with filter from runtime service failed", "filter", filter) + r.logErr(err, "ListContainers with filter from runtime service failed", "filter", filter) return nil, err } - klog.V(10).InfoS("[RemoteRuntimeService] ListContainers Response", "filter", filter, "containers", resp.Containers) + r.log(10, "[RemoteRuntimeService] ListContainers Response", "filter", filter, "containers", resp.Containers) return resp.Containers, nil } // ContainerStatus returns the container status. func (r *remoteRuntimeService) ContainerStatus(ctx context.Context, containerID string, verbose bool) (*runtimeapi.ContainerStatusResponse, error) { - klog.V(10).InfoS("[RemoteRuntimeService] ContainerStatus", "containerID", containerID, "timeout", r.timeout) + r.log(10, "[RemoteRuntimeService] ContainerStatus", "containerID", containerID, "timeout", r.timeout) ctx, cancel := context.WithTimeout(ctx, r.timeout) defer cancel() @@ -432,17 +442,17 @@ func (r *remoteRuntimeService) containerStatusV1(ctx context.Context, containerI if err != nil { // Don't spam the log with endless messages about the same failure. if r.logReduction.ShouldMessageBePrinted(err.Error(), containerID) { - klog.ErrorS(err, "ContainerStatus from runtime service failed", "containerID", containerID) + r.logErr(err, "ContainerStatus from runtime service failed", "containerID", containerID) } return nil, err } r.logReduction.ClearID(containerID) - klog.V(10).InfoS("[RemoteRuntimeService] ContainerStatus Response", "containerID", containerID, "status", resp.Status) + r.log(10, "[RemoteRuntimeService] ContainerStatus Response", "containerID", containerID, "status", resp.Status) status := resp.Status if resp.Status != nil { if err := verifyContainerStatus(status); err != nil { - klog.ErrorS(err, "verify ContainerStatus failed", "containerID", containerID) + r.logErr(err, "verify ContainerStatus failed", "containerID", containerID) return nil, err } } @@ -452,7 +462,7 @@ func (r *remoteRuntimeService) containerStatusV1(ctx context.Context, containerI // UpdateContainerResources updates a containers resource config func (r *remoteRuntimeService) UpdateContainerResources(ctx context.Context, containerID string, resources *runtimeapi.ContainerResources) (err error) { - klog.V(10).InfoS("[RemoteRuntimeService] UpdateContainerResources", "containerID", containerID, "timeout", r.timeout) + r.log(10, "[RemoteRuntimeService] UpdateContainerResources", "containerID", containerID, "timeout", r.timeout) ctx, cancel := context.WithTimeout(ctx, r.timeout) defer cancel() @@ -461,10 +471,10 @@ func (r *remoteRuntimeService) UpdateContainerResources(ctx context.Context, con Linux: resources.GetLinux(), Windows: resources.GetWindows(), }); err != nil { - klog.ErrorS(err, "UpdateContainerResources from runtime service failed", "containerID", containerID) + r.logErr(err, "UpdateContainerResources from runtime service failed", "containerID", containerID) return err } - klog.V(10).InfoS("[RemoteRuntimeService] UpdateContainerResources Response", "containerID", containerID) + r.log(10, "[RemoteRuntimeService] UpdateContainerResources Response", "containerID", containerID) return nil } @@ -472,7 +482,7 @@ func (r *remoteRuntimeService) UpdateContainerResources(ctx context.Context, con // 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) { - klog.V(10).InfoS("[RemoteRuntimeService] ExecSync", "containerID", containerID, "timeout", timeout) + r.log(10, "[RemoteRuntimeService] ExecSync", "containerID", containerID, "timeout", timeout) // Do not set timeout when timeout is 0. var cancel context.CancelFunc if timeout != 0 { @@ -496,7 +506,7 @@ func (r *remoteRuntimeService) execSyncV1(ctx context.Context, containerID strin } resp, err := r.runtimeClient.ExecSync(ctx, req) if err != nil { - klog.ErrorS(err, "ExecSync cmd from runtime service failed", "containerID", containerID, "cmd", cmd) + r.logErr(err, "ExecSync cmd from runtime service failed", "containerID", containerID, "cmd", cmd) // interpret DeadlineExceeded gRPC errors as timedout errors if status.Code(err) == codes.DeadlineExceeded { @@ -506,7 +516,7 @@ func (r *remoteRuntimeService) execSyncV1(ctx context.Context, containerID strin return nil, nil, err } - klog.V(10).InfoS("[RemoteRuntimeService] ExecSync Response", "containerID", containerID, "exitCode", resp.ExitCode) + r.log(10, "[RemoteRuntimeService] ExecSync Response", "containerID", containerID, "exitCode", resp.ExitCode) err = nil if resp.ExitCode != 0 { err = utilexec.CodeExitError{ @@ -520,7 +530,7 @@ func (r *remoteRuntimeService) execSyncV1(ctx context.Context, containerID strin // 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) { - klog.V(10).InfoS("[RemoteRuntimeService] Exec", "timeout", r.timeout) + r.log(10, "[RemoteRuntimeService] Exec", "timeout", r.timeout) ctx, cancel := context.WithTimeout(ctx, r.timeout) defer cancel() @@ -530,15 +540,15 @@ func (r *remoteRuntimeService) Exec(ctx context.Context, req *runtimeapi.ExecReq func (r *remoteRuntimeService) execV1(ctx context.Context, req *runtimeapi.ExecRequest) (*runtimeapi.ExecResponse, error) { resp, err := r.runtimeClient.Exec(ctx, req) if err != nil { - klog.ErrorS(err, "Exec cmd from runtime service failed", "containerID", req.ContainerId, "cmd", req.Cmd) + r.logErr(err, "Exec cmd from runtime service failed", "containerID", req.ContainerId, "cmd", req.Cmd) return nil, err } - klog.V(10).InfoS("[RemoteRuntimeService] Exec Response") + r.log(10, "[RemoteRuntimeService] Exec Response") if resp.Url == "" { errorMessage := "URL is not set" err := errors.New(errorMessage) - klog.ErrorS(err, "Exec failed") + r.logErr(err, "Exec failed") return nil, err } @@ -547,7 +557,7 @@ func (r *remoteRuntimeService) execV1(ctx context.Context, req *runtimeapi.ExecR // 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) { - klog.V(10).InfoS("[RemoteRuntimeService] Attach", "containerID", req.ContainerId, "timeout", r.timeout) + r.log(10, "[RemoteRuntimeService] Attach", "containerID", req.ContainerId, "timeout", r.timeout) ctx, cancel := context.WithTimeout(ctx, r.timeout) defer cancel() @@ -557,15 +567,15 @@ func (r *remoteRuntimeService) Attach(ctx context.Context, req *runtimeapi.Attac func (r *remoteRuntimeService) attachV1(ctx context.Context, req *runtimeapi.AttachRequest) (*runtimeapi.AttachResponse, error) { resp, err := r.runtimeClient.Attach(ctx, req) if err != nil { - klog.ErrorS(err, "Attach container from runtime service failed", "containerID", req.ContainerId) + r.logErr(err, "Attach container from runtime service failed", "containerID", req.ContainerId) return nil, err } - klog.V(10).InfoS("[RemoteRuntimeService] Attach Response", "containerID", req.ContainerId) + r.log(10, "[RemoteRuntimeService] Attach Response", "containerID", req.ContainerId) if resp.Url == "" { errorMessage := "URL is not set" err := errors.New(errorMessage) - klog.ErrorS(err, "Attach failed") + r.logErr(err, "Attach failed") return nil, err } return resp, nil @@ -573,7 +583,7 @@ func (r *remoteRuntimeService) attachV1(ctx context.Context, req *runtimeapi.Att // 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) { - klog.V(10).InfoS("[RemoteRuntimeService] PortForward", "podSandboxID", req.PodSandboxId, "port", req.Port, "timeout", r.timeout) + r.log(10, "[RemoteRuntimeService] PortForward", "podSandboxID", req.PodSandboxId, "port", req.Port, "timeout", r.timeout) ctx, cancel := context.WithTimeout(ctx, r.timeout) defer cancel() @@ -583,15 +593,15 @@ func (r *remoteRuntimeService) PortForward(ctx context.Context, req *runtimeapi. func (r *remoteRuntimeService) portForwardV1(ctx context.Context, req *runtimeapi.PortForwardRequest) (*runtimeapi.PortForwardResponse, error) { resp, err := r.runtimeClient.PortForward(ctx, req) if err != nil { - klog.ErrorS(err, "PortForward from runtime service failed", "podSandboxID", req.PodSandboxId) + r.logErr(err, "PortForward from runtime service failed", "podSandboxID", req.PodSandboxId) return nil, err } - klog.V(10).InfoS("[RemoteRuntimeService] PortForward Response", "podSandboxID", req.PodSandboxId) + r.log(10, "[RemoteRuntimeService] PortForward Response", "podSandboxID", req.PodSandboxId) if resp.Url == "" { errorMessage := "URL is not set" err := errors.New(errorMessage) - klog.ErrorS(err, "PortForward failed") + r.logErr(err, "PortForward failed") return nil, err } @@ -602,7 +612,7 @@ func (r *remoteRuntimeService) portForwardV1(ctx context.Context, req *runtimeap // 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) { - klog.V(10).InfoS("[RemoteRuntimeService] UpdateRuntimeConfig", "runtimeConfig", runtimeConfig, "timeout", r.timeout) + r.log(10, "[RemoteRuntimeService] UpdateRuntimeConfig", "runtimeConfig", runtimeConfig, "timeout", r.timeout) ctx, cancel := context.WithTimeout(ctx, r.timeout) defer cancel() @@ -614,14 +624,14 @@ func (r *remoteRuntimeService) UpdateRuntimeConfig(ctx context.Context, runtimeC }); err != nil { return err } - klog.V(10).InfoS("[RemoteRuntimeService] UpdateRuntimeConfig Response", "runtimeConfig", runtimeConfig) + r.log(10, "[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) { - klog.V(10).InfoS("[RemoteRuntimeService] Status", "timeout", r.timeout) + r.log(10, "[RemoteRuntimeService] Status", "timeout", r.timeout) ctx, cancel := context.WithTimeout(ctx, r.timeout) defer cancel() @@ -633,16 +643,16 @@ func (r *remoteRuntimeService) statusV1(ctx context.Context, verbose bool) (*run Verbose: verbose, }) if err != nil { - klog.ErrorS(err, "Status from runtime service failed") + r.logErr(err, "Status from runtime service failed") return nil, err } - klog.V(10).InfoS("[RemoteRuntimeService] Status Response", "status", resp.Status) + r.log(10, "[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) - klog.ErrorS(err, "Status failed") + r.logErr(err, "Status failed") return nil, err } @@ -651,7 +661,7 @@ func (r *remoteRuntimeService) statusV1(ctx context.Context, verbose bool) (*run // ContainerStats returns the stats of the container. func (r *remoteRuntimeService) ContainerStats(ctx context.Context, containerID string) (*runtimeapi.ContainerStats, error) { - klog.V(10).InfoS("[RemoteRuntimeService] ContainerStats", "containerID", containerID, "timeout", r.timeout) + r.log(10, "[RemoteRuntimeService] ContainerStats", "containerID", containerID, "timeout", r.timeout) ctx, cancel := context.WithTimeout(ctx, r.timeout) defer cancel() @@ -664,19 +674,19 @@ func (r *remoteRuntimeService) containerStatsV1(ctx context.Context, containerID }) if err != nil { if r.logReduction.ShouldMessageBePrinted(err.Error(), containerID) { - klog.ErrorS(err, "ContainerStats from runtime service failed", "containerID", containerID) + r.logErr(err, "ContainerStats from runtime service failed", "containerID", containerID) } return nil, err } r.logReduction.ClearID(containerID) - klog.V(10).InfoS("[RemoteRuntimeService] ContainerStats Response", "containerID", containerID, "stats", resp.GetStats()) + r.log(10, "[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) { - klog.V(10).InfoS("[RemoteRuntimeService] ListContainerStats", "filter", filter) + r.log(10, "[RemoteRuntimeService] ListContainerStats", "filter", filter) ctx, cancel := context.WithTimeout(ctx, r.timeout) defer cancel() @@ -688,17 +698,17 @@ func (r *remoteRuntimeService) listContainerStatsV1(ctx context.Context, filter Filter: filter, }) if err != nil { - klog.ErrorS(err, "ListContainerStats with filter from runtime service failed", "filter", filter) + r.logErr(err, "ListContainerStats with filter from runtime service failed", "filter", filter) return nil, err } - klog.V(10).InfoS("[RemoteRuntimeService] ListContainerStats Response", "filter", filter, "stats", resp.GetStats()) + r.log(10, "[RemoteRuntimeService] ListContainerStats Response", "filter", filter, "stats", resp.GetStats()) return resp.GetStats(), nil } // PodSandboxStats returns the stats of the pod. func (r *remoteRuntimeService) PodSandboxStats(ctx context.Context, podSandboxID string) (*runtimeapi.PodSandboxStats, error) { - klog.V(10).InfoS("[RemoteRuntimeService] PodSandboxStats", "podSandboxID", podSandboxID, "timeout", r.timeout) + r.log(10, "[RemoteRuntimeService] PodSandboxStats", "podSandboxID", podSandboxID, "timeout", r.timeout) ctx, cancel := context.WithTimeout(ctx, r.timeout) defer cancel() @@ -711,19 +721,19 @@ func (r *remoteRuntimeService) podSandboxStatsV1(ctx context.Context, podSandbox }) if err != nil { if r.logReduction.ShouldMessageBePrinted(err.Error(), podSandboxID) { - klog.ErrorS(err, "PodSandbox from runtime service failed", "podSandboxID", podSandboxID) + r.logErr(err, "PodSandbox from runtime service failed", "podSandboxID", podSandboxID) } return nil, err } r.logReduction.ClearID(podSandboxID) - klog.V(10).InfoS("[RemoteRuntimeService] PodSandbox Response", "podSandboxID", podSandboxID, "stats", resp.GetStats()) + r.log(10, "[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) { - klog.V(10).InfoS("[RemoteRuntimeService] ListPodSandboxStats", "filter", filter) + r.log(10, "[RemoteRuntimeService] ListPodSandboxStats", "filter", filter) // Set timeout, because runtimes are able to cache disk stats results ctx, cancel := context.WithTimeout(ctx, r.timeout) defer cancel() @@ -736,32 +746,32 @@ func (r *remoteRuntimeService) listPodSandboxStatsV1(ctx context.Context, filter Filter: filter, }) if err != nil { - klog.ErrorS(err, "ListPodSandboxStats with filter from runtime service failed", "filter", filter) + r.logErr(err, "ListPodSandboxStats with filter from runtime service failed", "filter", filter) return nil, err } - klog.V(10).InfoS("[RemoteRuntimeService] ListPodSandboxStats Response", "filter", filter, "stats", resp.GetStats()) + r.log(10, "[RemoteRuntimeService] ListPodSandboxStats Response", "filter", filter, "stats", resp.GetStats()) return resp.GetStats(), nil } // ReopenContainerLog reopens the container log file. func (r *remoteRuntimeService) ReopenContainerLog(ctx context.Context, containerID string) (err error) { - klog.V(10).InfoS("[RemoteRuntimeService] ReopenContainerLog", "containerID", containerID, "timeout", r.timeout) + r.log(10, "[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 { - klog.ErrorS(err, "ReopenContainerLog from runtime service failed", "containerID", containerID) + r.logErr(err, "ReopenContainerLog from runtime service failed", "containerID", containerID) return err } - klog.V(10).InfoS("[RemoteRuntimeService] ReopenContainerLog Response", "containerID", containerID) + r.log(10, "[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 { - klog.V(10).InfoS( + r.log(10, "[RemoteRuntimeService] CheckpointContainer", "options", options, @@ -793,7 +803,7 @@ func (r *remoteRuntimeService) CheckpointContainer(ctx context.Context, options ) if err != nil { - klog.ErrorS( + r.logErr( err, "CheckpointContainer from runtime service failed", "containerID", @@ -801,7 +811,7 @@ func (r *remoteRuntimeService) CheckpointContainer(ctx context.Context, options ) return err } - klog.V(10).InfoS( + r.log(10, "[RemoteRuntimeService] CheckpointContainer Response", "containerID", options.ContainerId, @@ -813,7 +823,7 @@ func (r *remoteRuntimeService) CheckpointContainer(ctx context.Context, options func (r *remoteRuntimeService) GetContainerEvents(containerEventsCh chan *runtimeapi.ContainerEventResponse, connectionEstablishedCallback func(runtimeapi.RuntimeService_GetContainerEventsClient)) error { containerEventsStreamingClient, err := r.runtimeClient.GetContainerEvents(context.Background(), &runtimeapi.GetEventsRequest{}) if err != nil { - klog.ErrorS(err, "GetContainerEvents failed to get streaming client") + r.logErr(err, "GetContainerEvents failed to get streaming client") return err } @@ -825,16 +835,16 @@ func (r *remoteRuntimeService) GetContainerEvents(containerEventsCh chan *runtim for { resp, err := containerEventsStreamingClient.Recv() if err == io.EOF { - klog.ErrorS(err, "container events stream is closed") + r.logErr(err, "container events stream is closed") return err } if err != nil { - klog.ErrorS(err, "failed to receive streaming container event") + r.logErr(err, "failed to receive streaming container event") return err } if resp != nil { containerEventsCh <- resp - klog.V(4).InfoS("container event received", "resp", resp) + r.log(4, "container event received", "resp", resp) } } } @@ -846,10 +856,10 @@ func (r *remoteRuntimeService) ListMetricDescriptors(ctx context.Context) ([]*ru resp, err := r.runtimeClient.ListMetricDescriptors(ctx, &runtimeapi.ListMetricDescriptorsRequest{}) if err != nil { - klog.ErrorS(err, "ListMetricDescriptors from runtime service failed") + r.logErr(err, "ListMetricDescriptors from runtime service failed") return nil, err } - klog.V(10).InfoS("[RemoteRuntimeService] ListMetricDescriptors Response", "stats", resp.GetDescriptors()) + r.log(10, "[RemoteRuntimeService] ListMetricDescriptors Response", "stats", resp.GetDescriptors()) return resp.GetDescriptors(), nil } @@ -861,10 +871,10 @@ func (r *remoteRuntimeService) ListPodSandboxMetrics(ctx context.Context) ([]*ru resp, err := r.runtimeClient.ListPodSandboxMetrics(ctx, &runtimeapi.ListPodSandboxMetricsRequest{}) if err != nil { - klog.ErrorS(err, "ListPodSandboxMetrics from runtime service failed") + r.logErr(err, "ListPodSandboxMetrics from runtime service failed") return nil, err } - klog.V(10).InfoS("[RemoteRuntimeService] ListPodSandboxMetrics Response", "stats", resp.GetPodMetrics()) + r.log(10, "[RemoteRuntimeService] ListPodSandboxMetrics Response", "stats", resp.GetPodMetrics()) return resp.GetPodMetrics(), nil } @@ -876,10 +886,10 @@ func (r *remoteRuntimeService) RuntimeConfig(ctx context.Context) (*runtimeapi.R resp, err := r.runtimeClient.RuntimeConfig(ctx, &runtimeapi.RuntimeConfigRequest{}) if err != nil { - klog.ErrorS(err, "RuntimeConfig from runtime service failed") + r.logErr(err, "RuntimeConfig from runtime service failed") return nil, err } - klog.V(10).InfoS("[RemoteRuntimeService] RuntimeConfigResponse", "linuxConfig", resp.GetLinux()) + r.log(10, "[RemoteRuntimeService] RuntimeConfigResponse", "linuxConfig", resp.GetLinux()) return resp, nil } diff --git a/pkg/kubelet/cri/remote/remote_runtime_test.go b/pkg/kubelet/cri/remote/remote_runtime_test.go index 059c5fa5b6d..69173fb5703 100644 --- a/pkg/kubelet/cri/remote/remote_runtime_test.go +++ b/pkg/kubelet/cri/remote/remote_runtime_test.go @@ -25,11 +25,13 @@ import ( sdktrace "go.opentelemetry.io/otel/sdk/trace" "go.opentelemetry.io/otel/sdk/trace/tracetest" oteltrace "go.opentelemetry.io/otel/trace" + "go.opentelemetry.io/otel/trace/noop" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" internalapi "k8s.io/cri-api/pkg/apis" apitest "k8s.io/cri-api/pkg/apis/testing" + "k8s.io/klog/v2" fakeremote "k8s.io/kubernetes/pkg/kubelet/cri/remote/fake" "k8s.io/kubernetes/pkg/kubelet/util" ) @@ -52,7 +54,8 @@ func createAndStartFakeRemoteRuntime(t *testing.T) (*fakeremote.RemoteRuntime, s } func createRemoteRuntimeService(endpoint string, t *testing.T) internalapi.RuntimeService { - runtimeService, err := NewRemoteRuntimeService(endpoint, defaultConnectionTimeout, oteltrace.NewNoopTracerProvider()) + logger := klog.Background() + runtimeService, err := NewRemoteRuntimeService(endpoint, defaultConnectionTimeout, noop.NewTracerProvider(), &logger) require.NoError(t, err) @@ -60,7 +63,8 @@ func createRemoteRuntimeService(endpoint string, t *testing.T) internalapi.Runti } func createRemoteRuntimeServiceWithTracerProvider(endpoint string, tp oteltrace.TracerProvider, t *testing.T) internalapi.RuntimeService { - runtimeService, err := NewRemoteRuntimeService(endpoint, defaultConnectionTimeout, tp) + logger := klog.Background() + runtimeService, err := NewRemoteRuntimeService(endpoint, defaultConnectionTimeout, tp, &logger) require.NoError(t, err) return runtimeService diff --git a/pkg/kubelet/cri/remote/utils.go b/pkg/kubelet/cri/remote/utils.go index ac3f35e45ad..ad7dfddbe01 100644 --- a/pkg/kubelet/cri/remote/utils.go +++ b/pkg/kubelet/cri/remote/utils.go @@ -20,6 +20,7 @@ import ( "fmt" runtimeapi "k8s.io/cri-api/pkg/apis/runtime/v1" + "k8s.io/klog/v2" ) // maxMsgSize use 16MB as the default message size limit. @@ -77,3 +78,17 @@ func verifyContainerStatus(status *runtimeapi.ContainerStatus) error { return nil } + +func log(logger *klog.Logger, level int, msg string, keyAndValues ...any) { + if logger == nil { + return + } + logger.V(level).Info(msg, keyAndValues...) +} + +func logErr(logger *klog.Logger, err error, msg string, keyAndValues ...any) { + if logger == nil { + return + } + logger.Error(err, msg, keyAndValues...) +} diff --git a/pkg/kubelet/kubelet.go b/pkg/kubelet/kubelet.go index aaf8a763dee..ebdd86317e3 100644 --- a/pkg/kubelet/kubelet.go +++ b/pkg/kubelet/kubelet.go @@ -328,10 +328,11 @@ func PreInitRuntimeService(kubeCfg *kubeletconfiginternal.KubeletConfiguration, tp = kubeDeps.TracerProvider } - if kubeDeps.RemoteRuntimeService, err = remote.NewRemoteRuntimeService(kubeCfg.ContainerRuntimeEndpoint, kubeCfg.RuntimeRequestTimeout.Duration, tp); err != nil { + logger := klog.Background() + if kubeDeps.RemoteRuntimeService, err = remote.NewRemoteRuntimeService(kubeCfg.ContainerRuntimeEndpoint, kubeCfg.RuntimeRequestTimeout.Duration, tp, &logger); err != nil { return err } - if kubeDeps.RemoteImageService, err = remote.NewRemoteImageService(remoteImageEndpoint, kubeCfg.RuntimeRequestTimeout.Duration, tp); err != nil { + if kubeDeps.RemoteImageService, err = remote.NewRemoteImageService(remoteImageEndpoint, kubeCfg.RuntimeRequestTimeout.Duration, tp, &logger); err != nil { return err } diff --git a/pkg/kubelet/kubelet_test.go b/pkg/kubelet/kubelet_test.go index 906ad11cc20..47b7292d695 100644 --- a/pkg/kubelet/kubelet_test.go +++ b/pkg/kubelet/kubelet_test.go @@ -57,6 +57,7 @@ import ( featuregatetesting "k8s.io/component-base/featuregate/testing" internalapi "k8s.io/cri-api/pkg/apis" runtimeapi "k8s.io/cri-api/pkg/apis/runtime/v1" + "k8s.io/klog/v2" "k8s.io/klog/v2/ktesting" "k8s.io/kubernetes/pkg/features" kubeletconfiginternal "k8s.io/kubernetes/pkg/kubelet/apis/config" @@ -3032,7 +3033,8 @@ func createAndStartFakeRemoteRuntime(t *testing.T) (*fakeremote.RemoteRuntime, s } func createRemoteRuntimeService(endpoint string, t *testing.T, tp oteltrace.TracerProvider) internalapi.RuntimeService { - runtimeService, err := remote.NewRemoteRuntimeService(endpoint, 15*time.Second, tp) + logger := klog.Background() + runtimeService, err := remote.NewRemoteRuntimeService(endpoint, 15*time.Second, tp, &logger) require.NoError(t, err) return runtimeService } @@ -3190,7 +3192,8 @@ func TestSyncPodSpans(t *testing.T) { fakeRuntime.ImageService.SetFakeImageSize(100) fakeRuntime.ImageService.SetFakeImages([]string{"test:latest"}) - imageSvc, err := remote.NewRemoteImageService(endpoint, 15*time.Second, tp) + logger := klog.Background() + imageSvc, err := remote.NewRemoteImageService(endpoint, 15*time.Second, tp, &logger) assert.NoError(t, err) kubelet.containerRuntime, err = kuberuntime.NewKubeGenericRuntimeManager( diff --git a/test/e2e_node/util.go b/test/e2e_node/util.go index f4ffe7c69b4..7bc7729cfd2 100644 --- a/test/e2e_node/util.go +++ b/test/e2e_node/util.go @@ -35,7 +35,7 @@ import ( "k8s.io/kubernetes/pkg/util/procfs" e2epod "k8s.io/kubernetes/test/e2e/framework/pod" - oteltrace "go.opentelemetry.io/otel/trace" + "go.opentelemetry.io/otel/trace/noop" v1 "k8s.io/api/core/v1" apiequality "k8s.io/apimachinery/pkg/api/equality" @@ -319,9 +319,10 @@ func logKubeletLatencyMetrics(ctx context.Context, metricNames ...string) { // getCRIClient connects CRI and returns CRI runtime service clients and image service client. func getCRIClient() (internalapi.RuntimeService, internalapi.ImageManagerService, error) { // connection timeout for CRI service connection + logger := klog.Background() const connectionTimeout = 2 * time.Minute runtimeEndpoint := framework.TestContext.ContainerRuntimeEndpoint - r, err := remote.NewRemoteRuntimeService(runtimeEndpoint, connectionTimeout, oteltrace.NewNoopTracerProvider()) + r, err := remote.NewRemoteRuntimeService(runtimeEndpoint, connectionTimeout, noop.NewTracerProvider(), &logger) if err != nil { return nil, nil, err } @@ -331,7 +332,7 @@ func getCRIClient() (internalapi.RuntimeService, internalapi.ImageManagerService //explicitly specified imageManagerEndpoint = framework.TestContext.ImageServiceEndpoint } - i, err := remote.NewRemoteImageService(imageManagerEndpoint, connectionTimeout, oteltrace.NewNoopTracerProvider()) + i, err := remote.NewRemoteImageService(imageManagerEndpoint, connectionTimeout, noop.NewTracerProvider(), &logger) if err != nil { return nil, nil, err }