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

• [x] Implement the Job Runner class.
• [x] Implement the Job Parameter class.
• [x] Implement the Job Runner logic.

Functional requirements

• [x] The job runner reads commands from the .commands index in PENDING status.

• [x] #126

• [x] 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

[f-galland]

Search results pagination can be achieved by means of two distinct methods:

1. Using SearchSourceBuilder's from() and size() which appear to be meant for user facing interfaces
2. Using Scroll and other related classes.

Solution 2 seems more robust (and is suggested for larger data batches).

I'm researching how official plugins handle iterating over the search result pages without blocking execution.

We have used the provided ThreadPool for this in past tests alongside simple while loops, but there seem to be more elegant solutions:

• https://github.com/opensearch-project/sql/blob/2.17.1.0/legacy/src/main/java/org/opensearch/sql/legacy/query/planner/physical/node/BatchPhysicalOperator.java

[f-galland]

As of commit 3fc33ea, the JobSchedulerExtension has been implemented as explained below:

• CommandManagerPlugin implements the JobSchedulerExtension interface
  • The overridden getJobRunner() method returns a Singleton instance of the CommandManagerJobRunner class
• CommandManagerJobRunner implements the ScheduledJobRunner interface, which provides the runJob() method.
• The runJob() method calls SearchJob's searchJobRunnable() which returns a Runnable.
  • The returned Runnable is submitted to a thread from ThreadPool.
• searchJobRunnable() is structured around a do while loop that issues a Point In Time search.
  • A handlePage() method will take each resulting SearchResponse (a page) and iterate over the SearchHits to update the command.status field from PENDING to SENT. This method also submits the command to the destination HTTP API.
• The runPitQuery() method is a wrapper around client.search(), using the synchronous variant of the method without any Future handling (this is all run in a separate thread at runJob() as described above anyway).
• pitSearchRequest() composes a SearchRequest object that consumes a SearchSourceBuilder object.
  • The SearchSourceBuilder object currently uses a hardcoded term query that looks for the value PENDING within the command.status field of the .commands index documents.
  • This search is sorted by order_id and timeout in order for the Point In Time pagination to work.
  • The SearchSourceBuilder object consumes a PointInTimeBuilder object that is handled by a custom PointInTime class.
  • If a null searchAfter Object[] is provided, this method returns a SearchRequest that can be used to pull the first page. Otherwise, it will create one that requests a subsequent page according to the Point In Time functionality.