import com.netflix.spectator.api.Registry; //导入依赖的package包/类
/**
 * The table service bean.
 *
 * @param connectorManager    connector manager
 * @param databaseService     database service
 * @param tagService          tag service
 * @param userMetadataService user metadata service
 * @param eventBus            Internal event bus
 * @param converterUtil       utility to convert to/from Dto to connector resources
 * @param registry             registry handle
 * @return The table service bean
 */
@Bean
public TableService tableService(
    final ConnectorManager connectorManager,
    final DatabaseService databaseService,
    final TagService tagService,
    final UserMetadataService userMetadataService,
    final MetacatEventBus eventBus,
    final ConverterUtil converterUtil,
    final Registry registry
) {
    return new TableServiceImpl(
        connectorManager,
        databaseService,
        tagService,
        userMetadataService,
        eventBus,
        converterUtil,
        registry
    );
}
 

import com.netflix.spectator.api.Registry; //导入依赖的package包/类
/**
 * Partition service bean.
 *
 * @param catalogService       catalog service
 * @param connectorManager     connector manager
 * @param tableService         table service
 * @param userMetadataService  user metadata service
 * @param threadServiceManager thread manager
 * @param config               configurations
 * @param eventBus             Internal event bus
 * @param converterUtil        utility to convert to/from Dto to connector resources
 * @param registry             registry handle
 * @return The partition service implementation to use
 */
@Bean
public PartitionService partitionService(
    final CatalogService catalogService,
    final ConnectorManager connectorManager,
    final TableService tableService,
    final UserMetadataService userMetadataService,
    final ThreadServiceManager threadServiceManager,
    final Config config,
    final MetacatEventBus eventBus,
    final ConverterUtil converterUtil,
    final Registry registry
) {
    return new PartitionServiceImpl(
        catalogService,
        connectorManager,
        tableService,
        userMetadataService,
        threadServiceManager,
        config,
        eventBus,
        converterUtil,
        registry
    );
}
 

import com.netflix.spectator.api.Registry; //导入依赖的package包/类
/**
 * The Catalog Thrift Service Factory.
 *
 * @param config               Application config to use
 * @param hiveConverters       Hive converters to use
 * @param metacatV1            The Metacat V1 API implementation to use
 * @param partitionV1          The Metacat Partition V1 API to use
 * @param registry             registry for spectator
 * @return The CatalogThriftServiceFactory
 */
@Bean
public CatalogThriftServiceFactory catalogThriftServiceFactory(
    final Config config,
    final HiveConverters hiveConverters,
    final MetacatV1 metacatV1,
    final PartitionV1 partitionV1,
    final Registry registry
) {
    return new CatalogThriftServiceFactoryImpl(
        config,
        hiveConverters,
        metacatV1,
        partitionV1,
        registry
    );
}
 

import com.netflix.spectator.api.Registry; //导入依赖的package包/类
/**
 * ElasticSearchMetric constructor.
 *
 * @param registry spectator registry
 */
public ElasticSearchMetric(final Registry registry) {
    this.elasticSearchDeleteFailureCounter = registry.counter(
        registry.createId(Metrics.CounterElasticSearchDelete.getMetricName())
            .withTags(Metrics.tagStatusFailureMap));
    this.elasticSearchBulkDeleteFailureCounter = registry.counter(
        registry.createId(
            Metrics.CounterElasticSearchBulkDelete.getMetricName())
            .withTags(Metrics.tagStatusFailureMap));
    this.elasticSearchUpdateFailureCounter = registry.counter(
        registry.createId(Metrics.CounterElasticSearchUpdate.getMetricName())
            .withTags(Metrics.tagStatusFailureMap));
    this.elasticSearchBulkUpdateFailureCounter = registry.counter(
        registry.createId(
            Metrics.CounterElasticSearchBulkUpdate.getMetricName()).withTags(Metrics.tagStatusFailureMap));
    this.elasticSearchLogFailureCounter = registry.counter(
        registry.createId(Metrics.CounterElasticSearchLog.getMetricName()).withTags(Metrics.tagStatusFailureMap));
    this.elasticSearchSaveFailureCounter = registry.counter(
        registry.createId(Metrics.CounterElasticSearchSave.getMetricName()).withTags(Metrics.tagStatusFailureMap));
    this.elasticSearchBulkSaveFailureCounter = registry.counter(
        registry.createId(Metrics.CounterElasticSearchBulkSave.getMetricName())
            .withTags(Metrics.tagStatusFailureMap));
}
 

import com.netflix.spectator.api.Registry; //导入依赖的package包/类
/**
 * Constructor.
 *
 * @param config              System config
 * @param eventBus            Event bus
 * @param catalogService      Catalog service
 * @param databaseService     Database service
 * @param tableService        Table service
 * @param partitionService    Partition service
 * @param userMetadataService User metadata service
 * @param tagService          Tag service
 * @param registry            registry of spectator
 * @param elasticSearchUtil   ElasticSearch client wrapper
 */
