Use container log manager in kubelet

cmd/kubelet/app/options/options.go

@@ -527,6 +527,8 @@ func AddKubeletConfigFlags(mainfs *pflag.FlagSet, c *kubeletconfig.KubeletConfig
 	fs.BoolVar(&c.MakeIPTablesUtilChains, "make-iptables-util-chains", c.MakeIPTablesUtilChains, "If true, kubelet will ensure iptables utility rules are present on host.")
 	fs.Int32Var(&c.IPTablesMasqueradeBit, "iptables-masquerade-bit", c.IPTablesMasqueradeBit, "The bit of the fwmark space to mark packets for SNAT. Must be within the range [0, 31]. Please match this parameter with corresponding parameter in kube-proxy.")
 	fs.Int32Var(&c.IPTablesDropBit, "iptables-drop-bit", c.IPTablesDropBit, "The bit of the fwmark space to mark packets for dropping. Must be within the range [0, 31].")
+	fs.StringVar(&c.ContainerLogMaxSize, "container-log-max-size", c.ContainerLogMaxSize, "<Warning: Alpha feature> Set the maximum size (e.g. 10Mi) of container log file before it is rotated.")
+	fs.Int32Var(&c.ContainerLogMaxFiles, "container-log-max-files", c.ContainerLogMaxFiles, "<Warning: Alpha feature> Set the maximum number of container log files that can be present for a container. The number must be >= 2.")
 
 	// Flags intended for testing, not recommended used in production environments.
 	fs.Int64Var(&c.MaxOpenFiles, "max-open-files", c.MaxOpenFiles, "Number of files that can be opened by Kubelet process.")

pkg/features/kube_features.go

@@ -250,6 +250,12 @@ const (
 	//
 	// Implement TokenRequest endpoint on service account resources.
 	TokenRequest utilfeature.Feature = "TokenRequest"
+
+	// owner: @Random-Liu
+	// alpha: v1.10
+	//
+	// Enable container log rotation for cri container runtime
+	CRIContainerLogRotation utilfeature.Feature = "CRIContainerLogRotation"
 )
 
 func init() {
@@ -293,6 +299,7 @@ var defaultKubernetesFeatureGates = map[utilfeature.Feature]utilfeature.FeatureS
 	HyperVContainer:                             {Default: false, PreRelease: utilfeature.Alpha},
 	NoDaemonSetScheduler:                        {Default: false, PreRelease: utilfeature.Alpha},
 	TokenRequest:                                {Default: false, PreRelease: utilfeature.Alpha},
+	CRIContainerLogRotation:                     {Default: false, PreRelease: utilfeature.Alpha},
 
 	// inherited features from generic apiserver, relisted here to get a conflict if it is changed
 	// unintentionally on either side:

pkg/kubelet/apis/cri/runtime/v1alpha2/api.proto

@@ -66,7 +66,9 @@ service RuntimeService {
     rpc UpdateContainerResources(UpdateContainerResourcesRequest) returns (UpdateContainerResourcesResponse) {}
     // ReopenContainerLog asks runtime to reopen the stdout/stderr log file
     // for the container. This is often called after the log file has been
-    // rotated.
+    // rotated. If the container is not running, container runtime can choose
+    // to either create a new log file and return nil, or return an error.
+    // Once it returns error, new container log file MUST NOT be created.
     rpc ReopenContainerLog(ReopenContainerLogRequest) returns (ReopenContainerLogResponse) {}
 
     // ExecSync runs a command in a container synchronously.

pkg/kubelet/apis/cri/services.go

@@ -53,7 +53,8 @@ type ContainerManager interface {
 	// 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.
+	// for the container. If it returns error, new container log file MUST NOT
+	// be created.
 	ReopenContainerLog(ContainerID string) error
 }
 

pkg/kubelet/apis/kubeletconfig/fuzzer/fuzzer.go

@@ -90,6 +90,8 @@ func Funcs(codecs runtimeserializer.CodecFactory) []interface{} {
 			obj.CgroupDriver = "cgroupfs"
 			obj.EnforceNodeAllocatable = v1beta1.DefaultNodeAllocatableEnforcement
 			obj.ManifestURLHeader = make(map[string][]string)
+			obj.ContainerLogMaxFiles = 5
+			obj.ContainerLogMaxSize = "10Mi"
 		},
 	}
 }

pkg/kubelet/apis/kubeletconfig/helpers_test.go

@@ -152,6 +152,8 @@ var (
 		"CgroupsPerQOS",
 		"ClusterDNS[*]",
 		"ClusterDomain",
+		"ContainerLogMaxFiles",
+		"ContainerLogMaxSize",
 		"ContentType",
 		"EnableContentionProfiling",
 		"EnableControllerAttachDetach",

pkg/kubelet/apis/kubeletconfig/types.go

@@ -240,6 +240,10 @@ type KubeletConfiguration struct {
 	FeatureGates map[string]bool
 	// Tells the Kubelet to fail to start if swap is enabled on the node.
 	FailSwapOn bool
+	// A quantity defines the maximum size of the container log file before it is rotated. For example: "5Mi" or "256Ki".
+	ContainerLogMaxSize string
+	// Maximum number of container log files that can be present for a container.
+	ContainerLogMaxFiles int32
 
 	/* following flags are meant for Node Allocatable */
 

pkg/kubelet/apis/kubeletconfig/v1beta1/defaults.go

@@ -192,6 +192,12 @@ func SetDefaults_KubeletConfiguration(obj *KubeletConfiguration) {
 	if obj.FailSwapOn == nil {
 		obj.FailSwapOn = utilpointer.BoolPtr(true)
 	}
+	if obj.ContainerLogMaxSize == "" {
+		obj.ContainerLogMaxSize = "10Mi"
+	}
+	if obj.ContainerLogMaxFiles == nil {
+		obj.ContainerLogMaxFiles = utilpointer.Int32Ptr(5)
+	}
 	if obj.EnforceNodeAllocatable == nil {
 		obj.EnforceNodeAllocatable = DefaultNodeAllocatableEnforcement
 	}

pkg/kubelet/apis/kubeletconfig/v1beta1/types.go

@@ -390,6 +390,14 @@ type KubeletConfiguration struct {
 	// Default: true
 	// +optional
 	FailSwapOn *bool `json:"failSwapOn,omitempty"`
+	// A quantity defines the maximum size of the container log file before it is rotated. For example: "5Mi" or "256Ki".
+	// Default: "10Mi"
+	// +optional
+	ContainerLogMaxSize string `json:"containerLogMaxSize,omitempty"`
+	// Maximum number of container log files that can be present for a container.
+	// Default: 5
+	// +optional
+	ContainerLogMaxFiles *int32 `json:"containerLogMaxFiles,omitempty"`
 
 	/* following flags are meant for Node Allocatable */
 

pkg/kubelet/kubelet.go

@@ -74,6 +74,7 @@ import (
 	"k8s.io/kubernetes/pkg/kubelet/kubeletconfig"
 	"k8s.io/kubernetes/pkg/kubelet/kuberuntime"
 	"k8s.io/kubernetes/pkg/kubelet/lifecycle"
+	"k8s.io/kubernetes/pkg/kubelet/logs"
 	"k8s.io/kubernetes/pkg/kubelet/metrics"
 	"k8s.io/kubernetes/pkg/kubelet/metrics/collectors"
 	"k8s.io/kubernetes/pkg/kubelet/network"
@@ -758,6 +759,21 @@ func NewMainKubelet(kubeCfg *kubeletconfiginternal.KubeletConfiguration,
 	}
 	klet.imageManager = imageManager
 
+	if containerRuntime == kubetypes.RemoteContainerRuntime && utilfeature.DefaultFeatureGate.Enabled(features.CRIContainerLogRotation) {
+		// setup containerLogManager for CRI container runtime
+		containerLogManager, err := logs.NewContainerLogManager(
+			klet.runtimeService,
+			kubeCfg.ContainerLogMaxSize,
+			int(kubeCfg.ContainerLogMaxFiles),
+		)
+		if err != nil {
+			return nil, fmt.Errorf("failed to initialize container log manager: %v", err)
+		}
+		klet.containerLogManager = containerLogManager
+	} else {
+		klet.containerLogManager = logs.NewStubContainerLogManager()
+	}
+
 	klet.statusManager = status.NewManager(klet.kubeClient, klet.podManager, klet)
 
 	if utilfeature.DefaultFeatureGate.Enabled(features.RotateKubeletServerCertificate) && kubeDeps.TLSOptions != nil {
@@ -993,6 +1009,9 @@ type Kubelet struct {
 	// Manager for image garbage collection.
 	imageManager images.ImageGCManager
 
+	// Manager for container logs.
+	containerLogManager logs.ContainerLogManager
+
 	// Secret manager.
 	secretManager secret.Manager
 
@@ -1335,6 +1354,9 @@ func (kl *Kubelet) initializeRuntimeDependentModules() {
 		// Fail kubelet and rely on the babysitter to retry starting kubelet.
 		glog.Fatalf("Failed to start ContainerManager %v", err)
 	}
+	// container log manager must start after container runtime is up to retrieve information from container runtime
+	// and inform container to reopen log file after log rotation.
+	kl.containerLogManager.Start()
 }
 
 // Run starts the kubelet reacting to config updates

pkg/kubelet/kubelet_test.go

@@ -52,6 +52,7 @@ import (
 	"k8s.io/kubernetes/pkg/kubelet/gpu"
 	"k8s.io/kubernetes/pkg/kubelet/images"
 	"k8s.io/kubernetes/pkg/kubelet/lifecycle"
+	"k8s.io/kubernetes/pkg/kubelet/logs"
 	"k8s.io/kubernetes/pkg/kubelet/network"
 	nettest "k8s.io/kubernetes/pkg/kubelet/network/testing"
 	"k8s.io/kubernetes/pkg/kubelet/pleg"
@@ -262,6 +263,7 @@ func newTestKubeletWithImageList(
 		fakeImageService: fakeRuntime,
 		ImageGCManager:   imageGCManager,
 	}
+	kubelet.containerLogManager = logs.NewStubContainerLogManager()
 	fakeClock := clock.NewFakeClock(time.Now())
 	kubelet.backOff = flowcontrol.NewBackOff(time.Second, time.Minute)
 	kubelet.backOff.Clock = fakeClock