Flink Metrics Dropwizard

<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-metrics-dropwizard</artifactId>
    <version>2.1.0</version>
</dependency>

import org.apache.flink.dropwizard.ScheduledDropwizardReporter;
import org.apache.flink.dropwizard.metrics.FlinkCounterWrapper;
import org.apache.flink.dropwizard.metrics.FlinkGaugeWrapper;
import org.apache.flink.dropwizard.metrics.FlinkMeterWrapper;
import org.apache.flink.dropwizard.metrics.FlinkHistogramWrapper;
import org.apache.flink.dropwizard.metrics.DropwizardMeterWrapper;
import org.apache.flink.dropwizard.metrics.DropwizardHistogramWrapper;

import org.apache.flink.dropwizard.ScheduledDropwizardReporter;
import org.apache.flink.dropwizard.metrics.FlinkCounterWrapper;
import org.apache.flink.metrics.Counter;
import org.apache.flink.metrics.SimpleCounter;
import org.apache.flink.metrics.MetricConfig;
import com.codahale.metrics.ScheduledReporter;
import com.codahale.metrics.ConsoleReporter;
import java.util.concurrent.TimeUnit;

// Create a custom reporter extending ScheduledDropwizardReporter
public class CustomDropwizardReporter extends ScheduledDropwizardReporter {
    @Override
    public ScheduledReporter getReporter(MetricConfig config) {
        return ConsoleReporter.forRegistry(registry)
            .convertRatesTo(TimeUnit.SECONDS)
            .convertDurationsTo(TimeUnit.MILLISECONDS)
            .build();
    }
}

// Use Flink counter with Dropwizard
Counter flinkCounter = new SimpleCounter();
FlinkCounterWrapper dropwizardCounter = new FlinkCounterWrapper(flinkCounter);

// Increment and get count
dropwizardCounter.inc();
dropwizardCounter.inc(5);
long count = dropwizardCounter.getCount(); // Returns 6

// Use Dropwizard meter with Flink
com.codahale.metrics.Meter dropwizardMeter = new com.codahale.metrics.Meter();
DropwizardMeterWrapper flinkMeter = new DropwizardMeterWrapper(dropwizardMeter);

// Mark events and get rate
flinkMeter.markEvent();
flinkMeter.markEvent(10);
double rate = flinkMeter.getRate();

/**
 * Base class for MetricReporter that wraps a Dropwizard Reporter.
 * Automatically handles metric type detection and wrapper application.
 */
@PublicEvolving
public abstract class ScheduledDropwizardReporter 
    implements MetricReporter, Scheduled, Reporter, CharacterFilter {
    
    // Configuration constants
    public static final String ARG_HOST = "host";
    public static final String ARG_PORT = "port"; 
    public static final String ARG_PREFIX = "prefix";
    public static final String ARG_CONVERSION_RATE = "rateConversion";
    public static final String ARG_CONVERSION_DURATION = "durationConversion";
    
    /**
     * Opens the reporter with the given configuration
     * @param config Metric configuration
     */
    public void open(MetricConfig config);
    
    /**
     * Closes the reporter and stops reporting
     */
    public void close();
    
    /**
     * Called when a metric is added to the system
     * @param metric The metric instance
     * @param metricName The metric name
     * @param group The metric group
     */
    public void notifyOfAddedMetric(Metric metric, String metricName, MetricGroup group);
    
    /**
     * Called when a metric is removed from the system
     * @param metric The metric instance
     * @param metricName The metric name  
     * @param group The metric group
     */
    public void notifyOfRemovedMetric(Metric metric, String metricName, MetricGroup group);
    
    /**
     * Filters invalid characters from metric names
     * @param metricName The metric name to filter
     * @return Filtered metric name
     */
    public String filterCharacters(String metricName);
    
    /**
     * Reports all metrics to the underlying Dropwizard reporter
     */
    public void report();
    
    /**
     * Abstract method to create the underlying Dropwizard ScheduledReporter
     * @param config Metric configuration
     * @return ScheduledReporter instance
     */
    public abstract ScheduledReporter getReporter(MetricConfig config);
}

