import com.netflix.servo.monitor.MonitorConfig; //导入依赖的package包/类
private static StatsTimer getTimer(String name, String...additionalTags) {
	String key = className + "." + name + "." + Joiner.on(",").join(additionalTags);
	return monitors.computeIfAbsent(key, k -> {
		Builder cb = MonitorConfig.builder(name).withTag("class", className).withTag("unit", TimeUnit.MILLISECONDS.name());
		for(int j = 0; j < additionalTags.length-1; j++) {
			String tk = additionalTags[j];
			String tv = additionalTags[j+1];
			cb.withTag(tk, tv);
			j++;
		}
		MonitorConfig config = cb.build();
		
		StatsTimer sm = new StatsTimer(config, statsConfig);
		registry.register(sm);
		return sm;
	});
}
 

import com.netflix.servo.monitor.MonitorConfig; //导入依赖的package包/类
private static BasicCounter getCounter(String name, String...additionalTags) {
	String key = className + "." + name + "."  + Joiner.on(",").join(additionalTags);
	return errors.computeIfAbsent(key, k -> {
		Builder cb = MonitorConfig.builder(name).withTag("class", className);
		for(int j = 0; j < additionalTags.length-1; j++) {
			String tk = additionalTags[j];
			String tv = additionalTags[j+1];
			cb.withTag(tk, tv);
			j++;
		}
		MonitorConfig config = cb.build();
		BasicCounter bc = new BasicCounter(config);
		registry.register(bc);
		return bc;
	});
}
 

import com.netflix.servo.monitor.MonitorConfig; //导入依赖的package包/类
private List<Metric> stripCommonTags(List<Metric> metrics) {
    Map<String, String> commonTags = NetflixTagKey.tagsFromEnvironment();
    List<Metric> result = Lists.newArrayListWithCapacity(metrics.size());

    for (Metric metric : metrics) {
        TagList origTags = metric.getConfig().getTags();
        MonitorConfig.Builder builder = MonitorConfig.builder(metric.getConfig().getName());
        for (Tag tag : origTags) {
            if (!commonTags.containsKey(tag.getKey())) {
                builder.withTag(tag);
            }
        }
        Metric noCommonTags = new Metric(builder.build(), metric.getTimestamp(), metric.getValue());
        result.add(noCommonTags);
    }
    return result;
}
 

import com.netflix.servo.monitor.MonitorConfig; //导入依赖的package包/类
private MonitorConfig keepTags(MonitorConfig monitorConfig, Set<String> tags) {
    boolean droppedSomeTags = false;

    SmallTagMap.Builder newTags = SmallTagMap.builder();
    for (Tag tag : monitorConfig.getTags()) {
        String tagName = tag.getKey();
        if (tags.contains(tagName)) {
            newTags.add(tag);
        } else {
            droppedSomeTags = true;
        }
    }

    if (droppedSomeTags) {
        return MonitorConfig.builder(monitorConfig.getName())
                .withTags(newTags)
                .withPublishingPolicy(monitorConfig.getPublishingPolicy())
                .build();
    } else {
        // avoid creating extra objects if we don't have to
        return monitorConfig;
    }
}
 

import com.netflix.servo.monitor.MonitorConfig; //导入依赖的package包/类
private MonitorConfig dropTags(MonitorConfig monitorConfig, Set<String> tagsToDrop) {
    SmallTagMap.Builder newTags = SmallTagMap.builder();
    boolean droppedSomeTags = false;
    for (Tag tag : monitorConfig.getTags()) {
        String tagName = tag.getKey();
        if (!tagsToDrop.contains(tagName)) {
            newTags.add(tag);
        } else {
            droppedSomeTags = true;
        }
    }

    if (droppedSomeTags) {
        return MonitorConfig.builder(monitorConfig.getName())
                .withTags(newTags)
                .withPublishingPolicy(monitorConfig.getPublishingPolicy())
                .build();
    } else {
        // avoid creating extra objects if we don't have to
        return monitorConfig;
    }
}
 

import com.netflix.servo.monitor.MonitorConfig; //导入依赖的package包/类
@SuppressWarnings("unchecked")
public static <T extends Number> T getNumberGauge(MonitorConfig config, T number) {
    NumberGauge v = GAUGES.get(config);
    if (v != null) {
        return (T) v.getValue(0);
    } else {
        NumberGauge gauge = new NumberGauge(config, number);
        NumberGauge prev = GAUGES.putIfAbsent(config, gauge);
        if (prev != null) {
            return (T) prev.getValue(0);
        } else {
            DefaultMonitorRegistry.getInstance().register(gauge);
            return (T) gauge.getValue(0);
        }
    }
}
 

import com.netflix.servo.monitor.MonitorConfig; //导入依赖的package包/类
/**
 * Write all messages in msgList to file writer, sync the file,
 * commit the queue and clear messages
 *
 * @param msgList
 * @throws java.io.IOException
 */
