Module hardeneks.namespace_based.reliability.applications
Expand source code
from ...resources import NamespacedResources
from ...report import (
print_pod_table,
print_service_table,
print_deployment_table,
)
def avoid_running_singleton_pods(namespaced_resources: NamespacedResources):
offenders = []
for pod in namespaced_resources.pods:
owner = pod.metadata.owner_references
if not owner:
offenders.append(pod)
if offenders:
print_pod_table(
offenders,
"[red]Avoid running pods without deployments.",
"Link: https://aws.github.io/aws-eks-best-practices/reliability/docs/application/#avoid-running-singleton-pods",
)
return offenders
def run_multiple_replicas(namespaced_resources: NamespacedResources):
offenders = []
for deployment in namespaced_resources.deployments:
if deployment.spec.replicas < 2:
offenders.append(deployment)
if offenders:
print_deployment_table(
offenders,
"[red]Avoid running single replica deployments",
"Link: https://aws.github.io/aws-eks-best-practices/reliability/docs/application/#run-multiple-replicas",
)
return offenders
def schedule_replicas_across_nodes(namespaced_resources: NamespacedResources):
offenders = []
for deployment in namespaced_resources.deployments:
spread = deployment.spec.template.spec.topology_spread_constraints
if not spread:
offenders.append(deployment)
else:
topology_keys = set([i.topology_key for i in spread])
if not set(["topology.kubernetes.io/zone"]).issubset(
topology_keys
):
offenders.append(deployment)
if offenders:
print_service_table(
offenders,
"[red]Spread replicas across AZs and Nodes",
"Link: https://aws.github.io/aws-eks-best-practices/reliability/docs/application/#schedule-replicas-across-nodes",
)
return offenders
def check_horizontal_pod_autoscaling_exists(
namespaced_resources: NamespacedResources,
):
offenders = []
hpas = [i.spec.scale_target_ref.name for i in namespaced_resources.hpas]
for deployment in namespaced_resources.deployments:
if deployment.metadata.name not in hpas:
offenders.append(deployment)
if offenders:
print_service_table(
offenders,
"[red]Deploy horizontal pod autoscaler for deployments",
"Link: https://aws.github.io/aws-eks-best-practices/reliability/docs/application/#horizontal-pod-autoscaler-hpa",
)
return offenders
def check_readiness_probes(namespaced_resources: NamespacedResources):
offenders = []
for pod in namespaced_resources.pods:
for container in pod.spec.containers:
if not container.readiness_probe:
offenders.append(pod)
if offenders:
print_pod_table(
offenders,
"[red]Define readiness probes for pods.",
"Link: https://aws.github.io/aws-eks-best-practices/reliability/docs/application/#use-readiness-probe-to-detect-partial-unavailability",
)
return offenders
def check_liveness_probes(namespaced_resources: NamespacedResources):
offenders = []
for pod in namespaced_resources.pods:
for container in pod.spec.containers:
if not container.liveness_probe:
offenders.append(pod)
if offenders:
print_pod_table(
offenders,
"[red]Define liveness probes for pods.",
"Link: https://aws.github.io/aws-eks-best-practices/reliability/docs/application/#use-liveness-probe-to-remove-unhealthy-pods",
)
return offenders
Functions
def avoid_running_singleton_pods(namespaced_resources: NamespacedResources)
-
Expand source code
def avoid_running_singleton_pods(namespaced_resources: NamespacedResources): offenders = [] for pod in namespaced_resources.pods: owner = pod.metadata.owner_references if not owner: offenders.append(pod) if offenders: print_pod_table( offenders, "[red]Avoid running pods without deployments.", "Link: https://aws.github.io/aws-eks-best-practices/reliability/docs/application/#avoid-running-singleton-pods", ) return offenders
def check_horizontal_pod_autoscaling_exists(namespaced_resources: NamespacedResources)
-
Expand source code
def check_horizontal_pod_autoscaling_exists( namespaced_resources: NamespacedResources, ): offenders = [] hpas = [i.spec.scale_target_ref.name for i in namespaced_resources.hpas] for deployment in namespaced_resources.deployments: if deployment.metadata.name not in hpas: offenders.append(deployment) if offenders: print_service_table( offenders, "[red]Deploy horizontal pod autoscaler for deployments", "Link: https://aws.github.io/aws-eks-best-practices/reliability/docs/application/#horizontal-pod-autoscaler-hpa", ) return offenders
def check_liveness_probes(namespaced_resources: NamespacedResources)
-
Expand source code
def check_liveness_probes(namespaced_resources: NamespacedResources): offenders = [] for pod in namespaced_resources.pods: for container in pod.spec.containers: if not container.liveness_probe: offenders.append(pod) if offenders: print_pod_table( offenders, "[red]Define liveness probes for pods.", "Link: https://aws.github.io/aws-eks-best-practices/reliability/docs/application/#use-liveness-probe-to-remove-unhealthy-pods", ) return offenders
def check_readiness_probes(namespaced_resources: NamespacedResources)
-
Expand source code
def check_readiness_probes(namespaced_resources: NamespacedResources): offenders = [] for pod in namespaced_resources.pods: for container in pod.spec.containers: if not container.readiness_probe: offenders.append(pod) if offenders: print_pod_table( offenders, "[red]Define readiness probes for pods.", "Link: https://aws.github.io/aws-eks-best-practices/reliability/docs/application/#use-readiness-probe-to-detect-partial-unavailability", ) return offenders
def run_multiple_replicas(namespaced_resources: NamespacedResources)
-
Expand source code
def run_multiple_replicas(namespaced_resources: NamespacedResources): offenders = [] for deployment in namespaced_resources.deployments: if deployment.spec.replicas < 2: offenders.append(deployment) if offenders: print_deployment_table( offenders, "[red]Avoid running single replica deployments", "Link: https://aws.github.io/aws-eks-best-practices/reliability/docs/application/#run-multiple-replicas", ) return offenders
def schedule_replicas_across_nodes(namespaced_resources: NamespacedResources)
-
Expand source code
def schedule_replicas_across_nodes(namespaced_resources: NamespacedResources): offenders = [] for deployment in namespaced_resources.deployments: spread = deployment.spec.template.spec.topology_spread_constraints if not spread: offenders.append(deployment) else: topology_keys = set([i.topology_key for i in spread]) if not set(["topology.kubernetes.io/zone"]).issubset( topology_keys ): offenders.append(deployment) if offenders: print_service_table( offenders, "[red]Spread replicas across AZs and Nodes", "Link: https://aws.github.io/aws-eks-best-practices/reliability/docs/application/#schedule-replicas-across-nodes", ) return offenders