kubernetes/k8s CRI 分析 - 容器运行时接口分析

// pkg/kubelet/config/flags.go// AddFlags adds flags to the container runtime, according to ContainerRuntimeOptions.func (s *ContainerRuntimeOptions) AddFlags(fs *pflag.FlagSet) {  dockerOnlyWarning := "This docker-specific flag only works when container-runtime is set to docker."
  // General settings.  fs.StringVar(&s.ContainerRuntime, "container-runtime", s.ContainerRuntime, "The container runtime to use. Possible values: 'docker', 'remote', 'rkt (deprecated)'.")  fs.StringVar(&s.RuntimeCgroups, "runtime-cgroups", s.RuntimeCgroups, "Optional absolute name of cgroups to create and run the runtime in.")  fs.BoolVar(&s.RedirectContainerStreaming, "redirect-container-streaming", s.RedirectContainerStreaming, "Enables container streaming redirect. If false, kubelet will proxy container streaming data between apiserver and container runtime; if true, kubelet will return an http redirect to apiserver, and apiserver will access container runtime directly. The proxy approach is more secure, but introduces some overhead. The redirect approach is more performant, but less secure because the connection between apiserver and container runtime may not be authenticated.")
  // Docker-specific settings.  fs.BoolVar(&s.ExperimentalDockershim, "experimental-dockershim", s.ExperimentalDockershim, "Enable dockershim only mode. In this mode, kubelet will only start dockershim without any other functionalities. This flag only serves test purpose, please do not use it unless you are conscious of what you are doing. [default=false]")  fs.MarkHidden("experimental-dockershim")  fs.StringVar(&s.DockershimRootDirectory, "experimental-dockershim-root-directory", s.DockershimRootDirectory, "Path to the dockershim root directory.")  fs.MarkHidden("experimental-dockershim-root-directory")  fs.StringVar(&s.PodSandboxImage, "pod-infra-container-image", s.PodSandboxImage, fmt.Sprintf("The image whose network/ipc namespaces containers in each pod will use. %s", dockerOnlyWarning))  fs.StringVar(&s.DockerEndpoint, "docker-endpoint", s.DockerEndpoint, fmt.Sprintf("Use this for the docker endpoint to communicate with. %s", dockerOnlyWarning))  fs.DurationVar(&s.ImagePullProgressDeadline.Duration, "image-pull-progress-deadline", s.ImagePullProgressDeadline.Duration, fmt.Sprintf("If no pulling progress is made before this deadline, the image pulling will be cancelled. %s", dockerOnlyWarning))  ...}

// cmd/kubelet/app/options/options.go// AddFlags adds flags for a specific KubeletFlags to the specified FlagSetfunc (f *KubeletFlags) AddFlags(mainfs *pflag.FlagSet) {    ...    fs.StringVar(&f.RemoteRuntimeEndpoint, "container-runtime-endpoint", f.RemoteRuntimeEndpoint, "[Experimental] The endpoint of remote runtime service. Currently unix socket endpoint is supported on Linux, while npipe and tcp endpoints are supported on windows.  Examples:'unix:///var/run/dockershim.sock', 'npipe:////./pipe/dockershim'")  fs.StringVar(&f.RemoteImageEndpoint, "image-service-endpoint", f.RemoteImageEndpoint, "[Experimental] The endpoint of remote image service. If not specified, it will be the same with container-runtime-endpoint by default. Currently unix socket endpoint is supported on Linux, while npipe and tcp endpoints are supported on windows.  Examples:'unix:///var/run/dockershim.sock', 'npipe:////./pipe/dockershim'")  ...}

