Added Simple LogRecordProcessor and refactored shutdown and forceFlush

api/src/OpenTelemetry/Exporter/LogRecord.hs

@@ -5,9 +5,18 @@ module OpenTelemetry.Exporter.LogRecord (
   logRecordExporterExport,
   logRecordExporterForceFlush,
   logRecordExporterShutdown,
+  ExportResult (..),
+  FlushResult (..),
+  flushErrorHandler,
+  takeWorstFlushResult,
+  exportResultToFlushResult,
   ShutdownResult (..),
+  shutdownErrorHandler,
+  takeWorstShutdownResult,
+  flushResultToShutdownResult,
 ) where
 
+import OpenTelemetry.Internal.Common.Types
 import OpenTelemetry.Internal.Logs.Types (
   LogRecordExporter,
   LogRecordExporterArguments (..),
@@ -16,5 +25,4 @@ import OpenTelemetry.Internal.Logs.Types (
   logRecordExporterShutdown,
   mkLogRecordExporter,
  )
-import OpenTelemetry.Processor.LogRecord (ShutdownResult (..))
 

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

@@ -10,13 +10,20 @@ module OpenTelemetry.Internal.Common.Types (
   AnyValue (..),
   ToValue (..),
   ShutdownResult (..),
+  shutdownErrorHandler,
+  takeWorstShutdownResult,
+  flushResultToShutdownResult,
   FlushResult (..),
+  flushErrorHandler,
+  takeWorstFlushResult,
+  exportResultToFlushResult,
   ExportResult (..),
 ) where
 
-import Control.Exception (SomeException)
+import Control.Exception (Exception, SomeException)
 import Data.ByteString (ByteString)
 import Data.Data (Data)
+import Data.Foldable (fold, toList)
 import qualified Data.HashMap.Strict as H
 import Data.Hashable (Hashable)
 import Data.Int (Int64)
@@ -159,22 +166,91 @@ instance ToValue AnyValue where
   toValue = id
 
 
-data ShutdownResult = ShutdownSuccess | ShutdownFailure | ShutdownTimeout
+data ShutdownResult
+  = ShutdownSuccess
+  | ShutdownTimeout
+  | ShutdownError [SomeException]
+
+
+{- | (<>) concatenates the error lists if both arguments are @ShutdownError@, otherwise it returns the worst of the two in this order
+- @ShutdownError@
+- @ShutdownTimeout@
+- @ShutdownSuccess@
+-}
+instance Semigroup ShutdownResult where
+  ShutdownError l <> ShutdownError r = ShutdownError (l <> r)
+  ShutdownError es <> _ = ShutdownError es
+  _ <> ShutdownError es = ShutdownError es
+  ShutdownTimeout <> _ = ShutdownTimeout
+  _ <> ShutdownTimeout = ShutdownTimeout
+  ShutdownSuccess <> ShutdownSuccess = ShutdownSuccess
+
+
+-- | mempty is ShutdownSuccess
+instance Monoid ShutdownResult where
+  mempty = ShutdownSuccess
+
+
+shutdownErrorHandler :: SomeException -> IO ShutdownResult
+shutdownErrorHandler = pure . ShutdownError . pure
+
+
+flushResultToShutdownResult :: FlushResult -> ShutdownResult
+flushResultToShutdownResult FlushSuccess = ShutdownSuccess
+flushResultToShutdownResult FlushTimeout = ShutdownTimeout
+flushResultToShutdownResult (FlushError es) = ShutdownError es
+
+
+takeWorstShutdownResult :: (Foldable t) => t ShutdownResult -> ShutdownResult
+takeWorstShutdownResult = fold
 
 
 -- | The outcome of a call to @OpenTelemetry.Trace.forceFlush@ or @OpenTelemetry.Logs.forceFlush@
 data FlushResult
-  = -- | One or more spans or @LogRecord@s did not export from all associated exporters
+  = -- | Flushing spans or @LogRecord@s to all associated exporters succeeded.
+    FlushSuccess
+  | -- | One or more spans or @LogRecord@s did not export from all associated exporters
     -- within the alotted timeframe.
     FlushTimeout
-  | -- | Flushing spans or @LogRecord@s to all associated exporters succeeded.
-    FlushSuccess
   | -- | One or more exporters failed to successfully export one or more
     -- unexported spans or @LogRecord@s.
-    FlushError
+    FlushError [SomeException]
   deriving (Show)
 
 
+{- | (<>) concatenates the error lists if both arguments are @FlushError@, otherwise it returns the worst of the two in this order
+- @FlushError@
+- @FlushTimeout@
+- @FlushSuccess@
+-}
+instance Semigroup FlushResult where
+  FlushError l <> FlushError r = FlushError (l <> r)
+  FlushError es <> _ = FlushError es
+  _ <> FlushError es = FlushError es
+  FlushTimeout <> _ = FlushTimeout
+  _ <> FlushTimeout = FlushTimeout
+  FlushSuccess <> FlushSuccess = FlushSuccess
+
+
+-- | mempty is FlushSuccess
+instance Monoid FlushResult where
+  mempty = FlushSuccess
+
+
+flushErrorHandler :: SomeException -> IO FlushResult
+flushErrorHandler = pure . FlushError . pure
+
+
+takeWorstFlushResult :: (Foldable t) => t FlushResult -> FlushResult
+takeWorstFlushResult = fold
+
+
+exportResultToFlushResult :: ExportResult -> FlushResult
+exportResultToFlushResult Success = FlushSuccess
+exportResultToFlushResult (Failure mErr) = FlushError $ toList mErr
+
+
 data ExportResult
   = Success
   | Failure (Maybe SomeException)
+  deriving (Show)

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

@@ -21,6 +21,7 @@ module OpenTelemetry.Internal.Logs.Core (
 
 import Control.Applicative
 import Control.Concurrent.Async
+import Control.Exception
 import Control.Monad
 import Control.Monad.Trans
 import Control.Monad.Trans.Maybe
@@ -108,41 +109,56 @@ setGlobalLoggerProvider :: (MonadIO m) => LoggerProvider -> m ()
 setGlobalLoggerProvider = liftIO . writeIORef globalLoggerProvider
 
 
+defaultShutdownTimeout :: Int
+defaultShutdownTimeout = 5_000_000
+
+
 {- | This method provides a way for provider to do any cleanup required.
 
  This will also trigger shutdowns on all internal processors.
 -}
-shutdownLoggerProvider :: (MonadIO m) => LoggerProvider -> m ()
-shutdownLoggerProvider LoggerProvider {loggerProviderProcessors} = liftIO $ do
-  asyncShutdownResults <- V.forM loggerProviderProcessors $ \processor -> do
-    logRecordProcessorShutdown processor
-  mapM_ wait asyncShutdownResults
+shutdownLoggerProvider
+  :: (MonadIO m)
+  => Maybe Int
+  -- ^ Optional timeout in microseconds, defaults to 5,000,000 (5s)
+  -> LoggerProvider
+  -> m ShutdownResult
+  -- ^ Result that denotes whether the shutdown action succeeded, failed, or timed out.
+shutdownLoggerProvider mtimeout LoggerProvider {loggerProviderProcessors} = liftIO $ do
+  mresult <-
+    timeout (fromMaybe defaultShutdownTimeout mtimeout) $
+      takeWorstShutdownResult
+        <$> forConcurrently
+          loggerProviderProcessors
+          (handle shutdownErrorHandler . logRecordProcessorShutdown)
+
+  case mresult of
+    Nothing -> pure ShutdownTimeout
+    Just res -> pure res
+
+
+defaultForceFlushTimeout :: Int
+defaultForceFlushTimeout = 5_000_000
 
 
 {- | This method provides a way for provider to immediately export all @LogRecord@s that have not yet
  been exported for all the internal processors.
 -}
 forceFlushLoggerProvider
   :: (MonadIO m)
-  => LoggerProvider
-  -> Maybe Int
+  => Maybe Int
   -- ^ Optional timeout in microseconds, defaults to 5,000,000 (5s)
+  -> LoggerProvider
   -> m FlushResult
   -- ^ Result that denotes whether the flush action succeeded, failed, or timed out.
-forceFlushLoggerProvider LoggerProvider {loggerProviderProcessors} mtimeout = liftIO $ do
-  jobs <- V.forM loggerProviderProcessors $ \processor -> async $ do
-    logRecordProcessorForceFlush processor
+forceFlushLoggerProvider mtimeout LoggerProvider {loggerProviderProcessors} = liftIO $ do
   mresult <-
-    timeout (fromMaybe 5_000_000 mtimeout) $
-      V.foldM
-        ( \status action -> do
-            res <- waitCatch action
-            pure $! case res of
-              Left _err -> FlushError
-              Right _ok -> status
-        )
-        FlushSuccess
-        jobs
+    timeout (fromMaybe defaultForceFlushTimeout mtimeout) $
+      takeWorstFlushResult
+        <$> forConcurrently
+          loggerProviderProcessors
+          (handle flushErrorHandler . logRecordProcessorForceFlush)
+
   case mresult of
     Nothing -> pure FlushTimeout
     Just res -> pure res

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

@@ -24,7 +24,6 @@ module OpenTelemetry.Internal.Logs.Types (
 ) where
 
 import Control.Concurrent (MVar, newMVar, withMVar)
-import Control.Concurrent.Async
 import Data.Function (on)
 import Data.HashMap.Strict (HashMap)
 import qualified Data.HashMap.Strict as H
@@ -33,7 +32,7 @@ import Data.Text (Text)
 import Data.Vector (Vector)
 import OpenTelemetry.Common (Timestamp, TraceFlags)
 import OpenTelemetry.Context.Types (Context)
-import OpenTelemetry.Internal.Common.Types (ExportResult, InstrumentationLibrary, ShutdownResult)
+import OpenTelemetry.Internal.Common.Types
 import OpenTelemetry.Internal.Trace.Id (SpanId, TraceId)
 import OpenTelemetry.LogAttributes
 import OpenTelemetry.Resource (MaterializedResources)
@@ -43,9 +42,9 @@ import OpenTelemetry.Resource (MaterializedResources)
 data LogRecordExporterArguments = LogRecordExporterArguments
   { logRecordExporterArgumentsExport :: Vector ReadableLogRecord -> IO ExportResult
   -- ^ See @logRecordExporterExport@ for documentation
-  , logRecordExporterArgumentsForceFlush :: IO ()
+  , logRecordExporterArgumentsForceFlush :: IO FlushResult
   -- ^ See @logRecordExporterArgumentsForceFlush@ for documentation
-  , logRecordExporterArgumentsShutdown :: IO ()
+  , logRecordExporterArgumentsShutdown :: IO ShutdownResult
   -- ^ See @logRecordExporterArgumentsShutdown@ for documentation
   }
 
@@ -97,7 +96,7 @@ the process after an invocation, but before the exporter exports the ReadlableLo
 ForceFlush SHOULD complete or abort within some timeout. ForceFlush can be implemented as a blocking API or an asynchronous API which
 notifies the caller via a callback or an event. OpenTelemetry SDK authors MAY decide if they want to make the flush timeout configurable.
 -}
-logRecordExporterForceFlush :: LogRecordExporter -> IO ()
+logRecordExporterForceFlush :: LogRecordExporter -> IO FlushResult
 logRecordExporterForceFlush = flip withMVar logRecordExporterArgumentsForceFlush . unExporter
 
 
@@ -109,7 +108,7 @@ allowed and SHOULD return a Failure result.
 Shutdown SHOULD NOT block indefinitely (e.g. if it attempts to flush the data and the destination is unavailable).
 OpenTelemetry SDK authors MAY decide if they want to make the shutdown timeout configurable.
 -}
