@Test public void testGetMeasurementsLike() throws Exception { StorageEngine engine = new MemStorageEngine(); engine.configure(conf, bgTasks); engine.writeDataPointWithLock(MiscUtils.buildDataPoint("test", "cpu", Arrays.asList("value"), Arrays.asList(Tag.newBuilder().setTagKey("test").setTagValue("2").build()), System.currentTimeMillis(), Arrays.asList(2L), Arrays.asList(false)), false); engine.writeDataPointWithLock(MiscUtils.buildDataPoint("test", "mem", Arrays.asList("value"), Arrays.asList(Tag.newBuilder().setTagKey("test").setTagValue("2").build()), System.currentTimeMillis() + 10, Arrays.asList(3L), Arrays.asList(false)), false); engine.writeDataPointWithLock(MiscUtils.buildDataPoint("test", "netm", Arrays.asList("value"), Arrays.asList(Tag.newBuilder().setTagKey("test").setTagValue("2").build()), System.currentTimeMillis() + 20, Arrays.asList(5L), Arrays.asList(false)), false); Set<String> result = engine.getMeasurementsLike("test", " "); assertEquals(3, result.size()); result = engine.getMeasurementsLike("test", "c.*"); assertEquals(1, result.size()); result = engine.getMeasurementsLike("test", ".*m.*"); assertEquals(2, result.size()); }
@Override public void configure(Map<String, String> conf, ScheduledExecutorService bgTaskPool) throws IOException { this.conf = conf; this.bgTaskPool = bgTaskPool; this.defaultRetentionHours = Integer .parseInt(conf.getOrDefault(RETENTION_HOURS, String.valueOf(DEFAULT_RETENTION_HOURS))); logger.info("Setting default timeseries retention hours policy to:" + defaultRetentionHours); databaseMap = new ConcurrentHashMap<>(); dbMetadataMap = new ConcurrentHashMap<>(); setCodecsForCompression(conf); try { archiver = (Archiver) Class.forName(conf.getOrDefault("archiver.class", NoneArchiver.class.getName())) .newInstance(); archiver.init(conf); } catch (InstantiationException | IllegalAccessException | ClassNotFoundException e) { logger.log(Level.SEVERE, "Failed to instantiate archiver", e); } this.defaultTimebucketSize = Integer .parseInt(conf.getOrDefault(DEFAULT_BUCKET_SIZE, String.valueOf(DEFAULT_TIME_BUCKET_CONSTANT))); if (bgTaskPool != null) { bgTaskPool.scheduleAtFixedRate(() -> { for (Entry<String, Map<String, Measurement>> measurementMap : databaseMap.entrySet()) { for (Entry<String, Measurement> measurementEntry : measurementMap.getValue().entrySet()) { Measurement value = measurementEntry.getValue(); try { } catch (Exception e) { logger.log(Level.SEVERE, "Failed collect garbage for measurement:" + value.getMeasurementName(), e); } } } }, Integer.parseInt(conf.getOrDefault(GC_FREQUENCY, DEFAULT_GC_FREQUENCY)), Integer.parseInt(conf.getOrDefault(GC_DELAY, DEFAULT_GC_DELAY)), TimeUnit.SECONDS); if (Boolean.parseBoolean(conf.getOrDefault(StorageEngine.COMPACTION_ENABLED, "false"))) { logger.info("Compaction is enabled"); bgTaskPool.scheduleAtFixedRate(() -> { for (Entry<String, Map<String, Measurement>> measurementMap : databaseMap.entrySet()) { for (Entry<String, Measurement> measurementEntry : measurementMap.getValue().entrySet()) { Measurement value = measurementEntry.getValue(); try { value.compact(); } catch (Exception e) { logger.log(Level.SEVERE, "Failed compaction for measurement:" + value.getMeasurementName(), e); } } } }, Integer.parseInt(conf.getOrDefault(COMPACTION_FREQUENCY, DEFAULT_COMPACTION_FREQUENCY)), Integer.parseInt(conf.getOrDefault(COMPACTION_DELAY, DEFAULT_COMPACTION_DELAY)), TimeUnit.SECONDS); } else { logger.warning("Compaction is disabled"); } } enableMetricsService(); }
MemStorageEngine implements StorageEngine { @Override public void configure(Map<String, String> conf, ScheduledExecutorService bgTaskPool) throws IOException { this.conf = conf; this.bgTaskPool = bgTaskPool; this.defaultRetentionHours = Integer .parseInt(conf.getOrDefault(RETENTION_HOURS, String.valueOf(DEFAULT_RETENTION_HOURS))); logger.info("Setting default timeseries retention hours policy to:" + defaultRetentionHours); databaseMap = new ConcurrentHashMap<>(); dbMetadataMap = new ConcurrentHashMap<>(); setCodecsForCompression(conf); try { archiver = (Archiver) Class.forName(conf.getOrDefault("archiver.class", NoneArchiver.class.getName())) .newInstance(); archiver.init(conf); } catch (InstantiationException | IllegalAccessException | ClassNotFoundException e) { logger.log(Level.SEVERE, "Failed to instantiate archiver", e); } this.defaultTimebucketSize = Integer .parseInt(conf.getOrDefault(DEFAULT_BUCKET_SIZE, String.valueOf(DEFAULT_TIME_BUCKET_CONSTANT))); if (bgTaskPool != null) { bgTaskPool.scheduleAtFixedRate(() -> { for (Entry<String, Map<String, Measurement>> measurementMap : databaseMap.entrySet()) { for (Entry<String, Measurement> measurementEntry : measurementMap.getValue().entrySet()) { Measurement value = measurementEntry.getValue(); try { } catch (Exception e) { logger.log(Level.SEVERE, "Failed collect garbage for measurement:" + value.getMeasurementName(), e); } } } }, Integer.parseInt(conf.getOrDefault(GC_FREQUENCY, DEFAULT_GC_FREQUENCY)), Integer.parseInt(conf.getOrDefault(GC_DELAY, DEFAULT_GC_DELAY)), TimeUnit.SECONDS); if (Boolean.parseBoolean(conf.getOrDefault(StorageEngine.COMPACTION_ENABLED, "false"))) { logger.info("Compaction is enabled"); bgTaskPool.scheduleAtFixedRate(() -> { for (Entry<String, Map<String, Measurement>> measurementMap : databaseMap.entrySet()) { for (Entry<String, Measurement> measurementEntry : measurementMap.getValue().entrySet()) { Measurement value = measurementEntry.getValue(); try { value.compact(); } catch (Exception e) { logger.log(Level.SEVERE, "Failed compaction for measurement:" + value.getMeasurementName(), e); } } } }, Integer.parseInt(conf.getOrDefault(COMPACTION_FREQUENCY, DEFAULT_COMPACTION_FREQUENCY)), Integer.parseInt(conf.getOrDefault(COMPACTION_DELAY, DEFAULT_COMPACTION_DELAY)), TimeUnit.SECONDS); } else { logger.warning("Compaction is disabled"); } } enableMetricsService(); } }
MemStorageEngine implements StorageEngine { @Override public void configure(Map<String, String> conf, ScheduledExecutorService bgTaskPool) throws IOException { this.conf = conf; this.bgTaskPool = bgTaskPool; this.defaultRetentionHours = Integer .parseInt(conf.getOrDefault(RETENTION_HOURS, String.valueOf(DEFAULT_RETENTION_HOURS))); logger.info("Setting default timeseries retention hours policy to:" + defaultRetentionHours); databaseMap = new ConcurrentHashMap<>(); dbMetadataMap = new ConcurrentHashMap<>(); setCodecsForCompression(conf); try { archiver = (Archiver) Class.forName(conf.getOrDefault("archiver.class", NoneArchiver.class.getName())) .newInstance(); archiver.init(conf); } catch (InstantiationException | IllegalAccessException | ClassNotFoundException e) { logger.log(Level.SEVERE, "Failed to instantiate archiver", e); } this.defaultTimebucketSize = Integer .parseInt(conf.getOrDefault(DEFAULT_BUCKET_SIZE, String.valueOf(DEFAULT_TIME_BUCKET_CONSTANT))); if (bgTaskPool != null) { bgTaskPool.scheduleAtFixedRate(() -> { for (Entry<String, Map<String, Measurement>> measurementMap : databaseMap.entrySet()) { for (Entry<String, Measurement> measurementEntry : measurementMap.getValue().entrySet()) { Measurement value = measurementEntry.getValue(); try { } catch (Exception e) { logger.log(Level.SEVERE, "Failed collect garbage for measurement:" + value.getMeasurementName(), e); } } } }, Integer.parseInt(conf.getOrDefault(GC_FREQUENCY, DEFAULT_GC_FREQUENCY)), Integer.parseInt(conf.getOrDefault(GC_DELAY, DEFAULT_GC_DELAY)), TimeUnit.SECONDS); if (Boolean.parseBoolean(conf.getOrDefault(StorageEngine.COMPACTION_ENABLED, "false"))) { logger.info("Compaction is enabled"); bgTaskPool.scheduleAtFixedRate(() -> { for (Entry<String, Map<String, Measurement>> measurementMap : databaseMap.entrySet()) { for (Entry<String, Measurement> measurementEntry : measurementMap.getValue().entrySet()) { Measurement value = measurementEntry.getValue(); try { value.compact(); } catch (Exception e) { logger.log(Level.SEVERE, "Failed compaction for measurement:" + value.getMeasurementName(), e); } } } }, Integer.parseInt(conf.getOrDefault(COMPACTION_FREQUENCY, DEFAULT_COMPACTION_FREQUENCY)), Integer.parseInt(conf.getOrDefault(COMPACTION_DELAY, DEFAULT_COMPACTION_DELAY)), TimeUnit.SECONDS); } else { logger.warning("Compaction is disabled"); } } enableMetricsService(); }  }
MemStorageEngine implements StorageEngine { @Override public void configure(Map<String, String> conf, ScheduledExecutorService bgTaskPool) throws IOException { this.conf = conf; this.bgTaskPool = bgTaskPool; this.defaultRetentionHours = Integer .parseInt(conf.getOrDefault(RETENTION_HOURS, String.valueOf(DEFAULT_RETENTION_HOURS))); logger.info("Setting default timeseries retention hours policy to:" + defaultRetentionHours); databaseMap = new ConcurrentHashMap<>(); dbMetadataMap = new ConcurrentHashMap<>(); setCodecsForCompression(conf); try { archiver = (Archiver) Class.forName(conf.getOrDefault("archiver.class", NoneArchiver.class.getName())) .newInstance(); archiver.init(conf); } catch (InstantiationException | IllegalAccessException | ClassNotFoundException e) { logger.log(Level.SEVERE, "Failed to instantiate archiver", e); } this.defaultTimebucketSize = Integer .parseInt(conf.getOrDefault(DEFAULT_BUCKET_SIZE, String.valueOf(DEFAULT_TIME_BUCKET_CONSTANT))); if (bgTaskPool != null) { bgTaskPool.scheduleAtFixedRate(() -> { for (Entry<String, Map<String, Measurement>> measurementMap : databaseMap.entrySet()) { for (Entry<String, Measurement> measurementEntry : measurementMap.getValue().entrySet()) { Measurement value = measurementEntry.getValue(); try { } catch (Exception e) { logger.log(Level.SEVERE, "Failed collect garbage for measurement:" + value.getMeasurementName(), e); } } } }, Integer.parseInt(conf.getOrDefault(GC_FREQUENCY, DEFAULT_GC_FREQUENCY)), Integer.parseInt(conf.getOrDefault(GC_DELAY, DEFAULT_GC_DELAY)), TimeUnit.SECONDS); if (Boolean.parseBoolean(conf.getOrDefault(StorageEngine.COMPACTION_ENABLED, "false"))) { logger.info("Compaction is enabled"); bgTaskPool.scheduleAtFixedRate(() -> { for (Entry<String, Map<String, Measurement>> measurementMap : databaseMap.entrySet()) { for (Entry<String, Measurement> measurementEntry : measurementMap.getValue().entrySet()) { Measurement value = measurementEntry.getValue(); try { value.compact(); } catch (Exception e) { logger.log(Level.SEVERE, "Failed compaction for measurement:" + value.getMeasurementName(), e); } } } }, Integer.parseInt(conf.getOrDefault(COMPACTION_FREQUENCY, DEFAULT_COMPACTION_FREQUENCY)), Integer.parseInt(conf.getOrDefault(COMPACTION_DELAY, DEFAULT_COMPACTION_DELAY)), TimeUnit.SECONDS); } else { logger.warning("Compaction is disabled"); } } enableMetricsService(); }  @Override void configure(Map<String, String> conf, ScheduledExecutorService bgTaskPool); void enableMetricsService(); @Override void updateTimeSeriesRetentionPolicy(String dbName, String measurementName, int retentionHours); @Override void updateTimeSeriesRetentionPolicy(String dbName, int retentionHours); @Override Map<String, Measurement> getOrCreateDatabase(String dbName); @Override Map<String, Measurement> getOrCreateDatabase(String dbName, int retentionPolicy, Map<String, String> conf); @Override Measurement getOrCreateMeasurement(String dbName, String measurementName); @Override boolean isMeasurementFieldFP(String dbName, String measurementName, String valueFieldName); @Override Set<String> getDatabases(); @Override void deleteAllData(); @Override boolean checkIfExists(String dbName); @Override void dropDatabase(String dbName); @Override void dropMeasurement(String dbName, String measurementName); @Override void startup(); @Override void shutdown(); Map<String, DBMetadata> getDbMetadataMap(); @Override Map<String, Map<String, Measurement>> getMeasurementMap(); @Override Map<String, Map<String, Measurement>> getDatabaseMap(); @Override int getDefaultTimebucketSize(); @Override Counter getCounter(); @Override Logger getLogger(); @Override Map<String, String> getConf(); }
MemStorageEngine implements StorageEngine { @Override public void configure(Map<String, String> conf, ScheduledExecutorService bgTaskPool) throws IOException { this.conf = conf; this.bgTaskPool = bgTaskPool; this.defaultRetentionHours = Integer .parseInt(conf.getOrDefault(RETENTION_HOURS, String.valueOf(DEFAULT_RETENTION_HOURS))); logger.info("Setting default timeseries retention hours policy to:" + defaultRetentionHours); databaseMap = new ConcurrentHashMap<>(); dbMetadataMap = new ConcurrentHashMap<>(); setCodecsForCompression(conf); try { archiver = (Archiver) Class.forName(conf.getOrDefault("archiver.class", NoneArchiver.class.getName())) .newInstance(); archiver.init(conf); } catch (InstantiationException | IllegalAccessException | ClassNotFoundException e) { logger.log(Level.SEVERE, "Failed to instantiate archiver", e); } this.defaultTimebucketSize = Integer .parseInt(conf.getOrDefault(DEFAULT_BUCKET_SIZE, String.valueOf(DEFAULT_TIME_BUCKET_CONSTANT))); if (bgTaskPool != null) { bgTaskPool.scheduleAtFixedRate(() -> { for (Entry<String, Map<String, Measurement>> measurementMap : databaseMap.entrySet()) { for (Entry<String, Measurement> measurementEntry : measurementMap.getValue().entrySet()) { Measurement value = measurementEntry.getValue(); try { } catch (Exception e) { logger.log(Level.SEVERE, "Failed collect garbage for measurement:" + value.getMeasurementName(), e); } } } }, Integer.parseInt(conf.getOrDefault(GC_FREQUENCY, DEFAULT_GC_FREQUENCY)), Integer.parseInt(conf.getOrDefault(GC_DELAY, DEFAULT_GC_DELAY)), TimeUnit.SECONDS); if (Boolean.parseBoolean(conf.getOrDefault(StorageEngine.COMPACTION_ENABLED, "false"))) { logger.info("Compaction is enabled"); bgTaskPool.scheduleAtFixedRate(() -> { for (Entry<String, Map<String, Measurement>> measurementMap : databaseMap.entrySet()) { for (Entry<String, Measurement> measurementEntry : measurementMap.getValue().entrySet()) { Measurement value = measurementEntry.getValue(); try { value.compact(); } catch (Exception e) { logger.log(Level.SEVERE, "Failed compaction for measurement:" + value.getMeasurementName(), e); } } } }, Integer.parseInt(conf.getOrDefault(COMPACTION_FREQUENCY, DEFAULT_COMPACTION_FREQUENCY)), Integer.parseInt(conf.getOrDefault(COMPACTION_DELAY, DEFAULT_COMPACTION_DELAY)), TimeUnit.SECONDS); } else { logger.warning("Compaction is disabled"); } } enableMetricsService(); }  @Override void configure(Map<String, String> conf, ScheduledExecutorService bgTaskPool); void enableMetricsService(); @Override void updateTimeSeriesRetentionPolicy(String dbName, String measurementName, int retentionHours); @Override void updateTimeSeriesRetentionPolicy(String dbName, int retentionHours); @Override Map<String, Measurement> getOrCreateDatabase(String dbName); @Override Map<String, Measurement> getOrCreateDatabase(String dbName, int retentionPolicy, Map<String, String> conf); @Override Measurement getOrCreateMeasurement(String dbName, String measurementName); @Override boolean isMeasurementFieldFP(String dbName, String measurementName, String valueFieldName); @Override Set<String> getDatabases(); @Override void deleteAllData(); @Override boolean checkIfExists(String dbName); @Override void dropDatabase(String dbName); @Override void dropMeasurement(String dbName, String measurementName); @Override void startup(); @Override void shutdown(); Map<String, DBMetadata> getDbMetadataMap(); @Override Map<String, Map<String, Measurement>> getMeasurementMap(); @Override Map<String, Map<String, Measurement>> getDatabaseMap(); @Override int getDefaultTimebucketSize(); @Override Counter getCounter(); @Override Logger getLogger(); @Override Map<String, String> getConf();  }