Added Simple LogRecordProcessor and refactored shutdown and forceFlush #136

evanlauer1 · 2024-07-09T23:58:39Z

Big Context

Logging Support is being added to hs-opentelemetry. Logging spec

Small (This PR) Context

Two built-in processors are necessary for a functional SDK: the simple processor and the batching processor. This PR implements the first of the two. The simple processor follows the spec for LogRecordProcessors.

Testing

stack build runs
New test suite written for Simple LogRecordProcessor passes

evanlauer1 · 2024-07-09T23:59:00Z

Warning

This pull request is not mergeable via GitHub because a downstack PR is open. Once all requirements are satisfied, merge this PR as a stack on Graphite.
Learn more

This stack of pull requests is managed by Graphite. Learn more about stacking.

Join @evanlauer1 and the rest of your teammates on Graphite

evanlauer1

Finished self-review

evanlauer1 · 2024-07-10T17:27:25Z

api/src/OpenTelemetry/Internal/Common/Types.hs

  FlushResult (..),
+  takeWorseFlushResult,
+  takeWorstFlushResult,
+  exportResultToFlushResult,


Helper functions to convert from ExportResult -> FlushResult and FlushResult -> ShutdownResult. These are used to propagate Failures and Timeouts up the chain of export to forceFlush to shutdown.

evanlauer1 · 2024-07-10T17:29:28Z

api/src/OpenTelemetry/Internal/Common/Types.hs

+
+exportResultToFlushResult :: ExportResult -> FlushResult
+exportResultToFlushResult Success = FlushSuccess
+exportResultToFlushResult (Failure _) = FlushError


I don't know of a better way to do this conversion than drop the Exception

Suggested change

exportResultToFlushResult (Failure _) = FlushError

exportResultToFlushResult (Failure mErr) = FlushError $ toList mErr

evanlauer1 · 2024-07-10T17:31:24Z

api/src/OpenTelemetry/Internal/Logs/Types.hs

@@ -58,7 +58,7 @@ data LogRecordExporter body = LogRecordExporter
  -- Result:
  -- Success - The batch has been successfully exported. For protocol exporters this typically means that the data is sent over the wire and delivered to the destination server.
  -- Failure - exporting failed. The batch must be dropped. For example, this can happen when the batch contains bad data and cannot be serialized.
-  , logRecordExporterForceFlush :: IO ()
+  , logRecordExporterForceFlush :: IO FlushResult


It may not be necessary for forceFlush to return a FlushResult at every step in the chain of Exporter to Processor to LoggerProvider but I think it makes sense for each step to be able to communicate a failure or timeout.

evanlauer1 · 2024-07-10T17:33:08Z

api/src/OpenTelemetry/Internal/Logs/Types.hs

@@ -111,7 +111,7 @@ data LogRecordProcessor body = LogRecordProcessor
  -- ^ Called when a LogRecord is emitted. This method is called synchronously on the thread that emitted the LogRecord, therefore it SHOULD NOT block or throw exceptions.
  --
  -- A LogRecordProcessor may freely modify logRecord for the duration of the OnEmit call. If logRecord is needed after OnEmit returns (i.e. for asynchronous processing) only reads are permitted.
-  , logRecordProcessorShutdown :: IO (Async ShutdownResult)
+  , logRecordProcessorShutdown :: IO ShutdownResult


It makes more sense to me for the LoggerProvider or a Processor to internally implement or choose not to implement concurrency than for it to be enforced by the types. I imagine there are situations where one would not want something to be asynchronous. Removing Async from the signature also allows combinators like withAsync and mapConcurrently to be used.

At least for tracing, the docs say this:

Shutdown can be implemented as a blocking API or an asynchronous API which notifies the caller via a callback or an event. OpenTelemetry client authors can decide if they want to make the shutdown timeout configurable.

I can appreciate the argument in the other direction, but would prefer to keep Async here to stay consistent with the SpanProcessor API https://hackage.haskell.org/package/hs-opentelemetry-api-0.1.0.0/docs/OpenTelemetry-Processor.html#t:Processor

Another reason I made the change is that having Async in the signature made it much more difficult to propagate and combine ExportResults, FlushResults, and ShutdownResults. I also figured that it would be good if either shutdown, forceFlush, and export were either all asynchronous or all synchronous.

