Implement the job-scheduler logic

[AlexRuiz7]

Description

As part of the command manager plugin development and in continuation of #65, we are going to implement the job-scheduler logic to prioritize the commands and send them to the Wazuh Server's Management API.

Plan

• Implement the Job Runner class.
• Implement the Job Parameter class.
• Implement the Job Runner logic.

Functional requirements

• The job runner reads commands from the .commands index in PENDING status.
• The job runner sorts the commands by the timeout field (lower timeout, higher priority).
• The job runner sends the commands to the Management API (needs HTTP service implementation).*
• The job runner logs each of its actions properly.

• The job runner sends the commands to an external function for its processing. For the time being, we can just print these commands. Once the HTTP service implementation is completed, we can assemble both pieces.

[f-galland]

Other plugins seem to interface with JobScheduler through its Service Provider Interface:

• https://github.com/opensearch-project/sql/blob/063015c8290551090b9025687fa40078d435adfb/plugin/src/main/java/org/opensearch/sql/plugin/SQLPlugin.java#L47

[f-galland]

It looks like the Plugin class (the main class inheriting from OpenSearch's Plugin) needs to implement JobSchedulerExtension.

• https://github.com/opensearch-project/sql/blob/063015c8290551090b9025687fa40078d435adfb/plugin/src/main/java/org/opensearch/sql/plugin/SQLPlugin.java#L116C63-L116C84

[f-galland]

A separate class implements ScheduledJobRunner's runJob() which pushes the task to its own thread:

• https://github.com/opensearch-project/sql/blob/063015c8290551090b9025687fa40078d435adfb/async-query/src/main/java/org/opensearch/sql/spark/scheduler/job/ScheduledAsyncQueryJobRunner.java#L103

A javadoc in this class reads as follows:

 * The job runner class for scheduling async query.
 *
 * <p>The job runner should be a singleton class if it uses OpenSearch client or other objects
 * passed from OpenSearch. Because when registering the job runner to JobScheduler plugin,
 * OpenSearch has not invoked plugins' createComponents() method. That is saying the plugin is not
 * completely initialized, and the OpenSearch {@link org.opensearch.client.Client}, {@link
 * ClusterService} and other objects are not available to plugin and this job runner.
 *
 * <p>So we have to move this job runner initialization to {@link Plugin} createComponents() method,
 * and using singleton job runner to ensure we register a usable job runner instance to JobScheduler
 * plugin.

[f-galland]

The SQL plugin uses a model class for scheduled jobs which implements ScheduledJobParameter from JobScheduler:

• https://github.com/opensearch-project/sql/blob/063015c8290551090b9025687fa40078d435adfb/async-query/src/main/java/org/opensearch/sql/spark/scheduler/model/ScheduledAsyncQueryJobRequest.java#L25

[AlexRuiz7]

That research was already performed in #65

[f-galland]

#65 's PR only added job-scheduler to the command manager's gradle task. Job scheduler classes are not really being used over there.

[f-galland]

SampleExtensionRestHandler:

• Receives POST call parameters
• Instantiates SampleJobParameters with parameters from POST call
• Indexes the SampleJobParameter as a json object.

SampleExtensionPlugin:

• Exposes:
  • getJobType(): Returns a string with the job type
  • getJobIndex(): Returns the name of the index that holds the scheduled jobs' parameters
  • getJobRunner(): Returns the singleton instance of the plugin's Runner class
  • getJobParser(): Returns a ScheduledJobParser object that can parse the task's parameters

SampleJobParameter:

• Implemets getters and setters for every job parameter
• Implements toXContent(), which is used to index the job

SampleJobRunner:

• Implements runJob() which contains the job's logic.
• Receives a ScheduledJobParameter, which gives it access to the task's details
• Receives a JobExecutionContext which allows it to acquire a lock during the task execution time window.
• The task itself is wrapped inside a Runnable object that gets submitted to an Opensearch thread.

It seems like the only proper way to schedule tasks using the job scheduler is to store them as documents to an index.

This is evidenced by the fact that the only call of runJob comes from the reschedule() method from the JobScheduler class. The job parameters to this runJob() call can be traced back to the sweep() method from the JobSweeper class in turn.
Lastly, the sweep() method seems to parse the job parameters from a provided index.

• JobScheduler's only usage of runJob()
• sweep() method parsing the ScheduledJobParameters off an index shard