feat(dataplane): ignored listeners with ignored labels in selector (#…

…8463)

Signed-off-by: Jakub Dyszkiewicz <jakub.dyszkiewicz@gmail.com>

api/mesh/v1alpha1/dataplane.proto

@@ -146,6 +146,19 @@ message Dataplane {
       // See https://kuma.io/docs/latest/documentation/health for more
       // information.
       ServiceProbe serviceProbe = 8;
+
+      enum State {
+        // Inbound is ready to serve the traffic.
+        Ready = 0;
+        // Inbound is not ready to serve the traffic.
+        NotReady = 1;
+        // Inbound is not created. It cannot be targeted by policies.
+        // However, a data plane proxy receives a certificate with identity of this inbound.
+        Ignored = 2;
+      }
+
+      // State describes the current state of the listener.
+      State state = 9;
     }
 
     // Outbound describes a service consumed by the data plane proxy.

api/mesh/v1alpha1/dataplane_helpers.go

@@ -186,6 +186,9 @@ func (n *Dataplane_Networking) ToInboundInterface(inbound *Dataplane_Networking_
 func (n *Dataplane_Networking) GetHealthyInbounds() []*Dataplane_Networking_Inbound {
 	var inbounds []*Dataplane_Networking_Inbound
 	for _, inbound := range n.GetInbound() {
+		if inbound.GetState() != Dataplane_Networking_Inbound_Ready {
+			continue
+		}
 		if inbound.Health != nil && !inbound.Health.Ready {
 			continue
 		}

api/mesh/v1alpha1/dataplane_helpers_test.go

@@ -143,7 +143,6 @@ var _ = Describe("Dataplane_Networking", func() {
 			networking := &Dataplane_Networking{
 				Inbound: []*Dataplane_Networking_Inbound{
 					{
-						Health:      nil,
 						Port:        8080,
 						ServicePort: 80,
 					},
@@ -152,26 +151,47 @@ var _ = Describe("Dataplane_Networking", func() {
 						Port:        9090,
 						ServicePort: 90,
 					},
+					{
+						State:       Dataplane_Networking_Inbound_Ready,
+						Port:        9091,
+						ServicePort: 91,
+					},
 					{
 						Health:      &Dataplane_Networking_Inbound_Health{Ready: false},
 						Port:        7070,
 						ServicePort: 70,
 					},
+					{
+						State:       Dataplane_Networking_Inbound_NotReady,
+						Port:        9092,
+						ServicePort: 92,
+					},
+					{
+						State:       Dataplane_Networking_Inbound_Ignored,
+						Port:        9093,
+						ServicePort: 93,
+					},
 				},
 			}
 
 			actual := networking.GetHealthyInbounds()
-			Expect(actual).To(HaveLen(2))
+			Expect(actual).To(HaveLen(3))
 			Expect(actual).To(ConsistOf(
 				&Dataplane_Networking_Inbound{
 					Health:      &Dataplane_Networking_Inbound_Health{Ready: true},
 					Port:        9090,
 					ServicePort: 90,
 				},
+				&Dataplane_Networking_Inbound{
+					State:       Dataplane_Networking_Inbound_Ready,
+					Port:        9091,
+					ServicePort: 91,
+				},
 				&Dataplane_Networking_Inbound{
 					Port:        8080,
 					ServicePort: 80,
-				}))
+				}),
+			)
 		})
 	})
 })

docs/generated/kuma-cp.md

@@ -369,6 +369,11 @@ runtime:
         tcAttachIface: "" # ENV: KUMA_RUNTIME_KUBERNETES_INJECTOR_EBPF_TC_ATTACH_IFACE
         # Path where compiled eBPF programs are placed
         programsSourcePath: /kuma/ebpf # ENV: KUMA_RUNTIME_KUBERNETES_INJECTOR_EBPF_PROGRAMS_SOURCE_PATH
