{-# 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.Job.Types (jobStatusToText)
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.Bifunctor (first)
import Data.Foldable (toList, traverse_)
import Data.HashMap.Strict (HashMap)
import Data.IORef (IORef, atomicModifyIORef')
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
  let withCached :: (ObservableResult Double -> Cached -> IO ())
-> [ObservableResult Double -> IO ()]
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 ())
-> [ObservableResult Double -> IO ()]
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 -> a
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
-> Text -> Text -> [ObservableResult Double -> IO ()] -> IO ()
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
-> Text
-> Text
-> (ObservableResult Double -> Snapshot -> IO ())
-> IO ()
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
-> Text
-> Text
-> (Snapshot -> [([(Text, Text)], Double)])
-> IO (Cached -> IO ())
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 (GaugeCache -> IORef (HashMap SeriesKey Baseline)
counterBaselines GaugeCache
cache) (MetricName -> Text
Name.metricName MetricName
name) Counter Double
counter (Cached -> Double
takenAt Cached
cached))
            (Snapshot -> [([(Text, Text)], Double)]
series (Cached -> Snapshot
reading Cached
cached))

  MetricName
-> Text
-> Text
-> (ObservableResult Double -> Snapshot -> IO ())
-> IO ()
reg MetricName
Name.QueueDepth Text
"{job}" Text
"Jobs in a queue by status" ((ObservableResult Double -> Snapshot -> IO ()) -> IO ())
-> (ObservableResult Double -> Snapshot -> IO ()) -> IO ()
forall a b. (a -> b) -> a -> b
$
    (Snapshot -> [([(Text, Text)], Double)])
-> ObservableResult Double -> Snapshot -> IO ()
forall {t :: * -> *} {a} {a}.
Foldable t =>
(a -> t ([(Text, Text)], a)) -> ObservableResult a -> a -> IO ()
observed ((Snapshot -> [([(Text, Text)], Double)])
 -> ObservableResult Double -> Snapshot -> IO ())
-> (Snapshot -> [([(Text, Text)], Double)])
-> ObservableResult Double
-> Snapshot
-> IO ()
forall a b. (a -> b) -> a -> b
$
      (Snapshot -> [QueueOverview])
-> (QueueOverview -> [([(Text, Text)], Double)])
-> Snapshot
-> [([(Text, Text)], Double)]
forall {t :: * -> *} {t} {a} {b}.
Foldable t =>
(t -> t a) -> (a -> [b]) -> t -> [b]
over Snapshot -> [QueueOverview]
queues ((QueueOverview -> [([(Text, Text)], Double)])
 -> Snapshot -> [([(Text, Text)], Double)])
-> (QueueOverview -> [([(Text, Text)], Double)])
-> Snapshot
-> [([(Text, Text)], Double)]
forall a b. (a -> b) -> a -> b
$ \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)]
statusCounts (QueueOverview -> QueueStats
overviewStats QueueOverview
overview)
        ]
  MetricName
-> Text
-> Text
-> (ObservableResult Double -> Snapshot -> IO ())
-> IO ()
reg MetricName
Name.QueueDepthByKind Text
"{job}" Text
"Jobs in a queue by payload variant" ((ObservableResult Double -> Snapshot -> IO ()) -> IO ())
-> (ObservableResult Double -> Snapshot -> IO ()) -> IO ()
forall a b. (a -> b) -> a -> b
$
    (Snapshot -> [([(Text, Text)], Double)])
-> ObservableResult Double -> Snapshot -> IO ()
forall {t :: * -> *} {a} {a}.
Foldable t =>
(a -> t ([(Text, Text)], a)) -> ObservableResult a -> a -> IO ()
observed ((Snapshot -> [([(Text, Text)], Double)])
 -> ObservableResult Double -> Snapshot -> IO ())
-> (Snapshot -> [([(Text, Text)], Double)])
-> ObservableResult Double
-> Snapshot
-> IO ()
forall a b. (a -> b) -> a -> b
$
      (Snapshot -> [QueueOverview])
-> (QueueOverview -> [([(Text, Text)], Double)])
-> Snapshot
-> [([(Text, Text)], Double)]
forall {t :: * -> *} {t} {a} {b}.
Foldable t =>
(t -> t a) -> (a -> [b]) -> t -> [b]
over Snapshot -> [QueueOverview]
queues ((QueueOverview -> [([(Text, Text)], Double)])
 -> Snapshot -> [([(Text, Text)], Double)])
