class GoogleMonitoringIngester extends DataIngester {

    void processResponse(ListTimeSeriesResponse response, List<String> allMetrics,
                         Map<String, Map<String, Record>> byTime) {

        for (TimeSeries ts : response.getTimeSeries()) {
            if (!allMetrics.contains(ts.getMetric().getType())) {
                ++skippedTimeSeries;
                continue;
            }

            // Extract attribute values from TimeSeries metric/resource labels. If some are
            // missing, skip this time series. Keep the keys sorted so that they correspond
            // to the attribute names recorded in the encoder.
            Map<String, String> attrMap = new TreeMap<>();
            boolean hasAllAttributes = true;

            for (String a : attributes) {
                String val = ts.getMetric().getLabels().get(a);
                if (val == null) {
                    val = ts.getResource().getLabels().get(a);
                    if (val == null) {
                        log.debug("Skipping TimeSeries due to missing attribute: " + a);
                        hasAllAttributes = false;
                        break;
                    }
                }
                attrMap.put(a, val);
            }

            if (!hasAllAttributes) {
                ++skippedTimeSeries;
                continue;
            }
            String attrKey = attrMap.toString();

            for (Point p : ts.getPoints()) {
                String timestamp = p.getInterval().getEndTime();

                byTime.putIfAbsent(timestamp, new HashMap<>());
                Record rec = byTime.get(timestamp).putIfAbsent(attrKey, new Record());
                if (rec == null) {
                    rec = byTime.get(timestamp).get(attrKey);
                    rec.attributes = attrMap;
                    rec.values = new HashMap<>();
                }

                double val;
                switch (ts.getValueType()) {
                    case "DOUBLE":
                        val = p.getValue().getDoubleValue();
                        break;
                    case "INT64":
                        val = p.getValue().getInt64Value();
                        break;
                    case "BOOL":
                        val = (p.getValue().getBoolValue()) ? 1.0 : 0.0;
                        break;
                    default:
                        continue;
                }

                rec.values.put(ts.getMetric().getType(), val);
            }
        }
    }

    public  GoogleMonitoringIngester(MacroBaseConf conf);

}