{-# LANGUAGE OverloadedStrings #-}

-- | Registration and refresh lifecycle for database gauges.
module Arbiter.Otel.Gauges
  ( startGauges
  , withGaugeLoop
  , reachabilityOf
  ) where

import Arbiter.Core.Job.Schema (SchemaName, TableName)
import Arbiter.Core.MonadArbiter (MonadArbiter)
import Arbiter.Core.Operations (Shared (..), micros)
import Arbiter.Worker (FailureGate, LogConfig (..), LogLevel (Warning), newFailureGate, reportOutcome)
import Control.Concurrent (threadDelay)
import Control.Concurrent.STM (atomically, readTVarIO)
import Control.Exception (SomeException)
import Control.Monad (forever)
import Data.Foldable (for_, traverse_)
import Data.Maybe (isNothing)
import Data.Text (Text)
import Data.Time (NominalDiffTime)
import GHC.Clock (getMonotonicTime)
import UnliftIO.Exception (bracket, finally)

import Arbiter.Otel.Gauges.Cache
  ( Cached (..)
  , GaugeCache (..)
  , live
  , newGaugeCache
  , publishSnapshot
  , retireCache
  , setReachable
  )
import Arbiter.Otel.Gauges.Coordination (RefreshSource (..), prepareRefreshSource)
import Arbiter.Otel.Gauges.Instruments (registerInstruments)
import Arbiter.Otel.Metrics (arbiterMeter)
import Arbiter.Otel.Telemetry qualified as Tel

-- | Register gauge instruments and return their refresh loop.
startGauges
  :: (MonadArbiter m)
  => Tel.Telemetry
  -> LogConfig
  -> (forall a. m a -> IO a)
  -> SchemaName
  -> [(TableName, [Text])]
  -> NominalDiffTime
  -> IO (IO ())
startGauges :: forall (m :: * -> *).
MonadArbiter m =>
Telemetry
-> LogConfig
-> (forall a. m a -> IO a)
-> Text
-> [(Text, [Text])]
-> NominalDiffTime
-> IO (IO ())
startGauges Telemetry
tel LogConfig
baseLog forall a. m a -> IO a
runDb Text
schema [(Text, [Text])]
queueKinds NominalDiffTime
refreshInterval = do
  (loop, stop) <- Telemetry
-> LogConfig
-> (forall a. m a -> IO a)
-> Text
-> [(Text, [Text])]
-> NominalDiffTime
-> IO (IO (), IO ())
forall (m :: * -> *).
MonadArbiter m =>
Telemetry
-> LogConfig
-> (forall a. m a -> IO a)
-> Text
-> [(Text, [Text])]
-> NominalDiffTime
-> IO (IO (), IO ())
prepareGauges Telemetry
tel LogConfig
baseLog m a -> IO a
forall a. m a -> IO a
runDb Text
schema [(Text, [Text])]
queueKinds NominalDiffTime
refreshInterval
  pure (loop `finally` stop)

-- | Run an action with registered gauge instruments and a refresh loop.
withGaugeLoop
  :: (MonadArbiter m)
  => Tel.Telemetry
  -> LogConfig
  -> (forall a. m a -> IO a)
  -> SchemaName
  -> [(TableName, [Text])]
  -> NominalDiffTime
  -> (IO () -> IO b)
  -> IO b
withGaugeLoop :: forall (m :: * -> *) b.
MonadArbiter m =>
Telemetry
-> LogConfig
-> (forall a. m a -> IO a)
-> Text
-> [(Text, [Text])]
-> NominalDiffTime
-> (IO () -> IO b)
-> IO b
withGaugeLoop Telemetry
tel LogConfig
baseLog forall a. m a -> IO a
runDb Text
schema [(Text, [Text])]
queueKinds NominalDiffTime
refreshInterval IO () -> IO b
use =
  IO (IO (), IO ())
-> ((IO (), IO ()) -> IO ()) -> ((IO (), IO ()) -> IO b) -> IO b
forall (m :: * -> *) a b c.
MonadUnliftIO m =>
m a -> (a -> m b) -> (a -> m c) -> m c
bracket (Telemetry
-> LogConfig
-> (forall a. m a -> IO a)
-> Text
-> [(Text, [Text])]
-> NominalDiffTime
-> IO (IO (), IO ())
forall (m :: * -> *).
MonadArbiter m =>
Telemetry
-> LogConfig
-> (forall a. m a -> IO a)
-> Text
-> [(Text, [Text])]
-> NominalDiffTime
-> IO (IO (), IO ())
prepareGauges Telemetry
tel LogConfig
baseLog m a -> IO a
forall a. m a -> IO a
runDb Text
schema [(Text, [Text])]
queueKinds NominalDiffTime
refreshInterval) (IO (), IO ()) -> IO ()
forall a b. (a, b) -> b
snd (IO () -> IO b
use (IO () -> IO b)
-> ((IO (), IO ()) -> IO ()) -> (IO (), IO ()) -> IO b
forall b c a. (b -> c) -> (a -> b) -> a -> c
. (IO (), IO ()) -> IO ()
forall a b. (a, b) -> a
fst)

prepareGauges
  :: (MonadArbiter m)
  => Tel.Telemetry
  -> LogConfig
  -> (forall a. m a -> IO a)
  -> SchemaName
  -> [(TableName, [Text])]
  -> NominalDiffTime
  -> IO (IO (), IO ())
prepareGauges :: forall (m :: * -> *).
MonadArbiter m =>
Telemetry
-> LogConfig
-> (forall a. m a -> IO a)
-> Text
-> [(Text, [Text])]
-> NominalDiffTime
-> IO (IO (), IO ())
prepareGauges Telemetry
tel LogConfig
baseLog forall a. m a -> IO a
runDb Text
schema [(Text, [Text])]
queueKinds NominalDiffTime
requestedInterval
  | Maybe ArbiterMeters -> Bool
forall a. Maybe a -> Bool
isNothing (Telemetry -> Maybe ArbiterMeters
Tel.meters Telemetry
tel) = (IO (), IO ()) -> IO (IO (), IO ())
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (() -> IO ()
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (), () -> IO ()
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure ())
  | Bool
otherwise = do
      cache <- Double -> IO GaugeCache
newGaugeCache (Double -> IO GaugeCache) -> IO Double -> IO GaugeCache
forall (m :: * -> *) a b. Monad m => (a -> m b) -> m a -> m b
=<< IO Double
getMonotonicTime
      advance <- arbiterMeter (Tel.provider tel) >>= \Meter
meter -> Meter -> GaugeCache -> IO (Cached -> IO ())
registerInstruments Meter
meter GaugeCache
cache
      source <-
        prepareRefreshSource
          logCfg
          runDb
          schema
          queueKinds
          refreshInterval
          freshnessWindow
          (cacheIsStale cache freshnessWindow)
      refreshGate <- newFailureGate
      pure (refreshLoop logCfg refreshGate source refreshInterval cache advance, atomically (retireCache cache))
  where
    refreshInterval :: NominalDiffTime
refreshInterval = NominalDiffTime -> NominalDiffTime -> NominalDiffTime
forall a. Ord a => a -> a -> a
max NominalDiffTime
minimumRefreshInterval NominalDiffTime
requestedInterval
    freshnessWindow :: NominalDiffTime
freshnessWindow = NominalDiffTime
freshnessWindowFactor NominalDiffTime -> NominalDiffTime -> NominalDiffTime
forall a. Num a => a -> a -> a
* NominalDiffTime
refreshInterval
    -- Gauge work has no worker-pool identity.
    logCfg :: LogConfig
logCfg = (Telemetry -> LogConfig -> LogConfig
Tel.telemetryLogConfig Telemetry
tel LogConfig
baseLog) {identityContext = []}

-- | The fastest refresh the loop runs at.
minimumRefreshInterval :: NominalDiffTime
minimumRefreshInterval :: NominalDiffTime
minimumRefreshInterval = NominalDiffTime
1

-- | The freshness window over the refresh interval. Covers a slow scan plus one
-- missed refresh.
freshnessWindowFactor :: NominalDiffTime
freshnessWindowFactor :: NominalDiffTime
freshnessWindowFactor = NominalDiffTime
3

refreshLoop :: LogConfig -> FailureGate -> RefreshSource -> NominalDiffTime -> GaugeCache -> (Cached -> IO ()) -> IO ()
refreshLoop :: LogConfig
-> FailureGate
-> RefreshSource
-> NominalDiffTime
-> GaugeCache
-> (Cached -> IO ())
-> IO ()
refreshLoop LogConfig
logCfg FailureGate
refreshGate RefreshSource
source NominalDiffTime
refreshInterval GaugeCache
cache Cached -> IO ()
advance = IO () -> IO ()
forall (f :: * -> *) a b. Applicative f => f a -> f b
forever (IO () -> IO ()) -> IO () -> IO ()
forall a b. (a -> b) -> a -> b
$ do
  started <- IO Double
getMonotonicTime
  refreshed <- runRefresh source
  now <- getMonotonicTime
  -- A failed scan keeps the last reading. A state change gets a log line.
  for_ (reachabilityOf refreshed) $ \Bool
reachable -> do
    LogConfig
-> LogLevel
-> FailureGate
-> Text
-> Either SomeException (Maybe (Shared Snapshot))
-> IO ()
forall (m :: * -> *) a.
MonadUnliftIO m =>
LogConfig
-> LogLevel
-> FailureGate
-> Text
-> Either SomeException a
-> m ()
reportOutcome LogConfig
logCfg LogLevel
Warning FailureGate
refreshGate Text
"Gauge refresh" Either SomeException (Maybe (Shared Snapshot))
refreshed
    STM () -> IO ()
forall a. STM a -> IO a
atomically (GaugeCache -> Bool -> STM ()
setReachable GaugeCache
cache Bool
reachable)
  traverse_ (traverse_ publish . (>>= stamp started now)) refreshed
  threadDelay (max (micros (minimumDelay source)) (micros (refreshInterval - realToFrac (now - started))))
  where
    publish :: Cached -> IO ()
publish Cached
cached = STM () -> IO ()
forall a. STM a -> IO a
atomically (GaugeCache -> Cached -> STM ()
publishSnapshot GaugeCache
cache Cached
cached) IO () -> IO () -> IO ()
forall a b. IO a -> IO b -> IO b
forall (m :: * -> *) a b. Monad m => m a -> m b -> m b
>> Cached -> IO ()
advance Cached
cached
    stamp :: Double -> Double -> Shared Snapshot -> Maybe Cached
stamp Double
started Double
now = \case
      Ran Snapshot
snap -> Cached -> Maybe Cached
forall a. a -> Maybe a
Just (Double -> Snapshot -> Cached
Cached Double
started Snapshot
snap)
      Published Double
age Snapshot
snap -> Cached -> Maybe Cached
forall a. a -> Maybe a
Just (Double -> Snapshot -> Cached
Cached (Double
now Double -> Double -> Double
forall a. Num a => a -> a -> a
- Double
age) Snapshot
snap)
      Unreadable Text
_ -> Maybe Cached
forall a. Maybe a
Nothing

-- | What a scan says about the database. An abandoned scan says nothing.
reachabilityOf :: Either SomeException (Maybe a) -> Maybe Bool
reachabilityOf :: forall a. Either SomeException (Maybe a) -> Maybe Bool
reachabilityOf = (SomeException -> Maybe Bool)
-> (Maybe a -> Maybe Bool)
-> Either SomeException (Maybe a)
-> Maybe Bool
forall a c b. (a -> c) -> (b -> c) -> Either a b -> c
either (Maybe Bool -> SomeException -> Maybe Bool
forall a b. a -> b -> a
const (Bool -> Maybe Bool
forall a. a -> Maybe a
Just Bool
False)) (Bool
True Bool -> Maybe a -> Maybe Bool
forall a b. a -> Maybe b -> Maybe a
forall (f :: * -> *) a b. Functor f => a -> f b -> f a
<$)

cacheIsStale :: GaugeCache -> NominalDiffTime -> IO Bool
cacheIsStale :: GaugeCache -> NominalDiffTime -> IO Bool
cacheIsStale GaugeCache
cache NominalDiffTime
maxAge = do
  now <- IO Double
getMonotonicTime
  cached <- live <$> readTVarIO (export cache)
  pure (maybe True (\Cached
lastCached -> Double
now Double -> Double -> Double
forall a. Num a => a -> a -> a
- Cached -> Double
takenAt Cached
lastCached Double -> Double -> Bool
forall a. Ord a => a -> a -> Bool
> NominalDiffTime -> Double
forall a b. (Real a, Fractional b) => a -> b
realToFrac NominalDiffTime
maxAge) cached)