-> (QueueOverview -> [([(Text, Text)], Double)])
-> Snapshot
-> [([(Text, Text)], Double)]
forall a b. (a -> b) -> a -> b
$ \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))
        ]
  MetricName
-> Text
-> Text
-> (ObservableResult Double -> Snapshot -> IO ())
-> IO ()
reg MetricName
Name.QueueOldestReadyAge Text
"s" Text
"Age of the oldest claimable job (0 = none ready)" ((ObservableResult Double -> Snapshot -> IO ()) -> IO ())
-> (ObservableResult Double -> Snapshot -> IO ()) -> IO ()
forall a b. (a -> b) -> a -> b
$
    (QueueStats -> Maybe Double)
-> ObservableResult Double -> Snapshot -> IO ()
forall {a}.
Num a =>
(QueueStats -> Maybe a) -> ObservableResult a -> Snapshot -> IO ()
perQueue QueueStats -> Maybe Double
oldestReadyAgeSeconds
  MetricName
-> Text
-> Text
-> (ObservableResult Double -> Snapshot -> IO ())
-> IO ()
reg MetricName
Name.QueueOldestInFlightAge Text
"s" Text
"Time the longest-running job has been leased (0 = none in flight)" ((ObservableResult Double -> Snapshot -> IO ()) -> IO ())
-> (ObservableResult Double -> Snapshot -> IO ()) -> IO ()
forall a b. (a -> b) -> a -> b
$
    (QueueStats -> Maybe Double)
-> ObservableResult Double -> Snapshot -> IO ()
forall {a}.
Num a =>
(QueueStats -> Maybe a) -> ObservableResult a -> Snapshot -> IO ()
perQueue QueueStats -> Maybe Double
oldestInFlightAgeSeconds
  -- Active and paused partition the pools with a fresh heartbeat. A queue's fleet is their sum.
  MetricName
-> Text
-> Text
-> (ObservableResult Double -> Snapshot -> IO ())
-> IO ()
reg MetricName
Name.Workers Text
"{worker}" Text
"Registered workers by state" ((ObservableResult Double -> Snapshot -> IO ()) -> IO ())
-> (ObservableResult Double -> Snapshot -> IO ()) -> IO ()
forall a b. (a -> b) -> a -> b
$
    (Snapshot -> [([(Text, Text)], Double)])
-> ObservableResult Double -> Snapshot -> IO ()
forall {t :: * -> *} {a} {a}.
Foldable t =>
(a -> t ([(Text, Text)], a)) -> ObservableResult a -> a -> IO ()
observed ((Snapshot -> [([(Text, Text)], Double)])
 -> ObservableResult Double -> Snapshot -> IO ())
-> (Snapshot -> [([(Text, Text)], Double)])
-> ObservableResult Double
-> Snapshot
-> IO ()
forall a b. (a -> b) -> a -> b
$
      (Snapshot -> [QueueOverview])
-> (QueueOverview -> [([(Text, Text)], Double)])
-> Snapshot
-> [([(Text, Text)], Double)]
forall {t :: * -> *} {t} {a} {b}.
Foldable t =>
(t -> t a) -> (a -> [b]) -> t -> [b]
over Snapshot -> [QueueOverview]
queues ((QueueOverview -> [([(Text, Text)], Double)])
 -> Snapshot -> [([(Text, Text)], Double)])
-> (QueueOverview -> [([(Text, Text)], Double)])
-> Snapshot
-> [([(Text, Text)], Double)]
forall a b. (a -> b) -> a -> b
$ \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.
  MetricName
-> Text
-> Text
-> (ObservableResult Double -> Snapshot -> IO ())
-> IO ()
reg MetricName
Name.AdmissionKeys Text
"{key}" Text
"Live admission keys, by policy" ((ObservableResult Double -> Snapshot -> IO ()) -> IO ())
-> (ObservableResult Double -> Snapshot -> IO ()) -> IO ()
forall a b. (a -> b) -> a -> b
$
    (ConcurrencyPolicyView -> Double)