/**
 * Wrapper that allows a Flink counter to be used as a DropWizard counter
 */
public class FlinkCounterWrapper extends com.codahale.metrics.Counter {
    
    /**
     * Creates a wrapper around a Flink counter
     * @param counter The Flink counter to wrap
     */
    public FlinkCounterWrapper(Counter counter);
    
    /**
     * Gets the current count value
     * @return The current count
     */
    public long getCount();
    
    /**
     * Increments the counter by 1
     */
    public void inc();
    
    /**
     * Increments the counter by the given amount
     * @param n Amount to increment by
     */
    public void inc(long n);
    
    /**
     * Decrements the counter by 1
     */
    public void dec();
    
    /**
     * Decrements the counter by the given amount
     * @param n Amount to decrement by
     */
    public void dec(long n);
}

/**
 * Wrapper that allows a Flink gauge to be used as a DropWizard gauge
 */
public class FlinkGaugeWrapper<T> implements com.codahale.metrics.Gauge<T> {
    
    /**
     * Creates a wrapper around a Flink gauge
     * @param gauge The Flink gauge to wrap
     */
    public FlinkGaugeWrapper(Gauge<T> gauge);
    
    /**
     * Gets the current gauge value
     * @return The current value
     */
    public T getValue();
    
    /**
     * Static factory method to create a wrapper from any gauge
     * @param gauge The gauge to wrap
     * @return FlinkGaugeWrapper instance
     */
    public static <T> FlinkGaugeWrapper<T> fromGauge(Gauge<?> gauge);
}

/**
 * Wrapper to use a Flink Meter as a Dropwizard Meter.
 * Note: Only one minute rate is supported, other rates return 0.
 */
public class FlinkMeterWrapper extends com.codahale.metrics.Meter {
    
    /**
     * Creates a wrapper around a Flink meter
     * @param meter The Flink meter to wrap
     */
    public FlinkMeterWrapper(Meter meter);
    
    /**
     * Creates a wrapper around a Flink meter with custom clock
     * @param meter The Flink meter to wrap
     * @param clock Custom clock for timing
     */
    public FlinkMeterWrapper(Meter meter, Clock clock);
    
    /**
     * Marks a single event
     */
    public void mark();
    
    /**
     * Marks multiple events
     * @param n Number of events to mark
     */
    public void mark(long n);
    
    /**
     * Gets the total event count
     * @return Total count of events
     */
    public long getCount();
    
    /**
     * Gets the one minute rate (delegates to Flink meter's getRate())
     * @return One minute rate
     */
    public double getOneMinuteRate();
    
    /**
     * Gets the five minute rate (not supported, returns 0)
     * @return Always returns 0.0
     */
    public double getFiveMinuteRate();
    
    /**
     * Gets the fifteen minute rate (not supported, returns 0)
     * @return Always returns 0.0
     */
    public double getFifteenMinuteRate();
    
    /**
     * Gets the mean rate (not supported, returns 0)
     * @return Always returns 0.0
     */
    public double getMeanRate();
}

/**
 * Wrapper to use a Flink Histogram as a Dropwizard Histogram
 */
public class FlinkHistogramWrapper extends com.codahale.metrics.Histogram {
    
    /**
     * Creates a wrapper around a Flink histogram
     * @param histogram The Flink histogram to wrap
     */
    public FlinkHistogramWrapper(Histogram histogram);
    
    /**
     * Updates the histogram with a new value
     * @param value The value to add
     */
    public void update(long value);
    
    /**
     * Gets the total number of values recorded
     * @return Count of recorded values
     */
    public long getCount();
    
    /**
     * Gets a snapshot of the histogram statistics
     * @return Snapshot wrapped for Dropwizard compatibility
     */
    public Snapshot getSnapshot();
}

