Release 1.22 camille rebase

pkg/kubelet/cadvisor/BUILD

@@ -0,0 +1,103 @@
+package(default_visibility = ["//visibility:public"])
+
+load(
+    "@io_bazel_rules_go//go:def.bzl",
+    "go_library",
+    "go_test",
+)
+
+go_library(
+    name = "go_default_library",
+    srcs = [
+        "cadvisor_linux.go",
+        "cadvisor_unsupported.go",
+        "cadvisor_windows.go",
+        "doc.go",
+        "helpers_linux.go",
+        "helpers_unsupported.go",
+        "types.go",
+        "util.go",
+    ],
+    importpath = "k8s.io/kubernetes/pkg/kubelet/cadvisor",
+    deps = [
+        "//pkg/apis/core/v1/helper:go_default_library",
+        "//pkg/kubelet/types:go_default_library",
+        "//staging/src/k8s.io/api/core/v1:go_default_library",
+        "//staging/src/k8s.io/apimachinery/pkg/api/resource:go_default_library",
+        "//vendor/github.com/google/cadvisor/events:go_default_library",
+        "//vendor/github.com/google/cadvisor/info/v1:go_default_library",
+        "//vendor/github.com/google/cadvisor/info/v2:go_default_library",
+    ] + select({
+        "@io_bazel_rules_go//go/platform:android": [
+            "//vendor/github.com/google/cadvisor/cache/memory:go_default_library",
+            "//vendor/github.com/google/cadvisor/container:go_default_library",
+            "//vendor/github.com/google/cadvisor/container/containerd/install:go_default_library",
+            "//vendor/github.com/google/cadvisor/container/crio/install:go_default_library",
+            "//vendor/github.com/google/cadvisor/container/docker/install:go_default_library",
+            "//vendor/github.com/google/cadvisor/container/systemd/install:go_default_library",
+            "//vendor/github.com/google/cadvisor/fs:go_default_library",
+            "//vendor/github.com/google/cadvisor/manager:go_default_library",
+            "//vendor/github.com/google/cadvisor/utils/cloudinfo/aws:go_default_library",
+            "//vendor/github.com/google/cadvisor/utils/cloudinfo/azure:go_default_library",
+            "//vendor/github.com/google/cadvisor/utils/cloudinfo/gce:go_default_library",
+            "//vendor/github.com/google/cadvisor/utils/sysfs:go_default_library",
+            "//vendor/k8s.io/klog:go_default_library",
+        ],
+        "@io_bazel_rules_go//go/platform:linux": [
+            "//vendor/github.com/google/cadvisor/cache/memory:go_default_library",
+            "//vendor/github.com/google/cadvisor/container:go_default_library",
+            "//vendor/github.com/google/cadvisor/container/containerd/install:go_default_library",
+            "//vendor/github.com/google/cadvisor/container/crio/install:go_default_library",
+            "//vendor/github.com/google/cadvisor/container/docker/install:go_default_library",
+            "//vendor/github.com/google/cadvisor/container/systemd/install:go_default_library",
+            "//vendor/github.com/google/cadvisor/fs:go_default_library",
+            "//vendor/github.com/google/cadvisor/manager:go_default_library",
+            "//vendor/github.com/google/cadvisor/utils/cloudinfo/aws:go_default_library",
+            "//vendor/github.com/google/cadvisor/utils/cloudinfo/azure:go_default_library",
+            "//vendor/github.com/google/cadvisor/utils/cloudinfo/gce:go_default_library",
+            "//vendor/github.com/google/cadvisor/utils/sysfs:go_default_library",
+            "//vendor/k8s.io/klog:go_default_library",
+        ],
+        "@io_bazel_rules_go//go/platform:windows": [
+            "//pkg/kubelet/winstats:go_default_library",
+        ],
+        "//conditions:default": [],
+    }),
+)
+
+go_test(
+    name = "go_default_test",
+    srcs = ["util_test.go"],
+    embed = [":go_default_library"],
+    deps = select({
+        "@io_bazel_rules_go//go/platform:android": [
+            "//staging/src/k8s.io/api/core/v1:go_default_library",
+            "//staging/src/k8s.io/apimachinery/pkg/api/resource:go_default_library",
+            "//vendor/github.com/google/cadvisor/container/crio:go_default_library",
+            "//vendor/github.com/google/cadvisor/info/v1:go_default_library",
+            "//vendor/github.com/stretchr/testify/assert:go_default_library",
+        ],
+        "@io_bazel_rules_go//go/platform:linux": [
+            "//staging/src/k8s.io/api/core/v1:go_default_library",
+            "//staging/src/k8s.io/apimachinery/pkg/api/resource:go_default_library",
+            "//vendor/github.com/google/cadvisor/info/v1:go_default_library",
+        ],
+        "//conditions:default": [],
+    }),
+)
+
+filegroup(
+    name = "package-srcs",
+    srcs = glob(["**"]),
+    tags = ["automanaged"],
+    visibility = ["//visibility:private"],
+)
+
+filegroup(
+    name = "all-srcs",
+    srcs = [
+        ":package-srcs",
+        "//pkg/kubelet/cadvisor/testing:all-srcs",
+    ],
+    tags = ["automanaged"],
+)

