Implement Content Updater JobScheduler

CHANGELOG.md

@@ -21,6 +21,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 - Implement RBAC "refresh" command [(#370)](https://github.com/wazuh/wazuh-indexer-plugins/pull/370)
 - Implement content update using Json Patch operations [(#362)](https://github.com/wazuh/wazuh-indexer-plugins/pull/362)
 - Implement CTI API client rate limit and enhanced response handling [(#363)](https://github.com/wazuh/wazuh-indexer-plugins/pull/363)
+- Implement Content Updater JobScheduler [(#387)](https://github.com/wazuh/wazuh-indexer-plugins/pull/387)
 - Add custom action to list modified plugins [(#388)](https://github.com/wazuh/wazuh-indexer-plugins/pull/388)
 - Expose settings for the Content Manager plugin [(#395)](https://github.com/wazuh/wazuh-indexer-plugins/pull/395)
 - Add content update to to the Content Manager init process [(#421)](https://github.com/wazuh/wazuh-indexer-plugins/pull/421)

plugins/content-manager/build.gradle

@@ -6,6 +6,7 @@ buildscript {
     ext {
         opensearch_version = System.getProperty("opensearch.version", "2.19.2-SNAPSHOT")
         opensearch_build = opensearch_version.replace("-SNAPSHOT","") + ".0"
+        job_scheduler_version = System.getProperty("job_scheduler.version", opensearch_build)
         wazuh_version = System.getProperty("version", "6.0.0")
         revision = System.getProperty("revision", "0")
     }
@@ -26,7 +27,7 @@ apply plugin: 'java'
 apply plugin: 'idea'
 apply plugin: 'eclipse'
 apply plugin: 'opensearch.opensearchplugin'
-apply plugin: 'opensearch.yaml-rest-test'
+apply plugin: 'opensearch.rest-test'
 apply plugin: 'opensearch.pluginzip'
 
 def pluginName = 'wazuh-indexer-content-manager'
@@ -69,6 +70,7 @@ opensearchplugin {
     name pluginName
     description pluginDescription
     classname "${projectPath}.${pathToPlugin}.${pluginClassName}"
+    extendedPlugins = ['opensearch-job-scheduler']
     licenseFile rootProject.file('LICENSE.txt')
     noticeFile rootProject.file('NOTICE.txt')
 }
@@ -89,6 +91,7 @@ dependencies {
     // Uncomment this line to load the command-manager plugin from the local Maven repository.
     // Keep it commented to allow GitHub Workflows to run the tests.
     // zipArchive group: 'com.wazuh', name:'wazuh-indexer-command-manager', version: "${wazuh_version}.${revision}"
+    compileOnly "org.opensearch:opensearch-job-scheduler-spi:${job_scheduler_version}"
 }
 
 // No need to validate license headers since spotless checks and applies it
@@ -121,41 +124,44 @@ dependencies {
     implementation 'com.google.code.gson:gson:2.8.9'
 }
 
-task integTest(type: RestIntegTestTask) {
-    description = "Run tests against a cluster"
-    testClassesDirs = sourceSets.test.output.classesDirs
-    classpath = sourceSets.test.runtimeClasspath
-}
-tasks.named("check").configure { dependsOn(integTest) }
-
 integTest {
     // The --debug-jvm command-line option makes the cluster debuggable; this makes the tests debuggable
     if (System.getProperty("test.debug") != null) {
         jvmArgs '-agentlib:jdwp=transport=dt_socket,server=y,suspend=y,address=*:5005'
     }
+    environment "IS_DEV", System.getenv("IS_DEV") ?: "false"
 }
 
 testClusters.integTest {
-    // add customized keystore
-    keystore 'indexer.username', 'admin'
-    keystore 'indexer.password', 'admin'
-    // Installs all registered zipArchive dependencies on integTest cluster nodes
-    configurations.zipArchive.asFileTree.each {
-      plugin(provider(new Callable<RegularFile>(){
-        @Override
-        RegularFile call() throws Exception {
-          return new RegularFile() {
-            @Override
-            File getAsFile() {
-              return it
-            }
+  testDistribution = "ARCHIVE"
+  // add customized keystore
+  keystore 'indexer.username', 'admin'
+  keystore 'indexer.password', 'admin'
+  // Installs all registered zipArchive dependencies on integTest cluster nodes
+  configurations.zipArchive.asFileTree.each {
+    plugin(provider(new Callable<RegularFile>(){
+      @Override
+      RegularFile call() throws Exception {
+        return new RegularFile() {
+          @Override
+          File getAsFile() {
+            return it
           }
         }
-      }))
+      }
+    }))
+  }
+  // This installs our plugin into the testClusters
+  plugin(project.tasks.bundlePlugin.archiveFile)
+  if (System.getProperty("cluster.debug") != null) {
+    def debugPort = 5005
+    nodes.forEach { node ->
+      node.jvmArgs("-agentlib:jdwp=transport=dt_socket,server=y,suspend=y,address=*:${debugPort}")
+      println "Debugging ${node.name} on port ${debugPort} — waiting for debugger to attach..."
+      debugPort += 1
     }
-    testDistribution = "ARCHIVE"
-    // This installs our plugin into the testClusters
-    plugin(project.tasks.bundlePlugin.archiveFile)
+  }
+  environment "IS_DEV", System.getenv("IS_DEV") ?: "false"
 }
 
 run {

plugins/content-manager/src/main/java/com/wazuh/contentmanager/ContentManagerPlugin.java

@@ -18,39 +18,64 @@
 
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
+import org.opensearch.action.index.IndexRequest;
+import org.opensearch.action.index.IndexResponse;
+import org.opensearch.action.support.WriteRequest;
 import org.opensearch.client.Client;
 import org.opensearch.cluster.metadata.IndexNameExpressionResolver;
 import org.opensearch.cluster.node.DiscoveryNode;
 import org.opensearch.cluster.service.ClusterService;
 import org.opensearch.common.settings.*;
+import org.opensearch.common.xcontent.json.JsonXContent;
 import org.opensearch.core.common.io.stream.NamedWriteableRegistry;
 import org.opensearch.core.xcontent.NamedXContentRegistry;
+import org.opensearch.core.xcontent.XContentParser;
+import org.opensearch.core.xcontent.XContentParserUtils;
 import org.opensearch.env.Environment;
 import org.opensearch.env.NodeEnvironment;
+import org.opensearch.jobscheduler.spi.JobSchedulerExtension;
+import org.opensearch.jobscheduler.spi.ScheduledJobParser;
+import org.opensearch.jobscheduler.spi.ScheduledJobRunner;
+import org.opensearch.jobscheduler.spi.schedule.IntervalSchedule;
+import org.opensearch.jobscheduler.spi.schedule.ScheduleParser;
+import org.opensearch.plugins.ActionPlugin;
 import org.opensearch.plugins.ClusterPlugin;
 import org.opensearch.plugins.Plugin;
 import org.opensearch.repositories.RepositoriesService;
 import org.opensearch.script.ScriptService;
 import org.opensearch.threadpool.ThreadPool;
 import org.opensearch.watcher.ResourceWatcherService;
 
+import java.io.IOException;
+import java.time.Instant;
+import java.time.temporal.ChronoUnit;
 import java.util.*;
+import java.util.concurrent.TimeUnit;
 import java.util.function.Supplier;
 
+import com.wazuh.contentmanager.client.CTIClient;
 import com.wazuh.contentmanager.index.ContentIndex;
 import com.wazuh.contentmanager.index.ContextIndex;
+import com.wazuh.contentmanager.jobscheduler.ContentUpdaterJobParameter;
+import com.wazuh.contentmanager.jobscheduler.ContentUpdaterJobRunner;
 import com.wazuh.contentmanager.settings.PluginSettings;
 import com.wazuh.contentmanager.utils.Privileged;
-import com.wazuh.contentmanager.utils.SnapshotManager;
 
 /** Main class of the Content Manager Plugin */
-public class ContentManagerPlugin extends Plugin implements ClusterPlugin {
+public class ContentManagerPlugin extends Plugin
+        implements ClusterPlugin, ActionPlugin, JobSchedulerExtension {
+
     private static final Logger log = LogManager.getLogger(ContentManagerPlugin.class);
+
+    /** Scheduled jobs index name. */
+    public static final String JOB_INDEX = ".content_updater_jobs";
+
+    /** Scheduled job ID. */
+    public static final String JOB_ID = "content_updater_job";
+
+    private Client client;
     private ContextIndex contextIndex;
     private ContentIndex contentIndex;
-    private SnapshotManager snapshotManager;
-    private ThreadPool threadPool;
-    private ClusterService clusterService;
 
     @Override
     public Collection<Object> createComponents(
@@ -65,13 +90,21 @@ public Collection<Object> createComponents(
             NamedWriteableRegistry namedWriteableRegistry,
             IndexNameExpressionResolver indexNameExpressionResolver,
             Supplier<RepositoriesService> repositoriesServiceSupplier) {
+
         PluginSettings.getInstance(environment.settings(), clusterService);
-        this.threadPool = threadPool;
-        this.clusterService = clusterService;
+
+        this.client = client;
         this.contextIndex = new ContextIndex(client);
         this.contentIndex = new ContentIndex(client);
-        this.snapshotManager =
-                new SnapshotManager(environment, this.contextIndex, this.contentIndex, new Privileged());
+        Privileged privileged = new Privileged();
+        ContentUpdaterJobRunner.getInstance(
+                privileged.doPrivilegedRequest(CTIClient::getInstance),
+                threadPool,
+                environment,
+                this.contextIndex,
+                contentIndex,
+                privileged);
+
         return Collections.emptyList();
     }
 
@@ -84,45 +117,37 @@ public Collection<Object> createComponents(
      */
     @Override
     public void onNodeStarted(DiscoveryNode localNode) {
-        // Only cluster managers are responsible for the initialization.
-        if (localNode.isClusterManagerNode()) {
-            if (this.clusterService.state().routingTable().hasIndex(ContentIndex.INDEX_NAME)) {
-                this.start();
-            }
-
-            // To be removed once we include the Job Scheduler.
-            this.clusterService.addListener(
-                    event -> {
-                        if (event.indicesCreated().contains(ContentIndex.INDEX_NAME)) {
-                            this.start();
-                        }
-                    });
+        // @TODO do we want an IS_DEV variable?
+        if (System.getenv("IS_DEV").equals("true") && localNode.isClusterManagerNode()) {
+            this.contextIndex.createIndex();
+            this.contentIndex.createIndex();
+        }
+        try {
+            log.info(
+                    "Scheduled content update job with status: [{}]", scheduleContentUpdateJob().getResult());
+        } catch (IOException e) {
+            log.error("Failed scheduling content update job: {}", e.getMessage());
         }
     }
 
+    private IndexResponse scheduleContentUpdateJob() throws IOException {
+        IntervalSchedule schedule = new IntervalSchedule(Instant.now(), 1, ChronoUnit.MINUTES);
+        ContentUpdaterJobParameter jobParameter =
+                new ContentUpdaterJobParameter("update_content", schedule);
+        IndexRequest indexRequest =
+                new IndexRequest()
+                        .index(JOB_INDEX)
+                        .id(JOB_ID)
+                        .source(jobParameter.toXContent(JsonXContent.contentBuilder(), null))
+                        .setRefreshPolicy(WriteRequest.RefreshPolicy.IMMEDIATE);
+        return this.client.index(indexRequest).actionGet(5, TimeUnit.SECONDS);
+    }
+
     /**
-     * Initialize. The initialization consists of:
+     * Returns the settings for the plugin.
      *
-     * <pre>
-     *     1. fetching the latest consumer's information from the CTI API.
-     *     2. initialize from a snapshot if the local consumer does not exist, or its offset is 0.
-     * </pre>
+     * @return the settings for the plugin
      */
-    private void start() {
-        try {
-            this.threadPool
-                    .generic()
-                    .execute(
-                            () -> {
-                                this.contextIndex.createIndex();
-                                this.snapshotManager.initialize();
-                            });
-        } catch (Exception e) {
-            // Log or handle exception
-            log.error("Error initializing snapshot helper: {}", e.getMessage(), e);
-        }
-    }
-
     @Override
     public List<Setting<?>> getSettings() {
         return Arrays.asList(
@@ -140,4 +165,64 @@ public List<Setting<?>> getSettings() {
                 PluginSettings.MAX_CONCURRENT_BULKS,
                 PluginSettings.MAX_ITEMS_PER_BULK);
     }
+
+    @Override
+    public String getJobType() {
+        return "content_updater";
+    }
+
+    @Override
+    public String getJobIndex() {
+        return ".content_updater_jobs";
+    }
+
+    @Override
+    public ScheduledJobRunner getJobRunner() {
+        return ContentUpdaterJobRunner.getInstance();
+    }
+
+    @Override
+    public ScheduledJobParser getJobParser() {
+        return (parser, id, jobDocVersion) -> {
+            ContentUpdaterJobParameter jobParameter = new ContentUpdaterJobParameter();
+            XContentParserUtils.ensureExpectedToken(
+                    XContentParser.Token.START_OBJECT, parser.nextToken(), parser);
+
+            while (!parser.nextToken().equals(XContentParser.Token.END_OBJECT)) {
+                String fieldName = parser.currentName();
+                parser.nextToken();
+                switch (fieldName) {
+                    case ContentUpdaterJobParameter.NAME_FIELD:
+                        jobParameter.setJobName(parser.text());
+                        break;
+                    case ContentUpdaterJobParameter.ENABLED_FIELD:
+                        jobParameter.setEnabled(parser.booleanValue());
+                        break;
+                    case ContentUpdaterJobParameter.ENABLED_TIME_FIELD:
+                        jobParameter.setEnabledTime(parseInstantValue(parser));
+                        break;
+                    case ContentUpdaterJobParameter.LAST_UPDATE_TIME_FIELD:
+                        jobParameter.setLastUpdateTime(parseInstantValue(parser));
+                        break;
+                    case ContentUpdaterJobParameter.SCHEDULE_FIELD:
+                        jobParameter.setSchedule(ScheduleParser.parse(parser));
+                        break;
+                    default:
+                        XContentParserUtils.throwUnknownToken(parser.currentToken(), parser.getTokenLocation());
+                }
+            }
+            return jobParameter;
+        };
+    }
+
+    private Instant parseInstantValue(XContentParser parser) throws IOException {
+        if (XContentParser.Token.VALUE_NULL.equals(parser.currentToken())) {
+            return null;
+        }
+        if (parser.currentToken().isValue()) {
+            return Instant.ofEpochMilli(parser.longValue());
+        }
+        XContentParserUtils.throwUnknownToken(parser.currentToken(), parser.getTokenLocation());
+        return null;
+    }
 }