diff --git a/src/main/java/com/scylladb/jmx/metrics/APIMetrics.java b/src/main/java/com/scylladb/jmx/metrics/APIMetrics.java
deleted file mode 100644
index b9e4fa3..0000000
--- a/src/main/java/com/scylladb/jmx/metrics/APIMetrics.java
+++ /dev/null
@@ -1,399 +0,0 @@
-package com.scylladb.jmx.metrics;
-
-/*
- * Copyright 2015 Cloudius Systems
- *
- * Modified by Cloudius Systems
- */
-
-import java.util.concurrent.TimeUnit;
-
-import com.yammer.metrics.core.APIMetricsRegistry;
-import com.yammer.metrics.core.Counter;
-import com.yammer.metrics.core.Gauge;
-import com.yammer.metrics.core.Histogram;
-import com.yammer.metrics.core.Meter;
-import com.yammer.metrics.core.MetricName;
-import com.yammer.metrics.core.Timer;
-import com.yammer.metrics.reporting.JmxReporter;
-import com.yammer.metrics.core.APIMeter;
-
-public class APIMetrics {
-    private static final APIMetricsRegistry DEFAULT_REGISTRY = new APIMetricsRegistry();
-    private static final Thread SHUTDOWN_HOOK = new Thread() {
-        public void run() {
-            JmxReporter.shutdownDefault();
-        }
-    };
-
-    static {
-        JmxReporter.startDefault(DEFAULT_REGISTRY);
-        Runtime.getRuntime().addShutdownHook(SHUTDOWN_HOOK);
-    }
-
-    private APIMetrics() { /* unused */
-    }
-
-    /**
-     * Given a new {@link com.yammer.metrics.core.Gauge}, registers it under the
-     * given class and name.
-     *
-     * @param klass
-     *            the class which owns the metric
-     * @param name
-     *            the name of the metric
-     * @param metric
-     *            the metric
-     * @param <T>
-     *            the type of the value returned by the metric
-     * @return {@code metric}
-     */
-    public static <T> Gauge<T> newGauge(Class<?> klass, String name,
-            Gauge<T> metric) {
-        return DEFAULT_REGISTRY.newGauge(klass, name, metric);
-    }
-
-    /**
-     * Given a new {@link com.yammer.metrics.core.Gauge}, registers it under the
-     * given class and name.
-     *
-     * @param klass
-     *            the class which owns the metric
-     * @param name
-     *            the name of the metric
-     * @param scope
-     *            the scope of the metric
-     * @param metric
-     *            the metric
-     * @param <T>
-     *            the type of the value returned by the metric
-     * @return {@code metric}
-     */
-    public static <T> Gauge<T> newGauge(Class<?> klass, String name,
-            String scope, Gauge<T> metric) {
-        return DEFAULT_REGISTRY.newGauge(klass, name, scope, metric);
-    }
-
-    /**
-     * Given a new {@link com.yammer.metrics.core.Gauge}, registers it under the
-     * given metric name.
-     *
-     * @param metricName
-     *            the name of the metric
-     * @param metric
-     *            the metric
-     * @param <T>
-     *            the type of the value returned by the metric
-     * @return {@code metric}
-     */
-    public static <T> Gauge<T> newGauge(MetricName metricName, Gauge<T> metric) {
-        return DEFAULT_REGISTRY.newGauge(metricName, metric);
-    }
-
-    /**
-     * Creates a new {@link com.yammer.metrics.core.Counter} and registers it
-     * under the given class and name.
-     *
-     * @param klass
-     *            the class which owns the metric
-     * @param name
-     *            the name of the metric
-     * @return a new {@link com.yammer.metrics.core.Counter}
-     */
-    public static Counter newCounter(String url, Class<?> klass, String name) {
-        return DEFAULT_REGISTRY.newCounter(url, klass, name);
-    }
-
-    /**
-     * Creates a new {@link com.yammer.metrics.core.Counter} and registers it
-     * under the given class and name.
-     *
-     * @param klass
-     *            the class which owns the metric
-     * @param name
-     *            the name of the metric
-     * @param scope
-     *            the scope of the metric
-     * @return a new {@link com.yammer.metrics.core.Counter}
-     */
-    public static Counter newCounter(String url, Class<?> klass, String name,
-            String scope) {
-        return DEFAULT_REGISTRY.newCounter(url, klass, name, scope);
-    }
-
-    /**
-     * Creates a new {@link com.yammer.metrics.core.Counter} and registers it
-     * under the given metric name.
-     *
-     * @param metricName
-     *            the name of the metric
-     * @return a new {@link com.yammer.metrics.core.Counter}
-     */
-    public static Counter newCounter(String url, MetricName metricName) {
-        return DEFAULT_REGISTRY.newCounter(url, metricName);
-    }
-
-    /**
-     * Creates a new {@link com.yammer.metrics.core.Histogram} and registers it
-     * under the given class and name.
-     *
-     * @param klass
-     *            the class which owns the metric
-     * @param name
-     *            the name of the metric
-     * @param biased
-     *            whether or not the histogram should be biased
-     * @return a new {@link com.yammer.metrics.core.Histogram}
-     */
-    public static Histogram newHistogram(String url, Class<?> klass,
-            String name, boolean biased) {
-        return DEFAULT_REGISTRY.newHistogram(url, klass, name, biased);
-    }
-
-    /**
-     * Creates a new {@link com.yammer.metrics.core.Histogram} and registers it
-     * under the given class, name, and scope.
-     *
-     * @param klass
-     *            the class which owns the metric
-     * @param name
-     *            the name of the metric
-     * @param scope
-     *            the scope of the metric
-     * @param biased
-     *            whether or not the histogram should be biased
-     * @return a new {@link com.yammer.metrics.core.Histogram}
-     */
-    public static Histogram newHistogram(String url, Class<?> klass,
-            String name, String scope, boolean biased) {
-        return DEFAULT_REGISTRY.newHistogram(url, klass, name, scope, biased);
-    }
-
-    /**
-     * Creates a new {@link com.yammer.metrics.core.Histogram} and registers it
-     * under the given metric name.
-     *
-     * @param metricName
-     *            the name of the metric
-     * @param biased
-     *            whether or not the histogram should be biased
-     * @return a new {@link com.yammer.metrics.core.Histogram}
-     */
-    public static Histogram newHistogram(String url, MetricName metricName,
-            boolean biased) {
-        return DEFAULT_REGISTRY.newHistogram(url, metricName, biased);
-    }
-
-    /**
-     * Creates a new non-biased {@link com.yammer.metrics.core.Histogram} and
-     * registers it under the given class and name.
-     *
-     * @param klass
-     *            the class which owns the metric
-     * @param name
-     *            the name of the metric
-     * @return a new {@link com.yammer.metrics.core.Histogram}
-     */
-    public static Histogram newHistogram(String url, Class<?> klass, String name) {
-        return DEFAULT_REGISTRY.newHistogram(url, klass, name);
-    }
-
-    /**
-     * Creates a new non-biased {@link com.yammer.metrics.core.Histogram} and
-     * registers it under the given class, name, and scope.
-     *
-     * @param klass
-     *            the class which owns the metric
-     * @param name
-     *            the name of the metric
-     * @param scope
-     *            the scope of the metric
-     * @return a new {@link com.yammer.metrics.core.Histogram}
-     */
-    public static Histogram newHistogram(String url, Class<?> klass,
-            String name, String scope) {
-        return DEFAULT_REGISTRY.newHistogram(url, klass, name, scope);
-    }
-
-    /**
-     * Creates a new non-biased {@link com.yammer.metrics.core.Histogram} and
-     * registers it under the given metric name.
-     *
-     * @param metricName
-     *            the name of the metric
-     * @return a new {@link com.yammer.metrics.core.Histogram}
-     */
-    public static Histogram newHistogram(String url, MetricName metricName) {
-        return newHistogram(url, metricName, false);
-    }
-
-    /**
-     * Creates a new {@link com.yammer.metrics.core.Meter} and registers it
-     * under the given class and name.
-     *
-     * @param klass
-     *            the class which owns the metric
-     * @param name
-     *            the name of the metric
-     * @param eventType
-     *            the plural name of the type of events the meter is measuring
-     *            (e.g., {@code "requests"})
-     * @param unit
-     *            the rate unit of the new meter
-     * @return a new {@link com.yammer.metrics.core.Meter}
-     */
-    public static APIMeter newMeter(String url, Class<?> klass, String name,
-            String eventType, TimeUnit unit) {
-        return DEFAULT_REGISTRY.newMeter(url, klass, name, eventType, unit);
-    }
-
-    /**
-     * Creates a new {@link com.yammer.metrics.core.Meter} and registers it
-     * under the given class, name, and scope.
-     *
-     * @param klass
-     *            the class which owns the metric
-     * @param name
-     *            the name of the metric
-     * @param scope
-     *            the scope of the metric
-     * @param eventType
-     *            the plural name of the type of events the meter is measuring
-     *            (e.g., {@code "requests"})
-     * @param unit
-     *            the rate unit of the new meter
-     * @return a new {@link com.yammer.metrics.core.Meter}
-     */
-    public static APIMeter newMeter(String url, Class<?> klass, String name,
-            String scope, String eventType, TimeUnit unit) {
-        return DEFAULT_REGISTRY.newMeter(url, klass, name, scope, eventType,
-                unit);
-    }
-
-    /**
-     * Creates a new {@link com.yammer.metrics.core.Meter} and registers it
-     * under the given metric name.
-     *
-     * @param metricName
-     *            the name of the metric
-     * @param eventType
-     *            the plural name of the type of events the meter is measuring
-     *            (e.g., {@code "requests"})
-     * @param unit
-     *            the rate unit of the new meter
-     * @return a new {@link com.yammer.metrics.core.Meter}
-     */
-    public static APIMeter newMeter(String url, MetricName metricName,
-            String eventType, TimeUnit unit) {
-        return DEFAULT_REGISTRY.newMeter(url, metricName, eventType, unit);
-    }
-
-    /**
-     * Creates a new {@link com.yammer.metrics.core.APITimer} and registers it
-     * under the given class and name.
-     *
-     * @param klass
-     *            the class which owns the metric
-     * @param name
-     *            the name of the metric
-     * @param durationUnit
-     *            the duration scale unit of the new timer
-     * @param rateUnit
-     *            the rate scale unit of the new timer
-     * @return a new {@link com.yammer.metrics.core.APITimer}
-     */
-    public static Timer newTimer(String url, Class<?> klass, String name,
-            TimeUnit durationUnit, TimeUnit rateUnit) {
-        return DEFAULT_REGISTRY.newTimer(url, klass, name, durationUnit, rateUnit);
-    }
-
-    /**
-     * Creates a new {@link com.yammer.metrics.core.APITimer} and registers it
-     * under the given class and name, measuring elapsed time in milliseconds
-     * and invocations per second.
-     *
-     * @param klass
-     *            the class which owns the metric
-     * @param name
-     *            the name of the metric
-     * @return a new {@link com.yammer.metrics.core.APITimer}
-     */
-    public static Timer newTimer(String url, Class<?> klass, String name) {
-        return DEFAULT_REGISTRY.newTimer(url, klass, name);
-    }
-
-    /**
-     * Creates a new {@link com.yammer.metrics.core.APITimer} and registers it
-     * under the given class, name, and scope.
-     *
-     * @param klass
-     *            the class which owns the metric
-     * @param name
-     *            the name of the metric
-     * @param scope
-     *            the scope of the metric
-     * @param durationUnit
-     *            the duration scale unit of the new timer
-     * @param rateUnit
-     *            the rate scale unit of the new timer
-     * @return a new {@link com.yammer.metrics.core.APITimer}
-     */
-    public static Timer newTimer(String url, Class<?> klass, String name, String scope,
-            TimeUnit durationUnit, TimeUnit rateUnit) {
-        return DEFAULT_REGISTRY.newTimer(url, klass, name, scope, durationUnit,
-                rateUnit);
-    }
-
-    /**
-     * Creates a new {@link com.yammer.metrics.core.APITimer} and registers it
-     * under the given class, name, and scope, measuring elapsed time in
-     * milliseconds and invocations per second.
-     *
-     * @param klass
-     *            the class which owns the metric
-     * @param name
-     *            the name of the metric
-     * @param scope
-     *            the scope of the metric
-     * @return a new {@link com.yammer.metrics.core.APITimer}
-     */
-    public static Timer newTimer(String url, Class<?> klass, String name, String scope) {
-        return DEFAULT_REGISTRY.newTimer(url, klass, name, scope);
-    }
-
-    /**
-     * Creates a new {@link com.yammer.metrics.core.APITimer} and registers it
-     * under the given metric name.
-     *
-     * @param metricName
-     *            the name of the metric
-     * @param durationUnit
-     *            the duration scale unit of the new timer
-     * @param rateUnit
-     *            the rate scale unit of the new timer
-     * @return a new {@link com.yammer.metrics.core.APITimer}
-     */
-    public static Timer newTimer(String url, MetricName metricName, TimeUnit durationUnit,
-            TimeUnit rateUnit) {
-        return DEFAULT_REGISTRY.newTimer(url, metricName, durationUnit, rateUnit);
-    }
-
-    /**
-     * Returns the (static) default registry.
-     *
-     * @return the metrics registry
-     */
-    public static APIMetricsRegistry defaultRegistry() {
-        return DEFAULT_REGISTRY;
-    }
-
-    /**
-     * Shuts down all thread pools for the default registry.
-     */
-    public static void shutdown() {
-        DEFAULT_REGISTRY.shutdown();
-        JmxReporter.shutdownDefault();
-        Runtime.getRuntime().removeShutdownHook(SHUTDOWN_HOOK);
-    }
-
-}
diff --git a/src/main/java/com/scylladb/jmx/metrics/MetricNameFactory.java b/src/main/java/com/scylladb/jmx/metrics/MetricNameFactory.java
deleted file mode 100644
index 9114711..0000000
--- a/src/main/java/com/scylladb/jmx/metrics/MetricNameFactory.java
+++ /dev/null
@@ -1,37 +0,0 @@
-/*
- * 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.
- */
-/*
- * Copyright 2015 Cloudius Systems
- *
- * Modified by Cloudius Systems
- */
-
-package com.scylladb.jmx.metrics;
-
-import com.yammer.metrics.core.MetricName;
-
-public interface MetricNameFactory
-{
-    /**
-     * Create a qualified name from given metric name.
-     *
-     * @param metricName part of qualified name.
-     * @return new String with given metric name.
-     */
-    MetricName createMetricName(String metricName);
-}
diff --git a/src/main/java/org/apache/cassandra/metrics/CASClientRequestMetrics.java b/src/main/java/org/apache/cassandra/metrics/CASClientRequestMetrics.java
index c1c54ce..7688a99 100644
--- a/src/main/java/org/apache/cassandra/metrics/CASClientRequestMetrics.java
+++ b/src/main/java/org/apache/cassandra/metrics/CASClientRequestMetrics.java
@@ -25,34 +25,20 @@
 
 package org.apache.cassandra.metrics;
 
