feat(controller): traefik v2 support

Author: duanhongyi

rootfs/api/models/app.py

@@ -196,14 +196,15 @@ def _refresh_ingress(self, hosts, tls_map, ssl_redirect):
             if ingress == "":
                 raise ServiceUnavailable('Empty hostname')
             try:
-                data = self._scheduler.ingress.get(namespace, ingress).json()
+                data = self._scheduler.ingress(
+                    settings.INGRESS_CLASS).get(namespace, ingress).json()
                 version = data["metadata"]["resourceVersion"]
-                self._scheduler.ingress.put(
-                    ingress, settings.INGRESS_CLASS, namespace, version, **kwargs)
+                self._scheduler.ingress(settings.INGRESS_CLASS).put(
+                    namespace, ingress, version, **kwargs)
             except KubeException:
                 self.log("creating Ingress {}".format(namespace), level=logging.INFO)
-                self._scheduler.ingress.create(
-                    ingress, settings.INGRESS_CLASS, namespace, **kwargs)
+                self._scheduler.ingress(settings.INGRESS_CLASS).create(
+                    namespace, ingress, **kwargs)
         except KubeException as e:
             raise ServiceUnavailable('Could not create Ingress in Kubernetes') from e
 
@@ -944,7 +945,7 @@ def routable(self, routable):
         else:
             try:
                 namespace = ingress = self.id
-                self._scheduler.ingress.delete(namespace, ingress)
+                self._scheduler.ingress(settings.INGRESS_CLASS).delete(namespace, ingress)
             except KubeException as e:
                 raise ServiceUnavailable(str(e)) from e
 

rootfs/scheduler/__init__.py

@@ -38,33 +38,41 @@ def get_session(k8s_api_verify_tls):
 
 
 class KubeHTTPClient(object):
+    abstract = False
+    api_version = 'v1'
+    api_prefix = 'api'
     # ISO-8601 which is used by kubernetes
     DATETIME_FORMAT = '%Y-%m-%dT%H:%M:%SZ'
 
     def __init__(self, url, k8s_api_verify_tls=True):
         global resource_mapping
         self.url = url
-        self.session = get_session(k8s_api_verify_tls)
+        self.k8s_api_verify_tls = k8s_api_verify_tls
+        self.session = get_session(self.k8s_api_verify_tls)
 
         # map the various k8s Resources to an internal property
         from scheduler.resources import Resource  # lazy load
         for res in Resource:
             name = str(res.__name__).lower()  # singular
             component = name + 's'  # make plural
             # check if component has already been processed
-            if component in resource_mapping:
+            if component in resource_mapping or res.abstract:
                 continue
 
             # get past recursion problems in case of self reference
             resource_mapping[component] = ''
-            resource_mapping[component] = res(self.url)
+            resource_mapping[component] = res(self.url, self.k8s_api_verify_tls)
             # map singular Resource name to the plural one
             resource_mapping[name] = component
             if res.short_name is not None:
                 # map short name to long name so a resource can be named rs
                 # but have the main object live at replicasets
                 resource_mapping[str(res.short_name).lower()] = component
 
+    def api(self, tmpl, *args):
+        """Return a fully-qualified Kubernetes API URL from a string template with args."""
+        return "/{}/{}".format(self.api_prefix, self.api_version) + tmpl.format(*args)
+
     def __getattr__(self, name):
         global resource_mapping
         if name in resource_mapping:

rootfs/scheduler/resources/ingress.py

@@ -1,22 +1,26 @@
 from scheduler.exceptions import KubeHTTPException
 from scheduler.resources import Resource
 
-MANIFEAT_CLASSES = {}
 
+class BaseIngress(Resource):
+    abstract = True
+    api_version = 'networking.k8s.io/v1'
+    api_prefix = 'apis'
+    short_name = 'ingress'
+    ingress_path = "/"
+    ingress_class = None
 
