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 2026/06/26 06:08:00 UTC
(flink) 02/04: [FLINK-39911][historyserver] Extract AbstractHistoryServerHandler from HistoryServerStaticFileServerHandler
This is an automated email from the ASF dual-hosted git repository.
reswqa pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit c1f4c5ea64c1862c9c915f5d071e1a455da5a306
Author: Zihao Chen <is...@gmail.com>
AuthorDate: Thu Jun 11 23:42:56 2026 +0800
[FLINK-39911][historyserver] Extract AbstractHistoryServerHandler from HistoryServerStaticFileServerHandler
---
...dler.java => AbstractHistoryServerHandler.java} | 227 +++++++++++---------
.../runtime/webmonitor/history/HistoryServer.java | 5 +-
.../HistoryServerStaticFileServerHandler.java | 228 +--------------------
.../history/AbstractHistoryServerHandlerTest.java | 179 ++++++++++++++++
.../HistoryServerStaticFileServerHandlerTest.java | 95 ---------
.../web-dashboard/src/app/app.interceptor.ts | 2 +-
6 files changed, 328 insertions(+), 408 deletions(-)
diff --git a/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/history/HistoryServerStaticFileServerHandler.java b/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/history/AbstractHistoryServerHandler.java
similarity index 62%
copy from flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/history/HistoryServerStaticFileServerHandler.java
copy to flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/history/AbstractHistoryServerHandler.java
index 40e5cb379d8..8b93ef01349 100644
--- a/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/history/HistoryServerStaticFileServerHandler.java
+++ b/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/history/AbstractHistoryServerHandler.java
@@ -9,23 +9,16 @@
*
* 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.
+ * 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.webmonitor.history;
-/**
- * *************************************************************************** This code is based on
- * the "HttpStaticFileServerHandler" from the Netty project's HTTP server example.
- *
- * <p>See http://netty.io and
- * https://github.com/netty/netty/blob/4.0/example/src/main/java/io/netty/example/http/file/HttpStaticFileServerHandler.java
- * ***************************************************************************
- */
import org.apache.flink.runtime.rest.NotFoundException;
import org.apache.flink.runtime.rest.handler.RestHandlerException;
import org.apache.flink.runtime.rest.handler.legacy.files.StaticFileServerHandler;
@@ -60,7 +53,6 @@ import java.io.RandomAccessFile;
import java.net.URI;
import java.net.URL;
import java.nio.file.Files;
-import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.Collections;
import java.util.Date;
@@ -74,46 +66,33 @@ import static org.apache.flink.shaded.netty4.io.netty.handler.codec.http.HttpRes
import static org.apache.flink.shaded.netty4.io.netty.handler.codec.http.HttpVersion.HTTP_1_1;
import static org.apache.flink.util.Preconditions.checkNotNull;
-/**
- * Simple file server handler used by the {@link HistoryServer} that serves requests to web
- * frontend's static files, such as HTML, CSS, JS or JSON files.
- *
- * <p>This code is based on the "HttpStaticFileServerHandler" from the Netty project's HTTP server
- * example.
- *
- * <p>This class is a copy of the {@link StaticFileServerHandler}. The differences are that the
- * request path is modified to end on ".json" if it does not have a filename extension; when
- * "index.html" is requested we load "index_hs.html" instead to inject the modified HistoryServer
- * WebInterface and that the caching of the "/joboverview" page is prevented.
- */
+/** Abstract base class for HistoryServer handlers. */
@ChannelHandler.Sharable
-public class HistoryServerStaticFileServerHandler
+public abstract class AbstractHistoryServerHandler<Entry>
extends SimpleChannelInboundHandler<RoutedRequest> {
- /** Default logger, if none is specified. */
- private static final Logger LOG =
- LoggerFactory.getLogger(HistoryServerStaticFileServerHandler.class);
-
- // ------------------------------------------------------------------------
+ private static final Logger LOG = LoggerFactory.getLogger(AbstractHistoryServerHandler.class);
/** The path in which the static documents are. */
- private final File rootPath;
+ protected final File rootPath;
+
+ protected final ArchiveStorage<Entry> archiveStorage;
- public HistoryServerStaticFileServerHandler(File rootPath) throws IOException {
+ protected AbstractHistoryServerHandler(ArchiveStorage<Entry> archiveStorage, File rootPath)
+ throws IOException {
+ this.archiveStorage = archiveStorage;
this.rootPath = checkNotNull(rootPath).getCanonicalFile();
}
- // ------------------------------------------------------------------------
- // Responses to requests
- // ------------------------------------------------------------------------
+ // -------------------------------------------------------------------------
+ // Netty lifecycle
+ // -------------------------------------------------------------------------
@Override
- public void channelRead0(ChannelHandlerContext ctx, RoutedRequest routedRequest)
+ public final void channelRead0(ChannelHandlerContext ctx, RoutedRequest routedRequest)
throws Exception {
- String requestPath = routedRequest.getPath();
-
try {
- respondWithFile(ctx, routedRequest.getRequest(), requestPath);
+ respondToRequest(ctx, routedRequest);
} catch (RestHandlerException rhe) {
HandlerUtils.sendErrorResponse(
ctx,
@@ -124,9 +103,33 @@ public class HistoryServerStaticFileServerHandler
}
}
- /** Response when running with leading JobManager. */
- private void respondWithFile(ChannelHandlerContext ctx, HttpRequest request, String requestPath)
- throws IOException, ParseException, RestHandlerException {
+ @Override
+ public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
+ if (ctx.channel().isActive()) {
+ LOG.error("Caught exception", cause);
+ HandlerUtils.sendErrorResponse(
+ ctx,
+ false,
+ new ErrorResponseBody("Internal server error."),
+ INTERNAL_SERVER_ERROR,
+ Collections.emptyMap());
+ }
+ }
+
+ // -------------------------------------------------------------------------
+ // Request handling
+ // -------------------------------------------------------------------------
+
+ /**
+ * Handles HTTP requests by routing to JSON resource or static file response.
+ *
+ * @param ctx Netty Channel context
+ * @param routedRequest The routed HTTP request
+ * @throws Exception Any exception that occurs during processing
+ */
+ protected void respondToRequest(ChannelHandlerContext ctx, RoutedRequest routedRequest)
+ throws Exception {
+ String requestPath = routedRequest.getPath();
// make sure we request the "index.html" in case there is a directory request
if (requestPath.endsWith("/")) {
@@ -135,49 +138,38 @@ public class HistoryServerStaticFileServerHandler
if (!requestPath.contains(".")) { // we assume that the path ends in either .html or .js
requestPath = requestPath + ".json";
- }
- // convert to absolute path
- final File file = new File(rootPath, requestPath);
-
- if (!file.exists()) {
- // file does not exist. Try to load it with the classloader
- ClassLoader cl = HistoryServerStaticFileServerHandler.class.getClassLoader();
-
- try (InputStream resourceStream = cl.getResourceAsStream("web" + requestPath)) {
- boolean success = false;
- try {
- if (resourceStream != null) {
- URL root = cl.getResource("web");
- URL requested = cl.getResource("web" + requestPath);
-
- if (root != null && requested != null) {
- URI rootURI = new URI(root.getPath()).normalize();
- URI requestedURI = new URI(requested.getPath()).normalize();
-
- // Check that we don't load anything from outside of the
- // expected scope.
- if (!rootURI.relativize(requestedURI).equals(requestedURI)) {
- LOG.debug("Loading missing file from classloader: {}", requestPath);
- // ensure that directory to file exists.
- file.getParentFile().mkdirs();
- Files.copy(resourceStream, file.toPath());
-
- success = true;
- }
- }
- }
- } catch (Throwable t) {
- LOG.error("error while responding", t);
- } finally {
- if (!success) {
- LOG.debug("Unable to load requested file {} from classloader", requestPath);
- throw new NotFoundException("File not found.");
- }
- }
+ LOG.debug("Responding to request for path {}", requestPath);
+ Entry resource = loadResource(requestPath);
+
+ if (resource == null) {
+ LOG.debug("Unable to load requested resource {}", requestPath);
+ throw new NotFoundException("Resource not found.");
+ }
+
+ respondWithResource(ctx, routedRequest.getRequest(), requestPath, resource);
+ } else {
+ File destFile = new File(rootPath, requestPath);
+ if (!destFile.exists()) {
+ tryLoadFromClassloader(destFile, requestPath);
}
+ responseWithFile(ctx, routedRequest.getRequest(), requestPath, destFile);
}
+ }
+
+ // -------------------------------------------------------------------------
+ // Response helpers
+ // -------------------------------------------------------------------------
+
+ /** Respond to the request with the resource. */
+ protected abstract void respondWithResource(
+ ChannelHandlerContext ctx, HttpRequest request, String requestPath, Entry resource)
+ throws Exception;
+ /** Respond to the request with the file. */
+ protected void responseWithFile(
+ ChannelHandlerContext ctx, HttpRequest request, String requestPath, File file)
+ throws Exception {
StaticFileServerHandler.checkFileValidity(file, rootPath, LOG);
// cache validation
@@ -267,16 +259,65 @@ public class HistoryServerStaticFileServerHandler
}
}
- @Override
- public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
- if (ctx.channel().isActive()) {
- LOG.error("Caught exception", cause);
- HandlerUtils.sendErrorResponse(
- ctx,
- false,
- new ErrorResponseBody("Internal server error."),
- INTERNAL_SERVER_ERROR,
- Collections.emptyMap());
+ // -------------------------------------------------------------------------
+ // Resource loading
+ // -------------------------------------------------------------------------
+
+ /**
+ * Loads the resource for the given request path from the archive storage.
+ *
+ * @param requestPath The request path
+ * @return The resource for the given request path, or null if not found
+ */
+ private Entry loadResource(String requestPath) throws Exception {
+ String requestKey = requestPath.startsWith("/") ? requestPath.substring(1) : requestPath;
+ if (archiveStorage.exists(requestKey)) {
+ return archiveStorage.getEntry(requestKey);
+ }
+ return null;
+ }
+
+ /**
+ * Attempts to load a missing resource from the classloader's {@code web/} resource directory
+ * and copies it to {@code destFile} on disk.
+ *
+ * @param destFile destination file
+ * @param requestPath relative resource path (e.g. {@code /index.html})
+ */
+ protected void tryLoadFromClassloader(File destFile, String requestPath) throws Exception {
+ ClassLoader cl = AbstractHistoryServerHandler.class.getClassLoader();
+
+ try (InputStream resourceStream = cl.getResourceAsStream("web" + requestPath)) {
+ boolean success = false;
+ try {
+ if (resourceStream != null) {
+ URL root = cl.getResource("web");
+ URL requested = cl.getResource("web" + requestPath);
+
+ if (root != null && requested != null) {
+ URI rootURI = new URI(root.getPath()).normalize();
+ URI requestedURI = new URI(requested.getPath()).normalize();
+
+ // Check that we don't load anything from outside of the
+ // expected scope.
+ if (!rootURI.relativize(requestedURI).equals(requestedURI)) {
+ LOG.debug("Loading missing resource from classloader: {}", requestPath);
+ // ensure that directory to file exists.
+ destFile.getParentFile().mkdirs();
+ Files.copy(resourceStream, destFile.toPath());
+
+ success = true;
+ }
+ }
+ }
+ } catch (Throwable t) {
+ LOG.error("error while responding", t);
+ }
+
+ if (!success) {
+ LOG.debug("Unable to load requested resource {} from classloader", requestPath);
+ throw new NotFoundException("Resource not found.");
+ }
}
}
}
diff --git a/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/history/HistoryServer.java b/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/history/HistoryServer.java
index 8ddcd57a402..ba836be15ef 100644
--- a/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/history/HistoryServer.java
+++ b/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/history/HistoryServer.java
@@ -363,7 +363,10 @@ public class HistoryServer {
new GeneratedLogUrlHandler(
CompletableFuture.completedFuture(pattern))));
- router.addGet("/:*", new HistoryServerStaticFileServerHandler(webDir));
+ router.addGet(
+ "/:*",
+ new HistoryServerStaticFileServerHandler(
+ (FileArchiveStorage) archiveStorage, webDir));
createDashboardConfigFile();
diff --git a/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/history/HistoryServerStaticFileServerHandler.java b/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/history/HistoryServerStaticFileServerHandler.java
index 40e5cb379d8..695f8edfffe 100644
--- a/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/history/HistoryServerStaticFileServerHandler.java
+++ b/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/history/HistoryServerStaticFileServerHandler.java
@@ -26,53 +26,14 @@ package org.apache.flink.runtime.webmonitor.history;
* https://github.com/netty/netty/blob/4.0/example/src/main/java/io/netty/example/http/file/HttpStaticFileServerHandler.java
* ***************************************************************************
*/
-import org.apache.flink.runtime.rest.NotFoundException;
-import org.apache.flink.runtime.rest.handler.RestHandlerException;
import org.apache.flink.runtime.rest.handler.legacy.files.StaticFileServerHandler;
-import org.apache.flink.runtime.rest.handler.router.RoutedRequest;
-import org.apache.flink.runtime.rest.handler.util.HandlerUtils;
-import org.apache.flink.runtime.rest.messages.ErrorResponseBody;
-import org.apache.flink.shaded.netty4.io.netty.channel.ChannelFuture;
-import org.apache.flink.shaded.netty4.io.netty.channel.ChannelFutureListener;
import org.apache.flink.shaded.netty4.io.netty.channel.ChannelHandler;
import org.apache.flink.shaded.netty4.io.netty.channel.ChannelHandlerContext;
-import org.apache.flink.shaded.netty4.io.netty.channel.DefaultFileRegion;
-import org.apache.flink.shaded.netty4.io.netty.channel.SimpleChannelInboundHandler;
-import org.apache.flink.shaded.netty4.io.netty.handler.codec.http.DefaultHttpResponse;
-import org.apache.flink.shaded.netty4.io.netty.handler.codec.http.HttpChunkedInput;
-import org.apache.flink.shaded.netty4.io.netty.handler.codec.http.HttpHeaderValues;
import org.apache.flink.shaded.netty4.io.netty.handler.codec.http.HttpRequest;
-import org.apache.flink.shaded.netty4.io.netty.handler.codec.http.HttpResponse;
-import org.apache.flink.shaded.netty4.io.netty.handler.codec.http.HttpUtil;
-import org.apache.flink.shaded.netty4.io.netty.handler.codec.http.LastHttpContent;
-import org.apache.flink.shaded.netty4.io.netty.handler.ssl.SslHandler;
-import org.apache.flink.shaded.netty4.io.netty.handler.stream.ChunkedFile;
-
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
import java.io.File;
-import java.io.FileNotFoundException;
import java.io.IOException;
-import java.io.InputStream;
-import java.io.RandomAccessFile;
-import java.net.URI;
-import java.net.URL;
-import java.nio.file.Files;
-import java.text.ParseException;
-import java.text.SimpleDateFormat;
-import java.util.Collections;
-import java.util.Date;
-import java.util.Locale;
-
-import static org.apache.flink.shaded.netty4.io.netty.handler.codec.http.HttpHeaderNames.CONNECTION;
-import static org.apache.flink.shaded.netty4.io.netty.handler.codec.http.HttpHeaderNames.IF_MODIFIED_SINCE;
-import static org.apache.flink.shaded.netty4.io.netty.handler.codec.http.HttpResponseStatus.INTERNAL_SERVER_ERROR;
-import static org.apache.flink.shaded.netty4.io.netty.handler.codec.http.HttpResponseStatus.NOT_FOUND;
-import static org.apache.flink.shaded.netty4.io.netty.handler.codec.http.HttpResponseStatus.OK;
-import static org.apache.flink.shaded.netty4.io.netty.handler.codec.http.HttpVersion.HTTP_1_1;
-import static org.apache.flink.util.Preconditions.checkNotNull;
/**
* Simple file server handler used by the {@link HistoryServer} that serves requests to web
@@ -87,20 +48,17 @@ import static org.apache.flink.util.Preconditions.checkNotNull;
* WebInterface and that the caching of the "/joboverview" page is prevented.
*/
@ChannelHandler.Sharable
-public class HistoryServerStaticFileServerHandler
- extends SimpleChannelInboundHandler<RoutedRequest> {
-
- /** Default logger, if none is specified. */
- private static final Logger LOG =
- LoggerFactory.getLogger(HistoryServerStaticFileServerHandler.class);
+public class HistoryServerStaticFileServerHandler extends AbstractHistoryServerHandler<File> {
// ------------------------------------------------------------------------
- /** The path in which the static documents are. */
- private final File rootPath;
-
public HistoryServerStaticFileServerHandler(File rootPath) throws IOException {
- this.rootPath = checkNotNull(rootPath).getCanonicalFile();
+ this(new FileArchiveStorage(rootPath), rootPath);
+ }
+
+ public HistoryServerStaticFileServerHandler(
+ FileArchiveStorage fileArchiveStorage, File rootPath) throws IOException {
+ super(fileArchiveStorage, rootPath);
}
// ------------------------------------------------------------------------
@@ -108,175 +66,9 @@ public class HistoryServerStaticFileServerHandler
// ------------------------------------------------------------------------
@Override
- public void channelRead0(ChannelHandlerContext ctx, RoutedRequest routedRequest)
+ protected void respondWithResource(
+ ChannelHandlerContext ctx, HttpRequest request, String requestPath, File file)
throws Exception {
- String requestPath = routedRequest.getPath();
-
- try {
- respondWithFile(ctx, routedRequest.getRequest(), requestPath);
- } catch (RestHandlerException rhe) {
- HandlerUtils.sendErrorResponse(
- ctx,
- routedRequest.getRequest(),
- new ErrorResponseBody(rhe.getMessage()),
- rhe.getHttpResponseStatus(),
- Collections.emptyMap());
- }
- }
-
- /** Response when running with leading JobManager. */
- private void respondWithFile(ChannelHandlerContext ctx, HttpRequest request, String requestPath)
- throws IOException, ParseException, RestHandlerException {
-
- // make sure we request the "index.html" in case there is a directory request
- if (requestPath.endsWith("/")) {
- requestPath = requestPath + "index.html";
- }
-
- if (!requestPath.contains(".")) { // we assume that the path ends in either .html or .js
- requestPath = requestPath + ".json";
- }
-
- // convert to absolute path
- final File file = new File(rootPath, requestPath);
-
- if (!file.exists()) {
- // file does not exist. Try to load it with the classloader
- ClassLoader cl = HistoryServerStaticFileServerHandler.class.getClassLoader();
-
- try (InputStream resourceStream = cl.getResourceAsStream("web" + requestPath)) {
- boolean success = false;
- try {
- if (resourceStream != null) {
- URL root = cl.getResource("web");
- URL requested = cl.getResource("web" + requestPath);
-
- if (root != null && requested != null) {
- URI rootURI = new URI(root.getPath()).normalize();
- URI requestedURI = new URI(requested.getPath()).normalize();
-
- // Check that we don't load anything from outside of the
- // expected scope.
- if (!rootURI.relativize(requestedURI).equals(requestedURI)) {
- LOG.debug("Loading missing file from classloader: {}", requestPath);
- // ensure that directory to file exists.
- file.getParentFile().mkdirs();
- Files.copy(resourceStream, file.toPath());
-
- success = true;
- }
- }
- }
- } catch (Throwable t) {
- LOG.error("error while responding", t);
- } finally {
- if (!success) {
- LOG.debug("Unable to load requested file {} from classloader", requestPath);
- throw new NotFoundException("File not found.");
- }
- }
- }
- }
-
- StaticFileServerHandler.checkFileValidity(file, rootPath, LOG);
-
- // cache validation
- final String ifModifiedSince = request.headers().get(IF_MODIFIED_SINCE);
- if (ifModifiedSince != null && !ifModifiedSince.isEmpty()) {
- SimpleDateFormat dateFormatter =
- new SimpleDateFormat(StaticFileServerHandler.HTTP_DATE_FORMAT, Locale.US);
- Date ifModifiedSinceDate = dateFormatter.parse(ifModifiedSince);
-
- // Only compare up to the second because the datetime format we send to the client
- // does not have milliseconds
- long ifModifiedSinceDateSeconds = ifModifiedSinceDate.getTime() / 1000;
- long fileLastModifiedSeconds = file.lastModified() / 1000;
- if (ifModifiedSinceDateSeconds == fileLastModifiedSeconds) {
- if (LOG.isDebugEnabled()) {
- LOG.debug(
- "Responding 'NOT MODIFIED' for file '" + file.getAbsolutePath() + '\'');
- }
-
- StaticFileServerHandler.sendNotModified(ctx);
- return;
- }
- }
-
- if (LOG.isDebugEnabled()) {
- LOG.debug("Responding with file '" + file.getAbsolutePath() + '\'');
- }
-
- // Don't need to close this manually. Netty's DefaultFileRegion will take care of it.
- final RandomAccessFile raf;
- try {
- raf = new RandomAccessFile(file, "r");
- } catch (FileNotFoundException e) {
- if (LOG.isDebugEnabled()) {
- LOG.debug("Could not find file {}.", file.getAbsolutePath());
- }
- HandlerUtils.sendErrorResponse(
- ctx,
- request,
- new ErrorResponseBody("File not found."),
- NOT_FOUND,
- Collections.emptyMap());
- return;
- }
-
- try {
- long fileLength = raf.length();
-
- HttpResponse response = new DefaultHttpResponse(HTTP_1_1, OK);
- StaticFileServerHandler.setContentTypeHeader(response, file);
-
- // the job overview should be updated as soon as possible
- if (!requestPath.equals("/joboverview.json")) {
- StaticFileServerHandler.setDateAndCacheHeaders(response, file);
- }
- if (HttpUtil.isKeepAlive(request)) {
- response.headers().set(CONNECTION, HttpHeaderValues.KEEP_ALIVE);
- }
- HttpUtil.setContentLength(response, fileLength);
-
- // write the initial line and the header.
- ctx.write(response);
-
- // write the content.
- ChannelFuture lastContentFuture;
- if (ctx.pipeline().get(SslHandler.class) == null) {
- ctx.write(
- new DefaultFileRegion(raf.getChannel(), 0, fileLength),
- ctx.newProgressivePromise());
- lastContentFuture = ctx.writeAndFlush(LastHttpContent.EMPTY_LAST_CONTENT);
- } else {
- lastContentFuture =
- ctx.writeAndFlush(
- new HttpChunkedInput(new ChunkedFile(raf, 0, fileLength, 8192)),
- ctx.newProgressivePromise());
- // HttpChunkedInput will write the end marker (LastHttpContent) for us.
- }
-
- // close the connection, if no keep-alive is needed
- if (!HttpUtil.isKeepAlive(request)) {
- lastContentFuture.addListener(ChannelFutureListener.CLOSE);
- }
- } catch (Exception e) {
- raf.close();
- LOG.error("Failed to serve file.", e);
- throw new RestHandlerException("Internal server error.", INTERNAL_SERVER_ERROR);
- }
- }
-
- @Override
- public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
- if (ctx.channel().isActive()) {
- LOG.error("Caught exception", cause);
- HandlerUtils.sendErrorResponse(
- ctx,
- false,
- new ErrorResponseBody("Internal server error."),
- INTERNAL_SERVER_ERROR,
- Collections.emptyMap());
- }
+ responseWithFile(ctx, request, requestPath, file);
}
}
diff --git a/flink-runtime-web/src/test/java/org/apache/flink/runtime/webmonitor/history/AbstractHistoryServerHandlerTest.java b/flink-runtime-web/src/test/java/org/apache/flink/runtime/webmonitor/history/AbstractHistoryServerHandlerTest.java
new file mode 100644
index 00000000000..7a7107da5d4
--- /dev/null
+++ b/flink-runtime-web/src/test/java/org/apache/flink/runtime/webmonitor/history/AbstractHistoryServerHandlerTest.java
@@ -0,0 +1,179 @@
+/*
+ * 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.webmonitor.history;
+
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.runtime.rest.handler.router.Router;
+import org.apache.flink.runtime.webmonitor.testutils.HttpUtils;
+import org.apache.flink.runtime.webmonitor.utils.WebFrontendBootstrap;
+import org.apache.flink.testutils.junit.extensions.parameterized.Parameter;
+import org.apache.flink.testutils.junit.extensions.parameterized.ParameterizedTestExtension;
+import org.apache.flink.testutils.junit.extensions.parameterized.Parameters;
+
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.TestTemplate;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.junit.jupiter.api.io.TempDir;
+import org.slf4j.LoggerFactory;
+
+import java.io.File;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.Collection;
+import java.util.Collections;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * Common HTTP-level tests for {@link AbstractHistoryServerHandler} subclasses. New subclasses can
+ * be exercised by adding an entry to {@link #handlerFactories()}.
+ */
+@ExtendWith(ParameterizedTestExtension.class)
+public class AbstractHistoryServerHandlerTest {
+
+ /** Factory that creates a concrete handler bound to the given web directory. */
+ @FunctionalInterface
+ public interface HandlerFactory {
+ AbstractHistoryServerHandler<?> create(File webDir) throws Exception;
+ }
+
+ /**
+ * Provides the concrete {@link AbstractHistoryServerHandler} implementations under test. To
+ * cover an additional handler simply add another factory here.
+ */
+ @Parameters(name = "handlerFactory={0}")
+ private static Collection<HandlerFactory> handlerFactories() {
+ HandlerFactory staticFileServerHandlerFactory = HistoryServerStaticFileServerHandler::new;
+ return Collections.singletonList(staticFileServerHandlerFactory);
+ }
+
+ @Parameter public HandlerFactory handlerFactory;
+
+ @TempDir private Path tmpDir;
+
+ private Path webDir;
+ private AbstractHistoryServerHandler<?> handler;
+ private WebFrontendBootstrap webUI;
+ private String baseUrl;
+
+ @BeforeEach
+ void setUp() throws Exception {
+ webDir = Files.createDirectory(tmpDir.resolve("webDir"));
+ final Path uploadDir = Files.createDirectory(tmpDir.resolve("uploadDir"));
+
+ handler = handlerFactory.create(webDir.toFile());
+ Router<?> router = new Router().addGet("/:*", handler);
+ webUI =
+ new WebFrontendBootstrap(
+ router,
+ LoggerFactory.getLogger(getClass()),
+ uploadDir.toFile(),
+ null,
+ "localhost",
+ 0,
+ new Configuration());
+ baseUrl = "http://localhost:" + webUI.getServerPort();
+ }
+
+ @AfterEach
+ void tearDown() {
+ if (webUI != null) {
+ webUI.shutdown();
+ }
+ }
+
+ /**
+ * Tests requests against static files served from the {@code web/} resources packaged with the
+ * handler module:
+ *
+ * <ul>
+ * <li>a request for {@code /} is rewritten to {@code /index.html};
+ * <li>{@code /index.html} is loaded via the classloader fallback when not present on disk;
+ * <li>a missing static file (e.g. {@code /hello.html}) results in 404.
+ * </ul>
+ */
+ @TestTemplate
+ void testRespondWithStaticFile() throws Exception {
+ // /index.html is loaded from the classloader (web/index.html) and served
+ Tuple2<Integer, String> index = HttpUtils.getFromHTTP(baseUrl + "/index.html");
+ assertThat(index.f0).isEqualTo(200);
+ assertThat(index.f1).contains("Apache Flink Web Dashboard");
+
+ // a trailing slash is rewritten to "/index.html"
+ Tuple2<Integer, String> index2 = HttpUtils.getFromHTTP(baseUrl + "/");
+ assertThat(index2).isEqualTo(index);
+
+ // a static file that is neither on disk nor in the classloader yields 404
+ Tuple2<Integer, String> missing = HttpUtils.getFromHTTP(baseUrl + "/hello.html");
+ assertThat(missing.f0).isEqualTo(404);
+ assertThat(missing.f1).contains("not found");
+ }
+
+ /**
+ * Tests file-system safety checks performed by {@code responseWithFile}:
+ *
+ * <ul>
+ * <li>a directory request returns 405;
+ * <li>a path that escapes the {@code webDir} returns 403.
+ * </ul>
+ */
+ @TestTemplate
+ void testRespondWithFile() throws Exception {
+ // requesting a directory rather than a file yields 405
+ Files.createDirectory(webDir.resolve("dir.json"));
+ Tuple2<Integer, String> dirNotFound = HttpUtils.getFromHTTP(baseUrl + "/dir.json");
+ assertThat(dirNotFound.f0).isEqualTo(405);
+ assertThat(dirNotFound.f1).contains("not found");
+
+ // requesting a file outside of the webDir is rejected with 403
+ Files.createFile(tmpDir.resolve("secret"));
+ Tuple2<Integer, String> outsideWebDir = HttpUtils.getFromHTTP(baseUrl + "/../secret");
+ assertThat(outsideWebDir.f0).isEqualTo(403);
+ assertThat(outsideWebDir.f1).contains("Forbidden");
+ }
+
+ /**
+ * Tests requests served via the {@link ArchiveStorage} resource.
+ *
+ * <ul>
+ * <li>an existing entry is returned with its content;
+ * <li>a missing entry results in 404.
+ * </ul>
+ */
+ @TestTemplate
+ void testRespondWithResource() throws Exception {
+ String resourcePath = "overviews/job1.json";
+ String resourceContent = "{\"job\":\"job1\"}";
+ handler.archiveStorage.putArchiveContent(resourcePath, resourceContent);
+
+ // request without an extension: handler will append ".json" and serve the
+ // entry from the ArchiveStorage via respondWithResource
+ Tuple2<Integer, String> resource = HttpUtils.getFromHTTP(baseUrl + "/overviews/job1");
+ assertThat(resource.f0).isEqualTo(200);
+ assertThat(resource.f1).isEqualTo(resourceContent);
+
+ // request a missing resource: archiveStorage returns null and the handler
+ // responds 404
+ Tuple2<Integer, String> missing = HttpUtils.getFromHTTP(baseUrl + "/hello");
+ assertThat(missing.f0).isEqualTo(404);
+ assertThat(missing.f1).contains("not found");
+ }
+}
diff --git a/flink-runtime-web/src/test/java/org/apache/flink/runtime/webmonitor/history/HistoryServerStaticFileServerHandlerTest.java b/flink-runtime-web/src/test/java/org/apache/flink/runtime/webmonitor/history/HistoryServerStaticFileServerHandlerTest.java
deleted file mode 100644
index d908cf633b8..00000000000
--- a/flink-runtime-web/src/test/java/org/apache/flink/runtime/webmonitor/history/HistoryServerStaticFileServerHandlerTest.java
+++ /dev/null
@@ -1,95 +0,0 @@
-/*
- * 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.webmonitor.history;
-
-import org.apache.flink.api.java.tuple.Tuple2;
-import org.apache.flink.configuration.Configuration;
-import org.apache.flink.runtime.rest.handler.router.Router;
-import org.apache.flink.runtime.webmonitor.testutils.HttpUtils;
-import org.apache.flink.runtime.webmonitor.utils.WebFrontendBootstrap;
-
-import org.junit.jupiter.api.Test;
-import org.junit.jupiter.api.io.TempDir;
-import org.slf4j.LoggerFactory;
-
-import java.nio.file.Files;
-import java.nio.file.Path;
-
-import static org.assertj.core.api.Assertions.assertThat;
-
-/** Tests for the HistoryServerStaticFileServerHandler. */
-class HistoryServerStaticFileServerHandlerTest {
-
- @Test
- void testRespondWithFile(@TempDir Path tmpDir) throws Exception {
- final Path webDir = Files.createDirectory(tmpDir.resolve("webDir"));
- final Path uploadDir = Files.createDirectory(tmpDir.resolve("uploadDir"));
-
- Router router =
- new Router()
- .addGet("/:*", new HistoryServerStaticFileServerHandler(webDir.toFile()));
- WebFrontendBootstrap webUI =
- new WebFrontendBootstrap(
- router,
- LoggerFactory.getLogger(HistoryServerStaticFileServerHandlerTest.class),
- uploadDir.toFile(),
- null,
- "localhost",
- 0,
- new Configuration());
-
- int port = webUI.getServerPort();
- try {
- // verify that 404 message is returned when requesting a non-existent file
- Tuple2<Integer, String> notFound404 =
- HttpUtils.getFromHTTP("http://localhost:" + port + "/hello");
- assertThat(notFound404.f0).isEqualTo(404);
- assertThat(notFound404.f1).contains("not found");
-
- // verify that a) a file can be loaded using the ClassLoader and b) that the
- // HistoryServer
- // index_hs.html is injected
- Tuple2<Integer, String> index =
- HttpUtils.getFromHTTP("http://localhost:" + port + "/index.html");
- assertThat(index.f0).isEqualTo(200);
- assertThat(index.f1).contains("Apache Flink Web Dashboard");
-
- // verify that index.html is appended if the request path ends on '/'
- Tuple2<Integer, String> index2 =
- HttpUtils.getFromHTTP("http://localhost:" + port + "/");
- assertThat(index2).isEqualTo(index);
-
- // verify that a 405 message is returned when requesting a directory
- Files.createDirectory(webDir.resolve("dir.json"));
- Tuple2<Integer, String> dirNotFound =
- HttpUtils.getFromHTTP("http://localhost:" + port + "/dir");
- assertThat(dirNotFound.f0).isEqualTo(405);
- assertThat(dirNotFound.f1).contains("not found");
-
- // verify that a 403 message is returned when requesting a file outside the webDir
- Files.createFile(tmpDir.resolve("secret"));
- Tuple2<Integer, String> dirOutsideDirectory =
- HttpUtils.getFromHTTP("http://localhost:" + port + "/../secret");
- assertThat(dirOutsideDirectory.f0).isEqualTo(403);
- assertThat(dirOutsideDirectory.f1).contains("Forbidden");
- } finally {
- webUI.shutdown();
- }
- }
-}
diff --git a/flink-runtime-web/web-dashboard/src/app/app.interceptor.ts b/flink-runtime-web/web-dashboard/src/app/app.interceptor.ts
index f7bbc12f9dd..cfc85d20091 100644
--- a/flink-runtime-web/web-dashboard/src/app/app.interceptor.ts
+++ b/flink-runtime-web/web-dashboard/src/app/app.interceptor.ts
@@ -41,7 +41,7 @@ export class AppInterceptor implements HttpInterceptor {
intercept(req: HttpRequest<unknown>, next: HttpHandler): Observable<HttpEvent<unknown>> {
// Error response from below url should be ignored
const ignoreErrorUrlEndsList = ['checkpoints/config', 'checkpoints'];
- const ignoreErrorMessage = ['File not found.'];
+ const ignoreErrorMessage = ['File not found.', 'Resource not found.'];
const option: NzNotificationDataOptions = {
nzDuration: 0,
nzStyle: { width: 'auto', 'white-space': 'pre-wrap' }