2
\$\begingroup\$

In a docker compose environment there are several apps that communicate with each other. One of these apps is a logger (a centralized logger). Every part of the system that needs to log something, will call the logger's APIs (which is written with gRPC).

The logger adds these log requests to a queue (a BlockingCollection) which will be consumed by workers. These workers will read from queue and store them in DB. In order to reduce DB calls, they will use a local buffer. That buffer will be flushed every (n) seconds or after getting (m) records.

These workers need to dynamically be created or disposed based on how many items are in the queue. For example for each (n) items there should be a worker.

This is the big picture:

enter image description here

And now the implementation,

The log entity:

public class ActivityLog
{
    public string Id { get; set; }
    public IPAddress IpAddress { get; set; }
    public string? Parameters { get; set; }
    // ...
}

gRPC proto contract:

syntax = "proto3";

import "google/protobuf/empty.proto";
import "google/protobuf/wrappers.proto";

package logger;

service Logger {
  rpc LogActivity (ActivityLogPayload) returns (google.protobuf.Empty);
}

message ActivityLogPayload {
  string ip_address = 1;
  google.protobuf.StringValue parameters = 2;
  // ...
}

gRPC service:

public class LoggerRpc : Logger.LoggerBase
{
    private readonly LogQueue queue;

    public LoggerRpc(LogQueue queue)
    {
        this.queue = queue;
    }

    public override Task<Empty> LogActivity(ActivityLogPayload request, ServerCallContext context)
    {
        var logEntity = request.ToEntity();
        queue.ActivityLogsCollection.Add(logEntity);
        return Task.FromResult(new Empty());
    }
}

The log queue:

public class LogQueue
{
    public BlockingCollection<ActivityLog> ActivityLogsCollection { get; }

    public LogQueue()
    {
        ActivityLogsCollection = new(10_000);
    }
}

The worker:

public class LogWorker : IDisposable
{
    private const int BufferThreshold = 100;
    private readonly TimeSpan BufferTimeLimit = TimeSpan.FromSeconds(5);
    private DateTimeOffset lastDbUpdate = DateTimeOffset.UtcNow;

    private readonly BlockingCollection<ActivityLog> queue;
    private readonly IServiceScopeFactory ssf;

    private CancellationTokenSource cts = new();
    private Task workerTask;

    public LogWorker(
        BlockingCollection<ActivityLog> queue,
        IServiceScopeFactory ssf)
    {
        this.queue = queue;
        this.ssf = ssf;

        workerTask = new Task(Work, TaskCreationOptions.LongRunning);
        workerTask.Start(TaskScheduler.Default);
    }

    private void Work()
    {
        using var scope = ssf.CreateScope();
        using var dbContext = scope.ServiceProvider.GetRequiredService<LogDbContext>();

        var localBuffer = new List<ActivityLog>();
        ActivityLog? nextItem;
        try
        {
            while (queue.TryTake(out nextItem, -1, cts.Token))
            {
                localBuffer.Add(nextItem);

                // make a DB call when we have 100 logs
                // or 5 seconds has passes from last DB call
                while (localBuffer.Count < BufferThreshold
                    && DateTime.UtcNow - lastDbUpdate < BufferTimeLimit)
                {
                    // wait for the remaining time of our 5 second window
                    var waitTime = (int)(BufferTimeLimit - (DateTime.UtcNow - lastDbUpdate)).TotalMilliseconds;
                    if (queue.TryTake(out nextItem, waitTime, cts.Token))
                        localBuffer.Add(nextItem);
                }

                dbContext.AddRange(localBuffer);
                dbContext.SaveChanges();

                localBuffer.Clear();
                lastDbUpdate = DateTime.UtcNow;
            }
        }
        catch (ObjectDisposedException)
        {
            // queue completed
        }
        catch(OperationCanceledException)
        {
            // worker is shutting down
        }
        finally
        {
            // clear the buffer
            if (localBuffer.Any())
            {
                dbContext.AddRange(localBuffer);
                dbContext.SaveChanges();
            }
        }
    }

