diff --git a/CHANGELOG.md b/CHANGELOG.md new file mode 100644 --- /dev/null +++ b/CHANGELOG.md @@ -0,0 +1,3 @@ +### 0.1.0.0 + +- Initial release. diff --git a/LICENSE b/LICENSE new file mode 100644 --- /dev/null +++ b/LICENSE @@ -0,0 +1,30 @@ +Copyright (c) 2021 Well-Typed + +All rights reserved. + +Redistribution and use in source and binary forms, with or without +modification, are permitted provided that the following conditions are met: + + * Redistributions of source code must retain the above copyright + notice, this list of conditions and the following disclaimer. + + * Redistributions in binary form must reproduce the above + copyright notice, this list of conditions and the following + disclaimer in the documentation and/or other materials provided + with the distribution. + + * Neither the name of Oleg Grenrus nor the names of other + contributors may be used to endorse or promote products derived + from this software without specific prior written permission. + +THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS +"AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT +LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR +A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT +OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, +SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT +LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, +DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY +THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT +(INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE +OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. diff --git a/app/Main.hs b/app/Main.hs new file mode 100644 --- /dev/null +++ b/app/Main.hs @@ -0,0 +1,8 @@ +{-# LANGUAGE NoImplicitPrelude #-} +{-# OPTIONS_GHC -Wno-missing-signatures #-} + +module Main where + +import GHC.Eventlog.Live.InfluxDB qualified (main) + +main = GHC.Eventlog.Live.InfluxDB.main diff --git a/eventlog-live-influxdb.cabal b/eventlog-live-influxdb.cabal new file mode 100644 --- /dev/null +++ b/eventlog-live-influxdb.cabal @@ -0,0 +1,127 @@ +cabal-version: 3.12 +name: eventlog-live-influxdb +version: 0.1.0.0 +synopsis: Stream eventlog data into InfluxDB. +description: + This executable supports live streaming of eventlog data into + [InfluxDB](https://www.influxdata.com) version 1.8. + + > Usage: eventlog-live-influxdb --eventlog-socket SOCKET + > [--eventlog-socket-timeout NUM] + > [--eventlog-socket-exponent NUM] + > [--batch-interval NUM] [--eventlog-log-file FILE] + > [-h Tcmdyrbi] + > [-v|--verbosity NUM|quiet|error|warning] + > [--influxdb-username USERNAME + > --influxdb-password PASSWORD] + > [--influxdb-retention-policy RETENTION_POLICY] + > [--influxdb-host HOST] [--influxdb-port HOST] + > [--influxdb-ssl] --influxdb-database DATABASE + > + > Available options: + > --eventlog-socket SOCKET Eventlog Unix socket. + > --eventlog-socket-timeout NUM + > Eventlog socket connection retry timeout in + > microseconds. + > --eventlog-socket-exponent NUM + > Eventlog socket connection retry timeout exponent. + > --batch-interval NUM Batch interval in microseconds. + > --eventlog-log-file FILE Use file to log binary eventlog data. + > -h Tcmdyrbi Heap profile breakdown. + > -v,--verbosity NUM|quiet|error|warning + > The verbosity threshold for logging. + > --influxdb-username USERNAME + > InfluxDB username + > --influxdb-password PASSWORD + > InfluxDB password + > --influxdb-retention-policy RETENTION_POLICY + > InfluxDB retention policy + > --influxdb-host HOST InfluxDB server host + > --influxdb-port HOST InfluxDB server host + > --influxdb-ssl InfluxDB server SSL + > --influxdb-database DATABASE + > InfluxDB database name + > --help Show this help text. + > --version Show version information + +license: AGPL-3.0-only +license-file: LICENSE +author: Wen Kokke +maintainer: wen@well-typed.com +copyright: (c) 2025 Well-Typed +build-type: Simple +category: Debug, Monitoring, System +extra-doc-files: CHANGELOG.md +tested-with: + GHC ==9.2.8 + || ==9.4.8 + || ==9.6.7 + || ==9.8.4 + || ==9.10.2 + || ==9.12.2 + +source-repository head + type: git + location: https://github.com/well-typed/eventlog-live.git + subdir: eventlog-live-influxdb + +common language + ghc-options: + -Wall -Wcompat -Widentities -Wprepositive-qualified-module + -Wredundant-constraints -Wunticked-promoted-constructors + -Wunused-packages + + default-language: Haskell2010 + default-extensions: + BangPatterns + DataKinds + DeriveFoldable + DeriveFunctor + DeriveTraversable + DerivingStrategies + DuplicateRecordFields + FlexibleContexts + FlexibleInstances + GADTs + GeneralizedNewtypeDeriving + ImportQualifiedPost + InstanceSigs + KindSignatures + LambdaCase + MultiParamTypeClasses + NoFieldSelectors + NumericUnderscores + OverloadedRecordDot + RankNTypes + RecordWildCards + ScopedTypeVariables + TupleSections + TypeApplications + TypeFamilies + +library + import: language + hs-source-dirs: src + exposed-modules: GHC.Eventlog.Live.InfluxDB + other-modules: PackageInfo_eventlog_live_influxdb + autogen-modules: PackageInfo_eventlog_live_influxdb + build-depends: + , base >=4.16 && <4.22 + , clock >=0.8.4 && <0.9 + , containers >=0.6 && <0.8 + , dlist >=1.0 && <1.1 + , eventlog-live >=0.1 && <0.2 + , eventlog-live:options + , eventlog-live:socket + , ghc-events >=0.20 && <0.21 + , influxdb >=1.9.3 && <1.10 + , lens-family >=2.1.3 && <2.2 + , machines >=0.7.4 && <0.8 + , optparse-applicative >=0.19 && <0.20 + , text >=1.2 && <2.2 + +executable eventlog-live-influxdb + import: language + main-is: Main.hs + hs-source-dirs: app + build-depends: eventlog-live-influxdb diff --git a/src/GHC/Eventlog/Live/InfluxDB.hs b/src/GHC/Eventlog/Live/InfluxDB.hs new file mode 100644 --- /dev/null +++ b/src/GHC/Eventlog/Live/InfluxDB.hs @@ -0,0 +1,413 @@ +{-# LANGUAGE OverloadedStrings #-} + +{- | +Module : GHC.Eventlog.Live.InfluxDB +Description : The implementation of @eventlog-live-influxdb@. +Stability : experimental +Portability : portable +-} +module GHC.Eventlog.Live.InfluxDB ( + main, +) where + +import Control.Applicative (Alternative ((<|>))) +import Control.Monad (unless) +import Control.Monad.IO.Class (MonadIO (..)) +import Data.DList (DList) +import Data.DList qualified as D +import Data.Machine (asParts, mapping, (<~)) +import Data.Machine.Fanout (fanout) +import Data.Machine.Plan (await) +import Data.Machine.Process (ProcessT, (~>)) +import Data.Machine.Type (repeatedly) +import Data.Map.Strict qualified as M +import Data.Maybe (catMaybes, mapMaybe) +import Data.String (IsString (..)) +import Data.Text (Text) +import Data.Text qualified as T +import Data.Version (showVersion) +import Data.Void (Void) +import Data.Word (Word32, Word64) +import Database.InfluxDB.Line qualified as I (Line (..)) +import Database.InfluxDB.Types qualified as I +import Database.InfluxDB.Write qualified as I +import GHC.Eventlog.Live.Machines +import GHC.Eventlog.Live.Options +import GHC.Eventlog.Live.Socket +import GHC.Eventlog.Live.Verbosity (Verbosity) +import GHC.RTS.Events (Event (..), HeapProfBreakdown (..)) +import Lens.Family2 (set, (^.)) +import Options.Applicative qualified as O +import Options.Applicative.Extra qualified as O (helperWith) +import PackageInfo_eventlog_live_influxdb qualified as EventlogLive +import System.Clock (TimeSpec, fromNanoSecs) + +-------------------------------------------------------------------------------- +-- Main function +-------------------------------------------------------------------------------- + +{- | +The main function for @eventlog-live-influxdb@. +-} +main :: IO () +main = do + Options{..} <- O.execParser optionsInfo + let toInfluxDB = + liftTick withStartTime + ~> sortByBatchTick (.value.evTime) + ~> liftTick + ( fanout + [ processThreadEvents verbosity + , processHeapEvents verbosity maybeHeapProfBreakdown + ] + ) + ~> batchByTick + ~> mapping D.toList + ~> influxDBWriter influxDBWriteParams + runWithEventlogSocket + eventlogSocket + eventlogSocketTimeout + eventlogSocketTimeoutExponent + batchInterval + Nothing -- chunk size (bytes) + maybeEventlogLogFile + toInfluxDB + +-------------------------------------------------------------------------------- +-- Thread events +-------------------------------------------------------------------------------- + +data OneOf a b c = A !a | B !b | C !c + +processThreadEvents :: + (MonadIO m) => + Verbosity -> + ProcessT m (WithStartTime Event) (DList (I.Line TimeSpec)) +processThreadEvents verbosity = + fanout + [ fanout + [ -- GCSpan + processGCSpans verbosity + ~> mapping (D.singleton . A) + , processThreadStateSpans' tryGetTimeUnixNano (.value) setWithStartTime'value verbosity + ~> fanout + [ -- MutatorSpan + asMutatorSpans' (.value) setWithStartTime'value + ~> mapping (D.singleton . B) + , -- ThreadStateSpan + mapping (D.singleton . C) + ] + ] + ~> asParts + ~> mapping repackCapabilityUsageSpanOrThreadStateSpan + ~> fanout + [ mapping leftToMaybe + ~> asParts + ~> fanout + [ -- CapabilityUsageMetric + processCapabilityUsageMetrics + ~> mapping (D.singleton . fromMetric "CapabilityUsage") + , -- CapabilityUsageSpan + mapping (D.singleton . fromSpan . (.value)) + ] + , -- ThreadStateSpan + mapping rightToMaybe + ~> asParts + ~> mapping (D.singleton . fromSpan . (.value)) + ] + , -- ThreadLabel + processThreadLabels + ~> mapping (D.singleton . fromThreadLabel) + ] + where + repackCapabilityUsageSpanOrThreadStateSpan = \case + A i -> Left $ fmap Left i + B i -> Left $ fmap Right i + C i -> Right i + +{- | +Internal helper. +Get the `Left` value, if any. +-} +leftToMaybe :: Either a b -> Maybe a +leftToMaybe = either Just (const Nothing) + +{- | +Internal helper. +Get the `Right` value, if any. +-} +rightToMaybe :: Either a b -> Maybe b +rightToMaybe = either (const Nothing) Just + +-------------------------------------------------------------------------------- +-- Heap events +-------------------------------------------------------------------------------- + +processHeapEvents :: + (MonadIO m) => + Verbosity -> + Maybe HeapProfBreakdown -> + ProcessT m (WithStartTime Event) (DList (I.Line TimeSpec)) +processHeapEvents verbosity maybeHeapProfBreakdown = + fanout + [ mapping (D.singleton . fromMetric "HeapAllocated") + <~ processHeapAllocatedData + , mapping (D.singleton . fromMetric "HeapSize") + <~ processHeapSizeData + , mapping (D.singleton . fromMetric "BlocksSize") + <~ processBlocksSizeData + , mapping (D.singleton . fromMetric "HeapLive") + <~ processHeapLiveData + , mapping + ( \i -> + D.fromList + [ fromMetric "MemCurrent" ((.current) <$> i) + , fromMetric "MemNeeded" ((.needed) <$> i) + , fromMetric "MemReturned" ((.returned) <$> i) + ] + ) + <~ processMemReturnData + , mapping (D.singleton . fromMetric "HeapProfSample") + <~ processHeapProfSampleData verbosity maybeHeapProfBreakdown + ] + +-------------------------------------------------------------------------------- +-- Interpreting metadata +-------------------------------------------------------------------------------- + +-------------------------------------------------------------------------------- +-- Interpreting spans + +class FromSpan v where + fromSpan :: v -> I.Line TimeSpec + +toTag :: (Show a) => a -> I.Key +toTag = I.Key . T.pack . show + +-------------------------------------------------------------------------------- +-- Interpret capability usage spans + +instance FromSpan CapabilityUsageSpan where + fromSpan :: CapabilityUsageSpan -> I.Line TimeSpec + fromSpan i = + I.Line "CapabilityUsageSpan" tagSet fieldSet timestamp + where + tagSet = + M.fromList + [ ("capability", toTag i.cap) + ] + fieldSet = + M.fromList + [ ("duration", toField $ duration i) + , ("category", toField $ showCapabilityUserCategory user) + , ("user", toField $ show user) + ] + where + user = capabilityUser i + timestamp = Just . fromNanoSecs . toInteger $ i.startTimeUnixNano + +-------------------------------------------------------------------------------- +-- Interpret thread state spans + +instance FromSpan ThreadStateSpan where + fromSpan :: ThreadStateSpan -> I.Line TimeSpec + fromSpan i = + I.Line "ThreadStateSpan" tagSet fieldSet timestamp + where + tagSet = + M.fromList + [ ("thread", toTag i.thread) + ] + fieldSet = + M.fromList . catMaybes $ + [ Just ("duration", toField $ duration i) + , Just ("category", toField $ showThreadStateCategory i.threadState) + , ("capability",) . toField <$> threadStateCap i.threadState + , ("status",) . toField . show <$> threadStateStatus i.threadState + ] + timestamp = Just . fromNanoSecs . toInteger $ i.startTimeUnixNano + +-------------------------------------------------------------------------------- +-- Interpreting attributes + +class IsField v where + toField :: v -> I.Field 'I.NonNullable + +instance IsField String where + toField :: String -> I.Field 'I.NonNullable + toField = toField . T.pack + +instance IsField Text where + toField :: Text -> I.Field 'I.NonNullable + toField = I.FieldString + +instance IsField Int where + toField :: Int -> I.Field 'I.NonNullable + toField = I.FieldInt . fromIntegral + +instance IsField Double where + toField :: Double -> I.Field 'I.NonNullable + toField = I.FieldFloat + +instance IsField Word32 where + toField :: Word32 -> I.Field 'I.NonNullable + toField = I.FieldInt . fromIntegral + +-- | __Warning__: This instance may cause overflow. +instance IsField Word64 where + toField :: Word64 -> I.Field 'I.NonNullable + toField = I.FieldInt . fromIntegral + +fromThreadLabel :: ThreadLabel -> I.Line TimeSpec +fromThreadLabel i = + I.Line "ThreadLabel" tagSet fieldSet timestamp + where + thread = I.Key . T.pack . show $ i.thread + tagSet = M.singleton "thread" thread + fieldSet = M.singleton "label" (toField . show $ i.threadlabel) + timestamp = Just . fromNanoSecs . toInteger $ i.startTimeUnixNano + +fromMetric :: (IsField v) => I.Measurement -> Metric v -> I.Line TimeSpec +fromMetric measurement@(I.Measurement measurementName) i = + I.Line measurement tagSet fieldSet timestamp + where + tagSet = M.fromList (mapMaybe (\(k, v) -> (I.Key k,) <$> fromAttrValue v) i.attr) + fieldSet = M.singleton (I.Key measurementName) (toField i.value) + timestamp = fromNanoSecs . toInteger <$> i.maybeTimeUnixNano + +fromAttrValue :: AttrValue -> Maybe I.Key +fromAttrValue = \case + AttrInt v -> Just . fromString . show $ v + AttrInt8 v -> Just . fromString . show $ v + AttrInt16 v -> Just . fromString . show $ v + AttrInt32 v -> Just . fromString . show $ v + AttrInt64 v -> Just . fromString . show $ v + AttrWord v -> Just . fromString . show $ v + AttrWord8 v -> Just . fromString . show $ v + AttrWord16 v -> Just . fromString . show $ v + AttrWord32 v -> Just . fromString . show $ v + AttrWord64 v -> Just . fromString . show $ v + AttrDouble v -> Just . fromString . show $ v + AttrText v -> Just . I.Key $ v + AttrNull -> Nothing + +-------------------------------------------------------------------------------- +-- InfluxDB Batch Writer +-------------------------------------------------------------------------------- + +influxDBWriter :: I.WriteParams -> ProcessT IO [I.Line TimeSpec] Void +influxDBWriter writeParams = repeatedly go + where + go = + await >>= \batch -> do + unless (null batch) $ do + liftIO (I.writeBatch writeParams batch) + +-------------------------------------------------------------------------------- +-- Options +-------------------------------------------------------------------------------- + +optionsInfo :: O.ParserInfo Options +optionsInfo = + O.info + ( optionsParser + O.<**> O.helperWith (O.long "help" <> O.help "Show this help text.") + O.<**> O.simpleVersioner (showVersion EventlogLive.version) + ) + O.idm + +data Options = Options + { eventlogSocket :: EventlogSocket + , eventlogSocketTimeout :: Double + , eventlogSocketTimeoutExponent :: Double + , batchInterval :: Int + , maybeEventlogLogFile :: Maybe FilePath + , maybeHeapProfBreakdown :: Maybe HeapProfBreakdown + , verbosity :: Verbosity + , influxDBWriteParams :: I.WriteParams + } + +optionsParser :: O.Parser Options +optionsParser = + Options + <$> eventlogSocketParser + <*> eventlogSocketTimeoutParser + <*> eventlogSocketTimeoutExponentParser + <*> batchIntervalParser + <*> O.optional eventlogLogFileParser + <*> O.optional heapProfBreakdownParser + <*> verbosityParser + <*> influxDBWriteParamsParser + +-------------------------------------------------------------------------------- +-- InfluxDB Configuration + +influxDBWriteParamsParser :: O.Parser I.WriteParams +influxDBWriteParamsParser = params4 + where + params0 = I.writeParams <$> influxDBDatabaseParser + params1 = set I.server <$> influxDBServerParser <*> params0 + params2 = set I.retentionPolicy <$> influxDBRetentionPolicyParser <*> params1 + params3 = set I.authentication <$> influxDBCredentialsParser <*> params2 + params4 = set I.precision I.Nanosecond <$> params3 + +influxDBDatabaseParser :: O.Parser I.Database +influxDBDatabaseParser = + O.strOption + ( O.long "influxdb-database" + <> O.metavar "DATABASE" + <> O.help "InfluxDB database name" + ) + +influxDBServerParser :: O.Parser I.Server +influxDBServerParser = + I.Server + <$> ( O.strOption + ( O.long "influxdb-host" + <> O.metavar "HOST" + <> O.help "InfluxDB server host" + ) + <|> pure (I.defaultServer ^. I.host) + ) + <*> ( O.option + O.auto + ( O.long "influxdb-port" + <> O.metavar "HOST" + <> O.help "InfluxDB server host" + ) + <|> pure (I.defaultServer ^. I.port) + ) + <*> ( O.flag + False + True + ( O.long "influxdb-ssl" + <> O.help "InfluxDB server SSL" + ) + <|> pure (I.defaultServer ^. I.ssl) + ) + +influxDBRetentionPolicyParser :: O.Parser (Maybe I.Key) +influxDBRetentionPolicyParser = + O.optional + ( O.strOption + ( O.long "influxdb-retention-policy" + <> O.metavar "RETENTION_POLICY" + <> O.help "InfluxDB retention policy" + ) + ) + +influxDBCredentialsParser :: O.Parser (Maybe I.Credentials) +influxDBCredentialsParser = + O.optional + ( I.Credentials + <$> O.strOption + ( O.long "influxdb-username" + <> O.metavar "USERNAME" + <> O.help "InfluxDB username" + ) + <*> O.strOption + ( O.long "influxdb-password" + <> O.metavar "PASSWORD" + <> O.help "InfluxDB password" + ) + )