@Test public void testConfiguratorLoadConfigsAndReceiveLiveChanges() throws Exception { dumpServerSetFiles(); File singerConfigFile = createSingerConfigFile(makeDirectorySingerConfigProperties()); createLogConfigPropertiesFile("ads.mohawk.properties", ImmutableMap.of( "writer.kafka.producerConfig.bootstrap.servers", "127.0.0.1:9092")); createLogConfigPropertiesFile("search.discovery.properties", ImmutableMap.of("writer.kafka.producerConfig.bootstrap.servers", "127.0.0.1:9092")); DirectorySingerConfigurator configurator = new DirectorySingerConfigurator(singerConfigFile .getParent()); SingerConfig singerConfig = configurator.parseSingerConfig(); final AtomicInteger exitCode = new AtomicInteger(-1); new SingerDirectoryWatcher(configurator.parseSingerConfig() , configurator, new SingerDirectoryWatcher.ExitManager() { @Override public void exit(int status) { exitCode.set(status); } }); assertEquals(2, singerConfig.getLogConfigsSize()); SingerLogConfig adsMohawk = singerConfig.logConfigs.get(0); assertEquals("ads.mohawk", adsMohawk.getName()); assertEquals("/mnt/log/singer", adsMohawk.getLogDir()); SingerLogConfig search = singerConfig.logConfigs.get(1); assertEquals("search.discovery", search.getName()); assertEquals("/mnt/log/singer", search.getLogDir()); { createLogConfigPropertiesFile("ads.ads.properties", ImmutableMap.of( "writer.kafka.producerConfig.bootstrap.servers", "127.0.0.1:9092")); Thread.sleep(2000); assertEquals(0, exitCode.get()); DirectorySingerConfigurator newConfigurator = new DirectorySingerConfigurator (singerConfigFile.getParent()); assertEquals(3, newConfigurator.parseSingerConfig().getLogConfigsSize()); } exitCode.set(-1); { createLogConfigPropertiesFile("ads.mohawk.properties", ImmutableMap.of("processor.batchSize", "300", "writer.kafka.producerConfig.bootstrap.servers", "127.0.0.1:9092")); Thread.sleep(2000); assertEquals(0, exitCode.get()); } }
@Override public SingerConfig parseSingerConfig() throws ConfigurationException { Preconditions.checkArgument(theDir.isDirectory() && theDir.exists(), "Singer configure directory %s is not an existing file directory", theDir.getPath()); SingerConfig singerConfig = LogConfigUtils.parseDirBasedSingerConfigHeader(FilenameUtils.concat( theDir.getPath(), SINGER_CONFIGURATION_FILE)); File logConfigDir = new File(FilenameUtils.concat(theDir.getPath(), SINGER_LOG_CONFIG_DIR)); Preconditions.checkArgument(logConfigDir.exists() && logConfigDir.isDirectory(), "%s is not a valid directory", logConfigDir.getPath()); FileFilter fileFilter = new RegexFileFilter(CONFIG_FILE_PATTERN); File[] files = logConfigDir.listFiles(fileFilter); Arrays.sort(files); for (File newFile : files) { try { LOG.info("Attempting to parse log config file:" + newFile.getAbsolutePath()); singerConfig.addToLogConfigs(LogConfigUtils.parseLogConfigFromFile(newFile)); } catch (ConfigurationException e) { Stats.incr(SingerMetrics.SINGER_CONFIGURATOR_CONFIG_ERRORS); LOG.error("Failed to parse Singer client config file {}, exception: {}. Skip and continue.", newFile.getPath(), ExceptionUtils.getFullStackTrace(e)); } } boolean useNewConfig = (System.getProperty("useNewConfig") != null); if(useNewConfig){ String newLogConfig; try { newLogConfig = getNewConfig(); } catch (Exception e) { LOG.info("Failed to get {} from S3, exception: {}. Skip and continue.", String.format("%s/%s", DATAPIPELINES_CONFIG_S3_BUCKET, DATAPIPELINES_CONFIG_S3_KEY), ExceptionUtils.getFullStackTrace(e)); return singerConfig; } try { SingerLogConfig[] logConfigs = LogConfigUtils.parseLogStreamConfigFromFile(newLogConfig); for (SingerLogConfig logConfig : logConfigs) { singerConfig.addToLogConfigs(logConfig); } } catch (ConfigurationException e) { Stats.incr(SingerMetrics.SINGER_CONFIGURATOR_CONFIG_ERRORS); LOG.info("Failed to parse Singer client config file {}, exception: {}. Terminating", DATAPIPELINES_CONFIG, ExceptionUtils.getFullStackTrace(e)); } catch (Exception e) { Stats.incr(SingerMetrics.SINGER_CONFIGURATOR_CONFIG_ERRORS_UNKNOWN); LOG.error("Error parsing configuration from file:" + newLogConfig, e); } } return singerConfig; }
DirectorySingerConfigurator implements SingerConfigurator { @Override public SingerConfig parseSingerConfig() throws ConfigurationException { Preconditions.checkArgument(theDir.isDirectory() && theDir.exists(), "Singer configure directory %s is not an existing file directory", theDir.getPath()); SingerConfig singerConfig = LogConfigUtils.parseDirBasedSingerConfigHeader(FilenameUtils.concat( theDir.getPath(), SINGER_CONFIGURATION_FILE)); File logConfigDir = new File(FilenameUtils.concat(theDir.getPath(), SINGER_LOG_CONFIG_DIR)); Preconditions.checkArgument(logConfigDir.exists() && logConfigDir.isDirectory(), "%s is not a valid directory", logConfigDir.getPath()); FileFilter fileFilter = new RegexFileFilter(CONFIG_FILE_PATTERN); File[] files = logConfigDir.listFiles(fileFilter); Arrays.sort(files); for (File newFile : files) { try { LOG.info("Attempting to parse log config file:" + newFile.getAbsolutePath()); singerConfig.addToLogConfigs(LogConfigUtils.parseLogConfigFromFile(newFile)); } catch (ConfigurationException e) { Stats.incr(SingerMetrics.SINGER_CONFIGURATOR_CONFIG_ERRORS); LOG.error("Failed to parse Singer client config file {}, exception: {}. Skip and continue.", newFile.getPath(), ExceptionUtils.getFullStackTrace(e)); } } boolean useNewConfig = (System.getProperty("useNewConfig") != null); if(useNewConfig){ String newLogConfig; try { newLogConfig = getNewConfig(); } catch (Exception e) { LOG.info("Failed to get {} from S3, exception: {}. Skip and continue.", String.format("%s/%s", DATAPIPELINES_CONFIG_S3_BUCKET, DATAPIPELINES_CONFIG_S3_KEY), ExceptionUtils.getFullStackTrace(e)); return singerConfig; } try { SingerLogConfig[] logConfigs = LogConfigUtils.parseLogStreamConfigFromFile(newLogConfig); for (SingerLogConfig logConfig : logConfigs) { singerConfig.addToLogConfigs(logConfig); } } catch (ConfigurationException e) { Stats.incr(SingerMetrics.SINGER_CONFIGURATOR_CONFIG_ERRORS); LOG.info("Failed to parse Singer client config file {}, exception: {}. Terminating", DATAPIPELINES_CONFIG, ExceptionUtils.getFullStackTrace(e)); } catch (Exception e) { Stats.incr(SingerMetrics.SINGER_CONFIGURATOR_CONFIG_ERRORS_UNKNOWN); LOG.error("Error parsing configuration from file:" + newLogConfig, e); } } return singerConfig; } }
DirectorySingerConfigurator implements SingerConfigurator { @Override public SingerConfig parseSingerConfig() throws ConfigurationException { Preconditions.checkArgument(theDir.isDirectory() && theDir.exists(), "Singer configure directory %s is not an existing file directory", theDir.getPath()); SingerConfig singerConfig = LogConfigUtils.parseDirBasedSingerConfigHeader(FilenameUtils.concat( theDir.getPath(), SINGER_CONFIGURATION_FILE)); File logConfigDir = new File(FilenameUtils.concat(theDir.getPath(), SINGER_LOG_CONFIG_DIR)); Preconditions.checkArgument(logConfigDir.exists() && logConfigDir.isDirectory(), "%s is not a valid directory", logConfigDir.getPath()); FileFilter fileFilter = new RegexFileFilter(CONFIG_FILE_PATTERN); File[] files = logConfigDir.listFiles(fileFilter); Arrays.sort(files); for (File newFile : files) { try { LOG.info("Attempting to parse log config file:" + newFile.getAbsolutePath()); singerConfig.addToLogConfigs(LogConfigUtils.parseLogConfigFromFile(newFile)); } catch (ConfigurationException e) { Stats.incr(SingerMetrics.SINGER_CONFIGURATOR_CONFIG_ERRORS); LOG.error("Failed to parse Singer client config file {}, exception: {}. Skip and continue.", newFile.getPath(), ExceptionUtils.getFullStackTrace(e)); } } boolean useNewConfig = (System.getProperty("useNewConfig") != null); if(useNewConfig){ String newLogConfig; try { newLogConfig = getNewConfig(); } catch (Exception e) { LOG.info("Failed to get {} from S3, exception: {}. Skip and continue.", String.format("%s/%s", DATAPIPELINES_CONFIG_S3_BUCKET, DATAPIPELINES_CONFIG_S3_KEY), ExceptionUtils.getFullStackTrace(e)); return singerConfig; } try { SingerLogConfig[] logConfigs = LogConfigUtils.parseLogStreamConfigFromFile(newLogConfig); for (SingerLogConfig logConfig : logConfigs) { singerConfig.addToLogConfigs(logConfig); } } catch (ConfigurationException e) { Stats.incr(SingerMetrics.SINGER_CONFIGURATOR_CONFIG_ERRORS); LOG.info("Failed to parse Singer client config file {}, exception: {}. Terminating", DATAPIPELINES_CONFIG, ExceptionUtils.getFullStackTrace(e)); } catch (Exception e) { Stats.incr(SingerMetrics.SINGER_CONFIGURATOR_CONFIG_ERRORS_UNKNOWN); LOG.error("Error parsing configuration from file:" + newLogConfig, e); } } return singerConfig; } DirectorySingerConfigurator(String singerConfigDir); }
DirectorySingerConfigurator implements SingerConfigurator { @Override public SingerConfig parseSingerConfig() throws ConfigurationException { Preconditions.checkArgument(theDir.isDirectory() && theDir.exists(), "Singer configure directory %s is not an existing file directory", theDir.getPath()); SingerConfig singerConfig = LogConfigUtils.parseDirBasedSingerConfigHeader(FilenameUtils.concat( theDir.getPath(), SINGER_CONFIGURATION_FILE)); File logConfigDir = new File(FilenameUtils.concat(theDir.getPath(), SINGER_LOG_CONFIG_DIR)); Preconditions.checkArgument(logConfigDir.exists() && logConfigDir.isDirectory(), "%s is not a valid directory", logConfigDir.getPath()); FileFilter fileFilter = new RegexFileFilter(CONFIG_FILE_PATTERN); File[] files = logConfigDir.listFiles(fileFilter); Arrays.sort(files); for (File newFile : files) { try { LOG.info("Attempting to parse log config file:" + newFile.getAbsolutePath()); singerConfig.addToLogConfigs(LogConfigUtils.parseLogConfigFromFile(newFile)); } catch (ConfigurationException e) { Stats.incr(SingerMetrics.SINGER_CONFIGURATOR_CONFIG_ERRORS); LOG.error("Failed to parse Singer client config file {}, exception: {}. Skip and continue.", newFile.getPath(), ExceptionUtils.getFullStackTrace(e)); } } boolean useNewConfig = (System.getProperty("useNewConfig") != null); if(useNewConfig){ String newLogConfig; try { newLogConfig = getNewConfig(); } catch (Exception e) { LOG.info("Failed to get {} from S3, exception: {}. Skip and continue.", String.format("%s/%s", DATAPIPELINES_CONFIG_S3_BUCKET, DATAPIPELINES_CONFIG_S3_KEY), ExceptionUtils.getFullStackTrace(e)); return singerConfig; } try { SingerLogConfig[] logConfigs = LogConfigUtils.parseLogStreamConfigFromFile(newLogConfig); for (SingerLogConfig logConfig : logConfigs) { singerConfig.addToLogConfigs(logConfig); } } catch (ConfigurationException e) { Stats.incr(SingerMetrics.SINGER_CONFIGURATOR_CONFIG_ERRORS); LOG.info("Failed to parse Singer client config file {}, exception: {}. Terminating", DATAPIPELINES_CONFIG, ExceptionUtils.getFullStackTrace(e)); } catch (Exception e) { Stats.incr(SingerMetrics.SINGER_CONFIGURATOR_CONFIG_ERRORS_UNKNOWN); LOG.error("Error parsing configuration from file:" + newLogConfig, e); } } return singerConfig; } DirectorySingerConfigurator(String singerConfigDir); @Override SingerConfig parseSingerConfig(); }
DirectorySingerConfigurator implements SingerConfigurator { @Override public SingerConfig parseSingerConfig() throws ConfigurationException { Preconditions.checkArgument(theDir.isDirectory() && theDir.exists(), "Singer configure directory %s is not an existing file directory", theDir.getPath()); SingerConfig singerConfig = LogConfigUtils.parseDirBasedSingerConfigHeader(FilenameUtils.concat( theDir.getPath(), SINGER_CONFIGURATION_FILE)); File logConfigDir = new File(FilenameUtils.concat(theDir.getPath(), SINGER_LOG_CONFIG_DIR)); Preconditions.checkArgument(logConfigDir.exists() && logConfigDir.isDirectory(), "%s is not a valid directory", logConfigDir.getPath()); FileFilter fileFilter = new RegexFileFilter(CONFIG_FILE_PATTERN); File[] files = logConfigDir.listFiles(fileFilter); Arrays.sort(files); for (File newFile : files) { try { LOG.info("Attempting to parse log config file:" + newFile.getAbsolutePath()); singerConfig.addToLogConfigs(LogConfigUtils.parseLogConfigFromFile(newFile)); } catch (ConfigurationException e) { Stats.incr(SingerMetrics.SINGER_CONFIGURATOR_CONFIG_ERRORS); LOG.error("Failed to parse Singer client config file {}, exception: {}. Skip and continue.", newFile.getPath(), ExceptionUtils.getFullStackTrace(e)); } } boolean useNewConfig = (System.getProperty("useNewConfig") != null); if(useNewConfig){ String newLogConfig; try { newLogConfig = getNewConfig(); } catch (Exception e) { LOG.info("Failed to get {} from S3, exception: {}. Skip and continue.", String.format("%s/%s", DATAPIPELINES_CONFIG_S3_BUCKET, DATAPIPELINES_CONFIG_S3_KEY), ExceptionUtils.getFullStackTrace(e)); return singerConfig; } try { SingerLogConfig[] logConfigs = LogConfigUtils.parseLogStreamConfigFromFile(newLogConfig); for (SingerLogConfig logConfig : logConfigs) { singerConfig.addToLogConfigs(logConfig); } } catch (ConfigurationException e) { Stats.incr(SingerMetrics.SINGER_CONFIGURATOR_CONFIG_ERRORS); LOG.info("Failed to parse Singer client config file {}, exception: {}. Terminating", DATAPIPELINES_CONFIG, ExceptionUtils.getFullStackTrace(e)); } catch (Exception e) { Stats.incr(SingerMetrics.SINGER_CONFIGURATOR_CONFIG_ERRORS_UNKNOWN); LOG.error("Error parsing configuration from file:" + newLogConfig, e); } } return singerConfig; } DirectorySingerConfigurator(String singerConfigDir); @Override SingerConfig parseSingerConfig(); static final String SINGER_LOG_CONFIG_PATH; @VisibleForTesting
  static final String SINGER_CONFIGURATION_FILE; @VisibleForTesting
  static final String SINGER_LOG_CONFIG_DIR; @VisibleForTesting
  static final String DATAPIPELINES_CONFIG; static final String CONFIG_FILE_PATTERN; static final String DATAPIPELINES_CONFIG_S3_BUCKET; static final String DATAPIPELINES_CONFIG_S3_KEY; }