-logRecordExporterShutdown :: LogRecordExporter -> IO ()
+logRecordExporterShutdown :: LogRecordExporter -> IO ShutdownResult
 logRecordExporterShutdown = flip withMVar logRecordExporterArgumentsShutdown . unExporter
 
 
@@ -143,7 +142,7 @@ data LogRecordProcessor = 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
   -- ^ Shuts down the processor. Called when SDK is shut down. This is an opportunity for processor to do any cleanup required.
   --
   -- Shutdown SHOULD be called only once for each LogRecordProcessor instance. After the call to Shutdown, subsequent calls to OnEmit are not allowed. SDKs SHOULD ignore these calls gracefully, if possible.
@@ -154,7 +153,7 @@ data LogRecordProcessor = LogRecordProcessor
   --
   -- Shutdown SHOULD complete or abort within some timeout. Shutdown can be implemented as a blocking API or an asynchronous API which notifies the caller via a callback or an event.
   -- OpenTelemetry SDK authors can decide if they want to make the shutdown timeout configurable.
-  , logRecordProcessorForceFlush :: IO ()
+  , logRecordProcessorForceFlush :: IO FlushResult
   -- ^ This is a hint to ensure that any tasks associated with LogRecords for which the LogRecordProcessor had already received events prior to the call to ForceFlush SHOULD be completed
   -- as soon as possible, preferably before returning from this method.
   --

