diff --git a/CHANGELOG.md b/CHANGELOG.md index 9fcc6dd70..6981b8dc3 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -16,10 +16,17 @@ The `Unreleased` section name is replaced by the expected version of next releas - `Equinox.DynamoStore`: Update `AWSSDK` dependencies to 4.x (knock-on effect of updating `FSharp.AWS.DynamoDB` dependency to `0.13.0-beta`) [#480](https://github.com/jet/equinox/pull/480) :pray: [@njlr](https://github.com/njlr) - `Equinox.SqlStreamStore.MsSql`: Update `System.IdentityModel.Tokens.Jwt` to remove transitive vulnerable defaults [#489](https://github.com/jet/equinox/pull/489) - `Equinox.EventStore`: Correct handling for `WrongExpectedVersion` when stream empty [#493](https://github.com/jet/equinox/pull/493) +- `Equinox.EventStore`, `Equinox.EventStoreDb`, `Equinox.SqlStreamStore`: Add `System.Diagnostics.Activity`-based OpenTelemetry tracing (following `Equinox.MessageDb` patterns). Store operations emit child Activities for granular metric capture via `ActivityListener` +- `Equinox.SqlStreamStore.Postgres`, `Equinox.SqlStreamStore.MsSql`, `Equinox.SqlStreamStore.MySql`: Use unconditional `ProjectReference` to `Equinox.SqlStreamStore` (previously conditional Debug/Release) + +### Removed + +- `Equinox.EventStore`: Removed `Logger` DU (`SerilogVerbose`/`SerilogNormal`/`CustomVerbose`/`CustomNormal`), `SerilogAdapter` type, and `?log: Logger` optional parameter from `EventStoreConnector` [**breaking**] ### Added - `Equinox.CosmosStore.Linq`: Add LINQ querying support for Indexed `u`nfolds [#450](https://github.com/jet/equinox/pull/450) +- `OtelToSerilogBridge`: Shared helper mapping OpenTelemetry Activity completions to equivalent Serilog structured log writes, providing a migration path for consumers with Serilog-based metric pipelines ## [4.1.2] - 2026-02-04 diff --git a/adr/0001-replace-serilog-with-opentelemetry-in-store-adapters.md b/adr/0001-replace-serilog-with-opentelemetry-in-store-adapters.md new file mode 100644 index 000000000..9b36d362b --- /dev/null +++ b/adr/0001-replace-serilog-with-opentelemetry-in-store-adapters.md @@ -0,0 +1,122 @@ +# ADR-0001: Replace Serilog Metric Capture with OpenTelemetry in Store Adapters + +## Status + +Accepted + +## Context + +Equinox store adapter packages (`Equinox.EventStore`, `Equinox.EventStoreDb`, `Equinox.SqlStreamStore`, and the `SqlStreamStore.*` database-specific packages) currently use Serilog as the primary mechanism for both structured logging and operational metric capture. The integration tests (`StoreIntegration.fs`, shared across all affected stores via conditional compilation) extract per-operation metrics by intercepting Serilog `LogEvent` emissions through a custom `ILogEventSink` (`LogCaptureBuffer`), then pattern-matching on `Log.Metric` discriminated union values embedded as `ScalarValue` properties. + +This approach has several drawbacks: + +1. **Tight coupling to Serilog internals** — Test assertions depend on the internal shape of Serilog `LogEvent.Properties`, `ScalarValue`, and `ILogEventSink`. This couples test infrastructure to a specific logging library's implementation details. + +2. **No standard telemetry integration** — Modern .NET observability is built on `System.Diagnostics.Activity` and the OpenTelemetry ecosystem. Consumers using OTel collectors, Jaeger, Zipkin, or other APM tools cannot benefit from store-level operation tracing without custom bridging. + +3. **Inconsistency across stores** — `Equinox.MessageDb` has already adopted `System.Diagnostics.Activity`-based tracing via `Activity.Current` tags (using the core `Equinox` `ActivitySource`), while the other stores have not. The integration test for MessageDb already sets up a `TracerProvider` with OTLP export. + +4. **Public API surface** — `Equinox.EventStore` exposes a `Logger` discriminated union (`SerilogVerbose | SerilogNormal | CustomVerbose | CustomNormal`) and a `SerilogAdapter` type as public API, creating a hard dependency on `Serilog.ILogger` at the connector level. The other stores do not expose Serilog in their public APIs but use it extensively internally. + +The core `Equinox` package will continue to depend on Serilog for the `ICategory.Load/Sync` interface and `Category.Stream` — this ADR does not propose changing that. Store packages will continue to receive `Serilog.ILogger` through these interfaces for human-readable structured logging. + +## Decision + +We will adopt the following changes, following and extending the pattern established by `Equinox.MessageDb`: + +### 1. Add per-store `ActivitySource` via shared `Tracing.fs` + +A single `Tracing.fs` file (physically in `src/Equinox.EventStoreDb/`) is linked into EventStore and SqlStreamStore packages via ``, following the same shared-file-with-`#if`-conditionals pattern used by `StoreIntegration.fs` and `Infrastructure.fs` in the test projects. The existing `DefineConstants` from the test projects (`STORE_EVENTSTOREDB`, `STORE_EVENTSTORE_LEGACY`) are reused — the same constants are added to the corresponding src `.fsproj` files. `SqlStreamStore` uses the `#else` fallback and requires no new constant. + +The shared file defines: +- A per-store `ActivitySource` selected via `#if STORE_EVENTSTOREDB` / `#elif STORE_EVENTSTORE_LEGACY` / `#else` blocks (e.g., `new ActivitySource("Equinox.EventStoreDb")`) +- Shared `ActivityExtensions` for domain-relevant Activity tags (shared across all stores) +- The `#else` branch provides the `Equinox.SqlStreamStore` namespace and ActivitySource name + +`Equinox.MessageDb`'s existing `Tracing.fs` remains separate — it references `Equinox.MessageDb.Core.StreamVersion`, a MessageDb-specific type. + +### 2. Instrument store operations with child Activities + +Each discrete store operation (read slice, read batch, write, read-last) starts a child Activity from the store's `ActivitySource`. Tags carry operational metadata: +- `eqx.expected_version`, `eqx.last_version`, `eqx.count`, `eqx.bytes` +- `eqx.conflict`, `eqx.event_types` (on write conflict) +- `eqx.direction` (forward/backward for reads) +- `eqx.batch_size`, `eqx.batches`, `eqx.start_position` + +These child Activities sit beneath the parent `Load`/`Sync` activities created by `Equinox.Core.Tracing.source` in `Category.fs`. Additionally, aggregate metrics are added to `Activity.Current` (the parent) via `IncMetric`, following the MessageDb pattern. + +### 3. Replace test metric capture with OTel Activity capture + +A new shared test helper (`ActivityCapture`) uses `System.Diagnostics.ActivityListener` to capture completed Activities from `Equinox.*` sources. The existing `EsAct` discriminated union is retained for assertion readability, but values are derived from Activity operation names and tags rather than Serilog `LogEvent` properties. + +### 4. Remove Serilog-specific public APIs + +- **`Equinox.EventStore`**: Remove the `Logger` DU, `SerilogAdapter` type, and `?log: Logger` parameter from `EventStoreConnector`. Users needing to configure the EventStore client logger can use the existing `?custom: ConnectionSettingsBuilder -> ConnectionSettingsBuilder` parameter. +- **`Equinox.EventStoreDb`**, **`Equinox.SqlStreamStore`**: No Serilog-specific public APIs exist; no removals needed. + +### 5. Provide OTel-to-Serilog bridge helper + +A shared helper (`OtelToSerilogBridge`) listens to Activity completions and emits equivalent Serilog log writes, allowing consumers migrating from Serilog-based metric consumption to see familiar output. + +### 6. Per-metric equivalence tests (Serilog ↔ OTel) + +For each `Log.Metric` case in each store (WriteSuccess, WriteConflict, Slice, Batch, ReadLast, etc.), add integration tests that perform dual capture — both Serilog `LogEvent` via `LogCaptureBuffer` and OTel `Activity` via `ActivityCapture` — and assert: +- Both yield the same `EsAct` classification +- Key metric values match (stream name, event count, bytes, direction, conflict status, version) +- The `OtelToSerilogBridge` produces equivalent output + +This provides per-event proof that every Serilog-derived assertion in `StoreIntegration.fs` has an equivalent OTel-derived assertion. + +### 7. Comment out `PackageValidationBaselineVersion` + +Since removing public API surface is a breaking change, `PackageValidationBaselineVersion` is commented out in the `.fsproj` files of all affected packages until a new baseline is established. + +### 8. Internal logging retained + +The `Log` module (including `Measurement`, `Metric`, `MetricEvent`, `InternalMetrics`, `prop`, `event`, `propEventData`, `propResolvedEvents`) remains in each store package. Internal `log.Information(...)` calls continue to emit structured Serilog log events. The `ILogger` flowing from `ICategory.Load/Sync` is still used. We are adding OTel tracing as a parallel telemetry path, not removing Serilog logging. + +## Consequences + +### Positive + +- **Standard observability**: Store operations appear as proper OTel spans in APM tools, with parent-child relationships (Decider → Load/Sync → ReadBatch/Write). +- **Consistent tracing across stores**: All stores (MessageDb, EventStoreDb, EventStore, SqlStreamStore) follow the same Activity-based pattern. +- **Decoupled test assertions**: Integration tests no longer depend on Serilog internals for metric validation. +- **Reduced public API surface**: Removing the `Logger` DU from `Equinox.EventStore` simplifies the connector API. + +### Negative + +- **Breaking change**: Removing `Logger` type from `Equinox.EventStore` requires consumers using `SerilogVerbose`/`SerilogNormal` to migrate to the `?custom` parameter or remove the logger argument. This necessitates a major or minor version bump with appropriate migration guidance. +- **Dual telemetry paths**: Both Serilog logging and OTel tracing coexist in the store code, adding some code volume. This is mitigated by the fact that they serve different purposes (human-readable logs vs. structured traces). +- **Package validation disabled**: Commenting out `PackageValidationBaselineVersion` temporarily removes automated API compatibility checking until a new baseline is set. + +### Neutral + +- **No impact on CosmosStore/DynamoStore**: These stores are not affected by this change. They may adopt the same pattern independently in the future. +- **TestOutput.fs unchanged**: The shared test output helper (used by all test projects) continues to use Serilog for rendering test output to xUnit `ITestOutputHelper`. This is orthogonal to the metric capture mechanism. +- **`Log.InternalMetrics.Stats.LogSink`**: The existing Serilog-based metric aggregation helper remains available for consumers. An OTel-based equivalent may be added separately. + +## References + +- `src/Equinox.MessageDb/Tracing.fs` — Established pattern for store-specific Activity extensions +- `src/Equinox/Tracing.fs` — Core `ActivitySource("Equinox")` and shared `ActivityExtensions` +- `src/Equinox/Category.fs` — Parent `Load`/`Sync` Activity creation +- `tests/Equinox.EventStoreDb.Integration/StoreIntegration.fs` — Shared integration tests with conditional compilation +- [CNCF OpenTelemetry .NET documentation](https://opentelemetry.io/docs/languages/dotnet/) +- [System.Diagnostics.Activity API](https://learn.microsoft.com/en-us/dotnet/api/system.diagnostics.activity) + +## Implementation Notes + +### F# preprocessor limitations + +F# does not support `#elif` for namespace declarations. The shared `Tracing.fs` uses `[] module internal` with separate `#if`/`#endif` blocks (not `#elif`) to select the namespace. + +### Test isolation via AsyncLocal + +`ActivityListener` registered via `ActivitySource.AddActivityListener()` is process-global — unlike Serilog's per-logger `ILogEventSink`, ALL listeners receive ALL activities from matching sources. When xUnit runs test classes in parallel, activities from one class's tests bleed into another's captures. + +The solution uses a single static `ActivityListener` per type with `AsyncLocal>` dispatch: the listener callback reads `AsyncLocal.Value` to find the current test's ops list, providing per-async-context isolation without process-level contention. + +### SqlStreamStore unconditional ProjectReference + +The `Equinox.SqlStreamStore.Postgres`, `.MsSql`, and `.MySql` wrapper projects previously used conditional references (`ProjectReference` in Debug, `PackageReference` in Release) to `Equinox.SqlStreamStore`. This was changed to unconditional `ProjectReference` to ensure locally-built `Tracing.fs` changes are always picked up. The previous pattern loaded the NuGet-published version in Release mode, which did not contain the new tracing code. diff --git a/samples/Store/Integration/OtelToSerilogBridge.fs b/samples/Store/Integration/OtelToSerilogBridge.fs new file mode 100644 index 000000000..478237734 --- /dev/null +++ b/samples/Store/Integration/OtelToSerilogBridge.fs @@ -0,0 +1,41 @@ +namespace global + +open System +open System.Diagnostics + +/// Maps completed store-specific Activities to equivalent Serilog log writes, +/// providing a migration path for consumers with existing Serilog-based metric pipelines. +/// Usage: Instantiate with a Serilog.ILogger, and all store Activities will be logged with +/// structured properties matching the original Serilog-based Log.Metric output. +type OtelToSerilogBridge(log: Serilog.ILogger) = + let listener = new ActivityListener( + ShouldListenTo = (fun source -> source.Name.StartsWith("Equinox.", StringComparison.Ordinal)), + Sample = (fun _ -> ActivitySamplingResult.AllDataAndRecorded), + ActivityStopped = (fun act -> + let tag name = act.GetTagItem(name) + let tagStrArr name = match tag name :?> string[] with null -> null | s -> s + let tagInt name = match tag name with :? int as i -> i | _ -> 0 + let count = tagInt "eqx.count" + let bytes = tagInt "eqx.bytes" + let op = act.OperationName + match op with + | "Append" -> + log.Information("Otel {Source} {Action:l} count={Count} bytes={Bytes}", + act.Source.Name, op, count, bytes) + | "AppendConflict" -> + let eventTypes = tagStrArr "eqx.event_types" + log.Information("Otel {Source} {Action:l} conflict eventTypes={EventTypes}", + act.Source.Name, op, eventTypes) + | "SliceForward" | "SliceBackward" -> + log.Information("Otel {Source} {Action:l} count={Count} bytes={Bytes}", + act.Source.Name, op, count, bytes) + | "BatchForward" | "BatchBackward" -> + let batches = tagInt "eqx.batches" + log.Information("Otel {Source} {Action:l} count={Count} bytes={Bytes} batches={Batches}", + act.Source.Name, op, count, bytes, batches) + | "ReadLast" -> + log.Information("Otel {Source} {Action:l} count={Count} bytes={Bytes}", + act.Source.Name, op, count, bytes) + | _ -> ())) + do ActivitySource.AddActivityListener(listener) + interface IDisposable with member _.Dispose() = listener.Dispose() diff --git a/src/Equinox.EventStore/Equinox.EventStore.fsproj b/src/Equinox.EventStore/Equinox.EventStore.fsproj index 82cd39f5d..713c50a40 100644 --- a/src/Equinox.EventStore/Equinox.EventStore.fsproj +++ b/src/Equinox.EventStore/Equinox.EventStore.fsproj @@ -2,12 +2,14 @@ net6.0 - 4.0.0 + + $(DefineConstants);STORE_EVENTSTORE_LEGACY + diff --git a/src/Equinox.EventStore/EventStore.fs b/src/Equinox.EventStore/EventStore.fs index c9db6daa8..7752bd9f4 100755 --- a/src/Equinox.EventStore/EventStore.fs +++ b/src/Equinox.EventStore/EventStore.fs @@ -1,10 +1,12 @@ namespace Equinox.EventStore open Equinox.Core +open Equinox.Core.Tracing open EventStore.ClientAPI open Serilog // NB must shadow EventStore.ClientAPI.ILogger open System open System.Collections.Generic +open System.Diagnostics type EventBody = ReadOnlyMemory @@ -145,20 +147,26 @@ module private Write = let private writeEventsLogged (conn: IEventStoreConnection) (streamName: string) (version: int64) (events: EventData[]) (log: ILogger) : Task = task { + let act = Activity.Current let log = if (not << log.IsEnabled) Events.LogEventLevel.Debug then log else log |> Log.propEventData "Json" events let bytes, count = eventDataBytes events, events.Length let log = log |> Log.prop "bytes" bytes + if act <> null then act.AddExpectedVersion(version).IncMetric(count, bytes) |> ignore let writeLog = log |> Log.prop "stream" streamName |> Log.prop "expectedVersion" version |> Log.prop "count" count let! t, result = (fun _ct -> writeEventsAsync writeLog conn streamName version events) |> Stopwatch.time CancellationToken.None let reqMetric: Log.Measurement = { stream = streamName; interval = t; bytes = bytes; count = count} let resultLog, evt = match result, reqMetric with | EsSyncResult.Written x, m -> + if act <> null then act.SetStatus(ActivityStatusCode.Ok).AddTag("eqx.new_version", x.NextExpectedVersion) |> ignore log |> Log.prop "nextExpectedVersion" x.NextExpectedVersion |> Log.prop "logPosition" x.LogPosition, Log.WriteSuccess m | EsSyncResult.Conflict actualVersion, m -> + let eventTypes = [| for x in events -> x.Type |] + if act <> null then act.RecordConflict().AddTag("eqx.event_types", eventTypes) |> ignore log |> Log.prop "actualVersion" actualVersion, Log.WriteConflict m (resultLog |> Log.event evt).Information("Ges{action:l} count={count} conflict={conflict}", "Write", events.Length, match evt with Log.WriteConflict _ -> true | _ -> false) + emitStoreOp (match evt with Log.WriteConflict _ -> "AppendConflict" | _ -> "Append") return result } let writeEvents (log: ILogger) retryPolicy (conn: IEventStoreConnection) (streamName: string) (version: int64) (events: EventData[]) @@ -178,13 +186,16 @@ module private Read = let (|ResolvedEventLen|) (x: ResolvedEvent) = match x.Event.Data, x.Event.Metadata with Log.BlobLen bytes, Log.BlobLen metaBytes -> bytes + metaBytes let private loggedReadSlice conn streamName direction batchSize startPos (log: ILogger): Task = task { + let act = Activity.Current let! t, slice = (fun _ct -> readSliceAsync conn streamName direction batchSize startPos) |> Stopwatch.time CancellationToken.None let bytes, count = slice.Events |> Array.sumBy (|ResolvedEventLen|), slice.Events.Length let reqMetric: Log.Measurement = { stream = streamName; interval = t; bytes = bytes; count = count} let evt = Log.Slice (direction, reqMetric) + if act <> null then act.IncMetric(count, bytes).AddDirection(string direction) |> ignore let log = if (not << log.IsEnabled) Events.LogEventLevel.Debug then log else log |> Log.propResolvedEvents "Json" slice.Events (log |> Log.prop "startPos" startPos |> Log.prop "bytes" bytes |> Log.event evt).Information("Ges{action:l} count={count} version={version}", "Read", count, slice.LastEventNumber) + emitStoreOp (match direction with Direction.Forward -> "SliceForward" | Direction.Backward -> "SliceBackward") return slice } let private readBatches (log: ILogger) (readSlice: int64 -> ILogger -> Task) @@ -211,14 +222,17 @@ module private Read = let resolvedEventBytes events = events |> Array.sumBy (|ResolvedEventLen|) let logBatchRead direction streamName t events batchSize version (log: ILogger) = + let act = Activity.Current let bytes, count = resolvedEventBytes events, events.Length let reqMetric: Log.Measurement = { stream = streamName; interval = t; bytes = bytes; count = count} let batches = (events.Length - 1) / batchSize + 1 let action = match direction with Direction.Forward -> "LoadF" | Direction.Backward -> "LoadB" + if act <> null then act.IncMetric(count, bytes).AddLastVersion(version).AddBatches(batches).AddDirection(string direction) |> ignore let evt = Log.Metric.Batch (direction, batches, reqMetric) (log |> Log.prop "bytes" bytes |> Log.event evt).Information( "Ges{action:l} stream={stream} count={count}/{batches} version={version}", action, streamName, count, batches, version) + emitStoreOp (match direction with Direction.Forward -> "BatchForward" | Direction.Backward -> "BatchBackward") let loadForwardsFrom (log: ILogger) retryPolicy conn batchSize maxPermittedBatchReads streamName startPosition: Task = task { let mergeBatches (batches: IAsyncEnumerable) = task { @@ -491,28 +505,6 @@ type EventStoreCategory<'event, 'state, 'req> = let cat = StoreCategory<'event, 'state, 'req>(context, codec, fold, initial, access) |> Caching.apply Token.isStale caching { inherit Equinox.Category<'event, 'state, 'req>(name, cat); __ = () }; val private __: unit -type private SerilogAdapter(log: ILogger) = - interface EventStore.ClientAPI.ILogger with - member _.Debug(format: string, args: obj []) = log.Debug(format, args) - member _.Debug(ex: exn, format: string, args: obj []) = log.Debug(ex, format, args) - member _.Info(format: string, args: obj []) = log.Information(format, args) - member _.Info(ex: exn, format: string, args: obj []) = log.Information(ex, format, args) - member _.Error(format: string, args: obj []) = log.Error(format, args) - member _.Error(ex: exn, format: string, args: obj []) = log.Error(ex, format, args) - -[] -type Logger = - | SerilogVerbose of ILogger - | SerilogNormal of ILogger - | CustomVerbose of EventStore.ClientAPI.ILogger - | CustomNormal of EventStore.ClientAPI.ILogger - member log.Configure(b: ConnectionSettingsBuilder) = - match log with - | SerilogVerbose logger -> b.EnableVerboseLogging().UseCustomLogger(SerilogAdapter(logger)) - | SerilogNormal logger -> b.UseCustomLogger(SerilogAdapter(logger)) - | CustomVerbose logger -> b.EnableVerboseLogging().UseCustomLogger(logger) - | CustomNormal logger -> b.UseCustomLogger(logger) - [] type NodePreference = /// Track master via gossip, writes direct, reads should immediately reflect writes, resync without backoff (highest load on master, good write perf) @@ -574,7 +566,7 @@ type ConnectionStrategy = type EventStoreConnector ( username, password, reqTimeout: TimeSpan, reqRetries: int, - [] ?log: Logger, [] ?heartbeatTimeout: TimeSpan, [] ?concurrentOperationsLimit, + [] ?heartbeatTimeout: TimeSpan, [] ?concurrentOperationsLimit, [] ?readRetryPolicy, [] ?writeRetryPolicy, [] ?gossipTimeout, [] ?clientConnectionTimeout, // Additional strings identifying the context of this connection; should provide enough context to disambiguate all potential connections to a cluster @@ -601,7 +593,6 @@ type EventStoreConnector |> fun s -> match heartbeatTimeout with Some v -> s.SetHeartbeatTimeout v | None -> s // default: 1500 ms |> fun s -> match gossipTimeout with Some v -> s.SetGossipTimeout v | None -> s // default: 1000 ms |> fun s -> match clientConnectionTimeout with Some v -> s.WithConnectionTimeoutOf v | None -> s // default: 1000 ms - |> fun s -> match log with Some log -> log.Configure s | None -> s |> fun s -> match custom with Some c -> c s | None -> s |> _.Build() diff --git a/src/Equinox.EventStoreDb/Equinox.EventStoreDb.fsproj b/src/Equinox.EventStoreDb/Equinox.EventStoreDb.fsproj index 65bdc76cc..e6f54c1af 100644 --- a/src/Equinox.EventStoreDb/Equinox.EventStoreDb.fsproj +++ b/src/Equinox.EventStoreDb/Equinox.EventStoreDb.fsproj @@ -2,12 +2,14 @@ net6.0 - 4.0.0 + + $(DefineConstants);STORE_EVENTSTOREDB + diff --git a/src/Equinox.EventStoreDb/EventStoreDb.fs b/src/Equinox.EventStoreDb/EventStoreDb.fs index c15f392f9..cb391508d 100644 --- a/src/Equinox.EventStoreDb/EventStoreDb.fs +++ b/src/Equinox.EventStoreDb/EventStoreDb.fs @@ -1,9 +1,11 @@ namespace Equinox.EventStoreDb open Equinox.Core +open Equinox.Core.Tracing open EventStore.Client open Serilog open System +open System.Diagnostics type EventBody = ReadOnlyMemory @@ -142,20 +144,26 @@ module private Write = let private writeEventsLogged (conn: EventStoreClient) (streamName: string) (version: int64) (events: EventData[]) (log: ILogger) ct : Task = task { + let act = Activity.Current let log = if (not << log.IsEnabled) Events.LogEventLevel.Debug then log else log |> Log.propEventData "Json" events let bytes, count = eventDataBytes events, events.Length let log = log |> Log.prop "bytes" bytes |> Log.prop "expectedVersion" version + if act <> null then act.AddExpectedVersion(version).IncMetric(count, bytes) |> ignore let writeLog = log |> Log.prop "stream" streamName |> Log.prop "count" count let! t, result = writeEventsAsync writeLog conn streamName version events |> Stopwatch.time ct let reqMetric: Log.Measurement = { stream = streamName; interval = t; bytes = bytes; count = count} let resultLog, evt = match result, reqMetric with | EsSyncResult.Written x, m -> + if act <> null then act.SetStatus(ActivityStatusCode.Ok).AddTag("eqx.new_version", x.NextExpectedVersion) |> ignore log |> Log.prop "nextExpectedVersion" x.NextExpectedVersion |> Log.prop "logPosition" x.LogPosition, Log.WriteSuccess m | EsSyncResult.Conflict actualVersion, m -> + let eventTypes = [| for x in events -> x.Type |] + if act <> null then act.RecordConflict().AddTag("eqx.event_types", eventTypes) |> ignore log |> Log.prop "actualVersion" actualVersion, Log.WriteConflict m (resultLog |> Log.event evt).Information("Esdb{action:l} count={count} conflict={conflict}", "Write", events.Length, match evt with Log.WriteConflict _ -> true | _ -> false) + emitStoreOp (match evt with Log.WriteConflict _ -> "AppendConflict" | _ -> "Append") return result } let writeEvents (log: ILogger) retryPolicy (conn: EventStoreClient) (streamName: string) (version: int64) (events: EventData[]) ct @@ -168,15 +176,18 @@ module private Read = let resolvedEventBytes (x: ResolvedEvent) = let Log.BlobLen bytes, Log.BlobLen metaBytes = x.Event.Data, x.Event.Metadata in bytes + metaBytes let resolvedEventsBytes events = events |> Array.sumBy resolvedEventBytes let private logBatchRead direction streamName t events batchSize version (log: ILogger) = + let act = Activity.Current let bytes, count = resolvedEventsBytes events, events.Length let reqMetric: Log.Measurement = { stream = streamName; interval = t; bytes = bytes; count = count} let batches = match batchSize with Some batchSize -> (events.Length - 1) / batchSize + 1 | None -> -1 let action = if direction = Direction.Forward then "LoadF" else "LoadB" + if act <> null then act.IncMetric(count, bytes).AddLastVersion(version).AddBatches(batches).AddDirection(string direction) |> ignore let log = if (not << log.IsEnabled) Events.LogEventLevel.Debug then log else log |> Log.propResolvedEvents "Json" events let evt = Log.Metric.Batch (direction, batches, reqMetric) (log |> Log.prop "bytes" bytes |> Log.event evt).Information( "Esdb{action:l} stream={stream} count={count}/{batches} version={version}", action, streamName, count, batches, version) + emitStoreOp (if direction = Direction.Forward then "BatchForward" else "BatchBackward") let private loadBackwardsUntilOrigin (log: ILogger) (conn: EventStoreClient) batchSize streamName (tryDecode, isOrigin) ct : Task = task { let res = conn.ReadStreamAsync(Direction.Backwards, streamName, StreamPosition.End, int64 batchSize, resolveLinkTos = false, cancellationToken = ct) diff --git a/src/Equinox.MessageDb/Tracing.fs b/src/Equinox.EventStoreDb/Tracing.fs similarity index 50% rename from src/Equinox.MessageDb/Tracing.fs rename to src/Equinox.EventStoreDb/Tracing.fs index 0d5e74aff..3cce4a6da 100644 --- a/src/Equinox.MessageDb/Tracing.fs +++ b/src/Equinox.EventStoreDb/Tracing.fs @@ -1,28 +1,65 @@ [] -module internal Equinox.MessageDb.Tracing +module internal +#if STORE_EVENTSTOREDB + Equinox.EventStoreDb.Tracing +#endif +#if STORE_EVENTSTORE_LEGACY + Equinox.EventStore.Tracing +#endif +#if STORE_MESSAGEDB + Equinox.MessageDb.Tracing +#endif +#if !STORE_EVENTSTOREDB && !STORE_EVENTSTORE_LEGACY && !STORE_MESSAGEDB + Equinox.SqlStreamStore.Tracing +#endif -open Equinox.MessageDb.Core open System.Diagnostics +#if STORE_MESSAGEDB +open Equinox.MessageDb.Core +#endif + +let [] private SourceName = +#if STORE_EVENTSTOREDB + "Equinox.EventStoreDb" +#endif +#if STORE_EVENTSTORE_LEGACY + "Equinox.EventStore" +#endif +#if STORE_MESSAGEDB + "Equinox.MessageDb" +#endif +#if !STORE_EVENTSTOREDB && !STORE_EVENTSTORE_LEGACY && !STORE_MESSAGEDB + "Equinox.SqlStreamStore" +#endif + +let source = new ActivitySource(SourceName) [] type ActivityExtensions private () = +#if STORE_MESSAGEDB [] static member AddExpectedVersion(act: Activity, version) = match version with StreamVersion v -> act.AddTag("eqx.expected_version", v) | Any -> act +#else + [] + static member AddExpectedVersion(act: Activity, version: int64) = + act.AddTag("eqx.expected_version", version) +#endif [] static member AddLastVersion(act: Activity, version: int64) = act.AddTag("eqx.last_version", version) - [] - static member AddBatchSize(act: Activity, size: int64) = - act.AddTag("eqx.batch_size", size) - [] static member AddBatches(act: Activity, batches: int) = act.AddTag("eqx.batches", batches) +#if STORE_MESSAGEDB + [] + static member AddBatchSize(act: Activity, size: int64) = + act.AddTag("eqx.batch_size", size) + [] static member AddStartPosition(act: Activity, pos: int64) = act.AddTag("eqx.start_position", pos) @@ -30,7 +67,17 @@ type ActivityExtensions private () = [] static member AddLoadMethod(act: Activity, method: string) = act.AddTag("eqx.load_method", method) +#else + [] + static member AddDirection(act: Activity, direction: string) = + act.AddTag("eqx.direction", direction) +#endif [] static member RecordConflict(act: Activity) = act.AddTag("eqx.conflict", true).SetStatus(ActivityStatusCode.Error, "WrongExpectedVersion") + +/// Emits a short-lived child Activity from the store-specific ActivitySource, enabling test capture via ActivityListener +let emitStoreOp (name: string) = + let a: Activity = source.StartActivity(name) + if a <> null then a.Dispose() diff --git a/src/Equinox.MessageDb/Equinox.MessageDb.fsproj b/src/Equinox.MessageDb/Equinox.MessageDb.fsproj index bcf42b4ab..4eabceeee 100644 --- a/src/Equinox.MessageDb/Equinox.MessageDb.fsproj +++ b/src/Equinox.MessageDb/Equinox.MessageDb.fsproj @@ -3,13 +3,14 @@ net6.0 4.0.0 + $(DefineConstants);STORE_MESSAGEDB - + diff --git a/src/Equinox.MessageDb/MessageDb.fs b/src/Equinox.MessageDb/MessageDb.fs index 8d376ad1c..ebcbbe01f 100644 --- a/src/Equinox.MessageDb/MessageDb.fs +++ b/src/Equinox.MessageDb/MessageDb.fs @@ -152,6 +152,7 @@ module private Write = log, Log.WriteConflict reqMetric (resultLog |> Log.event evt).Information("Mdb{action:l} count={count} conflict={conflict}", "Write", count, match evt with Log.WriteConflict _ -> true | _ -> false) + emitStoreOp (match evt with Log.WriteConflict _ -> "AppendConflict" | _ -> "Append") return result } let writeEvents log retryPolicy writer (category, streamId, streamName) version events onSync ct: Task = task { let call = writeEventsLogged writer streamName version events onSync @@ -188,6 +189,7 @@ module Read = let log = if not (log.IsEnabled Events.LogEventLevel.Debug) then log else log |> Log.propResolvedEvents "Json" slice.Messages (log |> Log.prop "startPos" startPos |> Log.prop "bytes" bytes |> Log.event evt).Information("Mdb{action:l} count={count} version={version}", "Read", count, slice.LastVersion) + emitStoreOp "SliceForward" return slice } let private readBatches (log: ILogger) fold originState tryDecode (readSlice: int64 -> int -> ILogger -> CancellationToken -> Task) @@ -223,6 +225,7 @@ module Read = (log |> Log.event evt).Information( "Mdb{action:l} stream={stream} count={count}/{batches} version={version}", action, streamName, events, batches, version) + emitStoreOp "BatchForward" let private logLastEventRead streamName t events (version: int64) (log: ILogger) = let bytes = resolvedEventBytes events @@ -234,6 +237,7 @@ module Read = (log |> Log.prop "bytes" bytes |> Log.event evt).Information( "Mdb{action:l} stream={stream} count={count} version={version}", "ReadL", streamName, count, version) + emitStoreOp "ReadLast" let internal loadLastEvent (log: ILogger) retryPolicy (reader: MessageDbReader) requiresLeader streamName eventType ct : Task[]> = task { diff --git a/src/Equinox.SqlStreamStore.MsSql/Equinox.SqlStreamStore.MsSql.fsproj b/src/Equinox.SqlStreamStore.MsSql/Equinox.SqlStreamStore.MsSql.fsproj index 108a6fa2d..ec618ee24 100644 --- a/src/Equinox.SqlStreamStore.MsSql/Equinox.SqlStreamStore.MsSql.fsproj +++ b/src/Equinox.SqlStreamStore.MsSql/Equinox.SqlStreamStore.MsSql.fsproj @@ -2,7 +2,7 @@ net6.0 - 4.0.0 + @@ -11,8 +11,7 @@ - - + diff --git a/src/Equinox.SqlStreamStore.MySql/Equinox.SqlStreamStore.MySql.fsproj b/src/Equinox.SqlStreamStore.MySql/Equinox.SqlStreamStore.MySql.fsproj index 0656da671..892dce557 100644 --- a/src/Equinox.SqlStreamStore.MySql/Equinox.SqlStreamStore.MySql.fsproj +++ b/src/Equinox.SqlStreamStore.MySql/Equinox.SqlStreamStore.MySql.fsproj @@ -2,7 +2,7 @@ net6.0 - 4.0.0 + @@ -11,8 +11,7 @@ - - + diff --git a/src/Equinox.SqlStreamStore.Postgres/Equinox.SqlStreamStore.Postgres.fsproj b/src/Equinox.SqlStreamStore.Postgres/Equinox.SqlStreamStore.Postgres.fsproj index 6d8b4d2f8..84052dc25 100644 --- a/src/Equinox.SqlStreamStore.Postgres/Equinox.SqlStreamStore.Postgres.fsproj +++ b/src/Equinox.SqlStreamStore.Postgres/Equinox.SqlStreamStore.Postgres.fsproj @@ -2,7 +2,7 @@ net6.0 - 4.0.0 + @@ -11,8 +11,7 @@ - - + diff --git a/src/Equinox.SqlStreamStore/Equinox.SqlStreamStore.fsproj b/src/Equinox.SqlStreamStore/Equinox.SqlStreamStore.fsproj index 26751da9e..ff5a79bdd 100644 --- a/src/Equinox.SqlStreamStore/Equinox.SqlStreamStore.fsproj +++ b/src/Equinox.SqlStreamStore/Equinox.SqlStreamStore.fsproj @@ -2,12 +2,13 @@ net6.0 - 4.0.0 + + diff --git a/src/Equinox.SqlStreamStore/SqlStreamStore.fs b/src/Equinox.SqlStreamStore/SqlStreamStore.fs index 1c49e7355..ffc7077be 100644 --- a/src/Equinox.SqlStreamStore/SqlStreamStore.fs +++ b/src/Equinox.SqlStreamStore/SqlStreamStore.fs @@ -1,11 +1,13 @@ namespace Equinox.SqlStreamStore open Equinox.Core +open Equinox.Core.Tracing open Serilog open SqlStreamStore open SqlStreamStore.Streams open System open System.Collections.Generic +open System.Diagnostics type EventBody = ReadOnlyMemory type EventData = NewStreamMessage @@ -148,20 +150,26 @@ module private Write = events |> Array.sumBy eventDataLen let private writeEventsLogged (conn: IEventStoreConnection) (streamName: string) (version: int64) (events: EventData[]) (log: ILogger) ct : Task = task { + let act = Activity.Current let log = if (not << log.IsEnabled) Events.LogEventLevel.Debug then log else log |> Log.propEventData "Json" events let bytes, count = eventDataBytes events, events.Length let log = log |> Log.prop "bytes" bytes + if act <> null then act.AddExpectedVersion(version).IncMetric(count, bytes) |> ignore let writeLog = log |> Log.prop "stream" streamName |> Log.prop "expectedVersion" version |> Log.prop "count" count let! t, result = writeEventsAsync writeLog conn streamName version events |> Stopwatch.time ct let reqMetric: Log.Measurement = { stream = streamName; interval = t; bytes = bytes; count = count} let resultLog, evt = match result, reqMetric with | EsSyncResult.Written x, m -> + if act <> null then act.SetStatus(ActivityStatusCode.Ok).AddTag("eqx.new_version", x.CurrentVersion) |> ignore log |> Log.prop "currentVersion" x.CurrentVersion |> Log.prop "currentPosition" x.CurrentPosition, Log.WriteSuccess m | EsSyncResult.ConflictUnknown, m -> + let eventTypes = [| for x in events -> x.Type |] + if act <> null then act.RecordConflict().AddTag("eqx.event_types", eventTypes) |> ignore log, Log.WriteConflict m (resultLog |> Log.event evt).Information("SqlEs{action:l} count={count} conflict={conflict}", "Write", events.Length, match evt with Log.WriteConflict _ -> true | _ -> false) + emitStoreOp (match evt with Log.WriteConflict _ -> "AppendConflict" | _ -> "Append") return result } let writeEvents (log: ILogger) retryPolicy (conn: IEventStoreConnection) (streamName: string) (version: int64) (events: EventData[]) ct : Task = @@ -177,13 +185,16 @@ module private Read = | Direction.Backward -> conn.ReadStreamBackwards(streamName, int startPos, batchSize, ct) let (|ResolvedEventLen|) (x: StreamMessage) = match x.JsonData, x.JsonMetadata with Log.StrLen bytes, Log.StrLen metaBytes -> bytes + metaBytes let private loggedReadSlice conn streamName direction batchSize startPos (log: ILogger) ct: Task = task { + let act = Activity.Current let! t, slice = readSliceAsync conn streamName direction batchSize startPos |> Stopwatch.time ct let bytes, count = slice.Messages |> Array.sumBy (|ResolvedEventLen|), slice.Messages.Length let reqMetric: Log.Measurement = { stream = streamName; interval = t; bytes = bytes; count = count } let evt = Log.Slice (direction, reqMetric) + if act <> null then act.IncMetric(count, bytes).AddDirection(string direction) |> ignore let log = if (not << log.IsEnabled) Events.LogEventLevel.Debug then log else log |> Log.propResolvedEvents "Json" slice.Messages (log |> Log.prop "startPos" startPos |> Log.prop "bytes" bytes |> Log.event evt).Information("SqlEs{action:l} count={count} version={version}", "Read", count, slice.LastStreamVersion) + emitStoreOp (match direction with Direction.Forward -> "SliceForward" | Direction.Backward -> "SliceBackward") return slice } let private readBatches (log: ILogger) (readSlice: int64 -> ILogger -> CancellationToken -> Task) (maxPermittedBatchReads: int option) (startPosition: int64) ct @@ -206,14 +217,17 @@ module private Read = loop 0 startPosition let resolvedEventBytes events = events |> Array.sumBy (|ResolvedEventLen|) let logBatchRead direction streamName t events batchSize version (log: ILogger) = + let act = Activity.Current let bytes, count = resolvedEventBytes events, events.Length let reqMetric: Log.Measurement = { stream = streamName; interval = t; bytes = bytes; count = count} let batches = (events.Length - 1)/batchSize + 1 let action = match direction with Direction.Forward -> "LoadF" | Direction.Backward -> "LoadB" + if act <> null then act.IncMetric(count, bytes).AddLastVersion(version).AddBatches(batches).AddDirection(string direction) |> ignore let evt = Log.Metric.Batch (direction, batches, reqMetric) (log |> Log.prop "bytes" bytes |> Log.event evt).Information( "SqlEs{action:l} stream={stream} count={count}/{batches} version={version}", action, streamName, count, batches, version) + emitStoreOp (match direction with Direction.Forward -> "BatchForward" | Direction.Backward -> "BatchBackward") let loadForwardsFrom (log: ILogger) retryPolicy conn batchSize maxPermittedBatchReads streamName startPosition ct : Task = task { let mergeBatches (batches: IAsyncEnumerable) = task { diff --git a/tests/Equinox.EventStore.Integration/Equinox.EventStore.Integration.fsproj b/tests/Equinox.EventStore.Integration/Equinox.EventStore.Integration.fsproj index 026c2f7a5..bdedfc686 100644 --- a/tests/Equinox.EventStore.Integration/Equinox.EventStore.Integration.fsproj +++ b/tests/Equinox.EventStore.Integration/Equinox.EventStore.Integration.fsproj @@ -8,6 +8,7 @@ + diff --git a/tests/Equinox.EventStoreDb.Integration/Equinox.EventStoreDb.Integration.fsproj b/tests/Equinox.EventStoreDb.Integration/Equinox.EventStoreDb.Integration.fsproj index ff29ba37e..b060976d5 100644 --- a/tests/Equinox.EventStoreDb.Integration/Equinox.EventStoreDb.Integration.fsproj +++ b/tests/Equinox.EventStoreDb.Integration/Equinox.EventStoreDb.Integration.fsproj @@ -8,6 +8,7 @@ + diff --git a/tests/Equinox.EventStoreDb.Integration/Infrastructure.fs b/tests/Equinox.EventStoreDb.Integration/Infrastructure.fs index 8270fa749..fa2b06a8a 100644 --- a/tests/Equinox.EventStoreDb.Integration/Infrastructure.fs +++ b/tests/Equinox.EventStoreDb.Integration/Infrastructure.fs @@ -3,6 +3,7 @@ open Domain open FsCheck.FSharp open System +open System.Diagnostics open FSharp.UMX module ArbMap = @@ -30,6 +31,21 @@ open Equinox.EventStoreDb open Equinox.EventStore #endif +[] +type EsAct = Append | AppendConflict | SliceForward | SliceBackward | BatchForward | BatchBackward | ReadLast + +[] +module OtelHelpers = + let tryEsAct = function + | "Append" -> Some EsAct.Append + | "AppendConflict" -> Some EsAct.AppendConflict + | "SliceForward" -> Some EsAct.SliceForward + | "SliceBackward" -> Some EsAct.SliceBackward + | "BatchForward" -> Some EsAct.BatchForward + | "BatchBackward" -> Some EsAct.BatchBackward + | "ReadLast" -> Some EsAct.ReadLast + | _ -> None + [] module SerilogHelpers = open Serilog.Events @@ -37,8 +53,6 @@ module SerilogHelpers = let (|SerilogScalar|_|): LogEventPropertyValue -> obj option = function | :? ScalarValue as x -> Some x.Value | _ -> None - [] - type EsAct = Append | AppendConflict | SliceForward | SliceBackward | BatchForward | BatchBackward | ReadLast let (|EsAction|) (evt: Log.Metric) = match evt with | Log.WriteSuccess _ -> EsAct.Append @@ -64,8 +78,38 @@ module SerilogHelpers = match e.Properties.TryGetValue name with | true, (SerilogScalar _ as s) -> Some s | _ -> None - let (|SerilogString|_|): LogEventPropertyValue -> string option = function SerilogScalar (:? string as y) -> Some y | _ -> None - let (|SerilogBool|_|): LogEventPropertyValue -> bool option = function SerilogScalar (:? bool as y) -> Some y | _ -> None + +/// Captures store-specific Activity emissions for test assertions, using AsyncLocal for per-test isolation +type StoreActivityCapture() = + static let storeSourceName = +#if STORE_EVENTSTOREDB + "Equinox.EventStoreDb" +#endif +#if STORE_EVENTSTORE_LEGACY + "Equinox.EventStore" +#endif +#if STORE_MESSAGEDB + "Equinox.MessageDb" +#endif +#if STORE_POSTGRES || STORE_MSSQL || STORE_MYSQL + "Equinox.SqlStreamStore" +#endif + static let currentOps = new System.Threading.AsyncLocal>() + static let gate = obj() + static let _listener = + let l = new ActivityListener( + ShouldListenTo = (fun source -> source.Name = storeSourceName), + Sample = (fun _ -> ActivitySamplingResult.AllDataAndRecorded), + ActivityStopped = (fun act -> + let ops = currentOps.Value + if ops <> null then lock gate (fun () -> ops.Add act.OperationName))) + ActivitySource.AddActivityListener(l) + l + let ops = ResizeArray() + do currentOps.Value <- ops + member _.Clear() = lock gate (fun () -> ops.Clear()) + member _.ExternalCalls = lock gate (fun () -> ops |> Seq.toList) |> List.choose tryEsAct + interface IDisposable with member _.Dispose() = currentOps.Value <- null type LogCapture() = inherit LogCaptureBuffer() @@ -74,6 +118,13 @@ type LogCapture() = type TestContext(testOutputHelper) = let output = TestOutput testOutputHelper - member x.CreateLoggerWithCapture() = - let capture = LogCapture() - output.CreateLogger(capture), capture + member _.CreateLogger() = output.CreateLogger() + + member _.CreateLoggerWithCapture() = + let capture = new StoreActivityCapture() + output.CreateLogger(), capture + + member _.CreateLoggerWithDualCapture() = + let serilogCapture = LogCapture() + let otelCapture = new StoreActivityCapture() + output.CreateLogger(serilogCapture), otelCapture, serilogCapture diff --git a/tests/Equinox.EventStoreDb.Integration/StoreIntegration.fs b/tests/Equinox.EventStoreDb.Integration/StoreIntegration.fs index 8a5e9295b..dbdf16c78 100644 --- a/tests/Equinox.EventStoreDb.Integration/StoreIntegration.fs +++ b/tests/Equinox.EventStoreDb.Integration/StoreIntegration.fs @@ -68,9 +68,9 @@ open Equinox.EventStore /// Connect to the locally running INSECURE 3-node EventStoreDB cluster (see docker-compose.yml) via legacy TCP, using gossip-based discovery. /// WARNING: Applies INSECURE overrides to match the docker-compose config. NEVER do this in staging or production. -let connectToLocalStore log = +let connectToLocalStore _ = let c = EventStoreConnector( - "admin", "changeit", reqTimeout=TimeSpan.FromSeconds 3., reqRetries=3, log=Logger.SerilogVerbose log, + "admin", "changeit", reqTimeout=TimeSpan.FromSeconds 3., reqRetries=3, tags=["I", Guid.NewGuid().ToString("N")], // INSECURE: Disable TLS for TCP data transfer, to match the docker-compose EVENTSTORE_INSECURE=true config. NEVER do this in production. custom = _.DisableTls()) @@ -88,9 +88,7 @@ type Category<'event, 'state, 'req> = EventStoreCategory<'event, 'state, 'req> let createContext connection batchSize = Context(connection, batchSize = batchSize) -#if NET let source = new ActivitySource("StoreIntegration") -#endif module SimplestThing = type Event = @@ -184,7 +182,7 @@ type GeneralTests(testOutputHelper) = .AddSource("Equinox") .AddSource("Equinox.MessageDb") .AddSource("StoreIntegration") - .AddSource("Npqsl") + .AddSource("Npgsql") .AddOtlpExporter(fun opts -> opts.Endpoint <- Uri("http://localhost:4317")) .Build() #endif @@ -201,9 +199,7 @@ type GeneralTests(testOutputHelper) = [] let ``Can roundtrip against Store, correctly batching the reads [without any optimizations]`` (ctx, skuId) = async { - #if NET use _ = source.StartActivity("Can roundtrip against Store, correctly batching the reads [without any optimizations]") - #endif let log, capture = output.CreateLoggerWithCapture() let! connection = connectToLocalStore log @@ -232,10 +228,8 @@ type GeneralTests(testOutputHelper) = [] let ``Can roundtrip against Store, managing sync conflicts by retrying [without any optimizations]`` (ctx, initialState) = async { - #if NET use _ = source.StartActivity("Can roundtrip against Store, managing sync conflicts by retrying [without any optimizations]") - #endif - let log1, capture1 = output.CreateLoggerWithCapture() + let log1, capture = output.CreateLoggerWithCapture() let! connection = connectToLocalStore log1 // Ensure batching is included at some point in the proceedings @@ -266,9 +260,7 @@ type GeneralTests(testOutputHelper) = let w3, s3 = eventWaitSet () let w4, s4 = eventWaitSet () let t1 = async { - #if NET use _ = source.StartActivity("Trx1") - #endif // Wait for other to have state, signal we have it, await conflict and handle let prepare = async { do! w0 @@ -280,13 +272,11 @@ type GeneralTests(testOutputHelper) = do! act prepare service1 sku12 12 // Signal conflict generated do! s4 } - let log2, capture2 = output.CreateLoggerWithCapture() + let log2 = output.CreateLogger() let context = createContext connection batchSize let service2 = Cart.createServiceWithoutOptimization log2 context let t2 = async { - #if NET use _ = source.StartActivity("Trx2") - #endif // Signal we have state, wait for other to do same, engineer conflict let prepare = async { do! s0 @@ -299,7 +289,7 @@ type GeneralTests(testOutputHelper) = do! s3 do! w4 } do! act prepare service2 sku22 22 } - // Act: Engineer the conflicts and applications, with logging into capture1 and capture2 + // Act: Engineer the conflicts and applications do! Async.Parallel [t1; t2] |> Async.Ignore // Load state @@ -310,9 +300,8 @@ type GeneralTests(testOutputHelper) = test <@ maybeInitialSku |> Option.forall (fun (skuId, quantity) -> has skuId quantity) && has sku11 11 && has sku12 12 && has sku21 21 && has sku22 22 @> - // Intended conflicts pertained - let hadConflict= function EsEvent (EsAction EsAct.AppendConflict) -> Some () | _ -> None - test <@ [1; 1] = [for c in [capture1; capture2] -> c.ChooseCalls hadConflict |> List.length] @> } + // Intended conflicts pertained (2 total: one per concurrent service) + test <@ 2 = (capture.ExternalCalls |> List.filter (fun a -> a = EsAct.AppendConflict) |> List.length) @> } #if STORE_EVENTSTOREDB // gRPC does not expose slice metrics let sliceBackward = [] @@ -329,9 +318,7 @@ type GeneralTests(testOutputHelper) = [] let ``Can correctly read and update against Store, with LatestKnownEvent Access Strategy`` id value = async { - #if NET use _ = source.StartActivity("Can correctly read and update against Store, with LatestKnownEvent Access Strategy") - #endif let log, capture = output.CreateLoggerWithCapture() let! client = connectToLocalStore log let service = ContactPreferences.createService log client @@ -352,9 +339,7 @@ type GeneralTests(testOutputHelper) = [] let ``Can roundtrip against Store, correctly caching to avoid redundant reads`` (ctx, skuId) = async { - #if NET use _ = source.StartActivity("Can roundtrip against Store, correctly caching to avoid redundant reads") - #endif let log, capture = output.CreateLoggerWithCapture() let! client = connectToLocalStore log let batchSize = 10 @@ -417,10 +402,9 @@ type GeneralTests(testOutputHelper) = [] let ``Version is 0-based`` () = async { - #if NET use _ = source.StartActivity("Version is 0-based") - #endif - let log, _ = output.CreateLoggerWithCapture() + let log, capture = output.CreateLoggerWithCapture() + use _capture = capture let! connection = connectToLocalStore log let batchSize = 3 @@ -459,9 +443,7 @@ type RollingSnapshotTests(testOutputHelper) = [] let ``Can roundtrip against Store, correctly compacting to avoid redundant reads`` (ctx, skuId) = async { - #if NET use _ = source.StartActivity("Can roundtrip against Store, correctly compacting to avoid redundant reads") - #endif let log, capture = output.CreateLoggerWithCapture() let! client = connectToLocalStore log let batchSize = 10 @@ -554,7 +536,7 @@ type AdjacentSnapshotTests(testOutputHelper) = .AddSource("Equinox") .AddSource("Equinox.MessageDb") .AddSource("StoreIntegration") - .AddSource("Npqsl") + .AddSource("Npgsql") .AddOtlpExporter(fun opts -> opts.Endpoint <- Uri("http://localhost:4317")) .Build() @@ -566,9 +548,7 @@ type AdjacentSnapshotTests(testOutputHelper) = [] let ``Can roundtrip against Store, correctly snapshotting to avoid redundant reads`` (ctx, skuId) = async { - #if NET use _ = source.StartActivity("Can roundtrip against Store, correctly snapshotting to avoid redundant reads") - #endif let log, capture = output.CreateLoggerWithCapture() let! client = connectToLocalStore log let batchSize = 10 @@ -659,3 +639,88 @@ type AdjacentSnapshotTests(testOutputHelper) = interface IDisposable with member _.Dispose() = sdk.Shutdown() |> ignore #endif + +/// Demonstrates that OTel Activity capture and Serilog LogEvent capture yield equivalent EsAct classifications, +/// proving the OTel migration is lossless. Also exercises the OtelToSerilogBridge helper. +type OtelSerilogMappingTests(testOutputHelper) = +#if STORE_MESSAGEDB + let sdk = + Sdk.CreateTracerProviderBuilder() + .SetResourceBuilder(ResourceBuilder.CreateDefault().AddService(serviceName = "mdbi-mapping")) + .AddSource("Equinox") + .AddSource("Equinox.MessageDb") + .AddSource("StoreIntegration") + .AddSource("Npgsql") + .AddOtlpExporter(fun opts -> opts.Endpoint <- Uri("http://localhost:4317")) + .Build() +#endif + let output = TestContext(testOutputHelper) + + [] + let ``OTel and Serilog both classify Append for a write operation`` (ctx, skuId) = async { + use _ = source.StartActivity "OTel and Serilog both classify Append for a write operation" + let log, otelCapture, serilogCapture = output.CreateLoggerWithDualCapture() + use _capture = otelCapture + let! connection = connectToLocalStore log + let context = createContext connection defaultBatchSize + let service = Cart.createServiceWithoutOptimization log context + let cartId = CartId.gen () + + do! service.SyncItems(cartId, false, [ (ctx, skuId, Some 1, None) ]) + + let acts = List.filter (fun a -> a = EsAct.Append) + let otelAppends = otelCapture.ExternalCalls |> acts + let serilogAppends = serilogCapture.ExternalCalls |> acts + test <@ otelAppends = serilogAppends && otelAppends.Length > 0 @> } + + [] + let ``OTel and Serilog both classify read operations consistently`` (ctx, skuId) = async { + use _ = source.StartActivity "OTel and Serilog both classify read operations consistently" + let log, otelCapture, serilogCapture = output.CreateLoggerWithDualCapture() + use _capture = otelCapture + let! connection = connectToLocalStore log + let context = createContext connection defaultBatchSize + let service = Cart.createServiceWithoutOptimization log context + let cartId = CartId.gen () + do! service.SyncItems(cartId, false, [ (ctx, skuId, Some 1, None) ]) + otelCapture.Clear() + serilogCapture.Clear() + + // Read the data that's been stashed + let! _ = service.Read cartId + + // Both should see at least a batch read operation; Both should agree on the type of read + let otelOps = otelCapture.ExternalCalls + let serilogOps = serilogCapture.ExternalCalls + let acts = List.choose (function EsAct.BatchForward | EsAct.BatchBackward | EsAct.ReadLast as a -> Some a | _ -> None) + test <@ acts otelOps = acts serilogOps && otelOps.Length > 0 @> } + + [] + let ``OtelToSerilogBridge emits log events matching store operations`` (ctx, skuId) = async { + use _ = source.StartActivity "OtelToSerilogBridge emits log events matching store operations" + let bridgeCapture = LogCaptureBuffer() + let bridgeLog = Serilog.LoggerConfiguration().WriteTo.Sink(bridgeCapture).CreateLogger() + use _bridge = new OtelToSerilogBridge(bridgeLog) + let log, capture = output.CreateLoggerWithCapture() + use _capture = capture + let! connection = connectToLocalStore log + let context = createContext connection defaultBatchSize + let service = Cart.createServiceWithoutOptimization log context + let cartId = CartId.gen () + + // Act: write then read + do! service.SyncItems(cartId, false, [ (ctx, skuId, Some 1, None) ]) + let! _ = service.Read cartId + + // The bridge should have emitted log events for store operations + let bridgeLogs = bridgeCapture.ChooseCalls(fun e -> + match e.Properties.TryGetValue("Action") with + | true, v -> Some (string v) + | _ -> None) + test <@ bridgeLogs.Length >= 2 @> + test <@ bridgeLogs |> List.exists _.Contains("Append") @> } + +#if STORE_MESSAGEDB + interface IDisposable with + member _.Dispose() = sdk.Shutdown() |> ignore +#endif diff --git a/tests/Equinox.MessageDb.Integration/Equinox.MessageDb.Integration.fsproj b/tests/Equinox.MessageDb.Integration/Equinox.MessageDb.Integration.fsproj index 2548a536f..487ac0a54 100644 --- a/tests/Equinox.MessageDb.Integration/Equinox.MessageDb.Integration.fsproj +++ b/tests/Equinox.MessageDb.Integration/Equinox.MessageDb.Integration.fsproj @@ -8,6 +8,7 @@ + diff --git a/tests/Equinox.SqlStreamStore.MsSql.Integration/Equinox.SqlStreamStore.MsSql.Integration.fsproj b/tests/Equinox.SqlStreamStore.MsSql.Integration/Equinox.SqlStreamStore.MsSql.Integration.fsproj index e041cace9..3b177a372 100644 --- a/tests/Equinox.SqlStreamStore.MsSql.Integration/Equinox.SqlStreamStore.MsSql.Integration.fsproj +++ b/tests/Equinox.SqlStreamStore.MsSql.Integration/Equinox.SqlStreamStore.MsSql.Integration.fsproj @@ -8,6 +8,7 @@ + diff --git a/tests/Equinox.SqlStreamStore.MySql.Integration/Equinox.SqlStreamStore.MySql.Integration.fsproj b/tests/Equinox.SqlStreamStore.MySql.Integration/Equinox.SqlStreamStore.MySql.Integration.fsproj index bca7a9020..99d391440 100644 --- a/tests/Equinox.SqlStreamStore.MySql.Integration/Equinox.SqlStreamStore.MySql.Integration.fsproj +++ b/tests/Equinox.SqlStreamStore.MySql.Integration/Equinox.SqlStreamStore.MySql.Integration.fsproj @@ -8,6 +8,7 @@ + diff --git a/tests/Equinox.SqlStreamStore.Postgres.Integration/Equinox.SqlStreamStore.Postgres.Integration.fsproj b/tests/Equinox.SqlStreamStore.Postgres.Integration/Equinox.SqlStreamStore.Postgres.Integration.fsproj index 24a2c82d0..5c0ed3ef5 100644 --- a/tests/Equinox.SqlStreamStore.Postgres.Integration/Equinox.SqlStreamStore.Postgres.Integration.fsproj +++ b/tests/Equinox.SqlStreamStore.Postgres.Integration/Equinox.SqlStreamStore.Postgres.Integration.fsproj @@ -8,6 +8,7 @@ +