locli: LogObjectSource for input from different backends; comments

Author: mgmeier

bench/locli/CHANGELOG.md

@@ -3,9 +3,11 @@
 ## 2.0 -- Dec 2024
 
 * New database (DB) persistence backend for log objects using serverless SQLite DBs
-* New CLI commands `prepare-db` and `unlog-db` to create and read from that persistence layer respectively
-* Tweak GC to mitigate high RAM requirements
-* New executable `locli-quick` which aims to be a development testbed for (upcoming) DB-backed quick queries.
+* Refactor current file persistence backend into its own module
+* New CLI commands `prepare-db` and `unlog-db` to create and read from DB persistence backend respectively
+* New sum type `LogObjectSource` to represent input from different backends (file or DB)
+* Tweak GC to mitigate high RAM requirements (for perf cluster analyses only)
+* New executable `locli-quick` which aims to be a development testbed for (upcoming) DB-backed quick queries
 
 ## 1.36 -- Nov 2024
 

bench/locli/src/Cardano/Analysis/API/Ground.hs

@@ -1,6 +1,7 @@
 {-# LANGUAGE DeriveAnyClass #-}
 {-# LANGUAGE DeriveDataTypeable #-}
 {-# LANGUAGE GeneralisedNewtypeDeriving #-}
+{-# LANGUAGE MultiWayIf #-}
 {-# LANGUAGE PolyKinds #-}
 {-# OPTIONS_GHC -Wno-orphans #-}
 module Cardano.Analysis.API.Ground
@@ -163,6 +164,32 @@ newtype OutputFile
   = OutputFile { unOutputFile :: FilePath }
   deriving (Show, Eq)
 
+data LogObjectSource =
+    LogObjectSourceJSON   JsonLogfile
+  | LogObjectSourceSQLite FilePath
+  | LogObjectSourceOther  FilePath
+  deriving (Show, Eq, Generic, NFData)
+
+logObjectSourceFile :: LogObjectSource -> FilePath
+logObjectSourceFile = \case
+  LogObjectSourceJSON j   -> unJsonLogfile j
+  LogObjectSourceSQLite f -> f
+  LogObjectSourceOther f  -> f
+
+toLogObjectSource :: FilePath -> LogObjectSource
+toLogObjectSource fp
+  | ext == ".sqlite" || ext == ".sqlite3" = LogObjectSourceSQLite fp
+  | ext == ".json"                        = LogObjectSourceJSON (JsonLogfile fp)
+  | otherwise                             = LogObjectSourceOther fp
+  where
+    ext = map toLower $ F.takeExtension fp
+
+instance FromJSON LogObjectSource where
+  parseJSON = withText "LogObjectSource" (pure . toLogObjectSource . T.unpack)
+
+instance ToJSON LogObjectSource where
+  toJSON = toJSON . logObjectSourceFile
+
 ---
 --- Orphans
 ---
@@ -210,6 +237,14 @@ optJsonLogfile optname desc =
       <> metavar "JSONLOGFILE"
       <> help desc
 
+optLogObjectSource :: String -> String -> Parser LogObjectSource
+optLogObjectSource optname desc =
+  fmap toLogObjectSource $
+    Opt.option Opt.str
+      $ long optname
+      <> metavar "JSONLOGFILE|SQLITE3LOGFILE"
+      <> help desc
+
 argJsonLogfile :: Parser JsonLogfile
 argJsonLogfile =
   JsonLogfile <$>
@@ -324,26 +359,26 @@ dumpObjects ident xs (JsonOutputFile f) = liftIO $ do
   withFile f WriteMode $ \hnd -> do
     forM_ xs $ LBS.hPutStrLn hnd . encode
 
-dumpAssociatedObjects :: ToJSON a => String -> [(JsonLogfile, a)] -> ExceptT Text IO ()
+dumpAssociatedObjects :: ToJSON a => String -> [(LogObjectSource, a)] -> ExceptT Text IO ()
 dumpAssociatedObjects ident xs = liftIO $
   flip mapConcurrently_ xs $
-    \(JsonLogfile f, x) ->
+    \(logObjectSourceFile -> f, x) ->
       withFile (replaceExtension f $ ident <> ".json") WriteMode $ \hnd ->
         LBS.hPutStrLn hnd $ encode x
 
 readAssociatedObjects :: forall a.
-  FromJSON a => String -> [JsonLogfile] -> ExceptT Text IO [(JsonLogfile, a)]
+  FromJSON a => String -> [JsonLogfile] -> ExceptT Text IO [(LogObjectSource, a)]
 readAssociatedObjects ident fs = firstExceptT T.pack . newExceptT . fmap (mapM sequence) $
   flip mapConcurrently fs $
     \jf@(JsonLogfile f) -> do
         x <- eitherDecode @a <$> LBS.readFile (replaceExtension f $ ident <> ".json")
         progress ident (Q f)
-        pure (jf, x)
+        pure (LogObjectSourceJSON jf, x)
 
-dumpAssociatedObjectStreams :: ToJSON a => String -> [(JsonLogfile, [a])] -> ExceptT Text IO ()
+dumpAssociatedObjectStreams :: ToJSON a => String -> [(LogObjectSource, [a])] -> ExceptT Text IO ()
 dumpAssociatedObjectStreams ident xss = liftIO $
   flip mapConcurrently_ xss $
-    \(JsonLogfile f, xs) -> do
+    \(logObjectSourceFile -> f, xs) -> do
       withFile (replaceExtension f $ ident <> ".json") WriteMode $ \hnd -> do
         forM_ xs $ LBS.hPutStrLn hnd . encode
 
@@ -353,9 +388,9 @@ dumpText ident xs (TextOutputFile f) = liftIO $ do
   withFile f WriteMode $ \hnd -> do
     forM_ xs $ hPutStrLn hnd
 
-dumpAssociatedTextStreams :: String -> [(JsonLogfile, [Text])] -> ExceptT Text IO ()
+dumpAssociatedTextStreams :: String -> [(LogObjectSource, [Text])] -> ExceptT Text IO ()
 dumpAssociatedTextStreams ident xss = liftIO $
   flip mapConcurrently_ xss $
-    \(JsonLogfile f, xs) -> do
+    \(logObjectSourceFile -> f, xs) -> do
       withFile (replaceExtension f $ ident <> ".txt") WriteMode $ \hnd -> do
         forM_ xs $ hPutStrLn hnd

bench/locli/src/Cardano/Analysis/BlockProp.hs

@@ -7,7 +7,7 @@
 #endif
 
 {-# OPTIONS_GHC -Wno-incomplete-patterns -Wno-name-shadowing #-}
-{-# OPTIONS_GHC -Wno-unused-imports -Wno-partial-fields -Wno-unused-matches -Wno-deprecations -Wno-unused-local-binds -Wno-incomplete-record-updates #-}
+{-# OPTIONS_GHC -Wno-unused-imports -Wno-partial-fields -Wno-unused-matches -Wno-incomplete-record-updates #-}
 
 {- HLINT ignore "Avoid lambda" -}
 {- HLINT ignore "Eta reduce" -}
@@ -293,13 +293,13 @@ beForgedAt :: BlockEvents -> UTCTime
 beForgedAt BlockEvents{beForge=BlockForge{..}} =
   bfForged `afterSlot` bfSlotStart
 
-buildMachViews :: Run -> [(JsonLogfile, [LogObject])] -> IO [(JsonLogfile, MachView)]
+buildMachViews :: Run -> [(LogObjectSource, [LogObject])] -> IO [(LogObjectSource, MachView)]
 buildMachViews run = mapConcurrentlyPure (fst &&& blockEventMapsFromLogObjects run)
 
 blockEventsAcceptance :: Genesis -> [ChainFilter] -> BlockEvents -> [(ChainFilter, Bool)]
 blockEventsAcceptance genesis flts be = flts <&> (id &&& testBlockEvents genesis be)
 
-rebuildChain :: Run -> [ChainFilter] -> [FilterName] -> [(JsonLogfile, MachView)] -> Chain
+rebuildChain :: Run -> [ChainFilter] -> [FilterName] -> [(LogObjectSource, MachView)] -> Chain
 rebuildChain run@Run{genesis} flts fltNames xs@(fmap snd -> machViews) =
   Chain
   { cDomSlots   = DataDomain
@@ -320,8 +320,8 @@ rebuildChain run@Run{genesis} flts fltNames xs@(fmap snd -> machViews) =
                 doRebuildChain (fmap deltifyEvents <$> eventMaps) tipHash
    (accepta, cRejecta) = partition (all snd . beAcceptance) cMainChain
 
-   blkSets :: (Set Hash, Set Hash)
-   blkSets@(acceptaBlocks, rejectaBlocks) =
+   acceptaBlocks, rejectaBlocks :: Set Hash
+   (acceptaBlocks, rejectaBlocks) =
      both (Set.fromList . fmap beBlock) (accepta, cRejecta)
    mvBlockStats :: MachView -> HostBlockStats
    mvBlockStats (fmap bfeBlock . mvForges -> fs) = HostBlockStats {..}
@@ -346,7 +346,7 @@ rebuildChain run@Run{genesis} flts fltNames xs@(fmap snd -> machViews) =
    finalBlockNo   = mbeBlockNo finalBlockEv
 
    tipHash        = rewindChain eventMaps finalBlockNo 1 (mbeBlock finalBlockEv)
-   tipBlock       = getBlockForge eventMaps finalBlockNo tipHash
+   _tipBlock      = getBlockForge eventMaps finalBlockNo tipHash
 
    computeChainBlockGaps :: [BlockEvents] -> [BlockEvents]
    computeChainBlockGaps [] = error "computeChainBlockGaps on an empty chain"
@@ -376,11 +376,12 @@ rebuildChain run@Run{genesis} flts fltNames xs@(fmap snd -> machViews) =
          ])
      & mapMbe id (error "Silly invariant failed.") (error "Silly invariant failed.")
 
-   adoptionMap    :: [Map Hash UTCTime]
-   adoptionMap    =  Map.mapMaybe (lazySMaybe . mbeAdopted) <$> eventMaps
+   adoptionMap  :: [Map Hash UTCTime]
+   adoptionMap  =  Map.mapMaybe (lazySMaybe . mbeAdopted) <$> eventMaps
 
-   heightHostMap      :: (Map BlockNo (Set Hash), Map Host (Set Hash))
-   heightHostMap@(heightMap, hostMap)
+   heightMap    :: Map BlockNo (Set Hash)
+   _hostMap     :: Map Host (Set Hash)
+   (heightMap, _hostMap)
      = foldr (\MachView{..} (accHeight, accHost) ->
                  (,)
                  (Map.foldr
@@ -589,11 +590,6 @@ blockProp run@Run{genesis} Chain{..} = do
                                                 & filter (not . isNaN))
     }
  where
-   ne :: String -> [a] -> [a]
-   ne desc = \case
-     [] -> error desc
-     xs -> xs
-
    hostBlockStats = Map.elems cHostBlockStats
 
    boFetchedCum :: BlockObservation -> NominalDiffTime
@@ -629,10 +625,10 @@ blockProp run@Run{genesis} Chain{..} = do
      cdfZ percs $ concatMap f cbes
 
 -- | Given a single machine's log object stream, recover its block map.
-blockEventMapsFromLogObjects :: Run -> (JsonLogfile, [LogObject]) -> MachView
-blockEventMapsFromLogObjects run (f@(unJsonLogfile -> fp), []) =
-  error $ mconcat ["0 LogObjects in ", fp]
-blockEventMapsFromLogObjects run (f@(unJsonLogfile -> fp), xs@(x:_)) =
+blockEventMapsFromLogObjects :: Run -> (LogObjectSource, [LogObject]) -> MachView
+blockEventMapsFromLogObjects run (f, []) =
+  error $ mconcat ["0 LogObjects in ", logObjectSourceFile f]
+blockEventMapsFromLogObjects run (f, xs@(x:_)) =
   foldl' (blockPropMachEventsStep run f) initial xs
  where
    initial =
@@ -648,8 +644,8 @@ blockEventMapsFromLogObjects run (f@(unJsonLogfile -> fp), xs@(x:_)) =
      , mvMemSnap      = SNothing
      }
 
-blockPropMachEventsStep :: Run -> JsonLogfile -> MachView -> LogObject -> MachView
-blockPropMachEventsStep run@Run{genesis} (JsonLogfile fp) mv@MachView{..} lo = case lo of
+blockPropMachEventsStep :: Run -> LogObjectSource -> MachView -> LogObject -> MachView
+blockPropMachEventsStep run@Run{genesis} _ mv@MachView{..} lo = case lo of
   -- 0. Notice (observer only)
   LogObject{loAt, loHost, loBody=LOChainSyncClientSeenHeader{loBlock,loBlockNo,loSlotNo}} ->
     let mbe0 = getBlock loBlock

bench/locli/src/Cardano/Analysis/MachPerf.hs

@@ -6,7 +6,7 @@
 {-# OPTIONS_GHC -Wno-x-partial #-}
 #endif
 
-{-# OPTIONS_GHC -Wno-incomplete-patterns -Wno-name-shadowing -Wno-orphans #-}
+{-# OPTIONS_GHC -Wno-incomplete-patterns -Wno-name-shadowing #-}
 {- HLINT ignore "Use head" -}
 {- HLINT ignore "Evaluate" -}
 
@@ -33,15 +33,15 @@ import Cardano.Unlog.Resources
 
 -- * 1. Collect SlotStats & RunScalars:
 --
-collectSlotStats :: Run -> [(JsonLogfile, [LogObject])]
-                 -> IO (Either Text [(JsonLogfile, (RunScalars, [SlotStats UTCTime]))])
+collectSlotStats :: Run -> [(LogObjectSource, [LogObject])]
+                 -> IO (Either Text [(LogObjectSource, (RunScalars, [SlotStats UTCTime]))])
 collectSlotStats run = fmap sequence <$> mapConcurrentlyPure (timelineFromLogObjects run)
 
 
-timelineFromLogObjects :: Run -> (JsonLogfile, [LogObject])
-                       -> Either Text (JsonLogfile, (RunScalars, [SlotStats UTCTime]))
-timelineFromLogObjects _ (JsonLogfile f, []) =
-  Left $ "timelineFromLogObjects:  zero logobjects from " <> pack f
+timelineFromLogObjects :: Run -> (LogObjectSource, [LogObject])
+                       -> Either Text (LogObjectSource, (RunScalars, [SlotStats UTCTime]))
+timelineFromLogObjects _ (f, []) =
+  Left $ "timelineFromLogObjects:  zero logobjects from " <> pack (logObjectSourceFile f)
 timelineFromLogObjects run@Run{genesis} (f, xs') =
   Right . (f,)
   $ foldl' (timelineStep run f) zeroTimelineAccum xs
@@ -107,7 +107,7 @@ timelineFromLogObjects run@Run{genesis} (f, xs') =
      , slLogObjects  = []
      }
 
-timelineStep :: Run -> JsonLogfile -> TimelineAccum -> LogObject -> TimelineAccum
+timelineStep :: Run -> LogObjectSource -> TimelineAccum -> LogObject -> TimelineAccum
 timelineStep Run{genesis} f accum@TimelineAccum{aSlotStats=cur:_, ..} lo =
   -- 1. skip pre-historic events not subject to performance analysis;
   --    Potentially _collapsingly huge_, depending on what portion of logs you get.
@@ -152,7 +152,7 @@ timelineStep Run{genesis} f accum@TimelineAccum{aSlotStats=cur:_, ..} lo =
              [ desc, " for a future slot=", show slot
              , " cur=", show (slSlot cur)
              , " host=", unpack . toText $ unHost host
-             , " file=", unJsonLogfile f
+             , " file=", logObjectSourceFile f
              ]
         else forExistingSlot slot acc x
   in
@@ -467,14 +467,14 @@ runSlotFilters ::
      NFData a =>
      Run
   -> [ChainFilter]
-  -> [(JsonLogfile, [SlotStats a])]
-  -> IO (DataDomain I SlotNo, [(JsonLogfile, [SlotStats a])])
+  -> [(LogObjectSource, [SlotStats a])]
+  -> IO (DataDomain I SlotNo, [(LogObjectSource, [SlotStats a])])
 runSlotFilters Run{genesis} flts slots =
   mapConcurrentlyPure (fmap $ filterSlotStats flts) slots
     <&> \filtered ->
           (,) (domain filtered) filtered
  where
-   domain :: [(JsonLogfile, [SlotStats a])] -> DataDomain I SlotNo
+   domain :: [(LogObjectSource, [SlotStats a])] -> DataDomain I SlotNo
    domain filtered = mkDataDomain
      ((CP.head samplePre  <&> slSlot) & fromMaybe 0)
      ((lastMay samplePre  <&> slSlot) & fromMaybe 0)
@@ -567,9 +567,9 @@ slotStatsSummary Run{genesis=Genesis{epochLength}} slots =
 
 -- * 4. Summarise SlotStats & SlotStatsSummary into MachPerf:
 --
-slotStatsMachPerf :: Run -> (JsonLogfile, [SlotStats NominalDiffTime]) -> Either Text (JsonLogfile, MachPerfOne)
-slotStatsMachPerf _ (JsonLogfile f, []) =
-  Left $ "slotStatsMachPerf:  zero filtered slots from " <> pack f
+slotStatsMachPerf :: Run -> (LogObjectSource, [SlotStats NominalDiffTime]) -> Either Text (LogObjectSource, MachPerfOne)
+slotStatsMachPerf _ (f, []) =
+  Left $ "slotStatsMachPerf:  zero filtered slots from " <> pack (logObjectSourceFile f)
 slotStatsMachPerf run (f, slots) =
   Right . (f,) $ MachPerf
   { mpVersion            = getLocliVersion

bench/locli/src/Cardano/Command.hs

@@ -68,7 +68,7 @@ data ChainCommand
   |            ReadChain    (JsonInputFile [BlockEvents])
   |        TimelineChain    RenderConfig TextOutputFile [TimelineComments BlockEvents]
 
-  |         CollectSlots    [JsonLogfile]
+  |         CollectSlots    [LogObjectSource]
   |            DumpSlotsRaw
   |          FilterSlots    [JsonFilterFile] [ChainFilter]
   |            DumpSlots
@@ -177,7 +177,7 @@ parseChainCommand =
    , op "collect-slots" "Collect per-slot performance stats"
      (CollectSlots
        <$> many
-           (optJsonLogfile    "ignore-log"   "Omit data from listed log files from perf statistics"))
+           (optLogObjectSource "ignore-log"   "Omit data from listed log sources from perf statistics"))
    , op "dump-slots-raw" "Dump unfiltered slot stats JSON streams, alongside input files"
      (DumpSlotsRaw & pure)
    , op "filter-slots" "Filter per-slot performance stats"
@@ -342,15 +342,15 @@ data State
   , sRunLogs          :: Maybe (RunLogs [LogObject])
   , sDomSlots         :: Maybe (DataDomain I SlotNo)
     -- propagation
-  , sMachViews        :: Maybe [(JsonLogfile, MachView)]
+  , sMachViews        :: Maybe [(LogObjectSource, MachView)]
   , sChain            :: Maybe Chain
   , sBlockProp        :: Maybe [BlockPropOne]
   , sMultiBlockProp   :: Maybe MultiBlockProp
     -- performance
-  , sSlotsRaw         :: Maybe [(JsonLogfile, [SlotStats NominalDiffTime])]
-  , sScalars          :: Maybe [(JsonLogfile, RunScalars)]
-  , sSlots            :: Maybe [(JsonLogfile, [SlotStats NominalDiffTime])]
-  , sMachPerf         :: Maybe [(JsonLogfile, MachPerfOne)]
+  , sSlotsRaw         :: Maybe [(LogObjectSource, [SlotStats NominalDiffTime])]
+  , sScalars          :: Maybe [(LogObjectSource, RunScalars)]
+  , sSlots            :: Maybe [(LogObjectSource, [SlotStats NominalDiffTime])]
+  , sMachPerf         :: Maybe [(LogObjectSource, MachPerfOne)]
   , sClusterPerf      :: Maybe [ClusterPerf]
   , sMultiClusterPerf :: Maybe MultiClusterPerf
     --
@@ -553,7 +553,7 @@ runChainCommand s@State{sRun=Just run, sRunLogs=Just (rlLogs -> objs)}
   c@(CollectSlots ignores) = do
   let nonIgnored = flip filter objs $ (`notElem` ignores) . fst
   forM_ ignores $
-    progress "perf-ignored-log" . R . unJsonLogfile
+    progress "perf-ignored-log" . R . logObjectSourceFile
   progress "slots" (Q $ printf "building %d slot timelines" $ length objs)
   performGC
   (scalars, slotsRaw) <-