聊聊flink的HistoryServer

序

本文主要研究一下flink的HistoryServer

HistoryServer

flink-1.7.2/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/history/HistoryServer.java

public class HistoryServer {

    private static final Logger LOG = LoggerFactory.getLogger(HistoryServer.class);

    private final Configuration config;

    private final String webAddress;
    private final int webPort;
    private final long webRefreshIntervalMillis;
    private final File webDir;

    private final HistoryServerArchiveFetcher archiveFetcher;

    @Nullable
    private final SSLHandlerFactory serverSSLFactory;
    private WebFrontendBootstrap netty;

    private final Object startupShutdownLock = new Object();
    private final AtomicBoolean shutdownRequested = new AtomicBoolean(false);
    private final Thread shutdownHook;

    public static void main(String[] args) throws Exception {
        ParameterTool pt = ParameterTool.fromArgs(args);
        String configDir = pt.getRequired("configDir");

        LOG.info("Loading configuration from {}", configDir);
        final Configuration flinkConfig = GlobalConfiguration.loadConfiguration(configDir);

        try {
            FileSystem.initialize(flinkConfig);
        } catch (IOException e) {
            throw new Exception("Error while setting the default filesystem scheme from configuration.", e);
        }

        // run the history server
        SecurityUtils.install(new SecurityConfiguration(flinkConfig));

        try {
            SecurityUtils.getInstalledContext().runSecured(new Callable<Integer>() {
                @Override
                public Integer call() throws Exception {
                    HistoryServer hs = new HistoryServer(flinkConfig);
                    hs.run();
                    return 0;
                }
            });
            System.exit(0);
        } catch (Throwable t) {
            final Throwable strippedThrowable = ExceptionUtils.stripException(t, UndeclaredThrowableException.class);
            LOG.error("Failed to run HistoryServer.", strippedThrowable);
            strippedThrowable.printStackTrace();
            System.exit(1);
        }
    }

    public HistoryServer(Configuration config) throws IOException, FlinkException {
        this(config, new CountDownLatch(0));
    }

    public HistoryServer(Configuration config, CountDownLatch numFinishedPolls) throws IOException, FlinkException {
        Preconditions.checkNotNull(config);
        Preconditions.checkNotNull(numFinishedPolls);

        this.config = config;
        if (config.getBoolean(HistoryServerOptions.HISTORY_SERVER_WEB_SSL_ENABLED) && SSLUtils.isRestSSLEnabled(config)) {
            LOG.info("Enabling SSL for the history server.");
            try {
                this.serverSSLFactory = SSLUtils.createRestServerSSLEngineFactory(config);
            } catch (Exception e) {
                throw new IOException("Failed to initialize SSLContext for the history server.", e);
            }
        } else {
            this.serverSSLFactory = null;
        }

        webAddress = config.getString(HistoryServerOptions.HISTORY_SERVER_WEB_ADDRESS);
        webPort = config.getInteger(HistoryServerOptions.HISTORY_SERVER_WEB_PORT);
        webRefreshIntervalMillis = config.getLong(HistoryServerOptions.HISTORY_SERVER_WEB_REFRESH_INTERVAL);

        String webDirectory = config.getString(HistoryServerOptions.HISTORY_SERVER_WEB_DIR);
        if (webDirectory == null) {
            webDirectory = System.getProperty("java.io.tmpdir") + File.separator + "flink-web-history-" + UUID.randomUUID();
        }
        webDir = new File(webDirectory);

        String refreshDirectories = config.getString(HistoryServerOptions.HISTORY_SERVER_ARCHIVE_DIRS);
        if (refreshDirectories == null) {
            throw new FlinkException(HistoryServerOptions.HISTORY_SERVER_ARCHIVE_DIRS + " was not configured.");
        }
        List<RefreshLocation> refreshDirs = new ArrayList<>();
        for (String refreshDirectory : refreshDirectories.split(",")) {
            try {
                Path refreshPath = WebMonitorUtils.validateAndNormalizeUri(new Path(refreshDirectory).toUri());
                FileSystem refreshFS = refreshPath.getFileSystem();
                refreshDirs.add(new RefreshLocation(refreshPath, refreshFS));
            } catch (Exception e) {
                // there's most likely something wrong with the path itself, so we ignore it from here on
                LOG.warn("Failed to create Path or FileSystem for directory '{}'. Directory will not be monitored.", refreshDirectory, e);
            }
        }

        if (refreshDirs.isEmpty()) {
            throw new FlinkException("Failed to validate any of the configured directories to monitor.");
        }

        long refreshIntervalMillis = config.getLong(HistoryServerOptions.HISTORY_SERVER_ARCHIVE_REFRESH_INTERVAL);
        archiveFetcher = new HistoryServerArchiveFetcher(refreshIntervalMillis, refreshDirs, webDir, numFinishedPolls);

        this.shutdownHook = ShutdownHookUtil.addShutdownHook(
            HistoryServer.this::stop,
            HistoryServer.class.getSimpleName(),
            LOG);
    }

