diff --git a/CHANGELOG.md b/CHANGELOG.md --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,4 +1,14 @@ -## [_Unreleased_](https://github.com/freckle/freckle-kafka/compare/v0.1.0.0...main) +## [_Unreleased_](https://github.com/freckle/freckle-kafka/compare/v0.1.1.0...main) + +## [v0.1.1.0](https://github.com/freckle/freckle-kafka/compare/v0.1.0.0...v0.1.1.0) + +- Add `Freckle.App.Kafka.Producer.Hooks`, with `ProducerHooks` for reacting to + librdkafka delivery reports and client errors, and `loggingProducerHooks` to + log failures (topic, partition, key and error; never the payload). Without + these, a message that fails after being enqueued is never reported. +- Add `createKafkaProducerPoolWith` and `withKafkaProducerPoolWith`, which + register `ProducerHooks` on every producer in the pool. The existing + functions are unchanged and register none. ## [v0.1.0.0](https://github.com/freckle/freckle-kafka/compare/v0.0.0.4...v0.1.0.0) diff --git a/freckle-kafka.cabal b/freckle-kafka.cabal --- a/freckle-kafka.cabal +++ b/freckle-kafka.cabal @@ -1,6 +1,6 @@ cabal-version: 1.18 name: freckle-kafka -version: 0.1.0.0 +version: 0.1.1.0 license: MIT license-file: LICENSE maintainer: Freckle Education @@ -24,6 +24,7 @@ Freckle.App.Kafka Freckle.App.Kafka.Consumer Freckle.App.Kafka.Producer + Freckle.App.Kafka.Producer.Hooks hs-source-dirs: library other-modules: Paths_freckle_kafka @@ -52,6 +53,7 @@ hs-opentelemetry-sdk >=0.0.3.6, hw-kafka-client >=5.0.0, lens >=5.1.1, + monad-logger-aeson >=0.4.0.4, mtl >=2.2.2, resource-pool >=0.4.0.0, text >=1.2.5.0, @@ -59,6 +61,44 @@ unliftio >=0.2.25.0, unordered-containers >=0.2.19.1, yesod-core >=1.6.24.2 + + if impl(ghc >=9.8) + ghc-options: + -Wno-missing-role-annotations -Wno-missing-poly-kind-signatures + +test-suite spec + type: exitcode-stdio-1.0 + main-is: Spec.hs + hs-source-dirs: test + other-modules: + Freckle.App.Kafka.Producer.HooksSpec + SpecHook + WhateverSpec + Paths_freckle_kafka + + default-language: GHC2021 + default-extensions: + DataKinds DeriveAnyClass DerivingVia DerivingStrategies GADTs + LambdaCase NoImplicitPrelude NoMonomorphismRestriction + OverloadedStrings TypeFamilies + + ghc-options: + -fignore-optim-changes -fwrite-ide-info -Weverything + -Wno-all-missed-specialisations -Wno-missing-exported-signatures + -Wno-missing-import-lists -Wno-missing-kind-signatures + -Wno-missing-local-signatures -Wno-missing-safe-haskell-mode + -Wno-monomorphism-restriction -Wno-prepositive-qualified-module + -Wno-safe -Wno-unsafe -threaded -rtsopts -with-rtsopts=-N + + build-depends: + base >=4.16.4.0 && <5, + containers >=0.6.5.1, + freckle-kafka, + hspec >=2.10.0, + hspec-junit-formatter >=1.1.2.1, + hw-kafka-client >=5.0.0, + monad-logger-aeson >=0.4.0.4, + text >=1.2.5.0 if impl(ghc >=9.8) ghc-options: diff --git a/library/Freckle/App/Kafka.hs b/library/Freckle/App/Kafka.hs --- a/library/Freckle/App/Kafka.hs +++ b/library/Freckle/App/Kafka.hs @@ -1,7 +1,9 @@ module Freckle.App.Kafka ( module Freckle.App.Kafka.Consumer , module Freckle.App.Kafka.Producer + , module Freckle.App.Kafka.Producer.Hooks ) where import Freckle.App.Kafka.Consumer import Freckle.App.Kafka.Producer +import Freckle.App.Kafka.Producer.Hooks diff --git a/library/Freckle/App/Kafka/Producer.hs b/library/Freckle/App/Kafka/Producer.hs --- a/library/Freckle/App/Kafka/Producer.hs +++ b/library/Freckle/App/Kafka/Producer.hs @@ -8,7 +8,9 @@ , KafkaProducerPool (..) , HasKafkaProducerPool (..) , withKafkaProducerPool + , withKafkaProducerPoolWith , createKafkaProducerPool + , createKafkaProducerPoolWith , produceKeyedOn , produce ) where @@ -37,6 +39,7 @@ import Data.Text qualified as T import Data.Time (NominalDiffTime) import Freckle.App.Env qualified as Env +import Freckle.App.Kafka.Producer.Hooks (ProducerHooks, producerHooksProperties) import Kafka.Producer import OpenTelemetry.Trace (SpanKind (..), defaultSpanArguments) import OpenTelemetry.Trace qualified as Trace @@ -113,9 +116,20 @@ -> KafkaProducerPoolConfig -> (Pool KafkaProducer -> m a) -> m a -withKafkaProducerPool addresses config = +withKafkaProducerPool = withKafkaProducerPoolWith mempty + +-- | 'withKafkaProducerPool', registering the given 'ProducerHooks' on every +-- producer in the pool +withKafkaProducerPoolWith + :: MonadUnliftIO m + => ProducerHooks + -> NonEmpty BrokerAddress + -> KafkaProducerPoolConfig + -> (Pool KafkaProducer -> m a) + -> m a +withKafkaProducerPoolWith hooks addresses config = bracket - (createKafkaProducerPool addresses config) + (createKafkaProducerPoolWith hooks addresses config) (liftIO . Pool.destroyAllResources) -- | Create a Kafka producer pool for the given broker addresses and config @@ -127,7 +141,20 @@ => NonEmpty BrokerAddress -> KafkaProducerPoolConfig -> m (Pool KafkaProducer) -createKafkaProducerPool addresses config = +createKafkaProducerPool = createKafkaProducerPoolWith mempty + +-- | 'createKafkaProducerPool', registering the given 'ProducerHooks' on every +-- producer in the pool +-- +-- See "Freckle.App.Kafka.Producer.Hooks" for logging failures and counting +-- events. +createKafkaProducerPoolWith + :: MonadIO m + => ProducerHooks + -> NonEmpty BrokerAddress + -> KafkaProducerPoolConfig + -> m (Pool KafkaProducer) +createKafkaProducerPoolWith hooks addresses config = liftIO $ Pool.newPool $ Pool.setNumStripes (Just $ kafkaProducerPoolConfigStripes config) $ @@ -142,7 +169,10 @@ ( \err -> Annotated.throw $ userError ("Failed to open kafka producer: " <> show err) ) pure - =<< newProducer (brokersList $ toList addresses) + =<< newProducer + ( brokersList (toList addresses) + <> producerHooksProperties hooks + ) produceKeyedOn :: ( MonadUnliftIO m diff --git a/library/Freckle/App/Kafka/Producer/Hooks.hs b/library/Freckle/App/Kafka/Producer/Hooks.hs new file mode 100644 --- /dev/null +++ b/library/Freckle/App/Kafka/Producer/Hooks.hs @@ -0,0 +1,111 @@ +-- | Observability hooks for Kafka producers +-- +-- By default, a producer created by 'Freckle.App.Kafka.Producer.createKafkaProducerPool' +-- only reports /local/ enqueue failures (the result of @produceMessage@). +-- Whether a message actually reached the broker is reported asynchronously by +-- librdkafka, through a delivery report, and client-level problems (broker +-- down, auth failure, ...) through an error callback. This module wraps both +-- so that failures can be logged and counted. +-- +-- /Note: librdkafka only runs these callbacks when the producer is polled./ +-- @hw-kafka-client@ polls after each @produceMessage@ and when the producer is +-- closed or flushed, but not in the background. A report for a message may +-- therefore be delayed until the next message is produced on the same +-- producer, or until the producer is closed. +module Freckle.App.Kafka.Producer.Hooks + ( ProducerHooks (..) + , loggingProducerHooks + , producerHooksProperties + ) where + +import Prelude + +import Blammo.Logging (Message ((:#)), logErrorNS, (.=)) +import Control.Monad.Logger.Aeson (LoggingT) +import Data.Text qualified as T +import Data.Text.Encoding (decodeUtf8With) +import Data.Text.Encoding.Error (lenientDecode) +import Kafka.Producer + ( DeliveryReport (..) + , KafkaError + , ProducerProperties + , ProducerRecord (..) + , TopicName (..) + , deliveryCallback + , errorCallback + , setCallback + ) +import UnliftIO.Exception (handleAny) + +-- | Actions to run on events reported by librdkafka +-- +-- These run on a librdkafka thread, so they should be quick, and they are +-- called in the order they were combined with '<>'. An exception thrown by a +-- hook is swallowed, so a faulty hook can't stop delivery reports being +-- processed. +data ProducerHooks = ProducerHooks + { onDeliveryReport :: DeliveryReport -> IO () + -- ^ Called once per message, with success or failure + -- + -- Use this to count sent and failed events, for example. + , onProducerError :: KafkaError -> String -> IO () + -- ^ Called for client-level errors, with librdkafka's message + } + +instance Semigroup ProducerHooks where + a <> b = + ProducerHooks + { onDeliveryReport = \r -> onDeliveryReport a r *> onDeliveryReport b r + , onProducerError = \e m -> onProducerError a e m *> onProducerError b e m + } + +instance Monoid ProducerHooks where + mempty = + ProducerHooks + { onDeliveryReport = \_ -> pure () + , onProducerError = \_ _ -> pure () + } + +-- | Hooks that log failures at error level, under the @kafka@ namespace +-- +-- Hooks run in 'IO' on a librdkafka thread, so the caller supplies how to run +-- logging there, such as @runLoggerLoggingT logger@ from Blammo. +-- +-- Successful deliveries are not logged. The event payload is never logged, +-- only the topic, partition and key. +loggingProducerHooks :: (LoggingT IO () -> IO ()) -> ProducerHooks +loggingProducerHooks runLogging = + ProducerHooks + { onDeliveryReport = \case + DeliverySuccess {} -> pure () + DeliveryFailure record err -> + runLogging + $ logErrorNS "kafka" + $ "Failed to deliver event" + :# [ "topic" .= unTopicName (prTopic record) + , "partition" .= T.pack (show (prPartition record)) + , "key" .= fmap (decodeUtf8With lenientDecode) (prKey record) + , "error" .= T.pack (show err) + ] + NoMessageError err -> + runLogging + $ logErrorNS "kafka" + $ "Delivery report without a message" + :# ["error" .= T.pack (show err)] + , onProducerError = \err message -> + runLogging + $ logErrorNS "kafka" + $ "Kafka producer error" + :# ["error" .= T.pack (show err), "message" .= T.pack message] + } + +-- | Register the hooks as callbacks on a producer +-- +-- Replaces the default delivery callback, which ignores reports. +producerHooksProperties :: ProducerHooks -> ProducerProperties +producerHooksProperties hooks = + setCallback (deliveryCallback $ safely . onDeliveryReport hooks) + <> setCallback (errorCallback $ \e m -> safely $ onProducerError hooks e m) + where + safely :: IO () -> IO () + safely = handleAny $ \_ -> pure () diff --git a/package.yaml b/package.yaml --- a/package.yaml +++ b/package.yaml @@ -1,5 +1,5 @@ name: freckle-kafka -version: 0.1.0.0 +version: 0.1.1.0 maintainer: Freckle Education category: Database github: freckle/freckle-kafka @@ -63,6 +63,7 @@ - hs-opentelemetry-sdk - hw-kafka-client - lens + - monad-logger-aeson - mtl - resource-pool >= 0.4.0.0 - text @@ -70,3 +71,17 @@ - unliftio - unordered-containers - yesod-core + +tests: + spec: + main: Spec.hs + source-dirs: test + ghc-options: -threaded -rtsopts "-with-rtsopts=-N" + dependencies: + - containers + - freckle-kafka + - hspec + - hspec-junit-formatter + - hw-kafka-client + - monad-logger-aeson + - text diff --git a/test/Freckle/App/Kafka/Producer/HooksSpec.hs b/test/Freckle/App/Kafka/Producer/HooksSpec.hs new file mode 100644 --- /dev/null +++ b/test/Freckle/App/Kafka/Producer/HooksSpec.hs @@ -0,0 +1,146 @@ +module Freckle.App.Kafka.Producer.HooksSpec + ( spec + ) where + +import Prelude + +import Control.Exception (bracket, throwIO) +import Control.Monad.Logger.Aeson (LogLevel (..), LoggingT, runLoggingT) +import Data.IORef (IORef, modifyIORef', newIORef, readIORef) +import Data.Map.Strict qualified as Map +import Data.Text (Text) +import Data.Text qualified as T +import Freckle.App.Kafka.Producer.Hooks +import Kafka.Consumer.Types (Offset (..)) +import Kafka.Producer +import System.Timeout (timeout) +import Test.Hspec + +spec :: Spec +spec = do + describe "ProducerHooks" $ do + it "runs both hooks, in order, when combined" $ do + ran <- newIORef ([] :: [String]) + let hook name = + mempty {onDeliveryReport = \_ -> modifyIORef' ran (name :)} + + onDeliveryReport (hook "first" <> hook "second") report + + readIORef ran `shouldReturn` ["second", "first"] + + describe "loggingProducerHooks" $ do + it "logs a delivery failure with topic and key, but not the payload" $ do + (run, logged) <- captureLogs + + onDeliveryReport (loggingProducerHooks run) + $ DeliveryFailure record (KafkaError "boom") + + readIORef logged >>= \case + [(source, level, message)] -> do + source `shouldBe` "kafka" + level `shouldBe` LevelError + message `shouldSatisfy` T.isInfixOf "Failed to deliver event" + message `shouldSatisfy` T.isInfixOf "answers" + message `shouldSatisfy` T.isInfixOf "student-123" + message `shouldSatisfy` (not . T.isInfixOf "SECRET-PAYLOAD") + other -> + expectationFailure $ "Expected one log, got " <> show (length other) + + it "does not log a successful delivery" $ do + (run, logged) <- captureLogs + + onDeliveryReport (loggingProducerHooks run) + $ DeliverySuccess record (Offset 0) + + readIORef logged `shouldReturn` [] + + it "logs a client error" $ do + (run, logged) <- captureLogs + + onProducerError (loggingProducerHooks run) (KafkaError "boom") "all down" + + readIORef logged >>= \case + [(source, level, message)] -> do + source `shouldBe` "kafka" + level `shouldBe` LevelError + message `shouldSatisfy` T.isInfixOf "Kafka producer error" + message `shouldSatisfy` T.isInfixOf "all down" + other -> + expectationFailure $ "Expected one log, got " <> show (length other) + + -- These use a real producer, but no broker: nothing listens on port 1, so + -- the message sits in librdkafka's queue until message.timeout.ms, then is + -- reported as failed. 'closeProducer' polls until the queue is empty, so + -- the delivery callback has run by the time 'withProducer' returns. + describe "producerHooksProperties" $ do + it "reports a failed delivery to the hook" $ do + reports <- newIORef [] + let hooks = + mempty {onDeliveryReport = \r -> modifyIORef' reports (r :)} + + withProducer (unreachable hooks) $ \producer -> + produceMessage producer record `shouldReturn` Nothing + + readIORef reports >>= \case + [DeliveryFailure failed _] -> do + prTopic failed `shouldBe` prTopic record + prKey failed `shouldBe` prKey record + prValue failed `shouldBe` prValue record + other -> + expectationFailure + $ "Expected one delivery failure, got " <> show other + + it "survives a hook that throws" $ do + let hooks = + mempty {onDeliveryReport = \_ -> throwIO $ userError "hook failed"} + + withProducer (unreachable hooks) $ \producer -> + produceMessage producer record `shouldReturn` Nothing + +record :: ProducerRecord +record = + ProducerRecord + { prTopic = "answers" + , prPartition = UnassignedPartition + , prKey = Just "student-123" + , prValue = Just "SECRET-PAYLOAD" + , prHeaders = mempty + } + +report :: DeliveryReport +report = NoMessageError $ KafkaError "boom" + +-- | A runner for 'loggingProducerHooks', and what it has logged so far +captureLogs + :: IO + ( LoggingT IO () -> IO () + , IORef [(Text, LogLevel, Text)] + ) +captureLogs = do + logged <- newIORef [] + let run action = + runLoggingT action $ \_ source level message -> + modifyIORef' logged ((source, level, T.pack (show message)) :) + pure (run, logged) + +-- | Producer settings for a broker that is not there, failing fast +unreachable :: ProducerHooks -> ProducerProperties +unreachable hooks = + brokersList ["localhost:1"] + -- Not 'sendTimeout': in hw-kafka-client 5.3.0 that sets a topic-level + -- property which is never applied, so messages would wait out the default + -- 5 minutes. The global property works. + <> extraProps (Map.singleton "delivery.timeout.ms" "500") + -- librdkafka logs every refused connection to stderr; only log emergencies + <> logLevel KafkaLogEmerg + <> producerHooksProperties hooks + +-- | Run an action with a producer, and flush it afterwards +-- +-- Fails, rather than hangs, if that takes more than 20 seconds. +withProducer :: ProducerProperties -> (KafkaProducer -> IO a) -> IO a +withProducer props action = + maybe (fail "Producer was not flushed within 20 seconds") pure + =<< timeout (20 * 1000000) (bracket open closeProducer action) + where + open = either throwIO pure =<< newProducer props diff --git a/test/Spec.hs b/test/Spec.hs new file mode 100644 --- /dev/null +++ b/test/Spec.hs @@ -0,0 +1,1 @@ +{-# OPTIONS_GHC -F -pgmF hspec-discover -Wno-missing-export-lists #-} diff --git a/test/SpecHook.hs b/test/SpecHook.hs new file mode 100644 --- /dev/null +++ b/test/SpecHook.hs @@ -0,0 +1,11 @@ +module SpecHook + ( hook + ) where + +import Prelude + +import Test.Hspec +import Test.Hspec.JUnit.Formatter.Env as Formatter + +hook :: Spec -> Spec +hook = Formatter.whenEnabled Formatter.add . parallel diff --git a/test/WhateverSpec.hs b/test/WhateverSpec.hs new file mode 100644 --- /dev/null +++ b/test/WhateverSpec.hs @@ -0,0 +1,8 @@ +module WhateverSpec (spec) where + +import Prelude + +import Test.Hspec + +spec :: Spec +spec = pure () -- Tests go here