pkg/kubelet/cadvisor/helpers_linux.go

@@ -38,13 +38,6 @@ func (i *imageFsInfoProvider) ImageFsInfoLabel() (string, error) {
 	switch i.runtime {
 	case types.DockerContainerRuntime:
 		return cadvisorfs.LabelDockerImages, nil
-	case types.RemoteContainerRuntime:
-		// This is a temporary workaround to get stats for cri-o from cadvisor
-		// and should be removed.
-		// Related to https://github.com/kubernetes/kubernetes/issues/51798
-		if i.runtimeEndpoint == CrioSocket || i.runtimeEndpoint == "unix://"+CrioSocket {
-			return cadvisorfs.LabelCrioImages, nil
-		}
 	}
 	return "", fmt.Errorf("no imagefs label for configured runtime")
 }

pkg/kubelet/cadvisor/util.go

@@ -21,20 +21,12 @@ import (
 
 	cadvisorapi "github.com/google/cadvisor/info/v1"
 	cadvisorapi2 "github.com/google/cadvisor/info/v2"
-	"k8s.io/api/core/v1"
+	v1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/resource"
 	v1helper "k8s.io/kubernetes/pkg/apis/core/v1/helper"
 	kubetypes "k8s.io/kubernetes/pkg/kubelet/types"
 )
 
