Versions Compared

Key

  • This line was added.
  • This line was removed.
  • Formatting was changed.

...

I propose introducing a new interface: Monitorable. If a plugin implements this interface, the withPluginMetrics() method will be called when the plugin is instantiated (after configure() if the plugin also implements Configurable). This will allow plugins to adds add their own metrics to the component (broker, producer, consumer, etc) that instantiated them.

Code Block
languagejava
titleMonitorable.java
package org.apache.kafka.common.metrics;

public interface Monitorable {

    /**
     * Set the PluginMetrics instance from the clientcomponent that instantiates the plugin.
     */
 Plugins can register voidand withPluginMetrics(PluginMetricsunregister metrics);

}

The PluginMetrics interface has methods to add and remove metrics and sensors. Plugins will only be able to remove metrics and sensors they created. Metrics created via this class will have their group set to "plugins" and include tags that uniquely identify the plugin.

Code Block
languagejava
public interface PluginMetrics extends Closeable {

    /**
     * Create a {@link MetricName} with the given name, description and tags. The group will be set to "plugins" using the given PluginMetrics
     * at any point in their lifecycle prior to their close method being called.
     *
     * Plugin must call the close() method on this PluginMetrics instance in their close()
     * Tagsmethod to uniquelyclear identifyall themetrics pluginsthey areregistered.
 automatically added to the provided tags*/
    void withPluginMetrics(PluginMetrics metrics);

}

The PluginMetrics interface has methods to add and remove metrics and sensors. Plugins will only be able to remove metrics and sensors they created. Metrics created via this class will have their group set to "plugins" and include tags that uniquely identify the plugin.

Code Block
languagejava
/**
 * Implementations are thread *safe @paramso nameplugins may use the PluginMetrics instance from  The name of the metric
     * @param description A human-readable description to include in the metricmultiple threads
 */
public interface PluginMetrics extends Closeable {

    /**
     * @paramCreate tagsa {@link MetricName} with the given name,  additional key/value attributes of the metricdescription and tags. The group will be set to "plugins"
     */
    MetricName metricName(String name, String description, Map<String, String> tags);

 Tags to uniquely identify the plugins are automatically added to the provided tags
     /**
     * Add@param aname metric to monitor an object that implements {@link MetricValueProvider}. This metric won't be associated with anyThe name of the metric
     * sensor.@param Thisdescription isA ahuman-readable waydescription to exposeinclude existingin values as metrics.the metric
     *
 @param tags      * @param metricNameadditional Thekey/value nameattributes of the metric
     */
 @param metricValueProvider The metric value provider associated with this metricMetricName metricName(String name, String description, Map<String, String> tags);

     * @throws IllegalArgumentException if/**
     * Add a metric with same name already exists.
     */to monitor an object that implements {@link MetricValueProvider}. This metric won't be associated with any
    void addMetric(MetricName metricName, MetricValueProvider<?> metricValueProvider);

    /**
     * Remove a metric if it exists* sensor. This is a way to expose existing values as metrics.
     *
     * @param metricName The name of the metric
     * @throws@param IllegalArgumentExceptionmetricValueProvider ifThe themetric pluginvalue hasprovider not already created a metric associated with this namemetric
     */
    void removeMetric @throws IllegalArgumentException if a metric with same name already exists.
     */
    void addMetric(MetricName metricName, MetricValueProvider<?> metricValueProvider);

    /**
     * CreateRemove a sensormetric withif theit givenexists.
 unique name. The name must*
 only be unique for the* plugin,@param sometricName differentThe pluginsname can useof the same names.metric
     *
 @throws IllegalArgumentException if the plugin has * @param name The sensornot already created a metric with this name
     * @return The sensor/
    void removeMetric(MetricName metricName);

     * @throws IllegalArgumentException if/**
     * Create a sensor with samethe namegiven alreadyunique existsname. forThe thisname plugin
must only be unique for */
the plugin, so different Sensorplugins sensor(String name);

    /**can use the same names.
     *
 Remove a sensor (if it* exists)@param andname itsThe associatedsensor metrics.name
     *
 @return The sensor
  *  @param name* The@throws nameIllegalArgumentException ofif thea sensor towith same bename removed
already exists for this plugin
 * @throws IllegalArgumentException if the*/
 plugin has not alreadySensor created a sensor with this namesensor(String name);

    /**
     */
 Remove a  voidsensor removeSensor(Stringif name);
}

The PluginMetrics interface implements Closeable. Calling the close() method removed all metrics and sensors created by this plugin. It will be the responsibility of the plugin that create metrics to call close() of their PluginMetrics instance to remove their metrics.