public ElasticSearchRefresh(
    @Nonnull @NonNull final Config config,
    @Nonnull @NonNull final MetacatEventBus eventBus,
    @Nonnull @NonNull final CatalogService catalogService,
    @Nonnull @NonNull final DatabaseService databaseService,
    @Nonnull @NonNull final TableService tableService,
    @Nonnull @NonNull final PartitionService partitionService,
    @Nonnull @NonNull final UserMetadataService userMetadataService,
    @Nonnull @NonNull final TagService tagService,
    @Nonnull @NonNull final ElasticSearchUtil elasticSearchUtil,
    @Nonnull @NonNull final Registry registry
) {
    this.config = config;
    this.eventBus = eventBus;
    this.catalogService = catalogService;
    this.databaseService = databaseService;
    this.tableService = tableService;
    this.partitionService = partitionService;
    this.userMetadataService = userMetadataService;
    this.tagService = tagService;
    this.elasticSearchUtil = elasticSearchUtil;
    this.registry = registry;
}
 

import com.netflix.spectator.api.Registry; //导入依赖的package包/类
/**
 * Constructor.
 *
 * @param connectorManager    connector manager
 * @param databaseService     database service
 * @param tagService          tag service
 * @param userMetadataService user metadata service
 * @param eventBus            Internal event bus
 * @param converterUtil       utility to convert to/from Dto to connector resources
 * @param registry             registry handle
 */
public TableServiceImpl(
    final ConnectorManager connectorManager,
    final DatabaseService databaseService,
    final TagService tagService,
    final UserMetadataService userMetadataService,
    final MetacatEventBus eventBus,
    final ConverterUtil converterUtil,
    final Registry registry
) {
    this.connectorManager = connectorManager;
    this.databaseService = databaseService;
    this.tagService = tagService;
    this.userMetadataService = userMetadataService;
    this.eventBus = eventBus;
    this.converterUtil = converterUtil;
    this.registry = registry;
}
 

import com.netflix.spectator.api.Registry; //导入依赖的package包/类
/**
 * Constructor.
 *
 * @param config         config
 * @param hiveConverters hive converter
 * @param metacatV1      Metacat V1 resource
 * @param partitionV1    Partition V1 resource
 * @param catalogName    catalog name
 * @param portNumber     port
 * @param registry       registry for spectator
 */
public CatalogThriftService(
    final Config config,
    final HiveConverters hiveConverters,
    final MetacatV1 metacatV1,
    final PartitionV1 partitionV1,
    final String catalogName,
    final int portNumber,
    final Registry registry
) {
    super(config, registry, portNumber, "thrift-pool-" + catalogName + "-" + portNumber + "-%d");
    this.hiveConverters = hiveConverters;
    this.metacatV1 = metacatV1;
    this.partitionV1 = partitionV1;
    this.catalogName = catalogName;
}
 

import com.netflix.spectator.api.Registry; //导入依赖的package包/类
/**
 * Constructor.
 *
 * @param config         config
 * @param hiveConverters hive converter
 * @param metacatV1      Metacat V1 resource
 * @param partitionV1    Partition V1 resource
 * @param catalogName    catalog name
 * @param registry       registry of spectator
 */
public CatalogThriftHiveMetastore(
    final Config config,
    final HiveConverters hiveConverters,
    final MetacatV1 metacatV1,
    final PartitionV1 partitionV1,
    final String catalogName,
    final Registry registry
) {
    super("CatalogThriftHiveMetastore");

    this.config = Preconditions.checkNotNull(config, "config is null");
    this.hiveConverters = Preconditions.checkNotNull(hiveConverters, "hive converters is null");
    this.v1 = Preconditions.checkNotNull(metacatV1, "metacat api is null");
    this.partV1 = Preconditions.checkNotNull(partitionV1, "partition api is null");
    this.catalogName = normalizeIdentifier(Preconditions.checkNotNull(catalogName, "catalog name is required"));
    this.registry = registry;
}
 

import com.netflix.spectator.api.Registry; //导入依赖的package包/类
/**
 * Constructor.
 *
 * @param config               config
 * @param hiveConverters       hive converter
 * @param metacatV1            Metacat V1
 * @param partitionV1          Partition V1
 * @param registry             registry for spectator
 */
public CatalogThriftServiceFactoryImpl(
    final Config config,
    final HiveConverters hiveConverters,
    final MetacatV1 metacatV1,
    final PartitionV1 partitionV1,
    final Registry registry
) {
    this.cache = CacheBuilder.newBuilder()
        .build(new CacheLoader<CacheKey, CatalogThriftService>() {
            public CatalogThriftService load(final CacheKey key) {
                return new CatalogThriftService(config, hiveConverters, metacatV1, partitionV1,
                    key.catalogName, key.portNumber, registry);
            }
        });
}
 

