kuma-cp: support <port>.service.kuma.io/protocol annotation on k8s as a way for users to indicate protocol of a service

CHANGELOG.md

@@ -4,6 +4,8 @@
 
 Changes:
 
+* feature: support `<port>.service.kuma.io/protocol` annotation on k8s as a way for users to indicate protocol of a service
+  [#575](https://github.com/Kong/kuma/pull/575)
 * feature: generate HTTP-specific inbound listeners for services tagged with `protocol: http`
   [#574](https://github.com/Kong/kuma/pull/574)
 * feature: support IPv6 in Dataplane resource

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

@@ -81,6 +81,9 @@ var _ = Describe("PodReconciler", func() {
 				ObjectMeta: kube_meta.ObjectMeta{
 					Namespace: "demo",
 					Name:      "example",
+					Annotations: map[string]string{
+						"80.service.kuma.io/protocol": "http",
+					},
 				},
 				Spec: kube_core.ServiceSpec{
 					Ports: []kube_core.ServicePort{
@@ -247,9 +250,11 @@ var _ = Describe("PodReconciler", func() {
             - interface: 192.168.0.1:8080:8080
               tags:
                 service: example.demo.svc:80
+                protocol: http
             - interface: 192.168.0.1:6060:6060
               tags:
                 service: example.demo.svc:6061
+                protocol: tcp
 `))
 	})
 
@@ -311,9 +316,11 @@ var _ = Describe("PodReconciler", func() {
             - interface: 192.168.0.1:8080:8080
               tags:
                 service: example.demo.svc:80
+                protocol: http
             - interface: 192.168.0.1:6060:6060
               tags:
                 service: example.demo.svc:6061
+                protocol: tcp
 `))
 	})
 })

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

@@ -11,6 +11,7 @@ import (
 	mesh_proto "github.com/Kong/kuma/api/mesh/v1alpha1"
 	injector_metadata "github.com/Kong/kuma/app/kuma-injector/pkg/injector/metadata"
 	"github.com/Kong/kuma/pkg/core"
+	mesh_core "github.com/Kong/kuma/pkg/core/resources/apis/mesh"
 	util_k8s "github.com/Kong/kuma/pkg/plugins/discovery/k8s/util"
 	mesh_k8s "github.com/Kong/kuma/pkg/plugins/resources/k8s/native/api/v1alpha1"
 	util_proto "github.com/Kong/kuma/pkg/util/proto"
@@ -62,7 +63,7 @@ func DataplaneFor(pod *kube_core.Pod, services []*kube_core.Service, others []*m
 		}
 		dataplane.Networking.Gateway = gateway
 	} else {
-		ifaces, err := InboundInterfacesFor(pod, services)
+		ifaces, err := InboundInterfacesFor(pod, services, false)
 		if err != nil {
 			return nil, err
 		}
@@ -79,7 +80,7 @@ func DataplaneFor(pod *kube_core.Pod, services []*kube_core.Service, others []*m
 }
 
 func GatewayFor(pod *kube_core.Pod, services []*kube_core.Service) (*mesh_proto.Dataplane_Networking_Gateway, error) {
-	interfaces, err := InboundInterfacesFor(pod, services)
+	interfaces, err := InboundInterfacesFor(pod, services, true)
 	if err != nil {
 		return nil, err
 	}
@@ -91,7 +92,7 @@ func GatewayFor(pod *kube_core.Pod, services []*kube_core.Service) (*mesh_proto.
 	}, nil
 }
 
-func InboundInterfacesFor(pod *kube_core.Pod, services []*kube_core.Service) ([]*mesh_proto.Dataplane_Networking_Inbound, error) {
+func InboundInterfacesFor(pod *kube_core.Pod, services []*kube_core.Service, isGateway bool) ([]*mesh_proto.Dataplane_Networking_Inbound, error) {
 	var ifaces []*mesh_proto.Dataplane_Networking_Inbound
 
 	for _, svc := range services {
@@ -111,7 +112,7 @@ func InboundInterfacesFor(pod *kube_core.Pod, services []*kube_core.Service) ([]
 				DataplanePort: uint32(containerPort),
 				WorkloadPort:  uint32(containerPort),
 			}
-			tags := InboundTagsFor(pod, svc, &svcPort)
+			tags := InboundTagsFor(pod, svc, &svcPort, isGateway)
 
 			ifaces = append(ifaces, &mesh_proto.Dataplane_Networking_Inbound{
 				Interface: iface.String(),
@@ -172,19 +173,39 @@ func OutboundInterfacesFor(others []*mesh_k8s.Dataplane, serviceGetter kube_clie
 	return ofaces, nil
 }
 
-func InboundTagsFor(pod *kube_core.Pod, svc *kube_core.Service, svcPort *kube_core.ServicePort) map[string]string {
+func InboundTagsFor(pod *kube_core.Pod, svc *kube_core.Service, svcPort *kube_core.ServicePort, isGateway bool) map[string]string {
 	tags := util_k8s.CopyStringMap(pod.Labels)
 	if tags == nil {
 		tags = make(map[string]string)
 	}
 	tags[mesh_proto.ServiceTag] = ServiceTagFor(svc, svcPort)
+	// notice that in case of a gateway it might be confusing to see a protocol tag
+	// since gateway proxies multiple services each with its own protocol
+	if !isGateway {
+		tags[mesh_proto.ProtocolTag] = ProtocolTagFor(svc, svcPort)
+	}
 	return tags
 }
 
 func ServiceTagFor(svc *kube_core.Service, svcPort *kube_core.ServicePort) string {
 	return fmt.Sprintf("%s.%s.svc:%d", svc.Name, svc.Namespace, svcPort.Port)
 }
 
+// ProtocolTagFor infers service protocol from a `<port>.service.kuma.io/protocol` annotation.
+func ProtocolTagFor(svc *kube_core.Service, svcPort *kube_core.ServicePort) string {
+	protocolAnnotation := fmt.Sprintf("%d.service.kuma.io/protocol", svcPort.Port)
+	protocolValue := svc.Annotations[protocolAnnotation]
+	if protocolValue == "" {
+		// if `<port>.service.kuma.io/protocol` annotation is missing or has an empty value
+		// we want Dataplane to have a `protocol: tcp` tag in order to get user's attention
+		return mesh_core.ProtocolTCP
+	}
+	// if `<port>.service.kuma.io/protocol` annotation is present but has an invalid value
+	// we still want Dataplane to have a `protocol: <value as is>` tag in order to make it clear
+	// to a user that at least `<port>.service.kuma.io/protocol` has an effect
+	return protocolValue
+}
+
 func ParseServiceFQDN(host string) (name string, namespace string, err error) {
 	// split host into <name>.<namespace>.svc
 	segments := strings.Split(host, ".")

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

@@ -196,6 +196,9 @@ var _ = Describe("PodToDataplane(..)", func() {
 					ObjectMeta: kube_meta.ObjectMeta{
 						Namespace: "demo",
 						Name:      "example",
+						Annotations: map[string]string{
+							"80.service.kuma.io/protocol": "http",
+						},
 					},
 					Spec: kube_core.ServiceSpec{
 						Ports: []kube_core.ServicePort{
@@ -222,6 +225,9 @@ var _ = Describe("PodToDataplane(..)", func() {
 					ObjectMeta: kube_meta.ObjectMeta{
 						Namespace: "playground",
 						Name:      "sample",
+						Annotations: map[string]string{
+							"7071.service.kuma.io/protocol": "MONGO",
+						},
 					},
 					Spec: kube_core.ServiceSpec{
 						Ports: []kube_core.ServicePort{
@@ -255,21 +261,25 @@ var _ = Describe("PodToDataplane(..)", func() {
                 - interface: 192.168.0.1:8080:8080
                   tags:
                     app: example
+                    protocol: http
                     service: example.demo.svc:80
                     version: "0.1"
                 - interface: 192.168.0.1:8443:8443
                   tags:
                     app: example
+                    protocol: tcp
                     service: example.demo.svc:443
                     version: "0.1"
                 - interface: 192.168.0.1:7070:7070
                   tags:
                     app: example
+                    protocol: MONGO
                     service: sample.playground.svc:7071
                     version: "0.1"
                 - interface: 192.168.0.1:6060:6060
                   tags:
                     app: example
+                    protocol: tcp
                     service: sample.playground.svc:6061
                     version: "0.1"
 `,
@@ -356,6 +366,7 @@ var _ = Describe("PodToDataplane(..)", func() {
                 - interface: 192.168.0.1:8080:8080
                   tags:
                     app: example
+                    protocol: tcp
                     service: example.demo.svc:80
                     version: "0.1"
                 outbound:
@@ -372,6 +383,9 @@ var _ = Describe("PodToDataplane(..)", func() {
 					ObjectMeta: kube_meta.ObjectMeta{
 						Namespace: "demo",
 						Name:      "example",
+						Annotations: map[string]string{
+							"80.service.kuma.io/protocol": "http", // should be ignored in case of a gateway
+						},
 					},
 					Spec: kube_core.ServiceSpec{
 						Ports: []kube_core.ServicePort{
@@ -463,8 +477,10 @@ var _ = Describe("MeshFor(..)", func() {
 var _ = Describe("InboundTagsFor(..)", func() {
 
 	type testCase struct {
-		podLabels map[string]string
-		expected  map[string]string
+		isGateway      bool
+		podLabels      map[string]string
+		svcAnnotations map[string]string
+		expected       map[string]string
 	}
 
 	DescribeTable("should combine Pod's labels with Service's FQDN and port",
@@ -483,6 +499,7 @@ var _ = Describe("InboundTagsFor(..)", func() {
 					Labels: map[string]string{
 						"more": "labels",
 					},
+					Annotations: given.svcAnnotations,
 				},
 				Spec: kube_core.ServiceSpec{
 					Ports: []kube_core.ServicePort{
@@ -499,15 +516,77 @@ var _ = Describe("InboundTagsFor(..)", func() {
 			}
 
 			// then
-			Expect(InboundTagsFor(pod, svc, &svc.Spec.Ports[0])).To(Equal(given.expected))
+			Expect(InboundTagsFor(pod, svc, &svc.Spec.Ports[0], given.isGateway)).To(Equal(given.expected))
 		},
 		Entry("Pod without labels", testCase{
+			isGateway: false,
 			podLabels: nil,
 			expected: map[string]string{
-				"service": "example.demo.svc:80",
+				"service":  "example.demo.svc:80",
+				"protocol": "tcp", // we want Kuma's default behaviour to be explicit to a user
 			},
 		}),
 		Entry("Pod with labels", testCase{
+			isGateway: false,
+			podLabels: map[string]string{
+				"app":     "example",
+				"version": "0.1",
+			},
+			expected: map[string]string{
+				"app":      "example",
+				"version":  "0.1",
+				"service":  "example.demo.svc:80",
+				"protocol": "tcp", // we want Kuma's default behaviour to be explicit to a user
+			},
+		}),
+		Entry("Pod with `service` label", testCase{
+			isGateway: false,
+			podLabels: map[string]string{
+				"service": "something",
+				"app":     "example",
+				"version": "0.1",
+			},
+			expected: map[string]string{
+				"app":      "example",
+				"version":  "0.1",
+				"service":  "example.demo.svc:80",
+				"protocol": "tcp", // we want Kuma's default behaviour to be explicit to a user
+			},
+		}),
+		Entry("Service with a `<port>.service.kuma.io/protocol` annotation and an unknown value", testCase{
+			isGateway: false,
+			podLabels: map[string]string{
+				"app":     "example",
+				"version": "0.1",
+			},
+			svcAnnotations: map[string]string{
+				"80.service.kuma.io/protocol": "not-yet-supported-protocol",
+			},
+			expected: map[string]string{
+				"app":      "example",
+				"version":  "0.1",
+				"service":  "example.demo.svc:80",
+				"protocol": "not-yet-supported-protocol", // we want Kuma's behaviour to be straightforward to a user (just copy annotation value "as is")
+			},
+		}),
+		Entry("Service with a `<port>.service.kuma.io/protocol` annotation and a known value", testCase{
+			isGateway: false,
+			podLabels: map[string]string{
+				"app":     "example",
+				"version": "0.1",
+			},
+			svcAnnotations: map[string]string{
+				"80.service.kuma.io/protocol": "http",
+			},
+			expected: map[string]string{
+				"app":      "example",
+				"version":  "0.1",
+				"service":  "example.demo.svc:80",
+				"protocol": "http",
+			},
+		}),
+		Entry("`gateway` Pod should not have a `protocol` tag", testCase{
+			isGateway: true,
 			podLabels: map[string]string{
 				"app":     "example",
 				"version": "0.1",
@@ -518,12 +597,15 @@ var _ = Describe("InboundTagsFor(..)", func() {
 				"service": "example.demo.svc:80",
 			},
 		}),
-		Entry("Pod with `service` label", testCase{
+		Entry("`gateway` Pod should not have a `protocol` tag even if `<port>.service.kuma.io/protocol` annotation is present", testCase{
+			isGateway: true,
 			podLabels: map[string]string{
-				"service": "something",
 				"app":     "example",
 				"version": "0.1",
 			},
+			svcAnnotations: map[string]string{
+				"80.service.kuma.io/protocol": "http",
+			},
 			expected: map[string]string{
 				"app":     "example",
 				"version": "0.1",
@@ -560,6 +642,82 @@ var _ = Describe("ServiceTagFor(..)", func() {
 	})
 })
 
+var _ = Describe("ProtocolTagFor(..)", func() {
+
+	type testCase struct {
+		annotations map[string]string
+		expected    string
+	}
+
+	DescribeTable("should infer service protocol from a `<port>.service.kuma.io/protocol` annotation",
+		func(given testCase) {
+			// given
+			svc := &kube_core.Service{
+				ObjectMeta: kube_meta.ObjectMeta{
+					Namespace:   "demo",
+					Name:        "example",
+					Annotations: given.annotations,
+				},
+				Spec: kube_core.ServiceSpec{
+					Ports: []kube_core.ServicePort{
+						{
+							Name: "http",
+							Port: 80,
+							TargetPort: kube_intstr.IntOrString{
+								Type:   kube_intstr.Int,
+								IntVal: 8080,
+							},
+						},
+					},
+				},
+			}
+
+			// expect
+			Expect(ProtocolTagFor(svc, &svc.Spec.Ports[0])).To(Equal(given.expected))
+		},
+		Entry("no `<port>.service.kuma.io/protocol` annotation", testCase{
+			annotations: nil,
+			expected:    "tcp", // we want Kuma's default behaviour to be explicit to a user
+		}),
+		Entry("`<port>.service.kuma.io/protocol` annotation has an empty value", testCase{
+			annotations: map[string]string{
+				"80.service.kuma.io/protocol": "",
+			},
+			expected: "tcp", // we want Kuma's default behaviour to be explicit to a user
+		}),
+		Entry("`<port>.service.kuma.io/protocol` annotation is for a different port", testCase{
+			annotations: map[string]string{
+				"8080.service.kuma.io/protocol": "http",
+			},
+			expected: "tcp", // we want Kuma's default behaviour to be explicit to a user
+		}),
+		Entry("`<port>.service.kuma.io/protocol` annotation has an unknown value", testCase{
+			annotations: map[string]string{
+				"80.service.kuma.io/protocol": "not-yet-supported-protocol",
+			},
+			expected: "not-yet-supported-protocol", // we want Kuma's behaviour to be straightforward to a user (just copy annotation value "as is")
+		}),
+		Entry("`<port>.service.kuma.io/protocol` annotation has a non-lowercase value", testCase{
+			annotations: map[string]string{
+				"80.service.kuma.io/protocol": "HtTp",
+			},
+			expected: "HtTp", // we want Kuma's behaviour to be straightforward to a user (just copy annotation value "as is")
+		}),
+		Entry("`<port>.service.kuma.io/protocol` annotation has a known value: http", testCase{
+			annotations: map[string]string{
+				"80.service.kuma.io/protocol": "http",
+			},
+			expected: "http",
+		}),
+		Entry("`<port>.service.kuma.io/protocol` annotation has a known value: tcp", testCase{
+			annotations: map[string]string{
+				"80.service.kuma.io/protocol": "tcp",
+			},
+			expected: "tcp",
+		}),
+	)
+})
+
 type fakeReader map[string]string
 
 var _ kube_client.Reader = fakeReader{}