Python multi-threaded kubernetes watcher
Restructuring and consolidation

Dealing with API client instanceInstead of generating a new client.AppsV1Api instance in each separate thread - create the instance in the main thread and pass shared API client to your threads constructors:

if __name__ == '__main__':
    ...
    v1 = client.AppsV1Api()
    thread.start_new_thread(watch_deployments, (v1,))
    thread.start_new_thread(watch_daemonsets, (v1))
    ...




Both target functions watch_deployments and watch_daemonsets perform essentially the same set of actions and differ only in specific v1.list_... routine.To eliminate duplication the common behavior is extracted into a separate function say _gather_watched_metadata that will accept a particular v1.list_... function object as callable:

def _gather_watched_metadata(list_func):
    w = watch.Watch()
    last_seen_version = list_func().metadata.resource_version
    while True:
        for item in w.stream(list_func, pretty="pretty", resource_version=last_seen_version):
            metadata = item["object"].metadata
            _ = {metadata.name: [item["object"].kind, metadata.namespace,
                                 metadata.resource_version,
                                 metadata.annotations]}
            depl_lst.put(_)


As can be seen in the above function, to avoid repetitive indexing of nested attributes like item["object"].metadata it is worth to extract it into a variable at once.

Now, both target functions become just as below:

def watch_deployments(v1):
    _gather_watched_metadata(v1.list_deployment_for_all_namespaces)


def watch_daemonsets(v1):
    _gather_watched_metadata(v1.list_daemon_set_for_all_namespaces)


That can be shortened even further: both watch_... functions can be eliminated and you may run just with _gather_watched_metadata function and specific API routines passed as an argument :

...
v1 = client.AppsV1Api()
thread.start_new_thread(_gather_watched_metadata, (v1.list_deployment_for_all_namespaces,))
thread.start_new_thread(_gather_watched_metadata, (v1.list_daemon_set_for_all_namespaces,))


But to apply the last "shortening" - is up to you ...



Consuming depl_lst queueInitiating for loop with range in this context:

while True:
    for i in range(depl_lst.qsize()):
        ...


is redundant as it's enough to check if queue is not empty:

while True:
    if not depl_lst.empty():
        current_obj = depl_lst.get()