-
Notifications
You must be signed in to change notification settings - Fork 10
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
- Loading branch information
1 parent
b0b8cc8
commit d002ace
Showing
4 changed files
with
276 additions
and
55 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
150 changes: 150 additions & 0 deletions
150
prometheus/src/main/java/com/itv/bucky/prometheus/PrometheusMetricsCollector.java
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,150 @@ | ||
| package com.itv.bucky.prometheus; | ||
|
|
||
| import java.util.Collections; | ||
| import java.util.concurrent.atomic.AtomicLong; | ||
| import java.util.function.Function; | ||
|
|
||
| import com.rabbitmq.client.Channel; | ||
| import com.rabbitmq.client.Connection; | ||
| import com.rabbitmq.client.impl.AbstractMetricsCollector; | ||
| import io.prometheus.client.Counter; | ||
| import io.prometheus.client.Gauge; | ||
| import io.prometheus.client.CollectorRegistry; | ||
|
|
||
| public class PrometheusMetricsCollector extends AbstractMetricsCollector { | ||
|
|
||
| private final Gauge connections; | ||
|
|
||
| private final Gauge channels; | ||
|
|
||
| private final Counter publishedMessages; | ||
|
|
||
| private final Counter failedToPublishMessages; | ||
|
|
||
| private final Counter ackedPublishedMessages; | ||
|
|
||
| private final Counter nackedPublishedMessages; | ||
|
|
||
| private final Counter unroutedPublishedMessages; | ||
|
|
||
| private final Counter consumedMessages; | ||
|
|
||
| private final Counter acknowledgedMessages; | ||
|
|
||
| private final Counter rejectedMessages; | ||
|
|
||
| public PrometheusMetricsCollector(final CollectorRegistry registry) { | ||
| this(registry, "rabbitmq"); | ||
| } | ||
|
|
||
| public PrometheusMetricsCollector(final CollectorRegistry registry, final String prefix) { | ||
| this.connections = Gauge.build().name(prefix + "_connections").create(); | ||
| registry.register(this.connections); | ||
|
|
||
| this.channels = Gauge.build() | ||
| .name(prefix + "_channels") | ||
| .create(); | ||
| registry.register(this.channels); | ||
|
|
||
| this.publishedMessages = Counter.build() | ||
| .name(prefix + "_published_messages") | ||
| .create(); | ||
| registry.register(this.publishedMessages); | ||
|
|
||
| this.failedToPublishMessages = Counter.build() | ||
| .name(prefix + "_failed_to_publish_messages") | ||
| .create(); | ||
| registry.register(failedToPublishMessages); | ||
|
|
||
| this.ackedPublishedMessages = Counter.build() | ||
| .name(prefix + "_acked_published_messages") | ||
| .create(); | ||
| registry.register(ackedPublishedMessages); | ||
|
|
||
| this.nackedPublishedMessages = Counter.build() | ||
| .name(prefix + "_nacked_published_messages") | ||
| .create(); | ||
| registry.register(nackedPublishedMessages); | ||
|
|
||
| this.unroutedPublishedMessages = Counter.build() | ||
| .name(prefix + "_unrouted_published_messages") | ||
| .create(); | ||
| registry.register(unroutedPublishedMessages); | ||
|
|
||
| this.consumedMessages = Counter.build() | ||
| .name(prefix + "_consumed_messages") | ||
| .create(); | ||
| registry.register(consumedMessages); | ||
|
|
||
| this.acknowledgedMessages = Counter.build() | ||
| .name(prefix + "_acknowledged_messages") | ||
| .create(); | ||
| registry.register(acknowledgedMessages); | ||
|
|
||
| this.rejectedMessages = Counter.build() | ||
| .name(prefix + "_rejected_messages") | ||
| .create(); | ||
| registry.register(rejectedMessages); | ||
|
|
||
| } | ||
|
|
||
| @Override | ||
| protected void incrementConnectionCount(Connection connection) { | ||
| connections.inc(); | ||
| } | ||
|
|
||
| @Override | ||
| protected void decrementConnectionCount(Connection connection) { | ||
| connections.dec(); | ||
| } | ||
|
|
||
| @Override | ||
| protected void incrementChannelCount(Channel channel) { | ||
| channels.inc(); | ||
| } | ||
|
|
||
| @Override | ||
| protected void decrementChannelCount(Channel channel) { | ||
| channels.dec(); | ||
| } | ||
|
|
||
| @Override | ||
| protected void markPublishedMessage() { | ||
| publishedMessages.inc(); | ||
| } | ||
|
|
||
| @Override | ||
| protected void markMessagePublishFailed() { | ||
| failedToPublishMessages.inc(); | ||
| } | ||
|
|
||
| @Override | ||
| protected void markConsumedMessage() { | ||
| consumedMessages.inc(); | ||
| } | ||
|
|
||
| @Override | ||
| protected void markAcknowledgedMessage() { | ||
| acknowledgedMessages.inc(); | ||
| } | ||
|
|
||
| @Override | ||
| protected void markRejectedMessage() { | ||
| rejectedMessages.inc(); | ||
| } | ||
|
|
||
| @Override | ||
| protected void markMessagePublishAcknowledged() { | ||
| ackedPublishedMessages.inc(); | ||
| } | ||
|
|
||
| @Override | ||
| protected void markMessagePublishNotAcknowledged() { | ||
| nackedPublishedMessages.inc(); | ||
| } | ||
|
|
||
| @Override | ||
| protected void markPublishedMessageUnrouted() { | ||
| unroutedPublishedMessages.inc(); | ||
| } | ||
| } |
29 changes: 29 additions & 0 deletions
29
prometheus/src/main/scala/com/itv/bucky/prometheus/PrometheusMetrics.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,29 @@ | ||
| // package com.itv.bucky.epimetheus | ||
|
|
||
| // import scala.language.higherKinds | ||
|
|
||
| // import com.itv.bucky.AmqpClient | ||
| // import io.chrisdavenport.epimetheus.CollectorRegistry | ||
| // import io.chrisdavenport.epimetheus.Counter | ||
| // import io.chrisdavenport.epimetheus.Gauge | ||
|
|
||
| // case class PrometheusMetrics[F[_]]( | ||
| // connections: Gauge[F], | ||
| // channels: Gauge[F], | ||
| // publishedMessages: Counter[F], | ||
| // consumedMessages: Counter[F], | ||
| // acknowledgedMessages: Counter[F], | ||
| // rejectedMessages: Counter[F], | ||
| // failedToPublishMessages: Counter[F], | ||
| // ackedPublishedMessages: Counter[F], | ||
| // nackedPublishedMessages: Counter[F], | ||
| // unroutedPublishedMessages: Counter[F] | ||
| // ) | ||
|
|
||
| // object Prometheus { | ||
|
|
||
| // def apply[F[_]](cr: CollectorRegistry[F])(amqpClient: AmqpClient[F]): AmqpClient[F] = | ||
| // amqpClient | ||
| // } | ||
|
|
||
|
|