Distributed log analysis system — single-process implementation

.gitignore

@@ -0,0 +1,20 @@
+# Build output
+bin/
+obj/
+
+# IDE / user files
+.vs/
+.vscode/
+.idea/
+*.user
+*.suo
+
+# Runtime artifacts produced by the app
+*.db
+*.db-shm
+*.db-wal
+alerts.log
+
+# OS
+.DS_Store
+Thumbs.db

HowToRun.md

@@ -0,0 +1,13 @@
+If you ran 'dotnet run' right now, the app would:
+  1. Initialize SQLite (logs.db created, schema
+  applied).
+  2. Start IngestionHost, which starts
+  FileIngestionSource.
+  3. Stream src/apache_log.txt → parse → push to
+  channel → batch-write to SQLite.
+  4. Sit idle once the file is exhausted, waiting
+  for more producers (there are none).
+  5. GET /logs would return 404 because we haven't
+  mapped the endpoints yet.
+  6. The alert worker is registered but throws on
+  first tick because it's still a stub.

LogAnalysis.slnx

@@ -0,0 +1,5 @@
+<Solution>
+  <Folder Name="/src/">
+    <Project Path="src/LogAnalysis.System/LogAnalysis.System.csproj" />
+  </Folder>
+</Solution>

src/LogAnalysis.System/Alerting/AlertWorker.cs

@@ -0,0 +1,14 @@
+using LogAnalysis.System.Storage;
+
+namespace LogAnalysis.System.Alerting;
+
+public sealed class AlertWorker(
+    IEnumerable<IAlertRule> rules,
+    IEnumerable<IAlertSink> sinks,
+    ILogRepository          repository,
+    IConfiguration          config,
+    ILogger<AlertWorker>    logger) : BackgroundService
+{
+    protected override Task ExecuteAsync(CancellationToken ct)
+        => throw new NotImplementedException();
+}

src/LogAnalysis.System/Alerting/FileAlertSink.cs

@@ -0,0 +1,9 @@
+namespace LogAnalysis.System.Alerting;
+
+public sealed class FileAlertSink(
+    IConfiguration              config,
+    ILogger<FileAlertSink>      logger) : IAlertSink
+{
+    public Task WriteAsync(Alert alert, CancellationToken ct)
+        => throw new NotImplementedException();
+}

src/LogAnalysis.System/Alerting/IAlertRule.cs

@@ -0,0 +1,12 @@
+using LogAnalysis.System.Storage;
+
+namespace LogAnalysis.System.Alerting;
+
+public interface IAlertRule
+{
+    string Name { get; }
+
+    Task<Alert?> EvaluateAsync(ILogRepository repository, DateTimeOffset now, CancellationToken ct);
+}
+
+public sealed record Alert(string RuleName, string Detail, DateTimeOffset FiredAt);

src/LogAnalysis.System/Alerting/IAlertSink.cs

@@ -0,0 +1,6 @@
+namespace LogAnalysis.System.Alerting;
+
+public interface IAlertSink
+{
+    Task WriteAsync(Alert alert, CancellationToken ct);
+}

src/LogAnalysis.System/Alerting/PatternRule.cs

@@ -0,0 +1,14 @@
+using LogAnalysis.System.Storage;
+
+namespace LogAnalysis.System.Alerting;
+
+public sealed class PatternRule(
+    string   name,
+    string   substring,
+    TimeSpan window) : IAlertRule
+{
+    public string Name => name;
+
+    public Task<Alert?> EvaluateAsync(ILogRepository repository, DateTimeOffset now, CancellationToken ct)
+        => throw new NotImplementedException();
+}

src/LogAnalysis.System/Alerting/ThresholdRule.cs

@@ -0,0 +1,17 @@
+using LogAnalysis.System.Domain;
+using LogAnalysis.System.Storage;
+using LogLevel = LogAnalysis.System.Domain.LogLevel;
+
+namespace LogAnalysis.System.Alerting;
+
+public sealed class ThresholdRule(
+    string   name,
+    LogLevel level,
+    long     threshold,
+    TimeSpan window) : IAlertRule
+{
+    public string Name => name;
+
+    public Task<Alert?> EvaluateAsync(ILogRepository repository, DateTimeOffset now, CancellationToken ct)
+        => throw new NotImplementedException();
+}

src/LogAnalysis.System/Api/QueryEndpoints.cs