-> (RateLimitPolicyView -> Double)
-> ObservableResult Double
-> Snapshot
-> IO ()
forall {a}.
(ConcurrencyPolicyView -> a)
-> (RateLimitPolicyView -> a)
-> ObservableResult a
-> Snapshot
-> IO ()
bothKinds (Int64 -> Double
forall a b. (Integral a, Num b) => a -> b
fromIntegral (Int64 -> Double)
-> (ConcurrencyPolicyView -> Int64)
-> ConcurrencyPolicyView
-> Double
forall b c a. (b -> c) -> (a -> b) -> a -> c
. ConcurrencyPolicyView -> Int64
Conc.keyCount) (Int64 -> Double
forall a b. (Integral a, Num b) => a -> b
fromIntegral (Int64 -> Double)
-> (RateLimitPolicyView -> Int64) -> RateLimitPolicyView -> Double
forall b c a. (b -> c) -> (a -> b) -> a -> c
. RateLimitPolicyView -> Int64
RL.bucketCount)
  MetricName
-> Text
-> Text
-> (ObservableResult Double -> Snapshot -> IO ())
-> IO ()
reg MetricName
Name.AdmissionLimit Text
"{slot}" Text
"Effective cap per key (concurrency slots, rate-limit tokens)" ((ObservableResult Double -> Snapshot -> IO ()) -> IO ())
-> (ObservableResult Double -> Snapshot -> IO ()) -> IO ()
forall a b. (a -> b) -> a -> b
$
    (ConcurrencyPolicyView -> Double)
-> (RateLimitPolicyView -> Double)
-> ObservableResult Double
-> Snapshot
-> IO ()
forall {a}.
(ConcurrencyPolicyView -> a)
-> (RateLimitPolicyView -> a)
-> ObservableResult a
-> Snapshot
-> IO ()
bothKinds (Int32 -> Double
forall a b. (Integral a, Num b) => a -> b
fromIntegral (Int32 -> Double)
-> (ConcurrencyPolicyView -> Int32)
-> ConcurrencyPolicyView
-> Double
forall b c a. (b -> c) -> (a -> b) -> a -> c
. ConcurrencyPolicyView -> Int32
effectiveLimit) RateLimitPolicyView -> Double
effectiveMaxTokens
  MetricName
-> Text
-> Text
-> (ObservableResult Double -> Snapshot -> IO ())
-> IO ()
reg MetricName
Name.AdmissionInFlight Text
"{job}" Text
"Jobs holding a concurrency slot, by policy" ((ObservableResult Double -> Snapshot -> IO ()) -> IO ())
-> (ObservableResult Double -> Snapshot -> IO ()) -> IO ()
forall a b. (a -> b) -> a -> b
$
    (ConcurrencyPolicyView -> Double)
-> ObservableResult Double -> Snapshot -> IO ()
forall {a}.
(ConcurrencyPolicyView -> a)
-> ObservableResult a -> Snapshot -> IO ()
perConcurrency (Int64 -> Double
forall a b. (Integral a, Num b) => a -> b
fromIntegral (Int64 -> Double)
-> (ConcurrencyPolicyView -> Int64)
-> ConcurrencyPolicyView
-> Double
forall b c a. (b -> c) -> (a -> b) -> a -> c
. ConcurrencyPolicyView -> Int64
Conc.totalInFlight)
  MetricName
-> Text
-> Text
-> (ObservableResult Double -> Snapshot -> IO ())
-> IO ()
reg MetricName
Name.AdmissionBusiestKey Text
"{job}" Text
"In-flight count of the fullest key, by policy" ((ObservableResult Double -> Snapshot -> IO ()) -> IO ())
-> (ObservableResult Double -> Snapshot -> IO ()) -> IO ()
forall a b. (a -> b) -> a -> b
$
    (ConcurrencyPolicyView -> Double)
-> ObservableResult Double -> Snapshot -> IO ()
forall {a}.
(ConcurrencyPolicyView -> a)
-> ObservableResult a -> Snapshot -> IO ()
perConcurrency (Int32 -> Double
forall a b. (Integral a, Num b) => a -> b
fromIntegral (Int32 -> Double)
-> (ConcurrencyPolicyView -> Int32)
-> ConcurrencyPolicyView
-> Double
forall b c a. (b -> c) -> (a -> b) -> a -> c
. Int32 -> Maybe Int32 -> Int32
forall a. a -> Maybe a -> a
fromMaybe Int32
0 (Maybe Int32 -> Int32)
-> (ConcurrencyPolicyView -> Maybe Int32)
-> ConcurrencyPolicyView
-> Int32
forall b c a. (b -> c) -> (a -> b) -> a -> c
. ConcurrencyPolicyView -> Maybe Int32
Conc.maxInFlight)
  MetricName
