feat: improve logging/metrics wrt. batch jobs

src/Jobs.hs

@@ -102,6 +102,7 @@ handleJobs foundation@UniWorX{..}
       jobCurrentCrontab <- liftIO $ newTVarIO Nothing
       jobHeldLocks <- liftIO $ newTVarIO Set.empty
       registerJobHeldLocksCount jobHeldLocks
+      registerJobWorkerQueueDepth appJobState
       atomically $ putTMVar appJobState JobState
         { jobContext = JobContext{..}
         , ..
@@ -452,6 +453,12 @@ handleJobs' wNum = C.mapM_ $ \jctl -> hoist delimitInternalState . withJobWorker
         JobHandlerException act -> do
           act & withJobWorkerState wNum (JobWorkerExecJob content)
           runDB $ setSerializable cleanup
+        JobHandlerAtomicWithFinalizer act fin -> do
+          res <- runDBJobs . setSerializable $ do
+            res <- act & withJobWorkerState wNum (JobWorkerExecJob content)
+            hoist lift cleanup
+            return res
+          fin res
     handleCmd JobCtlDetermineCrontab = do
       newCTab <- liftHandler . runDB $ setSerializable determineCrontab'
       -- logDebugS logIdent $ tshow newCTab

src/Jobs/Handler/Files.hs

@@ -34,11 +34,13 @@ import Handler.Utils.Files (sourceFileDB)
 
 
 dispatchJobPruneSessionFiles :: JobHandler UniWorX
-dispatchJobPruneSessionFiles = JobHandlerAtomic  . hoist lift $ do
+dispatchJobPruneSessionFiles = JobHandlerAtomicWithFinalizer act fin
+  where
+    act = hoist lift $ do
       now <- liftIO getCurrentTime
       expires <- getsYesod $ view _appSessionFilesExpire
-  n <- deleteWhereCount [ SessionFileTouched <. addUTCTime (- expires) now ]
+      deleteWhereCount [ SessionFileTouched <. addUTCTime (- expires) now ]
-  $logInfoS "PruneSessionFiles" [st|Deleted #{n} expired session files|]
+    fin n = $logInfoS "PruneSessionFiles" [st|Deleted #{n} expired session files|]
 
 
 
@@ -64,7 +66,9 @@ pruneUnreferencedFilesIntervalsCache :: TVar (Map Natural [(Maybe FileContentChu
 pruneUnreferencedFilesIntervalsCache = unsafePerformIO $ newTVarIO Map.empty
 
 dispatchJobPruneUnreferencedFiles :: Natural -> Natural -> Natural -> JobHandler UniWorX
-dispatchJobPruneUnreferencedFiles numIterations epoch iteration = JobHandlerAtomic . hoist lift $ do
+dispatchJobPruneUnreferencedFiles numIterations epoch iteration = JobHandlerAtomicWithFinalizer act fin
+  where
+    act = hoist lift $ do
       now <- liftIO getCurrentTime
       interval <- getsYesod $ view _appPruneUnreferencedFilesInterval
       keep <- fmap (max 0) . getsYesod $ view _appKeepUnreferencedFiles
@@ -171,9 +175,6 @@ dispatchJobPruneUnreferencedFiles numIterations epoch iteration = JobHandlerAtom
         .| C.mapM deleteEntry
         .| C.fold
 
-  when (deletedEntries > 0) $
-    $logInfoS "PruneUnreferencedFiles" [st|Deleted #{tshow deletedEntries} long-unreferenced files|]
-  
       let
         getChunkCandidates = E.selectSource . E.from $ \fileContentChunkUnreferenced -> do
           E.where_ $ fileContentChunkUnreferenced E.^. FileContentChunkUnreferencedSince E.<. E.val (addUTCTime (-keep) now)
@@ -198,7 +199,12 @@ dispatchJobPruneUnreferencedFiles numIterations epoch iteration = JobHandlerAtom
         .| C.mapM deleteChunk
         .| C.fold
 
-  when (deletedChunks > 0 || deletedChunkSize > 0) $
+      return (deletedEntries, deletedChunks, deletedChunkSize)
+
+    fin (deletedEntries, deletedChunks, deletedChunkSize) = do
+      observeDeletedUnreferencedFiles deletedEntries
+      $logInfoS "PruneUnreferencedFiles" [st|Deleted #{tshow deletedEntries} long-unreferenced files|]
+      observeDeletedUnreferencedChunks deletedChunks deletedChunkSize
       $logInfoS "PruneUnreferencedFiles" [st|Deleted #{tshow deletedChunks} chunks (#{tshow deletedChunkSize} bytes)|]
 
 
@@ -245,9 +251,9 @@ dispatchJobInjectFiles = JobHandlerException . maybeT (return ()) $ do
               .| maybe (C.map id) (takeWhileTime . (/ 2)) interval
               .| transPipe (lift . runDB . setSerializable) (persistentTokenBucketTakeC' TokenBucketInjectFiles $ views _1 Minio.oiSize)
               .| C.mapM (lift . injectOrDelete)
+              .| C.mapM (\res@(Sum inj, Sum sz) -> res <$ observeRechunkedFiles inj sz)
               .| C.fold
 
-  when (injectedFiles > 0 || injectedSize > 0) $
   $logInfoS "InjectFiles" [st|Injected #{tshow injectedFiles} files from upload cache into database (#{tshow injectedSize} bytes)|]
 
 
@@ -258,7 +264,9 @@ data RechunkFileException
   deriving anyclass (Exception)
     
 dispatchJobRechunkFiles :: JobHandler UniWorX
-dispatchJobRechunkFiles = JobHandlerAtomic . hoist lift $ do
+dispatchJobRechunkFiles = JobHandlerAtomicWithFinalizer act fin
+  where
+    act = hoist lift $ do
       interval <- getsYesod $ view _appRechunkFiles
       let
         getEntryCandidates = E.selectSource . E.from $ \fileContentEntry -> E.distinctOnOrderBy [E.asc $ fileContentEntry E.^. FileContentEntryHash] $ do
@@ -291,5 +299,7 @@ dispatchJobRechunkFiles = JobHandlerAtomic . hoist lift $ do
         .| C.mapM (uncurry rechunkFile)
         .| C.fold
 
-  when (rechunkedFiles > 0 || rechunkedSize > 0) $
+      return (rechunkedFiles, rechunkedSize)
+    fin (rechunkedFiles, rechunkedSize) = do
+      observeRechunkedFiles rechunkedFiles rechunkedSize
       $logInfoS "RechunkFiles" [st|Rechunked #{tshow rechunkedFiles} files in database (#{tshow rechunkedSize} bytes)|]

src/Jobs/Handler/PersonalisedSheetFiles.hs

@@ -8,8 +8,10 @@ import Database.Persist.Sql (deleteWhereCount)
 
 
 dispatchJobPruneFallbackPersonalisedSheetFilesKeys :: JobHandler UniWorX
-dispatchJobPruneFallbackPersonalisedSheetFilesKeys = JobHandlerAtomic . hoist lift $ do
+dispatchJobPruneFallbackPersonalisedSheetFilesKeys = JobHandlerAtomicWithFinalizer act fin
+  where
+    act = hoist lift $ do
       now <- liftIO getCurrentTime
       expires <- getsYesod $ view _appFallbackPersonalisedSheetFilesKeysExpire
-  n <- deleteWhereCount [ FallbackPersonalisedSheetFilesKeyGenerated <. addUTCTime (- expires) now ]
+      deleteWhereCount [ FallbackPersonalisedSheetFilesKeyGenerated <. addUTCTime (- expires) now ]
-  $logInfoS "PruneFallbackPersonalisedSheetFilesKeys" [st|Deleted #{n} expired fallback personalised sheet files keys|]
+    fin n = $logInfoS "PruneFallbackPersonalisedSheetFilesKeys" [st|Deleted #{n} expired fallback personalised sheet files keys|]

src/Jobs/Handler/PruneInvitations.hs

@@ -7,7 +7,9 @@ import Import
 import Database.Persist.Sql (deleteWhereCount)
 
 dispatchJobPruneInvitations :: JobHandler UniWorX
-dispatchJobPruneInvitations = JobHandlerAtomic . hoist lift $ do
+dispatchJobPruneInvitations = JobHandlerAtomicWithFinalizer act fin
+  where
+    act = hoist lift $ do
       now <- liftIO getCurrentTime
-  n <- deleteWhereCount [ InvitationExpiresAt <. Just now ]
+      deleteWhereCount [ InvitationExpiresAt <. Just now ]
-  $logInfoS "PruneInvitations" [st|Deleted #{n} expired invitations|]
+    fin n = $logInfoS "PruneInvitations" [st|Deleted #{n} expired invitations|]

src/Jobs/Handler/TransactionLog.hs

@@ -9,7 +9,9 @@ import Handler.Utils.DateTime
 import Database.Persist.Sql (updateWhereCount, deleteWhereCount)
 
 dispatchJobTruncateTransactionLog, dispatchJobDeleteTransactionLogIPs :: JobHandler UniWorX
-dispatchJobTruncateTransactionLog = JobHandlerAtomic . hoist lift $ do
+dispatchJobTruncateTransactionLog = JobHandlerAtomicWithFinalizer act fin
+  where
+    act = hoist lift $ do
       now <- liftIO getCurrentTime
       let localNow = utcToLocalTime now
           (localCurrentYear, _, _) = toGregorian $ localDay localNow
@@ -20,12 +22,14 @@ dispatchJobTruncateTransactionLog = JobHandlerAtomic . hoist lift $ do
             LTUUnique utc' _ -> utc'
             _other           -> startOfPreviousYear
 
-  n <- deleteWhereCount [ TransactionLogTime <. startOfPreviousYear' ]
+      deleteWhereCount [ TransactionLogTime <. startOfPreviousYear' ]
-  $logInfoS "TruncateTransactionLog" [st|Deleted #{n} transaction log entries|]
+    fin n = $logInfoS "TruncateTransactionLog" [st|Deleted #{n} transaction log entries|]
-dispatchJobDeleteTransactionLogIPs = JobHandlerAtomic . hoist lift $ do
+dispatchJobDeleteTransactionLogIPs = JobHandlerAtomicWithFinalizer act fin
+  where
+    act = hoist lift $ do
       now <- liftIO getCurrentTime
       retentionTime <- getsYesod $ view _appTransactionLogIPRetentionTime
       let cutoff = addUTCTime (- retentionTime) now
 
-  n <- updateWhereCount [ TransactionLogTime <. cutoff, TransactionLogRemote !=. Nothing ] [ TransactionLogRemote =. Nothing ]
+      updateWhereCount [ TransactionLogTime <. cutoff, TransactionLogRemote !=. Nothing ] [ TransactionLogRemote =. Nothing ]
-  $logInfoS "DeleteTransactionLogIPs" [st|Deleted #{n} IP entries from transaction log|]
+    fin n = $logInfoS "DeleteTransactionLogIPs" [st|Deleted #{n} IP entries from transaction log|]

src/Jobs/Types.hs

@@ -11,11 +11,10 @@ module Jobs.Types
   , JobContext(..)
   , JobState(..)
   , jobWorkerNames
-  , JobWorkerState(..)
+  , JobWorkerState(..), _jobWorkerJobCtl, _jobWorkerJob
-  , withJobWorkerState
   , JobWorkerId
   , showWorkerId, newWorkerId
-  , JobQueue, jqInsert, jqDequeue
+  , JobQueue, jqInsert, jqDequeue, jqDepth
   , JobPriority(..), prioritiseJob
   , jobNoQueueSame
   , module Cron
@@ -38,10 +37,6 @@ import qualified Data.Set as Set
 import Data.PQueue.Prio.Max (MaxPQueue)
 import qualified Data.PQueue.Prio.Max as PQ
 
-import Utils.Metrics (withJobWorkerStateLbls)
-
-import qualified Prometheus (Label4)
-
 import Cron (CronNextMatch(..), _MatchAsap, _MatchAt, _MatchNone)
 
 
@@ -179,7 +174,8 @@ type YesodJobDB site = ReaderT (YesodPersistBackend site) (WriterT (Set QueuedJo
 data JobHandler site
   = JobHandlerAtomic    (YesodJobDB site ())
   | JobHandlerException (HandlerFor site ())
-  deriving (Generic, Typeable)
+  | forall a. JobHandlerAtomicWithFinalizer (YesodJobDB site a) (a -> HandlerFor site ())
+  deriving (Typeable)
 
 makePrisms ''JobHandler
 
@@ -199,20 +195,6 @@ deriveJSON defaultOptions
   , sumEncoding = TaggedObject "state" "data"
   } ''JobWorkerState
 
-classifyJobWorkerState :: JobWorkerId -> JobWorkerState -> Prometheus.Label4
-classifyJobWorkerState wId jws = (showWorkerId wId, tag, maybe "n/a" pack mJobCtl, maybe "n/a" pack mJob)
-  where
-    Aeson.Object obj = Aeson.toJSON jws
-    Aeson.String tag = obj HashMap.! "state"
-    mJobCtl = asum
-      [ classifyJobCtl <$> jws ^? _jobWorkerJobCtl
-      , "perform" <$ jws ^? _jobWorkerJob
-      ]
-    mJob = classifyJob <$> jws ^? _jobWorkerJob
-
-withJobWorkerState :: (MonadIO m, MonadMask m) => JobWorkerId -> JobWorkerState -> m a -> m a
-withJobWorkerState wId newSt = withJobWorkerStateLbls $ classifyJobWorkerState wId newSt
-
 
 newtype JobWorkerId = JobWorkerId { jobWorkerUnique :: Unique }
   deriving (Eq, Ord)
@@ -273,6 +255,9 @@ jqInsert job = force . over _JobQueue $ PQ.insertBehind (prioritiseJob job) job
 jqDequeue :: JobQueue -> Maybe (JobCtl, JobQueue)
 jqDequeue = fmap ((\r@(_, q) -> q `deepseq` r) . over _2 JobQueue) . PQ.maxView . getJobQueue
 
+jqDepth :: Integral n => JobQueue -> n
+jqDepth = fromIntegral . PQ.size . getJobQueue
+
 
 data JobState = JobState
   { jobWorkers :: Map (Async ()) (TVar JobQueue)

src/Utils/Metrics.hs

@@ -1,14 +1,18 @@
+{-# OPTIONS_GHC -fno-warn-incomplete-uni-patterns #-}
+
 module Utils.Metrics
   ( withHealthReportMetrics
   , registerGHCMetrics
   , observeHTTPRequestLatency
   , registerReadyMetric
-  , registerJobHeldLocksCount
+  , withJobWorkerState
-  , withJobWorkerStateLbls
   , observeYesodCacheSize
   , observeFavouritesQuickActionsDuration
   , LoginOutcome(..), observeLoginOutcome
+  , registerJobHeldLocksCount
   , FileChunkStorage(..), observeSourcedChunk, observeSunkChunk
+  , observeDeletedUnreferencedFiles, observeDeletedUnreferencedChunks, observeInjectedFiles, observeRechunkedFiles
+  , registerJobWorkerQueueDepth
   ) where
 
 import Import.NoModel hiding (Vector, Info)
@@ -29,6 +33,11 @@ import Yesod.Core.Types (HandlerData(..), GHState(..))
 
 import qualified Data.Set as Set
 
+import Jobs.Types
+
+import qualified Data.Aeson as Aeson
+import qualified Data.HashMap.Strict as HashMap
+
 {-# ANN module ("HLint: ignore Use even" :: String) #-}
 
 
@@ -130,15 +139,75 @@ jobHeldLocksCount heldLocks = Metric $ return (MkJobHeldLocksCount, collectJobHe
 sourcedFileChunkSizes :: Vector Label1 Histogram
 sourcedFileChunkSizes = unsafeRegister . vector "storage" $ histogram info buckets
   where info = Info "uni2work_sourced_file_chunks_bytes"
-                    "Sizes of files chunks sourced"
+                    "Sizes of file chunks sourced"
-        buckets = 0 : histogramBuckets 1 20000000
+        buckets = 0 : histogramBuckets 1 1000000000
 
 {-# NOINLINE sunkFileChunkSizes #-}
 sunkFileChunkSizes :: Vector Label1 Histogram
 sunkFileChunkSizes = unsafeRegister . vector "storage" $ histogram info buckets
   where info = Info "uni2work_sunk_file_chunks_bytes"
-                    "Sizes of files chunks sunk"
+                    "Sizes of file chunks sunk"
-        buckets = 0 : histogramBuckets 1 100000000
+        buckets = 0 : histogramBuckets 1 1000000000
+
+{-# NOINLINE deletedUnreferencedFiles #-}
+deletedUnreferencedFiles :: Counter
+deletedUnreferencedFiles = unsafeRegister $ counter info
+  where info = Info "uni2work_deleted_unreferenced_files_count"
+                    "Number of unreferenced files deleted"
+
+{-# NOINLINE deletedUnreferencedChunks #-}
+deletedUnreferencedChunks :: Counter
+deletedUnreferencedChunks = unsafeRegister $ counter info
+  where info = Info "uni2work_deleted_unreferenced_chunks_count"
+                    "Number of unreferenced chunks deleted"
+
+{-# NOINLINE deletedUnreferencedChunksBytes #-}
+deletedUnreferencedChunksBytes :: Counter
+deletedUnreferencedChunksBytes = unsafeRegister $ counter info
+  where info = Info "uni2work_deleted_unreferenced_chunks_bytes"
+                    "Size of unreferenced chunks deleted"
+
+{-# NOINLINE injectedFiles #-}
+injectedFiles :: Counter
+injectedFiles = unsafeRegister $ counter info
+  where info = Info "uni2work_injected_files_count"
+                    "Number of files injected from upload cache into database"
+
+{-# NOINLINE injectedFilesBytes #-}
+injectedFilesBytes :: Counter
+injectedFilesBytes = unsafeRegister $ counter info
+  where info = Info "uni2work_injected_files_bytes"
+                    "Size of files injected from upload cache into database"
+  
+{-# NOINLINE rechunkedFiles #-}
+rechunkedFiles :: Counter
+rechunkedFiles = unsafeRegister $ counter info
+  where info = Info "uni2work_rechunked_files_count"
+                    "Number of files rechunked within database"
+
+{-# NOINLINE rechunkedFilesBytes #-}
+rechunkedFilesBytes :: Counter
+rechunkedFilesBytes = unsafeRegister $ counter info
+  where info = Info "uni2work_rechunked_files_bytes"
+                    "Size of files rechunked within database"
+
+data JobWorkerQueueDepth = MkJobWorkerQueueDepth
+
+jobWorkerQueueDepth :: TMVar JobState -> Metric JobWorkerQueueDepth
+jobWorkerQueueDepth jSt = Metric $ return (MkJobWorkerQueueDepth, collectJobWorkerQueueDepth)
+  where
+    collectJobWorkerQueueDepth = maybeT (return []) $ do
+      wQueues <- hoist atomically $ do
+        JobState{..} <- MaybeT $ tryReadTMVar jSt
+        flip ifoldMapM jobWorkers $ \wAsync wQueue
+          -> lift $ pure . (jobWorkerName wAsync, ) . jqDepth <$> readTVar wQueue
+      return [ SampleGroup info GaugeType
+                 [ Sample "uni2work_queued_jobs_count" [("worker", showWorkerId wName)] . encodeUtf8 $ tshow wDepth
+                 | (wName, wDepth) <- wQueues
+                 ]
+             ]
+    info = Info "uni2work_queued_jobs_count"
+                "Number of JobQueue entries in this Uni2work-instance"
 
 withHealthReportMetrics :: MonadIO m => m HealthReport -> m HealthReport
 withHealthReportMetrics act = do
@@ -181,6 +250,20 @@ observeHTTPRequestLatency classifyHandler app req respond' = do
 registerReadyMetric :: MonadIO m => m ()
 registerReadyMetric = liftIO $ void . register . readyMetric =<< getPOSIXTime
 
+classifyJobWorkerState :: JobWorkerId -> JobWorkerState -> Prometheus.Label4
+classifyJobWorkerState wId jws = (showWorkerId wId, tag, maybe "n/a" pack mJobCtl, maybe "n/a" pack mJob)
+  where
+    Aeson.Object obj = Aeson.toJSON jws
+    Aeson.String tag = obj HashMap.! "state"
+    mJobCtl = asum
+      [ classifyJobCtl <$> jws ^? _jobWorkerJobCtl
+      , "perform" <$ jws ^? _jobWorkerJob
+      ]
+    mJob = classifyJob <$> jws ^? _jobWorkerJob
+
+withJobWorkerState :: (MonadIO m, MonadMask m) => JobWorkerId -> JobWorkerState -> m a -> m a
+withJobWorkerState wId newSt = withJobWorkerStateLbls $ classifyJobWorkerState wId newSt
+
 withJobWorkerStateLbls :: (MonadIO m, MonadMask m) => Label4 -> m a -> m a
 withJobWorkerStateLbls newLbls act = do
   liftIO $ withLabel jobWorkerStateTransitions newLbls incCounter
@@ -240,3 +323,24 @@ observeSourcedChunk store = liftIO . observeChunkSize sourcedFileChunkSizes stor
 
 observeChunkSize :: Vector Label1 Histogram -> FileChunkStorage -> Integer -> IO ()
 observeChunkSize metric (toPathPiece -> storageLabel) = withLabel metric storageLabel . flip observe . fromInteger
+
+observeDeletedUnreferencedFiles :: MonadIO m => Natural -> m ()
+observeDeletedUnreferencedFiles = liftIO . void . addCounter deletedUnreferencedFiles . fromIntegral
+
+observeDeletedUnreferencedChunks :: MonadIO m => Natural -> Word64 -> m ()
+observeDeletedUnreferencedChunks num size = liftIO $ do
+  void . addCounter deletedUnreferencedChunks $ fromIntegral num
+  void . addCounter deletedUnreferencedChunksBytes $ fromIntegral size
+
+observeInjectedFiles :: MonadIO m => Natural -> Word64 -> m ()
+observeInjectedFiles num size = liftIO $ do
+  void . addCounter injectedFiles $ fromIntegral num
+  void . addCounter injectedFilesBytes $ fromIntegral size
+
+observeRechunkedFiles :: MonadIO m => Natural -> Word64 -> m ()
+observeRechunkedFiles num size = liftIO $ do
+  void . addCounter rechunkedFiles $ fromIntegral num
+  void . addCounter rechunkedFilesBytes $ fromIntegral size
+
+registerJobWorkerQueueDepth :: MonadIO m => TMVar JobState -> m ()
+registerJobWorkerQueueDepth = liftIO . void . register . jobWorkerQueueDepth