api/src/OpenTelemetry/Logs/Core.hs

@@ -7,7 +7,9 @@ module OpenTelemetry.Logs.Core (
   setGlobalLoggerProvider,
   getGlobalLoggerProvider,
   shutdownLoggerProvider,
+  ShutdownResult (..),
   forceFlushLoggerProvider,
+  FlushResult (..),
 
   -- * @Logger@ operations
   InstrumentationLibrary (..),
@@ -16,10 +18,13 @@ module OpenTelemetry.Logs.Core (
 
   -- * @LogRecord@ operations
   ReadableLogRecord,
+  mkReadableLogRecord,
   ReadWriteLogRecord,
   IsReadableLogRecord (..),
   IsReadWriteLogRecord (..),
+  ImmutableLogRecord (..),
   LogRecordArguments (..),
+  emptyLogRecordArguments,
   AnyValue (..),
   ToValue (..),
   SeverityNumber (..),

api/src/OpenTelemetry/Processor/LogRecord.hs

@@ -1,6 +1,12 @@
 module OpenTelemetry.Processor.LogRecord (
   LogRecordProcessor (..),
+  FlushResult (..),
+  flushErrorHandler,
+  takeWorstFlushResult,
   ShutdownResult (..),
+  shutdownErrorHandler,
+  takeWorstShutdownResult,
+  flushResultToShutdownResult,
 ) where
 
 import OpenTelemetry.Internal.Common.Types

api/src/OpenTelemetry/Trace/Core.hs

@@ -871,7 +871,7 @@ forceFlushTracerProvider TracerProvider {..} mtimeout = liftIO $ do
         ( \status action -> do
             res <- waitCatch action
             pure $! case res of
-              Left _err -> FlushError
+              Left err -> FlushError $ pure err
               Right _ok -> status
         )
         FlushSuccess

sdk/hs-opentelemetry-sdk.cabal

@@ -105,6 +105,7 @@ test-suite hs-opentelemetry-sdk-test
   other-modules:
       OpenTelemetry.BaggageSpec
       OpenTelemetry.ContextSpec
+      OpenTelemetry.LogRecordProcessorSpec
       OpenTelemetry.ResourceSpec
       OpenTelemetry.TraceSpec
       Paths_hs_opentelemetry_sdk

sdk/src/OpenTelemetry/Processor/Batch/Span.hs

@@ -343,8 +343,8 @@ batchProcessor BatchTimeoutConfig {..} exporter = liftIO $ do
                   ShutdownTimeout
                 Just er ->
                   case er of
-                    Left _ ->
-                      ShutdownFailure
+                    Left e ->
+                      ShutdownError $ pure e
                     Right _ ->
                       ShutdownSuccess
       }