feat(*) statefulset support (#901)

* fix(kuma-cp) statefulset support

Signed-off-by: Ilya Lobkov <ilya.lobkov@konghq.com>

Makefile.e2e.mk

@@ -136,7 +136,7 @@ define verify_example_outbound
 	@echo "Checking number of Outbound requests via Envoy ..."
 	test $$( $(1) \
 		wget -qO- http://localhost:9901/stats/prometheus | \
-		grep 'envoy_cluster_upstream_rq_total{envoy_cluster_name="pass_through"}' | \
+		grep 'envoy_cluster_upstream_rq_total{envoy_cluster_name="outbound_passthrough"}' | \
 		awk '{print $$2}' | tr -d [:space:] \
 	) -ge 1
 	@echo "Check passed!"

api/mesh/v1alpha1/dataplane.proto

@@ -91,11 +91,16 @@ message Dataplane {
     // TransparentProxying describes configuration for transparent proxying.
     message TransparentProxying {
 
-      // Port on which all traffic is being transparently redirected.
-      uint32 redirect_port = 1 [ (validate.rules).uint32 = {lte : 65535} ];
+      // Port on which all inbound traffic is being transparently redirected.
+      uint32 redirect_port_inbound = 1
+          [ (validate.rules).uint32 = {lte : 65535} ];
+
+      // Port on which all outbound traffic is being transparently redirected.
+      uint32 redirect_port_outbound = 2
+          [ (validate.rules).uint32 = {lte : 65535} ];
 
       // List of services that will be access directly via IP:PORT
-      repeated string direct_access_services = 2;
+      repeated string direct_access_services = 3;
     }
 
     // Gateway describes configuration of gateway of the dataplane.

api/mesh/v1alpha1/dataplane_helpers.go

@@ -18,6 +18,8 @@ const (
 	// Optional tag that has a reserved meaning in Kuma.
 	// If absent, Kuma will treat application's protocol as opaque TCP.
 	ProtocolTag = "protocol"
+	// InstanceTag is set only for Dataplanes that implements headless services
+	InstanceTag = "instance"
 )
 
 type InboundInterface struct {

pkg/api-server/config_ws_test.go

@@ -173,7 +173,8 @@ var _ = Describe("Config WS", func() {
                     "successThreshold": 1,
                     "timeoutSeconds": 3
                   },
-                  "redirectPort": 15001,
+                  "redirectPortInbound": 15006,
+                  "redirectPortOutbound": 15001,
                   "resources": {
                     "limits": {
                       "cpu": "1000m",

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

@@ -186,7 +186,8 @@ runtime:
       cniEnabled: false
       sidecarContainer:
         image: kuma/kuma-dp:latest
-        redirectPort: 15001
+        redirectPortInbound: 15006
+        redirectPortOutbound: 15001
         uid: 5678
         gid: 5678
         adminPort: 9901

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

@@ -18,12 +18,13 @@ func DefaultKubernetesRuntimeConfig() *KubernetesRuntimeConfig {
 		Injector: Injector{
 			CNIEnabled: false,
 			SidecarContainer: SidecarContainer{
-				Image:        "kuma/kuma-dp:latest",
-				RedirectPort: 15001,
-				UID:          5678,
-				GID:          5678,
-				AdminPort:    9901,
-				DrainTime:    30 * time.Second,
+				Image:                "kuma/kuma-dp:latest",
+				RedirectPortInbound:  15006,
+				RedirectPortOutbound: 15001,
+				UID:                  5678,
+				GID:                  5678,
+				AdminPort:            9901,
+				DrainTime:            30 * time.Second,
 
 				ReadinessProbe: SidecarReadinessProbe{
 					InitialDelaySeconds: 1,
@@ -90,8 +91,10 @@ type Injector struct {
 type SidecarContainer struct {
 	// Image name.
 	Image string `yaml:"image,omitempty" envconfig:"kuma_runtime_kubernetes_injector_sidecar_container_image"`
-	// Redirect port.
-	RedirectPort uint32 `yaml:"redirectPort,omitempty" envconfig:"kuma_runtime_kubernetes_injector_sidecar_container_redirect_port"`
+	// Redirect port for inbound traffic.
+	RedirectPortInbound uint32 `yaml:"redirectPortInbound,omitempty" envconfig:"kuma_runtime_kubernetes_injector_sidecar_container_redirect_port_inbound"`
+	// Redirect port for outbound traffic.
+	RedirectPortOutbound uint32 `yaml:"redirectPortOutbound,omitempty" envconfig:"kuma_runtime_kubernetes_injector_sidecar_container_redirect_port_outbound"`
 	// User ID.
 	UID int64 `yaml:"uid,omitempty" envconfig:"kuma_runtime_kubernetes_injector_sidecar_container_uid"`
 	// Group ID.
@@ -223,8 +226,11 @@ func (c *SidecarContainer) Validate() (errs error) {
 	if c.Image == "" {
 		errs = multierr.Append(errs, errors.Errorf(".Image must be non-empty"))
 	}
-	if 65535 < c.RedirectPort {
-		errs = multierr.Append(errs, errors.Errorf(".RedirectPort must be in the range [0, 65535]"))
+	if 65535 < c.RedirectPortInbound {
+		errs = multierr.Append(errs, errors.Errorf(".RedirectPortInbound must be in the range [0, 65535]"))
+	}
+	if 65535 < c.RedirectPortOutbound {
+		errs = multierr.Append(errs, errors.Errorf(".RedirectPortOutbound must be in the range [0, 65535]"))
 	}
 	if 65535 < c.AdminPort {
 		errs = multierr.Append(errs, errors.Errorf(".AdminPort must be in the range [0, 65535]"))

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

@@ -30,7 +30,8 @@ var _ = Describe("Config", func() {
 		Expect(cfg.AdmissionServer.CertDir).To(Equal("/var/secret/kuma-cp"))
 		// and
 		Expect(cfg.Injector.SidecarContainer.Image).To(Equal("kuma-sidecar:latest"))
-		Expect(cfg.Injector.SidecarContainer.RedirectPort).To(Equal(uint32(1234)))
+		Expect(cfg.Injector.SidecarContainer.RedirectPortOutbound).To(Equal(uint32(1234)))
+		Expect(cfg.Injector.SidecarContainer.RedirectPortInbound).To(Equal(uint32(1236)))
 		Expect(cfg.Injector.SidecarContainer.UID).To(Equal(int64(2345)))
 		Expect(cfg.Injector.SidecarContainer.GID).To(Equal(int64(3456)))
 		Expect(cfg.Injector.SidecarContainer.AdminPort).To(Equal(uint32(45678)))
@@ -81,6 +82,6 @@ var _ = Describe("Config", func() {
 		err := config.Load(filepath.Join("testdata", "invalid-config.input.yaml"), &cfg)
 
 		// then
-		Expect(err.Error()).To(Equal(`Invalid configuration: .AdmissionServer is not valid: .Port must be in the range [0, 65535]; .CertDir should not be empty; .Injector is not valid: .SidecarContainer is not valid: .Image must be non-empty; .RedirectPort must be in the range [0, 65535]; .AdminPort must be in the range [0, 65535]; .DrainTime must be positive; .ReadinessProbe is not valid: .InitialDelaySeconds must be >= 1; .TimeoutSeconds must be >= 1; .PeriodSeconds must be >= 1; .SuccessThreshold must be >= 1; .FailureThreshold must be >= 1; .LivenessProbe is not valid: .InitialDelaySeconds must be >= 1; .TimeoutSeconds must be >= 1; .PeriodSeconds must be >= 1; .FailureThreshold must be >= 1; .Resources is not valid: .Requests is not valid: .CPU is not valid: quantities must match the regular expression '^([+-]?[0-9.]+)([eEinumkKMGTP]*[-+]?[0-9]*)$'; .Memory is not valid: quantities must match the regular expression '^([+-]?[0-9.]+)([eEinumkKMGTP]*[-+]?[0-9]*)$'; .Limits is not valid: .CPU is not valid: quantities must match the regular expression '^([+-]?[0-9.]+)([eEinumkKMGTP]*[-+]?[0-9]*)$'; .Memory is not valid: quantities must match the regular expression '^([+-]?[0-9.]+)([eEinumkKMGTP]*[-+]?[0-9]*)$'; .InitContainer is not valid: .Image must be non-empty`))
+		Expect(err.Error()).To(Equal(`Invalid configuration: .AdmissionServer is not valid: .Port must be in the range [0, 65535]; .CertDir should not be empty; .Injector is not valid: .SidecarContainer is not valid: .Image must be non-empty; .RedirectPortInbound must be in the range [0, 65535]; .RedirectPortOutbound must be in the range [0, 65535]; .AdminPort must be in the range [0, 65535]; .DrainTime must be positive; .ReadinessProbe is not valid: .InitialDelaySeconds must be >= 1; .TimeoutSeconds must be >= 1; .PeriodSeconds must be >= 1; .SuccessThreshold must be >= 1; .FailureThreshold must be >= 1; .LivenessProbe is not valid: .InitialDelaySeconds must be >= 1; .TimeoutSeconds must be >= 1; .PeriodSeconds must be >= 1; .FailureThreshold must be >= 1; .Resources is not valid: .Requests is not valid: .CPU is not valid: quantities must match the regular expression '^([+-]?[0-9.]+)([eEinumkKMGTP]*[-+]?[0-9]*)$'; .Memory is not valid: quantities must match the regular expression '^([+-]?[0-9.]+)([eEinumkKMGTP]*[-+]?[0-9]*)$'; .Limits is not valid: .CPU is not valid: quantities must match the regular expression '^([+-]?[0-9.]+)([eEinumkKMGTP]*[-+]?[0-9]*)$'; .Memory is not valid: quantities must match the regular expression '^([+-]?[0-9.]+)([eEinumkKMGTP]*[-+]?[0-9]*)$'; .InitContainer is not valid: .Image must be non-empty`))
 	})
 })

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

@@ -6,7 +6,8 @@ injector:
   cniEnabled: false
   sidecarContainer:
     image: kuma/kuma-dp:latest
-    redirectPort: 15001
+    redirectPortOutbound: 15001
+    redirectPortInbound: 15006
     uid: 5678
     gid: 5678
     adminPort: 9901

pkg/config/plugins/runtime/k8s/testdata/invalid-config.input.yaml

@@ -5,7 +5,8 @@ admissionServer:
 injector:
   sidecarContainer:
     image:
-    redirectPort: 423456
+    redirectPortOutbound: 423456
+    redirectPortInbound: 423457
     uid: -1
     gid: -2
     adminPort: 523456

pkg/config/plugins/runtime/k8s/testdata/valid-config.input.yaml

@@ -6,7 +6,8 @@ injector:
   cniEnabled: true
   sidecarContainer:
     image: kuma-sidecar:latest
-    redirectPort: 1234
+    redirectPortOutbound: 1234
+    redirectPortInbound: 1236
     uid: 2345
     gid: 3456
     adminPort: 45678

pkg/plugins/discovery/k8s/controllers/endpoints.go

@@ -9,8 +9,9 @@ import (
 )
 
 type Endpoint struct {
-	Address string
-	Port    uint32
+	Address  string
+	Port     uint32
+	Instance string
 }
 
 type EndpointsByService map[string][]Endpoint
@@ -41,7 +42,8 @@ func endpointsByService(dataplanes []*mesh_k8s.Dataplane) EndpointsByService {
 				continue
 			}
 			endpoint := Endpoint{
-				Port: inbound.Port,
+				Port:     inbound.Port,
+				Instance: inbound.GetTags()[mesh_proto.InstanceTag],
 			}
 			if inbound.Address != "" {
 				endpoint.Address = inbound.Address