authentik.outposts.controllers.k8s.service

  1"""Kubernetes Service Reconciler"""
  2
  3from typing import TYPE_CHECKING
  4
  5from kubernetes.client import CoreV1Api, V1ObjectMeta, V1Service, V1ServicePort, V1ServiceSpec
  6
  7from authentik.outposts.controllers.base import FIELD_MANAGER
  8from authentik.outposts.controllers.k8s.base import KubernetesObjectReconciler
  9from authentik.outposts.controllers.k8s.deployment import DeploymentReconciler
 10from authentik.outposts.controllers.k8s.triggers import NeedsUpdate
 11from authentik.outposts.controllers.k8s.utils import compare_ports
 12
 13if TYPE_CHECKING:
 14    from authentik.outposts.controllers.kubernetes import KubernetesController
 15
 16
 17class ServiceReconciler(KubernetesObjectReconciler[V1Service]):
 18    """Kubernetes Service Reconciler"""
 19
 20    def __init__(self, controller: KubernetesController) -> None:
 21        super().__init__(controller)
 22        self.api = CoreV1Api(controller.client)
 23
 24    @staticmethod
 25    def reconciler_name() -> str:
 26        return "service"
 27
 28    def reconcile(self, current: V1Service, reference: V1Service):
 29        compare_ports(current.spec.ports, reference.spec.ports)
 30        # run the base reconcile last, as that will probably raise NeedsUpdate
 31        # after an authentik update. However the ports might have also changed during
 32        # the update, so this causes the service to be re-created with higher
 33        # priority than being updated.
 34        if current.spec.selector != reference.spec.selector:
 35            raise NeedsUpdate()
 36        if current.spec.type != reference.spec.type:
 37            raise NeedsUpdate()
 38        super().reconcile(current, reference)
 39
 40    def get_reference_object(self) -> V1Service:
 41        """Get deployment object for outpost"""
 42        meta = self.get_object_meta(name=self.name)
 43        ports = []
 44        for port in self.controller.deployment_ports:
 45            ports.append(
 46                V1ServicePort(
 47                    name=port.name,
 48                    port=port.port,
 49                    protocol=port.protocol.upper(),
 50                    target_port=port.inner_port or port.port,
 51                )
 52            )
 53        if self.is_embedded:
 54            selector_labels = {
 55                "app.kubernetes.io/name": "authentik",
 56                "app.kubernetes.io/component": "server",
 57            }
 58        else:
 59            selector_labels = DeploymentReconciler(self.controller).get_pod_meta()
 60        return V1Service(
 61            metadata=meta,
 62            spec=V1ServiceSpec(
 63                ports=ports,
 64                selector=selector_labels,
 65                type=self.controller.outpost.config.kubernetes_service_type,
 66            ),
 67        )
 68
 69    def create(self, reference: V1Service):
 70        return self.api.create_namespaced_service(
 71            self.namespace, reference, field_manager=FIELD_MANAGER
 72        )
 73
 74    def delete(self, reference: V1Service):
 75        return self.api.delete_namespaced_service(reference.metadata.name, self.namespace)
 76
 77    def retrieve(self) -> V1Service:
 78        return self.api.read_namespaced_service(self.name, self.namespace)
 79
 80    def update(self, current: V1Service, reference: V1Service):
 81        return self.api.patch_namespaced_service(
 82            current.metadata.name,
 83            self.namespace,
 84            reference,
 85            field_manager=FIELD_MANAGER,
 86        )
 87
 88
 89class MetricsServiceReconciler(ServiceReconciler):
 90    @property
 91    def noop(self) -> bool:
 92        return self.is_embedded
 93
 94    @staticmethod
 95    def reconciler_name() -> str:
 96        return "service-metrics"
 97
 98    @property
 99    def name(self):
100        name_suffix = "-metrics"
101        name = super().name
102        return name[: 63 - len(name_suffix)] + name_suffix
103
104    def get_object_meta(self, **kwargs) -> V1ObjectMeta:
105        meta: V1ObjectMeta = super().get_object_meta(**kwargs)
106        meta.labels["goauthentik.io/service-type"] = "metrics"
107        return meta
108
109    def get_reference_object(self) -> V1Service:
110        """Get deployment object for outpost"""
111        meta = self.get_object_meta(name=self.name)
112        ports = []
113        for port in self.controller.metrics_ports:
114            ports.append(
115                V1ServicePort(
116                    name=port.name,
117                    port=port.port,
118                    protocol=port.protocol.upper(),
119                    target_port=port.inner_port or port.port,
120                )
121            )
122        selector_labels = DeploymentReconciler(self.controller).get_pod_meta()
123        return V1Service(
124            metadata=meta,
125            spec=V1ServiceSpec(
126                ports=ports,
127                selector=selector_labels,
128                type="ClusterIP",
129            ),
130        )