diff --git a/EonaCat.LogStack.LogClient/EonaCat.LogStack.LogClient.csproj b/EonaCat.LogStack.LogClient/EonaCat.LogStack.LogClient.csproj
index 3bfd282..1393c9d 100644
--- a/EonaCat.LogStack.LogClient/EonaCat.LogStack.LogClient.csproj
+++ b/EonaCat.LogStack.LogClient/EonaCat.LogStack.LogClient.csproj
@@ -25,7 +25,7 @@
-
+
diff --git a/EonaCat.LogStack.WindowsEventLogFlow/EonaCat.LogStack.Flows.WindowsEventLog.csproj b/EonaCat.LogStack.WindowsEventLogFlow/EonaCat.LogStack.Flows.WindowsEventLog.csproj
index eeb6a69..5aad612 100644
--- a/EonaCat.LogStack.WindowsEventLogFlow/EonaCat.LogStack.Flows.WindowsEventLog.csproj
+++ b/EonaCat.LogStack.WindowsEventLogFlow/EonaCat.LogStack.Flows.WindowsEventLog.csproj
@@ -35,8 +35,8 @@
-
-
+
+
diff --git a/EonaCat.LogStack/EonaCat.LogStack.csproj b/EonaCat.LogStack/EonaCat.LogStack.csproj
index 6482035..3628e05 100644
--- a/EonaCat.LogStack/EonaCat.LogStack.csproj
+++ b/EonaCat.LogStack/EonaCat.LogStack.csproj
@@ -14,7 +14,7 @@ It features a rich fluent API for routing log events to dozens of destinations f
EonaCat (Jeroen Saey)
EonaCat;Logger;EonaCatLogStack;Log;Writer;Flows;LogStack;Memory;Speed;Jeroen;Saey
- 0.0.3
+ 0.0.4
README.md
True
LICENSE
@@ -25,7 +25,7 @@ It features a rich fluent API for routing log events to dozens of destinations f
- 0.0.3+{chash:10}.{c:ymd}
+ 0.0.4+{chash:10}.{c:ymd}
true
true
v[0-9]*
@@ -36,7 +36,7 @@ It features a rich fluent API for routing log events to dozens of destinations f
- 0.0.3
+ 0.0.4
EonaCat.LogStack
EonaCat.LogStack
https://git.saey.me/EonaCat/EonaCat.LogStack
@@ -66,18 +66,18 @@ It features a rich fluent API for routing log events to dozens of destinations f
-
-
+
+
all
runtime; build; native; contentfiles; analyzers; buildtransitive
-
-
-
+
+
+
-
+
diff --git a/EonaCat.LogStack/EonaCatLoggerCore/Flows/EncryptedFileFlow.cs b/EonaCat.LogStack/EonaCatLoggerCore/Flows/EncryptedFileFlow.cs
index 1e589b3..f62eec7 100644
--- a/EonaCat.LogStack/EonaCatLoggerCore/Flows/EncryptedFileFlow.cs
+++ b/EonaCat.LogStack/EonaCatLoggerCore/Flows/EncryptedFileFlow.cs
@@ -1,9 +1,14 @@
-using EonaCat.LogStack.Core;
+using EonaCat.LogStack.Core;
using EonaCat.LogStack.EonaCatLogStackCore;
+using EonaCat.LogStack.EonaCatLogStackCore.Policies;
using System;
using System.Collections.Concurrent;
+using System.Collections.Generic;
+using System.Diagnostics;
using System.IO;
+using System.IO.Compression;
using System.Linq;
+using System.Runtime.CompilerServices;
using System.Security.Cryptography;
using System.Text;
using System.Threading;
@@ -34,17 +39,37 @@ namespace EonaCat.LogStack.Flows
private const int KeySize = 32; // AES-256
private const int Pbkdf2Iter = 100000;
- private readonly BlockingCollection _queue;
+ private const int QueueCapacity = 8192;
+
+ private readonly BlockingCollection _queue;
private readonly CancellationTokenSource _cts = new CancellationTokenSource();
private readonly Thread _writerThread;
private readonly Thread _flushThread;
+ private readonly Thread _retentionThread;
+ private readonly Stopwatch _uptime = Stopwatch.StartNew();
+
+ private struct QueueEntry
+ {
+ public string Line;
+ public LogLevel Level;
+ }
private readonly string _directory;
private readonly string _filePrefix;
private readonly string _password;
private readonly long _maxFileSize;
+ private readonly long _maxDirectorySize;
+ private readonly FileRetentionPolicy _retention;
private readonly int _flushIntervalMs;
+ private readonly int _batchSize;
private readonly TimestampMode _timestampMode;
+ private readonly BackpressureStrategy _backpressure;
+ private readonly FileOutputFormat _outputFormat;
+ private readonly CompressionFormat _compressionFormat;
+ private readonly string _template;
+ private readonly bool _useCategoryRouting;
+ private readonly HashSet _logLevelsForSeparateFiles;
+ private readonly long _maxMemoryBytes;
private readonly object _lock = new object();
private FileStream _currentStream;
@@ -55,15 +80,85 @@ namespace EonaCat.LogStack.Flows
private long _totalWritten;
private long _totalRotations;
+ private long _totalBytesWritten;
+ private long _currentMemoryBytes;
+
+ public bool IgnoreConsoleErrors { get; set; }
+
+ private volatile SamplingPolicy _samplingPolicy;
+ private volatile Action _onDrop;
+ private volatile Action _onRotate;
+ private List> _secondaryWriters;
+ private readonly object _secondaryWritersLock = new object();
+
+ private readonly List> _filters = new List>();
+ private readonly object _filtersLock = new object();
+
+ private readonly ConcurrentDictionary _deduplicationCache
+ = new ConcurrentDictionary(StringComparer.Ordinal);
+ private TimeSpan _deduplicationWindow = TimeSpan.Zero;
+ private volatile bool _deduplicationEnabled;
+
+ private string _dateFormat = "yyyyMMdd";
+ private volatile Exception _lastError;
+ private long _lastErrorTimestamp;
+ private long _totalErrors;
+
+ private readonly List>> _enrichers
+ = new List>>();
+
+ // Rate limiting
+ private int _maxEventsPerSecond;
+ private bool _rateLimitEnabled => _maxEventsPerSecond > 0;
+ private long _rateLimitWindowStart;
+ private int _rateLimitCounter;
+ private readonly object _rateLimitLock = new object();
+
+ // Auto-flush on error
+ private volatile bool _autoFlushOnError;
+
+ // Scoped properties (AsyncLocal for ambient context)
+ private static readonly AsyncLocal> _scopeProperties
+ = new AsyncLocal>();
+
+ private static readonly Dictionary LevelStrings =
+ new Dictionary
+ {
+ { LogLevel.Trace, "TRACE" },
+ { LogLevel.Debug, "DEBUG" },
+ { LogLevel.Information, "INFO" },
+ { LogLevel.Warning, "WARN" },
+ { LogLevel.Error, "ERROR" },
+ { LogLevel.Critical, "CRITICAL" },
+ };
+
+ private static readonly string CachedMachineName = Environment.MachineName;
+ private static readonly int CachedPid = Process.GetCurrentProcess().Id;
+
+ private List> _compiledTemplate;
+ private readonly Dictionary> _customTokens
+ = new Dictionary>(StringComparer.OrdinalIgnoreCase);
+
+ private int _correlationSeed;
public EncryptedFileFlow(
string directory,
string password,
string filePrefix = "encrypted_log",
long maxFileSize = 50L * 1024 * 1024,
+ long maxDirectorySize = 2L * 1024 * 1024 * 1024,
+ FileRetentionPolicy retention = null,
int flushIntervalMs = 3000,
+ int batchSize = 1,
LogLevel minimumLevel = LogLevel.Trace,
- TimestampMode tsMode = TimestampMode.Utc)
+ bool useCategoryRouting = false,
+ LogLevel[] logLevelsForSeparateFiles = null,
+ TimestampMode tsMode = TimestampMode.Utc,
+ BackpressureStrategy backpressure = BackpressureStrategy.DropOldest,
+ FileOutputFormat outputFormat = FileOutputFormat.Text,
+ CompressionFormat compression = CompressionFormat.GZip,
+ string template = "[{ts}] [Host: {host}] [Category: {category}] [Thread: {thread}] [{logtype}] {message}{props}",
+ long maxMemoryBytes = 20 * 1024 * 1024)
: base("EncryptedFile:" + directory, minimumLevel)
{
if (directory == null)
@@ -81,12 +176,34 @@ namespace EonaCat.LogStack.Flows
throw new ArgumentNullException("filePrefix");
}
+ if (template == null)
+ {
+ throw new ArgumentNullException("template");
+ }
+
+ CheckForProcessTermination();
+
_directory = directory;
_password = password;
_filePrefix = filePrefix;
_maxFileSize = maxFileSize;
+ _maxDirectorySize = maxDirectorySize > 0 ? maxDirectorySize : 10L * maxFileSize;
+ _retention = retention ?? new FileRetentionPolicy();
_flushIntervalMs = flushIntervalMs;
+ _batchSize = batchSize <= 0 ? 1 : batchSize;
_timestampMode = tsMode;
+ _backpressure = backpressure;
+ _outputFormat = outputFormat;
+ _compressionFormat = compression;
+ _template = template;
+ _useCategoryRouting = useCategoryRouting;
+ _maxMemoryBytes = maxMemoryBytes;
+
+ _logLevelsForSeparateFiles = logLevelsForSeparateFiles != null
+ ? new HashSet(logLevelsForSeparateFiles)
+ : new HashSet();
+
+ CompileTemplate(template);
// Resolve relative path
if (_directory.StartsWith("./", StringComparison.Ordinal))
@@ -96,12 +213,12 @@ namespace EonaCat.LogStack.Flows
Directory.CreateDirectory(_directory);
- _queue = new BlockingCollection(new ConcurrentQueue(), 8192);
+ _queue = new BlockingCollection(new ConcurrentQueue(), QueueCapacity);
_writerThread = new Thread(WriterLoop)
{
IsBackground = true,
- Name = "EncryptedFileFlow.Writer",
+ Name = "EncryptedFileFlow.Writer[" + filePrefix + "]",
Priority = ThreadPriority.AboveNormal
};
_writerThread.Start();
@@ -109,10 +226,298 @@ namespace EonaCat.LogStack.Flows
_flushThread = new Thread(FlushLoop)
{
IsBackground = true,
- Name = "EncryptedFileFlow.Flush",
+ Name = "EncryptedFileFlow.Flush[" + filePrefix + "]",
Priority = ThreadPriority.BelowNormal
};
_flushThread.Start();
+
+ _retentionThread = new Thread(RetentionLoop)
+ {
+ IsBackground = true,
+ Name = "EncryptedFileFlow.Retention[" + filePrefix + "]",
+ Priority = ThreadPriority.BelowNormal
+ };
+ _retentionThread.Start();
+ }
+
+ private void CheckForProcessTermination()
+ {
+ AppDomain.CurrentDomain.ProcessExit += (_, __) =>
+ {
+ try
+ {
+ DisposeAsync().AsTask().GetAwaiter().GetResult();
+ }
+ catch
+ {
+ // Do nothing
+ }
+ };
+
+ Console.CancelKeyPress += (_, e) =>
+ {
+ try
+ {
+ DisposeAsync().AsTask().GetAwaiter().GetResult();
+ }
+ catch
+ {
+ // Do nothing
+ }
+ };
+ }
+
+ /// Add an ambient property enricher applied to every event.
+ public EncryptedFileFlow EnrichWith(string key, Func valueFactory)
+ {
+ if (key == null)
+ {
+ throw new ArgumentNullException("key");
+ }
+
+ if (valueFactory == null)
+ {
+ throw new ArgumentNullException("valueFactory");
+ }
+
+ _enrichers.Add(new KeyValuePair>(key, valueFactory));
+ return this;
+ }
+
+ /// Add a static ambient property.
+ public EncryptedFileFlow EnrichWith(string key, object value)
+ {
+ return EnrichWith(key, _ => value);
+ }
+
+ /// Configure sampling: only log 1 in events
+ /// that match .
+ public EncryptedFileFlow WithSampling(int rate, Func predicate = null)
+ {
+ _samplingPolicy = new SamplingPolicy { Rate = rate, Predicate = predicate };
+ return this;
+ }
+
+ /// Callback invoked when an event is dropped due to backpressure.
+ public EncryptedFileFlow OnEventDropped(Action callback)
+ {
+ _onDrop = callback;
+ return this;
+ }
+
+ /// Callback invoked with the archived path after each file rotation.
+ public EncryptedFileFlow OnFileRotated(Action callback)
+ {
+ _onRotate = callback;
+ return this;
+ }
+
+ /// Fan-out: also invoke for every formatted log line.
+ public EncryptedFileFlow AddSecondaryWriter(Action writer)
+ {
+ if (writer == null)
+ {
+ throw new ArgumentNullException("writer");
+ }
+
+ lock (_secondaryWritersLock)
+ {
+ if (_secondaryWriters == null)
+ {
+ _secondaryWriters = new List>();
+ }
+
+ _secondaryWriters.Add(writer);
+ }
+ return this;
+ }
+
+ /// Change the minimum log level at runtime (thread-safe).
+ public void SetMinimumLevel(LogLevel level)
+ {
+ MinimumLevel = level;
+ }
+
+ /// Register a custom template token (e.g. {mytoken}).
+ public EncryptedFileFlow RegisterToken(string name, Action formatter)
+ {
+ if (formatter == null)
+ {
+ throw new ArgumentNullException("formatter");
+ }
+
+ _customTokens[name] = formatter;
+ return this;
+ }
+
+ /// Returns the current estimated memory usage of the queue in bytes.
+ public long GetMemoryPressureBytes()
+ {
+ return Interlocked.Read(ref _currentMemoryBytes);
+ }
+
+ /// Add a custom filter predicate. Events are logged only if ALL filters return true.
+ public EncryptedFileFlow WithFilter(Func predicate)
+ {
+ if (predicate == null)
+ {
+ throw new ArgumentNullException("predicate");
+ }
+
+ lock (_filtersLock)
+ {
+ _filters.Add(predicate);
+ }
+ return this;
+ }
+
+ /// Enable deduplication: suppress identical messages within the given time window.
+ public EncryptedFileFlow WithDeduplication(TimeSpan window)
+ {
+ if (window <= TimeSpan.Zero)
+ {
+ throw new ArgumentOutOfRangeException("window", "Deduplication window must be positive.");
+ }
+
+ _deduplicationWindow = window;
+ _deduplicationEnabled = true;
+ return this;
+ }
+
+ /// Configure a custom date format for log file names (default: yyyyMMdd).
+ public EncryptedFileFlow WithDateFormat(string dateFormat)
+ {
+ if (string.IsNullOrWhiteSpace(dateFormat))
+ {
+ throw new ArgumentNullException("dateFormat");
+ }
+
+ _dateFormat = dateFormat;
+ return this;
+ }
+
+ /// Limit the flow to a maximum number of events per second. Events exceeding the limit are dropped.
+ public EncryptedFileFlow WithRateLimit(int maxEventsPerSecond)
+ {
+ _maxEventsPerSecond = maxEventsPerSecond;
+ return this;
+ }
+
+ /// When enabled, the file stream is flushed immediately after writing Error or Critical level events.
+ public EncryptedFileFlow WithAutoFlushOnError(bool enabled = true)
+ {
+ _autoFlushOnError = enabled;
+ return this;
+ }
+
+ /// Push scoped properties that will be included in all log events written on the current async context.
+ public IDisposable BeginScope(params KeyValuePair[] properties)
+ {
+ var previous = _scopeProperties.Value;
+ var merged = previous != null
+ ? new Dictionary(previous)
+ : new Dictionary();
+
+ foreach (var kv in properties)
+ {
+ merged[kv.Key] = kv.Value;
+ }
+
+ _scopeProperties.Value = merged;
+ return new ScopeDisposable(previous);
+ }
+
+ /// Push a single scoped property.
+ public IDisposable BeginScope(string key, object value)
+ {
+ return BeginScope(new KeyValuePair(key, value));
+ }
+
+ /// Returns the current queue depth (number of pending events).
+ public int GetQueueDepth()
+ {
+ return _queue.Count;
+ }
+
+ /// Generates a fingerprint hash for an exception to assist with grouping.
+ public static string GetExceptionFingerprint(Exception ex)
+ {
+ if (ex == null)
+ {
+ return null;
+ }
+
+ string source = string.Concat(
+ ex.GetType().FullName, "|",
+ ex.TargetSite?.Name ?? string.Empty, "|",
+ ex.StackTrace != null && ex.StackTrace.Length > 0
+ ? ex.StackTrace.Substring(0, Math.Min(200, ex.StackTrace.Length))
+ : string.Empty);
+
+ // Simple FNV-1a hash
+ unchecked
+ {
+ uint hash = 2166136261;
+ foreach (char c in source)
+ {
+ hash ^= c;
+ hash *= 16777619;
+ }
+ return hash.ToString("x8");
+ }
+ }
+
+ private sealed class ScopeDisposable : IDisposable
+ {
+ private readonly Dictionary _previous;
+
+ public ScopeDisposable(Dictionary previous)
+ {
+ _previous = previous;
+ }
+
+ public void Dispose()
+ {
+ _scopeProperties.Value = _previous;
+ }
+ }
+
+ /// Returns true if the flow is healthy (no recent errors and writer thread alive).
+ public bool IsHealthy()
+ {
+ if (!IsEnabled)
+ {
+ return false;
+ }
+
+ if (!_writerThread.IsAlive)
+ {
+ return false;
+ }
+
+ long lastErr = Interlocked.Read(ref _lastErrorTimestamp);
+ if (lastErr > 0)
+ {
+ TimeSpan since = TimeSpan.FromTicks(DateTime.UtcNow.Ticks - lastErr);
+ if (since < TimeSpan.FromMinutes(1))
+ {
+ return false;
+ }
+ }
+
+ return true;
+ }
+
+ /// Returns the last error encountered by the writer, or null if none.
+ public Exception GetLastError()
+ {
+ return _lastError;
+ }
+
+ /// Returns the total number of write errors encountered.
+ public long GetTotalErrors()
+ {
+ return Interlocked.Read(ref _totalErrors);
}
///
@@ -201,12 +606,16 @@ namespace EonaCat.LogStack.Flows
public LogStats GetStats()
{
- return new LogStats(
- Interlocked.Read(ref _totalWritten),
- Interlocked.Read(ref DroppedCount),
- Interlocked.Read(ref _totalRotations), 0, 0);
+ long written = Interlocked.Read(ref BlastedCount);
+ long dropped = Interlocked.Read(ref DroppedCount);
+ long bytes = Interlocked.Read(ref _totalBytesWritten);
+ long rots = Interlocked.Read(ref _totalRotations);
+ double elapsed = _uptime.Elapsed.TotalSeconds;
+ double wps = elapsed > 0 ? written / elapsed : 0;
+ return new LogStats(written, dropped, rots, bytes, wps);
}
+ [MethodImpl(MethodImplOptions.AggressiveInlining)]
public override Task BlastAsync(
LogEvent logEvent,
CancellationToken cancellationToken = default(CancellationToken))
@@ -216,7 +625,35 @@ namespace EonaCat.LogStack.Flows
return Task.FromResult(WriteResult.LevelFiltered);
}
- return Task.FromResult(TryEnqueue(Format(logEvent)));
+ SamplingPolicy sp = _samplingPolicy;
+ if (sp != null && !sp.ShouldLog(logEvent))
+ {
+ return Task.FromResult(WriteResult.LevelFiltered);
+ }
+
+ if (!PassesFilters(logEvent))
+ {
+ return Task.FromResult(WriteResult.LevelFiltered);
+ }
+
+ if (_deduplicationEnabled && IsDuplicate(logEvent))
+ {
+ return Task.FromResult(WriteResult.LevelFiltered);
+ }
+
+ if (_rateLimitEnabled && !TryPassRateLimit())
+ {
+ Interlocked.Increment(ref DroppedCount);
+ Action drop = _onDrop;
+ if (drop != null)
+ {
+ drop(logEvent);
+ }
+
+ return Task.FromResult(WriteResult.Dropped);
+ }
+
+ return Task.FromResult(TryEnqueue(logEvent));
}
public override Task BlastBatchAsync(
@@ -229,18 +666,51 @@ namespace EonaCat.LogStack.Flows
}
WriteResult result = WriteResult.Success;
- foreach (LogEvent e in logEvents.ToArray())
+ SamplingPolicy sp = _samplingPolicy;
+ ReadOnlySpan span = logEvents.Span;
+
+ for (int i = 0; i < span.Length; i++)
{
+ LogEvent e = span[i];
if (e.Level < MinimumLevel)
{
continue;
}
- if (TryEnqueue(Format(e)) == WriteResult.Dropped)
+ if (sp != null && !sp.ShouldLog(e))
+ {
+ continue;
+ }
+
+ if (!PassesFilters(e))
+ {
+ continue;
+ }
+
+ if (_deduplicationEnabled && IsDuplicate(e))
+ {
+ continue;
+ }
+
+ if (_rateLimitEnabled && !TryPassRateLimit())
+ {
+ Interlocked.Increment(ref DroppedCount);
+ Action drop = _onDrop;
+ if (drop != null)
+ {
+ drop(e);
+ }
+
+ result = WriteResult.Dropped;
+ continue;
+ }
+
+ if (TryEnqueue(e) == WriteResult.Dropped)
{
result = WriteResult.Dropped;
}
}
+
return Task.FromResult(result);
}
@@ -258,11 +728,17 @@ namespace EonaCat.LogStack.Flows
public override async ValueTask DisposeAsync()
{
+ if (!IsEnabled)
+ {
+ return;
+ }
+
IsEnabled = false;
_queue.CompleteAdding();
_cts.Cancel();
_writerThread.Join(TimeSpan.FromSeconds(5));
_flushThread.Join(TimeSpan.FromSeconds(2));
+ _retentionThread.Join(TimeSpan.FromSeconds(2));
lock (_lock) { CloseCurrentFile(); }
_cts.Dispose();
_queue.Dispose();
@@ -275,32 +751,32 @@ namespace EonaCat.LogStack.Flows
{
while (!_queue.IsCompleted)
{
- string line;
- try { line = _queue.Take(_cts.Token); }
+ QueueEntry entry;
+ try { entry = _queue.Take(_cts.Token); }
catch (OperationCanceledException) { break; }
catch (InvalidOperationException) { break; }
- WriteEncrypted(line);
+ WriteEncryptedLine(entry);
- string extra;
+ QueueEntry extra;
int batch = 0;
- while (batch < 256 && _queue.TryTake(out extra))
+ while (batch < _batchSize && _queue.TryTake(out extra))
{
- WriteEncrypted(extra);
+ WriteEncryptedLine(extra);
batch++;
}
}
}
catch (Exception ex)
{
- Console.Error.WriteLine("[EncryptedFileFlow] Writer error: " + ex.Message);
+ WriteToConsoleError("[EncryptedFileFlow] Writer error: " + ex.Message);
}
finally
{
- string remaining;
+ QueueEntry remaining;
while (_queue.TryTake(out remaining))
{
- WriteEncrypted(remaining);
+ WriteEncryptedLine(remaining);
}
lock (_lock) { CloseCurrentFile(); }
@@ -324,8 +800,12 @@ namespace EonaCat.LogStack.Flows
}
}
- private void WriteEncrypted(string line)
+ private void WriteEncryptedLine(QueueEntry entry)
{
+ // Deduct memory estimate
+ long size = 200L + (entry.Line != null ? entry.Line.Length * 2 : 0);
+ Interlocked.Add(ref _currentMemoryBytes, -size);
+
lock (_lock)
{
DateTime today = _timestampMode == TimestampMode.Local
@@ -337,22 +817,45 @@ namespace EonaCat.LogStack.Flows
if (_currentStream != null)
{
Interlocked.Increment(ref _totalRotations);
+ Action onRotate = _onRotate;
+ if (onRotate != null && _currentPath != null)
+ {
+ try { onRotate(_currentPath); }
+ catch { /* Do nothing */ }
+ }
}
CloseCurrentFile();
OpenNewFile(today);
}
- byte[] plain = Encoding.UTF8.GetBytes(line);
- byte[] cipher = _encryptor.TransformFinalBlock(plain, 0, plain.Length);
- byte[] lenBuf = BitConverter.GetBytes(cipher.Length);
+ try
+ {
+ byte[] plain = Encoding.UTF8.GetBytes(entry.Line);
+ byte[] cipher = _encryptor.TransformFinalBlock(plain, 0, plain.Length);
+ byte[] lenBuf = BitConverter.GetBytes(cipher.Length);
- _currentStream.Write(lenBuf, 0, 4);
- _currentStream.Write(cipher, 0, cipher.Length);
- _currentSize += 4 + cipher.Length;
+ _currentStream.Write(lenBuf, 0, 4);
+ _currentStream.Write(cipher, 0, cipher.Length);
+ _currentSize += 4 + cipher.Length;
- Interlocked.Increment(ref _totalWritten);
- Interlocked.Increment(ref BlastedCount);
+ Interlocked.Increment(ref _totalWritten);
+ Interlocked.Increment(ref BlastedCount);
+ Interlocked.Add(ref _totalBytesWritten, 4 + cipher.Length);
+
+ // Auto-flush on error/critical
+ if (_autoFlushOnError && entry.Level >= LogLevel.Error)
+ {
+ try { _currentStream.Flush(true); } catch { /* ignore */ }
+ }
+ }
+ catch (Exception ex)
+ {
+ _lastError = ex;
+ Interlocked.Increment(ref _totalErrors);
+ Interlocked.Exchange(ref _lastErrorTimestamp, DateTime.UtcNow.Ticks);
+ WriteToConsoleError("[EncryptedFileFlow] Write error: " + ex.Message);
+ }
}
}
@@ -361,7 +864,7 @@ namespace EonaCat.LogStack.Flows
_currentDate = date;
_currentPath = Path.Combine(
_directory,
- _filePrefix + "_" + Environment.MachineName + "_" + date.ToString("yyyyMMdd") + ".eona");
+ _filePrefix + "_" + Environment.MachineName + "_" + date.ToString(_dateFormat) + ".eona");
bool isNew = !File.Exists(_currentPath) || new FileInfo(_currentPath).Length == 0;
@@ -420,43 +923,412 @@ namespace EonaCat.LogStack.Flows
}
}
- private WriteResult TryEnqueue(string line)
+ private WriteResult TryEnqueue(LogEvent logEvent)
{
- if (_queue.TryAdd(line))
+ long size = EstimateSize(logEvent);
+ long current = Interlocked.Read(ref _currentMemoryBytes);
+
+ if (current + size > _maxMemoryBytes)
{
- return WriteResult.Success;
+ Interlocked.Increment(ref DroppedCount);
+ Action drop = _onDrop;
+ if (drop != null)
+ {
+ drop(logEvent);
+ }
+
+ return WriteResult.Dropped;
}
- Interlocked.Increment(ref DroppedCount);
- return WriteResult.Dropped;
+ string line = Format(logEvent);
+ var entry = new QueueEntry { Line = line, Level = logEvent.Level };
+
+ if (!_queue.TryAdd(entry))
+ {
+ Interlocked.Increment(ref DroppedCount);
+ Action drop = _onDrop;
+ if (drop != null)
+ {
+ drop(logEvent);
+ }
+
+ return WriteResult.Dropped;
+ }
+
+ Interlocked.Add(ref _currentMemoryBytes, size);
+ Interlocked.Increment(ref BlastedCount);
+
+ // Fan-out to secondary writers
+ try
+ {
+ if (_secondaryWriters != null)
+ {
+ lock (_secondaryWritersLock)
+ {
+ foreach (Action writer in _secondaryWriters)
+ {
+ try { writer(logEvent, line); }
+ catch { /* Do nothing */ }
+ }
+ }
+ }
+ }
+ catch { /* Do nothing */ }
+
+ return WriteResult.Success;
+ }
+
+ [MethodImpl(MethodImplOptions.AggressiveInlining)]
+ private bool TryPassRateLimit()
+ {
+ long now = DateTime.UtcNow.Ticks;
+ lock (_rateLimitLock)
+ {
+ long elapsed = now - _rateLimitWindowStart;
+ if (elapsed >= TimeSpan.TicksPerSecond)
+ {
+ _rateLimitWindowStart = now;
+ _rateLimitCounter = 1;
+ return true;
+ }
+
+ if (_rateLimitEnabled && _rateLimitCounter >= _maxEventsPerSecond)
+ {
+ return false;
+ }
+
+ _rateLimitCounter++;
+ return true;
+ }
+ }
+
+ [MethodImpl(MethodImplOptions.AggressiveInlining)]
+ private bool PassesFilters(LogEvent log)
+ {
+ lock (_filtersLock)
+ {
+ for (int i = 0; i < _filters.Count; i++)
+ {
+ if (!_filters[i](log))
+ {
+ return false;
+ }
+ }
+ }
+ return true;
+ }
+
+ private bool IsDuplicate(LogEvent log)
+ {
+ string key = string.Concat(
+ log.Level.ToString(), "|",
+ log.Category ?? string.Empty, "|",
+ log.Message.Length > 0 ? log.Message.ToString() : string.Empty);
+
+ long nowTicks = DateTime.UtcNow.Ticks;
+ long windowTicks = _deduplicationWindow.Ticks;
+
+ long existing;
+ if (_deduplicationCache.TryGetValue(key, out existing))
+ {
+ if (nowTicks - existing < windowTicks)
+ {
+ return true;
+ }
+ }
+
+ _deduplicationCache[key] = nowTicks;
+
+ if (_deduplicationCache.Count > 10000)
+ {
+ CleanDeduplicationCache(nowTicks, windowTicks);
+ }
+
+ return false;
+ }
+
+ private void CleanDeduplicationCache(long nowTicks, long windowTicks)
+ {
+ foreach (var kvp in _deduplicationCache)
+ {
+ if (nowTicks - kvp.Value >= windowTicks)
+ {
+ long removed;
+ _deduplicationCache.TryRemove(kvp.Key, out removed);
+ }
+ }
+ }
+
+ private void WriteToConsoleError(string text)
+ {
+ if (IgnoreConsoleErrors)
+ {
+ return;
+ }
+
+ Console.Error.WriteLine(text);
}
private string Format(LogEvent log)
{
- DateTime ts = LogEvent.GetDateTime(log.Timestamp);
var sb = new StringBuilder(256);
- sb.Append(ts.ToString("yyyy-MM-dd HH:mm:ss.fff"));
- sb.Append(" [").Append(LevelString(log.Level)).Append("] ");
- sb.Append(log.Category ?? string.Empty);
- sb.Append(": ");
- sb.Append(log.Message.Length > 0 ? log.Message.ToString() : string.Empty);
-
- if (log.Exception != null)
+ foreach (Action action in _compiledTemplate)
{
- sb.Append(" | EX: ").Append(log.Exception.GetType().Name)
- .Append(": ").Append(log.Exception.Message);
+ action(log, sb);
+ }
+ return sb.ToString();
+ }
+
+ private void AppendProperties(LogEvent log, StringBuilder sb)
+ {
+ var scopeProps = _scopeProperties.Value;
+ bool hasEnrichers = _enrichers.Count > 0;
+ bool hasProps = log.Properties.Count > 0;
+ bool hasScope = scopeProps != null && scopeProps.Count > 0;
+ if (!hasEnrichers && !hasProps && !hasScope)
+ {
+ return;
}
- if (log.Properties.Count > 0)
+ sb.Append(" {");
+ bool first = true;
+
+ foreach (KeyValuePair> kv in _enrichers)
{
- sb.Append(" |");
- foreach (var kv in log.Properties.ToArray())
+ if (!first)
{
- sb.Append(' ').Append(kv.Key).Append('=')
+ sb.Append(", ");
+ }
+
+ first = false;
+ object val = kv.Value(log);
+ sb.Append(kv.Key).Append('=').Append(val != null ? val.ToString() : "null");
+ }
+
+ foreach (var property in log.Properties)
+ {
+ if (!first)
+ {
+ sb.Append(", ");
+ }
+
+ first = false;
+ sb.Append(property.Key).Append('=')
+ .Append(property.Value != null ? property.Value.ToString() : "null");
+ }
+
+ if (hasScope)
+ {
+ foreach (var kv in scopeProps)
+ {
+ if (!first)
+ {
+ sb.Append(", ");
+ }
+
+ first = false;
+ sb.Append(kv.Key).Append('=')
.Append(kv.Value != null ? kv.Value.ToString() : "null");
}
}
- return sb.ToString();
+
+ sb.Append('}');
+ }
+
+ private void CompileTemplate(string template)
+ {
+ _compiledTemplate = new List>();
+ int pos = 0;
+
+ while (pos < template.Length)
+ {
+ int open = template.IndexOf('{', pos);
+ if (open < 0)
+ {
+ string lit = template.Substring(pos);
+ _compiledTemplate.Add((_, sb) => sb.Append(lit));
+ break;
+ }
+
+ if (open > pos)
+ {
+ string lit = template.Substring(pos, open - pos);
+ _compiledTemplate.Add((_, sb) => sb.Append(lit));
+ }
+
+ int close = template.IndexOf('}', open);
+ if (close < 0)
+ {
+ string lit = template.Substring(open);
+ _compiledTemplate.Add((_, sb) => sb.Append(lit));
+ break;
+ }
+
+ string token = template.Substring(open + 1, close - (open + 1));
+ _compiledTemplate.Add(ResolveToken(token));
+ pos = close + 1;
+ }
+ }
+
+ private Action ResolveToken(string token)
+ {
+ switch (token.ToLowerInvariant())
+ {
+ case "ts":
+ return (log, sb) =>
+ sb.Append(LogEvent.GetDateTime(log.Timestamp).ToString("yyyy-MM-dd HH:mm:ss.fff"));
+ case "tz":
+ return (log, sb) =>
+ sb.Append(_timestampMode == TimestampMode.Local
+ ? TimeZoneInfo.Local.StandardName : "UTC");
+ case "host":
+ return (log, sb) => sb.Append(CachedMachineName);
+ case "category":
+ return (log, sb) => { if (log.Category != null) { sb.Append(log.Category); } };
+ case "thread":
+ return (log, sb) => sb.Append(Thread.CurrentThread.ManagedThreadId);
+ case "logtype":
+ return (log, sb) =>
+ {
+ string s;
+ sb.Append(LevelStrings.TryGetValue(log.Level, out s) ? s : log.Level.ToString());
+ };
+ case "message":
+ return (log, sb) =>
+ {
+ if (log.Message.Length > 0)
+ {
+ sb.Append(log.Message.ToString());
+ }
+ };
+ case "exception":
+ return (log, sb) =>
+ {
+ if (log.Exception != null)
+ {
+ sb.Append(log.Exception.ToString());
+ }
+ };
+ case "props":
+ return (log, sb) => AppendProperties(log, sb);
+ case "newline":
+ return (log, sb) => sb.AppendLine();
+ case "pid":
+ return (log, sb) => sb.Append(CachedPid);
+ case "traceid":
+ return (log, sb) =>
+ {
+ if (log.TraceId != default(ActivityTraceId))
+ {
+ sb.Append(log.TraceId.ToHexString());
+ }
+ };
+ case "spanid":
+ return (log, sb) =>
+ {
+ if (log.SpanId != default(ActivitySpanId))
+ {
+ sb.Append(log.SpanId.ToHexString());
+ }
+ };
+ default:
+ return BuildCustomOrLiteralToken(token);
+ }
+ }
+
+ private Action BuildCustomOrLiteralToken(string token)
+ {
+ string name = token;
+ return (log, sb) =>
+ {
+ Action custom;
+ if (_customTokens.TryGetValue(name, out custom))
+ {
+ custom(log, sb);
+ }
+ else
+ {
+ sb.Append('{').Append(name).Append('}');
+ }
+ };
+ }
+
+ private void RetentionLoop()
+ {
+ while (!_cts.Token.IsCancellationRequested)
+ {
+ try
+ {
+ if (_cts.Token.WaitHandle.WaitOne(TimeSpan.FromMinutes(15)))
+ {
+ break;
+ }
+
+ ApplyRetention();
+ }
+ catch (Exception ex)
+ {
+ WriteToConsoleError("[EncryptedFileFlow] Retention error: " + ex.Message);
+ }
+ }
+ }
+
+ private void ApplyRetention()
+ {
+ try
+ {
+ DirectoryInfo dir = new DirectoryInfo(_directory);
+ if (!dir.Exists)
+ {
+ return;
+ }
+
+ FileInfo[] files = dir.GetFiles("*.eona")
+ .OrderByDescending(f => f.LastWriteTimeUtc)
+ .ToArray();
+
+ long totalBytes = 0;
+ int kept = 0;
+
+ foreach (FileInfo f in files)
+ {
+ bool tooOld = _retention.MaxAgeDays > 0
+ && (DateTime.UtcNow - f.LastWriteTimeUtc).TotalDays > _retention.MaxAgeDays;
+ bool tooMany = _retention.MaxRolledFiles > 0 && kept >= _retention.MaxRolledFiles;
+ bool tooLarge = _retention.MaxTotalArchiveBytes > 0
+ && totalBytes + f.Length > _retention.MaxTotalArchiveBytes;
+ bool directoryTooLarge = totalBytes + f.Length > _maxDirectorySize;
+
+ if (tooOld || tooMany || tooLarge || directoryTooLarge)
+ {
+ try { f.Delete(); } catch { /* ignore */ }
+ }
+ else
+ {
+ totalBytes += f.Length;
+ kept++;
+ }
+ }
+ }
+ catch (Exception ex)
+ {
+ WriteToConsoleError("[EncryptedFileFlow] Retention error: " + ex.Message);
+ }
+ }
+
+ [MethodImpl(MethodImplOptions.AggressiveInlining)]
+ private static long EstimateSize(LogEvent log)
+ {
+ long s = 200L
+ + log.Message.Length * 2
+ + (log.Category != null ? log.Category.Length * 2 : 0)
+ + log.Properties.Count * 40;
+ if (log.Exception != null)
+ {
+ s += 2048;
+ }
+
+ return s;
}
private static string LevelString(LogLevel level)
diff --git a/EonaCat.LogStack/EonaCatLoggerCore/Flows/FileFlow.cs b/EonaCat.LogStack/EonaCatLoggerCore/Flows/FileFlow.cs
index 06b9142..4dcdbdf 100644
--- a/EonaCat.LogStack/EonaCatLoggerCore/Flows/FileFlow.cs
+++ b/EonaCat.LogStack/EonaCatLoggerCore/Flows/FileFlow.cs
@@ -1,5 +1,4 @@
-using EonaCat.Json;
-using EonaCat.LogStack.Core;
+using EonaCat.LogStack.Core;
using EonaCat.LogStack.EonaCatLogStackCore;
using EonaCat.LogStack.EonaCatLogStackCore.Policies;
using System;
@@ -42,6 +41,10 @@ namespace EonaCat.LogStack.Flows
{ LogLevel.Critical, "CRITICAL" },
};
+ private static readonly char[] CsvSpecialChars = { ',', '"', '\n', '\r' };
+ private static readonly string CachedMachineName = Environment.MachineName;
+ private static readonly int CachedPid = Process.GetCurrentProcess().Id;
+
private const string CsvHeader = "timestamp,level,category,message,exception,properties\r\n";
private readonly BlockingCollection _queue;
@@ -57,6 +60,7 @@ namespace EonaCat.LogStack.Flows
private readonly string _directory;
private readonly string _filePrefix;
private readonly long _maxFileSize;
+ private readonly long _maxDirectorySize;
private readonly FileRetentionPolicy _retention;
private readonly TimestampMode _timestampMode;
private readonly TimeSpan _flushInterval;
@@ -74,11 +78,38 @@ namespace EonaCat.LogStack.Flows
private readonly object _secondaryWritersLock = new object();
private int _correlationSeed;
+ private readonly List> _filters = new List>();
+ private readonly object _filtersLock = new object();
+
+ private readonly ConcurrentDictionary _deduplicationCache
+ = new ConcurrentDictionary(StringComparer.Ordinal);
+ private TimeSpan _deduplicationWindow = TimeSpan.Zero;
+ private volatile bool _deduplicationEnabled;
+
+ private string _dateFormat = "yyyyMMdd";
+ private volatile Exception _lastError;
+ private long _lastErrorTimestamp;
+ private long _totalErrors;
+
private readonly List>> _enrichers
= new List>>();
private long _currentMemoryBytes;
+ // Rate limiting
+ private int _maxEventsPerSecond;
+ private bool _rateLimitEnabled => _maxEventsPerSecond > 0;
+ private long _rateLimitWindowStart;
+ private int _rateLimitCounter;
+ private readonly object _rateLimitLock = new object();
+
+ // Auto-flush on error
+ private volatile bool _autoFlushOnError;
+
+ // Scoped properties (AsyncLocal for ambient context)
+ private static readonly AsyncLocal> _scopeProperties
+ = new AsyncLocal>();
+
private long _totalBytesWritten;
private long _totalRotations;
@@ -121,6 +152,7 @@ namespace EonaCat.LogStack.Flows
string directory,
string filePrefix = "log",
long maxFileSize = 200 * 1024 * 1024,
+ long maxDirectorySize = 2L * 1024 * 1024 * 1024,
FileRetentionPolicy retention = null,
int flushIntervalMs = 2000,
int batchSize = 1,
@@ -157,6 +189,7 @@ namespace EonaCat.LogStack.Flows
_filePrefix = filePrefix;
_template = template;
_maxFileSize = maxFileSize;
+ _maxDirectorySize = maxDirectorySize > 0 ? maxDirectorySize : 10L * maxFileSize;
_retention = retention ?? new FileRetentionPolicy();
_timestampMode = timestampMode;
_useCategoryRouting = useCategoryRouting;
@@ -318,6 +351,176 @@ namespace EonaCat.LogStack.Flows
MinimumLevel = level;
}
+ /// Add a custom filter predicate. Events are logged only if ALL filters return true.
+ public FileFlow WithFilter(Func predicate)
+ {
+ if (predicate == null)
+ {
+ throw new ArgumentNullException("predicate");
+ }
+
+ lock (_filtersLock)
+ {
+ _filters.Add(predicate);
+ }
+ return this;
+ }
+
+ /// Enable deduplication: suppress identical messages within the given time window.
+ public FileFlow WithDeduplication(TimeSpan window)
+ {
+ if (window <= TimeSpan.Zero)
+ {
+ throw new ArgumentOutOfRangeException("window", "Deduplication window must be positive.");
+ }
+
+ _deduplicationWindow = window;
+ _deduplicationEnabled = true;
+ return this;
+ }
+
+ /// Configure a custom date format for log file names (default: yyyyMMdd).
+ public FileFlow WithDateFormat(string dateFormat)
+ {
+ if (string.IsNullOrWhiteSpace(dateFormat))
+ {
+ throw new ArgumentNullException("dateFormat");
+ }
+
+ _dateFormat = dateFormat;
+ return this;
+ }
+
+ /// Limit the flow to a maximum number of events per second. Events exceeding the limit are dropped.
+ public FileFlow WithRateLimit(int maxEventsPerSecond)
+ {
+ _maxEventsPerSecond = maxEventsPerSecond;
+ return this;
+ }
+
+ /// When enabled, the file stream is flushed immediately after writing Error or Critical level events.
+ public FileFlow WithAutoFlushOnError(bool enabled = true)
+ {
+ _autoFlushOnError = enabled;
+ return this;
+ }
+
+ /// Push scoped properties that will be included in all log events written on the current async context.
+ public IDisposable BeginScope(params KeyValuePair[] properties)
+ {
+ var previous = _scopeProperties.Value;
+ var merged = previous != null
+ ? new Dictionary(previous)
+ : new Dictionary();
+
+ foreach (var kv in properties)
+ {
+ merged[kv.Key] = kv.Value;
+ }
+
+ _scopeProperties.Value = merged;
+ return new ScopeDisposable(previous);
+ }
+
+ /// Push a single scoped property.
+ public IDisposable BeginScope(string key, object value)
+ {
+ return BeginScope(new KeyValuePair(key, value));
+ }
+
+ /// Returns the current queue depth (number of pending events).
+ public int GetQueueDepth()
+ {
+ return _queue.Count;
+ }
+
+ /// Returns the current estimated memory usage of the queue in bytes.
+ public long GetMemoryPressureBytes()
+ {
+ return Interlocked.Read(ref _currentMemoryBytes);
+ }
+
+ /// Generates a fingerprint hash for an exception to assist with grouping.
+ public static string GetExceptionFingerprint(Exception ex)
+ {
+ if (ex == null)
+ {
+ return null;
+ }
+
+ string source = string.Concat(
+ ex.GetType().FullName, "|",
+ ex.TargetSite?.Name ?? string.Empty, "|",
+ ex.StackTrace != null && ex.StackTrace.Length > 0
+ ? ex.StackTrace.Substring(0, Math.Min(200, ex.StackTrace.Length))
+ : string.Empty);
+
+ // Simple FNV-1a hash
+ unchecked
+ {
+ uint hash = 2166136261;
+ foreach (char c in source)
+ {
+ hash ^= c;
+ hash *= 16777619;
+ }
+ return hash.ToString("x8");
+ }
+ }
+
+ private sealed class ScopeDisposable : IDisposable
+ {
+ private readonly Dictionary _previous;
+
+ public ScopeDisposable(Dictionary previous)
+ {
+ _previous = previous;
+ }
+
+ public void Dispose()
+ {
+ _scopeProperties.Value = _previous;
+ }
+ }
+
+ /// Returns true if the flow is healthy (no recent errors and writer thread alive).
+ public bool IsHealthy()
+ {
+ if (!IsEnabled)
+ {
+ return false;
+ }
+
+ if (!_writerThread.IsAlive)
+ {
+ return false;
+ }
+
+ long lastErr = Interlocked.Read(ref _lastErrorTimestamp);
+ if (lastErr > 0)
+ {
+ TimeSpan since = TimeSpan.FromTicks(DateTime.UtcNow.Ticks - lastErr);
+ if (since < TimeSpan.FromMinutes(1))
+ {
+ return false;
+ }
+ }
+
+ return true;
+ }
+
+ /// Returns the last error encountered by the writer, or null if none.
+ public Exception GetLastError()
+ {
+ return _lastError;
+ }
+
+ /// Returns the total number of write errors encountered.
+ public long GetTotalErrors()
+ {
+ return Interlocked.Read(ref _totalErrors);
+ }
+
/// Returns live throughput and health metrics.
public LogStats GetStats()
{
@@ -347,6 +550,28 @@ namespace EonaCat.LogStack.Flows
return Task.FromResult(WriteResult.LevelFiltered);
}
+ if (!PassesFilters(logEvent))
+ {
+ return Task.FromResult(WriteResult.LevelFiltered);
+ }
+
+ if (_deduplicationEnabled && IsDuplicate(logEvent))
+ {
+ return Task.FromResult(WriteResult.LevelFiltered);
+ }
+
+ if (_rateLimitEnabled && !TryPassRateLimit())
+ {
+ Interlocked.Increment(ref DroppedCount);
+ Action drop = _onDrop;
+ if (drop != null)
+ {
+ drop(logEvent);
+ }
+
+ return Task.FromResult(WriteResult.Dropped);
+ }
+
return Task.FromResult(TryEnqueue(logEvent));
}
@@ -361,9 +586,11 @@ namespace EonaCat.LogStack.Flows
WriteResult result = WriteResult.Success;
SamplingPolicy sp = _samplingPolicy;
+ ReadOnlySpan span = logEvents.Span;
- foreach (LogEvent e in logEvents.ToArray())
+ for (int i = 0; i < span.Length; i++)
{
+ LogEvent e = span[i];
if (e.Level < MinimumLevel)
{
continue;
@@ -374,6 +601,29 @@ namespace EonaCat.LogStack.Flows
continue;
}
+ if (!PassesFilters(e))
+ {
+ continue;
+ }
+
+ if (_deduplicationEnabled && IsDuplicate(e))
+ {
+ continue;
+ }
+
+ if (_rateLimitEnabled && !TryPassRateLimit())
+ {
+ Interlocked.Increment(ref DroppedCount);
+ Action drop = _onDrop;
+ if (drop != null)
+ {
+ drop(e);
+ }
+
+ result = WriteResult.Dropped;
+ continue;
+ }
+
if (TryEnqueue(e) == WriteResult.Dropped)
{
result = WriteResult.Dropped;
@@ -418,6 +668,88 @@ namespace EonaCat.LogStack.Flows
return WriteResult.Success;
}
+ [MethodImpl(MethodImplOptions.AggressiveInlining)]
+ private bool TryPassRateLimit()
+ {
+ long now = DateTime.UtcNow.Ticks;
+ lock (_rateLimitLock)
+ {
+ long elapsed = now - _rateLimitWindowStart;
+ if (elapsed >= TimeSpan.TicksPerSecond)
+ {
+ _rateLimitWindowStart = now;
+ _rateLimitCounter = 1;
+ return true;
+ }
+
+ if (_rateLimitEnabled && _rateLimitCounter >= _maxEventsPerSecond)
+ {
+ return false;
+ }
+
+ _rateLimitCounter++;
+ return true;
+ }
+ }
+
+ [MethodImpl(MethodImplOptions.AggressiveInlining)]
+ private bool PassesFilters(LogEvent log)
+ {
+ lock (_filtersLock)
+ {
+ for (int i = 0; i < _filters.Count; i++)
+ {
+ if (!_filters[i](log))
+ {
+ return false;
+ }
+ }
+ }
+ return true;
+ }
+
+ private bool IsDuplicate(LogEvent log)
+ {
+ string key = string.Concat(
+ log.Level.ToString(), "|",
+ log.Category ?? string.Empty, "|",
+ log.Message.Length > 0 ? log.Message.ToString() : string.Empty);
+
+ long nowTicks = DateTime.UtcNow.Ticks;
+ long windowTicks = _deduplicationWindow.Ticks;
+
+ long existing;
+ if (_deduplicationCache.TryGetValue(key, out existing))
+ {
+ if (nowTicks - existing < windowTicks)
+ {
+ return true;
+ }
+ }
+
+ _deduplicationCache[key] = nowTicks;
+
+ // Periodic cleanup: remove expired entries when cache grows large
+ if (_deduplicationCache.Count > 10000)
+ {
+ CleanDeduplicationCache(nowTicks, windowTicks);
+ }
+
+ return false;
+ }
+
+ private void CleanDeduplicationCache(long nowTicks, long windowTicks)
+ {
+ foreach (var kvp in _deduplicationCache)
+ {
+ if (nowTicks - kvp.Value >= windowTicks)
+ {
+ long removed;
+ _deduplicationCache.TryRemove(kvp.Key, out removed);
+ }
+ }
+ }
+
public override Task FlushAsync(CancellationToken cancellationToken = default(CancellationToken))
{
@@ -642,6 +974,9 @@ namespace EonaCat.LogStack.Flows
}
catch (Exception ex)
{
+ _lastError = ex;
+ Interlocked.Increment(ref _totalErrors);
+ Interlocked.Exchange(ref _lastErrorTimestamp, DateTime.UtcNow.Ticks);
try
{
WriteToConsoleError("[FileFlow] Write error for '" + path + "': " + ex.Message);
@@ -651,6 +986,19 @@ namespace EonaCat.LogStack.Flows
}
Interlocked.Add(ref _totalBytesWritten, line.Length + 1);
+
+ // Auto-flush on error/critical
+ if (_autoFlushOnError && (log.Level >= LogLevel.Error))
+ {
+ lock (_fileLock)
+ {
+ OpenFile autoFlushOf;
+ if (_openFiles.TryGetValue(path, out autoFlushOf))
+ {
+ try { autoFlushOf.Writer.Flush(); } catch { /* ignore */ }
+ }
+ }
+ }
}
catch (Exception ex)
{
@@ -719,11 +1067,30 @@ namespace EonaCat.LogStack.Flows
sb.Append("\",");
sb.Append("\"host\":\"");
- AppendJsonEscaped(Environment.MachineName, sb);
+ AppendJsonEscaped(CachedMachineName, sb);
sb.Append("\",");
sb.Append("\"pid\":");
- sb.Append(Process.GetCurrentProcess().Id);
+ sb.Append(CachedPid);
+ sb.Append(',');
+
+ // Distributed tracing
+ if (log.TraceId != default(ActivityTraceId))
+ {
+ sb.Append("\"traceId\":\"");
+ sb.Append(log.TraceId.ToHexString());
+ sb.Append("\",");
+ }
+
+ if (log.SpanId != default(ActivitySpanId))
+ {
+ sb.Append("\"spanId\":\"");
+ sb.Append(log.SpanId.ToHexString());
+ sb.Append("\",");
+ }
+
+ sb.Append("\"threadId\":");
+ sb.Append(log.ThreadId);
sb.Append(',');
}
@@ -777,7 +1144,7 @@ namespace EonaCat.LogStack.Flows
sb.Append('"');
}
- foreach (var property in log.Properties.ToArray())
+ foreach (var property in log.Properties)
{
if (!first)
{
@@ -927,7 +1294,7 @@ namespace EonaCat.LogStack.Flows
sb.Append("");
}
- foreach (var property in log.Properties.ToArray())
+ foreach (var property in log.Properties)
{
if (string.IsNullOrEmpty(property.Key))
{
@@ -1025,7 +1392,7 @@ namespace EonaCat.LogStack.Flows
object val = kv.Value(log);
AppendCsvInner(val != null ? val.ToString() : "null", sb);
}
- foreach (var property in log.Properties.ToArray())
+ foreach (var property in log.Properties)
{
if (!first)
{
@@ -1047,7 +1414,7 @@ namespace EonaCat.LogStack.Flows
value = string.Empty;
}
- bool needsQuote = value.IndexOfAny(new[] { ',', '"', '\n', '\r' }) >= 0;
+ bool needsQuote = value.IndexOfAny(CsvSpecialChars) >= 0;
if (needsQuote)
{
sb.Append('"');
@@ -1253,7 +1620,20 @@ namespace EonaCat.LogStack.Flows
for (int i = maxFiles - 1; i >= 1; i--)
{
string src = Path.Combine(dir, baseName + "_" + i + _fileExtension);
+ string srcGz = src + ".gz";
string dst = Path.Combine(dir, baseName + "_" + (i + 1) + _fileExtension);
+ string dstGz = dst + ".gz";
+
+ // Move .gz variant if it exists
+ if (File.Exists(srcGz))
+ {
+ if (File.Exists(dstGz))
+ {
+ File.Delete(dstGz);
+ }
+ File.Move(srcGz, dstGz);
+ }
+
if (!File.Exists(src))
{
continue;
@@ -1261,8 +1641,6 @@ namespace EonaCat.LogStack.Flows
if (File.Exists(dst))
{
- _compressionQueue.Enqueue(dst);
- _compressionSignal.Release(1);
File.Delete(dst);
}
File.Move(src, dst);
@@ -1275,6 +1653,14 @@ namespace EonaCat.LogStack.Flows
}
File.Move(filePath, archive);
+
+ // Queue the archived file for compression
+ if (_compressionFormat != CompressionFormat.None)
+ {
+ _compressionQueue.Enqueue(archive);
+ _compressionSignal.Release(1);
+ }
+
return archive;
}
catch (Exception ex)
@@ -1343,6 +1729,9 @@ namespace EonaCat.LogStack.Flows
gz.Write(buffer, 0, read);
}
}
+
+ // Delete the original uncompressed file after successful compression
+ try { File.Delete(path); } catch { /* ignore */ }
}
private void PeriodicFlushLoop()
@@ -1361,16 +1750,12 @@ namespace EonaCat.LogStack.Flows
break;
}
- OpenFile[] snapshot;
lock (_fileLock)
{
- snapshot = new OpenFile[_openFiles.Count];
- _openFiles.Values.CopyTo(snapshot, 0);
- }
-
- foreach (OpenFile of in snapshot)
- {
- try { of.Writer.Flush(); } catch { /* ignore */ }
+ foreach (OpenFile of in _openFiles.Values)
+ {
+ try { of.Writer.Flush(); } catch { /* ignore */ }
+ }
}
}
catch (ThreadInterruptedException) { break; }
@@ -1405,6 +1790,7 @@ namespace EonaCat.LogStack.Flows
}
FileInfo[] files = dir.GetFiles("*" + _fileExtension)
+ .Concat(dir.GetFiles("*" + _fileExtension + ".gz"))
.OrderByDescending(f => f.LastWriteTimeUtc)
.ToArray();
@@ -1418,8 +1804,9 @@ namespace EonaCat.LogStack.Flows
bool tooMany = _retention.MaxRolledFiles > 0 && kept >= _retention.MaxRolledFiles;
bool tooLarge = _retention.MaxTotalArchiveBytes > 0
&& totalBytes + f.Length > _retention.MaxTotalArchiveBytes;
+ bool directoryTooLarge = totalBytes + f.Length > _maxDirectorySize;
- if (tooOld || tooMany || tooLarge)
+ if (tooOld || tooMany || tooLarge || directoryTooLarge)
{
try { f.Delete(); } catch { /* ignore */ }
}
@@ -1460,7 +1847,7 @@ namespace EonaCat.LogStack.Flows
return Path.Combine(
_directory,
- prefix + "_" + Environment.MachineName + "_" + date.ToString("yyyyMMdd") + _fileExtension);
+ string.Concat(prefix, "_", CachedMachineName, "_", date.ToString(_dateFormat), _fileExtension));
}
private void SetFileExtension(FileOutputFormat fmt)
@@ -1488,7 +1875,7 @@ namespace EonaCat.LogStack.Flows
+ log.Properties.Count * 40;
if (log.Exception != null)
{
- s += log.Exception.ToString().Length * 2;
+ s += 2048; // Avoid calling Exception.ToString() just for size estimation
}
return s;
@@ -1541,7 +1928,7 @@ namespace EonaCat.LogStack.Flows
sb.Append(_timestampMode == TimestampMode.Local
? TimeZoneInfo.Local.StandardName : "UTC");
case "host":
- return (log, sb) => sb.Append(Environment.MachineName);
+ return (log, sb) => sb.Append(CachedMachineName);
case "category":
return (log, sb) => { if (log.Category != null) { sb.Append(log.Category); } };
case "thread":
@@ -1573,7 +1960,23 @@ namespace EonaCat.LogStack.Flows
case "newline":
return (log, sb) => sb.AppendLine();
case "pid":
- return (log, sb) => sb.Append(Process.GetCurrentProcess().Id);
+ return (log, sb) => sb.Append(CachedPid);
+ case "traceid":
+ return (log, sb) =>
+ {
+ if (log.TraceId != default(ActivityTraceId))
+ {
+ sb.Append(log.TraceId.ToHexString());
+ }
+ };
+ case "spanid":
+ return (log, sb) =>
+ {
+ if (log.SpanId != default(ActivitySpanId))
+ {
+ sb.Append(log.SpanId.ToHexString());
+ }
+ };
default:
return BuildCustomOrLiteralToken(token);
}
@@ -1598,9 +2001,11 @@ namespace EonaCat.LogStack.Flows
private void AppendProperties(LogEvent log, StringBuilder sb)
{
+ var scopeProps = _scopeProperties.Value;
bool hasEnrichers = _enrichers.Count > 0;
bool hasProps = log.Properties.Count > 0;
- if (!hasEnrichers && !hasProps)
+ bool hasScope = scopeProps != null && scopeProps.Count > 0;
+ if (!hasEnrichers && !hasProps && !hasScope)
{
return;
}
@@ -1620,7 +2025,7 @@ namespace EonaCat.LogStack.Flows
sb.Append(kv.Key).Append('=').Append(val != null ? val.ToString() : "null");
}
- foreach (var property in log.Properties.ToArray())
+ foreach (var property in log.Properties)
{
if (!first)
{
@@ -1632,6 +2037,21 @@ namespace EonaCat.LogStack.Flows
.Append(property.Value != null ? property.Value.ToString() : "null");
}
+ if (hasScope)
+ {
+ foreach (var kv in scopeProps)
+ {
+ if (!first)
+ {
+ sb.Append(", ");
+ }
+
+ first = false;
+ sb.Append(kv.Key).Append('=')
+ .Append(kv.Value != null ? kv.Value.ToString() : "null");
+ }
+ }
+
sb.Append('}');
}
}
diff --git a/EonaCat.LogStack/LogBuilder.cs b/EonaCat.LogStack/LogBuilder.cs
index 85cbd00..d0d6d4f 100644
--- a/EonaCat.LogStack/LogBuilder.cs
+++ b/EonaCat.LogStack/LogBuilder.cs
@@ -95,6 +95,7 @@ public sealed class LogBuilder
string directory,
string filePrefix = "log",
long maxFileSize = 100 * 1024 * 1024,
+ long maxDirectorySize = 2L * 1024 * 1024 * 1024,
FileRetentionPolicy fileRetentionPolicy = null,
int flushIntervalInMilliSeconds = 2000,
bool useCategoryRouting = false,
@@ -110,6 +111,7 @@ public sealed class LogBuilder
directory,
filePrefix,
maxFileSize,
+ maxDirectorySize,
fileRetentionPolicy,
flushIntervalInMilliSeconds,
batchSize,
@@ -140,9 +142,14 @@ public sealed class LogBuilder
password,
filePrefix,
maxFileSize,
- flushIntervalInMilliSeconds,
- minimumLevel,
- _timestampMode));
+ maxDirectorySize: 2L * 1024 * 1024 * 1024,
+ retention: fileRetentionPolicy,
+ flushIntervalMs: flushIntervalInMilliSeconds,
+ batchSize: 1,
+ minimumLevel: minimumLevel,
+ useCategoryRouting: useCategoryRouting,
+ logLevelsForSeparateFiles: logLevelsForSeparateFiles,
+ tsMode: _timestampMode));
return this;
}
diff --git a/Testers/EonaCat.LogStack.Test.Web/EonaCat.LogStack.Test.Web.csproj b/Testers/EonaCat.LogStack.Test.Web/EonaCat.LogStack.Test.Web.csproj
index c65ca1a..95cedb5 100644
--- a/Testers/EonaCat.LogStack.Test.Web/EonaCat.LogStack.Test.Web.csproj
+++ b/Testers/EonaCat.LogStack.Test.Web/EonaCat.LogStack.Test.Web.csproj
@@ -7,7 +7,7 @@
-
+
all
runtime; build; native; contentfiles; analyzers; buildtransitive