@Test public void testValidateConfigChanges() throws Exception { File singerConfigFile = createSingerConfigFile(makeWrongSingerConfigProperties(4)); DirectorySingerConfigurator configurator = new DirectorySingerConfigurator(singerConfigFile.getParent()); try { configurator.parseSingerConfig(); assertEquals(true,false); } catch (ConfigurationException x) { assertEquals(SingerConfigDef.MONITOR_INTERVAL_IN_SECS + IS_MISSING_ELEMENT, x.getMessage()); } File singerConfigFileTwo = createSingerConfigFile(makeWrongSingerConfigProperties(5)); DirectorySingerConfigurator configuratorTwo = new DirectorySingerConfigurator(singerConfigFileTwo .getParent()); try { configuratorTwo.parseSingerConfig(); assertEquals(true,false); } catch (ConfigurationException x) { assertEquals("singer.setLogConfigPollIntervalSecs" + IS_MISSING_ELEMENT, x.getMessage()); } File singerConfigFileThree = createSingerConfigFile(makeWrongSingerConfigProperties(3)); DirectorySingerConfigurator configuratorThree = new DirectorySingerConfigurator(singerConfigFileThree .getParent()); try { configuratorThree.parseSingerConfig(); assertEquals(true,false); } catch (ConfigurationException x) { assertEquals("monitorIntervalInSecs" + IS_MISSING_ELEMENT + "\n" + "singer.setLogConfigPollIntervalSecs" + IS_MISSING_ELEMENT, x.getMessage()); } }
@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; }