Create an AsyncStream based OTel Logger w/ HB2 example

Examples/Server/Sources/ServerExample/ServerExample.swift

@@ -16,18 +16,14 @@ import Logging
 import Metrics
 @_spi(Metrics) import OTel
 @_spi(Metrics) import OTLPGRPC
+@_spi(Logging) import OTel
+@_spi(Logging) import OTLPGRPC
+import ServiceLifecycle
 import Tracing
 
 @main
 enum ServerMiddlewareExample {
     static func main() async throws {
-        // Bootstrap the logging backend with the OTel metadata provider which includes span IDs in logging messages.
-        LoggingSystem.bootstrap { label in
-            var handler = StreamLogHandler.standardError(label: label, metadataProvider: .otel)
-            handler.logLevel = .trace
-            return handler
-        }
-
         // Configure OTel resource detection to automatically apply helpful attributes to events.
         let environment = OTelEnvironment.detected()
         let resourceDetection = OTelResourceDetection(detectors: [
@@ -51,14 +47,28 @@ enum ServerMiddlewareExample {
         )
         MetricsSystem.bootstrap(OTLPMetricsFactory(registry: registry))
 
+        // Bootstrap the logging backend with the OTel metadata provider which includes span IDs in logging messages.
+        let logExporter = try OTLPGRPCLogExporter(
+            configuration: .init(environment: environment)
+        )
+
+        let logger = OTelStreamingLogger(
+            resource: resource,
+            exporter: logExporter,
+            logLevel: .trace
+        )
+        LoggingSystem.bootstrap { label in
+            return logger
+        }
+
         // Bootstrap the tracing backend to export traces periodically in OTLP/gRPC.
-        let exporter = try OTLPGRPCSpanExporter(configuration: .init(environment: environment))
-        let processor = OTelBatchSpanProcessor(exporter: exporter, configuration: .init(environment: environment))
+        let traceExporter = try OTLPGRPCSpanExporter(configuration: .init(environment: environment))
+        let traceProcessor = OTelBatchSpanProcessor(exporter: traceExporter, configuration: .init(environment: environment))
         let tracer = OTelTracer(
             idGenerator: OTelRandomIDGenerator(),
             sampler: OTelConstantSampler(isOn: true),
             propagator: OTelW3CPropagator(),
-            processor: processor,
+            processor: traceProcessor,
             environment: environment,
             resource: resource
         )
@@ -69,11 +79,14 @@ enum ServerMiddlewareExample {
         router.middlewares.add(HBTracingMiddleware())
         router.middlewares.add(HBMetricsMiddleware())
         router.middlewares.add(HBLogRequestsMiddleware(.info))
-        router.get("hello") { _, _ in "hello" }
+        router.get("hello") { _, context in
+            context.logger.info("Someone visited me, at last!")
+            return "hello"
+        }
         var app = HBApplication(router: router)
 
-        // Add the tracer lifecycle service to the HTTP server service group and start the application.
-        app.addServices(metrics, tracer)
+        // Add the logger, metrics and tracer lifecycle services to the HTTP server service group and start the application.
+        app.addServices(logger, metrics, tracer)
         try await app.runService()
     }
 }

Examples/Server/docker/otel-collector-config.yaml

@@ -5,8 +5,8 @@ receivers:
         endpoint: "otel-collector:4317"
 
 exporters:
-  debug:  # Data sources: traces, metrics, logs
-    verbosity: detailed
+  logging:
+    loglevel: debug
 
   prometheus:  # Data sources: metrics
     endpoint: "otel-collector:7070"
@@ -24,6 +24,11 @@ service:
       exporters: [prometheus, debug]
     traces:
       receivers: [otlp]
-      exporters: [otlp/jaeger, debug]
+      exporters: [otlp/jaeger]
+
+    logs:
+      receivers: [otlp]
+      processors: []
+      exporters: [logging]
 
 # yaml-language-server: $schema=https://raw.githubusercontent.com/srikanthccv/otelcol-jsonschema/main/schema.json

Sources/OTLPGRPC/Logging/OTLPGRPCLogEntryExporter.swift

@@ -0,0 +1,175 @@
+//===----------------------------------------------------------------------===//
+//
+// This source file is part of the Swift OTel open source project
+//
+// Copyright (c) 2024 Moritz Lang and the Swift OTel project authors
+// Licensed under Apache License v2.0
+//
+// See LICENSE.txt for license information
+//
+// SPDX-License-Identifier: Apache-2.0
+//
+//===----------------------------------------------------------------------===//
+
+import GRPC
+import Logging
+import NIO
+import NIOHPACK
+import NIOSSL
+@_spi(Logging) import OTel
+@_spi(Logging) import OTLPCore
+
+/// Exports logs to an OTel collector using OTLP/gRPC.
+@_spi(Logging)
+public final class OTLPGRPCLogEntryExporter: OTelLogEntryExporter {
+    private let configuration: OTLPGRPCLogEntryExporterConfiguration
+    private let connection: ClientConnection
+    private let client: Opentelemetry_Proto_Collector_Logs_V1_LogsServiceAsyncClient
+    private let logger = Logger(label: String(describing: OTLPGRPCLogEntryExporter.self))
+
+    public init(
+        configuration: OTLPGRPCLogEntryExporterConfiguration,
+        group: EventLoopGroup = MultiThreadedEventLoopGroup.singleton,
+        requestLogger: Logger = ._otelDisabled,
+        backgroundActivityLogger: Logger = ._otelDisabled
+    ) {
+        self.configuration = configuration
+
+        var connectionConfiguration = ClientConnection.Configuration.default(
+            target: .host(configuration.endpoint.host, port: configuration.endpoint.port),
+            eventLoopGroup: group
+        )
+
+        if configuration.endpoint.isInsecure {
+            logger.debug("Using insecure connection.", metadata: [
+                "host": "\(configuration.endpoint.host)",
+                "port": "\(configuration.endpoint.port)",
+            ])
+        }
+
+        // TODO: Support OTEL_EXPORTER_OTLP_CERTIFICATE
+        // TODO: Support OTEL_EXPORTER_OTLP_CLIENT_KEY
+        // TODO: Support OTEL_EXPORTER_OTLP_CLIENT_CERTIFICATE
+
+        var headers = configuration.headers
+        if !headers.isEmpty {
+            logger.trace("Configured custom request headers.", metadata: [
+                "keys": .array(headers.map { "\($0.name)" }),
+            ])
+        }
+        headers.replaceOrAdd(name: "user-agent", value: "OTel-OTLP-Exporter-Swift/\(OTelLibrary.version)")
+
+        connectionConfiguration.backgroundActivityLogger = backgroundActivityLogger
+        connection = ClientConnection(configuration: connectionConfiguration)
+
+        client = Opentelemetry_Proto_Collector_Logs_V1_LogsServiceAsyncClient(
+            channel: connection,
+            defaultCallOptions: .init(customMetadata: headers, logger: requestLogger)
+        )
+    }
+
+    public func export(_ batch: some Collection<OTelLogEntry> & Sendable) async throws {
+        if case .shutdown = connection.connectivity.state {
+            throw OTelLogEntryExporterAlreadyShutDownError()
+        }
+
+        guard !batch.isEmpty else { return }
+
+        let request = Opentelemetry_Proto_Collector_Logs_V1_ExportLogsServiceRequest.with { request in
+            request.resourceLogs = [
+                Opentelemetry_Proto_Logs_V1_ResourceLogs.with { resourceLog in
+                    resourceLog.scopeLogs = [
+                        Opentelemetry_Proto_Logs_V1_ScopeLogs.with { scopeLog in
+                            scopeLog.logRecords = batch.map { log in
+                                Opentelemetry_Proto_Logs_V1_LogRecord.with { logRecord in
+                                    logRecord.timeUnixNano = log.timeNanosecondsSinceEpoch
+                                    logRecord.observedTimeUnixNano = log.timeNanosecondsSinceEpoch
+                                    logRecord.severityNumber = switch log.level {
+                                    case .trace: .trace
+                                    case .debug: .debug
+                                    case .info: .info
+                                    case .notice: .info4
+                                    case .warning: .warn
+                                    case .error: .error
+                                    case .critical: .fatal
+                                    }
+                                    logRecord.severityText = switch log.level {
+                                    case .trace: "TRACE"
+                                    case .debug: "DEUG"
+                                    case .info: "INFO"
+                                    case .notice: "NOTICE"
+                                    case .warning: "WARNING"
+                                    case .error: "ERROR"
+                                    case .critical: "CRITICAL"
+                                    }
+                                    if let metadata = log.metadata {
+                                        logRecord.attributes = .init(metadata)
+                                    }
+                                    logRecord.body = .with { body in
+                                        body.stringValue = log.body
+                                    }
+                                }
+                            }
+                        }
+                    ]
+                }
+            ]
+        }
+
+        _ = try await client.export(request)
+    }
+
+    public func forceFlush() async throws {}
+
+    public func shutdown() async {
+        let promise = connection.eventLoop.makePromise(of: Void.self)
+
+        connection.closeGracefully(deadline: .now() + .milliseconds(500), promise: promise)
+
+        try? await promise.futureResult.get()
+    }
+}
+
+@_spi(Logging)
+extension [Opentelemetry_Proto_Common_V1_KeyValue] {
+    package init(_ metadata: Logger.Metadata) {
+        self = metadata.map { key, value in
+            return .with { attribute in
+                attribute.key = key
+                attribute.value = .init(value)
+            }
+        }
+    }
+}
+
+@_spi(Logging)
+extension Opentelemetry_Proto_Common_V1_KeyValueList {
+    package init(_ metadata: Logger.Metadata) {
+        self = .with { keyValueList in
+            keyValueList.values = .init(metadata)
+        }
+    }
+}
+
+@_spi(Logging)
+extension Opentelemetry_Proto_Common_V1_AnyValue {
+    package init(_ value: Logger.Metadata.Value) {
+        self = .with { attributeValue in
+            attributeValue.value = switch value {
+            case .string(let string): .stringValue(string)
+            case .stringConvertible(let stringConvertible): .stringValue(stringConvertible.description)
+            case .dictionary(let metadata): .kvlistValue(.init(metadata))
+            case .array(let values): .arrayValue(.init(values))
+            }
+        }
+    }
+}
+
+@_spi(Logging)
+extension Opentelemetry_Proto_Common_V1_ArrayValue {
+    package init(_ values: [Logger.Metadata.Value]) {
+        self = .with { valueList in
+            valueList.values = values.map(Opentelemetry_Proto_Common_V1_AnyValue.init)
+        }
+    }
+}

Sources/OTLPGRPC/Logging/OTLPGRPCLogEntryExporterConfiguration.swift

@@ -0,0 +1,70 @@
+//===----------------------------------------------------------------------===//
+//
+// This source file is part of the Swift OTel open source project
+//
+// Copyright (c) 2024 Moritz Lang and the Swift OTel project authors
+// Licensed under Apache License v2.0
+//
+// See LICENSE.txt for license information
+//
+// SPDX-License-Identifier: Apache-2.0
+//
+//===----------------------------------------------------------------------===//
+
+import NIOHPACK
+import OTel
+
+@_spi(Logging)
+public struct OTLPGRPCLogEntryExporterConfiguration: Sendable {
+    let endpoint: OTLPGRPCEndpoint
+    let headers: HPACKHeaders
+
+    /// Create a configuration for an ``OTLPGRPCLogEntryExporter``.
+    ///
+    /// - Parameters:
+    ///   - environment: The environment variables.
+    ///   - endpoint: An optional endpoint string that takes precedence over any environment values. Defaults to `localhost:4317` if `nil`.
+    ///   - shouldUseAnInsecureConnection: Whether to use an insecure connection in the absence of a scheme inside an endpoint configuration value.
+    ///   - headers: Optional headers that take precedence over any headers configured via environment values.
+    public init(
+        environment: OTelEnvironment,
+        endpoint: String? = nil,
+        shouldUseAnInsecureConnection: Bool? = nil,
+        headers: HPACKHeaders? = nil
+    ) throws {
+        let shouldUseAnInsecureConnection = try environment.value(
+            programmaticOverride: shouldUseAnInsecureConnection,
+            signalSpecificKey: "OTEL_EXPORTER_OTLP_LOGGING_INSECURE",
+            sharedKey: "OTEL_EXPORTER_OTLP_INSECURE"
+        ) ?? false
+
+        let programmaticEndpoint: OTLPGRPCEndpoint? = try {
+            guard let endpoint else { return nil }
+            return try OTLPGRPCEndpoint(urlString: endpoint, isInsecure: shouldUseAnInsecureConnection)
+        }()
+
+        self.endpoint = try environment.value(
+            programmaticOverride: programmaticEndpoint,
+            signalSpecificKey: "OTEL_EXPORTER_OTLP_LOGGING_ENDPOINT",
+            sharedKey: "OTEL_EXPORTER_OTLP_ENDPOINT",
+            transformValue: { value in
+                do {
+                    return try OTLPGRPCEndpoint(urlString: value, isInsecure: shouldUseAnInsecureConnection)
+                } catch {
+                    // TODO: Log
+                    return nil
+                }
+            }
+        ) ?? .default
+
+        self.headers = try environment.value(
+            programmaticOverride: headers,
+            signalSpecificKey: "OTEL_EXPORTER_OTLP_LOGGING_HEADERS",
+            sharedKey: "OTEL_EXPORTER_OTLP_HEADERS",
+            transformValue: { value in
+                guard let keyValuePairs = OTelEnvironment.headers(parsingValue: value) else { return nil }
+                return HPACKHeaders(keyValuePairs)
+            }
+        ) ?? [:]
+    }
+}

Sources/OTel/Logging/Exporting/OTelLogEntryExporter.swift

@@ -0,0 +1,40 @@
+//===----------------------------------------------------------------------===//
+//
+// This source file is part of the Swift OTel open source project
+//
+// Copyright (c) 2024 Moritz Lang and the Swift OTel project authors
+// Licensed under Apache License v2.0
+//
+// See LICENSE.txt for license information
+//
+// SPDX-License-Identifier: Apache-2.0
+//
+//===----------------------------------------------------------------------===//
+
+/// A span exporter receives batches of processed spans to export them, e.g. by sending them over the network.
+///
+/// [OpenTelemetry specification: Span exporter](https://github.com/open-telemetry/opentelemetry-specification/blob/v1.20.0/specification/trace/sdk.md#span-exporter)
+@_spi(Logging)
+public protocol OTelLogEntryExporter: Sendable {
+    /// Export the given batch of spans.
+    ///
+    /// - Parameter batch: A batch of spans to export.
+    func export(_ batch: some Collection<OTelLogEntry> & Sendable) async throws
+
+    /// Force the span exporter to export any previously received spans as soon as possible.
+    func forceFlush() async throws
+
+    /// Shut down the span exporter.
+    ///a
+    /// This method gives exporters a chance to wrap up existing work such as finishing in-flight exports while not allowing new ones anymore.
+    /// Once this method returns, the exporter is to be considered shut down and further invocations of ``export(_:)``
+    /// are expected to fail.
+    func shutdown() async
+}
+
+/// An error indicating that a given exporter has already been shut down while receiving an additional batch of spans to export.
+@_spi(Logging)
+public struct OTelLogEntryExporterAlreadyShutDownError: Error {
+    /// Initialize the error.
+    public init() {}
+}