GoogleMonitoringIngester extends DataIngester { MBStream<Datum> convertToStream(Map<String, Map<String, Record>> byTime) { MBStream<Datum> stream = new MBStream<>(); for (Entry<String, Map<String, Record>> entry : byTime.entrySet()) { for (Record rec : entry.getValue().values()) { try { stream.add(processRecord(rec)); ++pointsAdded; } catch (Exception e) { ++skippedPoints; } } } return stream; } GoogleMonitoringIngester(MacroBaseConf conf); @Override MBStream<Datum> getStream(); static final String GOOGLE_MONITORING_QUERIES; static final String GOOGLE_MONITORING_START_TIME; static final String GOOGLE_MONITORING_END_TIME; }
@Test public void testConvertToStream() throws Exception { Map<String, Map<String, Record>> byTime = new TreeMap<>(); Record record1 = new Record(); record1.attributes = new TreeMap<>(); record1.attributes.put("attr1", "foo"); record1.attributes.put("attr2", "bar"); record1.values = new HashMap<>(); record1.values.put("custom.googleapis.com/test", 77.0); Record record2 = new Record(); record2.attributes = new TreeMap<>(); record2.attributes.put("attr1", "foo"); record2.attributes.put("attr2", "baz"); record2.values = new HashMap<>(); record2.values.put("custom.googleapis.com/test", 88.0); Record record3 = new Record(); record3.attributes = new TreeMap<>(); record3.attributes.put("attr1", "foo"); record3.attributes.put("attr2", "bar"); record3.values = new HashMap<>(); Record record4 = new Record(); record4.attributes = new TreeMap<>(); record4.attributes.put("attr1", "foo"); record4.attributes.put("attr2", "bar"); record4.values = new HashMap<>(); record4.values.put("custom.googleapis.com/test", 99.0); Map<String, Record> recordMap1 = new HashMap<>(); recordMap1.put("attr-map-key1", record1); recordMap1.put("attr-map-key2", record2); Map<String, Record> recordMap2 = new HashMap<>(); recordMap2.put("attr-map-key1", record3); Map<String, Record> recordMap3 = new HashMap<>(); recordMap3.put("attr-map-key1", record4); byTime.put("2016-08-08T00:00:00Z", recordMap1); byTime.put("2016-08-08T00:00:01Z", recordMap2); byTime.put("2016-08-08T00:00:02Z", recordMap3); GoogleMonitoringIngester ingester = new GoogleMonitoringIngester(conf); MBStream<Datum> stream = ingester.convertToStream(byTime); List<Datum> data = stream.drain(); assertEquals(3, data.size(), 0.0); assertEquals(77, data.get(0).metrics().getEntry(0), 0.0); assertEquals(0, data.get(0).attributes().get(0), 0.0); assertEquals(1, data.get(0).attributes().get(1), 0.0); assertEquals(88, data.get(1).metrics().getEntry(0), 0.0); assertEquals(0, data.get(1).attributes().get(0), 0.0); assertEquals(2, data.get(1).attributes().get(1), 0.0); assertEquals(99, data.get(2).metrics().getEntry(0), 0.0); assertEquals(0, data.get(2).attributes().get(0), 0.0); assertEquals(1, data.get(2).attributes().get(1), 0.0); }