Diagnostics and Instrumentation

So far, Marten has diagnostics, command logging, and unit of work life cycle tracking.

For information on accessing and previewing the database schema objects generated by Marten, see Marten and Postgres Schema

Disabling Npgsql Logging 7.0

The built in Npgsql logging is turned on by default in Marten, so to disable that logging so you can actually glean some value from your logs without blowing up the storage costs for your logging provider, use this flag:

var builder = Host.CreateDefaultBuilder();
builder.ConfigureServices(services =>
{
    services.AddMarten(opts =>
    {
        opts.Connection(ConnectionSource.ConnectionString);

        // Disable the absurdly verbose Npgsql logging
        opts.DisableNpgsqlLogging = true;

        opts.Events.AppendMode = EventAppendMode.Quick;
        opts.Events.UseIdentityMapForAggregates = true;

        opts.Projections.Add<DaemonTests.TestingSupport.TripProjection>(ProjectionLifecycle.Inline);
    });
});

^{snippet source | anchor}

The Marten team will be considering reversing the default for this behavior in Marten 8.

Listening for Document Store Events

INFO

All of the functionality in this section was added as part of Marten v0.8

Marten has a facility for listening and even intercepting document persistence events with the IDocumentSessionListener interface:

public interface IChangeListener
{
    /// <summary>
    /// Used to carry out actions on potentially changed projected documents generated and updated
    /// during the execution of asynchronous projections. This will give you "at most once" delivery guarantees
    /// </summary>
    /// <param name="session"></param>
    /// <param name="commit"></param>
    /// <param name="token"></param>
    /// <returns></returns>
    Task AfterCommitAsync(IDocumentSession session, IChangeSet commit, CancellationToken token);

    /// <summary>
    /// Used to carry out actions on potentially changed projected documents generated and updated
    /// during the execution of asynchronous projections. This will execute *before* database changes
    /// are committed. Use this for "at least once" delivery guarantees.
    /// </summary>
    /// <param name="session"></param>
    /// <param name="commit"></param>
    /// <param name="token"></param>
    /// <returns></returns>
    Task BeforeCommitAsync(IDocumentSession session, IChangeSet commit, CancellationToken token);
}

/// <summary>
///     Used to listen to and intercept operations within an IDocumentSession.SaveChanges()/SaveChangesAsync()
///     operation
/// </summary>
public interface IDocumentSessionListener
{
    /// <summary>
    ///     After an IDocumentSession is committed
    /// </summary>
    /// <param name="session"></param>
    /// <param name="commit"></param>
    /// <param name="token"></param>
    /// <returns></returns>
    Task AfterCommitAsync(IDocumentSession session, IChangeSet commit, CancellationToken token);

    /// <summary>
    ///     Called just after IDocumentSession.SaveChanges() is called, but before
    ///     any database calls are made
    /// </summary>
    /// <param name="session"></param>
    void BeforeSaveChanges(IDocumentSession session);

    /// <summary>
    ///     Called just after IDocumentSession.SaveChanges() is called,
    ///     but before any database calls are made
    /// </summary>
    /// <param name="session"></param>
    /// <param name="token"></param>
    /// <returns></returns>
    Task BeforeSaveChangesAsync(IDocumentSession session, CancellationToken token);

    /// <summary>
    ///     After an IDocumentSession is committed
    /// </summary>
    /// <param name="session"></param>
    /// <param name="commit"></param>
    void AfterCommit(IDocumentSession session, IChangeSet commit);

    /// <summary>
    ///     Called after a document is loaded
    /// </summary>
    void DocumentLoaded(object id, object document);

    /// <summary>
    ///     Called after a document is explicitly added to a session
    ///     as a staged insert or update
    /// </summary>
    void DocumentAddedForStorage(object id, object document);
}

^{snippet source | anchor}

You can build and inject your own listeners by adding them to the StoreOptions object you use to configure a DocumentStore:

var stub1 = new StubDocumentSessionListener();
var stub2 = new StubDocumentSessionListener();

using (var store = SeparateStore(_ =>
       {
           _.Connection(ConnectionSource.ConnectionString);
           _.AutoCreateSchemaObjects = AutoCreate.All;

           _.Listeners.Add(stub1);
           _.Listeners.Add(stub2);
       }))

^{snippet source | anchor}

var stub1 = new StubDocumentSessionListener();
var stub2 = new StubDocumentSessionListener();

using (var store = SeparateStore(_ =>
       {
           _.Connection(ConnectionSource.ConnectionString);
           _.AutoCreateSchemaObjects = AutoCreate.All;
       }))