// cmd/kubelet/app/options/options.go// NewKubeletFlags will create a new KubeletFlags with default valuesfunc NewKubeletFlags() *KubeletFlags {  remoteRuntimeEndpoint := ""  if runtime.GOOS == "linux" {    remoteRuntimeEndpoint = "unix:///var/run/dockershim.sock"  } else if runtime.GOOS == "windows" {    remoteRuntimeEndpoint = "npipe:////./pipe/dockershim"  }
  return &KubeletFlags{    EnableServer:                        true,    ContainerRuntimeOptions:             *NewContainerRuntimeOptions(),    CertDirectory:                       "/var/lib/kubelet/pki",    RootDirectory:                       defaultRootDir,    MasterServiceNamespace:              metav1.NamespaceDefault,    MaxContainerCount:                   -1,    MaxPerPodContainerCount:             1,    MinimumGCAge:                        metav1.Duration{Duration: 0},    NonMasqueradeCIDR:                   "10.0.0.0/8",    RegisterSchedulable:                 true,    ExperimentalKernelMemcgNotification: false,    RemoteRuntimeEndpoint:               remoteRuntimeEndpoint,    NodeLabels:                          make(map[string]string),    VolumePluginDir:                     "/usr/libexec/kubernetes/kubelet-plugins/volume/exec/",    RegisterNode:                        true,    SeccompProfileRoot:                  filepath.Join(defaultRootDir, "seccomp"),    // prior to the introduction of this flag, there was a hardcoded cap of 50 images    NodeStatusMaxImages:         50,    EnableCAdvisorJSONEndpoints: true,  }}

// cmd/kubelet/app/options/container_runtime.go// NewContainerRuntimeOptions will create a new ContainerRuntimeOptions with// default values.func NewContainerRuntimeOptions() *config.ContainerRuntimeOptions {  dockerEndpoint := ""  if runtime.GOOS != "windows" {    dockerEndpoint = "unix:///var/run/docker.sock"  }
  return &config.ContainerRuntimeOptions{    ContainerRuntime:           kubetypes.DockerContainerRuntime,    RedirectContainerStreaming: false,    DockerEndpoint:             dockerEndpoint,    DockershimRootDirectory:    "/var/lib/dockershim",    PodSandboxImage:            defaultPodSandboxImage,    ImagePullProgressDeadline:  metav1.Duration{Duration: 1 * time.Minute},    ExperimentalDockershim:     false,
    //Alpha feature    CNIBinDir:   "/opt/cni/bin",    CNIConfDir:  "/etc/cni/net.d",    CNICacheDir: "/var/lib/cni/cache",  }}

// pkg/kubelet/kubelet.gofunc NewMainKubelet(...) {    ...    if remoteRuntimeEndpoint != "" {    // remoteImageEndpoint is same as remoteRuntimeEndpoint if not explicitly specified    if remoteImageEndpoint == "" {      remoteImageEndpoint = remoteRuntimeEndpoint    }  }  ...}

// staging/src/k8s.io/cri-api/pkg/apis/services.go// RuntimeService interface should be implemented by a container runtime.// The methods should be thread-safe.type RuntimeService interface {  RuntimeVersioner  ContainerManager  PodSandboxManager  ContainerStatsManager
  // UpdateRuntimeConfig updates runtime configuration if specified  UpdateRuntimeConfig(runtimeConfig *runtimeapi.RuntimeConfig) error  // Status returns the status of the runtime.  Status() (*runtimeapi.RuntimeStatus, error)}

// staging/src/k8s.io/cri-api/pkg/apis/services.go// RuntimeVersioner contains methods for runtime name, version and API version.type RuntimeVersioner interface {  // Version returns the runtime name, runtime version and runtime API version  Version(apiVersion string) (*runtimeapi.VersionResponse, error)}

// staging/src/k8s.io/cri-api/pkg/apis/services.go// ContainerManager contains methods to manipulate containers managed by a// container runtime. The methods are thread-safe.type ContainerManager interface {  // CreateContainer creates a new container in specified PodSandbox.  CreateContainer(podSandboxID string, config *runtimeapi.ContainerConfig, sandboxConfig *runtimeapi.PodSandboxConfig) (string, error)  // StartContainer starts the container.  StartContainer(containerID string) error  // StopContainer stops a running container with a grace period (i.e., timeout).  StopContainer(containerID string, timeout int64) error  // RemoveContainer removes the container.  RemoveContainer(containerID string) error  // ListContainers lists all containers by filters.  ListContainers(filter *runtimeapi.ContainerFilter) ([]*runtimeapi.Container, error)  // ContainerStatus returns the status of the container.  ContainerStatus(containerID string) (*runtimeapi.ContainerStatus, error)  // UpdateContainerResources updates the cgroup resources for the container.  UpdateContainerResources(containerID string, resources *runtimeapi.LinuxContainerResources) error  // 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.  ExecSync(containerID string, cmd []string, timeout time.Duration) (stdout []byte, stderr []byte, err error)  // Exec prepares a streaming endpoint to execute a command in the container, and returns the address.  Exec(*runtimeapi.ExecRequest) (*runtimeapi.ExecResponse, error)  // Attach prepares a streaming endpoint to attach to a running container, and returns the address.  Attach(req *runtimeapi.AttachRequest) (*runtimeapi.AttachResponse, error)  // ReopenContainerLog asks runtime to reopen the stdout/stderr log file  // for the container. If it returns error, new container log file MUST NOT  // be created.  ReopenContainerLog(ContainerID string) error}

// staging/src/k8s.io/cri-api/pkg/apis/services.go// PodSandboxManager contains methods for operating on PodSandboxes. The methods// are thread-safe.type PodSandboxManager interface {  // RunPodSandbox creates and starts a pod-level sandbox. Runtimes should ensure  // the sandbox is in ready state.  RunPodSandbox(config *runtimeapi.PodSandboxConfig, runtimeHandler string) (string, error)  // StopPodSandbox stops the sandbox. If there are any running containers in the  // sandbox, they should be force terminated.  StopPodSandbox(podSandboxID string) error  // RemovePodSandbox removes the sandbox. If there are running containers in the  // sandbox, they should be forcibly removed.  RemovePodSandbox(podSandboxID string) error  // PodSandboxStatus returns the Status of the PodSandbox.  PodSandboxStatus(podSandboxID string) (*runtimeapi.PodSandboxStatus, error)  // ListPodSandbox returns a list of Sandbox.  ListPodSandbox(filter *runtimeapi.PodSandboxFilter) ([]*runtimeapi.PodSandbox, error)  // PortForward prepares a streaming endpoint to forward ports from a PodSandbox, and returns the address.  PortForward(*runtimeapi.PortForwardRequest) (*runtimeapi.PortForwardResponse, error)}

// staging/src/k8s.io/cri-api/pkg/apis/services.go// ContainerStatsManager contains methods for retrieving the container// statistics.type ContainerStatsManager interface {  // ContainerStats returns stats of the container. If the container does not  // exist, the call returns an error.  ContainerStats(containerID string) (*runtimeapi.ContainerStats, error)  // ListContainerStats returns stats of all running containers.  ListContainerStats(filter *runtimeapi.ContainerStatsFilter) ([]*runtimeapi.ContainerStats, error)}

// staging/src/k8s.io/cri-api/pkg/apis/services.go// ImageManagerService interface should be implemented by a container image// manager.// The methods should be thread-safe.type ImageManagerService interface {  // ListImages lists the existing images.  ListImages(filter *runtimeapi.ImageFilter) ([]*runtimeapi.Image, error)  // ImageStatus returns the status of the image.  ImageStatus(image *runtimeapi.ImageSpec) (*runtimeapi.Image, error)  // PullImage pulls an image with the authentication config.  PullImage(image *runtimeapi.ImageSpec, auth *runtimeapi.AuthConfig, podSandboxConfig *runtimeapi.PodSandboxConfig) (string, error)  // RemoveImage removes the image.  RemoveImage(image *runtimeapi.ImageSpec) error  // ImageFsInfo returns information of the filesystem that is used to store images.  ImageFsInfo() ([]*runtimeapi.FilesystemUsage, error)}

// pkg/kubelet/dockershim/docker_service.go// CRIService includes all methods necessary for a CRI server.type CRIService interface {  runtimeapi.RuntimeServiceServer  runtimeapi.ImageServiceServer  Start() error}
// DockerService is an interface that embeds the new RuntimeService and// ImageService interfaces.type DockerService interface {  CRIService
  // For serving streaming calls.  http.Handler
  // For supporting legacy features.  DockerLegacyService}

// pkg/kubelet/remote/remote_runtime.go// RemoteRuntimeService is a gRPC implementation of internalapi.RuntimeService.type RemoteRuntimeService struct {  timeout       time.Duration  runtimeClient runtimeapi.RuntimeServiceClient  // Cache last per-container error message to reduce log spam  logReduction *logreduction.LogReduction}

// pkg/kubelet/remote/remote_image.go// RemoteImageService is a gRPC implementation of internalapi.ImageManagerService.type RemoteImageService struct {  timeout     time.Duration  imageClient runtimeapi.ImageServiceClient}

// pkg/kubelet/dockershim/remote/docker_server.go// DockerServer is the grpc server of dockershim.type DockerServer struct {  // endpoint is the endpoint to serve on.  endpoint string  // service is the docker service which implements runtime and image services.  service dockershim.CRIService  // server is the grpc server.  server *grpc.Server}

// pkg/kubelet/kubelet.gofunc NewMainKubelet(...) {    ...    switch containerRuntime {    // （1）初始化并启动内置容器运行时服务端dockershim  case kubetypes.DockerContainerRuntime:    // Create and start the CRI shim running as a grpc server.    streamingConfig := getStreamingConfig(kubeCfg, kubeDeps, crOptions)    ds, err := dockershim.NewDockerService(kubeDeps.DockerClientConfig, crOptions.PodSandboxImage, streamingConfig,      &pluginSettings, runtimeCgroups, kubeCfg.CgroupDriver, crOptions.DockershimRootDirectory, !crOptions.RedirectContainerStreaming)    if err != nil {      return nil, err    }    if crOptions.RedirectContainerStreaming {      klet.criHandler = ds    }
    // The unix socket for kubelet <-> dockershim communication.    klog.V(5).Infof("RemoteRuntimeEndpoint: %q, RemoteImageEndpoint: %q",      remoteRuntimeEndpoint,      remoteImageEndpoint)    klog.V(2).Infof("Starting the GRPC server for the docker CRI shim.")    server := dockerremote.NewDockerServer(remoteRuntimeEndpoint, ds)    if err := server.Start(); err != nil {      return nil, err    }
    // Create dockerLegacyService when the logging driver is not supported.    supported, err := ds.IsCRISupportedLogDriver()    if err != nil {      return nil, err    }    if !supported {      klet.dockerLegacyService = ds      legacyLogProvider = ds    }  case kubetypes.RemoteContainerRuntime:    // No-op.    break  default:    return nil, fmt.Errorf("unsupported CRI runtime: %q", containerRuntime)  }  // （2）初始化容器运行时CRI shim客户端  runtimeService, imageService, err := getRuntimeAndImageServices(remoteRuntimeEndpoint, remoteImageEndpoint, kubeCfg.RuntimeRequestTimeout)  if err != nil {    return nil, err  }  klet.runtimeService = runtimeService
  if utilfeature.DefaultFeatureGate.Enabled(features.RuntimeClass) && kubeDeps.KubeClient != nil {    klet.runtimeClassManager = runtimeclass.NewManager(kubeDeps.KubeClient)  }    // （3）初始化```GenericRuntimeManager```，用于容器运行时的管理  runtime, err := kuberuntime.NewKubeGenericRuntimeManager(    kubecontainer.FilterEventRecorder(kubeDeps.Recorder),    klet.livenessManager,    klet.startupManager,    seccompProfileRoot,    containerRefManager,    machineInfo,    klet,    kubeDeps.OSInterface,    klet,    httpClient,    imageBackOff,    kubeCfg.SerializeImagePulls,    float32(kubeCfg.RegistryPullQPS),    int(kubeCfg.RegistryBurst),    kubeCfg.CPUCFSQuota,    kubeCfg.CPUCFSQuotaPeriod,    runtimeService,    imageService,    kubeDeps.ContainerManager.InternalContainerLifecycle(),    legacyLogProvider,    klet.runtimeClassManager,  )  if err != nil {    return nil, err  }  klet.containerRuntime = runtime  klet.streamingRuntime = runtime  klet.runner = runtime  ...}

// pkg/kubelet/dockershim/docker_service.go// NewDockerService creates a new `DockerService` struct.// NOTE: Anything passed to DockerService should be eventually handled in another way when we switch to running the shim as a different process.func NewDockerService(config *ClientConfig, podSandboxImage string, streamingConfig *streaming.Config, pluginSettings *NetworkPluginSettings,  cgroupsName string, kubeCgroupDriver string, dockershimRootDir string, startLocalStreamingServer bool) (DockerService, error) {    // （1）创建docker的客户端  client := NewDockerClientFromConfig(config)
  c := libdocker.NewInstrumentedInterface(client)
  checkpointManager, err := checkpointmanager.NewCheckpointManager(filepath.Join(dockershimRootDir, sandboxCheckpointDir))  if err != nil {    return nil, err  }    // （2）构建```dockerService struct```  ds := &dockerService{    client:          c,    os:              kubecontainer.RealOS{},    podSandboxImage: podSandboxImage,    streamingRuntime: &streamingRuntime{      client:      client,      execHandler: &NativeExecHandler{},    },    containerManager:          cm.NewContainerManager(cgroupsName, client),    checkpointManager:         checkpointManager,    startLocalStreamingServer: startLocalStreamingServer,    networkReady:              make(map[string]bool),    containerCleanupInfos:     make(map[string]*containerCleanupInfo),  }
  // check docker version compatibility.  if err = ds.checkVersionCompatibility(); err != nil {    return nil, err  }
  // create streaming server if configured.  if streamingConfig != nil {    var err error    ds.streamingServer, err = streaming.NewServer(*streamingConfig, ds.streamingRuntime)    if err != nil {      return nil, err    }  }
  // Determine the hairpin mode.  if err := effectiveHairpinMode(pluginSettings); err != nil {    // This is a non-recoverable error. Returning it up the callstack will just    // lead to retries of the same failure, so just fail hard.    return nil, err  }  klog.Infof("Hairpin mode set to %q", pluginSettings.HairpinMode)    // （3）初始化CNI网络配置  // dockershim currently only supports CNI plugins.  pluginSettings.PluginBinDirs = cni.SplitDirs(pluginSettings.PluginBinDirString)  cniPlugins := cni.ProbeNetworkPlugins(pluginSettings.PluginConfDir, pluginSettings.PluginCacheDir, pluginSettings.PluginBinDirs)  cniPlugins = append(cniPlugins, kubenet.NewPlugin(pluginSettings.PluginBinDirs, pluginSettings.PluginCacheDir))  netHost := &dockerNetworkHost{    &namespaceGetter{ds},    &portMappingGetter{ds},  }  plug, err := network.InitNetworkPlugin(cniPlugins, pluginSettings.PluginName, netHost, pluginSettings.HairpinMode, pluginSettings.NonMasqueradeCIDR, pluginSettings.MTU)  if err != nil {    return nil, fmt.Errorf("didn't find compatible CNI plugin with given settings %+v: %v", pluginSettings, err)  }  ds.network = network.NewPluginManager(plug)  klog.Infof("Docker cri networking managed by %v", plug.Name())
  // NOTE: cgroup driver is only detectable in docker 1.11+  cgroupDriver := defaultCgroupDriver  dockerInfo, err := ds.client.Info()  klog.Infof("Docker Info: %+v", dockerInfo)  if err != nil {    klog.Errorf("Failed to execute Info() call to the Docker client: %v", err)    klog.Warningf("Falling back to use the default driver: %q", cgroupDriver)  } else if len(dockerInfo.CgroupDriver) == 0 {    klog.Warningf("No cgroup driver is set in Docker")    klog.Warningf("Falling back to use the default driver: %q", cgroupDriver)  } else {    cgroupDriver = dockerInfo.CgroupDriver  }  if len(kubeCgroupDriver) != 0 && kubeCgroupDriver != cgroupDriver {    return nil, fmt.Errorf("misconfiguration: kubelet cgroup driver: %q is different from docker cgroup driver: %q", kubeCgroupDriver, cgroupDriver)  }  klog.Infof("Setting cgroupDriver to %s", cgroupDriver)  ds.cgroupDriver = cgroupDriver  ds.versionCache = cache.NewObjectCache(    func() (interface{}, error) {      return ds.getDockerVersion()    },    versionCacheTTL,  )
  // Register prometheus metrics.  metrics.Register()
  return ds, nil}

// pkg/kubelet/dockershim/docker_service.go// NewDockerClientFromConfig create a docker client from given configure// return nil if nil configure is given.func NewDockerClientFromConfig(config *ClientConfig) libdocker.Interface {  if config != nil {    // Create docker client.    client := libdocker.ConnectToDockerOrDie(      config.DockerEndpoint,      config.RuntimeRequestTimeout,      config.ImagePullProgressDeadline,      config.WithTraceDisabled,      config.EnableSleep,    )    return client  }
  return nil}

