Metrics Refactoring (#118)

* Break out key generation so that it can be reused across MetricsRecorder instances * Break out time recording so that it can be reused across MetricRecorder implementations

Author: stanlemon

src/main/java/com/salesforce/storm/spout/dynamic/metrics/KeyBuilder.java

@@ -0,0 +1,81 @@
+/*
+ * Copyright (c) 2017, 2018, Salesforce.com, Inc.
+ * All rights reserved.
+ *
+ * Redistribution and use in source and binary forms, with or without modification, are permitted provided that the
+ * following conditions are met:
+ *
+ * * Redistributions of source code must retain the above copyright notice, this list of conditions and the following
+ * disclaimer.
+ *
+ * * Redistributions in binary form must reproduce the above copyright notice, this list of conditions and the following
+ * disclaimer in the documentation and/or other materials provided with the distribution.
+ *
+ * * Neither the name of Salesforce.com nor the names of its contributors may be used to endorse or promote products
+ * derived from this software without specific prior written permission.
+ *
+ * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES,
+ * INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
+ * DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+ * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR
+ * SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY,
+ * WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE
+ * USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ */
+
+package com.salesforce.storm.spout.dynamic.metrics;
+
+import org.slf4j.helpers.MessageFormatter;
+
+/**
+ * Takes a {@link MetricDefinition} and generates a String that can be used as a metric key.
+ */
+class KeyBuilder {
+
+ private final String keyPrefix;
+
+ /**
+ * Create a new {@link KeyBuilder} with the specified prefix.
+ *
+ * Prefix is optional, if you don't need it simply pass in null to the builder.
+ *
+ * @param keyPrefix prefix to use when building keys from {@link MetricDefinition} instances
+ */
+ KeyBuilder(final String keyPrefix) {
+ this.keyPrefix = keyPrefix;
+ }
+
+ /**
+ * Build metric keys from {@link MetricDefinition} instances with the optionally supplied parameters.
+ *
+ * Strings are formatted using {@link MessageFormatter}, so if you have parameters "foo" and "bar" for a metric with the name of "test"
+ * than you could use a {@link CustomMetric} with "test.{}.{}" in the constructor and pass parameters "foo" and "bar" to make the full
+ * metric key of "test.foo.bar".
+ *
+ * @return format string like: "prefix.metric_name.param1.param2"
+ */
+ String build(final MetricDefinition metric, final Object[] parameters) {
+ final StringBuilder keyBuilder = new StringBuilder();
+
+ // Conditionally add key prefix.
+ if (getKeyPrefix() != null && !getKeyPrefix().isEmpty()) {
+ keyBuilder
+ .append(getKeyPrefix())
+ .append(".");
+ }
+
+ // Our default implementation should include the simple class name in the key
+ keyBuilder.append(
+ MessageFormatter.arrayFormat(metric.getKey(), parameters).getMessage()
+ );
+ return keyBuilder.toString();
+ }
+
+ /**
+ * Get key prefix.
+ * @return key prefix
+ */
+ private String getKeyPrefix() {
+ return keyPrefix;
+ }
+}

src/main/java/com/salesforce/storm/spout/dynamic/metrics/LogRecorder.java

@@ -28,9 +28,7 @@
 import org.apache.storm.task.TopologyContext;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
-import org.slf4j.helpers.MessageFormatter;
 
-import java.time.Clock;
 import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
 
@@ -42,35 +40,17 @@
 public class LogRecorder implements MetricsRecorder {
 
  private static final Logger logger = LoggerFactory.getLogger(LogRecorder.class);
- private final Map<String, Long> counters = new ConcurrentHashMap<>();
- private final Map<String, Object> assignedValues = new ConcurrentHashMap<>();
 
- // For storing timer start values
- private final Map<String, Long> timerStartValues = new ConcurrentHashMap<>();
+ private final KeyBuilder keyBuilder = new KeyBuilder(null);
 
- @Override
- public void open(Map spoutConfig, TopologyContext topologyContext) {
- }
-
- @Override
- public void close() {
- // Noop
- }
+ private final Map<String, Long> counters = new ConcurrentHashMap<>();
+ private final Map<String, Object> assignedValues = new ConcurrentHashMap<>();
 
- @Override
- public void count(final MetricDefinition metric) {
- countBy(metric, 1L, new Object[0]);
- }
+ private TimerManager timerManager;
 
  @Override
- public void count(final MetricDefinition metric, final Object... metricParameters) {
- countBy(metric, 1L, metricParameters);
-
- }
-
- @Override
- public void countBy(final MetricDefinition metric, final long incrementBy) {
- countBy(metric, incrementBy, new Object[0]);
+ public void open(final Map<String, Object> config, final TopologyContext topologyContext) {
+ timerManager = new TimerManager();
  }
 
  @Override
@@ -90,41 +70,22 @@ public void assignValue(final MetricDefinition metric, final Object value, final
  logger.debug("[ASSIGNED] {} => {}", key, value);
  }
 
- @Override
- public void assignValue(final MetricDefinition metric, final Object value) {
- assignValue(metric, value, new Object[0]);
- }
-
  @Override
  public void startTimer(final MetricDefinition metric, final Object... metricParameters) {
  final String key = generateKey(metric, metricParameters);
- timerStartValues.put(key, Clock.systemUTC().millis());
- }
-
- @Override
- public void startTimer(final MetricDefinition metric) {
- startTimer(metric, new Object[0]);
+ timerManager.start(key);
  }
 
  @Override
  public long stopTimer(final MetricDefinition metric, final Object... metricParameters) {
- final long stopTime = Clock.systemUTC().millis();
-
+ // Build key from the metric
  final String key = generateKey(metric, metricParameters);
- final Long startTime = timerStartValues.get(key);
 
- if (startTime == null) {
- logger.warn("Could not find timer key {}", key);
- return -1;
- }
- final long elapsedTimeMs = stopTime - startTime;
- logger.debug("[TIMER] {} + {}ms", key, elapsedTimeMs);
- return elapsedTimeMs;
- }
+ final long elapsedMs = timerManager.stop(key);
 
- @Override
- public long stopTimer(final MetricDefinition metric) {
- return stopTimer(metric, new Object[0]);
+  logger.debug("[TIMER] {} + {}ms", key, elapsedMs);
+
+ return elapsedMs;
  }
 
  @Override
@@ -134,6 +95,6 @@ public void recordTimer(final MetricDefinition metric, final long elapsedTimeMs,
  }
 
  private String generateKey(final MetricDefinition metric, final Object[] parameters) {
- return MessageFormatter.format(metric.getKey(), parameters).getMessage();
+ return keyBuilder.build(metric, parameters);
  }
 }

src/main/java/com/salesforce/storm/spout/dynamic/metrics/MetricsRecorder.java

@@ -42,33 +42,41 @@ public interface MetricsRecorder {
  * @param spoutConfig spout configuration.
  * @param topologyContext topology context.
  */
- void open(final Map<String, Object> spoutConfig, final TopologyContext topologyContext);
+ default void open(final Map<String, Object> spoutConfig, final TopologyContext topologyContext) {
+ }
 
  /**
  * Perform any cleanup.
  */
- void close();
+ default void close() {
+ }
 
  /**
  * Count a metric, given a name, increments it by 1.
  * @param metric metric definition.
  */
- void count(final MetricDefinition metric);
+ default void count(final MetricDefinition metric) {
+ countBy(metric, 1L, new Object[0]);
+ }
 
  /**
  * Count a metric, given a name, increments it by 1.
  * @param metric metric definition.
  * @param metricParameters when a {@link MetricDefinition} supports interpolation on it's key, for example "foo.{}.bar" the {}
  * can be replace with the supplied parameters.
  */
- void count(final MetricDefinition metric, final Object... metricParameters);
+ default void count(final MetricDefinition metric, final Object... metricParameters) {
+ countBy(metric, 1L, metricParameters);
+ }
 
  /**
  * Count a metric, given a name, increments it by value.
  * @param metric metric definition.
  * @param incrementBy amount to increment the metric by.
  */
- void countBy(final MetricDefinition metric, final long incrementBy);
+ default void countBy(final MetricDefinition metric, final long incrementBy) {
+ countBy(metric, incrementBy, new Object[0]);
+ }
 
  /**
  * Count a metric, given a name and increments by a specific amount.
@@ -93,7 +101,9 @@ public interface MetricsRecorder {
  * @param metric metric definition.
  * @param value value to be assigned.
  */
- void assignValue(final MetricDefinition metric, final Object value);
+ default void assignValue(final MetricDefinition metric, final Object value) {
+ assignValue(metric, value, new Object[0]);
+ }
 
  /**
  * Starts a timer for the given sourceClass and metricName.
@@ -107,7 +117,9 @@ public interface MetricsRecorder {
  * Starts a timer for the given sourceClass and metricName.
  * @param metric metric definition.
  */
- void startTimer(final MetricDefinition metric);
+ default void startTimer(final MetricDefinition metric) {
+ startTimer(metric, new Object[0]);
+ }
 
  /**
  * Stops and records a timer for the given sourceClass and metricName.
@@ -123,7 +135,9 @@ public interface MetricsRecorder {
  * @param metric metric definition.
  * @return How long elapsed in the timer, in milliseconds.
  */
- long stopTimer(final MetricDefinition metric);
+ default long stopTimer(final MetricDefinition metric) {
+ return stopTimer(metric, new Object[0]);
+ }
 
  /**
  * Record a timer passing in the elapsed time.