subprojects/s4-core/src/main/java/org/apache/s4/core/util/S4Metrics.java - incubator-retired-s4 - Git at Google

 /**
  * Licensed to the Apache Software Foundation (ASF) under one
  * or more contributor license agreements.  See the NOTICE file
  * distributed with this work for additional information
  * regarding copyright ownership.  The ASF licenses this file
  * to you under the Apache License, Version 2.0 (the
  * "License"); you may not use this file except in compliance
  * with the License.  You may obtain a copy of the License at
  *
  *     http://www.apache.org/licenses/LICENSE-2.0
  *
  * Unless required by applicable law or agreed to in writing, software
  * distributed under the License is distributed on an "AS IS" BASIS,
  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
 package org.apache.s4.core.util;

 import java.io.File;
 import java.util.List;
 import java.util.Map;
 import java.util.concurrent.TimeUnit;
 import java.util.regex.Matcher;
 import java.util.regex.Pattern;

 import org.apache.s4.base.Emitter;
 import org.apache.s4.comm.topology.Assignment;
 import org.apache.s4.core.ProcessingElement;
 import org.apache.s4.core.ReceiverImpl;
 import org.apache.s4.core.RemoteSender;
 import org.apache.s4.core.SenderImpl;
 import org.apache.s4.core.Stream;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;

 import com.beust.jcommander.internal.Lists;
 import com.beust.jcommander.internal.Maps;
 import com.google.common.base.Strings;
 import com.google.common.cache.LoadingCache;
 import com.google.inject.Inject;
 import com.google.inject.Singleton;
 import com.google.inject.name.Named;
 import com.yammer.metrics.Metrics;
 import com.yammer.metrics.core.Gauge;
 import com.yammer.metrics.core.Meter;
 import com.yammer.metrics.reporting.ConsoleReporter;
 import com.yammer.metrics.reporting.CsvReporter;

 /**
  * Utility class for centralizing system runtime metrics, such as information about event processing rates, cache
  * eviction etc...
  */
 @Singleton
 public class S4Metrics {

     private static Logger logger = LoggerFactory.getLogger(S4Metrics.class);

     static final Pattern METRICS_CONFIG_PATTERN = Pattern
             .compile("(csv:.+|console):(\\d+):(DAYS|HOURS|MICROSECONDS|MILLISECONDS|MINUTES|NANOSECONDS|SECONDS)");

     @Inject
     Emitter emitter;

     @Inject
     Assignment assignment;

     @Inject(optional = true)
     @Named("s4.metrics.config")
     String metricsConfig;

     static List<Meter> partitionSenderMeters = Lists.newArrayList();

     private final Meter eventMeter = Metrics.newMeter(ReceiverImpl.class, "received-events", "event-count",
             TimeUnit.SECONDS);
     private final Meter bytesMeter = Metrics.newMeter(ReceiverImpl.class, "received-bytes", "bytes-count",
             TimeUnit.SECONDS);

     private final Meter localEventsMeter = Metrics.newMeter(Stream.class, "sent-local", "sent-local", TimeUnit.SECONDS);
     private final Meter remoteEventsMeter = Metrics.newMeter(Stream.class, "sent-remote", "sent-remote",
             TimeUnit.SECONDS);

     private Meter[] senderMeters;

     private final Map<String, Meter> dequeuingStreamMeters = Maps.newHashMap();
     private final Map<String, Meter> droppedStreamMeters = Maps.newHashMap();
     private final Map<String, Meter> streamQueueFullMeters = Maps.newHashMap();
     private final Meter droppedInSenderMeter = Metrics.newMeter(SenderImpl.class, "dropped@sender", "dropped@sender",
             TimeUnit.SECONDS);
     private final Meter droppedInRemoteSenderMeter = Metrics.newMeter(SenderImpl.class, "dropped@remote-sender",
             "dropped@remote-sender", TimeUnit.SECONDS);

     private final Map<String, Meter[]> remoteSenderMeters = Maps.newHashMap();

     @Inject
     private void init() {

         if (Strings.isNullOrEmpty(metricsConfig)) {
             logger.info("Metrics reporting not configured");
         } else {
             Matcher matcher = METRICS_CONFIG_PATTERN.matcher(metricsConfig);
             if (!matcher.matches()) {
                 logger.error(
                         "Invalid metrics configuration [{}]. Metrics configuration must match the pattern [{}]. Metrics reporting disabled.",
                         metricsConfig, METRICS_CONFIG_PATTERN);
             } else {
                 String group1 = matcher.group(1);

                 if (group1.startsWith("csv")) {
                     String outputDir = group1.substring("csv:".length());
                     long period = Long.valueOf(matcher.group(2));
                     TimeUnit timeUnit = TimeUnit.valueOf(matcher.group(3));
                     logger.info("Reporting metrics through csv files in directory [{}] with frequency of [{}] [{}]",
                             new String[] { outputDir, String.valueOf(period), timeUnit.name() });
                     CsvReporter.enable(new File(outputDir), period, timeUnit);
                 } else {
                     long period = Long.valueOf(matcher.group(2));
                     TimeUnit timeUnit = TimeUnit.valueOf(matcher.group(3));
                     logger.info("Reporting metrics on the console with frequency of [{}] [{}]",
                             new String[] { String.valueOf(period), timeUnit.name() });
                     ConsoleReporter.enable(period, timeUnit);
                 }
             }
         }

         senderMeters = new Meter[emitter.getPartitionCount()];
         // int localPartitionId = assignment.assignClusterNode().getPartition();
         for (int i = 0; i < senderMeters.length; i++) {
             senderMeters[i] = Metrics.newMeter(SenderImpl.class, "sender", "sent-to-" + (i), TimeUnit.SECONDS);
         }
         Metrics.newGauge(Stream.class, "local-vs-remote", new Gauge<Double>() {
             @Override
             public Double value() {
                 // this will return NaN if divider is zero
                 return localEventsMeter.oneMinuteRate() / remoteEventsMeter.oneMinuteRate();
             }
         });

     }

     public void createCacheGauges(ProcessingElement prototype, final LoadingCache<String, ProcessingElement> cache) {

         Metrics.newGauge(prototype.getClass(), prototype.getClass().getName() + "-cache-entries", new Gauge<Long>() {

             @Override
             public Long value() {
                 return cache.size();
             }
         });
         Metrics.newGauge(prototype.getClass(), prototype.getClass().getName() + "-cache-evictions", new Gauge<Long>() {

             @Override
             public Long value() {
                 return cache.stats().evictionCount();
             }
         });
         Metrics.newGauge(prototype.getClass(), prototype.getClass().getName() + "-cache-misses", new Gauge<Long>() {

             @Override
             public Long value() {
                 return cache.stats().missCount();
             }
         });
     }

     public void receivedEventFromCommLayer(int bytes) {
         eventMeter.mark();
         bytesMeter.mark(bytes);
     }

     public void queueIsFull(String name) {
         streamQueueFullMeters.get(name).mark();

     }

     public void sentEvent(int partition) {
         remoteEventsMeter.mark();
         try {
             senderMeters[partition].mark();
         } catch (NullPointerException e) {
             logger.warn("Sender meter not ready for partition {}", partition);
         } catch (ArrayIndexOutOfBoundsException e) {
             logger.warn("Partition {} does not exist", partition);
         }
     }

     public void droppedEventInSender() {
         droppedInSenderMeter.mark();
     }

     public void droppedEventInRemoteSender() {
         droppedInRemoteSenderMeter.mark();
     }

     public void sentLocal() {
         localEventsMeter.mark();
     }

     public void createStreamMeters(String name) {
         // TODO avoid maps to avoid map lookups?
         dequeuingStreamMeters.put(name,
                 Metrics.newMeter(Stream.class, "dequeued@" + name, "dequeued", TimeUnit.SECONDS));
         droppedStreamMeters.put(name, Metrics.newMeter(Stream.class, "dropped@" + name, "dropped", TimeUnit.SECONDS));
         streamQueueFullMeters.put(name,
                 Metrics.newMeter(Stream.class, "stream-full@" + name, "stream-full", TimeUnit.SECONDS));
     }

     public void dequeuedEvent(String name) {
         dequeuingStreamMeters.get(name).mark();
     }

     public void droppedEvent(String streamName) {
         droppedStreamMeters.get(streamName).mark();
     }

     public void createRemoteStreamMeters(String remoteClusterName, int partitionCount) {
         Meter[] meters = new Meter[partitionCount];
         for (int i = 0; i < partitionCount; i++) {
             meters[i] = Metrics.newMeter(RemoteSender.class, "remote-sender@" + remoteClusterName + "@partition-" + i,
                     "sent", TimeUnit.SECONDS);
         }
         synchronized (remoteSenderMeters) {
             remoteSenderMeters.put(remoteClusterName, meters);
         }

     }

     public void sentEventToRemoteCluster(String remoteClusterName, int partition) {
         remoteSenderMeters.get(remoteClusterName)[partition].mark();
     }

     public static class CheckpointingMetrics {

         static Meter rejectedSerializationTask = Metrics.newMeter(CheckpointingMetrics.class,
                 "checkpointing-rejected-serialization-task", "checkpointing-rejected-serialization-task",
                 TimeUnit.SECONDS);
         static Meter rejectedStorageTask = Metrics.newMeter(CheckpointingMetrics.class,
                 "checkpointing-rejected-storage-task", "checkpointing-rejected-storage-task", TimeUnit.SECONDS);
         static Meter fetchedCheckpoint = Metrics.newMeter(CheckpointingMetrics.class,
                 "checkpointing-fetched-checkpoint", "checkpointing-fetched-checkpoint", TimeUnit.SECONDS);
         static Meter fetchedCheckpointFailure = Metrics.newMeter(CheckpointingMetrics.class,
                 "checkpointing-fetched-checkpoint-failed", "checkpointing-fetched-checkpoint-failed", TimeUnit.SECONDS);

         public static void rejectedSerializationTask() {
             rejectedSerializationTask.mark();
         }

         public static void rejectedStorageTask() {
             rejectedStorageTask.mark();
         }

         public static void fetchedCheckpoint() {
             fetchedCheckpoint.mark();
         }

         public static void checkpointFetchFailed() {
             fetchedCheckpointFailure.mark();
         }
     }

 }
	/**
	* Licensed to the Apache Software Foundation (ASF) under one
	* or more contributor license agreements. See the NOTICE file
	* distributed with this work for additional information
	* regarding copyright ownership. The ASF licenses this file
	* to you under the Apache License, Version 2.0 (the
	* "License"); you may not use this file except in compliance
	* with the License. You may obtain a copy of the License at
	*
	* http://www.apache.org/licenses/LICENSE-2.0
	*
	* Unless required by applicable law or agreed to in writing, software
	* distributed under the License is distributed on an "AS IS" BASIS,
	* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
	* See the License for the specific language governing permissions and
	* limitations under the License.
	*/
	package org.apache.s4.core.util;

	import java.io.File;
	import java.util.List;
	import java.util.Map;
	import java.util.concurrent.TimeUnit;
	import java.util.regex.Matcher;
	import java.util.regex.Pattern;

	import org.apache.s4.base.Emitter;
	import org.apache.s4.comm.topology.Assignment;
	import org.apache.s4.core.ProcessingElement;
	import org.apache.s4.core.ReceiverImpl;
	import org.apache.s4.core.RemoteSender;
	import org.apache.s4.core.SenderImpl;
	import org.apache.s4.core.Stream;
	import org.slf4j.Logger;
	import org.slf4j.LoggerFactory;

	import com.beust.jcommander.internal.Lists;
	import com.beust.jcommander.internal.Maps;
	import com.google.common.base.Strings;
	import com.google.common.cache.LoadingCache;
	import com.google.inject.Inject;
	import com.google.inject.Singleton;
	import com.google.inject.name.Named;
	import com.yammer.metrics.Metrics;
	import com.yammer.metrics.core.Gauge;
	import com.yammer.metrics.core.Meter;
	import com.yammer.metrics.reporting.ConsoleReporter;
	import com.yammer.metrics.reporting.CsvReporter;

	/**
	* Utility class for centralizing system runtime metrics, such as information about event processing rates, cache
	* eviction etc...
	*/
	@Singleton
	public class S4Metrics {

	private static Logger logger = LoggerFactory.getLogger(S4Metrics.class);

	static final Pattern METRICS_CONFIG_PATTERN = Pattern
	.compile("(csv:.+\|console):(\\d+):(DAYS\|HOURS\|MICROSECONDS\|MILLISECONDS\|MINUTES\|NANOSECONDS\|SECONDS)");

	@Inject
	Emitter emitter;

	@Inject
	Assignment assignment;

	@Inject(optional = true)
	@Named("s4.metrics.config")
	String metricsConfig;

	static List<Meter> partitionSenderMeters = Lists.newArrayList();

	private final Meter eventMeter = Metrics.newMeter(ReceiverImpl.class, "received-events", "event-count",
	TimeUnit.SECONDS);
	private final Meter bytesMeter = Metrics.newMeter(ReceiverImpl.class, "received-bytes", "bytes-count",
	TimeUnit.SECONDS);

	private final Meter localEventsMeter = Metrics.newMeter(Stream.class, "sent-local", "sent-local", TimeUnit.SECONDS);
	private final Meter remoteEventsMeter = Metrics.newMeter(Stream.class, "sent-remote", "sent-remote",
	TimeUnit.SECONDS);

	private Meter[] senderMeters;

	private final Map<String, Meter> dequeuingStreamMeters = Maps.newHashMap();
	private final Map<String, Meter> droppedStreamMeters = Maps.newHashMap();
	private final Map<String, Meter> streamQueueFullMeters = Maps.newHashMap();
	private final Meter droppedInSenderMeter = Metrics.newMeter(SenderImpl.class, "dropped@sender", "dropped@sender",
	TimeUnit.SECONDS);
	private final Meter droppedInRemoteSenderMeter = Metrics.newMeter(SenderImpl.class, "dropped@remote-sender",
	"dropped@remote-sender", TimeUnit.SECONDS);

	private final Map<String, Meter[]> remoteSenderMeters = Maps.newHashMap();

	@Inject
	private void init() {

	if (Strings.isNullOrEmpty(metricsConfig)) {
	logger.info("Metrics reporting not configured");
	} else {
	Matcher matcher = METRICS_CONFIG_PATTERN.matcher(metricsConfig);
	if (!matcher.matches()) {
	logger.error(
	"Invalid metrics configuration [{}]. Metrics configuration must match the pattern [{}]. Metrics reporting disabled.",
	metricsConfig, METRICS_CONFIG_PATTERN);
	} else {
	String group1 = matcher.group(1);

	if (group1.startsWith("csv")) {
	String outputDir = group1.substring("csv:".length());
	long period = Long.valueOf(matcher.group(2));
	TimeUnit timeUnit = TimeUnit.valueOf(matcher.group(3));
	logger.info("Reporting metrics through csv files in directory [{}] with frequency of [{}] [{}]",
	new String[] { outputDir, String.valueOf(period), timeUnit.name() });
	CsvReporter.enable(new File(outputDir), period, timeUnit);
	} else {
	long period = Long.valueOf(matcher.group(2));
	TimeUnit timeUnit = TimeUnit.valueOf(matcher.group(3));
	logger.info("Reporting metrics on the console with frequency of [{}] [{}]",
	new String[] { String.valueOf(period), timeUnit.name() });
	ConsoleReporter.enable(period, timeUnit);
	}
	}
	}

	senderMeters = new Meter[emitter.getPartitionCount()];
	// int localPartitionId = assignment.assignClusterNode().getPartition();
	for (int i = 0; i < senderMeters.length; i++) {
	senderMeters[i] = Metrics.newMeter(SenderImpl.class, "sender", "sent-to-" + (i), TimeUnit.SECONDS);
	}
	Metrics.newGauge(Stream.class, "local-vs-remote", new Gauge<Double>() {
	@Override
	public Double value() {
	// this will return NaN if divider is zero
	return localEventsMeter.oneMinuteRate() / remoteEventsMeter.oneMinuteRate();
	}
	});

	}

	public void createCacheGauges(ProcessingElement prototype, final LoadingCache<String, ProcessingElement> cache) {

	Metrics.newGauge(prototype.getClass(), prototype.getClass().getName() + "-cache-entries", new Gauge<Long>() {

	@Override
	public Long value() {
	return cache.size();
	}
	});
	Metrics.newGauge(prototype.getClass(), prototype.getClass().getName() + "-cache-evictions", new Gauge<Long>() {

	@Override
	public Long value() {
	return cache.stats().evictionCount();
	}
	});
	Metrics.newGauge(prototype.getClass(), prototype.getClass().getName() + "-cache-misses", new Gauge<Long>() {

	@Override
	public Long value() {
	return cache.stats().missCount();
	}
	});
	}

	public void receivedEventFromCommLayer(int bytes) {
	eventMeter.mark();
	bytesMeter.mark(bytes);
	}

	public void queueIsFull(String name) {
	streamQueueFullMeters.get(name).mark();

	}

	public void sentEvent(int partition) {
	remoteEventsMeter.mark();
	try {
	senderMeters[partition].mark();
	} catch (NullPointerException e) {
	logger.warn("Sender meter not ready for partition {}", partition);
	} catch (ArrayIndexOutOfBoundsException e) {
	logger.warn("Partition {} does not exist", partition);
	}
	}

	public void droppedEventInSender() {
	droppedInSenderMeter.mark();
	}

	public void droppedEventInRemoteSender() {
	droppedInRemoteSenderMeter.mark();
	}

	public void sentLocal() {
	localEventsMeter.mark();
	}

	public void createStreamMeters(String name) {
	// TODO avoid maps to avoid map lookups?
	dequeuingStreamMeters.put(name,
	Metrics.newMeter(Stream.class, "dequeued@" + name, "dequeued", TimeUnit.SECONDS));
	droppedStreamMeters.put(name, Metrics.newMeter(Stream.class, "dropped@" + name, "dropped", TimeUnit.SECONDS));
	streamQueueFullMeters.put(name,
	Metrics.newMeter(Stream.class, "stream-full@" + name, "stream-full", TimeUnit.SECONDS));
	}

	public void dequeuedEvent(String name) {
	dequeuingStreamMeters.get(name).mark();
	}

	public void droppedEvent(String streamName) {
	droppedStreamMeters.get(streamName).mark();
	}

	public void createRemoteStreamMeters(String remoteClusterName, int partitionCount) {
	Meter[] meters = new Meter[partitionCount];
	for (int i = 0; i < partitionCount; i++) {
	meters[i] = Metrics.newMeter(RemoteSender.class, "remote-sender@" + remoteClusterName + "@partition-" + i,
	"sent", TimeUnit.SECONDS);
	}
	synchronized (remoteSenderMeters) {
	remoteSenderMeters.put(remoteClusterName, meters);
	}

	}

	public void sentEventToRemoteCluster(String remoteClusterName, int partition) {
	remoteSenderMeters.get(remoteClusterName)[partition].mark();
	}

	public static class CheckpointingMetrics {

	static Meter rejectedSerializationTask = Metrics.newMeter(CheckpointingMetrics.class,
	"checkpointing-rejected-serialization-task", "checkpointing-rejected-serialization-task",
	TimeUnit.SECONDS);
	static Meter rejectedStorageTask = Metrics.newMeter(CheckpointingMetrics.class,
	"checkpointing-rejected-storage-task", "checkpointing-rejected-storage-task", TimeUnit.SECONDS);
	static Meter fetchedCheckpoint = Metrics.newMeter(CheckpointingMetrics.class,
	"checkpointing-fetched-checkpoint", "checkpointing-fetched-checkpoint", TimeUnit.SECONDS);
	static Meter fetchedCheckpointFailure = Metrics.newMeter(CheckpointingMetrics.class,
	"checkpointing-fetched-checkpoint-failed", "checkpointing-fetched-checkpoint-failed", TimeUnit.SECONDS);

	public static void rejectedSerializationTask() {
	rejectedSerializationTask.mark();
	}

	public static void rejectedStorageTask() {
	rejectedStorageTask.mark();
	}

	public static void fetchedCheckpoint() {
	fetchedCheckpoint.mark();
	}

	public static void checkpointFetchFailed() {
	fetchedCheckpointFailure.mark();
	}
	}

	}