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/incubator-celeborn.git
The following commit(s) were added to refs/heads/main by this push:
new 438cdf674 [CELEBORN-973] Improve HttpRequestHandler handle HTTP
request with base, master and worker
438cdf674 is described below
commit 438cdf6747a240a5fd6a59e5f8a8d2bfeaac7076
Author: SteNicholas <[email protected]>
AuthorDate: Thu Oct 12 15:08:15 2023 +0800
[CELEBORN-973] Improve HttpRequestHandler handle HTTP request with base,
master and worker
### What changes were proposed in this pull request?
The code that `HttpRequestHandler` handles HTTP request could be improved
with handling HTTP request with base, master and worker.
### Why are the changes needed?
Improves `HttpRequestHandler` handle HTTP request with base, master and
worker.
### Does this PR introduce _any_ user-facing change?
No.
### How was this patch tested?
Internal tests.
Closes #1977 from SteNicholas/http-request-handler.
Authored-by: SteNicholas <[email protected]>
Signed-off-by: Cheng Pan <[email protected]>
---
.../server/common/http/HttpRequestHandler.scala | 106 ++++++++++++++++-----
1 file changed, 80 insertions(+), 26 deletions(-)
diff --git
a/service/src/main/scala/org/apache/celeborn/server/common/http/HttpRequestHandler.scala
b/service/src/main/scala/org/apache/celeborn/server/common/http/HttpRequestHandler.scala
index 2115aef14..d574947ed 100644
---
a/service/src/main/scala/org/apache/celeborn/server/common/http/HttpRequestHandler.scala
+++
b/service/src/main/scala/org/apache/celeborn/server/common/http/HttpRequestHandler.scala
@@ -27,14 +27,18 @@ import org.apache.celeborn.common.internal.Logging
import org.apache.celeborn.common.metrics.sink.PrometheusHttpRequestHandler
import org.apache.celeborn.server.common.{HttpService, Service}
+/**
+ * A handler for the REST API that defines how to handle the HTTP request
given a message.
+ *
+ * @param service The service of HTTP server.
+ * @param uri The uri of HTTP request.
+ */
@Sharable
class HttpRequestHandler(
service: HttpService,
prometheusHttpRequestHandler: PrometheusHttpRequestHandler)
extends SimpleChannelInboundHandler[FullHttpRequest] with Logging {
- private val INVALID = "invalid"
-
override def channelReadComplete(ctx: ChannelHandlerContext): Unit = {
ctx.flush()
}
@@ -43,11 +47,11 @@ class HttpRequestHandler(
val uri = req.uri()
val msg = handleRequest(uri)
val response = msg match {
- case INVALID =>
+ case "invalid" =>
if (prometheusHttpRequestHandler != null) {
prometheusHttpRequestHandler.handleRequest(uri)
} else {
- s"invalid uri ${uri}"
+ s"invalid uri $uri"
}
case _ => msg
}
@@ -56,46 +60,96 @@ class HttpRequestHandler(
HttpVersion.HTTP_1_1,
HttpResponseStatus.OK,
Unpooled.copiedBuffer(response, CharsetUtil.UTF_8))
- res.headers().set(HttpHeaderNames.CONTENT_TYPE, "text/html;
charset=UTF-8");
- ctx.writeAndFlush(res).addListener(ChannelFutureListener.CLOSE);
+ res.headers().set(HttpHeaderNames.CONTENT_TYPE, "text/html; charset=UTF-8")
+ ctx.writeAndFlush(res).addListener(ChannelFutureListener.CLOSE)
+ }
+
+ private def handleRequest(uri: String): String = {
+ if (service.serviceName == Service.MASTER) {
+ new MasterRequestHandler(service, uri).handle()
+ } else {
+ new WorkerRequestHandler(service, uri).handle()
+ }
}
+}
+
+/**
+ * A basic handler for the REST API that defines how to handle the HTTP
request.
+ *
+ * @param service The service of HTTP server.
+ * @param uri The uri of HTTP request.
+ */
+class BaseRequestHandler(service: HttpService, uri: String) extends Logging {
- def handleRequest(uri: String): String = {
- val (path, parameters) = HttpUtils.parseUrl(uri)
+ val (path, parameters) = HttpUtils.parseUrl(uri)
+
+ def handle(): String = {
path match {
case "/conf" =>
service.getConf
- case "/masterGroupInfo" if service.serviceName == Service.MASTER =>
- service.getMasterGroupInfo
case "/workerInfo" =>
service.getWorkerInfo
- case "/lostWorkers" if service.serviceName == Service.MASTER =>
- service.getLostWorkers
- case "/excludedWorkers" if service.serviceName == Service.MASTER =>
- service.getExcludedWorkers
- case "/shutdownWorkers" if service.serviceName == Service.MASTER =>
- service.getShutdownWorkers
case "/threadDump" =>
service.getThreadDump
- case "/hostnames" if service.serviceName == Service.MASTER =>
- service.getHostnameList
- case "/applications" if service.serviceName == Service.MASTER =>
- service.getApplicationList
case "/shuffles" =>
service.getShuffleList
case "/listTopDiskUsedApps" =>
service.listTopDiskUseApps
- case "/listPartitionLocationInfo" if service.serviceName ==
Service.WORKER =>
+ case _ => "invalid"
+ }
+ }
+}
+
+/**
+ * A handler for the REST API that defines how to handle the HTTP request from
master.
+ *
+ * @param service The service of HTTP server.
+ * @param uri The uri of HTTP request from master.
+ */
+class MasterRequestHandler(service: HttpService, uri: String)
+ extends BaseRequestHandler(service, uri) with Logging {
+
+ override def handle(): String = {
+ path match {
+ case "/masterGroupInfo" =>
+ service.getMasterGroupInfo
+ case "/lostWorkers" =>
+ service.getLostWorkers
+ case "/excludedWorkers" =>
+ service.getExcludedWorkers
+ case "/shutdownWorkers" =>
+ service.getShutdownWorkers
+ case "/hostnames" =>
+ service.getHostnameList
+ case "/applications" =>
+ service.getApplicationList
+ case _ => super.handle()
+ }
+ }
+}
+
+/**
+ * A handler for the REST API that defines how to handle the HTTP request from
worker.
+ *
+ * @param service The service of HTTP server.
+ * @param uri The uri of HTTP request from worker.
+ */
+class WorkerRequestHandler(service: HttpService, uri: String)
+ extends BaseRequestHandler(service, uri) with Logging {
+
+ override def handle(): String = {
+ path match {
+ case "/listPartitionLocationInfo" =>
service.listPartitionLocationInfo
- case "/unavailablePeers" if service.serviceName == Service.WORKER =>
+ case "/unavailablePeers" =>
service.getUnavailablePeers
- case "/isShutdown" if service.serviceName == Service.WORKER =>
+ case "/isShutdown" =>
service.isShutdown
- case "/isRegistered" if service.serviceName == Service.WORKER =>
+ case "/isRegistered" =>
service.isRegistered
- case "/exit" if service.serviceName == Service.WORKER =>
+ case "/exit" =>
service.exit(parameters.getOrElse("TYPE", ""))
- case _ => INVALID
+ case _ => super.handle()
}
}
}