+      # IgnoredServiceSelectorLabels defines a list ignored labels in Service selector.
+      # If Pod matches a Service with ignored labels, but does not match it fully, it gets Ignored inbound.
+      # It is useful when you change Service selector and expect traffic to be sent immediately.
+      # An example of this is ArgoCD's BlueGreen deployment and "rollouts-pod-template-hash" selector.
+      ignoredServiceSelectorLabels: [] # ENV: KUMA_RUNTIME_KUBERNETES_INJECTOR_IGNORED_SERVICE_SELECTOR_LABELS
     marshalingCacheExpirationTime: 5m # ENV: KUMA_RUNTIME_KUBERNETES_MARSHALING_CACHE_EXPIRATION_TIME
     # Kubernetes's resources reconciliation concurrency configuration
     controllersConcurrency:

docs/generated/raw/kuma-cp.yaml

@@ -366,6 +366,11 @@ runtime:
         tcAttachIface: "" # ENV: KUMA_RUNTIME_KUBERNETES_INJECTOR_EBPF_TC_ATTACH_IFACE
         # Path where compiled eBPF programs are placed
         programsSourcePath: /kuma/ebpf # ENV: KUMA_RUNTIME_KUBERNETES_INJECTOR_EBPF_PROGRAMS_SOURCE_PATH
+      # IgnoredServiceSelectorLabels defines a list ignored labels in Service selector.
+      # If Pod matches a Service with ignored labels, but does not match it fully, it gets Ignored inbound.
+      # It is useful when you change Service selector and expect traffic to be sent immediately.
+      # An example of this is ArgoCD's BlueGreen deployment and "rollouts-pod-template-hash" selector.
+      ignoredServiceSelectorLabels: [] # ENV: KUMA_RUNTIME_KUBERNETES_INJECTOR_IGNORED_SERVICE_SELECTOR_LABELS
     marshalingCacheExpirationTime: 5m # ENV: KUMA_RUNTIME_KUBERNETES_MARSHALING_CACHE_EXPIRATION_TIME
     # Kubernetes's resources reconciliation concurrency configuration
     controllersConcurrency:

docs/generated/raw/protos/Dataplane.json

@@ -136,6 +136,25 @@
                     "$ref": "#/definitions/kuma.mesh.v1alpha1.Dataplane.Networking.Inbound.ServiceProbe",
                     "additionalProperties": true,
                     "description": "ServiceProbe defines parameters for probing the service next to sidecar. When service probe is defined, Envoy will periodically health check the application next to it and report the status to the control plane. On Kubernetes, Kuma deployments rely on Kubernetes probes so this is not used. See https://kuma.io/docs/latest/documentation/health for more information."
+                },
+                "state": {
+                    "enum": [
+                        "Ready",
+                        0,
+                        "NotReady",
+                        1,
+                        "Ignored",
+                        2
+                    ],
+                    "oneOf": [
+                        {
+                            "type": "string"
+                        },
+                        {
+                            "type": "integer"
+                        }
+                    ],
+                    "title": "State"
                 }
             },
             "additionalProperties": true,

docs/generated/raw/protos/DataplaneOverview.json

@@ -152,6 +152,25 @@
                     "$ref": "#/definitions/kuma.mesh.v1alpha1.Dataplane.Networking.Inbound.ServiceProbe",
                     "additionalProperties": true,
                     "description": "ServiceProbe defines parameters for probing the service next to sidecar. When service probe is defined, Envoy will periodically health check the application next to it and report the status to the control plane. On Kubernetes, Kuma deployments rely on Kubernetes probes so this is not used. See https://kuma.io/docs/latest/documentation/health for more information."
+                },
+                "state": {
+                    "enum": [
+                        "Ready",
+                        0,
+                        "NotReady",
+                        1,
+                        "Ignored",
+                        2
+                    ],
+                    "oneOf": [
+                        {
+                            "type": "string"
+                        },
+                        {
+                            "type": "integer"
+                        }
+                    ],
+                    "title": "State"
                 }
             },
             "additionalProperties": true,

docs/generated/resources/proxy_dataplane.md

@@ -141,6 +141,16 @@
             - `tcp` (optional)
 
                 Tcp checker tries to establish tcp connection with destination    
