Table of Contents |
---|
Status
Current state: [Under Discussion]Accepted
Discussion thread: here
Vote thread: here
JIRA:
Jira | ||||||
---|---|---|---|---|---|---|
|
Please keep the discussion on the mailing list rather than commenting on the wiki (wiki discussions get unwieldy fast).
Motivation
Current Kafka authorizer plugin cannot access to Kafka broker metrics information as plugin. Kafka authorizer plugins Kafka authorizer plugins need access to runtime broker Metrics instance to add additional metrics. There is disconnection for between how to manage Metrics between broker and authorizer plugins in current Kafka implementation. The authorizer The authorizer plugins as plugins as plugin of broker could use same Metrics instance as in broker, so authorization plugin so authorizer plugins need not manage tasks like creating and configuring Metrics and JmxReporter. With the feature of this KIP, Authorizer authorizer plugin can use broker's Metrics to manage and create Kafka instance to add additional metrics very easy.
Public Interfaces
AuthorizerServerInfo
interface provides runtime broker configuration to authorization plugins including cluster resource, broker id, cluster id and endpoint information. A new method 'Metrics metrics()' for accessing to broker Metrics information in broker is added to this interfaceDefine a new interface Monitorable in package org.apache.kafka.common. The authorizer plugins can implement the interface and broker can pass the instance of Metrics to Authorizer. Other broker'd or client's plugins could potentially implement the interface in the future and get the broker's or client's Metrics instance to add additional metrics from sub-components. Support for other plugins will be addressed later in another KIP if necessary. This KIP will focus on exposing Kafka Metrics in Authorizer only.
Code Block | |||||
---|---|---|---|---|---|
| |||||
packagepackage org.apache.kafka.server.authorizer; import java.util.Collection; import org.apache.kafka.common.ClusterResource; import org.apache.kafka.common.Endpoint; import org.apache.kafka.common.annotation.InterfaceStability; import org.apache.kafka.common.metrics.Metrics; /** * RuntimeInterface broker configuration metadata providedfor plugins to authorizersget during start up.Metrics instance */ @InterfaceStability.Evolving public interface AuthorizerServerInfoMonitorable { /** * Returns cluster metadata for the broker running this authorizer including cluster id. */ ClusterResource clusterResource(); /** * Returns broker id. This may be a generated broker id if `broker.id` was not configured. */ int brokerId(); /**Metrcis * Returns endpoints for all listeners including the advertised host and port to which * the listener is bound. */ Collection<Endpoint> endpoints(); /** * Returns the inter-broker endpoint. This is one of the endpoints returned by {@link #endpoints()}. */ Endpoint interBrokerEndpoint(); // Confluent extensions to Authorizer API /** * Returns the Get the instance of {@link Metrics} in this broker. */ default void monitor(Metrics metrics() { return null; } } |
Proposed Changes
return;
};
} |
The org.apache.kafka.server.authorizer.Authorizer interface need be updated to extend Monitorable Interface.Both object and class of Broker in kafka.cluster.Broker.scala will be updated to accept a new parameter referring to Kafka Metrics.
Code Block | ||||
---|---|---|---|---|
| ||||
package org.apache.kafka.server. | cluster.Brokerobject Broker {
private[cluster] case class ServerInfo(clusterResource: ClusterResource,
brokerId: Int,
endpoints: util.List[Endpoint],
interBrokerEndpoint: Endpoint,
override val metrics: Metrics) extends AuthorizerServerInfo
}
case class Broker(id: Int, endPoints: Seq[EndPoint], rack: Option[String]) {
......
def toServerInfo(clusterId: String, config: KafkaConfig, metrics: Metrics): AuthorizerServerInfo = {
val clusterResource: ClusterResource = new ClusterResource(clusterId)
val interBrokerEndpoint: Endpoint = endPoint(config.interBrokerListenerName).toJava
val brokerEndpoints: util.List[Endpoint] = endPoints.toList.map(_.toJava).asJava
Broker.ServerInfo(clusterResource, id, brokerEndpoints, interBrokerEndpoint, metrics)
}
}
authorizer.Authorizer
...
import org.apache.kafka.common.Monitorable;
...
public interface Authorizer extends Configurable, Closeable, Monitorable {
...
} |
The following metrics will be added:
Full Name | Type | Description |
---|---|---|
kafka.server:type=kafka.security.authorizer.metrics,name=acls-total-count | 32-bit gauge | Total acls created in the broker |
kafka.server:type=kafka.security.authorizer.metrics,name=authorization-request-rate-per-minute | Rate per minute | Total number of authorization requests per minute |
kafka.server:type=kafka.security.authorizer.metrics,name=authorization-allowed-rate-per-minute | Rate per minute | Total number of authorization allowed per minute |
kafka.server:type=kafka.security.authorizer.metrics,name=authorization-denied-rate-per-minute | Rate per minute | Total number of authorization denied per minute |
Proposed Changes
kafka.server.KafkaServer will be updated to pass instance of Metrics to start method in Authorizer Object.
Code Block | ||||
---|---|---|---|---|
| ||||
/* Get the authorizer and initialize it if one is specified.*/ authorizer = config.authorizer authorizer.foreach(_.configure(config.originals)) val authorizerFutures: Map[Endpoint, CompletableFuture[Void]] = authorizer match { case Some(authZ) => authZ.start(brokerInfo.broker.toServerInfo(clusterId, config, metrics)).asScala.map { case (ep, cs) => ep -> cs.toCompletableFuture } case None => brokerInfo.broker.endPoints.map { ep => ep.toJava -> CompletableFuture.completedFuture[Void](null) }.toMap }.foreach(_.monitor(metrics)) ... |
kafka.security.authorizer.AclAuthorizer will be updated for adding following Kafka metrics
Metric "acl-total-count" for reporting total acls created.
Metric "authorization-request-rate-per-hour" for reporting total number of authorization requests per hour
Metric "authorization-allowed-rate-per-hour" for reporting total number of authorization allowed per hour
Metric "authorization-denied-rate-per-hour" for reporting total number of authorization denied per hour to collect and add Authorizer metrics.
Code Block | ||||
---|---|---|---|---|
| ||||
class AclAuthorizer extends Authorizer with Logging { ...... private var metricsGroupauthorizerMetrics: MetricsGroupAuthorizerMetrics = _ ...... override def startmonitor(serverInfometrics: AuthorizerServerInfoMetrics): util.Map[Endpoint, _ <: CompletionStage[Void]]Unit = { metricsGroupauthorizerMetrics = new MetricsGroupAuthorizerMetrics(serverInfo.metrics()) serverInfo.endpoints.asScala.map { endpoint => endpoint -> CompletableFuture.completedFuture[Void](null) }.toMap.asJava } ...... private def authorizeAction(requestContext: AuthorizableRequestContext, action: Action): AuthorizationResult = { ...... // Record metrics if (authorized) { metricsGroup.recordAuthorizationAllowed() } else { metricsGroup.recordAuthorizationDenied() } metricsGroup.recordAuthorizationRequest() logAuditMessage(requestContext, action, logAuditMessage(requestContext, action, authorized) authorizerMetrics.recordAuthorizerMetrics(authorized) if (authorized) AuthorizationResult.ALLOWED else AuthorizationResult.DENIED } ... class MetricsGroupAuthorizerMetrics(metrics: Metrics) { val GROUP_NAME = "acl-authorizer-kafka.security.authorizer.metrics" val authorizationAllowedSensor = metrics.sensor("aclauthorizer-authorization-allowed") authorizationAllowedSensor.add(metrics.metricName("authorization-allowed-rate-per-hourminute", GROUP_NAME, "The number of authoization allowed per hour"), new Rate(TimeUnit.HOURSMINUTES, new WindowedCount())) val authorizationDeniedSensor = metrics.sensor("aclauthorizer-authorization-denied") authorizationDeniedSensor.add(metrics.metricName("authorization-denied-rate-per-hourminute", GROUP_NAME, "The number of authoization denied per hour"), new Rate(TimeUnit.HOURSMINUTES, new WindowedCount())) val authorizationRequestSensor = metrics.sensor("aclauthorizer-authorization-request") authorizationRequestSensor.add(metrics.metricName("authorization-request-rate-per-hourminute", GROUP_NAME, "The number of authoization request per hour"), new Rate(TimeUnit.HOURSMINUTES, new WindowedCount())) metrics.addMetric(metrics.metricName("aclacls-total-count", GROUP_NAME, "The number of acls defined"), (config, now) => aclCache.size) def recordAuthorizationAllowedrecordAuthorizerMetrics(authorized: Boolean): Unit = { authorizationAllowedSensor.recordif (authorized) { } authorizationAllowedSensor.record() def recordAuthorizationDenied(): Unit = } else { authorizationDeniedSensor.record() } def recordAuthorizationRequest(): Unit = { } authorizationRequestSensor.record() } } } |
Compatibility, Deprecation, and Migration Plan
Broker will start Authorizer plugins with AuthorizerServerInfo containing the new method 'Metrics metrics()' call authorizer's monitor(metrics: Metrics) which has default implementation after authorizer configured. Old version of Authorizer plugins will not call the new authorizer plugins doesn't call the monitor method, so it is backward to old version of Authorizer authorizer plugins.
Test Plan
Test total number of acls created match with the number from metrics
Test authorization request rate per minute
Test authorization allowed rate per minute
Test authorization denied rate per minute
Rejected Alternatives
Authorizer plugins create and configure own Metrics and JmxReport. Using this alternative approach make collecting Kafka metrics in Authorizer plugin separate from broker metrics that cause disconnection. Users need implement and manage own Metrics and JmsReport with extra effort.
Pass the instance of Metrics through AuthorizerServerInfo interface. The instance of Metrics should not be part of Authorizer server information. We use an interface to expose Kafka Metrics so that other broker's or client's plugins could potentially implement the interface.