Communication mechanism for js

build.gradle

@@ -138,6 +138,8 @@ repositories {
 
 dependencies {
     implementation project(path: ":${rootProject.name}-spi", configuration: 'shadow')
+    implementation group: 'com.google.guava', name: 'guava', version:'31.0.1-jre'
+    implementation group: 'com.google.guava', name: 'failureaccess', version:'1.0.1'
     javaRestTestImplementation project.sourceSets.main.runtimeClasspath
 }
 

src/main/java/org/opensearch/jobscheduler/JobSchedulerPlugin.java

@@ -8,6 +8,17 @@
  */
 package org.opensearch.jobscheduler;
 
+import org.opensearch.action.ActionRequest;
+import org.opensearch.action.ActionResponse;
+import org.opensearch.cluster.node.DiscoveryNodes;
+import org.opensearch.common.settings.Setting;
+import org.opensearch.common.settings.Settings;
+import org.opensearch.common.settings.ClusterSettings;
+import org.opensearch.common.settings.IndexScopedSettings;
+import org.opensearch.common.settings.SettingsFilter;
+import org.opensearch.jobscheduler.model.JobDetails;
+import org.opensearch.jobscheduler.rest.RestGetJobIndexAction;
+import org.opensearch.jobscheduler.rest.RestGetJobTypeAction;
 import org.opensearch.jobscheduler.scheduler.JobScheduler;
 import org.opensearch.jobscheduler.spi.JobSchedulerExtension;
 import org.opensearch.jobscheduler.spi.ScheduledJobParser;
@@ -23,35 +34,43 @@
 import org.opensearch.cluster.service.ClusterService;
 import org.opensearch.common.ParseField;
 import org.opensearch.common.io.stream.NamedWriteableRegistry;
-import org.opensearch.common.settings.Setting;
-import org.opensearch.common.settings.Settings;
 import org.opensearch.common.util.concurrent.OpenSearchExecutors;
 import org.opensearch.common.xcontent.NamedXContentRegistry;
 import org.opensearch.env.Environment;
 import org.opensearch.env.NodeEnvironment;
 import org.opensearch.index.IndexModule;
+import org.opensearch.jobscheduler.transport.GetJobIndexAction;
+import org.opensearch.jobscheduler.transport.GetJobIndexTransportAction;
+import org.opensearch.jobscheduler.transport.GetJobTypeAction;
+import org.opensearch.jobscheduler.transport.GetJobTypeTransportAction;
+import org.opensearch.plugins.ActionPlugin;
 import org.opensearch.plugins.ExtensiblePlugin;
 import org.opensearch.plugins.Plugin;
 import org.opensearch.repositories.RepositoriesService;
+import org.opensearch.rest.RestController;
 import org.opensearch.script.ScriptService;
 import org.opensearch.threadpool.ExecutorBuilder;
 import org.opensearch.threadpool.FixedExecutorBuilder;
 import org.opensearch.threadpool.ThreadPool;
 import org.opensearch.watcher.ResourceWatcherService;
 
-import java.util.ArrayList;
-import java.util.Collection;
-import java.util.Collections;
 import java.util.HashMap;
 import java.util.HashSet;
-import java.util.List;
 import java.util.Map;
 import java.util.Set;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
+import java.util.Arrays;
+import java.util.ArrayList;
 import java.util.function.Supplier;
 
-public class JobSchedulerPlugin extends Plugin implements ExtensiblePlugin {
+import com.google.common.collect.ImmutableList;
+
+public class JobSchedulerPlugin extends Plugin implements ActionPlugin, ExtensiblePlugin {
 
     public static final String OPEN_DISTRO_JOB_SCHEDULER_THREAD_POOL_NAME = "open_distro_job_scheduler";
+    public static final String JS_BASE_URI = "/_plugins/_job_scheduler";
 
     private static final Logger log = LogManager.getLogger(JobSchedulerPlugin.class);
 
@@ -60,10 +79,12 @@ public class JobSchedulerPlugin extends Plugin implements ExtensiblePlugin {
     private LockService lockService;
     private Map<String, ScheduledJobProvider> indexToJobProviders;
     private Set<String> indicesToListen;
+    private Map<String, JobDetails> jobDetailsHashMap;
 
     public JobSchedulerPlugin() {
         this.indicesToListen = new HashSet<>();
         this.indexToJobProviders = new HashMap<>();
+        this.jobDetailsHashMap = new HashMap<>();
     }
 
     @Override
@@ -185,4 +206,27 @@ private JobSweeper initSweeper(
     ) {
         return new JobSweeper(settings, client, clusterService, threadPool, registry, this.indexToJobProviders, scheduler, lockService);
     }
+
+    @Override
+    public List getRestHandlers(
+        Settings settings,
+        RestController restController,
+        ClusterSettings clusterSettings,
+        IndexScopedSettings indexScopedSettings,
+        SettingsFilter settingsFilter,
+        IndexNameExpressionResolver indexNameExpressionResolver,
+        Supplier<DiscoveryNodes> nodesInCluster
+    ) {
+        RestGetJobIndexAction restGetJobIndexAction = new RestGetJobIndexAction(jobDetailsHashMap);
+        RestGetJobTypeAction restGetJobTypeAction = new RestGetJobTypeAction(jobDetailsHashMap);
+        return ImmutableList.of(restGetJobIndexAction, restGetJobTypeAction);
+    }
+
+    @Override
+    public List<ActionHandler<? extends ActionRequest, ? extends ActionResponse>> getActions() {
+        return Arrays.asList(
+            new ActionHandler<>(GetJobIndexAction.INSTANCE, GetJobIndexTransportAction.class),
+            new ActionHandler<>(GetJobTypeAction.INSTANCE, GetJobTypeTransportAction.class)
+        );
+    }
 }

src/main/java/org/opensearch/jobscheduler/constant/CommonValue.java

@@ -0,0 +1,14 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ * The OpenSearch Contributors require contributions made to
+ * this file be licensed under the Apache-2.0 license or a
+ * compatible open source license.
+ */
+package org.opensearch.jobscheduler.constant;
+
+public class CommonValue {
+
+    public static String EXTERNAL_ACTION_PREFIX = "cluster:admin/opendistro/js/";
+}

src/main/java/org/opensearch/jobscheduler/model/JobDetails.java

@@ -0,0 +1,164 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ * The OpenSearch Contributors require contributions made to
+ * this file be licensed under the Apache-2.0 license or a
+ * compatible open source license.
+ */
+package org.opensearch.jobscheduler.model;
+
+import org.opensearch.common.xcontent.ToXContent;
+import org.opensearch.common.xcontent.ToXContentObject;
+import org.opensearch.common.xcontent.XContentBuilder;
+import org.opensearch.common.xcontent.XContentParser;
+
+import java.io.IOException;
+import java.util.Objects;
+
+import static org.opensearch.common.xcontent.XContentParserUtils.ensureExpectedToken;
+
+/**
+ * This model class stores the job details of the extension.
+ */
+public class JobDetails implements ToXContentObject {
+
+    private String jobIndex;
+
+    private String jobType;
+
+    private String jobParserAction;
+
+    private String jobRunnerAction;
+
+    public static final String JOB_INDEX = "job_index";
+    public static final String JOB_TYPE = "job_type";
+    public static final String JOB_PARSER_ACTION = "job_parser_action";
+    public static final String JOB_RUNNER_ACTION = "job_runner_action";
+
+    public JobDetails() {}
+
+    public JobDetails(String jobIndex, String jobType, String jobParserAction, String jobRunnerAction) {
+        this.jobIndex = jobIndex;
+        this.jobType = jobType;
+        this.jobParserAction = jobParserAction;
+        this.jobRunnerAction = jobRunnerAction;
+    }
+
+    @Override
+    public XContentBuilder toXContent(XContentBuilder builder, ToXContent.Params params) throws IOException {
+        XContentBuilder xContentBuilder = builder.startObject();
+        if (jobIndex != null) {
+            xContentBuilder.field(JOB_INDEX, jobIndex);
+        }
+        if (jobType != null) {
+            xContentBuilder.field(JOB_TYPE, jobType);
+        }
+        if (jobParserAction != null) {
+            xContentBuilder.field(JOB_PARSER_ACTION, jobParserAction);
+        }
+        if (jobRunnerAction != null) {
+            xContentBuilder.field(JOB_RUNNER_ACTION, jobRunnerAction);
+        }
+        return xContentBuilder.endObject();
+    }
+
+    public static JobDetails parse(XContentParser parser) throws IOException {
+        String jobIndex = null;
+        String jobType = null;
+        String jobParserAction = null;
+        String jobRunnerAction = null;
+
+        ensureExpectedToken(XContentParser.Token.START_OBJECT, parser.currentToken(), parser);
+
+        while (parser.nextToken() != XContentParser.Token.END_OBJECT) {
+            String fieldName = parser.currentName();
+            parser.nextToken();
+            switch (fieldName) {
+                case JOB_INDEX:
+                    jobIndex = parser.text();
+                    break;
+                case JOB_TYPE:
+                    jobType = parser.text();
+                    break;
+                case JOB_PARSER_ACTION:
+                    jobParserAction = parser.text();
+                    break;
+                case JOB_RUNNER_ACTION:
+                    jobRunnerAction = parser.text();
+                    break;
+                default:
+                    parser.skipChildren();
+                    break;
+            }
+        }
+
+        return new JobDetails(jobIndex, jobType, jobParserAction, jobRunnerAction);
+    }
+
+    public String getJobIndex() {
+        return jobIndex;
+    }
+
+    public void setJobIndex(String jobIndex) {
+        this.jobIndex = jobIndex;
+    }
+
+    public String getJobType() {
+        return jobType;
+    }
+
+    public void setJobType(String jobType) {
+        this.jobType = jobType;
+    }
+
+    public String getJobParserAction() {
+        return jobParserAction;
+    }
+
+    public void setJobParserAction(String jobParserAction) {
+        this.jobParserAction = jobParserAction;
+    }
+
+    public String getJobRunnerAction() {
+        return jobRunnerAction;
+    }
+
+    public void setJobRunnerAction(String jobRunnerAction) {
+        this.jobRunnerAction = jobRunnerAction;
+    }
+
+    @Override
+    public boolean equals(Object o) {
+        if (this == o) return true;
+        if (o == null || getClass() != o.getClass()) return false;
+        JobDetails that = (JobDetails) o;
+        return Objects.equals(jobIndex, that.jobIndex)
+            && Objects.equals(jobType, that.jobType)
+            && jobParserAction.equals(that.jobParserAction)
+            && jobRunnerAction.equals(that.jobRunnerAction);
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(jobIndex, jobType, jobParserAction, jobRunnerAction);
+    }
+
+    @Override
+    public String toString() {
+        return "JobDetails{"
+            + "jobIndex='"
+            + jobIndex
+            + '\''
+            + ", jobType='"
+            + jobType
+            + '\''
+            + ", jobParserAction='"
+            + jobParserAction
+            + '\''
+            + ", jobRunnerAction='"
+            + jobRunnerAction
+            + '\''
+            + '}';
+    }
+}

src/main/java/org/opensearch/jobscheduler/rest/RestGetJobIndexAction.java

@@ -0,0 +1,94 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ * The OpenSearch Contributors require contributions made to
+ * this file be licensed under the Apache-2.0 license or a
+ * compatible open source license.
+ */
+package org.opensearch.jobscheduler.rest;
+
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+import org.opensearch.client.node.NodeClient;
+import org.opensearch.common.xcontent.ToXContent;
+import org.opensearch.common.xcontent.XContentParser;
+import org.opensearch.jobscheduler.JobSchedulerPlugin;
+import org.opensearch.jobscheduler.model.JobDetails;
+import org.opensearch.jobscheduler.transport.GetJobDetailsResponse;
+import org.opensearch.jobscheduler.transport.GetJobIndexAction;
+import org.opensearch.jobscheduler.transport.GetJobIndexRequest;
+
+import org.opensearch.rest.BaseRestHandler;
+import org.opensearch.rest.RestChannel;
+import org.opensearch.rest.RestRequest;
+import org.opensearch.rest.RestStatus;
+import org.opensearch.rest.RestResponse;
+import org.opensearch.rest.BytesRestResponse;
+import org.opensearch.rest.action.RestResponseListener;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+
+import static java.util.Arrays.asList;
+import static java.util.Collections.unmodifiableList;
+import static org.opensearch.common.xcontent.XContentParserUtils.ensureExpectedToken;
+import static org.opensearch.rest.RestRequest.Method.PUT;
+
+/**
+ * This class consists of the REST handler to GET job index from extensions.
+ */
+public class RestGetJobIndexAction extends BaseRestHandler {
+
+    public static final String GET_JOB_INDEX_ACTION = "get_job_index_action";
+
+    private final Logger logger = LogManager.getLogger(RestGetJobIndexAction.class);
+
+    private Map<String, JobDetails> jobDetailsHashMap;
+
+    public RestGetJobIndexAction(Map<String, JobDetails> jobDetailsHashMap) {
+        this.jobDetailsHashMap = jobDetailsHashMap;
+    }
+
+    @Override
+    public String getName() {
+        return GET_JOB_INDEX_ACTION;
+    }
+
+    @Override
+    public List<Route> routes() {
+        return unmodifiableList(
+            asList(new Route(PUT, String.format(Locale.ROOT, "%s/%s", JobSchedulerPlugin.JS_BASE_URI, "_get/_job_index")))
+        );
+    }
+
+    @Override
+    protected RestChannelConsumer prepareRequest(RestRequest restRequest, NodeClient client) throws IOException {
+        XContentParser parser = restRequest.contentParser();
+        ensureExpectedToken(XContentParser.Token.START_OBJECT, parser.nextToken(), parser);
+
+        GetJobIndexRequest getJobIndexRequest = GetJobIndexRequest.parse(parser);
+
+        JobDetails jobDetails = jobDetailsHashMap.getOrDefault(getJobIndexRequest.getExtensionId(), new JobDetails());
+        jobDetails.setJobIndex(getJobIndexRequest.getJobIndex());
+        jobDetails.setJobParserAction(getJobIndexRequest.getJobParserAction());
+        jobDetails.setJobRunnerAction(getJobIndexRequest.getJobRunnerAction());
+
+        jobDetailsHashMap.put(getJobIndexRequest.getExtensionId(), jobDetails);
+
+        return channel -> client.execute(GetJobIndexAction.INSTANCE, getJobIndexRequest, getJobIndexResponse(channel, RestStatus.OK));
+
+    }
+
+    private RestResponseListener<GetJobDetailsResponse> getJobIndexResponse(RestChannel channel, RestStatus status) {
+        return new RestResponseListener<>(channel) {
+            @Override
+            public RestResponse buildResponse(GetJobDetailsResponse getJobDetailsResponse) throws Exception {
+                return new BytesRestResponse(status, getJobDetailsResponse.toXContent(channel.newBuilder(), ToXContent.EMPTY_PARAMS));
+            }
+        };
+    }
+
+}