@@ -0,0 +1,21 @@
+using LogAnalysis.System.Storage;
+
+namespace LogAnalysis.System.Api;
+
+public static class QueryEndpoints
+{
+    public static IEndpointRouteBuilder MapQueryEndpoints(this IEndpointRouteBuilder app)
+    {
+        // GET /logs?startTime&endTime&level&service&limit&cursor
+        // GET /logs/aggregate?startTime&endTime&groupBy=level|service
+
+        // --- Production ingestion path (commented out for the demo) ----------
+        // POST /logs: accepts a JSON LogEntry, returns 202 Accepted, writes
+        // straight into the same LogChannel that FileIngestionSource uses.
+        // The pipeline downstream is source-agnostic, so enabling this is a
+        // one-method change. The demo replays src/apache_log.txt instead.
+        // ---------------------------------------------------------------------
+
+        return app;
+    }
+}

src/LogAnalysis.System/Domain/LogEntry.cs

@@ -0,0 +1,7 @@
+namespace LogAnalysis.System.Domain;
+
+public sealed record LogEntry(
+    DateTimeOffset Timestamp,
+    LogLevel       Level,
+    string         Service,
+    string         Message);

src/LogAnalysis.System/Domain/LogLevel.cs

@@ -0,0 +1,10 @@
+namespace LogAnalysis.System.Domain;
+
+public enum LogLevel
+{
+    Debug = 0,
+    Info  = 1,
+    Warn  = 2,
+    Error = 3,
+    Fatal = 4
+}

src/LogAnalysis.System/Ingestion/ApacheLogParser.cs

@@ -0,0 +1,39 @@
+using System.Globalization;
+using System.Text.RegularExpressions;
+using LogAnalysis.System.Domain;
+using LogLevel = LogAnalysis.System.Domain.LogLevel;
+
+namespace LogAnalysis.System.Ingestion;
+
+public sealed partial class ApacheLogParser : ILogParser
+{
+    private const string DateFormat = "dd/MMM/yyyy:HH:mm:ss zzz";
+    private const string ServiceTag = "apache";
+
+    [GeneratedRegex(
+        @"^\S+ \S+ \S+ \[(?<ts>[^\]]+)\] ""(?<req>[^""]*)"" (?<status>\d{3}) ",
+        RegexOptions.Compiled)]
+    private static partial Regex Pattern();
+
+    public bool CanParse(string line) => Pattern().IsMatch(line);
+
+    public LogEntry? Parse(string line)
+    {
+        var m = Pattern().Match(line);
+        if (!m.Success) return null;
+
+        if (!DateTimeOffset.TryParseExact(
+                m.Groups["ts"].Value, DateFormat,
+                CultureInfo.InvariantCulture, DateTimeStyles.None, out var ts))
+        {
+            ts = DateTimeOffset.UtcNow;
+        }
+
+        var status = int.Parse(m.Groups["status"].Value, CultureInfo.InvariantCulture);
+        var level  = status >= 500 ? LogLevel.Error
+                   : status >= 400 ? LogLevel.Warn
+                   :                 LogLevel.Info;
+
+        return new LogEntry(ts.ToUniversalTime(), level, ServiceTag, m.Groups["req"].Value);
+    }
+}

src/LogAnalysis.System/Ingestion/FileIngestionSource.cs

@@ -0,0 +1,44 @@
+using System.Runtime.CompilerServices;
+
+namespace LogAnalysis.System.Ingestion;
+
+public sealed class FileIngestionSource(
+    IngestionPipeline               pipeline,
+    IConfiguration                  config,
+    ILogger<FileIngestionSource>    logger) : IIngestionSource
+{
+    public async Task RunAsync(CancellationToken ct)
+    {
+        var path = config["Ingestion:FilePath"];
+        if (string.IsNullOrWhiteSpace(path))
+        {
+            logger.LogInformation("FileIngestionSource disabled — no Ingestion:FilePath configured");
+            return;
+        }
+        if (!File.Exists(path))
+        {
+            logger.LogWarning("Log file not found: {Path}", path);
+            return;
+        }
+
+        logger.LogInformation("Replaying {Path}", path);
+
+        var count = 0;
+        await foreach (var line in ReadLinesAsync(path, ct))
+        {
+            await pipeline.IngestAsync(line, ct);
+            count++;
+        }
+
+        logger.LogInformation("File ingestion complete: {Count} line(s) from {Path}", count, path);
+    }
+
+    private static async IAsyncEnumerable<string> ReadLinesAsync(
+        string path, [EnumeratorCancellation] CancellationToken ct)
+    {
+        using var reader = new StreamReader(path);
+        string? line;
+        while ((line = await reader.ReadLineAsync(ct)) is not null)
+            yield return line;
+    }
+}