+
+        - `state` (optional, enum)
+
+            State describes the current state of the listener.
+
+            - `Ready`
+
+            - `NotReady`
+
+            - `Ignored`    
 
     - `outbound` (optional, repeated)
 

pkg/config/app/kuma-cp/kuma-cp.defaults.yaml

@@ -366,6 +366,11 @@ runtime:
         tcAttachIface: "" # ENV: KUMA_RUNTIME_KUBERNETES_INJECTOR_EBPF_TC_ATTACH_IFACE
         # Path where compiled eBPF programs are placed
         programsSourcePath: /kuma/ebpf # ENV: KUMA_RUNTIME_KUBERNETES_INJECTOR_EBPF_PROGRAMS_SOURCE_PATH
+      # IgnoredServiceSelectorLabels defines a list ignored labels in Service selector.
+      # If Pod matches a Service with ignored labels, but does not match it fully, it gets Ignored inbound.
+      # It is useful when you change Service selector and expect traffic to be sent immediately.
+      # An example of this is ArgoCD's BlueGreen deployment and "rollouts-pod-template-hash" selector.
+      ignoredServiceSelectorLabels: [] # ENV: KUMA_RUNTIME_KUBERNETES_INJECTOR_IGNORED_SERVICE_SELECTOR_LABELS
     marshalingCacheExpirationTime: 5m # ENV: KUMA_RUNTIME_KUBERNETES_MARSHALING_CACHE_EXPIRATION_TIME
     # Kubernetes's resources reconciliation concurrency configuration
     controllersConcurrency:

pkg/config/loader_test.go

