This repository has been archived on 2024-05-31. You can view files and clone it, but cannot push or open issues or pull requests.
authentik/passbook/outposts/controllers/k8s/service.py

61 lines
2.0 KiB
Python
Raw Normal View History

2020-10-14 15:49:09 +00:00
"""Kubernetes Service Reconciler"""
from typing import TYPE_CHECKING
2020-10-14 15:49:09 +00:00
from kubernetes.client import CoreV1Api, V1Service, V1ServicePort, V1ServiceSpec
from passbook.outposts.controllers.k8s.base import (
KubernetesObjectReconciler,
NeedsUpdate,
)
if TYPE_CHECKING:
from passbook.outposts.controllers.kubernetes import KubernetesController
2020-10-14 15:49:09 +00:00
class ServiceReconciler(KubernetesObjectReconciler[V1Service]):
"""Kubernetes Service Reconciler"""
def __init__(self, controller: "KubernetesController") -> None:
super().__init__(controller)
2020-10-14 15:49:09 +00:00
self.api = CoreV1Api()
@property
def name(self) -> str:
return f"passbook-outpost-{self.controller.outpost.name}"
2020-10-14 15:49:09 +00:00
def reconcile(self, current: V1Service, reference: V1Service):
if len(current.spec.ports) != len(reference.spec.ports):
raise NeedsUpdate()
for port in reference.spec.ports:
if port not in current.spec.ports:
raise NeedsUpdate()
def get_reference_object(self) -> V1Service:
"""Get deployment object for outpost"""
meta = self.get_object_meta(name=self.name)
2020-10-14 15:49:09 +00:00
ports = []
for port_name, port in self.controller.deployment_ports.items():
2020-10-14 15:49:09 +00:00
ports.append(V1ServicePort(name=port_name, port=port))
return V1Service(
metadata=meta,
spec=V1ServiceSpec(ports=ports, selector=meta.labels, type="ClusterIP"),
)
def create(self, reference: V1Service):
return self.api.create_namespaced_service(self.namespace, reference)
def delete(self, reference: V1Service):
return self.api.delete_namespaced_service(
reference.metadata.name, self.namespace
)
def retrieve(self) -> V1Service:
return self.api.read_namespaced_service(
f"passbook-outpost-{self.controller.outpost.name}", self.namespace
2020-10-14 15:49:09 +00:00
)
def update(self, current: V1Service, reference: V1Service):
return self.api.patch_namespaced_service(
current.metadata.name, self.namespace, reference
)