Resolve named ports in Ingress (#4813)

Update IngressProcessor to resolve named ports Signed-off-by: Anton Ustyuzhanin <[email protected]> Signed-off-by: Hamzah Qudsi <[email protected]>
emissary-ingress · Feb 3, 2023 · 7a6edf7 · 7a6edf7
1 parent f70b5ae
commit 7a6edf7
Show file tree

Hide file tree

Showing 6 changed files with 154 additions and 8 deletions.
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -101,7 +101,14 @@ it will be removed; but as it won't be user-visible this isn't considered a brea
   hostname, an incorrect configuration  was generated with an sni match including the port. This has
   been fixed and the correct envoy configuration is being generated. ([fix: hostname port issue])
 
+- Change: Previously, specifying backend ports by name in Ingress was not supported and would result
+  in defaulting to port 80. This allows emissary-ingress to now resolve port names for backend
+  services. If the port number cannot be resolved by the name (e.g named port in the Service doesn't
+  exist) then it defaults back  to the original behavior. (Thanks to <a
+  href="https://github.com/antonu17">Anton Ustyuzhanin</a>!). ([#4809])
+
 [fix: hostname port issue]: https://github.com/emissary-ingress/emissary/pull/4816
+[#4809]: https://github.com/emissary-ingress/emissary/pull/4809
 
 ## [3.4.0] January 03, 2023
 [3.4.0]: https://github.com/emissary-ingress/emissary/compare/v3.3.0...v3.4.0

diff --git a/docs/releaseNotes.yml b/docs/releaseNotes.yml
@@ -54,6 +54,18 @@ items:
         - title: "fix: hostname port issue"
           link: https://github.com/emissary-ingress/emissary/pull/4816
 
+      - title: Add support for resolving port names in Ingress resource 
+        type: change
+        body: >-
+          Previously, specifying backend ports by name in Ingress was not supported and would result in defaulting
+          to port 80. This allows emissary-ingress to now resolve port names for backend services. If the port number
+          cannot be resolved by the name (e.g named port in the Service doesn't exist) then it defaults back 
+          to the original behavior.
+          (Thanks to <a href="https://github.com/antonu17">Anton Ustyuzhanin</a>!).
+        github:
+        - title: "#4809"
+          link: https://github.com/emissary-ingress/emissary/pull/4809
+
   - version: 3.4.0
     prevVersion: 3.3.0
     date: '2023-01-03'

diff --git a/python/ambassador/fetch/dependency.py b/python/ambassador/fetch/dependency.py
@@ -3,6 +3,7 @@
 from typing import (
     Any,
     Collection,
+    Dict,
     Iterator,
     Mapping,
     MutableSet,
@@ -13,7 +14,7 @@
     TypeVar,
 )
 
-from .k8sobject import KubernetesObject
+from .k8sobject import KubernetesObject, KubernetesObjectKey
 
 
 class Dependency(Protocol):
@@ -35,9 +36,11 @@ class ServiceDependency(Dependency):
     """
 
     ambassador_service: Optional[KubernetesObject]
+    discovered_services: Dict[KubernetesObjectKey, KubernetesObject]
 
     def __init__(self) -> None:
         self.ambassador_service = None
+        self.discovered_services = {}
 
     def watt_key(self) -> str:
         return "service"

diff --git a/python/ambassador/fetch/ingress.py b/python/ambassador/fetch/ingress.py
@@ -2,7 +2,7 @@
 
 from ..config import Config
 from .dependency import IngressClassesDependency, SecretDependency, ServiceDependency
-from .k8sobject import KubernetesGVK, KubernetesObject
+from .k8sobject import KubernetesGVK, KubernetesObject, KubernetesObjectKey
 from .k8sprocessor import ManagedKubernetesProcessor
 from .resource import NormalizedResource, ResourceManager
 
@@ -104,6 +104,23 @@ def _update_status(self, obj: KubernetesObject) -> None:
                 f"Not reconciling Ingress {obj.name}: observed and current statuses are in sync"
             )
 
+    def _try_resolve_service_port_number(self, namespace, service_name, service_port):
+        self.logger.debug(f"Resolving named port '{service_port}' in service '{service_name}'")
+
+        key = KubernetesObjectKey(KubernetesGVK("v1", "Service"), namespace, service_name)
+        k8s_svc: Optional[KubernetesObject]
+        k8s_svc = self.service_dep.discovered_services.get(key, None)
+        if not k8s_svc:
+            self.logger.debug(f"Could not find service '{service_name}'")
+            return service_port
+
+        for port in k8s_svc.spec.get("ports", []):
+            if service_port == port.get("name", None):
+                return port.get("port", service_port)
+
+        self.logger.debug(f"Could not find port '{service_port}' in service '{service_name}'")
+        return service_port
+
     def _process(self, obj: KubernetesObject) -> None:
         ingress_class_name = obj.spec.get("ingressClassName", "")
 
@@ -221,6 +238,13 @@ def _process(self, obj: KubernetesObject) -> None:
                 service_port = path_backend.get("servicePort", None)
                 path_location = path.get("path", "/")
 
+                try:
+                    service_port = int(service_port)
+                except:
+                    service_port = self._try_resolve_service_port_number(
+                        obj.namespace, service_name, service_port
+                    )
+
                 if not service_name or not service_port or not path_location:
                     continue
 

diff --git a/python/ambassador/fetch/service.py b/python/ambassador/fetch/service.py
@@ -40,14 +40,12 @@ class InternalServiceProcessor(ManagedKubernetesProcessor):
 
     service_dep: ServiceDependency
     helm_chart: Optional[str]
-    discovered_services: Dict[KubernetesObjectKey, KubernetesObject]
 
     def __init__(self, manager: ResourceManager) -> None:
         super().__init__(manager)
 
         self.service_dep = self.deps.provide(ServiceDependency)
         self.helm_chart = None
-        self.discovered_services = {}
 
     def kinds(self) -> FrozenSet[KubernetesGVK]:
         return frozenset([KubernetesGVK("v1", "Service")])
@@ -93,7 +91,7 @@ def _process(self, obj: KubernetesObject) -> None:
                 f"not saving Kubernetes Service {obj.name}.{obj.namespace} with no ports"
             )
         else:
-            self.discovered_services[obj.key] = obj
+            self.service_dep.discovered_services[obj.key] = obj
 
             if self._is_ambassador_service(obj):
                 self.logger.debug(f"Found Ambassador service: {obj.name}")
@@ -204,6 +202,7 @@ class ServiceProcessor(ManagedKubernetesProcessor):
     Ambassador service resources.
     """
 
+    service_dep: ServiceDependency
     services: InternalServiceProcessor
     endpoints: InternalEndpointsProcessor
     delegate: AggregateKubernetesProcessor
@@ -212,6 +211,7 @@ class ServiceProcessor(ManagedKubernetesProcessor):
     def __init__(self, manager: ResourceManager, watch_only: bool = False):
         super().__init__(manager)
 
+        self.service_dep = self.deps.want(ServiceDependency)
         self.services = InternalServiceProcessor(manager)
         self.endpoints = InternalEndpointsProcessor(manager)
         self.delegate = AggregateKubernetesProcessor([self.services, self.endpoints])
@@ -226,7 +226,7 @@ def _process(self, obj: KubernetesObject) -> None:
     def finalize(self) -> None:
         self.delegate.finalize()
 
-        # The point here is to sort out self.services.discovered_services and
+        # The point here is to sort out self.service_dep.discovered_services and
         # self.endpoints.discovered_endpoints and turn them into proper
         # Ambassador Service resources. This is a bit annoying, because of the
         # annoyances of Kubernetes, but we'll give it a go.
@@ -251,12 +251,12 @@ def finalize(self) -> None:
         # od = {
         #     'elements': [ x.as_dict() for x in self.elements ],
         #     'k8s_endpoints': self.endpoints.discovered_endpoints,
-        #     'k8s_services': self.services.discovered_services,
+        #     'k8s_services': self.service_dep.discovered_services,
         # }
         #
         # self.logger.debug("==== FINALIZE START\n%s" % dump_json(od, pretty=True))
 
-        for k8s_svc in self.services.discovered_services.values():
+        for k8s_svc in self.service_dep.discovered_services.values():
             key = f"{k8s_svc.name}.{k8s_svc.namespace}"
 
             target_ports = {}

diff --git a/python/tests/unit/test_fetch.py b/python/tests/unit/test_fetch.py
@@ -12,6 +12,7 @@
 logger = logging.getLogger("ambassador")
 
 from ambassador import Config
+from ambassador.fetch import ResourceFetcher
 from ambassador.fetch.ambassador import AmbassadorProcessor
 from ambassador.fetch.dependency import (
     DependencyManager,
@@ -278,6 +279,105 @@ def test_mapping_v1(self):
         assert mapping.prefix == valid_mapping_v1.spec["prefix"]
         assert mapping.service == valid_mapping_v1.spec["service"]
 
+    def test_ingress_with_named_port(self):
+        yaml = """
+---
+apiVersion: v1
+kind: Service
+metadata:
+  name: quote
+  namespace: default
+spec:
+  type: ClusterIP
+  ports:
+  - name: http
+    port: 3000
+    protocol: TCP
+    targetPort: 3000
+  selector:
+    app: quote
+---
+apiVersion: extensions/v1beta1
+kind: Ingress
+metadata:
+  annotations:
+    getambassador.io/ambassador-id: default
+    kubernetes.io/ingress.class: ambassador
+  name: quote
+  namespace: default
+spec:
+  rules:
+  - http:
+      paths:
+      - path: /
+        pathType: ImplementationSpecific
+        backend:
+          serviceName: quote
+          servicePort: http
+      - path: /metrics
+        pathType: ImplementationSpecific
+        backend:
+          serviceName: quote
+          servicePort: metrics
+      - path: /health
+        pathType: ImplementationSpecific
+        backend:
+          serviceName: quote
+          servicePort: 9000
+      - path: /missed-name
+        pathType: ImplementationSpecific
+        backend:
+          serviceName: missed
+          servicePort: missed
+      - path: /missed-number
+        pathType: ImplementationSpecific
+        backend:
+          serviceName: missed
+          servicePort: 8080
+status:
+  loadBalancer: {}
+"""
+        aconf = Config()
+        logger.setLevel(logging.DEBUG)
+
+        fetcher = ResourceFetcher(logger, aconf)
+        fetcher.parse_yaml(yaml, True)
+
+        mgr = fetcher.manager
+        assert len(mgr.elements) == 6
+
+        aconf.load_all(fetcher.sorted())
+        assert len(aconf.errors) == 0
+
+        mappings = aconf.get_config("mappings")
+        assert mappings
+        assert len(mappings) == 5
+
+        mapping_root = mappings.get("quote-0-0")
+        assert mapping_root
+        assert mapping_root.prefix == "/"
+        assert mapping_root.service == "quote.default:3000"
+
+        mapping_metrics = mappings.get("quote-0-1")
+        assert mapping_metrics
+        assert mapping_metrics.prefix == "/metrics"
+        assert mapping_metrics.service == "quote.default:metrics"
+
+        mapping_health = mappings.get("quote-0-2")
+        assert mapping_health
+        assert mapping_health.prefix == "/health"
+        assert mapping_health.service == "quote.default:9000"
+
+        mapping_missed_name = mappings.get("quote-0-3")
+        assert mapping_missed_name
+        assert mapping_missed_name.prefix == "/missed-name"
+        assert mapping_missed_name.service == "missed.default:missed"
+
+        mapping_missed_number = mappings.get("quote-0-4")
+        assert mapping_missed_number
+        assert mapping_missed_number.prefix == "/missed-number"
+        assert mapping_missed_number.service == "missed.default:8080"
+
 
 class TestAggregateKubernetesProcessor:
     def test_aggregation(self):