def drain_pods()

in gamelift-daemon/src/main.py [0:0]

• 32 lines of code
• 5 McCabe index (conditional complexity)

def drain_pods():
    """This method evicts all Kubernetes pods from the specified node that are not in the kube-system namespace."""
    field_selector = 'spec.nodeName=' + ec2_metadata.private_hostname
    try:
        pods = core_v1_client.list_pod_for_all_namespaces(watch=False, field_selector=field_selector)
    except ApiException as e:
        print(f'Exception when calling CoreV1Api->list_pod_for_all_namespaces: {e}\n', flush=True)
    # Create a filtered list of pods not in the kube-system namespace
    filtered_pods = filter(lambda x: x.metadata.namespace != 'kube-system', pods.items)

    for pod in filtered_pods:
        print(f'Deleting pod {pod.metadata.name} in namespace {pod.metadata.namespace}', flush=True)
        if 'grace_period' in globals():
            body = {
                'apiVersion': 'policy/v1beta1',
                'kind': 'Eviction',
                'metadata': {
                    'name': pod.metadata.name,
                    'namespace': pod.metadata.namespace,
                    'grace_period_seconds': grace_period
                }
            }
        else:
            body = {
                'apiVersion': 'policy/v1beta1',
                'kind': 'Eviction',
                'metadata': {
                    'name': pod.metadata.name,
                    'namespace': pod.metadata.namespace
                }
            }
        try:
            core_v1_client.create_namespaced_pod_eviction(pod.metadata.name, pod.metadata.namespace, body)
        except ApiException as e:
            print(f'Exception when calling CoreV1Api->create_namespaced_pod_eviction: {e}\n', flush=True)