    public void Dispose()
    {
        cts.Cancel();
        workerTask.Wait();
    }
}

Worker manager:

public class LogWorkerManager : IHostedService
{
    private const int MinimumWorkersCount = 2;
    private const int LogsThreshold = 200;
    private readonly TimeSpan TimerInterval = TimeSpan.FromSeconds(2);

    private readonly List<LogWorker> workers = new();

    private readonly LogQueue queue;
    private readonly IServiceScopeFactory ssf;

    Timer? timer;

    public LogWorkerManager(
        LogQueue queue,
        IServiceScopeFactory ssf)
    {
        this.queue = queue;
        this.ssf = ssf;
    }

    private void MonitorQueue(object? state)
    {
        var workersNeeded = (queue.ActivityLogsCollection.Count / LogsThreshold) + 1;
        workersNeeded = Math.Max(workersNeeded, MinimumWorkersCount);
        if (workers.Count > workersNeeded)
        {
            // gradually reduce workers
            var worker = workers.Last();
            worker.Dispose();
            workers.Remove(worker);
        }
        else if (workers.Count < workersNeeded)
        {
            for (int i = 0; i < workersNeeded - workers.Count; i++)
            {
                var worker = new LogWorker(queue.ActivityLogsCollection, ssf);
                workers.Add(worker);
            }
        }
    }

    public Task StartAsync(CancellationToken cancellationToken)
    {
        timer = new(MonitorQueue, null, 0, (int)TimerInterval.TotalMilliseconds);
        return Task.CompletedTask;
    }

    public Task StopAsync(CancellationToken cancellationToken)
    {
        timer?.Change(Timeout.Infinite, Timeout.Infinite);
        timer?.Dispose();
        Parallel.ForEach(workers, worker => worker.Dispose());
        return Task.CompletedTask;
    }
}

DI:

builder.Services.AddSingleton<LogQueue>();
builder.Services.AddHostedService<LogWorkerManager>();
\$\endgroup\$
5
  • \$\begingroup\$ (Best of luck. I wince reading Manager.) \$\endgroup\$
    – greybeard
    Jul 11, 2022 at 6:51
  • \$\begingroup\$ LogWorker performs work with the database. These are IO operations, they should be performed asynchronously: DbContext.SaveChangesAsync and so on. Implement the IAsyncDisposable Interface. Don't use workerTask.Wait(); in the Dispose method. Use await workerTask in DisposeAsync method. \$\endgroup\$ Jul 12, 2022 at 4:05
  • \$\begingroup\$ Do we need to inject IServiceScopeFactory or can we do with injecting DbContext directly into the class constructor? \$\endgroup\$ Jul 12, 2022 at 4:12
  • 2
    \$\begingroup\$ Please do not edit the question, especially the code, after an answer has been posted. Changing the question may cause answer invalidation. Everyone needs to be able to see what the reviewer was referring to. What to do after the question has been answered. In this particular case it might be better to ask a follow up question with a link back to this question. \$\endgroup\$
    – pacmaninbw
    Jul 13, 2022 at 15:07
  • \$\begingroup\$ If the insert performance is still not good enough you could take a look at the EFCore.BulkExtensions NuGet package. Beware of the supported platforms. \$\endgroup\$
    – Johnbot
    Jul 14, 2022 at 9:00

1 Answer 1

3
\$\begingroup\$

This looks like an XY-problem.

The only work done in the worker is the DB Insert. If the worker can't keep up it's because the DB can't keep up. By creating additional workers the work is just moved to a new queue without a limit.


DbContext is not designed to be long living. The items added to the context will continue to be tracked until cleared or disposed.

This could be the source of poor performance.

\$\endgroup\$
0

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge you have read our privacy policy.

Not the answer you're looking for? Browse other questions tagged or ask your own question.