github/kubernetes
1
0
Fork 0
mirror of https://github.com/k3s-io/kubernetes.git synced 2025-07-31 07:20:13 +00:00
Code Issues Projects Releases Wiki Activity

Merge pull request #124741 from saschagrunert/remote-runtime-logging

Browse Source 
Make remote runtime and image service logging independent
This commit is contained in:
Kubernetes Prow Robot 2024-05-13 08:52:32 -07:00 committed by GitHub
commit 63e85caac5
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
9 changed files with 173 additions and 122 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
cmd/kubemark/app/hollow_node.go
View File

@ -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)
}

38
pkg/kubelet/cri/remote/remote_image.go
View File

@ -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

8
pkg/kubelet/cri/remote/remote_image_test.go
View File

@ -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

200
pkg/kubelet/cri/remote/remote_runtime.go
View File

@ -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
}

8
pkg/kubelet/cri/remote/remote_runtime_test.go
View File

@ -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

15
pkg/kubelet/cri/remote/utils.go
View File

@ -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...)
}

5
pkg/kubelet/kubelet.go
View File

@ -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
}

7
pkg/kubelet/kubelet_test.go
View File

@ -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(

7
test/e2e_node/util.go
View File

@ -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
}