-import com.scylladb.jmx.metrics.APIMetrics;
-import com.yammer.metrics.core.*;
+import javax.management.MalformedObjectNameException;
 
+// TODO: In StorageProxy
 public class CASClientRequestMetrics extends ClientRequestMetrics {
 
-    public final Histogram contention;
-    /* Used only for write */
-    public final Counter conditionNotMet;
-
-    public final Counter unfinishedCommit;
-
-    public CASClientRequestMetrics(String url, String scope) {
-        super(url, scope);
-        contention = APIMetrics.newHistogram(url + "contention",
-                factory.createMetricName("ContentionHistogram"), true);
-        conditionNotMet = APIMetrics.newCounter(url + "condition_not_met",
-                factory.createMetricName("ConditionNotMet"));
-        unfinishedCommit = APIMetrics.newCounter(url + "unfinished_commit",
-                factory.createMetricName("UnfinishedCommit"));
+    public CASClientRequestMetrics(String scope, String url) {
+        super(scope, url);
     }
 
-    public void release() {
-        super.release();
-        APIMetrics.defaultRegistry().removeMetric(
-                factory.createMetricName("ContentionHistogram"));
-        APIMetrics.defaultRegistry().removeMetric(
-                factory.createMetricName("ConditionNotMet"));
-        APIMetrics.defaultRegistry().removeMetric(
-                factory.createMetricName("UnfinishedCommit"));
+    @Override
+    public void register(MetricsRegistry registry) throws MalformedObjectNameException {
+        super.register(registry);
+        registry.register(() -> registry.histogram(uri + "/contention", true), names("ContentionHistogram"));
+        registry.register(() -> registry.counter(uri + "/condition_not_met"), names("ConditionNotMet"));
+        registry.register(() -> registry.counter(uri + "/unfinished_commit"), names("UnfinishedCommit"));
     }
 }
diff --git a/src/main/java/org/apache/cassandra/metrics/CacheMetrics.java b/src/main/java/org/apache/cassandra/metrics/CacheMetrics.java
index 6ff459c..d47d28f 100644
--- a/src/main/java/org/apache/cassandra/metrics/CacheMetrics.java
+++ b/src/main/java/org/apache/cassandra/metrics/CacheMetrics.java
@@ -23,44 +23,20 @@
  */
 package org.apache.cassandra.metrics;
 
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicLong;
-
-import com.scylladb.jmx.api.APIClient;
-import com.scylladb.jmx.metrics.APIMetrics;
-import com.scylladb.jmx.metrics.DefaultNameFactory;
-import com.scylladb.jmx.metrics.MetricNameFactory;
-import com.yammer.metrics.core.Gauge;
-import com.yammer.metrics.core.APIMeter;
+import javax.management.MalformedObjectNameException;
 
 /**
  * Metrics for {@code ICache}.
  */
-public class CacheMetrics {
-    /** Cache capacity in bytes */
-    public final Gauge<Long> capacity;
-    /** Total number of cache hits */
-    public final APIMeter hits;
-    /** Total number of cache requests */
-    public final APIMeter requests;
-    /** cache hit rate */
-    public final Gauge<Double> hitRate;
-    /** Total size of cache, in bytes */
-    public final Gauge<Long> size;
-    /** Total number of cache entries */
-    public final Gauge<Integer> entries;
+public class CacheMetrics implements Metrics {
 
-    private final AtomicLong lastRequests = new AtomicLong(0);
-    private final AtomicLong lastHits = new AtomicLong(0);
+    private final String type;
+    private final String url;
 
-    private APIClient c = new APIClient();
-
-    private String getURL(String url, String value) {
-        if (url == null || value == null) {
-            return null;
-        }
-        return "/cache_service/metrics/" + url + value;
+    private String compose(String value) {
+        return "/cache_service/metrics/" + url + "/" + value;
     }
+
     /**
      * Create metrics for given cache.
      *
@@ -70,70 +46,21 @@ public class CacheMetrics {
      *            Cache to measure metrics
      */
     public CacheMetrics(String type, final String url) {
+        this.type = type;
+        this.url = url;
+    }
+
+    @Override
+    public void register(MetricsRegistry registry) throws MalformedObjectNameException {
         MetricNameFactory factory = new DefaultNameFactory("Cache", type);
 
+        registry.register(() -> registry.gauge(compose("capacity")), factory.createMetricName("Capacity"));
+        registry.register(() -> registry.meter(compose("hits_moving_avrage")), factory.createMetricName("Hits"));
+        registry.register(() -> registry.meter(compose("requests_moving_avrage")),
+                factory.createMetricName("Requests"));
 
-        capacity = APIMetrics.newGauge(factory.createMetricName("Capacity"),
-                new Gauge<Long>() {
-                    String u = getURL(url, "/capacity");
-                    public Long value() {
-                        if (u == null) {
-                            return 0L;
-                        }
-                        return c.getLongValue(u);
-                    }
-                });
-        hits = APIMetrics.newMeter(getURL(url, "/hits_moving_avrage"), factory.createMetricName("Hits"), "hits",
-                TimeUnit.SECONDS);
-        requests = APIMetrics.newMeter(getURL(url, "/requests_moving_avrage"), factory.createMetricName("Requests"),
-                "requests", TimeUnit.SECONDS);
-        hitRate = APIMetrics.newGauge(factory.createMetricName("HitRate"),
-                new Gauge<Double>() {
-                    String u = getURL(url, "/hit_rate");
-                    @Override
-                    public Double value() {
-                        if (u == null) {
-                            return 0.0;
-                        }
-                        return c.getDoubleValue(u);
-                    }
-                });
-        size = APIMetrics.newGauge(factory.createMetricName("Size"),
-                new Gauge<Long>() {
-                    String u = getURL(url, "/size");
-                    public Long value() {
-                        if (u == null) {
-                            return 0L;
-                        }
-                        return c.getLongValue(u);
-                    }
-                });
-        entries = APIMetrics.newGauge(factory.createMetricName("Entries"),
-                new Gauge<Integer>() {
-                    String u = getURL(url, "/entries");
-                    public Integer value() {
-                        if (u == null) {
-                            return 0;
-                        }
-                        return c.getIntValue(u);
-                    }
-                });
+        registry.register(() -> registry.gauge(Double.class, compose("hit_rate")), factory.createMetricName("HitRate"));
+        registry.register(() -> registry.gauge(compose("size")), factory.createMetricName("Size"));
+        registry.register(() -> registry.gauge(Integer.class, compose("entries")), factory.createMetricName("Entries"));
     }
-
-    // for backward compatibility
-    @Deprecated
-    public double getRecentHitRate() {
-        long r = requests.count();
-        long h = hits.count();
-        try
-        {
-            return ((double)(h - lastHits.get())) / (r - lastRequests.get());
-        }
-        finally
-        {
-            lastRequests.set(r);
-            lastHits.set(h);
-        }
-    }
-
 }
diff --git a/src/main/java/org/apache/cassandra/metrics/ClientRequestMetrics.java b/src/main/java/org/apache/cassandra/metrics/ClientRequestMetrics.java
index 6fb8718..9644e3f 100644
--- a/src/main/java/org/apache/cassandra/metrics/ClientRequestMetrics.java
+++ b/src/main/java/org/apache/cassandra/metrics/ClientRequestMetrics.java
@@ -27,52 +27,17 @@
 
 package org.apache.cassandra.metrics;
 
-import java.util.concurrent.TimeUnit;
-
-import com.scylladb.jmx.metrics.APIMetrics;
-import com.scylladb.jmx.metrics.DefaultNameFactory;
-import com.yammer.metrics.Metrics;
-import com.yammer.metrics.core.Counter;
-import com.yammer.metrics.core.Meter;
+import javax.management.MalformedObjectNameException;
 
 public class ClientRequestMetrics extends LatencyMetrics {
-    @Deprecated
-    public static final Counter readTimeouts = Metrics
-            .newCounter(DefaultNameFactory.createMetricName(
-                    "ClientRequestMetrics", "ReadTimeouts", null));
-    @Deprecated
-    public static final Counter writeTimeouts = Metrics
-            .newCounter(DefaultNameFactory.createMetricName(
-                    "ClientRequestMetrics", "WriteTimeouts", null));
-    @Deprecated
-    public static final Counter readUnavailables = Metrics
-            .newCounter(DefaultNameFactory.createMetricName(
-                    "ClientRequestMetrics", "ReadUnavailables", null));
-    @Deprecated
-    public static final Counter writeUnavailables = Metrics
-            .newCounter(DefaultNameFactory.createMetricName(
-                    "ClientRequestMetrics", "WriteUnavailables", null));
-
-    public final Meter timeouts;
-    public final Meter unavailables;
-
-    public ClientRequestMetrics(String url, String scope) {
-        super(url, "ClientRequest", scope);
-
-        timeouts = APIMetrics.newMeter(url + "/timeouts_rates",
-                factory.createMetricName("Timeouts"), "timeouts",
-                TimeUnit.SECONDS);
-        unavailables = APIMetrics.newMeter(url + "/unavailables_rates",
-                factory.createMetricName("Unavailables"), "unavailables",
-                TimeUnit.SECONDS);
-
+    public ClientRequestMetrics(String scope, String url) {
+        super("ClientRequest", scope, url);
     }
 
-    public void release() {
-        super.release();
-        APIMetrics.defaultRegistry().removeMetric(
-                factory.createMetricName("Timeouts"));
-        APIMetrics.defaultRegistry().removeMetric(
-                factory.createMetricName("Unavailables"));
+    @Override
+    public void register(MetricsRegistry registry) throws MalformedObjectNameException {
+        super.register(registry);
+        registry.register(() -> registry.meter(uri + "/timeouts_rates"), names("Timeouts"));
+        registry.register(() -> registry.meter(uri + "/unavailables_rates"), names("Unavailables"));
     }
 }
diff --git a/src/main/java/org/apache/cassandra/metrics/ColumnFamilyMetrics.java b/src/main/java/org/apache/cassandra/metrics/ColumnFamilyMetrics.java
deleted file mode 100644
index ffd86d8..0000000
--- a/src/main/java/org/apache/cassandra/metrics/ColumnFamilyMetrics.java
+++ /dev/null
@@ -1,576 +0,0 @@
-/*
- * 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.
- */
-
-/*
- * Copyright 2015 Cloudius Systems
- *
- * Modified by Cloudius Systems
- */
-package org.apache.cassandra.metrics;
-
-import java.util.HashSet;
-import java.util.Set;
-import java.util.concurrent.ConcurrentMap;
-import java.util.concurrent.TimeUnit;
-
-import org.apache.cassandra.db.ColumnFamilyStore;
-
-import com.google.common.collect.Maps;
-import com.google.common.collect.Sets;
-import com.scylladb.jmx.api.APIClient;
-import com.scylladb.jmx.metrics.APIMetrics;
-import com.scylladb.jmx.metrics.MetricNameFactory;
-import com.scylladb.jmx.utils.RecentEstimatedHistogram;
-import com.yammer.metrics.Metrics;
-import com.yammer.metrics.core.*;
-
-/**
- * Metrics for {@link ColumnFamilyStore}.
- */
-public class ColumnFamilyMetrics {
-    private APIClient c = new APIClient();
-    /**
-     * Total amount of data stored in the memtable that resides on-heap,
-     * including column related overhead and overwritten rows.
-     */
-    public final Gauge<Long> memtableOnHeapSize;
-    /**
-     * Total amount of data stored in the memtable that resides off-heap,
-     * including column related overhead and overwritten rows.
-     */
-    public final Gauge<Long> memtableOffHeapSize;
-    /**
-     * Total amount of live data stored in the memtable, excluding any data
-     * structure overhead
-     */
-    public final Gauge<Long> memtableLiveDataSize;
-    /**
-     * Total amount of data stored in the memtables (2i and pending flush
-     * memtables included) that resides on-heap.
-     */
-    public final Gauge<Long> allMemtablesOnHeapSize;
-    /**
-     * Total amount of data stored in the memtables (2i and pending flush
-     * memtables included) that resides off-heap.
-     */
-    public final Gauge<Long> allMemtablesOffHeapSize;
-    /**
-     * Total amount of live data stored in the memtables (2i and pending flush
-     * memtables included) that resides off-heap, excluding any data structure
-     * overhead
-     */
-    public final Gauge<Long> allMemtablesLiveDataSize;
-    /** Total number of columns present in the memtable. */
-    public final Gauge<Long> memtableColumnsCount;
-    /** Number of times flush has resulted in the memtable being switched out. */
-    public final Counter memtableSwitchCount;
-    /** Current compression ratio for all SSTables */
-    public final Gauge<Double> compressionRatio;
-    /** Histogram of estimated row size (in bytes). */
-    public final Gauge<long[]> estimatedRowSizeHistogram;
-    /** Approximate number of keys in table. */
-    public final Gauge<Long> estimatedRowCount;
-    /** Histogram of estimated number of columns. */
-    public final Gauge<long[]> estimatedColumnCountHistogram;
-    /** Histogram of the number of sstable data files accessed per read */
-    public final ColumnFamilyHistogram sstablesPerReadHistogram;
-    /** (Local) read metrics */
-    public final LatencyMetrics readLatency;
-    /** (Local) range slice metrics */
-    public final LatencyMetrics rangeLatency;
-    /** (Local) write metrics */
-    public final LatencyMetrics writeLatency;
-    /** Estimated number of tasks pending for this column family */
-    public final Counter pendingFlushes;
-    /** Estimate of number of pending compactios for this CF */
-    public final Gauge<Integer> pendingCompactions;
-    /** Number of SSTables on disk for this CF */
-    public final Gauge<Integer> liveSSTableCount;
-    /** Disk space used by SSTables belonging to this CF */
-    public final Counter liveDiskSpaceUsed;
-    /**
-     * Total disk space used by SSTables belonging to this CF, including
-     * obsolete ones waiting to be GC'd
-     */
-    public final Counter totalDiskSpaceUsed;
-    /** Size of the smallest compacted row */
-    public final Gauge<Long> minRowSize;
-    /** Size of the largest compacted row */
-    public final Gauge<Long> maxRowSize;
-    /** Size of the smallest compacted row */
-    public final Gauge<Long> meanRowSize;
-    /** Number of false positives in bloom filter */
-    public final Gauge<Long> bloomFilterFalsePositives;
-    /** Number of false positives in bloom filter from last read */
-    public final Gauge<Long> recentBloomFilterFalsePositives;
-    /** False positive ratio of bloom filter */
-    public final Gauge<Double> bloomFilterFalseRatio;
-    /** False positive ratio of bloom filter from last read */
-    public final Gauge<Double> recentBloomFilterFalseRatio;
-    /** Disk space used by bloom filter */
-    public final Gauge<Long> bloomFilterDiskSpaceUsed;
-    /** Off heap memory used by bloom filter */
-    public final Gauge<Long> bloomFilterOffHeapMemoryUsed;
-    /** Off heap memory used by index summary */
-    public final Gauge<Long> indexSummaryOffHeapMemoryUsed;
-    /** Off heap memory used by compression meta data */
-    public final Gauge<Long> compressionMetadataOffHeapMemoryUsed;
-    /** Key cache hit rate for this CF */
-    public final Gauge<Double> keyCacheHitRate;
-    /** Tombstones scanned in queries on this CF */
-    public final ColumnFamilyHistogram tombstoneScannedHistogram;
-    /** Live cells scanned in queries on this CF */
-    public final ColumnFamilyHistogram liveScannedHistogram;
-    /** Column update time delta on this CF */
-    public final ColumnFamilyHistogram colUpdateTimeDeltaHistogram;
-    /** Disk space used by snapshot files which */
-    public final Gauge<Long> trueSnapshotsSize;
-    /** Row cache hits, but result out of range */
-    public final Counter rowCacheHitOutOfRange;
-    /** Number of row cache hits */
-    public final Counter rowCacheHit;
-    /** Number of row cache misses */
-    public final Counter rowCacheMiss;
-    /** CAS Prepare metrics */
-    public final LatencyMetrics casPrepare;
-    /** CAS Propose metrics */
-    public final LatencyMetrics casPropose;
-    /** CAS Commit metrics */
-    public final LatencyMetrics casCommit;
-
-    public final Timer coordinatorReadLatency;
-    public final Timer coordinatorScanLatency;
-
-    /** Time spent waiting for free memtable space, either on- or off-heap */
-    public final Timer waitingOnFreeMemtableSpace;
-
-    private final MetricNameFactory factory;
-    private static final MetricNameFactory globalNameFactory = new AllColumnFamilyMetricNameFactory();
-
-    public final Counter speculativeRetries;
-
-    // for backward compatibility
-    @Deprecated
-    public final EstimatedHistogramWrapper sstablesPerRead;
-    // it should not be called directly
-    @Deprecated
-    protected final RecentEstimatedHistogram recentSSTablesPerRead = new RecentEstimatedHistogram(35);
-    private String cfName;
-
-    public final static LatencyMetrics globalReadLatency = new LatencyMetrics(
-            "/column_family/metrics/read_latency", globalNameFactory, "Read");
-    public final static LatencyMetrics globalWriteLatency = new LatencyMetrics(
-            "/column_family/metrics/write_latency", globalNameFactory, "Write");
-    public final static LatencyMetrics globalRangeLatency = new LatencyMetrics(
-            "/column_family/metrics/range_latency", globalNameFactory, "Range");
-
-    /**
-     * stores metrics that will be rolled into a single global metric
-     */
-    public final static ConcurrentMap<String, Set<Metric>> allColumnFamilyMetrics = Maps
-            .newConcurrentMap();
-
-    /**
-     * Stores all metric names created that can be used when unregistering
-     */
-    public final static Set<String> all = Sets.newHashSet();
-
-    /**
-     * Creates metrics for given {@link ColumnFamilyStore}.
-     *
-     * @param cfs
-     *            ColumnFamilyStore to measure metrics
-     */
-    public ColumnFamilyMetrics(final ColumnFamilyStore cfs) {
-        factory = new ColumnFamilyMetricNameFactory(cfs);
-        cfName = cfs.getCFName();
-        memtableColumnsCount = createColumnFamilyGauge(
-                "/column_family/metrics/memtable_columns_count",
-                "MemtableColumnsCount");
-        memtableOnHeapSize = createColumnFamilyGauge(
-                "/column_family/metrics/memtable_on_heap_size",
-                "MemtableOnHeapSize");
-        memtableOffHeapSize = createColumnFamilyGauge(
-                "/column_family/metrics/memtable_off_heap_size",
-                "MemtableOffHeapSize");
-        memtableLiveDataSize = createColumnFamilyGauge(
-                "/column_family/metrics/memtable_live_data_size",
-                "MemtableLiveDataSize");
-        allMemtablesOnHeapSize = createColumnFamilyGauge(
-                "/column_family/metrics/all_memtables_on_heap_size",
-                "AllMemtablesHeapSize");
-        allMemtablesOffHeapSize = createColumnFamilyGauge(
-                "/column_family/metrics/all_memtables_off_heap_size",
-                "AllMemtablesOffHeapSize");
-        allMemtablesLiveDataSize = createColumnFamilyGauge(
-                "/column_family/metrics/all_memtables_live_data_size",
-                "AllMemtablesLiveDataSize");
-        memtableSwitchCount = createColumnFamilyCounter(
-                "/column_family/metrics/memtable_switch_count",
-                "MemtableSwitchCount");
-        estimatedRowSizeHistogram = Metrics.newGauge(
-                factory.createMetricName("EstimatedRowSizeHistogram"),
-                new Gauge<long[]>() {
-                    public long[] value() {
-                        return c.getEstimatedHistogramAsLongArrValue("/column_family/metrics/estimated_row_size_histogram/"
-                                + cfName);
-                    }
-                });
-        estimatedRowCount= Metrics.newGauge(
-                factory.createMetricName("EstimatedRowCount"),
-                new Gauge<Long>() {
-                    public Long value() {
-                        return c.getLongValue("/column_family/metrics/estimated_row_count/"
-                                + cfName);
-                    }
-                });
-
-        estimatedColumnCountHistogram = Metrics.newGauge(
-                factory.createMetricName("EstimatedColumnCountHistogram"),
-                new Gauge<long[]>() {
-                    public long[] value() {
-                        return c.getEstimatedHistogramAsLongArrValue("/column_family/metrics/estimated_column_count_histogram/"
-                                + cfName);
-                    }
-                });
-        sstablesPerReadHistogram = createColumnFamilyHistogram(
-                "/column_family/metrics/sstables_per_read_histogram",
-                "SSTablesPerReadHistogram");
-        compressionRatio = createColumnFamilyGauge("CompressionRatio",
-                new Gauge<Double>() {
-                    public Double value() {
-                        return c.getDoubleValue("/column_family/metrics/compression_ratio/"
-                                + cfName);
-                    }
-                }, new Gauge<Double>() // global gauge
-                {
-                    public Double value() {
-                        return c.getDoubleValue("/column_family/metrics/compression_ratio/");
-                    }
-                });
-        readLatency = new LatencyMetrics("/column_family/metrics/read_latency",
-                cfName, factory, "Read");
-        writeLatency = new LatencyMetrics(
-                "/column_family/metrics/write_latency", cfName, factory,
-                "Write");
-        rangeLatency = new LatencyMetrics(
-                "/column_family/metrics/range_latency", cfName, factory,
-                "Range");
-        pendingFlushes = createColumnFamilyCounter(
-                "/column_family/metrics/pending_flushes", "PendingFlushes");
-        pendingCompactions = createColumnFamilyGaugeInt(
-                "/column_family/metrics/pending_compactions",
-                "PendingCompactions");
-        liveSSTableCount = createColumnFamilyGaugeInt(
-                "/column_family/metrics/live_ss_table_count",
-                "LiveSSTableCount");
-        liveDiskSpaceUsed = createColumnFamilyCounter(
-                "/column_family/metrics/live_disk_space_used",
-                "LiveDiskSpaceUsed");
-        totalDiskSpaceUsed = createColumnFamilyCounter(
-                "/column_family/metrics/total_disk_space_used",
-                "TotalDiskSpaceUsed");
-        minRowSize = createColumnFamilyGauge(
-                "/column_family/metrics/min_row_size", "MinRowSize");
-        maxRowSize = createColumnFamilyGauge(
-                "/column_family/metrics/max_row_size", "MaxRowSize");
-        meanRowSize = createColumnFamilyGauge(
-                "/column_family/metrics/mean_row_size", "MeanRowSize");
-        bloomFilterFalsePositives = createColumnFamilyGauge(
-                "/column_family/metrics/bloom_filter_false_positives",
-                "BloomFilterFalsePositives");
-        recentBloomFilterFalsePositives = createColumnFamilyGauge(
-                "/column_family/metrics/recent_bloom_filter_false_positives",
-                "RecentBloomFilterFalsePositives");
-        bloomFilterFalseRatio = createColumnFamilyGaugeDouble(
-                "/column_family/metrics/bloom_filter_false_ratio",
-                "BloomFilterFalseRatio");
-        recentBloomFilterFalseRatio = createColumnFamilyGaugeDouble(
-                "/column_family/metrics/recent_bloom_filter_false_ratio",
-                "RecentBloomFilterFalseRatio");
-        bloomFilterDiskSpaceUsed = createColumnFamilyGauge(
-                "/column_family/metrics/bloom_filter_disk_space_used",
-                "BloomFilterDiskSpaceUsed");
-        bloomFilterOffHeapMemoryUsed = createColumnFamilyGauge(
-                "/column_family/metrics/bloom_filter_off_heap_memory_used",
-                "BloomFilterOffHeapMemoryUsed");
-        indexSummaryOffHeapMemoryUsed = createColumnFamilyGauge(
-                "/column_family/metrics/index_summary_off_heap_memory_used",
-                "IndexSummaryOffHeapMemoryUsed");
-        compressionMetadataOffHeapMemoryUsed = createColumnFamilyGauge(
-                "/column_family/metrics/compression_metadata_off_heap_memory_used",
-                "CompressionMetadataOffHeapMemoryUsed");
-        speculativeRetries = createColumnFamilyCounter(
-                "/column_family/metrics/speculative_retries",
-                "SpeculativeRetries");
-        keyCacheHitRate = Metrics.newGauge(
-                factory.createMetricName("KeyCacheHitRate"),
-                new Gauge<Double>() {
-                    @Override
-                    public Double value() {
-                        return c.getDoubleValue("/column_family/metrics/key_cache_hit_rate/"
-                                + cfName);
-                    }
-                });
-        tombstoneScannedHistogram = createColumnFamilyHistogram(
-                "/column_family/metrics/tombstone_scanned_histogram",
-                "TombstoneScannedHistogram");
-        liveScannedHistogram = createColumnFamilyHistogram(
-                "/column_family/metrics/live_scanned_histogram",
-                "LiveScannedHistogram");
-        colUpdateTimeDeltaHistogram = createColumnFamilyHistogram(
-                "/column_family/metrics/col_update_time_delta_histogram",
-                "ColUpdateTimeDeltaHistogram");
-        coordinatorReadLatency = APIMetrics.newTimer("/column_family/metrics/coordinator/read/" + cfName,
-                factory.createMetricName("CoordinatorReadLatency"),
-                TimeUnit.MICROSECONDS, TimeUnit.SECONDS);
-        coordinatorScanLatency = APIMetrics.newTimer("/column_family/metrics/coordinator/scan/" + cfName,
-                factory.createMetricName("CoordinatorScanLatency"),
-                TimeUnit.MICROSECONDS, TimeUnit.SECONDS);
-        waitingOnFreeMemtableSpace = APIMetrics.newTimer("/column_family/metrics/waiting_on_free_memtable/" + cfName,
-                factory.createMetricName("WaitingOnFreeMemtableSpace"),
-                TimeUnit.MICROSECONDS, TimeUnit.SECONDS);
-
-        trueSnapshotsSize = createColumnFamilyGauge(
-                "/column_family/metrics/snapshots_size", "SnapshotsSize");
-        rowCacheHitOutOfRange = createColumnFamilyCounter(
-                "/column_family/metrics/row_cache_hit_out_of_range",
-                "RowCacheHitOutOfRange");
-        rowCacheHit = createColumnFamilyCounter(
-                "/column_family/metrics/row_cache_hit", "RowCacheHit");
-        rowCacheMiss = createColumnFamilyCounter(
-                "/column_family/metrics/row_cache_miss", "RowCacheMiss");
-
-        casPrepare = new LatencyMetrics("/column_family/metrics/cas_prepare/"
-                + cfName, factory, "CasPrepare");
-        casPropose = new LatencyMetrics("/column_family/metrics/cas_propose/"
-                + cfName, factory, "CasPropose");
-        casCommit = new LatencyMetrics("/column_family/metrics/cas_commit/"
-                + cfName, factory, "CasCommit");
-        sstablesPerRead = new EstimatedHistogramWrapper("/column_family/metrics/sstables_per_read_histogram/" + cfName);
-    }
-
-    /**
-     * Release all associated metrics.
-     */
-    public void release() {
-        for (String name : all) {
-            allColumnFamilyMetrics.get(name).remove(
-                    Metrics.defaultRegistry().allMetrics()
-                            .get(factory.createMetricName(name)));
-            Metrics.defaultRegistry().removeMetric(
-                    factory.createMetricName(name));
-        }
-        readLatency.release();
-        writeLatency.release();
-        rangeLatency.release();
-        Metrics.defaultRegistry().removeMetric(
-                factory.createMetricName("EstimatedRowSizeHistogram"));
-        Metrics.defaultRegistry().removeMetric(
-                factory.createMetricName("EstimatedColumnCountHistogram"));
-        Metrics.defaultRegistry().removeMetric(
-                factory.createMetricName("KeyCacheHitRate"));
-        Metrics.defaultRegistry().removeMetric(
-                factory.createMetricName("CoordinatorReadLatency"));
-        Metrics.defaultRegistry().removeMetric(
-                factory.createMetricName("CoordinatorScanLatency"));
-        Metrics.defaultRegistry().removeMetric(
-                factory.createMetricName("WaitingOnFreeMemtableSpace"));
-    }
-
-    /**
-     * Create a gauge that will be part of a merged version of all column
-     * families. The global gauge will merge each CF gauge by adding their
-     * values
-     */
-    protected Gauge<Double> createColumnFamilyGaugeDouble(final String url,
-            final String name) {
-        Gauge<Double> gauge = new Gauge<Double>() {
-            public Double value() {
-                return c.getDoubleValue(url + "/" + cfName);
-            }
-        };
-        return createColumnFamilyGauge(url, name, gauge);
-    }
-
-    /**
-     * Create a gauge that will be part of a merged version of all column
-     * families. The global gauge will merge each CF gauge by adding their
-     * values
-     */
-    protected Gauge<Long> createColumnFamilyGauge(final String url, final String name) {
-        Gauge<Long> gauge = new Gauge<Long>() {
-            public Long value() {
-                return (long)c.getDoubleValue(url + "/" + cfName);
-            }
-        };
-        return createColumnFamilyGauge(url, name, gauge);
-    }
-
-    /**
-     * Create a gauge that will be part of a merged version of all column
-     * families. The global gauge will merge each CF gauge by adding their
-     * values
-     */
-    protected Gauge<Integer> createColumnFamilyGaugeInt(final String url,
-            final String name) {
-        Gauge<Integer> gauge = new Gauge<Integer>() {
-            public Integer value() {
-                return (int)c.getDoubleValue(url + "/" + cfName);
-            }
-        };
-        return createColumnFamilyGauge(url, name, gauge);
-    }
-
-    /**
-     * Create a gauge that will be part of a merged version of all column
-     * families. The global gauge will merge each CF gauge by adding their
-     * values
-     */
-    protected <T extends Number> Gauge<T> createColumnFamilyGauge(final String url,
-            final String name, Gauge<T> gauge) {
-        return createColumnFamilyGauge(name, gauge, new Gauge<Long>() {
-            public Long value() {
-                // This is an optimiztion, call once for all column families
-                // instead
-                // of iterating over all of them
-                return c.getLongValue(url);
-            }
-        });
-    }
-
-    /**
-     * Create a gauge that will be part of a merged version of all column
-     * families. The global gauge is defined as the globalGauge parameter
-     */
-    protected <G, T> Gauge<T> createColumnFamilyGauge(String name,
-            Gauge<T> gauge, Gauge<G> globalGauge) {
-        Gauge<T> cfGauge = APIMetrics.newGauge(factory.createMetricName(name),
-                gauge);
-        if (register(name, cfGauge)) {
-            Metrics.newGauge(globalNameFactory.createMetricName(name),
-                    globalGauge);
-        }
-        return cfGauge;
-    }
-
-    /**
-     * Creates a counter that will also have a global counter thats the sum of
-     * all counters across different column families
-     */
-    protected Counter createColumnFamilyCounter(final String url, final String name) {
-        Counter cfCounter = APIMetrics.newCounter(url + "/" + cfName,
-                factory.createMetricName(name));
-        if (register(name, cfCounter)) {
-            Metrics.newGauge(globalNameFactory.createMetricName(name),
-                    new Gauge<Long>() {
-                        public Long value() {
-                            // This is an optimiztion, call once for all column
-                            // families instead
-                            // of iterating over all of them
-                            return c.getLongValue(url);
-                        }
-                    });
-        }
-        return cfCounter;
-    }
-
-    /**
-     * Create a histogram-like interface that will register both a CF, keyspace
-     * and global level histogram and forward any updates to both
-     */
-    protected ColumnFamilyHistogram createColumnFamilyHistogram(String url,
-            String name) {
-        Histogram cfHistogram = APIMetrics.newHistogram(url + "/" + cfName,
-                factory.createMetricName(name), true);
-        register(name, cfHistogram);
-
-        // TBD add keyspace and global histograms
-        // keyspaceHistogram,
-        // Metrics.newHistogram(globalNameFactory.createMetricName(name),
-        // true));
-        return new ColumnFamilyHistogram(cfHistogram, null, null);
-    }
-
-    /**
-     * Registers a metric to be removed when unloading CF.
-     *
-     * @return true if first time metric with that name has been registered
-     */
-    private boolean register(String name, Metric metric) {
-        boolean ret = allColumnFamilyMetrics.putIfAbsent(name,
-                new HashSet<Metric>()) == null;
-        allColumnFamilyMetrics.get(name).add(metric);
-        all.add(name);
-        return ret;
-    }
-
-    public long[] getRecentSSTablesPerRead() {
-        return recentSSTablesPerRead
-                .getBuckets(sstablesPerRead.getBuckets(false));
-    }
-
-    public class ColumnFamilyHistogram {
-        public final Histogram[] all;
-        public final Histogram cf;
-
-        private ColumnFamilyHistogram(Histogram cf, Histogram keyspace,
-                Histogram global) {
-            this.cf = cf;
-            this.all = new Histogram[] { cf, keyspace, global };
-        }
-    }
-
-    class ColumnFamilyMetricNameFactory implements MetricNameFactory {
-        private final String keyspaceName;
-        private final String columnFamilyName;
-        private final boolean isIndex;
-
-        ColumnFamilyMetricNameFactory(ColumnFamilyStore cfs) {
-            this.keyspaceName = cfs.getKeyspace();
-            this.columnFamilyName = cfs.getColumnFamilyName();
-            isIndex = cfs.isIndex();
-        }
-
-        public MetricName createMetricName(String metricName) {
-            String groupName = ColumnFamilyMetrics.class.getPackage().getName();
-            String type = isIndex ? "IndexColumnFamily" : "ColumnFamily";
-
-            StringBuilder mbeanName = new StringBuilder();
-            mbeanName.append(groupName).append(":");
-            mbeanName.append("type=").append(type);
-            mbeanName.append(",keyspace=").append(keyspaceName);
-            mbeanName.append(",scope=").append(columnFamilyName);
-            mbeanName.append(",name=").append(metricName);
-            return new MetricName(groupName, type, metricName, keyspaceName
-                    + "." + columnFamilyName, mbeanName.toString());
-        }
-    }
-
-    static class AllColumnFamilyMetricNameFactory implements MetricNameFactory {
-        public MetricName createMetricName(String metricName) {
-            String groupName = ColumnFamilyMetrics.class.getPackage().getName();
-            StringBuilder mbeanName = new StringBuilder();
-            mbeanName.append(groupName).append(":");
-            mbeanName.append("type=ColumnFamily");
-            mbeanName.append(",name=").append(metricName);
-            return new MetricName(groupName, "ColumnFamily", metricName, "all",
-                    mbeanName.toString());
-        }
-    }
-}
diff --git a/src/main/java/org/apache/cassandra/metrics/CommitLogMetrics.java b/src/main/java/org/apache/cassandra/metrics/CommitLogMetrics.java
index e3a1aa9..9a7be67 100644
--- a/src/main/java/org/apache/cassandra/metrics/CommitLogMetrics.java
+++ b/src/main/java/org/apache/cassandra/metrics/CommitLogMetrics.java
@@ -23,65 +23,38 @@
  */
 package org.apache.cassandra.metrics;
 
-import com.scylladb.jmx.api.APIClient;
-import com.scylladb.jmx.metrics.APIMetrics;
-import com.scylladb.jmx.metrics.DefaultNameFactory;
-import com.scylladb.jmx.metrics.MetricNameFactory;
-import com.yammer.metrics.core.Gauge;
-import com.yammer.metrics.core.Timer;
-
-import java.util.concurrent.TimeUnit;
+import javax.management.MalformedObjectNameException;
 
 /**
  * Metrics for commit log
  */
-public class CommitLogMetrics {
-    public static final MetricNameFactory factory = new DefaultNameFactory(
-            "CommitLog");
-    private APIClient c = new APIClient();
-
-    /** Number of completed tasks */
-    public final Gauge<Long> completedTasks;
-    /** Number of pending tasks */
-    public final Gauge<Long> pendingTasks;
-    /** Current size used by all the commit log segments */
-    public final Gauge<Long> totalCommitLogSize;
-    /**
-     * Time spent waiting for a CLS to be allocated - under normal conditions
-     * this should be zero
-     */
-    public final Timer waitingOnSegmentAllocation;
-    /**
-     * The time spent waiting on CL sync; for Periodic this is only occurs when
-     * the sync is lagging its sync interval
-     */
-    public final Timer waitingOnCommit;
-
+public class CommitLogMetrics implements Metrics {
     public CommitLogMetrics() {
-        completedTasks = APIMetrics.newGauge(
-                factory.createMetricName("CompletedTasks"), new Gauge<Long>() {
-                    public Long value() {
-                        return c.getLongValue("/commitlog/metrics/completed_tasks");
-                    }
-                });
-        pendingTasks = APIMetrics.newGauge(
-                factory.createMetricName("PendingTasks"), new Gauge<Long>() {
-                    public Long value() {
-                        return c.getLongValue("/commitlog/metrics/pending_tasks");
-                    }
-                });
-        totalCommitLogSize = APIMetrics.newGauge(
-                factory.createMetricName("TotalCommitLogSize"),
-                new Gauge<Long>() {
-                    public Long value() {
-                        return c.getLongValue("/commitlog/metrics/total_commit_log_size");
-                    }
-                });
-        waitingOnSegmentAllocation = APIMetrics.newTimer("/commit_log/metrics/waiting_on_segment_allocation",
-                factory.createMetricName("WaitingOnSegmentAllocation"),
-                TimeUnit.MICROSECONDS, TimeUnit.SECONDS);
-        waitingOnCommit = APIMetrics.newTimer("/commit_log/metrics/waiting_on_commit",
-                factory.createMetricName("WaitingOnCommit"),
-                TimeUnit.MICROSECONDS, TimeUnit.SECONDS);
+    }
+
+    @Override
+    public void register(MetricsRegistry registry) throws MalformedObjectNameException {
+        MetricNameFactory factory = new DefaultNameFactory("CommitLog");
+        /** Number of completed tasks */
+        registry.register(() -> registry.gauge("/commitlog/metrics/completed_tasks"),
+                factory.createMetricName("CompletedTasks"));
+        /** Number of pending tasks */
+        registry.register(() -> registry.gauge("/commitlog/metrics/pending_tasks"),
+                factory.createMetricName("PendingTasks"));
+        /** Current size used by all the commit log segments */
+        registry.register(() -> registry.gauge("/commitlog/metrics/total_commit_log_size"),
+                factory.createMetricName("TotalCommitLogSize"));
+        /**
+         * Time spent waiting for a CLS to be allocated - under normal
+         * conditions this should be zero
+         */
+        registry.register(() -> registry.timer("/commitlog/metrics/waiting_on_segment_allocation"),
+                factory.createMetricName("WaitingOnSegmentAllocation"));
+        /**
+         * The time spent waiting on CL sync; for Periodic this is only occurs
+         * when the sync is lagging its sync interval
+         */
+        registry.register(() -> registry.timer("/commitlog/metrics/waiting_on_commit"),
+                factory.createMetricName("WaitingOnCommit"));
     }
 }
diff --git a/src/main/java/org/apache/cassandra/metrics/CompactionMetrics.java b/src/main/java/org/apache/cassandra/metrics/CompactionMetrics.java
index d996ee2..afd53d3 100644
--- a/src/main/java/org/apache/cassandra/metrics/CompactionMetrics.java
+++ b/src/main/java/org/apache/cassandra/metrics/CompactionMetrics.java
@@ -23,52 +23,30 @@
  */
 package org.apache.cassandra.metrics;
 
-import java.util.concurrent.TimeUnit;
-
-import com.scylladb.jmx.api.APIClient;
-import com.scylladb.jmx.metrics.APIMetrics;
-import com.scylladb.jmx.metrics.DefaultNameFactory;
-import com.scylladb.jmx.metrics.MetricNameFactory;
-import com.yammer.metrics.core.Counter;
-import com.yammer.metrics.core.Gauge;
-import com.yammer.metrics.core.APIMeter;
+import javax.management.MalformedObjectNameException;
 
 /**
  * Metrics for compaction.
  */
-public class CompactionMetrics {
-    public static final MetricNameFactory factory = new DefaultNameFactory(
-            "Compaction");
-    private APIClient c = new APIClient();
-    /** Estimated number of compactions remaining to perform */
-    public final Gauge<Integer> pendingTasks;
-    /** Number of completed compactions since server [re]start */
-    public final Gauge<Long> completedTasks;
-    /** Total number of compactions since server [re]start */
-    public final APIMeter totalCompactionsCompleted;
-    /** Total number of bytes compacted since server [re]start */
-    public final Counter bytesCompacted;
-
+public class CompactionMetrics implements Metrics {
     public CompactionMetrics() {
+    }
 
-        pendingTasks = APIMetrics.newGauge(
-                factory.createMetricName("PendingTasks"), new Gauge<Integer>() {
-                    public Integer value() {
-                        return c.getIntValue("/compaction_manager/metrics/pending_tasks");
-                    }
-                });
-        completedTasks = APIMetrics.newGauge(
-                factory.createMetricName("CompletedTasks"), new Gauge<Long>() {
-                    public Long value() {
-                        return c.getLongValue("/compaction_manager/metrics/completed_tasks");
-                    }
-                });
-        totalCompactionsCompleted = APIMetrics.newMeter(
-                "/compaction_manager/metrics/total_compactions_completed",
-                factory.createMetricName("TotalCompactionsCompleted"),
-                "compaction completed", TimeUnit.SECONDS);
-        bytesCompacted = APIMetrics.newCounter(
-                "/compaction_manager/metrics/bytes_compacted",
+    @Override
+    public void register(MetricsRegistry registry) throws MalformedObjectNameException {
+        MetricNameFactory factory = new DefaultNameFactory("Compaction");
+        /** Estimated number of compactions remaining to perform */
+        registry.register(() -> registry.gauge(Integer.class, "/compaction_manager/metrics/pending_tasks"),
+                factory.createMetricName("PendingTasks"));
+        /** Number of completed compactions since server [re]start */
+        registry.register(() -> registry.gauge("/compaction_manager/metrics/completed_tasks"),
+                factory.createMetricName("CompletedTasks"));
+        /** Total number of compactions since server [re]start */
+        registry.register(() -> registry.meter("/compaction_manager/metrics/total_compactions_completed"),
+                factory.createMetricName("TotalCompactionsCompleted"));
+        /** Total number of bytes compacted since server [re]start */
+        registry.register(() -> registry.meter("/compaction_manager/metrics/bytes_compacted"),
                 factory.createMetricName("BytesCompacted"));
     }
+
 }
diff --git a/src/main/java/com/scylladb/jmx/metrics/DefaultNameFactory.java b/src/main/java/org/apache/cassandra/metrics/DefaultNameFactory.java
similarity index 71%
rename from src/main/java/com/scylladb/jmx/metrics/DefaultNameFactory.java
rename to src/main/java/org/apache/cassandra/metrics/DefaultNameFactory.java
index 1ba99d8..654a1a1 100644
--- a/src/main/java/com/scylladb/jmx/metrics/DefaultNameFactory.java
+++ b/src/main/java/org/apache/cassandra/metrics/DefaultNameFactory.java
@@ -15,15 +15,10 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
+package org.apache.cassandra.metrics;
 
-/*
- * Copyright 2015 Cloudius Systems
- *
- * Modified by Cloudius Systems
- */
-package com.scylladb.jmx.metrics;
-
-import com.yammer.metrics.core.MetricName;
+import javax.management.MalformedObjectNameException;
+import javax.management.ObjectName;
 
 /**
  * MetricNameFactory that generates default MetricName of metrics.
@@ -43,19 +38,14 @@ public class DefaultNameFactory implements MetricNameFactory {
         this.scope = scope;
     }
 
-    public MetricName createMetricName(String metricName) {
+    @Override
+    public ObjectName createMetricName(String metricName) throws MalformedObjectNameException {
         return createMetricName(type, metricName, scope);
     }
 
-    public static MetricName createMetricName(String type, String metricName,
-            String scope) {
-        return new MetricName(GROUP_NAME, type, metricName, scope,
-                createDefaultMBeanName(type, metricName, scope));
-    }
-
-    protected static String createDefaultMBeanName(String type, String name,
-            String scope) {
-        final StringBuilder nameBuilder = new StringBuilder();
+    public static ObjectName createMetricName(String type, String name, String scope)
+            throws MalformedObjectNameException {
+        StringBuilder nameBuilder = new StringBuilder();
         nameBuilder.append(GROUP_NAME);
         nameBuilder.append(":type=");
         nameBuilder.append(type);
@@ -67,6 +57,6 @@ public class DefaultNameFactory implements MetricNameFactory {
             nameBuilder.append(",name=");
             nameBuilder.append(name);
         }
-        return nameBuilder.toString();
+        return new ObjectName(nameBuilder.toString());
     }
 }
diff --git a/src/main/java/org/apache/cassandra/metrics/DroppedMessageMetrics.java b/src/main/java/org/apache/cassandra/metrics/DroppedMessageMetrics.java
index fe9140a..e67e0c1 100644
--- a/src/main/java/org/apache/cassandra/metrics/DroppedMessageMetrics.java
+++ b/src/main/java/org/apache/cassandra/metrics/DroppedMessageMetrics.java
@@ -24,42 +24,27 @@
 
 package org.apache.cassandra.metrics;
 
-import java.util.concurrent.TimeUnit;
+import javax.management.MalformedObjectNameException;
 
 import org.apache.cassandra.net.MessagingService;
 
-import com.scylladb.jmx.metrics.APIMetrics;
-import com.scylladb.jmx.metrics.DefaultNameFactory;
-import com.scylladb.jmx.metrics.MetricNameFactory;
-import com.yammer.metrics.core.APIMeter;
-
 /**
  * Metrics for dropped messages by verb.
  */
-public class DroppedMessageMetrics {
-    /** Number of dropped messages */
-    public final APIMeter dropped;
-
-    private long lastDropped = 0;
+public class DroppedMessageMetrics implements Metrics {
+    private final MessagingService.Verb verb;
 
     public DroppedMessageMetrics(MessagingService.Verb verb) {
-        MetricNameFactory factory = new DefaultNameFactory("DroppedMessage",
-                verb.toString());
-        dropped = (APIMeter) APIMetrics.newMeter(null,
-                factory.createMetricName("Dropped"), "dropped",
-                TimeUnit.SECONDS);
-        dropped.stop();
+        this.verb = verb;
     }
 
-    @Deprecated
-    public int getRecentlyDropped() {
-        long currentDropped = dropped.count();
-        long recentlyDropped = currentDropped - lastDropped;
-        lastDropped = currentDropped;
-        return (int) recentlyDropped;
-    }
+    @Override
+    public void register(MetricsRegistry registry) throws MalformedObjectNameException {
+        MetricNameFactory factory = new DefaultNameFactory("DroppedMessage", verb.toString());
+        /** Number of dropped messages */
+        // TODO: this API url does not exist. Add meter calls for verbs.
+        registry.register(() -> registry.meter("/messaging_service/messages/dropped/" + verb),
+                factory.createMetricName("Dropped"));
 
-    public APIMeter getMeter() {
-        return dropped;
     }
 }
diff --git a/src/main/java/org/apache/cassandra/metrics/EstimatedHistogramWrapper.java b/src/main/java/org/apache/cassandra/metrics/EstimatedHistogramWrapper.java
deleted file mode 100644
index 3b511fb..0000000
--- a/src/main/java/org/apache/cassandra/metrics/EstimatedHistogramWrapper.java
+++ /dev/null
@@ -1,55 +0,0 @@
-package org.apache.cassandra.metrics;
-/*
- * Copyright (C) 2015 ScyllaDB
- */
-
-/*
- * This file is part of Scylla.
- *
- * Scylla is free software: you can redistribute it and/or modify
- * it under the terms of the GNU Affero General Public License as published by
- * the Free Software Foundation, either version 3 of the License, or
- * (at your option) any later version.
- *
- * Scylla is distributed in the hope that it will be useful,
- * but WITHOUT ANY WARRANTY; without even the implied warranty of
- * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
- * GNU General Public License for more details.
- *
- * You should have received a copy of the GNU General Public License
- * along with Scylla.  If not, see <http://www.gnu.org/licenses/>.
- */
-
-import javax.ws.rs.core.MultivaluedMap;
-
-import com.scylladb.jmx.api.APIClient;
-import com.scylladb.jmx.utils.EstimatedHistogram;
-
-public class EstimatedHistogramWrapper {
-    private APIClient c = new APIClient();
-    private String url;
-    private MultivaluedMap<String, String> queryParams;
-    private static final int DURATION = 50;
-    private int duration;
-    public EstimatedHistogramWrapper(String url, MultivaluedMap<String, String> queryParams, int duration) {
-        this.url = url;
-        this.queryParams = queryParams;
-        this.duration = duration;
-
-    }
-    public EstimatedHistogramWrapper(String url) {
-        this(url, null, DURATION);
-
-    }
-    public EstimatedHistogramWrapper(String url, int duration) {
-        this(url, null, duration);
-
-    }
-    public EstimatedHistogram get() {
-        return c.getEstimatedHistogram(url, queryParams, duration);
-    }
-
-    public long[] getBuckets(boolean reset) {
-        return get().getBuckets(reset);
-    }
-}
diff --git a/src/main/java/org/apache/cassandra/metrics/LatencyMetrics.java b/src/main/java/org/apache/cassandra/metrics/LatencyMetrics.java
index 1f4a9b4..4346f6a 100644
--- a/src/main/java/org/apache/cassandra/metrics/LatencyMetrics.java
+++ b/src/main/java/org/apache/cassandra/metrics/LatencyMetrics.java
@@ -23,41 +23,19 @@
  */
 package org.apache.cassandra.metrics;
 
-import java.util.List;
-import java.util.concurrent.TimeUnit;
+import java.util.Arrays;
 
-import com.google.common.collect.ImmutableList;
-import com.google.common.collect.Lists;
-import com.scylladb.jmx.metrics.APIMetrics;
-import com.scylladb.jmx.metrics.DefaultNameFactory;
-import com.scylladb.jmx.metrics.MetricNameFactory;
-import com.scylladb.jmx.utils.RecentEstimatedHistogram;
-import com.yammer.metrics.core.Counter;
-import com.yammer.metrics.core.Timer;
+import javax.management.MalformedObjectNameException;
+import javax.management.ObjectName;
 
 /**
  * Metrics about latencies
  */
-public class LatencyMetrics {
-    /** Latency */
-    public final Timer latency;
-    /** Total latency in micro sec */
-    public final Counter totalLatency;
-
-    /** parent metrics to replicate any updates to **/
-    private List<LatencyMetrics> parents = Lists.newArrayList();
-
-    protected final MetricNameFactory factory;
+public class LatencyMetrics implements Metrics {
+    protected final MetricNameFactory[] factories;
     protected final String namePrefix;
-
-    @Deprecated public EstimatedHistogramWrapper totalLatencyHistogram;
-    /*
-     * It should not be called directly, use the getRecentLatencyHistogram
-     */
-    @Deprecated protected final RecentEstimatedHistogram recentLatencyHistogram = new RecentEstimatedHistogram();
-
-    protected long lastLatency;
-    protected long lastOpCount;
+    protected final String uri;
+    protected final String param;
 
     /**
      * Create LatencyMetrics with given group, type, and scope. Name prefix for
@@ -68,8 +46,8 @@ public class LatencyMetrics {
      * @param scope
      *            Scope
      */
-    public LatencyMetrics(String url, String type, String scope) {
-        this(url, type, "", scope);
+    public LatencyMetrics(String type, String scope, String uri) {
+        this(type, "", scope, uri, null);
     }
 
     /**
@@ -83,83 +61,35 @@ public class LatencyMetrics {
      * @param scope
      *            Scope of metrics
      */
-    public LatencyMetrics(String url, String type, String namePrefix,
-            String scope) {
-        this(url, new DefaultNameFactory(type, scope), namePrefix);
+    public LatencyMetrics(String type, String namePrefix, String scope, String uri, String param) {
+        this(namePrefix, uri, param, new DefaultNameFactory(type, scope));
     }
 
-    /**
-     * Create LatencyMetrics with given group, type, prefix to append to each
-     * metric name, and scope.
-     *
-     * @param factory
-     *            MetricName factory to use
-     * @param namePrefix
-     *            Prefix to append to each metric name
-     */
-    public LatencyMetrics(String url, MetricNameFactory factory,
-            String namePrefix) {
-        this(url, null, factory, namePrefix);
+    public LatencyMetrics(String namePrefix, String uri, MetricNameFactory... factories) {
+        this(namePrefix, uri, null, factories);
     }
 
-    public LatencyMetrics(String url, String paramName,
-            MetricNameFactory factory, String namePrefix) {
-        this.factory = factory;
+    public LatencyMetrics(String namePrefix, String uri, String param, MetricNameFactory... factories) {
+        this.factories = factories;
         this.namePrefix = namePrefix;
-
-        paramName = (paramName == null)? "" : "/" + paramName;
-        latency = APIMetrics.newTimer(url + "/moving_average_histogram" + paramName,
-                factory.createMetricName(namePrefix + "Latency"),
-                TimeUnit.MICROSECONDS, TimeUnit.SECONDS);
-        totalLatency = APIMetrics.newCounter(url +  paramName,
-                factory.createMetricName(namePrefix + "TotalLatency"));
-        totalLatencyHistogram = new EstimatedHistogramWrapper(url + "/estimated_histogram" + paramName);
+        this.uri = uri;
+        this.param = param;
     }
 
-    /**
-     * Create LatencyMetrics with given group, type, prefix to append to each
-     * metric name, and scope. Any updates to this will also run on parent
-     *
-     * @param factory
-     *            MetricName factory to use
-     * @param namePrefix
-     *            Prefix to append to each metric name
-     * @param parents
-     *            any amount of parents to replicate updates to
-     */
-    public LatencyMetrics(String url, MetricNameFactory factory,
-            String namePrefix, LatencyMetrics... parents) {
-        this(url, factory, namePrefix);
-        this.parents.addAll(ImmutableList.copyOf(parents));
+    protected ObjectName[] names(String suffix) throws MalformedObjectNameException {
+        return Arrays.stream(factories).map(f -> {
+            try {
+                return f.createMetricName(namePrefix + suffix);
+            } catch (MalformedObjectNameException e) {
+                throw new RuntimeException(e); // dung...
+            }
+        }).toArray(size -> new ObjectName[size]);
     }
 
-    /** takes nanoseconds **/
-    public void addNano(long nanos) {
-        // the object is only updated from the API
-    }
-
-    public void release() {
-        APIMetrics.defaultRegistry()
-                .removeMetric(factory.createMetricName(namePrefix + "Latency"));
-        APIMetrics.defaultRegistry().removeMetric(
-                factory.createMetricName(namePrefix + "TotalLatency"));
-    }
-
-    @Deprecated
-    public synchronized double getRecentLatency() {
-        long ops = latency.count();
-        long n = totalLatency.count();
-        if (ops == lastOpCount)
-            return 0;
-        try {
-            return ((double) n - lastLatency) / (ops - lastOpCount);
-        } finally {
-            lastLatency = n;
-            lastOpCount = ops;
-        }
-    }
-
-    public long[] getRecentLatencyHistogram() {
-        return recentLatencyHistogram.getBuckets(totalLatencyHistogram.getBuckets(false));
+    @Override
+    public void register(MetricsRegistry registry) throws MalformedObjectNameException {
+        String paramName = (param == null) ? "" : "/" + param;
+        registry.register(() -> registry.timer(uri + "/moving_average_histogram" + paramName), names("Latency"));
+        registry.register(() -> registry.counter(uri + paramName), names("TotalLatency"));
     }
 }
diff --git a/src/main/java/org/apache/cassandra/metrics/StorageMetrics.java b/src/main/java/org/apache/cassandra/metrics/StorageMetrics.java
index aa87e99..d717412 100644
--- a/src/main/java/org/apache/cassandra/metrics/StorageMetrics.java
+++ b/src/main/java/org/apache/cassandra/metrics/StorageMetrics.java
@@ -23,27 +23,21 @@
  */
 package org.apache.cassandra.metrics;
 
-import com.scylladb.jmx.metrics.APIMetrics;
-import com.scylladb.jmx.metrics.DefaultNameFactory;
-import com.scylladb.jmx.metrics.MetricNameFactory;
-import com.yammer.metrics.core.Counter;
+import javax.management.MalformedObjectNameException;
 
 /**
  * Metrics related to Storage.
  */
-public class StorageMetrics {
-    private static final MetricNameFactory factory = new DefaultNameFactory(
-            "Storage");
-
-    public static final Counter load = APIMetrics.newCounter(
-            "/storage_service/metrics/load", factory.createMetricName("Load"));
-    public static final Counter exceptions = APIMetrics.newCounter(
-            "/storage_service/metrics/exceptions",
-            factory.createMetricName("Exceptions"));
-    public static final Counter totalHintsInProgress = APIMetrics.newCounter(
-            "/storage_service/metrics/hints_in_progress",
-            factory.createMetricName("TotalHintsInProgress"));
-    public static final Counter totalHints = APIMetrics.newCounter(
-            "/storage_service/metrics/total_hints",
-            factory.createMetricName("TotalHints"));
+public class StorageMetrics implements Metrics {
+    @Override
+    public void register(MetricsRegistry registry) throws MalformedObjectNameException {
+        MetricNameFactory factory = new DefaultNameFactory("Storage");
+        registry.register(() -> registry.counter("/storage_service/metrics/load"), factory.createMetricName("Load"));
+        registry.register(() -> registry.counter("/storage_service/metrics/exceptions"),
+                factory.createMetricName("Exceptions"));
+        registry.register(() -> registry.counter("/storage_service/metrics/hints_in_progress"),
+                factory.createMetricName("TotalHintsInProgress"));
+        registry.register(() -> registry.counter("/storage_service/metrics/total_hints"),
+                factory.createMetricName("TotalHints"));
+    }
 }
diff --git a/src/main/java/org/apache/cassandra/metrics/StreamingMetrics.java b/src/main/java/org/apache/cassandra/metrics/StreamingMetrics.java
index ce389ce..b2eb35f 100644
--- a/src/main/java/org/apache/cassandra/metrics/StreamingMetrics.java
+++ b/src/main/java/org/apache/cassandra/metrics/StreamingMetrics.java
@@ -23,84 +23,87 @@
  */
 package org.apache.cassandra.metrics;
 
+import static java.util.Arrays.asList;
+import static java.util.Collections.emptySet;
+import static org.apache.cassandra.metrics.DefaultNameFactory.createMetricName;
+
 import java.net.InetAddress;
-import java.util.HashMap;
+import java.net.UnknownHostException;
 import java.util.HashSet;
-import java.util.Map;
 import java.util.Set;
-import java.util.Timer;
-import java.util.TimerTask;
 
 import javax.json.JsonArray;
+import javax.management.MBeanServer;
+import javax.management.MalformedObjectNameException;
+import javax.management.ObjectName;
 
 import com.scylladb.jmx.api.APIClient;
-import com.scylladb.jmx.metrics.APIMetrics;
-import com.scylladb.jmx.metrics.DefaultNameFactory;
-import com.scylladb.jmx.metrics.MetricNameFactory;
-import com.yammer.metrics.core.Counter;
+import com.scylladb.jmx.metrics.APIMBean;
 
 /**
  * Metrics for streaming.
  */
-public class StreamingMetrics
-{
+public class StreamingMetrics {
     public static final String TYPE_NAME = "Streaming";
-    private static final Map<String, StreamingMetrics> instances = new HashMap<String, StreamingMetrics>();
-    static final int INTERVAL = 1000; //update every 1second
 
-    private static Timer timer = new Timer("Streaming Metrics");
+    private static final HashSet<ObjectName> globalNames;
 
-    public static final Counter activeStreamsOutbound = APIMetrics.newCounter("/stream_manager/metrics/outbound", DefaultNameFactory.createMetricName(TYPE_NAME, "ActiveOutboundStreams", null));
-    public static final Counter totalIncomingBytes = APIMetrics.newCounter("/stream_manager/metrics/incoming", DefaultNameFactory.createMetricName(TYPE_NAME, "TotalIncomingBytes", null));
-    public static final Counter totalOutgoingBytes = APIMetrics.newCounter("/stream_manager/metrics/outgoing", DefaultNameFactory.createMetricName(TYPE_NAME, "TotalOutgoingBytes", null));
-    public final Counter incomingBytes;
-    public final Counter outgoingBytes;
-    private static APIClient s_c = new APIClient();
-
-    public static void register_mbeans() {
-        TimerTask taskToExecute = new CheckRegistration();
-        timer.scheduleAtFixedRate(taskToExecute, 100, INTERVAL);
-    }
-
-    public StreamingMetrics(final InetAddress peer)
-    {
-        MetricNameFactory factory = new DefaultNameFactory("Streaming", peer.getHostAddress().replaceAll(":", "."));
-        incomingBytes = APIMetrics.newCounter("/stream_manager/metrics/incoming/" + peer,factory.createMetricName("IncomingBytes"));
-        outgoingBytes= APIMetrics.newCounter("/stream_manager/metrics/outgoing/" + peer, factory.createMetricName("OutgoingBytes"));
-    }
-
-    public static boolean checkRegistration() {
+    static {
         try {
-            JsonArray streams = s_c.getJsonArray("/stream_manager/");
-            Set<String> all = new HashSet<String>();
-            for (int i = 0; i < streams.size(); i ++) {
-                JsonArray sessions = streams.getJsonObject(i).getJsonArray("sessions");
-                for (int j = 0; j < sessions.size(); j++) {
-                    String name = sessions.getJsonObject(j).getString("peer");
-                    if (!instances.containsKey(name)) {
-                        StreamingMetrics metrics = new StreamingMetrics(InetAddress.getByName(name));
-                        instances.put(name, metrics);
-                    }
-                    all.add(name);
-                }
-            }
-            //removing deleted stream
-            for (String n : instances.keySet()) {
-                if (! all.contains(n)) {
-                    instances.remove(n);
-                }
-            }
-        } catch (Exception e) {
-            // ignoring exceptions, will retry on the next interval
-            return false;
+            globalNames = new HashSet<ObjectName>(asList(createMetricName(TYPE_NAME, "ActiveOutboundStreams", null),
+                    createMetricName(TYPE_NAME, "TotalIncomingBytes", null),
+                    createMetricName(TYPE_NAME, "TotalOutgoingBytes", null)));
+        } catch (MalformedObjectNameException e) {
+            throw new Error(e);
         }
-        return true;
+    };
+
+    private StreamingMetrics() {
     }
 
-    private static final class CheckRegistration extends TimerTask {
-        @Override
-        public void run() {
-            checkRegistration();
+    private static boolean isStreamingName(ObjectName n) {
+        return TYPE_NAME.equals(n.getKeyProperty("type"));
+    }
+
+    public static void unregister(APIClient client, MBeanServer server) throws MalformedObjectNameException {
+        APIMBean.checkRegistration(server, emptySet(), StreamingMetrics::isStreamingName, (n) -> null);
+    }
+
+    public static boolean checkRegistration(APIClient client, MBeanServer server)
+            throws MalformedObjectNameException, UnknownHostException {
+
+        Set<ObjectName> all = new HashSet<ObjectName>(globalNames);
+        JsonArray streams = client.getJsonArray("/stream_manager/");
+        for (int i = 0; i < streams.size(); i++) {
+            JsonArray sessions = streams.getJsonObject(i).getJsonArray("sessions");
+            for (int j = 0; j < sessions.size(); j++) {
+                String peer = sessions.getJsonObject(j).getString("peer");
+                String scope = InetAddress.getByName(peer).getHostAddress().replaceAll(":", ".");
+                all.add(createMetricName(TYPE_NAME, "IncomingBytes", scope));
+                all.add(createMetricName(TYPE_NAME, "OutgoingBytes", scope));
+            }
         }
+
+        MetricsRegistry registry = new MetricsRegistry(client, server);
+        return APIMBean.checkRegistration(server, all, StreamingMetrics::isStreamingName, n -> {
+            String scope = n.getKeyProperty("scope");
+            String name = n.getKeyProperty("name");
+
+            String url = null;
+            if ("ActiveOutboundStreams".equals(name)) {
+                url = "/stream_manager/metrics/outbound";
+            } else if ("IncomingBytes".equals(name) || "TotalIncomingBytes".equals(name)) {
+                url = "/stream_manager/metrics/incoming";
+            } else if ("OutgoingBytes".equals(name) || "TotalOutgoingBytes".equals(name)) {
+                url = "/stream_manager/metrics/outgoing";
+            }
+            if (url == null) {
+                throw new IllegalArgumentException();
+            }
+            if (scope != null) {
+                url = url + "/" + scope;
+            }
+            return registry.counter(url);
+        });
     }
 }