outposts: fix compare_ports to support both service and container ports
Signed-off-by: Jens Langhammer <jens.langhammer@beryju.org>
This commit is contained in:
parent
91227b1e96
commit
ef2eed0bdf
|
@ -1,6 +1,7 @@
|
||||||
"""k8s utils"""
|
"""k8s utils"""
|
||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
|
|
||||||
|
from kubernetes.client.models.v1_container_port import V1ContainerPort
|
||||||
from kubernetes.client.models.v1_service_port import V1ServicePort
|
from kubernetes.client.models.v1_service_port import V1ServicePort
|
||||||
from kubernetes.config.incluster_config import SERVICE_TOKEN_FILENAME
|
from kubernetes.config.incluster_config import SERVICE_TOKEN_FILENAME
|
||||||
|
|
||||||
|
@ -16,19 +17,28 @@ def get_namespace() -> str:
|
||||||
return "default"
|
return "default"
|
||||||
|
|
||||||
|
|
||||||
def compare_port(current: V1ServicePort, reference: V1ServicePort) -> bool:
|
def compare_port(
|
||||||
|
current: V1ServicePort | V1ContainerPort, reference: V1ServicePort | V1ContainerPort
|
||||||
|
) -> bool:
|
||||||
"""Compare a single port"""
|
"""Compare a single port"""
|
||||||
if current.name != reference.name:
|
if current.name != reference.name:
|
||||||
return False
|
return False
|
||||||
|
if current.protocol != reference.protocol:
|
||||||
|
return False
|
||||||
|
if isinstance(current, V1ServicePort) and isinstance(reference, V1ServicePort):
|
||||||
# We only care about the target port
|
# We only care about the target port
|
||||||
if current.target_port != reference.target_port:
|
if current.target_port != reference.target_port:
|
||||||
return False
|
return False
|
||||||
if current.protocol != reference.protocol:
|
if isinstance(current, V1ContainerPort) and isinstance(reference, V1ContainerPort):
|
||||||
|
# We only care about the target port
|
||||||
|
if current.container_port != reference.container_port:
|
||||||
return False
|
return False
|
||||||
return True
|
return True
|
||||||
|
|
||||||
|
|
||||||
def compare_ports(current: list[V1ServicePort], reference: list[V1ServicePort]):
|
def compare_ports(
|
||||||
|
current: list[V1ServicePort | V1ContainerPort], reference: list[V1ServicePort | V1ContainerPort]
|
||||||
|
):
|
||||||
"""Compare ports of a list"""
|
"""Compare ports of a list"""
|
||||||
if len(current) != len(reference):
|
if len(current) != len(reference):
|
||||||
raise NeedsRecreate()
|
raise NeedsRecreate()
|
||||||
|
|
Reference in a new issue