You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by gu...@apache.org on 2022/07/06 09:10:09 UTC
[flink] 04/04: [FLINK-28311][rest] Introduce JobManagerJobEnvironmentHandler
This is an automated email from the ASF dual-hosted git repository.
guoyangze pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit adfe97deb5cea83c9c0b2af6dc295c6aa33ab96c
Author: Yangze Guo <ka...@gmail.com>
AuthorDate: Mon Jul 4 15:10:31 2022 +0800
[FLINK-28311][rest] Introduce JobManagerJobEnvironmentHandler
This closes #20150.
---
.../shortcodes/generated/rest_v1_dispatcher.html | 106 +++++++++++++++++++++
docs/static/generated/rest_v1_dispatcher.yml | 18 ++++
.../src/test/resources/rest_api_v1.snapshot | 61 ++++++++++++
.../job/JobManagerJobEnvironmentHandler.java | 82 ++++++++++++++++
.../job/JobManagerJobEnvironmentHeaders.java | 80 ++++++++++++++++
.../runtime/webmonitor/WebMonitorEndpoint.java | 13 +++
6 files changed, 360 insertions(+)
diff --git a/docs/layouts/shortcodes/generated/rest_v1_dispatcher.html b/docs/layouts/shortcodes/generated/rest_v1_dispatcher.html
index 64ebceeae89..3fb81e43024 100644
--- a/docs/layouts/shortcodes/generated/rest_v1_dispatcher.html
+++ b/docs/layouts/shortcodes/generated/rest_v1_dispatcher.html
@@ -3097,6 +3097,112 @@ Using 'curl' you can upload a jar via 'curl -X POST -H "Expect:" -F "jarfile=@pa
</tr>
</tbody>
</table>
+<table class="rest-api table table-bordered">
+ <tbody>
+ <tr>
+ <td class="text-left" colspan="2"><h5><strong>/jobs/:jobid/jobmanager/environment</strong></h5></td>
+ </tr>
+ <tr>
+ <td class="text-left" style="width: 20%">Verb: <code>GET</code></td>
+ <td class="text-left">Response code: <code>200 OK</code></td>
+ </tr>
+ <tr>
+ <td colspan="2">Returns the jobmanager's environment variables of a specific job.</td>
+ </tr>
+ <tr>
+ <td colspan="2">Path parameters</td>
+ </tr>
+ <tr>
+ <td colspan="2">
+ <ul>
+<li><code>jobid</code> - 32-character hexadecimal string value that identifies a job.</li>
+ </ul>
+ </td>
+ </tr>
+ <tr>
+ <td colspan="2">
+ <div class="book-expand">
+ <label>
+ <div class="book-expand-head flex justify-between">
+ <span>Request</span>
+ <span>▾</span>
+ </div>
+ <input type="checkbox" class="hidden">
+ <div class="book-expand-content markdown-inner">
+ <pre>
+ <code>
+{} </code>
+ </pre>
+ </div>
+ </label>
+ </div>
+ </td>
+ </tr>
+ <tr>
+ <td colspan="2">
+ <div class="book-expand">
+ <label>
+ <div class="book-expand-head flex justify-between">
+ <span>Response</span>
+ <span>▾</span>
+ </div>
+ <input type="checkbox" class="hidden">
+ <div class="book-expand-content markdown-inner">
+ <pre>
+ <code>
+{
+ "type" : "object",
+ "id" : "urn:jsonschema:org:apache:flink:runtime:rest:messages:EnvironmentInfo",
+ "properties" : {
+ "classpath" : {
+ "type" : "array",
+ "items" : {
+ "type" : "string"
+ }
+ },
+ "environment" : {
+ "type" : "array",
+ "items" : {
+ "type" : "object",
+ "id" : "urn:jsonschema:org:apache:flink:runtime:rest:messages:EnvironmentInfo:EnvironmentVariableItem",
+ "properties" : {
+ "key" : {
+ "type" : "string"
+ },
+ "value" : {
+ "type" : "string"
+ }
+ }
+ }
+ },
+ "jvm" : {
+ "type" : "object",
+ "id" : "urn:jsonschema:org:apache:flink:runtime:rest:messages:EnvironmentInfo:JVMInfo",
+ "properties" : {
+ "arch" : {
+ "type" : "string"
+ },
+ "options" : {
+ "type" : "array",
+ "items" : {
+ "type" : "string"
+ }
+ },
+ "version" : {
+ "type" : "string"
+ }
+ }
+ }
+ }
+} </code>
+ </pre>
+ </div>
+ </label>
+ </div>
+ </td>
+ </tr>
+ </tbody>
+</table>
<table class="rest-api table table-bordered">
<tbody>
<tr>
diff --git a/docs/static/generated/rest_v1_dispatcher.yml b/docs/static/generated/rest_v1_dispatcher.yml
index ecb6b964f33..b495a75290b 100644
--- a/docs/static/generated/rest_v1_dispatcher.yml
+++ b/docs/static/generated/rest_v1_dispatcher.yml
@@ -659,6 +659,24 @@ paths:
type: boolean
items:
$ref: '#/components/schemas/ConfigurationInfoEntry'
+ /jobs/{jobid}/jobmanager/environment:
+ get:
+ description: Returns the jobmanager's environment variables of a specific job.
+ operationId: getJobManagerJobEnvironment
+ parameters:
+ - name: jobid
+ in: path
+ description: 32-character hexadecimal string value that identifies a job.
+ required: true
+ schema:
+ $ref: '#/components/schemas/JobID'
+ responses:
+ "200":
+ description: The request was successful.
+ content:
+ application/json:
+ schema:
+ $ref: '#/components/schemas/EnvironmentInfo'
/jobs/{jobid}/metrics:
get:
description: Provides access to job metrics.
diff --git a/flink-runtime-web/src/test/resources/rest_api_v1.snapshot b/flink-runtime-web/src/test/resources/rest_api_v1.snapshot
index 87aedec6d18..2de7c030c29 100644
--- a/flink-runtime-web/src/test/resources/rest_api_v1.snapshot
+++ b/flink-runtime-web/src/test/resources/rest_api_v1.snapshot
@@ -1918,6 +1918,67 @@
}
}
}
+ }, {
+ "url" : "/jobs/:jobid/jobmanager/environment",
+ "method" : "GET",
+ "status-code" : "200 OK",
+ "file-upload" : false,
+ "path-parameters" : {
+ "pathParameters" : [ {
+ "key" : "jobid"
+ } ]
+ },
+ "query-parameters" : {
+ "queryParameters" : [ ]
+ },
+ "request" : {
+ "type" : "any"
+ },
+ "response" : {
+ "type" : "object",
+ "id" : "urn:jsonschema:org:apache:flink:runtime:rest:messages:EnvironmentInfo",
+ "properties" : {
+ "environment" : {
+ "type" : "array",
+ "items" : {
+ "type" : "object",
+ "id" : "urn:jsonschema:org:apache:flink:runtime:rest:messages:EnvironmentInfo:EnvironmentVariableItem",
+ "properties" : {
+ "key" : {
+ "type" : "string"
+ },
+ "value" : {
+ "type" : "string"
+ }
+ }
+ }
+ },
+ "jvm" : {
+ "type" : "object",
+ "id" : "urn:jsonschema:org:apache:flink:runtime:rest:messages:EnvironmentInfo:JVMInfo",
+ "properties" : {
+ "version" : {
+ "type" : "string"
+ },
+ "arch" : {
+ "type" : "string"
+ },
+ "options" : {
+ "type" : "array",
+ "items" : {
+ "type" : "string"
+ }
+ }
+ }
+ },
+ "classpath" : {
+ "type" : "array",
+ "items" : {
+ "type" : "string"
+ }
+ }
+ }
+ }
}, {
"url" : "/jobs/:jobid/metrics",
"method" : "GET",
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/job/JobManagerJobEnvironmentHandler.java b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/job/JobManagerJobEnvironmentHandler.java
new file mode 100644
index 00000000000..ba2ff0ba494
--- /dev/null
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/job/JobManagerJobEnvironmentHandler.java
@@ -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.flink.runtime.rest.handler.job;
+
+import org.apache.flink.api.common.time.Time;
+import org.apache.flink.runtime.rest.handler.AbstractRestHandler;
+import org.apache.flink.runtime.rest.handler.HandlerRequest;
+import org.apache.flink.runtime.rest.handler.RestHandlerException;
+import org.apache.flink.runtime.rest.messages.EmptyRequestBody;
+import org.apache.flink.runtime.rest.messages.EnvironmentInfo;
+import org.apache.flink.runtime.rest.messages.JobIDPathParameter;
+import org.apache.flink.runtime.rest.messages.JobMessageParameters;
+import org.apache.flink.runtime.rest.messages.MessageHeaders;
+import org.apache.flink.runtime.rest.messages.job.JobManagerJobEnvironmentHeaders;
+import org.apache.flink.runtime.scheduler.ExecutionGraphInfo;
+import org.apache.flink.runtime.webmonitor.RestfulGateway;
+import org.apache.flink.runtime.webmonitor.history.ArchivedJson;
+import org.apache.flink.runtime.webmonitor.history.JsonArchivist;
+import org.apache.flink.runtime.webmonitor.retriever.GatewayRetriever;
+
+import javax.annotation.Nonnull;
+
+import java.io.IOException;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+
+/** Handler which serves the jobmanager's environment variables of a specific job. */
+public class JobManagerJobEnvironmentHandler
+ extends AbstractRestHandler<
+ RestfulGateway, EmptyRequestBody, EnvironmentInfo, JobMessageParameters>
+ implements JsonArchivist {
+ private final EnvironmentInfo environmentInfo;
+
+ public JobManagerJobEnvironmentHandler(
+ GatewayRetriever<? extends RestfulGateway> leaderRetriever,
+ Time timeout,
+ Map<String, String> responseHeaders,
+ MessageHeaders<EmptyRequestBody, EnvironmentInfo, JobMessageParameters>
+ messageHeaders) {
+ super(leaderRetriever, timeout, responseHeaders, messageHeaders);
+
+ this.environmentInfo = EnvironmentInfo.create();
+ }
+
+ @Override
+ protected CompletableFuture<EnvironmentInfo> handleRequest(
+ @Nonnull HandlerRequest<EmptyRequestBody> request, @Nonnull RestfulGateway gateway)
+ throws RestHandlerException {
+ return CompletableFuture.completedFuture(environmentInfo);
+ }
+
+ @Override
+ public Collection<ArchivedJson> archiveJsonWithPath(ExecutionGraphInfo executionGraphInfo)
+ throws IOException {
+ return Collections.singletonList(
+ new ArchivedJson(
+ JobManagerJobEnvironmentHeaders.getInstance()
+ .getTargetRestEndpointURL()
+ .replace(
+ ':' + JobIDPathParameter.KEY,
+ executionGraphInfo.getJobId().toString()),
+ environmentInfo));
+ }
+}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/job/JobManagerJobEnvironmentHeaders.java b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/job/JobManagerJobEnvironmentHeaders.java
new file mode 100644
index 00000000000..628b3ceb818
--- /dev/null
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/job/JobManagerJobEnvironmentHeaders.java
@@ -0,0 +1,80 @@
+/*
+ * 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.flink.runtime.rest.messages.job;
+
+import org.apache.flink.runtime.rest.HttpMethodWrapper;
+import org.apache.flink.runtime.rest.messages.EmptyRequestBody;
+import org.apache.flink.runtime.rest.messages.EnvironmentInfo;
+import org.apache.flink.runtime.rest.messages.JobMessageParameters;
+import org.apache.flink.runtime.rest.messages.MessageHeaders;
+
+import org.apache.flink.shaded.netty4.io.netty.handler.codec.http.HttpResponseStatus;
+
+/**
+ * Message headers for the {@link
+ * org.apache.flink.runtime.rest.handler.job.JobManagerJobEnvironmentHandler}.
+ */
+public class JobManagerJobEnvironmentHeaders
+ implements MessageHeaders<EmptyRequestBody, EnvironmentInfo, JobMessageParameters> {
+ private static final JobManagerJobEnvironmentHeaders INSTANCE =
+ new JobManagerJobEnvironmentHeaders();
+
+ public static final String JOBMANAGER_JOB_ENV_REST_PATH = "/jobs/:jobid/jobmanager/environment";
+
+ private JobManagerJobEnvironmentHeaders() {}
+
+ @Override
+ public Class<EmptyRequestBody> getRequestClass() {
+ return EmptyRequestBody.class;
+ }
+
+ @Override
+ public HttpMethodWrapper getHttpMethod() {
+ return HttpMethodWrapper.GET;
+ }
+
+ @Override
+ public String getTargetRestEndpointURL() {
+ return JOBMANAGER_JOB_ENV_REST_PATH;
+ }
+
+ @Override
+ public Class<EnvironmentInfo> getResponseClass() {
+ return EnvironmentInfo.class;
+ }
+
+ @Override
+ public HttpResponseStatus getResponseStatusCode() {
+ return HttpResponseStatus.OK;
+ }
+
+ @Override
+ public JobMessageParameters getUnresolvedMessageParameters() {
+ return new JobMessageParameters();
+ }
+
+ public static JobManagerJobEnvironmentHeaders getInstance() {
+ return INSTANCE;
+ }
+
+ @Override
+ public String getDescription() {
+ return "Returns the jobmanager's environment variables of a specific job.";
+ }
+}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/webmonitor/WebMonitorEndpoint.java b/flink-runtime/src/main/java/org/apache/flink/runtime/webmonitor/WebMonitorEndpoint.java
index b3fead6af48..cf5ad90abcc 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/webmonitor/WebMonitorEndpoint.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/webmonitor/WebMonitorEndpoint.java
@@ -51,6 +51,7 @@ import org.apache.flink.runtime.rest.handler.job.JobExceptionsHandler;
import org.apache.flink.runtime.rest.handler.job.JobExecutionResultHandler;
import org.apache.flink.runtime.rest.handler.job.JobIdsHandler;
import org.apache.flink.runtime.rest.handler.job.JobManagerJobConfigurationHandler;
+import org.apache.flink.runtime.rest.handler.job.JobManagerJobEnvironmentHandler;
import org.apache.flink.runtime.rest.handler.job.JobPlanHandler;
import org.apache.flink.runtime.rest.handler.job.JobStatusHandler;
import org.apache.flink.runtime.rest.handler.job.JobVertexAccumulatorsHandler;
@@ -125,6 +126,7 @@ import org.apache.flink.runtime.rest.messages.cluster.JobManagerThreadDumpHeader
import org.apache.flink.runtime.rest.messages.cluster.ShutdownHeaders;
import org.apache.flink.runtime.rest.messages.job.JobDetailsHeaders;
import org.apache.flink.runtime.rest.messages.job.JobManagerJobConfigurationHeaders;
+import org.apache.flink.runtime.rest.messages.job.JobManagerJobEnvironmentHeaders;
import org.apache.flink.runtime.rest.messages.job.JobStatusInfoHeaders;
import org.apache.flink.runtime.rest.messages.job.SubtaskCurrentAttemptDetailsHeaders;
import org.apache.flink.runtime.rest.messages.job.SubtaskExecutionAttemptAccumulatorsHeaders;
@@ -328,6 +330,13 @@ public class WebMonitorEndpoint<T extends RestfulGateway> extends RestServerEndp
responseHeaders,
JobManagerEnvironmentHeaders.getInstance());
+ JobManagerJobEnvironmentHandler jobManagerJobEnvironmentHandler =
+ new JobManagerJobEnvironmentHandler(
+ leaderRetriever,
+ timeout,
+ responseHeaders,
+ JobManagerJobEnvironmentHeaders.getInstance());
+
JobConfigHandler jobConfigHandler =
new JobConfigHandler(
leaderRetriever,
@@ -683,6 +692,10 @@ public class WebMonitorEndpoint<T extends RestfulGateway> extends RestServerEndp
Tuple2.of(
jobManagerEnvironmentHandler.getMessageHeaders(),
jobManagerEnvironmentHandler));
+ handlers.add(
+ Tuple2.of(
+ jobManagerJobEnvironmentHandler.getMessageHeaders(),
+ jobManagerJobEnvironmentHandler));
handlers.add(Tuple2.of(dashboardConfigHandler.getMessageHeaders(), dashboardConfigHandler));
handlers.add(Tuple2.of(jobIdsHandler.getMessageHeaders(), jobIdsHandler));
handlers.add(Tuple2.of(jobStatusHandler.getMessageHeaders(), jobStatusHandler));