Skip to content

KAFKA-10748: Add IP connection rate throttling metric #9685

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
dajac merged 2 commits into from
Dec 10, 2020
Merged

KAFKA-10748: Add IP connection rate throttling metric #9685

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
e40b1d3
Adds IP throttle metric
splett2 Dec 4, 2020
ad144db
update comment
splett2 Dec 9, 2020
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
36 changes: 22 additions & 14 deletions core/src/main/scala/kafka/network/SocketServer.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1327,6 +1327,8 @@ object ConnectionQuotas {
private val ConnectionRateSensorName = "Connection-Accept-Rate"
private val ConnectionRateMetricName = "connection-accept-rate"
private val IpMetricTag = "ip"
private val ListenerThrottlePrefix = ""
private val IpThrottlePrefix = "ip-"

private case class ListenerQuotaEntity(listenerName: String) extends ConnectionQuotaEntity {
override def sensorName: String = s"$ConnectionRateSensorName-$listenerName"
Expand Down Expand Up @@ -1574,7 +1576,7 @@ class ConnectionQuotas(config: KafkaConfig, time: Time, metrics: Metrics) extend
val throttleTimeMs = math.max(minThrottleTimeMs, listenerThrottleTimeMs)
// record throttle time due to hitting connection rate quota
if (throttleTimeMs > 0) {
listenerQuota.connectionRateThrottleSensor.record(throttleTimeMs.toDouble, timeMs)
listenerQuota.listenerConnectionRateThrottleSensor.record(throttleTimeMs.toDouble, timeMs)
}
throttleTimeMs
}
Expand All @@ -1590,19 +1592,23 @@ class ConnectionQuotas(config: KafkaConfig, time: Time, metrics: Metrics) extend
}

