# Observability (/integrations/masstransit/how-to/observability) The MassTransit.KubeMQ transport surfaces all three observability pillars — health, traces, and metrics — through the standard .NET diagnostics APIs. Health flows through MassTransit's built-in health check, which maps the KubeMQ connection state to an ASP.NET Core health status. Traces use MassTransit's `ActivitySource` and the W3C trace context that the transport propagates over KubeMQ Tags. Metrics are emitted on a dedicated `MassTransit.KubeMQ` meter that you can subscribe to with OpenTelemetry or a raw `MeterListener`. This page wires up each pillar with runnable code, then closes with guidance on which signal to watch for which symptom. ## Prerequisites [#prerequisites] * A MassTransit bus already configured with `UsingKubeMQ` (see [Configuration](/integrations/masstransit/how-to/configuration)) * ASP.NET Core health checks (`AddHealthChecks`) registered, if you want to expose `/health` ## Health Checks [#health-checks] MassTransit automatically registers a health check for the bus and every receive endpoint when you call `AddMassTransit`. You do not register a transport-specific check — expose the built-in one through the ASP.NET Core health endpoint: ```csharp title="Program.cs" using MassTransit; var builder = WebApplication.CreateBuilder(args); builder.Services.AddMassTransit(x => { x.UsingKubeMQ((ctx, cfg) => { cfg.Host("localhost", 50000); cfg.ReceiveEndpoint("orders", e => { e.ConfigureKubeMQ(t => { }); }); }); }); var app = builder.Build(); app.MapHealthChecks("/health"); app.Run(); ``` A running KubeMQ broker is required for the bus to report `Healthy`. Start one in Docker with the gRPC port exposed: Port `50000` is the gRPC port the transport connects to. Port `9090` is the shared HTTP server that hosts the REST, CloudEvents, MCP, and A2A connectors. ### Health-State Mapping [#health-state-mapping] The transport monitors the KubeMQ client's `ConnectionState` and maps each state onto a MassTransit health status. The built-in check reflects this mapping directly: | KubeMQ State | MassTransit Health | Meaning | | -------------- | ------------------ | ---------------------------------------- | | `Ready` | **Healthy** | Connection is active and operational | | `Connecting` | **Degraded** | Initial connection in progress | | `Reconnecting` | **Degraded** | Lost connection, attempting to reconnect | | `Closed` | **Unhealthy** | Connection permanently closed | | `Idle` | **Unhealthy** | Not connected | A `Degraded` status during startup or a transient network blip is expected — the transport reports `Reconnecting` while the KubeMQ SDK retries internally, and returns to `Healthy` once the connection is re-established. Treat `Unhealthy` (`Closed`/`Idle`) as the only state that should fail a liveness probe. ### Checking Health Programmatically [#checking-health-programmatically] For dashboards or custom probes you can read the connection state directly. The internal `KubeMQConnectionContextSupervisor` exposes three boolean properties that mirror the table above: ```csharp title="ConnectionState.cs" // These properties reflect the current connection state supervisor.IsReady // true when ConnectionState == Ready supervisor.IsDegraded // true when Connecting or Reconnecting supervisor.IsUnhealthy // true when Closed or Idle ``` In application code you normally go one level higher and query the bus itself. `IBusControl.CheckHealth()` returns a `BusHealthResult` carrying the overall `Status` plus per-endpoint entries — this is the recommended way to build a custom `IHealthCheck` that wraps bus health: ```csharp title="BusHealthCheck.cs" public class BusHealthCheck(IBusControl busControl, ILogger logger) : IHealthCheck { public Task CheckHealthAsync( HealthCheckContext context, CancellationToken cancellationToken = default) { var healthResult = busControl.CheckHealth(); logger.LogInformation("Bus health status: {Status}", healthResult.Status); foreach (var (key, endpointResult) in healthResult.Endpoints) { logger.LogInformation(" Endpoint '{Key}': Status={Status}, Description={Description}", key, endpointResult.Status, endpointResult.Description); } return Task.FromResult(healthResult.Status switch { BusHealthStatus.Healthy => HealthCheckResult.Healthy(healthResult.Description), BusHealthStatus.Degraded => HealthCheckResult.Degraded(healthResult.Description), _ => HealthCheckResult.Unhealthy(healthResult.Description) }); } } ``` Register it alongside (or instead of) the built-in check. Use a distinct name so it does not collide with MassTransit's own `masstransit-bus` entry: ```csharp title="Program.cs" builder.Services.AddHealthChecks() .AddCheck("masstransit-bus-custom"); ``` `busControl.CheckHealth()` works against any MassTransit transport — it queries the bus and endpoint pipeline, so the same code is portable if you later add a second transport. ## Distributed Tracing [#distributed-tracing] The transport propagates W3C trace context across services by writing the active trace onto KubeMQ Tags on every outbound message: | MassTransit / W3C | KubeMQ Tag | | ----------------- | ---------------- | | `traceparent` | `MT-TraceParent` | | `tracestate` | `MT-TraceState` | On the receive side those tags are read back and used to restore the trace context, so a span started in the producer continues into the consumer. The tags integrate with MassTransit's own `ActivitySource` (named `"MassTransit"`), giving you end-to-end traces across producers and consumers without any KubeMQ-specific instrumentation. The simplest way to observe activities in a console app is an `ActivityListener`. In production you would replace this with the OpenTelemetry SDK and `AddSource("MassTransit")`, but the listener makes the propagation visible without extra packages: ```csharp title="DistributedTracing.cs" using System.Diagnostics; // Capture every activity (MassTransit's own spans plus your custom ones) var capturedActivities = new List(); using var activityListener = new ActivityListener { ShouldListenTo = source => true, Sample = (ref ActivityCreationOptions _) => ActivitySamplingResult.AllDataAndRecorded, ActivityStarted = activity => capturedActivities.Add(activity), }; ActivitySource.AddActivityListener(activityListener); ``` Inside a consumer, `Activity.Current` reflects the restored trace context — the `TraceId` matches the producer's, confirming propagation worked across the KubeMQ hop: ```csharp title="FullObsConsumer.cs" public class FullObsConsumer(ILogger logger) : IConsumer { public Task Consume(ConsumeContext context) { var activity = Activity.Current; logger.LogInformation( "Received order: OrderId={OrderId}, Product={Product}, Amount={Amount:C}", context.Message.OrderId, context.Message.ProductName, context.Message.Amount); if (activity != null) { logger.LogInformation(" [Tracing] TraceId={TraceId}, SpanId={SpanId}", activity.TraceId, activity.SpanId); } return Task.CompletedTask; } } ``` ## Metrics [#metrics] The transport publishes its own instruments on a meter named `MassTransit.KubeMQ`. Subscribe to that meter with the OpenTelemetry metrics SDK (`AddMeter("MassTransit.KubeMQ")`) or, for a self-contained example, a `MeterListener`. The meter exposes six instruments: | Metric | Type | Unit | Description | | ----------------------------------- | --------- | -------- | --------------------------------- | | `kubemq.transport.poll.duration` | Histogram | ms | Duration of queue poll operations | | `kubemq.transport.poll.messages` | Histogram | messages | Messages received per poll batch | | `kubemq.transport.poll.empty` | Counter | polls | Count of empty poll responses | | `kubemq.transport.send.duration` | Histogram | ms | Duration of queue send operations | | `kubemq.transport.publish.duration` | Histogram | ms | Duration of publish operations | | `kubemq.transport.errors` | Counter | errors | Transport-level error count | The histograms record `double` measurements (durations) except `poll.messages`, which records `int`, and the two counters record `long`. A `MeterListener` therefore needs a callback per value type. The example below subscribes to every instrument on the `MassTransit.KubeMQ` meter and routes measurements by name: ```csharp title="TransportMetrics.cs" using System.Diagnostics.Metrics; var sendDurations = new List(); var pollDurations = new List(); var pollMessageCounts = new List(); var publishDurations = new List(); long emptyPolls = 0; long errors = 0; using var listener = new MeterListener(); listener.InstrumentPublished = (instrument, meterListener) => { // Subscribe to all instruments from the MassTransit.KubeMQ meter if (instrument.Meter.Name == "MassTransit.KubeMQ") meterListener.EnableMeasurementEvents(instrument); }; listener.SetMeasurementEventCallback((instrument, measurement, tags, state) => { switch (instrument.Name) { case "kubemq.transport.send.duration": sendDurations.Add(measurement); break; case "kubemq.transport.poll.duration": pollDurations.Add(measurement); break; case "kubemq.transport.publish.duration": publishDurations.Add(measurement); break; } }); listener.SetMeasurementEventCallback((instrument, measurement, tags, state) => { if (instrument.Name == "kubemq.transport.poll.messages") pollMessageCounts.Add(measurement); }); listener.SetMeasurementEventCallback((instrument, measurement, tags, state) => { switch (instrument.Name) { case "kubemq.transport.poll.empty": Interlocked.Add(ref emptyPolls, measurement); break; case "kubemq.transport.errors": Interlocked.Add(ref errors, measurement); break; } }); listener.Start(); ``` The instruments are recorded internally by the transport on every poll, send, and publish — you only subscribe to them. The meter version is `1.0.0`. A poll that returns zero messages records `poll.duration`, a `poll.messages` value of `0`, and increments `poll.empty`. ## Full Setup [#full-setup] For a single host that combines all three pillars, register the bus, add a health check that wraps `IBusControl.CheckHealth()`, and wire up an `ActivityListener` plus a `MeterListener` from a hosted service. ### Register the bus and a bus-health check [#register-the-bus-and-a-bus-health-check] ```csharp title="Program.cs" using MassTransit; var builder = Host.CreateApplicationBuilder(args); builder.Services.AddMassTransit(x => { x.AddConsumer(); x.UsingKubeMQ((context, cfg) => { cfg.Host("localhost", 50000); cfg.ReceiveEndpoint("obs-full-queue", e => { e.ConfigureKubeMQ(t => { }); }); }); }); // Register health check for bus health builder.Services.AddHealthChecks() .AddCheck("masstransit-bus"); builder.Services.AddHostedService(); var host = builder.Build(); await host.RunAsync(); ``` ### Wrap bus health in an IHealthCheck [#wrap-bus-health-in-an-ihealthcheck] ```csharp title="FullSetupBusHealthCheck.cs" public class FullSetupBusHealthCheck(IBusControl busControl) : IHealthCheck { public Task CheckHealthAsync( HealthCheckContext context, CancellationToken cancellationToken = default) { var healthResult = busControl.CheckHealth(); return Task.FromResult(healthResult.Status switch { BusHealthStatus.Healthy => HealthCheckResult.Healthy(healthResult.Description), BusHealthStatus.Degraded => HealthCheckResult.Degraded(healthResult.Description), _ => HealthCheckResult.Unhealthy(healthResult.Description) }); } } ``` ### Wire tracing, metrics, and health in one place [#wire-tracing-metrics-and-health-in-one-place] The hosted service registers the `ActivityListener`, the `MeterListener`, and resolves the `HealthCheckService` before sending messages, then reports each pillar. ```csharp title="FullObsSender.cs" protected override async Task ExecuteAsync(CancellationToken stoppingToken) { await Task.Delay(2000, stoppingToken); // --- Pillar 1: Distributed Tracing --- var capturedActivities = new List(); using var activityListener = new ActivityListener { ShouldListenTo = source => true, Sample = (ref ActivityCreationOptions _) => ActivitySamplingResult.AllDataAndRecorded, ActivityStarted = activity => capturedActivities.Add(activity), }; ActivitySource.AddActivityListener(activityListener); // --- Pillar 2: Transport Metrics --- var sendDurations = new List(); long errorCount = 0; using var meterListener = new MeterListener(); meterListener.InstrumentPublished = (instrument, ml) => { if (instrument.Meter.Name == "MassTransit.KubeMQ") ml.EnableMeasurementEvents(instrument); }; meterListener.SetMeasurementEventCallback((instrument, measurement, tags, state) => { if (instrument.Name == "kubemq.transport.send.duration") sendDurations.Add(measurement); }); meterListener.SetMeasurementEventCallback((instrument, measurement, tags, state) => { if (instrument.Name == "kubemq.transport.errors") Interlocked.Add(ref errorCount, measurement); }); meterListener.Start(); // --- Pillar 3: Health Checks --- var healthCheckService = serviceProvider.GetRequiredService(); var report = await healthCheckService.CheckHealthAsync(stoppingToken); logger.LogInformation("[Health] Initial bus health: {Status}", report.Status); // ... send messages inside activitySource.StartActivity(...) spans ... // Flush any pending observable instruments before reading metrics meterListener.RecordObservableInstruments(); logger.LogInformation("[Metrics] Send durations: {Count} measurements", sendDurations.Count); logger.LogInformation("[Metrics] Errors: {Count}", Interlocked.Read(ref errorCount)); report = await healthCheckService.CheckHealthAsync(stoppingToken); logger.LogInformation("[Health] Final bus health: {Status}", report.Status); } ``` Call `meterListener.RecordObservableInstruments()` before reading aggregated values — it flushes any pending observable measurements so your snapshot reflects the latest poll and send activity. ## Multi-Bus Observability [#multi-bus-observability] When you run multiple bus instances (multi-bus), each bus gets its own `IKubeMQClient` connection, and its health check is independent. The `MassTransit.KubeMQ` meter is shared, so metrics from all buses aggregate onto the same instruments. To tell connections and health entries apart, give each bus a distinct `ClientId` — it identifies the connection in the KubeMQ server logs: ```csharp title="Program.cs" cfg.Host("localhost", 50000, h => { h.ClientId = "orders-bus"; }); ``` Each bus reports its own health entry, so a `MapHealthChecks("/health")` endpoint aggregates the status of every connection: if one bus is `Reconnecting` while another is `Ready`, the overall report is `Degraded`. ## Recommended Log Levels [#recommended-log-levels] For routine diagnostics, raise the transport and SDK categories to `Debug` while keeping the rest of the application at `Information`: ```json title="appsettings.json" { "Logging": { "LogLevel": { "Default": "Information", "MassTransit": "Debug", "MassTransit.KubeMQTransport": "Debug", "KubeMQ": "Debug" } } } ``` At `Debug`, the SDK logs `ConnectionState` transitions and the receive transport logs each poll cycle, which is enough to correlate health changes and empty-poll counts with broker behavior. For exception-level diagnostics — faults, retries, and `_error` channel routing — see the [Error Handling & DLQ](/integrations/masstransit/how-to/error-handling-dlq) guide. ## Which Signal for Which Symptom [#which-signal-for-which-symptom] Once the meter is wired up, use this table to map a symptom to the instrument or signal that explains it: | Symptom | What to watch | | ------------------------------------------ | --------------------------------------------------------------------------------------------------------------------------- | | Idle queues / no work flowing | `kubemq.transport.poll.empty` climbing while `poll.messages` stays at `0` — the receiver is polling but nothing is enqueued | | Transport failures (send/connection drops) | `kubemq.transport.errors` incrementing; cross-check the health status for `Degraded`/`Unhealthy` | | High send latency | `kubemq.transport.send.duration` trending up | | High publish latency | `kubemq.transport.publish.duration` trending up | | High poll / receive latency | `kubemq.transport.poll.duration` trending up — also check network latency to the broker | For high poll latency specifically, the transport tuning levers are `MaxPollMessages` (raise it to receive more messages per batch) and `PollTimeoutSeconds` (lower it for faster empty-queue turnaround). Track `kubemq.transport.poll.duration` to confirm the change moved the trend in the right direction. ## Related [#related]