[worker] fix starting pingAlive thread once

[sbocahu]

Proposed changes

• Start pingAlive once in the main loop
• Use consume() generator instead of callable + basic_consume to handle graceful stops
• handle graceful stops for sigint (control-c) or sigterm
• improved exception handling
• rejection of faulty messages in AMQP listening mode (if message is not json)

Checklist

• I consider the submitted work as finished
• I tested the code for its functionality
• I wrote test cases for the relevant uses case (coverage and e2e)
• I added/update the relevant documentation (either on github or on notion)
• Where necessary I refactored code to improve the overall quality

Further comments

Before this change, a worker would create a pingAlive thread per rabbitmq queue, which is useless & consumes resources.
Also, on exceptions (eg. a message which is not valid json), the consuption would fail and restart, creating a new pingAlive thread. Exceptions during ingestion would then lead to an ever increasing number of these pingAlive threads.
A comment suggested graceful stop was handled but it was incomplete/incorrect.

[sbocahu]

just got an exception that looks not great. converting to draft

[sbocahu]

did not test much this part but I ran into issues with external import connectors. looking at the code, it seems to be wrong to use listen_callback_uri as a condition to create an ApiConsumer because it is always set up by the connector helper (https://github.com/OpenCTI-Platform/client-python/blob/073609f05b968455e9ecb905612bdedb523c14dd/pycti/connector/opencti_connector_helper.py#L1105)

[sbocahu]

resolving with OpenCTI-Platform/client-python#867

[sbocahu]

just got an exception that looks not great. converting to draft

the exception was due to a control-C/sigterm & introduced earlier with the addition of the threadpool.
I reworked the Consumer object to handle graceful stop, which was not the case previously (despite the comment saying so).

TODO:

• rework apiConsumer for graceful stop
• check the condition around launching an ApiConsumer
• handle sigkill similarly to sigint

[codecov]

Codecov Report

All modified and coverable lines are covered by tests ✅

Project coverage is 65.44%. Comparing base (6d44674) to head (aaa8957).
Report is 11 commits behind head on master.

Additional details and impacted files

@@           Coverage Diff           @@
##           master   #10428   +/-   ##
=======================================
  Coverage   65.44%   65.44%           
=======================================
  Files         672      672           
  Lines       67031    67031           
  Branches     7347     7354    +7     
=======================================
  Hits        43870    43870           
  Misses      23161    23161           

☔ View full report in Codecov by Sentry.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.
• 📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.

[sbocahu]

sibbling PR in pycti: OpenCTI-Platform/client-python#867

[aHenryJard]

@sbocahu I was about to test locally but there is a conflict with current master, I'm not sure on how to resolve it. Could you rebase and manage the conflit ?

[sbocahu]

done

[aHenryJard]

Tested locally, all good to me. I used some connectors, and also stopped opencti for a while to see if the worker is correctly taking back work when opencti is up again and it's all good.

[richard-julien]

Please dont merge until my review

[richard-julien]

Need review and change