Communication mechanism for js

build.gradle

@@ -136,6 +136,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

@@ -5,6 +5,15 @@
 
 package org.opensearch.jobscheduler;
 
+import org.opensearch.action.ActionRequest;
+import org.opensearch.action.ActionResponse;
+import org.opensearch.client.node.NodeClient;
+import org.opensearch.cluster.node.DiscoveryNodes;
+import org.opensearch.common.settings.*;
+import org.opensearch.extensions.ExtensionsSettings;
+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;
@@ -20,36 +29,39 @@
 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.BaseRestHandler;
+import org.opensearch.rest.RestController;
+import org.opensearch.rest.RestHandler;
+import org.opensearch.rest.RestRequest;
 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.io.IOException;
+import java.util.*;
 import java.util.function.Supplier;
 
+import com.google.common.collect.ImmutableList;
 
-public class JobSchedulerPlugin extends Plugin implements ExtensiblePlugin {
+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);
 
@@ -58,10 +70,12 @@ public class JobSchedulerPlugin extends Plugin implements ExtensiblePlugin {
     private LockService lockService;
     private Map<String, ScheduledJobProvider> indexToJobProviders;
     private Set<String> indicesToListen;
+    private HashMap<String, JobDetails> jobDetailsHashMap;
 
     public JobSchedulerPlugin() {
         this.indicesToListen = new HashSet<>();
         this.indexToJobProviders = new HashMap<>();
+        this.jobDetailsHashMap=new HashMap<>();
     }
 
     @Override
@@ -155,4 +169,28 @@ private JobSweeper initSweeper(Settings settings, Client client, ClusterService
         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,13 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package org.opensearch.jobscheduler.constant;
+
+public class CommonValue {
+
+    public static Integer NO_SCHEMA_VERSION = 0;
+    public static String INTERNAL_ACTION_PREFIX = "cluster:admin/opendistro/jsinternal/";
+    public static String EXTERNAL_ACTION_PREFIX = "cluster:admin/opendistro/js/";
+}

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

@@ -0,0 +1,148 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+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;
+
+public class JobDetails implements ToXContentObject {
+
+    private String jobIndex;
+
+    private String jobType;
+
+    private String jobParamAction;
+
+    private String jobRunnerAction;
+
+
+    public static final String JOB_INDEX = "job_index";
+    public static final String JOB_TYPE = "job_type";
+    public static final String JOB_PARAM_ACTION= "job_param_action";
+    public static final String JOB_RUNNER_ACTION = "job_runner_action";
+
+
+    public JobDetails(){}
+    public JobDetails(String jobIndex, String jobType, String jobParamAction, String jobRunnerAction) {
+        this.jobIndex = jobIndex;
+        this.jobType = jobType;
+        this.jobParamAction = jobParamAction;
+        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 (jobParamAction != null) {
+            xContentBuilder.field(JOB_PARAM_ACTION, jobParamAction);
+        }
+        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 jobParamAction = 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_PARAM_ACTION:
+                    jobParamAction = parser.text();
+                    break;
+                case JOB_RUNNER_ACTION:
+                    jobRunnerAction = parser.text();
+                    break;
+                default:
+                    parser.skipChildren();
+                    break;
+            }
+        }
+
+        return new JobDetails(jobIndex,jobType,jobParamAction,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 getJobParamAction() {
+        return jobParamAction;
+    }
+
+    public void setJobParamAction(String jobParamAction) {
+        this.jobParamAction = jobParamAction;
+    }
+
+    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) && jobParamAction.equals(that.jobParamAction) && jobRunnerAction.equals(that.jobRunnerAction);
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(jobIndex, jobType, jobParamAction, jobRunnerAction);
+    }
+
+    @Override
+    public String toString() {
+        return "JobDetails{" +
+                "jobIndex='" + jobIndex + '\'' +
+                ", jobType='" + jobType + '\'' +
+                ", jobParamAction='" + jobParamAction + '\'' +
+                ", jobRunnerAction='" + jobRunnerAction + '\'' +
+                '}';
+    }
+}

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

@@ -0,0 +1,88 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+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.*;
+import org.opensearch.rest.action.RestResponseListener;
+
+import java.io.IOException;
+import java.util.HashMap;
+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;
+
+public class RestGetJobIndexAction extends BaseRestHandler {
+
+    private static final String GET_JOB_INDEX_ACTION = "get_job_index_action";
+
+    private final Logger logger = LogManager.getLogger(RestGetJobIndexAction.class);
+
+    private HashMap<String, JobDetails> jobDetailsHashMap;
+
+    public RestGetJobIndexAction(HashMap<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.setJobParamAction(getJobIndexRequest.getJobParamAction());
+        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 {
+                BytesRestResponse bytesRestResponse = new BytesRestResponse(
+                        status,
+                        getJobDetailsResponse.toXContent(channel.newBuilder(), ToXContent.EMPTY_PARAMS)
+                );
+                return bytesRestResponse;
+            }
+        };
+    }
+
+}