packages feed

freckle-kafka 0.1.0.0 → 0.1.1.0

raw patch · 10 files changed

+381/−7 lines, 10 filesdep +freckle-kafkadep +hspecdep +hspec-junit-formatter

Dependencies added: freckle-kafka, hspec, hspec-junit-formatter, monad-logger-aeson

Files

CHANGELOG.md view
@@ -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)
freckle-kafka.cabal view
@@ -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:
library/Freckle/App/Kafka.hs view
@@ -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
library/Freckle/App/Kafka/Producer.hs view
@@ -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
+ library/Freckle/App/Kafka/Producer/Hooks.hs view
@@ -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 ()
package.yaml view
@@ -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
+ test/Freckle/App/Kafka/Producer/HooksSpec.hs view
@@ -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
+ test/Spec.hs view
@@ -0,0 +1,1 @@+{-# OPTIONS_GHC -F -pgmF hspec-discover -Wno-missing-export-lists #-}
+ test/SpecHook.hs view
@@ -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
+ test/WhateverSpec.hs view
@@ -0,0 +1,8 @@+module WhateverSpec (spec) where++import Prelude++import Test.Hspec++spec :: Spec+spec = pure () -- Tests go here