DefaultDiscoveryService extends DefaultService implements DiscoveryService { @Override public List<MetricQuery> getMatchingQueries(MetricQuery query) { requireNotDisposed(); SystemAssert.requireArgument(query != null, "Metric query cannot be null."); List<MetricQuery> expandedQueryList = null; long start = System.nanoTime(); if (DiscoveryService.isWildcardQuery(query)) { _logger.debug(MessageFormat.format("MetricQuery {0} contains wildcards. Will match against schema records.", query)); int limit = 0; int noOfTimeseriesAllowed = DiscoveryService.maxTimeseriesAllowed(query, _maxDataPointsPerQuery); if(noOfTimeseriesAllowed == 0) { DiscoveryService.throwMaximumDatapointsExceededException(query, _maxDataPointsPerQuery, _enforceDatapointsLimit, _monitorService, _logger); } Map<String, MetricQuery> queries = new HashMap<>(); if (query.getTags() == null || query.getTags().isEmpty()) { MetricSchemaRecordQuery schemaQuery = new MetricSchemaRecordQuery.MetricSchemaRecordQueryBuilder().namespace(query.getNamespace()) .scope(query.getScope()) .metric(query.getMetric()) .tagKey("*") .tagValue("*") .limit(limit) .page(1) .build(); List<MetricSchemaRecord> records = _schemaService.get(schemaQuery); for (MetricSchemaRecord record : records) { String identifier = _getIdentifier(record); if (!queries.containsKey(identifier)) { if (queries.size() == noOfTimeseriesAllowed) { DiscoveryService.throwMaximumDatapointsExceededException(query, _maxDataPointsPerQuery, _enforceDatapointsLimit, _monitorService, _logger); } MetricQuery mq = new MetricQuery(record.getScope(), record.getMetric(), null, 0L, 1L); mq.setNamespace(record.getNamespace()); _copyRemainingProperties(mq, query); queries.put(identifier, mq); } } expandedQueryList = new ArrayList<>(queries.values()); } else { Map<String, Integer> timeseriesCount = new HashMap<>(); for (Entry<String, String> tag : query.getTags().entrySet()) { MetricSchemaRecordQuery schemaQuery = new MetricSchemaRecordQuery.MetricSchemaRecordQueryBuilder().namespace(query.getNamespace()) .scope(query.getScope()) .metric(query.getMetric()) .tagKey(tag.getKey()) .tagValue(tag.getValue()) .limit(limit) .page(1) .build(); boolean containsWildcard = SchemaService.containsWildcard(query.getScope()) || SchemaService.containsWildcard(query.getMetric()) || SchemaService.containsWildcard(query.getNamespace()) || SchemaService.containsWildcard(tag.getKey()) || SchemaService.containsWildcard(tag.getValue()); List<MetricSchemaRecord> records; if(!containsWildcard) { records = Arrays.asList(new MetricSchemaRecord(query.getNamespace(), query.getScope(), query.getMetric(), tag.getKey(), tag.getValue())); } else { records = _schemaService.get(schemaQuery); } for (MetricSchemaRecord record : records) { if (_getTotalTimeseriesCount(timeseriesCount) == noOfTimeseriesAllowed) { DiscoveryService.throwMaximumDatapointsExceededException(query, _maxDataPointsPerQuery, _enforceDatapointsLimit, _monitorService, _logger); } String identifier = _getIdentifier(record); if (queries.containsKey(identifier)) { MetricQuery mq = queries.get(identifier); if (mq.getTags().containsKey(record.getTagKey())) { String oldValue = mq.getTag(record.getTagKey()); String newValue = oldValue + "|" + record.getTagValue(); mq.setTag(record.getTagKey(), newValue); } else { mq.setTag(record.getTagKey(), record.getTagValue()); } timeseriesCount.put(identifier, DiscoveryService.numApproxTimeseriesForQuery(mq)); } else { Map<String, String> tags = new HashMap<String, String>(); tags.put(record.getTagKey(), record.getTagValue()); MetricQuery mq = new MetricQuery(record.getScope(), record.getMetric(), tags, 0L, 1L); mq.setNamespace(record.getNamespace()); _copyRemainingProperties(mq, query); queries.put(identifier, mq); timeseriesCount.put(identifier, 1); } } } expandedQueryList = new ArrayList<>(queries.size()); for(Map.Entry<String, MetricQuery> entry : queries.entrySet()) { MetricQuery q = entry.getValue(); if(q.getTags().size() == query.getTags().size()) { expandedQueryList.add(q); } } } } else { _logger.debug(MessageFormat.format("MetricQuery {0} does not have any wildcards", query)); expandedQueryList = Arrays.asList(query); } _logger.debug("Time to get matching queries in ms: " + (System.nanoTime() - start) / 1000000); _logMatchedQueries(expandedQueryList); return expandedQueryList; } @Inject  DefaultDiscoveryService(SchemaService schemaService, SystemConfiguration config, MonitorService monitorService); @Override List<MetricSchemaRecord> filterRecords(SchemaQuery query); @Override List<MetricSchemaRecord> getUniqueRecords(MetricSchemaRecordQuery query, RecordType type); @Override List<String> browseRecords(MetricSchemaRecordQuery query, RecordType type, int indexLevel); @Override List<MetricQuery> getMatchingQueries(MetricQuery query);  }
@Test public void testWildcardQueriesMatchWithinLimit() { SchemaService schemaServiceMock = mock(SchemaService.class); List<MetricSchemaRecord> records = new ArrayList<>(); records.add(new MetricSchemaRecord(null, "scope0", "metric0", "source", "unittest")); records.add(new MetricSchemaRecord(null, "scope1", "metric1", "source", "unittest")); when(schemaServiceMock.get(any(MetricSchemaRecordQuery.class))).thenReturn(records); DefaultDiscoveryService discoveryService = new DefaultDiscoveryService(schemaServiceMock, system.getConfiguration(), monitorService); Map<String, String> tags = new HashMap<String, String>(); tags.put("source", "unittest"); MetricQuery query = new MetricQuery("scope[0|1]", "metric[0|1]", tags, 1L, 2L); List<MetricQuery> queries = discoveryService.getMatchingQueries(query); assertEquals(2, queries.size()); assertEquals(new MetricQuery("scope0", "metric0", tags, 1L, 2L), queries.get(0)); assertEquals(new MetricQuery("scope1", "metric1", tags, 1L, 2L), queries.get(1)); }
@Test public void testWildcardQueriesMatchMultipleTags() { SchemaService schemaServiceMock = mock(SchemaService.class); MetricSchemaRecordQuery queryForTag1 = new MetricSchemaRecordQueryBuilder().scope("scope0") .metric("metric0") .tagKey("source") .tagValue("unittest0") .limit(500) .build(); MetricSchemaRecordQuery queryForTag2 = new MetricSchemaRecordQueryBuilder().scope("scope0") .metric("metric0") .tagKey("device") .tagValue("device[1]") .limit(500) .build(); when(schemaServiceMock.get(queryForTag2)).thenReturn(new ArrayList<>()); DefaultDiscoveryService discoveryService = new DefaultDiscoveryService(schemaServiceMock, system.getConfiguration(), monitorService); Map<String, String> tags = new HashMap<String, String>(); tags.put("source", "unittest0"); tags.put("device", "device[1]"); MetricQuery query = new MetricQuery("scope0", "metric0", tags, 1L, 2L); List<MetricQuery> matchedQueries = discoveryService.getMatchingQueries(query); assertTrue(matchedQueries.isEmpty()); }
@Test public void testWildcardQueriesMatchMultipleTags1() { SchemaService schemaServiceMock = mock(SchemaService.class); MetricSchemaRecordQuery queryForTag1 = new MetricSchemaRecordQueryBuilder().scope("scope0") .metric("metric0") .tagKey("source") .tagValue("unittest0") .limit(500) .build(); MetricSchemaRecordQuery queryForTag2 = new MetricSchemaRecordQueryBuilder().scope("scope0") .metric("metric0") .tagKey("device") .tagValue("device[1]") .limit(500) .build(); DefaultDiscoveryService discoveryService = new DefaultDiscoveryService(schemaServiceMock, system.getConfiguration(), monitorService); Map<String, String> tags = new HashMap<String, String>(); tags.put("source", "unittest0"); tags.put("device", "device[1]"); MetricQuery query = new MetricQuery("scope?", "metric0", tags, 1L, 2L); List<MetricQuery> matchedQueries = discoveryService.getMatchingQueries(query); assertTrue(matchedQueries.isEmpty()); }
@Test(expected = WildcardExpansionLimitExceededException.class) public void testWildcardQueriesMatchExceedingLimit() { SchemaService schemaServiceMock = mock(SchemaService.class); List<MetricSchemaRecord> records = new ArrayList<>(); records.add(new MetricSchemaRecord(null, "scope", "metric0", "source", "unittest")); records.add(new MetricSchemaRecord(null, "scope", "metric1", "source", "unittest")); records.add(new MetricSchemaRecord(null, "scope", "metric2", "source", "unittest")); records.add(new MetricSchemaRecord(null, "scope", "metric3", "source", "unittest")); records.add(new MetricSchemaRecord(null, "scope", "metric4", "source", "unittest")); records.add(new MetricSchemaRecord(null, "scope", "metric5", "source", "unittest")); records.add(new MetricSchemaRecord(null, "scope", "metric6", "source", "unittest")); records.add(new MetricSchemaRecord(null, "scope", "metric7", "source", "unittest")); records.add(new MetricSchemaRecord(null, "scope", "metric8", "source", "unittest")); records.add(new MetricSchemaRecord(null, "scope", "metric9", "source", "unittest")); records.add(new MetricSchemaRecord(null, "scope", "metric10", "source", "unittest")); records.add(new MetricSchemaRecord(null, "scope", "metric11", "source", "unittest")); records.add(new MetricSchemaRecord(null, "scope", "metric12", "source", "unittest")); records.add(new MetricSchemaRecord(null, "scope", "metric13", "source", "unittest")); records.add(new MetricSchemaRecord(null, "scope", "metric14", "source", "unittest")); records.add(new MetricSchemaRecord(null, "scope", "metric15", "source", "unittest")); records.add(new MetricSchemaRecord(null, "scope", "metric16", "source", "unittest")); records.add(new MetricSchemaRecord(null, "scope", "metric17", "source", "unittest")); records.add(new MetricSchemaRecord(null, "scope", "metric18", "source", "unittest")); records.add(new MetricSchemaRecord(null, "scope", "metric19", "source", "unittest")); records.add(new MetricSchemaRecord(null, "scope", "metric20", "source", "unittest")); records.add(new MetricSchemaRecord(null, "scope", "metric21", "source", "unittest")); records.add(new MetricSchemaRecord(null, "scope", "metric22", "source", "unittest")); records.add(new MetricSchemaRecord(null, "scope", "metric23", "source", "unittest")); records.add(new MetricSchemaRecord(null, "scope", "metric24", "source", "unittest")); records.add(new MetricSchemaRecord(null, "scope", "metric25", "source", "unittest")); records.add(new MetricSchemaRecord(null, "scope", "metric26", "source", "unittest")); records.add(new MetricSchemaRecord(null, "scope", "metric27", "source", "unittest")); records.add(new MetricSchemaRecord(null, "scope", "metric28", "source", "unittest")); records.add(new MetricSchemaRecord(null, "scope", "metric29", "source", "unittest")); when(schemaServiceMock.get(any(MetricSchemaRecordQuery.class))).thenReturn(records); DefaultDiscoveryService discoveryService = new DefaultDiscoveryService(schemaServiceMock, system.getConfiguration(), monitorService); Map<String, String> tags = new HashMap<String, String>(); tags.put("source", "unittest"); MetricQuery query = new MetricQuery("scope", "metric*", null, System.currentTimeMillis() - (250 * 24 * 60 * 60 * 1000L), System.currentTimeMillis()); List<MetricQuery> queries = discoveryService.getMatchingQueries(query); assertEquals(30, queries.size()); }
@Test public void testWildcardQueriesMatchWithDownsampling() { SchemaService schemaServiceMock = mock(SchemaService.class); List<MetricSchemaRecord> records = new ArrayList<>(); records.add(new MetricSchemaRecord(null, "scope", "metric0", "source", "unittest")); records.add(new MetricSchemaRecord(null, "scope", "metric1", "source", "unittest")); records.add(new MetricSchemaRecord(null, "scope", "metric2", "source", "unittest")); records.add(new MetricSchemaRecord(null, "scope", "metric3", "source", "unittest")); records.add(new MetricSchemaRecord(null, "scope", "metric4", "source", "unittest")); records.add(new MetricSchemaRecord(null, "scope", "metric5", "source", "unittest")); records.add(new MetricSchemaRecord(null, "scope", "metric6", "source", "unittest")); records.add(new MetricSchemaRecord(null, "scope", "metric7", "source", "unittest")); records.add(new MetricSchemaRecord(null, "scope", "metric8", "source", "unittest")); records.add(new MetricSchemaRecord(null, "scope", "metric9", "source", "unittest")); records.add(new MetricSchemaRecord(null, "scope", "metric10", "source", "unittest")); records.add(new MetricSchemaRecord(null, "scope", "metric11", "source", "unittest")); records.add(new MetricSchemaRecord(null, "scope", "metric12", "source", "unittest")); records.add(new MetricSchemaRecord(null, "scope", "metric13", "source", "unittest")); records.add(new MetricSchemaRecord(null, "scope", "metric14", "source", "unittest")); records.add(new MetricSchemaRecord(null, "scope", "metric15", "source", "unittest")); records.add(new MetricSchemaRecord(null, "scope", "metric16", "source", "unittest")); records.add(new MetricSchemaRecord(null, "scope", "metric17", "source", "unittest")); records.add(new MetricSchemaRecord(null, "scope", "metric18", "source", "unittest")); records.add(new MetricSchemaRecord(null, "scope", "metric19", "source", "unittest")); records.add(new MetricSchemaRecord(null, "scope", "metric20", "source", "unittest")); records.add(new MetricSchemaRecord(null, "scope", "metric21", "source", "unittest")); records.add(new MetricSchemaRecord(null, "scope", "metric22", "source", "unittest")); records.add(new MetricSchemaRecord(null, "scope", "metric23", "source", "unittest")); records.add(new MetricSchemaRecord(null, "scope", "metric24", "source", "unittest")); records.add(new MetricSchemaRecord(null, "scope", "metric25", "source", "unittest")); records.add(new MetricSchemaRecord(null, "scope", "metric26", "source", "unittest")); records.add(new MetricSchemaRecord(null, "scope", "metric27", "source", "unittest")); records.add(new MetricSchemaRecord(null, "scope", "metric28", "source", "unittest")); records.add(new MetricSchemaRecord(null, "scope", "metric29", "source", "unittest")); when(schemaServiceMock.get(any(MetricSchemaRecordQuery.class))).thenReturn(records); DefaultDiscoveryService discoveryService = new DefaultDiscoveryService(schemaServiceMock, system.getConfiguration(), monitorService); Map<String, String> tags = new HashMap<String, String>(); tags.put("source", "unittest"); MetricQuery query = new MetricQuery("scope", "metric*", null, System.currentTimeMillis() - (100 * 24 * 60 * 60 * 1000L), System.currentTimeMillis()); query.setDownsampler(Aggregator.AVG); query.setDownsamplingPeriod(5 * 60 * 1000L); List<MetricQuery> queries = discoveryService.getMatchingQueries(query); assertEquals(30, queries.size()); }
@Test public void testWildcardQueriesNoMatch() { SchemaService schemaServiceMock = mock(SchemaService.class); List<MetricSchemaRecord> records = new ArrayList<>(); when(schemaServiceMock.get(any(MetricSchemaRecordQuery.class))).thenReturn(records); DefaultDiscoveryService discoveryService = new DefaultDiscoveryService(schemaServiceMock, system.getConfiguration(), monitorService); Map<String, String> tags = new HashMap<String, String>(); tags.put("source", "unittest"); MetricQuery query = new MetricQuery("sdfg*", "ymdasdf*", tags, 1L, 2L); List<MetricQuery> queries = discoveryService.getMatchingQueries(query); assertEquals(0, queries.size()); }
@Test public void testNonWildcardQuery() { SchemaService schemaServiceMock = mock(SchemaService.class); List<MetricSchemaRecord> records = new ArrayList<>(); DefaultDiscoveryService discoveryService = new DefaultDiscoveryService(schemaServiceMock, system.getConfiguration(), monitorService); Map<String, String> tags = new HashMap<String, String>(); tags.put("recordType", "A"); MetricQuery query = new MetricQuery("system", "runtime", null, 1L, 2L); List<MetricQuery> queries = discoveryService.getMatchingQueries(query); assertEquals(1, queries.size()); assertEquals(query, queries.get(0)); }