    @VisibleForTesting
    int getWebPort() {
        return netty.getServerPort();
    }

    public void run() {
        try {
            start();
            new CountDownLatch(1).await();
        } catch (Exception e) {
            LOG.error("Failure while running HistoryServer.", e);
        } finally {
            stop();
        }
    }

    // ------------------------------------------------------------------------
    // Life-cycle
    // ------------------------------------------------------------------------

    void start() throws IOException, InterruptedException {
        synchronized (startupShutdownLock) {
            LOG.info("Starting history server.");

            Files.createDirectories(webDir.toPath());
            LOG.info("Using directory {} as local cache.", webDir);

            Router router = new Router();
            router.addGet("/:*", new HistoryServerStaticFileServerHandler(webDir));

            if (!webDir.exists() && !webDir.mkdirs()) {
                throw new IOException("Failed to create local directory " + webDir.getAbsoluteFile() + ".");
            }

            createDashboardConfigFile();

            archiveFetcher.start();

            netty = new WebFrontendBootstrap(router, LOG, webDir, serverSSLFactory, webAddress, webPort, config);
        }
    }

    void stop() {
        if (shutdownRequested.compareAndSet(false, true)) {
            synchronized (startupShutdownLock) {
                LOG.info("Stopping history server.");

                try {
                    netty.shutdown();
                } catch (Throwable t) {
                    LOG.warn("Error while shutting down WebFrontendBootstrap.", t);
                }

                archiveFetcher.stop();

                try {
                    LOG.info("Removing web dashboard root cache directory {}", webDir);
                    FileUtils.deleteDirectory(webDir);
                } catch (Throwable t) {
                    LOG.warn("Error while deleting web root directory {}", webDir, t);
                }

                LOG.info("Stopped history server.");

                // Remove shutdown hook to prevent resource leaks
                ShutdownHookUtil.removeShutdownHook(shutdownHook, getClass().getSimpleName(), LOG);
            }
        }
    }

    // ------------------------------------------------------------------------
    // File generation
    // ------------------------------------------------------------------------

    static FileWriter createOrGetFile(File folder, String name) throws IOException {
        File file = new File(folder, name + ".json");
        if (!file.exists()) {
            Files.createFile(file.toPath());
        }
        FileWriter fr = new FileWriter(file);
        return fr;
    }

    private void createDashboardConfigFile() throws IOException {
        try (FileWriter fw = createOrGetFile(webDir, "config")) {
            fw.write(createConfigJson(DashboardConfiguration.from(webRefreshIntervalMillis, ZonedDateTime.now())));
            fw.flush();
        } catch (IOException ioe) {
            LOG.error("Failed to write config file.");
            throw ioe;
        }
    }

    private static String createConfigJson(DashboardConfiguration dashboardConfiguration) throws IOException {
        StringWriter writer = new StringWriter();
        JsonGenerator gen = JsonFactory.JACKSON_FACTORY.createGenerator(writer);

        gen.writeStartObject();
        gen.writeNumberField(DashboardConfiguration.FIELD_NAME_REFRESH_INTERVAL, dashboardConfiguration.getRefreshInterval());
        gen.writeNumberField(DashboardConfiguration.FIELD_NAME_TIMEZONE_OFFSET, dashboardConfiguration.getTimeZoneOffset());
        gen.writeStringField(DashboardConfiguration.FIELD_NAME_TIMEZONE_NAME, dashboardConfiguration.getTimeZoneName());
        gen.writeStringField(DashboardConfiguration.FIELD_NAME_FLINK_VERSION, dashboardConfiguration.getFlinkVersion());
        gen.writeStringField(DashboardConfiguration.FIELD_NAME_FLINK_REVISION, dashboardConfiguration.getFlinkRevision());

        gen.writeEndObject();

        gen.close();

        return writer.toString();
    }

