torn umbra
I have a function that is being run in 3 worker threads, they keep running until an exception occurs after which the worker thread that was executing it is killed. after a while all my worker threads are dead.
def process_tasks(worker_id):
try:
while True:
payload = task_queue.get()
if payload is None: # Exit signal
break
logger.debug("Worker %d processing task %s :", worker_id, payload.get('pod'))
command = f"kubectl --kubeconfig {path} exec '{payload.get('pod')}' -- /usr/bin/thread-dump"
result = subprocess.run(
command,
check=True,
shell=True,
capture_output=True
)
logger.info("stdout: %s", result.stdout.decode('utf-8'))
task_queue.task_done()
except KeyError as key_error:
logger.error("KeyError: %s", key_error, exc_info=True)
except subprocess.CalledProcessError as e:
logger.error("CalledProcessError: %s", e)
logger.error("stdout: %s", e.stdout.decode('utf-8'))
logger.error("stderr: %s", e.stderr.decode('utf-8'))
except Exception as e:
logger.error("Exception: %s", e, exc_info=True)
finally:
task_queue.task_done()
@app.post("/enqueue", dependencies=[Depends(k8s_utils.kubeconfig_status)])
async def enqueue_task(alert_data: PromAlertData):
logger.debug(f"Received alert data for alert {alert_data.alertname} in namespace {alert_data.namespace}, pod {alert_data.pod}")
task_queue.put(alert_data.model_dump())
return {"message": "Task added to queue"}
@app.on_event("startup")
def startup():
startscript()
global worker_threads
worker_threads = []
for _ in range(3):
worker_thread = threading.Thread(target=process_tasks, args=(_,), daemon=True)
worker_thread.start()
worker_threads.append(worker_thread)```
Need help in fixing this behavior
barren isle
jaunty atlas
autumn pollen