// pkg/kubelet/dockershim/libdocker/client.go// ConnectToDockerOrDie creates docker client connecting to docker daemon.// If the endpoint passed in is "fake://", a fake docker client// will be returned. The program exits if error occurs. The requestTimeout// is the timeout for docker requests. If timeout is exceeded, the request// will be cancelled and throw out an error. If requestTimeout is 0, a default// value will be applied.func ConnectToDockerOrDie(dockerEndpoint string, requestTimeout, imagePullProgressDeadline time.Duration,  withTraceDisabled bool, enableSleep bool) Interface {  if dockerEndpoint == FakeDockerEndpoint {    fakeClient := NewFakeDockerClient()    if withTraceDisabled {      fakeClient = fakeClient.WithTraceDisabled()    }
    if enableSleep {      fakeClient.EnableSleep = true    }    return fakeClient  }  client, err := getDockerClient(dockerEndpoint)  if err != nil {    klog.Fatalf("Couldn't connect to docker: %v", err)  }  klog.Infof("Start docker client with request timeout=%v", requestTimeout)  return newKubeDockerClient(client, requestTimeout, imagePullProgressDeadline)}

// pkg/kubelet/dockershim/remote/docker_server.go// NewDockerServer creates the dockershim grpc server.func NewDockerServer(endpoint string, s dockershim.CRIService) *DockerServer {  return &DockerServer{    endpoint: endpoint,    service:  s,  }}

