diff --git a/CHANGELOG.md b/CHANGELOG.md new file mode 100644 --- /dev/null +++ b/CHANGELOG.md @@ -0,0 +1,2 @@ +# consumers-metrics-prometheus-1.0.0.0 (2025-03-03) +* Initial release. diff --git a/LICENSE b/LICENSE new file mode 100644 --- /dev/null +++ b/LICENSE @@ -0,0 +1,30 @@ +Copyright (c) 2024, Scrive + +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 Scrive 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/README.md b/README.md new file mode 100644 --- /dev/null +++ b/README.md @@ -0,0 +1,8 @@ +# consumers-metrics-prometheus + +[![Hackage version](https://img.shields.io/hackage/v/consumers-metrics-prometheus.svg?label=Hackage)](https://hackage.haskell.org/package/consumers-metrics-prometheus) +[![Haskell-CI](https://github.com/scrive/consumers/actions/workflows/haskell-ci.yml/badge.svg?branch=master)](https://github.com/scrive/consumers/actions/workflows/haskell-ci.yml) + +Provides seamless instrumentation of your existing `consumers` using Prometheus. + +See [`consumers`](https://hackage.haskell.org/package/consumers) for usage. diff --git a/consumers-metrics-prometheus.cabal b/consumers-metrics-prometheus.cabal new file mode 100644 --- /dev/null +++ b/consumers-metrics-prometheus.cabal @@ -0,0 +1,57 @@ +cabal-version: 3.0 +name: consumers-metrics-prometheus +version: 1.0.0.0 +synopsis: Prometheus metrics for the consumers library + +description: Provides seamless instrumentation of your existing + consumers using Prometheus + +homepage: https://github.com/scrive/consumers +license: BSD-3-Clause +license-file: LICENSE +extra-source-files: CHANGELOG.md, README.md +author: Scrive AB +maintainer: Andrzej Rybczak , + Jonathan Jouty +copyright: Scrive AB +category: Concurrency, Database +build-type: Simple +tested-with: GHC == { 8.10.7, 9.0.2, 9.2.8, 9.4.8, 9.6.6, 9.8.4, 9.10.1, 9.12.1 } + +bug-reports: https://github.com/scrive/consumers/issues +source-repository head + type: git + location: https://github.com/scrive/consumers.git + +common language + ghc-options: -Wall -Wcompat -Werror=prepositive-qualified-module + + default-language: Haskell2010 + default-extensions: OverloadedStrings + , DisambiguateRecordFields + , DuplicateRecordFields + , FlexibleContexts + , NamedFieldPuns + , ImportQualifiedPost + , NumericUnderscores + , RecordWildCards + , ScopedTypeVariables + +library + import: language + + exposed-modules: Database.PostgreSQL.Consumers.Instrumented + + build-depends: base >= 4.14 && < 5 + , consumers >= 2.3 + , exceptions >= 0.10 + , hpqtypes >= 1.11 + , lifted-base >= 0.2 + , log-base >= 0.11 + , monad-control >= 1.0 + , monad-time >= 0.4 + , safe-exceptions >= 0.1.7 + , transformers-base >= 0.4 + , prometheus-client >= 1.1.0 + + hs-source-dirs: src diff --git a/src/Database/PostgreSQL/Consumers/Instrumented.hs b/src/Database/PostgreSQL/Consumers/Instrumented.hs new file mode 100644 --- /dev/null +++ b/src/Database/PostgreSQL/Consumers/Instrumented.hs @@ -0,0 +1,265 @@ +-- | +-- Provides seamless instrumentation of your existing @consumers@ using Prometheus +-- (see the [`consumers`](https://hackage.haskell.org/package/consumers) library +-- for usage). +module Database.PostgreSQL.Consumers.Instrumented + ( -- * Instrument + runInstrumentedConsumer + + -- ** Configuration + , defaultConsumerMetricsConfig + , ConsumerMetricsConfig (..) + + -- ** Metrics + , ConsumerMetrics + , registerConsumerMetrics + ) where + +import Control.Concurrent.Lifted +import Control.Exception.Safe +import Control.Monad.Base +import Control.Monad.Catch (ExitCase (..)) +import Control.Monad.Time +import Control.Monad.Trans.Control +import Database.PostgreSQL.Consumers (ConsumerConfig (..), Result (..), runConsumer) +import Database.PostgreSQL.PQTypes +import Log +import Prometheus qualified as Prom + +data ConsumerMetricsConfig = ConsumerMetricsConfig + { collectSeconds :: Int + -- ^ Collection interval in seconds + , jobExecutionBuckets :: [Prom.Bucket] + -- ^ Buckets to use for the 'jobExecution' histogram + , collectDegradeThresholdSeconds :: Double + -- ^ @logAttention@ and graceful degrade if collection takes longer than @x@ seconds + , collectDegradeSeconds :: Int + -- ^ Degraded collection interval in seconds + } + +-- | Hopefully sensible defaults that you can use +-- +-- @ +-- ConsumerMetricsConfig +-- { collectSeconds = 15 +-- , jobExecutionBuckets = [0.01, 0.05, 0.1, 0.5, 1, 2, 4, 8, 16, 32, 64, 128, 256, 512] +-- , collectDegradeThresholdSeconds = 0.1 +-- , collectDegradeSeconds = 60 +-- } +-- @ +defaultConsumerMetricsConfig :: ConsumerMetricsConfig +defaultConsumerMetricsConfig = + ConsumerMetricsConfig + { collectSeconds = 15 + , jobExecutionBuckets = [0.01, 0.05, 0.1, 0.5, 1, 2, 4, 8, 16, 32, 64, 128, 256, 512] + , collectDegradeThresholdSeconds = 0.1 + , collectDegradeSeconds = 60 + } + +-- | Metrics store for the following Prometheus metrics: +-- +-- @ +-- # HELP consumers_job_execution_seconds Execution time of jobs in seconds, by job_name, includes the job_result +-- # TYPE consumers_job_execution_seconds histogram +-- +-- # HELP consumers_jobs_reserved_total The total number of job reserved, by job_name +-- # TYPE consumers_jobs_reserved_total counter +-- +-- # HELP consumers_jobs_overdue The current number of jobs overdue, by job_name +-- # TYPE consumers_jobs_overdue gauge +-- +-- # HELP consumers_job_info The number of workers registered for a given job_name +-- # TYPE consumers_job_info gauge +-- @ +data ConsumerMetrics = ConsumerMetrics + { collectSeconds :: Int + , collectDegradeThresholdSeconds :: Double + , collectDegradeSeconds :: Int + , jobInfo :: Prom.Vector Prom.Label1 Prom.Gauge + , jobsOverdue :: Prom.Vector Prom.Label1 Prom.Gauge + , jobsReserved :: Prom.Vector Prom.Label1 Prom.Counter + , jobsExecution :: Prom.Vector Prom.Label2 Prom.Histogram + } + +registerConsumerMetrics :: MonadBaseControl IO m => ConsumerMetricsConfig -> m ConsumerMetrics +registerConsumerMetrics ConsumerMetricsConfig {..} = liftBase $ do + jobInfo <- + Prom.register + . Prom.vector "job_name" + $ Prom.gauge + Prom.Info + { metricName = "consumers_job_info" + , metricHelp = "The number of workers registered for a given job_name" + } + jobsOverdue <- + Prom.register + . Prom.vector "job_name" + $ Prom.gauge + Prom.Info + { metricName = "consumers_jobs_overdue" + , metricHelp = "The current number of jobs overdue, by job_name" + } + jobsReserved <- + Prom.register + . Prom.vector "job_name" + $ Prom.counter + Prom.Info + { metricName = "consumers_jobs_reserved_total" + , metricHelp = "The total number of job reserved, by job_name" + } + jobsExecution <- + Prom.register + . Prom.vector ("job_name", "job_result") + $ Prom.histogram + Prom.Info + { metricName = "consumers_job_execution_seconds" + , metricHelp = "Execution time of jobs in seconds, by job_name, includes the job_result" + } + jobExecutionBuckets + pure $ ConsumerMetrics {..} + +-- | Run a 'ConsumerConfig', but with instrumentation added. +-- +-- This should be used in place of 'runConsumer'. +-- Use 'registerConsumerMetrics' to create the metrics. +-- +-- A thread will spawned to collect "queue" metrics every 'collectSeconds', and +-- an altered @ccProcessJob@ will be run that collects "job" metrics. +-- +-- See 'ConsumerMetrics' for more details on the metrics collected. +runInstrumentedConsumer + :: forall m idx job + . ( Eq idx + , Show idx + , FromSQL idx + , ToSQL idx + , MonadBaseControl IO m + , MonadMask m + , MonadTime m + , MonadLog m + ) + => ConsumerMetrics + -> ConsumerConfig m idx job + -> ConnectionSourceM m + -> m (m ()) +runInstrumentedConsumer metrics config connSource = do + let instrumentedConfig = instrumentConsumerConfig metrics config + cleanupConsumer <- runConsumer instrumentedConfig connSource + tidMetricsCollection <- runMetricsCollection connSource metrics config + pure $ do + cleanupConsumer + killThread tidMetricsCollection + +-- | Spawns a new thread to run "queue" metrics collection for the given configuration +runMetricsCollection + :: forall m idx job + . ( MonadBaseControl IO m + , MonadMask m + , MonadTime m + , MonadLog m + ) + => ConnectionSourceM m + -> ConsumerMetrics + -> ConsumerConfig m idx job + -> m ThreadId +runMetricsCollection connSource metrics@ConsumerMetrics {..} config = localDomain "metrics-collection" $ fork collectLoop + where + collectLoop = do + seconds <- handleAny handleEx collect + threadDelay (seconds * 1_000_000) + collectLoop + + handleEx e = do + logAttention "Exception while running metrics-collection" $ + object + [ "exception" .= show e + , "rerun_seconds" .= collectDegradeSeconds + ] + pure collectDegradeSeconds + + collect = do + logInfo_ "Collecting consumer metrics" + t1 <- monotonicTime + collectMetrics connSource metrics config + t2 <- monotonicTime + let runtime = t2 - t1 + -- Graceful degrade if things take too long + if runtime < collectDegradeThresholdSeconds + then pure collectSeconds + else do + logAttention "Consumer metrics collection took long" $ + object + [ "runtime" .= runtime + , "threshold" .= collectDegradeThresholdSeconds + , "rerun_seconds" .= collectDegradeSeconds + ] + pure collectDegradeSeconds + +-- | Collect and report "queue" metrics for a given configuration +collectMetrics + :: ( MonadBaseControl IO m + , MonadMask m + , MonadThrow m + ) + => ConnectionSourceM m + -> ConsumerMetrics + -> ConsumerConfig m idx job + -> m () +collectMetrics connSource ConsumerMetrics {..} ConsumerConfig {ccJobsTable, ccConsumersTable} = runDBT connSource defaultTransactionSettings $ do + let jobName = unRawSQL ccJobsTable + + info <- do + runSQL_ $ + "SELECT count(id)::float8 FROM " + <> raw ccConsumersTable + <> " WHERE name =" unRawSQL ccJobsTable + fetchOne runIdentity + liftBase $ Prom.withLabel jobInfo jobName (`Prom.setGauge` info) + + overdue <- do + runSQL_ $ + "SELECT count(id)::float8 FROM " + <> raw ccJobsTable + <> " WHERE run_at <= now() AND reserved_by IS NULL" + fetchOne runIdentity + liftBase $ Prom.withLabel jobsOverdue jobName (`Prom.setGauge` overdue) + +-- | Alter a configuration to collect "job" metrics on 'ccProcessJob' +instrumentConsumerConfig + :: ( MonadBaseControl IO m + , MonadMask m + , MonadTime m + , MonadLog m + ) + => ConsumerMetrics + -> ConsumerConfig m idx job + -> ConsumerConfig m idx job +instrumentConsumerConfig ConsumerMetrics {..} ConsumerConfig {..} = + ConsumerConfig {ccProcessJob = ccProcessJob', ..} + where + jobName = unRawSQL ccJobsTable + + -- Alter the `ccProcessJob` function by adding instrumentation: + -- First we increment the 'metrics.jobsReserved' counter (while handling + -- any potential exceptions). Then we use `generalBracket` to acquire the + -- start time, process the job, and finally report the job metrics as the + -- "release" action. Any exceptions in `ccProcessJob` will be re-raised by + -- `generalBracket`. However, we need to handle exceptions in the + -- "release" action as we don't want those exceptions to propagate upwards + -- to the consumer's `ccOnException` (and thus potentially change the + -- result of the job). + ccProcessJob' job = do + handleAny handleEx . liftBase $ Prom.withLabel jobsReserved jobName Prom.incCounter + fst <$> generalBracket monotonicTime reportJob (const $ ccProcessJob job) + + reportJob t1 jobExit = handleAny handleEx $ do + t2 <- monotonicTime + let duration = t2 - t1 + resultLabel = case jobExit of + ExitCaseSuccess (Ok _) -> "ok" + ExitCaseSuccess (Failed _) -> "failed" + ExitCaseException _ -> "exception" + ExitCaseAbort -> "abort" + liftBase $ Prom.withLabel jobsExecution (jobName, resultLabel) (`Prom.observe` duration) + + handleEx e = logAttention "Exception while instrumenting job" $ object ["exception" .= show e]