^{snippet source | anchor}

The listeners can be used to modify an IDocumentSession and its related unit of work just before persisting. Marten itself will be using this mechanism internally to perform projections in the future.

The following fake, sample listener demonstrates how you can query into the pending changes before making a transactional commit, and also how to query what was done after a commit is made:

// DocumentSessionListenerBase is a helper abstract class in Marten
// with empty implementations of each method you may find helpful
public class SimpleSessionListener: DocumentSessionListenerBase
{
    public override void BeforeSaveChanges(IDocumentSession session)
    {
        // Use pending changes to preview what is about to be
        // persisted
        var pending = session.PendingChanges;

        // Careful here, Marten can only sort documents into "inserts" or "updates" based
        // on whether or not Marten had to assign a new Id to that document upon DocumentStore()
        pending.InsertsFor<User>()
            .Each(user => Debug.WriteLine($"New user: {user.UserName}"));

        pending.UpdatesFor<User>()
            .Each(user => Debug.WriteLine($"Updated user {user.UserName}"));

        pending.DeletionsFor<User>()
            .Each(d => Debug.WriteLine(d));

        // This is a convenience method to find all the pending events
        // organized into streams that will be appended to the event store
        pending.Streams()
            .Each(s => Debug.WriteLine(s));
    }

    public override void AfterCommit(IDocumentSession session, IChangeSet commit)
    {
        // See what was just persisted, and possibly carry out post
        // commit actions

        var last = commit;

        last.Updated.Each(x => Debug.WriteLine($"{x} was updated"));
        last.Deleted.Each(x => Debug.WriteLine($"{x} was deleted"));
        last.Inserted.Each(x => Debug.WriteLine($"{x} was inserted"));
    }
}

^{snippet source | anchor}

As of Marten 1.4, you can also register IDocumentSessionListener objects scoped to a particular session with the DocumentStore.OpenSession(SessionOptions) signature.

As of Marten v5, separate listeners will need to be registered for Document Store and Async Daemon. Adding listeners for Async Daemon are covered in the next section.

Listening for Async Daemon Events

Use AsyncListeners to register session listeners that will ONLY be applied within the asynchronous daemon updates.

INFO

Listeners will never get activated during projection rebuilds to safe guard against any side effects.

A sample listener:

public class FakeListener: IChangeListener
{
    public List<IChangeSet> Befores = new();
    public IList<IChangeSet> Changes = new List<IChangeSet>();

    public Task AfterCommitAsync(IDocumentSession session, IChangeSet commit, CancellationToken token)
    {
        session.ShouldNotBeNull();
        Changes.Add(commit);
        return Task.CompletedTask;
    }

    public Task BeforeCommitAsync(IDocumentSession session, IChangeSet commit, CancellationToken token)
    {
        session.ShouldNotBeNull();
        Befores.Add(commit);

        Changes.Count.ShouldBeLessThan(Befores.Count);

        return Task.CompletedTask;
    }
}

^{snippet source | anchor}

Wiring a Async Daemon listener:

var listener = new FakeListener();
StoreOptions(x =>
{
    x.Projections.Add(new TripProjectionWithCustomName(), ProjectionLifecycle.Async);
    x.Projections.AsyncListeners.Add(listener);
});

^{snippet source | anchor}

Custom Logging

Marten v0.8 comes with a new mechanism to plug in custom logging to the IDocumentStore, IQuerySession, and IDocumentSession activity:

/// <summary>
///     Records command usage, schema changes, and sessions within Marten
/// </summary>
public interface IMartenLogger
{
    /// <summary>
    ///     Called when the session is initialized
    /// </summary>
    /// <param name="session"></param>
    /// <returns></returns>
    IMartenSessionLogger StartSession(IQuerySession session);

    /// <summary>
    ///     Capture any DDL executed at runtime by Marten
    /// </summary>
    /// <param name="sql"></param>
    void SchemaChange(string sql);
}

/// <summary>
///     Use to create custom logging within an IQuerySession or IDocumentSession
/// </summary>
public interface IMartenSessionLogger
{
    /// <summary>
    ///     Log a command that executed successfully
    /// </summary>
    /// <param name="command"></param>
    void LogSuccess(NpgsqlCommand command);

    /// <summary>
    ///     Log a command that failed
    /// </summary>
    /// <param name="command"></param>
    /// <param name="ex"></param>
    void LogFailure(NpgsqlCommand command, Exception ex);

