LogAggregationFileControllerFactory { public LogAggregationFileControllerFactory(Configuration conf) { this.conf = conf; Collection<String> fileControllers = conf.getStringCollection( YarnConfiguration.LOG_AGGREGATION_FILE_FORMATS); List<String> controllerClassName = new ArrayList<>(); Map<String, String> controllerChecker = new HashMap<>(); for (String fileController : fileControllers) { Preconditions.checkArgument(validateAggregatedFileControllerName( fileController), "The FileControllerName: " + fileController + " set in " + YarnConfiguration.LOG_AGGREGATION_FILE_FORMATS +" is invalid." + "The valid File Controller name should only " + "contain a-zA-Z0-9_ and can not start with numbers"); String remoteDirStr = String.format( YarnConfiguration.LOG_AGGREGATION_REMOTE_APP_LOG_DIR_FMT, fileController); String remoteDir = conf.get(remoteDirStr); boolean defaultRemoteDir = false; if (remoteDir == null || remoteDir.isEmpty()) { remoteDir = conf.get(YarnConfiguration.NM_REMOTE_APP_LOG_DIR, YarnConfiguration.DEFAULT_NM_REMOTE_APP_LOG_DIR); defaultRemoteDir = true; } String suffixStr = String.format( YarnConfiguration.LOG_AGGREGATION_REMOTE_APP_LOG_DIR_SUFFIX_FMT, fileController); String suffix = conf.get(suffixStr); boolean defaultSuffix = false; if (suffix == null || suffix.isEmpty()) { suffix = conf.get(YarnConfiguration.NM_REMOTE_APP_LOG_DIR_SUFFIX, YarnConfiguration.DEFAULT_NM_REMOTE_APP_LOG_DIR_SUFFIX); defaultSuffix = true; } String dirSuffix = remoteDir + "-" + suffix; if (controllerChecker.containsKey(dirSuffix)) { if (defaultRemoteDir && defaultSuffix) { String fileControllerStr = controllerChecker.get(dirSuffix); List<String> controllersList = new ArrayList<>(); controllersList.add(fileControllerStr); controllersList.add(fileController); fileControllerStr = StringUtils.join(controllersList, ","); controllerChecker.put(dirSuffix, fileControllerStr); } else { String conflictController = controllerChecker.get(dirSuffix); throw new RuntimeException("The combined value of " + remoteDirStr + " and " + suffixStr + " should not be the same as the value" + " set for " + conflictController); } } else { controllerChecker.put(dirSuffix, fileController); } String classKey = String.format( YarnConfiguration.LOG_AGGREGATION_FILE_CONTROLLER_FMT, fileController); String className = conf.get(classKey); if (className == null || className.isEmpty()) { throw new RuntimeException("No class configured for " + fileController); } controllerClassName.add(className); Class<? extends LogAggregationFileController> sClass = conf.getClass( classKey, null, LogAggregationFileController.class); if (sClass == null) { throw new RuntimeException("No class defined for " + fileController); } LogAggregationFileController s = ReflectionUtils.newInstance( sClass, conf); if (s == null) { throw new RuntimeException("No object created for " + controllerClassName); } s.initialize(conf, fileController); controllers.add(s); } } LogAggregationFileControllerFactory(Configuration conf); LogAggregationFileController getFileControllerForWrite(); LogAggregationFileController getFileControllerForRead(
      ApplicationId appId, String appOwner); @Private @VisibleForTesting LinkedList<LogAggregationFileController> getConfiguredLogAggregationFileControllerList();  }
@Test(timeout = 10000) public void testLogAggregationFileControllerFactory() throws Exception { ApplicationId appId = ApplicationId.newInstance( System.currentTimeMillis(), 1); String appOwner = "test"; String remoteLogRootDir = "target/app-logs/"; Configuration conf = new YarnConfiguration(); conf.setBoolean(YarnConfiguration.LOG_AGGREGATION_ENABLED, true); conf.set(YarnConfiguration.NM_REMOTE_APP_LOG_DIR, remoteLogRootDir); conf.set(YarnConfiguration.NM_REMOTE_APP_LOG_DIR_SUFFIX, "log"); FileSystem fs = FileSystem.get(conf); LogAggregationFileControllerFactory factory = new LogAggregationFileControllerFactory(conf); LinkedList<LogAggregationFileController> list = factory .getConfiguredLogAggregationFileControllerList(); assertTrue(list.size() == 1); assertTrue(list.getFirst() instanceof LogAggregationTFileController); assertTrue(factory.getFileControllerForWrite() instanceof LogAggregationTFileController); Path logPath = list.getFirst().getRemoteAppLogDir(appId, appOwner); try { if (fs.exists(logPath)) { fs.delete(logPath, true); } assertTrue(fs.mkdirs(logPath)); Writer writer = new FileWriter(new File(logPath.toString(), "testLog")); writer.write("test"); writer.close(); assertTrue(factory.getFileControllerForRead(appId, appOwner) instanceof LogAggregationTFileController); } finally { fs.delete(logPath, true); } conf.set(YarnConfiguration.LOG_AGGREGATION_FILE_FORMATS, "TestLogAggregationFileController"); try { factory = new LogAggregationFileControllerFactory(conf); fail(); } catch (Exception ex) { } conf.set(YarnConfiguration.LOG_AGGREGATION_FILE_FORMATS, "TestLogAggregationFileController,TFile"); conf.setClass( "yarn.log-aggregation.file-controller.TestLogAggregationFileController" + ".class", TestLogAggregationFileController.class, LogAggregationFileController.class); conf.set( "yarn.log-aggregation.TestLogAggregationFileController" + ".remote-app-log-dir", remoteLogRootDir); conf.set( "yarn.log-aggregation.TestLogAggregationFileController" + ".remote-app-log-dir-suffix", "testLog"); factory = new LogAggregationFileControllerFactory(conf); list = factory.getConfiguredLogAggregationFileControllerList(); assertTrue(list.size() == 2); assertTrue(list.getFirst() instanceof TestLogAggregationFileController); assertTrue(list.getLast() instanceof LogAggregationTFileController); assertTrue(factory.getFileControllerForWrite() instanceof TestLogAggregationFileController); logPath = list.getFirst().getRemoteAppLogDir(appId, appOwner); try { if (fs.exists(logPath)) { fs.delete(logPath, true); } assertTrue(fs.mkdirs(logPath)); Writer writer = new FileWriter(new File(logPath.toString(), "testLog")); writer.write("test"); writer.close(); assertTrue(factory.getFileControllerForRead(appId, appOwner) instanceof TestLogAggregationFileController); } finally { fs.delete(logPath, true); } }