Merge 42468bd into 84e0451

README.md

@@ -23,6 +23,7 @@ Example Usage
         OpenTsdbReporter.forRegistry(environment.metrics())
                         .prefixedWith(environment.getName())
                         .withTags(ImmutableMap.of("other", "tags")) // static tags included with every metric
+                        .withDeduplicator(100, 30) // Deduplicator metrics
                         // .withBatchSize(10) // optional batching. unbounded by default. likely need to tune this.
                         .build(opentsdb)
                         .start(15L, TimeUnit.SECONDS); // tune your reporting interval
@@ -40,6 +41,7 @@ Tagged Metric Registry
     OpenTsdbReporter.forRegistry(metrics)
         .withTags(tags)
         .withBatchSize(5)
+        .withDeduplicator(100, 30)
 		.build(OpenTsdb.forService("http://opentsdb/")
 		.create())
 		.start(30L, TimeUnit.SECONDS);

pom.xml

@@ -6,7 +6,7 @@
         <artifactId>oss-parent</artifactId>
         <version>9</version>
     </parent>
-
+    
     <groupId>com.github.sps.metrics</groupId>
     <artifactId>metrics-opentsdb</artifactId>
     <version>1.1.1-SNAPSHOT</version>
@@ -22,7 +22,7 @@
     <properties>
         <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
         <project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
-        <metrics.version>3.1.0</metrics.version>
+        <metrics.version>3.2.0</metrics.version>
         <jersey.version>2.22.2</jersey.version>
         <mockito.version>1.10.17</mockito.version>
         <slf4j.version>1.7.5</slf4j.version>
@@ -72,6 +72,11 @@
             <artifactId>jersey-media-json-jackson</artifactId>
             <version>${jersey.version}</version>
         </dependency>
+        <dependency>
+            <groupId>com.google.guava</groupId>
+            <artifactId>guava</artifactId>
+            <version>24.1-jre</version>
+        </dependency>
         <dependency>
             <groupId>junit</groupId>
             <artifactId>junit</artifactId>
@@ -216,13 +221,13 @@
                 </executions>
             </plugin>
             <plugin>
-                <groupId>org.eluder.coveralls</groupId>
-                <artifactId>coveralls-maven-plugin</artifactId>
-                <version>3.0.1</version>
-                <configuration>
-                    <repoToken>${COVERALLS_REPO_TOKEN}</repoToken>
-                </configuration>
-            </plugin>
+               <groupId>org.eluder.coveralls</groupId>
+               <artifactId>coveralls-maven-plugin</artifactId>
+               <version>3.0.1</version>
+               <configuration>
+                   <repoToken>${COVERALLS_REPO_TOKEN}</repoToken>
+               </configuration>
+           </plugin>
         </plugins>
     </build>
 

src/main/java/com/github/sps/metrics/DefaultMetricsChecker.java

@@ -0,0 +1,92 @@
+package com.github.sps.metrics;
+
+import com.codahale.metrics.Counting;
+import com.codahale.metrics.Gauge;
+import com.google.common.base.Charsets;
+import com.google.common.cache.Cache;
+import com.google.common.cache.CacheBuilder;
+import com.google.common.hash.Funnel;
+import com.google.common.hash.Hasher;
+import com.google.common.hash.Hashing;
+import com.google.common.hash.PrimitiveSink;
+
+import java.util.Map;
+import java.util.concurrent.Callable;
+import java.util.concurrent.TimeUnit;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class DefaultMetricsChecker {
+
+    public boolean isDuplicate(String key, Counting metric, Map<String, String> tagsToUse) {
+        return false;
+    }
+
+    public boolean isDuplicate(String key, Gauge metric, Map<String, String> tagsToUse) {
+        return false;
+    }
+
+    public static class DeduplicatorMetricsChecker extends DefaultMetricsChecker {
+        private Logger logger = LoggerFactory.getLogger(DeduplicatorMetricsChecker.class.getName());
+
+        private final Cache<String, Number> deDuplicateMetrics;
+
+        private final Funnel<Map<String, String>> mapFunnel = new Funnel<Map<String, String>>() {
+            @Override
+            public void funnel(Map<String, String> from, PrimitiveSink into) {
+                for (Map.Entry<String, String> entry : from.entrySet()) {
+                    into.putString(entry.getKey(), Charsets.UTF_8);
+                    into.putString(entry.getValue(), Charsets.UTF_8);
+                }
+            }
+        };
+
+        public DeduplicatorMetricsChecker(long maxCapacity, int ttl) {
+            this.deDuplicateMetrics = CacheBuilder.newBuilder()
+                                                  .maximumSize(maxCapacity)
+                                                  .expireAfterWrite(ttl, TimeUnit.MINUTES).build();
+        }
+
+        protected boolean isDuplicate(String key, Map<String, String> tagsToUse,
+                                      Callable<Number> callableLong) {
+            final Hasher hc = Hashing.murmur3_128().newHasher().putString(key, Charsets.UTF_8);
+            if (!tagsToUse.isEmpty()) {
+                hc.putObject(tagsToUse, mapFunnel);
+            }
+            final String hashKey = hc.hash().toString();
+            final Number prevMetric = deDuplicateMetrics.getIfPresent(hashKey);
+            try {
+                final Number number = callableLong.call();
+                if (number.equals(prevMetric)) {
+                    return true;
+                }
+                deDuplicateMetrics.put(hashKey, number);
+            } catch (Exception e) {
+                logger.warn(e.getMessage(), e);
+            }
+            return false;
+        }
+
+        @Override
+        public boolean isDuplicate(String key, final Counting metric, Map<String, String> tagsToUse) {
+            return isDuplicate(key, tagsToUse, new Callable<Number>() {
+                @Override
+                public Number call() throws Exception {
+                    return metric.getCount();
+                }
+            });
+        }
+
+        @Override
+        public boolean isDuplicate(String key, final Gauge metric, Map<String, String> tagsToUse) {
+            return isDuplicate(key, tagsToUse, new Callable<Number>() {
+                @Override
+                public Number call() throws Exception {
+                    return (Number) metric.getValue();
+                }
+            });
+        }
+
+    }
+}