Also, the way I rewrote shutdownLoggerProvider seems cleaner and more abstract (using forConcurrently) than the way it was written for shutdownTracerProvider before. Plus, people defining a LogRecordProcessor no longer have to worry about the asynchronous part because it is handled by the LoggerProvider.

Let me know your thoughts.

evanlauer1 · 2024-07-10T17:35:17Z

api/src/OpenTelemetry/Logs/Core.hs

@@ -124,41 +126,56 @@ setGlobalLoggerProvider :: (MonadIO m) => LoggerProvider body -> m ()
 setGlobalLoggerProvider = liftIO . writeIORef globalLoggerProvider


+defaultShutdownTimeout :: Int
+defaultShutdownTimeout = 5_000_000


shutdown previously never timed out. I used the same shutdown timeout as forceFlush, but there is no reason that 5,000,000 microseconds is a magic number. There might be a better default timeout.

evanlauer1 · 2024-07-10T17:46:22Z

sdk/src/OpenTelemetry/LogRecordProcessor/Simple.hs

+
+{- | This is an implementation of LogRecordProcessor which passes finished logs and passes the export-friendly ReadableLogRecord
+representation to the configured LogRecordExporter, as soon as they are finished.
+-}


This implementation mirrors SpanProcessor.Simple but with a lot of refactoring. It also adds propagating ExportResults to FlushResults to ShutdownResults.

evanlauer1 · 2024-07-10T17:48:18Z

sdk/src/OpenTelemetry/LogRecordProcessor/Simple.hs

+          shutdownResult <- logRecordExporterShutdown
+
+          pure $ takeWorseShutdownResult shutdownResult $ flushResultToShutdownResult flushResult
+      , logRecordProcessorForceFlush


SpanProcessor.Simple does not implement forceFlush. Instead, it has a defacto force flush in shutdown.

In this implementation, forceFlush is implemented and shutdown calls forceFlush as it's supposed to.

evanlauer1 · 2024-07-10T17:51:03Z

sdk/test/OpenTelemetry/LogRecordProcessorSpec.hs

+      lr <- emitLogRecord l (emptyLogRecordArguments ("a bad one" :: String))
+      logRecordProcessorOnEmit processorNoShutdown lr Context.empty
+      numExportsNoShutdown <- readIORef numExportsNoShutdownRef
+      numExportsNoShutdown `shouldBe` 3


OnEmit calls should do nothing after shutdown.

evanlauer1 · 2024-07-10T17:52:04Z

sdk/test/OpenTelemetry/LogRecordProcessorSpec.hs

+      exportRes <- logRecordExporterExport testExporter H.empty
+      exportRes `shouldSatisfy` \case
+        Success -> False
+        Failure _ -> True


This makes sure that the Exporter's shutdown method was called.

evanlauer1 · 2024-07-10T17:52:46Z

sdk/test/OpenTelemetry/LogRecordProcessorSpec.hs

+      shutdownLoggerProvider Nothing lp
+
+      numExports <- readIORef numExportsRef
+      numExports `shouldBe` 3


This ensures that each emitted logRecord made it to the exporter.

iand675

Mostly feedback on how you're propagating the errors, but overall LGTM

iand675 · 2024-07-31T18:00:50Z

api/src/OpenTelemetry/Internal/Common/Types.hs

@@ -175,6 +200,25 @@ data FlushResult
  deriving (Show)


I'd recommend tweaking FlushError a bit here to carry exceptions with it:

| FlushError [SomeException]

iand675 · 2024-07-31T18:04:56Z

api/src/OpenTelemetry/Internal/Common/Types.hs

@@ -175,6 +200,25 @@ data FlushResult
  deriving (Show)


+-- | Returns @FlushError@ if either argument is @FlushError@, @FlushTimeout@ if either argument is @FlushTimeout@, and @FlushSuccess@ otherwise.
+takeWorseFlushResult :: FlushResult -> FlushResult -> FlushResult


If you take my advice on changing FlushError, you can make this into (<>) by implementing the Semigroup instance:

-- | Returns @FlushError@ if either argument is @FlushError@, @FlushTimeout@ -- if either argument is @FlushTimeout@, and @FlushSuccess@ otherwise. instance Semigroup FlushResult where (<>) (FlushError l) (FlushError r) = FlushError (l <> r) (<>) (FlushError l) _ = FlushError l (<>) _ (FlushError r) = FlushError r (<>) FlushTimeout _ = FlushTimeout (<>) _ FlushTimeout = FlushTimeout (<>) FlushSuccess FlushSuccess = FlushSuccess instance Monoid FlushResult where mempty = FlushSuccess

