Metrics

default.nix

@@ -1,7 +1,8 @@
 { mkDerivation, aeson, annotated-exception, base, bytestring
-, containers, directory, exceptions, filepath, generic-lens, katip
-, lens, lens-aeson, lib, mtl, optparse-generic, pandoc, pipes, stm
-, streamly, streamly-core, text, time, unliftio, wreq
+, containers, directory, ekg-core, exceptions, filepath
+, generic-lens, katip, lens, lens-aeson, lib, monad-control, mtl
+, optparse-generic, pandoc, pipes, retry, stm, streamly
+, streamly-core, text, time, transformers-base, unliftio, wreq
 }:
 mkDerivation {
   pname = "wallabag-tagger";
@@ -11,9 +12,9 @@ mkDerivation {
   isExecutable = true;
   libraryHaskellDepends = [
     aeson annotated-exception base bytestring containers directory
-    exceptions filepath generic-lens katip lens lens-aeson mtl
-    optparse-generic pandoc pipes stm streamly streamly-core text time
-    unliftio wreq
+    ekg-core exceptions filepath generic-lens katip lens lens-aeson
+    monad-control mtl optparse-generic pandoc pipes retry stm streamly
+    streamly-core text time transformers-base unliftio wreq
   ];
   executableHaskellDepends = [ base ];
   testHaskellDepends = [ base ];

src/AI.hs

@@ -3,19 +3,35 @@
 
 module AI where
 
-import Control.Lens (re, traversed, (&), (.~), (^.), (^?))
-import Control.Monad.Reader (liftIO)
+import Control.Exception.Annotated (checkpointCallStack)
+import Control.Lens (re, traversed, view, (&), (.~), (^.), (^?))
+import Control.Monad.Catch (MonadCatch)
+import Control.Monad.Reader (MonadReader, liftIO)
 import Creds
 import Data.Aeson (object, (.=))
 import Data.Aeson.Lens (key, _Array, _String)
 import Data.Text (Text)
 import Data.Text.Strict.Lens (utf8)
-import Network.Wreq (defaults, header, postWith, responseBody)
+import HTTP (HTTPMetric, Retry (SafeTransient), Trace, postWith, tracerMetrics)
+import M (HasOpenrouterMetrics, openrouterMetrics)
+import Network.Wreq (defaults, header, responseBody)
+import Tracer (Tracer (..))
 import UnliftIO (MonadIO)
-import Control.Exception.Annotated (checkpointCallStack)
-import Control.Monad.Catch (MonadCatch)
 
-completions :: (MonadIO m, MonadCatch m) => Creds -> Text -> m (Maybe Text)
+mkTracerMetrics :: (MonadReader r m, HasOpenrouterMetrics r HTTPMetric) => m (Tracer HTTP.Trace)
+mkTracerMetrics = do
+  m <- view openrouterMetrics
+  pure $ HTTP.tracerMetrics m
+
+completions ::
+  ( MonadIO m,
+    MonadCatch m,
+    MonadReader r0 m,
+    HasOpenrouterMetrics r0 HTTPMetric
+  ) =>
+  Creds ->
+  Text ->
+  m (Maybe Text)
 completions c prompt = checkpointCallStack $ do
   let opts = defaults & header "Authorization" .~ ["Bearer " <> c ^. openrouterKey . re utf8]
   let j =
@@ -24,5 +40,6 @@ completions c prompt = checkpointCallStack $ do
             "messages"
               .= [object ["role" .= ("user" :: Text), "content" .= prompt]]
           ]
-  r <- liftIO $ postWith opts "https://openrouter.ai/api/v1/chat/completions" j
+  metrics <- mkTracerMetrics
+  r <- liftIO $ postWith metrics SafeTransient opts "https://openrouter.ai/api/v1/chat/completions" j
   pure $ r ^? responseBody . key "choices" . _Array . traversed . key "message" . key "content" . _String

src/HTTP.hs

@@ -0,0 +1,125 @@
+{-# LANGUAGE LambdaCase #-}
+
+module HTTP
+  ( post,
+    postWith,
+    get,
+    getWith,
+    Retry (..),
+    Trace (..),
+    HTTPMetric(..),
+    tracerMetrics
+  )
+where
+
+import Control.Exception.Annotated (catch, throw)
+import Control.Monad.Catch (Handler (Handler), SomeException)
+import Control.Retry (RetryStatus (RetryStatus, rsIterNumber), recovering, retryPolicyDefault)
+import Data.ByteString.Lazy (ByteString)
+import Data.Time.Clock.POSIX (POSIXTime, getPOSIXTime)
+import Network.HTTP.Client (HttpException (..), HttpExceptionContent (..), Response (..), method)
+import Network.HTTP.Types (methodGet, methodHead, status408, status429, status500, status502, status503, status504)
+import Network.Wreq qualified as Wreq
+import Network.Wreq.Types qualified as Wreq
+import Tracer
+import System.Metrics.Distribution (Distribution)
+import System.Metrics.Counter (Counter)
+import qualified System.Metrics.Counter as Counter
+import qualified System.Metrics.Distribution as Distribution
+
+data Trace
+  = Retrying Int
+  | Failing SomeException
+  | Success POSIXTime POSIXTime
+  deriving (Show)
+
+data Retry = SafeTransient | Transient | NoRetry
+  deriving (Eq)
+
+retryRules :: Retry -> [RetryStatus -> Handler IO Bool]
+retryRules NoRetry = []
+retryRules rules =
+  [ const $ Handler $ \case
+      HttpExceptionRequest req ConnectionClosed -> pure $ safe req
+      HttpExceptionRequest req ConnectionTimeout -> pure $ safe req
+      HttpExceptionRequest req ResponseTimeout -> pure $ safe req
+      HttpExceptionRequest req (StatusCodeException res _) -> pure (safe req && retryStatus res)
+      _ -> pure False
+  ]
+  where
+    safe req = rules == SafeTransient || method req == methodGet || method req == methodHead
+    retryStatus res =
+      responseStatus res == status503
+        || responseStatus res == status408
+        || responseStatus res == status429
+        || responseStatus res == status500
+        || responseStatus res == status502
+        || responseStatus res == status504
+
+traceRetry :: Tracer Trace -> RetryStatus -> IO ()
+traceRetry tracer RetryStatus {rsIterNumber = n} | n > 0 = trace tracer $ Retrying n
+traceRetry _wtracer _rs = pure ()
+
+post :: (Wreq.Postable a) => Tracer Trace -> Retry -> String -> a -> IO (Wreq.Response ByteString)
+post tracer rules url postable = do
+  start <- getPOSIXTime
+  res <-
+    recovering
+      retryPolicyDefault
+      (retryRules rules)
+      (\rs -> traceRetry tracer rs >> Wreq.post url postable)
+      `catch` (\e -> trace tracer (Failing e) >> throw e)
+  end <- getPOSIXTime
+  trace tracer $ Success start end
+  pure res
+
+postWith :: (Wreq.Postable a) => Tracer Trace -> Retry -> Wreq.Options -> String -> a -> IO (Wreq.Response ByteString)
+postWith tracer rules opts url postable = do
+  start <- getPOSIXTime
+  res <-
+    recovering
+      retryPolicyDefault
+      (retryRules rules)
+      (\rs -> traceRetry tracer rs >> Wreq.postWith opts url postable)
+      `catch` (\e -> trace tracer (Failing e) >> throw e)
+  end <- getPOSIXTime
+  trace tracer $ Success start end
+  pure res
+
+get :: Tracer Trace -> Retry -> String -> IO (Response ByteString)
+get tracer rules url = do
+  start <- getPOSIXTime
+  res <-
+    recovering
+      retryPolicyDefault
+      (retryRules rules)
+      (\rs -> traceRetry tracer rs >> Wreq.get url)
+      `catch` (\e -> trace tracer (Failing e) >> throw e)
+  end <- getPOSIXTime
+  trace tracer $ Success start end
+  pure res
+
+getWith :: Tracer Trace -> Retry -> Wreq.Options -> String -> IO (Response ByteString)
+getWith tracer rules opts url = do
+  start <- getPOSIXTime
+  res <-
+    recovering
+      retryPolicyDefault
+      (retryRules rules)
+      (\rs -> traceRetry tracer rs >> Wreq.getWith opts url)
+      `catch` (\e -> trace tracer (Failing e) >> throw e)
+  end <- getPOSIXTime
+  trace tracer $ Success start end
+  pure res
+
+data HTTPMetric = HTTPMetric
+  { httpRequests :: Distribution,
+    httpRetries :: Counter,
+    httpFailures :: Counter
+  }
+
+tracerMetrics :: HTTPMetric -> Tracer HTTP.Trace
+tracerMetrics m = Tracer $ \case
+  HTTP.Retrying _ -> Counter.inc (httpRetries m)
+  HTTP.Failing _ -> Counter.inc (httpFailures m)
+  HTTP.Success start end -> Distribution.add (httpRequests m) (realToFrac end - realToFrac start)

src/M.hs

@@ -1,8 +1,8 @@
 {-# LANGUAGE FunctionalDependencies #-}
 {-# LANGUAGE OverloadedStrings #-}
 {-# LANGUAGE TemplateHaskell #-}
-{-# OPTIONS_GHC -Wall #-}
 {-# LANGUAGE UndecidableInstances #-}
+{-# OPTIONS_GHC -Wall #-}
 
 module M where
 
@@ -10,23 +10,28 @@ import Control.Concurrent.STM
 import Control.Exception (bracket)
 import Control.Lens (over, view)
 import Control.Lens.TH (makeFields)
+import Control.Monad.Base (MonadBase)
 import Control.Monad.Catch (MonadCatch, MonadThrow)
 import Control.Monad.Reader (MonadIO, MonadReader, ReaderT, local, runReaderT)
+import Control.Monad.Trans.Control (MonadBaseControl)
 import Creds
 import GHC.Generics (Generic)
+import HTTP (HTTPMetric (HTTPMetric))
 import Katip qualified as K
 import System.IO (stdout)
+import System.Metrics qualified as EKG
 import UnliftIO (MonadUnliftIO)
-import Wallabag (Auth, Auth' (..))
-import Control.Monad.Trans.Control (MonadBaseControl)
-import Control.Monad.Base (MonadBase)
+import Wallabag.Auth (Auth, Auth' (..))
+import qualified Data.HashMap.Strict as HM
 
 data S = S
   { _sTokenCache :: Auth,
     _sLogNamespace :: K.Namespace,
     _sLogEnv :: K.LogEnv,
     _sLogContext :: K.LogContexts,
-    _sCreds :: Creds
+    _sCreds :: Creds,
+    _sWallabagMetrics :: HTTPMetric,
+    _sOpenrouterMetrics :: HTTPMetric
   }
   deriving (Generic)
 
@@ -61,9 +66,23 @@ instance K.KatipContext M where
 
 initialS :: K.Severity -> (S -> IO a) -> IO a
 initialS severity f = do
+  store <- EKG.newStore
+  wbMetrics <-
+    HTTPMetric
+      <$> EKG.createDistribution "wallabag.requests" store
+      <*> EKG.createCounter "wallabag.retries" store
+      <*> EKG.createCounter "wallabag.failures" store
+  orMetrics <-
+    HTTPMetric
+      <$> EKG.createDistribution "openrouter.requests" store
+      <*> EKG.createCounter "openrouter.retries" store
+      <*> EKG.createCounter "openrouter.failures" store
   c <- getCreds
   handleScribe <- K.mkHandleScribe K.ColorIfTerminal stdout (K.permitItem severity) K.V2
   let makeLogEnv = K.registerScribe "stdout" handleScribe K.defaultScribeSettings =<< K.initLogEnv "wallabag" "production"
-  bracket makeLogEnv K.closeScribes $ \le -> do
+  res <- bracket makeLogEnv K.closeScribes $ \le -> do
     cache <- newTVarIO Unauthenticated
-    f (S cache "main" le mempty c)
+    f (S cache "main" le mempty c wbMetrics orMetrics)
+  -- TODO: This is just for debug purposes
+  EKG.sampleAll store >>= mapM_ print . HM.toList
+  pure res

src/MyLib.hs

@@ -16,10 +16,10 @@ import Data.Text.Strict.Lens (utf8)
 import Katip qualified as K
 import M
 import Options.Generic (Generic, ParseRecord (..), getRecord, unHelpful, type (<?>))
+import Streamly.Data.Fold.Prelude qualified as Fold
+import Streamly.Data.Stream.Prelude qualified as Stream
 import Text.Pandoc (PandocError, def, readHtml, runPure, writeMarkdown)
 import Wallabag
-import qualified Streamly.Data.Fold.Prelude as Fold
-import qualified Streamly.Data.Stream.Prelude as Stream
 
 data Options = Options
   { full :: Bool <?> "Go through all the entries",
@@ -51,8 +51,9 @@ run :: K.Severity -> Scan -> IO ()
 run sev scan = initialS sev $ \s -> do
   runM s $ do
     Stream.fold Fold.drain $
-      sourceEntries scan & markdownEntry & processableEntries &
-      Stream.parMapM (Stream.maxThreads 5) doWork
+      sourceEntries scan
+        & markdownEntry & processableEntries
+        & Stream.parMapM (Stream.maxThreads 5) doWork
 
 test :: IO ()
 test = run K.DebugS Quick
@@ -110,6 +111,8 @@ hardLabels =
       "AI"
     ]
 
+
+
 defaultMain :: IO ()
 defaultMain = do
   opts <- getRecord "Wallabag tagger"

src/Tracer.hs

@@ -0,0 +1,9 @@
+{-# LANGUAGE DerivingVia #-}
+module Tracer where
+import Data.Functor.Contravariant
+
+
+newtype Tracer a = Tracer { trace :: a -> IO () }
+  deriving Contravariant via Op (IO ())
+  deriving (Semigroup, Monoid) via (a -> IO ())
+

src/Wallabag.hs

@@ -1,44 +1,38 @@
 {-# LANGUAGE FunctionalDependencies #-}
-{-# LANGUAGE StrictData #-}
+{-# LANGUAGE LambdaCase #-}
 {-# LANGUAGE OverloadedStrings #-}
 {-# LANGUAGE ScopedTypeVariables #-}
+{-# LANGUAGE StrictData #-}
 {-# LANGUAGE TemplateHaskell #-}
 {-# OPTIONS_GHC -Wall #-}
 
 module Wallabag where
 
-import Control.Concurrent.STM (TVar, atomically, readTVar, retry, writeTVar)
+import Control.Concurrent.STM (atomically, readTVar, retry, writeTVar)
 import Control.Exception (Exception)
+import Control.Exception.Annotated (Annotation (..), checkpoint, throw)
 import Control.Lens (re, to, traversed, view, (&), (.~), (^.), (^..), (^?))
 import Control.Lens.TH (makeFields)
-import Control.Monad.Catch (MonadThrow, MonadCatch)
-import Control.Exception.Annotated (throw, checkpoint, Annotation (..))
+import Control.Monad.Catch (MonadCatch, MonadThrow)
 import Control.Monad.Reader (MonadReader, liftIO)
 import Creds
 import Data.Aeson (Value)
 import Data.Aeson.Lens (key, _Array, _Integer, _String)
 import Data.Generics.Product (HasType (..))
+import Data.Maybe (mapMaybe)
 import Data.Text (Text)
 import Data.Text qualified as T
 import Data.Text.Strict.Lens (unpacked, utf8)
-import Data.Time (UTCTime, addUTCTime, getCurrentTime, secondsToNominalDiffTime)
-import Katip qualified as K
-import Network.Wreq (FormParam (..), defaults, getWith, header, post, postWith, responseBody)
-import UnliftIO (MonadIO)
+import Data.Time (addUTCTime, getCurrentTime, secondsToNominalDiffTime)
 import GHC.Stack (HasCallStack)
-import qualified Streamly.Data.Stream.Prelude as Stream
-import Data.Maybe (mapMaybe)
-
-data Auth'
-  = Unauthenticated
-  | Authenticating
-  | Auth
-      { accessToken :: Text,
-        expires :: UTCTime
-      }
-  deriving (Show)
-
-type Auth = TVar Auth'
+import HTTP (HTTPMetric, Retry (SafeTransient), Trace, getWith, post, postWith, tracerMetrics)
+import Katip qualified as K
+import M (HasWallabagMetrics, wallabagMetrics)
+import Network.Wreq (FormParam (..), defaults, header, responseBody)
+import Streamly.Data.Stream.Prelude qualified as Stream
+import Tracer (Tracer (..))
+import UnliftIO (MonadIO)
+import Wallabag.Auth
 
 data Entry = Entry
   { _entryIdentifier :: Integer,
@@ -50,7 +44,14 @@ data Entry = Entry
 
 makeFields ''Entry
 
-authenticate :: (MonadReader r m, MonadIO m, K.KatipContext m) => Creds -> m (Maybe Auth')
+authenticate ::
+  ( MonadReader r m,
+    MonadIO m,
+    K.KatipContext m,
+    HasWallabagMetrics r HTTPMetric
+  ) =>
+  Creds ->
+  m (Maybe Auth')
 authenticate c = do
   K.logFM K.DebugS "Authenticating"
   let form =
@@ -60,7 +61,8 @@ authenticate c = do
           "username" := c ^. username,
           "password" := c ^. password
         ]
-  r <- liftIO $ post "https://app.wallabag.it/oauth/v2/token" form
+  metrics <- mkTracerMetrics
+  r <- liftIO $ post metrics SafeTransient "https://app.wallabag.it/oauth/v2/token" form
   now <- liftIO getCurrentTime
   let expiresIn = r ^? responseBody . key "expires_in" . _Integer . to fromIntegral . to secondsToNominalDiffTime
       expireTime = fmap (`addUTCTime` now) expiresIn
@@ -70,7 +72,15 @@ data WallabagException7g = MissingTokenException deriving (Show)
 
 instance Exception WallabagException7g
 
-getToken :: (HasType Creds r, HasType Auth r, MonadReader r m, MonadIO m, K.KatipContext m) => m (Maybe Text)
+getToken ::
+  ( HasType Creds r,
+    HasType Auth r,
+    MonadReader r m,
+    MonadIO m,
+    K.KatipContext m,
+    HasWallabagMetrics r HTTPMetric
+  ) =>
+  m (Maybe Text)
 getToken = do
   now <- liftIO getCurrentTime
   cache <- view (typed @Auth)
@@ -101,7 +111,8 @@ sourceEntries ::
     K.KatipContext m,
     MonadThrow m,
     MonadCatch m,
-    HasCallStack
+    HasCallStack,
+    HasWallabagMetrics r HTTPMetric
   ) =>
   Scan ->
   Stream.Stream m Entry
@@ -114,10 +125,11 @@ sourceEntries s = Stream.unfoldrM go (Just startUrl) & Stream.concatMap (Stream.
     go :: Maybe String -> m (Maybe ([Value], Maybe String))
     go Nothing = pure Nothing
     go (Just url) = checkpoint (Annotation url) $ do
+      metrics <- mkTracerMetrics
       t <- maybe (throw MissingTokenException) pure =<< getToken
       let opts = defaults & header "Authorization" .~ ["Bearer " <> t ^. re utf8]
       K.logFM K.DebugS $ K.ls ("Fetching entries '" <> url <> "'")
-      r <- liftIO $ getWith opts url
+      r <- liftIO $ getWith metrics SafeTransient opts url
       let items = r ^.. responseBody . key "_embedded" . key "items" . _Array . traversed
       let next = r ^? responseBody . key "_links" . key "next" . key "href" . _String . unpacked
       pure $ Just (items, next)
@@ -129,6 +141,11 @@ sourceEntries s = Stream.unfoldrM go (Just startUrl) & Stream.concatMap (Stream.
         <*> Just (v ^.. key "tags" . _Array . traversed . key "label" . _String)
         <*> v ^? key "title" . _String
 
+mkTracerMetrics :: (MonadReader r m, HasWallabagMetrics r HTTPMetric) => m (Tracer HTTP.Trace)
+mkTracerMetrics = do
+  m <- view wallabagMetrics
+  pure $ HTTP.tracerMetrics m
+
 applyTags ::
   ( HasType Auth r,
     HasType Creds r,
@@ -137,7 +154,8 @@ applyTags ::
     HasCallStack,
     K.KatipContext m,
     MonadThrow m,
-    MonadCatch m
+    MonadCatch m,
+    HasWallabagMetrics r HTTPMetric
   ) =>
   Entry ->
   m ()
@@ -146,5 +164,6 @@ applyTags e = checkpoint (Annotation $ e ^. identifier) $ do
   let opts = defaults & header "Authorization" .~ ["Bearer " <> t ^. re utf8]
   K.logFM K.InfoS (K.ls ("Applying tags for '" <> e ^. title <> "': " <> T.pack (show (e ^. tags))))
   let url = "https://app.wallabag.it/api/entries/" <> show (e ^. identifier) <> "/tags"
-  _ <- liftIO $ postWith opts url ["tags" := T.intercalate "," (e ^. tags)]
+  metrics <- mkTracerMetrics
+  _ <- liftIO $ postWith metrics SafeTransient opts url ["tags" := T.intercalate "," (e ^. tags)]
   pure ()

src/Wallabag/Auth.hs

@@ -0,0 +1,16 @@
+module Wallabag.Auth where
+
+import Control.Concurrent.STM (TVar)
+import Data.Text (Text)
+import Data.Time (UTCTime)
+
+data Auth'
+  = Unauthenticated
+  | Authenticating
+  | Auth
+      { accessToken :: Text,
+        expires :: UTCTime
+      }
+  deriving (Show)
+
+type Auth = TVar Auth'

wallabag-tagger.cabal

@@ -22,16 +22,24 @@ library
                       Creds
                       M
                       Wallabag
+                      Wallabag.Auth
                       MyLib
+                      HTTP
+                      Tracer
     -- other-modules:
     -- other-extensions:
     build-depends:    base ^>=4.18.2.1
                     , wreq
                     , lens
                     , katip
+                    , http-client
+                    , http-types
+                    , ekg-core
+                    , retry
                     , pipes
                     , streamly
                     , streamly-core
+                    , unordered-containers
                     , monad-control
                     , transformers-base
                     , pandoc