diff --git a/.claude/rules/configuration.md b/.claude/rules/configuration.md index 7296ea5..4d3272b 100644 --- a/.claude/rules/configuration.md +++ b/.claude/rules/configuration.md @@ -34,4 +34,6 @@ Proposed names (descriptive: once implemented, the code is the reference). | `Llm:Anthropic:ApiKey` | Matching.Worker | user-secrets only, never in the repo; required only when `Provider = Anthropic` | | `Llm:Anthropic:Model`, `Llm:Anthropic:MaxOutputTokens` | Matching.Worker | `claude-haiku-5-5`, 8000; the ceiling includes adaptive thinking | | `Llm:Anthropic:MaxRetries` | Matching.Worker | the SDK's retries of 408/409/429/5xx and connection failures; 2 | +| `OTEL_EXPORTER_OTLP_ENDPOINT` | all three hosts | unset = no telemetry at all (tests, CI). Set only in the `launchSettings.json` profiles (`http://localhost:4317`, Aspire Dashboard), never in `appsettings*.json`: integration tests run in `Development` and load those files | +| `OTEL_SERVICE_NAME` | all three hosts | `ingress-api`, `sync-worker`, `matching-worker`, next to the endpoint in `launchSettings.json`; required when the endpoint is set | | `Llm:Fake:Mode` | Matching.Worker | `Valid`, `InvalidJson`, `OutOfRange`, `Hang`; default when the CV has no `[llm:]` marker | diff --git a/.claude/rules/messaging.md b/.claude/rules/messaging.md index 0cb1231..2cb9a74 100644 --- a/.claude/rules/messaging.md +++ b/.claude/rules/messaging.md @@ -34,6 +34,9 @@ paths: No attributes, no payload body: the envelope is a notification. Workers fetch current state from the API (notify-then-fetch). `Contracts` references nothing. +Transport metadata goes in `EventData.Properties`, never in the envelope. The only one is `traceparent` +(`EventProperties.TraceParent`, ADR 010): the relay's send span, parent of every consumer attempt span. + ## Ingress.Api 1. Read the raw body (enable buffering). Verify signature v2 through the verifier interface from diff --git a/.claude/rules/persistence.md b/.claude/rules/persistence.md index ba23017..45a8b33 100644 --- a/.claude/rules/persistence.md +++ b/.claude/rules/persistence.md @@ -46,7 +46,8 @@ configurations take the schema as a constructor parameter. ## Tables (minimal columns, descriptive once migrations exist) -- `Outbox`: `Id`, `EventId`, `CandidateId`, `Type`, `Envelope` (json), `CreatedAt`, `PublishedAt` (null until published). +- `Outbox`: `Id`, `EventId`, `CandidateId`, `Type`, `Envelope` (json), `TraceParent` (W3C, nullable, ADR 010), + `CreatedAt`, `PublishedAt` (null until published). Filtered index on `PublishedAt IS NULL`. - `Inbox`: `Id`, `Source`, `EventId`, `ProcessedAt`. - `ParkingLot`: `Id`, `Source` (nullable), `EventId` (nullable), `PartitionId`, `SequenceNumber`, `Envelope`, diff --git a/.claude/rules/testing.md b/.claude/rules/testing.md index 7bf7612..187501c 100644 --- a/.claude/rules/testing.md +++ b/.claude/rules/testing.md @@ -14,6 +14,7 @@ paths: | `TalentSync.Sync.Application.Tests`, `TalentSync.Matching.Application.Tests` | Unit, no I/O | Use cases: each fetch result and port result → use case result. Ports replaced by small hand-written stubs | | `TalentSync.Infrastructure.Teamtailor.Tests` | Unit, no network | Signature v2, webhook parsing, stage mapping, response mapping (404 → Deleted, 4xx → Rejected). Internals via `InternalsVisibleTo` | | `TalentSync.Infrastructure.Messaging.Tests` | Unit, no I/O | Consumer skeleton: inbox and parking lot mapping, failure classification, handler loop | +| `TalentSync.Infrastructure.Observability.Tests` | Unit, no network | `AddObservability` with the SDK enabled on a test server (the request span is a root), `SpanExportPolicy` in front of an in-memory exporter. Internals via `InternalsVisibleTo` | | `TalentSync.Sync.Infrastructure.Tests` | Unit, no network | CRM client: status mapping (404 on delete = done, 4xx → Rejected), HTTP pipeline. Internals via `InternalsVisibleTo` | | `TalentSync.Matching.Evaluation.Tests` | Unit, no network | Prompt evaluation (ADR 009): metric calculations, golden set hash, case file validation, report format, the CI guard for the evaluation report. Never calls a model | | `TalentSync.IntegrationTests` | Acceptance | Scenarios from `docs/acceptance-scenarios.md` | @@ -62,6 +63,13 @@ paths: The fixture sets `Llm:Timeout` to 2 s for M3. The circuit breaker is shared by all scenarios; only M3 fails a call, which stays below the configured failure ratio. +## Traces + +Scenario O1 records spans with `SpanRecorder` (`Infrastructure/`), an `ActivityListener` on `TalentSync.*` and +`Microsoft.AspNetCore` (request spans): all hosts run in the test process, so no exporter is needed. Our spans are +matched by the `talentsync.event_id` tag, the request span by the webhook span's parent id. O1 runs without the SDK; +the SDK's own handling of the request span is covered by `EdgeTraceTests` in `TalentSync.Infrastructure.Observability.Tests`. + ## Prompt evaluation guard `EvaluationReportGuardTests` fails when `eval/reports/` has no report for the current `PromptVersion` and golden set diff --git a/Directory.Packages.props b/Directory.Packages.props index c7748a8..84af14c 100644 --- a/Directory.Packages.props +++ b/Directory.Packages.props @@ -15,6 +15,7 @@ + @@ -22,11 +23,18 @@ + + + + + + + diff --git a/README.md b/README.md index f556443..9dc7c93 100644 --- a/README.md +++ b/README.md @@ -58,7 +58,8 @@ dotnet user-secrets set "Llm:Provider" "Anthropic" --project src/Matching/Talent To click through the whole flow by hand (success, CRM outage, candidate without a CV), start the containers and the three hosts with `deploy/run-local.sh` (Git Bash; on Windows each host opens in its own Windows Terminal tab), then use -[`docs/e2e.http`](docs/e2e.http). +[`docs/e2e.http`](docs/e2e.http). Traces, metrics and logs of the three hosts are in the Aspire Dashboard at +http://localhost:18888 (ADR 010). ## Simplifications diff --git a/TalentSync.slnx b/TalentSync.slnx index 9a49c56..ea2555a 100644 --- a/TalentSync.slnx +++ b/TalentSync.slnx @@ -20,6 +20,7 @@ + @@ -30,6 +31,7 @@ + diff --git a/deploy/docker-compose.yml b/deploy/docker-compose.yml index c914641..436b19e 100644 --- a/deploy/docker-compose.yml +++ b/deploy/docker-compose.yml @@ -1,5 +1,5 @@ # Local infrastructure: SQL Server, Event Hubs emulator and Azurite (emulator metadata, later worker checkpoints), -# plus the three simulators. The TalentSync hosts run on the host with `dotnet run`. +# the three simulators and the Aspire Dashboard. The TalentSync hosts run on the host with `dotnet run`. # Secrets come from deploy/.env (gitignored, copy deploy/.env.example). name: talentsync @@ -86,5 +86,16 @@ services: ports: - "5075:8080" + # Traces, metrics and logs of the three hosts (ADR 010), UI on http://localhost:18888. They export over OTLP/gRPC to + # 4317 (OTEL_EXPORTER_OTLP_ENDPOINT in their launchSettings.json). Data is in memory and is lost on restart. + # Anonymous access: local only, never expose this port. + aspire-dashboard: + image: mcr.microsoft.com/aspire/dashboard:13.6.1 + environment: + ASPIRE_DASHBOARD_UNSECURED_ALLOW_ANONYMOUS: "true" + ports: + - "18888:18888" + - "4317:18889" + volumes: sql-data: diff --git a/deploy/run-local.sh b/deploy/run-local.sh index cce9fd4..b4a439d 100644 --- a/deploy/run-local.sh +++ b/deploy/run-local.sh @@ -30,13 +30,13 @@ if command -v wt.exe >/dev/null 2>&1; then cmd /k dotnet run --no-build --launch-profile http --project "${projects[$i]}" echo "Started ${titles[$i]}" done - echo "Click through the flow with docs/e2e.http." + echo "Click through the flow with docs/e2e.http. Traces and logs: http://localhost:18888" else trap 'kill $(jobs -p) 2>/dev/null' INT TERM EXIT for i in "${!titles[@]}"; do dotnet run --no-build --launch-profile http --project "${projects[$i]}" 2>&1 | sed -u "s/^/[${titles[$i]}] /" & echo "Started ${titles[$i]}" done - echo "Click through the flow with docs/e2e.http. Ctrl+C stops all three hosts." + echo "Click through the flow with docs/e2e.http. Traces and logs: http://localhost:18888. Ctrl+C stops all three hosts." wait fi diff --git a/docs/acceptance-scenarios.md b/docs/acceptance-scenarios.md index 85020a1..cc14f4c 100644 --- a/docs/acceptance-scenarios.md +++ b/docs/acceptance-scenarios.md @@ -193,3 +193,17 @@ Scenario: I4 Stale timestamp is rejected Then Ingress responds 401 And ingress.Outbox has no row for that EventId ``` + +## Observability + +The test records spans in process with an `ActivityListener`; no exporter, no dashboard. + +```gherkin +# Illustrates: ADR 010 (one trace per webhook, propagated across the outbox and Event Hubs) +Scenario: O1 One webhook produces one trace through Ingress, the relay and both workers + When candidate "c-o1" applies in FakeTeamtailor + Then the request span of its candidate.create webhook is the root of a trace + And the webhook span of that event is a child of the request span + And the send span of that event is a child of the webhook span + And every process span of that event in "sync-worker" and in "matching-worker" is a child of the send span +``` diff --git a/docs/adr/010-observability.md b/docs/adr/010-observability.md index 1b62864..e06beb0 100644 --- a/docs/adr/010-observability.md +++ b/docs/adr/010-observability.md @@ -14,32 +14,76 @@ at Event Hubs (the envelope carries ids only) and in the retry loop (one call ca ## Decision - **OpenTelemetry over OTLP**, W3C trace context. Locally the Aspire Dashboard runs in docker compose. The exporter - is registered only when `OTEL_EXPORTER_OTLP_ENDPOINT` is set, so tests and CI export nothing. Simulators are not - instrumented: they stand in for external systems. + is registered only when `OTEL_EXPORTER_OTLP_ENDPOINT` is set, so tests and CI export nothing; `OTEL_SERVICE_NAME` + is then required. Simulators are not instrumented: they stand in for external systems. - **One project, `TalentSync.Infrastructure.Observability`** (`AddObservability`), referenced only by the hosts. - It holds one concern, so it is not a Common project (ADR 001). Our spans and metrics use the BCL `ActivitySource` - and `Meter` where the work happens; the SDK subscribes to `TalentSync.*`. No telemetry in `*.Domain` or `*.Application`. -- **Explicit propagation, one trace per webhook.** Ingress stores the trace context in `ingress.Outbox.TraceParent`. - The relay's publish span is a child of it and writes `traceparent` to `EventData.Properties`, never to the envelope. + It holds one concern, so it is not a Common project (ADR 001). The SDK subscribes to `TalentSync.*`. + No telemetry in `*.Domain` or `*.Application`. +- **One telemetry class per context, where the work happens:** `IngressTelemetry`, `ConsumerTelemetry` + (`Infrastructure.Messaging`), `MatchingTelemetry`. Each holds a static BCL `ActivitySource` and its instruments, + created through `IMeterFactory`, is a singleton injected into the code it observes, and never throws: an unknown + outcome is tagged `unknown`, never an exception that would fail the event. +- **The LLM call is instrumented by `OpenTelemetryChatClient`** (`Microsoft.Extensions.AI`, gen_ai semantic + conventions: the `chat` span, token usage and duration) under the source name `TalentSync.Matching`. +- **SQL calls are spans** through the SqlClient instrumentation, which also sees EF Core's commands. +- **Names follow the semantic conventions** where they exist: `send ats-events` and `process ats-events` with + `messaging.*` attributes, `error.type`, `gen_ai.*`. Our own attributes are `talentsync..` + (`talentsync.consumer.outcome`, `talentsync.enrichment.outcome`); ids are `talentsync.event_id`, + `talentsync.event_type`, `talentsync.candidate_id` in spans and logs alike. +- **Explicit propagation, one trace per webhook.** Our hosts are the edge and ignore the trace context of incoming + requests: the sender's context is untrusted, and one sender trace can carry several webhooks. So the webhook + request span is the root of a new trace, and the webhook span is its child. + Ingress stores the webhook span's context in `ingress.Outbox.TraceParent`. + The relay's send span is a child of it and writes `traceparent` to `EventData.Properties`, never to the envelope. The consumer span is its child. A missing or invalid value starts a new trace and never fails the event. - **One consumer span per attempt**, tagged with the outcome (`handled`, `skipped_duplicate`, `parked`, ...). A span is exported when it ends, so one span around the retry loop would stay invisible during an outage. -- **Logs carry the trace** (OpenTelemetry logging provider) and a scope with `EventId`, `EventType`, `CandidateId`. -- **Invariant 14 covers telemetry:** ids only; never the CV, the prompt, model output or response bodies. +- **Metrics** answer what a trace cannot: consumer attempts by outcome, event age from webhook to effect, outbox + publish delay, enrichment calls by outcome, and the gen_ai token usage and call duration. The LLM call is a span; + a reused stored suggestion has none. Our histograms are in seconds and set their bucket boundaries as instrument + advice: the SDK's default buckets are sized for milliseconds. +- **Logs carry the trace** (OpenTelemetry logging provider) and a scope with the event's ids, named like the span tags. +- **Invariant 14 covers telemetry:** ids only; never the CV, the prompt, model output or response bodies. A failed span + carries the error status and `error.type`, never the exception message, which can echo a request: the message and + stack trace are in the log, which carries the trace id. `SpanExportPolicy` clears the status description of every + span before export, ours and the libraries' alike. `OpenTelemetryChatClient.EnableSensitiveData` is set to false + explicitly, so `OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT` cannot turn it on. +- **A client span without a parent is not exported** (`SpanExportPolicy`): the relay polling the outbox, migrations. + It belongs to no flow, and every poll would be a trace. +- **The checkpoint is our own span**, `checkpoint ats-events` (a settle operation), a child of the attempt that writes + it, with the Blob container that holds it in `talentsync.checkpoint.container`. The Blob SDK's HTTP requests are filtered out by their user agent: ownership polling and a checkpoint's + update-or-create round trips, whose first 404 per partition is expected, are the SDK's internals. ## Consequences -- One trace shows the request, the outbox delay, both consumer groups and every HTTP and LLM call. +- One trace shows the request, the outbox delay, both consumer groups, every HTTP, SQL and LLM call, and each + checkpoint. - A relay double publish shows up in the same trace as a `skipped_duplicate` branch. A webhook that Teamtailor sends twice is a new trace, found by the `EventId` tag. - One more column and migration in `ingress`. Rows published before it start their own trace. - Consumer spans can start long after their parent ended (outage, replay). Valid, but long traces are harder to read. -- OpenTelemetry packages in one project, one more container in compose. +- OpenTelemetry packages in one project, one more container in compose, `Microsoft.Extensions.AI` in + `Matching.Infrastructure`. The gen_ai conventions are still experimental, so their names can change with the package. +- Our hosts ignore the sender's trace context only because the ASP.NET Core instrumentation does not extract it again + when the propagator is exactly `TraceContextPropagator`. That is the instrumentation's internal behaviour, pinned by + `EdgeTraceTests` with the SDK enabled; scenario O1 runs without it. +- HTTP client metrics include the Blob SDK's polling: unlike spans, they cannot be filtered per request. +- The checkpoint store's latency is in the `checkpoint` span only; a slow Blob request has no span of its own. ## Alternatives considered - **The Event Hubs SDK's own propagation.** Rejected: experimental switch, hidden in a library. - **Trace context in `EventEnvelope`.** Rejected: the envelope is a business contract, not transport metadata. - **A span link instead of a parent at the outbox.** Rejected: the dashboard would split one webhook into separate traces. +- **Continuing the sender's trace.** Rejected: an external caller would choose our trace id and its sampling. +- **A new trace at the webhook span, linked to the request span.** Rejected: dashboards show links only in span + details, so the request and the flow it starts would be separate traces. - **Setup duplicated in each host.** Rejected: three copies would drift. +- **A hand-written gen_ai decorator.** Rejected: `OpenTelemetryChatClient` implements the conventions, metrics + included, for one package. +- **EF Core instrumentation.** Rejected: still a beta package; the SqlClient instrumentation is stable and sees EF's + commands. +- **Static `Meter`s.** Rejected: `IMeterFactory` ties the meters to the host's container, and tests create their own. +- **The exception on the span (`AddException`).** Rejected: its message can carry data from the request (invariant 14), + and the log already holds it under the same trace id. - **Seq or Grafana.** Seq covers logs only; Grafana is heavier. Production only needs another OTLP endpoint. diff --git a/docs/architecture.md b/docs/architecture.md index e39f166..fb604b9 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -20,6 +20,7 @@ Three deployables, split by runtime behaviour, not by business capability. They | Event Hubs | Hub `ats-events`, ordered log per partition, independent read position per consumer group | Dead-letter (there is no DLQ) | | | Azurite | Checkpoints and partition ownership | | | | Simulators | Stand-ins for Teamtailor, CRM, Matching Platform; break on demand | Production behaviour | in-memory | +| Aspire Dashboard | Local view of the hosts' traces, metrics and logs over OTLP (ADR 010) | Run in tests or CI | in-memory | ## Repository layout @@ -40,6 +41,10 @@ src/ stage mapping from config, 404 → Deleted, 4xx / unmappable → Rejected, signature v2, webhook parsing, public AddTeamtailorWebhooks() (Ingress) and AddTeamtailorApi() (workers). → Recruitment.Domain + TalentSync.Infrastructure.Observability AddObservability(): OpenTelemetry traces, metrics and logs over OTLP, + subscribes to every TalentSync.* ActivitySource and Meter, HTTP and SQL + instrumentation, SpanExportPolicy (ADR 010). + Only hosts reference it. References nothing in src/ TalentSync.Infrastructure.Messaging EventProcessorClient hosting, consumer handler skeleton (retry loop, failure classification), HandlerResult, Inbox and ParkingLot entities with schema-parameterised EF config. → Contracts @@ -77,6 +82,7 @@ tests/ TalentSync.Matching.Infrastructure.Tests Matching Platform client: request body, status mapping, resilience TalentSync.Infrastructure.Teamtailor.Tests signature v2, webhook parsing, stage mapping, 404 → Deleted, 4xx → Rejected TalentSync.Infrastructure.Messaging.Tests consumer skeleton: inbox / parking lot mapping, failure classification, handler loop + TalentSync.Infrastructure.Observability.Tests request span is a root with the SDK enabled, SpanExportPolicy TalentSync.Sync.Infrastructure.Tests CRM client: status mapping, 404 on delete = done, resilience TalentSync.Matching.Evaluation.Tests prompt evaluation: metrics, golden set hash, case files, report TalentSync.IntegrationTests acceptance scenarios (see .claude/rules/testing.md) @@ -85,7 +91,7 @@ eval/ offline evaluation of the enrichment cases/ golden set: one JSON per fictional CV reports/ committed reports; local/ is gitignored run-eval.sh runs the evaluation locally; provider, runs, output and model as variables at the top -deploy/ docker-compose.yml (infra + simulators), simulator.Dockerfile, +deploy/ docker-compose.yml (infra + simulators + Aspire Dashboard), simulator.Dockerfile, eventhubs/Config.json, run-local.sh (containers + the three hosts) docs/ domain.md, architecture.md, acceptance-scenarios.md, diagrams.md, adr/ (index: adr/README.md), e2e.http (end-to-end demo by hand) @@ -105,7 +111,7 @@ Code is split by context first, then by layer, with a project wherever the compi | Layer | Shared | Ingress | Sync | Matching | |---|---|---|---|---| | Presentation + composition root | | `Ingress.Api` | `Sync.Worker` | `Matching.Worker` | -| Infrastructure | `Infrastructure.Teamtailor`, `Infrastructure.Messaging` | (inside `Ingress.Api`) | `Sync.Infrastructure` | `Matching.Infrastructure` | +| Infrastructure | `Infrastructure.Teamtailor`, `Infrastructure.Messaging`, `Infrastructure.Observability` | (inside `Ingress.Api`) | `Sync.Infrastructure` | `Matching.Infrastructure` | | Application | | | `Sync.Application` | `Matching.Application` | | Domain | `Recruitment.Domain` | | | `Matching.Domain` | | Contract between processes | `Contracts` | | | | diff --git a/docs/diagrams.md b/docs/diagrams.md index b4eec13..0e32fd9 100644 --- a/docs/diagrams.md +++ b/docs/diagrams.md @@ -26,6 +26,7 @@ flowchart TB mp["FakeMatchingPlatform"] llm["LLM / FakeChatClient"] blob[("Azurite: checkpoints")] + dash["Aspire Dashboard
traces, metrics, logs"] tt -->|webhook| ep --> outbox --> relay --> eh eh --> sh @@ -39,6 +40,9 @@ flowchart TB mh -->|upsert profile| mp sh -.-> blob mh -.-> blob + ing -.->|OTLP| dash + sync -.->|OTLP| dash + mat -.->|OTLP| dash ``` ## Key flow: application qualified @@ -66,6 +70,36 @@ sequenceDiagram MW->>MW: insert inbox, checkpoint ``` +The same flow is one trace in the Aspire Dashboard (ADR 010). Each webhook request is its root; the outbox and +Event Hubs hops are carried by `ingress.Outbox.TraceParent` and the `traceparent` event property. Both consumer +groups appear as branches, Sync.Worker's skipping the event. Span names as the dashboard shows them; the right-hand +column is a comment: + +``` +POST /webhooks/teamtailor ingress-api +└─ webhook job_application.update ingress-api + ├─ INSERT ingress.Outbox + └─ send ats-events ingress-api, after the outbox delay + ├─ process ats-events matching-worker, handled + │ ├─ SELECT [matching].[Inbox] + │ ├─ GET teamtailor /v1/job-applications + │ ├─ GET teamtailor /v1/candidates + │ ├─ SELECT [matching].[EnrichmentResults] + │ ├─ enrich enriched; only if no stored result + │ │ └─ chat claude-haiku-5-5 model, tokens; "chat" with the Fake provider + │ ├─ INSERT matching.EnrichmentResults + │ ├─ PUT matching-platform /profiles + │ ├─ INSERT matching.Inbox + │ └─ checkpoint ats-events + └─ process ats-events sync-worker, skipped_unhandled_type + └─ checkpoint ats-events +``` + +An `INSERT` span carries no table name: EF Core prefixes its commands with `SET` statements, and the SqlClient +instrumentation then summarises the command as `INSERT`. The table is in `db.query.text`; values are parameters, never +literals. `checkpoint` is our span around the event processor's checkpoint; the Blob SDK's own requests are not +traced. + ## Consumer handler: failure paths ```mermaid diff --git a/src/Ingress/TalentSync.Ingress.Api/DependencyInjection.cs b/src/Ingress/TalentSync.Ingress.Api/DependencyInjection.cs index 1d84437..52b811d 100644 --- a/src/Ingress/TalentSync.Ingress.Api/DependencyInjection.cs +++ b/src/Ingress/TalentSync.Ingress.Api/DependencyInjection.cs @@ -1,7 +1,9 @@ +using System.Diagnostics.Metrics; using Azure.Messaging.EventHubs; using Azure.Messaging.EventHubs.Producer; using Microsoft.EntityFrameworkCore; using Microsoft.Extensions.DependencyInjection.Extensions; +using TalentSync.Infrastructure.Observability; using TalentSync.Infrastructure.Teamtailor; using TalentSync.Ingress.Api.Outbox; using TalentSync.Ingress.Api.Persistence; @@ -42,11 +44,13 @@ public static IServiceCollection AddIngress(this IServiceCollection services, IC .Bind(configuration.GetSection(OutboxRelayOptions.SectionName)) .Validate(o => o.IsValid(), $"'{OutboxRelayOptions.SectionName}' needs BatchSize > 0, PollInterval > 0 and MaxBackoff >= PollInterval.") .ValidateOnStart(); + services.AddSingleton(provider => new IngressTelemetry(provider.GetRequiredService(), hubName)); services.AddSingleton(); services.AddHostedService(); services.TryAddSingleton(TimeProvider.System); services.AddTeamtailorWebhooks(configuration); + services.AddObservability(configuration); return services; } diff --git a/src/Ingress/TalentSync.Ingress.Api/IngressTelemetry.cs b/src/Ingress/TalentSync.Ingress.Api/IngressTelemetry.cs new file mode 100644 index 0000000..95a9e20 --- /dev/null +++ b/src/Ingress/TalentSync.Ingress.Api/IngressTelemetry.cs @@ -0,0 +1,90 @@ +using System.Diagnostics; +using System.Diagnostics.Metrics; +using TalentSync.Ingress.Api.Outbox; + +namespace TalentSync.Ingress.Api; + +/// +/// Spans and metrics of the webhook and the relay (ADR 010). Tags hold ids only, a failure only its exception type +/// (invariant 14). Never throws. +/// +internal sealed class IngressTelemetry +{ + public const string Name = "TalentSync.Ingress"; + + private const string EventIdTag = "talentsync.event_id"; + private const string EventTypeTag = "talentsync.event_type"; + private const string CandidateIdTag = "talentsync.candidate_id"; + + private static readonly ActivitySource Source = new(Name); + + private readonly string _hubName; + private readonly Histogram _publishDelay; + + public IngressTelemetry(IMeterFactory meterFactory, string hubName) + { + _hubName = hubName; + + // Milliseconds while the relay keeps up, up to hours while Event Hubs is down; the SDK's default buckets are + // sized for milliseconds. + _publishDelay = meterFactory.Create(Name).CreateHistogram( + "talentsync.outbox.publish_delay", + unit: "s", + description: "Time from storing a webhook to publishing it.", + tags: null, + advice: new InstrumentAdvice { HistogramBucketBoundaries = [0.01, 0.05, 0.1, 0.25, 0.5, 1, 2.5, 5, 10, 30, 60, 300, 900, 3600] }); + } + + /// A child of the webhook request span, which is the root of the trace. + public Activity? StartWebhook(string eventId, string eventName, string candidateId) + { + var activity = Source.StartActivity($"webhook {eventName}"); + activity?.SetTag(EventIdTag, eventId); + activity?.SetTag(EventTypeTag, eventName); + activity?.SetTag(CandidateIdTag, candidateId); + return activity; + } + + /// + /// Continues the webhook's trace across the outbox. A row without a stored context starts a new trace: the relay + /// loop has no ambient activity to fall back on. + /// + public Activity? StartSend(OutboxMessage message) + { + ActivityContext.TryParse(message.TraceParent, null, out var webhook); + var activity = Source.StartActivity($"send {_hubName}", ActivityKind.Producer, webhook); + activity?.SetTag("messaging.system", "eventhubs"); + activity?.SetTag("messaging.operation.type", "send"); + activity?.SetTag("messaging.operation.name", "send"); + activity?.SetTag("messaging.destination.name", _hubName); + activity?.SetTag("messaging.message.id", message.EventId); + activity?.SetTag(EventIdTag, message.EventId); + activity?.SetTag(EventTypeTag, message.Type); + activity?.SetTag(CandidateIdTag, message.CandidateId); + return activity; + } + + public void RecordPublished(OutboxMessage message) + { + if (message.PublishedAt is { } publishedAt) + { + _publishDelay.Record( + (publishedAt - message.CreatedAt).TotalSeconds, + new KeyValuePair(EventTypeTag, message.Type)); + } + } + + /// The exception type only: its message and stack trace go to the log, which carries the trace id. + public void Fail(Activity? activity, Exception exception) + { + activity?.SetTag("error.type", exception.GetType().FullName); + activity?.SetStatus(ActivityStatusCode.Error); + } + + /// + /// The value to store or send as a traceparent. Null without a listener, and for a non-W3C id, whose length + /// is unbounded and would fail the outbox insert. + /// + public string? TraceParentOf(Activity? activity) => + activity is { IdFormat: ActivityIdFormat.W3C } ? activity.Id : null; +} diff --git a/src/Ingress/TalentSync.Ingress.Api/Outbox/OutboxMessage.cs b/src/Ingress/TalentSync.Ingress.Api/Outbox/OutboxMessage.cs index b623229..732d9a8 100644 --- a/src/Ingress/TalentSync.Ingress.Api/Outbox/OutboxMessage.cs +++ b/src/Ingress/TalentSync.Ingress.Api/Outbox/OutboxMessage.cs @@ -22,6 +22,12 @@ public sealed class OutboxMessage /// UTC, from . public required DateTime CreatedAt { get; init; } + /// + /// W3C trace context of the webhook span, so the publish span continues its trace across the outbox (ADR 010). + /// Null without a listener, and for rows stored before the column existed. + /// + public string? TraceParent { get; init; } + /// UTC. Null until the relay has published the row. public DateTime? PublishedAt { get; set; } } diff --git a/src/Ingress/TalentSync.Ingress.Api/Outbox/OutboxPublisher.cs b/src/Ingress/TalentSync.Ingress.Api/Outbox/OutboxPublisher.cs index 228d012..3f19ca6 100644 --- a/src/Ingress/TalentSync.Ingress.Api/Outbox/OutboxPublisher.cs +++ b/src/Ingress/TalentSync.Ingress.Api/Outbox/OutboxPublisher.cs @@ -1,14 +1,30 @@ using Azure.Messaging.EventHubs; using Azure.Messaging.EventHubs.Producer; +using TalentSync.Contracts; namespace TalentSync.Ingress.Api.Outbox; /// Sends one outbox row to Event Hubs with partition key candidateId (invariant 10). -internal sealed class OutboxPublisher(EventHubProducerClient producer) +internal sealed class OutboxPublisher(EventHubProducerClient producer, IngressTelemetry telemetry) { - public Task PublishAsync(OutboxMessage message, CancellationToken cancellationToken) => - producer.SendAsync( - [new EventData(BinaryData.FromString(message.Envelope)) { ContentType = "application/json", MessageId = message.EventId }], - new SendEventOptions { PartitionKey = message.CandidateId }, - cancellationToken); + public async Task PublishAsync(OutboxMessage message, CancellationToken cancellationToken) + { + using var activity = telemetry.StartSend(message); + + var eventData = new EventData(BinaryData.FromString(message.Envelope)) { ContentType = "application/json", MessageId = message.EventId }; + if (telemetry.TraceParentOf(activity) is { } traceParent) + { + eventData.Properties[EventProperties.TraceParent] = traceParent; + } + + try + { + await producer.SendAsync([eventData], new SendEventOptions { PartitionKey = message.CandidateId }, cancellationToken); + } + catch (Exception exception) + { + telemetry.Fail(activity, exception); + throw; + } + } } diff --git a/src/Ingress/TalentSync.Ingress.Api/Outbox/OutboxRelay.cs b/src/Ingress/TalentSync.Ingress.Api/Outbox/OutboxRelay.cs index 3e5b15b..789795b 100644 --- a/src/Ingress/TalentSync.Ingress.Api/Outbox/OutboxRelay.cs +++ b/src/Ingress/TalentSync.Ingress.Api/Outbox/OutboxRelay.cs @@ -19,6 +19,7 @@ namespace TalentSync.Ingress.Api.Outbox; internal sealed class OutboxRelay( IDbContextFactory dbContextFactory, OutboxPublisher publisher, + IngressTelemetry telemetry, IOptions options, TimeProvider timeProvider, ILogger logger) : BackgroundService @@ -98,6 +99,7 @@ private async Task RelayBatchAsync(CancellationToken stoppingToken) row.PublishedAt = timeProvider.GetUtcNow().UtcDateTime; published++; + telemetry.RecordPublished(row); logger.LogInformation("Published {EventId} ({EventName}) for candidate {CandidateId}", row.EventId, row.Type, row.CandidateId); } diff --git a/src/Ingress/TalentSync.Ingress.Api/Persistence/Migrations/20261015131911_AddOutboxTraceParent.Designer.cs b/src/Ingress/TalentSync.Ingress.Api/Persistence/Migrations/20261015131911_AddOutboxTraceParent.Designer.cs new file mode 100644 index 0000000..6e26e9a --- /dev/null +++ b/src/Ingress/TalentSync.Ingress.Api/Persistence/Migrations/20261015131911_AddOutboxTraceParent.Designer.cs @@ -0,0 +1,80 @@ +// +using System; +using Microsoft.EntityFrameworkCore; +using Microsoft.EntityFrameworkCore.Infrastructure; +using Microsoft.EntityFrameworkCore.Metadata; +using Microsoft.EntityFrameworkCore.Migrations; +using Microsoft.EntityFrameworkCore.Storage.ValueConversion; +using TalentSync.Ingress.Api.Persistence; + +#nullable disable + +namespace TalentSync.Ingress.Api.Persistence.Migrations +{ + [DbContext(typeof(IngressDbContext))] + [Migration("20261015131911_AddOutboxTraceParent")] + partial class AddOutboxTraceParent + { + /// + protected override void BuildTargetModel(ModelBuilder modelBuilder) + { +#pragma warning disable 612, 618 + modelBuilder + .HasDefaultSchema("ingress") + .HasAnnotation("ProductVersion", "10.0.12") + .HasAnnotation("Relational:MaxIdentifierLength", 128); + + SqlServerModelBuilderExtensions.UseIdentityColumns(modelBuilder); + + modelBuilder.Entity("TalentSync.Ingress.Api.Outbox.OutboxMessage", b => + { + b.Property("Id") + .ValueGeneratedOnAdd() + .HasColumnType("bigint"); + + SqlServerPropertyBuilderExtensions.UseIdentityColumn(b.Property("Id")); + + b.Property("CandidateId") + .IsRequired() + .HasMaxLength(100) + .HasColumnType("nvarchar(100)"); + + b.Property("CreatedAt") + .HasColumnType("datetime2"); + + b.Property("Envelope") + .IsRequired() + .HasColumnType("json"); + + b.Property("EventId") + .IsRequired() + .HasMaxLength(64) + .IsUnicode(false) + .HasColumnType("char(64)") + .IsFixedLength(); + + b.Property("PublishedAt") + .HasColumnType("datetime2"); + + b.Property("TraceParent") + .HasMaxLength(55) + .IsUnicode(false) + .HasColumnType("varchar(55)"); + + b.Property("Type") + .IsRequired() + .HasMaxLength(100) + .HasColumnType("nvarchar(100)"); + + b.HasKey("Id"); + + b.HasIndex("Id") + .HasDatabaseName("IX_Outbox_Unpublished") + .HasFilter("[PublishedAt] IS NULL"); + + b.ToTable("Outbox", "ingress"); + }); +#pragma warning restore 612, 618 + } + } +} diff --git a/src/Ingress/TalentSync.Ingress.Api/Persistence/Migrations/20261015131911_AddOutboxTraceParent.cs b/src/Ingress/TalentSync.Ingress.Api/Persistence/Migrations/20261015131911_AddOutboxTraceParent.cs new file mode 100644 index 0000000..f5d061e --- /dev/null +++ b/src/Ingress/TalentSync.Ingress.Api/Persistence/Migrations/20261015131911_AddOutboxTraceParent.cs @@ -0,0 +1,32 @@ +using Microsoft.EntityFrameworkCore.Migrations; + +#nullable disable + +namespace TalentSync.Ingress.Api.Persistence.Migrations +{ + /// + public partial class AddOutboxTraceParent : Migration + { + /// + protected override void Up(MigrationBuilder migrationBuilder) + { + migrationBuilder.AddColumn( + name: "TraceParent", + schema: "ingress", + table: "Outbox", + type: "varchar(55)", + unicode: false, + maxLength: 55, + nullable: true); + } + + /// + protected override void Down(MigrationBuilder migrationBuilder) + { + migrationBuilder.DropColumn( + name: "TraceParent", + schema: "ingress", + table: "Outbox"); + } + } +} diff --git a/src/Ingress/TalentSync.Ingress.Api/Persistence/Migrations/IngressDbContextModelSnapshot.cs b/src/Ingress/TalentSync.Ingress.Api/Persistence/Migrations/IngressDbContextModelSnapshot.cs index 6b6386c..c0bd5c7 100644 --- a/src/Ingress/TalentSync.Ingress.Api/Persistence/Migrations/IngressDbContextModelSnapshot.cs +++ b/src/Ingress/TalentSync.Ingress.Api/Persistence/Migrations/IngressDbContextModelSnapshot.cs @@ -53,6 +53,11 @@ protected override void BuildModel(ModelBuilder modelBuilder) b.Property("PublishedAt") .HasColumnType("datetime2"); + b.Property("TraceParent") + .HasMaxLength(55) + .IsUnicode(false) + .HasColumnType("varchar(55)"); + b.Property("Type") .IsRequired() .HasMaxLength(100) diff --git a/src/Ingress/TalentSync.Ingress.Api/Persistence/OutboxMessageConfiguration.cs b/src/Ingress/TalentSync.Ingress.Api/Persistence/OutboxMessageConfiguration.cs index 31b39a3..7b617cf 100644 --- a/src/Ingress/TalentSync.Ingress.Api/Persistence/OutboxMessageConfiguration.cs +++ b/src/Ingress/TalentSync.Ingress.Api/Persistence/OutboxMessageConfiguration.cs @@ -19,6 +19,8 @@ public void Configure(EntityTypeBuilder builder) builder.Property(m => m.Envelope).HasColumnType("json"); builder.Property(m => m.CreatedAt).HasColumnType("datetime2"); builder.Property(m => m.PublishedAt).HasColumnType("datetime2"); + // Version 00 of W3C traceparent is exactly 55 characters; only W3C ids are stored. + builder.Property(m => m.TraceParent).HasMaxLength(55).IsUnicode(false); // The relay reads only unpublished rows in Id order; published rows drop out of the index. builder.HasIndex(m => m.Id) diff --git a/src/Ingress/TalentSync.Ingress.Api/Properties/launchSettings.json b/src/Ingress/TalentSync.Ingress.Api/Properties/launchSettings.json index ae08841..c1d4986 100644 --- a/src/Ingress/TalentSync.Ingress.Api/Properties/launchSettings.json +++ b/src/Ingress/TalentSync.Ingress.Api/Properties/launchSettings.json @@ -7,7 +7,9 @@ "launchBrowser": true, "applicationUrl": "http://localhost:5065", "environmentVariables": { - "ASPNETCORE_ENVIRONMENT": "Development" + "ASPNETCORE_ENVIRONMENT": "Development", + "OTEL_EXPORTER_OTLP_ENDPOINT": "http://localhost:4317", + "OTEL_SERVICE_NAME": "ingress-api" } }, "https": { @@ -16,7 +18,9 @@ "launchBrowser": true, "applicationUrl": "https://localhost:7291;http://localhost:5065", "environmentVariables": { - "ASPNETCORE_ENVIRONMENT": "Development" + "ASPNETCORE_ENVIRONMENT": "Development", + "OTEL_EXPORTER_OTLP_ENDPOINT": "http://localhost:4317", + "OTEL_SERVICE_NAME": "ingress-api" } } } diff --git a/src/Ingress/TalentSync.Ingress.Api/TalentSync.Ingress.Api.csproj b/src/Ingress/TalentSync.Ingress.Api/TalentSync.Ingress.Api.csproj index 9a56c8d..ca6a413 100644 --- a/src/Ingress/TalentSync.Ingress.Api/TalentSync.Ingress.Api.csproj +++ b/src/Ingress/TalentSync.Ingress.Api/TalentSync.Ingress.Api.csproj @@ -10,6 +10,7 @@ + diff --git a/src/Ingress/TalentSync.Ingress.Api/Webhooks/TeamtailorWebhookEndpoint.cs b/src/Ingress/TalentSync.Ingress.Api/Webhooks/TeamtailorWebhookEndpoint.cs index 917357b..b92f3b0 100644 --- a/src/Ingress/TalentSync.Ingress.Api/Webhooks/TeamtailorWebhookEndpoint.cs +++ b/src/Ingress/TalentSync.Ingress.Api/Webhooks/TeamtailorWebhookEndpoint.cs @@ -25,6 +25,7 @@ private static async Task HandleAsync( HttpRequest request, IWebhookParser parser, IngressDbContext db, + IngressTelemetry telemetry, TimeProvider timeProvider, ILoggerFactory loggerFactory, CancellationToken cancellationToken) @@ -42,7 +43,7 @@ private static async Task HandleAsync( switch (parser.Parse(rawBody)) { case WebhookParseResult.Parsed parsed: - return await StoreAsync(parsed.Notification, rawBody, db, timeProvider, logger, cancellationToken); + return await StoreAsync(parsed.Notification, rawBody, db, telemetry, timeProvider, logger, cancellationToken); case WebhookParseResult.Ignored ignored: // No consumer for this event (e.g. job.*): acknowledge so the sender does not treat it as a failure. @@ -64,6 +65,7 @@ private static async Task StoreAsync( WebhookNotification notification, byte[] rawBody, IngressDbContext db, + IngressTelemetry telemetry, TimeProvider timeProvider, ILogger logger, CancellationToken cancellationToken) @@ -71,6 +73,8 @@ private static async Task StoreAsync( var eventId = Convert.ToHexStringLower(SHA256.HashData(rawBody)); var receivedAt = timeProvider.GetUtcNow(); + using var activity = telemetry.StartWebhook(eventId, notification.EventName, notification.CandidateId); + var envelope = new EventEnvelope { EventId = eventId, @@ -90,6 +94,7 @@ private static async Task StoreAsync( Type = notification.EventName, Envelope = EventEnvelopeJson.Serialize(envelope), CreatedAt = receivedAt.UtcDateTime, + TraceParent = telemetry.TraceParentOf(activity), }); try @@ -102,6 +107,7 @@ private static async Task StoreAsync( { // Known limitation (ADR 004): Teamtailor may not retry a 503, so this event is lost. // The exception carries no PII: the row holds ids only and sensitive data logging is off. + telemetry.Fail(activity, exception); logger.LogError(exception, "Could not store webhook {EventId} ({EventName}) in the outbox", eventId, notification.EventName); return Results.StatusCode(StatusCodes.Status503ServiceUnavailable); } diff --git a/src/Matching/TalentSync.Matching.Infrastructure/DependencyInjection.cs b/src/Matching/TalentSync.Matching.Infrastructure/DependencyInjection.cs index 477e8e4..fafb2ca 100644 --- a/src/Matching/TalentSync.Matching.Infrastructure/DependencyInjection.cs +++ b/src/Matching/TalentSync.Matching.Infrastructure/DependencyInjection.cs @@ -1,3 +1,4 @@ +using System.Diagnostics.Metrics; using Anthropic; using Microsoft.EntityFrameworkCore; using Microsoft.Extensions.AI; @@ -34,7 +35,8 @@ public static IServiceCollection AddMatchingInfrastructure(this IServiceCollecti // Short blips only: attempt timeout, a few retries, circuit breaker; long outages are waited out by the // handler loop. Retrying the PUT is harmless: it is keyed by the candidate id, so a retry after a timeout // whose request did reach the platform repeats the same effect (invariant 2). - services.AddHttpClient((provider, client) => + // The client name labels its HTTP spans and resilience logs. + services.AddHttpClient("matching-platform", (provider, client) => client.BaseAddress = provider.GetRequiredService>().Value.BaseUrl) .AddStandardResilienceHandler(section.GetSection("Resilience")); @@ -56,14 +58,27 @@ public static IServiceCollection AddMatchingInfrastructure(this IServiceCollecti services.AddSingleton(provider => { var options = provider.GetRequiredService>().Value; - return options.Provider switch + IChatClient inner = options.Provider switch { LlmOptions.FakeProvider => provider.GetRequiredService(), LlmOptions.AnthropicProvider => CreateAnthropicChatClient(options.Anthropic), _ => throw new InvalidOperationException($"Unknown '{LlmOptions.SectionName}:Provider'."), }; + + // A gen_ai "chat" span with token usage, plus the gen_ai duration and token metrics (ADR 010). Set + // explicitly: OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT=true would otherwise put the CV and the + // model's answer into the span (invariant 14). + return inner.AsBuilder() + .UseOpenTelemetry(sourceName: MatchingTelemetry.Name, configure: client => client.EnableSensitiveData = false) + .Build(provider); }); - services.AddSingleton(); + + services.AddSingleton(provider => new MatchingTelemetry(provider.GetRequiredService())); + + // Singletons, so the circuit breaker sees every call of the worker. + services.AddSingleton(); + services.AddSingleton(provider => new TelemetryEnrichmentService( + provider.GetRequiredService(), provider.GetRequiredService())); return services; } diff --git a/src/Matching/TalentSync.Matching.Infrastructure/Llm/TelemetryEnrichmentService.cs b/src/Matching/TalentSync.Matching.Infrastructure/Llm/TelemetryEnrichmentService.cs new file mode 100644 index 0000000..e9cdc4e --- /dev/null +++ b/src/Matching/TalentSync.Matching.Infrastructure/Llm/TelemetryEnrichmentService.cs @@ -0,0 +1,25 @@ +using TalentSync.Matching.Application.Ports; + +namespace TalentSync.Matching.Infrastructure.Llm; + +/// +/// Decorates with an enrich span and a counter of outcomes (ADR 010). Under the +/// span, the chat span of OpenTelemetryChatClient separates the provider's answer from our validation: +/// invalid_output with a successful chat span means the model answered and our rules rejected it. +/// +/// +/// The use case calls the service only when no stored suggestion exists, so a reused suggestion has no span: in a +/// trace, a profile upsert without an enrich span before it means the stored suggestion was used. +/// +internal sealed class TelemetryEnrichmentService(IEnrichmentService inner, MatchingTelemetry telemetry) : IEnrichmentService +{ + public string PromptVersion => inner.PromptVersion; + + public async Task EnrichAsync(string cvText, CancellationToken cancellationToken) + { + using var activity = telemetry.StartEnrichment(inner.PromptVersion); + var outcome = await inner.EnrichAsync(cvText, cancellationToken); + telemetry.EndEnrichment(activity, outcome); + return outcome; + } +} diff --git a/src/Matching/TalentSync.Matching.Infrastructure/MatchingTelemetry.cs b/src/Matching/TalentSync.Matching.Infrastructure/MatchingTelemetry.cs new file mode 100644 index 0000000..10c582c --- /dev/null +++ b/src/Matching/TalentSync.Matching.Infrastructure/MatchingTelemetry.cs @@ -0,0 +1,56 @@ +using System.Diagnostics; +using System.Diagnostics.Metrics; +using TalentSync.Matching.Application.Ports; + +namespace TalentSync.Matching.Infrastructure; + +/// +/// Spans and metrics of Matching's own work (ADR 010). The LLM call itself is instrumented by +/// OpenTelemetryChatClient under the same name. Tags hold the outcome and the prompt version: never the prompt, +/// the CV or the model's answer (invariant 14). Never throws. +/// +internal sealed class MatchingTelemetry +{ + public const string Name = "TalentSync.Matching"; + + private const string OutcomeTag = "talentsync.enrichment.outcome"; + + private static readonly ActivitySource Source = new(Name); + + private readonly Counter _enrichments; + + public MatchingTelemetry(IMeterFactory meterFactory) + { + _enrichments = meterFactory.Create(Name).CreateCounter( + "talentsync.enrichment.calls", unit: "{call}", description: "Enrichment attempts by outcome."); + } + + public Activity? StartEnrichment(string promptVersion) + { + var activity = Source.StartActivity("enrich"); + activity?.SetTag("talentsync.enrichment.prompt_version", promptVersion); + return activity; + } + + public void EndEnrichment(Activity? activity, EnrichmentOutcome outcome) + { + var name = NameOf(outcome); + activity?.SetTag(OutcomeTag, name); + if (outcome is EnrichmentOutcome.Failed) + { + // No error.type: a failed enrichment is an outcome, not an exception (D5). + activity?.SetStatus(ActivityStatusCode.Error); + } + + _enrichments.Add(1, new KeyValuePair(OutcomeTag, name)); + } + + private static string NameOf(EnrichmentOutcome outcome) => outcome switch + { + EnrichmentOutcome.Enriched => "enriched", + EnrichmentOutcome.Failed { Reason: EnrichmentFailure.Timeout } => "timeout", + EnrichmentOutcome.Failed { Reason: EnrichmentFailure.Unavailable } => "unavailable", + EnrichmentOutcome.Failed { Reason: EnrichmentFailure.InvalidOutput } => "invalid_output", + _ => "unknown", + }; +} diff --git a/src/Matching/TalentSync.Matching.Infrastructure/TalentSync.Matching.Infrastructure.csproj b/src/Matching/TalentSync.Matching.Infrastructure/TalentSync.Matching.Infrastructure.csproj index bb2283e..fcda4dd 100644 --- a/src/Matching/TalentSync.Matching.Infrastructure/TalentSync.Matching.Infrastructure.csproj +++ b/src/Matching/TalentSync.Matching.Infrastructure/TalentSync.Matching.Infrastructure.csproj @@ -1,7 +1,7 @@ - + diff --git a/src/Matching/TalentSync.Matching.Worker/DependencyInjection.cs b/src/Matching/TalentSync.Matching.Worker/DependencyInjection.cs index 954070b..e44e2c2 100644 --- a/src/Matching/TalentSync.Matching.Worker/DependencyInjection.cs +++ b/src/Matching/TalentSync.Matching.Worker/DependencyInjection.cs @@ -1,4 +1,5 @@ using TalentSync.Infrastructure.Messaging; +using TalentSync.Infrastructure.Observability; using TalentSync.Infrastructure.Teamtailor; using TalentSync.Matching.Application.UseCases; using TalentSync.Matching.Infrastructure; @@ -17,6 +18,7 @@ public static IServiceCollection AddMatchingWorker(this IServiceCollection servi services.AddScoped(); services.AddMessaging(configuration); services.AddHealthChecks(); + services.AddObservability(configuration); return services; } diff --git a/src/Matching/TalentSync.Matching.Worker/Properties/launchSettings.json b/src/Matching/TalentSync.Matching.Worker/Properties/launchSettings.json index 10387c7..c90f120 100644 --- a/src/Matching/TalentSync.Matching.Worker/Properties/launchSettings.json +++ b/src/Matching/TalentSync.Matching.Worker/Properties/launchSettings.json @@ -7,7 +7,9 @@ "launchBrowser": false, "applicationUrl": "http://localhost:5067", "environmentVariables": { - "ASPNETCORE_ENVIRONMENT": "Development" + "ASPNETCORE_ENVIRONMENT": "Development", + "OTEL_EXPORTER_OTLP_ENDPOINT": "http://localhost:4317", + "OTEL_SERVICE_NAME": "matching-worker" } } } diff --git a/src/Matching/TalentSync.Matching.Worker/TalentSync.Matching.Worker.csproj b/src/Matching/TalentSync.Matching.Worker/TalentSync.Matching.Worker.csproj index eecd474..4a5e13d 100644 --- a/src/Matching/TalentSync.Matching.Worker/TalentSync.Matching.Worker.csproj +++ b/src/Matching/TalentSync.Matching.Worker/TalentSync.Matching.Worker.csproj @@ -13,6 +13,7 @@ + diff --git a/src/Shared/TalentSync.Contracts/EventProperties.cs b/src/Shared/TalentSync.Contracts/EventProperties.cs new file mode 100644 index 0000000..ad36dc4 --- /dev/null +++ b/src/Shared/TalentSync.Contracts/EventProperties.cs @@ -0,0 +1,11 @@ +namespace TalentSync.Contracts; + +/// +/// Application properties of a published event: transport metadata, kept out of , +/// which is the business contract. Consumers ignore properties they do not know (invariant 11). +/// +public static class EventProperties +{ + /// W3C trace context of the publish span (ADR 010). Missing or invalid starts a new trace. + public const string TraceParent = "traceparent"; +} diff --git a/src/Shared/TalentSync.Infrastructure.Messaging/DependencyInjection.cs b/src/Shared/TalentSync.Infrastructure.Messaging/DependencyInjection.cs index e848fd9..a798082 100644 --- a/src/Shared/TalentSync.Infrastructure.Messaging/DependencyInjection.cs +++ b/src/Shared/TalentSync.Infrastructure.Messaging/DependencyInjection.cs @@ -1,3 +1,4 @@ +using System.Diagnostics.Metrics; using Azure.Messaging.EventHubs; using Azure.Storage.Blobs; using Microsoft.EntityFrameworkCore; @@ -27,14 +28,13 @@ public static IServiceCollection AddMessaging(this IServic .Validate(o => o.IsValid(), $"'{ConsumerOptions.SectionName}' needs InitialBackoff > 0, MaxBackoff >= InitialBackoff and MaxUnclassifiedAttempts > 0.") .ValidateOnStart(); + var hubName = Required(configuration, "EventHubs:HubName"); + var consumerGroup = Required(configuration, "EventHubs:ConsumerGroup"); var checkpointContainer = new BlobContainerClient( Required(configuration, "Checkpoints:BlobConnectionString"), Required(configuration, "Checkpoints:Container")); var processor = new EventProcessorClient( - checkpointContainer, - Required(configuration, "EventHubs:ConsumerGroup"), - Required(configuration, "EventHubs:ConnectionString"), - Required(configuration, "EventHubs:HubName")); + checkpointContainer, consumerGroup, Required(configuration, "EventHubs:ConnectionString"), hubName); // Singletons (invariant 9): the processor holds the AMQP connection and the partition ownership. services.AddSingleton(checkpointContainer); @@ -45,6 +45,13 @@ public static IServiceCollection AddMessaging(this IServic services.AddScoped>(); services.AddScoped(); services.TryAddSingleton(TimeProvider.System); + // IMeterFactory comes from the host (the generic host registers it). + services.AddSingleton(provider => new ConsumerTelemetry( + provider.GetRequiredService(), + provider.GetRequiredService(), + hubName, + consumerGroup, + checkpointContainer.Name)); return services; } diff --git a/src/Shared/TalentSync.Infrastructure.Messaging/EventHubs/EventProcessorHost.cs b/src/Shared/TalentSync.Infrastructure.Messaging/EventHubs/EventProcessorHost.cs index c287043..d0bde90 100644 --- a/src/Shared/TalentSync.Infrastructure.Messaging/EventHubs/EventProcessorHost.cs +++ b/src/Shared/TalentSync.Infrastructure.Messaging/EventHubs/EventProcessorHost.cs @@ -4,6 +4,7 @@ using Microsoft.Extensions.DependencyInjection; using Microsoft.Extensions.Hosting; using Microsoft.Extensions.Logging; +using TalentSync.Contracts; using TalentSync.Infrastructure.Messaging.Handling; namespace TalentSync.Infrastructure.Messaging.EventHubs; @@ -51,7 +52,8 @@ private async Task OnProcessEventAsync(ProcessEventArgs args) await using var scope = scopeFactory.CreateAsyncScope(); var loop = scope.ServiceProvider.GetRequiredService(); - var received = new ReceivedEvent(args.Partition.PartitionId, args.Data.SequenceNumber, args.Data.EventBody.ToMemory()); + var traceParent = args.Data.Properties.TryGetValue(EventProperties.TraceParent, out var value) ? value as string : null; + var received = new ReceivedEvent(args.Partition.PartitionId, args.Data.SequenceNumber, args.Data.EventBody.ToMemory(), traceParent); await loop.ProcessAsync(received, args.UpdateCheckpointAsync, args.CancellationToken); } diff --git a/src/Shared/TalentSync.Infrastructure.Messaging/Handling/AttemptOutcome.cs b/src/Shared/TalentSync.Infrastructure.Messaging/Handling/AttemptOutcome.cs new file mode 100644 index 0000000..e49f513 --- /dev/null +++ b/src/Shared/TalentSync.Infrastructure.Messaging/Handling/AttemptOutcome.cs @@ -0,0 +1,25 @@ +namespace TalentSync.Infrastructure.Messaging.Handling; + +/// How one attempt of at one event ended. +internal enum AttemptOutcome +{ + /// The effect is done and the inbox row written. + Handled, + + /// The event is already in the inbox: a redelivery (invariant 1). + SkippedDuplicate, + + /// The worker does not handle this event type. + SkippedUnhandledType, + + Parked, + + /// Retried for as long as the outage lasts (invariant 8). + TransientFailure, + + /// Retried until the attempt limit, then parked. + UnclassifiedFailure, + + /// Shutdown or a lost partition: no checkpoint, the next owner of the partition processes the event again. + Cancelled, +} diff --git a/src/Shared/TalentSync.Infrastructure.Messaging/Handling/ConsumerTelemetry.cs b/src/Shared/TalentSync.Infrastructure.Messaging/Handling/ConsumerTelemetry.cs new file mode 100644 index 0000000..2759eb1 --- /dev/null +++ b/src/Shared/TalentSync.Infrastructure.Messaging/Handling/ConsumerTelemetry.cs @@ -0,0 +1,151 @@ +using System.Diagnostics; +using System.Diagnostics.Metrics; +using TalentSync.Contracts; + +namespace TalentSync.Infrastructure.Messaging.Handling; + +/// +/// Spans and metrics of (ADR 010). One span per attempt, not per event: a span is +/// exported when it ends, so one span around the whole retry loop would stay invisible for the length of an outage. +/// Tags hold ids only, a failure only its exception type (invariant 14). Never throws: telemetry must not turn a +/// handled event into a failed one. +/// +internal sealed class ConsumerTelemetry +{ + public const string Name = "TalentSync.Messaging"; + + // Also the keys of the loop's log scope, so logs and spans of one event share their attribute names. + public const string EventIdTag = "talentsync.event_id"; + public const string EventTypeTag = "talentsync.event_type"; + public const string CandidateIdTag = "talentsync.candidate_id"; + + private const string ConsumerGroupTag = "messaging.consumer.group.name"; + private const string OutcomeTag = "talentsync.consumer.outcome"; + + private static readonly ActivitySource Source = new(Name); + + private readonly string _hubName; + private readonly string _consumerGroup; + private readonly string _checkpointContainer; + private readonly TimeProvider _timeProvider; + private readonly Counter _attempts; + private readonly Histogram _eventAge; + + public ConsumerTelemetry( + IMeterFactory meterFactory, TimeProvider timeProvider, string hubName, string consumerGroup, string checkpointContainer) + { + _checkpointContainer = checkpointContainer; + _hubName = hubName; + _consumerGroup = consumerGroup; + _timeProvider = timeProvider; + + var meter = meterFactory.Create(Name); + _attempts = meter.CreateCounter( + "talentsync.consumer.attempts", unit: "{attempt}", description: "Processing attempts by outcome."); + + // From the webhook reaching Ingress to the effect being done: the outbox, the hub, the partition queue and the + // handler together. Notify-then-fetch makes this the delay before the CRM or the profile shows the change. + // Seconds normally, up to hours after an outage; the SDK's default buckets are sized for milliseconds. + _eventAge = meter.CreateHistogram( + "talentsync.consumer.event_age", + unit: "s", + description: "Time from receiving a webhook to handling its event.", + tags: null, + advice: new InstrumentAdvice { HistogramBucketBoundaries = [0.1, 0.25, 0.5, 1, 2.5, 5, 10, 30, 60, 300, 900, 3600] }); + } + + /// + /// Starts the span of one attempt as a child of the relay's send span. A missing or invalid traceparent + /// starts a new trace and never fails the event. + /// + public Attempt StartAttempt(ReceivedEvent received, EventEnvelope? envelope, int number) + { + ActivityContext.TryParse(received.TraceParent, null, out var send); + var activity = Source.StartActivity($"process {_hubName}", ActivityKind.Consumer, send); + activity?.SetTag("messaging.system", "eventhubs"); + activity?.SetTag("messaging.operation.type", "process"); + activity?.SetTag("messaging.operation.name", "process"); + activity?.SetTag("messaging.destination.name", _hubName); + activity?.SetTag("messaging.destination.partition.id", received.PartitionId); + activity?.SetTag(ConsumerGroupTag, _consumerGroup); + activity?.SetTag("messaging.message.id", envelope?.EventId); + activity?.SetTag(EventIdTag, envelope?.EventId); + activity?.SetTag(EventTypeTag, envelope?.Type); + activity?.SetTag(CandidateIdTag, envelope?.CandidateId); + activity?.SetTag("talentsync.consumer.attempt", number); + return new Attempt(this, activity, envelope); + } + + /// + /// Wraps the processor's checkpoint in a checkpoint {hub} span, a child of the attempt that writes it + /// (semantic conventions: a settle operation). The Blob SDK's own requests are filtered out of the traces. + /// + public Func TraceCheckpoint(Func checkpoint, ReceivedEvent received) => + async cancellationToken => + { + using var activity = Source.StartActivity($"checkpoint {_hubName}", ActivityKind.Client); + activity?.SetTag("messaging.system", "eventhubs"); + activity?.SetTag("messaging.operation.type", "settle"); + activity?.SetTag("messaging.operation.name", "checkpoint"); + activity?.SetTag("messaging.destination.name", _hubName); + activity?.SetTag("messaging.destination.partition.id", received.PartitionId); + activity?.SetTag(ConsumerGroupTag, _consumerGroup); + // Where the processor writes it: a Blob container (Azurite locally). The name only, never the connection string. + activity?.SetTag("talentsync.checkpoint.container", _checkpointContainer); + try + { + await checkpoint(cancellationToken); + } + catch (Exception exception) + { + activity?.SetTag("error.type", exception.GetType().FullName); + activity?.SetStatus(ActivityStatusCode.Error); + throw; + } + }; + + private static string NameOf(AttemptOutcome outcome) => outcome switch + { + AttemptOutcome.Handled => "handled", + AttemptOutcome.SkippedDuplicate => "skipped_duplicate", + AttemptOutcome.SkippedUnhandledType => "skipped_unhandled_type", + AttemptOutcome.Parked => "parked", + AttemptOutcome.TransientFailure => "transient_failure", + AttemptOutcome.UnclassifiedFailure => "unclassified_failure", + AttemptOutcome.Cancelled => "cancelled", + _ => "unknown", + }; + + /// One attempt. Never null, also without a listener, so the loop needs no null checks. + internal sealed class Attempt(ConsumerTelemetry telemetry, Activity? activity, EventEnvelope? envelope) : IDisposable + { + public void End(AttemptOutcome outcome) + { + var name = NameOf(outcome); + activity?.SetTag(OutcomeTag, name); + telemetry._attempts.Add( + 1, + new KeyValuePair(ConsumerGroupTag, telemetry._consumerGroup), + new KeyValuePair(EventTypeTag, envelope?.Type), + new KeyValuePair(OutcomeTag, name)); + + if (outcome == AttemptOutcome.Handled && envelope is not null) + { + telemetry._eventAge.Record( + (telemetry._timeProvider.GetUtcNow() - envelope.ReceivedAt).TotalSeconds, + new KeyValuePair(ConsumerGroupTag, telemetry._consumerGroup), + new KeyValuePair(EventTypeTag, envelope.Type)); + } + } + + /// The exception type only: its message and stack trace go to the log, which carries the trace id. + public void Fail(Exception exception, AttemptOutcome outcome) + { + activity?.SetTag("error.type", exception.GetType().FullName); + activity?.SetStatus(ActivityStatusCode.Error); + End(outcome); + } + + public void Dispose() => activity?.Dispose(); + } +} diff --git a/src/Shared/TalentSync.Infrastructure.Messaging/Handling/EventProcessingLoop.cs b/src/Shared/TalentSync.Infrastructure.Messaging/Handling/EventProcessingLoop.cs index c9c4dbf..f3feea9 100644 --- a/src/Shared/TalentSync.Infrastructure.Messaging/Handling/EventProcessingLoop.cs +++ b/src/Shared/TalentSync.Infrastructure.Messaging/Handling/EventProcessingLoop.cs @@ -19,6 +19,7 @@ internal sealed class EventProcessingLoop( IEventHandler handler, IConsumerStore store, IOptions options, + ConsumerTelemetry telemetry, TimeProvider timeProvider, ILogger logger) { @@ -47,23 +48,35 @@ internal sealed class EventProcessingLoop( /// Cancelled when the processor shuts down or loses the partition. public async Task ProcessAsync(ReceivedEvent received, Func checkpoint, CancellationToken cancellationToken) { + checkpoint = telemetry.TraceCheckpoint(checkpoint, received); var envelope = TryReadEnvelope(received.Body.Span); var failedAttempts = 0; var unclassifiedAttempts = 0; + // Every log of the use case and the clients carries these, without passing them around. Named like the span + // tags, so one attribute finds an event's logs and spans. + using var logScope = envelope is null ? null : logger.BeginScope(new Dictionary + { + [ConsumerTelemetry.EventIdTag] = envelope.EventId, + [ConsumerTelemetry.EventTypeTag] = envelope.Type, + [ConsumerTelemetry.CandidateIdTag] = envelope.CandidateId, + }); + try { await _retry.ExecuteAsync(async token => { + using var attempt = telemetry.StartAttempt(received, envelope, failedAttempts + 1); try { - await HandleOnceAsync(received, envelope, checkpoint, token); + attempt.End(await HandleOnceAsync(received, envelope, checkpoint, token)); } catch (Exception exception) when (!IsShutdown(exception, token)) { failedAttempts++; if (FailureClassifier.IsTransient(exception)) { + attempt.Fail(exception, AttemptOutcome.TransientFailure); logger.LogWarning(exception, "Transient failure on {EventId} (partition {PartitionId}, sequence {SequenceNumber}), attempt {Attempt}", envelope?.EventId, received.PartitionId, received.SequenceNumber, failedAttempts); throw; @@ -76,9 +89,16 @@ await _retry.ExecuteAsync(async token => if (unclassifiedAttempts >= _options.MaxUnclassifiedAttempts && await TryParkAsync(received, envelope, $"retry limit exceeded: {exception.GetType().FullName}", checkpoint, token)) { + attempt.Fail(exception, AttemptOutcome.Parked); return; } + attempt.Fail(exception, AttemptOutcome.UnclassifiedFailure); + throw; + } + catch (OperationCanceledException) when (token.IsCancellationRequested) + { + attempt.End(AttemptOutcome.Cancelled); throw; } }, cancellationToken); @@ -93,7 +113,7 @@ await _retry.ExecuteAsync(async token => private static bool IsShutdown(Exception exception, CancellationToken cancellationToken) => exception is OperationCanceledException && cancellationToken.IsCancellationRequested; - private async Task HandleOnceAsync( + private async Task HandleOnceAsync( ReceivedEvent received, EventEnvelope? envelope, Func checkpoint, CancellationToken cancellationToken) { if (envelope is null) @@ -101,14 +121,19 @@ private async Task HandleOnceAsync( await store.ParkAsync(NewParkedEvent(received, envelope: null, UnreadableEnvelope), cancellationToken); await checkpoint(cancellationToken); logger.LogWarning("Parked unreadable event (partition {PartitionId}, sequence {SequenceNumber})", received.PartitionId, received.SequenceNumber); - return; + return AttemptOutcome.Parked; + } + + if (!handler.Handles(envelope.Type)) + { + await checkpoint(cancellationToken); + return AttemptOutcome.SkippedUnhandledType; } - if (!handler.Handles(envelope.Type) - || await store.IsHandledAsync(envelope.Source, envelope.EventId, cancellationToken)) + if (await store.IsHandledAsync(envelope.Source, envelope.EventId, cancellationToken)) { await checkpoint(cancellationToken); - return; + return AttemptOutcome.SkippedDuplicate; } var result = await handler.HandleAsync(envelope, cancellationToken); @@ -120,7 +145,7 @@ private async Task HandleOnceAsync( await checkpoint(cancellationToken); logger.LogWarning("Parked {EventId} ({EventType}) for candidate {CandidateId}: {Reason}", envelope.EventId, envelope.Type, envelope.CandidateId, permanent.Reason); - return; + return AttemptOutcome.Parked; case HandlerResult.DoneResult: // After the effect, never before (invariant 3). A crash here repeats the effect on redelivery, @@ -130,7 +155,7 @@ await store.MarkHandledAsync( cancellationToken); await checkpoint(cancellationToken); logger.LogInformation("Handled {EventId} ({EventType}) for candidate {CandidateId}", envelope.EventId, envelope.Type, envelope.CandidateId); - return; + return AttemptOutcome.Handled; default: throw new InvalidOperationException($"Unknown handler result {result.GetType().Name}."); diff --git a/src/Shared/TalentSync.Infrastructure.Messaging/Handling/ReceivedEvent.cs b/src/Shared/TalentSync.Infrastructure.Messaging/Handling/ReceivedEvent.cs index b75a3b1..ee64d89 100644 --- a/src/Shared/TalentSync.Infrastructure.Messaging/Handling/ReceivedEvent.cs +++ b/src/Shared/TalentSync.Infrastructure.Messaging/Handling/ReceivedEvent.cs @@ -4,4 +4,6 @@ namespace TalentSync.Infrastructure.Messaging.Handling; /// An event as read from one partition, independent of the Event Hubs SDK types. /// and identify it even when its body is not an envelope. /// -internal sealed record ReceivedEvent(string PartitionId, long SequenceNumber, ReadOnlyMemory Body); +/// The traceparent property set by the relay (ADR 010); null when missing. +internal sealed record ReceivedEvent( + string PartitionId, long SequenceNumber, ReadOnlyMemory Body, string? TraceParent); diff --git a/src/Shared/TalentSync.Infrastructure.Observability/DependencyInjection.cs b/src/Shared/TalentSync.Infrastructure.Observability/DependencyInjection.cs new file mode 100644 index 0000000..c62e1e7 --- /dev/null +++ b/src/Shared/TalentSync.Infrastructure.Observability/DependencyInjection.cs @@ -0,0 +1,104 @@ +using System.Diagnostics; +using Microsoft.Extensions.Configuration; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.DependencyInjection.Extensions; +using Microsoft.Extensions.Http; +using OpenTelemetry; +using OpenTelemetry.Context.Propagation; +using OpenTelemetry.Exporter; +using OpenTelemetry.Logs; +using OpenTelemetry.Metrics; +using OpenTelemetry.Resources; +using OpenTelemetry.Trace; + +namespace TalentSync.Infrastructure.Observability; + +/// Exports traces, metrics and logs of one host over OTLP (ADR 010). +public static class DependencyInjection +{ + /// Every ActivitySource and Meter of ours is named with this prefix. + public const string SourcePrefix = "TalentSync."; + + private const string EndpointKey = "OTEL_EXPORTER_OTLP_ENDPOINT"; + private const string ServiceNameKey = "OTEL_SERVICE_NAME"; + + /// + /// Registers no exporter unless OTEL_EXPORTER_OTLP_ENDPOINT is set: without one every span would be + /// created and dropped, and tests and CI would try to reach a collector that is not there. + /// + public static IServiceCollection AddObservability(this IServiceCollection services, IConfiguration configuration) + { + // Registered with or without an exporter: whose context a request continues is a trust decision, not an export + // setting, and tests must see the same traces as a running system. ASP.NET Core resolves it from DI. + services.AddSingleton(new EdgeTraceContextPropagator()); + + var endpoint = configuration[EndpointKey]; + if (string.IsNullOrWhiteSpace(endpoint)) + { + return services; + } + + // The SDK would fall back to "unknown_service", and three hosts would merge into one in the dashboard. + var serviceName = configuration[ServiceNameKey] + ?? throw new InvalidOperationException($"Missing configuration '{ServiceNameKey}'."); + var collector = new Uri(endpoint); + + // With the SDK's default (W3C trace context and baggage) the ASP.NET Core instrumentation extracts the incoming + // context a second time and replaces the request span with one parented to the sender's trace. With exactly + // TraceContextPropagator it trusts the request span, which EdgeTraceContextPropagator has made a root. That is + // the instrumentation's internal behaviour, pinned by EdgeTraceTests. Process-wide, like the default it replaces. + Sdk.SetDefaultTextMapPropagator(new TraceContextPropagator()); + + services.AddOpenTelemetry() + .ConfigureResource(resource => resource.AddService(serviceName)) + .WithTracing(tracing => tracing + // Every request is a root (EdgeTraceContextPropagator), so ParentBased would sample everything too; + // AlwaysOn says so without depending on that. + .SetSampler(new AlwaysOnSampler()) + .AddSource(SourcePrefix + "*") + // Liveness probes would bury the flow. + .AddAspNetCoreInstrumentation(options => options.Filter = context => !context.Request.Path.StartsWithSegments("/health")) + .AddHttpClientInstrumentation(options => + { + // The Blob SDK's requests are its internals (ownership polling, a checkpoint's update-or-create + // round trips with an expected 404). The consumer's own checkpoint span says what happened. + options.FilterHttpRequestMessage = request => !IsAzureSdk(request); + options.EnrichWithHttpRequestMessage = HttpPeerName.Enrich; + }) + // EF Core sends its commands through SqlClient. Parameter values are never recorded + // (SetDbQueryParameters is off) and EF does not inline them, so db.query.text holds no PII (invariant 14). + .AddSqlClientInstrumentation() + // Processors run in the order they are added: the policy must see a span before the exporter does. + .AddProcessor(new SpanExportPolicy()) + .AddOtlpExporter(options => UseCollector(options, collector))) + .WithMetrics(metrics => metrics + .AddMeter(SourcePrefix + "*") + .AddAspNetCoreInstrumentation() + // Includes the Blob SDK's polling of checkpoints and ownership, told apart by server.address: unlike + // the spans, these metrics cannot be filtered per request. + .AddHttpClientInstrumentation() + .AddSqlClientInstrumentation() + .AddMeter("System.Runtime") + .AddOtlpExporter(options => UseCollector(options, collector))) + .WithLogging( + logging => logging.AddOtlpExporter(options => UseCollector(options, collector)), + options => + { + options.IncludeScopes = true; + options.IncludeFormattedMessage = true; + }); + + services.TryAddEnumerable(ServiceDescriptor.Singleton()); + + return services; + } + + private static bool IsAzureSdk(HttpRequestMessage request) => + request.Headers.UserAgent.ToString().StartsWith("azsdk-net-", StringComparison.Ordinal); + + private static void UseCollector(OtlpExporterOptions options, Uri collector) + { + options.Endpoint = collector; + options.Protocol = OtlpExportProtocol.Grpc; + } +} diff --git a/src/Shared/TalentSync.Infrastructure.Observability/EdgeTraceContextPropagator.cs b/src/Shared/TalentSync.Infrastructure.Observability/EdgeTraceContextPropagator.cs new file mode 100644 index 0000000..21a0669 --- /dev/null +++ b/src/Shared/TalentSync.Infrastructure.Observability/EdgeTraceContextPropagator.cs @@ -0,0 +1,27 @@ +using System.Diagnostics; + +namespace TalentSync.Infrastructure.Observability; + +/// +/// Ignores the trace context of incoming HTTP requests, so every request to our hosts starts a new trace (ADR 010). +/// Our hosts are the edge: their only callers are webhook senders and probes, whose context is untrusted, and one +/// sender trace can carry several webhooks. Outgoing requests keep the default W3C injection. +/// +internal sealed class EdgeTraceContextPropagator : DistributedContextPropagator +{ + private static readonly DistributedContextPropagator Outgoing = CreateDefaultPropagator(); + + public override IReadOnlyCollection Fields => Outgoing.Fields; + + public override void Inject(Activity? activity, object? carrier, PropagatorSetterCallback? setter) => + Outgoing.Inject(activity, carrier, setter); + + public override void ExtractTraceIdAndState( + object? carrier, PropagatorGetterCallback? getter, out string? traceId, out string? traceState) + { + traceId = null; + traceState = null; + } + + public override IEnumerable>? ExtractBaggage(object? carrier, PropagatorGetterCallback? getter) => null; +} diff --git a/src/Shared/TalentSync.Infrastructure.Observability/HttpPeerName.cs b/src/Shared/TalentSync.Infrastructure.Observability/HttpPeerName.cs new file mode 100644 index 0000000..5642a78 --- /dev/null +++ b/src/Shared/TalentSync.Infrastructure.Observability/HttpPeerName.cs @@ -0,0 +1,58 @@ +using System.Diagnostics; +using Microsoft.Extensions.Http; + +namespace TalentSync.Infrastructure.Observability; + +/// +/// Names an outgoing HTTP span after the system it calls, e.g. PUT crm /contacts instead of PUT. +/// The system is the name of the IHttpClientFactory client that sent the request, so the adapters only name +/// their clients and never reference this project. +/// +internal static class HttpPeerName +{ + private static readonly HttpRequestOptionsKey Key = new("talentsync.peer"); + + /// + /// Runs before the span exists: the span is started at the bottom of the handler chain, so the name travels on + /// the request. + /// + internal sealed class Filter : IHttpMessageHandlerBuilderFilter + { + public Action Configure(Action next) => builder => + { + next(builder); + if (!string.IsNullOrEmpty(builder.Name)) + { + builder.AdditionalHandlers.Insert(0, new Handler(builder.Name)); + } + }; + } + + public static void Enrich(Activity activity, HttpRequestMessage request) + { + if (request.Options.TryGetValue(Key, out var peer)) + { + activity.DisplayName = $"{request.Method} {peer} {Resource(request.RequestUri)}"; + activity.SetTag("peer.service", peer); + } + } + + // The path without its last segment, which is the id in every API we call (/contacts/{id}, /v1/candidates/{id}). + // A heuristic for /{collection}/{id} only: a call without an id, such as a list, would lose its collection name. + // It keeps the span name readable and free of ids. + private static string Resource(Uri? uri) + { + var path = uri?.AbsolutePath.TrimEnd('/') ?? ""; + var lastSlash = path.LastIndexOf('/'); + return lastSlash > 0 ? path[..lastSlash] : path; + } + + private sealed class Handler(string peer) : DelegatingHandler + { + protected override Task SendAsync(HttpRequestMessage request, CancellationToken cancellationToken) + { + request.Options.Set(Key, peer); + return base.SendAsync(request, cancellationToken); + } + } +} diff --git a/src/Shared/TalentSync.Infrastructure.Observability/SpanExportPolicy.cs b/src/Shared/TalentSync.Infrastructure.Observability/SpanExportPolicy.cs new file mode 100644 index 0000000..44a09f0 --- /dev/null +++ b/src/Shared/TalentSync.Infrastructure.Observability/SpanExportPolicy.cs @@ -0,0 +1,29 @@ +using System.Diagnostics; +using OpenTelemetry; + +namespace TalentSync.Infrastructure.Observability; + +/// +/// Decides what of a span leaves the process (ADR 010), for our spans and the libraries' alike. Registered before the +/// exporter, which skips a span that is not recorded. +/// +internal sealed class SpanExportPolicy : BaseProcessor +{ + public override void OnEnd(Activity activity) + { + // A client span without a parent belongs to no flow: the relay polling the outbox, migrations. Each poll would + // be a trace of its own and bury the flow. Calls inside a flow (inbox, parking lot) have a parent and are kept. + if (activity.Kind == ActivityKind.Client && activity.ParentSpanId == default) + { + activity.ActivityTraceFlags &= ~ActivityTraceFlags.Recorded; + return; + } + + // Invariant 14: a status description is an exception message, which can echo a request (a provider's error, + // a duplicate key). The status code and error.type stay; the message is in the log, which carries the trace id. + if (activity.StatusDescription is not null) + { + activity.SetStatus(activity.Status); + } + } +} diff --git a/src/Shared/TalentSync.Infrastructure.Observability/TalentSync.Infrastructure.Observability.csproj b/src/Shared/TalentSync.Infrastructure.Observability/TalentSync.Infrastructure.Observability.csproj new file mode 100644 index 0000000..9579ad8 --- /dev/null +++ b/src/Shared/TalentSync.Infrastructure.Observability/TalentSync.Infrastructure.Observability.csproj @@ -0,0 +1,15 @@ + + + + + + + + + + + + + + + diff --git a/src/Shared/TalentSync.Infrastructure.Teamtailor/DependencyInjection.cs b/src/Shared/TalentSync.Infrastructure.Teamtailor/DependencyInjection.cs index e22b119..6909cc9 100644 --- a/src/Shared/TalentSync.Infrastructure.Teamtailor/DependencyInjection.cs +++ b/src/Shared/TalentSync.Infrastructure.Teamtailor/DependencyInjection.cs @@ -46,7 +46,8 @@ public static IServiceCollection AddTeamtailorApi(this IServiceCollection servic // HTTP pipeline, outermost first: one rate-limit retry, then the standard handler (total timeout, retry, // circuit breaker, attempt timeout) for short blips. Long outages are waited out by the handler loop, which // runs this whole pipeline again on every attempt. - var apiClient = services.AddHttpClient((provider, client) => + // The client name labels its HTTP spans and resilience logs. + var apiClient = services.AddHttpClient("teamtailor", (provider, client) => { var options = provider.GetRequiredService>().Value; client.BaseAddress = options.BaseUrl; diff --git a/src/Sync/TalentSync.Sync.Infrastructure/DependencyInjection.cs b/src/Sync/TalentSync.Sync.Infrastructure/DependencyInjection.cs index 1018170..3dba466 100644 --- a/src/Sync/TalentSync.Sync.Infrastructure/DependencyInjection.cs +++ b/src/Sync/TalentSync.Sync.Infrastructure/DependencyInjection.cs @@ -26,7 +26,8 @@ public static IServiceCollection AddSyncInfrastructure(this IServiceCollection s // Short blips only: attempt timeout, a few retries, circuit breaker; long outages are waited out by the // handler loop. The standard handler retries PUT and DELETE too. Harmless here: both are keyed by the // candidate id, so a retry after a timeout whose request did reach the CRM repeats the same effect (invariant 2). - services.AddHttpClient((provider, client) => + // The client name labels its HTTP spans and resilience logs. + services.AddHttpClient("crm", (provider, client) => client.BaseAddress = provider.GetRequiredService>().Value.BaseUrl) .AddStandardResilienceHandler(section.GetSection("Resilience")); diff --git a/src/Sync/TalentSync.Sync.Worker/DependencyInjection.cs b/src/Sync/TalentSync.Sync.Worker/DependencyInjection.cs index 1d8367e..00ed7f8 100644 --- a/src/Sync/TalentSync.Sync.Worker/DependencyInjection.cs +++ b/src/Sync/TalentSync.Sync.Worker/DependencyInjection.cs @@ -1,4 +1,5 @@ using TalentSync.Infrastructure.Messaging; +using TalentSync.Infrastructure.Observability; using TalentSync.Infrastructure.Teamtailor; using TalentSync.Sync.Application.UseCases; using TalentSync.Sync.Infrastructure; @@ -17,6 +18,7 @@ public static IServiceCollection AddSyncWorker(this IServiceCollection services, services.AddScoped(); services.AddMessaging(configuration); services.AddHealthChecks(); + services.AddObservability(configuration); return services; } diff --git a/src/Sync/TalentSync.Sync.Worker/Properties/launchSettings.json b/src/Sync/TalentSync.Sync.Worker/Properties/launchSettings.json index 71e9d83..adfd567 100644 --- a/src/Sync/TalentSync.Sync.Worker/Properties/launchSettings.json +++ b/src/Sync/TalentSync.Sync.Worker/Properties/launchSettings.json @@ -7,7 +7,9 @@ "launchBrowser": false, "applicationUrl": "http://localhost:5066", "environmentVariables": { - "ASPNETCORE_ENVIRONMENT": "Development" + "ASPNETCORE_ENVIRONMENT": "Development", + "OTEL_EXPORTER_OTLP_ENDPOINT": "http://localhost:4317", + "OTEL_SERVICE_NAME": "sync-worker" } } } diff --git a/src/Sync/TalentSync.Sync.Worker/TalentSync.Sync.Worker.csproj b/src/Sync/TalentSync.Sync.Worker/TalentSync.Sync.Worker.csproj index ee8d4ae..c0e1259 100644 --- a/src/Sync/TalentSync.Sync.Worker/TalentSync.Sync.Worker.csproj +++ b/src/Sync/TalentSync.Sync.Worker/TalentSync.Sync.Worker.csproj @@ -13,6 +13,7 @@ + diff --git a/tests/TalentSync.Infrastructure.Messaging.Tests/EventProcessingLoopTests.cs b/tests/TalentSync.Infrastructure.Messaging.Tests/EventProcessingLoopTests.cs index f1ec72c..4cdd512 100644 --- a/tests/TalentSync.Infrastructure.Messaging.Tests/EventProcessingLoopTests.cs +++ b/tests/TalentSync.Infrastructure.Messaging.Tests/EventProcessingLoopTests.cs @@ -1,3 +1,4 @@ +using System.Diagnostics; using System.Net; using System.Text; using Microsoft.Extensions.Logging.Abstractions; @@ -208,6 +209,38 @@ await loop.ProcessAsync(Received(Serialize(Envelope())), _ => Assert.Equal(1, _checkpoints); } + [Fact] + public async Task Checkpoint_IsASpanOfTheAttempt() + { + using var spans = new SpanCollector(); + + await ProcessAsync(Envelope()); + + var process = Assert.Single(spans.Stopped, span => span.DisplayName == "process test-hub"); + var checkpoint = Assert.Single(spans.Stopped, span => span.DisplayName == "checkpoint test-hub"); + Assert.Equal(process.SpanId, checkpoint.ParentSpanId); + Assert.Equal(ActivityStatusCode.Unset, checkpoint.Status); + Assert.Equal("test-checkpoints", checkpoint.GetTagItem("talentsync.checkpoint.container")); + } + + [Fact] + public async Task FailedCheckpoint_MarksOnlyItsOwnSpan() + { + using var spans = new SpanCollector(); + var failures = 1; + + await Loop().ProcessAsync( + Received(Serialize(Envelope())), + _ => failures-- > 0 ? throw new TimeoutException() : Task.CompletedTask, + TestContext.Current.CancellationToken); + + var checkpoints = spans.Stopped.Where(span => span.DisplayName == "checkpoint test-hub").ToList(); + Assert.Equal(2, checkpoints.Count); + Assert.Equal(ActivityStatusCode.Error, checkpoints[0].Status); + Assert.Equal(typeof(TimeoutException).FullName, checkpoints[0].GetTagItem("error.type")); + Assert.Equal(ActivityStatusCode.Unset, checkpoints[1].Status); + } + private Task ProcessAsync(EventEnvelope envelope) => ProcessAsync(Serialize(envelope)); private Task ProcessAsync(byte[] body) => @@ -228,10 +261,11 @@ private Task CountCheckpoint(CancellationToken cancellationToken) MaxBackoff = maxBackoff, MaxUnclassifiedAttempts = MaxUnclassifiedAttempts, }), + new ConsumerTelemetry(new TestMeterFactory(), timeProvider ?? TimeProvider.System, "test-hub", "test-consumer-group", "test-checkpoints"), timeProvider ?? TimeProvider.System, NullLogger.Instance); - private static ReceivedEvent Received(byte[] body) => new("0", 42, body); + private static ReceivedEvent Received(byte[] body) => new("0", 42, body, TraceParent: null); private static byte[] Serialize(EventEnvelope envelope) => Encoding.UTF8.GetBytes(EventEnvelopeJson.Serialize(envelope)); diff --git a/tests/TalentSync.Infrastructure.Messaging.Tests/SpanCollector.cs b/tests/TalentSync.Infrastructure.Messaging.Tests/SpanCollector.cs new file mode 100644 index 0000000..27ae208 --- /dev/null +++ b/tests/TalentSync.Infrastructure.Messaging.Tests/SpanCollector.cs @@ -0,0 +1,27 @@ +using System.Collections.Concurrent; +using System.Diagnostics; +using TalentSync.Infrastructure.Messaging.Handling; + +namespace TalentSync.Infrastructure.Messaging.Tests; + +/// Records the consumer skeleton's spans in the order they end. +internal sealed class SpanCollector : IDisposable +{ + private readonly ConcurrentQueue _stopped = new(); + private readonly ActivityListener _listener; + + public SpanCollector() + { + _listener = new ActivityListener + { + ShouldListenTo = source => source.Name == ConsumerTelemetry.Name, + Sample = (ref ActivityCreationOptions _) => ActivitySamplingResult.AllDataAndRecorded, + ActivityStopped = _stopped.Enqueue, + }; + ActivitySource.AddActivityListener(_listener); + } + + public IReadOnlyList Stopped => [.. _stopped]; + + public void Dispose() => _listener.Dispose(); +} diff --git a/tests/TalentSync.Infrastructure.Messaging.Tests/TestMeterFactory.cs b/tests/TalentSync.Infrastructure.Messaging.Tests/TestMeterFactory.cs new file mode 100644 index 0000000..47fa231 --- /dev/null +++ b/tests/TalentSync.Infrastructure.Messaging.Tests/TestMeterFactory.cs @@ -0,0 +1,18 @@ +using System.Diagnostics.Metrics; + +namespace TalentSync.Infrastructure.Messaging.Tests; + +/// Creates real meters, so the telemetry code runs; nothing listens to them. +internal sealed class TestMeterFactory : IMeterFactory +{ + private readonly List _meters = []; + + public Meter Create(MeterOptions options) + { + var meter = new Meter(options); + _meters.Add(meter); + return meter; + } + + public void Dispose() => _meters.ForEach(meter => meter.Dispose()); +} diff --git a/tests/TalentSync.Infrastructure.Observability.Tests/EdgeTraceTests.cs b/tests/TalentSync.Infrastructure.Observability.Tests/EdgeTraceTests.cs new file mode 100644 index 0000000..3735b0d --- /dev/null +++ b/tests/TalentSync.Infrastructure.Observability.Tests/EdgeTraceTests.cs @@ -0,0 +1,53 @@ +using System.Diagnostics; +using Microsoft.AspNetCore.Builder; +using Microsoft.AspNetCore.Hosting; +using Microsoft.AspNetCore.Http; +using Microsoft.AspNetCore.TestHost; +using Microsoft.Extensions.Configuration; + +namespace TalentSync.Infrastructure.Observability.Tests; + +/// +/// AddObservability with the SDK enabled, as in a running host. Scenario O1 runs without an endpoint, so only +/// this test covers the ASP.NET Core instrumentation, which extracts the incoming context a second time and would +/// replace the request span before the endpoint runs. +/// +public sealed class EdgeTraceTests +{ + [Fact] + public async Task RequestWithSenderTraceContext_StartsANewTrace() + { + var cancellationToken = TestContext.Current.CancellationToken; + + var builder = WebApplication.CreateSlimBuilder(); + builder.WebHost.UseTestServer(); + builder.Configuration.AddInMemoryCollection(new Dictionary + { + // Nothing listens there: the exporter fails quietly, and the test reads the span in process. + ["OTEL_EXPORTER_OTLP_ENDPOINT"] = "http://127.0.0.1:9", + ["OTEL_SERVICE_NAME"] = "observability-tests", + }); + builder.Services.AddObservability(builder.Configuration); + + await using var app = builder.Build(); + Activity? requestSpan = null; + app.MapGet("/webhook", () => + { + requestSpan = Activity.Current; + return Results.Ok(); + }); + await app.StartAsync(cancellationToken); + + var senderTrace = ActivityTraceId.CreateRandom(); + using var request = new HttpRequestMessage(HttpMethod.Get, "/webhook"); + request.Headers.Add("traceparent", $"00-{senderTrace}-{ActivitySpanId.CreateRandom()}-01"); + using var response = await app.GetTestClient().SendAsync(request, cancellationToken); + response.EnsureSuccessStatusCode(); + + Assert.NotNull(requestSpan); + Assert.Equal(ActivityKind.Server, requestSpan.Kind); + Assert.True(requestSpan.Recorded); + Assert.Equal(default, requestSpan.ParentSpanId); + Assert.NotEqual(senderTrace, requestSpan.TraceId); + } +} diff --git a/tests/TalentSync.Infrastructure.Observability.Tests/SpanExportPolicyTests.cs b/tests/TalentSync.Infrastructure.Observability.Tests/SpanExportPolicyTests.cs new file mode 100644 index 0000000..fc0d9a1 --- /dev/null +++ b/tests/TalentSync.Infrastructure.Observability.Tests/SpanExportPolicyTests.cs @@ -0,0 +1,62 @@ +using System.Diagnostics; +using OpenTelemetry; +using OpenTelemetry.Trace; + +namespace TalentSync.Infrastructure.Observability.Tests; + +/// The policy in front of an exporter, registered in the same order as in AddObservability. +public sealed class SpanExportPolicyTests : IDisposable +{ + private readonly ActivitySource _source = new($"{nameof(SpanExportPolicyTests)}.{Guid.NewGuid()}"); + private readonly List _exported = []; + private readonly TracerProvider _provider; + + public SpanExportPolicyTests() + { + _provider = Sdk.CreateTracerProviderBuilder() + .AddSource(_source.Name) + .AddProcessor(new SpanExportPolicy()) + .AddInMemoryExporter(_exported) + .Build(); + } + + public void Dispose() + { + _provider.Dispose(); + _source.Dispose(); + } + + [Fact] + public void ClientSpanWithoutParent_IsNotExported() + { + using (_source.StartActivity("poll outbox", ActivityKind.Client)) + { + } + + Assert.Empty(_exported); + } + + [Fact] + public void ClientSpanInsideAFlow_IsExported() + { + using (_source.StartActivity("process")) + using (_source.StartActivity("insert inbox", ActivityKind.Client)) + { + } + + Assert.Contains(_exported, span => span.DisplayName == "insert inbox"); + } + + [Fact] + public void FailedSpan_KeepsItsStatusButNotTheMessage() + { + using (var activity = _source.StartActivity("insert outbox")) + { + activity?.SetStatus(ActivityStatusCode.Error, "Violation of UNIQUE KEY constraint. The duplicate key value is (alice@example.com)."); + } + + var span = Assert.Single(_exported); + Assert.Equal(ActivityStatusCode.Error, span.Status); + Assert.Null(span.StatusDescription); + } +} diff --git a/tests/TalentSync.Infrastructure.Observability.Tests/TalentSync.Infrastructure.Observability.Tests.csproj b/tests/TalentSync.Infrastructure.Observability.Tests/TalentSync.Infrastructure.Observability.Tests.csproj new file mode 100644 index 0000000..4ca9f73 --- /dev/null +++ b/tests/TalentSync.Infrastructure.Observability.Tests/TalentSync.Infrastructure.Observability.Tests.csproj @@ -0,0 +1,23 @@ + + + + Exe + false + + + + + + + + + + + + + + + + + + diff --git a/tests/TalentSync.IntegrationTests/Infrastructure/HostRouter.cs b/tests/TalentSync.IntegrationTests/Infrastructure/HostRouter.cs index 75702f6..e32a8f3 100644 --- a/tests/TalentSync.IntegrationTests/Infrastructure/HostRouter.cs +++ b/tests/TalentSync.IntegrationTests/Infrastructure/HostRouter.cs @@ -1,3 +1,4 @@ +using System.Diagnostics; using Microsoft.AspNetCore.TestHost; namespace TalentSync.IntegrationTests.Infrastructure; @@ -12,12 +13,26 @@ public sealed class HostRouter(IReadOnlyDictionary servers) private readonly Dictionary _invokers = servers.ToDictionary(server => server.Key, server => new HttpMessageInvoker(server.Value.CreateHandler())); - protected override Task SendAsync(HttpRequestMessage request, CancellationToken cancellationToken) + protected override async Task SendAsync(HttpRequestMessage request, CancellationToken cancellationToken) { var host = request.RequestUri?.Host ?? throw new InvalidOperationException("Request without a URI."); - return _invokers.TryGetValue(host, out var invoker) - ? invoker.SendAsync(request, cancellationToken) - : throw new InvalidOperationException($"No in-process server for host '{host}'."); + if (!_invokers.TryGetValue(host, out var invoker)) + { + throw new InvalidOperationException($"No in-process server for host '{host}'."); + } + + // Behave like the network for trace context, so scenario O1 can show that a host ignores its caller's context. + // SocketsHttpHandler, which this handler replaces, writes the context to the request headers; and the in-process + // server must not inherit the caller's span as an ambient parent, which no network passes on. The change to + // Activity.Current stays inside this async method. + if (Activity.Current is { } activity) + { + DistributedContextPropagator.Current.Inject(activity, request, static (carrier, name, value) => + ((HttpRequestMessage)carrier!).Headers.TryAddWithoutValidation(name, value)); + } + + Activity.Current = null; + return await invoker.SendAsync(request, cancellationToken); } protected override void Dispose(bool disposing) diff --git a/tests/TalentSync.IntegrationTests/Infrastructure/SpanRecorder.cs b/tests/TalentSync.IntegrationTests/Infrastructure/SpanRecorder.cs new file mode 100644 index 0000000..9e4746a --- /dev/null +++ b/tests/TalentSync.IntegrationTests/Infrastructure/SpanRecorder.cs @@ -0,0 +1,35 @@ +using System.Collections.Concurrent; +using System.Diagnostics; + +namespace TalentSync.IntegrationTests.Infrastructure; + +/// +/// Records our spans and the ASP.NET Core request spans (ADR 010). Every host runs in this process, so one listener +/// sees the spans of all of them, without an exporter. The listener is process-wide: while it lives, the other hosts' +/// events create spans too. +/// +public sealed class SpanRecorder : IDisposable +{ + private readonly ConcurrentQueue _stopped = new(); + private readonly ActivityListener _listener; + + public SpanRecorder() + { + _listener = new ActivityListener + { + ShouldListenTo = source => source.Name.StartsWith("TalentSync.", StringComparison.Ordinal) || source.Name == "Microsoft.AspNetCore", + Sample = (ref ActivityCreationOptions _) => ActivitySamplingResult.AllDataAndRecorded, + ActivityStopped = _stopped.Enqueue, + }; + ActivitySource.AddActivityListener(_listener); + } + + /// Ended spans tagged with this event id. + public IReadOnlyList ForEvent(string eventId) => + _stopped.Where(span => span.GetTagItem("talentsync.event_id") as string == eventId).ToList(); + + /// The ended span with this id, e.g. the parent of a recorded span; null when it has not ended. + public Activity? Find(ActivitySpanId spanId) => _stopped.FirstOrDefault(span => span.SpanId == spanId); + + public void Dispose() => _listener.Dispose(); +} diff --git a/tests/TalentSync.IntegrationTests/Observability/ObservabilityScenarios.cs b/tests/TalentSync.IntegrationTests/Observability/ObservabilityScenarios.cs new file mode 100644 index 0000000..114f007 --- /dev/null +++ b/tests/TalentSync.IntegrationTests/Observability/ObservabilityScenarios.cs @@ -0,0 +1,65 @@ +using System.Diagnostics; +using Microsoft.EntityFrameworkCore; +using TalentSync.Contracts; +using TalentSync.IntegrationTests.Infrastructure; + +namespace TalentSync.IntegrationTests.Observability; + +/// Scenario O1 of docs/acceptance-scenarios.md. +[Collection(SystemCollection.Name)] +public sealed class ObservabilityScenarios(SystemFixture fixture) +{ + private const string SyncWorker = "sync-worker"; + private const string MatchingWorker = "matching-worker"; + private const string IngressSource = "TalentSync.Ingress"; + + [Fact] + public async Task O1_OneWebhookProducesOneTraceThroughIngressTheRelayAndBothWorkers() + { + const string candidateId = "c-o1"; + using var recorder = new SpanRecorder(); + + await fixture.FakeTeamtailor.ApplyAsync(candidateId, "Olga", "Wrona", "olga.wrona@example.com"); + var eventId = await fixture.QueryIngressDbAsync(db => db.Outbox + .Where(m => m.CandidateId == candidateId && m.Type == EventTypes.CandidateCreate) + .Select(m => m.EventId) + .SingleAsync()); + + // Not part of the Then: spans are recorded when they end, so wait until both consumer groups are done. + var spans = await Eventually.GetAsync( + () => + { + var recorded = recorder.ForEvent(eventId); + return Task.FromResult(HasProcessSpan(recorded, SyncWorker) && HasProcessSpan(recorded, MatchingWorker) ? recorded : null); + }, + $"process spans of {eventId} in both consumer groups"); + + var webhook = Assert.Single(spans, span => span.Source.Name == IngressSource && span.Kind == ActivityKind.Internal); + var request = recorder.Find(webhook.ParentSpanId); + Assert.NotNull(request); + Assert.Equal(ActivityKind.Server, request.Kind); + Assert.Equal(webhook.TraceId, request.TraceId); + Assert.Equal(default, request.ParentSpanId); + + var send = Assert.Single(spans, span => span.Source.Name == IngressSource && span.Kind == ActivityKind.Producer); + Assert.Equal(webhook.TraceId, send.TraceId); + Assert.Equal(webhook.SpanId, send.ParentSpanId); + + foreach (var consumerGroup in new[] { SyncWorker, MatchingWorker }) + { + var processSpans = ProcessSpans(spans, consumerGroup); + Assert.NotEmpty(processSpans); + Assert.All(processSpans, process => + { + Assert.Equal(webhook.TraceId, process.TraceId); + Assert.Equal(send.SpanId, process.ParentSpanId); + }); + } + } + + private static bool HasProcessSpan(IReadOnlyList spans, string consumerGroup) => ProcessSpans(spans, consumerGroup).Count > 0; + + private static List ProcessSpans(IReadOnlyList spans, string consumerGroup) => + spans.Where(span => span.Kind == ActivityKind.Consumer + && span.GetTagItem("messaging.consumer.group.name") as string == consumerGroup).ToList(); +} diff --git a/tests/TalentSync.Matching.Infrastructure.Tests/LlmProviderRegistrationTests.cs b/tests/TalentSync.Matching.Infrastructure.Tests/LlmProviderRegistrationTests.cs index 9bc84b0..cf6fde8 100644 --- a/tests/TalentSync.Matching.Infrastructure.Tests/LlmProviderRegistrationTests.cs +++ b/tests/TalentSync.Matching.Infrastructure.Tests/LlmProviderRegistrationTests.cs @@ -15,11 +15,24 @@ public sealed class LlmProviderRegistrationTests : IDisposable public void Dispose() => _providers.ForEach(p => p.Dispose()); [Fact] - public void Fake_ResolvesFakeChatClient() + public void Fake_ResolvesTheFakeChatClientSingletonBehindTelemetry() { var provider = Build(new() { ["Llm:Provider"] = "Fake" }); - Assert.IsType(provider.GetRequiredService()); + var client = provider.GetRequiredService(); + + // The same instance the integration tests count calls on (scenario M5). + Assert.IsType(client); + Assert.Same(provider.GetRequiredService(), client.GetService()); + } + + [Fact] + public void Telemetry_NeverCapturesMessageContent() + { + var provider = Build(new() { ["Llm:Provider"] = "Fake" }); + + // Invariant 14: the CV and the model's answer stay out of spans, whatever the environment says. + Assert.False(provider.GetRequiredService().GetService()?.EnableSensitiveData); } [Fact]