New methods will be added to AbstractConfig so new plugin instances implementing Monitorable can get a PluginMetrics instance.

it exists) and its associated metrics.
     *
     * @param name The name of the sensor to be removed
     * @throws IllegalArgumentException if the plugin has not already created a sensor with this name
     */
    void removeSensor(String name);
}

The PluginMetrics interface implements Closeable. Calling the close() method removed all metrics and sensors created by this plugin. It will be the responsibility of the plugin that creates metrics to call close() of the PluginMetrics instance they were given to remove their metrics.

New methods will be added to AbstractConfig so new plugin instances implementing Monitorable can get a PluginMetrics instance.

Code Block
languagejava
titleAbstractConfig.java
/**
 * Get a configured instance of the give class specified by the given configuration key. If the object implements
 * Configurable configure it using the configuration. If the object implements Monitorable, set the appropriate PluginMetrics for that instance.
 *
 * @param key The configuration key for the class.
 * @param t The interface the class should implement.
 * @param metrics The metrics registry to use to build the PluginMetrics instance if the class implements Monitorable.
 * @return A configured instance of the class.
 */
public <T> T getConfiguredInstance(String key, Class<T> t, Metrics metrics);

/**
 * Get a configured instance of the give class specified by the given configuration key. If the object implements
 * Configurable configure it using the configuration. If the object implements Monitorable, set the appropriate PluginMetrics for that instance.
 *
 * @param key The configuration key for
Code Block
languagejava
titleAbstractConfig.java
/**
 * Get a configured instance of the give class specified by the given configuration key. If the object implements
 * Configurable configure it using the configuration. If the object implements Monitorable, set the appropriate PluginMetrics for that instance.
 *
 * @param key The configuration key for the class.
 * @param t The interface the class should implement.
 * @param metrics The metrics registry to use to build the PluginMetrics instance if the class implements Monitorable.
 * @return A configured instance of the class.
 */
public <T> T getConfiguredInstance(String key, Class<T> t, Metrics metrics);

/**
 * Get a configured instance of the give class specified by the given configuration key. If the object implements
 * Configurable configure it using the configuration. If the object implements Monitorable, set the appropriate PluginMetrics for that instance.
 *
 * @param key The configuration key for the class.
 * @param t The interface the class should implement.
 * @param configOverrides override origin configs.
 * @param metrics The metrics registry to use to build the PluginMetrics instance if the class implements Monitorable.
 * @return A configured instance of the class.
 */
public <T> T getConfiguredInstance(String key, Class<T> t, Map<String, Object> configOverrides, Metrics metrics);

/**
 * Get a list of configured instances of the given class specified by the given configuration key. The configuration
 * may specify either null or an empty string to indicate no configured instances. If an instance implements Monitorable, 
 * set the appropriate PluginMetrics for that instance. In both cases, this method returns an empty list@param t The interface the class should implement.
 * @param configOverrides override origin configs.
 * @param metrics The metrics registry to use to build the PluginMetrics instance if the class implements Monitorable.
 * @return A configured instance of the class.
 */
public <T> T getConfiguredInstance(String key, Class<T> t, Map<String, Object> configOverrides, Metrics metrics);

/**
 * Get a list of configured instances of the given class specified by the given configuration key. The configuration
 * may specify either null or an empty string to indicate no configured instances.
 If *an @paraminstance keyimplements TheMonitorable, configuration
 key* forset the class.
 appropriate PluginMetrics for that instance. In both cases, this method returns an empty list to indicate no configured instances.
 * @param key The configuration key for the class.
 * @param t The interface the class should implement.
 * @param configOverrides Configuration overrides to use.
 * @param metrics The metrics registry to use to build the PluginMetrics instances for each class that implements Monitorable.
 * @return The list of configured instances.
 */
public <T> List<T> getConfiguredInstances(String key, Class<T> t, Map<String, Object> configOverrides, Metrics metrics);

/**
 * Get a list of configured instances of the given class specified by the given configuration key. The configuration
 * may specify either null or an empty string to indicate no configured instances. If an instance implements Monitorable, 
 * set the appropriate PluginMetrics for that instance. In both cases, this method returns an empty list to indicate no configured instances.
 * @param key The configuration key for the classes.   
 * @param classNames The list of class names of the instances to create.
 * @param t The interface the class should implement.
 * @param configOverrides Configuration overrides to use.
 * @param metrics The metrics registry to use to build the PluginMetrics instances for each class that implements Monitorable.
 * @return The list of configured instances.
 */