@Override
protected void write(List<Message> msgList) throws IOException {
    for (Message msg : msgList) {
        writer.writeTo(msg);

        String routingKey = normalizeRoutingKey(msg);

        DynamicCounter.increment(
                MonitorConfig.builder("writtenMessages")
                        .withTag(TagKey.DATA_SOURCE, routingKey)
                        .build());
        ++writtenMessages;
        DynamicCounter.increment(
                MonitorConfig.builder("writtenBytes")
                        .withTag(TagKey.DATA_SOURCE, routingKey)
                        .build(), msg.getPayload().length);
        writtenBytes += msg.getPayload().length;

        messageWrittenInRotation = true;
    }

    writer.sync();

    throughput.increment(msgList.size());
}
 

import com.netflix.servo.monitor.MonitorConfig; //导入依赖的package包/类
public static int incrementMessageCount(String counterName, String app, Iterable<Message> messages, List<AsyncSuroClient.Listener> listeners) {
    int count = 0;
    for (Message message : messages) {
        DynamicCounter.increment(
                MonitorConfig.builder(counterName)
                        .withTag(TagKey.APP, app)
                        .withTag(TagKey.DATA_SOURCE, message.getRoutingKey())
                        .build());
        ++count;
    }

    for (AsyncSuroClient.Listener listener : listeners) {
        listener.sentCallback(count);
    }

    return count;
}
 

import com.netflix.servo.monitor.MonitorConfig; //导入依赖的package包/类
@Override
public List<MessageContainer> parse(String data) {
    List<MessageContainer> messages = new ArrayList<MessageContainer>();

    try {
        Map<String, Object> blob = jsonMapper.readValue(data, S3Consumer.typeReference);
        List<Map<String, Object>> records = (List<Map<String, Object>>) blob.get("Records");
        for (Map<String, Object> record : records) {
            messages.add(new DefaultMessageContainer(
                    new Message(routingKey, jsonMapper.writeValueAsBytes(record)),
                    jsonMapper));
        }
    } catch (Exception e) {
        log.error("Exception on parsing: " + e.getMessage(), e);
        DynamicCounter.increment(
                MonitorConfig.builder("recordParseError").withTag("parserType", TYPE).build());
    }

    return messages;
}
 

import com.netflix.servo.monitor.MonitorConfig; //导入依赖的package包/类
@Override
public void writeTo(final MessageContainer message) {
    queuedRecords.incrementAndGet();
    DynamicCounter.increment(
        MonitorConfig
            .builder("queuedRecord")
            .withTag(TagKey.ROUTING_KEY, message.getRoutingKey())
            .build());
    runRecordCounterListener();

    if (metadataFetchedTopicSet.contains(message.getRoutingKey())) {
        sendMessage(message);
    } else {
        if(!metadataWaitingQueue.offer(message)) {
            dropMessage(message.getRoutingKey(), "metadataWaitingQueueFull");
        }
    }
}
 

import com.netflix.servo.monitor.MonitorConfig; //导入依赖的package包/类
public static LongGauge getLongGauge(String name) {
    LongGauge gauge = (LongGauge) monitorMap.get(name);
    if (gauge == null) {
        writeLock.lock();
        try {
            if (monitorMap.containsKey(name)) {
                gauge = (LongGauge) monitorMap.get(name);
            } else {
                gauge = new LongGauge(MonitorConfig.builder(name).withTag(OWNER).build());
                monitorMap.put(name, gauge);
                DefaultMonitorRegistry.getInstance().register(gauge);
            }
        } finally {
            writeLock.unlock();
        }
    }
    return gauge;
}
 

import com.netflix.servo.monitor.MonitorConfig; //导入依赖的package包/类
public static MonitorConfig getMonitorConfig(String name, String appName, String cacheName, String serverGroup, String metric, String reason, String status) {
    Builder builder = MonitorConfig.builder(name).withTag("APP", appName).withTag("METRIC", metric).withTag(OWNER);
    if (cacheName != null && cacheName.length() > 0) {
        builder = builder.withTag("CACHE", cacheName);
    }
    if (serverGroup != null && serverGroup.length() > 0) {
        builder = builder.withTag("ServerGroup", serverGroup);
    }
    if (reason != null && reason.length() > 0) {
        builder = builder.withTag("Reason", reason);
    }
    if (status != null && status.length() > 0) {
        builder = builder.withTag("Status", status);
    }
    return builder.build();
}
 

