devbox
繁依Fanyi0
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

聊聊flink的HistoryServer

作者：繁依Fanyi0 | 2024-05-12 10:45:27

赞

踩

failed to access job archive location for path hdfs

序

本文主要研究一下flink的HistoryServer

HistoryServer

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

  1. public class HistoryServer {
  2.  
  3. 	private static final Logger LOG = LoggerFactory.getLogger(HistoryServer.class);
  4.  
  5. 	private final Configuration config;
  6.  
  7. 	private final String webAddress;
  8. 	private final int webPort;
  9. 	private final long webRefreshIntervalMillis;
  10. 	private final File webDir;
  11.  
  12. 	private final HistoryServerArchiveFetcher archiveFetcher;
  13.  
  14. 	@Nullable
  15. 	private final SSLHandlerFactory serverSSLFactory;
  16. 	private WebFrontendBootstrap netty;
  17.  
  18. 	private final Object startupShutdownLock = new Object();
  19. 	private final AtomicBoolean shutdownRequested = new AtomicBoolean(false);
  20. 	private final Thread shutdownHook;
  21.  
  22. 	public static void main(String[] args) throws Exception {
  23. 		ParameterTool pt = ParameterTool.fromArgs(args);
  24. 		String configDir = pt.getRequired("configDir");
  25.  
  26. 		LOG.info("Loading configuration from {}", configDir);
  27. 		final Configuration flinkConfig = GlobalConfiguration.loadConfiguration(configDir);
  28.  
  29. 		try {
  30. 			FileSystem.initialize(flinkConfig);
  31. 		} catch (IOException e) {
  32. 			throw new Exception("Error while setting the default filesystem scheme from configuration.", e);
  33. 		}
  34.  
  35. 		// run the history server
  36. 		SecurityUtils.install(new SecurityConfiguration(flinkConfig));
  37.  
  38. 		try {
  39. 			SecurityUtils.getInstalledContext().runSecured(new Callable<Integer>() {
  40. 				@Override
  41. 				public Integer call() throws Exception {
  42. 					HistoryServer hs = new HistoryServer(flinkConfig);
  43. 					hs.run();
  44. 					return 0;
  45. 				}
  46. 			});
  47. 			System.exit(0);
  48. 		} catch (Throwable t) {
  49. 			final Throwable strippedThrowable = ExceptionUtils.stripException(t, UndeclaredThrowableException.class);
  50. 			LOG.error("Failed to run HistoryServer.", strippedThrowable);
  51. 			strippedThrowable.printStackTrace();
  52. 			System.exit(1);
  53. 		}
  54. 	}
  55.  
  56. 	public HistoryServer(Configuration config) throws IOException, FlinkException {
  57. 		this(config, new CountDownLatch(0));
  58. 	}
  59.  
  60. 	public HistoryServer(Configuration config, CountDownLatch numFinishedPolls) throws IOException, FlinkException {
  61. 		Preconditions.checkNotNull(config);
  62. 		Preconditions.checkNotNull(numFinishedPolls);
  63.  
  64. 		this.config = config;
  65. 		if (config.getBoolean(HistoryServerOptions.HISTORY_SERVER_WEB_SSL_ENABLED) && SSLUtils.isRestSSLEnabled(config)) {
  66. 			LOG.info("Enabling SSL for the history server.");
  67. 			try {
  68. 				this.serverSSLFactory = SSLUtils.createRestServerSSLEngineFactory(config);
  69. 			} catch (Exception e) {
  70. 				throw new IOException("Failed to initialize SSLContext for the history server.", e);
  71. 			}
  72. 		} else {
  73. 			this.serverSSLFactory = null;
  74. 		}
  75.  
  76. 		webAddress = config.getString(HistoryServerOptions.HISTORY_SERVER_WEB_ADDRESS);
  77. 		webPort = config.getInteger(HistoryServerOptions.HISTORY_SERVER_WEB_PORT);
  78. 		webRefreshIntervalMillis = config.getLong(HistoryServerOptions.HISTORY_SERVER_WEB_REFRESH_INTERVAL);
  79.  
  80. 		String webDirectory = config.getString(HistoryServerOptions.HISTORY_SERVER_WEB_DIR);
  81. 		if (webDirectory == null) {
  82. 			webDirectory = System.getProperty("java.io.tmpdir") + File.separator + "flink-web-history-" + UUID.randomUUID();
  83. 		}
  84. 		webDir = new File(webDirectory);
  85.  
  86. 		String refreshDirectories = config.getString(HistoryServerOptions.HISTORY_SERVER_ARCHIVE_DIRS);
  87. 		if (refreshDirectories == null) {
  88. 			throw new FlinkException(HistoryServerOptions.HISTORY_SERVER_ARCHIVE_DIRS + " was not configured.");
  89. 		}
  90. 		List<RefreshLocation> refreshDirs = new ArrayList<>();
  91. 		for (String refreshDirectory : refreshDirectories.split(",")) {
  92. 			try {
  93. 				Path refreshPath = WebMonitorUtils.validateAndNormalizeUri(new Path(refreshDirectory).toUri());
  94. 				FileSystem refreshFS = refreshPath.getFileSystem();
  95. 				refreshDirs.add(new RefreshLocation(refreshPath, refreshFS));
  96. 			} catch (Exception e) {
  97. 				// there's most likely something wrong with the path itself, so we ignore it from here on
  98. 				LOG.warn("Failed to create Path or FileSystem for directory '{}'. Directory will not be monitored.", refreshDirectory, e);
  99. 			}
  100. 		}
  102. 		if (refreshDirs.isEmpty()) {
  103. 			throw new FlinkException("Failed to validate any of the configured directories to monitor.");
  104. 		}
  106. 		long refreshIntervalMillis = config.getLong(HistoryServerOptions.HISTORY_SERVER_ARCHIVE_REFRESH_INTERVAL);
  107. 		archiveFetcher = new HistoryServerArchiveFetcher(refreshIntervalMillis, refreshDirs, webDir, numFinishedPolls);
  109. 		this.shutdownHook = ShutdownHookUtil.addShutdownHook(
  110. 			HistoryServer.this::stop,
  111. 			HistoryServer.class.getSimpleName(),
  112. 			LOG);
  113. 	}
  115. 	@VisibleForTesting
  116. 	int getWebPort() {
  117. 		return netty.getServerPort();
  118. 	}
  120. 	public void run() {
  121. 		try {
  122. 			start();
  123. 			new CountDownLatch(1).await();
  124. 		} catch (Exception e) {
  125. 			LOG.error("Failure while running HistoryServer.", e);
  126. 		} finally {
  127. 			stop();
  128. 		}
  129. 	}
  131. 	// ------------------------------------------------------------------------
  132. 	// Life-cycle
  133. 	// ------------------------------------------------------------------------
  135. 	void start() throws IOException, InterruptedException {
  136. 		synchronized (startupShutdownLock) {
  137. 			LOG.info("Starting history server.");
  139. 			Files.createDirectories(webDir.toPath());
  140. 			LOG.info("Using directory {} as local cache.", webDir);
  142. 			Router router = new Router();
  143. 			router.addGet("/:*", new HistoryServerStaticFileServerHandler(webDir));
  145. 			if (!webDir.exists() && !webDir.mkdirs()) {
  146. 				throw new IOException("Failed to create local directory " + webDir.getAbsoluteFile() + ".");
  147. 			}
  149. 			createDashboardConfigFile();
  151. 			archiveFetcher.start();
  153. 			netty = new WebFrontendBootstrap(router, LOG, webDir, serverSSLFactory, webAddress, webPort, config);
  154. 		}
  155. 	}
  157. 	void stop() {
  158. 		if (shutdownRequested.compareAndSet(false, true)) {
  159. 			synchronized (startupShutdownLock) {
  160. 				LOG.info("Stopping history server.");
  162. 				try {
  163. 					netty.shutdown();
  164. 				} catch (Throwable t) {
  165. 					LOG.warn("Error while shutting down WebFrontendBootstrap.", t);
  166. 				}
  168. 				archiveFetcher.stop();
  170. 				try {
  171. 					LOG.info("Removing web dashboard root cache directory {}", webDir);
  172. 					FileUtils.deleteDirectory(webDir);
  173. 				} catch (Throwable t) {
  174. 					LOG.warn("Error while deleting web root directory {}", webDir, t);
  175. 				}
  177. 				LOG.info("Stopped history server.");
  179. 				// Remove shutdown hook to prevent resource leaks
  180. 				ShutdownHookUtil.removeShutdownHook(shutdownHook, getClass().getSimpleName(), LOG);
  181. 			}
  182. 		}
  183. 	}
  185. 	// ------------------------------------------------------------------------
  186. 	// File generation
  187. 	// ------------------------------------------------------------------------
  189. 	static FileWriter createOrGetFile(File folder, String name) throws IOException {
  190. 		File file = new File(folder, name + ".json");
  191. 		if (!file.exists()) {
  192. 			Files.createFile(file.toPath());
  193. 		}
  194. 		FileWriter fr = new FileWriter(file);
  195. 		return fr;
  196. 	}
  198. 	private void createDashboardConfigFile() throws IOException {
  199. 		try (FileWriter fw = createOrGetFile(webDir, "config")) {
  200. 			fw.write(createConfigJson(DashboardConfiguration.from(webRefreshIntervalMillis, ZonedDateTime.now())));
  201. 			fw.flush();
  202. 		} catch (IOException ioe) {
  203. 			LOG.error("Failed to write config file.");
  204. 			throw ioe;
  205. 		}
  206. 	}
  208. 	private static String createConfigJson(DashboardConfiguration dashboardConfiguration) throws IOException {
  209. 		StringWriter writer = new StringWriter();
  210. 		JsonGenerator gen = JsonFactory.JACKSON_FACTORY.createGenerator(writer);
  212. 		gen.writeStartObject();
  213. 		gen.writeNumberField(DashboardConfiguration.FIELD_NAME_REFRESH_INTERVAL, dashboardConfiguration.getRefreshInterval());
  214. 		gen.writeNumberField(DashboardConfiguration.FIELD_NAME_TIMEZONE_OFFSET, dashboardConfiguration.getTimeZoneOffset());
  215. 		gen.writeStringField(DashboardConfiguration.FIELD_NAME_TIMEZONE_NAME, dashboardConfiguration.getTimeZoneName());
  216. 		gen.writeStringField(DashboardConfiguration.FIELD_NAME_FLINK_VERSION, dashboardConfiguration.getFlinkVersion());
  217. 		gen.writeStringField(DashboardConfiguration.FIELD_NAME_FLINK_REVISION, dashboardConfiguration.getFlinkRevision());
  219. 		gen.writeEndObject();
  221. 		gen.close();
  223. 		return writer.toString();
  224. 	}
  226. 	/**
  227. 	 * Container for the {@link Path} and {@link FileSystem} of a refresh directory.
  228. 	 */
  229. 	static class RefreshLocation {
  230. 		private final Path path;
  231. 		private final FileSystem fs;
  233. 		private RefreshLocation(Path path, FileSystem fs) {
  234. 			this.path = path;
  235. 			this.fs = fs;
  236. 		}
  238. 		public Path getPath() {
  239. 			return path;
  240. 		}
  242. 		public FileSystem getFs() {
  243. 			return fs;
  244. 		}
  245. 	}
  246. }
  247. 复制代码
  • 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

  1. class HistoryServerArchiveFetcher {
  2.  
  3. 	private static final Logger LOG = LoggerFactory.getLogger(HistoryServerArchiveFetcher.class);
  4.  
  5. 	private static final JsonFactory jacksonFactory = new JsonFactory();
  6. 	private static final ObjectMapper mapper = new ObjectMapper();
  7.  
  8. 	private final ScheduledExecutorService executor = Executors.newSingleThreadScheduledExecutor(
  9. 		new ExecutorThreadFactory("Flink-HistoryServer-ArchiveFetcher"));
  10. 	private final JobArchiveFetcherTask fetcherTask;
  11. 	private final long refreshIntervalMillis;
  12.  
  13. 	HistoryServerArchiveFetcher(long refreshIntervalMillis, List<HistoryServer.RefreshLocation> refreshDirs, File webDir, CountDownLatch numFinishedPolls) {
  14. 		this.refreshIntervalMillis = refreshIntervalMillis;
  15. 		this.fetcherTask = new JobArchiveFetcherTask(refreshDirs, webDir, numFinishedPolls);
  16. 		if (LOG.isInfoEnabled()) {
  17. 			for (HistoryServer.RefreshLocation refreshDir : refreshDirs) {
  18. 				LOG.info("Monitoring directory {} for archived jobs.", refreshDir.getPath());
  19. 			}
  20. 		}
  21. 	}
  22.  
  23. 	void start() {
  24. 		executor.scheduleWithFixedDelay(fetcherTask, 0, refreshIntervalMillis, TimeUnit.MILLISECONDS);
  25. 	}
  26.  
  27. 	void stop() {
  28. 		executor.shutdown();
  29.  
  30. 		try {
  31. 			if (!executor.awaitTermination(1, TimeUnit.SECONDS)) {
  32. 				executor.shutdownNow();
  33. 			}
  34. 		} catch (InterruptedException ignored) {
  35. 			executor.shutdownNow();
  36. 		}
  37. 	}
  38.  
  39. 	/**
  40. 	 * {@link TimerTask} that polls the directories configured as {@link HistoryServerOptions#HISTORY_SERVER_ARCHIVE_DIRS} for
  41. 	 * new job archives.
  42. 	 */
  43. 	static class JobArchiveFetcherTask extends TimerTask {
  44.  
  45. 		private final List<HistoryServer.RefreshLocation> refreshDirs;
  46. 		private final CountDownLatch numFinishedPolls;
  47.  
  48. 		/** Cache of all available jobs identified by their id. */
  49. 		private final Set<String> cachedArchives;
  50.  
  51. 		private final File webDir;
  52. 		private final File webJobDir;
  53. 		private final File webOverviewDir;
  54.  
  55. 		private static final String JSON_FILE_ENDING = ".json";
  56.  
  57. 		JobArchiveFetcherTask(List<HistoryServer.RefreshLocation> refreshDirs, File webDir, CountDownLatch numFinishedPolls) {
  58. 			this.refreshDirs = checkNotNull(refreshDirs);
  59. 			this.numFinishedPolls = numFinishedPolls;
  60. 			this.cachedArchives = new HashSet<>();
  61. 			this.webDir = checkNotNull(webDir);
  62. 			this.webJobDir = new File(webDir, "jobs");
  63. 			webJobDir.mkdir();
  64. 			this.webOverviewDir = new File(webDir, "overviews");
  65. 			webOverviewDir.mkdir();
  66. 		}
  67.  
  68. 		@Override
  69. 		public void run() {
  70. 			try {
  71. 				for (HistoryServer.RefreshLocation refreshLocation : refreshDirs) {
  72. 					Path refreshDir = refreshLocation.getPath();
  73. 					FileSystem refreshFS = refreshLocation.getFs();
  74.  
  75. 					// contents of /:refreshDir
  76. 					FileStatus[] jobArchives;
  77. 					try {
  78. 						jobArchives = refreshFS.listStatus(refreshDir);
  79. 					} catch (IOException e) {
  80. 						LOG.error("Failed to access job archive location for path {}.", refreshDir, e);
  81. 						continue;
  82. 					}
  83. 					if (jobArchives == null) {
  84. 						continue;
  85. 					}
  86. 					boolean updateOverview = false;
  87. 					for (FileStatus jobArchive : jobArchives) {
  88. 						Path jobArchivePath = jobArchive.getPath();
  89. 						String jobID = jobArchivePath.getName();
  90. 						try {
  91. 							JobID.fromHexString(jobID);
  92. 						} catch (IllegalArgumentException iae) {
  93. 							LOG.debug("Archive directory {} contained file with unexpected name {}. Ignoring file.",
  94. 								refreshDir, jobID, iae);
  95. 							continue;
  96. 						}
  97. 						if (cachedArchives.add(jobID)) {
  98. 							try {
  99. 								for (ArchivedJson archive : FsJobArchivist.getArchivedJsons(jobArchive.getPath())) {
  100. 									String path = archive.getPath();
  101. 									String json = archive.getJson();
  102.  
  103. 									File target;
  104. 									if (path.equals(JobsOverviewHeaders.URL)) {
  105. 										target = new File(webOverviewDir, jobID + JSON_FILE_ENDING);
  106. 									} else if (path.equals("/joboverview")) { // legacy path
  107. 										json = convertLegacyJobOverview(json);
  108. 										target = new File(webOverviewDir, jobID + JSON_FILE_ENDING);
  109. 									} else {
  110. 										target = new File(webDir, path + JSON_FILE_ENDING);
  111. 									}
  112.  
  113. 									java.nio.file.Path parent = target.getParentFile().toPath();
  114.  
  115. 									try {
  116. 										Files.createDirectories(parent);
  117. 									} catch (FileAlreadyExistsException ignored) {
  118. 										// there may be left-over directories from the previous attempt
  119. 									}
  120.  
  121. 									java.nio.file.Path targetPath = target.toPath();
  122.  
  123. 									// We overwrite existing files since this may be another attempt at fetching this archive.
  124. 									// Existing files may be incomplete/corrupt.
  125. 									Files.deleteIfExists(targetPath);
  126.  
  127. 									Files.createFile(target.toPath());
  128. 									try (FileWriter fw = new FileWriter(target)) {
  129. 										fw.write(json);
  130. 										fw.flush();
  131. 									}
  132. 								}
  133. 								updateOverview = true;
  134. 							} catch (IOException e) {
  135. 								LOG.error("Failure while fetching/processing job archive for job {}.", jobID, e);
  136. 								// Make sure we attempt to fetch the archive again
  137. 								cachedArchives.remove(jobID);
  138. 								// Make sure we do not include this job in the overview
  139. 								try {
  140. 									Files.delete(new File(webOverviewDir, jobID + JSON_FILE_ENDING).toPath());
  141. 								} catch (IOException ioe) {
  142. 									LOG.debug("Could not delete file from overview directory.", ioe);
  143. 								}
  144.  
  145. 								// Clean up job files we may have created
  146. 								File jobDirectory = new File(webJobDir, jobID);
  147. 								try {
  148. 									FileUtils.deleteDirectory(jobDirectory);
  149. 								} catch (IOException ioe) {
  150. 									LOG.debug("Could not clean up job directory.", ioe);
  151. 								}
  152. 							}
  153. 						}
  154. 					}
  155. 					if (updateOverview) {
  156. 						updateJobOverview(webOverviewDir, webDir);
  157. 					}
  158. 				}
  159. 			} catch (Exception e) {
  160. 				LOG.error("Critical failure while fetching/processing job archives.", e);
  161. 			}
  162. 			numFinishedPolls.countDown();
  163. 		}
  164. 	}
  165.  
  166. 	private static String convertLegacyJobOverview(String legacyOverview) throws IOException {
  167. 		JsonNode root = mapper.readTree(legacyOverview);
  168. 		JsonNode finishedJobs = root.get("finished");
  169. 		JsonNode job = finishedJobs.get(0);
  170.  
  171. 		JobID jobId = JobID.fromHexString(job.get("jid").asText());
  172. 		String name = job.get("name").asText();
  173. 		JobStatus state = JobStatus.valueOf(job.get("state").asText());
  174.  
  175. 		long startTime = job.get("start-time").asLong();
  176. 		long endTime = job.get("end-time").asLong();
  177. 		long duration = job.get("duration").asLong();
  178. 		long lastMod = job.get("last-modification").asLong();
  179.  
  180. 		JsonNode tasks = job.get("tasks");
  181. 		int numTasks = tasks.get("total").asInt();
  182. 		int pending = tasks.get("pending").asInt();
  183. 		int running = tasks.get("running").asInt();
  184. 		int finished = tasks.get("finished").asInt();
  185. 		int canceling = tasks.get("canceling").asInt();
  186. 		int canceled = tasks.get("canceled").asInt();
  187. 		int failed = tasks.get("failed").asInt();
  188.  
  189. 		int[] tasksPerState = new int[ExecutionState.values().length];
  190. 		// pending is a mix of CREATED/SCHEDULED/DEPLOYING
  191. 		// to maintain the correct number of task states we have to pick one of them
  192. 		tasksPerState[ExecutionState.SCHEDULED.ordinal()] = pending;
  193. 		tasksPerState[ExecutionState.RUNNING.ordinal()] = running;
  194. 		tasksPerState[ExecutionState.FINISHED.ordinal()] = finished;
  195. 		tasksPerState[ExecutionState.CANCELING.ordinal()] = canceling;
  196. 		tasksPerState[ExecutionState.CANCELED.ordinal()] = canceled;
  197. 		tasksPerState[ExecutionState.FAILED.ordinal()] = failed;
  198.  
  199. 		JobDetails jobDetails = new JobDetails(jobId, name, startTime, endTime, duration, state, lastMod, tasksPerState, numTasks);
  200. 		MultipleJobsDetails multipleJobsDetails = new MultipleJobsDetails(Collections.singleton(jobDetails));
  201.  
  202. 		StringWriter sw = new StringWriter();
  203. 		mapper.writeValue(sw, multipleJobsDetails);
  204. 		return sw.toString();
  205. 	}
  206.  
  207. 	/**
  208. 	 * This method replicates the JSON response that would be given by the JobsOverviewHandler when
  209. 	 * listing both running and finished jobs.
  210. 	 *
  211. 	 * <p>Every job archive contains a joboverview.json file containing the same structure. Since jobs are archived on
  212. 	 * their own however the list of finished jobs only contains a single job.
  213. 	 *
  214. 	 * <p>For the display in the HistoryServer WebFrontend we have to combine these overviews.
  215. 	 */
  216. 	private static void updateJobOverview(File webOverviewDir, File webDir) {
  217. 		try (JsonGenerator gen = jacksonFactory.createGenerator(HistoryServer.createOrGetFile(webDir, JobsOverviewHeaders.URL))) {
  218. 			File[] overviews = new File(webOverviewDir.getPath()).listFiles();
  219. 			if (overviews != null) {
  220. 				Collection<JobDetails> allJobs = new ArrayList<>(overviews.length);
  221. 				for (File overview : overviews) {
  222. 					MultipleJobsDetails subJobs = mapper.readValue(overview, MultipleJobsDetails.class);
  223. 					allJobs.addAll(subJobs.getJobs());
  224. 				}
  225. 				mapper.writeValue(gen, new MultipleJobsDetails(allJobs));
  226. 			}
  227. 		} catch (IOException ioe) {
  228. 			LOG.error("Failed to update job overview.", ioe);
  229. 		}
  230. 	}
  231. }
  232. 复制代码
  • 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

  1. public class WebFrontendBootstrap {
  2. 	private final Router router;
  3. 	private final Logger log;
  4. 	private final File uploadDir;
  5. 	private final ServerBootstrap bootstrap;
  6. 	private final Channel serverChannel;
  7. 	private final String restAddress;
  8.  
  9. 	public WebFrontendBootstrap(
  10. 			Router router,
  11. 			Logger log,
  12. 			File directory,
  13. 			@Nullable SSLHandlerFactory serverSSLFactory,
  14. 			String configuredAddress,
  15. 			int configuredPort,
  16. 			final Configuration config) throws InterruptedException, UnknownHostException {
  17.  
  18. 		this.router = Preconditions.checkNotNull(router);
  19. 		this.log = Preconditions.checkNotNull(log);
  20. 		this.uploadDir = directory;
  21.  
  22. 		ChannelInitializer<SocketChannel> initializer = new ChannelInitializer<SocketChannel>() {
  23.  
  24. 			@Override
  25. 			protected void initChannel(SocketChannel ch) {
  26. 				RouterHandler handler = new RouterHandler(WebFrontendBootstrap.this.router, new HashMap<>());
  27.  
  28. 				// SSL should be the first handler in the pipeline
  29. 				if (serverSSLFactory != null) {
  30. 					ch.pipeline().addLast("ssl", serverSSLFactory.createNettySSLHandler());
  31. 				}
  32.  
  33. 				ch.pipeline()
  34. 					.addLast(new HttpServerCodec())
  35. 					.addLast(new ChunkedWriteHandler())
  36. 					.addLast(new HttpRequestHandler(uploadDir))
  37. 					.addLast(handler.getName(), handler)
  38. 					.addLast(new PipelineErrorHandler(WebFrontendBootstrap.this.log));
  39. 			}
  40. 		};
  41.  
  42. 		NioEventLoopGroup bossGroup   = new NioEventLoopGroup(1);
  43. 		NioEventLoopGroup workerGroup = new NioEventLoopGroup();
  44.  
  45. 		this.bootstrap = new ServerBootstrap();
  46. 		this.bootstrap
  47. 			.group(bossGroup, workerGroup)
  48. 			.channel(NioServerSocketChannel.class)
  49. 			.childHandler(initializer);
  50.  
  51. 		ChannelFuture ch;
  52. 		if (configuredAddress == null) {
  53. 			ch = this.bootstrap.bind(configuredPort);
  54. 		} else {
  55. 			ch = this.bootstrap.bind(configuredAddress, configuredPort);
  56. 		}
  57. 		this.serverChannel = ch.sync().channel();
  58.  
  59. 		InetSocketAddress bindAddress = (InetSocketAddress) serverChannel.localAddress();
  60.  
  61. 		InetAddress inetAddress = bindAddress.getAddress();
  62. 		final String address;
  63.  
  64. 		if (inetAddress.isAnyLocalAddress()) {
  65. 			address = config.getString(JobManagerOptions.ADDRESS, InetAddress.getLocalHost().getHostName());
  66. 		} else {
  67. 			address = inetAddress.getHostAddress();
  68. 		}
  69.  
  70. 		int port = bindAddress.getPort();
  71.  
  72. 		this.log.info("Web frontend listening at {}" + ':' + "{}", address, port);
  73.  
  74. 		final String protocol = serverSSLFactory != null ? "https://" : "http://";
  75.  
  76. 		this.restAddress = protocol + address + ':' + port;
  77. 	}
  78.  
  79. 	public ServerBootstrap getBootstrap() {
  80. 		return bootstrap;
  81. 	}
  82.  
  83. 	public int getServerPort() {
  84. 		Channel server = this.serverChannel;
  85. 		if (server != null) {
  86. 			try {
  87. 				return ((InetSocketAddress) server.localAddress()).getPort();
  88. 			}
  89. 			catch (Exception e) {
  90. 				log.error("Cannot access local server port", e);
  91. 			}
  92. 		}
  93.  
  94. 		return -1;
  95. 	}
  96.  
  97. 	public String getRestAddress() {
  98. 		return restAddress;
  99. 	}
  100.  
  101. 	public void shutdown() {
  102. 		if (this.serverChannel != null) {
  103. 			this.serverChannel.close().awaitUninterruptibly();
  104. 		}
  105. 		if (bootstrap != null) {
  106. 			if (bootstrap.group() != null) {
  107. 				bootstrap.group().shutdownGracefully();
  108. 			}
  109. 			if (bootstrap.childGroup() != null) {
  110. 				bootstrap.childGroup().shutdownGracefully();
  111. 			}
  112. 		}
  113. 	}
  114. }
  115. 复制代码
  • 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

转载于:https://juejin.im/post/5c8482ce5188257a323f52b9

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/繁依Fanyi0/article/detail/558643
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号