    /**
     * Container for the {@link Path} and {@link FileSystem} of a refresh directory.
     */
    static class RefreshLocation {
        private final Path path;
        private final FileSystem fs;

        private RefreshLocation(Path path, FileSystem fs) {
            this.path = path;
            this.fs = fs;
        }

        public Path getPath() {
            return path;
        }

        public FileSystem getFs() {
            return fs;
        }
    }
}

• HistoryServer提供了finished jobs的相关查询功能；构造器从配置中读取historyserver.web.address、historyserver.web.port( 默认8082 )、historyserver.web.refresh-interval( 默认10秒 )、historyserver.web.tmpdir、historyserver.archive.fs.dir、historyserver.archive.fs.refresh-interval( 默认10秒 )，然后创建了HistoryServerArchiveFetcher
• 其run方法主要是调用start方法，该方法主要是启动HistoryServerArchiveFetcher，然后创建WebFrontendBootstrap
• 构造器使用ShutdownHookUtil.addShutdownHook注册了ShutdownHook，在shutdown时执行stop方法，stop方法主要是调用WebFrontendBootstrap的shutdown方法以及HistoryServerArchiveFetcher的stop方法，然后清理webDir，移除shutdownHook

HistoryServerArchiveFetcher

flink-1.7.2/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/history/HistoryServerArchiveFetcher.java

class HistoryServerArchiveFetcher {

    private static final Logger LOG = LoggerFactory.getLogger(HistoryServerArchiveFetcher.class);

    private static final JsonFactory jacksonFactory = new JsonFactory();
    private static final ObjectMapper mapper = new ObjectMapper();

    private final ScheduledExecutorService executor = Executors.newSingleThreadScheduledExecutor(
        new ExecutorThreadFactory("Flink-HistoryServer-ArchiveFetcher"));
    private final JobArchiveFetcherTask fetcherTask;
    private final long refreshIntervalMillis;

    HistoryServerArchiveFetcher(long refreshIntervalMillis, List<HistoryServer.RefreshLocation> refreshDirs, File webDir, CountDownLatch numFinishedPolls) {
        this.refreshIntervalMillis = refreshIntervalMillis;
        this.fetcherTask = new JobArchiveFetcherTask(refreshDirs, webDir, numFinishedPolls);
        if (LOG.isInfoEnabled()) {
            for (HistoryServer.RefreshLocation refreshDir : refreshDirs) {
                LOG.info("Monitoring directory {} for archived jobs.", refreshDir.getPath());
            }
        }
    }

    void start() {
        executor.scheduleWithFixedDelay(fetcherTask, 0, refreshIntervalMillis, TimeUnit.MILLISECONDS);
    }

    void stop() {
        executor.shutdown();

        try {
            if (!executor.awaitTermination(1, TimeUnit.SECONDS)) {
                executor.shutdownNow();
            }
        } catch (InterruptedException ignored) {
            executor.shutdownNow();
        }
    }

    /**
     * {@link TimerTask} that polls the directories configured as {@link HistoryServerOptions#HISTORY_SERVER_ARCHIVE_DIRS} for
     * new job archives.
     */
    static class JobArchiveFetcherTask extends TimerTask {

        private final List<HistoryServer.RefreshLocation> refreshDirs;
        private final CountDownLatch numFinishedPolls;

        /** Cache of all available jobs identified by their id. */
        private final Set<String> cachedArchives;

        private final File webDir;
        private final File webJobDir;
        private final File webOverviewDir;

        private static final String JSON_FILE_ENDING = ".json";

        JobArchiveFetcherTask(List<HistoryServer.RefreshLocation> refreshDirs, File webDir, CountDownLatch numFinishedPolls) {
            this.refreshDirs = checkNotNull(refreshDirs);
            this.numFinishedPolls = numFinishedPolls;
            this.cachedArchives = new HashSet<>();
            this.webDir = checkNotNull(webDir);
            this.webJobDir = new File(webDir, "jobs");
            webJobDir.mkdir();
            this.webOverviewDir = new File(webDir, "overviews");
            webOverviewDir.mkdir();
        }

