feat(k8s): sidecar containers

Signed-off-by: Mike Beaumont <[email protected]>

pkg/plugins/runtime/k8s/controllers/inbound_converter.go

@@ -46,14 +46,14 @@ func inboundForService(zone string, pod *kube_core.Pod, service *kube_core.Servi
 		// to figure out which container implements which service. Since we know container we can check its status
 		// and map it to the Dataplane health
 		if container != nil {
-			if cs := util_k8s.FindContainerStatus(pod, container.Name); cs != nil && !cs.Ready {
+			if cs := util_k8s.FindContainerStatus(container.Name, pod.Status.ContainerStatuses); cs != nil && !cs.Ready {
 				state = mesh_proto.Dataplane_Networking_Inbound_NotReady
 				health.Ready = false
 			}
 		}
 
 		// also we're checking whether kuma-sidecar container is ready
-		if cs := util_k8s.FindContainerStatus(pod, util_k8s.KumaSidecarContainerName); cs != nil && !cs.Ready {
+		if cs := util_k8s.FindContainerStatus(util_k8s.KumaSidecarContainerName, pod.Status.ContainerStatuses, pod.Status.InitContainerStatuses); cs != nil && !cs.Ready {
 			state = mesh_proto.Dataplane_Networking_Inbound_NotReady
 			health.Ready = false
 		}
@@ -98,15 +98,15 @@ func inboundForServiceless(zone string, pod *kube_core.Pod, name string) *mesh_p
 
 	for _, container := range pod.Spec.Containers {
 		if container.Name != util_k8s.KumaSidecarContainerName {
-			if cs := util_k8s.FindContainerStatus(pod, container.Name); cs != nil && !cs.Ready {
+			if cs := util_k8s.FindContainerStatus(container.Name, pod.Status.ContainerStatuses); cs != nil && !cs.Ready {
 				state = mesh_proto.Dataplane_Networking_Inbound_NotReady
 				health.Ready = false
 			}
 		}
 	}
 
 	// also we're checking whether kuma-sidecar container is ready
-	if cs := util_k8s.FindContainerStatus(pod, util_k8s.KumaSidecarContainerName); cs != nil && !cs.Ready {
+	if cs := util_k8s.FindContainerStatus(util_k8s.KumaSidecarContainerName, pod.Status.ContainerStatuses, pod.Status.InitContainerStatuses); cs != nil && !cs.Ready {
 		state = mesh_proto.Dataplane_Networking_Inbound_NotReady
 		health.Ready = false
 	}

pkg/plugins/runtime/k8s/controllers/pod_status_controller.go

@@ -24,6 +24,7 @@ import (
 )
 
 // PodStatusReconciler tracks pods status changes and signals kuma-dp when it has to complete
+// but only when Kuma isn't using the SidecarContainer feature
 type PodStatusReconciler struct {
 	kube_client.Client
 	kube_record.EventRecorder

pkg/plugins/runtime/k8s/plugin.go

@@ -3,7 +3,9 @@ package k8s
 import (
 	"fmt"
 
+	"github.com/Masterminds/semver/v3"
 	"github.com/pkg/errors"
+	"k8s.io/client-go/discovery"
 	kube_ctrl "sigs.k8s.io/controller-runtime"
 	kube_webhook "sigs.k8s.io/controller-runtime/pkg/webhook"
 	kube_admission "sigs.k8s.io/controller-runtime/pkg/webhook/admission"
@@ -29,9 +31,11 @@ import (
 	"github.com/kumahq/kuma/pkg/plugins/runtime/k8s/webhooks/injector"
 )
 
-var log = core.Log.WithName("plugin").WithName("runtime").WithName("k8s")
-
-var _ core_plugins.RuntimePlugin = &plugin{}
+var (
+	log                                                = core.Log.WithName("plugin").WithName("runtime").WithName("k8s")
+	sidecarContainerVersion                            = semver.New(1, 29, 0, "", "")
+	_                       core_plugins.RuntimePlugin = &plugin{}
+)
 
 type plugin struct{}
 
@@ -302,10 +306,26 @@ func addValidators(mgr kube_ctrl.Manager, rt core_runtime.Runtime, converter k8s
 func addMutators(mgr kube_ctrl.Manager, rt core_runtime.Runtime, converter k8s_common.Converter) error {
 	if rt.Config().Mode != config_core.Global {
 		address := fmt.Sprintf("https://%s.%s:%d", rt.Config().Runtime.Kubernetes.ControlPlaneServiceName, rt.Config().Store.Kubernetes.SystemNamespace, rt.Config().DpServer.Port)
+		kubeConfig := mgr.GetConfig()
+		discClient, err := discovery.NewDiscoveryClientForConfig(kubeConfig)
+		if err != nil {
+			return err
+		}
+		k8sVersion, err := discClient.ServerVersion()
+		if err != nil {
+			return err
+		}
+		var sidecarContainersEnabled bool
+		if v, err := semver.NewVersion(
+			fmt.Sprintf("%s.%s.0", k8sVersion.Major, k8sVersion.Minor),
+		); err == nil && !v.LessThan(sidecarContainerVersion) {
+			sidecarContainersEnabled = true
+		}
 		kumaInjector, err := injector.New(
 			rt.Config().Runtime.Kubernetes.Injector,
 			address,
 			mgr.GetClient(),
+			sidecarContainersEnabled,
 			converter,
 			rt.Config().GetEnvoyAdminPort(),
 			rt.Config().Store.Kubernetes.SystemNamespace,

pkg/plugins/runtime/k8s/util/util.go

@@ -2,6 +2,7 @@ package util
 
 import (
 	"fmt"
+	"slices"
 	"sort"
 
 	"github.com/go-logr/logr"
@@ -122,8 +123,8 @@ func FindPort(pod *kube_core.Pod, svcPort *kube_core.ServicePort) (int, *kube_co
 	return 0, nil, fmt.Errorf("no suitable port for manifest: %s", pod.UID)
 }
 
-func FindContainerStatus(pod *kube_core.Pod, containerName string) *kube_core.ContainerStatus {
-	for _, cs := range pod.Status.ContainerStatuses {
+func FindContainerStatus(containerName string, status []kube_core.ContainerStatus, otherStatuses ...[]kube_core.ContainerStatus) *kube_core.ContainerStatus {
+	for _, cs := range append(status, slices.Concat(otherStatuses...)...) {
 		if cs.Name == containerName {
 			return &cs
 		}

pkg/plugins/runtime/k8s/webhooks/injector/injector.go

@@ -40,6 +40,7 @@ func New(
 	cfg runtime_k8s.Injector,
 	controlPlaneURL string,
 	client kube_client.Client,
+	sidecarContainersEnabled bool,
 	converter k8s_common.Converter,
 	envoyAdminPort uint32,
 	systemNamespace string,
@@ -53,23 +54,25 @@ func New(
 		caCert = string(bytes)
 	}
 	return &KumaInjector{
-		cfg:              cfg,
-		client:           client,
-		converter:        converter,
-		defaultAdminPort: envoyAdminPort,
+		cfg:                      cfg,
+		client:                   client,
+		sidecarContainersEnabled: sidecarContainersEnabled,
+		converter:                converter,
+		defaultAdminPort:         envoyAdminPort,
 		proxyFactory: containers.NewDataplaneProxyFactory(controlPlaneURL, caCert, envoyAdminPort,
 			cfg.SidecarContainer.DataplaneContainer, cfg.BuiltinDNS, cfg.SidecarContainer.WaitForDataplaneReady),
 		systemNamespace: systemNamespace,
 	}, nil
 }
 
 type KumaInjector struct {
-	cfg              runtime_k8s.Injector
-	client           kube_client.Client
-	converter        k8s_common.Converter
-	proxyFactory     *containers.DataplaneProxyFactory
-	defaultAdminPort uint32
-	systemNamespace  string
+	cfg                      runtime_k8s.Injector
+	client                   kube_client.Client
+	sidecarContainersEnabled bool
+	converter                k8s_common.Converter
+	proxyFactory             *containers.DataplaneProxyFactory
+	defaultAdminPort         uint32
+	systemNamespace          string
 }
 
 func (i *KumaInjector) InjectKuma(ctx context.Context, pod *kube_core.Pod) error {
@@ -121,9 +124,6 @@ func (i *KumaInjector) InjectKuma(ctx context.Context, pod *kube_core.Pod) error
 		pod.Annotations[kube_podcmd.DefaultContainerAnnotationName] = pod.Spec.Containers[0].Name
 	}
 
-	// inject sidecar as first container
-	pod.Spec.Containers = append([]kube_core.Container{patchedContainer}, pod.Spec.Containers...)
-
 	annotations, err := i.NewAnnotations(pod, meshName, logger)
 	if err != nil {
 		return errors.Wrap(err, "could not generate annotations for pod")
@@ -172,6 +172,15 @@ func (i *KumaInjector) InjectKuma(ctx context.Context, pod *kube_core.Pod) error
 		}
 	}
 
+	if i.sidecarContainersEnabled {
+		// inject sidecar after init
+		patchedContainer.RestartPolicy = pointer.To(kube_core.ContainerRestartPolicyAlways)
+		pod.Spec.InitContainers = append(pod.Spec.InitContainers, patchedContainer)
+	} else {
+		// inject sidecar as first container
+		pod.Spec.Containers = append([]kube_core.Container{patchedContainer}, pod.Spec.Containers...)
+	}
+
 	if err := i.overrideHTTPProbes(pod); err != nil {
 		return err
 	}

pkg/plugins/runtime/k8s/webhooks/injector/injector_test.go

@@ -32,10 +32,11 @@ var _ = Describe("Injector", func() {
 	)
 
 	type testCase struct {
-		num       string
-		mesh      string
-		cfgFile   string
-		namespace string
+		num                      string
+		mesh                     string
+		cfgFile                  string
+		namespace                string
+		sidecarContainersEnabled bool
 	}
 
 	BeforeAll(func() {
@@ -85,7 +86,7 @@ spec:
 			var cfg conf.Injector
 			Expect(config.Load(filepath.Join("testdata", given.cfgFile), &cfg)).To(Succeed())
 			cfg.CaCertFile = caCertPath
-			injector, err := inject.New(cfg, "http://kuma-control-plane.kuma-system:5681", k8sClient, k8s.NewSimpleConverter(), 9901, systemNamespace)
+			injector, err := inject.New(cfg, "http://kuma-control-plane.kuma-system:5681", k8sClient, given.sidecarContainersEnabled, k8s.NewSimpleConverter(), 9901, systemNamespace)
 			Expect(err).ToNot(HaveOccurred())
 
 			// and create mesh
@@ -682,7 +683,7 @@ spec:
 			var cfg conf.Injector
 			Expect(config.Load(filepath.Join("testdata", given.cfgFile), &cfg)).To(Succeed())
 			cfg.CaCertFile = caCertPath
-			injector, err := inject.New(cfg, "http://kuma-control-plane.kuma-system:5681", k8sClient, k8s.NewSimpleConverter(), 9901, systemNamespace)
+			injector, err := inject.New(cfg, "http://kuma-control-plane.kuma-system:5681", k8sClient, given.sidecarContainersEnabled, k8s.NewSimpleConverter(), 9901, systemNamespace)
 			Expect(err).ToNot(HaveOccurred())
 
 			// and create mesh
@@ -788,7 +789,7 @@ spec:
 			var cfg conf.Injector
 			Expect(config.Load(filepath.Join("testdata", given.cfgFile), &cfg)).To(Succeed())
 			cfg.CaCertFile = caCertPath
-			injector, err := inject.New(cfg, "http://kuma-control-plane.kuma-system:5681", k8sClient, k8s.NewSimpleConverter(), 9901, systemNamespace)
+			injector, err := inject.New(cfg, "http://kuma-control-plane.kuma-system:5681", k8sClient, given.sidecarContainersEnabled, k8s.NewSimpleConverter(), 9901, systemNamespace)
 			Expect(err).ToNot(HaveOccurred())
 
 			// and create mesh

pkg/plugins/runtime/k8s/webhooks/injector/precheck.go

@@ -3,6 +3,7 @@ package injector
 import (
 	"context"
 	"fmt"
+	"slices"
 
 	"github.com/go-logr/logr"
 	"github.com/pkg/errors"
@@ -87,7 +88,7 @@ func (i *KumaInjector) needToInject(pod *kube_core.Pod, ns *kube_core.Namespace)
 		return false, nil
 	}
 
-	for _, container := range pod.Spec.Containers {
+	for _, container := range slices.Concat(pod.Spec.Containers, pod.Spec.InitContainers) {
 		if container.Name == k8s_util.KumaSidecarContainerName {
 			log.V(1).Info("pod already has Kuma sidecar")
 			return false, nil