// pkg/kubelet/kubelet.gofunc getRuntimeAndImageServices(remoteRuntimeEndpoint string, remoteImageEndpoint string, runtimeRequestTimeout metav1.Duration) (internalapi.RuntimeService, internalapi.ImageManagerService, error) {  rs, err := remote.NewRemoteRuntimeService(remoteRuntimeEndpoint, runtimeRequestTimeout.Duration)  if err != nil {    return nil, nil, err  }  is, err := remote.NewRemoteImageService(remoteImageEndpoint, runtimeRequestTimeout.Duration)  if err != nil {    return nil, nil, err  }  return rs, is, err}

// pkg/kubelet/remote/remote_runtime.go// NewRemoteRuntimeService creates a new internalapi.RuntimeService.func NewRemoteRuntimeService(endpoint string, connectionTimeout time.Duration) (internalapi.RuntimeService, error) {  klog.V(3).Infof("Connecting to runtime service %s", endpoint)  addr, dailer, err := util.GetAddressAndDialer(endpoint)  if err != nil {    return nil, err  }  ctx, cancel := context.WithTimeout(context.Background(), connectionTimeout)  defer cancel()
  conn, err := grpc.DialContext(ctx, addr, grpc.WithInsecure(), grpc.WithDialer(dailer), grpc.WithDefaultCallOptions(grpc.MaxCallRecvMsgSize(maxMsgSize)))  if err != nil {    klog.Errorf("Connect remote runtime %s failed: %v", addr, err)    return nil, err  }
  return &RemoteRuntimeService{    timeout:       connectionTimeout,    runtimeClient: runtimeapi.NewRuntimeServiceClient(conn),    logReduction:  logreduction.NewLogReduction(identicalErrorDelay),  }, nil}