-> Text
-> Text
-> (ObservableResult Double -> Snapshot -> IO ())
-> IO ()
reg MetricName
Name.AdmissionTokens Text
"{token}" Text
"Rate-limit tokens left across a policy's buckets" ((ObservableResult Double -> Snapshot -> IO ()) -> IO ())
-> (ObservableResult Double -> Snapshot -> IO ()) -> IO ()
forall a b. (a -> b) -> a -> b
$
    (Snapshot -> [([(Text, Text)], Double)])
-> ObservableResult Double -> Snapshot -> IO ()
forall {t :: * -> *} {a} {a}.
Foldable t =>
(a -> t ([(Text, Text)], a)) -> ObservableResult a -> a -> IO ()
observed ((Snapshot -> [([(Text, Text)], Double)])
 -> ObservableResult Double -> Snapshot -> IO ())
-> (Snapshot -> [([(Text, Text)], Double)])
-> ObservableResult Double
-> Snapshot
-> IO ()
forall a b. (a -> b) -> a -> b
$
      (Snapshot -> [RateLimitPolicyView])
-> (RateLimitPolicyView -> [([(Text, Text)], Double)])
-> Snapshot
-> [([(Text, Text)], Double)]
forall {t :: * -> *} {t} {a} {b}.
Foldable t =>
(t -> t a) -> (a -> [b]) -> t -> [b]
over Snapshot -> [RateLimitPolicyView]
rateLimits ((RateLimitPolicyView -> [([(Text, Text)], Double)])
 -> Snapshot -> [([(Text, Text)], Double)])
-> (RateLimitPolicyView -> [([(Text, Text)], Double)])
-> Snapshot
-> [([(Text, Text)], Double)]
forall a b. (a -> b) -> a -> b
$ \RateLimitPolicyView
policy ->
        [ ([(Text
"policy", RateLimitPolicyView -> Text
RL.prefix RateLimitPolicyView
policy), (Text
"stat", Text
stat)], Double -> Maybe Double -> Double
forall a. a -> Maybe a -> a
fromMaybe Double
0 Maybe Double
tokens)
        | (Text
stat, Maybe Double
tokens) <- [(Text
"min", RateLimitPolicyView -> Maybe Double
RL.minTokens RateLimitPolicyView
policy), (Text
"avg", RateLimitPolicyView -> Maybe Double
RL.avgTokens RateLimitPolicyView
policy)]
        ]

  MetricName
-> Text
-> Text
-> (ObservableResult Double -> Snapshot -> IO ())
-> IO ()
reg MetricName
Name.PgTableDeadTuples Text
"{tuple}" Text
"Dead tuples pending vacuum" ((ObservableResult Double -> Snapshot -> IO ()) -> IO ())
-> (ObservableResult Double -> Snapshot -> IO ()) -> IO ()
forall a b. (a -> b) -> a -> b
$ (PgTableHealth -> Double)
-> ObservableResult Double -> Snapshot -> IO ()
forall {a}.
(PgTableHealth -> a) -> ObservableResult a -> Snapshot -> IO ()
perTable (Int64 -> Double
forall a b. (Integral a, Num b) => a -> b
fromIntegral (Int64 -> Double)
-> (PgTableHealth -> Int64) -> PgTableHealth -> Double
forall b c a. (b -> c) -> (a -> b) -> a -> c
. PgTableHealth -> Int64
Health.deadTup)
  MetricName
-> Text
-> Text
-> (ObservableResult Double -> Snapshot -> IO ())
-> IO ()
reg MetricName
Name.PgTableLiveTuples Text
"{tuple}" Text
"Estimated live tuples" ((ObservableResult Double -> Snapshot -> IO ()) -> IO ())
-> (ObservableResult Double -> Snapshot -> IO ()) -> IO ()
forall a b. (a -> b) -> a -> b
$ (PgTableHealth -> Double)
-> ObservableResult Double -> Snapshot -> IO ()
forall {a}.
(PgTableHealth -> a) -> ObservableResult a -> Snapshot -> IO ()
perTable (Int64 -> Double
forall a b. (Integral a, Num b) => a -> b
fromIntegral (Int64 -> Double)
-> (PgTableHealth -> Int64) -> PgTableHealth -> Double
forall b c a. (b -> c) -> (a -> b) -> a -> c
. PgTableHealth -> Int64
Health.liveTup)
  MetricName