import com.netflix.spectator.api.Registry; //导入依赖的package包/类
private static void checkBucketCounter(Registry registry, String mode) throws Exception {
  LongFunction<String> f = null;
  switch (mode) {
    case "age":             f = BucketFunctions.age(500, TimeUnit.MILLISECONDS);        break;
    case "ageBiasOld":      f = BucketFunctions.ageBiasOld(500, TimeUnit.MILLISECONDS); break;
    case "latency":         f = BucketFunctions.latency(500, TimeUnit.MILLISECONDS);    break;
    case "latencyBiasSlow": f = BucketFunctions.latency(500, TimeUnit.MILLISECONDS);    break;
    case "bytes":           f = BucketFunctions.bytes(500);                             break;
    case "decimal":         f = BucketFunctions.decimal(500);                           break;
    default: throw new IllegalStateException("unkown mode: " + mode);
  }
  BucketCounter bc = BucketCounter.get(registry, registry.createId("bucket-counter-" + mode), f);
  for (int i = 0; i < 1000; ++i) {
    bc.record(TimeUnit.MILLISECONDS.toNanos(i));
  }
}
 

import com.netflix.spectator.api.Registry; //导入依赖的package包/类
@Inject
public Server(Registry registry) throws IOException {
  this.registry = registry;
  requestLatency = registry.timer("server.requestLatency");
  requestSize = registry.distributionSummary("server.requestSize");

  ThreadPoolExecutor executor = (ThreadPoolExecutor)
      Executors.newFixedThreadPool(10, r -> new Thread(r, "HttpServer"));
  registry.gauge("server.threadsBusy", executor, ThreadPoolExecutor::getActiveCount);
  registry.gauge("server.threadsMax",  executor, ThreadPoolExecutor::getMaximumPoolSize);
  registry.gauge("server.queueSize",   executor, e -> e.getQueue().size());

  httpServer = HttpServer.create(new InetSocketAddress(PORT), 100);
  httpServer.setExecutor(executor);
  httpServer.createContext("/", this);
  httpServer.start();

  LOGGER.info("server started on port " + PORT);
}
 

import com.netflix.spectator.api.Registry; //导入依赖的package包/类
public static void main(String[] args) throws Exception {
  GuiceHelper helper = new GuiceHelper();
  helper.start(new AbstractModule() {
    @Override protected void configure() {
      bind(Clock.class).toInstance(Clock.SYSTEM);
      bind(Server.class).asEagerSingleton();
      bind(ReportersManager.class).asEagerSingleton();
    }

    @Provides
    @Singleton
    private MetricRegistry providesCodaRegistry() {
      return new MetricRegistry();
    }

    @Provides
    @Singleton
    private Registry providesRegistry(Clock clock, MetricRegistry codaRegistry) {
      return new MetricsRegistry(clock, codaRegistry);
    }
  });

  helper.addShutdownHook();
}
 

import com.netflix.spectator.api.Registry; //导入依赖的package包/类
public static void main(String[] args) throws Exception {
  initMetricsExtensions();
  initMetricsPublishing();

  GuiceHelper helper = new GuiceHelper();
  helper.start(new AbstractModule() {
    @Override protected void configure() {
      bind(Clock.class).toInstance(Clock.SYSTEM);
      bind(Server.class).asEagerSingleton();
    }

    @Provides
    @Singleton
    private Registry providesRegistry(Clock clock) {
      Registry r = new ServoRegistry(clock);
      Spectator.globalRegistry().add(r);
      return r;
    }
  });

  helper.addShutdownHook();
}
 

import com.netflix.spectator.api.Registry; //导入依赖的package包/类
@Test
public void readLatency() throws Exception {
  Registry r = new DefaultRegistry(new ManualClock());
  List<JmxConfig> configs = configs();

  JmxData data = timer("keyspace=test,name=ReadLatency,scope=foo,type=ColumnFamily", 0);
  List<Measurement> ms = measure(r, configs, data);
  Assert.assertEquals(7, ms.size());
  Assert.assertEquals(
      50.0e-4,
      Utils.first(ms, "statistic", "percentile_50").value(),
      1e-12);

  data = timer("keyspace=test,name=ReadLatency,scope=foo,type=ColumnFamily", 1);
  ms = measure(r, configs, data);
  Assert.assertEquals(7, ms.size());
  Assert.assertEquals(
      50.01e-4,
      Utils.first(ms, "statistic", "percentile_50").value(),
      1e-12);
}
 

import com.netflix.spectator.api.Registry; //导入依赖的package包/类
/**
 * Convert a Spectator Meter type into a Stackdriver Metric kind.
 */
