allegro-internal/flex-roadmap#819 returned other metrics

allegro · Oct 19, 2024 · c191571 · c191571
1 parent 84e0174
commit c191571
Show file tree

Hide file tree

Showing 3 changed files with 25 additions and 0 deletions.
diff --git a/...ain/kotlin/pl/allegro/tech/servicemesh/envoycontrol/synchronization/GlobalStateChanges.kt b/...ain/kotlin/pl/allegro/tech/servicemesh/envoycontrol/synchronization/GlobalStateChanges.kt
@@ -4,6 +4,8 @@ import io.micrometer.core.instrument.MeterRegistry
 import pl.allegro.tech.servicemesh.envoycontrol.services.ClusterStateChanges
 import pl.allegro.tech.servicemesh.envoycontrol.services.MultiClusterState
 import pl.allegro.tech.servicemesh.envoycontrol.services.MultiClusterState.Companion.toMultiClusterState
+import pl.allegro.tech.servicemesh.envoycontrol.utils.CHECKPOINT_TAG
+import pl.allegro.tech.servicemesh.envoycontrol.utils.SERVICES_STATE_METRIC
 import pl.allegro.tech.servicemesh.envoycontrol.utils.logSuppressedError
 import pl.allegro.tech.servicemesh.envoycontrol.utils.measureBuffer
 import pl.allegro.tech.servicemesh.envoycontrol.utils.onBackpressureLatestMeasured
@@ -44,6 +46,9 @@ class GlobalStateChanges(
             .logSuppressedError("combineLatest() suppressed exception")
             .measureBuffer("global-service-changes-combinator", meterRegistry)
             .checkpoint("global-service-changes-emitted")
+            .name(SERVICES_STATE_METRIC)
+            .tag(CHECKPOINT_TAG, "combined")
+            .metrics()
     }
 
     private fun combinedExperimentalFlow(
@@ -70,7 +75,14 @@ class GlobalStateChanges(
             .logSuppressedError("combineLatest() suppressed exception")
             .measureBuffer("global-service-changes-combine-latest", meterRegistry)
             .checkpoint("global-service-changes-emitted")
+            .name(SERVICES_STATE_METRIC)
+            .tag(CHECKPOINT_TAG, "emitted")
+            .metrics()
             .onBackpressureLatestMeasured("global-service-changes-backpressure", meterRegistry)
             .publishOn(scheduler, 1)
+            .checkpoint("global-service-changes-published")
+            .name(SERVICES_STATE_METRIC)
+            .tag(CHECKPOINT_TAG, "published")
+            .metrics()
     }
 }
diff --git a/...lin/pl/allegro/tech/servicemesh/envoycontrol/synchronization/RemoteClusterStateChanges.kt b/...lin/pl/allegro/tech/servicemesh/envoycontrol/synchronization/RemoteClusterStateChanges.kt
@@ -3,6 +3,8 @@ package pl.allegro.tech.servicemesh.envoycontrol.synchronization
 import pl.allegro.tech.servicemesh.envoycontrol.EnvoyControlProperties
 import pl.allegro.tech.servicemesh.envoycontrol.services.ClusterStateChanges
 import pl.allegro.tech.servicemesh.envoycontrol.services.MultiClusterState
+import pl.allegro.tech.servicemesh.envoycontrol.utils.CHECKPOINT_TAG
+import pl.allegro.tech.servicemesh.envoycontrol.utils.SERVICES_STATE_METRIC
 import reactor.core.publisher.Flux
 
 class RemoteClusterStateChanges(
@@ -14,4 +16,7 @@ class RemoteClusterStateChanges(
             .getChanges(properties.sync.pollingInterval)
             .startWith(MultiClusterState.empty())
             .distinctUntilChanged()
+            .name(SERVICES_STATE_METRIC)
+            .tag(CHECKPOINT_TAG, "cross-dc")
+            .metrics()
 }
diff --git a/...n/kotlin/pl/allegro/tech/servicemesh/envoycontrol/consul/services/ConsulServiceChanges.kt b/...n/kotlin/pl/allegro/tech/servicemesh/envoycontrol/consul/services/ConsulServiceChanges.kt
@@ -13,7 +13,9 @@ import pl.allegro.tech.servicemesh.envoycontrol.logger
 import pl.allegro.tech.servicemesh.envoycontrol.server.ReadinessStateHandler
 import pl.allegro.tech.servicemesh.envoycontrol.services.ServiceInstances
 import pl.allegro.tech.servicemesh.envoycontrol.services.ServicesState
+import pl.allegro.tech.servicemesh.envoycontrol.utils.CHECKPOINT_TAG
 import pl.allegro.tech.servicemesh.envoycontrol.utils.ENVOY_CONTROL_WARM_UP_METRIC
+import pl.allegro.tech.servicemesh.envoycontrol.utils.SERVICES_STATE_METRIC
 import pl.allegro.tech.servicemesh.envoycontrol.utils.measureDiscardedItems
 import reactor.core.publisher.Flux
 import reactor.core.publisher.FluxSink
@@ -54,7 +56,13 @@ class ConsulServiceChanges(
         )
             .measureDiscardedItems("consul-service-changes", metrics.meterRegistry)
             .checkpoint("consul-service-changes-emitted")
+            .name(SERVICES_STATE_METRIC)
+            .tag(CHECKPOINT_TAG, "consul-emitted")
+            .metrics()
             .checkpoint("consul-service-changes-emitted-distinct")
+            .name(SERVICES_STATE_METRIC)
+            .tag(CHECKPOINT_TAG, "consul-distinct")
+            .metrics()
             .doOnCancel {
                 logger.warn("Cancelling watching consul service changes")
                 watcher.close()