-> Text
-> Text
-> (ObservableResult Double -> Snapshot -> IO ())
-> IO ()
reg MetricName
Name.PgTableAutovacuumAge Text
"s" Text
"Seconds since last (auto)vacuum, absent until one runs" ((ObservableResult Double -> Snapshot -> IO ()) -> IO ())
-> (ObservableResult Double -> Snapshot -> IO ()) -> IO ()
forall a b. (a -> b) -> a -> b
$
    (PgTableHealth -> Maybe Double)
-> ObservableResult Double -> Snapshot -> IO ()
forall {t :: * -> *} {a}.
Foldable t =>
(PgTableHealth -> t a) -> ObservableResult a -> Snapshot -> IO ()
perTableMaybe PgTableHealth -> Maybe Double
Health.autovacuumAge
  MetricName
-> Text
-> Text
-> (ObservableResult Double -> Snapshot -> IO ())
-> IO ()
reg MetricName
Name.PgTableSizeBytes Text
"By" Text
"Total relation size" ((ObservableResult Double -> Snapshot -> IO ()) -> IO ())
-> (ObservableResult Double -> Snapshot -> IO ()) -> IO ()
forall a b. (a -> b) -> a -> b
$ (PgTableHealth -> Double)
-> ObservableResult Double -> Snapshot -> IO ()
forall {a}.
(PgTableHealth -> a) -> ObservableResult a -> Snapshot -> IO ()
perTable (Int64 -> Double
forall a b. (Integral a, Num b) => a -> b
fromIntegral (Int64 -> Double)
-> (PgTableHealth -> Int64) -> PgTableHealth -> Double
forall b c a. (b -> c) -> (a -> b) -> a -> c
. PgTableHealth -> Int64
Health.totalBytes)
  MetricName
-> Text
-> Text
-> (ObservableResult Double -> Snapshot -> IO ())
-> IO ()
reg MetricName
Name.PgTableXidAge Text
"{transaction}" Text
"Transaction-id age of the table (wraparound headroom)" ((ObservableResult Double -> Snapshot -> IO ()) -> IO ())
-> (ObservableResult Double -> Snapshot -> IO ()) -> IO ()
forall a b. (a -> b) -> a -> b
$
    (PgTableHealth -> Maybe Double)
-> ObservableResult Double -> Snapshot -> IO ()
forall {t :: * -> *} {a}.
Foldable t =>
(PgTableHealth -> t a) -> ObservableResult a -> Snapshot -> IO ()
perTableMaybe ((Int64 -> Double) -> Maybe Int64 -> Maybe Double
forall a b. (a -> b) -> Maybe a -> Maybe b
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
fmap Int64 -> Double
forall a b. (Integral a, Num b) => a -> b
fromIntegral (Maybe Int64 -> Maybe Double)
-> (PgTableHealth -> Maybe Int64) -> PgTableHealth -> Maybe Double
forall b c a. (b -> c) -> (a -> b) -> a -> c
. PgTableHealth -> Maybe Int64
Health.xidAge)
  MetricName
-> Text
-> Text
-> (ObservableResult Double -> Snapshot -> IO ())
-> IO ()
reg MetricName
Name.PgDbConnections Text
"{connection}" Text
"Backends by state, across the whole database" ((ObservableResult Double -> Snapshot -> IO ()) -> IO ())
-> (ObservableResult Double -> Snapshot -> IO ()) -> IO ()
forall a b. (a -> b) -> a -> b
$
    Text
-> (PgDbHealth -> [(Text, Double)])
-> ObservableResult Double
-> Snapshot
-> IO ()
forall {a}.
Text
-> (PgDbHealth -> [(Text, a)])
-> ObservableResult a
-> Snapshot
-> IO ()
perDbBy Text
"state" (((Text, Int64) -> (Text, Double))
-> [(Text, Int64)] -> [(Text, Double)]
forall a b. (a -> b) -> [a] -> [b]
map ((Int64 -> Double) -> (Text, Int64) -> (Text, Double)
forall a b. (a -> b) -> (Text, a) -> (Text, b)
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
fmap Int64 -> Double
forall a b. (Integral a, Num b) => a -> b
fromIntegral) ([(Text, Int64)] -> [(Text, Double)])
-> (PgDbHealth -> [(Text, Int64)])
-> PgDbHealth
-> [(Text, Double)]
forall b c a. (b -> c) -> (a -> b) -> a -> c
. PgDbHealth -> [(Text, Int64)]
forall {a}. IsString a => PgDbHealth -> [(a, Int64)]
connCounts)
  MetricName
