authentik.outposts.controllers.k8s.service_monitor

  1"""Kubernetes Prometheus ServiceMonitor Reconciler"""
  2
  3from dataclasses import asdict, dataclass, field
  4from typing import TYPE_CHECKING
  5
  6from dacite.core import from_dict
  7from kubernetes.client import ApiextensionsV1Api, CustomObjectsApi
  8
  9from authentik.outposts.controllers.base import FIELD_MANAGER
 10from authentik.outposts.controllers.k8s.base import KubernetesObjectReconciler
 11from authentik.outposts.controllers.k8s.service import MetricsServiceReconciler
 12from authentik.outposts.controllers.k8s.triggers import NeedsUpdate
 13
 14if TYPE_CHECKING:
 15    from authentik.outposts.controllers.kubernetes import KubernetesController
 16
 17
 18@dataclass(slots=True)
 19class PrometheusServiceMonitorSpecEndpoint:
 20    """Prometheus ServiceMonitor endpoint spec"""
 21
 22    port: str
 23    path: str = field(default="/metrics")
 24
 25
 26@dataclass(slots=True)
 27class PrometheusServiceMonitorSpecSelector:
 28    """Prometheus ServiceMonitor selector spec"""
 29
 30    matchLabels: dict
 31
 32
 33@dataclass(slots=True)
 34class PrometheusServiceMonitorSpec:
 35    """Prometheus ServiceMonitor spec"""
 36
 37    endpoints: list[PrometheusServiceMonitorSpecEndpoint]
 38
 39    selector: PrometheusServiceMonitorSpecSelector
 40
 41
 42@dataclass(slots=True)
 43class PrometheusServiceMonitorMetadata:
 44    """Prometheus ServiceMonitor metadata"""
 45
 46    name: str
 47    namespace: str
 48    labels: dict = field(default_factory=dict)
 49
 50
 51@dataclass(slots=True)
 52class PrometheusServiceMonitor:
 53    """Prometheus ServiceMonitor"""
 54
 55    apiVersion: str
 56    kind: str
 57    metadata: PrometheusServiceMonitorMetadata
 58    spec: PrometheusServiceMonitorSpec
 59
 60    def to_dict(self):
 61        """`to_dict` to conform to how the kubernetes client converts objects to dicts"""
 62        return asdict(self)
 63
 64
 65CRD_NAME = "servicemonitors.monitoring.coreos.com"
 66CRD_GROUP = "monitoring.coreos.com"
 67CRD_VERSION = "v1"
 68CRD_PLURAL = "servicemonitors"
 69
 70
 71class PrometheusServiceMonitorReconciler(KubernetesObjectReconciler[PrometheusServiceMonitor]):
 72    """Kubernetes Prometheus ServiceMonitor Reconciler"""
 73
 74    def __init__(self, controller: KubernetesController) -> None:
 75        super().__init__(controller)
 76        self.api_ex = ApiextensionsV1Api(controller.client)
 77        self.api = CustomObjectsApi(controller.client)
 78
 79    @staticmethod
 80    def reconciler_name() -> str:
 81        return "prometheus servicemonitor"
 82
 83    def reconcile(self, current: PrometheusServiceMonitor, reference: PrometheusServiceMonitor):
 84        if current.spec.selector.matchLabels != reference.spec.selector.matchLabels:
 85            raise NeedsUpdate()
 86        super().reconcile(current, reference)
 87
 88    @property
 89    def noop(self) -> bool:
 90        if not self._crd_exists():
 91            self.logger.debug("CRD doesn't exist")
 92            return True
 93        return self.is_embedded
 94
 95    def _crd_exists(self) -> bool:
 96        """Check if the Prometheus ServiceMonitor exists"""
 97        return bool(
 98            len(
 99                self.api_ex.list_custom_resource_definition(
100                    field_selector=f"metadata.name={CRD_NAME}"
101                ).items
102            )
103        )
104
105    def get_reference_object(self) -> PrometheusServiceMonitor:
106        """Get service monitor object for outpost"""
107        return PrometheusServiceMonitor(
108            apiVersion=f"{CRD_GROUP}/{CRD_VERSION}",
109            kind="ServiceMonitor",
110            metadata=PrometheusServiceMonitorMetadata(
111                name=self.name,
112                namespace=self.namespace,
113                labels=self.get_object_meta().labels,
114            ),
115            spec=PrometheusServiceMonitorSpec(
116                endpoints=[
117                    PrometheusServiceMonitorSpecEndpoint(
118                        port="http-metrics",
119                    )
120                ],
121                selector=PrometheusServiceMonitorSpecSelector(
122                    matchLabels=MetricsServiceReconciler(self.controller)
123                    .get_object_meta(name=self.name)
124                    .labels,
125                ),
126            ),
127        )
128
129    def create(self, reference: PrometheusServiceMonitor):
130        return self.api.create_namespaced_custom_object(
131            group=CRD_GROUP,
132            version=CRD_VERSION,
133            plural=CRD_PLURAL,
134            namespace=self.namespace,
135            body=asdict(reference),
136            field_manager=FIELD_MANAGER,
137        )
138
139    def delete(self, reference: PrometheusServiceMonitor):
140        return self.api.delete_namespaced_custom_object(
141            group=CRD_GROUP,
142            version=CRD_VERSION,
143            namespace=self.namespace,
144            plural=CRD_PLURAL,
145            name=self.name,
146        )
147
148    def retrieve(self) -> PrometheusServiceMonitor:
149        return from_dict(
150            PrometheusServiceMonitor,
151            self.api.get_namespaced_custom_object(
152                group=CRD_GROUP,
153                version=CRD_VERSION,
154                namespace=self.namespace,
155                plural=CRD_PLURAL,
156                name=self.name,
157            ),
158        )
159
160    def update(self, current: PrometheusServiceMonitor, reference: PrometheusServiceMonitor):
161        return self.api.patch_namespaced_custom_object(
162            group=CRD_GROUP,
163            version=CRD_VERSION,
164            namespace=self.namespace,
165            plural=CRD_PLURAL,
166            name=self.name,
167            body=asdict(reference),
168            field_manager=FIELD_MANAGER,
169        )