-const (
-	// CrioSocket is the path to the CRI-O socket.
-	// Please keep this in sync with the one in:
-	// github.com/google/cadvisor/tree/master/container/crio/client.go
-	CrioSocket = "/var/run/crio/crio.sock"
-)
-
-// CapacityFromMachineInfo returns the capacity of the resources from the machine info.
 func CapacityFromMachineInfo(info *cadvisorapi.MachineInfo) v1.ResourceList {
 	c := v1.ResourceList{
 		v1.ResourceCPU: *resource.NewMilliQuantity(
@@ -74,6 +66,5 @@ func EphemeralStorageCapacityFromFsInfo(info cadvisorapi2.FsInfo) v1.ResourceLis
 // be removed. Related issue:
 // https://github.com/kubernetes/kubernetes/issues/51798
 func UsingLegacyCadvisorStats(runtime, runtimeEndpoint string) bool {
-	return (runtime == kubetypes.DockerContainerRuntime && goruntime.GOOS == "linux") ||
-		runtimeEndpoint == CrioSocket || runtimeEndpoint == "unix://"+CrioSocket
+	return (runtime == kubetypes.DockerContainerRuntime && goruntime.GOOS == "linux")
 }

pkg/kubelet/cadvisor/util_test.go

@@ -22,9 +22,7 @@ import (
 	"reflect"
 	"testing"
 
-	"github.com/google/cadvisor/container/crio"
 	info "github.com/google/cadvisor/info/v1"
-	"github.com/stretchr/testify/assert"
 	"k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/resource"
 )
@@ -51,7 +49,3 @@ func TestCapacityFromMachineInfoWithHugePagesEnable(t *testing.T) {
 		t.Errorf("when set hugepages true, got resource list %v, want %v", actual, expected)
 	}
 }
-
-func TestCrioSocket(t *testing.T) {
-	assert.EqualValues(t, CrioSocket, crio.CrioSocket, "CrioSocket in this package must equal the one in github.com/google/cadvisor/container/crio/client.go")
-}

pkg/kubelet/kubelet.go

@@ -5,7 +5,7 @@ Licensed under the Apache License, Version 2.0 (the "License");
 you may not use this file except in compliance with the License.
 You may obtain a copy of the License at
 
-    http://www.apache.org/licenses/LICENSE-2.0
+	http://www.apache.org/licenses/LICENSE-2.0
 
 Unless required by applicable law or agreed to in writing, software
 distributed under the License is distributed on an "AS IS" BASIS,
@@ -59,7 +59,7 @@ import (
 	cloudprovider "k8s.io/cloud-provider"
 	"k8s.io/component-helpers/apimachinery/lease"
 	internalapi "k8s.io/cri-api/pkg/apis"
-	"k8s.io/klog/v2"
+	"k8s.io/klog"
 	pluginwatcherapi "k8s.io/kubelet/pkg/apis/pluginregistration/v1"
 	statsapi "k8s.io/kubelet/pkg/apis/stats/v1alpha1"
 	api "k8s.io/kubernetes/pkg/apis/core"
@@ -2318,10 +2318,19 @@ func (kl *Kubelet) HandlePodReconcile(pods []*v1.Pod) {
 		// to the pod manager.
 		kl.podManager.UpdatePod(pod)
 
+		sidecarsStatus := status.GetSidecarsStatus(pod)
+
 		// Reconcile Pod "Ready" condition if necessary. Trigger sync pod for reconciliation.
 		if status.NeedToReconcilePodReadiness(pod) {
 			mirrorPod, _ := kl.podManager.GetMirrorPodByPod(pod)
 			kl.dispatchWork(pod, kubetypes.SyncPodSync, mirrorPod, start)
+		} else if sidecarsStatus.ContainersWaiting {
+			// if containers aren't running and the sidecars are all ready trigger a sync so that the containers get started
+			if sidecarsStatus.SidecarsPresent && sidecarsStatus.SidecarsReady {
+				klog.Infof("Pod: %s: sidecars: sidecars are ready, dispatching work", format.Pod(pod))
+				mirrorPod, _ := kl.podManager.GetMirrorPodByPod(pod)
+				kl.dispatchWork(pod, kubetypes.SyncPodSync, mirrorPod, start)
+			}
 		}
 
 		// After an evicted pod is synced, all dead containers in the pod can be removed.

pkg/kubelet/kuberuntime/kuberuntime_container.go

@@ -5,7 +5,7 @@ Licensed under the Apache License, Version 2.0 (the "License");
 you may not use this file except in compliance with the License.
 You may obtain a copy of the License at
 
-    http://www.apache.org/licenses/LICENSE-2.0
+	http://www.apache.org/licenses/LICENSE-2.0
 
 Unless required by applicable law or agreed to in writing, software
 distributed under the License is distributed on an "AS IS" BASIS,
@@ -39,7 +39,7 @@ import (
 	grpcstatus "google.golang.org/grpc/status"
 
 	"github.com/armon/circbuf"
-	"k8s.io/klog/v2"
+	"k8s.io/klog"
 
 	v1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -284,9 +284,9 @@ func (m *kubeGenericRuntimeManager) startContainer(podSandboxID string, podSandb
 			klog.ErrorS(handlerErr, "Failed to execute PostStartHook", "pod", klog.KObj(pod),
 				"podUID", pod.UID, "containerName", container.Name, "containerID", kubeContainerID.String())
 			m.recordContainerEvent(pod, container, kubeContainerID.ID, v1.EventTypeWarning, events.FailedPostStartHook, msg)
-			if err := m.killContainer(pod, kubeContainerID, container.Name, "FailedPostStartHook", reasonFailedPostStartHook, nil); err != nil {
-				klog.ErrorS(err, "Failed to kill container", "pod", klog.KObj(pod),
-					"podUID", pod.UID, "containerName", container.Name, "containerID", kubeContainerID.String())
+			if err := m.killContainer(pod, kubeContainerID, container.Name, "FailedPostStartHook", reasonFailedPostStartHook, 0); err != nil {
+				klog.Errorf("Failed to kill container %q(id=%q) in pod %q: %v, %v",
+					container.Name, kubeContainerID.String(), format.Pod(pod), ErrPostStartHook, err)
 			}
 			return msg, ErrPostStartHook
 		}
@@ -631,6 +631,12 @@ func (m *kubeGenericRuntimeManager) restoreSpecsFromContainerLabels(containerID
 
 	l := getContainerInfoFromLabels(s.Labels)
 	a := getContainerInfoFromAnnotations(s.Annotations)
+
+	annotations := make(map[string]string)
+	if a.Sidecar {
+		annotations[fmt.Sprintf("sidecars.lyft.net/container-lifecycle-%s", l.ContainerName)] = "Sidecar"
+	}
+
 	// Notice that the followings are not full spec. The container killing code should not use
 	// un-restored fields.
 	pod = &v1.Pod{
@@ -639,6 +645,7 @@ func (m *kubeGenericRuntimeManager) restoreSpecsFromContainerLabels(containerID
 			Name:                       l.PodName,
 			Namespace:                  l.PodNamespace,
 			DeletionGracePeriodSeconds: a.PodDeletionGracePeriod,
+			Annotations:                annotations,
 		},
 		Spec: v1.PodSpec{
 			TerminationGracePeriodSeconds: a.PodTerminationGracePeriod,
@@ -660,12 +667,19 @@ func (m *kubeGenericRuntimeManager) restoreSpecsFromContainerLabels(containerID
 // killContainer kills a container through the following steps:
 // * Run the pre-stop lifecycle hooks (if applicable).
 // * Stop the container.
-func (m *kubeGenericRuntimeManager) killContainer(pod *v1.Pod, containerID kubecontainer.ContainerID, containerName string, message string, reason containerKillReason, gracePeriodOverride *int64) error {
+func (m *kubeGenericRuntimeManager) killContainer(pod *v1.Pod, containerID kubecontainer.ContainerID, containerName string, message string, reason containerKillReason, gracePeriodDuration time.Duration) error {
 	var containerSpec *v1.Container
 	if pod != nil {
 		if containerSpec = kubecontainer.GetContainerSpec(pod, containerName); containerSpec == nil {
-			return fmt.Errorf("failed to get containerSpec %q (id=%q) in pod %q when killing container for reason %q",
-				containerName, containerID.String(), format.Pod(pod), message)
+			// after a kubelet restart, it's not 100% certain that the
+			// pod we're given has the container we need in the spec
+			// -- we try to recover that here.
+			restoredPod, restoredContainer, err := m.restoreSpecsFromContainerLabels(containerID)
+			if err != nil {
+				return fmt.Errorf("failed to get containerSpec %q(id=%q) in pod %q when killing container for reason %q. error: %v",
+					containerName, containerID.String(), format.Pod(pod), message, err)
+			}
+			pod, containerSpec = restoredPod, restoredContainer
 		}
 	} else {
 		// Restore necessary information if one of the specs is nil.
@@ -714,10 +728,9 @@ func (m *kubeGenericRuntimeManager) killContainer(pod *v1.Pod, containerID kubec
 	if gracePeriod < minimumGracePeriodInSeconds {
 		gracePeriod = minimumGracePeriodInSeconds
 	}
-	if gracePeriodOverride != nil {
-		gracePeriod = *gracePeriodOverride
-		klog.V(3).InfoS("Killing container with a grace period override", "pod", klog.KObj(pod), "podUID", pod.UID,
-			"containerName", containerName, "containerID", containerID.String(), "gracePeriod", gracePeriod)
+	if gracePeriodDuration > 0 {
+		gracePeriod = int64(gracePeriodDuration.Seconds())
+		klog.V(3).Infof("Killing container %q, but using %d second grace period override", containerID, gracePeriod)
 	}
 
 	klog.V(2).InfoS("Killing container with a grace period", "pod", klog.KObj(pod), "podUID", pod.UID,
@@ -736,18 +749,54 @@ func (m *kubeGenericRuntimeManager) killContainer(pod *v1.Pod, containerID kubec
 }
 
 // killContainersWithSyncResult kills all pod's containers with sync results.
-func (m *kubeGenericRuntimeManager) killContainersWithSyncResult(pod *v1.Pod, runningPod kubecontainer.Pod, gracePeriodOverride *int64) (syncResults []*kubecontainer.SyncResult) {
+func (m *kubeGenericRuntimeManager) killContainersWithSyncResult(pod *v1.Pod, runningPod kubecontainer.Pod, gracePeriodDuration time.Duration) (syncResults []*kubecontainer.SyncResult) {
+	// split out sidecars and non-sidecars
+	var (
+		sidecars    []*kubecontainer.Container
+		nonSidecars []*kubecontainer.Container
+	)
+	for _, container := range runningPod.Containers {
+		if isSidecar(pod, container.Name) {
+			sidecars = append(sidecars, container)
+		} else {
+			nonSidecars = append(nonSidecars, container)
+		}
+	}
+
 	containerResults := make(chan *kubecontainer.SyncResult, len(runningPod.Containers))
-	wg := sync.WaitGroup{}
+	// non-sidecars first
+	start := time.Now()
+	klog.Infof("Pod: %s, killContainersWithSyncResult: killing %d non-sidecars, %s termination period", runningPod.Name, len(nonSidecars), gracePeriodDuration)
+	nonSidecarsWg := sync.WaitGroup{}
+	nonSidecarsWg.Add(len(nonSidecars))
+	for _, container := range nonSidecars {
+		go func(container *kubecontainer.Container) {
+			defer utilruntime.HandleCrash()
+			defer nonSidecarsWg.Done()
+			killContainerResult := kubecontainer.NewSyncResult(kubecontainer.KillContainer, container.Name)
+			if err := m.killContainer(pod, container.ID, container.Name, "Need to kill Pod", gracePeriodDuration); err != nil {
+				killContainerResult.Fail(kubecontainer.ErrKillContainer, err.Error())
+			}
+			containerResults <- killContainerResult
+		}(container)
+	}
+	nonSidecarsWg.Wait()
 
-	wg.Add(len(runningPod.Containers))
-	for _, container := range runningPod.Containers {
+	gracePeriodDuration = gracePeriodDuration - time.Since(start)
+	if gracePeriodDuration < 0 {
+		gracePeriodDuration = 0
+	}
+
+	// then sidecars
+	klog.Infof("Pod: %s, killContainersWithSyncResult: killing %d sidecars, %s left", runningPod.Name, len(sidecars), gracePeriodDuration)
+	wg := sync.WaitGroup{}
+	wg.Add(len(sidecars))
+	for _, container := range sidecars {
 		go func(container *kubecontainer.Container) {
 			defer utilruntime.HandleCrash()
 			defer wg.Done()
-
 			killContainerResult := kubecontainer.NewSyncResult(kubecontainer.KillContainer, container.Name)
-			if err := m.killContainer(pod, container.ID, container.Name, "", reasonUnknown, gracePeriodOverride); err != nil {
+			if err := m.killContainer(pod, container.ID, container.Name, "Need to kill Pod", reasonUnknown, gracePeriodDuration); err != nil {
 				killContainerResult.Fail(kubecontainer.ErrKillContainer, err.Error())
 				// Use runningPod for logging as the pod passed in could be *nil*.
 				klog.ErrorS(err, "Kill container failed", "pod", klog.KRef(runningPod.Namespace, runningPod.Name), "podUID", runningPod.ID,
@@ -757,6 +806,7 @@ func (m *kubeGenericRuntimeManager) killContainersWithSyncResult(pod *v1.Pod, ru
 		}(container)
 	}
 	wg.Wait()
+
 	close(containerResults)
 
 	for containerResult := range containerResults {