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 <[email protected]> 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' }