        @Override
        public void run() {
            try {
                for (HistoryServer.RefreshLocation refreshLocation : refreshDirs) {
                    Path refreshDir = refreshLocation.getPath();
                    FileSystem refreshFS = refreshLocation.getFs();

                    // contents of /:refreshDir
                    FileStatus[] jobArchives;
                    try {
                        jobArchives = refreshFS.listStatus(refreshDir);
                    } catch (IOException e) {
                        LOG.error("Failed to access job archive location for path {}.", refreshDir, e);
                        continue;
                    }
                    if (jobArchives == null) {
                        continue;
                    }
                    boolean updateOverview = false;
                    for (FileStatus jobArchive : jobArchives) {
                        Path jobArchivePath = jobArchive.getPath();
                        String jobID = jobArchivePath.getName();
                        try {
                            JobID.fromHexString(jobID);
                        } catch (IllegalArgumentException iae) {
                            LOG.debug("Archive directory {} contained file with unexpected name {}. Ignoring file.",
                                refreshDir, jobID, iae);
                            continue;
                        }
                        if (cachedArchives.add(jobID)) {
                            try {
                                for (ArchivedJson archive : FsJobArchivist.getArchivedJsons(jobArchive.getPath())) {
                                    String path = archive.getPath();
                                    String json = archive.getJson();

                                    File target;
                                    if (path.equals(JobsOverviewHeaders.URL)) {
                                        target = new File(webOverviewDir, jobID + JSON_FILE_ENDING);
                                    } else if (path.equals("/joboverview")) { // legacy path
                                        json = convertLegacyJobOverview(json);
                                        target = new File(webOverviewDir, jobID + JSON_FILE_ENDING);
                                    } else {
                                        target = new File(webDir, path + JSON_FILE_ENDING);
                                    }

                                    java.nio.file.Path parent = target.getParentFile().toPath();

                                    try {
                                        Files.createDirectories(parent);
                                    } catch (FileAlreadyExistsException ignored) {
                                        // there may be left-over directories from the previous attempt
                                    }

                                    java.nio.file.Path targetPath = target.toPath();

                                    // We overwrite existing files since this may be another attempt at fetching this archive.
                                    // Existing files may be incomplete/corrupt.
                                    Files.deleteIfExists(targetPath);

                                    Files.createFile(target.toPath());
                                    try (FileWriter fw = new FileWriter(target)) {
                                        fw.write(json);
                                        fw.flush();
                                    }
                                }
                                updateOverview = true;
                            } catch (IOException e) {
                                LOG.error("Failure while fetching/processing job archive for job {}.", jobID, e);
                                // Make sure we attempt to fetch the archive again
                                cachedArchives.remove(jobID);
                                // Make sure we do not include this job in the overview
                                try {
                                    Files.delete(new File(webOverviewDir, jobID + JSON_FILE_ENDING).toPath());
                                } catch (IOException ioe) {
                                    LOG.debug("Could not delete file from overview directory.", ioe);
                                }

                                // Clean up job files we may have created
                                File jobDirectory = new File(webJobDir, jobID);
                                try {
                                    FileUtils.deleteDirectory(jobDirectory);
                                } catch (IOException ioe) {
                                    LOG.debug("Could not clean up job directory.", ioe);
                                }
                            }
                        }
                    }
                    if (updateOverview) {
                        updateJobOverview(webOverviewDir, webDir);
                    }
                }
            } catch (Exception e) {
                LOG.error("Critical failure while fetching/processing job archives.", e);
            }
            numFinishedPolls.countDown();
        }
    }