-class BaseManifest(object):
-
-    def manifest(self, api_version, ingress, ingress_class, namespace, **kwargs):
-        path = "/*" if ingress_class in ("gce", "alb") else "/"
+    def manifest(self, ingress, **kwargs):
         hosts, tls = kwargs.pop("hosts", None), kwargs.pop("tls", None)
         version = kwargs.pop("version", None)
         data = {
             "kind": "Ingress",
-            "apiVersion": api_version,
+            "apiVersion": self.api_version,
             "metadata": {
                 "name": ingress,
                 "annotations": {
                     "kubernetes.io/tls-acme": "true",
+                    "kubernetes.io/ingress.class": self.ingress_class
                 }
             },
             "spec": {}
@@ -27,7 +31,7 @@ def manifest(self, api_version, ingress, ingress_class, namespace, **kwargs):
                 "http": {
                     "paths": [
                         {
-                            "path": path,
+                            "path": self.ingress_path,
                             "pathType": "Prefix",
                             "backend": {
                                 "service": {
@@ -41,72 +45,12 @@ def manifest(self, api_version, ingress, ingress_class, namespace, **kwargs):
                     ]
                 }
             } for host in hosts]
-        if ingress_class:
-            data["metadata"]["annotations"].update({
-                "kubernetes.io/ingress.class": ingress_class
-            })
         if tls:
             data["spec"]["tls"] = tls
         if version:
             data["metadata"]["resourceVersion"] = version
         return data
 
-
-class NginxManifest(BaseManifest):
-
-    def manifest(self, api_version, ingress, ingress_class, namespace, **kwargs):
-        data = BaseManifest.manifest(
-            self, api_version, ingress, ingress_class, namespace, **kwargs)
-        if "allowlist" in kwargs:
-            allowlist = ", ".join(kwargs.pop("allowlist"))
-            data["metadata"]["annotations"].update({
-                "nginx.ingress.kubernetes.io/whitelist-source-range": allowlist
-            })
-        if "ssl_redirect" in kwargs:
-            ssl_redirect = kwargs.pop("ssl_redirect")
-            data["metadata"]["annotations"].update({
-                "nginx.ingress.kubernetes.io/ssl-redirect": ssl_redirect
-            })
-        return data
-
-
-MANIFEAT_CLASSES["nginx"] = NginxManifest
-
-
-class TraefikManifest(BaseManifest):
-
-    def manifest(self, api_version, ingress, ingress_class, namespace, **kwargs):
-        data = BaseManifest.manifest(
-            self, api_version, ingress, ingress_class, namespace, **kwargs)
-        if "allowlist" in kwargs:
-            allowlist = ", ".join(kwargs.pop("allowlist"))
-            data["metadata"]["annotations"].update({
-                "ingress.kubernetes.io/whitelist-x-forwarded-for": "true",
-                "traefik.ingress.kubernetes.io/whitelist-source-range": allowlist
-            })
-        if "ssl_redirect" in kwargs:
-            ssl_redirect = kwargs.pop("ssl_redirect")
-            data["metadata"]["annotations"].update({
-                "ingress.kubernetes.io/ssl-redirect": ssl_redirect
-            })
-        return data
-
-
-MANIFEAT_CLASSES["traefik"] = TraefikManifest
-
-
-class Ingress(Resource):
-
-    api_version = 'networking.k8s.io/v1'
-    api_prefix = 'apis'
-    short_name = 'ingress'
-
-    @staticmethod
-    def manifest(api_version, ingress, ingress_class, namespace, **kwargs):
-        return MANIFEAT_CLASSES.get(ingress_class, BaseManifest)().manifest(
-            api_version, ingress, ingress_class, namespace, **kwargs
-        )
-
     def get(self, namespace, ingress=None, **kwargs):
         """
         Fetch a single Ingress or a list of Ingresses
@@ -124,28 +68,28 @@ def get(self, namespace, ingress=None, **kwargs):
 
         return response
 
-    def create(self, ingress, ingress_class, namespace, **kwargs):
+    def create(self, namespace, ingress, **kwargs):
         url = self.api("/namespaces/{}/ingresses", namespace)
-        data = self.manifest(self.api_version, ingress, ingress_class, namespace, **kwargs)
+        data = self.manifest(ingress, **kwargs)
         response = self.http_post(url, json=data)
 
-        if not response.status_code == 201:
+        if self.unhealthy(response.status_code):
             raise KubeHTTPException(response, "create Ingress {}".format(namespace))
 
         return response
 
-    def put(self, ingress, ingress_class, namespace, version, **kwargs):
+    def put(self, namespace, ingress, version, **kwargs):
         url = self.api("/namespaces/{}/ingresses/{}", namespace, ingress)
         kwargs["version"] = version
-        data = self.manifest(self.api_version, ingress, ingress_class, namespace, **kwargs)
+        data = self.manifest(ingress, **kwargs)
         response = self.http_put(url, json=data)
 
         if self.unhealthy(response.status_code):
             raise KubeHTTPException(response, "put Ingress {}".format(namespace))
 
         return response
 
-    def patch(self, ingress, namespace, data):
+    def patch(self, namespace, ingress, data, **kwargs):
         url = self.api("/namespaces/{}/ingresses/{}", namespace, ingress)
         response = self.http_put(url, json=data)
 
@@ -161,3 +105,157 @@ def delete(self, namespace, ingress):
             raise KubeHTTPException(response, 'delete Ingress "{}"', namespace)
 
         return response
+
+
+class IngressFactory(Resource):
+
+    short_name = 'ingress'
+    ingress_class_map = {
+        "default": BaseIngress
+    }
+
+    def __call__(self, ingress_name):
+        ingress_cls = self.ingress_class_map.get(ingress_name, self.ingress_class_map["default"])
+        ingress_cls.ingress_class = ingress_name
+        return ingress_cls(self.url, self.k8s_api_verify_tls)
+
+    @classmethod
+    def register(cls, ingress_name, ingress_cls):
+        cls.ingress_class_map[ingress_name] = ingress_cls
+
+
+class NginxIngress(BaseIngress):
+
+    def manifest(self, ingress, **kwargs):
+        data = BaseIngress.manifest(self, ingress, **kwargs)
+        if "allowlist" in kwargs:
+            allowlist = ", ".join(kwargs.pop("allowlist"))
+            data["metadata"]["annotations"].update({
+                "nginx.ingress.kubernetes.io/whitelist-source-range": allowlist
+            })
+        if "ssl_redirect" in kwargs:
+            ssl_redirect = kwargs.pop("ssl_redirect")
+            data["metadata"]["annotations"].update({
+                "nginx.ingress.kubernetes.io/ssl-redirect": ssl_redirect
+            })
+        return data
+
+
+class TraefikIngress(BaseIngress):
+
+    class Middleware(Resource):
+        abstract = True
+        api_version = 'traefik.containo.us/v1alpha1'
+        api_prefix = 'apis'
+
+        def manifest(self, name, allowlist, ssl_redirect, resource_version=None):
+            manifest = {
+                "apiVersion": self.api_version,
+                "kind": "Middleware",
+                "metadata": {
+                    "name": name
+                },
+                "spec": {}
+            }
+            if allowlist:
+                manifest["spec"]["ipWhiteList"] = {
+                    "sourceRange": allowlist
+                }
+            if ssl_redirect:
+                manifest["spec"]["redirectScheme"] = {
+                    "scheme": "https",
+                    "permanent": True
+                }
+            if resource_version:
+                manifest["metadata"]["resourceVersion"] = resource_version
+            return manifest
+
+        def get(self, namespace, name=None, **kwargs):
+            """
+            Fetch a single Middleware or a list of Middlewares
+            """
+            if name is not None:
+                url = self.api("/namespaces/{}/middlewares/{}", namespace, name)
+            else:
+                url = self.api("/namespaces/{}/middlewares", namespace)
+            return self.http_get(url, params=self.query_params(**kwargs))
+
+        def create(self, namespace, name, allowlist, ssl_redirect):
+            url = self.api("/namespaces/{}/middlewares", namespace)
+            data = self.manifest(name, allowlist, ssl_redirect)
+            response = self.http_put(url, json=data)
+            if self.unhealthy(response.status_code):
+                raise KubeHTTPException(response, "create middleware {}".format(namespace))
+            return response
+
+        def put(self, namespace, name, allowlist, ssl_redirect, resource_version):
+            url = self.api("/namespaces/{}/middlewares/{}", namespace, name)
+            data = self.manifest(allowlist, ssl_redirect, resource_version)
+            response = self.http_put(url, json=data)
+            if self.unhealthy(response.status_code):
+                raise KubeHTTPException(response, "put middleware {}".format(namespace))
+            return response
+
+        def create_or_put(self, namespace, name, allowlist, ssl_redirect):
+            response = self.get(namespace, name)
+            if response.status_code == 404:
+                return self.create(namespace, name, allowlist, ssl_redirect)
+            elif self.unhealthy(response.status_code):
+                raise KubeHTTPException(response, "get middleware {}".format(namespace))
+            resource_version = response.json()["metadata"]["resourceVersion"]
+            return self.put(namespace, name, allowlist, ssl_redirect, resource_version)
+
+        def delete(self, namespace, name):
+            url = self.api("/namespaces/{}/middlewares/{}", namespace, name)
+            response = self.http_delete(url)
+            if self.unhealthy(response.status_code):
+                raise KubeHTTPException(response, 'delete middlewares "{}"', namespace)
+
+            return response
+
+    def __init__(self, url, k8s_api_verify_tls=True):
+        super().__init__(url, k8s_api_verify_tls)
+        self.middleware = self.Middleware(url, k8s_api_verify_tls)
+
+    def create(self, namespace, ingress, **kwargs):
+        response = super().create(ingress, namespace, **kwargs)
+        if "allowlist" in kwargs or "ssl_redirect" in kwargs:
+            self.middleware.create_or_put(
+                namespace, ingress,
+                kwargs.get("allowlist"), kwargs.get("ssl_redirect")
+            )
+        return response
+
+    def put(self, namespace, ingress, version, **kwargs):
+        response = super().put(ingress, namespace, version, **kwargs)
+        if "allowlist" in kwargs or "ssl_redirect" in kwargs:
+            self.middleware.create_or_put(
+                namespace, ingress,
+                kwargs.get("allowlist"), kwargs.get("ssl_redirect")
+            )
+        return response
+
+    def patch(self, namespace, ingress, data, **kwargs):
+        response = super().patch(ingress, namespace, data, **kwargs)
+        if "allowlist" in kwargs or "ssl_redirect" in kwargs:
+            self.middleware.create_or_put(
+                namespace, ingress,
+                kwargs.get("allowlist"), kwargs.get("ssl_redirect")
+            )
+        return response
+
+    def delete(self, namespace, ingress):
+        response = super().delete(namespace, ingress)
+        if not self.unhealthy(self.middleware.get(namespace, ingress)):
+            self.middleware.delete(namespace, ingress)
+        return response
+
+
+class WildcardPathIngress(BaseIngress):
+    ingress_path = "/*"
+
+
+IngressFactory.register("nginx", NginxIngress)
+IngressFactory.register("traefik", TraefikIngress)
+IngressFactory.register("gce", WildcardPathIngress)
+IngressFactory.register("alb", WildcardPathIngress)