CachedTSDBService extends DefaultService implements TSDBService { @Override public Map<MetricQuery, List<Metric>> getMetrics(List<MetricQuery> queries) { List<MetricQuery> queryList = new ArrayList<MetricQuery>(queries); Map<MetricQuery, List<Metric>> result = new HashMap<MetricQuery, List<Metric>>(); List<MetricQuery> filterMetricQueries = new ArrayList<MetricQuery>(); Map<MetricQuery, MetricQueryTimestamp> map = new HashMap<MetricQuery, MetricQueryTimestamp>(); for (MetricQuery query : queryList) { if (compulsoryCacheMiss(query)) { _logger.debug("Not using cache for this metric query"); filterMetricQueries.add(query); } } queryList.removeAll(filterMetricQueries); if (!queryList.isEmpty()) { long beforeTime = System.currentTimeMillis(); List<MetricQueryTimestamp> uncached = _getCachedMetricValues(queryList, result); long afterTime = System.currentTimeMillis(); _logger.info("Time spent in _getCachedMetricValues: {}", afterTime - beforeTime); for (MetricQueryTimestamp queryWithTimestamp : uncached) { MetricQuery metricQuery = queryWithTimestamp.getMetricQuery(); metricQuery.setEndTimestamp(queryWithTimestamp.orignalEndTimestamp); filterMetricQueries.add(metricQuery); map.put(metricQuery, queryWithTimestamp); } } Map<MetricQuery, List<Metric>> metricsMap = _defaultTsdbService.getMetrics(filterMetricQueries); for (Entry<MetricQuery, List<Metric>> entry : metricsMap.entrySet()) { MetricQuery metricQuery = entry.getKey(); MetricQueryTimestamp queryWithTimestamp = map.get(metricQuery); if (!compulsoryCacheMiss(metricQuery)) { Thread cacheInsertThread = new CacheInsertThread(entry.getValue(), metricQuery); cacheInsertThread.start(); List<Metric> metrics = new ArrayList<Metric>(); long beforeTime = System.currentTimeMillis(); for (Metric metric : entry.getValue()) { Metric tempMetric = new Metric(metric); Map<Long, Double> filteredDatapoints = new LinkedHashMap<>(); for (Map.Entry<Long, Double> datapoint : metric.getDatapoints().entrySet()) { if (datapoint.getKey() >= queryWithTimestamp.getOriginalStartTimestamp()) { filteredDatapoints.put(datapoint.getKey(), datapoint.getValue()); } } tempMetric.setDatapoints(filteredDatapoints); metrics.add(tempMetric); } result.put(metricQuery, metrics); _logger.info("Time spent in trimming metric values: {}", System.currentTimeMillis() - beforeTime); } else { result.put(metricQuery, entry.getValue()); } } return result; } @Inject private  CachedTSDBService(SystemConfiguration config, MonitorService monitorService,
    		CacheService cacheService, @NamedBinding TSDBService tsdbService); @Override Map<MetricQuery, List<Metric>> getMetrics(List<MetricQuery> queries); @Override void dispose(); @Override void putMetrics(List<Metric> metrics); @Override void putHistograms(List<Histogram> histograms); @Override void putAnnotations(List<Annotation> annotations); @Override List<Annotation> getAnnotations(List<AnnotationQuery> queries);  }
@Test public void testWhenCacheMissShouldCallTSDBServiceGetMetricsMethod() throws Exception { long startTime = System.currentTimeMillis()-60*60*1000, endTime = System.currentTimeMillis(); MetricQuery query = new MetricQuery("scope", "metric", new HashMap<>(), startTime, endTime); cachedTSDBService.getMetrics(Arrays.asList(query)); verify(tsdbService,times(1)).getMetrics(any()); }
@SuppressWarnings("unchecked") @Test public void testWhenCacheMissShouldReturnMetricsFromOnlyTSDBNShouldNotCallCacheService() { long startTime = System.currentTimeMillis()-60*60*1000, endTime = System.currentTimeMillis(); MetricQuery query = new MetricQuery("scope", "metric", new HashMap<>(), startTime, endTime); Metric metric = new Metric("scope", "metric"); Map<Long, Double> dps = new HashMap<>(); dps.put(startTime,1d); metric.setDatapoints(dps); Map<MetricQuery, List<Metric>> metricsFromTSDB = new HashMap<>(); metricsFromTSDB.put(query, Arrays.asList(metric)); when(tsdbService.getMetrics(any())).thenReturn(metricsFromTSDB); Map<MetricQuery, List<Metric>> actual = cachedTSDBService.getMetrics(Arrays.asList(query)); assertEquals(metricsFromTSDB, actual); verify(cacheService, never()).getRange(anySet(), anyInt(), anyInt()); }
@SuppressWarnings("unchecked") @Test public void testWhenMetricsExistInCacheShouldReturnMetricsFromBothCacheNTSDB() throws Exception { long currTime = System.currentTimeMillis(); long startTime = currTime-4*24*60*60*1000, endTime = currTime; MetricQuery query = new MetricQuery("scope", "metric", new HashMap<>(), startTime, endTime); Metric lastHourMetric= new Metric("scope", "metric"); Map<Long, Double> lastHourDps = new HashMap<>(); lastHourDps.put(endTime, 2d); lastHourMetric.setDatapoints(lastHourDps); Map<MetricQuery, List<Metric>> lastHourMetrics = mock(Map.class); when(lastHourMetrics.get(any())).thenReturn(Arrays.asList(lastHourMetric)); when(tsdbService.getMetrics(any())).thenReturn(lastHourMetrics); StringBuilder cacheKey = new StringBuilder(); cacheKey.append(startTime); cacheKey.append(":null:scope:metric:{}:AVG:AVG:300000"); String cacheKeyValue = "{\"metatagsRecord\":null,\"datapoints\":{\""+ startTime+ "\":1}," + "\"metric\":\"metric\",\"units\":null,\"namespace\":null,\"query\":{\"aggregator\":\"AVG\",\"downsampler\":\"AVG\"," + "\"downsamplingPeriod\":300000,\"metric\":\"metric\",\"startTimestamp\":1544659200000,\"endTimestamp\":1544816134802," + "\"scope\":\"scope\",\"tags\":{}},\"displayName\":null,\"scope\":\"scope\",\"uid\":null,\"tags\":{}}"; Map<String, Object> metricsFromCache = new HashMap<>(); metricsFromCache.put(cacheKey.toString(), Arrays.asList(cacheKeyValue)); when(cacheService.getRange(anySet(), anyInt(), anyInt())).thenReturn(metricsFromCache); Map<MetricQuery, List<Metric>> actual = cachedTSDBService.getMetrics(Arrays.asList(query)); Metric expectedMetric= new Metric("scope", "metric"); Map<Long, Double> dps = new HashMap<>(); dps.put(startTime, 1d); dps.put(endTime, 2d); expectedMetric.setDatapoints(dps); List<Metric> expected = new ArrayList<>(); expected.add(expectedMetric); assertEquals(expected, actual.get(query)); }