iand675 · 2024-07-31T18:05:48Z

api/src/OpenTelemetry/Internal/Common/Types.hs

+
+
+takeWorstFlushResult :: (Foldable t) => t FlushResult -> FlushResult
+takeWorstFlushResult = foldr takeWorseFlushResult FlushSuccess


If you go the semigroup / monoid route outlined above, this becomes

Suggested change

takeWorstFlushResult = foldr takeWorseFlushResult FlushSuccess

takeWorstFlushResult = fold

iand675 · 2024-07-31T18:07:23Z

api/src/OpenTelemetry/Internal/Common/Types.hs

+
+exportResultToFlushResult :: ExportResult -> FlushResult
+exportResultToFlushResult Success = FlushSuccess
+exportResultToFlushResult (Failure _) = FlushError


Suggested change

exportResultToFlushResult (Failure _) = FlushError

exportResultToFlushResult (Failure mErr) = FlushError $ toList mErr

iand675 · 2024-07-31T18:11:39Z

api/src/OpenTelemetry/Internal/Logs/Types.hs

@@ -111,7 +111,7 @@ data LogRecordProcessor body = LogRecordProcessor
  -- ^ Called when a LogRecord is emitted. This method is called synchronously on the thread that emitted the LogRecord, therefore it SHOULD NOT block or throw exceptions.
  --
  -- A LogRecordProcessor may freely modify logRecord for the duration of the OnEmit call. If logRecord is needed after OnEmit returns (i.e. for asynchronous processing) only reads are permitted.
-  , logRecordProcessorShutdown :: IO (Async ShutdownResult)
+  , logRecordProcessorShutdown :: IO ShutdownResult


At least for tracing, the docs say this:

Shutdown can be implemented as a blocking API or an asynchronous API which notifies the caller via a callback or an event. OpenTelemetry client authors can decide if they want to make the shutdown timeout configurable.

I can appreciate the argument in the other direction, but would prefer to keep Async here to stay consistent with the SpanProcessor API https://hackage.haskell.org/package/hs-opentelemetry-api-0.1.0.0/docs/OpenTelemetry-Processor.html#t:Processor

iand675 · 2024-07-31T18:13:50Z

api/src/OpenTelemetry/Internal/Common/Types.hs