    /// <summary>
    ///     Log a command that executed successfully
    /// </summary>
    /// <param name="batch"></param>
    void LogSuccess(NpgsqlBatch batch);

    /// <summary>
    ///     Log a batch that failed
    /// </summary>
    /// <param name="batch"></param>
    /// <param name="ex"></param>
    void LogFailure(NpgsqlBatch batch, Exception ex);

    /// <summary>
    /// Log a message for generic errors
    /// </summary>
    /// <param name="ex"></param>
    /// <param name="message"></param>
    /// <param name="batch"></param>
    void LogFailure(Exception ex, string message);

    /// <summary>
    ///     Called immediately after committing an IDocumentSession
    ///     through SaveChanges() or SaveChangesAsync()
    /// </summary>
    /// <param name="session"></param>
    /// <param name="commit"></param>
    void RecordSavedChanges(IDocumentSession session, IChangeSet commit);

    /// <summary>
    ///     Called just before a command is to be executed. Use this to create
    ///     performance logging of Marten operations
    /// </summary>
    /// <param name="command"></param>
    public void OnBeforeExecute(NpgsqlCommand command);

    /// <summary>
    ///     Called just before a command is to be executed. Use this to create
    ///     performance logging of Marten operations
    /// </summary>
    /// <param name="command"></param>
    public void OnBeforeExecute(NpgsqlBatch batch);
}

^{snippet source | anchor}

To apply these logging abstractions, you can either plug your own IMartenLogger into the StoreOptions object and allow that default logger to create the individual session loggers:

var store = DocumentStore.For(_ =>
{
    _.Logger(new ConsoleMartenLogger());
});

^{snippet source | anchor}

You can also directly apply a session logger to any IQuerySession or IDocumentSession like this:

using var session = store.LightweightSession();
// Replace the logger for only this one session
session.Logger = new RecordingLogger();

^{snippet source | anchor}

The session logging is a different abstraction specifically so that you could track database commands issued per session. In effect, my own shop is going to use this capability to understand what HTTP endpoints or service bus message handlers are being unnecessarily chatty in their database interactions. We also hope that the contextual logging of commands per document session makes it easier to understand how our systems behave.

public class ConsoleMartenLogger: IMartenLogger, IMartenSessionLogger
{
    private Stopwatch? _stopwatch;

    public IMartenSessionLogger StartSession(IQuerySession session)
    {
        return this;
    }

    public void SchemaChange(string sql)
    {
        Console.WriteLine("Executing DDL change:");
        Console.WriteLine(sql);
        Console.WriteLine();
    }

    public void LogSuccess(NpgsqlCommand command)
    {
        Console.WriteLine(command.CommandText);
        foreach (var p in command.Parameters.OfType<NpgsqlParameter>())
            Console.WriteLine($"  {p.ParameterName}: {GetParameterValue(p)}");
    }

    public void LogSuccess(NpgsqlBatch batch)
    {
        foreach (var command in batch.BatchCommands)
        {
            Console.WriteLine(command.CommandText);
            foreach (var p in command.Parameters.OfType<NpgsqlParameter>())
                Console.WriteLine($"  {p.ParameterName}: {GetParameterValue(p)}");
        }
    }

    private static object? GetParameterValue(NpgsqlParameter p)
    {
        if (p.Value is IList enumerable)
        {
            var result = "";
            for (var i = 0; i < Math.Min(enumerable.Count, 5); i++)
            {
                result += $"[{i}] {enumerable[i]}; ";
            }
            if (enumerable.Count > 5) result += $" + {enumerable.Count - 5} more";
            return result;
        }
        return p.Value;
    }

    public void LogFailure(NpgsqlCommand command, Exception ex)
    {
        Console.WriteLine("Postgresql command failed!");
        Console.WriteLine(command.CommandText);
        foreach (var p in command.Parameters.OfType<NpgsqlParameter>())
            Console.WriteLine($"  {p.ParameterName}: {p.Value}");
        Console.WriteLine(ex);
    }

    public void LogFailure(NpgsqlBatch batch, Exception ex)
    {
        Console.WriteLine("Postgresql command failed!");
        foreach (var command in batch.BatchCommands)
        {
            Console.WriteLine(command.CommandText);
            foreach (var p in command.Parameters.OfType<NpgsqlParameter>())
                Console.WriteLine($"  {p.ParameterName}: {p.Value}");
        }

        Console.WriteLine(ex);
    }

