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