diff --git a/src/main/scala/com/samstarling/prometheusfinagle/PrometheusStatsReceiver.scala b/src/main/scala/com/samstarling/prometheusfinagle/PrometheusStatsReceiver.scala index 5b74fb4..d0834ef 100644 --- a/src/main/scala/com/samstarling/prometheusfinagle/PrometheusStatsReceiver.scala +++ b/src/main/scala/com/samstarling/prometheusfinagle/PrometheusStatsReceiver.scala @@ -36,40 +36,49 @@ class PrometheusStatsReceiver(registry: CollectorRegistry, namespace: String, ti override def counter(verbosity: Verbosity, name: String*): Counter = { val (metricName, labels) = extractLabels(name) - val c = counters - .getOrElseUpdate(metricName, this.synchronized { newCounter(metricName, labels.keys.toSeq) }) - .labels(labels.values.toSeq: _*) + val counter = this.synchronized { + counters + .getOrElseUpdate(metricName, newCounter(metricName, labels.keys.toSeq)) + .labels(labels.values.toSeq: _*) + } new Counter { override def incr(delta: Long): Unit = { - c.inc(delta) + counter.inc(delta) } } } override def stat(verbosity: Verbosity, name: String*): Stat = { val (metricName, labels) = extractLabels(name) - val s = summaries + val summary = summaries .getOrElseUpdate(metricName, this.synchronized { newSummary(metricName, labels.keys.toSeq) }) .labels(labels.values.toSeq: _*) new Stat { override def add(value: Float): Unit = { - s.observe(value) + summary.observe(value) } } } - override def addGauge(verbosity: Verbosity, name: String*)( - f: => Float): Gauge = { + override def addGauge(verbosity: Verbosity, name: String*)(f: => Float): Gauge = { val (metricName, labels) = extractLabels(name) - gauges - .getOrElseUpdate(metricName, this.synchronized { newGauge(metricName, labels.keys.toSeq) }) val labelValues = labels.values.toSeq - gaugeChilds - .getOrElseUpdate((metricName, labelValues), this.synchronized { gauges(metricName).labels(labelValues: _*) }) - .set(f) // Set once initially + + this.synchronized { + gauges + .getOrElseUpdate(metricName, newGauge(metricName, labels.keys.toSeq)) + } + + this.synchronized { + gaugeChilds + .getOrElseUpdate((metricName, labelValues), gauges(metricName).labels(labelValues: _*)) + .set(f) // Set once initially + } + gaugeProviders.update((metricName, labelValues), () => f) + new Gauge { override def remove(): Unit = gaugeProviders.remove((metricName, labelValues)) } diff --git a/src/test/scala/com/samstarling/prometheusfinagle/PrometheusStatsReceiverRaceTest.scala b/src/test/scala/com/samstarling/prometheusfinagle/PrometheusStatsReceiverRaceTest.scala index 7f62fa5..35c49d3 100644 --- a/src/test/scala/com/samstarling/prometheusfinagle/PrometheusStatsReceiverRaceTest.scala +++ b/src/test/scala/com/samstarling/prometheusfinagle/PrometheusStatsReceiverRaceTest.scala @@ -10,11 +10,10 @@ class PrometheusStatsReceiverRaceTest extends UnitTest { "PrometheusStatsReceiver#counters" should { - "handle creating and incrementing concurrently nicely" in { val registry = new CollectorRegistry(true) val statsReceiver = new PrometheusStatsReceiver(registry).scope("test") - val cf: Seq[Future[Unit]] = (1 to threadCount) map { n => + val cf: Seq[Future[Unit]] = (1 to threadCount) map { _ => pool { statsReceiver.counter("my_counter").incr(1) } @@ -39,19 +38,15 @@ class PrometheusStatsReceiverRaceTest extends UnitTest { Await.result(joinedFutures, Duration(100, TimeUnit.MILLISECONDS)) registry.getSampleValue("finagle_my_counter", Array("serviceName"), Array("test")) === threadCount } - - } "PrometheusStatsReceiver#counters#Gauges" should { - - "reflect gauge value after creation" in { val registry = new CollectorRegistry(true) val mockTimer = new MockTimer val statsReceiver = new PrometheusStatsReceiver(registry, "finagle", mockTimer, Duration(10, TimeUnit.SECONDS)).scope("test") - Time.withCurrentTimeFrozen { timeCtl => + Time.withCurrentTimeFrozen { _ => var gaugeResult = 42 statsReceiver.addGauge("my_gauge") { gaugeResult @@ -98,6 +93,5 @@ class PrometheusStatsReceiverRaceTest extends UnitTest { registry.getSampleValue("finagle_my_gauge", Array("serviceName"), Array("test")) === 8 } } - } }