def get_daemonset_term_handler()

in standalone/aws_eks_k8s_client.py [0:0]

• 19 lines of code
• 6 McCabe index (conditional complexity)

def get_daemonset_term_handler(session,region,account_id, cluster_name, ds_id="aws-node-termination-handler") -> dict:
    api_client = get_k8s_api_client(session,region,account_id, cluster_name)
    api_apps = apps_v1_api.AppsV1Api(api_client)

    result_dict = {'term_handler_available':'NO', 'node_selector_spot':'NO'}

    print("get daemonset - accessing k8s api through client:   " )
    #print("type=" + str(type(api_apps.read_namespaced_daemon_set("aws-node", 'kube-system'))))
    #ds_id = "aws-node"
    try:
        resp = api_apps.read_namespaced_daemon_set(ds_id, 'kube-system')
        result_dict['term_handler_available']='YES'

        # get node selector. assign points if it is lifecycle=Ec2Spot
        node_selector = resp.spec.template.spec.node_selector
        print("Node selector:" + str(node_selector))
        if node_selector == "lifecycle=Ec2Spot":
            result_dict['node_selector_spot'] = 'YES'
    except ApiException as ex:
        print("ApiException:"+ str(ex.reason))
        print("ApiException:" + str(ex))
        #assign negative points if node termination handler has not be deployed.
    except Exception as excp:
        result_dict['term_handler_available'] = 'EXCEPTION'
        print("Exception:" + str(excp))
    return result_dict