import com.netflix.servo.monitor.MonitorConfig; //导入依赖的package包/类
public EVCacheNodeImpl(SocketAddress sa, SocketChannel c, int bufSize, BlockingQueue<Operation> rq,
        BlockingQueue<Operation> wq, BlockingQueue<Operation> iq,
        long opQueueMaxBlockTimeMillis, boolean waitForAuth, long dt, long at, ConnectionFactory fa, String appName,
        int id, ServerGroup serverGroup, long stTime) {
    super(sa, c, bufSize, rq, wq, iq, Long.valueOf(opQueueMaxBlockTimeMillis), waitForAuth, dt, at, fa);

    this.id = id;
    this._appName = appName;
    this._serverGroup = serverGroup;
    setConnectTime(stTime);
    this.readQ = rq;
    this.inputQueue = iq;
    this.sendMetrics = EVCacheConfig.getInstance().getDynamicBooleanProperty("EVCacheNodeImpl." + appName + ".sendMetrics", false);
    this.tags = BasicTagList.of("ServerGroup", _serverGroup.getName(), "APP", appName, "Id", String.valueOf(id), EVCacheMetricsFactory.OWNER.getKey(), EVCacheMetricsFactory.OWNER.getValue());
    this.hostName = ((InetSocketAddress) getSocketAddress()).getHostName();
    this.metricPrefix = "EVCacheNode";
    this.baseConfig = MonitorConfig.builder(metricPrefix).build();
    baseTags = BasicTagList.concat(tags, BasicTagList.of("HOST", hostName));
    setupMonitoring();
}
 

import com.netflix.servo.monitor.MonitorConfig; //导入依赖的package包/类
public ProducerInvocationMonitor(String operationName) {
  super(operationName, String.format(MetricsConst.PRODUCER_PREFIX_TEMPLATE, operationName));
  this.waitInQueue = new BasicCounter(MonitorConfig.builder(this.getPrefix() + ".waitInQueue.count").build());
  this.lifeTimeInQueue = new TimerMonitor(this.getPrefix() + ".lifeTimeInQueue");
  this.executionTime = new TimerMonitor(this.getPrefix() + ".executionTime");
  this.producerLatency = new TimerMonitor(this.getPrefix() + ".producerLatency");
  this.producerCall = new CallMonitor(this.getPrefix() + ".producerCall");
}
 

import com.netflix.servo.monitor.MonitorConfig; //导入依赖的package包/类
public TimerMonitor(String prefix) {
  this.prefix = prefix;
  total = new StepCounter(MonitorConfig.builder(prefix + ".total").build());
  count = new StepCounter(MonitorConfig.builder(prefix + ".count").build());
  min = new MinGauge(MonitorConfig.builder(prefix + ".min").build());
  max = new MaxGauge(MonitorConfig.builder(prefix + ".max").build());
}
 

import com.netflix.servo.monitor.MonitorConfig; //导入依赖的package包/类
/**
 * Get the initial metrics and register with servo.
 */
public void initMetricsPublishing() {
  /* list of monitors */
  List<Monitor<?>> monitors = getMetricsMonitors();
  MonitorConfig commandMetricsConfig = MonitorConfig.builder("metrics").build();
  BasicCompositeMonitor commandMetricsMonitor = new BasicCompositeMonitor(commandMetricsConfig, monitors);
  DefaultMonitorRegistry.getInstance().register(commandMetricsMonitor);
}
 

import com.netflix.servo.monitor.MonitorConfig; //导入依赖的package包/类
/**
 * Get the total requests and failed requests for instance level.
 *
 * @param metricsName Name of the metrics
 * @param metricToEvaluate observable method to be called for preparation of metrics.
 * @return Guage metrics
 */
protected Monitor<Number> getRequestValuesGaugeMonitor(final String metricsName,
    final Func0<Number> metricToEvaluate) {
  return new GaugeMetric(MonitorConfig.builder(metricsName).build()) {

    @Override
    public Number getValue() {
      return metricToEvaluate.call();
    }
  };
}
 

import com.netflix.servo.monitor.MonitorConfig; //导入依赖的package包/类
/**
 * Get the total requests and failed requests for each producer.
 *
 * @param name Name of the metrics
 * @param metricToEvaluate observable method to be called for preparation of metrics.
 * @return Guage metrics
 */
protected Monitor<String> getInfoMetricsOperationalAndInstance(final String name,
    final Func0<String> metricToEvaluate) {
  return new InformationalMetric(MonitorConfig.builder(name).build()) {
    @Override
    public String getValue() {
      return metricToEvaluate.call();
    }
  };
}
 

import com.netflix.servo.monitor.MonitorConfig; //导入依赖的package包/类
@Override
protected boolean removeEldestEntry(Map.Entry<MonitorConfig, CounterValue> eldest) {
    final long now = System.currentTimeMillis();
    final long lastMod = eldest.getValue().getTimestamp();
    final boolean expired = (now - lastMod > (2 * POLLING_INTERVAL_MS));
    if (expired) {
        LOGGER.debug("heartbeat interval exceeded, expiring {}", eldest.getKey());
    }
    return expired;
}
 

import com.netflix.servo.monitor.MonitorConfig; //导入依赖的package包/类
@Override
public boolean matches(MonitorConfig config) {
    try {
        return query.call().apply(config);
    } catch (Exception e) {
        return false;
    }
}