    private static String convertLegacyJobOverview(String legacyOverview) throws IOException {
        JsonNode root = mapper.readTree(legacyOverview);
        JsonNode finishedJobs = root.get("finished");
        JsonNode job = finishedJobs.get(0);

        JobID jobId = JobID.fromHexString(job.get("jid").asText());
        String name = job.get("name").asText();
        JobStatus state = JobStatus.valueOf(job.get("state").asText());

        long startTime = job.get("start-time").asLong();
        long endTime = job.get("end-time").asLong();
        long duration = job.get("duration").asLong();
        long lastMod = job.get("last-modification").asLong();

        JsonNode tasks = job.get("tasks");
        int numTasks = tasks.get("total").asInt();
        int pending = tasks.get("pending").asInt();
        int running = tasks.get("running").asInt();
        int finished = tasks.get("finished").asInt();
        int canceling = tasks.get("canceling").asInt();
        int canceled = tasks.get("canceled").asInt();
        int failed = tasks.get("failed").asInt();

        int[] tasksPerState = new int[ExecutionState.values().length];
        // pending is a mix of CREATED/SCHEDULED/DEPLOYING
        // to maintain the correct number of task states we have to pick one of them
        tasksPerState[ExecutionState.SCHEDULED.ordinal()] = pending;
        tasksPerState[ExecutionState.RUNNING.ordinal()] = running;
        tasksPerState[ExecutionState.FINISHED.ordinal()] = finished;
        tasksPerState[ExecutionState.CANCELING.ordinal()] = canceling;
        tasksPerState[ExecutionState.CANCELED.ordinal()] = canceled;
        tasksPerState[ExecutionState.FAILED.ordinal()] = failed;

        JobDetails jobDetails = new JobDetails(jobId, name, startTime, endTime, duration, state, lastMod, tasksPerState, numTasks);
        MultipleJobsDetails multipleJobsDetails = new MultipleJobsDetails(Collections.singleton(jobDetails));

        StringWriter sw = new StringWriter();
        mapper.writeValue(sw, multipleJobsDetails);
        return sw.toString();
    }

    /**
     * This method replicates the JSON response that would be given by the JobsOverviewHandler when
     * listing both running and finished jobs.
     *
     * <p>Every job archive contains a joboverview.json file containing the same structure. Since jobs are archived on
     * their own however the list of finished jobs only contains a single job.
     *
     * <p>For the display in the HistoryServer WebFrontend we have to combine these overviews.
     */
    private static void updateJobOverview(File webOverviewDir, File webDir) {
        try (JsonGenerator gen = jacksonFactory.createGenerator(HistoryServer.createOrGetFile(webDir, JobsOverviewHeaders.URL))) {
            File[] overviews = new File(webOverviewDir.getPath()).listFiles();
            if (overviews != null) {
                Collection<JobDetails> allJobs = new ArrayList<>(overviews.length);
                for (File overview : overviews) {
                    MultipleJobsDetails subJobs = mapper.readValue(overview, MultipleJobsDetails.class);
                    allJobs.addAll(subJobs.getJobs());
                }
                mapper.writeValue(gen, new MultipleJobsDetails(allJobs));
            }
        } catch (IOException ioe) {
            LOG.error("Failed to update job overview.", ioe);
        }
    }
}

• HistoryServerArchiveFetcher主要是以historyserver.archive.fs.refresh-interval的时间间隔从historyserver.archive.fs.dir目录拉取job archives；它内部创建了JobArchiveFetcherTask来执行这个任务
• JobArchiveFetcherTask继承了jdk的TimerTask，其run方法就是遍历refreshDirs，然后执行FileSystem.listStatus，然后使用FsJobArchivist.getArchivedJsons获取ArchivedJson根据不同path写入到指定文件
• 如果path是/jobs/overview，则写入webDir/overviews/jobID.json文件；如果path是/joboverview，则先调用convertLegacyJobOverview转换json，然后再写入webDir/overviews/jobID.json文件；其他的path则写入webDir/path.json文件

WebFrontendBootstrap

flink-1.7.2/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/utils/WebFrontendBootstrap.java

public class WebFrontendBootstrap {
    private final Router router;
    private final Logger log;
    private final File uploadDir;
    private final ServerBootstrap bootstrap;
    private final Channel serverChannel;
    private final String restAddress;

