[client] Connector queuing auto backpressure - Scheduler (#6325)

[Megafredo]

Proposed changes

• Adding query "read" connector
• Update query "ping" connector
• Setting up a scheduler and process in helper
• New variable environment "queue_threshold", default = 500 Mo
• New variable environment "duration_period"

Related issues

• Some connectors will pause based on queues threshold opencti#6325

Checklist

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

Further comments

If this is a relatively large or complex change, kick off the discussion by explaining why you chose the solution you did and what alternatives you considered, etc...

[helene-nguyen]

Changing the duration_period from "PT5S" to "0" raised an error

  duration_period: '0'

Doing this works

Suggested change

	td = timedelta_adapter.validate_python(duration_period)
	td = timedelta_adapter.validate_python(int(duration_period))

BUT if we doing this, cannot handle the value in ISO8601 format anymore:

But the value of duration_period in int works if it's 0

  duration_period: 0

Let's talk about it :)

[helene-nguyen]

Tested locally and everything seems good to me, here are different cases I've tested:

• For duration_period time in ISO8601 format:
  • Case "0" in string, considered as run_and_terminate process => works
  • Case "PT0S" in string, considered as run_and_terminate process => works
  • Case 0 in int, considered as run_and_terminate process => works
  • Case "P0S" (deliberate mistake), an error is raised because it is not the good format
  • Case "PT3S" (for a scheduled period time of 3 seconds) =>everything works as expected

Steps to Reproduce and my understanding:

1. Stop the Worker:

   • Begin by stopping the worker process.

2. Run the Connector:

   • Start the connector process.

3. Scheduler Checks duration_period Configuration:

   • The scheduler will review the duration_period setting.

4. Main Process Execution:

   • The main process of the connector will start.
   • We must wait for this process to complete before entering the scheduler configuration and initialization.

5. Scheduler Activation:

   • The scheduler activates once the specified period time is reached.

6. Queue Size Threshold Check:

   • The scheduler checks the queue size threshold, defined as queue_threshold.
   • If no environment variable is set, the default threshold is 500MB.

7. Queue Size Condition:

   • If the connector's queue size exceeds the queue threshold, the message callback process (connector's main process) will pause.
   • The process will not resume until the queue is ingested and reduced sufficiently, allowing it to restart during the next scheduler check.

---

During the test, the connector was modified to run the scheduler as following:

    def process_message(self):
        """Main process of the connector."""
        [...]

    def run(self):
        self.helper.schedule_iso(
            message_callback=self.process_message,
            duration_period=self.config.duration_period,
        )

[helene-nguyen]

Could we add information as log about the next run here ?

[SouadHadjiat]

minor comment: since we get only one connector, I suggest to rename the query name to GetConnector without the s. Or just Connector

[SouadHadjiat]

this dependency seem to break connectors that use older versions of pydantic (1.10.17)