src/LogAnalysis.System/Ingestion/IIngestionSource.cs

@@ -0,0 +1,6 @@
+namespace LogAnalysis.System.Ingestion;
+
+public interface IIngestionSource
+{
+    Task RunAsync(CancellationToken ct);
+}

src/LogAnalysis.System/Ingestion/ILogParser.cs

@@ -0,0 +1,9 @@
+using LogAnalysis.System.Domain;
+
+namespace LogAnalysis.System.Ingestion;
+
+public interface ILogParser
+{
+    bool CanParse(string line);
+    LogEntry? Parse(string line);
+}

src/LogAnalysis.System/Ingestion/IngestionHost.cs

@@ -0,0 +1,33 @@
+namespace LogAnalysis.System.Ingestion;
+
+public sealed class IngestionHost(
+    IEnumerable<IIngestionSource> sources,
+    ILogger<IngestionHost>        logger) : BackgroundService
+{
+    protected override Task ExecuteAsync(CancellationToken ct)
+    {
+        var tasks = sources.Select(s => RunSafe(s, ct)).ToArray();
+
+        if (tasks.Length == 0)
+        {
+            logger.LogInformation("No ingestion sources registered");
+            return Task.CompletedTask;
+        }
+
+        logger.LogInformation("Starting {Count} ingestion source(s)", tasks.Length);
+        return Task.WhenAll(tasks);
+    }
+
+    private async Task RunSafe(IIngestionSource source, CancellationToken ct)
+    {
+        try
+        {
+            await source.RunAsync(ct);
+        }
+        catch (OperationCanceledException) { }
+        catch (Exception ex)
+        {
+            logger.LogError(ex, "Ingestion source {Type} failed", source.GetType().Name);
+        }
+    }
+}

src/LogAnalysis.System/Ingestion/IngestionPipeline.cs

@@ -0,0 +1,32 @@
+using LogAnalysis.System.Pipeline;
+
+namespace LogAnalysis.System.Ingestion;
+
+public sealed class IngestionPipeline(
+    IEnumerable<ILogParser>      parsers,
+    LogChannel                   channel,
+    ILogger<IngestionPipeline>   logger)
+{
+    private readonly ILogParser[] _parsers = parsers.ToArray();
+
+    public async ValueTask IngestAsync(string line, CancellationToken ct)
+    {
+        if (string.IsNullOrWhiteSpace(line)) return;
+
+        foreach (var parser in _parsers)
+        {
+            if (!parser.CanParse(line)) continue;
+
+            var entry = parser.Parse(line);
+            if (entry is null) continue;
+
+            await channel.Writer.WriteAsync(entry, ct);
+            return;
+        }
+
+        logger.LogDebug("No parser matched line: {Preview}", Preview(line));
+    }
+
+    private static string Preview(string line) =>
+        line.Length <= 80 ? line : line[..80];
+}

src/LogAnalysis.System/LogAnalysis.System.csproj

@@ -0,0 +1,14 @@
+<Project Sdk="Microsoft.NET.Sdk.Web">
+
+  <PropertyGroup>
+    <TargetFramework>net10.0</TargetFramework>
+    <Nullable>enable</Nullable>
+    <ImplicitUsings>enable</ImplicitUsings>
+    <RootNamespace>LogAnalysis.System</RootNamespace>
+  </PropertyGroup>
+
+  <ItemGroup>
+    <PackageReference Include="Microsoft.Data.Sqlite" Version="9.0.0" />
+  </ItemGroup>
+
+</Project>

src/LogAnalysis.System/Pipeline/LogChannel.cs

@@ -0,0 +1,20 @@
+using System.Threading.Channels;
+using LogAnalysis.System.Domain;
+
+namespace LogAnalysis.System.Pipeline;
+
+public sealed class LogChannel
+{
+    private const int Capacity = 10_000;
+
+    private readonly Channel<LogEntry> _channel = Channel.CreateBounded<LogEntry>(
+        new BoundedChannelOptions(Capacity)
+        {
+            FullMode     = BoundedChannelFullMode.DropOldest,
+            SingleReader = true,
+            SingleWriter = false
+        });
+
+    public ChannelWriter<LogEntry> Writer => _channel.Writer;
+    public ChannelReader<LogEntry> Reader => _channel.Reader;
+}