    public WebFrontendBootstrap(
            Router router,
            Logger log,
            File directory,
            @Nullable SSLHandlerFactory serverSSLFactory,
            String configuredAddress,
            int configuredPort,
            final Configuration config) throws InterruptedException, UnknownHostException {

        this.router = Preconditions.checkNotNull(router);
        this.log = Preconditions.checkNotNull(log);
        this.uploadDir = directory;

        ChannelInitializer<SocketChannel> initializer = new ChannelInitializer<SocketChannel>() {

            @Override
            protected void initChannel(SocketChannel ch) {
                RouterHandler handler = new RouterHandler(WebFrontendBootstrap.this.router, new HashMap<>());

                // SSL should be the first handler in the pipeline
                if (serverSSLFactory != null) {
                    ch.pipeline().addLast("ssl", serverSSLFactory.createNettySSLHandler());
                }

                ch.pipeline()
                    .addLast(new HttpServerCodec())
                    .addLast(new ChunkedWriteHandler())
                    .addLast(new HttpRequestHandler(uploadDir))
                    .addLast(handler.getName(), handler)
                    .addLast(new PipelineErrorHandler(WebFrontendBootstrap.this.log));
            }
        };

        NioEventLoopGroup bossGroup   = new NioEventLoopGroup(1);
        NioEventLoopGroup workerGroup = new NioEventLoopGroup();

        this.bootstrap = new ServerBootstrap();
        this.bootstrap
            .group(bossGroup, workerGroup)
            .channel(NioServerSocketChannel.class)
            .childHandler(initializer);

        ChannelFuture ch;
        if (configuredAddress == null) {
            ch = this.bootstrap.bind(configuredPort);
        } else {
            ch = this.bootstrap.bind(configuredAddress, configuredPort);
        }
        this.serverChannel = ch.sync().channel();

        InetSocketAddress bindAddress = (InetSocketAddress) serverChannel.localAddress();

        InetAddress inetAddress = bindAddress.getAddress();
        final String address;

        if (inetAddress.isAnyLocalAddress()) {
            address = config.getString(JobManagerOptions.ADDRESS, InetAddress.getLocalHost().getHostName());
        } else {
            address = inetAddress.getHostAddress();
        }

        int port = bindAddress.getPort();

        this.log.info("Web frontend listening at {}" + ':' + "{}", address, port);

        final String protocol = serverSSLFactory != null ? "https://" : "http://";

        this.restAddress = protocol + address + ':' + port;
    }

    public ServerBootstrap getBootstrap() {
        return bootstrap;
    }

    public int getServerPort() {
        Channel server = this.serverChannel;
        if (server != null) {
            try {
                return ((InetSocketAddress) server.localAddress()).getPort();
            }
            catch (Exception e) {
                log.error("Cannot access local server port", e);
            }
        }

        return -1;
    }

    public String getRestAddress() {
        return restAddress;
    }

    public void shutdown() {
        if (this.serverChannel != null) {
            this.serverChannel.close().awaitUninterruptibly();
        }
        if (bootstrap != null) {
            if (bootstrap.group() != null) {
                bootstrap.group().shutdownGracefully();
            }
            if (bootstrap.childGroup() != null) {
                bootstrap.childGroup().shutdownGracefully();
            }
        }
    }
}

• WebFrontendBootstrap使用netty启动了一个http server，其pipeline有HttpServerCodec、ChunkedWriteHandler、HttpRequestHandler、RouterHandler、PipelineErrorHandler；其中这里的RouterHandler的Router有个GET的route，其使用的是HistoryServerStaticFileServerHandler，用于给HistoryServer提供静态文件服务

小结

• HistoryServer提供了finished jobs的相关查询功能；其主要由HistoryServerArchiveFetcher以及WebFrontendBootstrap两部分组成；其run方法主要是调用start方法，该方法主要是启动HistoryServerArchiveFetcher，然后创建WebFrontendBootstrap
• HistoryServerArchiveFetcher主要是以historyserver.archive.fs.refresh-interval的时间间隔从historyserver.archive.fs.dir目录拉取job archives；它内部创建了JobArchiveFetcherTask来执行这个任务；JobArchiveFetcherTask继承了jdk的TimerTask，其run方法就是遍历refreshDirs，然后执行FileSystem.listStatus，然后使用FsJobArchivist.getArchivedJsons获取ArchivedJson根据不同path写入到指定文件
• WebFrontendBootstrap使用netty启动了一个http server，其pipeline有HttpServerCodec、ChunkedWriteHandler、HttpRequestHandler、RouterHandler、PipelineErrorHandler；其中这里的RouterHandler的Router有个GET的route，其使用的是HistoryServerStaticFileServerHandler，用于给HistoryServer提供静态文件服务

doc

• HistoryServer