LCOV - code coverage report
Current view: top level - proxy/src - metrics.rs (source / functions) Coverage Total Hit
Test: 07bee600374ccd486c69370d0972d9035964fe68.info Lines: 70.7 % 191 135
Test Date: 2025-02-20 13:11:02 Functions: 76.8 % 56 43

            Line data    Source code
       1              : use std::sync::{Arc, OnceLock};
       2              : 
       3              : use lasso::ThreadedRodeo;
       4              : use measured::label::{
       5              :     FixedCardinalitySet, LabelGroupSet, LabelName, LabelSet, LabelValue, StaticLabelSet,
       6              : };
       7              : use measured::metric::histogram::Thresholds;
       8              : use measured::metric::name::MetricName;
       9              : use measured::{
      10              :     Counter, CounterVec, FixedCardinalityLabel, Gauge, Histogram, HistogramVec, LabelGroup,
      11              :     MetricGroup,
      12              : };
      13              : use metrics::{CounterPairAssoc, CounterPairVec, HyperLogLog, HyperLogLogVec};
      14              : use tokio::time::{self, Instant};
      15              : 
      16              : use crate::control_plane::messages::ColdStartInfo;
      17              : use crate::error::ErrorKind;
      18              : 
      19           44 : #[derive(MetricGroup)]
      20              : #[metric(new(thread_pool: Arc<ThreadPoolMetrics>))]
      21              : pub struct Metrics {
      22              :     #[metric(namespace = "proxy")]
      23              :     #[metric(init = ProxyMetrics::new(thread_pool))]
      24              :     pub proxy: ProxyMetrics,
      25              : 
      26              :     #[metric(namespace = "wake_compute_lock")]
      27              :     pub wake_compute_lock: ApiLockMetrics,
      28              : }
      29              : 
      30              : static SELF: OnceLock<Metrics> = OnceLock::new();
      31              : impl Metrics {
      32            0 :     pub fn install(thread_pool: Arc<ThreadPoolMetrics>) {
      33            0 :         SELF.set(Metrics::new(thread_pool))
      34            0 :             .ok()
      35            0 :             .expect("proxy metrics must not be installed more than once");
      36            0 :     }
      37              : 
      38          143 :     pub fn get() -> &'static Self {
      39          143 :         #[cfg(test)]
      40          143 :         return SELF.get_or_init(|| Metrics::new(Arc::new(ThreadPoolMetrics::new(0))));
      41          143 : 
      42          143 :         #[cfg(not(test))]
      43          143 :         SELF.get()
      44          143 :             .expect("proxy metrics must be installed by the main() function")
      45          143 :     }
      46              : }
      47              : 
      48           44 : #[derive(MetricGroup)]
      49              : #[metric(new(thread_pool: Arc<ThreadPoolMetrics>))]
      50              : pub struct ProxyMetrics {
      51              :     #[metric(flatten)]
      52              :     pub db_connections: CounterPairVec<NumDbConnectionsGauge>,
      53              :     #[metric(flatten)]
      54              :     pub client_connections: CounterPairVec<NumClientConnectionsGauge>,
      55              :     #[metric(flatten)]
      56              :     pub connection_requests: CounterPairVec<NumConnectionRequestsGauge>,
      57              :     #[metric(flatten)]
      58              :     pub http_endpoint_pools: HttpEndpointPools,
      59              :     #[metric(flatten)]
      60              :     pub cancel_channel_size: CounterPairVec<CancelChannelSizeGauge>,
      61              : 
      62              :     /// Time it took for proxy to establish a connection to the compute endpoint.
      63              :     // largest bucket = 2^16 * 0.5ms = 32s
      64              :     #[metric(metadata = Thresholds::exponential_buckets(0.0005, 2.0))]
      65              :     pub compute_connection_latency_seconds: HistogramVec<ComputeConnectionLatencySet, 16>,
      66              : 
      67              :     /// Time it took for proxy to receive a response from control plane.
      68              :     #[metric(
      69              :         // largest bucket = 2^16 * 0.2ms = 13s
      70              :         metadata = Thresholds::exponential_buckets(0.0002, 2.0),
      71              :     )]
      72              :     pub console_request_latency: HistogramVec<ConsoleRequestSet, 16>,
      73              : 
      74              :     /// Time it takes to acquire a token to call console plane.
      75              :     // largest bucket = 3^16 * 0.05ms = 2.15s
      76              :     #[metric(metadata = Thresholds::exponential_buckets(0.00005, 3.0))]
      77              :     pub control_plane_token_acquire_seconds: Histogram<16>,
      78              : 
      79              :     /// Size of the HTTP request body lengths.
      80              :     // smallest bucket = 16 bytes
      81              :     // largest bucket = 4^12 * 16 bytes = 256MB
      82              :     #[metric(metadata = Thresholds::exponential_buckets(16.0, 4.0))]
      83              :     pub http_conn_content_length_bytes: HistogramVec<StaticLabelSet<HttpDirection>, 12>,
      84              : 
      85              :     /// Time it takes to reclaim unused connection pools.
      86              :     #[metric(metadata = Thresholds::exponential_buckets(1e-6, 2.0))]
      87              :     pub http_pool_reclaimation_lag_seconds: Histogram<16>,
      88              : 
      89              :     /// Number of opened connections to a database.
      90              :     pub http_pool_opened_connections: Gauge,
      91              : 
      92              :     /// Number of cache hits/misses for allowed ips.
      93              :     pub allowed_ips_cache_misses: CounterVec<StaticLabelSet<CacheOutcome>>,
      94              : 
      95              :     /// Number of allowed ips
      96              :     #[metric(metadata = Thresholds::with_buckets([0.0, 1.0, 2.0, 3.0, 4.0, 5.0, 10.0, 20.0, 50.0, 100.0]))]
      97              :     pub allowed_ips_number: Histogram<10>,
      98              : 
      99              :     /// Number of cache hits/misses for VPC endpoint IDs.
     100              :     pub vpc_endpoint_id_cache_stats: CounterVec<StaticLabelSet<CacheOutcome>>,
     101              : 
     102              :     /// Number of cache hits/misses for access blocker flags.
     103              :     pub access_blocker_flags_cache_stats: CounterVec<StaticLabelSet<CacheOutcome>>,
     104              : 
     105              :     /// Number of allowed VPC endpoints IDs
     106              :     #[metric(metadata = Thresholds::with_buckets([0.0, 1.0, 2.0, 3.0, 4.0, 5.0, 10.0, 20.0, 50.0, 100.0]))]
     107              :     pub allowed_vpc_endpoint_ids: Histogram<10>,
     108              : 
     109              :     /// Number of connections (per sni).
     110              :     pub accepted_connections_by_sni: CounterVec<StaticLabelSet<SniKind>>,
     111              : 
     112              :     /// Number of connection failures (per kind).
     113              :     pub connection_failures_total: CounterVec<StaticLabelSet<ConnectionFailureKind>>,
     114              : 
     115              :     /// Number of wake-up failures (per kind).
     116              :     pub connection_failures_breakdown: CounterVec<ConnectionFailuresBreakdownSet>,
     117              : 
     118              :     /// Number of bytes sent/received between all clients and backends.
     119              :     pub io_bytes: CounterVec<StaticLabelSet<Direction>>,
     120              : 
     121              :     /// Number of errors by a given classification.
     122              :     pub errors_total: CounterVec<StaticLabelSet<crate::error::ErrorKind>>,
     123              : 
     124              :     /// Number of cancellation requests (per found/not_found).
     125              :     pub cancellation_requests_total: CounterVec<CancellationRequestSet>,
     126              : 
     127              :     /// Number of errors by a given classification
     128              :     pub redis_errors_total: CounterVec<RedisErrorsSet>,
     129              : 
     130              :     /// Number of TLS handshake failures
     131              :     pub tls_handshake_failures: Counter,
     132              : 
     133              :     /// Number of connection requests affected by authentication rate limits
     134              :     pub requests_auth_rate_limits_total: Counter,
     135              : 
     136              :     /// HLL approximate cardinality of endpoints that are connecting
     137              :     pub connecting_endpoints: HyperLogLogVec<StaticLabelSet<Protocol>, 32>,
     138              : 
     139              :     /// Number of endpoints affected by errors of a given classification
     140              :     pub endpoints_affected_by_errors: HyperLogLogVec<StaticLabelSet<crate::error::ErrorKind>, 32>,
     141              : 
     142              :     /// Number of endpoints affected by authentication rate limits
     143              :     pub endpoints_auth_rate_limits: HyperLogLog<32>,
     144              : 
     145              :     /// Number of invalid endpoints (per protocol, per rejected).
     146              :     pub invalid_endpoints_total: CounterVec<InvalidEndpointsSet>,
     147              : 
     148              :     /// Number of retries (per outcome, per retry_type).
     149              :     #[metric(metadata = Thresholds::with_buckets([0.0, 1.0, 2.0, 3.0, 4.0, 5.0, 6.0, 7.0, 8.0]))]
     150              :     pub retries_metric: HistogramVec<RetriesMetricSet, 9>,
     151              : 
     152              :     /// Number of events consumed from redis (per event type).
     153              :     pub redis_events_count: CounterVec<StaticLabelSet<RedisEventsCount>>,
     154              : 
     155              :     #[metric(namespace = "connect_compute_lock")]
     156              :     pub connect_compute_lock: ApiLockMetrics,
     157              : 
     158              :     #[metric(namespace = "scram_pool")]
     159              :     #[metric(init = thread_pool)]
     160              :     pub scram_pool: Arc<ThreadPoolMetrics>,
     161              : }
     162              : 
     163           88 : #[derive(MetricGroup)]
     164              : #[metric(new())]
     165              : pub struct ApiLockMetrics {
     166              :     /// Number of semaphores registered in this api lock
     167              :     pub semaphores_registered: Counter,
     168              :     /// Number of semaphores unregistered in this api lock
     169              :     pub semaphores_unregistered: Counter,
     170              :     /// Time it takes to reclaim unused semaphores in the api lock
     171              :     #[metric(metadata = Thresholds::exponential_buckets(1e-6, 2.0))]
     172              :     pub reclamation_lag_seconds: Histogram<16>,
     173              :     /// Time it takes to acquire a semaphore lock
     174              :     #[metric(metadata = Thresholds::exponential_buckets(1e-4, 2.0))]
     175              :     pub semaphore_acquire_seconds: Histogram<16>,
     176              : }
     177              : 
     178              : impl Default for ApiLockMetrics {
     179           88 :     fn default() -> Self {
     180           88 :         Self::new()
     181           88 :     }
     182              : }
     183              : 
     184              : #[derive(FixedCardinalityLabel, Copy, Clone)]
     185              : #[label(singleton = "direction")]
     186              : pub enum HttpDirection {
     187              :     Request,
     188              :     Response,
     189              : }
     190              : 
     191              : #[derive(FixedCardinalityLabel, Copy, Clone)]
     192              : #[label(singleton = "direction")]
     193              : pub enum Direction {
     194              :     Tx,
     195              :     Rx,
     196              : }
     197              : 
     198              : #[derive(FixedCardinalityLabel, Clone, Copy, Debug)]
     199              : #[label(singleton = "protocol")]
     200              : pub enum Protocol {
     201              :     Http,
     202              :     Ws,
     203              :     Tcp,
     204              :     SniRouter,
     205              : }
     206              : 
     207              : impl Protocol {
     208            0 :     pub fn as_str(self) -> &'static str {
     209            0 :         match self {
     210            0 :             Protocol::Http => "http",
     211            0 :             Protocol::Ws => "ws",
     212            0 :             Protocol::Tcp => "tcp",
     213            0 :             Protocol::SniRouter => "sni_router",
     214              :         }
     215            0 :     }
     216              : }
     217              : 
     218              : impl std::fmt::Display for Protocol {
     219            0 :     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
     220            0 :         f.write_str(self.as_str())
     221            0 :     }
     222              : }
     223              : 
     224              : #[derive(FixedCardinalityLabel, Copy, Clone)]
     225              : pub enum Bool {
     226              :     True,
     227              :     False,
     228              : }
     229              : 
     230              : #[derive(FixedCardinalityLabel, Copy, Clone)]
     231              : #[label(singleton = "outcome")]
     232              : pub enum Outcome {
     233              :     Success,
     234              :     Failed,
     235              : }
     236              : 
     237              : #[derive(FixedCardinalityLabel, Copy, Clone)]
     238              : #[label(singleton = "outcome")]
     239              : pub enum CacheOutcome {
     240              :     Hit,
     241              :     Miss,
     242              : }
     243              : 
     244           88 : #[derive(LabelGroup)]
     245              : #[label(set = ConsoleRequestSet)]
     246              : pub struct ConsoleRequest<'a> {
     247              :     #[label(dynamic_with = ThreadedRodeo, default)]
     248              :     pub request: &'a str,
     249              : }
     250              : 
     251              : #[derive(MetricGroup, Default)]
     252              : pub struct HttpEndpointPools {
     253              :     /// Number of endpoints we have registered pools for
     254              :     pub http_pool_endpoints_registered_total: Counter,
     255              :     /// Number of endpoints we have unregistered pools for
     256              :     pub http_pool_endpoints_unregistered_total: Counter,
     257              : }
     258              : 
     259              : pub struct HttpEndpointPoolsGuard<'a> {
     260              :     dec: &'a Counter,
     261              : }
     262              : 
     263              : impl Drop for HttpEndpointPoolsGuard<'_> {
     264            2 :     fn drop(&mut self) {
     265            2 :         self.dec.inc();
     266            2 :     }
     267              : }
     268              : 
     269              : impl HttpEndpointPools {
     270            2 :     pub fn guard(&self) -> HttpEndpointPoolsGuard<'_> {
     271            2 :         self.http_pool_endpoints_registered_total.inc();
     272            2 :         HttpEndpointPoolsGuard {
     273            2 :             dec: &self.http_pool_endpoints_unregistered_total,
     274            2 :         }
     275            2 :     }
     276              : }
     277              : pub struct NumDbConnectionsGauge;
     278              : impl CounterPairAssoc for NumDbConnectionsGauge {
     279              :     const INC_NAME: &'static MetricName = MetricName::from_str("opened_db_connections_total");
     280              :     const DEC_NAME: &'static MetricName = MetricName::from_str("closed_db_connections_total");
     281              :     const INC_HELP: &'static str = "Number of opened connections to a database.";
     282              :     const DEC_HELP: &'static str = "Number of closed connections to a database.";
     283              :     type LabelGroupSet = StaticLabelSet<Protocol>;
     284              : }
     285              : pub type NumDbConnectionsGuard<'a> = metrics::MeasuredCounterPairGuard<'a, NumDbConnectionsGauge>;
     286              : 
     287              : pub struct NumClientConnectionsGauge;
     288              : impl CounterPairAssoc for NumClientConnectionsGauge {
     289              :     const INC_NAME: &'static MetricName = MetricName::from_str("opened_client_connections_total");
     290              :     const DEC_NAME: &'static MetricName = MetricName::from_str("closed_client_connections_total");
     291              :     const INC_HELP: &'static str = "Number of opened connections from a client.";
     292              :     const DEC_HELP: &'static str = "Number of closed connections from a client.";
     293              :     type LabelGroupSet = StaticLabelSet<Protocol>;
     294              : }
     295              : pub type NumClientConnectionsGuard<'a> =
     296              :     metrics::MeasuredCounterPairGuard<'a, NumClientConnectionsGauge>;
     297              : 
     298              : pub struct NumConnectionRequestsGauge;
     299              : impl CounterPairAssoc for NumConnectionRequestsGauge {
     300              :     const INC_NAME: &'static MetricName = MetricName::from_str("accepted_connections_total");
     301              :     const DEC_NAME: &'static MetricName = MetricName::from_str("closed_connections_total");
     302              :     const INC_HELP: &'static str = "Number of client connections accepted.";
     303              :     const DEC_HELP: &'static str = "Number of client connections closed.";
     304              :     type LabelGroupSet = StaticLabelSet<Protocol>;
     305              : }
     306              : pub type NumConnectionRequestsGuard<'a> =
     307              :     metrics::MeasuredCounterPairGuard<'a, NumConnectionRequestsGauge>;
     308              : 
     309              : pub struct CancelChannelSizeGauge;
     310              : impl CounterPairAssoc for CancelChannelSizeGauge {
     311              :     const INC_NAME: &'static MetricName = MetricName::from_str("opened_msgs_cancel_channel_total");
     312              :     const DEC_NAME: &'static MetricName = MetricName::from_str("closed_msgs_cancel_channel_total");
     313              :     const INC_HELP: &'static str = "Number of processing messages in the cancellation channel.";
     314              :     const DEC_HELP: &'static str = "Number of closed messages in the cancellation channel.";
     315              :     type LabelGroupSet = StaticLabelSet<RedisMsgKind>;
     316              : }
     317              : pub type CancelChannelSizeGuard<'a> = metrics::MeasuredCounterPairGuard<'a, CancelChannelSizeGauge>;
     318              : 
     319          264 : #[derive(LabelGroup)]
     320              : #[label(set = ComputeConnectionLatencySet)]
     321              : pub struct ComputeConnectionLatencyGroup {
     322              :     protocol: Protocol,
     323              :     cold_start_info: ColdStartInfo,
     324              :     outcome: ConnectOutcome,
     325              :     excluded: LatencyExclusions,
     326              : }
     327              : 
     328              : #[derive(FixedCardinalityLabel, Copy, Clone)]
     329              : pub enum LatencyExclusions {
     330              :     Client,
     331              :     ClientAndCplane,
     332              :     ClientCplaneCompute,
     333              :     ClientCplaneComputeRetry,
     334              : }
     335              : 
     336              : #[derive(FixedCardinalityLabel, Copy, Clone)]
     337              : #[label(singleton = "kind")]
     338              : pub enum SniKind {
     339              :     Sni,
     340              :     NoSni,
     341              :     PasswordHack,
     342              : }
     343              : 
     344              : #[derive(FixedCardinalityLabel, Copy, Clone)]
     345              : #[label(singleton = "kind")]
     346              : pub enum ConnectionFailureKind {
     347              :     ComputeCached,
     348              :     ComputeUncached,
     349              : }
     350              : 
     351          176 : #[derive(LabelGroup)]
     352              : #[label(set = ConnectionFailuresBreakdownSet)]
     353              : pub struct ConnectionFailuresBreakdownGroup {
     354              :     pub kind: ErrorKind,
     355              :     pub retry: Bool,
     356              : }
     357              : 
     358           88 : #[derive(LabelGroup, Copy, Clone)]
     359              : #[label(set = RedisErrorsSet)]
     360              : pub struct RedisErrors<'a> {
     361              :     #[label(dynamic_with = ThreadedRodeo, default)]
     362              :     pub channel: &'a str,
     363              : }
     364              : 
     365              : #[derive(FixedCardinalityLabel, Copy, Clone)]
     366              : pub enum CancellationOutcome {
     367              :     NotFound,
     368              :     Found,
     369              :     RateLimitExceeded,
     370              : }
     371              : 
     372          132 : #[derive(LabelGroup)]
     373              : #[label(set = CancellationRequestSet)]
     374              : pub struct CancellationRequest {
     375              :     pub kind: CancellationOutcome,
     376              : }
     377              : 
     378              : #[derive(Clone, Copy)]
     379              : pub enum Waiting {
     380              :     Cplane,
     381              :     Client,
     382              :     Compute,
     383              :     RetryTimeout,
     384              : }
     385              : 
     386              : #[derive(FixedCardinalityLabel, Copy, Clone)]
     387              : #[label(singleton = "kind")]
     388              : #[allow(clippy::enum_variant_names)]
     389              : pub enum RedisMsgKind {
     390              :     HSet,
     391              :     HSetMultiple,
     392              :     HGet,
     393              :     HGetAll,
     394              :     HDel,
     395              : }
     396              : 
     397              : #[derive(Default)]
     398              : struct Accumulated {
     399              :     cplane: time::Duration,
     400              :     client: time::Duration,
     401              :     compute: time::Duration,
     402              :     retry: time::Duration,
     403              : }
     404              : 
     405              : pub struct LatencyTimer {
     406              :     // time since the stopwatch was started
     407              :     start: time::Instant,
     408              :     // time since the stopwatch was stopped
     409              :     stop: Option<time::Instant>,
     410              :     // accumulated time on the stopwatch
     411              :     accumulated: Accumulated,
     412              :     // label data
     413              :     protocol: Protocol,
     414              :     cold_start_info: ColdStartInfo,
     415              :     outcome: ConnectOutcome,
     416              : 
     417              :     skip_reporting: bool,
     418              : }
     419              : 
     420              : impl LatencyTimer {
     421           70 :     pub fn new(protocol: Protocol) -> Self {
     422           70 :         Self {
     423           70 :             start: time::Instant::now(),
     424           70 :             stop: None,
     425           70 :             accumulated: Accumulated::default(),
     426           70 :             protocol,
     427           70 :             cold_start_info: ColdStartInfo::Unknown,
     428           70 :             // assume failed unless otherwise specified
     429           70 :             outcome: ConnectOutcome::Failed,
     430           70 :             skip_reporting: false,
     431           70 :         }
     432           70 :     }
     433              : 
     434            0 :     pub(crate) fn noop(protocol: Protocol) -> Self {
     435            0 :         Self {
     436            0 :             start: time::Instant::now(),
     437            0 :             stop: None,
     438            0 :             accumulated: Accumulated::default(),
     439            0 :             protocol,
     440            0 :             cold_start_info: ColdStartInfo::Unknown,
     441            0 :             // assume failed unless otherwise specified
     442            0 :             outcome: ConnectOutcome::Failed,
     443            0 :             skip_reporting: true,
     444            0 :         }
     445            0 :     }
     446              : 
     447           28 :     pub fn unpause(&mut self, start: Instant, waiting_for: Waiting) {
     448           28 :         let dur = start.elapsed();
     449           28 :         match waiting_for {
     450            0 :             Waiting::Cplane => self.accumulated.cplane += dur,
     451           15 :             Waiting::Client => self.accumulated.client += dur,
     452            7 :             Waiting::Compute => self.accumulated.compute += dur,
     453            6 :             Waiting::RetryTimeout => self.accumulated.retry += dur,
     454              :         }
     455           28 :     }
     456              : 
     457            0 :     pub fn cold_start_info(&mut self, cold_start_info: ColdStartInfo) {
     458            0 :         self.cold_start_info = cold_start_info;
     459            0 :     }
     460              : 
     461            4 :     pub fn success(&mut self) {
     462            4 :         // stop the stopwatch and record the time that we have accumulated
     463            4 :         self.stop = Some(time::Instant::now());
     464            4 : 
     465            4 :         // success
     466            4 :         self.outcome = ConnectOutcome::Success;
     467            4 :     }
     468              : }
     469              : 
     470              : #[derive(FixedCardinalityLabel, Clone, Copy, Debug)]
     471              : pub enum ConnectOutcome {
     472              :     Success,
     473              :     Failed,
     474              : }
     475              : 
     476              : impl Drop for LatencyTimer {
     477           70 :     fn drop(&mut self) {
     478           70 :         if self.skip_reporting {
     479            0 :             return;
     480           70 :         }
     481           70 : 
     482           70 :         let duration = self
     483           70 :             .stop
     484           70 :             .unwrap_or_else(time::Instant::now)
     485           70 :             .duration_since(self.start);
     486           70 : 
     487           70 :         let metric = &Metrics::get().proxy.compute_connection_latency_seconds;
     488           70 : 
     489           70 :         // Excluding client communication from the accumulated time.
     490           70 :         metric.observe(
     491           70 :             ComputeConnectionLatencyGroup {
     492           70 :                 protocol: self.protocol,
     493           70 :                 cold_start_info: self.cold_start_info,
     494           70 :                 outcome: self.outcome,
     495           70 :                 excluded: LatencyExclusions::Client,
     496           70 :             },
     497           70 :             duration
     498           70 :                 .saturating_sub(self.accumulated.client)
     499           70 :                 .as_secs_f64(),
     500           70 :         );
     501           70 : 
     502           70 :         // Exclude client and cplane communication from the accumulated time.
     503           70 :         let accumulated_total = self.accumulated.client + self.accumulated.cplane;
     504           70 :         metric.observe(
     505           70 :             ComputeConnectionLatencyGroup {
     506           70 :                 protocol: self.protocol,
     507           70 :                 cold_start_info: self.cold_start_info,
     508           70 :                 outcome: self.outcome,
     509           70 :                 excluded: LatencyExclusions::ClientAndCplane,
     510           70 :             },
     511           70 :             duration.saturating_sub(accumulated_total).as_secs_f64(),
     512           70 :         );
     513           70 : 
     514           70 :         // Exclude client cplane, compue communication from the accumulated time.
     515           70 :         let accumulated_total =
     516           70 :             self.accumulated.client + self.accumulated.cplane + self.accumulated.compute;
     517           70 :         metric.observe(
     518           70 :             ComputeConnectionLatencyGroup {
     519           70 :                 protocol: self.protocol,
     520           70 :                 cold_start_info: self.cold_start_info,
     521           70 :                 outcome: self.outcome,
     522           70 :                 excluded: LatencyExclusions::ClientCplaneCompute,
     523           70 :             },
     524           70 :             duration.saturating_sub(accumulated_total).as_secs_f64(),
     525           70 :         );
     526           70 : 
     527           70 :         // Exclude client cplane, compue, retry communication from the accumulated time.
     528           70 :         let accumulated_total = self.accumulated.client
     529           70 :             + self.accumulated.cplane
     530           70 :             + self.accumulated.compute
     531           70 :             + self.accumulated.retry;
     532           70 :         metric.observe(
     533           70 :             ComputeConnectionLatencyGroup {
     534           70 :                 protocol: self.protocol,
     535           70 :                 cold_start_info: self.cold_start_info,
     536           70 :                 outcome: self.outcome,
     537           70 :                 excluded: LatencyExclusions::ClientCplaneComputeRetry,
     538           70 :             },
     539           70 :             duration.saturating_sub(accumulated_total).as_secs_f64(),
     540           70 :         );
     541           70 :     }
     542              : }
     543              : 
     544              : impl From<bool> for Bool {
     545            3 :     fn from(value: bool) -> Self {
     546            3 :         if value {
     547            2 :             Bool::True
     548              :         } else {
     549            1 :             Bool::False
     550              :         }
     551            3 :     }
     552              : }
     553              : 
     554          220 : #[derive(LabelGroup)]
     555              : #[label(set = InvalidEndpointsSet)]
     556              : pub struct InvalidEndpointsGroup {
     557              :     pub protocol: Protocol,
     558              :     pub rejected: Bool,
     559              :     pub outcome: ConnectOutcome,
     560              : }
     561              : 
     562          176 : #[derive(LabelGroup)]
     563              : #[label(set = RetriesMetricSet)]
     564              : pub struct RetriesMetricGroup {
     565              :     pub outcome: ConnectOutcome,
     566              :     pub retry_type: RetryType,
     567              : }
     568              : 
     569              : #[derive(FixedCardinalityLabel, Clone, Copy, Debug)]
     570              : pub enum RetryType {
     571              :     WakeCompute,
     572              :     ConnectToCompute,
     573              : }
     574              : 
     575              : #[derive(FixedCardinalityLabel, Clone, Copy, Debug)]
     576              : #[label(singleton = "event")]
     577              : pub enum RedisEventsCount {
     578              :     EndpointCreated,
     579              :     BranchCreated,
     580              :     ProjectCreated,
     581              :     CancelSession,
     582              :     PasswordUpdate,
     583              :     AllowedIpsUpdate,
     584              :     AllowedVpcEndpointIdsUpdateForProjects,
     585              :     AllowedVpcEndpointIdsUpdateForAllProjectsInOrg,
     586              :     BlockPublicOrVpcAccessUpdate,
     587              : }
     588              : 
     589              : pub struct ThreadPoolWorkers(usize);
     590              : #[derive(Copy, Clone)]
     591              : pub struct ThreadPoolWorkerId(pub usize);
     592              : 
     593              : impl LabelValue for ThreadPoolWorkerId {
     594            0 :     fn visit<V: measured::label::LabelVisitor>(&self, v: V) -> V::Output {
     595            0 :         v.write_int(self.0 as i64)
     596            0 :     }
     597              : }
     598              : 
     599              : impl LabelGroup for ThreadPoolWorkerId {
     600            0 :     fn visit_values(&self, v: &mut impl measured::label::LabelGroupVisitor) {
     601            0 :         v.write_value(LabelName::from_str("worker"), self);
     602            0 :     }
     603              : }
     604              : 
     605              : impl LabelGroupSet for ThreadPoolWorkers {
     606              :     type Group<'a> = ThreadPoolWorkerId;
     607              : 
     608          100 :     fn cardinality(&self) -> Option<usize> {
     609          100 :         Some(self.0)
     610          100 :     }
     611              : 
     612            6 :     fn encode_dense(&self, value: Self::Unique) -> Option<usize> {
     613            6 :         Some(value)
     614            6 :     }
     615              : 
     616            0 :     fn decode_dense(&self, value: usize) -> Self::Group<'_> {
     617            0 :         ThreadPoolWorkerId(value)
     618            0 :     }
     619              : 
     620              :     type Unique = usize;
     621              : 
     622            6 :     fn encode(&self, value: Self::Group<'_>) -> Option<Self::Unique> {
     623            6 :         Some(value.0)
     624            6 :     }
     625              : 
     626            0 :     fn decode(&self, value: &Self::Unique) -> Self::Group<'_> {
     627            0 :         ThreadPoolWorkerId(*value)
     628            0 :     }
     629              : }
     630              : 
     631              : impl LabelSet for ThreadPoolWorkers {
     632              :     type Value<'a> = ThreadPoolWorkerId;
     633              : 
     634            0 :     fn dynamic_cardinality(&self) -> Option<usize> {
     635            0 :         Some(self.0)
     636            0 :     }
     637              : 
     638            0 :     fn encode(&self, value: Self::Value<'_>) -> Option<usize> {
     639            0 :         (value.0 < self.0).then_some(value.0)
     640            0 :     }
     641              : 
     642            0 :     fn decode(&self, value: usize) -> Self::Value<'_> {
     643            0 :         ThreadPoolWorkerId(value)
     644            0 :     }
     645              : }
     646              : 
     647              : impl FixedCardinalitySet for ThreadPoolWorkers {
     648            0 :     fn cardinality(&self) -> usize {
     649            0 :         self.0
     650            0 :     }
     651              : }
     652              : 
     653           50 : #[derive(MetricGroup)]
     654              : #[metric(new(workers: usize))]
     655              : pub struct ThreadPoolMetrics {
     656              :     #[metric(init = CounterVec::with_label_set(ThreadPoolWorkers(workers)))]
     657              :     pub worker_task_turns_total: CounterVec<ThreadPoolWorkers>,
     658              :     #[metric(init = CounterVec::with_label_set(ThreadPoolWorkers(workers)))]
     659              :     pub worker_task_skips_total: CounterVec<ThreadPoolWorkers>,
     660              : }
        

Generated by: LCOV version 2.1-beta