public <T> List<T> getConfiguredInstances(String key, List<String> classNames, Class<T> t, Map<String, Object> configOverrides, Metrics metrics)

...

For Connectors and Converters, the name of the connector (connector) will be added as a tag (connector). Tasks will add the connector name and the of the connector (connector) and the task id (task) added as tags. Transformations and Predicates will have the connector name, the task id and their alias (aliastransformation/predicate) added as tags. For example for a task: kafka.connect:

  • for a task: kafka.connect:type=plugins,connector=my-sink,task=0
  • for a predicate: kafka.connect:type=plugins,connector=my-sink,task=0,predicate=my-predicate

For configurations that accept a list of classes, for example interceptor.classes, if the same class is provided multiple times, their metrics may collide. This is deemed highly unlikely to occur as there are no use cases where providing multiple times the same class is useful.

This proposal supersedes KIP-608 which only aimed at providing this functionality to Authorizer plugins. KIP-608 was adopted in 2020 but never implemented. This KIP proposes to alter the names of the metrics from KIP-608 to the following so all plugins use the same naming format:

Full NameTypeDescription
kafka.server:type=plugins,config=authorizer.class.name,class=MyAuthorizer,name=acls-total-countGaugeTotal ACLs created in the broker
kafka.server:type=plugins,config=authorizer.class.name,class=MyAuthorizer,name=authorization-request-rate-per-minuteRateTotal number of authorization requests per minute
kafka.server:type=plugins,config=authorizer.class.name,class=

...

MyAuthorizer,

...

name=authorization-allowed-rate-per-minuteRateTotal number of authorization allowed per minute
kafka.server:type=plugins,config=authorizer.class.name,class=MyAuthorizer,name=authorization-denied-rate-per-minuteRateTotal number of authorization denied per minute 

...

Supported plugins

The goal is allow this feature to be used by all plugins that are Closeable, AutoCloseable or have a close() methods apart from MetricsReporter since instances are created before the Metrics instance. Also all Connect connector plugins will be supported. The javadoc of all supported plugins will be updated to mention they are able to implement the Monitorable interface to define their own metrics.

...

ClassConfigReason
KafkaMetricsReporterkafka.metrics.reportersThis interface is technically not part of the public API. Since MetricsReporter will not support it, it makes sense to not add it to this one either.
ConsumerPartitionAssignorpartition.assignment.strategyThis interface does not have a close() method. This interface is being deprecated by KIP-848, hence I don't propose updating it.
DeserializationExceptionHandlerdefault.deserialization.exception.handlerThis interface does not have a close() method.
ProductionExceptionHandlerdefault.production.exception.handlerThis interface does not have a close() method.
TimestampExtractordefault.timestamp.extractorThis interface does not have a close() method.
RocksDBConfigSetterrocksdb.config.setterThis interface does not have a close() method.
SecurityProviderCreatorsecurity.providersThis interface does not have a close() method. The instance is passed to java.security.Security and we don't control its lifecycle.
ReplicationPolicyreplication.policy.classMirrorMaker currently uses its own mechanism to emit metrics. This interface does not have a close() method.
ConfigPropertyFilterconfig.property.filter.classMirrorMaker currently uses its own mechanism to emit metrics.
GroupFiltergroup.filter.classMirrorMaker currently uses its own mechanism to emit metrics.
TopicFiltertopic.filter.classMirrorMaker currently uses its own mechanism to emit metrics.
ForwardingAdminforwarding.admin.classMirrorMaker currently uses its own mechanism to emit metrics.

...

Code Block
languagejava
public class MyInterceptor<K, V> implements ProducerInterceptor<K, V>, Monitorable {

    private Sensor sensor;
    private PluginMetrics metrics;

    public void withPluginMetrics(PluginMetrics metrics) {
        this.metrics = metrics;
        sensor = metrics.sensor("onSend");
        MetricName rate = metrics.metricName("rate", "Average number of calls per second.", Collections.emptyMap());
        MetricName total = metrics.metricName("total", "Total number of calls.", Collections.emptyMap());
        sensor.add(rate, new Rate());
        sensor.add(total, new CumulativeCount());
    }

    @Override
    public ProducerRecord<K, V> onSend(ProducerRecord<K, V> record) {
        sensor.record();
        return record;
    }

    @Override
    public void close() {
        try {
            if (metrics != null) metrics.close();
        } catch (IOException e) { // Even though ProducerInterceptor extends AutoCloseable which has "void close() throws Exception;", ProducerInterceptor has its own close() method without "throws"!
            throw new RuntimeException(e);
        }
    }
}

...