Merge pull request kubernetes#128031 from HirazawaUi/kep-4656

[Kubeadm] KEP-4656: Add kubelet instance configuration to configure CRI socket for each node

Author: k8s-ci-robot

cmd/kubeadm/app/cmd/phases/init/kubelet.go

@@ -23,9 +23,11 @@ import (
 
 	"k8s.io/klog/v2"
 
+	kubeletconfig "k8s.io/kubelet/config/v1beta1"
 	"k8s.io/kubernetes/cmd/kubeadm/app/cmd/options"
 	"k8s.io/kubernetes/cmd/kubeadm/app/cmd/phases/workflow"
 	cmdutil "k8s.io/kubernetes/cmd/kubeadm/app/cmd/util"
+	"k8s.io/kubernetes/cmd/kubeadm/app/features"
 	kubeletphase "k8s.io/kubernetes/cmd/kubeadm/app/phases/kubelet"
 )
 
@@ -76,6 +78,16 @@ func runKubeletStart(c workflow.RunData) error {
 		return errors.Wrap(err, "error writing a dynamic environment file for the kubelet")
 	}
 
+	// Write the instance kubelet configuration file to disk.
+	if features.Enabled(data.Cfg().FeatureGates, features.NodeLocalCRISocket) {
+		kubeletConfig := &kubeletconfig.KubeletConfiguration{
+			ContainerRuntimeEndpoint: data.Cfg().NodeRegistration.CRISocket,
+		}
+		if err := kubeletphase.WriteInstanceConfigToDisk(kubeletConfig, data.KubeletDir()); err != nil {
+			return errors.Wrap(err, "error writing instance kubelet configuration to disk")
+		}
+	}
+
 	// Write the kubelet configuration file to disk.
 	if err := kubeletphase.WriteConfigToDisk(&data.Cfg().ClusterConfiguration, data.KubeletDir(), data.PatchesDir(), data.OutputWriter()); err != nil {
 		return errors.Wrap(err, "error writing kubelet configuration to disk")

cmd/kubeadm/app/cmd/phases/init/uploadconfig.go

@@ -30,6 +30,7 @@ import (
 	"k8s.io/kubernetes/cmd/kubeadm/app/cmd/phases/workflow"
 	cmdutil "k8s.io/kubernetes/cmd/kubeadm/app/cmd/util"
 	kubeadmconstants "k8s.io/kubernetes/cmd/kubeadm/app/constants"
+	"k8s.io/kubernetes/cmd/kubeadm/app/features"
 	kubeletphase "k8s.io/kubernetes/cmd/kubeadm/app/phases/kubelet"
 	patchnodephase "k8s.io/kubernetes/cmd/kubeadm/app/phases/patchnode"
 	"k8s.io/kubernetes/cmd/kubeadm/app/phases/uploadconfig"
@@ -127,9 +128,11 @@ func runUploadKubeletConfig(c workflow.RunData) error {
 		return errors.Wrap(err, "error creating kubelet configuration ConfigMap")
 	}
 
-	klog.V(1).Infoln("[upload-config] Preserving the CRISocket information for the control-plane node")
-	if err := patchnodephase.AnnotateCRISocket(client, cfg.NodeRegistration.Name, cfg.NodeRegistration.CRISocket); err != nil {
-		return errors.Wrap(err, "Error writing Crisocket information for the control-plane node")
+	if !features.Enabled(cfg.FeatureGates, features.NodeLocalCRISocket) {
+		klog.V(1).Infoln("[upload-config] Preserving the CRISocket information for the control-plane node")
+		if err := patchnodephase.AnnotateCRISocket(client, cfg.NodeRegistration.Name, cfg.NodeRegistration.CRISocket); err != nil {
+			return errors.Wrap(err, "error writing CRISocket for this node")
+		}
 	}
 	return nil
 }

cmd/kubeadm/app/cmd/phases/join/kubelet.go

@@ -232,6 +232,16 @@ func runKubeletStartJoinPhase(c workflow.RunData) (returnErr error) {
 		fmt.Println("[kubelet-start] Would stop the kubelet")
 	}
 
+	// Write the instance kubelet configuration file to disk.
+	if features.Enabled(initCfg.FeatureGates, features.NodeLocalCRISocket) {
+		kubeletConfig := &kubeletconfig.KubeletConfiguration{
+			ContainerRuntimeEndpoint: data.Cfg().NodeRegistration.CRISocket,
+		}
+		if err := kubeletphase.WriteInstanceConfigToDisk(kubeletConfig, data.KubeletDir()); err != nil {
+			return errors.Wrap(err, "error writing instance kubelet configuration to disk")
+		}
+	}
+
 	// Write the configuration for the kubelet (using the bootstrap token credentials) to disk so the kubelet can start
 	if err := kubeletphase.WriteConfigToDisk(&initCfg.ClusterConfiguration, data.KubeletDir(), data.PatchesDir(), data.OutputWriter()); err != nil {
 		return err
@@ -323,9 +333,11 @@ func runKubeletWaitBootstrapPhase(c workflow.RunData) (returnErr error) {
 		return err
 	}
 
-	klog.V(1).Infoln("[kubelet-start] preserving the crisocket information for the node")
-	if err := patchnodephase.AnnotateCRISocket(client, cfg.NodeRegistration.Name, cfg.NodeRegistration.CRISocket); err != nil {
-		return errors.Wrap(err, "error uploading crisocket")
+	if !features.Enabled(initCfg.ClusterConfiguration.FeatureGates, features.NodeLocalCRISocket) {
+		klog.V(1).Infoln("[kubelet-start] preserving the crisocket information for the node")
+		if err := patchnodephase.AnnotateCRISocket(client, cfg.NodeRegistration.Name, cfg.NodeRegistration.CRISocket); err != nil {
+			return errors.Wrap(err, "error writing CRISocket for this node")
+		}
 	}
 
 	return nil

cmd/kubeadm/app/cmd/phases/upgrade/apply/uploadconfig.go

@@ -28,6 +28,7 @@ import (
 	kubeadmapi "k8s.io/kubernetes/cmd/kubeadm/app/apis/kubeadm"
 	"k8s.io/kubernetes/cmd/kubeadm/app/cmd/options"
 	"k8s.io/kubernetes/cmd/kubeadm/app/cmd/phases/workflow"
+	"k8s.io/kubernetes/cmd/kubeadm/app/features"
 	kubeletphase "k8s.io/kubernetes/cmd/kubeadm/app/phases/kubelet"
 	patchnodephase "k8s.io/kubernetes/cmd/kubeadm/app/phases/patchnode"
 	"k8s.io/kubernetes/cmd/kubeadm/app/phases/uploadconfig"
@@ -107,9 +108,11 @@ func runUploadKubeletConfig(c workflow.RunData) error {
 		return errors.Wrap(err, "error creating kubelet configuration ConfigMap")
 	}
 
-	klog.V(1).Infoln("[upgrade/upload-config] Preserving the CRISocket information for this control-plane node")
-	if err := patchnodephase.AnnotateCRISocket(client, cfg.NodeRegistration.Name, cfg.NodeRegistration.CRISocket); err != nil {
-		return errors.Wrap(err, "error writing Crisocket information for the control-plane node")
+	if !features.Enabled(cfg.ClusterConfiguration.FeatureGates, features.NodeLocalCRISocket) {
+		klog.V(1).Infoln("[upgrade/upload-config] Preserving the CRISocket information for this control-plane node")
+		if err := patchnodephase.AnnotateCRISocket(client, cfg.NodeRegistration.Name, cfg.NodeRegistration.CRISocket); err != nil {
+			return errors.Wrap(err, "error writing CRISocket for this node")
+		}
 	}
 
 	return nil

cmd/kubeadm/app/componentconfigs/kubelet.go

@@ -20,6 +20,7 @@ import (
 	"path/filepath"
 
 	"github.com/pkg/errors"
+
 	clientset "k8s.io/client-go/kubernetes"
 	"k8s.io/klog/v2"
 	kubeletconfig "k8s.io/kubelet/config/v1beta1"

cmd/kubeadm/app/constants/constants.go

@@ -306,6 +306,10 @@ const (
 	// This file should exist under KubeletRunDirectory
 	KubeletConfigurationFileName = "config.yaml"
 
+	// KubeletInstanceConfigurationFileName is the name of the kubelet instance configuration file written
+	// to all nodes. This file should exist under KubeletRunDirectory.
+	KubeletInstanceConfigurationFileName = "instance-config.yaml"
+
 	// KubeletEnvFileName is a file "kubeadm init" writes at runtime. Using that interface, kubeadm can customize certain
 	// kubelet flags conditionally based on the environment at runtime. Also, parameters given to the configuration file
 	// might be passed through this file. "kubeadm init" writes one variable, with the name ${KubeletEnvFileVariableName}.

cmd/kubeadm/app/features/features.go

@@ -40,6 +40,8 @@ const (
 	WaitForAllControlPlaneComponents = "WaitForAllControlPlaneComponents"
 	// ControlPlaneKubeletLocalMode is expected to be in alpha in v1.31, beta in v1.32
 	ControlPlaneKubeletLocalMode = "ControlPlaneKubeletLocalMode"
+	// NodeLocalCRISocket is expected to be in alpha in v1.32, beta in v1.33, ga in v1.35
+	NodeLocalCRISocket = "NodeLocalCRISocket"
 )
 
 // InitFeatureGates are the default feature gates for the init command
@@ -56,6 +58,7 @@ var InitFeatureGates = FeatureList{
 	EtcdLearnerMode:                  {FeatureSpec: featuregate.FeatureSpec{Default: true, PreRelease: featuregate.GA, LockToDefault: true}},
 	WaitForAllControlPlaneComponents: {FeatureSpec: featuregate.FeatureSpec{Default: false, PreRelease: featuregate.Alpha}},
 	ControlPlaneKubeletLocalMode:     {FeatureSpec: featuregate.FeatureSpec{Default: false, PreRelease: featuregate.Alpha}},
+	NodeLocalCRISocket:               {FeatureSpec: featuregate.FeatureSpec{Default: false, PreRelease: featuregate.Alpha}},
 }
 
 // Feature represents a feature being gated

cmd/kubeadm/app/phases/kubelet/config.go

@@ -27,13 +27,15 @@ import (
 	v1 "k8s.io/api/core/v1"
 	rbac "k8s.io/api/rbac/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/types"
 	clientset "k8s.io/client-go/kubernetes"
 	kubeletconfig "k8s.io/kubelet/config/v1beta1"
 	"sigs.k8s.io/yaml"
 
 	kubeadmapi "k8s.io/kubernetes/cmd/kubeadm/app/apis/kubeadm"
 	"k8s.io/kubernetes/cmd/kubeadm/app/componentconfigs"
 	kubeadmconstants "k8s.io/kubernetes/cmd/kubeadm/app/constants"
+	"k8s.io/kubernetes/cmd/kubeadm/app/features"
 	kubeadmutil "k8s.io/kubernetes/cmd/kubeadm/app/util"
 	"k8s.io/kubernetes/cmd/kubeadm/app/util/apiclient"
 	"k8s.io/kubernetes/cmd/kubeadm/app/util/patches"
@@ -66,7 +68,21 @@ func WriteConfigToDisk(cfg *kubeadmapi.ClusterConfiguration, kubeletDir, patches
 		}
 	}
 
-	return writeConfigBytesToDisk(kubeletBytes, kubeletDir)
+	if features.Enabled(cfg.FeatureGates, features.NodeLocalCRISocket) {
+		file := filepath.Join(kubeletDir, kubeadmconstants.KubeletInstanceConfigurationFileName)
+		kubeletBytes, err = applyKubeletConfigPatchFromFile(kubeletBytes, file, output)
+		if err != nil {
+			return errors.Wrapf(err, "could not apply kubelet instance configuration as a patch from %q", file)
+		}
+	}
+	return writeConfigBytesToDisk(kubeletBytes, kubeletDir, kubeadmconstants.KubeletConfigurationFileName)
+}
+
+// WriteInstanceConfigToDisk writes the container runtime endpoint configuration
+// to the instance configuration file in the specified kubelet directory.
+func WriteInstanceConfigToDisk(cfg *kubeletconfig.KubeletConfiguration, kubeletDir string) error {
+	instanceFileContent := fmt.Sprintf("containerRuntimeEndpoint: %q\n", cfg.ContainerRuntimeEndpoint)
+	return writeConfigBytesToDisk([]byte(instanceFileContent), kubeletDir, kubeadmconstants.KubeletInstanceConfigurationFileName)
 }
 
 // ApplyPatchesToConfig applies the patches located in patchesDir to the KubeletConfiguration stored
@@ -188,8 +204,8 @@ func createConfigMapRBACRules(client clientset.Interface) error {
 }
 
 // writeConfigBytesToDisk writes a byte slice down to disk at the specific location of the kubelet config file
-func writeConfigBytesToDisk(b []byte, kubeletDir string) error {
-	configFile := filepath.Join(kubeletDir, kubeadmconstants.KubeletConfigurationFileName)
+func writeConfigBytesToDisk(b []byte, kubeletDir, fileName string) error {
+	configFile := filepath.Join(kubeletDir, fileName)
 	fmt.Printf("[kubelet-start] Writing kubelet configuration to file %q\n", configFile)
 
 	// creates target folder if not already exists
@@ -198,7 +214,7 @@ func writeConfigBytesToDisk(b []byte, kubeletDir string) error {
 	}
 
 	if err := os.WriteFile(configFile, b, 0644); err != nil {
-		return errors.Wrapf(err, "failed to write kubelet configuration to the file %q", configFile)
+		return errors.Wrapf(err, "failed to write kubelet configuration file %q", configFile)
 	}
 	return nil
 }
@@ -225,3 +241,45 @@ func applyKubeletConfigPatches(kubeletBytes []byte, patchesDir string, output io
 	}
 	return kubeletBytes, nil
 }
+
+// applyKubeletConfigPatchFromFile applies a single patch file to the kubelet configuration bytes.
+func applyKubeletConfigPatchFromFile(kubeletConfigBytes []byte, patchFilePath string, output io.Writer) ([]byte, error) {
+	// Get the patch data from the file.
+	data, err := os.ReadFile(patchFilePath)
+	if err != nil {
+		return nil, errors.Wrapf(err, "could not read patch file %q", patchFilePath)
+	}
+
+	patchSet, err := patches.CreatePatchSet(patches.KubeletConfiguration, types.StrategicMergePatchType, string(data))
+	if err != nil {
+		return nil, err
+	}
+
+	patchManager := patches.NewPatchManager([]*patches.PatchSet{patchSet}, []string{patches.KubeletConfiguration}, output)
+
+	// Always convert the target data to JSON.
+	patchData, err := yaml.YAMLToJSON(kubeletConfigBytes)
+	if err != nil {
+		return nil, err
+	}
+
+	// Define the patch target.
+	patchTarget := &patches.PatchTarget{
+		Name:                      patches.KubeletConfiguration,
+		StrategicMergePatchObject: kubeletconfig.KubeletConfiguration{},
+		Data:                      patchData,
+	}
+
+	err = patchManager.ApplyPatchesToTarget(patchTarget)
+	if err != nil {
+		return nil, err
+	}
+
+	// Convert the patched data back to YAML and return it.
+	kubeletConfigBytes, err = yaml.JSONToYAML(patchTarget.Data)
+	if err != nil {
+		return nil, errors.Wrap(err, "failed to convert patched data to YAML")
+	}
+
+	return kubeletConfigBytes, nil
+}

cmd/kubeadm/app/phases/kubelet/config_test.go

@@ -24,6 +24,8 @@ import (
 	"path/filepath"
 	"testing"
 
+	"github.com/stretchr/testify/assert"
+
 	v1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
@@ -109,6 +111,74 @@ func TestApplyKubeletConfigPatches(t *testing.T) {
 	}
 }
 
+func TestApplyKubeletConfigPatchFromFile(t *testing.T) {
+	const kubeletConfigGVK = "apiVersion: kubelet.config.k8s.io/v1beta1\nkind: KubeletConfiguration\n"
+
+	tests := []struct {
+		name           string
+		kubeletConfig  []byte
+		patchContent   []byte
+		expectError    bool
+		expectedResult []byte
+	}{
+		{
+			name:           "apply new field",
+			kubeletConfig:  []byte(kubeletConfigGVK),
+			patchContent:   []byte("containerRuntimeEndpoint: unix:///run/containerd/containerd.sock"),
+			expectError:    false,
+			expectedResult: []byte("apiVersion: kubelet.config.k8s.io/v1beta1\ncontainerRuntimeEndpoint: unix:///run/containerd/containerd.sock\nkind: KubeletConfiguration\n"),
+		},
+		{
+			name:           "overwrite existing field",
+			kubeletConfig:  []byte(kubeletConfigGVK + "containerRuntimeEndpoint: unix:///run/crio/crio.sock\n"),
+			patchContent:   []byte("containerRuntimeEndpoint: unix:///run/containerd/containerd.sock"),
+			expectError:    false,
+			expectedResult: []byte("apiVersion: kubelet.config.k8s.io/v1beta1\ncontainerRuntimeEndpoint: unix:///run/containerd/containerd.sock\nkind: KubeletConfiguration\n"),
+		},
+		{
+			name:          "invalid patch contents",
+			kubeletConfig: []byte(kubeletConfigGVK),
+			patchContent:  []byte("invalid-patch-content"),
+			expectError:   true,
+		},
+		{
+			name:           "empty patch file",
+			kubeletConfig:  []byte(kubeletConfigGVK),
+			patchContent:   []byte(""),
+			expectError:    false,
+			expectedResult: []byte(kubeletConfigGVK),
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			output := io.Discard
+
+			// Create a temporary file to store the patch content.
+			patchFile, err := os.CreateTemp("", "instance-config-*.yml")
+			if err != nil {
+				t.Errorf("Error creating temporary file: %v", err)
+			}
+			defer func() {
+				_ = patchFile.Close()
+				_ = os.Remove(patchFile.Name())
+			}()
+
+			_, err = patchFile.Write(tt.patchContent)
+			if err != nil {
+				t.Errorf("Error writing instance config to file: %v", err)
+			}
+
+			// Apply the patch.
+			result, err := applyKubeletConfigPatchFromFile(tt.kubeletConfig, patchFile.Name(), output)
+			if !tt.expectError && err != nil {
+				t.Errorf("Unexpected error: %v", err)
+			}
+			assert.Equal(t, tt.expectedResult, result)
+		})
+	}
+}
+
 func TestApplyPatchesToConfig(t *testing.T) {
 	const (
 		expectedAddress = "barfoo"