MetricsServiceImpl.java 文件源码

public void startUp(Session session, String keyspace, boolean resetDb, boolean createSchema,
        HawkularMetricRegistry metricRegistry) {
    session.execute("USE " + keyspace);
    log.infoKeyspaceUsed(keyspace);
    metricsTasks = MoreExecutors.listeningDecorator(Executors.newFixedThreadPool(4, new MetricsThreadFactory()));
    loadDataRetentions();

    this.metricRegistry = metricRegistry;

    pointsInserter = ImmutableMap
            .<MetricType<?>, Func1<Observable<? extends Metric<?>>, Observable<Integer>>>builder()
            .put(GAUGE, metric -> {
                @SuppressWarnings("unchecked")
                Observable<Metric<Double>> gauge = (Observable<Metric<Double>>) metric;
                return dataAccess.insertData(gauge);
            })
            .put(COUNTER, metric -> {
                @SuppressWarnings("unchecked")
                Observable<Metric<Long>> counter = (Observable<Metric<Long>>) metric;
                return dataAccess.insertData(counter);
            })
            .put(AVAILABILITY, metric -> {
                @SuppressWarnings("unchecked")
                Observable<Metric<AvailabilityType>> avail = (Observable<Metric<AvailabilityType>>) metric;
                return dataAccess.insertData(avail);
            })
            .put(STRING, metric -> {
                @SuppressWarnings("unchecked")
                Observable<Metric<String>> string = (Observable<Metric<String>>) metric;
                return dataAccess.insertStringDatas(string, this::getTTL, maxStringSize);
            })
            .build();

    dataPointFinders = ImmutableMap
            .<MetricType<?>, Func6<? extends MetricId<?>, Long, Long, Integer, Order, Integer,
                    Observable<Row>>>builder()
            .put(STRING, (metricId, start, end, limit, order, pageSize) -> {
                @SuppressWarnings("unchecked")
                MetricId<String> stringId = (MetricId<String>) metricId;
                return dataAccess.findStringData(stringId, start, end, limit, order, pageSize);
            })
            .build();

    dataPointMappers = ImmutableMap.<MetricType<?>, Func1<Row, ? extends DataPoint<?>>> builder()
            .put(GAUGE, Functions::getGaugeDataPoint)
            .put(AVAILABILITY, Functions::getAvailabilityDataPoint)
            .put(COUNTER, Functions::getCounterDataPoint)
            .put(STRING, Functions::getStringDataPoint)
            .build();

    tempDataPointMappers = ImmutableMap.<MetricType<?>, Func1<Row, ? extends DataPoint<?>>> builder()
            .put(GAUGE, Functions::getTempGaugeDataPoint)
            .put(COUNTER, Functions::getTempCounterDataPoint)
            .put(AVAILABILITY, Functions::getTempAvailabilityDataPoint)
            .build();

    initConfiguration(session);
    setDefaultTTL(session, keyspace);
    initMetrics();

    verifyAndCreateTempTables();

    tagQueryParser = new SimpleTagQueryParser(this.dataAccess, this);
    expresssionTagQueryParser = new ExpressionTagQueryParser(this.dataAccess, this);
}