-> Text
-> Text
-> (ObservableResult Double -> Snapshot -> IO ())
-> IO ()
reg MetricName
Name.PgDbOldestTransactionAge Text
"s" Text
"Age of the oldest open transaction, across the whole database" ((ObservableResult Double -> Snapshot -> IO ()) -> IO ())
-> (ObservableResult Double -> Snapshot -> IO ()) -> IO ()
forall a b. (a -> b) -> a -> b
$
    (PgDbHealth -> Double)
-> ObservableResult Double -> Snapshot -> IO ()
forall {a}.
(PgDbHealth -> a) -> ObservableResult a -> Snapshot -> IO ()
perDb PgDbHealth -> Double
Health.oldestTxnAge
  MetricName
-> Text
-> Text
-> (ObservableResult Double -> Snapshot -> IO ())
-> IO ()
reg MetricName
Name.PgDbOldestQueryAge Text
"s" Text
"Age of the oldest running query, across the whole database" ((ObservableResult Double -> Snapshot -> IO ()) -> IO ())
-> (ObservableResult Double -> Snapshot -> IO ()) -> IO ()
forall a b. (a -> b) -> a -> b
$
    (PgDbHealth -> Double)
-> ObservableResult Double -> Snapshot -> IO ()
forall {a}.
(PgDbHealth -> a) -> ObservableResult a -> Snapshot -> IO ()
perDb PgDbHealth -> Double
Health.oldestQueryAge
  MetricName
-> Text
-> Text
-> (ObservableResult Double -> Snapshot -> IO ())
-> IO ()
reg MetricName
Name.PgDbBackends Text
"{backend}" Text
"Backends connected to the database" ((ObservableResult Double -> Snapshot -> IO ()) -> IO ())
-> (ObservableResult Double -> Snapshot -> IO ()) -> IO ()
forall a b. (a -> b) -> a -> b
$
    (PgDbHealth -> Double)
-> ObservableResult Double -> Snapshot -> IO ()
forall {a}.
(PgDbHealth -> a) -> ObservableResult a -> Snapshot -> IO ()
perDb (Int64 -> Double
forall a b. (Integral a, Num b) => a -> b
fromIntegral (Int64 -> Double) -> (PgDbHealth -> Int64) -> PgDbHealth -> Double
forall b c a. (b -> c) -> (a -> b) -> a -> c
. PgDbHealth -> Int64
Health.numBackends)

  advances <-
    [IO (Cached -> IO ())] -> IO [Cached -> IO ()]
