{-# LANGUAGE OverloadedStrings #-}

-- | OpenTelemetry instruments backed by the gauge snapshot cache.
module Arbiter.Otel.Gauges.Instruments (registerInstruments) where

import Arbiter.Core.Concurrency.Stats qualified as Conc (ConcurrencyPolicyView (..))
import Arbiter.Core.Health qualified as Health
import Arbiter.Core.Operations
  ( QueueOverview (..)
  , QueueStats (..)
  , queueStatusCounts
  )
import Arbiter.Core.RateLimit.Stats qualified as RL (RateLimitPolicyView (..))
import Control.Concurrent.STM (readTVarIO)
import Control.Monad (void)
import Data.Foldable (toList, traverse_)
import Data.HashMap.Strict (HashMap)
import Data.HashMap.Strict qualified as HM
import Data.IORef (IORef, atomicModifyIORef', newIORef)
import Data.Map.Strict qualified as Map
import Data.Maybe (fromMaybe)
import Data.Text (Text)
import GHC.Clock (getMonotonicTime)
import OpenTelemetry.Metric.Core
  ( Counter
  , Meter
  , counterAdd
  , defaultAdvisoryParameters
  , meterCreateCounterDouble
  , meterCreateObservableGaugeDouble
  , observe
  )

import Arbiter.Otel.Gauges.Cache
  ( Baseline
  , Cached (..)
  , GaugeCache (..)
  , SeriesKey
  , Snapshot (..)
  , lastScan
  , live
  , riseSince
  )
import Arbiter.Otel.MetricNames qualified as Name
import Arbiter.Otel.Metrics (attrs, concurrencyKind, rateLimitKind)

-- | Register instruments that read the last shared snapshot, and return the action
-- that carries a freshly published reading onto the counters.
registerInstruments :: Meter -> GaugeCache -> IO (Cached -> IO ())
registerInstruments :: Meter -> GaugeCache -> IO (Cached -> IO ())
registerInstruments Meter
meter GaugeCache
cache = do
  baselines <- HashMap SeriesKey Baseline
-> IO (IORef (HashMap SeriesKey Baseline))
forall a. a -> IO (IORef a)
newIORef HashMap SeriesKey Baseline
forall k v. HashMap k v
HM.empty
  let withCached ObservableResult Double -> Cached -> IO ()
emit = [\ObservableResult Double
res -> TVar Export -> IO Export
forall a. TVar a -> IO a
readTVarIO (GaugeCache -> TVar Export
export GaugeCache
cache) IO Export -> (Export -> IO ()) -> IO ()
forall a b. IO a -> (a -> IO b) -> IO b
forall (m :: * -> *) a b. Monad m => m a -> (a -> m b) -> m b
>>= (Cached -> IO ()) -> Maybe Cached -> IO ()
forall (t :: * -> *) (f :: * -> *) a b.
(Foldable t, Applicative f) =>
(a -> f b) -> t a -> f ()
traverse_ (ObservableResult Double -> Cached -> IO ()
emit ObservableResult Double
res) (Maybe Cached -> IO ())
-> (Export -> Maybe Cached) -> Export -> IO ()
forall b c a. (b -> c) -> (a -> b) -> a -> c
. Export -> Maybe Cached
live]
      callback ObservableResult Double -> Snapshot -> IO ()
emit = (ObservableResult Double -> Cached -> IO ())
-> [ObservableResult Double -> IO ()]
withCached (\ObservableResult Double
res -> ObservableResult Double -> Snapshot -> IO ()
emit ObservableResult Double
res (Snapshot -> IO ()) -> (Cached -> Snapshot) -> Cached -> IO ()
forall b c a. (b -> c) -> (a -> b) -> a -> c
. Cached -> Snapshot
reading)
      -- Every replica exports the winner's reading. Aggregate with max.
      shared a
desc = a
desc a -> a -> a
forall a. Semigroup a => a -> a -> a
<> a
" (shared reading, do not sum across replicas)"
      regGauge MetricName
name Text
unit Text
desc [ObservableResult Double -> IO ()]
cbs =
        IO (ObservableGauge Double) -> IO ()
forall (f :: * -> *) a. Functor f => f a -> f ()
void (IO (ObservableGauge Double) -> IO ())
-> IO (ObservableGauge Double) -> IO ()
forall a b. (a -> b) -> a -> b
$
          Meter
-> Text
-> Maybe Text
-> Maybe Text
-> AdvisoryParameters
-> [ObservableResult Double -> IO ()]
-> IO (ObservableGauge Double)
meterCreateObservableGaugeDouble
            Meter
meter
            (MetricName -> Text
Name.metricName MetricName
name)
            (Text -> Maybe Text
forall a. a -> Maybe a
Just Text
unit)
            (Text -> Maybe Text
forall a. a -> Maybe a
Just Text
desc)
            AdvisoryParameters
defaultAdvisoryParameters
            [ObservableResult Double -> IO ()]
cbs
      reg MetricName
name Text
unit Text
desc ObservableResult Double -> Snapshot -> IO ()
emit = MetricName
-> Text -> Text -> [ObservableResult Double -> IO ()] -> IO ()
regGauge MetricName
name Text
unit (Text -> Text
forall {a}. (Semigroup a, IsString a) => a -> a
shared Text
desc) ((ObservableResult Double -> Snapshot -> IO ())
-> [ObservableResult Double -> IO ()]
callback ObservableResult Double -> Snapshot -> IO ()
emit)
      regCounter MetricName
name Text
unit Text
desc Snapshot -> [([(Text, Text)], Double)]
series = do
        counter <-
          Meter
-> Text
-> Maybe Text
-> Maybe Text
-> AdvisoryParameters
-> IO (Counter Double)
meterCreateCounterDouble
            Meter
meter
            (MetricName -> Text
Name.metricName MetricName
name)
            (Text -> Maybe Text
forall a. a -> Maybe a
Just Text
unit)
            (Text -> Maybe Text
forall a. a -> Maybe a
Just (Text -> Text
forall {a}. (Semigroup a, IsString a) => a -> a
shared Text
desc))
            AdvisoryParameters
defaultAdvisoryParameters
        pure $ \Cached
cached ->
          (([(Text, Text)], Double) -> IO ())
-> [([(Text, Text)], Double)] -> IO ()
forall (t :: * -> *) (f :: * -> *) a b.
(Foldable t, Applicative f) =>
(a -> f b) -> t a -> f ()
traverse_
            (IORef (HashMap SeriesKey Baseline)
-> Text
-> Counter Double
-> Double
-> ([(Text, Text)], Double)
-> IO ()
addRise IORef (HashMap SeriesKey Baseline)
baselines (MetricName -> Text
Name.metricName MetricName
name) Counter Double
counter (Cached -> Double
takenAt Cached
cached))
            (Snapshot -> [([(Text, Text)], Double)]
series (Cached -> Snapshot
reading Cached
cached))

  reg Name.QueueDepth "{job}" "Jobs in a queue by status" $
    observed $
      over queues $ \QueueOverview
overview ->
        [ ([(Text
"queue", QueueOverview -> Text
overviewQueue QueueOverview
overview), (Text
"status", Text
status)], Int64 -> Double
forall a b. (Integral a, Num b) => a -> b
fromIntegral Int64
count)
        | (Text
status, Int64
count) <- QueueStats -> [(Text, Int64)]
queueStatusCounts (QueueOverview -> QueueStats
overviewStats QueueOverview
overview)
        ]
  reg Name.QueueDepthByKind "{job}" "Jobs in a queue by payload variant" $
    observed $
      over queues $ \QueueOverview
overview ->
        [ ([(Text
"queue", QueueOverview -> Text
overviewQueue QueueOverview
overview), (Text
"kind", Text
kind)], Int64 -> Double
forall a b. (Integral a, Num b) => a -> b
fromIntegral Int64
count)
        | (Text
kind, Int64
count) <- Map Text Int64 -> [(Text, Int64)]
forall k a. Map k a -> [(k, a)]
Map.toList (QueueStats -> Map Text Int64
kindCounts (QueueOverview -> QueueStats
overviewStats QueueOverview
overview))
        ]
  reg Name.QueueOldestReadyAge "s" "Time the oldest ready or blocked job has waited, 0 when none is waiting" $
    perQueue oldestReadyAgeSeconds
  reg Name.QueueOldestInFlightAge "s" "Time the longest-running job has been leased, 0 when none is in flight" $
    perQueue oldestInFlightAgeSeconds
  -- Active and paused partition the live workers. A queue's fleet is their sum.
  reg Name.Workers "{worker}" "Live workers by state" $
    observed $
      over queues $ \QueueOverview
overview ->
        let paused :: Int64
paused = QueueOverview -> Int64
overviewWorkersPaused QueueOverview
overview
         in [ ([(Text
"queue", QueueOverview -> Text
overviewQueue QueueOverview
overview), (Text
"state", Text
"active")], Int64 -> Double
forall a b. (Integral a, Num b) => a -> b
fromIntegral (Int64 -> Int64 -> Int64
forall a. Ord a => a -> a -> a
max Int64
0 (QueueOverview -> Int64
overviewWorkersLive QueueOverview
overview Int64 -> Int64 -> Int64
forall a. Num a => a -> a -> a
- Int64
paused)))
            , ([(Text
"queue", QueueOverview -> Text
overviewQueue QueueOverview
overview), (Text
"state", Text
"paused")], Int64 -> Double
forall a b. (Integral a, Num b) => a -> b
fromIntegral Int64
paused)
            ]

  -- Keyed by policy prefix.
  reg Name.AdmissionKeys "{key}" "Live admission keys, by policy" $
    bothKinds (fromIntegral . Conc.keyCount) (fromIntegral . RL.bucketCount)
  reg Name.AdmissionLimit "{admission}" "Effective cap per key, in concurrency slots or rate-limit tokens" $
    bothKinds (fromIntegral . effectiveLimit) effectiveMaxTokens
  reg Name.AdmissionInFlight "{job}" "Jobs holding a concurrency slot, by policy" $
    perConcurrency (fromIntegral . Conc.totalInFlight)
  reg Name.AdmissionBusiestKey "{job}" "In-flight count of the fullest key, by policy" $
    perConcurrency (fromIntegral . fromMaybe 0 . Conc.maxInFlight)
  reg Name.AdmissionTokens "{token}" "Rate-limit tokens left across a policy's buckets" $
    observed $
      over rateLimits $ \RateLimitPolicyView
policy ->
        [ ([(Text
"policy", RateLimitPolicyView -> Text
RL.prefix RateLimitPolicyView
policy), (Text
"stat", Text
stat)], Double
tokens)
        | (Text
stat, Just Double
tokens) <- [(Text
"min", RateLimitPolicyView -> Maybe Double
RL.minTokens RateLimitPolicyView
policy), (Text
"avg", RateLimitPolicyView -> Maybe Double
RL.avgTokens RateLimitPolicyView
policy)]
        ]

  reg Name.PgTableDeadTuples "{tuple}" "Dead tuples pending vacuum" $ perTable (fromIntegral . Health.deadTup)
  reg Name.PgTableLiveTuples "{tuple}" "Estimated live tuples" $ perTable (fromIntegral . Health.liveTup)
  reg Name.PgTableVacuumAge "s" "Seconds since the last vacuum or autovacuum, absent until one runs" $
    perTableMaybe Health.vacuumAge
  reg Name.PgTableSize "By" "Total relation size" $ perTable (fromIntegral . Health.totalBytes)
  reg Name.PgTableXidAge "{transaction}" "Transaction-id age of the table" $
    perTableMaybe (fmap fromIntegral . Health.xidAge)
  reg Name.PgDbConnections "{connection}" "Backends by state, across the whole database" $
    perDbBy "state" (map (fmap fromIntegral) . connCounts)
  reg Name.PgDbOldestTransactionAge "s" "Age of the oldest open transaction, across the whole database" $
    perDb Health.oldestTxnAge
  reg Name.PgDbOldestQueryAge "s" "Age of the oldest running query, across the whole database" $
    perDb Health.oldestQueryAge
  reg Name.PgDbBackends "{backend}" "Backends connected to the database" $
    perDb (fromIntegral . Health.numBackends)

  advances <-
    sequence
      [ regCounter Name.PgTableScans "{scan}" "Table scans, by the access path they took" $
          perTableTotals "path" (\PgTableHealth
tableHealth -> [(Text
"seq", PgTableHealth -> Double
Health.seqScan PgTableHealth
tableHealth), (Text
"index", PgTableHealth -> Double
Health.idxScan PgTableHealth
tableHealth)])
      , regCounter Name.PgTableBlocks "{block}" "Block reads for the table, by whether they hit the cache" $
          perTableTotals "source" (\PgTableHealth
tableHealth -> [(Text
"hit", PgTableHealth -> Double
Health.blksHit PgTableHealth
tableHealth), (Text
"disk", PgTableHealth -> Double
Health.blksRead PgTableHealth
tableHealth)])
      ]

  -- Absent until the first scan.
  regGauge
    Name.DbReachable
    "1"
    "1 when the last gauge scan reached the database, 0 when it failed"
    [ \ObservableResult Double
res ->
        TVar (Maybe Bool) -> IO (Maybe Bool)
forall a. TVar a -> IO a
readTVarIO (GaugeCache -> TVar (Maybe Bool)
databaseReachable GaugeCache
cache) IO (Maybe Bool) -> (Maybe Bool -> IO ()) -> IO ()
forall a b. IO a -> (a -> IO b) -> IO b
forall (m :: * -> *) a b. Monad m => m a -> (a -> m b) -> m b
>>= (Bool -> IO ()) -> Maybe Bool -> IO ()
forall (t :: * -> *) (f :: * -> *) a b.
(Foldable t, Applicative f) =>
(a -> f b) -> t a -> f ()
traverse_ (\Bool
reachable -> ObservableResult Double -> Double -> Attributes -> IO ()
forall a. ObservableResult a -> a -> Attributes -> IO ()
observe ObservableResult Double
res (if Bool
reachable then Double
1 else Double
0) ([(Text, Text)] -> Attributes
attrs []))
    ]

  -- How far behind the exported readings have fallen. Before the first scan it counts
  -- from registration. A stopped loop leaves the other gauges holding their last
  -- reading. This gauge tells that reading apart from a fresh one.
  regGauge
    Name.GaugesAge
    "s"
    "Seconds since the exported readings were scanned"
    [ \ObservableResult Double
res -> do
        now <- IO Double
getMonotonicTime
        scanned <- lastScan <$> readTVarIO (export cache)
        observe res (now - fromMaybe (registeredAt cache) scanned) (attrs [])
    ]

  pure (\Cached
cached -> ((Cached -> IO ()) -> IO ()) -> [Cached -> IO ()] -> IO ()
forall (t :: * -> *) (f :: * -> *) a b.
(Foldable t, Applicative f) =>
(a -> f b) -> t a -> f ()
traverse_ ((Cached -> IO ()) -> Cached -> IO ()
forall a b. (a -> b) -> a -> b
$ Cached
cached) [Cached -> IO ()]
advances)
  where
    effectiveLimit :: ConcurrencyPolicyView -> Int32
effectiveLimit ConcurrencyPolicyView
policy = Int32 -> Maybe Int32 -> Int32
forall a. a -> Maybe a -> a
fromMaybe (ConcurrencyPolicyView -> Int32
Conc.defaultLimit ConcurrencyPolicyView
policy) (ConcurrencyPolicyView -> Maybe Int32
Conc.overrideLimit ConcurrencyPolicyView
policy)
    effectiveMaxTokens :: RateLimitPolicyView -> Double
effectiveMaxTokens RateLimitPolicyView
policy = Double -> Maybe Double -> Double
forall a. a -> Maybe a -> a
fromMaybe (RateLimitPolicyView -> Double
RL.defaultMaxTokens RateLimitPolicyView
policy) (RateLimitPolicyView -> Maybe Double
RL.overrideMaxTokens RateLimitPolicyView
policy)
    connCounts :: PgDbHealth -> [(a, Int64)]
connCounts PgDbHealth
dbHealth =
      [ (a
"active", PgDbHealth -> Int64
Health.connActive PgDbHealth
dbHealth)
      , (a
"idle", PgDbHealth -> Int64
Health.connIdle PgDbHealth
dbHealth)
      , (a
"idle_in_transaction", PgDbHealth -> Int64
Health.connIdleInTxn PgDbHealth
dbHealth)
      , (a
"idle_in_transaction_aborted", PgDbHealth -> Int64
Health.connIdleInTxnAborted PgDbHealth
dbHealth)
      , (a
"blocked", PgDbHealth -> Int64
Health.connBlocked PgDbHealth
dbHealth)
      , (a
"other", PgDbHealth -> Int64
Health.connOther PgDbHealth
dbHealth)
      ]
    -- Every series is a list of rows picked out of the snapshot, each row labelled
    -- and valued. Counters hand that list over, gauges observe it.
    over :: (t -> t a) -> (a -> [b]) -> t -> [b]
over t -> t a
pick a -> [b]
label t
snap = (a -> [b]) -> t a -> [b]
forall (t :: * -> *) a b. Foldable t => (a -> [b]) -> t a -> [b]
concatMap a -> [b]
label (t -> t a
pick t
snap)
    observed :: (a -> t ([(Text, Text)], a)) -> ObservableResult a -> a -> IO ()
observed a -> t ([(Text, Text)], a)
rows ObservableResult a
res = (([(Text, Text)], a) -> IO ()) -> t ([(Text, Text)], a) -> IO ()
forall (t :: * -> *) (f :: * -> *) a b.
(Foldable t, Applicative f) =>
(a -> f b) -> t a -> f ()
traverse_ (\([(Text, Text)]
kvs, a
value) -> ObservableResult a -> a -> Attributes -> IO ()
forall a. ObservableResult a -> a -> Attributes -> IO ()
observe ObservableResult a
res a
value ([(Text, Text)] -> Attributes
attrs [(Text, Text)]
kvs)) (t ([(Text, Text)], a) -> IO ())
-> (a -> t ([(Text, Text)], a)) -> a -> IO ()
forall b c a. (b -> c) -> (a -> b) -> a -> c
. a -> t ([(Text, Text)], a)
rows
    perConcurrency :: (ConcurrencyPolicyView -> a)
-> ObservableResult a -> Snapshot -> IO ()
perConcurrency ConcurrencyPolicyView -> a
field = (Snapshot -> [([(Text, Text)], a)])
-> ObservableResult a -> Snapshot -> IO ()
forall {t :: * -> *} {a} {a}.
Foldable t =>
(a -> t ([(Text, Text)], a)) -> ObservableResult a -> a -> IO ()
observed ((Snapshot -> [ConcurrencyPolicyView])
-> (ConcurrencyPolicyView -> [([(Text, Text)], a)])
-> Snapshot
-> [([(Text, Text)], a)]
forall {t :: * -> *} {t} {a} {b}.
Foldable t =>
(t -> t a) -> (a -> [b]) -> t -> [b]
over Snapshot -> [ConcurrencyPolicyView]
concurrency (\ConcurrencyPolicyView
policy -> [([(Text
"policy", ConcurrencyPolicyView -> Text
Conc.prefix ConcurrencyPolicyView
policy)], ConcurrencyPolicyView -> a
field ConcurrencyPolicyView
policy)]))
    bothKinds :: (ConcurrencyPolicyView -> a)
-> (RateLimitPolicyView -> a)
-> ObservableResult a
-> Snapshot
-> IO ()
bothKinds ConcurrencyPolicyView -> a
concField RateLimitPolicyView -> a
rateField = (Snapshot -> [([(Text, Text)], a)])
-> ObservableResult a -> Snapshot -> IO ()
forall {t :: * -> *} {a} {a}.
Foldable t =>
(a -> t ([(Text, Text)], a)) -> ObservableResult a -> a -> IO ()
observed ((Snapshot -> [([(Text, Text)], a)])
 -> ObservableResult a -> Snapshot -> IO ())
-> (Snapshot -> [([(Text, Text)], a)])
-> ObservableResult a
-> Snapshot
-> IO ()
forall a b. (a -> b) -> a -> b
$ \Snapshot
snap ->
      [([(Text
"policy_kind", Text
concurrencyKind), (Text
"policy", ConcurrencyPolicyView -> Text
Conc.prefix ConcurrencyPolicyView
policy)], ConcurrencyPolicyView -> a
concField ConcurrencyPolicyView
policy) | ConcurrencyPolicyView
policy <- Snapshot -> [ConcurrencyPolicyView]
concurrency Snapshot
snap]
        [([(Text, Text)], a)]
-> [([(Text, Text)], a)] -> [([(Text, Text)], a)]
forall a. Semigroup a => a -> a -> a
<> [([(Text
"policy_kind", Text
rateLimitKind), (Text
"policy", RateLimitPolicyView -> Text
RL.prefix RateLimitPolicyView
policy)], RateLimitPolicyView -> a
rateField RateLimitPolicyView
policy) | RateLimitPolicyView
policy <- Snapshot -> [RateLimitPolicyView]
rateLimits Snapshot
snap]
    dbOf :: Snapshot -> [PgDbHealth]
dbOf = Maybe PgDbHealth -> [PgDbHealth]
forall a. Maybe a -> [a]
forall (t :: * -> *) a. Foldable t => t a -> [a]
toList (Maybe PgDbHealth -> [PgDbHealth])
-> (Snapshot -> Maybe PgDbHealth) -> Snapshot -> [PgDbHealth]
forall b c a. (b -> c) -> (a -> b) -> a -> c
. Snapshot -> Maybe PgDbHealth
db
    perTableTotals :: a
-> (PgTableHealth -> [(Text, b)]) -> Snapshot -> [([(a, Text)], b)]
perTableTotals a
label PgTableHealth -> [(Text, b)]
pairs =
      (Snapshot -> [PgTableHealth])
-> (PgTableHealth -> [([(a, Text)], b)])
-> Snapshot
-> [([(a, Text)], b)]
forall {t :: * -> *} {t} {a} {b}.
Foldable t =>
(t -> t a) -> (a -> [b]) -> t -> [b]
over
        Snapshot -> [PgTableHealth]
tables
        (\PgTableHealth
tableHealth -> [([(a
"table", PgTableHealth -> Text
Health.table PgTableHealth
tableHealth), (a
label, Text
key)], b
value) | (Text
key, b
value) <- PgTableHealth -> [(Text, b)]
pairs PgTableHealth
tableHealth])
    perTable :: (PgTableHealth -> a) -> ObservableResult a -> Snapshot -> IO ()
perTable PgTableHealth -> a
field = (PgTableHealth -> Maybe a)
-> ObservableResult a -> Snapshot -> IO ()
forall {t :: * -> *} {a}.
Foldable t =>
(PgTableHealth -> t a) -> ObservableResult a -> Snapshot -> IO ()
perTableMaybe (a -> Maybe a
forall a. a -> Maybe a
Just (a -> Maybe a) -> (PgTableHealth -> a) -> PgTableHealth -> Maybe a
forall b c a. (b -> c) -> (a -> b) -> a -> c
. PgTableHealth -> a
field)
    perTableMaybe :: (PgTableHealth -> t a) -> ObservableResult a -> Snapshot -> IO ()
perTableMaybe PgTableHealth -> t a
field =
      (Snapshot -> [([(Text, Text)], a)])
-> ObservableResult a -> Snapshot -> IO ()
forall {t :: * -> *} {a} {a}.
Foldable t =>
(a -> t ([(Text, Text)], a)) -> ObservableResult a -> a -> IO ()
observed
        ((Snapshot -> [PgTableHealth])
-> (PgTableHealth -> [([(Text, Text)], a)])
-> Snapshot
-> [([(Text, Text)], a)]
forall {t :: * -> *} {t} {a} {b}.
Foldable t =>
(t -> t a) -> (a -> [b]) -> t -> [b]
over Snapshot -> [PgTableHealth]
tables (\PgTableHealth
tableHealth -> [([(Text
"table", PgTableHealth -> Text
Health.table PgTableHealth
tableHealth)], a
value) | a
value <- t a -> [a]
forall a. t a -> [a]
forall (t :: * -> *) a. Foldable t => t a -> [a]
toList (PgTableHealth -> t a
field PgTableHealth
tableHealth)]))
    perQueue :: (QueueStats -> Maybe a) -> ObservableResult a -> Snapshot -> IO ()
perQueue QueueStats -> Maybe a
field =
      (Snapshot -> [([(Text, Text)], a)])
-> ObservableResult a -> Snapshot -> IO ()
forall {t :: * -> *} {a} {a}.
Foldable t =>
(a -> t ([(Text, Text)], a)) -> ObservableResult a -> a -> IO ()
observed
        ((Snapshot -> [QueueOverview])
-> (QueueOverview -> [([(Text, Text)], a)])
-> Snapshot
-> [([(Text, Text)], a)]
forall {t :: * -> *} {t} {a} {b}.
Foldable t =>
(t -> t a) -> (a -> [b]) -> t -> [b]
over Snapshot -> [QueueOverview]
queues (\QueueOverview
overview -> [([(Text
"queue", QueueOverview -> Text
overviewQueue QueueOverview
overview)], a -> Maybe a -> a
forall a. a -> Maybe a -> a
fromMaybe a
0 (QueueStats -> Maybe a
field (QueueOverview -> QueueStats
overviewStats QueueOverview
overview)))]))
    perDb :: (PgDbHealth -> a) -> ObservableResult a -> Snapshot -> IO ()
perDb PgDbHealth -> a
field = (Snapshot -> [([(Text, Text)], a)])
-> ObservableResult a -> Snapshot -> IO ()
forall {t :: * -> *} {a} {a}.
Foldable t =>
(a -> t ([(Text, Text)], a)) -> ObservableResult a -> a -> IO ()
observed ((Snapshot -> [PgDbHealth])
-> (PgDbHealth -> [([(Text, Text)], a)])
-> Snapshot
-> [([(Text, Text)], a)]
forall {t :: * -> *} {t} {a} {b}.
Foldable t =>
(t -> t a) -> (a -> [b]) -> t -> [b]
over Snapshot -> [PgDbHealth]
dbOf (\PgDbHealth
dbHealth -> [([], PgDbHealth -> a
field PgDbHealth
dbHealth)]))
    perDbBy :: Text
-> (PgDbHealth -> [(Text, a)])
-> ObservableResult a
-> Snapshot
-> IO ()
perDbBy Text
label PgDbHealth -> [(Text, a)]
pairs = (Snapshot -> [([(Text, Text)], a)])
-> ObservableResult a -> Snapshot -> IO ()
forall {t :: * -> *} {a} {a}.
Foldable t =>
(a -> t ([(Text, Text)], a)) -> ObservableResult a -> a -> IO ()
observed ((Snapshot -> [PgDbHealth])
-> (PgDbHealth -> [([(Text, Text)], a)])
-> Snapshot
-> [([(Text, Text)], a)]
forall {t :: * -> *} {t} {a} {b}.
Foldable t =>
(t -> t a) -> (a -> [b]) -> t -> [b]
over Snapshot -> [PgDbHealth]
dbOf (\PgDbHealth
dbHealth -> [([(Text
label, Text
key)], a
value) | (Text
key, a
value) <- PgDbHealth -> [(Text, a)]
pairs PgDbHealth
dbHealth]))

-- | Count an absolute total's rise since the scan it was last counted from.
addRise
  :: IORef (HashMap SeriesKey Baseline)
  -> Text
  -> Counter Double
  -> Double
  -- ^ Monotonic time the reading was scanned at.
  -> ([(Text, Text)], Double)
  -> IO ()
addRise :: IORef (HashMap SeriesKey Baseline)
-> Text
-> Counter Double
-> Double
-> ([(Text, Text)], Double)
-> IO ()
addRise IORef (HashMap SeriesKey Baseline)
baselines Text
name Counter Double
counter Double
scannedAt ([(Text, Text)]
kvs, Double
total) = do
  rise <- IORef (HashMap SeriesKey Baseline)
-> (HashMap SeriesKey Baseline
    -> (HashMap SeriesKey Baseline, Double))
-> IO Double
forall a b. IORef a -> (a -> (a, b)) -> IO b
atomicModifyIORef' IORef (HashMap SeriesKey Baseline)
baselines (SeriesKey
-> Double
-> Double
-> HashMap SeriesKey Baseline
-> (HashMap SeriesKey Baseline, Double)
riseSince (Text
name, [(Text, Text)]
kvs) Double
scannedAt Double
total)
  counterAdd counter rise (attrs kvs)