diff --git a/CHANGELOG.md b/CHANGELOG.md index cd0af3932371d..1def0448ad24c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,6 +5,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/), ## [Unreleased 2.x] ### Added +- [Offline Nodes] Adds offline-tasks library containing various interfaces to be used for Offline Background Tasks. ([#13574](https://github.com/opensearch-project/OpenSearch/pull/13574)) - Fix for hasInitiatedFetching to fix allocation explain and manual reroute APIs (([#14972](https://github.com/opensearch-project/OpenSearch/pull/14972)) - [Workload Management] Add queryGroupId to Task ([14708](https://github.com/opensearch-project/OpenSearch/pull/14708)) - Add setting to ignore throttling nodes for allocation of unassigned primaries in remote restore ([#14991](https://github.com/opensearch-project/OpenSearch/pull/14991)) diff --git a/libs/task-commons/build.gradle b/libs/task-commons/build.gradle new file mode 100644 index 0000000000000..dde3acd8effcf --- /dev/null +++ b/libs/task-commons/build.gradle @@ -0,0 +1,25 @@ +/* + * 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. + * + * Modifications Copyright OpenSearch Contributors. See + * GitHub history for details. + */ + +dependencies { + api project(':libs:opensearch-common') + + testImplementation "com.carrotsearch.randomizedtesting:randomizedtesting-runner:${versions.randomizedrunner}" + testImplementation "junit:junit:${versions.junit}" + testImplementation "org.hamcrest:hamcrest:${versions.hamcrest}" + testImplementation(project(":test:framework")) { + exclude group: 'org.opensearch', module: 'opensearch-task-commons' + } +} + +tasks.named('forbiddenApisMain').configure { + replaceSignatureFiles 'jdk-signatures' +} diff --git a/libs/task-commons/src/main/java/org/opensearch/task/commons/clients/TaskListRequest.java b/libs/task-commons/src/main/java/org/opensearch/task/commons/clients/TaskListRequest.java new file mode 100644 index 0000000000000..625e15dfb3b6d --- /dev/null +++ b/libs/task-commons/src/main/java/org/opensearch/task/commons/clients/TaskListRequest.java @@ -0,0 +1,103 @@ +/* + * 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.task.commons.clients; + +import org.opensearch.task.commons.task.TaskStatus; +import org.opensearch.task.commons.task.TaskType; +import org.opensearch.task.commons.worker.WorkerNode; + +/** + * Request object for listing tasks + */ +public class TaskListRequest { + + /** + * Filters listTasks response by specific task status' + */ + private TaskStatus[] taskStatus; + + /** + * Filter listTasks response by specific task types + */ + private TaskType[] taskTypes; + + /** + * Filter listTasks response by specific worker node + */ + private WorkerNode workerNodes; + + /** + * Depicts the start page number for the list call. + * + * @see TaskManagerClient#listTasks(TaskListRequest) + */ + private int startPageNumber; + + /** + * Depicts the page size for the list call. + * + * @see TaskManagerClient#listTasks(TaskListRequest) + */ + private int pageSize; + + /** + * Default constructor + */ + public TaskListRequest() {} + + /** + * Update task types to filter with in the request + * @param taskTypes TaskType[] + * @return ListTaskRequest + */ + public TaskListRequest taskType(TaskType... taskTypes) { + this.taskTypes = taskTypes; + return this; + } + + /** + * Update task status to filter with in the request + * @param taskStatus TaskStatus[] + * @return ListTaskRequest + */ + public TaskListRequest taskType(TaskStatus... taskStatus) { + this.taskStatus = taskStatus; + return this; + } + + /** + * Update worker node to filter with in the request + * @param workerNode WorkerNode + * @return ListTaskRequest + */ + private TaskListRequest workerNode(WorkerNode workerNode) { + this.workerNodes = workerNode; + return this; + } + + /** + * Update page number to start with when fetching the list of tasks + * @param startPageNumber startPageNumber + * @return ListTaskRequest + */ + public TaskListRequest startPageNumber(int startPageNumber) { + this.startPageNumber = startPageNumber; + return this; + } + + /** + * Update page size for the list tasks response + * @param pageSize int + * @return ListTaskRequest + */ + public TaskListRequest pageSize(int pageSize) { + this.pageSize = pageSize; + return this; + } +} diff --git a/libs/task-commons/src/main/java/org/opensearch/task/commons/clients/TaskManagerClient.java b/libs/task-commons/src/main/java/org/opensearch/task/commons/clients/TaskManagerClient.java new file mode 100644 index 0000000000000..23ad8eabdc365 --- /dev/null +++ b/libs/task-commons/src/main/java/org/opensearch/task/commons/clients/TaskManagerClient.java @@ -0,0 +1,69 @@ +/* + * 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.task.commons.clients; + +import org.opensearch.common.annotation.ExperimentalApi; +import org.opensearch.task.commons.task.Task; +import org.opensearch.task.commons.task.TaskId; +import org.opensearch.task.commons.worker.WorkerNode; + +import java.util.List; + +/** + * Client used to interact with Task Store/Queue. + * + * TODO: TaskManager can be something not running an opensearch process. + * We need to come up with a way to allow this interface to be used with in and out opensearch as well + * + * @opensearch.experimental + */ +@ExperimentalApi +public interface TaskManagerClient { + + /** + * Get task from TaskStore/Queue + * + * @param taskId TaskId of the task to be retrieved + * @return Task corresponding to TaskId + */ + Task getTask(TaskId taskId); + + /** + * Update task in TaskStore/Queue + * + * @param task Task to be updated + */ + void updateTask(Task task); + + /** + * Mark task as cancelled. + * Ongoing Tasks can be cancelled as well if the corresponding worker supports cancellation + * + * @param taskId TaskId of the task to be cancelled + */ + void cancelTask(TaskId taskId); + + /** + * List all tasks applying all the filters present in listTaskRequest + * + * @param taskListRequest TaskListRequest + * @return list of all the task matching the filters in listTaskRequest + */ + List listTasks(TaskListRequest taskListRequest); + + /** + * Assign Task to a particular WorkerNode. This ensures no 2 worker Nodes work on the same task. + * This API can be used in both pull and push models of task assignment. + * + * @param taskId TaskId of the task to be assigned + * @param node WorkerNode task is being assigned to + * @return true if task is assigned successfully, false otherwise + */ + boolean assignTask(TaskId taskId, WorkerNode node); +} diff --git a/libs/task-commons/src/main/java/org/opensearch/task/commons/clients/TaskProducerClient.java b/libs/task-commons/src/main/java/org/opensearch/task/commons/clients/TaskProducerClient.java new file mode 100644 index 0000000000000..18f3421f9aa97 --- /dev/null +++ b/libs/task-commons/src/main/java/org/opensearch/task/commons/clients/TaskProducerClient.java @@ -0,0 +1,24 @@ +/* + * 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.task.commons.clients; + +import org.opensearch.task.commons.task.Task; + +/** + * Producer interface used to submit new tasks for execution on worker nodes. + */ +public interface TaskProducerClient { + + /** + * Submit a new task to TaskStore/Queue + * + * @param task Task to be submitted for execution on offline nodes + */ + void submitTask(Task task); +} diff --git a/libs/task-commons/src/main/java/org/opensearch/task/commons/clients/TaskWorkerClient.java b/libs/task-commons/src/main/java/org/opensearch/task/commons/clients/TaskWorkerClient.java new file mode 100644 index 0000000000000..59abe189434dd --- /dev/null +++ b/libs/task-commons/src/main/java/org/opensearch/task/commons/clients/TaskWorkerClient.java @@ -0,0 +1,38 @@ +/* + * 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.task.commons.clients; + +import org.opensearch.task.commons.task.Task; +import org.opensearch.task.commons.task.TaskId; + +import java.util.List; + +/** + * Consumer interface used to find new tasks assigned to a {@code WorkerNode} for execution. + */ +public interface TaskWorkerClient { + + /** + * List all tasks assigned to a WorkerNode. + * Useful when the implementation uses a separate store for Task assignments to Worker nodes + * + * @param taskListRequest TaskListRequest + * @return list of all tasks assigned to a WorkerNode + */ + List getAssignedTasks(TaskListRequest taskListRequest); + + /** + * Sends task heart beat to Task Store/Queue + * + * @param taskId TaskId of Task to send heartbeat for + * @param timestamp timestamp of heartbeat to be recorded in TaskStore/Queue + */ + void sendTaskHeartbeat(TaskId taskId, long timestamp); + +} diff --git a/libs/task-commons/src/main/java/org/opensearch/task/commons/clients/package-info.java b/libs/task-commons/src/main/java/org/opensearch/task/commons/clients/package-info.java new file mode 100644 index 0000000000000..1329f3888248c --- /dev/null +++ b/libs/task-commons/src/main/java/org/opensearch/task/commons/clients/package-info.java @@ -0,0 +1,12 @@ +/* + * 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. + */ + +/** + * Contains task client related classes + */ +package org.opensearch.task.commons.clients; diff --git a/libs/task-commons/src/main/java/org/opensearch/task/commons/package-info.java b/libs/task-commons/src/main/java/org/opensearch/task/commons/package-info.java new file mode 100644 index 0000000000000..4cb773ace62ce --- /dev/null +++ b/libs/task-commons/src/main/java/org/opensearch/task/commons/package-info.java @@ -0,0 +1,12 @@ +/* + * 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. + */ + +/** + * Contains offline tasks related classes + */ +package org.opensearch.task.commons; diff --git a/libs/task-commons/src/main/java/org/opensearch/task/commons/task/Task.java b/libs/task-commons/src/main/java/org/opensearch/task/commons/task/Task.java new file mode 100644 index 0000000000000..7ad567b57bd42 --- /dev/null +++ b/libs/task-commons/src/main/java/org/opensearch/task/commons/task/Task.java @@ -0,0 +1,361 @@ +/* + * 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.task.commons.task; + +import org.opensearch.common.Nullable; +import org.opensearch.common.annotation.ExperimentalApi; +import org.opensearch.task.commons.worker.WorkerNode; + +/** + * A Background Task to be run on Offline Node. + */ +@ExperimentalApi +public class Task { + + /** + * Task identifier used to uniquely identify a Task + */ + private final TaskId taskId; + + /** + * Depicts latest state of the Task + */ + private final TaskStatus taskStatus; + + /** + * Various params to used for Task execution + */ + private final TaskParams params; + + /** + * Type/Category of the Task + */ + private final TaskType taskType; + + /** + * Worker Node on which the Task is to be executed + */ + private final WorkerNode assignedNode; + + /** + * Timestamp at which the Task was created + */ + private final long createdAt; + + /** + * Timestamp at which the Task was assigned to a worker + */ + private final long assignedAt; + + /** + * Timestamp at which the Task was started execution on worker + */ + private final long startedAt; + + /** + * Timestamp at which the Task was either completed/failed/cancelled + */ + private final long completedAt; + + /** + * Timestamp at which last heartbeat was sent by the worker + */ + private final long lastHeartbeatAt; + + /** + * Constructor for Task + * + * @param taskId Task identifier + * @param taskStatus Task status + * @param params Task Params + * @param taskType Task Type + * @param createdAt Timestamp at which the Task was created + * @param assignedAt Timestamp at which the Task was assigned to a worker + * @param startedAt Timestamp at which the Task was started execution on worker + * @param completedAt Timestamp at which the Task was either completed/failed/cancelled + * @param lastHeartbeatAt Timestamp at which last heartbeat was sent by the worker + * @param assignedNode Worker Node on which the Task is to be executed + */ + public Task( + TaskId taskId, + TaskStatus taskStatus, + TaskParams params, + TaskType taskType, + long createdAt, + @Nullable long assignedAt, + @Nullable long startedAt, + @Nullable long completedAt, + @Nullable long lastHeartbeatAt, + @Nullable WorkerNode assignedNode + ) { + this.taskId = taskId; + this.taskStatus = taskStatus; + this.params = params; + this.taskType = taskType; + this.createdAt = createdAt; + this.assignedAt = assignedAt; + this.startedAt = startedAt; + this.completedAt = completedAt; + this.lastHeartbeatAt = lastHeartbeatAt; + this.assignedNode = assignedNode; + } + + /** + * Get TaskId + * @return TaskId + */ + public TaskId getTaskId() { + return taskId; + } + + /** + * Get TaskStatus + * @return TaskStatus + */ + public TaskStatus getTaskStatus() { + return taskStatus; + } + + /** + * Get TaskParams + * @return TaskParams + */ + public TaskParams getParams() { + return params; + } + + /** + * Get TaskType + * @return TaskType + */ + public TaskType getTaskType() { + return taskType; + } + + /** + * Get Task Creation Time + * @return createdAt + */ + public long getCreatedAt() { + return createdAt; + } + + /** + * Get Task Assignment Time + * @return assignedAt + */ + public long getAssignedAt() { + return assignedAt; + } + + /** + * Get Task Start Time + * @return startedAt + */ + public long getStartedAt() { + return startedAt; + } + + /** + * Get Task Completion Time + * @return completedAt + */ + public long getCompletedAt() { + return completedAt; + } + + /** + * Get Last Heartbeat Time + * @return lastHeartbeatAt + */ + public long getLastHeartbeatAt() { + return lastHeartbeatAt; + } + + /** + * Get Task Assigned Node + * @return assignedNode + */ + public WorkerNode getAssignedNode() { + return assignedNode; + } + + /** + * Builder class for Task. + */ + public static class Builder { + /** + * Task identifier used to uniquely identify a Task + */ + private final TaskId taskId; + + /** + * Depicts latest state of the Task + */ + private TaskStatus taskStatus; + + /** + * Various params to used for Task execution + */ + private final TaskParams params; + + /** + * Type/Category of the Task + */ + private final TaskType taskType; + + /** + * Type/Category of the Task + */ + private WorkerNode assignedNode; + + /** + * Timestamp at which the Task was created + */ + private final long createdAt; + + /** + * Timestamp at which the Task was assigned to a worker + */ + private long assignedAt; + + /** + * Timestamp at which the Task was started execution on worker + */ + private long startedAt; + + /** + * Timestamp at which the Task was either completed/failed/cancelled + */ + private long completedAt; + + /** + * Timestamp at which last heartbeat was sent by the worker + */ + private long lastHeartbeatAt; + + /** + * Constructor for Task Builder + * + * @param taskId Task identifier + * @param taskStatus Task status + * @param params Task Params + * @param taskType Task Type + * @param createdAt Task Creation Time + */ + private Builder(TaskId taskId, TaskStatus taskStatus, TaskParams params, TaskType taskType, long createdAt) { + this.taskId = taskId; + this.taskStatus = taskStatus; + this.params = params; + this.taskType = taskType; + this.createdAt = createdAt; + } + + /** + * Build Builder from Task + * @param task Task to build from + * @return Task.Builder + */ + public static Builder builder(Task task) { + Builder builder = new Builder( + task.getTaskId(), + task.getTaskStatus(), + task.getParams(), + task.getTaskType(), + task.getCreatedAt() + ); + builder.assignedAt(task.getAssignedAt()); + builder.startedAt(task.getStartedAt()); + builder.completedAt(task.getCompletedAt()); + builder.lastHeartbeatAt(task.getLastHeartbeatAt()); + builder.assignedNode(task.getAssignedNode()); + return builder; + } + + /** + * Build Builder from various Task attributes + * @param taskId Task identifier + * @param taskStatus TaskStatus + * @param params TaskParams + * @param taskType TaskType + * @param createdAt Task Creation Time + * @return Task.Builder + */ + public static Builder builder(TaskId taskId, TaskStatus taskStatus, TaskParams params, TaskType taskType, long createdAt) { + return new Builder(taskId, taskStatus, params, taskType, createdAt); + } + + /** + * Set Task Assignment Time + * @param assignedAt Timestamp at which the Task was assigned to a worker + */ + public void assignedAt(long assignedAt) { + this.assignedAt = assignedAt; + } + + /** + * Set Task Start Time + * @param startedAt Timestamp at which the Task was started execution on worker + */ + public void startedAt(long startedAt) { + this.startedAt = startedAt; + } + + /** + * Set Task Completion Time + * @param completedAt Timestamp at which the Task was either completed/failed/cancelled + */ + public void completedAt(long completedAt) { + this.completedAt = completedAt; + } + + /** + * Set Task Last Heartbeat Time for the task + * @param lastHeartbeatAt Timestamp at which last heartbeat was sent by the worker + */ + public void lastHeartbeatAt(long lastHeartbeatAt) { + this.lastHeartbeatAt = lastHeartbeatAt; + } + + /** + * Update the Task Status + * @param taskStatus {@link TaskStatus} - current status of the Task + */ + public void taskStatus(TaskStatus taskStatus) { + this.taskStatus = taskStatus; + } + + /** + * Set Task Assigned Node + * @param node Worker Node on which the Task is to be executed + */ + public void assignedNode(WorkerNode node) { + this.assignedNode = node; + } + + /** + * Build Task from Builder + * @return Task + */ + public Task build() { + return new Task( + taskId, + taskStatus, + params, + taskType, + createdAt, + assignedAt, + startedAt, + completedAt, + lastHeartbeatAt, + assignedNode + ); + } + } +} diff --git a/libs/task-commons/src/main/java/org/opensearch/task/commons/task/TaskId.java b/libs/task-commons/src/main/java/org/opensearch/task/commons/task/TaskId.java new file mode 100644 index 0000000000000..7fb7e1536dc17 --- /dev/null +++ b/libs/task-commons/src/main/java/org/opensearch/task/commons/task/TaskId.java @@ -0,0 +1,55 @@ +/* + * 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.task.commons.task; + +import org.opensearch.common.annotation.ExperimentalApi; + +import java.util.Objects; + +/** + * Class encapsulating Task identifier + */ +@ExperimentalApi +public class TaskId { + + /** + * Identified of the Task + */ + private final String id; + + /** + * Constructor to initialize TaskId + * @param id String value of Task id + */ + public TaskId(String id) { + this.id = id; + } + + /** + * Get id value + * @return id + */ + public String getValue() { + return id; + } + + @Override + public boolean equals(Object obj) { + if (obj == null || getClass() != obj.getClass()) { + return false; + } + TaskId other = (TaskId) obj; + return this.id.equals(other.id); + } + + @Override + public int hashCode() { + return Objects.hash(id); + } +} diff --git a/libs/task-commons/src/main/java/org/opensearch/task/commons/task/TaskParams.java b/libs/task-commons/src/main/java/org/opensearch/task/commons/task/TaskParams.java new file mode 100644 index 0000000000000..ac6126a02cf24 --- /dev/null +++ b/libs/task-commons/src/main/java/org/opensearch/task/commons/task/TaskParams.java @@ -0,0 +1,23 @@ +/* + * 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.task.commons.task; + +import org.opensearch.common.annotation.ExperimentalApi; + +/** + * Base class for all TaskParams implementation of various TaskTypes + */ +@ExperimentalApi +public abstract class TaskParams { + + /** + * Default constructor + */ + public TaskParams() {} +} diff --git a/libs/task-commons/src/main/java/org/opensearch/task/commons/task/TaskStatus.java b/libs/task-commons/src/main/java/org/opensearch/task/commons/task/TaskStatus.java new file mode 100644 index 0000000000000..e44e5c7fee620 --- /dev/null +++ b/libs/task-commons/src/main/java/org/opensearch/task/commons/task/TaskStatus.java @@ -0,0 +1,49 @@ +/* + * 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.task.commons.task; + +import org.opensearch.common.annotation.ExperimentalApi; + +/** + * Task status enum + */ +@ExperimentalApi +public enum TaskStatus { + + /** + * TaskStatus of a Task which is not yet assigned to or picked up by a worker + */ + UNASSIGNED, + + /** + * TaskStatus of a Task which is assigned to or picked up by a worker but hasn't started execution yet. + * This status confirms that a worker will execute this task and no other worker should pick it up. + */ + ASSIGNED, + + /** + * TaskStatus of an in progress Task + */ + ACTIVE, + + /** + * TaskStatus of a finished Task + */ + SUCCESS, + + /** + * TaskStatus of a Task which failed in 1 or more attempts + */ + FAILED, + + /** + * TaskStatus of a cancelled Task + */ + CANCELLED +} diff --git a/libs/task-commons/src/main/java/org/opensearch/task/commons/task/TaskType.java b/libs/task-commons/src/main/java/org/opensearch/task/commons/task/TaskType.java new file mode 100644 index 0000000000000..ef267a86e3da4 --- /dev/null +++ b/libs/task-commons/src/main/java/org/opensearch/task/commons/task/TaskType.java @@ -0,0 +1,27 @@ +/* + * 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.task.commons.task; + +import org.opensearch.common.annotation.ExperimentalApi; + +/** + * Enum for task type + */ +@ExperimentalApi +public enum TaskType { + /** + * For all segment merge related tasks + */ + MERGE, + + /** + * For all snapshot related tasks + */ + SNAPSHOT +} diff --git a/libs/task-commons/src/main/java/org/opensearch/task/commons/task/package-info.java b/libs/task-commons/src/main/java/org/opensearch/task/commons/task/package-info.java new file mode 100644 index 0000000000000..e99e4079cc7af --- /dev/null +++ b/libs/task-commons/src/main/java/org/opensearch/task/commons/task/package-info.java @@ -0,0 +1,12 @@ +/* + * 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. + */ + +/** + * Contains tasks related classes + */ +package org.opensearch.task.commons.task; diff --git a/libs/task-commons/src/main/java/org/opensearch/task/commons/worker/TaskWorker.java b/libs/task-commons/src/main/java/org/opensearch/task/commons/worker/TaskWorker.java new file mode 100644 index 0000000000000..6d9fd1fed23e4 --- /dev/null +++ b/libs/task-commons/src/main/java/org/opensearch/task/commons/worker/TaskWorker.java @@ -0,0 +1,29 @@ +/* + * 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.task.commons.worker; + +import org.opensearch.common.annotation.ExperimentalApi; +import org.opensearch.task.commons.task.Task; + +/** + * Task Worker that executes the Task + * + * @opensearch.experimental + */ +@ExperimentalApi +public interface TaskWorker { + + /** + * Execute the Task + * + * @param task Task to be executed + */ + void executeTask(Task task); + +} diff --git a/libs/task-commons/src/main/java/org/opensearch/task/commons/worker/WorkerNode.java b/libs/task-commons/src/main/java/org/opensearch/task/commons/worker/WorkerNode.java new file mode 100644 index 0000000000000..1b9426d472184 --- /dev/null +++ b/libs/task-commons/src/main/java/org/opensearch/task/commons/worker/WorkerNode.java @@ -0,0 +1,96 @@ +/* + * 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.task.commons.worker; + +import java.util.Objects; + +/** + * Represents a worker node in the fleet + */ +public class WorkerNode { + /** + * The unique identifier of the worker node. + */ + private final String id; + + /** + * The name of the worker node. + */ + private final String name; + + /** + * The IP address of the worker node. + */ + private final String ip; + + /** + * Creates a new worker node with the given ID, name, and IP address. + * + * @param id The unique identifier of the worker node. + * @param name The name of the worker node. + * @param ip The IP address of the worker node. + */ + private WorkerNode(String id, String name, String ip) { + this.id = id; + this.name = name; + this.ip = ip; + } + + /** + * Creates a new worker node with the given ID, name, and IP address. + * + * @param id The unique identifier of the worker node. + * @param name The name of the worker node. + * @param ip The IP address of the worker node. + * @return The created worker node. + */ + public static WorkerNode createWorkerNode(String id, String name, String ip) { + return new WorkerNode(id, name, ip); + } + + /** + * Returns the unique identifier of the worker node. + * + * @return The ID of the worker node. + */ + public String getId() { + return id; + } + + /** + * Returns the name of the worker node. + * + * @return The name of the worker node. + */ + public String getName() { + return name; + } + + /** + * Returns the IP address of the worker node. + * + * @return The IP address of the worker node. + */ + public String getIp() { + return ip; + } + + @Override + public boolean equals(Object o) { + if (this == o) return true; + if (o == null || getClass() != o.getClass()) return false; + WorkerNode that = (WorkerNode) o; + return Objects.equals(id, that.id) && Objects.equals(name, that.name) && Objects.equals(ip, that.ip); + } + + @Override + public int hashCode() { + return Objects.hash(id, name, ip); + } +} diff --git a/libs/task-commons/src/main/java/org/opensearch/task/commons/worker/package-info.java b/libs/task-commons/src/main/java/org/opensearch/task/commons/worker/package-info.java new file mode 100644 index 0000000000000..d74fa30e8b661 --- /dev/null +++ b/libs/task-commons/src/main/java/org/opensearch/task/commons/worker/package-info.java @@ -0,0 +1,12 @@ +/* + * 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. + */ + +/** + * Contains task worker related classes + */ +package org.opensearch.task.commons.worker; diff --git a/libs/task-commons/src/test/java/org/opensearch/task/commons/mocks/MockTaskParams.java b/libs/task-commons/src/test/java/org/opensearch/task/commons/mocks/MockTaskParams.java new file mode 100644 index 0000000000000..21d2a3dd725b6 --- /dev/null +++ b/libs/task-commons/src/test/java/org/opensearch/task/commons/mocks/MockTaskParams.java @@ -0,0 +1,25 @@ +/* + * 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.task.commons.mocks; + +import org.opensearch.task.commons.task.TaskParams; + +public class MockTaskParams extends TaskParams { + + private final String value; + + public MockTaskParams(String mockValue) { + super(); + value = mockValue; + } + + public String getValue() { + return value; + } +} diff --git a/libs/task-commons/src/test/java/org/opensearch/task/commons/task/TaskIdTests.java b/libs/task-commons/src/test/java/org/opensearch/task/commons/task/TaskIdTests.java new file mode 100644 index 0000000000000..e58382e9fc056 --- /dev/null +++ b/libs/task-commons/src/test/java/org/opensearch/task/commons/task/TaskIdTests.java @@ -0,0 +1,53 @@ +/* + * 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.task.commons.task; + +import org.opensearch.test.OpenSearchTestCase; + +/** + * Tests for {@link TaskId} + */ +public class TaskIdTests extends OpenSearchTestCase { + + public void testConstructorAndGetValue() { + TaskId taskId = new TaskId("123"); + assertEquals("123", taskId.getValue()); + } + + public void testEqualsWithSameId() { + TaskId taskId1 = new TaskId("456"); + TaskId taskId2 = new TaskId("456"); + assertEquals(taskId1, taskId2); + } + + public void testEqualsWithDifferentId() { + TaskId taskId1 = new TaskId("789"); + TaskId taskId2 = new TaskId("987"); + assertNotEquals(taskId1, taskId2); + } + + public void testEqualsWithNull() { + TaskId taskId = new TaskId("abc"); + assertNotEquals(null, taskId); + } + + public void testEqualsWithDifferentClass() { + TaskId taskId = new TaskId("def"); + assertNotEquals(taskId, new Object()); + } + + public void testHashCode() { + TaskId taskId1 = new TaskId("456"); + TaskId taskId2 = new TaskId("456"); + assertEquals(taskId1.hashCode(), taskId2.hashCode()); + + TaskId taskId3 = new TaskId("4567"); + assertNotEquals(taskId1.hashCode(), taskId3.hashCode()); + } +} diff --git a/libs/task-commons/src/test/java/org/opensearch/task/commons/task/TaskTests.java b/libs/task-commons/src/test/java/org/opensearch/task/commons/task/TaskTests.java new file mode 100644 index 0000000000000..37c5f543daa02 --- /dev/null +++ b/libs/task-commons/src/test/java/org/opensearch/task/commons/task/TaskTests.java @@ -0,0 +1,147 @@ +/* + * 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.task.commons.task; + +import org.opensearch.task.commons.mocks.MockTaskParams; +import org.opensearch.task.commons.worker.WorkerNode; +import org.opensearch.test.OpenSearchTestCase; + +/** + * Test for {@link Task} + */ +public class TaskTests extends OpenSearchTestCase { + + public void testTaskConstructorAndGetters() { + TaskId taskId = new TaskId("123"); + TaskStatus taskStatus = TaskStatus.UNASSIGNED; + TaskParams params = new MockTaskParams("mock"); + TaskType taskType = TaskType.MERGE; + long createdAt = System.currentTimeMillis(); + long assignedAt = createdAt + 1000; + long startedAt = createdAt + 2000; + long completedAt = createdAt + 3000; + long lastHeartbeatAt = createdAt + 2500; + WorkerNode assignedNode = WorkerNode.createWorkerNode("node1", "nodeip", "nodename"); + + Task task = new Task( + taskId, + taskStatus, + params, + taskType, + createdAt, + assignedAt, + startedAt, + completedAt, + lastHeartbeatAt, + assignedNode + ); + + assertEquals(taskId, task.getTaskId()); + assertEquals(taskStatus, task.getTaskStatus()); + assertEquals(params, task.getParams()); + assertEquals(taskType, task.getTaskType()); + assertEquals(createdAt, task.getCreatedAt()); + assertEquals(assignedAt, task.getAssignedAt()); + assertEquals(startedAt, task.getStartedAt()); + assertEquals(completedAt, task.getCompletedAt()); + assertEquals(lastHeartbeatAt, task.getLastHeartbeatAt()); + assertEquals(assignedNode, task.getAssignedNode()); + } + + public void testBuilderFromTask() { + TaskId taskId = new TaskId("123"); + TaskStatus taskStatus = TaskStatus.UNASSIGNED; + TaskParams params = new MockTaskParams("mock"); + TaskType taskType = TaskType.MERGE; + long createdAt = System.currentTimeMillis(); + long assignedAt = createdAt + 1000; + long startedAt = createdAt + 2000; + long completedAt = createdAt + 3000; + long lastHeartbeatAt = createdAt + 2500; + WorkerNode assignedNode = WorkerNode.createWorkerNode("node1", "nodeip", "nodename"); + + Task originalTask = new Task( + taskId, + taskStatus, + params, + taskType, + createdAt, + assignedAt, + startedAt, + completedAt, + lastHeartbeatAt, + assignedNode + ); + + Task.Builder builder = Task.Builder.builder(originalTask); + Task newTask = builder.build(); + + assertEquals(originalTask.getTaskId(), newTask.getTaskId()); + assertEquals(originalTask.getTaskStatus(), newTask.getTaskStatus()); + assertEquals(originalTask.getParams(), newTask.getParams()); + assertEquals(originalTask.getTaskType(), newTask.getTaskType()); + assertEquals(originalTask.getCreatedAt(), newTask.getCreatedAt()); + assertEquals(originalTask.getAssignedAt(), newTask.getAssignedAt()); + assertEquals(originalTask.getStartedAt(), newTask.getStartedAt()); + assertEquals(originalTask.getCompletedAt(), newTask.getCompletedAt()); + assertEquals(originalTask.getLastHeartbeatAt(), newTask.getLastHeartbeatAt()); + assertEquals(originalTask.getAssignedNode(), newTask.getAssignedNode()); + } + + public void testBuilderFromAttributes() { + TaskId taskId = new TaskId("123"); + TaskStatus taskStatus = TaskStatus.UNASSIGNED; + TaskParams params = new MockTaskParams("mock"); + TaskType taskType = TaskType.MERGE; + long createdAt = System.currentTimeMillis(); + + Task.Builder builder = Task.Builder.builder(taskId, taskStatus, params, taskType, createdAt); + builder.assignedAt(createdAt + 1000); + builder.startedAt(createdAt + 2000); + builder.completedAt(createdAt + 3000); + builder.lastHeartbeatAt(createdAt + 2500); + builder.assignedNode(WorkerNode.createWorkerNode("node1", "nodeip", "nodename")); + builder.taskStatus(TaskStatus.ACTIVE); + + Task task = builder.build(); + + assertEquals(taskId, task.getTaskId()); + assertEquals(TaskStatus.ACTIVE, task.getTaskStatus()); + assertEquals(params, task.getParams()); + assertEquals(taskType, task.getTaskType()); + assertEquals(createdAt, task.getCreatedAt()); + assertEquals(createdAt + 1000, task.getAssignedAt()); + assertEquals(createdAt + 2000, task.getStartedAt()); + assertEquals(createdAt + 3000, task.getCompletedAt()); + assertEquals(createdAt + 2500, task.getLastHeartbeatAt()); + assertEquals(WorkerNode.createWorkerNode("node1", "nodeip", "nodename"), task.getAssignedNode()); + } + + public void testBuilderWithNullOptionalFields() { + TaskId taskId = new TaskId("123"); + TaskStatus taskStatus = TaskStatus.UNASSIGNED; + TaskParams params = new MockTaskParams("mock"); + TaskType taskType = TaskType.MERGE; + long createdAt = System.currentTimeMillis(); + + Task.Builder builder = Task.Builder.builder(taskId, taskStatus, params, taskType, createdAt); + Task task = builder.build(); + + assertEquals(taskId, task.getTaskId()); + assertEquals(taskStatus, task.getTaskStatus()); + assertEquals(params, task.getParams()); + assertEquals(taskType, task.getTaskType()); + assertEquals(createdAt, task.getCreatedAt()); + assertEquals(0, task.getAssignedAt()); + assertEquals(0, task.getStartedAt()); + assertEquals(0, task.getCompletedAt()); + assertEquals(0, task.getLastHeartbeatAt()); + assertNull(task.getAssignedNode()); + } +} diff --git a/libs/task-commons/src/test/java/org/opensearch/task/commons/worker/WorkerNodeTests.java b/libs/task-commons/src/test/java/org/opensearch/task/commons/worker/WorkerNodeTests.java new file mode 100644 index 0000000000000..07570bf947a27 --- /dev/null +++ b/libs/task-commons/src/test/java/org/opensearch/task/commons/worker/WorkerNodeTests.java @@ -0,0 +1,52 @@ +/* + * 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.task.commons.worker; + +import org.opensearch.test.OpenSearchTestCase; + +/** + * Tests for {@link WorkerNode} + */ +public class WorkerNodeTests extends OpenSearchTestCase { + + public void testCreateWorkerNode() { + WorkerNode worker = WorkerNode.createWorkerNode("1", "Worker1", "192.168.1.1"); + assertNotNull(worker); + assertEquals("1", worker.getId()); + assertEquals("Worker1", worker.getName()); + assertEquals("192.168.1.1", worker.getIp()); + } + + public void testEquality() { + WorkerNode worker1 = WorkerNode.createWorkerNode("5", "Worker5", "192.168.1.5"); + WorkerNode worker2 = WorkerNode.createWorkerNode("5", "Worker5", "192.168.1.5"); + WorkerNode worker3 = WorkerNode.createWorkerNode("6", "Worker6", "192.168.1.6"); + + assertEquals(worker1, worker2); + assertNotEquals(worker1, worker3); + assertNotEquals(worker2, worker3); + } + + public void testHashCode() { + WorkerNode worker1 = WorkerNode.createWorkerNode("7", "Worker7", "192.168.1.7"); + WorkerNode worker2 = WorkerNode.createWorkerNode("7", "Worker7", "192.168.1.7"); + + assertEquals(worker1.hashCode(), worker2.hashCode()); + } + + public void testNotEqualToNull() { + WorkerNode worker = WorkerNode.createWorkerNode("8", "Worker8", "192.168.1.8"); + assertNotEquals(null, worker); + } + + public void testNotEqualToDifferentClass() { + WorkerNode worker = WorkerNode.createWorkerNode("9", "Worker9", "192.168.1.9"); + assertNotEquals(worker, new Object()); + } +} diff --git a/server/build.gradle b/server/build.gradle index 429af5d0ac258..5facc73dff968 100644 --- a/server/build.gradle +++ b/server/build.gradle @@ -68,6 +68,7 @@ dependencies { api project(':libs:opensearch-x-content') api project(":libs:opensearch-geo") api project(":libs:opensearch-telemetry") + api project(":libs:opensearch-task-commons") compileOnly project(':libs:opensearch-plugin-classloader') diff --git a/server/src/main/java/org/opensearch/common/util/FeatureFlags.java b/server/src/main/java/org/opensearch/common/util/FeatureFlags.java index 9d57e6939e3ae..e2554d61116ad 100644 --- a/server/src/main/java/org/opensearch/common/util/FeatureFlags.java +++ b/server/src/main/java/org/opensearch/common/util/FeatureFlags.java @@ -72,6 +72,11 @@ public class FeatureFlags { */ public static final String REMOTE_PUBLICATION_EXPERIMENTAL = "opensearch.experimental.feature.remote_store.publication.enabled"; + /** + * Gates the functionality of background task execution. + */ + public static final String BACKGROUND_TASK_EXECUTION_EXPERIMENTAL = "opensearch.experimental.feature.task.background.enabled"; + public static final Setting REMOTE_STORE_MIGRATION_EXPERIMENTAL_SETTING = Setting.boolSetting( REMOTE_STORE_MIGRATION_EXPERIMENTAL, false, diff --git a/server/src/main/java/org/opensearch/node/Node.java b/server/src/main/java/org/opensearch/node/Node.java index cbed8dfea8cc4..409f84354a8b1 100644 --- a/server/src/main/java/org/opensearch/node/Node.java +++ b/server/src/main/java/org/opensearch/node/Node.java @@ -215,6 +215,7 @@ import org.opensearch.plugins.SearchPlugin; import org.opensearch.plugins.SecureSettingsFactory; import org.opensearch.plugins.SystemIndexPlugin; +import org.opensearch.plugins.TaskManagerClientPlugin; import org.opensearch.plugins.TelemetryAwarePlugin; import org.opensearch.plugins.TelemetryPlugin; import org.opensearch.ratelimitting.admissioncontrol.AdmissionControlService; @@ -239,6 +240,7 @@ import org.opensearch.snapshots.SnapshotShardsService; import org.opensearch.snapshots.SnapshotsInfoService; import org.opensearch.snapshots.SnapshotsService; +import org.opensearch.task.commons.clients.TaskManagerClient; import org.opensearch.tasks.Task; import org.opensearch.tasks.TaskCancellationMonitoringService; import org.opensearch.tasks.TaskCancellationMonitoringSettings; @@ -298,6 +300,7 @@ import java.util.stream.Stream; import static java.util.stream.Collectors.toList; +import static org.opensearch.common.util.FeatureFlags.BACKGROUND_TASK_EXECUTION_EXPERIMENTAL; import static org.opensearch.common.util.FeatureFlags.TELEMETRY; import static org.opensearch.env.NodeEnvironment.collectFileCacheDataPath; import static org.opensearch.index.ShardIndexingPressureSettings.SHARD_INDEXING_PRESSURE_ENABLED_ATTRIBUTE_KEY; @@ -1314,6 +1317,13 @@ protected Node( .flatMap(List::stream) .collect(toList()); + final Optional taskManagerClientOptional = FeatureFlags.isEnabled(BACKGROUND_TASK_EXECUTION_EXPERIMENTAL) + ? pluginsService.filterPlugins(TaskManagerClientPlugin.class) + .stream() + .map(plugin -> plugin.getTaskManagerClient(client, clusterService, threadPool)) + .findFirst() + : Optional.empty(); + final PersistentTasksExecutorRegistry registry = new PersistentTasksExecutorRegistry(tasksExecutors); final PersistentTasksClusterService persistentTasksClusterService = new PersistentTasksClusterService( settings, @@ -1420,6 +1430,8 @@ protected Node( b.bind(PersistedStateRegistry.class).toInstance(persistedStateRegistry); b.bind(SegmentReplicationStatsTracker.class).toInstance(segmentReplicationStatsTracker); b.bind(SearchRequestOperationsCompositeListenerFactory.class).toInstance(searchRequestOperationsCompositeListenerFactory); + + taskManagerClientOptional.ifPresent(value -> b.bind(TaskManagerClient.class).toInstance(value)); }); injector = modules.createInjector(); diff --git a/server/src/main/java/org/opensearch/plugins/TaskManagerClientPlugin.java b/server/src/main/java/org/opensearch/plugins/TaskManagerClientPlugin.java new file mode 100644 index 0000000000000..3f511bd4c48d2 --- /dev/null +++ b/server/src/main/java/org/opensearch/plugins/TaskManagerClientPlugin.java @@ -0,0 +1,27 @@ +/* + * 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.plugins; + +import org.opensearch.client.Client; +import org.opensearch.cluster.service.ClusterService; +import org.opensearch.common.annotation.ExperimentalApi; +import org.opensearch.task.commons.clients.TaskManagerClient; +import org.opensearch.threadpool.ThreadPool; + +/** + * Plugin to provide an implementation of Task client + */ +@ExperimentalApi +public interface TaskManagerClientPlugin { + + /** + * Get the task client. + */ + TaskManagerClient getTaskManagerClient(Client client, ClusterService clusterService, ThreadPool threadPool); +} diff --git a/server/src/main/java/org/opensearch/plugins/TaskWorkerPlugin.java b/server/src/main/java/org/opensearch/plugins/TaskWorkerPlugin.java new file mode 100644 index 0000000000000..3ac96ed989425 --- /dev/null +++ b/server/src/main/java/org/opensearch/plugins/TaskWorkerPlugin.java @@ -0,0 +1,36 @@ +/* + * 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.plugins; + +import org.opensearch.client.Client; +import org.opensearch.cluster.service.ClusterService; +import org.opensearch.common.annotation.ExperimentalApi; +import org.opensearch.task.commons.task.TaskType; +import org.opensearch.task.commons.worker.TaskWorker; +import org.opensearch.threadpool.ThreadPool; + +/** + * Plugin for providing TaskWorkers for Offline Nodes + */ +@ExperimentalApi +public interface TaskWorkerPlugin { + + /** + * Get the new TaskWorker for a TaskType + * + * @return TaskWorker to execute Tasks on Offline Nodes + */ + TaskWorker getTaskWorker(Client client, ClusterService clusterService, ThreadPool threadPool); + + /** + * Get the TaskType for this TaskWorker + * @return TaskType + */ + TaskType getTaskType(); +}