    public void LogFailure(Exception ex, string message)
    {
        Console.WriteLine("Failure: " + message);
        Console.WriteLine(ex.ToString());
    }

    public void RecordSavedChanges(IDocumentSession session, IChangeSet commit)
    {
        _stopwatch?.Stop();

        var lastCommit = commit;
        Console.WriteLine(
            $"Persisted {lastCommit.Updated.Count()} updates in {_stopwatch?.ElapsedMilliseconds ?? 0} ms, {lastCommit.Inserted.Count()} inserts, and {lastCommit.Deleted.Count()} deletions");
    }

    public void OnBeforeExecute(NpgsqlCommand command)
    {
        _stopwatch = new Stopwatch();
        _stopwatch.Start();
    }

    public void OnBeforeExecute(NpgsqlBatch batch)
    {
        _stopwatch = new Stopwatch();
        _stopwatch.Start();
    }
}

^{snippet source | anchor}

Accessing Diagnostics

All the diagnostics are going to be exposed off of the IDocumentStore.Diagnostics property. Today, the only capabilities are to get a preview of the generated storage code or a preview of the ADO.NET code that will be generated for a LINQ query.

Previewing LINQ Queries

Let's say that we have a small document type called Trade:

public class Trade
{
    public int Id { get; set; }

    [DuplicateField]
    public double Value { get; set; }
}

^{snippet source | anchor}

The [DuplicateField] attribute directs Marten to duplicate the value of Value into a separate database field for more efficient querying. Now, let's say that we want to search for every Trade document with a value of over 2,000, but we want to see the SQL query that Marten will build for that query first:

// store is the active IDocumentStore
var queryable = theStore.QuerySession().Query<Trade>().Where(x => x.Value > 2000);
var cmd = queryable.ToCommand(FetchType.FetchMany);

Debug.WriteLine(cmd.CommandText);

^{snippet source | anchor}

The sql string in our debug window for the code above is:

select d.data from mt_doc_trade as d where d.value > :arg0

Session Specific Logging

The IMartenLogger can be swapped out on any IQuerySession or IDocumentSession like this example from the unit tests:

// We frequently use this special marten logger per
// session to pipe Marten logging to the xUnit.Net output
theSession.Logger = new TestOutputMartenLogger(_output);

^{snippet source | anchor}

Previewing the PostgreSQL Query Plan

Marten has a helper to find and preview the PostgreSQL EXPLAIN plan for a Linq query. Our hope is that this will be a valuable aid to teams who need face performance problems while using Marten. The syntax for fetching the EXPLAIN plan for the Linq query from the previous section is shown below:

// Explain() is an extension method off of IQueryable<T>
var plan = queryable.Explain();
Console.WriteLine($"NodeType: {plan.NodeType}");
Console.WriteLine($"RelationName: {plan.RelationName}");
Console.WriteLine($"Alias: {plan.Alias}");
Console.WriteLine($"StartupCost: {plan.StartupCost}");
Console.WriteLine($"TotalCost: {plan.TotalCost}");
Console.WriteLine($"PlanRows: {plan.PlanRows}");
Console.WriteLine($"PlanWidth: {plan.PlanWidth}");

^{snippet source | anchor}

The console output for the code below (on my box) was:

NodeType: Seq Scan
RelationName: mt_doc_trade
Alias: d
StartupCost: 0
TotalCost: 24.13
PlanRows: 377
PlanWidth: 36

Request Counting and Thresholds

Marten has several facilities for improving system performance by reducing the number of network round trips to the server, but the first step maybe to just understand what kinds of operations are being chatty in the first place. To that end, Marten exposes the request count for each IQuerySession or IDocumentSession that simply tells you how many commands have been issued to Postgresql by that session:

using (var session = theStore.QuerySession())
{
    var users = session.Query<User>().ToList();
    var count = session.Query<User>().Count();
    var any = session.Query<User>().Any();

    session.RequestCount.ShouldBe(3);
}

^{snippet source | anchor}

At this point, Marten does not have any built in support for asserting requests per session thresholds like other tools. While I think that we are uncomfortable with that functionality ever being turned on in production, it should be easily feasible to build those kinds of automated threshold testing like "fail the test if there were more than 25 requests issued for any given HTTP request."

Getting PostgreSQL server version

Marten provides a helper method to fetch the PostgreSQL server version exposed via IDocumentStore.Diagnostics. This is helpful to enable feature toggles based on features available in PostgreSQL versions or perform any diagnostics.

var pgVersion = theStore.Diagnostics.GetPostgresVersion();

^{snippet source | anchor}