/**
 * Wrapper to use a Dropwizard Meter as a Flink Meter
 */
public class DropwizardMeterWrapper implements Meter {
    
    /**
     * Creates a wrapper around a Dropwizard meter
     * @param meter The Dropwizard meter to wrap
     */
    public DropwizardMeterWrapper(com.codahale.metrics.Meter meter);
    
    /**
     * Gets access to the underlying Dropwizard meter
     * @return The wrapped Dropwizard meter
     */
    public com.codahale.metrics.Meter getDropwizardMeter();
    
    /**
     * Marks a single event
     */
    public void markEvent();
    
    /**
     * Marks multiple events
     * @param n Number of events to mark
     */
    public void markEvent(long n);
    
    /**
     * Gets the rate (one minute rate from Dropwizard meter)
     * @return Current rate
     */
    public double getRate();
    
    /**
     * Gets the total event count
     * @return Total count of events
     */
    public long getCount();
}

/**
 * Wrapper to use a Dropwizard Histogram as a Flink Histogram
 */
public class DropwizardHistogramWrapper implements Histogram {
    
    /**
     * Creates a wrapper around a Dropwizard histogram
     * @param dropwizardHistogram The Dropwizard histogram to wrap
     */
    public DropwizardHistogramWrapper(com.codahale.metrics.Histogram dropwizardHistogram);
    
    /**
     * Gets access to the underlying Dropwizard histogram
     * @return The wrapped Dropwizard histogram
     */
    public com.codahale.metrics.Histogram getDropwizardHistogram();
    
    /**
     * Updates the histogram with a new value
     * @param value The value to add
     */
    public void update(long value);
    
    /**
     * Gets the total number of values recorded
     * @return Count of recorded values
     */
    public long getCount();
    
    /**
     * Gets histogram statistics compatible with Flink
     * @return HistogramStatistics instance
     */
    public HistogramStatistics getStatistics();
}

/**
 * Dropwizard histogram statistics implementation for DropwizardHistogramWrapper
 */
class DropwizardHistogramStatistics extends HistogramStatistics {
    
    /**
     * Gets the value at the specified quantile
     * @param quantile The quantile (0.0 to 1.0)
     * @return Value at the quantile
     */
    public double getQuantile(double quantile);
    
    /**
     * Gets all recorded values
     * @return Array of all values
     */
    public long[] getValues();
    
    /**
     * Gets the number of recorded values
     * @return Size of the dataset
     */
    public int size();
    
    /**
     * Gets the arithmetic mean of all values
     * @return Mean value
     */
    public double getMean();
    
    /**
     * Gets the standard deviation
     * @return Standard deviation
     */
    public double getStdDev();
    
    /**
     * Gets the maximum value
     * @return Maximum value
     */
    public long getMax();
    
    /**
     * Gets the minimum value
     * @return Minimum value
     */
    public long getMin();
}

/**
 * Wrapper to use Flink's HistogramStatistics as a Dropwizard Snapshot
 */
class HistogramStatisticsWrapper extends Snapshot {
    
    /**
     * Gets the value at the specified quantile
     * @param quantile The quantile (0.0 to 1.0)
     * @return Value at the quantile
     */
    public double getValue(double quantile);
    
    /**
     * Gets all recorded values
     * @return Array of all values
     */
    public long[] getValues();
    
    /**
     * Gets the number of recorded values
     * @return Size of the dataset
     */
    public int size();
    
    /**
     * Gets the maximum value
     * @return Maximum value
     */
    public long getMax();
    
    /**
     * Gets the arithmetic mean of all values
     * @return Mean value
     */
    public double getMean();
    
    /**
     * Gets the minimum value
     * @return Minimum value
     */
    public long getMin();
    
    /**
     * Gets the standard deviation
     * @return Standard deviation
     */
    public double getStdDev();
    
    /**
     * Dumps all histogram values to an output stream
     * @param output OutputStream to write values to
     */
    public void dump(OutputStream output);
}