// pkg/kubelet/remote/remote_runtime.go// NewRemoteImageService creates a new internalapi.ImageManagerService.func NewRemoteImageService(endpoint string, connectionTimeout time.Duration) (internalapi.ImageManagerService, error) {  klog.V(3).Infof("Connecting to image service %s", endpoint)  addr, dailer, err := util.GetAddressAndDialer(endpoint)  if err != nil {    return nil, err  }
  ctx, cancel := context.WithTimeout(context.Background(), connectionTimeout)  defer cancel()
  conn, err := grpc.DialContext(ctx, addr, grpc.WithInsecure(), grpc.WithDialer(dailer), grpc.WithDefaultCallOptions(grpc.MaxCallRecvMsgSize(maxMsgSize)))  if err != nil {    klog.Errorf("Connect remote image service %s failed: %v", addr, err)    return nil, err  }
  return &RemoteImageService{    timeout:     connectionTimeout,    imageClient: runtimeapi.NewImageServiceClient(conn),  }, nil}

// pkg/kubelet/kuberuntime/kuberuntime_manager.go// NewKubeGenericRuntimeManager creates a new kubeGenericRuntimeManagerfunc NewKubeGenericRuntimeManager(  recorder record.EventRecorder,  livenessManager proberesults.Manager,  startupManager proberesults.Manager,  seccompProfileRoot string,  containerRefManager *kubecontainer.RefManager,  machineInfo *cadvisorapi.MachineInfo,  podStateProvider podStateProvider,  osInterface kubecontainer.OSInterface,  runtimeHelper kubecontainer.RuntimeHelper,  httpClient types.HttpGetter,  imageBackOff *flowcontrol.Backoff,  serializeImagePulls bool,  imagePullQPS float32,  imagePullBurst int,  cpuCFSQuota bool,  cpuCFSQuotaPeriod metav1.Duration,  runtimeService internalapi.RuntimeService,  imageService internalapi.ImageManagerService,  internalLifecycle cm.InternalContainerLifecycle,  legacyLogProvider LegacyLogProvider,  runtimeClassManager *runtimeclass.Manager,) (KubeGenericRuntime, error) {  kubeRuntimeManager := &kubeGenericRuntimeManager{    recorder:            recorder,    cpuCFSQuota:         cpuCFSQuota,    cpuCFSQuotaPeriod:   cpuCFSQuotaPeriod,    seccompProfileRoot:  seccompProfileRoot,    livenessManager:     livenessManager,    startupManager:      startupManager,    containerRefManager: containerRefManager,    machineInfo:         machineInfo,    osInterface:         osInterface,    runtimeHelper:       runtimeHelper,    runtimeService:      newInstrumentedRuntimeService(runtimeService),    imageService:        newInstrumentedImageManagerService(imageService),    keyring:             credentialprovider.NewDockerKeyring(),    internalLifecycle:   internalLifecycle,    legacyLogProvider:   legacyLogProvider,    runtimeClassManager: runtimeClassManager,    logReduction:        logreduction.NewLogReduction(identicalErrorDelay),  }
  typedVersion, err := kubeRuntimeManager.runtimeService.Version(kubeRuntimeAPIVersion)  if err != nil {    klog.Errorf("Get runtime version failed: %v", err)    return nil, err  }
  // Only matching kubeRuntimeAPIVersion is supported now  // TODO: Runtime API machinery is under discussion at https://github.com/kubernetes/kubernetes/issues/28642  if typedVersion.Version != kubeRuntimeAPIVersion {    klog.Errorf("Runtime api version %s is not supported, only %s is supported now",      typedVersion.Version,      kubeRuntimeAPIVersion)    return nil, ErrVersionNotSupported  }
  kubeRuntimeManager.runtimeName = typedVersion.RuntimeName  klog.Infof("Container runtime %s initialized, version: %s, apiVersion: %s",    typedVersion.RuntimeName,    typedVersion.RuntimeVersion,    typedVersion.RuntimeApiVersion)
  // If the container logs directory does not exist, create it.  // TODO: create podLogsRootDirectory at kubelet.go when kubelet is refactored to  // new runtime interface  if _, err := osInterface.Stat(podLogsRootDirectory); os.IsNotExist(err) {    if err := osInterface.MkdirAll(podLogsRootDirectory, 0755); err != nil {      klog.Errorf("Failed to create directory %q: %v", podLogsRootDirectory, err)    }  }
  kubeRuntimeManager.imagePuller = images.NewImageManager(    kubecontainer.FilterEventRecorder(recorder),    kubeRuntimeManager,    imageBackOff,    serializeImagePulls,    imagePullQPS,    imagePullBurst)  kubeRuntimeManager.runner = lifecycle.NewHandlerRunner(httpClient, kubeRuntimeManager, kubeRuntimeManager)  kubeRuntimeManager.containerGC = newContainerGC(runtimeService, podStateProvider, kubeRuntimeManager)
  kubeRuntimeManager.versionCache = cache.NewObjectCache(    func() (interface{}, error) {      return kubeRuntimeManager.getTypedVersion()    },    versionCacheTTL,  )
  return kubeRuntimeManager, nil}