@@ -217,6 +217,7 @@ var _ = Describe("Config loader", func() {
 			Expect(cfg.Runtime.Kubernetes.Injector.EBPF.CgroupPath).To(Equal("/faz/daz/zaz"))
 			Expect(cfg.Runtime.Kubernetes.Injector.EBPF.TCAttachIface).To(Equal("veth1"))
 			Expect(cfg.Runtime.Kubernetes.Injector.EBPF.ProgramsSourcePath).To(Equal("/kuma/baz"))
+			Expect(cfg.Runtime.Kubernetes.Injector.IgnoredServiceSelectorLabels).To(Equal([]string{"x", "y"}))
 			Expect(cfg.Runtime.Kubernetes.NodeTaintController.CniNamespace).To(Equal("kuma-system"))
 			Expect(cfg.Runtime.Kubernetes.ControllersConcurrency.PodController).To(Equal(10))
 			Expect(cfg.Runtime.Kubernetes.ClientConfig.Qps).To(Equal(100))
@@ -534,6 +535,7 @@ runtime:
         cgroupPath: /faz/daz/zaz
         tcAttachIface: veth1
         programsSourcePath: /kuma/baz
+      ignoredServiceSelectorLabels: ["x", "y"]
     controllersConcurrency: 
       podController: 10
     clientConfig:
@@ -839,6 +841,7 @@ tracing:
 				"KUMA_RUNTIME_KUBERNETES_INJECTOR_EBPF_CGROUP_PATH":                                        "/faz/daz/zaz",
 				"KUMA_RUNTIME_KUBERNETES_INJECTOR_EBPF_TC_ATTACH_IFACE":                                    "veth1",
 				"KUMA_RUNTIME_KUBERNETES_INJECTOR_EBPF_PROGRAMS_SOURCE_PATH":                               "/kuma/baz",
+				"KUMA_RUNTIME_KUBERNETES_INJECTOR_IGNORED_SERVICE_SELECTOR_LABELS":                         "x,y",
 				"KUMA_RUNTIME_KUBERNETES_VIRTUAL_PROBES_ENABLED":                                           "false",
 				"KUMA_RUNTIME_KUBERNETES_VIRTUAL_PROBES_PORT":                                              "1111",
 				"KUMA_RUNTIME_KUBERNETES_EXCEPTIONS_LABELS":                                                "openshift.io/build.name:value1,openshift.io/deployer-pod-for.name:value2",

pkg/config/plugins/runtime/k8s/config.go

@@ -90,6 +90,7 @@ func DefaultKubernetesRuntimeConfig() *KubernetesRuntimeConfig {
 				CgroupPath:           "/sys/fs/cgroup",
 				ProgramsSourcePath:   "/kuma/ebpf",
 			},
+			IgnoredServiceSelectorLabels: []string{},
 		},
 		MarshalingCacheExpirationTime: config_types.Duration{Duration: 5 * time.Minute},
 		NodeTaintController: NodeTaintController{
@@ -208,6 +209,11 @@ type Injector struct {
 	// EBPF is a configuration for ebpf if transparent proxy should be installed
 	// using ebpf instead of iptables
 	EBPF EBPF `json:"ebpf"`
+	// IgnoredServiceSelectorLabels defines a list ignored labels in Service selector.
+	// If Pod matches a Service with ignored labels, but does not match it fully, it gets Ignored inbound.
+	// It is useful when you change Service selector and expect traffic to be sent immediately.
+	// An example of this is ArgoCD's BlueGreen deployment and "rollouts-pod-template-hash" selector.
+	IgnoredServiceSelectorLabels []string `json:"ignoredServiceSelectorLabels" envconfig:"KUMA_RUNTIME_KUBERNETES_INJECTOR_IGNORED_SERVICE_SELECTOR_LABELS"`
 }
 
 // Exceptions defines list of exceptions for Kuma injection

pkg/config/plugins/runtime/k8s/testdata/default-config.golden.yaml

@@ -25,6 +25,7 @@ injector:
     labels:
       openshift.io/build.name: '*'
       openshift.io/deployer-pod-for.name: '*'
+  ignoredServiceSelectorLabels: []
   initContainer:
     image: kuma/kuma-init:latest
   sidecarContainer:

pkg/core/managers/apis/dataplane/dataplane_manager.go

@@ -112,8 +112,12 @@ func (m *dataplaneManager) setGatewayClusterTag(dp *core_mesh.DataplaneResource)
 
 func (m *dataplaneManager) setHealth(dp *core_mesh.DataplaneResource) {
 	for _, inbound := range dp.Spec.Networking.Inbound {
-		if inbound.ServiceProbe != nil && inbound.Health == nil {
-			inbound.Health = &mesh_proto.Dataplane_Networking_Inbound_Health{Ready: false}
+		if inbound.ServiceProbe != nil {
+			inbound.State = mesh_proto.Dataplane_Networking_Inbound_NotReady
+			// write health for backwards compatibility with Kuma 2.5 and older
+			inbound.Health = &mesh_proto.Dataplane_Networking_Inbound_Health{
+				Ready: false,
+			}
 		}
 	}
 }

pkg/core/managers/apis/dataplane/dataplane_manager_test.go

@@ -214,7 +214,6 @@ var _ = Describe("Dataplane Manager", func() {
 		actual := core_mesh.NewDataplaneResource()
 		err = s.Get(context.Background(), actual, store.GetByKey("dp1", "default"))
 		Expect(err).ToNot(HaveOccurred())
-		Expect(actual.Spec.Networking.Inbound[0].Health).ToNot(BeNil())
-		Expect(actual.Spec.Networking.Inbound[0].Health.Ready).To(BeFalse())
+		Expect(actual.Spec.Networking.Inbound[0].State).To(Equal(mesh_proto.Dataplane_Networking_Inbound_NotReady))
 	})
 })

pkg/core/resources/apis/mesh/dataplane_overview_helpers.go

@@ -24,7 +24,7 @@ func (t *DataplaneOverviewResource) GetStatus() (Status, []string) {
 	var errs []string
 
 	for _, inbound := range t.Spec.Dataplane.Networking.Inbound {
-		if inbound.Health != nil && !inbound.Health.Ready {
+		if (inbound.Health != nil && !inbound.Health.Ready) || inbound.State == mesh_proto.Dataplane_Networking_Inbound_NotReady {
 			errs = append(errs, fmt.Sprintf("inbound[port=%d,svc=%s] is not ready", inbound.Port, inbound.Tags[mesh_proto.ServiceTag]))
 		}
 	}

pkg/hds/tracker/callbacks.go

@@ -208,9 +208,18 @@ func (t *tracker) updateDataplane(streamID xds.StreamID, healthMap map[uint32]bo
 		} else {
 			workloadHealth = envoyHealth
 		}
-		if inbound.Health == nil || inbound.Health.Ready != workloadHealth {
+		if workloadHealth && inbound.State == mesh_proto.Dataplane_Networking_Inbound_NotReady {
+			inbound.State = mesh_proto.Dataplane_Networking_Inbound_Ready
+			// write health for backwards compatibility with Kuma 2.5 and older
 			inbound.Health = &mesh_proto.Dataplane_Networking_Inbound_Health{
-				Ready: workloadHealth,
+				Ready: true,
+			}
+			changed = true
+		} else if !workloadHealth && inbound.State == mesh_proto.Dataplane_Networking_Inbound_Ready {
+			inbound.State = mesh_proto.Dataplane_Networking_Inbound_NotReady
+			// write health for backwards compatibility with Kuma 2.5 and older
+			inbound.Health = &mesh_proto.Dataplane_Networking_Inbound_Health{
+				Ready: false,
 			}
 			changed = true
 		}

pkg/insights/resyncer.go

@@ -477,6 +477,9 @@ func (r *resyncer) createOrUpdateServiceInsight(
 		}
 
 		for _, inbound := range networking.GetInbound() {
+			if inbound.State == mesh_proto.Dataplane_Networking_Inbound_Ignored {
+				continue
+			}
 			// address port is empty to save space in the resource. It will be filled by the server on API response
 			populateInsight(mesh_proto.ServiceInsight_Service_internal, insight, inbound.GetService(), status, backend, "")
 		}
@@ -610,6 +613,9 @@ func (r *resyncer) createOrUpdateMeshInsight(
 		}
 
 		for _, inbound := range networking.GetInbound() {
+			if inbound.State == mesh_proto.Dataplane_Networking_Inbound_Ignored {
+				continue
+			}
 			internalServices[inbound.GetService()] = struct{}{}
 		}
 	}

pkg/plugins/policies/core/matchers/dataplane.go

@@ -153,6 +153,9 @@ func resolveMeshHTTPRouteRef(refMeta core_model.ResourceMeta, refName string, mh
 func inboundsSelectedByTags(tags map[string]string, dpp *core_mesh.DataplaneResource, gateway *core_mesh.MeshGatewayResource) []core_rules.InboundListener {
 	result := []core_rules.InboundListener{}
 	for _, inbound := range dpp.Spec.GetNetworking().GetInbound() {
+		if inbound.State == mesh_proto.Dataplane_Networking_Inbound_Ignored {
+			continue
+		}
 		if mesh_proto.TagSelector(tags).Matches(inbound.Tags) {
 			intf := dpp.Spec.GetNetworking().ToInboundInterface(inbound)
 			result = append(result, core_rules.InboundListener{

pkg/plugins/policies/core/matchers/testdata/matchedpolicies/dataplanepolicies/07.dataplane.yaml

@@ -0,0 +1,11 @@
+type: Dataplane
+mesh: mesh-1
+name: dp-1
+networking:
+  address: 1.1.1.1
+  inbound:
+    - port: 8080
+      tags:
+        kuma.io/service: web
+        version: v1
+      state: Ignored

pkg/plugins/policies/core/matchers/testdata/matchedpolicies/dataplanepolicies/07.golden.yaml

@@ -0,0 +1,3 @@
+items: []
+next: null
+total: 0

pkg/plugins/policies/core/matchers/testdata/matchedpolicies/dataplanepolicies/07.policies.yaml

@@ -0,0 +1,15 @@
+# 07. policies do not select anything because inbound is ignored
+type: MeshTrafficPermission
+mesh: mesh-1
+name: mtp-1
+spec:
+  targetRef:
+    kind: MeshServiceSubset
+    name: web
+    tags:
+      version: v1
+  from:
+    - targetRef:
+        kind: Mesh
+      default:
+        action: Allow