Core Metric Types

/**
 * A Counter is a Metric that measures a count.
 */
public interface Counter extends Metric {
    /** Increment the current count by 1. */
    void inc();
    
    /**
     * Increment the current count by the given value.
     * @param n value to increment the current count by
     */
    void inc(long n);
    
    /** Decrement the current count by 1. */
    void dec();
    
    /**
     * Decrement the current count by the given value.
     * @param n value to decrement the current count by
     */
    void dec(long n);
    
    /**
     * Returns the current count.
     * @return current count
     */
    long getCount();
}

// Create and use a counter
Counter processedRecords = metricGroup.counter("records-processed");

// Increment operations
processedRecords.inc();           // +1
processedRecords.inc(10);         // +10

// Decrement operations  
processedRecords.dec();           // -1
processedRecords.dec(5);          // -5

// Read current value
long currentCount = processedRecords.getCount();

/**
 * A Gauge is a Metric that calculates a specific value at a point in time.
 * @param <T> the type of the measured value
 */
public interface Gauge<T> extends Metric {
    /**
     * Calculates and returns the measured value.
     * @return calculated value
     */
    T getValue();
}

// Create gauge for queue size
Gauge<Integer> queueSizeGauge = new Gauge<Integer>() {
    @Override
    public Integer getValue() {
        return taskQueue.size();
    }
};
metricGroup.gauge("queue-size", queueSizeGauge);

// Create gauge for memory usage
Gauge<Double> memoryUsageGauge = () -> {
    Runtime runtime = Runtime.getRuntime();
    long used = runtime.totalMemory() - runtime.freeMemory();
    return (double) used / runtime.maxMemory();
};
metricGroup.gauge("memory-usage-ratio", memoryUsageGauge);

// Lambda expression gauge
Gauge<String> statusGauge = () -> processor.getCurrentStatus().name();
metricGroup.gauge("processor-status", statusGauge);

/**
 * Metric for measuring throughput.
 */
public interface Meter extends Metric {
    /** Mark occurrence of an event. */
    void markEvent();
    
    /**
     * Mark occurrence of multiple events.
     * @param n number of events occurred
     */
    void markEvent(long n);
    
    /**
     * Returns the current rate of events per second.
     * @return current rate of events per second
     */
    double getRate();
    
    /**
     * Get number of events marked on the meter.
     * @return number of events marked on the meter
     */
    long getCount();
}

// Create a meter with 60-second time window
Meter throughputMeter = new MeterView(60);
metricGroup.meter("throughput", throughputMeter);

// Mark single events
throughputMeter.markEvent();

// Mark batch events
throughputMeter.markEvent(batchSize);

// Read current rate and total count
double currentRate = throughputMeter.getRate();    // events/second
long totalEvents = throughputMeter.getCount();     // total events marked

/**
 * Histogram interface to be used with Flink's metrics system.
 * The histogram allows to record values, get the current count of recorded 
 * values and create histogram statistics for the currently seen elements.
 */
public interface Histogram extends Metric {
    /**
     * Update the histogram with the given value.
     * @param value Value to update the histogram with
     */
    void update(long value);
    
    /**
     * Get the count of seen elements.
     * @return Count of seen elements
     */
    long getCount();
    
    /**
     * Create statistics for the currently recorded elements.
     * @return Statistics about the currently recorded elements
     */
    HistogramStatistics getStatistics();
}

/**
 * Histogram statistics represent the current snapshot of elements 
 * recorded in the histogram.
 */
public abstract class HistogramStatistics {
    /**
     * Returns the value for the given quantile based on the histogram statistics.
     * @param quantile Quantile to calculate the value for
     * @return Value for the given quantile
     */
    public abstract double getQuantile(double quantile);
    
    /**
     * Returns the elements of the statistics' sample.
     * @return Elements of the statistics' sample
     */
    public abstract long[] getValues();
    
    /**
     * Returns the size of the statistics' sample.
     * @return Size of the statistics' sample
     */
    public abstract int size();
    
    /**
     * Returns the mean of the histogram values.
     * @return Mean of the histogram values
     */
    public abstract double getMean();
    
    /**
     * Returns the standard deviation of the distribution.
     * @return Standard deviation of histogram distribution
     */
    public abstract double getStdDev();
    
    /**
     * Returns the maximum value of the histogram.
     * @return Maximum value of the histogram
     */
    public abstract long getMax();
    
    /**
     * Returns the minimum value of the histogram.
     * @return Minimum value of the histogram
     */
    public abstract long getMin();
}

// Create and use histogram
Histogram latencyHistogram = // ... custom histogram implementation
metricGroup.histogram("request-latency", latencyHistogram);

// Record values
latencyHistogram.update(45);      // 45ms latency
latencyHistogram.update(67);      // 67ms latency
latencyHistogram.update(23);      // 23ms latency

// Get statistics
HistogramStatistics stats = latencyHistogram.getStatistics();
double p95 = stats.getQuantile(0.95);        // 95th percentile
double mean = stats.getMean();               // average latency
double stdDev = stats.getStdDev();          // standard deviation
long max = stats.getMax();                  // maximum latency
long min = stats.getMin();                  // minimum latency
int sampleSize = stats.size();              // number of samples

/**
 * Common super interface for all metrics.
 */
public interface Metric {
    /**
     * Returns the metric type. Default implementation throws 
     * UnsupportedOperationException for custom metric types.
     * @return MetricType enum value
     */
    default MetricType getMetricType() {
        throw new UnsupportedOperationException("Custom metric types are not supported.");
    }
}

/**
 * Enum describing the different metric types.
 */
public enum MetricType {
    COUNTER,
    METER, 
    GAUGE,
    HISTOGRAM
}

// Check metric type
if (someMetric.getMetricType() == MetricType.COUNTER) {
    Counter counter = (Counter) someMetric;
    long count = counter.getCount();
}

// Type-specific handling
switch (metric.getMetricType()) {
    case COUNTER:
        handleCounter((Counter) metric);
        break;
    case GAUGE:
        handleGauge((Gauge<?>) metric);
        break;
    case METER:
        handleMeter((Meter) metric);
        break;
    case HISTOGRAM:
        handleHistogram((Histogram) metric);
        break;
}