@@ -10,7 +10,13 @@ module OpenTelemetry.Internal.Common.Types (
  AnyValue (..),
  ToValue (..),
  ShutdownResult (..),
+  takeWorseShutdownResult,
+  takeWorstShutdownResult,
+  flushResultToShutdownResult,
  FlushResult (..),


This is a bit of a driveby request on my part, but it would be nice to reexport FlushResult so that callers of forceFlushTracerProvider can actually use the type.

iand675 · 2024-07-31T18:15:16Z

sdk/src/OpenTelemetry/Processor/Simple/LogRecord.hs

+
+            exporterFlushRes <- logRecordExporterForceFlush exporter
+
+            pure $ takeWorseFlushResult exporterFlushRes chanFlushRes


Suggested change

pure $ takeWorseFlushResult exporterFlushRes chanFlushRes

pure (exporterFlushRes <> chanFlushRes)

…ces, and made error handlers

evanlauer1

Finished self-review for changes to FlushResult and ShutdownResult

evanlauer1 · 2024-07-31T19:41:08Z

api/src/OpenTelemetry/Internal/Common/Types.hs

+
+
+shutdownErrorHandler :: SomeException -> IO ShutdownResult
+shutdownErrorHandler = pure . ShutdownError . pure


used with handle and catch from Control.Exception

evanlauer1 · 2024-07-31T19:42:21Z

api/src/OpenTelemetry/Internal/Common/Types.hs

+
+
+takeWorstShutdownResult :: (Foldable t) => t ShutdownResult -> ShutdownResult
+takeWorstShutdownResult = fold


Developers may be able to use fold instead of this function, but I think takeWorsShutdownResult better communicates the meaning of the action and fold might be less well known

michaelpj · 2024-11-12T11:17:39Z

Hi @evanlauer1, whats' the status of this? Would you like me to give it a review?

evanlauer1 force-pushed the add-log-record-exporters branch from 88286a6 to 548bed8 Compare July 10, 2024 01:02

evanlauer1 force-pushed the add-simple-log-record-processor branch from 2ee2412 to 8d06716 Compare July 10, 2024 01:02

evanlauer1 commented Jul 10, 2024

View reviewed changes

evanlauer1 force-pushed the add-simple-log-record-processor branch from bc412e8 to dc76362 Compare July 10, 2024 19:38

evanlauer1 force-pushed the add-log-record-exporters branch from f415f88 to 6797029 Compare July 10, 2024 19:58

evanlauer1 force-pushed the add-simple-log-record-processor branch from 1c75434 to 608e2b0 Compare July 10, 2024 19:58

evanlauer1 force-pushed the add-log-record-exporters branch from 6797029 to e5db1ec Compare July 15, 2024 20:49

evanlauer1 force-pushed the add-simple-log-record-processor branch from 608e2b0 to 3a0df02 Compare July 15, 2024 20:49

evanlauer1 force-pushed the add-log-record-exporters branch from e5db1ec to e6fa1b9 Compare July 15, 2024 21:01

evanlauer1 force-pushed the add-simple-log-record-processor branch from 3a0df02 to 444d7f7 Compare July 15, 2024 21:01

evanlauer1 force-pushed the add-log-record-exporters branch from e6fa1b9 to 8ee8167 Compare July 15, 2024 22:55

evanlauer1 force-pushed the add-simple-log-record-processor branch from 444d7f7 to 45e0207 Compare July 15, 2024 22:55

evanlauer1 force-pushed the add-log-record-exporters branch from 8ee8167 to 7b85d46 Compare July 17, 2024 22:30

evanlauer1 force-pushed the add-simple-log-record-processor branch from 51a0d36 to 33a94d5 Compare July 17, 2024 22:30

evanlauer1 force-pushed the add-log-record-exporters branch from 7b85d46 to 2377db3 Compare July 17, 2024 23:35

evanlauer1 force-pushed the add-simple-log-record-processor branch from 33a94d5 to 693ac4c Compare July 17, 2024 23:35

evanlauer1 force-pushed the add-log-record-exporters branch from 2377db3 to a23f5bf Compare July 18, 2024 00:08

evanlauer1 force-pushed the add-simple-log-record-processor branch from 693ac4c to 7cb5e40 Compare July 18, 2024 00:08

evanlauer1 force-pushed the add-log-record-exporters branch from a23f5bf to 6fe460b Compare July 22, 2024 18:29

evanlauer1 force-pushed the add-simple-log-record-processor branch from 7cb5e40 to 939738e Compare July 22, 2024 18:29

evanlauer1 force-pushed the add-log-record-exporters branch from 6fe460b to b3274c6 Compare July 22, 2024 19:57

evanlauer1 force-pushed the add-simple-log-record-processor branch from f7d80a7 to cf6b2ca Compare July 22, 2024 19:57

evanlauer1 force-pushed the add-log-record-exporters branch from b3274c6 to 7bfaada Compare July 22, 2024 20:02

evanlauer1 force-pushed the add-simple-log-record-processor branch from cf6b2ca to 793d987 Compare July 22, 2024 20:03

evanlauer1 force-pushed the add-log-record-exporters branch from 7bfaada to 98e0d0c Compare July 24, 2024 00:13

evanlauer1 force-pushed the add-simple-log-record-processor branch from 793d987 to 4d35dc5 Compare July 24, 2024 00:13

evanlauer1 force-pushed the add-log-record-exporters branch from 98e0d0c to 2b1ff85 Compare July 24, 2024 00:23

evanlauer1 force-pushed the add-simple-log-record-processor branch from 4d35dc5 to b67e4e7 Compare July 24, 2024 00:23

evanlauer1 force-pushed the add-log-record-exporters branch from 2b1ff85 to cdabce7 Compare July 24, 2024 00:28

evanlauer1 force-pushed the add-simple-log-record-processor branch from b67e4e7 to 88ca14d Compare July 24, 2024 00:29

evanlauer1 force-pushed the add-simple-log-record-processor branch 2 times, most recently from 5ef10f4 to 486717e Compare July 31, 2024 00:41

evanlauer1 force-pushed the add-log-record-exporters branch from 8f00ae6 to 6d08447 Compare July 31, 2024 17:37

evanlauer1 force-pushed the add-simple-log-record-processor branch from 486717e to d669016 Compare July 31, 2024 17:37

evanlauer1 force-pushed the add-log-record-exporters branch from 6d08447 to af20283 Compare July 31, 2024 17:38

evanlauer1 force-pushed the add-simple-log-record-processor branch from d669016 to 304f604 Compare July 31, 2024 17:38

evanlauer1 added 8 commits July 31, 2024 11:14

Added Simple LogRecordProcessor and refactored shutdown and forceFlush

e404a2f

working on tests

77a9b64

Finished writing tests

a3fbc19

Fixed infinite loop because of bracket

fa40778

Removed HashMap from logRecordExporterExport and updated tests

b9df7ce

Fixed Simple Processor to work with upstream changes

9b6b234

Updated processor tests with emptyLogRecordArguments updates

6a12572

Export ImmutableLogRecord from Core.hs

7c35830

iand675 requested changes Jul 31, 2024

View reviewed changes

Fixed to work with changes to LogRecordExporterArguments

58f8451

evanlauer1 force-pushed the add-simple-log-record-processor branch from 304f604 to 58f8451 Compare July 31, 2024 18:22

Added SomeException to Flush and Shutdown Results, made Monoid instan…

9250ec2

…ces, and made error handlers

evanlauer1 commented Jul 31, 2024

View reviewed changes

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Added Simple LogRecordProcessor and refactored shutdown and forceFlush #136

Added Simple LogRecordProcessor and refactored shutdown and forceFlush #136

evanlauer1 commented Jul 9, 2024 •

edited

Loading

evanlauer1 commented Jul 9, 2024

evanlauer1 left a comment

evanlauer1 Jul 10, 2024

evanlauer1 Jul 10, 2024

iand675 Jul 31, 2024

evanlauer1 Jul 10, 2024

evanlauer1 Jul 10, 2024

iand675 Jul 31, 2024

evanlauer1 Jul 31, 2024

evanlauer1 Jul 10, 2024

evanlauer1 Jul 10, 2024

evanlauer1 Jul 10, 2024

evanlauer1 Jul 10, 2024

evanlauer1 Jul 10, 2024

evanlauer1 Jul 10, 2024

iand675 left a comment

iand675 Jul 31, 2024

evanlauer1 Jul 31, 2024

iand675 Jul 31, 2024

iand675 Jul 31, 2024

iand675 Jul 31, 2024

iand675 Jul 31, 2024

iand675 Jul 31, 2024

iand675 Jul 31, 2024

evanlauer1 left a comment

evanlauer1 Jul 31, 2024

evanlauer1 Jul 31, 2024

michaelpj commented Nov 12, 2024

	exportResultToFlushResult (Failure _) = FlushError
	exportResultToFlushResult (Failure mErr) = FlushError $ toList mErr



		takeWorstFlushResult :: (Foldable t) => t FlushResult -> FlushResult
		takeWorstFlushResult = foldr takeWorseFlushResult FlushSuccess

	takeWorstFlushResult = foldr takeWorseFlushResult FlushSuccess
	takeWorstFlushResult = fold


		exporterFlushRes <- logRecordExporterForceFlush exporter

		pure $ takeWorseFlushResult exporterFlushRes chanFlushRes

	pure $ takeWorseFlushResult exporterFlushRes chanFlushRes
	pure (exporterFlushRes <> chanFlushRes)



		shutdownErrorHandler :: SomeException -> IO ShutdownResult
		shutdownErrorHandler = pure . ShutdownError . pure



		takeWorstShutdownResult :: (Foldable t) => t ShutdownResult -> ShutdownResult
		takeWorstShutdownResult = fold

Added Simple LogRecordProcessor and refactored shutdown and forceFlush #136

Are you sure you want to change the base?

Added Simple LogRecordProcessor and refactored shutdown and forceFlush #136

Conversation

evanlauer1 commented Jul 9, 2024 • edited Loading

Big Context

Small (This PR) Context

Testing

evanlauer1 commented Jul 9, 2024

evanlauer1 left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

iand675 left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

evanlauer1 left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

michaelpj commented Nov 12, 2024

evanlauer1 commented Jul 9, 2024 •

edited

Loading