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

• A MassTransit bus already configured with UsingKubeMQ (see Configuration)
• ASP.NET Core health checks (AddHealthChecks) registered, if you want to expose /health

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:

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:

docker run -d \  --name kubemq \  -p 50000:50000 \  -p 9090:9090 \  -e KUBEMQ_TOKEN=YOUR_LICENSE_KEY \  europe-docker.pkg.dev/kubemq/images/kubemq:next

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

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:

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:

// 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:

public class BusHealthCheck(IBusControl busControl, ILogger<BusHealthCheck> logger) : IHealthCheck
{
    public Task<HealthCheckResult> 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:

builder.Services.AddHealthChecks()
    .AddCheck<BusHealthCheck>("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

The transport propagates W3C trace context across services by writing the active trace onto KubeMQ Tags on every outbound message:

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:

using System.Diagnostics;

// Capture every activity (MassTransit's own spans plus your custom ones)
var capturedActivities = new List<Activity>();
using var activityListener = new ActivityListener
{
    ShouldListenTo = source => true,
    Sample = (ref ActivityCreationOptions<ActivityContext> _) => 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:

public class FullObsConsumer(ILogger<FullObsConsumer> logger) : IConsumer<FullObsOrder>
{
    public Task Consume(ConsumeContext<FullObsOrder> 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

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:

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:

using System.Diagnostics.Metrics;

var sendDurations = new List<double>();
var pollDurations = new List<double>();
var pollMessageCounts = new List<int>();
var publishDurations = new List<double>();
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<double>((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<int>((instrument, measurement, tags, state) =>
{
    if (instrument.Name == "kubemq.transport.poll.messages")
        pollMessageCounts.Add(measurement);
});

listener.SetMeasurementEventCallback<long>((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();

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.

using MassTransit;

var builder = Host.CreateApplicationBuilder(args);

builder.Services.AddMassTransit(x =>
{
    x.AddConsumer<FullObsConsumer>();
    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<FullSetupBusHealthCheck>("masstransit-bus");

builder.Services.AddHostedService<FullObsSender>();

var host = builder.Build();
await host.RunAsync();

public class FullSetupBusHealthCheck(IBusControl busControl) : IHealthCheck
{
    public Task<HealthCheckResult> 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)
        });
    }
}

protected override async Task ExecuteAsync(CancellationToken stoppingToken)
{
    await Task.Delay(2000, stoppingToken);

    // --- Pillar 1: Distributed Tracing ---
    var capturedActivities = new List<Activity>();
    using var activityListener = new ActivityListener
    {
        ShouldListenTo = source => true,
        Sample = (ref ActivityCreationOptions<ActivityContext> _) => ActivitySamplingResult.AllDataAndRecorded,
        ActivityStarted = activity => capturedActivities.Add(activity),
    };
    ActivitySource.AddActivityListener(activityListener);

    // --- Pillar 2: Transport Metrics ---
    var sendDurations = new List<double>();
    long errorCount = 0;

    using var meterListener = new MeterListener();
    meterListener.InstrumentPublished = (instrument, ml) =>
    {
        if (instrument.Meter.Name == "MassTransit.KubeMQ")
            ml.EnableMeasurementEvents(instrument);
    };
    meterListener.SetMeasurementEventCallback<double>((instrument, measurement, tags, state) =>
    {
        if (instrument.Name == "kubemq.transport.send.duration")
            sendDurations.Add(measurement);
    });
    meterListener.SetMeasurementEventCallback<long>((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<HealthCheckService>();
    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);
}

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:

cfg.Host("localhost", 50000, h =>
{
    h.ClientId = "orders-bus";
});

Recommended Log Levels

For routine diagnostics, raise the transport and SDK categories to Debug while keeping the rest of the application at Information:

{
  "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 guide.

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:

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