public String meterToKind(Registry registry, Meter meter) {
  if (meter instanceof Counter) {
    return "CUMULATIVE";
  }

  if (registry.counters().anyMatch(m -> m.id().equals(meter.id()))) {
    return "CUMULATIVE";
  }

  if (meterIsTimer(registry, meter)) {
      return "CUMULATIVE";
  }

  return "GAUGE";
}
 

import com.netflix.spectator.api.Registry; //导入依赖的package包/类
/**
 * Add a TimeSeries for each appropriate meter measurement.
 */
void addMeterToTimeSeries(
      Registry registry, Meter meter, List<TimeSeries> tsList) {
  Iterable<Measurement> measurements = meter.measure();
  boolean applyFilter = true;

  if (cache.meterIsTimer(registry, meter)) {
     measurements = transformTimerMeasurements(measurements);
     applyFilter = false;
  }
  for (Measurement measurement : measurements) {
    if (applyFilter && !measurementFilter.test(measurement)) {
      continue;
    }

    String descriptorType = cache.idToDescriptorType(measurement.id());
    tsList.add(measurementToTimeSeries(descriptorType, registry, meter, measurement));
  }
}
 

import com.netflix.spectator.api.Registry; //导入依赖的package包/类
/**
 * Creates a timer for tracking long running tasks.
 *
 * @param registry
 *     Registry to use.
 * @param id
 *     Identifier for the metric being registered.
 * @return
 *     Timer instance.
 */
public static LongTaskTimer get(Registry registry, Id id) {
  ConcurrentMap<Id, Object> state = registry.state();
  Object obj = Utils.computeIfAbsent(state, id, i -> {
    LongTaskTimer timer = new LongTaskTimer(registry, id);
    PolledMeter.using(registry)
        .withId(id)
        .withTag(Statistic.activeTasks)
        .monitorValue(timer, LongTaskTimer::activeTasks);
    PolledMeter.using(registry)
        .withId(id)
        .withTag(Statistic.duration)
        .monitorValue(timer, t -> t.duration() / NANOS_PER_SECOND);
    return timer;
  });
  if (!(obj instanceof LongTaskTimer)) {
    Utils.propagateTypeError(registry, id, LongTaskTimer.class, obj.getClass());
    obj = new LongTaskTimer(new NoopRegistry(), id);
  }
  return (LongTaskTimer) obj;
}
 

import com.netflix.spectator.api.Registry; //导入依赖的package包/类
@Override protected void update(Registry registry) {
  double sum = Double.NaN;
  Iterator<ValueEntry<T>> iter = pairs.iterator();
  while (iter.hasNext()) {
    final ValueEntry<T> pair = iter.next();
    final T obj = pair.ref.get();
    if (obj != null) {
      double v = pair.f.applyAsDouble(obj);
      if (!Double.isNaN(v)) {
        sum = Double.isNaN(sum) ? v : sum + v;
      }
    } else {
      iter.remove();
    }
  }
  gauge.set(sum);
}
 

import com.netflix.spectator.api.Registry; //导入依赖的package包/类
@Test
public void basic() {
  Registry r = new DefaultRegistry();
  BucketDistributionSummary c = BucketDistributionSummary.get(
      r, r.createId("test"), BucketFunctions.latency(4, TimeUnit.SECONDS));

  c.record(TimeUnit.MILLISECONDS.toNanos(3750));
  Assert.assertEquals(1, r.distributionSummaries().count());
  Assert.assertEquals(1, sum(r, "test"));

  c.record(TimeUnit.MILLISECONDS.toNanos(4221));
  Assert.assertEquals(2, r.distributionSummaries().count());
  Assert.assertEquals(2, sum(r, "test"));

  c.record(TimeUnit.MILLISECONDS.toNanos(4221));
  Assert.assertEquals(2, r.distributionSummaries().count());
  Assert.assertEquals(3, sum(r, "test"));
}
 

import com.netflix.spectator.api.Registry; //导入依赖的package包/类
@Test
public void basic() {
  Registry r = new DefaultRegistry();
  BucketTimer t = BucketTimer.get(
      r, r.createId("test"), BucketFunctions.latency(4, TimeUnit.SECONDS));

  t.record(3750, TimeUnit.MILLISECONDS);
  Assert.assertEquals(1, r.timers().count());
  Assert.assertEquals(1, sum(r, "test"));

  t.record(4221, TimeUnit.MILLISECONDS);
  Assert.assertEquals(2, r.timers().count());
  Assert.assertEquals(2, sum(r, "test"));

  t.record(4221, TimeUnit.MILLISECONDS);
  Assert.assertEquals(2, r.timers().count());
  Assert.assertEquals(3, sum(r, "test"));
}