forall (t :: * -> *) (m :: * -> *) a.
(Traversable t, Monad m) =>
t (m a) -> m (t a)
forall (m :: * -> *) a. Monad m => [m a] -> m [a]
sequence
      [ MetricName
-> Text
-> Text
-> (Snapshot -> [([(Text, Text)], Double)])
-> IO (Cached -> IO ())
regCounter MetricName
Name.PgTableScans Text
"{scan}" Text
"Table scans, by the access path they took" ((Snapshot -> [([(Text, Text)], Double)]) -> IO (Cached -> IO ()))
-> (Snapshot -> [([(Text, Text)], Double)]) -> IO (Cached -> IO ())
forall a b. (a -> b) -> a -> b
$
          Text
-> (PgTableHealth -> [(Text, Double)])
-> Snapshot
-> [([(Text, Text)], Double)]
forall {a} {b}.
IsString a =>
a
-> (PgTableHealth -> [(Text, b)]) -> Snapshot -> [([(a, Text)], b)]
perTableTotals Text
"path" (\PgTableHealth
tableHealth -> [(Text
"seq", PgTableHealth -> Double
Health.seqScan PgTableHealth
tableHealth), (Text
"index", PgTableHealth -> Double
Health.idxScan PgTableHealth
tableHealth)])
      , MetricName
-> Text
-> Text
-> (Snapshot -> [([(Text, Text)], Double)])
-> IO (Cached -> IO ())
regCounter MetricName
Name.PgTableBlocks Text
"{block}" Text
"Block reads for the table, by whether they hit the cache" ((Snapshot -> [([(Text, Text)], Double)]) -> IO (Cached -> IO ()))
-> (Snapshot -> [([(Text, Text)], Double)]) -> IO (Cached -> IO ())
forall a b. (a -> b) -> a -> b
$
          Text
-> (PgTableHealth -> [(Text, Double)])
-> Snapshot
-> [([(Text, Text)], Double)]
forall {a} {b}.
IsString a =>
a
-> (PgTableHealth -> [(Text, b)]) -> Snapshot -> [([(a, Text)], b)]
perTableTotals Text
"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
    "{status}"
    "1 when the last health 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)
    statusCounts :: QueueStats -> [(Text, Int64)]
statusCounts = ((JobStatus, Int64) -> (Text, Int64))
-> [(JobStatus, Int64)] -> [(Text, Int64)]
forall a b. (a -> b) -> [a] -> [b]
map ((JobStatus -> Text) -> (JobStatus, Int64) -> (Text, Int64)
forall a b c. (a -> b) -> (a, c) -> (b, c)
forall (p :: * -> * -> *) a b c.
Bifunctor p =>
(a -> b) -> p a c -> p b c
first JobStatus -> Text
jobStatusToText) ([(JobStatus, Int64)] -> [(Text, Int64)])
-> (QueueStats -> [(JobStatus, Int64)])
-> QueueStats
-> [(Text, Int64)]
forall b c a. (b -> c) -> (a -> b) -> a -> c
. QueueStats -> [(JobStatus, Int64)]
queueStatusCounts
    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])
    perDbTotals :: a -> (PgDbHealth -> [(b, b)]) -> Snapshot -> [([(a, b)], b)]
perDbTotals a
label PgDbHealth -> [(b, b)]
pairs = (Snapshot -> [PgDbHealth])
-> (PgDbHealth -> [([(a, b)], b)]) -> Snapshot -> [([(a, b)], b)]
forall {t :: * -> *} {t} {a} {b}.
Foldable t =>
(t -> t a) -> (a -> [b]) -> t -> [b]
over Snapshot -> [PgDbHealth]
dbOf (\PgDbHealth
dbHealth -> [([(a
label, b
key)], b
value) | (b
key, b
value) <- PgDbHealth -> [(b, b)]
pairs PgDbHealth
dbHealth])
    dbTotal :: (PgDbHealth -> b) -> Snapshot -> [([a], b)]
dbTotal PgDbHealth -> b
field = (Snapshot -> [PgDbHealth])
-> (PgDbHealth -> [([a], b)]) -> Snapshot -> [([a], b)]
forall {t :: * -> *} {t} {a} {b}.
Foldable t =>
(t -> t a) -> (a -> [b]) -> t -> [b]
over Snapshot -> [PgDbHealth]
dbOf (\PgDbHealth
dbHealth -> [([], PgDbHealth -> b
field PgDbHealth
dbHealth)])
    perTable :: (PgTableHealth -> a) -> ObservableResult a -> Snapshot -> IO ()
perTable PgTableHealth -> 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)], PgTableHealth -> a
field PgTableHealth
tableHealth)]))
    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 = (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 ())
-> ((PgDbHealth -> a) -> Snapshot -> [([(Text, Text)], a)])
-> (PgDbHealth -> a)
-> ObservableResult a
-> Snapshot
-> IO ()
forall b c a. (b -> c) -> (a -> b) -> a -> c
. (PgDbHealth -> a) -> Snapshot -> [([(Text, Text)], a)]
forall {b} {a}. (PgDbHealth -> b) -> Snapshot -> [([a], b)]
dbTotal
    perDbBy :: Text
-> (PgDbHealth -> [(Text, a)])
-> ObservableResult a
-> Snapshot
-> IO ()
perDbBy Text
label = (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 ())
-> ((PgDbHealth -> [(Text, a)])
    -> Snapshot -> [([(Text, Text)], a)])
-> (PgDbHealth -> [(Text, a)])
-> ObservableResult a
-> Snapshot
-> IO ()
forall b c a. (b -> c) -> (a -> b) -> a -> c
. Text
-> (PgDbHealth -> [(Text, a)]) -> Snapshot -> [([(Text, Text)], a)]
forall {a} {b} {b}.
a -> (PgDbHealth -> [(b, b)]) -> Snapshot -> [([(a, b)], b)]
perDbTotals Text
label

-- | 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)