Python multi-threaded kubernetes watcher

Question

My program (which eventually will be containerised) should be able to do the following:

• continuously listening for deployment and daemonsets in kubernetes (2 threads)
• collect info needed (like annotations)
• use the info collected in order to create objects in a remote database

I could use some peer review for the thread part, which looks like this currently:

#first thread to watch for deployments
def watch_deployments():
 v1 = client.AppsV1Api()
 w = watch.Watch()
 last_seen_version = v1.list_deployment_for_all_namespaces().metadata.resource_version
 while True:
 for item in w.stream(v1.list_deployment_for_all_namespaces, pretty="pretty", resource_version=last_seen_version):
 _ = {item["object"].metadata.name:[item["object"].kind, item["object"].metadata.namespace, item["object"].metadata.resource_version, item["object"].metadata.annotations]}
 depl_lst.put(_)
def watch_daemonsets():
 v1 = client.AppsV1Api()
 w = watch.Watch()
 last_seen_version = v1.list_daemon_set_for_all_namespaces().metadata.resource_version
 while True:
 for item in w.stream(v1.list_daemon_set_for_all_namespaces, pretty="pretty", resource_version=last_seen_version):
 _ = {item["object"].metadata.name:[item["object"].kind, item["object"].metadata.namespace, item["object"].metadata.resource_version, item["object"].metadata.annotations]}
 depl_lst.put(_)
if __name__ == '__main__':
 current_obj = {}
 depl_lst = Queue()
 thread.start_new_thread(watch_deployments, ())
 thread.start_new_thread(watch_daemonsets, ())
 while True:
 for i in range(depl_lst.qsize()):
 current_obj = depl_lst.get()
 #add object is the function in order to create the item in the remote database, not listed here
 add_object()
 depl_lst.task_done()

Its the same thing with what is being done here but in this case, the asyncio was being used: https://medium.com/@sebgoa/kubernets-async-watches-b8fa8a7ebfd4

Answer 1

Restructuring and consolidation

Dealing with API client instance
Instead 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 queue
Initiating 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()