You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@celeborn.apache.org by ch...@apache.org on 2024/07/11 02:57:11 UTC
(celeborn) branch main updated: [CELEBORN-1477][CIP-9] Refine the celeborn RESTful APIs
This is an automated email from the ASF dual-hosted git repository.
chengpan pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/celeborn.git
The following commit(s) were added to refs/heads/main by this push:
new d698a69ed [CELEBORN-1477][CIP-9] Refine the celeborn RESTful APIs
d698a69ed is described below
commit d698a69edc51f783d20e4726039aa849136dfb6a
Author: Fei Wang <fw...@ebay.com>
AuthorDate: Thu Jul 11 10:57:00 2024 +0800
[CELEBORN-1477][CIP-9] Refine the celeborn RESTful APIs
### What changes were proposed in this pull request?
This PR is for [CIP-9 Refine the celeborn RESTful APIs](https://docs.google.com/document/d/1LV2vV-w3XtlbJj2Vi4J77mt4IYCr40-8A_JncZLsHqs/edit?usp=sharing).
We leverage [openapi-generator](https://github.com/OpenAPITools/openapi-generator) to generate the client and model code.
### Why are the changes needed?
Celeborn has implemented RESTful APIs for monitoring and administrative operations on both master and worker endpoints. These APIs enable tasks such as configuration checks, status viewing of master/worker nodes, worker decommissioning/recommissioning, and more. They provide crucial insights and support for DevOps.
The primary concern with the existing API is the response content type, which is `text/plain` rather than the more widely accepted `application/json`. This mismatch makes integration with DevOps tools challenging, as these tools typically require JSON-formatted responses for seamless parsing and automation.
And I also saw the need for REST API evolution in[ Apache Celeborn CLI Proposal](https://cwiki.apache.org/confluence/display/CELEBORN/CIP-7+Celeborn+CLI).
### Does this PR introduce _any_ user-facing change?
This pr introduce a new API namespace: `/api/v1`. This approach allows us to maintain the current API for compatibility while offering an improved version.
### How was this patch tested?
UT.
Closes #2599 from turboFei/cip_9_openapi.
Lead-authored-by: Fei Wang <fw...@ebay.com>
Co-authored-by: Fei Wang <cn...@gmail.com>
Signed-off-by: Cheng Pan <ch...@apache.org>
---
LICENSE-binary | 1 +
.../celeborn/common/meta/AppDiskUsageMetric.scala | 6 +
.../common/meta/WorkerPartitionLocationInfo.scala | 4 +-
dev/deps/dependencies-server | 3 +
.../celeborn/service/deploy/master/Master.scala | 53 +-
.../deploy/master/http/api/ApiMasterResource.scala | 24 +-
.../master/http/api/v1/ApiV1MasterResource.scala | 29 +-
.../master/http/api/v1/ApplicationResource.scala | 97 +++
.../deploy/master/http/api/v1/MasterResource.scala | 81 +++
.../master/http/api/v1/ShuffleResource.scala | 49 ++
.../deploy/master/http/api/v1/WorkerResource.scala | 131 ++++
.../http/api/v1/ApiV1MasterResourceSuite.scala | 138 ++++
openapi/openapi-client/pom.xml | 202 ++++++
.../src/main/openapi3/master_rest_v1.yaml | 712 +++++++++++++++++++++
.../src/main/openapi3/worker_rest_v1.yaml | 627 ++++++++++++++++++
openapi/openapi-model/pom.xml | 165 +++++
.../java/org/apache/celeborn/rest/v1/JSON.java | 7 +-
pom.xml | 37 ++
project/CelebornBuild.scala | 150 ++++-
project/plugins.sbt | 3 +
service/pom.xml | 10 +
.../celeborn/server/common/HttpService.scala | 6 +-
.../server/common/http/api/ApiBaseResource.scala | 2 +
.../server/common/http/api/ApiRootResource.scala | 2 +-
.../common/http/api/CelebornOpenApiResource.scala | 2 +-
.../server/common/http/api/OpenAPIConfig.scala | 17 +-
.../server/common/http/api/v1/ApiUtils.scala | 131 ++++
.../common/http/api/v1/ApiV1BaseResource.scala | 59 ++
.../server/common/http/api/v1/ConfResource.scala | 135 ++++
.../http/api/v1/ApiV1BaseResourceSuite.scala | 46 ++
worker/pom.xml | 6 +
.../deploy/worker/http/api/ApiWorkerResource.scala | 2 +
.../worker/http/api/v1/ApiV1WorkerResource.scala | 18 +-
.../worker/http/api/v1/ApplicationResource.scala | 74 +++
.../worker/http/api/v1/ShuffleResource.scala | 78 +++
.../deploy/worker/http/api/v1/WorkerResource.scala | 88 +++
.../http/api/v1/ApiV1OpenapiClientSuite.scala | 112 ++++
.../api/v1/ApiV1WorkerOpenapiClientSuite.scala | 89 +++
.../http/api/v1/ApiV1WorkerResourceSuite.scala | 82 +++
39 files changed, 3416 insertions(+), 62 deletions(-)
diff --git a/LICENSE-binary b/LICENSE-binary
index 5c03a8339..221f632dd 100644
--- a/LICENSE-binary
+++ b/LICENSE-binary
@@ -298,6 +298,7 @@ org.glassfish.jersey.media:jersey-media-json-jackson
org.glassfish.jersey.media:jersey-media-multipart
org.javassist:javassist
org.mybatis:mybatis
+org.openapitools:jackson-databind-nullable
org.roaringbitmap:RoaringBitmap
org.rocksdb:rocksdbjni
org.scala-lang:scala-library
diff --git a/common/src/main/scala/org/apache/celeborn/common/meta/AppDiskUsageMetric.scala b/common/src/main/scala/org/apache/celeborn/common/meta/AppDiskUsageMetric.scala
index 92e08b5e0..497c73ce5 100644
--- a/common/src/main/scala/org/apache/celeborn/common/meta/AppDiskUsageMetric.scala
+++ b/common/src/main/scala/org/apache/celeborn/common/meta/AppDiskUsageMetric.scala
@@ -185,4 +185,10 @@ class AppDiskUsageMetric(conf: CelebornConf) extends Logging {
}
}
}
+
+ def topSnapshots(): Seq[AppDiskUsageSnapShot] = {
+ snapShots.take(snapshotCount)
+ .filter(_ != null)
+ .filter(_.topNItems.exists(_ != null))
+ }
}
diff --git a/common/src/main/scala/org/apache/celeborn/common/meta/WorkerPartitionLocationInfo.scala b/common/src/main/scala/org/apache/celeborn/common/meta/WorkerPartitionLocationInfo.scala
index 18d9fe10a..96ff20971 100644
--- a/common/src/main/scala/org/apache/celeborn/common/meta/WorkerPartitionLocationInfo.scala
+++ b/common/src/main/scala/org/apache/celeborn/common/meta/WorkerPartitionLocationInfo.scala
@@ -31,8 +31,8 @@ class WorkerPartitionLocationInfo extends Logging {
// key: ShuffleKey, values: (uniqueId -> PartitionLocation))
type PartitionInfo = ConcurrentHashMap[String, ConcurrentHashMap[String, PartitionLocation]]
- private val primaryPartitionLocations = new PartitionInfo
- private val replicaPartitionLocations = new PartitionInfo
+ private[celeborn] val primaryPartitionLocations = new PartitionInfo
+ private[celeborn] val replicaPartitionLocations = new PartitionInfo
def shuffleKeySet: util.HashSet[String] = {
val shuffleKeySet = new util.HashSet[String]()
diff --git a/dev/deps/dependencies-server b/dev/deps/dependencies-server
index edc800ee2..33b965865 100644
--- a/dev/deps/dependencies-server
+++ b/dev/deps/dependencies-server
@@ -34,6 +34,7 @@ hk2-locator/2.6.1//hk2-locator-2.6.1.jar
hk2-utils/2.6.1//hk2-utils-2.6.1.jar
jackson-annotations/2.15.3//jackson-annotations-2.15.3.jar
jackson-core/2.15.3//jackson-core-2.15.3.jar
+jackson-databind-nullable/0.2.6//jackson-databind-nullable-0.2.6.jar
jackson-databind/2.15.3//jackson-databind-2.15.3.jar
jackson-dataformat-yaml/2.13.2//jackson-dataformat-yaml-2.13.2.jar
jackson-datatype-jsr310/2.13.2//jackson-datatype-jsr310-2.13.2.jar
@@ -134,10 +135,12 @@ scala-reflect/2.12.18//scala-reflect-2.12.18.jar
slf4j-api/1.7.36//slf4j-api-1.7.36.jar
snakeyaml/2.2//snakeyaml-2.2.jar
snappy-java/1.1.10.5//snappy-java-1.1.10.5.jar
+swagger-annotations/1.6.11//swagger-annotations-1.6.11.jar
swagger-annotations/2.2.1//swagger-annotations-2.2.1.jar
swagger-core/2.2.1//swagger-core-2.2.1.jar
swagger-integration/2.2.1//swagger-integration-2.2.1.jar
swagger-jaxrs2/2.2.1//swagger-jaxrs2-2.2.1.jar
+swagger-models/1.6.11//swagger-models-1.6.11.jar
swagger-models/2.2.1//swagger-models-2.2.1.jar
swagger-ui/4.9.1//swagger-ui-4.9.1.jar
zstd-jni/1.5.2-1//zstd-jni-1.5.2-1.jar
diff --git a/master/src/main/scala/org/apache/celeborn/service/deploy/master/Master.scala b/master/src/main/scala/org/apache/celeborn/service/deploy/master/Master.scala
index 811e03b0c..2ce5fc83f 100644
--- a/master/src/main/scala/org/apache/celeborn/service/deploy/master/Master.scala
+++ b/master/src/main/scala/org/apache/celeborn/service/deploy/master/Master.scala
@@ -134,7 +134,7 @@ private[celeborn] class Master(
}
private val rackResolver = new CelebornRackResolver(conf)
- private val statusSystem =
+ private[celeborn] val statusSystem =
if (conf.haEnabled) {
val sys = new HAMasterMetaManager(internalRpcEnvInUse, conf, rackResolver)
val handler = new MetaHandler(sys)
@@ -1183,33 +1183,32 @@ private[celeborn] class Master(
statusSystem.workers.asScala.mkString("\n")
}
- override def handleWorkerEvent(workerEventType: String, workers: String): String = {
- val sb = new StringBuilder
- if (workerEventType.isEmpty || workers.isEmpty) {
- return sb.append(
- s"handle eventType failed as eventType: $workerEventType or workers: $workers has empty value").toString()
- }
-
- sb.append("============================ Handle Worker Event =============================\n")
- val workerArray = workers.split(",").filter(_.nonEmpty)
+ override def handleWorkerEvent(
+ workerEventType: String,
+ workers: Seq[WorkerInfo]): HandleResponse = {
+ val sb = new StringBuilder()
try {
val workerEventResponse = self.askSync[PbWorkerEventResponse](WorkerEventRequest(
- workerArray.map(WorkerInfo.fromUniqueId).toList.asJava,
+ workers.asJava,
workerEventType,
MasterClient.genRequestId()))
if (workerEventResponse.getSuccess) {
- sb.append(s"handle $workerEventType for ${workerArray.mkString(",")} successfully")
+ sb.append(
+ s"handle $workerEventType for ${workers.map(_.readableAddress()).mkString(",")} successfully")
} else {
- sb.append(s"handle $workerEventType for ${workerArray.mkString(",")} failed")
+ sb.append(
+ s"handle $workerEventType for ${workers.map(_.readableAddress).mkString(",")} failed")
}
+ workerEventResponse.getSuccess -> sb.toString()
} catch {
case e: Throwable =>
val message =
- s"handle $workerEventType for ${workerArray.mkString(",")} failed, message: ${e.getMessage}"
+ s"handle $workerEventType for ${workers.map(_.readableAddress()).mkString(
+ ",")} failed, message: ${e.getMessage}"
logError(message, e)
sb.append(message)
+ false -> sb.toString()
}
- sb.append("\n").toString()
}
override def getWorkerInfo: String = {
@@ -1289,30 +1288,30 @@ private[celeborn] class Master(
sb.toString()
}
- override def exclude(addWorkers: String, removeWorkers: String): String = {
+ override def exclude(
+ addWorkers: Seq[WorkerInfo],
+ removeWorkers: Seq[WorkerInfo]): HandleResponse = {
val sb = new StringBuilder
- sb.append("============================ Add/Remove Excluded Workers Manually =============================\n")
- val workersToAdd = addWorkers.split(",").filter(_.nonEmpty).map(WorkerInfo.fromUniqueId).toList
- val workersToRemove =
- removeWorkers.split(",").filter(_.nonEmpty).map(WorkerInfo.fromUniqueId).toList
val workerExcludeResponse = self.askSync[PbWorkerExcludeResponse](WorkerExclude(
- workersToAdd.asJava,
- workersToRemove.asJava,
+ addWorkers.asJava,
+ removeWorkers.asJava,
MasterClient.genRequestId()))
if (workerExcludeResponse.getSuccess) {
sb.append(
- s"Excluded workers add ${workersToAdd.mkString(",")} and remove ${workersToRemove.mkString(",")} successfully.\n")
+ s"Excluded workers add ${addWorkers.map(_.readableAddress).mkString(
+ ",")} and remove ${removeWorkers.map(_.readableAddress).mkString(",")} successfully.\n")
} else {
sb.append(
- s"Failed to Exclude workers add ${workersToAdd.mkString(",")} and remove ${workersToRemove.mkString(",")}.\n")
+ s"Failed to Exclude workers add ${addWorkers.map(_.readableAddress).mkString(
+ ",")} and remove ${removeWorkers.map(_.readableAddress).mkString(",")}.\n")
}
val unknownExcludedWorkers =
- (workersToAdd ++ workersToRemove).filter(!statusSystem.workers.contains(_))
+ (addWorkers ++ removeWorkers).filter(!statusSystem.workers.contains(_))
if (unknownExcludedWorkers.nonEmpty) {
sb.append(
- s"Unknown worker ${unknownExcludedWorkers.mkString(",")}. Workers in Master:\n$getWorkers.")
+ s"Unknown workers ${unknownExcludedWorkers.map(_.readableAddress).mkString(",")}. Workers in Master:\n$getWorkers.")
}
- sb.toString()
+ workerExcludeResponse.getSuccess -> sb.toString()
}
private def isMasterActive: Int = {
diff --git a/master/src/main/scala/org/apache/celeborn/service/deploy/master/http/api/ApiMasterResource.scala b/master/src/main/scala/org/apache/celeborn/service/deploy/master/http/api/ApiMasterResource.scala
index 935255f4e..0e32de2f3 100644
--- a/master/src/main/scala/org/apache/celeborn/service/deploy/master/http/api/ApiMasterResource.scala
+++ b/master/src/main/scala/org/apache/celeborn/service/deploy/master/http/api/ApiMasterResource.scala
@@ -22,9 +22,13 @@ import javax.ws.rs.core.MediaType
import io.swagger.v3.oas.annotations.media.Content
import io.swagger.v3.oas.annotations.responses.ApiResponse
+import io.swagger.v3.oas.annotations.tags.Tag
+import org.apache.commons.lang3.StringUtils
+import org.apache.celeborn.common.meta.WorkerInfo
import org.apache.celeborn.server.common.http.api.ApiRequestContext
+@Tag(name = "Deprecated")
@Path("/")
class ApiMasterResource extends ApiRequestContext {
@@ -103,7 +107,15 @@ class ApiMasterResource extends ApiRequestContext {
def exclude(
@FormParam("add") addWorkers: String,
@FormParam("remove") removeWorkers: String): String = {
- httpService.exclude(normalizeParam(addWorkers), normalizeParam(removeWorkers))
+ val sb = new StringBuilder
+ sb.append("============================ Add/Remove Excluded Workers Manually =============================\n")
+ val workersToAdd =
+ normalizeParam(addWorkers).split(",").filter(_.nonEmpty).map(WorkerInfo.fromUniqueId).toList
+ val workersToRemove =
+ normalizeParam(removeWorkers).split(",").filter(_.nonEmpty).map(
+ WorkerInfo.fromUniqueId).toList
+ sb.append(httpService.exclude(workersToAdd, workersToRemove)._2)
+ sb.toString()
}
@Path("/sendWorkerEvent")
@@ -117,6 +129,14 @@ class ApiMasterResource extends ApiRequestContext {
def sendWorkerEvent(
@FormParam("type") eventType: String,
@FormParam("workers") workers: String): String = {
- httpService.handleWorkerEvent(normalizeParam(eventType), normalizeParam(workers))
+ val sb = new StringBuilder
+ if (StringUtils.isEmpty(eventType) || StringUtils.isEmpty(workers)) {
+ return sb.append(
+ s"handle eventType failed as eventType: $eventType or workers: $workers has empty value").toString()
+ }
+ sb.append("============================ Handle Worker Event =============================\n")
+ val workerList = workers.split(",").filter(_.nonEmpty).map(WorkerInfo.fromUniqueId)
+ sb.append(httpService.handleWorkerEvent(normalizeParam(eventType), workerList)._2)
+ sb.toString()
}
}
diff --git a/service/src/main/scala/org/apache/celeborn/server/common/http/api/OpenAPIConfig.scala b/master/src/main/scala/org/apache/celeborn/service/deploy/master/http/api/v1/ApiV1MasterResource.scala
similarity index 58%
copy from service/src/main/scala/org/apache/celeborn/server/common/http/api/OpenAPIConfig.scala
copy to master/src/main/scala/org/apache/celeborn/service/deploy/master/http/api/v1/ApiV1MasterResource.scala
index b9355ac00..e80916710 100644
--- a/service/src/main/scala/org/apache/celeborn/server/common/http/api/OpenAPIConfig.scala
+++ b/master/src/main/scala/org/apache/celeborn/service/deploy/master/http/api/v1/ApiV1MasterResource.scala
@@ -15,20 +15,23 @@
* limitations under the License.
*/
-package org.apache.celeborn.server.common.http.api
+package org.apache.celeborn.service.deploy.master.http.api.v1
-import org.glassfish.jersey.server.ResourceConfig
+import javax.ws.rs.Path
-class OpenAPIConfig extends ResourceConfig {
- packages(OpenAPIConfig.packages: _*)
- register(classOf[CelebornOpenApiResource])
- register(classOf[CelebornScalaObjectMapper])
- register(classOf[RestExceptionMapper])
-}
+import org.apache.celeborn.server.common.http.api.ApiRequestContext
+
+@Path("/api/v1")
+class ApiV1MasterResource extends ApiRequestContext {
+ @Path("shuffles")
+ def shuffles: Class[ShuffleResource] = classOf[ShuffleResource]
+
+ @Path("applications")
+ def applications: Class[ApplicationResource] = classOf[ApplicationResource]
+
+ @Path("masters")
+ def masters: Class[MasterResource] = classOf[MasterResource]
-object OpenAPIConfig {
- val packages = Seq(
- "org.apache.celeborn.server.common.http.api",
- "org.apache.celeborn.service.deploy.master.http.api",
- "org.apache.celeborn.service.deploy.worker.http.api")
+ @Path("workers")
+ def workers: Class[WorkerResource] = classOf[WorkerResource]
}
diff --git a/master/src/main/scala/org/apache/celeborn/service/deploy/master/http/api/v1/ApplicationResource.scala b/master/src/main/scala/org/apache/celeborn/service/deploy/master/http/api/v1/ApplicationResource.scala
new file mode 100644
index 000000000..23b230728
--- /dev/null
+++ b/master/src/main/scala/org/apache/celeborn/service/deploy/master/http/api/v1/ApplicationResource.scala
@@ -0,0 +1,97 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.celeborn.service.deploy.master.http.api.v1
+
+import javax.ws.rs.{Consumes, GET, Path, Produces}
+import javax.ws.rs.core.MediaType
+
+import scala.collection.JavaConverters._
+
+import io.swagger.v3.oas.annotations.media.{Content, Schema}
+import io.swagger.v3.oas.annotations.responses.ApiResponse
+import io.swagger.v3.oas.annotations.tags.Tag
+
+import org.apache.celeborn.common.util.Utils
+import org.apache.celeborn.rest.v1.model.{AppDiskUsageData, AppDiskUsageSnapshotData, AppDiskUsageSnapshotsResponse, ApplicationHeartbeatData, ApplicationsHeartbeatResponse, HostnamesResponse}
+import org.apache.celeborn.server.common.http.api.ApiRequestContext
+import org.apache.celeborn.service.deploy.master.Master
+
+@Tag(name = "Application")
+@Produces(Array(MediaType.APPLICATION_JSON))
+@Consumes(Array(MediaType.APPLICATION_JSON))
+class ApplicationResource extends ApiRequestContext {
+ private def statusSystem = httpService.asInstanceOf[Master].statusSystem
+
+ @ApiResponse(
+ responseCode = "200",
+ content = Array(new Content(
+ mediaType = MediaType.APPLICATION_JSON,
+ schema = new Schema(implementation = classOf[ApplicationsHeartbeatResponse]))),
+ description = "List all running application's ids of the cluster.")
+ @GET
+ def applications(): ApplicationsHeartbeatResponse = {
+ new ApplicationsHeartbeatResponse()
+ .applications(
+ statusSystem.appHeartbeatTime.asScala.map { case (appId, heartbeat) =>
+ new ApplicationHeartbeatData()
+ .appId(appId)
+ .lastHeartbeatTimestamp(heartbeat)
+ }.toSeq.asJava)
+ }
+
+ @ApiResponse(
+ responseCode = "200",
+ content = Array(new Content(
+ mediaType = MediaType.APPLICATION_JSON,
+ schema = new Schema(implementation = classOf[AppDiskUsageSnapshotsResponse]))),
+ description =
+ "List the top disk usage application ids. It will return the top disk usage application ids for the cluster.")
+ @Path("/top_disk_usages")
+ @GET
+ def topDiskUsedApplications(): AppDiskUsageSnapshotsResponse = {
+ new AppDiskUsageSnapshotsResponse()
+ .snapshots(
+ statusSystem.appDiskUsageMetric.topSnapshots().map { snapshot =>
+ new AppDiskUsageSnapshotData()
+ .start(
+ snapshot.startSnapShotTime)
+ .end(
+ snapshot.endSnapShotTime)
+ .topNItems(
+ snapshot.topNItems.map { usage =>
+ new AppDiskUsageData()
+ .appId(usage.appId)
+ .estimatedUsage(usage.estimatedUsage)
+ .estimatedUsageStr(Utils.bytesToString(usage.estimatedUsage))
+ }.toSeq.asJava)
+ }.asJava)
+ }
+
+ @ApiResponse(
+ responseCode = "200",
+ content = Array(new Content(
+ mediaType = MediaType.APPLICATION_JSON,
+ schema = new Schema(implementation = classOf[HostnamesResponse]))),
+ description =
+ "List all running application's LifecycleManager's hostnames of the cluster.")
+ @Path("/hostnames")
+ @GET
+ def hostnames(): HostnamesResponse = {
+ new HostnamesResponse().hostnames(statusSystem.hostnameSet.asScala.toSeq.asJava)
+ }
+}
diff --git a/master/src/main/scala/org/apache/celeborn/service/deploy/master/http/api/v1/MasterResource.scala b/master/src/main/scala/org/apache/celeborn/service/deploy/master/http/api/v1/MasterResource.scala
new file mode 100644
index 000000000..94cd25700
--- /dev/null
+++ b/master/src/main/scala/org/apache/celeborn/service/deploy/master/http/api/v1/MasterResource.scala
@@ -0,0 +1,81 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.celeborn.service.deploy.master.http.api.v1
+
+import javax.ws.rs.{BadRequestException, Consumes, GET, Produces}
+import javax.ws.rs.core.MediaType
+
+import scala.collection.JavaConverters._
+
+import io.swagger.v3.oas.annotations.media.{Content, Schema}
+import io.swagger.v3.oas.annotations.responses.ApiResponse
+import io.swagger.v3.oas.annotations.tags.Tag
+import org.apache.ratis.proto.RaftProtos.RaftPeerRole
+
+import org.apache.celeborn.rest.v1.model.{MasterCommitData, MasterInfoResponse, MasterLeader}
+import org.apache.celeborn.server.common.http.api.ApiRequestContext
+import org.apache.celeborn.service.deploy.master.Master
+import org.apache.celeborn.service.deploy.master.clustermeta.ha.HAMasterMetaManager
+
+@Tag(name = "Master")
+@Produces(Array(MediaType.APPLICATION_JSON))
+@Consumes(Array(MediaType.APPLICATION_JSON))
+class MasterResource extends ApiRequestContext {
+ private def master = httpService.asInstanceOf[Master]
+
+ @ApiResponse(
+ responseCode = "200",
+ content = Array(new Content(
+ mediaType = MediaType.APPLICATION_JSON,
+ schema = new Schema(implementation = classOf[MasterInfoResponse]))),
+ description =
+ "List master group information of the service. It will list all master's LEADER, FOLLOWER information.")
+ @GET
+ def masterGroupInfo: MasterInfoResponse = {
+ if (master.conf.haEnabled) {
+ val groupInfo =
+ master.statusSystem.asInstanceOf[HAMasterMetaManager].getRatisServer.getGroupInfo
+ val leader = Option(groupInfo.getRoleInfoProto).map { roleInfo =>
+ if (roleInfo.getRole == RaftPeerRole.LEADER) {
+ roleInfo.getSelf
+ } else {
+ Option(roleInfo.getFollowerInfo).map(_.getLeaderInfo.getId).orNull
+ }
+ }.orNull
+ val masterLeader = Option(leader).map { _ =>
+ new MasterLeader()
+ .id(leader.getId.toString)
+ .address(leader.getAddress)
+ }.orNull
+ val masterCommitDataList = groupInfo.getCommitInfos.asScala.map { commitInfo =>
+ new MasterCommitData()
+ .commitIndex(commitInfo.getCommitIndex)
+ .id(commitInfo.getServer.getId.toString)
+ .address(commitInfo.getServer.getAddress)
+ .clientAddress(commitInfo.getServer.getClientAddress)
+ .startUpRole(commitInfo.getServer.getStartupRole.toString)
+ }
+ new MasterInfoResponse()
+ .groupId(groupInfo.getGroup.getGroupId.getUuid.toString)
+ .leader(masterLeader)
+ .masterCommitInfo(masterCommitDataList.toSeq.asJava)
+ } else {
+ throw new BadRequestException("HA is not enabled")
+ }
+ }
+}
diff --git a/master/src/main/scala/org/apache/celeborn/service/deploy/master/http/api/v1/ShuffleResource.scala b/master/src/main/scala/org/apache/celeborn/service/deploy/master/http/api/v1/ShuffleResource.scala
new file mode 100644
index 000000000..4b7319d46
--- /dev/null
+++ b/master/src/main/scala/org/apache/celeborn/service/deploy/master/http/api/v1/ShuffleResource.scala
@@ -0,0 +1,49 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.celeborn.service.deploy.master.http.api.v1
+
+import javax.ws.rs.{Consumes, GET, Produces}
+import javax.ws.rs.core.MediaType
+
+import scala.collection.JavaConverters._
+
+import io.swagger.v3.oas.annotations.media.{Content, Schema}
+import io.swagger.v3.oas.annotations.responses.ApiResponse
+import io.swagger.v3.oas.annotations.tags.Tag
+
+import org.apache.celeborn.rest.v1.model.ShufflesResponse
+import org.apache.celeborn.server.common.http.api.ApiRequestContext
+import org.apache.celeborn.service.deploy.master.Master
+
+@Tag(name = "Shuffle")
+@Produces(Array(MediaType.APPLICATION_JSON))
+@Consumes(Array(MediaType.APPLICATION_JSON))
+class ShuffleResource extends ApiRequestContext {
+ private def statusSystem = httpService.asInstanceOf[Master].statusSystem
+ @ApiResponse(
+ responseCode = "200",
+ content = Array(new Content(
+ mediaType = MediaType.APPLICATION_JSON,
+ schema = new Schema(implementation = classOf[ShufflesResponse]))),
+ description =
+ "List all running shuffle keys of the service. It will return all running shuffle's key of the cluster.")
+ @GET
+ def shuffles: ShufflesResponse = {
+ new ShufflesResponse().shuffleIds(statusSystem.registeredShuffle.asScala.toSeq.asJava)
+ }
+}
diff --git a/master/src/main/scala/org/apache/celeborn/service/deploy/master/http/api/v1/WorkerResource.scala b/master/src/main/scala/org/apache/celeborn/service/deploy/master/http/api/v1/WorkerResource.scala
new file mode 100644
index 000000000..969a28c0e
--- /dev/null
+++ b/master/src/main/scala/org/apache/celeborn/service/deploy/master/http/api/v1/WorkerResource.scala
@@ -0,0 +1,131 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.celeborn.service.deploy.master.http.api.v1
+
+import javax.ws.rs.{BadRequestException, Consumes, GET, Path, POST, Produces}
+import javax.ws.rs.core.MediaType
+
+import scala.collection.JavaConverters._
+
+import io.swagger.v3.oas.annotations.media.{Content, Schema}
+import io.swagger.v3.oas.annotations.responses.ApiResponse
+import io.swagger.v3.oas.annotations.tags.Tag
+
+import org.apache.celeborn.rest.v1.model.{ExcludeWorkerRequest, HandleResponse, SendWorkerEventRequest, WorkerEventData, WorkerEventInfoData, WorkerEventsResponse, WorkersResponse, WorkerTimestampData}
+import org.apache.celeborn.server.common.http.api.ApiRequestContext
+import org.apache.celeborn.server.common.http.api.v1.ApiUtils
+import org.apache.celeborn.service.deploy.master.Master
+
+@Tag(name = "Worker")
+@Produces(Array(MediaType.APPLICATION_JSON))
+@Consumes(Array(MediaType.APPLICATION_JSON))
+class WorkerResource extends ApiRequestContext {
+ private def statusSystem = httpService.asInstanceOf[Master].statusSystem
+
+ @ApiResponse(
+ responseCode = "200",
+ content = Array(new Content(
+ mediaType = MediaType.APPLICATION_JSON,
+ schema = new Schema(implementation = classOf[WorkersResponse]))),
+ description =
+ "List worker information of the service. It will list all registered workers' information.")
+ @GET
+ def workers: WorkersResponse = {
+ new WorkersResponse()
+ .workers(statusSystem.workers.asScala.map(ApiUtils.workerData).toSeq.asJava)
+ .lostWorkers(statusSystem.lostWorkers.asScala.toSeq.sortBy(_._2)
+ .map(kv =>
+ new WorkerTimestampData().worker(ApiUtils.workerData(kv._1)).timestamp(kv._2)).asJava)
+ .excludedWorkers(
+ (statusSystem.excludedWorkers.asScala ++ statusSystem.manuallyExcludedWorkers.asScala)
+ .map(ApiUtils.workerData).toSeq.asJava)
+ .manualExcludedWorkers(statusSystem.manuallyExcludedWorkers.asScala.map(
+ ApiUtils.workerData).toSeq.asJava)
+ .shutdownWorkers(statusSystem.shutdownWorkers.asScala.map(
+ ApiUtils.workerData).toSeq.asJava)
+ .decommissioningWorkers(statusSystem.decommissionWorkers.asScala.map(
+ ApiUtils.workerData).toSeq.asJava)
+ }
+
+ @ApiResponse(
+ responseCode = "200",
+ content = Array(new Content(
+ mediaType = MediaType.APPLICATION_JSON,
+ schema = new Schema(implementation = classOf[HandleResponse]))),
+ description =
+ "Excluded workers of the master add or remove the worker manually given worker id. The parameter add or remove specifies the excluded workers to add or remove.")
+ @Path("/exclude")
+ @POST
+ def excludeWorker(request: ExcludeWorkerRequest): HandleResponse = {
+ val (success, msg) = httpService.exclude(
+ request.getAdd.asScala.map(ApiUtils.toWorkerInfo).toSeq,
+ request.getRemove.asScala.map(ApiUtils.toWorkerInfo).toSeq)
+ new HandleResponse().success(success).message(msg)
+ }
+
+ @ApiResponse(
+ responseCode = "200",
+ content = Array(new Content(
+ mediaType = MediaType.APPLICATION_JSON,
+ schema = new Schema(
+ implementation = classOf[WorkerEventsResponse]))),
+ description = "List all worker event infos of the master.")
+ @Path("/events")
+ @GET
+ def workerEvents(): WorkerEventsResponse = {
+ new WorkerEventsResponse().workerEvents(
+ statusSystem.workerEventInfos.asScala.map { case (worker, event) =>
+ new WorkerEventData()
+ .worker(
+ ApiUtils.workerData(worker)).event(
+ new WorkerEventInfoData()
+ .eventType(event.getEventType.toString)
+ .eventTime(event.getEventStartTime))
+ }.toSeq.asJava)
+ }
+
+ @ApiResponse(
+ responseCode = "200",
+ content = Array(new Content(
+ mediaType = MediaType.APPLICATION_JSON,
+ schema = new Schema(implementation = classOf[HandleResponse]))),
+ description =
+ "For Master(Leader) can send worker event to manager workers. Legal types are 'None', 'Immediately', 'Decommission', 'DecommissionThenIdle', 'Graceful', 'Recommission'.")
+ @Path("/events")
+ @POST
+ def sendWorkerEvents(request: SendWorkerEventRequest): HandleResponse = {
+ if (request.getEventType == SendWorkerEventRequest.EventTypeEnum.NONE || request.getWorkers.isEmpty) {
+ throw new BadRequestException(
+ s"eventType(${request.getEventType}) and workers(${request.getWorkers}) are required")
+ }
+ val workers = request.getWorkers.asScala.map(ApiUtils.toWorkerInfo).toSeq
+ val (filteredWorkers, unknownWorkers) = workers.partition(statusSystem.workers.contains)
+ if (filteredWorkers.isEmpty) {
+ throw new BadRequestException(
+ s"None of the workers are known: ${unknownWorkers.map(_.readableAddress).mkString(", ")}")
+ }
+ val (success, msg) = httpService.handleWorkerEvent(request.getEventType.toString, workers)
+ val finalMsg =
+ if (unknownWorkers.isEmpty) {
+ msg
+ } else {
+ s"${msg}\n(Unknown workers: ${unknownWorkers.map(_.readableAddress).mkString(", ")})"
+ }
+ new HandleResponse().success(success).message(finalMsg)
+ }
+}
diff --git a/master/src/test/scala/org/apache/celeborn/service/deploy/master/http/api/v1/ApiV1MasterResourceSuite.scala b/master/src/test/scala/org/apache/celeborn/service/deploy/master/http/api/v1/ApiV1MasterResourceSuite.scala
new file mode 100644
index 000000000..74ca872c3
--- /dev/null
+++ b/master/src/test/scala/org/apache/celeborn/service/deploy/master/http/api/v1/ApiV1MasterResourceSuite.scala
@@ -0,0 +1,138 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.celeborn.service.deploy.master.http.api.v1
+
+import java.util.Collections
+import javax.servlet.http.HttpServletResponse
+import javax.ws.rs.client.Entity
+import javax.ws.rs.core.MediaType
+
+import com.google.common.io.Files
+
+import org.apache.celeborn.common.CelebornConf
+import org.apache.celeborn.common.util.{CelebornExitKind, Utils}
+import org.apache.celeborn.rest.v1.model.{AppDiskUsageSnapshotsResponse, ApplicationsResponse, ExcludeWorkerRequest, HandleResponse, HostnamesResponse, SendWorkerEventRequest, ShufflesResponse, WorkerEventsResponse, WorkerId, WorkersResponse}
+import org.apache.celeborn.server.common.HttpService
+import org.apache.celeborn.server.common.http.api.v1.ApiV1BaseResourceSuite
+import org.apache.celeborn.service.deploy.master.{Master, MasterArguments}
+
+class ApiV1MasterResourceSuite extends ApiV1BaseResourceSuite {
+ private var master: Master = _
+
+ override protected def httpService: HttpService = master
+
+ def getTmpDir(): String = {
+ val tmpDir = Files.createTempDir()
+ tmpDir.deleteOnExit()
+ tmpDir.getAbsolutePath
+ }
+
+ override def beforeAll(): Unit = {
+ val randomMasterPort = Utils.selectRandomPort(1024, 65535)
+ val randomHttpPort = randomMasterPort + 1
+ celebornConf.set(CelebornConf.HA_ENABLED.key, "false")
+ celebornConf.set(CelebornConf.HA_MASTER_RATIS_STORAGE_DIR.key, getTmpDir())
+ celebornConf.set(CelebornConf.WORKER_STORAGE_DIRS.key, getTmpDir())
+ celebornConf.set(CelebornConf.MASTER_HTTP_HOST.key, "127.0.0.1")
+ celebornConf.set(CelebornConf.MASTER_HTTP_PORT.key, randomHttpPort.toString)
+
+ val args = Array("-h", "localhost", "-p", randomMasterPort.toString)
+
+ val masterArgs = new MasterArguments(args, celebornConf)
+ master = new Master(celebornConf, masterArgs)
+ new Thread() {
+ override def run(): Unit = {
+ master.initialize()
+ }
+ }.start()
+ super.beforeAll()
+ }
+
+ override def afterAll(): Unit = {
+ super.afterAll()
+ master.stop(CelebornExitKind.EXIT_IMMEDIATELY)
+ master.rpcEnv.shutdown()
+ }
+
+ test("shuffle resource") {
+ val response = webTarget.path("shuffles").request(MediaType.APPLICATION_JSON).get()
+ assert(HttpServletResponse.SC_OK == response.getStatus)
+ assert(response.readEntity(classOf[ShufflesResponse]).getShuffleIds.isEmpty)
+ }
+
+ test("application resource") {
+ var response = webTarget.path("applications").request(MediaType.APPLICATION_JSON).get()
+ assert(HttpServletResponse.SC_OK == response.getStatus)
+ assert(response.readEntity(classOf[ApplicationsResponse]).getApplications.isEmpty)
+
+ response =
+ webTarget.path("applications/top_disk_usages").request(MediaType.APPLICATION_JSON).get()
+ assert(HttpServletResponse.SC_OK == response.getStatus)
+ assert(response.readEntity(classOf[AppDiskUsageSnapshotsResponse]).getSnapshots.isEmpty)
+
+ response = webTarget.path("applications/hostnames").request(MediaType.APPLICATION_JSON).get()
+ assert(HttpServletResponse.SC_OK == response.getStatus)
+ assert(response.readEntity(classOf[HostnamesResponse]).getHostnames.isEmpty)
+ }
+
+ test("master resource") {
+ val response = webTarget.path("masters").request(MediaType.APPLICATION_JSON).get()
+ assert(HttpServletResponse.SC_BAD_REQUEST == response.getStatus)
+ assert(response.readEntity(classOf[String]).contains("HA is not enabled"))
+ }
+
+ test("worker resource") {
+ var response = webTarget.path("workers").request(MediaType.APPLICATION_JSON).get()
+ assert(HttpServletResponse.SC_OK == response.getStatus)
+ val workersResponse = response.readEntity(classOf[WorkersResponse])
+ assert(workersResponse.getWorkers.isEmpty)
+ assert(workersResponse.getLostWorkers.isEmpty)
+ assert(workersResponse.getExcludedWorkers.isEmpty)
+ assert(workersResponse.getShutdownWorkers.isEmpty)
+ assert(workersResponse.getDecommissioningWorkers.isEmpty)
+
+ val worker = new WorkerId()
+ .host("unknown.celeborn")
+ .rpcPort(0)
+ .pushPort(0)
+ .fetchPort(0)
+ .replicatePort(0)
+ val excludeWorkerRequest =
+ new ExcludeWorkerRequest()
+ .add(Collections.singletonList(worker))
+ .remove(Collections.singletonList(worker))
+ response = webTarget.path("workers/exclude").request(MediaType.APPLICATION_JSON).post(
+ Entity.entity(excludeWorkerRequest, MediaType.APPLICATION_JSON))
+ assert(HttpServletResponse.SC_OK == response.getStatus)
+ assert(response.readEntity(classOf[HandleResponse]).getMessage.contains(
+ "Unknown workers Host:unknown.celeborn"))
+
+ response = webTarget.path("workers/events").request(MediaType.APPLICATION_JSON).get()
+ assert(HttpServletResponse.SC_OK == response.getStatus)
+ assert(response.readEntity(classOf[WorkerEventsResponse]).getWorkerEvents.isEmpty)
+
+ val sendWorkerEventRequest = new SendWorkerEventRequest()
+ .eventType(SendWorkerEventRequest.EventTypeEnum.DECOMMISSION)
+ .workers(Collections.singletonList(worker))
+ response = webTarget.path("workers/events").request(MediaType.APPLICATION_JSON).post(
+ Entity.entity(sendWorkerEventRequest, MediaType.APPLICATION_JSON))
+ assert(HttpServletResponse.SC_BAD_REQUEST == response.getStatus)
+ assert(response.readEntity(classOf[String]).contains(
+ "None of the workers are known"))
+ }
+}
diff --git a/openapi/openapi-client/pom.xml b/openapi/openapi-client/pom.xml
new file mode 100644
index 000000000..ffe09c774
--- /dev/null
+++ b/openapi/openapi-client/pom.xml
@@ -0,0 +1,202 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+ ~ Licensed to the Apache Software Foundation (ASF) under one or more
+ ~ contributor license agreements. See the NOTICE file distributed with
+ ~ this work for additional information regarding copyright ownership.
+ ~ The ASF licenses this file to You under the Apache License, Version 2.0
+ ~ (the "License"); you may not use this file except in compliance with
+ ~ the License. You may obtain a copy of the License at
+ ~
+ ~ http://www.apache.org/licenses/LICENSE-2.0
+ ~
+ ~ Unless required by applicable law or agreed to in writing, software
+ ~ distributed under the License is distributed on an "AS IS" BASIS,
+ ~ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ ~ See the License for the specific language governing permissions and
+ ~ limitations under the License.
+ -->
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+ <modelVersion>4.0.0</modelVersion>
+
+ <parent>
+ <groupId>org.apache.celeborn</groupId>
+ <artifactId>celeborn-parent_${scala.binary.version}</artifactId>
+ <version>${project.version}</version>
+ <relativePath>../../pom.xml</relativePath>
+ </parent>
+
+ <artifactId>celeborn-openapi-client_${scala.binary.version}</artifactId>
+ <packaging>jar</packaging>
+ <name>Celeborn OpenApi Client</name>
+
+ <dependencyManagement>
+ <dependencies>
+ <dependency>
+ <groupId>org.glassfish.jersey.core</groupId>
+ <artifactId>jersey-client</artifactId>
+ <version>${jersey.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>com.fasterxml.jackson.datatype</groupId>
+ <artifactId>jackson-datatype-jsr310</artifactId>
+ <version>${jackson.version}</version>
+ </dependency>
+ </dependencies>
+ </dependencyManagement>
+
+ <dependencies>
+ <dependency>
+ <groupId>io.swagger</groupId>
+ <artifactId>swagger-annotations</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>io.swagger</groupId>
+ <artifactId>swagger-models</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.openapitools</groupId>
+ <artifactId>jackson-databind-nullable</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>com.google.code.findbugs</groupId>
+ <artifactId>jsr305</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.glassfish.jersey.core</groupId>
+ <artifactId>jersey-client</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.glassfish.jersey.media</groupId>
+ <artifactId>jersey-media-json-jackson</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>com.fasterxml.jackson.core</groupId>
+ <artifactId>jackson-annotations</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>com.fasterxml.jackson.core</groupId>
+ <artifactId>jackson-databind</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>com.fasterxml.jackson.datatype</groupId>
+ <artifactId>jackson-datatype-jsr310</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.glassfish.jersey.media</groupId>
+ <artifactId>jersey-media-multipart</artifactId>
+ </dependency>
+ </dependencies>
+
+ <build>
+ <plugins>
+ <plugin>
+ <groupId>org.openapitools</groupId>
+ <artifactId>openapi-generator-maven-plugin</artifactId>
+ <!--
+ please keep the schema definitions consistent between master_rest_v1.yaml and worker_rest_v1.yaml
+ -->
+ <executions>
+ <execution>
+ <id>generate-java-master-rest-v1</id>
+ <goals>
+ <goal>generate</goal>
+ </goals>
+ <phase>generate-sources</phase>
+ <configuration>
+ <inputSpec>${project.basedir}/src/main/openapi3/master_rest_v1.yaml</inputSpec>
+ <output>${project.build.directory}/generated-sources/java</output>
+ <apiPackage>org.apache.celeborn.rest.v1.master</apiPackage>
+ <modelPackage>org.apache.celeborn.rest.v1.master.model</modelPackage>
+ <invokerPackage>org.apache.celeborn.rest.v1.master.invoker</invokerPackage>
+ <generatorName>java</generatorName>
+ <generateApiTests>false</generateApiTests>
+ <generateModelTests>false</generateModelTests>
+ <groupId>${groupId}</groupId>
+ <artifactId>${artifactId}</artifactId>
+ <artifactVersion>${project.version}</artifactVersion>
+ <addCompileSourceRoot>false</addCompileSourceRoot>
+ <cleanupOutput>true</cleanupOutput>
+ <configOptions>
+ <licenseName>Apache License 2.0</licenseName>
+ <groupId>${groupId}</groupId>
+ <artifactId>${artifactId}</artifactId>
+ <artifactVersion>${project.version}</artifactVersion>
+ <dateLibrary>java8</dateLibrary>
+ <useGzipFeature>true</useGzipFeature>
+ <!--
+ When using jdk11+, use the following line to generate native java code.
+ <library>native</library>
+ -->
+ <library>jersey2</library>
+ <!--
+ When upgrading openapi-generator to 7.x+, which does not support JDK8,
+ use the following line to annotations with io.swagger.core.v3:swagger-annotations.
+ <annotationLibrary>swagger2</annotationLibrary>
+ -->
+ <annotationLibrary>swagger1</annotationLibrary>
+ </configOptions>
+ </configuration>
+ </execution>
+ <execution>
+ <id>generate-java-worker-rest-v1</id>
+ <goals>
+ <goal>generate</goal>
+ </goals>
+ <phase>generate-sources</phase>
+ <configuration>
+ <inputSpec>${project.basedir}/src/main/openapi3/worker_rest_v1.yaml</inputSpec>
+ <output>${project.build.directory}/generated-sources/java</output>
+ <apiPackage>org.apache.celeborn.rest.v1.worker</apiPackage>
+ <modelPackage>org.apache.celeborn.rest.v1.worker.model</modelPackage>
+ <invokerPackage>org.apache.celeborn.rest.v1.worker.invoker</invokerPackage>
+ <generatorName>java</generatorName>
+ <generateApiTests>false</generateApiTests>
+ <generateModelTests>false</generateModelTests>
+ <groupId>${groupId}</groupId>
+ <artifactId>${artifactId}</artifactId>
+ <artifactVersion>${project.version}</artifactVersion>
+ <addCompileSourceRoot>false</addCompileSourceRoot>
+ <configOptions>
+ <licenseName>Apache License 2.0</licenseName>
+ <groupId>${groupId}</groupId>
+ <artifactId>${artifactId}</artifactId>
+ <artifactVersion>${project.version}</artifactVersion>
+ <dateLibrary>java8</dateLibrary>
+ <useGzipFeature>true</useGzipFeature>
+ <!--
+ When using jdk11+, use the following line to generate native java code.
+ <library>native</library>
+ -->
+ <library>jersey2</library>
+ <!--
+ When upgrading openapi-generator to 7.x+, which does not support JDK8,
+ use the following line to annotations with io.swagger.core.v3:swagger-annotations.
+ <annotationLibrary>swagger2</annotationLibrary>
+ -->
+ <annotationLibrary>swagger1</annotationLibrary>
+ </configOptions>
+ </configuration>
+ </execution>
+ </executions>
+ </plugin>
+ <plugin>
+ <groupId>org.codehaus.mojo</groupId>
+ <artifactId>build-helper-maven-plugin</artifactId>
+ <executions>
+ <execution>
+ <id>add-source</id>
+ <goals>
+ <goal>add-source</goal>
+ </goals>
+ <phase>generate-sources</phase>
+ <configuration>
+ <sources>
+ <source>${project.build.directory}/generated-sources/java/src/main/java</source>
+ </sources>
+ </configuration>
+ </execution>
+ </executions>
+ </plugin>
+ </plugins>
+ </build>
+</project>
diff --git a/openapi/openapi-client/src/main/openapi3/master_rest_v1.yaml b/openapi/openapi-client/src/main/openapi3/master_rest_v1.yaml
new file mode 100644
index 000000000..73a860913
--- /dev/null
+++ b/openapi/openapi-client/src/main/openapi3/master_rest_v1.yaml
@@ -0,0 +1,712 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+openapi: 3.0.0
+servers:
+ - url: http://localhost:9098/
+ description: Localhost reference server
+security:
+ - basic: []
+info:
+ title: Celeborn REST API
+ license:
+ name: Apache 2.0
+ url: https://www.apache.org/licenses/LICENSE-2.0.html
+ version: 1.0.0
+paths:
+ /api/v1/conf:
+ get:
+ tags:
+ - Conf
+ operationId: getConf
+ description: List the conf setting.
+ responses:
+ "200":
+ description: The request was successful.
+ content:
+ application/json:
+ schema:
+ $ref: '#/components/schemas/ConfResponse'
+
+ /api/v1/conf/dynamic:
+ get:
+ tags:
+ - Conf
+ operationId: getDynamicConf
+ description: |
+ List the dynamic configs.
+ The parameter level specifies the config level of dynamic configs.
+ The parameter tenant specifies the tenant id of TENANT or TENANT_USER level.
+ The parameter name specifies the user name of TENANT_USER level.
+ Meanwhile, either none or all of the parameter tenant and name are specified for TENANT_USER level.
+ parameters:
+ - name: level
+ in: query
+ description: the config level of dynamic configs.
+ required: false
+ schema:
+ type: string
+ - name: tenant
+ in: query
+ description: the tenant id of TENANT or TENANT_USER level.
+ required: false
+ schema:
+ type: string
+ - name: name
+ in: query
+ description: the user name of TENANT_USER level.
+ required: false
+ schema:
+ type: string
+ responses:
+ "200":
+ description: The request was successful.
+ content:
+ application/json:
+ schema:
+ $ref: '#/components/schemas/DynamicConfigResponse'
+ "503":
+ description: Dynamic configuration is disabled.
+
+ /api/v1/thread_dump:
+ get:
+ operationId: getThreadDump
+ description: List the current thread dump.
+ responses:
+ "200":
+ description: The request was successful.
+ content:
+ application/json:
+ schema:
+ $ref: '#/components/schemas/ThreadStackResponse'
+
+ /api/v1/shuffles:
+ get:
+ tags:
+ - Shuffle
+ operationId: getShuffles
+ description: List all running shuffle keys of the service. It will return all running shuffle's key of the cluster.
+ responses:
+ "200":
+ description: The request was successful.
+ content:
+ application/json:
+ schema:
+ $ref: '#/components/schemas/ShufflesResponse'
+
+ /api/v1/masters:
+ get:
+ tags:
+ - Master
+ operationId: getMasterGroupInfo
+ description: |
+ List master group information of the service.
+ It will list all master's LEADER, FOLLOWER information.
+ responses:
+ "200":
+ description: The request was successful.
+ content:
+ application/json:
+ schema:
+ $ref: '#/components/schemas/MasterInfoResponse'
+ "400":
+ description: HA is not enabled.
+
+ /api/v1/workers:
+ get:
+ tags:
+ - Worker
+ operationId: getWorkers
+ description: |
+ List worker information of the service. It will list all registered workers,
+ lost workers, excluded workers, shutdown workers and decommission workers information.
+ responses:
+ "200":
+ description: The request was successful.
+ content:
+ application/json:
+ schema:
+ $ref: '#/components/schemas/WorkersResponse'
+
+ /api/v1/workers/exclude:
+ post:
+ tags:
+ - Worker
+ operationId: excludeWorker
+ description: |
+ Excluded workers of the master add or remove the worker manually given worker id.
+ The parameter add or remove specifies the excluded workers to add or remove.
+ requestBody:
+ content:
+ application/json:
+ schema:
+ $ref: '#/components/schemas/ExcludeWorkerRequest'
+ responses:
+ "200":
+ description: The request was successful.
+ content:
+ application/json:
+ schema:
+ $ref: '#/components/schemas/HandleResponse'
+
+ /api/v1/workers/events:
+ get:
+ tags:
+ - Worker
+ operationId: getWorkerEvents
+ description: List all worker event infos of the master.
+ responses:
+ "200":
+ description: The request was successful.
+ content:
+ application/json:
+ schema:
+ $ref: '#/components/schemas/WorkerEventsResponse'
+ post:
+ tags:
+ - Worker
+ operationId: sendWorkerEvent
+ description: |
+ For Master(Leader) can send worker event to manager workers.
+ Legal types are 'None', 'Immediately', 'Decommission', 'DecommissionThenIdle', 'Graceful', 'Recommission'.
+ requestBody:
+ content:
+ application/json:
+ schema:
+ $ref: '#/components/schemas/SendWorkerEventRequest'
+ responses:
+ "200":
+ description: The request was successful.
+ content:
+ application/json:
+ schema:
+ $ref: '#/components/schemas/HandleResponse'
+
+ /api/v1/applications:
+ get:
+ tags:
+ - Application
+ operationId: getApplications
+ description: List all running application's ids of the cluster.
+ responses:
+ "200":
+ description: The request was successful.
+ content:
+ application/json:
+ schema:
+ $ref: '#/components/schemas/ApplicationsHeartbeatResponse'
+
+ /api/v1/applications/top_disk_usages:
+ get:
+ tags:
+ - Application
+ operationId: getApplicationsDiskUsageSnapshots
+ description: |
+ List the top disk usage application ids.
+ It will return the top disk usage application ids for the cluster.
+ responses:
+ "200":
+ description: The request was successful.
+ content:
+ application/json:
+ schema:
+ $ref: '#/components/schemas/AppDiskUsageSnapshotsResponse'
+
+ /api/v1/applications/hostnames:
+ get:
+ tags:
+ - Application
+ operationId: getApplicationHostNames
+ description: List all running application's LifecycleManager's hostnames of the cluster.
+ responses:
+ "200":
+ description: The request was successful.
+ content:
+ application/json:
+ schema:
+ $ref: '#/components/schemas/HostnamesResponse'
+
+components:
+ schemas:
+ ConfigData:
+ type: object
+ properties:
+ name:
+ type: string
+ description: The config name.
+ value:
+ type: string
+ description: The config value.
+ required:
+ - name
+ - value
+
+ ConfResponse:
+ type: object
+ properties:
+ configs:
+ type: array
+ items:
+ $ref: '#/components/schemas/ConfigData'
+
+ DynamicConfig:
+ type: object
+ properties:
+ level:
+ type: string
+ description: the config level of dynamic configs.
+ enum:
+ - SYSTEM
+ - TENANT
+ - TENANT_USER
+ desc:
+ type: string
+ description: additional description of the dynamic config, such as tenantId and user identifier.
+ configs:
+ type: array
+ description: the config items.
+ items:
+ $ref: '#/components/schemas/ConfigData'
+
+ DynamicConfigResponse:
+ type: object
+ properties:
+ configs:
+ type: array
+ items:
+ $ref: '#/components/schemas/DynamicConfig'
+
+ ThreadStack:
+ type: object
+ properties:
+ threadId:
+ type: integer
+ format: int64
+ description: The id of the thread.
+ threadName:
+ type: string
+ description: The name of the thread.
+ threadState:
+ type: string
+ description: The state of the thread.
+ stackTrace:
+ type: array
+ description: The stacktrace of the thread.
+ items:
+ type: string
+ blockedByThreadId:
+ type: integer
+ format: int64
+ description: The id of the thread that the current thread is blocked by.
+ blockedByLock:
+ type: string
+ description: The lock that the current thread is blocked by.
+ holdingLocks:
+ type: array
+ description: The locks that the current thread is holding.
+ items:
+ type: string
+ required:
+ - threadId
+ - threadName
+ - threadState
+ - stackTrace
+
+ ThreadStackResponse:
+ type: object
+ properties:
+ threadStacks:
+ type: array
+ description: The thread stacks.
+ items:
+ $ref: '#/components/schemas/ThreadStack'
+ required:
+ - threadStacks
+
+ ShufflesResponse:
+ type: object
+ properties:
+ shuffleIds:
+ type: array
+ description: The shuffle keys.
+ items:
+ type: string
+
+ ApplicationHeartbeatData:
+ type: object
+ properties:
+ appId:
+ type: string
+ description: The id of the application.
+ lastHeartbeatTimestamp:
+ type: integer
+ format: int64
+ description: The last heartbeat timestamp of the application.
+ required:
+ - appId
+ - lastHeartbeatTimestamp
+
+ ApplicationsHeartbeatResponse:
+ type: object
+ properties:
+ applications:
+ type: array
+ description: The applications information.
+ items:
+ $ref: '#/components/schemas/ApplicationHeartbeatData'
+ required:
+ - applications
+
+ HostnamesResponse:
+ type: object
+ properties:
+ hostnames:
+ type: array
+ description: The hostnames of the applications.
+ items:
+ type: string
+ required:
+ - hostnames
+
+ AppDiskUsageData:
+ type: object
+ properties:
+ appId:
+ type: string
+ description: The id of the application.
+ estimatedUsage:
+ type: integer
+ format: int64
+ description: The application disk usage.
+ estimatedUsageStr:
+ type: string
+ description: The application disk usage in string type.
+ required:
+ - appId
+ - estimatedUsage
+ - estimatedUsageStr
+
+
+ AppDiskUsageSnapshotData:
+ type: object
+ properties:
+ start:
+ type: integer
+ format: int64
+ description: The start timestamp of the snapshot.
+ end:
+ type: integer
+ format: int64
+ description: The end timestamp of the snapshot.
+ topNItems:
+ type: array
+ description: The top N app disk usages.
+ items:
+ $ref: '#/components/schemas/AppDiskUsageData'
+
+ AppDiskUsageSnapshotsResponse:
+ type: object
+ properties:
+ snapshots:
+ type: array
+ description: The disk usage snapshots.
+ items:
+ $ref: '#/components/schemas/AppDiskUsageSnapshotData'
+
+ MasterLeader:
+ type: object
+ properties:
+ id:
+ type: string
+ description: The id of the master leader raft peer.
+ address:
+ type: string
+ description: The raft RPC address of the master leader.
+ required:
+ - id
+ - address
+
+ MasterCommitData:
+ type: object
+ properties:
+ commitIndex:
+ type: integer
+ format: int64
+ description: The raft commit index of the master.
+ id:
+ type: string
+ description: The id of the raft peer.
+ address:
+ type: string
+ description: The raft RPC address of the peer for server-server communication.
+ clientAddress:
+ type: string
+ description: The raft RPC address of the peer for client operations.
+ startUpRole:
+ type: string
+ description: The raft start up role of the master.
+ required:
+ - commitIndex
+ - id
+ - address
+ - clientAddress
+ - startUpRole
+
+ MasterInfoResponse:
+ type: object
+ properties:
+ groupId:
+ type: string
+ description: The group id of the master raft server.
+ leader:
+ $ref: '#/components/schemas/MasterLeader'
+ masterCommitInfo:
+ type: array
+ description: The raft commit info of the master.
+ items:
+ $ref: '#/components/schemas/MasterCommitData'
+ required:
+ - groupId
+
+ WorkerData:
+ type: object
+ properties:
+ host:
+ type: string
+ description: The host of the worker.
+ rpcPort:
+ type: integer
+ format: int32
+ description: The rpc port of the worker.
+ pushPort:
+ type: integer
+ format: int32
+ description: The push port of the worker.
+ fetchPort:
+ type: integer
+ format: int32
+ description: The fetch port of the worker.
+ replicatePort:
+ type: integer
+ format: int32
+ description: The replicate port of the worker.
+ internalPort:
+ type: integer
+ format: int32
+ description: The internal port of the worker.
+ slotUsed:
+ type: integer
+ format: int64
+ description: The slot used of the worker.
+ lastHeartbeatTimestamp:
+ type: integer
+ format: int64
+ description: The last heartbeat timestamp of the worker.
+ heartbeatElapsedSeconds:
+ type: integer
+ format: int64
+ description: The elapsed seconds since the last heartbeat of the worker.
+ diskInfos:
+ type: object
+ description: A map of disk name and disk info.
+ additionalProperties:
+ type: string
+ resourceConsumption:
+ type: object
+ description: A map of identifier and resource consumption.
+ additionalProperties:
+ type: string
+ workerRef:
+ type: string
+ description: The reference of the worker.
+ workerState:
+ type: string
+ description: The state of the worker.
+ workerStateStartTime:
+ type: integer
+ format: int64
+ description: The start time of the worker state.
+ required:
+ - host
+ - rpcPort
+ - pushPort
+ - fetchPort
+ - replicatePort
+
+ WorkerTimestampData:
+ type: object
+ properties:
+ worker:
+ $ref: '#/components/schemas/WorkerData'
+ timestamp:
+ type: integer
+ format: int64
+ required:
+ - worker
+ - timestamp
+
+ WorkersResponse:
+ type: object
+ properties:
+ workers:
+ type: array
+ description: The registered workers.
+ items:
+ $ref: '#/components/schemas/WorkerData'
+ lostWorkers:
+ type: array
+ description: The lost workers.
+ items:
+ $ref: '#/components/schemas/WorkerTimestampData'
+ excludedWorkers:
+ type: array
+ description: The excluded workers.
+ items:
+ $ref: '#/components/schemas/WorkerData'
+ manualExcludedWorkers:
+ type: array
+ description: The manual excluded workers.
+ items:
+ $ref: '#/components/schemas/WorkerData'
+ shutdownWorkers:
+ type: array
+ description: The shutdown workers.
+ items:
+ $ref: '#/components/schemas/WorkerData'
+ decommissioningWorkers:
+ type: array
+ description: The decommissioning workers.
+ items:
+ $ref: '#/components/schemas/WorkerData'
+ required:
+ - workers
+
+ WorkerEventInfoData:
+ type: object
+ properties:
+ eventType:
+ type: string
+ description: The type of the event.
+ eventTime:
+ type: integer
+ format: int64
+ description: The start time of the event.
+ required:
+ - eventType
+ - eventTime
+
+ WorkerEventData:
+ type: object
+ properties:
+ worker:
+ $ref: '#/components/schemas/WorkerData'
+ event:
+ $ref: '#/components/schemas/WorkerEventInfoData'
+ required:
+ - worker
+ - event
+
+ WorkerEventsResponse:
+ type: object
+ properties:
+ workerEvents:
+ type: array
+ description: The worker events.
+ items:
+ $ref: '#/components/schemas/WorkerEventData'
+
+ WorkerId:
+ type: object
+ properties:
+ host:
+ type: string
+ description: The host of the worker.
+ rpcPort:
+ type: integer
+ format: int32
+ description: The rpc port of the worker.
+ pushPort:
+ type: integer
+ format: int32
+ description: The push port of the worker.
+ fetchPort:
+ type: integer
+ format: int32
+ description: The fetch port of the worker.
+ replicatePort:
+ type: integer
+ format: int32
+ description: The replicate port of the worker.
+ required:
+ - host
+ - rpcPort
+ - pushPort
+ - fetchPort
+ - replicatePort
+
+ ExcludeWorkerRequest:
+ type: object
+ properties:
+ add:
+ type: array
+ description: The workers to be added to the excluded workers.
+ items:
+ $ref: '#/components/schemas/WorkerId'
+ remove:
+ type: array
+ description: The workers to be removed from the excluded workers.
+ items:
+ $ref: '#/components/schemas/WorkerId'
+
+ SendWorkerEventRequest:
+ type: object
+ properties:
+ eventType:
+ type: string
+ default: None
+ description: The type of the event.
+ enum:
+ - Immediately
+ - Decommission
+ - DecommissionThenIdle
+ - Graceful
+ - Recommission
+ - None
+ workers:
+ type: array
+ description: The workers to send the event.
+ items:
+ $ref: '#/components/schemas/WorkerId'
+ required:
+ - eventType
+ - worker
+
+ HandleResponse:
+ type: object
+ properties:
+ success:
+ type: boolean
+ description: The result of the request.
+ message:
+ type: string
+ description: The message of the request.
+ required:
+ - success
+
+ securitySchemes:
+ basic:
+ type: http
+ scheme: basic
diff --git a/openapi/openapi-client/src/main/openapi3/worker_rest_v1.yaml b/openapi/openapi-client/src/main/openapi3/worker_rest_v1.yaml
new file mode 100644
index 000000000..e854f0a2c
--- /dev/null
+++ b/openapi/openapi-client/src/main/openapi3/worker_rest_v1.yaml
@@ -0,0 +1,627 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+openapi: 3.0.0
+servers:
+ - url: http://localhost:9096/
+ description: Localhost reference server
+security:
+ - basic: []
+info:
+ title: Celeborn REST API
+ license:
+ name: Apache 2.0
+ url: https://www.apache.org/licenses/LICENSE-2.0.html
+ version: 1.0.0
+paths:
+ /api/v1/conf:
+ get:
+ tags:
+ - Conf
+ operationId: getConf
+ description: List the conf setting.
+ responses:
+ "200":
+ description: The request was successful.
+ content:
+ application/json:
+ schema:
+ $ref: '#/components/schemas/ConfResponse'
+
+ /api/v1/conf/dynamic:
+ get:
+ tags:
+ - Conf
+ operationId: getDynamicConf
+ description: |
+ List the dynamic configs.
+ The parameter level specifies the config level of dynamic configs.
+ The parameter tenant specifies the tenant id of TENANT or TENANT_USER level.
+ The parameter name specifies the user name of TENANT_USER level.
+ Meanwhile, either none or all of the parameter tenant and name are specified for TENANT_USER level.
+ parameters:
+ - name: level
+ in: query
+ description: the config level of dynamic configs.
+ required: false
+ schema:
+ type: string
+ - name: tenant
+ in: query
+ description: the tenant id of TENANT or TENANT_USER level.
+ required: false
+ schema:
+ type: string
+ - name: name
+ in: query
+ description: the user name of TENANT_USER level.
+ required: false
+ schema:
+ type: string
+ responses:
+ "200":
+ description: The request was successful.
+ content:
+ application/json:
+ schema:
+ $ref: '#/components/schemas/DynamicConfigResponse'
+ "503":
+ description: Dynamic configuration is disabled.
+
+ /api/v1/thread_dump:
+ get:
+ operationId: getThreadDump
+ description: List the current thread dump.
+ responses:
+ "200":
+ description: The request was successful.
+ content:
+ application/json:
+ schema:
+ $ref: '#/components/schemas/ThreadStackResponse'
+
+ /api/v1/shuffles:
+ get:
+ tags:
+ - Shuffle
+ operationId: getShuffles
+ description: |
+ List all the running shuffle keys of the worker.
+ It only return keys of shuffles running in that worker.
+ responses:
+ "200":
+ description: The request was successful.
+ content:
+ application/json:
+ schema:
+ $ref: '#/components/schemas/ShufflesResponse'
+
+ /api/v1/shuffles/partitions:
+ get:
+ tags:
+ - Shuffle
+ operationId: getShufflePartitions
+ description: |
+ List all the living shuffle PartitionLocation information in the worker.
+ responses:
+ "200":
+ description: The request was successful.
+ content:
+ application/json:
+ schema:
+ $ref: '#/components/schemas/ShufflePartitionsResponse'
+
+ /api/v1/workers:
+ get:
+ tags:
+ - Worker
+ operationId: getWorkerInfo
+ description: List the worker information.
+ responses:
+ "200":
+ description: The request was successful.
+ content:
+ application/json:
+ schema:
+ $ref: '#/components/schemas/WorkerInfoResponse'
+
+ /api/v1/workers/unavailable_peers:
+ get:
+ tags:
+ - Worker
+ operationId: unavailablePeers
+ description: |
+ List the unavailable peers of the worker, this always means the worker connect to the peer failed.
+ responses:
+ "200":
+ description: The request was successful.
+ content:
+ application/json:
+ schema:
+ $ref: '#/components/schemas/UnAvailablePeersResponse'
+
+ /api/v1/workers/exit:
+ post:
+ tags:
+ - Worker
+ operationId: workerExit
+ description: |
+ Trigger this worker to exit.
+ Legal exit types are 'Decommission', 'Graceful' and 'Immediately'.
+ requestBody:
+ content:
+ application/json:
+ schema:
+ $ref: '#/components/schemas/WorkerExitRequest'
+ responses:
+ "200":
+ description: The request was successful.
+ content:
+ application/json:
+ schema:
+ $ref: '#/components/schemas/HandleResponse'
+
+ /api/v1/applications:
+ get:
+ tags:
+ - Application
+ operationId: getApplicationList
+ description: |
+ List all running application's ids of the worker.
+ It only return application ids running in that worker.
+ responses:
+ "200":
+ description: The request was successful.
+ content:
+ application/json:
+ schema:
+ $ref: '#/components/schemas/ApplicationsResponse'
+
+ /api/v1/applications/top_disk_usages:
+ get:
+ tags:
+ - Application
+ operationId: getApplicationsDiskUsage
+ description: |
+ List the top disk usage application ids.
+ It will return the top disk usage application ids for the cluster.
+ responses:
+ "200":
+ description: The request was successful.
+ content:
+ application/json:
+ schema:
+ $ref: '#/components/schemas/AppDiskUsagesResponse'
+
+components:
+ schemas:
+ ConfigData:
+ type: object
+ properties:
+ name:
+ type: string
+ description: The config name.
+ value:
+ type: string
+ description: The config value.
+ required:
+ - name
+ - value
+
+ ConfResponse:
+ type: object
+ properties:
+ configs:
+ type: array
+ items:
+ $ref: '#/components/schemas/ConfigData'
+
+ DynamicConfig:
+ type: object
+ properties:
+ level:
+ type: string
+ description: the config level of dynamic configs.
+ enum:
+ - SYSTEM
+ - TENANT
+ - TENANT_USER
+ desc:
+ type: string
+ description: additional description of the dynamic config, such as tenantId and user identifier.
+ configs:
+ type: array
+ description: the config items.
+ items:
+ $ref: '#/components/schemas/ConfigData'
+
+ DynamicConfigResponse:
+ type: object
+ properties:
+ configs:
+ type: array
+ items:
+ $ref: '#/components/schemas/DynamicConfig'
+
+ ThreadStack:
+ type: object
+ properties:
+ threadId:
+ type: integer
+ format: int64
+ description: The id of the thread.
+ threadName:
+ type: string
+ description: The name of the thread.
+ threadState:
+ type: string
+ description: The state of the thread.
+ stackTrace:
+ type: array
+ description: The stacktrace of the thread.
+ items:
+ type: string
+ blockedByThreadId:
+ type: integer
+ format: int64
+ description: The id of the thread that the current thread is blocked by.
+ blockedByLock:
+ type: string
+ description: The lock that the current thread is blocked by.
+ holdingLocks:
+ type: array
+ description: The locks that the current thread is holding.
+ items:
+ type: string
+ required:
+ - threadId
+ - threadName
+ - threadState
+ - stackTrace
+
+ ThreadStackResponse:
+ type: object
+ properties:
+ threadStacks:
+ type: array
+ description: The thread stacks.
+ items:
+ $ref: '#/components/schemas/ThreadStack'
+ required:
+ - threadStacks
+
+ ShufflesResponse:
+ type: object
+ properties:
+ shuffleIds:
+ type: array
+ description: The shuffle keys.
+ items:
+ type: string
+
+ ApplicationsResponse:
+ type: object
+ properties:
+ applications:
+ type: array
+ description: The applications information.
+ items:
+ type: string
+ required:
+ - applications
+
+ AppDiskUsageData:
+ type: object
+ properties:
+ appId:
+ type: string
+ description: The id of the application.
+ estimatedUsage:
+ type: integer
+ format: int64
+ description: The application disk usage.
+ estimatedUsageStr:
+ type: string
+ description: The application disk usage in string type.
+ required:
+ - appId
+ - estimatedUsage
+ - estimatedUsageStr
+
+ AppDiskUsagesResponse:
+ type: object
+ properties:
+ appDiskUsages:
+ type: array
+ description: The application disk usages.
+ items:
+ $ref: '#/components/schemas/AppDiskUsageData'
+
+ WorkerInfoResponse:
+ type: object
+ properties:
+ host:
+ type: string
+ description: The host of the worker.
+ rpcPort:
+ type: integer
+ format: int32
+ description: The rpc port of the worker.
+ pushPort:
+ type: integer
+ format: int32
+ description: The push port of the worker.
+ fetchPort:
+ type: integer
+ format: int32
+ description: The fetch port of the worker.
+ replicatePort:
+ type: integer
+ format: int32
+ description: The replicate port of the worker.
+ internalPort:
+ type: integer
+ format: int32
+ description: The internal port of the worker.
+ slotUsed:
+ type: integer
+ format: int64
+ description: The slot used of the worker.
+ lastHeartbeatTimestamp:
+ type: integer
+ format: int64
+ description: The last heartbeat timestamp of the worker.
+ heartbeatElapsedSeconds:
+ type: integer
+ format: int64
+ description: The elapsed seconds since the last heartbeat of the worker.
+ diskInfos:
+ type: object
+ description: A map of disk name and disk info.
+ additionalProperties:
+ type: string
+ resourceConsumption:
+ type: object
+ description: A map of identifier and resource consumption.
+ additionalProperties:
+ type: string
+ workerRef:
+ type: string
+ description: The reference of the worker.
+ workerState:
+ type: string
+ description: The state of the worker.
+ workerStateStartTime:
+ type: integer
+ format: int64
+ description: The start time of the worker state.
+ isRegistered:
+ type: boolean
+ description: The registration status of the worker.
+ isShutdown:
+ type: boolean
+ description: The shutdown status of the worker.
+ isDecommissioning:
+ type: boolean
+ description: The decommission status of the worker.
+ required:
+ - host
+ - rpcPort
+ - pushPort
+ - fetchPort
+ - replicatePort
+ - internalPort
+ - isRegistered
+ - isShutdown
+ - isDecommissioning
+
+ WorkerData:
+ type: object
+ properties:
+ host:
+ type: string
+ description: The host of the worker.
+ rpcPort:
+ type: integer
+ format: int32
+ description: The rpc port of the worker.
+ pushPort:
+ type: integer
+ format: int32
+ description: The push port of the worker.
+ fetchPort:
+ type: integer
+ format: int32
+ description: The fetch port of the worker.
+ replicatePort:
+ type: integer
+ format: int32
+ description: The replicate port of the worker.
+ internalPort:
+ type: integer
+ format: int32
+ description: The internal port of the worker.
+ slotUsed:
+ type: integer
+ format: int64
+ description: The slot used of the worker.
+ lastHeartbeatTimestamp:
+ type: integer
+ format: int64
+ description: The last heartbeat timestamp of the worker.
+ heartbeatElapsedSeconds:
+ type: integer
+ format: int64
+ description: The elapsed seconds since the last heartbeat of the worker.
+ diskInfos:
+ type: object
+ description: A map of disk name and disk info.
+ additionalProperties:
+ type: string
+ resourceConsumption:
+ type: object
+ description: A map of identifier and resource consumption.
+ additionalProperties:
+ type: string
+ workerRef:
+ type: string
+ description: The reference of the worker.
+ workerState:
+ type: string
+ description: The state of the worker.
+ workerStateStartTime:
+ type: integer
+ format: int64
+ description: The start time of the worker state.
+ required:
+ - host
+ - rpcPort
+ - pushPort
+ - fetchPort
+ - replicatePort
+
+ WorkerTimestampData:
+ type: object
+ properties:
+ worker:
+ $ref: '#/components/schemas/WorkerData'
+ timestamp:
+ type: integer
+ format: int64
+ required:
+ - worker
+ - timestamp
+
+ UnAvailablePeersResponse:
+ type: object
+ properties:
+ peers:
+ type: array
+ description: The unavailable peers.
+ items:
+ $ref: '#/components/schemas/WorkerTimestampData'
+
+ PartitionLocationData:
+ type: object
+ properties:
+ idEpoch:
+ type: string
+ description: The id and epoch.
+ hostAndPorts:
+ type: string
+ description: The host-rpcPort-pushPort-fetchPort-replicatePort.
+ mode:
+ type: string
+ description: partition mode.
+ enum:
+ - PRIMARY
+ - REPLICA
+ peer:
+ type: string
+ description: The peer address.
+ storage:
+ type: string
+ description: The storage hint.
+ enum:
+ - MEMORY
+ - HDD
+ - SSD
+ - HDFS
+ - OSS
+ mapIdBitMap:
+ type: string
+ description: The map id bitmap hint.
+
+ PartitionLocations:
+ type: object
+ default: {}
+ additionalProperties:
+ $ref: '#/components/schemas/PartitionLocationData'
+
+ ShufflePartitionLocations:
+ type: object
+ default: {}
+ additionalProperties:
+ $ref: '#/components/schemas/PartitionLocations'
+
+ ShufflePartitionsResponse:
+ type: object
+ properties:
+ primaryPartitions:
+ $ref: '#/components/schemas/ShufflePartitionLocations'
+ default: { }
+ replicaPartitions:
+ $ref: '#/components/schemas/ShufflePartitionLocations'
+ default: { }
+
+ WorkerId:
+ type: object
+ properties:
+ host:
+ type: string
+ description: The host of the worker.
+ rpcPort:
+ type: integer
+ format: int32
+ description: The rpc port of the worker.
+ pushPort:
+ type: integer
+ format: int32
+ description: The push port of the worker.
+ fetchPort:
+ type: integer
+ format: int32
+ description: The fetch port of the worker.
+ replicatePort:
+ type: integer
+ format: int32
+ description: The replicate port of the worker.
+ required:
+ - host
+ - rpcPort
+ - pushPort
+ - fetchPort
+ - replicatePort
+
+ WorkerExitRequest:
+ type: object
+ properties:
+ type:
+ type: string
+ default: None
+ description: The type of the worker exit request.
+ enum:
+ - Decommission
+ - Graceful
+ - Immediately
+ - None
+
+ HandleResponse:
+ type: object
+ properties:
+ success:
+ type: boolean
+ description: The result of the request.
+ message:
+ type: string
+ description: The message of the request.
+ required:
+ - success
+
+ securitySchemes:
+ basic:
+ type: http
+ scheme: basic
diff --git a/openapi/openapi-model/pom.xml b/openapi/openapi-model/pom.xml
new file mode 100644
index 000000000..da87d1857
--- /dev/null
+++ b/openapi/openapi-model/pom.xml
@@ -0,0 +1,165 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+ ~ Licensed to the Apache Software Foundation (ASF) under one or more
+ ~ contributor license agreements. See the NOTICE file distributed with
+ ~ this work for additional information regarding copyright ownership.
+ ~ The ASF licenses this file to You under the Apache License, Version 2.0
+ ~ (the "License"); you may not use this file except in compliance with
+ ~ the License. You may obtain a copy of the License at
+ ~
+ ~ http://www.apache.org/licenses/LICENSE-2.0
+ ~
+ ~ Unless required by applicable law or agreed to in writing, software
+ ~ distributed under the License is distributed on an "AS IS" BASIS,
+ ~ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ ~ See the License for the specific language governing permissions and
+ ~ limitations under the License.
+ -->
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+ <modelVersion>4.0.0</modelVersion>
+
+ <parent>
+ <groupId>org.apache.celeborn</groupId>
+ <artifactId>celeborn-parent_${scala.binary.version}</artifactId>
+ <version>${project.version}</version>
+ <relativePath>../../pom.xml</relativePath>
+ </parent>
+
+ <artifactId>celeborn-openapi-model_${scala.binary.version}</artifactId>
+ <packaging>jar</packaging>
+ <name>Celeborn OpenApi Model</name>
+
+ <dependencies>
+ <dependency>
+ <groupId>io.swagger</groupId>
+ <artifactId>swagger-annotations</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>io.swagger</groupId>
+ <artifactId>swagger-models</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.openapitools</groupId>
+ <artifactId>jackson-databind-nullable</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>com.google.code.findbugs</groupId>
+ <artifactId>jsr305</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.glassfish.jersey.media</groupId>
+ <artifactId>jersey-media-json-jackson</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>com.fasterxml.jackson.core</groupId>
+ <artifactId>jackson-annotations</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>com.fasterxml.jackson.core</groupId>
+ <artifactId>jackson-databind</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.glassfish.jersey.media</groupId>
+ <artifactId>jersey-media-multipart</artifactId>
+ </dependency>
+ </dependencies>
+
+ <build>
+ <plugins>
+ <plugin>
+ <groupId>org.openapitools</groupId>
+ <artifactId>openapi-generator-maven-plugin</artifactId>
+ <!--
+ please keep the schema definitions consistent between master_rest_v1.yaml and worker_rest_v1.yaml
+ -->
+ <executions>
+ <execution>
+ <id>generate-java-master-model-rest-v1</id>
+ <goals>
+ <goal>generate</goal>
+ </goals>
+ <phase>generate-sources</phase>
+ <configuration>
+ <inputSpec>${project.basedir}/../openapi-client/src/main/openapi3/master_rest_v1.yaml</inputSpec>
+ <output>${project.build.directory}/generated-sources/java</output>
+ <modelPackage>org.apache.celeborn.rest.v1.model</modelPackage>
+ <generatorName>java</generatorName>
+ <skipValidateSpec>true</skipValidateSpec>
+ <generateApis>false</generateApis>
+ <generateApiTests>false</generateApiTests>
+ <generateModelTests>false</generateModelTests>
+ <generateSupportingFiles>false</generateSupportingFiles>
+ <groupId>${groupId}</groupId>
+ <artifactId>${artifactId}</artifactId>
+ <artifactVersion>${project.version}</artifactVersion>
+ <addCompileSourceRoot>false</addCompileSourceRoot>
+ <cleanupOutput>true</cleanupOutput>
+ <skipValidateSpec>true</skipValidateSpec>
+ <configOptions>
+ <licenseName>Apache License 2.0</licenseName>
+ <groupId>${groupId}</groupId>
+ <artifactId>${artifactId}</artifactId>
+ <artifactVersion>${project.version}</artifactVersion>
+ <dateLibrary>java8</dateLibrary>
+ <useGzipFeature>true</useGzipFeature>
+ <library>jersey2</library>
+ <annotationLibrary>swagger1</annotationLibrary>
+ </configOptions>
+ </configuration>
+ </execution>
+ <execution>
+ <id>generate-java-worker-model-rest-v1</id>
+ <goals>
+ <goal>generate</goal>
+ </goals>
+ <phase>generate-sources</phase>
+ <configuration>
+ <inputSpec>${project.basedir}/../openapi-client/src/main/openapi3/worker_rest_v1.yaml</inputSpec>
+ <output>${project.build.directory}/generated-sources/java</output>
+ <modelPackage>org.apache.celeborn.rest.v1.model</modelPackage>
+ <generatorName>java</generatorName>
+ <skipValidateSpec>true</skipValidateSpec>
+ <generateApis>false</generateApis>
+ <generateApiTests>false</generateApiTests>
+ <generateModelTests>false</generateModelTests>
+ <generateSupportingFiles>false</generateSupportingFiles>
+ <groupId>${groupId}</groupId>
+ <artifactId>${artifactId}</artifactId>
+ <artifactVersion>${project.version}</artifactVersion>
+ <addCompileSourceRoot>false</addCompileSourceRoot>
+ <skipValidateSpec>true</skipValidateSpec>
+ <configOptions>
+ <licenseName>Apache License 2.0</licenseName>
+ <groupId>${groupId}</groupId>
+ <artifactId>${artifactId}</artifactId>
+ <artifactVersion>${project.version}</artifactVersion>
+ <dateLibrary>java8</dateLibrary>
+ <useGzipFeature>true</useGzipFeature>
+ <library>jersey2</library>
+ <annotationLibrary>swagger1</annotationLibrary>
+ </configOptions>
+ </configuration>
+ </execution>
+ </executions>
+ </plugin>
+ <plugin>
+ <groupId>org.codehaus.mojo</groupId>
+ <artifactId>build-helper-maven-plugin</artifactId>
+ <executions>
+ <execution>
+ <id>add-source</id>
+ <goals>
+ <goal>add-source</goal>
+ </goals>
+ <phase>generate-sources</phase>
+ <configuration>
+ <sources>
+ <source>${project.build.directory}/generated-sources/java/src/main/java</source>
+ </sources>
+ </configuration>
+ </execution>
+ </executions>
+ </plugin>
+ </plugins>
+ </build>
+</project>
diff --git a/project/plugins.sbt b/openapi/openapi-model/src/main/java/org/apache/celeborn/rest/v1/JSON.java
similarity index 83%
copy from project/plugins.sbt
copy to openapi/openapi-model/src/main/java/org/apache/celeborn/rest/v1/JSON.java
index 235aefd80..ee1f4bd75 100644
--- a/project/plugins.sbt
+++ b/openapi/openapi-model/src/main/java/org/apache/celeborn/rest/v1/JSON.java
@@ -15,8 +15,7 @@
* limitations under the License.
*/
-addSbtPlugin("com.eed3si9n" % "sbt-assembly" % "2.1.1")
+package org.apache.celeborn.rest.v1;
-addSbtPlugin("com.thesamet" % "sbt-protoc" % "1.0.6")
-
-addSbtPlugin("com.github.sbt" % "sbt-pgp" % "2.2.1")
+// placeholder for JSON java file, referenced by model java files in unused imports
+public class JSON {}
diff --git a/pom.xml b/pom.xml
index 2f4b95eba..e93b7d442 100644
--- a/pom.xml
+++ b/pom.xml
@@ -31,6 +31,8 @@
<name>Celeborn Project Parent POM</name>
<modules>
+ <module>openapi/openapi-client</module>
+ <module>openapi/openapi-model</module>
<module>common</module>
<module>client</module>
<module>service</module>
@@ -116,6 +118,9 @@
<jersey.version>2.39.1</jersey.version>
<jetty.version>9.4.52.v20230823</jetty.version>
<jakarta.servlet-api.version>4.0.4</jakarta.servlet-api.version>
+ <!-- openapi-generator dependencies -->
+ <openapitools.jackson-databind-nullable.version>0.2.6</openapitools.jackson-databind-nullable.version>
+ <openapitools.swagger.version>1.6.11</openapitools.swagger.version>
<!-- org.fusesource.leveldbjni will be used except on arm64 platform. -->
<leveldbjni.group>org.fusesource.leveldbjni</leveldbjni.group>
@@ -138,6 +143,9 @@
<maven.plugin.spotless.version>2.24.1</maven.plugin.spotless.version>
<maven.plugin.surefire.version>3.0.0-M7</maven.plugin.surefire.version>
<maven.plugin.silencer.version>1.7.13</maven.plugin.silencer.version>
+ <maven.plugin.resources.version>3.3.1</maven.plugin.resources.version>
+ <!-- This was the last version to support Java 8 -->
+ <openapi.generator.version>6.6.0</openapi.generator.version>
<!-- Allow modules to enable / disable certain build plugins easily. -->
<testJarPhase>prepare-package</testJarPhase>
@@ -483,6 +491,30 @@
<version>${jakarta.servlet-api.version}</version>
</dependency>
+ <dependency>
+ <groupId>org.openapitools</groupId>
+ <artifactId>jackson-databind-nullable</artifactId>
+ <version>${openapitools.jackson-databind-nullable.version}</version>
+ <exclusions>
+ <exclusion>
+ <groupId>com.fasterxml.jackson.core</groupId>
+ <artifactId>jackson-databind</artifactId>
+ </exclusion>
+ </exclusions>
+ </dependency>
+
+ <dependency>
+ <groupId>io.swagger</groupId>
+ <artifactId>swagger-annotations</artifactId>
+ <version>${openapitools.swagger.version}</version>
+ </dependency>
+
+ <dependency>
+ <groupId>io.swagger</groupId>
+ <artifactId>swagger-models</artifactId>
+ <version>${openapitools.swagger.version}</version>
+ </dependency>
+
<dependency>
<groupId>org.glassfish.jersey.core</groupId>
<artifactId>jersey-server</artifactId>
@@ -1112,6 +1144,11 @@
</annotationProcessorPaths>
</configuration>
</plugin>
+ <plugin>
+ <groupId>org.openapitools</groupId>
+ <artifactId>openapi-generator-maven-plugin</artifactId>
+ <version>${openapi.generator.version}</version>
+ </plugin>
</plugins>
</pluginManagement>
diff --git a/project/CelebornBuild.scala b/project/CelebornBuild.scala
index 1a4b7beeb..00809d8fa 100644
--- a/project/CelebornBuild.scala
+++ b/project/CelebornBuild.scala
@@ -15,7 +15,7 @@
* limitations under the License.
*/
-import java.nio.file.Files
+import java.nio.file.{Files, Path}
import java.nio.charset.StandardCharsets.UTF_8
import java.util.Locale
@@ -23,6 +23,8 @@ import scala.util.Properties
import scala.xml._
import scala.xml.transform._
+import org.openapitools.generator.sbt.plugin.OpenApiGeneratorPlugin
+import org.openapitools.generator.sbt.plugin.OpenApiGeneratorPlugin.autoImport._
import sbtassembly.AssemblyPlugin.autoImport._
import sbtprotoc.ProtocPlugin.autoImport._
@@ -73,6 +75,8 @@ object Dependencies {
val jerseyVersion = "2.39.1"
val jettyVersion = "9.4.52.v20230823"
val jakartaServeletApiVersion = "4.0.4"
+ val openApiToolsJacksonBindNullableVersion = "0.2.6"
+ val openApiToolsSwaggerVersion = "1.6.11"
// For SSL support
val bouncycastleVersion = "1.77"
@@ -139,6 +143,7 @@ object Dependencies {
val jacksonCore = "com.fasterxml.jackson.core" % "jackson-core" % jacksonVersion
val jacksonAnnotations = "com.fasterxml.jackson.core" % "jackson-annotations" % jacksonVersion
val jacksonModule = "com.fasterxml.jackson.module" %% "jackson-module-scala" % jacksonVersion
+ val jacksonDataTypeJsr310 = "com.fasterxml.jackson.datatype" % "jackson-datatype-jsr310" % jacksonVersion
val scalaReflect = "org.scala-lang" % "scala-reflect" % projectScalaVersion
val slf4jApi = "org.slf4j" % "slf4j-api" % slf4jVersion
val slf4jJulToSlf4j = "org.slf4j" % "jul-to-slf4j" % slf4jVersion
@@ -156,6 +161,7 @@ object Dependencies {
val jakartaServletApi = "jakarta.servlet" % "jakarta.servlet-api" % jakartaServeletApiVersion
val jerseyServer = "org.glassfish.jersey.core" % "jersey-server" % jerseyVersion excludeAll(
ExclusionRule("jakarta.xml.bind", "jakarta.xml.bind-api"))
+ val jerseyClient = "org.glassfish.jersey.core" % "jersey-client" % jerseyVersion
val jerseyContainerServletCore = "org.glassfish.jersey.containers" % "jersey-container-servlet-core" % jerseyVersion
val jerseyHk2 = "org.glassfish.jersey.inject" % "jersey-hk2" % jerseyVersion
val jerseyMediaJsonJackson = "org.glassfish.jersey.media" % "jersey-media-json-jackson" % jerseyVersion
@@ -165,6 +171,10 @@ object Dependencies {
ExclusionRule("org.javassist", "javassist"),
ExclusionRule("jakarta.activation", "jakarta.activation-api"))
val swaggerUi = "org.webjars" % "swagger-ui" % swaggerUiVersion
+ val openApiToolsJacksonBindNullable = "org.openapitools" % "jackson-databind-nullable" % openApiToolsJacksonBindNullableVersion excludeAll(
+ ExclusionRule("com.fasterxml.jackson.core", "jackson-databind"))
+ val openApiToolsSwaggerAnnotations = "io.swagger" % "swagger-annotations" % openApiToolsSwaggerVersion
+ val openApiToolsSwaggerModels = "io.swagger" % "swagger-models" % openApiToolsSwaggerVersion
// Test dependencies
// https://www.scala-sbt.org/1.x/docs/Testing.html
@@ -327,6 +337,12 @@ object CelebornCommonSettings {
object CelebornBuild extends sbt.internal.BuildDef {
override def projectDefinitions(baseDirectory: File): Seq[Project] = {
Seq(
+ CelebornOpenApi.openapiInternalMasterClient,
+ CelebornOpenApi.openapiInternalWorkerClient,
+ CelebornOpenApi.openapiClient,
+ CelebornOpenApi.openapiInternalMasterModel,
+ CelebornOpenApi.openapiInternalWorkerModel,
+ CelebornOpenApi.openapiModel,
CelebornCommon.common,
CelebornClient.client,
CelebornService.service,
@@ -503,6 +519,7 @@ object CelebornClient {
object CelebornService {
lazy val service = Project("celeborn-service", file("service"))
.dependsOn(CelebornCommon.common % "test->test;compile->compile")
+ .dependsOn(CelebornOpenApi.openapiModel)
.settings (
commonSettings,
libraryDependencies ++= Seq(
@@ -566,6 +583,7 @@ object CelebornWorker {
.dependsOn(CelebornService.service % "test->test;compile->compile")
.dependsOn(CelebornClient.client % "test->compile")
.dependsOn(CelebornMaster.master % "test->compile")
+ .dependsOn(CelebornOpenApi.openapiClient % Test)
.settings (
commonSettings,
libraryDependencies ++= Seq(
@@ -1231,3 +1249,133 @@ object MRClientProjects {
}
)
}
+
+object CelebornOpenApi {
+ val openApiSpecDir = "openapi/openapi-client/src/main/openapi3"
+ val openApiClientOutputDir = "openapi/openapi-client/target/generated-sources/java"
+ val openApiModelOutputDir = "openapi/openapi-model/target/generated-sources/java"
+
+ def traverseJavaSourceFiles(outputDir: String): Seq[File] = {
+ Files.walk((file(outputDir) / "/src/main/java").toPath, Int.MaxValue)
+ .toArray.map(_.asInstanceOf[Path].toFile)
+ .toSeq.filter(_.isFile)
+ }
+
+ val commonOpenApiClientSettings = Seq(
+ openApiGeneratorName := "java",
+ openApiOutputDir := openApiClientOutputDir,
+ openApiGenerateApiTests := SettingDisabled,
+ openApiGenerateModelTests := SettingDisabled,
+ openApiAdditionalProperties := Map(
+ "dateLibrary" -> "java8",
+ "useGzipFeature" -> "true",
+ "library" -> "jersey2",
+ "annotationLibrary" -> "swagger1"
+ )
+ )
+
+ lazy val openapiInternalMasterClient = Project("celeborn-openapi-master-client", file("openapi/openapi-client/target/master"))
+ .enablePlugins(OpenApiGeneratorPlugin)
+ .settings(
+ commonSettings,
+ openApiInputSpec := (file(openApiSpecDir) / "master_rest_v1.yaml").toString,
+ openApiApiPackage := "org.apache.celeborn.rest.v1.master",
+ openApiModelPackage := "org.apache.celeborn.rest.v1.master.model",
+ openApiInvokerPackage := "org.apache.celeborn.rest.v1.master.invoker",
+ commonOpenApiClientSettings
+ )
+
+ lazy val openapiInternalWorkerClient = Project("celeborn-openapi-worker-client", file("openapi/openapi-client/target/worker"))
+ .enablePlugins(OpenApiGeneratorPlugin)
+ .settings(
+ commonSettings,
+ openApiInputSpec := (file(openApiSpecDir) / "worker_rest_v1.yaml").toString,
+ openApiApiPackage := "org.apache.celeborn.rest.v1.worker",
+ openApiModelPackage := "org.apache.celeborn.rest.v1.worker.model",
+ openApiInvokerPackage := "org.apache.celeborn.rest.v1.worker.invoker",
+ commonOpenApiClientSettings
+ )
+
+ lazy val openapiClient = Project("celeborn-openapi-client", file("openapi/openapi-client"))
+ .enablePlugins(OpenApiGeneratorPlugin)
+ .dependsOn(openapiInternalMasterClient, openapiInternalWorkerClient)
+ .settings(
+ commonSettings,
+ libraryDependencies ++= Seq(
+ Dependencies.openApiToolsSwaggerAnnotations,
+ Dependencies.openApiToolsSwaggerModels,
+ Dependencies.openApiToolsJacksonBindNullable,
+ Dependencies.findbugsJsr305,
+ Dependencies.jerseyClient,
+ Dependencies.jerseyMediaJsonJackson,
+ Dependencies.jacksonAnnotations,
+ Dependencies.jacksonDatabind,
+ Dependencies.jacksonDataTypeJsr310,
+ Dependencies.jerseyMediaMultipart
+ ),
+ Compile / sourceGenerators += Def.task {
+ traverseJavaSourceFiles(openApiClientOutputDir)
+ }.dependsOn(
+ Def.task {
+ println("Cleaning up openapi client output directory: " + openApiClientOutputDir)
+ file(openApiClientOutputDir).delete()
+ },
+ openapiInternalMasterClient / Compile / openApiGenerate,
+ openapiInternalWorkerClient / Compile / openApiGenerate
+ )
+ )
+
+ val commonOpenApiModelSettings = Seq(
+ openApiGeneratorName := "java",
+ openApiOutputDir := openApiModelOutputDir,
+ openApiModelPackage := "org.apache.celeborn.rest.v1.model",
+ openApiGenerateApiTests := SettingDisabled,
+ openApiGenerateModelTests := SettingDisabled,
+ openApiAdditionalProperties := Map("library" -> "jersey2", "annotationLibrary" -> "swagger1"),
+ openApiGlobalProperties := Map("apis" -> "false", "models" -> "", "supportingFiles" -> "false")
+ )
+
+ lazy val openapiInternalMasterModel = Project("celeborn-openapi-master-model", file("openapi/openapi-model/target/master"))
+ .enablePlugins(OpenApiGeneratorPlugin)
+ .settings(
+ commonSettings,
+ openApiInputSpec := (file(openApiSpecDir) / "master_rest_v1.yaml").toString,
+ commonOpenApiModelSettings
+ )
+
+ lazy val openapiInternalWorkerModel = Project("celeborn-openapi-worker-model", file("openapi/openapi-model/target/worker"))
+ .enablePlugins(OpenApiGeneratorPlugin)
+ .settings(
+ commonSettings,
+ openApiInputSpec := (file(openApiSpecDir) / "worker_rest_v1.yaml").toString,
+ commonOpenApiModelSettings
+ )
+
+ lazy val openapiModel = Project("celeborn-openapi-model", file("openapi/openapi-model"))
+ .enablePlugins(OpenApiGeneratorPlugin)
+ .dependsOn(openapiInternalMasterModel, openapiInternalWorkerModel)
+ .settings(
+ commonSettings,
+ libraryDependencies ++= Seq(
+ Dependencies.openApiToolsSwaggerAnnotations,
+ Dependencies.openApiToolsSwaggerModels,
+ Dependencies.openApiToolsJacksonBindNullable,
+ Dependencies.findbugsJsr305,
+ Dependencies.jerseyMediaJsonJackson,
+ Dependencies.jacksonAnnotations,
+ Dependencies.jacksonDatabind,
+ Dependencies.jacksonDataTypeJsr310,
+ Dependencies.jerseyMediaMultipart
+ ),
+ Compile / sourceGenerators += Def.task {
+ traverseJavaSourceFiles(openApiModelOutputDir)
+ }.dependsOn(
+ Def.task {
+ println("Cleaning up openapi model output directory: " + openApiModelOutputDir)
+ file(openApiModelOutputDir).delete()
+ },
+ openapiInternalMasterModel / Compile / openApiGenerate,
+ openapiInternalWorkerModel / Compile / openApiGenerate
+ )
+ )
+}
diff --git a/project/plugins.sbt b/project/plugins.sbt
index 235aefd80..31aaf3812 100644
--- a/project/plugins.sbt
+++ b/project/plugins.sbt
@@ -20,3 +20,6 @@ addSbtPlugin("com.eed3si9n" % "sbt-assembly" % "2.1.1")
addSbtPlugin("com.thesamet" % "sbt-protoc" % "1.0.6")
addSbtPlugin("com.github.sbt" % "sbt-pgp" % "2.2.1")
+
+// This was the last version to support Java 8
+addSbtPlugin("org.openapitools" % "sbt-openapi-generator" % "5.0.1")
diff --git a/service/pom.xml b/service/pom.xml
index 8cc0721a5..16b4fac4e 100644
--- a/service/pom.xml
+++ b/service/pom.xml
@@ -65,6 +65,11 @@
<artifactId>celeborn-common_${scala.binary.version}</artifactId>
<version>${project.version}</version>
</dependency>
+ <dependency>
+ <groupId>org.apache.celeborn</groupId>
+ <artifactId>celeborn-openapi-model_${scala.binary.version}</artifactId>
+ <version>${project.version}</version>
+ </dependency>
<dependency>
<groupId>org.apache.celeborn</groupId>
<artifactId>celeborn-common_${scala.binary.version}</artifactId>
@@ -164,6 +169,11 @@
<artifactId>jetty-proxy</artifactId>
</dependency>
+ <dependency>
+ <groupId>org.openapitools</groupId>
+ <artifactId>jackson-databind-nullable</artifactId>
+ </dependency>
+
<!-- Test dependencies -->
<dependency>
<groupId>org.mockito</groupId>
diff --git a/service/src/main/scala/org/apache/celeborn/server/common/HttpService.scala b/service/src/main/scala/org/apache/celeborn/server/common/HttpService.scala
index 87a621320..613e7d0ea 100644
--- a/service/src/main/scala/org/apache/celeborn/server/common/HttpService.scala
+++ b/service/src/main/scala/org/apache/celeborn/server/common/HttpService.scala
@@ -26,6 +26,7 @@ import org.eclipse.jetty.servlet.FilterHolder
import org.apache.celeborn.common.CelebornConf
import org.apache.celeborn.common.internal.Logging
+import org.apache.celeborn.common.meta.WorkerInfo
import org.apache.celeborn.common.util.Utils
import org.apache.celeborn.server.common.http.HttpServer
import org.apache.celeborn.server.common.http.api.ApiRootResource
@@ -33,6 +34,7 @@ import org.apache.celeborn.server.common.http.authentication.{AuthenticationFilt
import org.apache.celeborn.server.common.service.config.ConfigLevel
abstract class HttpService extends Service with Logging {
+ type HandleResponse = (Boolean, String)
private var httpServer: HttpServer = _
@@ -169,7 +171,7 @@ abstract class HttpService extends Service with Logging {
def getHostnameList: String = throw new UnsupportedOperationException()
- def exclude(addWorkers: String, removeWorkers: String): String =
+ def exclude(addWorkers: Seq[WorkerInfo], removeWorkers: Seq[WorkerInfo]): HandleResponse =
throw new UnsupportedOperationException()
def listPartitionLocationInfo: String = throw new UnsupportedOperationException()
@@ -184,7 +186,7 @@ abstract class HttpService extends Service with Logging {
def exit(exitType: String): String = throw new UnsupportedOperationException()
- def handleWorkerEvent(workerEventType: String, workers: String): String =
+ def handleWorkerEvent(workerEventType: String, workers: Seq[WorkerInfo]): HandleResponse =
throw new UnsupportedOperationException()
def getWorkerEventInfo(): String = throw new UnsupportedOperationException()
diff --git a/service/src/main/scala/org/apache/celeborn/server/common/http/api/ApiBaseResource.scala b/service/src/main/scala/org/apache/celeborn/server/common/http/api/ApiBaseResource.scala
index a945c284f..b9dbecbfe 100644
--- a/service/src/main/scala/org/apache/celeborn/server/common/http/api/ApiBaseResource.scala
+++ b/service/src/main/scala/org/apache/celeborn/server/common/http/api/ApiBaseResource.scala
@@ -22,7 +22,9 @@ import javax.ws.rs.core.MediaType
import io.swagger.v3.oas.annotations.media.Content
import io.swagger.v3.oas.annotations.responses.ApiResponse
+import io.swagger.v3.oas.annotations.tags.Tag
+@Tag(name = "Deprecated")
@Path("/")
private[api] class ApiBaseResource extends ApiRequestContext {
def service: String = httpService.serviceName
diff --git a/service/src/main/scala/org/apache/celeborn/server/common/http/api/ApiRootResource.scala b/service/src/main/scala/org/apache/celeborn/server/common/http/api/ApiRootResource.scala
index 6d69c1b58..acdf6dc33 100644
--- a/service/src/main/scala/org/apache/celeborn/server/common/http/api/ApiRootResource.scala
+++ b/service/src/main/scala/org/apache/celeborn/server/common/http/api/ApiRootResource.scala
@@ -25,7 +25,7 @@ import org.apache.celeborn.server.common.HttpService
private[celeborn] object ApiRootResource {
def getServletHandler(rs: HttpService): ServletContextHandler = {
- val openapiConf: ResourceConfig = new OpenAPIConfig
+ val openapiConf: ResourceConfig = new OpenAPIConfig(rs.serviceName)
val holder = new ServletHolder(new ServletContainer(openapiConf))
val handler = new ServletContextHandler(ServletContextHandler.NO_SESSIONS)
handler.setContextPath("/")
diff --git a/service/src/main/scala/org/apache/celeborn/server/common/http/api/CelebornOpenApiResource.scala b/service/src/main/scala/org/apache/celeborn/server/common/http/api/CelebornOpenApiResource.scala
index 5559ed90d..5ef00627c 100644
--- a/service/src/main/scala/org/apache/celeborn/server/common/http/api/CelebornOpenApiResource.scala
+++ b/service/src/main/scala/org/apache/celeborn/server/common/http/api/CelebornOpenApiResource.scala
@@ -53,7 +53,7 @@ class CelebornOpenApiResource extends BaseOpenApiResource with ApiRequestContext
val ctx: OpenApiContext = new CelebornJaxrsOpenApiContextBuilder()
.servletConfig(config)
.application(app)
- .resourcePackages(OpenAPIConfig.packages.toSet.asJava)
+ .resourcePackages(OpenAPIConfig.packages(httpService.serviceName).toSet.asJava)
.configLocation(configLocation)
.openApiConfiguration(openApiConfiguration)
.ctxId(ctxId)
diff --git a/service/src/main/scala/org/apache/celeborn/server/common/http/api/OpenAPIConfig.scala b/service/src/main/scala/org/apache/celeborn/server/common/http/api/OpenAPIConfig.scala
index b9355ac00..feda12be5 100644
--- a/service/src/main/scala/org/apache/celeborn/server/common/http/api/OpenAPIConfig.scala
+++ b/service/src/main/scala/org/apache/celeborn/server/common/http/api/OpenAPIConfig.scala
@@ -19,16 +19,21 @@ package org.apache.celeborn.server.common.http.api
import org.glassfish.jersey.server.ResourceConfig
-class OpenAPIConfig extends ResourceConfig {
- packages(OpenAPIConfig.packages: _*)
+import org.apache.celeborn.server.common.Service
+
+class OpenAPIConfig(serviceName: String) extends ResourceConfig {
+ packages(OpenAPIConfig.packages(serviceName): _*)
register(classOf[CelebornOpenApiResource])
register(classOf[CelebornScalaObjectMapper])
register(classOf[RestExceptionMapper])
}
object OpenAPIConfig {
- val packages = Seq(
- "org.apache.celeborn.server.common.http.api",
- "org.apache.celeborn.service.deploy.master.http.api",
- "org.apache.celeborn.service.deploy.worker.http.api")
+ val packages = Map(
+ Service.MASTER -> Seq(
+ "org.apache.celeborn.server.common.http.api",
+ "org.apache.celeborn.service.deploy.master.http.api"),
+ Service.WORKER -> Seq(
+ "org.apache.celeborn.server.common.http.api",
+ "org.apache.celeborn.service.deploy.worker.http.api"))
}
diff --git a/service/src/main/scala/org/apache/celeborn/server/common/http/api/v1/ApiUtils.scala b/service/src/main/scala/org/apache/celeborn/server/common/http/api/v1/ApiUtils.scala
new file mode 100644
index 000000000..0b2802035
--- /dev/null
+++ b/service/src/main/scala/org/apache/celeborn/server/common/http/api/v1/ApiUtils.scala
@@ -0,0 +1,131 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.celeborn.server.common.http.api.v1
+
+import java.util.concurrent.TimeUnit
+
+import scala.collection.JavaConverters._
+
+import org.apache.celeborn.common.meta.{WorkerInfo, WorkerStatus}
+import org.apache.celeborn.common.protocol.{PartitionLocation, StorageInfo}
+import org.apache.celeborn.common.protocol.PbWorkerStatus.State
+import org.apache.celeborn.rest.v1.model._
+import org.apache.celeborn.rest.v1.model.PartitionLocationData.{ModeEnum, StorageEnum}
+
+object ApiUtils {
+ def workerData(workerInfo: WorkerInfo): WorkerData = {
+ val (diskInfos, slots) =
+ if (workerInfo.diskInfos == null) {
+ Map.empty[String, String] -> 0L
+ } else {
+ workerInfo.diskInfos.asScala.map { case (disk, diskInfo) =>
+ disk -> diskInfo.toString()
+ }.toMap -> workerInfo.usedSlots()
+ }
+ val userResourceConsumption =
+ if (workerInfo.userResourceConsumption == null) {
+ Map.empty[String, String]
+ } else {
+ workerInfo.userResourceConsumption.asScala.map { case (user, resourceConsumption) =>
+ user.toString -> resourceConsumption.toString()
+ }.toMap
+ }
+
+ new WorkerData()
+ .host(workerInfo.host)
+ .rpcPort(workerInfo.rpcPort)
+ .pushPort(workerInfo.pushPort)
+ .fetchPort(workerInfo.fetchPort)
+ .replicatePort(workerInfo.replicatePort)
+ .internalPort(workerInfo.internalPort)
+ .slotUsed(slots)
+ .lastHeartbeatTimestamp(workerInfo.lastHeartbeat)
+ .heartbeatElapsedSeconds(TimeUnit.MILLISECONDS.toSeconds(
+ System.currentTimeMillis() - workerInfo.lastHeartbeat))
+ .diskInfos(diskInfos.asJava)
+ .resourceConsumption(userResourceConsumption.asJava)
+ .workerRef(Option(workerInfo.endpoint).map(_.toString).orNull)
+ .workerState(workerInfo.workerStatus.getState.toString)
+ .workerStateStartTime(workerInfo.workerStatus.getStateStartTime)
+ }
+
+ def workerInfoResponse(
+ workerInfo: WorkerInfo,
+ currentStatus: WorkerStatus,
+ isShutdown: Boolean,
+ isRegistered: Boolean): WorkerInfoResponse = {
+ val workerData = ApiUtils.workerData(workerInfo)
+ new WorkerInfoResponse()
+ .host(workerData.getHost)
+ .rpcPort(workerData.getRpcPort)
+ .pushPort(workerData.getPushPort)
+ .fetchPort(workerData.getFetchPort)
+ .replicatePort(workerData.getReplicatePort)
+ .internalPort(workerData.getInternalPort)
+ .slotUsed(workerData.getSlotUsed)
+ .lastHeartbeatTimestamp(workerData.getLastHeartbeatTimestamp)
+ .heartbeatElapsedSeconds(workerData.getHeartbeatElapsedSeconds)
+ .diskInfos(workerData.getDiskInfos)
+ .resourceConsumption(workerData.getResourceConsumption)
+ .workerRef(workerData.getWorkerRef)
+ .workerState(currentStatus.getState.toString)
+ .workerStateStartTime(currentStatus.getStateStartTime)
+ .isShutdown(isShutdown)
+ .isRegistered(isRegistered)
+ .isDecommissioning(
+ isShutdown && (
+ currentStatus.getState == State.InDecommission ||
+ currentStatus.getState == State.InDecommissionThenIdle))
+ }
+
+ def toWorkerInfo(workerId: WorkerId): WorkerInfo = {
+ new WorkerInfo(
+ workerId.getHost,
+ workerId.getRpcPort,
+ workerId.getPushPort,
+ workerId.getFetchPort,
+ workerId.getReplicatePort)
+ }
+
+ def partitionLocationData(partitionLocation: PartitionLocation): PartitionLocationData = {
+ val locationData = new PartitionLocationData()
+ .idEpoch(partitionLocation.getId + "-" + partitionLocation.getEpoch)
+ .hostAndPorts(partitionLocation.hostAndPorts())
+ partitionLocation.getMode match {
+ case PartitionLocation.Mode.PRIMARY =>
+ locationData.mode(ModeEnum.PRIMARY)
+ case PartitionLocation.Mode.REPLICA =>
+ locationData.mode(ModeEnum.REPLICA)
+ }
+ Option(partitionLocation.getPeer).map(_.hostAndPorts).foreach(locationData.peer)
+ partitionLocation.getStorageInfo.getType match {
+ case StorageInfo.Type.MEMORY =>
+ locationData.storage(StorageEnum.MEMORY)
+ case StorageInfo.Type.HDD =>
+ locationData.storage(StorageEnum.HDD)
+ case StorageInfo.Type.SSD =>
+ locationData.storage(StorageEnum.SSD)
+ case StorageInfo.Type.HDFS =>
+ locationData.storage(StorageEnum.HDFS)
+ case StorageInfo.Type.OSS =>
+ locationData.storage(StorageEnum.OSS)
+ }
+ Option(partitionLocation.getMapIdBitMap).map(_.toString).foreach(locationData.mapIdBitMap)
+ locationData
+ }
+}
diff --git a/service/src/main/scala/org/apache/celeborn/server/common/http/api/v1/ApiV1BaseResource.scala b/service/src/main/scala/org/apache/celeborn/server/common/http/api/v1/ApiV1BaseResource.scala
new file mode 100644
index 000000000..b0d451403
--- /dev/null
+++ b/service/src/main/scala/org/apache/celeborn/server/common/http/api/v1/ApiV1BaseResource.scala
@@ -0,0 +1,59 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.celeborn.server.common.http.api.v1
+
+import javax.ws.rs.{GET, Path}
+import javax.ws.rs.core.MediaType
+
+import scala.collection.JavaConverters._
+
+import io.swagger.v3.oas.annotations.media.{Content, Schema}
+import io.swagger.v3.oas.annotations.responses.ApiResponse
+
+import org.apache.celeborn.common.util.Utils
+import org.apache.celeborn.rest.v1.model.{ThreadStack, ThreadStackResponse}
+import org.apache.celeborn.server.common.http.api.ApiRequestContext
+
+@Path("/api/v1")
+class ApiV1BaseResource extends ApiRequestContext {
+ @Path("conf")
+ def conf: Class[ConfResource] = classOf[ConfResource]
+
+ @Path("/thread_dump")
+ @ApiResponse(
+ responseCode = "200",
+ content = Array(new Content(
+ mediaType = MediaType.APPLICATION_JSON,
+ schema = new Schema(
+ implementation = classOf[ThreadStackResponse]))),
+ description = "List the current thread dump.")
+ @GET
+ def threadDump(): ThreadStackResponse = {
+ new ThreadStackResponse()
+ .threadStacks(Utils.getThreadDump().map { threadStack =>
+ new ThreadStack()
+ .threadId(threadStack.threadId)
+ .threadName(threadStack.threadName)
+ .threadState(threadStack.threadState.toString)
+ .stackTrace(threadStack.stackTrace.elems.asJava)
+ .blockedByThreadId(threadStack.blockedByThreadId.getOrElse(null.asInstanceOf[Long]): Long)
+ .blockedByLock(threadStack.blockedByLock)
+ .holdingLocks(threadStack.holdingLocks.asJava)
+ }.asJava)
+ }
+}
diff --git a/service/src/main/scala/org/apache/celeborn/server/common/http/api/v1/ConfResource.scala b/service/src/main/scala/org/apache/celeborn/server/common/http/api/v1/ConfResource.scala
new file mode 100644
index 000000000..01b60c688
--- /dev/null
+++ b/service/src/main/scala/org/apache/celeborn/server/common/http/api/v1/ConfResource.scala
@@ -0,0 +1,135 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.celeborn.server.common.http.api.v1
+
+import javax.ws.rs.{Consumes, GET, Path, Produces, QueryParam, ServiceUnavailableException}
+import javax.ws.rs.core.MediaType
+
+import scala.collection.JavaConverters._
+
+import io.swagger.v3.oas.annotations.media.{Content, Schema}
+import io.swagger.v3.oas.annotations.responses.ApiResponse
+import io.swagger.v3.oas.annotations.tags.Tag
+import org.apache.commons.lang3.StringUtils
+
+import org.apache.celeborn.common.CelebornConf
+import org.apache.celeborn.common.util.Utils
+import org.apache.celeborn.rest.v1.model.{ConfigData, ConfResponse, DynamicConfig, DynamicConfigResponse}
+import org.apache.celeborn.rest.v1.model.DynamicConfig.LevelEnum
+import org.apache.celeborn.server.common.http.api.ApiRequestContext
+import org.apache.celeborn.server.common.service.config.ConfigLevel
+
+@Tag(name = "Conf")
+@Produces(Array(MediaType.APPLICATION_JSON))
+@Consumes(Array(MediaType.APPLICATION_JSON))
+private[api] class ConfResource extends ApiRequestContext {
+ private def configService = httpService.configService
+
+ @ApiResponse(
+ responseCode = "200",
+ content = Array(new Content(
+ mediaType = MediaType.APPLICATION_JSON,
+ schema = new Schema(implementation = classOf[ConfResponse]))),
+ description = "List the conf setting")
+ @GET
+ def conf: ConfResponse = {
+ new ConfResponse()
+ .configs(
+ Utils.redact(httpService.conf, httpService.conf.getAll).sortBy(_._1).map { case (n, v) =>
+ new ConfigData().name(n).value(v)
+ }.asJava)
+ }
+
+ @ApiResponse(
+ responseCode = "200",
+ content = Array(new Content(
+ mediaType = MediaType.APPLICATION_JSON,
+ schema = new Schema(implementation = classOf[DynamicConfigResponse]))),
+ description = "List the dynamic configs. " +
+ "The parameter level specifies the config level of dynamic configs. " +
+ "The parameter tenant specifies the tenant id of TENANT or TENANT_USER level. " +
+ "The parameter name specifies the user name of TENANT_USER level. " +
+ "Meanwhile, either none or all of the parameter tenant and name are specified for TENANT_USER level.")
+ @Path("/dynamic")
+ @GET
+ def dynamicConf(
+ @QueryParam("level") level: String,
+ @QueryParam("tenant") tenant: String,
+ @QueryParam("name") name: String): DynamicConfigResponse = {
+ if (configService == null) {
+ throw new ServiceUnavailableException(
+ s"Dynamic configuration is disabled. Please check whether to config" +
+ s" `${CelebornConf.DYNAMIC_CONFIG_STORE_BACKEND.key}`.")
+ } else {
+ if (StringUtils.isEmpty(level)) {
+ new DynamicConfigResponse()
+ .configs(ConfigLevel.values().flatMap { configLevel =>
+ getDynamicConfig(configLevel.name(), tenant, name)
+ }.toSeq.asJava)
+ } else {
+ new DynamicConfigResponse()
+ .configs(getDynamicConfig(level, tenant, name).asJava)
+ }
+ }
+ }
+
+ private def getDynamicConfig(level: String, tenant: String, name: String): Seq[DynamicConfig] = {
+ if (ConfigLevel.SYSTEM.name().equalsIgnoreCase(level)) {
+ val config = configService.getSystemConfigFromCache.getConfigs.asScala
+ Seq(new DynamicConfig()
+ .level(LevelEnum.SYSTEM)
+ .configs(config.toSeq.sortBy(_._1).map { case (n, v) =>
+ new ConfigData().name(n).value(v)
+ }.asJava))
+ } else if (ConfigLevel.TENANT.name().equalsIgnoreCase(level)) {
+ val tenantConfigs =
+ if (StringUtils.isEmpty(tenant)) {
+ configService.listRawTenantConfigsFromCache().asScala
+ } else {
+ List(configService.getRawTenantConfigFromCache(tenant))
+ }
+ tenantConfigs.sortBy(_.getTenantId).map { tenantConfig =>
+ new DynamicConfig()
+ .level(LevelEnum.TENANT)
+ .desc(s"Tenant: ${tenantConfig.getTenantId}")
+ .configs(tenantConfig.getConfigs.asScala.toSeq.sortBy(_._1).map { case (n, v) =>
+ new ConfigData().name(n).value(v)
+ }.asJava)
+ }.toSeq
+ } else if (ConfigLevel.TENANT_USER.name().equalsIgnoreCase(level)) {
+ val tenantUserConfigs =
+ if (StringUtils.isEmpty(tenant) && StringUtils.isEmpty(name)) {
+ configService.listRawTenantUserConfigsFromCache().asScala
+ } else if (tenant.nonEmpty && name.nonEmpty) {
+ List(configService.getRawTenantUserConfigFromCache(tenant, name))
+ } else {
+ List()
+ }
+ tenantUserConfigs.sortBy(_.getTenantId).map { tenantUserConfig =>
+ new DynamicConfig()
+ .level(LevelEnum.TENANT_USER)
+ .desc(s"Tenant: ${tenantUserConfig.getTenantId}, User: ${tenantUserConfig.getName}")
+ .configs(tenantUserConfig.getConfigs.asScala.toSeq.sortBy(_._1).map { case (n, v) =>
+ new ConfigData().name(n).value(v)
+ }.asJava)
+ }.toSeq
+ } else {
+ Seq.empty[DynamicConfig]
+ }
+ }
+}
diff --git a/service/src/test/scala/org/apache/celeborn/server/common/http/api/v1/ApiV1BaseResourceSuite.scala b/service/src/test/scala/org/apache/celeborn/server/common/http/api/v1/ApiV1BaseResourceSuite.scala
new file mode 100644
index 000000000..1f4a9a19a
--- /dev/null
+++ b/service/src/test/scala/org/apache/celeborn/server/common/http/api/v1/ApiV1BaseResourceSuite.scala
@@ -0,0 +1,46 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.celeborn.server.common.http.api.v1
+
+import java.net.URI
+import javax.servlet.http.HttpServletResponse
+import javax.ws.rs.core.{MediaType, UriBuilder}
+
+import org.apache.celeborn.rest.v1.model.{ConfResponse, ThreadStackResponse}
+import org.apache.celeborn.server.common.http.HttpTestHelper
+
+abstract class ApiV1BaseResourceSuite extends HttpTestHelper {
+ override protected lazy val baseUri: URI =
+ UriBuilder.fromUri(s"http://${httpService.connectionUrl}/api/v1").build()
+
+ test("conf resource") {
+ var response = webTarget.path("conf").request(MediaType.APPLICATION_JSON).get()
+ assert(HttpServletResponse.SC_OK == response.getStatus)
+ assert(!response.readEntity(classOf[ConfResponse]).getConfigs.isEmpty)
+
+ response = webTarget.path("conf/dynamic").request(MediaType.APPLICATION_JSON).get()
+ assert(HttpServletResponse.SC_SERVICE_UNAVAILABLE == response.getStatus)
+ assert(response.readEntity(classOf[String]).contains("Dynamic configuration is disabled."))
+ }
+
+ test("thread_dump") {
+ val response = webTarget.path("thread_dump").request(MediaType.APPLICATION_JSON).get()
+ assert(HttpServletResponse.SC_OK == response.getStatus)
+ assert(!response.readEntity(classOf[ThreadStackResponse]).getThreadStacks.isEmpty)
+ }
+}
diff --git a/worker/pom.xml b/worker/pom.xml
index 31c624e3b..2f9b3280c 100644
--- a/worker/pom.xml
+++ b/worker/pom.xml
@@ -38,6 +38,12 @@
<artifactId>celeborn-service_${scala.binary.version}</artifactId>
<version>${project.version}</version>
</dependency>
+ <dependency>
+ <groupId>org.apache.celeborn</groupId>
+ <artifactId>celeborn-openapi-client_${scala.binary.version}</artifactId>
+ <version>${project.version}</version>
+ <scope>test</scope>
+ </dependency>
<dependency>
<groupId>${leveldbjni.group}</groupId>
<artifactId>leveldbjni-all</artifactId>
diff --git a/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/http/api/ApiWorkerResource.scala b/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/http/api/ApiWorkerResource.scala
index 7dff9348c..d6c64f68f 100644
--- a/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/http/api/ApiWorkerResource.scala
+++ b/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/http/api/ApiWorkerResource.scala
@@ -22,9 +22,11 @@ import javax.ws.rs.core.MediaType
import io.swagger.v3.oas.annotations.media.Content
import io.swagger.v3.oas.annotations.responses.ApiResponse
+import io.swagger.v3.oas.annotations.tags.Tag
import org.apache.celeborn.server.common.http.api.ApiRequestContext
+@Tag(name = "Deprecated")
@Path("/")
class ApiWorkerResource extends ApiRequestContext {
diff --git a/project/plugins.sbt b/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/http/api/v1/ApiV1WorkerResource.scala
similarity index 61%
copy from project/plugins.sbt
copy to worker/src/main/scala/org/apache/celeborn/service/deploy/worker/http/api/v1/ApiV1WorkerResource.scala
index 235aefd80..6721a5b74 100644
--- a/project/plugins.sbt
+++ b/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/http/api/v1/ApiV1WorkerResource.scala
@@ -15,8 +15,20 @@
* limitations under the License.
*/
-addSbtPlugin("com.eed3si9n" % "sbt-assembly" % "2.1.1")
+package org.apache.celeborn.service.deploy.worker.http.api.v1
-addSbtPlugin("com.thesamet" % "sbt-protoc" % "1.0.6")
+import javax.ws.rs.Path
-addSbtPlugin("com.github.sbt" % "sbt-pgp" % "2.2.1")
+import org.apache.celeborn.server.common.http.api.ApiRequestContext
+
+@Path("/api/v1")
+class ApiV1WorkerResource extends ApiRequestContext {
+ @Path("shuffles")
+ def shuffles: Class[ShuffleResource] = classOf[ShuffleResource]
+
+ @Path("applications")
+ def applications: Class[ApplicationResource] = classOf[ApplicationResource]
+
+ @Path("workers")
+ def workers: Class[WorkerResource] = classOf[WorkerResource]
+}
diff --git a/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/http/api/v1/ApplicationResource.scala b/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/http/api/v1/ApplicationResource.scala
new file mode 100644
index 000000000..06fa9faea
--- /dev/null
+++ b/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/http/api/v1/ApplicationResource.scala
@@ -0,0 +1,74 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.celeborn.service.deploy.worker.http.api.v1
+
+import javax.ws.rs.{Consumes, GET, Path, Produces}
+import javax.ws.rs.core.MediaType
+
+import scala.collection.JavaConverters._
+
+import io.swagger.v3.oas.annotations.media.{Content, Schema}
+import io.swagger.v3.oas.annotations.responses.ApiResponse
+import io.swagger.v3.oas.annotations.tags.Tag
+
+import org.apache.celeborn.common.util.Utils
+import org.apache.celeborn.rest.v1.model.{AppDiskUsageData, AppDiskUsagesResponse, ApplicationsResponse}
+import org.apache.celeborn.server.common.http.api.ApiRequestContext
+import org.apache.celeborn.service.deploy.worker.Worker
+
+@Tag(name = "Application")
+@Produces(Array(MediaType.APPLICATION_JSON))
+@Consumes(Array(MediaType.APPLICATION_JSON))
+class ApplicationResource extends ApiRequestContext {
+ private def worker = httpService.asInstanceOf[Worker]
+ private def storageManager = worker.storageManager
+
+ @ApiResponse(
+ responseCode = "200",
+ content = Array(new Content(
+ mediaType = MediaType.APPLICATION_JSON,
+ schema = new Schema(implementation = classOf[ApplicationsResponse]))),
+ description =
+ "List all running application's ids of the worker. It only return application ids running in that worker.")
+ @GET
+ def applications(): ApplicationsResponse = {
+ new ApplicationsResponse()
+ .applications(worker.workerInfo.getApplicationIdSet.asScala.toSeq.asJava)
+ }
+
+ @ApiResponse(
+ responseCode = "200",
+ content = Array(new Content(
+ mediaType = MediaType.APPLICATION_JSON,
+ schema = new Schema(implementation = classOf[AppDiskUsagesResponse]))),
+ description =
+ "List the top disk usage application ids. It will return the top disk usage application ids for the cluster.")
+ @Path("/top_disk_usages")
+ @GET
+ def topDiskUsedApplications(): AppDiskUsagesResponse = {
+ new AppDiskUsagesResponse()
+ .appDiskUsages(
+ storageManager.topAppDiskUsage.asScala.map { case (appId, diskUsage) =>
+ new AppDiskUsageData()
+ .appId(appId)
+ .estimatedUsage(diskUsage)
+ .estimatedUsageStr(Utils.bytesToString(diskUsage))
+ }.toSeq.asJava)
+
+ }
+}
diff --git a/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/http/api/v1/ShuffleResource.scala b/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/http/api/v1/ShuffleResource.scala
new file mode 100644
index 000000000..a62cf745c
--- /dev/null
+++ b/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/http/api/v1/ShuffleResource.scala
@@ -0,0 +1,78 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.celeborn.service.deploy.worker.http.api.v1
+
+import javax.ws.rs.{Consumes, GET, Path, Produces}
+import javax.ws.rs.core.MediaType
+
+import scala.collection.JavaConverters._
+
+import io.swagger.v3.oas.annotations.media.{Content, Schema}
+import io.swagger.v3.oas.annotations.responses.ApiResponse
+import io.swagger.v3.oas.annotations.tags.Tag
+
+import org.apache.celeborn.rest.v1.model.{ShufflePartitionsResponse, ShufflesResponse}
+import org.apache.celeborn.server.common.http.api.ApiRequestContext
+import org.apache.celeborn.server.common.http.api.v1.ApiUtils
+import org.apache.celeborn.service.deploy.worker.Worker
+
+@Tag(name = "Shuffle")
+@Produces(Array(MediaType.APPLICATION_JSON))
+@Consumes(Array(MediaType.APPLICATION_JSON))
+class ShuffleResource extends ApiRequestContext {
+ private def worker = httpService.asInstanceOf[Worker]
+
+ @ApiResponse(
+ responseCode = "200",
+ content = Array(new Content(
+ mediaType = MediaType.APPLICATION_JSON,
+ schema = new Schema(implementation = classOf[ShufflesResponse]))),
+ description =
+ "List all the running shuffle keys of the worker. It only return keys of shuffles running in that worker.")
+ @GET
+ def shuffles(): ShufflesResponse = {
+ new ShufflesResponse()
+ .shuffleIds(worker.storageManager.shuffleKeySet().asScala.toSeq.asJava)
+
+ }
+
+ @ApiResponse(
+ responseCode = "200",
+ content = Array(new Content(
+ mediaType = MediaType.APPLICATION_JSON,
+ schema = new Schema(
+ implementation = classOf[ShufflePartitionsResponse]))),
+ description = "List all the living shuffle PartitionLocation information in the worker.")
+ @Path("/partitions")
+ @GET
+ def partitions(): ShufflePartitionsResponse = {
+ new ShufflePartitionsResponse()
+ .primaryPartitions(
+ worker.partitionLocationInfo.primaryPartitionLocations.asScala.map { case (k, v) =>
+ k -> v.asScala.map { case (id, location) =>
+ id -> ApiUtils.partitionLocationData(location)
+ }.asJava
+ }.asJava)
+ .replicaPartitions(
+ worker.partitionLocationInfo.replicaPartitionLocations.asScala.map { case (k, v) =>
+ k -> v.asScala.map { case (id, location) =>
+ id -> ApiUtils.partitionLocationData(location)
+ }.asJava
+ }.asJava)
+ }
+}
diff --git a/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/http/api/v1/WorkerResource.scala b/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/http/api/v1/WorkerResource.scala
new file mode 100644
index 000000000..31229c6b3
--- /dev/null
+++ b/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/http/api/v1/WorkerResource.scala
@@ -0,0 +1,88 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.celeborn.service.deploy.worker.http.api.v1
+
+import javax.ws.rs.{Consumes, GET, Path, POST, Produces}
+import javax.ws.rs.core.MediaType
+
+import scala.collection.JavaConverters._
+
+import io.swagger.v3.oas.annotations.media.{Content, Schema}
+import io.swagger.v3.oas.annotations.responses.ApiResponse
+import io.swagger.v3.oas.annotations.tags.Tag
+
+import org.apache.celeborn.rest.v1.model.{HandleResponse, UnAvailablePeersResponse, WorkerExitRequest, WorkerInfoResponse, WorkerTimestampData}
+import org.apache.celeborn.server.common.http.api.ApiRequestContext
+import org.apache.celeborn.server.common.http.api.v1.ApiUtils
+import org.apache.celeborn.service.deploy.worker.Worker
+
+@Tag(name = "Worker")
+@Produces(Array(MediaType.APPLICATION_JSON))
+@Consumes(Array(MediaType.APPLICATION_JSON))
+class WorkerResource extends ApiRequestContext {
+ private def worker = httpService.asInstanceOf[Worker]
+
+ @ApiResponse(
+ responseCode = "200",
+ content = Array(new Content(
+ mediaType = MediaType.APPLICATION_JSON,
+ schema = new Schema(
+ implementation = classOf[WorkerInfoResponse]))),
+ description = "List the worker information.")
+ @GET
+ def workers(): WorkerInfoResponse = {
+ ApiUtils.workerInfoResponse(
+ worker.workerInfo,
+ worker.workerStatusManager.currentWorkerStatus,
+ worker.shutdown.get(),
+ worker.registered.get())
+ }
+
+ @ApiResponse(
+ responseCode = "200",
+ content = Array(new Content(
+ mediaType = MediaType.APPLICATION_JSON,
+ schema = new Schema(implementation = classOf[UnAvailablePeersResponse]))),
+ description =
+ "List the unavailable peers of the worker, this always means the worker connect to the peer failed.")
+ @Path("/unavailable_peers")
+ @GET
+ def unavailablePeerWorkers(): UnAvailablePeersResponse = {
+ new UnAvailablePeersResponse()
+ .peers(
+ worker.unavailablePeers.asScala.map { case (worker, lastTimeout) =>
+ new WorkerTimestampData().worker(ApiUtils.workerData(worker)).timestamp(lastTimeout)
+ }.toSeq.asJava)
+ }
+
+ @ApiResponse(
+ responseCode = "200",
+ content = Array(new Content(
+ mediaType = MediaType.APPLICATION_JSON,
+ schema = new Schema(
+ implementation = classOf[HandleResponse]))),
+ description =
+ "Trigger this worker to exit. Legal exit types are 'Decommission', 'Graceful' and 'Immediately'.")
+ @POST
+ @Path("exit")
+ def exit(request: WorkerExitRequest): HandleResponse = {
+ new HandleResponse()
+ .success(true)
+ .message(httpService.exit(request.getType.toString))
+ }
+}
diff --git a/worker/src/test/scala/org/apache/celeborn/service/deploy/worker/http/api/v1/ApiV1OpenapiClientSuite.scala b/worker/src/test/scala/org/apache/celeborn/service/deploy/worker/http/api/v1/ApiV1OpenapiClientSuite.scala
new file mode 100644
index 000000000..1c4a1454d
--- /dev/null
+++ b/worker/src/test/scala/org/apache/celeborn/service/deploy/worker/http/api/v1/ApiV1OpenapiClientSuite.scala
@@ -0,0 +1,112 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.celeborn.service.deploy.worker.http.api.v1
+
+import java.util.Collections
+import javax.servlet.http.HttpServletResponse
+
+import org.apache.celeborn.rest.v1.master._
+import org.apache.celeborn.rest.v1.master.invoker._
+import org.apache.celeborn.rest.v1.master.model.{ExcludeWorkerRequest, SendWorkerEventRequest, WorkerId}
+import org.apache.celeborn.rest.v1.master.model.SendWorkerEventRequest.EventTypeEnum
+
+class ApiV1OpenapiClientSuite extends ApiV1WorkerOpenapiClientSuite {
+ private var masterApiClient: ApiClient = _
+
+ override def beforeAll(): Unit = {
+ super.beforeAll()
+ masterApiClient = new ApiClient().setBasePath(s"http://${master.connectionUrl}")
+ }
+
+ test("master: default api") {
+ val api = new DefaultApi(masterApiClient)
+ assert(!api.getThreadDump.getThreadStacks.isEmpty)
+ }
+
+ test("master: conf api") {
+ val api = new ConfApi(masterApiClient)
+ assert(!api.getConf.getConfigs.isEmpty)
+ val e = intercept[ApiException](api.getDynamicConf("", "", ""))
+ assert(e.getCode == HttpServletResponse.SC_SERVICE_UNAVAILABLE)
+ assert(e.getMessage.contains("Dynamic configuration is disabled"))
+ }
+
+ test("master: application api") {
+ val api = new ApplicationApi(masterApiClient)
+ assert(api.getApplications.getApplications.isEmpty)
+ assert(api.getApplicationHostNames.getHostnames.isEmpty)
+ assert(api.getApplicationsDiskUsageSnapshots.getSnapshots.isEmpty)
+ }
+
+ test("master: master api") {
+ val api = new MasterApi(masterApiClient)
+ val e = intercept[ApiException](api.getMasterGroupInfo)
+ assert(e.getCode == HttpServletResponse.SC_BAD_REQUEST)
+ assert(e.getMessage.contains("HA is not enabled"))
+ }
+
+ test("master: shuffle api") {
+ val api = new ShuffleApi(masterApiClient)
+ assert(api.getShuffles.getShuffleIds.isEmpty)
+ }
+
+ test("master: worker api") {
+ val api = new WorkerApi(masterApiClient)
+ var workersResponse = api.getWorkers
+ assert(!workersResponse.getWorkers.isEmpty)
+ assert(workersResponse.getLostWorkers.isEmpty)
+ assert(workersResponse.getExcludedWorkers.isEmpty)
+ assert(workersResponse.getManualExcludedWorkers.isEmpty)
+ assert(workersResponse.getShutdownWorkers.isEmpty)
+ assert(workersResponse.getDecommissioningWorkers.isEmpty)
+
+ val workerData = workersResponse.getWorkers.get(0)
+ val workerId = new WorkerId()
+ .host(workerData.getHost)
+ .rpcPort(workerData.getRpcPort)
+ .pushPort(workerData.getPushPort)
+ .fetchPort(workerData.getFetchPort)
+ .replicatePort(workerData.getReplicatePort)
+ var handleResponse = api.excludeWorker(
+ new ExcludeWorkerRequest().addAddItem(workerId).remove(Collections.emptyList()))
+ assert(handleResponse.getSuccess)
+
+ workersResponse = api.getWorkers
+ assert(!workersResponse.getWorkers.isEmpty)
+ assert(!workersResponse.getExcludedWorkers.isEmpty)
+ assert(!workersResponse.getManualExcludedWorkers.isEmpty)
+
+ handleResponse = api.excludeWorker(
+ new ExcludeWorkerRequest().addRemoveItem(workerId).add(Collections.emptyList()))
+ assert(handleResponse.getSuccess)
+
+ workersResponse = api.getWorkers
+ assert(!workersResponse.getWorkers.isEmpty)
+ assert(workersResponse.getExcludedWorkers.isEmpty)
+ assert(workersResponse.getManualExcludedWorkers.isEmpty)
+
+ assert(api.getWorkerEvents.getWorkerEvents.isEmpty)
+
+ handleResponse = api.sendWorkerEvent(
+ new SendWorkerEventRequest().addWorkersItem(workerId).eventType(
+ EventTypeEnum.DECOMMISSIONTHENIDLE))
+ assert(handleResponse.getSuccess)
+
+ assert(!api.getWorkerEvents.getWorkerEvents.isEmpty)
+ }
+}
diff --git a/worker/src/test/scala/org/apache/celeborn/service/deploy/worker/http/api/v1/ApiV1WorkerOpenapiClientSuite.scala b/worker/src/test/scala/org/apache/celeborn/service/deploy/worker/http/api/v1/ApiV1WorkerOpenapiClientSuite.scala
new file mode 100644
index 000000000..4b307be49
--- /dev/null
+++ b/worker/src/test/scala/org/apache/celeborn/service/deploy/worker/http/api/v1/ApiV1WorkerOpenapiClientSuite.scala
@@ -0,0 +1,89 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.celeborn.service.deploy.worker.http.api.v1
+
+import javax.servlet.http.HttpServletResponse
+
+import org.apache.celeborn.CelebornFunSuite
+import org.apache.celeborn.common.CelebornConf
+import org.apache.celeborn.rest.v1.worker._
+import org.apache.celeborn.rest.v1.worker.invoker.{ApiClient, ApiException}
+import org.apache.celeborn.service.deploy.MiniClusterFeature
+import org.apache.celeborn.service.deploy.master.Master
+import org.apache.celeborn.service.deploy.worker.Worker
+
+abstract class ApiV1WorkerOpenapiClientSuite extends CelebornFunSuite with MiniClusterFeature {
+ private val celebornConf = new CelebornConf()
+ protected var master: Master = _
+ protected var worker: Worker = _
+ private var workerApiClient: ApiClient = _
+
+ override def beforeAll(): Unit = {
+ logInfo("test initialized, setup celeborn mini cluster")
+ val (m, w) =
+ setupMiniClusterWithRandomPorts(workerConf = celebornConf.getAll.toMap, workerNum = 1)
+ master = m
+ worker = w.head
+ super.beforeAll()
+ workerApiClient = new ApiClient().setBasePath(s"http://${worker.connectionUrl}")
+ }
+
+ override def afterAll(): Unit = {
+ super.afterAll()
+ logInfo("all test complete, stop celeborn mini cluster")
+ shutdownMiniCluster()
+ }
+
+ test("worker: default api") {
+ val api = new DefaultApi(workerApiClient)
+ assert(!api.getThreadDump.getThreadStacks.isEmpty)
+ }
+
+ test("worker: conf api") {
+ val api = new ConfApi(workerApiClient)
+ assert(!api.getConf.getConfigs.isEmpty)
+ val e = intercept[ApiException](api.getDynamicConf("", "", ""))
+ assert(e.getCode == HttpServletResponse.SC_SERVICE_UNAVAILABLE)
+ assert(e.getMessage.contains("Dynamic configuration is disabled"))
+ }
+
+ test("worker: application api") {
+ val api = new ApplicationApi(workerApiClient)
+ assert(api.getApplicationList.getApplications.isEmpty)
+ assert(api.getApplicationsDiskUsage.getAppDiskUsages.isEmpty)
+ }
+
+ test("worker: shuffle api") {
+ val api = new ShuffleApi(workerApiClient)
+ assert(api.getShuffles.getShuffleIds.isEmpty)
+ assert(api.getShufflePartitions.getPrimaryPartitions.isEmpty)
+ assert(api.getShufflePartitions.getReplicaPartitions.isEmpty)
+ }
+
+ test("worker: worker api") {
+ val api = new WorkerApi(workerApiClient)
+ val workerInfo = api.getWorkerInfo
+ assert(!workerInfo.getIsShutdown)
+ assert(workerInfo.getIsRegistered)
+ assert(!workerInfo.getIsDecommissioning)
+
+ assert(api.unavailablePeers().getPeers.isEmpty)
+
+ // assert(api.workerExit(new WorkerExitRequest()).getSuccess)
+ }
+}
diff --git a/worker/src/test/scala/org/apache/celeborn/service/deploy/worker/http/api/v1/ApiV1WorkerResourceSuite.scala b/worker/src/test/scala/org/apache/celeborn/service/deploy/worker/http/api/v1/ApiV1WorkerResourceSuite.scala
new file mode 100644
index 000000000..d30c8f4be
--- /dev/null
+++ b/worker/src/test/scala/org/apache/celeborn/service/deploy/worker/http/api/v1/ApiV1WorkerResourceSuite.scala
@@ -0,0 +1,82 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.celeborn.service.deploy.worker.http.api.v1
+
+import javax.servlet.http.HttpServletResponse
+import javax.ws.rs.core.MediaType
+
+import org.apache.celeborn.rest.v1.model.{AppDiskUsagesResponse, ApplicationsResponse, ShufflePartitionsResponse, ShufflesResponse, UnAvailablePeersResponse, WorkerInfoResponse}
+import org.apache.celeborn.server.common.HttpService
+import org.apache.celeborn.server.common.http.api.v1.ApiV1BaseResourceSuite
+import org.apache.celeborn.service.deploy.MiniClusterFeature
+import org.apache.celeborn.service.deploy.worker.Worker
+
+class ApiV1WorkerResourceSuite extends ApiV1BaseResourceSuite with MiniClusterFeature {
+ private var worker: Worker = _
+ override protected def httpService: HttpService = worker
+
+ override def beforeAll(): Unit = {
+ logInfo("test initialized, setup celeborn mini cluster")
+ val (m, w) =
+ setupMiniClusterWithRandomPorts(workerConf = celebornConf.getAll.toMap, workerNum = 1)
+ worker = w.head
+ super.beforeAll()
+ }
+
+ override def afterAll(): Unit = {
+ super.afterAll()
+ logInfo("all test complete, stop celeborn mini cluster")
+ shutdownMiniCluster()
+ }
+
+ test("shuffle resource") {
+ var response = webTarget.path("shuffles").request(MediaType.APPLICATION_JSON).get()
+ assert(HttpServletResponse.SC_OK == response.getStatus)
+ assert(response.readEntity(classOf[ShufflesResponse]).getShuffleIds.isEmpty)
+
+ response = webTarget.path("shuffles/partitions").request(MediaType.APPLICATION_JSON).get()
+ assert(HttpServletResponse.SC_OK == response.getStatus)
+ val partitions = response.readEntity(classOf[ShufflePartitionsResponse])
+ assert(partitions.getPrimaryPartitions.isEmpty)
+ assert(partitions.getReplicaPartitions.isEmpty)
+ }
+
+ test("application resource") {
+ var response = webTarget.path("applications").request(MediaType.APPLICATION_JSON).get()
+ assert(HttpServletResponse.SC_OK == response.getStatus)
+ assert(response.readEntity(classOf[ApplicationsResponse]).getApplications.isEmpty)
+
+ response =
+ webTarget.path("applications/top_disk_usages").request(MediaType.APPLICATION_JSON).get()
+ assert(HttpServletResponse.SC_OK == response.getStatus)
+ assert(response.readEntity(classOf[AppDiskUsagesResponse]).getAppDiskUsages.isEmpty)
+ }
+
+ test("worker resource") {
+ var response = webTarget.path("workers").request(MediaType.APPLICATION_JSON).get()
+ assert(HttpServletResponse.SC_OK == response.getStatus)
+ val workerData = response.readEntity(classOf[WorkerInfoResponse])
+ assert(workerData.getIsRegistered)
+ assert(!workerData.getIsShutdown)
+ assert(!workerData.getIsDecommissioning)
+
+ response = webTarget.path("workers/unavailable_peers").request(MediaType.APPLICATION_JSON).get()
+ assert(HttpServletResponse.SC_OK == response.getStatus)
+ assert(response.readEntity(classOf[UnAvailablePeersResponse]).getPeers.isEmpty)
+ }
+}