/**
* To avoid over-recording listener/broker connection rate, we un-record a listener and broker connection
* if the IP gets throttled.
* Record IP throttle time on the corresponding listener. To avoid over-recording listener/broker connection rate, we
* also un-record the listener and broker connection if the IP gets throttled.
*
* @param listenerName listener to un-record connection
* @param throttleMs IP throttle time to record for listener
* @param timeMs current time in milliseconds
*/
private def unrecordListenerConnection(listenerName: ListenerName, timeMs: Long): Unit = {
private def updateListenerMetrics(listenerName: ListenerName, throttleMs: Long, timeMs: Long): Unit = {
if (!protectedListener(listenerName)) {
brokerConnectionRateSensor.record(-1.0, timeMs, false)
}
maxConnectionsPerListener
.get(listenerName)
.foreach(_.connectionRateSensor.record(-1.0, timeMs, false))
.foreach { listenerQuota =>
listenerQuota.ipConnectionRateThrottleSensor.record(throttleMs.toDouble, timeMs)
listenerQuota.connectionRateSensor.record(-1.0, timeMs, false)
}
}

/**
Expand All @@ -1626,7 +1632,7 @@ class ConnectionQuotas(config: KafkaConfig, time: Time, metrics: Metrics) extend
trace(s"Throttling $address for $throttleMs ms")
// unrecord the connection since we won't accept the connection
sensor.record(-1.0, timeMs, false)
unrecordListenerConnection(listenerName, timeMs)
updateListenerMetrics(listenerName, throttleMs, timeMs)
throw new ConnectionThrottledException(address, timeMs, throttleMs)
}
}
Expand Down Expand Up @@ -1702,7 +1708,8 @@ class ConnectionQuotas(config: KafkaConfig, time: Time, metrics: Metrics) extend
class ListenerConnectionQuota(lock: Object, listener: ListenerName) extends ListenerReconfigurable with AutoCloseable {
@volatile private var _maxConnections = Int.MaxValue
private[network] val connectionRateSensor = getOrCreateConnectionRateQuotaSensor(Int.MaxValue, ListenerQuotaEntity(listener.value))
private[network] val connectionRateThrottleSensor = createConnectionRateThrottleSensor()
private[network] val listenerConnectionRateThrottleSensor = createConnectionRateThrottleSensor(ListenerThrottlePrefix)
private[network] val ipConnectionRateThrottleSensor = createConnectionRateThrottleSensor(IpThrottlePrefix)

def maxConnections: Int = _maxConnections

Expand Down Expand Up @@ -1737,7 +1744,8 @@ class ConnectionQuotas(config: KafkaConfig, time: Time, metrics: Metrics) extend

def close(): Unit = {
metrics.removeSensor(connectionRateSensor.name)
metrics.removeSensor(connectionRateThrottleSensor.name)
metrics.removeSensor(listenerConnectionRateThrottleSensor.name)
metrics.removeSensor(ipConnectionRateThrottleSensor.name)
}

private def maxConnections(configs: util.Map[String, _]): Int = {
Expand All @@ -1749,13 +1757,13 @@ class ConnectionQuotas(config: KafkaConfig, time: Time, metrics: Metrics) extend
}

/**
* Creates sensor for tracking the average throttle time on this listener due to hitting connection rate quota.
* The average is out of all throttle times > 0, which is consistent with the bandwidth and request quota throttle
* time metrics.
* Creates sensor for tracking the average throttle time on this listener due to hitting broker/listener connection
* rate or IP connection rate quota. The average is out of all throttle times > 0, which is consistent with the
* bandwidth and request quota throttle time metrics.
*/
private def createConnectionRateThrottleSensor(): Sensor = {
val sensor = metrics.sensor(s"ConnectionRateThrottleTime-${listener.value}")
val metricName = metrics.metricName("connection-accept-throttle-time",
private def createConnectionRateThrottleSensor(throttlePrefix: String): Sensor = {
val sensor = metrics.sensor(s"${throttlePrefix}ConnectionRateThrottleTime-${listener.value}")
val metricName = metrics.metricName(s"${throttlePrefix}connection-accept-throttle-time",
MetricsGroup,
"Tracking average throttle-time, out of non-zero throttle times, per listener",
Map(ListenerMetricTag -> listener.value).asJava)
Expand Down
25 changes: 24 additions & 1 deletion core/src/test/scala/unit/kafka/network/ConnectionQuotasTest.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -451,6 +451,8 @@ class ConnectionQuotasTest {

// acceptor shouldn't block for IP rate throttling
verifyNoBlockedPercentRecordedOnAllListeners()
// no IP throttle time should be recorded on any listeners
listeners.values.map(_.listenerName).foreach(verifyIpThrottleTimeOnListener(_, expectThrottle = false))
}

@Test
Expand All @@ -469,15 +471,21 @@ class ConnectionQuotasTest {
val numConnections = 80
acceptConnectionsAndVerifyRate(connectionQuotas, externalListener, numConnections, connCreateIntervalMs, ipConnectionRateLimit,
1, expectIpThrottle = true)
verifyIpThrottleTimeOnListener(externalListener.listenerName, expectThrottle = true)

// verify that default quota applies to IPs without a quota override
connectionQuotas.updateIpConnectionRateQuota(None, Some(ipConnectionRateLimit))
val adminListener = listeners("ADMIN")
// listener shouldn't have any IP throttle time recorded
verifyIpThrottleTimeOnListener(adminListener.listenerName, expectThrottle = false)
acceptConnectionsAndVerifyRate(connectionQuotas, adminListener, numConnections, connCreateIntervalMs, ipConnectionRateLimit,
1, expectIpThrottle = true)
verifyIpThrottleTimeOnListener(adminListener.listenerName, expectThrottle = true)

// acceptor shouldn't block for IP rate throttling
verifyNoBlockedPercentRecordedOnAllListeners()
// replication listener shouldn't have any IP throttling recorded
verifyIpThrottleTimeOnListener(listeners("REPLICATION").listenerName, expectThrottle = false)
}

@Test
Expand Down Expand Up @@ -757,8 +765,10 @@ class ConnectionQuotasTest {
0, metricValue(listenerConnRateMetric(listenerName.value)), eps)
assertNotNull(s"Expected connection-accept-throttle-time metric to exist for listener ${listenerName.value}",
listenerConnThrottleMetric(listenerName.value))
assertEquals(s"Connection throttle metric for listener ${listenerName.value}",
assertEquals(s"Listener connection throttle metric for listener ${listenerName.value}",
0, metricValue(listenerConnThrottleMetric(listenerName.value)).toLong)
assertEquals(s"Ip connection throttle metric for listener ${listenerName.value}",
0, metricValue(ipConnThrottleMetric(listenerName.value)).toLong)
}
verifyNoBlockedPercentRecordedOnAllListeners()
assertEquals("Broker-wide connection acceptance rate metric", 0, metricValue(brokerConnRateMetric()), eps)
Expand All @@ -780,6 +790,11 @@ class ConnectionQuotasTest {
}
}

private def verifyIpThrottleTimeOnListener(listener: ListenerName, expectThrottle: Boolean): Unit = {
assertEquals(s"IP connection throttle recorded for listener ${listener.value}", expectThrottle,
metricValue(ipConnThrottleMetric(listener.value)).toLong > 0)
}

private def verifyOnlyNonInterBrokerListenersBlockedPercentRecorded(): Unit = {
blockedPercentMeters.foreach { case (name, meter) =>
name match {
Expand All @@ -806,6 +821,14 @@ class ConnectionQuotasTest {
metrics.metric(metricName)
}

private def ipConnThrottleMetric(listener: String): KafkaMetric = {
val metricName = metrics.metricName(
"ip-connection-accept-throttle-time",
SocketServer.MetricsGroup,
Collections.singletonMap(Processor.ListenerMetricTag, listener))
metrics.metric(metricName)
}

private def listenerConnRateMetric(listener: String) : KafkaMetric = {
val metricName = metrics.metricName(
"connection-accept-rate",
Expand Down