-
Notifications
You must be signed in to change notification settings - Fork 369
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Support JMX metrics for HTTP sink (#196)
* Support JMX metrics for HTTP sink Adds the counts for success, request average. * Brings HDR histogram as opposed to the Percentiles from Kafka. the Lib is easier to use Manually sets the attributes * Goes back on the JMX attributes name, since SetAttribute fails. * Fix the integration tests compilation error * Fix the integration tests to avoid using the same sink name, since it will fail on registering the JMX metrics --------- Co-authored-by: stheppi <[email protected]>
- Loading branch information
Showing
12 changed files
with
496 additions
and
47 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
66 changes: 66 additions & 0 deletions
66
...c/main/scala/io/lenses/streamreactor/connect/http/sink/metrics/HttpSinkMetricsMBean.scala
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,66 @@ | ||
| /* | ||
| * Copyright 2017-2025 Lenses.io Ltd | ||
| * | ||
| * Licensed under the Apache License, Version 2.0 (the "License"); | ||
| * you may not use this file except in compliance with the License. | ||
| * You may obtain a copy of the License at | ||
| * | ||
| * http://www.apache.org/licenses/LICENSE-2.0 | ||
| * | ||
| * Unless required by applicable law or agreed to in writing, software | ||
| * distributed under the License is distributed on an "AS IS" BASIS, | ||
| * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. | ||
| * See the License for the specific language governing permissions and | ||
| * limitations under the License. | ||
| */ | ||
| package io.lenses.streamreactor.connect.http.sink.metrics | ||
|
|
||
| import org.HdrHistogram.Histogram | ||
|
|
||
| import java.util.concurrent.atomic.LongAdder | ||
|
|
||
| trait HttpSinkMetricsMBean { | ||
| def get2xxCount: Long | ||
| def get4xxCount: Long | ||
| def get5xxCount: Long | ||
| def getOtherErrorsCount: Long | ||
| def recordRequestTime(time: Long): Unit | ||
| def increment2xxCount(): Unit | ||
| def increment4xxCount(): Unit | ||
| def increment5xxCount(): Unit | ||
| def incrementOtherErrorsCount(): Unit | ||
| def getP50RequestTimeMs: Long | ||
| def getP95RequestTimeMs: Long | ||
| def getP99RequestTimeMs: Long | ||
| } | ||
|
|
||
| class HttpSinkMetrics extends HttpSinkMetricsMBean { | ||
| private val successCount = new LongAdder() | ||
| private val error4xxCount = new LongAdder() | ||
| private val error5xxCount = new LongAdder() | ||
| private val otherErrorsCount = new LongAdder() | ||
|
|
||
| // Sets the maximum value to record in the histogram | ||
| private val MaxValueMillis = 3 * 60 * 60 * 1000L // 3 hours | ||
| private val histogram = new Histogram(MaxValueMillis, 3) | ||
|
|
||
| def increment2xxCount(): Unit = successCount.increment() | ||
| def increment4xxCount(): Unit = error4xxCount.increment() | ||
| def increment5xxCount(): Unit = error5xxCount.increment() | ||
|
|
||
| def recordRequestTime(millis: Long): Unit = | ||
| histogram.recordValue(math.min(millis, MaxValueMillis)) | ||
|
|
||
| override def get2xxCount: Long = successCount.sum() | ||
| override def get4xxCount: Long = error4xxCount.sum() | ||
| override def get5xxCount: Long = error5xxCount.sum() | ||
| override def getOtherErrorsCount: Long = otherErrorsCount.sum() | ||
| override def incrementOtherErrorsCount(): Unit = otherErrorsCount.increment() | ||
|
|
||
| override def getP50RequestTimeMs: Long = | ||
| histogram.getValueAtPercentile(50.0) | ||
| override def getP95RequestTimeMs: Long = | ||
| histogram.getValueAtPercentile(95.0) | ||
| override def getP99RequestTimeMs: Long = | ||
| histogram.getValueAtPercentile(99.0) | ||
| } |
36 changes: 36 additions & 0 deletions
36
...p/src/main/scala/io/lenses/streamreactor/connect/http/sink/metrics/MetricsRegistrar.scala
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,36 @@ | ||
| /* | ||
| * Copyright 2017-2025 Lenses.io Ltd | ||
| * | ||
| * Licensed under the Apache License, Version 2.0 (the "License"); | ||
| * you may not use this file except in compliance with the License. | ||
| * You may obtain a copy of the License at | ||
| * | ||
| * http://www.apache.org/licenses/LICENSE-2.0 | ||
| * | ||
| * Unless required by applicable law or agreed to in writing, software | ||
| * distributed under the License is distributed on an "AS IS" BASIS, | ||
| * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. | ||
| * See the License for the specific language governing permissions and | ||
| * limitations under the License. | ||
| */ | ||
| package io.lenses.streamreactor.connect.http.sink.metrics | ||
|
|
||
| import java.lang.management.ManagementFactory | ||
| import javax.management.ObjectName | ||
|
|
||
| object MetricsRegistrar { | ||
|
|
||
| val NameTemplate = "io.lenses.streamreactor.connect.http.sink:type=metrics,name=%s" | ||
|
|
||
| /** | ||
| * Register the metrics MBean exposing the count on 200, 400, 500 and other response codes as well as the http request time percentiles | ||
| * @param metrics | ||
| * @param sinkName | ||
| */ | ||
| def registerMetricsMBean(metrics: HttpSinkMetricsMBean, sinkName: String): Unit = { | ||
| val mbs = ManagementFactory.getPlatformMBeanServer | ||
| val objectName = new ObjectName(s"io.lenses.streamreactor.connect.http.sink:type=metrics,name=$sinkName") | ||
| mbs.registerMBean(metrics, objectName) | ||
| () | ||
| } | ||
| } |
Oops, something went wrong.