diff --git a/EonaCat.LogStack/EonaCatLoggerCore/Flows/AdvancedFilterFlow.cs b/EonaCat.LogStack/EonaCatLoggerCore/Flows/AdvancedFilterFlow.cs new file mode 100644 index 0000000..1037806 --- /dev/null +++ b/EonaCat.LogStack/EonaCatLoggerCore/Flows/AdvancedFilterFlow.cs @@ -0,0 +1,178 @@ +using EonaCat.LogStack.Core; +using System; +using System.Collections.Generic; +using System.Threading; +using System.Threading.Tasks; + +namespace EonaCat.LogStack.Flows +{ + // This file is part of the EonaCat project(s) which is released under the Apache License. + // See the LICENSE file or go to https://EonaCat.com/License for full license details. + + /// + /// A flow that filters log events based on custom predicates before passing to an inner flow. + /// Useful for applying complex filtering logic beyond simple log level filtering. + /// + public sealed class AdvancedFilterFlow : FlowBase + { + private readonly IFlow _innerFlow; + private readonly List> _filters; + + public AdvancedFilterFlow( + IFlow innerFlow, + LogLevel minimumLevel = LogLevel.Trace) + : base(innerFlow?.Name + "_Filter" ?? "AdvancedFilterFlow", minimumLevel) + { + _innerFlow = innerFlow ?? throw new ArgumentNullException(nameof(innerFlow)); + _filters = new List>(); + } + + /// + /// Adds a filter predicate. All predicates must return true for the log event to pass through. + /// + public void AddFilter(Predicate filter) + { + if (filter == null) + { + throw new ArgumentNullException(nameof(filter)); + } + + lock (_filters) + { + _filters.Add(filter); + } + } + + /// + /// Adds a filter that matches messages containing specific text (case-insensitive). + /// + public void AddMessageContainsFilter(string text) + { + if (string.IsNullOrEmpty(text)) + { + throw new ArgumentNullException(nameof(text)); + } + + var lowerText = text.ToLowerInvariant(); + AddFilter(logEvent => logEvent.Message.ToString().ToLowerInvariant().Contains(lowerText)); + } + + /// + /// Adds a filter that matches specific log categories. + /// + public void AddCategoryFilter(params string[] categories) + { + if (categories == null || categories.Length == 0) + { + throw new ArgumentException("At least one category must be specified", nameof(categories)); + } + + var categorySet = new HashSet(categories, StringComparer.OrdinalIgnoreCase); + AddFilter(logEvent => categorySet.Contains(logEvent.Category)); + } + + /// + /// Adds a filter that excludes specific log categories. + /// + public void AddExcludeCategoryFilter(params string[] categories) + { + if (categories == null || categories.Length == 0) + { + throw new ArgumentException("At least one category must be specified", nameof(categories)); + } + + var categorySet = new HashSet(categories, StringComparer.OrdinalIgnoreCase); + AddFilter(logEvent => !categorySet.Contains(logEvent.Category)); + } + + /// + /// Adds a filter for exceptions. + /// + public void AddExceptionFilter(bool onlyWithExceptions = true) + { + AddFilter(logEvent => onlyWithExceptions ? logEvent.Exception != null : logEvent.Exception == null); + } + + private bool PassesAllFilters(LogEvent logEvent) + { + lock (_filters) + { + foreach (var filter in _filters) + { + if (!filter(logEvent)) + { + return false; + } + } + } + + return true; + } + + public override async Task BlastAsync(LogEvent logEvent, CancellationToken cancellationToken = default) + { + if (!IsEnabled || !IsLogLevelEnabled(logEvent)) + { + return WriteResult.LevelFiltered; + } + + if (!PassesAllFilters(logEvent)) + { + Interlocked.Increment(ref DroppedCount); + return WriteResult.Success; + } + + var result = await _innerFlow.BlastAsync(logEvent, cancellationToken).ConfigureAwait(false); + if (result == WriteResult.Success) + { + Interlocked.Increment(ref BlastedCount); + } + else + { + Interlocked.Increment(ref DroppedCount); + } + + return result; + } + + public override async Task BlastBatchAsync(ReadOnlyMemory logEvents, CancellationToken cancellationToken = default) + { + if (!IsEnabled) + { + return WriteResult.FlowDisabled; + } + + var filteredEvents = new List(); + foreach (var logEvent in logEvents.Span) + { + if (IsLogLevelEnabled(logEvent) && PassesAllFilters(logEvent)) + { + filteredEvents.Add(logEvent); + Interlocked.Increment(ref BlastedCount); + } + else + { + Interlocked.Increment(ref DroppedCount); + } + } + + if (filteredEvents.Count > 0) + { + return await _innerFlow.BlastBatchAsync(filteredEvents.ToArray(), cancellationToken).ConfigureAwait(false); + } + + return WriteResult.Success; + } + + public override async Task FlushAsync(CancellationToken cancellationToken = default) + { + await _innerFlow.FlushAsync(cancellationToken).ConfigureAwait(false); + } + + public override async ValueTask DisposeAsync() + { + await _innerFlow.DisposeAsync().ConfigureAwait(false); + await base.DisposeAsync().ConfigureAwait(false); + } + } +} diff --git a/EonaCat.LogStack/EonaCatLoggerCore/Flows/AggregationFlow.cs b/EonaCat.LogStack/EonaCatLoggerCore/Flows/AggregationFlow.cs new file mode 100644 index 0000000..9448363 --- /dev/null +++ b/EonaCat.LogStack/EonaCatLoggerCore/Flows/AggregationFlow.cs @@ -0,0 +1,189 @@ +using EonaCat.LogStack.Core; +using System; +using System.Collections.Generic; +using System.Linq; +using System.Threading; +using System.Threading.Tasks; + +namespace EonaCat.LogStack.Flows +{ + // This file is part of the EonaCat project(s) which is released under the Apache License. + // See the LICENSE file or go to https://EonaCat.com/License for full license details. + + /// + /// A flow that aggregates log events and periodically emits summary logs. + /// Groups logs by category or level and provides statistics. + /// + public sealed class AggregationFlow : FlowBase + { + private readonly IFlow _innerFlow; + private readonly TimeSpan _aggregationInterval; + private readonly Dictionary _stats; + private readonly Timer _flushTimer; + private readonly object _statsLock = new(); + private bool _disposed; + + public AggregationFlow( + IFlow innerFlow, + int aggregationIntervalMs = 60000, + LogLevel minimumLevel = LogLevel.Trace) + : base(innerFlow?.Name + "_Aggregated" ?? "AggregationFlow", minimumLevel) + { + _innerFlow = innerFlow ?? throw new ArgumentNullException(nameof(innerFlow)); + _aggregationInterval = TimeSpan.FromMilliseconds(aggregationIntervalMs > 0 ? aggregationIntervalMs : 60000); + _stats = new Dictionary(); + _flushTimer = new Timer(FlushAggregationCallback, null, _aggregationInterval, _aggregationInterval); + } + + private class AggregationStats + { + public int Count { get; set; } + public int ErrorCount { get; set; } + public int WarningCount { get; set; } + public DateTime FirstOccurrence { get; set; } + public DateTime LastOccurrence { get; set; } + public HashSet Messages { get; } = new HashSet(); + } + + public override async Task BlastAsync(LogEvent logEvent, CancellationToken cancellationToken = default) + { + if (!IsEnabled || !IsLogLevelEnabled(logEvent)) + { + return WriteResult.LevelFiltered; + } + + lock (_statsLock) + { + var key = $"{logEvent.Category}_{logEvent.Level}"; + + if (!_stats.TryGetValue(key, out var stats)) + { + stats = new AggregationStats { FirstOccurrence = DateTime.UtcNow }; + _stats[key] = stats; + } + + stats.Count++; + stats.LastOccurrence = DateTime.UtcNow; + + if (logEvent.Level >= LogLevel.Error) + { + stats.ErrorCount++; + } + else if (logEvent.Level == LogLevel.Warning) + { + stats.WarningCount++; + } + + var messageStr = logEvent.Message.ToString(); + if (!string.IsNullOrEmpty(messageStr) && stats.Messages.Count < 10) + { + stats.Messages.Add(messageStr); + } + } + + Interlocked.Increment(ref BlastedCount); + return WriteResult.Success; + } + + public override async Task BlastBatchAsync(ReadOnlyMemory logEvents, CancellationToken cancellationToken = default) + { + if (!IsEnabled) + { + return WriteResult.FlowDisabled; + } + + var eventsArray = logEvents.ToArray(); + foreach (var logEvent in eventsArray) + { + if (IsLogLevelEnabled(logEvent)) + { + await BlastAsync(logEvent, cancellationToken).ConfigureAwait(false); + } + } + + return WriteResult.Success; + } + + public override async Task FlushAsync(CancellationToken cancellationToken = default) + { + await EmitAggregationSummaryAsync(cancellationToken).ConfigureAwait(false); + } + + private async void FlushAggregationCallback(object? state) + { + if (!_disposed) + { + await EmitAggregationSummaryAsync(CancellationToken.None).ConfigureAwait(false); + } + } + + private async Task EmitAggregationSummaryAsync(CancellationToken cancellationToken) + { + Dictionary currentStats; + lock (_statsLock) + { + if (_stats.Count == 0) + { + return; + } + + currentStats = new Dictionary(_stats); + _stats.Clear(); + } + + var summaryMessages = new List(); + summaryMessages.Add($"=== Log Aggregation Summary (Period: {_aggregationInterval.TotalSeconds:F0}s) ==="); + + foreach (var kvp in currentStats.OrderBy(x => x.Key)) + { + var key = kvp.Key; + var stats = kvp.Value; + var duration = (stats.LastOccurrence - stats.FirstOccurrence).TotalSeconds; + + summaryMessages.Add( + $"Category-Level: {key} | Count: {stats.Count} | " + + $"Errors: {stats.ErrorCount} | Warnings: {stats.WarningCount} | " + + $"Duration: {duration:F2}s"); + + if (stats.Messages.Count > 0) + { + summaryMessages.Add($" Sample messages: {string.Join("; ", stats.Messages.Take(3))}"); + } + } + + var summaryMessage = string.Join(Environment.NewLine, summaryMessages); + + var summaryEvent = new LogEvent + { + Message = summaryMessage.AsMemory(), + Category = "AggregationFlow", + Level = LogLevel.Information, + Timestamp = LogEvent.CreateTimestamp(DateTime.UtcNow), + Exception = null, + Properties = new Dictionary + { + { "AggregationType", "Summary" }, + { "StatsCount", currentStats.Count } + } + }; + + try + { + await _innerFlow.BlastAsync(summaryEvent, cancellationToken).ConfigureAwait(false); + } + catch (Exception ex) + { + System.Diagnostics.Debug.WriteLine($"AggregationFlow summary emit error: {ex.Message}"); + } + } + + public override async ValueTask DisposeAsync() + { + _disposed = true; + _flushTimer?.Dispose(); + await EmitAggregationSummaryAsync(default).ConfigureAwait(false); + await _innerFlow.DisposeAsync().ConfigureAwait(false); + await base.DisposeAsync().ConfigureAwait(false); + } + } +} diff --git a/EonaCat.LogStack/EonaCatLoggerCore/Flows/BufferedFlow.cs b/EonaCat.LogStack/EonaCatLoggerCore/Flows/BufferedFlow.cs new file mode 100644 index 0000000..e2f9a58 --- /dev/null +++ b/EonaCat.LogStack/EonaCatLoggerCore/Flows/BufferedFlow.cs @@ -0,0 +1,158 @@ +using EonaCat.LogStack.Core; +using System; +using System.Collections.Generic; +using System.Threading; +using System.Threading.Tasks; + +namespace EonaCat.LogStack.Flows +{ + // This file is part of the EonaCat project(s) which is released under the Apache License. + // See the LICENSE file or go to https://EonaCat.com/License for full license details. + + /// + /// A flow that buffers log events and flushes them on a schedule or when a size threshold is reached. + /// Useful for improving performance by batching writes to an underlying flow. + /// + public sealed class BufferedFlow : FlowBase + { + private readonly IFlow _innerFlow; + private readonly int _bufferSize; + private readonly TimeSpan _flushInterval; + private readonly List _buffer; + private readonly SemaphoreSlim _semaphore; + private readonly Timer _flushTimer; + private readonly object _bufferLock = new(); + private bool _disposed; + + public BufferedFlow( + IFlow innerFlow, + int bufferSize = 1000, + int flushIntervalMs = 5000, + LogLevel minimumLevel = LogLevel.Trace) + : base(innerFlow?.Name + "_Buffered" ?? "BufferedFlow", minimumLevel) + { + _innerFlow = innerFlow ?? throw new ArgumentNullException(nameof(innerFlow)); + _bufferSize = bufferSize > 0 ? bufferSize : throw new ArgumentException("Buffer size must be greater than 0", nameof(bufferSize)); + _flushInterval = TimeSpan.FromMilliseconds(flushIntervalMs > 0 ? flushIntervalMs : 5000); + _buffer = new List(_bufferSize); + _semaphore = new SemaphoreSlim(1, 1); + _flushTimer = new Timer(FlushTimerCallback, null, _flushInterval, _flushInterval); + } + + public override async Task BlastAsync(LogEvent logEvent, CancellationToken cancellationToken = default) + { + if (!IsEnabled || !IsLogLevelEnabled(logEvent)) + { + return WriteResult.LevelFiltered; + } + + await _semaphore.WaitAsync(cancellationToken).ConfigureAwait(false); + try + { + lock (_bufferLock) + { + _buffer.Add(logEvent); + Interlocked.Increment(ref BlastedCount); + + if (_buffer.Count >= _bufferSize) + { + // Return a fire-and-forget task to flush + _ = FlushInternalAsync(cancellationToken); + } + } + + return WriteResult.Success; + } + finally + { + _semaphore.Release(); + } + } + + public override async Task BlastBatchAsync(ReadOnlyMemory logEvents, CancellationToken cancellationToken = default) + { + if (!IsEnabled) + { + return WriteResult.FlowDisabled; + } + + await _semaphore.WaitAsync(cancellationToken).ConfigureAwait(false); + try + { + lock (_bufferLock) + { + foreach (var logEvent in logEvents.Span) + { + if (IsLogLevelEnabled(logEvent)) + { + _buffer.Add(logEvent); + Interlocked.Increment(ref BlastedCount); + } + } + + if (_buffer.Count >= _bufferSize) + { + _ = FlushInternalAsync(cancellationToken); + } + } + + return WriteResult.Success; + } + finally + { + _semaphore.Release(); + } + } + + public override async Task FlushAsync(CancellationToken cancellationToken = default) + { + await FlushInternalAsync(cancellationToken).ConfigureAwait(false); + } + + private async Task FlushInternalAsync(CancellationToken cancellationToken) + { + List eventsToFlush; + lock (_bufferLock) + { + if (_buffer.Count == 0) + { + return; + } + + eventsToFlush = new List(_buffer); + _buffer.Clear(); + } + + if (eventsToFlush.Count > 0) + { + try + { + await _innerFlow.BlastBatchAsync(eventsToFlush.ToArray(), cancellationToken).ConfigureAwait(false); + } + catch (Exception ex) + { + Interlocked.Add(ref DroppedCount, eventsToFlush.Count); + System.Diagnostics.Debug.WriteLine($"BufferedFlow flush error: {ex.Message}"); + } + } + } + + private void FlushTimerCallback(object? state) + { + if (!_disposed) + { + _ = FlushInternalAsync(CancellationToken.None); + } + } + + public override async ValueTask DisposeAsync() + { + _disposed = true; + _flushTimer?.Dispose(); + await FlushAsync(default).ConfigureAwait(false); + _semaphore?.Dispose(); + await _innerFlow.DisposeAsync().ConfigureAwait(false); + await base.DisposeAsync().ConfigureAwait(false); + } + } +} diff --git a/EonaCat.LogStack/EonaCatLoggerCore/Flows/DuplicateDetectionFlow.cs b/EonaCat.LogStack/EonaCatLoggerCore/Flows/DuplicateDetectionFlow.cs new file mode 100644 index 0000000..895aa1c --- /dev/null +++ b/EonaCat.LogStack/EonaCatLoggerCore/Flows/DuplicateDetectionFlow.cs @@ -0,0 +1,182 @@ +using EonaCat.LogStack.Core; +using System; +using System.Collections.Generic; +using System.Threading; +using System.Threading.Tasks; + +namespace EonaCat.LogStack.Flows +{ + // This file is part of the EonaCat project(s) which is released under the Apache License. + // See the LICENSE file or go to https://EonaCat.com/License for full license details. + + /// + /// A flow that detects and suppresses duplicate log events. + /// Uses content-based hashing to identify duplicates within a rolling time window. + /// + public sealed class DuplicateDetectionFlow : FlowBase + { + private readonly IFlow _innerFlow; + private readonly int _windowSizeMs; + private readonly Dictionary _seenHashes; + private readonly Timer _pruneTimer; + private readonly object _hashLock = new(); + private bool _disposed; + + private class DuplicateEntry + { + public DateTime FirstSeen { get; set; } + public int Count { get; set; } + } + + public DuplicateDetectionFlow( + IFlow innerFlow, + int windowSizeMs = 300000, + int pruneIntervalMs = 60000, + LogLevel minimumLevel = LogLevel.Trace) + : base(innerFlow?.Name + "_DuplicateDetection" ?? "DuplicateDetectionFlow", minimumLevel) + { + _innerFlow = innerFlow ?? throw new ArgumentNullException(nameof(innerFlow)); + _windowSizeMs = windowSizeMs > 0 ? windowSizeMs : 300000; + _seenHashes = new Dictionary(); + _pruneTimer = new Timer(PruneOldEntriesCallback, null, pruneIntervalMs > 0 ? pruneIntervalMs : 60000, pruneIntervalMs > 0 ? pruneIntervalMs : 60000); + } + + private int ComputeHash(LogEvent logEvent) + { + unchecked + { + int hash = 17; + hash = hash * 31 + logEvent.Message.ToString().GetHashCode(); + hash = hash * 31 + (logEvent.Category?.GetHashCode() ?? 0); + hash = hash * 31 + logEvent.Level.GetHashCode(); + hash = hash * 31 + (logEvent.Exception?.GetType().Name.GetHashCode() ?? 0); + return hash; + } + } + + private bool IsDuplicate(LogEvent logEvent, out bool isNewException) + { + isNewException = false; + var hash = ComputeHash(logEvent); + + lock (_hashLock) + { + var now = DateTime.UtcNow; + + if (_seenHashes.TryGetValue(hash, out var entry)) + { + var age = (now - entry.FirstSeen).TotalMilliseconds; + if (age < _windowSizeMs) + { + entry.Count++; + return true; + } + else + { + _seenHashes[hash] = new DuplicateEntry { FirstSeen = now, Count = 1 }; + return false; + } + } + else + { + _seenHashes[hash] = new DuplicateEntry { FirstSeen = now, Count = 1 }; + return false; + } + } + } + + private void PruneOldEntriesCallback(object? state) + { + if (!_disposed) + { + lock (_hashLock) + { + var now = DateTime.UtcNow; + var keysToRemove = new List(); + + foreach (var kvp in _seenHashes) + { + var age = (now - kvp.Value.FirstSeen).TotalMilliseconds; + if (age >= _windowSizeMs) + { + keysToRemove.Add(kvp.Key); + } + } + + foreach (var key in keysToRemove) + { + _seenHashes.Remove(key); + } + } + } + } + + public override async Task BlastAsync(LogEvent logEvent, CancellationToken cancellationToken = default) + { + if (!IsEnabled || !IsLogLevelEnabled(logEvent)) + { + return WriteResult.LevelFiltered; + } + + if (IsDuplicate(logEvent, out _)) + { + Interlocked.Increment(ref DroppedCount); + return WriteResult.Success; + } + + var result = await _innerFlow.BlastAsync(logEvent, cancellationToken).ConfigureAwait(false); + if (result == WriteResult.Success) + { + Interlocked.Increment(ref BlastedCount); + } + else + { + Interlocked.Increment(ref DroppedCount); + } + + return result; + } + + public override async Task BlastBatchAsync(ReadOnlyMemory logEvents, CancellationToken cancellationToken = default) + { + if (!IsEnabled) + { + return WriteResult.FlowDisabled; + } + + var uniqueEvents = new List(); + foreach (var logEvent in logEvents.Span) + { + if (IsLogLevelEnabled(logEvent) && !IsDuplicate(logEvent, out _)) + { + uniqueEvents.Add(logEvent); + Interlocked.Increment(ref BlastedCount); + } + else + { + Interlocked.Increment(ref DroppedCount); + } + } + + if (uniqueEvents.Count > 0) + { + return await _innerFlow.BlastBatchAsync(uniqueEvents.ToArray(), cancellationToken).ConfigureAwait(false); + } + + return WriteResult.Success; + } + + public override async Task FlushAsync(CancellationToken cancellationToken = default) + { + await _innerFlow.FlushAsync(cancellationToken).ConfigureAwait(false); + } + + public override async ValueTask DisposeAsync() + { + _disposed = true; + _pruneTimer?.Dispose(); + await _innerFlow.DisposeAsync().ConfigureAwait(false); + await base.DisposeAsync().ConfigureAwait(false); + } + } +} diff --git a/EonaCat.LogStack/EonaCatLoggerCore/Flows/FileFlow.cs b/EonaCat.LogStack/EonaCatLoggerCore/Flows/FileFlow.cs index 10ac1fa..fe0e59e 100644 --- a/EonaCat.LogStack/EonaCatLoggerCore/Flows/FileFlow.cs +++ b/EonaCat.LogStack/EonaCatLoggerCore/Flows/FileFlow.cs @@ -51,6 +51,7 @@ namespace EonaCat.LogStack.Flows private readonly BlockingCollection _queue; private readonly ConcurrentQueue _compressionQueue = new ConcurrentQueue(); private readonly CancellationTokenSource _cts = new CancellationTokenSource(); + private volatile bool _isDisposing; // Track disposal state to prevent accessing disposed objects private readonly Thread _writerThread; private readonly Thread _compressionThread; private readonly Task _flushTask; @@ -217,25 +218,36 @@ namespace EonaCat.LogStack.Flows // Fall back to a per-user temp directory. try { - Directory.CreateDirectory(_directory); + // Use DirectoryPermissionHelper to ensure directory with proper permissions + if (!Helpers.DirectoryPermissionHelper.EnsureDirectoryHierarchy(_directory)) + { + throw new UnauthorizedAccessException($"Cannot write to directory: {Helpers.DirectoryPermissionHelper.GetAccessIssueDiagnosis(_directory)}"); + } } - catch + catch (Exception ex) { try { var processId = Process.GetCurrentProcess().Id; var newDirectory = Path.Combine(Path.GetTempPath(), "EonaCat.LogStack", processId.ToString()); - Directory.CreateDirectory(newDirectory); - OnDirectoryException?.Invoke(this, $"FileFlow: Could not create directory: '{_directory}', using directory '{newDirectory}' instead"); - _directory = newDirectory; + + if (Helpers.DirectoryPermissionHelper.EnsureDirectoryHierarchy(newDirectory)) + { + OnDirectoryException?.Invoke(this, $"FileFlow: Could not create directory: '{_directory}' ({ex.Message}), using directory '{newDirectory}' instead"); + _directory = newDirectory; + } + else + { + throw new InvalidOperationException($"Failed to create fallback directory: {newDirectory}"); + } } catch { var newDirectory = Path.GetTempPath(); - OnDirectoryException?.Invoke(this, $"FileFlow: Could not create directory: '{_directory}', using directory '{newDirectory}' instead"); + OnDirectoryException?.Invoke(this, $"FileFlow: Could not create any writable directory, falling back to temp: '{newDirectory}'. Original error: {ex.Message}"); _directory = newDirectory; - // Last resort: disable file output by pointing to a safe-ish temp path. + // Last resort: disable file output by pointing to temp path. // The writer thread still runs and swallows failures. } } @@ -815,13 +827,15 @@ namespace EonaCat.LogStack.Flows } IsEnabled = false; + _isDisposing = true; // Signal all threads that disposal is in progress _queue.CompleteAdding(); _cts.Cancel(); - _writerThread.Join(2000); + // Give threads more time to gracefully exit (increased from 2000ms to 5000ms) + _writerThread.Join(5000); _compressionSignal.Release(); - _compressionThread.Join(2000); + _compressionThread.Join(5000); lock (_fileLock) { @@ -842,7 +856,16 @@ namespace EonaCat.LogStack.Flows _openFiles.Clear(); } - _cts.Dispose(); + // Only dispose the CTS after all threads have been signaled and given time to exit + try + { + _cts.Dispose(); + } + catch (ObjectDisposedException) + { + // Already disposed, ignore + } + _compressionSignal.Dispose(); _queue.Dispose(); @@ -857,8 +880,18 @@ namespace EonaCat.LogStack.Flows LogEvent e; try { + // Check if CTS is disposed before using it + if (_isDisposing) + { + break; + } e = _queue.Take(_cts.Token); } + catch (ObjectDisposedException) + { + // CancellationTokenSource was disposed; gracefully exit + break; + } catch (OperationCanceledException) { break; } catch (InvalidOperationException) { break; } @@ -1013,7 +1046,16 @@ namespace EonaCat.LogStack.Flows Interlocked.Exchange(ref _lastErrorTimestamp, DateTime.UtcNow.Ticks); try { - WriteToConsoleError("[FileFlow] Write error for '" + path + "': " + ex.Message); + var diagnosis = ""; + if (ex is UnauthorizedAccessException || ex is System.IO.IOException) + { + var dir = Path.GetDirectoryName(path); + if (!string.IsNullOrEmpty(dir)) + { + diagnosis = " | " + Helpers.DirectoryPermissionHelper.GetAccessIssueDiagnosis(dir); + } + } + WriteToConsoleError("[FileFlow] Write error for '" + path + "': " + ex.Message + diagnosis); } catch { /* Do nothing */ } } @@ -1511,7 +1553,11 @@ namespace EonaCat.LogStack.Flows string dir = Path.GetDirectoryName(path); if (!string.IsNullOrEmpty(dir) && !Directory.Exists(dir)) { - Directory.CreateDirectory(dir); + // Use DirectoryPermissionHelper to ensure directory with permissions + if (!Helpers.DirectoryPermissionHelper.EnsureDirectoryHierarchy(dir)) + { + throw new UnauthorizedAccessException($"Cannot create directory: {Helpers.DirectoryPermissionHelper.GetAccessIssueDiagnosis(dir)}"); + } } fs = new FileStream( @@ -1584,7 +1630,16 @@ namespace EonaCat.LogStack.Flows _openFiles.Remove(path); } - WriteToConsoleError("[FileFlow] Failed to open '" + path + "': " + ex.Message); + var diagnosis = ""; + if (ex is UnauthorizedAccessException || ex is System.IO.IOException) + { + var dir = Path.GetDirectoryName(path); + if (!string.IsNullOrEmpty(dir)) + { + diagnosis = " Diagnosis: " + Helpers.DirectoryPermissionHelper.GetAccessIssueDiagnosis(dir); + } + } + WriteToConsoleError("[FileFlow] Failed to open '" + path + "': " + ex.Message + diagnosis); } } catch (Exception ex) @@ -1710,8 +1765,21 @@ namespace EonaCat.LogStack.Flows { try { + // Check for disposal before attempting to use the token + if (_isDisposing) + { + DrainCompressionQueue(); + return; + } + _compressionSignal.Wait(_cts.Token); } + catch (ObjectDisposedException) + { + // Handle the case where CTS is disposed while waiting + DrainCompressionQueue(); + return; + } catch (OperationCanceledException) { DrainCompressionQueue(); @@ -1802,46 +1870,85 @@ namespace EonaCat.LogStack.Flows private void PeriodicFlushLoop() { - while (!_cts.Token.IsCancellationRequested) + try { - try + while (!_isDisposing && !_cts.Token.IsCancellationRequested) { - long mem = Interlocked.Read(ref _currentMemoryBytes); - int delay = mem > (long)(_maxMemoryBytes * 0.8) - ? 250 - : (int)_flushInterval.TotalMilliseconds; - - if (_cts.Token.WaitHandle.WaitOne(delay)) + try { - break; - } + long mem = Interlocked.Read(ref _currentMemoryBytes); + int delay = mem > (long)(_maxMemoryBytes * 0.8) + ? 250 + : (int)_flushInterval.TotalMilliseconds; - lock (_fileLock) - { - foreach (OpenFile of in _openFiles.Values) + // Take precaution before using token + if (_isDisposing) { - try { of.Writer.Flush(); } catch { /* ignore */ } + break; + } + + if (_cts.Token.WaitHandle.WaitOne(delay)) + { + break; + } + + lock (_fileLock) + { + foreach (OpenFile of in _openFiles.Values) + { + try { of.Writer.Flush(); } catch { /* ignore */ } + } } } + catch (ObjectDisposedException) + { + // Token source was disposed; exit gracefully + break; + } + catch (ThreadInterruptedException) { break; } + catch (Exception ex) + { + WriteToConsoleError("[FileFlow] Flush error: " + ex.Message); + } } - catch (ThreadInterruptedException) { break; } - catch (Exception ex) - { - WriteToConsoleError("[FileFlow] Flush error: " + ex.Message); - } + } + catch (Exception ex) + { + WriteToConsoleError("[FileFlow] PeriodicFlushLoop error: " + ex.Message); } } private void RetentionLoop() { - while (!_cts.Token.IsCancellationRequested) + try { - if (_cts.Token.WaitHandle.WaitOne(TimeSpan.FromMinutes(15))) + while (!_isDisposing && !_cts.Token.IsCancellationRequested) { - break; - } + try + { + // Check disposal state before accessing token + if (_isDisposing) + { + break; + } - ApplyRetention(); + if (_cts.Token.WaitHandle.WaitOne(TimeSpan.FromMinutes(15))) + { + break; + } + + ApplyRetention(); + } + catch (ObjectDisposedException) + { + // Token source was disposed; exit gracefully + break; + } + } + } + catch (Exception ex) + { + WriteToConsoleError("[FileFlow] RetentionLoop error: " + ex.Message); } } diff --git a/EonaCat.LogStack/EonaCatLoggerCore/Flows/MetricsAggregatorFlow.cs b/EonaCat.LogStack/EonaCatLoggerCore/Flows/MetricsAggregatorFlow.cs new file mode 100644 index 0000000..5ec3f3b --- /dev/null +++ b/EonaCat.LogStack/EonaCatLoggerCore/Flows/MetricsAggregatorFlow.cs @@ -0,0 +1,229 @@ +using EonaCat.LogStack.Core; +using System; +using System.Collections.Generic; +using System.Diagnostics; +using System.Linq; +using System.Threading; +using System.Threading.Tasks; + +namespace EonaCat.LogStack.Flows +{ + // This file is part of the EonaCat project(s) which is released under the Apache License. + // See the LICENSE file or go to https://EonaCat.com/License for full license details. + + /// + /// A flow that collects performance metrics about log processing. + /// Tracks throughput, latency, and other performance indicators. + /// + public sealed class MetricsAggregatorFlow : FlowBase + { + private readonly object _metricsLock = new(); + private long _totalEventsProcessed; + private long _totalEventsDropped; + private long _totalLatencyMs; + private long _minLatencyMs = long.MaxValue; + private long _maxLatencyMs = 0; + private DateTime _metricsStartTime; + private readonly Queue _recentLatencies; + private readonly int _maxRecentLatencies; + + public MetricsAggregatorFlow( + int maxRecentLatencies = 100, + LogLevel minimumLevel = LogLevel.Trace) + : base("MetricsAggregatorFlow", minimumLevel) + { + _maxRecentLatencies = maxRecentLatencies > 0 ? maxRecentLatencies : 100; + _recentLatencies = new Queue(_maxRecentLatencies); + _metricsStartTime = DateTime.UtcNow; + } + + public override async Task BlastAsync(LogEvent logEvent, CancellationToken cancellationToken = default) + { + if (!IsEnabled || !IsLogLevelEnabled(logEvent)) + { + return WriteResult.LevelFiltered; + } + + var sw = Stopwatch.StartNew(); + try + { + sw.Stop(); + RecordMetric(success: true, latencyMs: sw.ElapsedMilliseconds); + Interlocked.Increment(ref BlastedCount); + return WriteResult.Success; + } + catch (Exception) + { + sw.Stop(); + RecordMetric(success: false, latencyMs: sw.ElapsedMilliseconds); + Interlocked.Increment(ref DroppedCount); + return WriteResult.Failed; + } + } + + public override async Task BlastBatchAsync(ReadOnlyMemory logEvents, CancellationToken cancellationToken = default) + { + if (!IsEnabled) + { + return WriteResult.FlowDisabled; + } + + var sw = Stopwatch.StartNew(); + try + { + int processedCount = 0; + foreach (var logEvent in logEvents.Span) + { + if (IsLogLevelEnabled(logEvent)) + { + processedCount++; + } + } + + sw.Stop(); + RecordMetric(success: true, latencyMs: sw.ElapsedMilliseconds); + Interlocked.Add(ref BlastedCount, processedCount); + Interlocked.Add(ref DroppedCount, logEvents.Length - processedCount); + return WriteResult.Success; + } + catch (Exception) + { + sw.Stop(); + RecordMetric(success: false, latencyMs: sw.ElapsedMilliseconds); + Interlocked.Add(ref DroppedCount, logEvents.Length); + return WriteResult.Failed; + } + } + + private void RecordMetric(bool success, long latencyMs) + { + lock (_metricsLock) + { + if (success) + { + _totalEventsProcessed++; + _totalLatencyMs += latencyMs; + _minLatencyMs = Math.Min(_minLatencyMs, latencyMs); + _maxLatencyMs = Math.Max(_maxLatencyMs, latencyMs); + + if (_recentLatencies.Count >= _maxRecentLatencies) + { + _recentLatencies.Dequeue(); + } + _recentLatencies.Enqueue(latencyMs); + } + else + { + _totalEventsDropped++; + } + } + } + + /// + /// Gets current performance metrics snapshot. + /// + public MetricsSnapshot GetMetricsSnapshot() + { + lock (_metricsLock) + { + var totalProcessed = _totalEventsProcessed; + var totalDropped = _totalEventsDropped; + var totalLatency = _totalLatencyMs; + var minLatency = _minLatencyMs == long.MaxValue ? 0 : _minLatencyMs; + var maxLatency = _maxLatencyMs; + var recentLatencies = _recentLatencies.ToArray(); + + var avgLatency = totalProcessed > 0 ? totalLatency / totalProcessed : 0; + var medianLatency = recentLatencies.Length > 0 + ? recentLatencies[recentLatencies.Length / 2] + : 0; + + var elapsed = DateTime.UtcNow - _metricsStartTime; + var eventsPerSecond = elapsed.TotalSeconds > 0 + ? totalProcessed / elapsed.TotalSeconds + : 0; + + return new MetricsSnapshot + { + TotalProcessed = totalProcessed, + TotalDropped = totalDropped, + AverageLatencyMs = avgLatency, + MinLatencyMs = minLatency, + MaxLatencyMs = maxLatency, + MedianLatencyMs = medianLatency, + EventsPerSecond = eventsPerSecond, + ElapsedTime = elapsed, + CollectedAt = DateTime.UtcNow + }; + } + } + + /// + /// Resets metrics counters. + /// + public void ResetMetrics() + { + lock (_metricsLock) + { + _totalEventsProcessed = 0; + _totalEventsDropped = 0; + _totalLatencyMs = 0; + _minLatencyMs = long.MaxValue; + _maxLatencyMs = 0; + _recentLatencies.Clear(); + _metricsStartTime = DateTime.UtcNow; + } + } + + public override async Task FlushAsync(CancellationToken cancellationToken = default) + { + // Metrics flow doesn't buffer events, nothing to flush + return; + } + + public override ValueTask DisposeAsync() + { + return base.DisposeAsync(); + } + } + + /// + /// A snapshot of performance metrics at a point in time. + /// + public sealed class MetricsSnapshot + { + /// Total number of log events processed + public long TotalProcessed { get; set; } + + /// Total number of log events dropped + public long TotalDropped { get; set; } + + /// Average latency in milliseconds + public long AverageLatencyMs { get; set; } + + /// Minimum latency in milliseconds + public long MinLatencyMs { get; set; } + + /// Maximum latency in milliseconds + public long MaxLatencyMs { get; set; } + + /// Median latency in milliseconds + public long MedianLatencyMs { get; set; } + + /// Number of events processed per second + public double EventsPerSecond { get; set; } + + /// Total elapsed time since metrics started + public TimeSpan ElapsedTime { get; set; } + + /// When this snapshot was collected + public DateTime CollectedAt { get; set; } + + public override string ToString() + { + return $"Metrics[Processed={TotalProcessed}, Dropped={TotalDropped}, " + + $"AvgLat={AverageLatencyMs}ms, MinLat={MinLatencyMs}ms, MaxLat={MaxLatencyMs}ms, " + + $"Median={MedianLatencyMs}ms, Rate={EventsPerSecond:F2} eps, Elapsed={ElapsedTime.TotalSeconds:F2}s]"; + } + } +} diff --git a/EonaCat.LogStack/EonaCatLoggerCore/Flows/PipelineFlow.cs b/EonaCat.LogStack/EonaCatLoggerCore/Flows/PipelineFlow.cs new file mode 100644 index 0000000..adc832c --- /dev/null +++ b/EonaCat.LogStack/EonaCatLoggerCore/Flows/PipelineFlow.cs @@ -0,0 +1,253 @@ +using EonaCat.LogStack.Core; +using System; +using System.Collections.Generic; +using System.Threading; +using System.Threading.Tasks; + +namespace EonaCat.LogStack.Flows +{ + // This file is part of the EonaCat project(s) which is released under the Apache License. + // See the LICENSE file or go to https://EonaCat.com/License for full license details. + + /// + /// A flow that chains multiple flows together, passing log events through them in sequence. + /// Useful for composing complex logging behavior from simple components. + /// + public sealed class PipelineFlow : FlowBase + { + private readonly List _flows; + private readonly StopOnError _stopOnError; + + public enum StopOnError + { + /// Continue to next flow even if current flow fails + Continue = 0, + + /// Stop pipeline if any flow fails + StopImmediate = 1, + + /// Stop pipeline if core flows fail, continue for non-critical flows + StopOnCritical = 2 + } + + public PipelineFlow( + LogLevel minimumLevel = LogLevel.Trace, + StopOnError stopOnError = StopOnError.Continue) + : base("PipelineFlow", minimumLevel) + { + _flows = new List(); + _stopOnError = stopOnError; + } + + /// + /// Adds a flow to the pipeline. Flows are executed in the order they are added. + /// + public PipelineFlow AddFlow(IFlow flow) + { + if (flow == null) + { + throw new ArgumentNullException(nameof(flow)); + } + + lock (_flows) + { + _flows.Add(flow); + } + + return this; + } + + /// + /// Adds multiple flows to the pipeline. + /// + public PipelineFlow AddFlows(params IFlow[] flows) + { + if (flows == null) + { + throw new ArgumentNullException(nameof(flows)); + } + + foreach (var flow in flows) + { + AddFlow(flow); + } + + return this; + } + + public override async Task BlastAsync(LogEvent logEvent, CancellationToken cancellationToken = default) + { + if (!IsEnabled || !IsLogLevelEnabled(logEvent)) + { + return WriteResult.LevelFiltered; + } + + List flowsToExecute; + lock (_flows) + { + flowsToExecute = new List(_flows); + } + + if (flowsToExecute.Count == 0) + { + Interlocked.Increment(ref DroppedCount); + return WriteResult.Success; + } + + var overallResult = WriteResult.Success; + + foreach (var flow in flowsToExecute) + { + try + { + if (flow.IsEnabled && logEvent.Level >= flow.MinimumLevel) + { + var result = await flow.BlastAsync(logEvent, cancellationToken).ConfigureAwait(false); + + if (result != WriteResult.Success) + { + overallResult = result; + + if (_stopOnError == StopOnError.StopImmediate) + { + break; + } + } + } + } + catch (Exception ex) + { + System.Diagnostics.Debug.WriteLine($"PipelineFlow error in {flow.Name}: {ex.Message}"); + overallResult = WriteResult.Failed; + + if (_stopOnError == StopOnError.StopImmediate) + { + break; + } + } + } + + if (overallResult == WriteResult.Success) + { + Interlocked.Increment(ref BlastedCount); + } + else + { + Interlocked.Increment(ref DroppedCount); + } + + return overallResult; + } + + public override async Task BlastBatchAsync(ReadOnlyMemory logEvents, CancellationToken cancellationToken = default) + { + if (!IsEnabled) + { + return WriteResult.FlowDisabled; + } + + List flowsToExecute; + lock (_flows) + { + flowsToExecute = new List(_flows); + } + + if (flowsToExecute.Count == 0) + { + Interlocked.Add(ref DroppedCount, logEvents.Length); + return WriteResult.Success; + } + + var overallResult = WriteResult.Success; + var logsProcessed = 0; + + foreach (var flow in flowsToExecute) + { + try + { + if (flow.IsEnabled) + { + var result = await flow.BlastBatchAsync(logEvents, cancellationToken).ConfigureAwait(false); + + if (result != WriteResult.Success) + { + overallResult = result; + + if (_stopOnError == StopOnError.StopImmediate) + { + break; + } + } + + logsProcessed += logEvents.Length; + } + } + catch (Exception ex) + { + System.Diagnostics.Debug.WriteLine($"PipelineFlow batch error in {flow.Name}: {ex.Message}"); + overallResult = WriteResult.Failed; + + if (_stopOnError == StopOnError.StopImmediate) + { + break; + } + } + } + + if (logsProcessed > 0) + { + Interlocked.Add(ref BlastedCount, logsProcessed); + } + else + { + Interlocked.Add(ref DroppedCount, logEvents.Length); + } + + return overallResult; + } + + public override async Task FlushAsync(CancellationToken cancellationToken = default) + { + List flowsToFlush; + lock (_flows) + { + flowsToFlush = new List(_flows); + } + + foreach (var flow in flowsToFlush) + { + try + { + await flow.FlushAsync(cancellationToken).ConfigureAwait(false); + } + catch (Exception ex) + { + System.Diagnostics.Debug.WriteLine($"PipelineFlow flush error in {flow.Name}: {ex.Message}"); + } + } + } + + public override async ValueTask DisposeAsync() + { + List flowsToDispose; + lock (_flows) + { + flowsToDispose = new List(_flows); + } + + foreach (var flow in flowsToDispose) + { + try + { + await flow.DisposeAsync().ConfigureAwait(false); + } + catch (Exception ex) + { + System.Diagnostics.Debug.WriteLine($"PipelineFlow dispose error in {flow.Name}: {ex.Message}"); + } + } + + await base.DisposeAsync().ConfigureAwait(false); + } + } +} diff --git a/EonaCat.LogStack/EonaCatLoggerCore/Flows/RotatingFileFlow.cs b/EonaCat.LogStack/EonaCatLoggerCore/Flows/RotatingFileFlow.cs new file mode 100644 index 0000000..f89d750 --- /dev/null +++ b/EonaCat.LogStack/EonaCatLoggerCore/Flows/RotatingFileFlow.cs @@ -0,0 +1,247 @@ +using EonaCat.LogStack.Core; +using System; +using System.IO; +using System.Text; +using System.Threading; +using System.Threading.Tasks; + +namespace EonaCat.LogStack.Flows +{ + // This file is part of the EonaCat project(s) which is released under the Apache License. + // See the LICENSE file or go to https://EonaCat.com/License for full license details. + + /// + /// A flow that writes logs to files with automatic date-based rotation. + /// Creates new log files daily, hourly, or at custom intervals without external dependencies. + /// + public sealed class RotatingFileFlow : FlowBase + { + public enum RotationFrequency + { + Hourly = 0, + Daily = 1, + Monthly = 2 + } + + private readonly string _baseDirectory; + private readonly string _fileNamePattern; + private readonly RotationFrequency _frequency; + private readonly Encoding _encoding; + private readonly object _fileLock = new(); + private StreamWriter? _currentWriter; + private string _currentFileName = string.Empty; + private DateTime _lastRotationCheck = DateTime.MinValue; + + public RotatingFileFlow( + string baseDirectory, + string fileNamePattern = "logs-{date}.txt", + RotationFrequency frequency = RotationFrequency.Daily, + Encoding? encoding = null, + LogLevel minimumLevel = LogLevel.Trace) + : base("RotatingFileFlow", minimumLevel) + { + _baseDirectory = baseDirectory ?? throw new ArgumentNullException(nameof(baseDirectory)); + _fileNamePattern = fileNamePattern ?? throw new ArgumentNullException(nameof(fileNamePattern)); + _frequency = frequency; + _encoding = encoding ?? Encoding.UTF8; + + try + { + if (!Directory.Exists(_baseDirectory)) + { + Directory.CreateDirectory(_baseDirectory); + } + } + catch (Exception ex) + { + System.Diagnostics.Debug.WriteLine($"RotatingFileFlow directory creation error: {ex.Message}"); + } + } + + private string GetCurrentFileName() + { + var now = DateTime.Now; + string datePart = _frequency switch + { + RotationFrequency.Hourly => now.ToString("yyyy-MM-dd_HH"), + RotationFrequency.Monthly => now.ToString("yyyy-MM"), + RotationFrequency.Daily => now.ToString("yyyy-MM-dd"), + _ => now.ToString("yyyy-MM-dd") + }; + + var fileName = _fileNamePattern.Replace("{date}", datePart); + return Path.Combine(_baseDirectory, fileName); + } + + private bool ShouldRotate() + { + var now = DateTime.Now; + var timeSinceLastCheck = now - _lastRotationCheck; + + return _frequency switch + { + RotationFrequency.Hourly => timeSinceLastCheck.TotalHours >= 1, + RotationFrequency.Monthly => timeSinceLastCheck.TotalDays >= 30, + RotationFrequency.Daily => timeSinceLastCheck.TotalDays >= 1, + _ => false + }; + } + + private async Task EnsureWriterAsync() + { + var targetFileName = GetCurrentFileName(); + + if (_currentWriter != null && _currentFileName == targetFileName && !ShouldRotate()) + { + return; + } + + _lastRotationCheck = DateTime.Now; + + try + { + _currentWriter?.Dispose(); + + _currentFileName = targetFileName; + var fileStream = new FileStream( + _currentFileName, + FileMode.Append, + FileAccess.Write, + FileShare.Read, + 4096, + FileOptions.SequentialScan); + + _currentWriter = new StreamWriter(fileStream, _encoding, 4096, false); + } + catch (Exception ex) + { + System.Diagnostics.Debug.WriteLine($"RotatingFileFlow writer creation error: {ex.Message}"); + throw; + } + } + + public override async Task BlastAsync(LogEvent logEvent, CancellationToken cancellationToken = default) + { + if (!IsEnabled || !IsLogLevelEnabled(logEvent)) + { + return WriteResult.LevelFiltered; + } + + lock (_fileLock) + { + try + { + EnsureWriterAsync().Wait(cancellationToken); + + if (_currentWriter == null) + { + Interlocked.Increment(ref DroppedCount); + return WriteResult.Failed; + } + + var formattedMessage = FormatLogEvent(logEvent); + _currentWriter.WriteLine(formattedMessage); + _currentWriter.Flush(); + + Interlocked.Increment(ref BlastedCount); + return WriteResult.Success; + } + catch (Exception ex) + { + Interlocked.Increment(ref DroppedCount); + System.Diagnostics.Debug.WriteLine($"RotatingFileFlow write error: {ex.Message}"); + return WriteResult.Failed; + } + } + } + + public override async Task BlastBatchAsync(ReadOnlyMemory logEvents, CancellationToken cancellationToken = default) + { + if (!IsEnabled) + { + return WriteResult.FlowDisabled; + } + + lock (_fileLock) + { + try + { + EnsureWriterAsync().Wait(cancellationToken); + + if (_currentWriter == null) + { + Interlocked.Add(ref DroppedCount, logEvents.Length); + return WriteResult.Failed; + } + + foreach (var logEvent in logEvents.Span) + { + if (IsLogLevelEnabled(logEvent)) + { + var formattedMessage = FormatLogEvent(logEvent); + _currentWriter.WriteLine(formattedMessage); + Interlocked.Increment(ref BlastedCount); + } + else + { + Interlocked.Increment(ref DroppedCount); + } + } + + _currentWriter.Flush(); + return WriteResult.Success; + } + catch (Exception ex) + { + Interlocked.Add(ref DroppedCount, logEvents.Length); + System.Diagnostics.Debug.WriteLine($"RotatingFileFlow batch write error: {ex.Message}"); + return WriteResult.Failed; + } + } + } + + private string FormatLogEvent(LogEvent logEvent) + { + var sb = new StringBuilder(); + var dateTime = LogEvent.GetDateTime(logEvent.Timestamp); + sb.Append($"[{dateTime:yyyy-MM-dd HH:mm:ss.fff}] "); + sb.Append($"[{logEvent.Level}] "); + if (!string.IsNullOrEmpty(logEvent.Category)) + { + sb.Append($"[{logEvent.Category}] "); + } + sb.Append(logEvent.Message); + + if (logEvent.Exception != null) + { + sb.AppendLine(); + sb.Append("Exception: "); + sb.Append(logEvent.Exception.GetType().Name); + sb.Append(": "); + sb.Append(logEvent.Exception.Message); + } + + return sb.ToString(); + } + + public override async Task FlushAsync(CancellationToken cancellationToken = default) + { + lock (_fileLock) + { + _currentWriter?.Flush(); + } + } + + public override async ValueTask DisposeAsync() + { + lock (_fileLock) + { + _currentWriter?.Flush(); + _currentWriter?.Dispose(); + _currentWriter = null; + } + + await base.DisposeAsync().ConfigureAwait(false); + } + } +} diff --git a/EonaCat.LogStack/EonaCatLoggerCore/Flows/TransformFlow.cs b/EonaCat.LogStack/EonaCatLoggerCore/Flows/TransformFlow.cs new file mode 100644 index 0000000..50ccbc1 --- /dev/null +++ b/EonaCat.LogStack/EonaCatLoggerCore/Flows/TransformFlow.cs @@ -0,0 +1,107 @@ +using EonaCat.LogStack.Core; +using System; +using System.Collections.Generic; +using System.Threading; +using System.Threading.Tasks; + +namespace EonaCat.LogStack.Flows +{ + // This file is part of the EonaCat project(s) which is released under the Apache License. + // See the LICENSE file or go to https://EonaCat.com/License for full license details. + + /// + /// A flow that transforms log events before passing them to an inner flow. + /// Allows modification of messages, properties, and other attributes. + /// + public sealed class TransformFlow : FlowBase + { + private readonly IFlow _innerFlow; + private readonly Func _transformer; + + public TransformFlow( + IFlow innerFlow, + Func transformer, + LogLevel minimumLevel = LogLevel.Trace) + : base(innerFlow?.Name + "_Transform" ?? "TransformFlow", minimumLevel) + { + _innerFlow = innerFlow ?? throw new ArgumentNullException(nameof(innerFlow)); + _transformer = transformer ?? throw new ArgumentNullException(nameof(transformer)); + } + + public override async Task BlastAsync(LogEvent logEvent, CancellationToken cancellationToken = default) + { + if (!IsEnabled || !IsLogLevelEnabled(logEvent)) + { + return WriteResult.LevelFiltered; + } + + try + { + var transformedEvent = _transformer(logEvent); + var result = await _innerFlow.BlastAsync(transformedEvent, cancellationToken).ConfigureAwait(false); + if (result == WriteResult.Success) + { + Interlocked.Increment(ref BlastedCount); + } + else + { + Interlocked.Increment(ref DroppedCount); + } + + return result; + } + catch (Exception ex) + { + Interlocked.Increment(ref DroppedCount); + System.Diagnostics.Debug.WriteLine($"TransformFlow error: {ex.Message}"); + return WriteResult.Failed; + } + } + + public override async Task BlastBatchAsync(ReadOnlyMemory logEvents, CancellationToken cancellationToken = default) + { + if (!IsEnabled) + { + return WriteResult.FlowDisabled; + } + + var eventsArray = logEvents.ToArray(); + var transformedEvents = new List(); + foreach (var logEvent in eventsArray) + { + if (IsLogLevelEnabled(logEvent)) + { + try + { + var transformed = _transformer(logEvent); + transformedEvents.Add(transformed); + Interlocked.Increment(ref BlastedCount); + } + catch (Exception ex) + { + Interlocked.Increment(ref DroppedCount); + System.Diagnostics.Debug.WriteLine($"TransformFlow batch error: {ex.Message}"); + } + } + } + + if (transformedEvents.Count > 0) + { + return await _innerFlow.BlastBatchAsync(transformedEvents.ToArray(), cancellationToken).ConfigureAwait(false); + } + + return WriteResult.Success; + } + + public override async Task FlushAsync(CancellationToken cancellationToken = default) + { + await _innerFlow.FlushAsync(cancellationToken).ConfigureAwait(false); + } + + public override async ValueTask DisposeAsync() + { + await _innerFlow.DisposeAsync().ConfigureAwait(false); + await base.DisposeAsync().ConfigureAwait(false); + } + } +} diff --git a/EonaCat.LogStack/Helpers/DirectoryPermissionHelper.cs b/EonaCat.LogStack/Helpers/DirectoryPermissionHelper.cs new file mode 100644 index 0000000..0bccec8 --- /dev/null +++ b/EonaCat.LogStack/Helpers/DirectoryPermissionHelper.cs @@ -0,0 +1,221 @@ +using System; +using System.IO; +using System.Runtime.InteropServices; + +namespace EonaCat.LogStack.Helpers +{ + /// + /// Utility class for managing directory permissions and ensuring write access. + /// Handles both Windows and Unix-like systems gracefully. + /// + public static class DirectoryPermissionHelper + { + /// + /// Checks if the current process has write access to the specified directory. + /// Returns true if writable, false otherwise. Never throws. + /// + public static bool CanWrite(string dirPath) + { + if (string.IsNullOrEmpty(dirPath)) + { + return false; + } + + try + { + if (!Directory.Exists(dirPath)) + { + // Check if we can create it + var parentPath = Path.GetDirectoryName(dirPath); + if (string.IsNullOrEmpty(parentPath) || parentPath == dirPath) + { + return false; + } + return CanWrite(parentPath); + } + + // Try to create a temporary file to verify write access + var testFile = Path.Combine(dirPath, ".permission_test_" + Guid.NewGuid().ToString(".")); + try + { + using (var fs = File.Create(testFile, 1, FileOptions.DeleteOnClose)) + { + fs.WriteByte(0); + } + return true; + } + catch + { + return false; + } + } + catch + { + return false; + } + } + + /// + /// Attempts to create a directory and fix permissions if needed. + /// Never throws - logs errors and returns false if it cannot fix permissions. + /// + public static bool EnsureDirectory(string dirPath) + { + if (string.IsNullOrEmpty(dirPath)) + { + return false; + } + + try + { + // Create directory if it doesn't exist + if (!Directory.Exists(dirPath)) + { + Directory.CreateDirectory(dirPath); + } + + // Verify we have write access + if (CanWrite(dirPath)) + { + return true; + } + + // Try to fix permissions (Windows-specific via attribute clearing) + if (RuntimeInformation.IsOSPlatform(OSPlatform.Windows)) + { + try + { + TryFixWindowsPermissions(dirPath); + return CanWrite(dirPath); + } + catch + { + // Permission fixing failed, but that's okay - we tried + } + } + + // For Unix-like systems, we can't easily fix permissions without external tools + // Just return whether we can write + return CanWrite(dirPath); + } + catch + { + return false; + } + } + + /// + /// Attempts to ensure all parent directories have write access. + /// Creates missing directories. + /// + public static bool EnsureDirectoryHierarchy(string dirPath) + { + if (string.IsNullOrEmpty(dirPath)) + { + return false; + } + + try + { + var current = Path.GetDirectoryName(dirPath); + var stack = new System.Collections.Generic.Stack(); + + // Build directory hierarchy + while (!string.IsNullOrEmpty(current) && current != Path.GetPathRoot(current)) + { + if (Directory.Exists(current)) + { + break; + } + stack.Push(current); + current = Path.GetDirectoryName(current); + } + + // Create missing directories from root to target + while (stack.Count > 0) + { + var dir = stack.Pop(); + if (!Directory.Exists(dir)) + { + Directory.CreateDirectory(dir); + } + } + + // Ensure final target directory + return EnsureDirectory(dirPath); + } + catch + { + return false; + } + } + + /// + /// Attempts to grant write permissions on Windows by clearing the read-only attribute. + /// This is a best-effort operation and may fail on restricted systems. + /// + private static void TryFixWindowsPermissions(string dirPath) + { + try + { + var dirInfo = new DirectoryInfo(dirPath); + + // Clear read-only attribute if set + if ((dirInfo.Attributes & FileAttributes.ReadOnly) != 0) + { + dirInfo.Attributes &= ~FileAttributes.ReadOnly; + } + + // Note: Full ACL manipulation would require System.Security.AccessControl + // which is a separate NuGet package. We're using the basic attribute approach + // which covers most common permission issues. + } + catch + { + // Silently fail - we tried our best + throw; + } + } + + /// + /// Gets a safe description of why directory access failed. + /// + public static string GetAccessIssueDiagnosis(string dirPath) + { + try + { + if (string.IsNullOrEmpty(dirPath)) + { + return "Directory path is null or empty"; + } + + if (!Directory.Exists(dirPath)) + { + var parent = Path.GetDirectoryName(dirPath); + if (string.IsNullOrEmpty(parent)) + { + return $"Cannot determine parent directory: {dirPath}"; + } + if (!Directory.Exists(parent)) + { + return $"Parent directory does not exist: {parent}"; + } + return $"Directory does not exist and parent is not writable: {dirPath}"; + } + + var dirInfo = new DirectoryInfo(dirPath); + + if ((dirInfo.Attributes & FileAttributes.ReadOnly) != 0) + { + return $"Directory is read-only: {dirPath}"; + } + + return $"Directory exists but no write permission: {dirPath}. Check file system permissions."; + } + catch (Exception ex) + { + return $"Error diagnosing directory: {ex.Message}"; + } + } + } +} diff --git a/EonaCat.LogStack/LogBuilder.cs b/EonaCat.LogStack/LogBuilder.cs index eb6ac33..a8fa3f2 100644 --- a/EonaCat.LogStack/LogBuilder.cs +++ b/EonaCat.LogStack/LogBuilder.cs @@ -12,6 +12,7 @@ using System.Linq; using System.Net.Http; using System.Net.Security; using System.Security.Cryptography.X509Certificates; +using System.Text; namespace EonaCat.LogStack.Configuration; @@ -1274,6 +1275,121 @@ public sealed class LogBuilder return this; } + /// + /// Writes log events to files with automatic date-based rotation. + /// Creates new log files daily, hourly, or at custom intervals. + /// + public LogBuilder WriteToRotatingFileFlow( + string baseDirectory, + string fileNamePattern = "logs-{date}.txt", + RotatingFileFlow.RotationFrequency frequency = RotatingFileFlow.RotationFrequency.Daily, + Encoding? encoding = null, + LogLevel minimumLevel = LogLevel.Trace) + { + _flows.Add(new RotatingFileFlow( + baseDirectory, + fileNamePattern, + frequency, + encoding, + minimumLevel)); + return this; + } + + /// + /// Chains multiple flows together, passing log events through them in sequence. + /// Useful for composing complex logging behavior from simple components. + /// + public LogBuilder WriteToPipelineFlow( + LogLevel minimumLevel = LogLevel.Trace, + PipelineFlow.StopOnError stopOnError = PipelineFlow.StopOnError.Continue) + { + _flows.Add(new PipelineFlow(minimumLevel, stopOnError)); + return this; + } + + /// + /// Adds a metrics aggregator flow that collects and reports logging metrics. + /// + public LogBuilder WriteToMetricsAggregatorFlow( + int maxRecentLatencies = 100, + LogLevel minimumLevel = LogLevel.Trace) + { + _flows.Add(new MetricsAggregatorFlow(maxRecentLatencies, minimumLevel)); + return this; + } + + /// + /// Adds duplicate detection to prevent repeated identical log messages within a time window. + /// + public LogBuilder WriteToDuplicateDetectionFlow( + IFlow innerFlow, + int windowSizeMs = 300000, + int pruneIntervalMs = 60000, + LogLevel minimumLevel = LogLevel.Trace) + { + _flows.Add(new DuplicateDetectionFlow( + innerFlow, + windowSizeMs, + pruneIntervalMs, + minimumLevel)); + return this; + } + + /// + /// Transforms log events using a custom function before forwarding to an inner flow. + /// + public LogBuilder WriteToTransformFlow( + IFlow innerFlow, + Func transformer, + LogLevel minimumLevel = LogLevel.Trace) + { + _flows.Add(new TransformFlow(innerFlow, transformer, minimumLevel)); + return this; + } + + /// + /// Aggregates log events by category and reports statistics periodically. + /// + public LogBuilder WriteToAggregationFlow( + IFlow innerFlow, + int aggregationIntervalMs = 60000, + LogLevel minimumLevel = LogLevel.Trace) + { + _flows.Add(new AggregationFlow( + innerFlow, + aggregationIntervalMs, + minimumLevel)); + return this; + } + + /// + /// Adds advanced filtering capabilities using predicates to selectively process log events. + /// + public LogBuilder WriteToAdvancedFilterFlow( + IFlow innerFlow, + LogLevel minimumLevel = LogLevel.Trace) + { + _flows.Add(new AdvancedFilterFlow(innerFlow, minimumLevel)); + return this; + } + + /// + /// Buffers log events and flushes them in batches to reduce I/O operations. + /// + public LogBuilder WriteToBufferedFlow( + IFlow innerFlow, + int bufferSize = 1000, + int flushIntervalMs = 5000, + LogLevel minimumLevel = LogLevel.Trace) + { + _flows.Add(new BufferedFlow( + innerFlow, + bufferSize, + flushIntervalMs, + minimumLevel)); + return this; + } + /// /// Adds the which captures caller member/file/line. /// diff --git a/EonaCat.LogStack/Server.cs b/EonaCat.LogStack/Server.cs index 43e77bc..5041c9d 100644 --- a/EonaCat.LogStack/Server.cs +++ b/EonaCat.LogStack/Server.cs @@ -147,6 +147,12 @@ namespace EonaCat.LogStack.Server _isRunning = true; Metrics.StartedAt = DateTime.UtcNow; + // Initialize log directory with permission checking + if (!InitializeLogDirectory()) + { + Console.WriteLine("[EonaCat Server] WARNING: Could not initialize log directory. Logs will be dropped if write fails."); + } + var bind = ipAddress ?? _options.BindAddress ?? IPAddress.Any; var tasks = new List(); @@ -184,6 +190,40 @@ namespace EonaCat.LogStack.Server await Task.WhenAll(tasks).ConfigureAwait(false); } + /// + /// Initializes the log directory during startup, ensuring write permissions. + /// Returns true if successful, false if there were issues (but doesn't prevent startup). + /// + private bool InitializeLogDirectory() + { + try + { + var logDir = _options.LogsRootDirectory; + + if (!Helpers.DirectoryPermissionHelper.EnsureDirectoryHierarchy(logDir)) + { + Console.WriteLine($"[EonaCat Server] WARNING: Could not fully initialize log directory: {logDir}"); + Console.WriteLine($"[EonaCat Server] Diagnosis: {Helpers.DirectoryPermissionHelper.GetAccessIssueDiagnosis(logDir)}"); + return false; + } + + if (!Helpers.DirectoryPermissionHelper.CanWrite(logDir)) + { + Console.WriteLine($"[EonaCat Server] WARNING: Log directory exists but is not writable: {logDir}"); + Console.WriteLine($"[EonaCat Server] Please check file system permissions."); + return false; + } + + Console.WriteLine($"[EonaCat Server] Log directory ready: {Path.GetFullPath(logDir)}"); + return true; + } + catch (Exception ex) + { + Console.WriteLine($"[EonaCat Server] Error initializing log directory: {ex.GetType().Name}: {ex.Message}"); + return false; + } + } + /// Gracefully stop all transports. public void Stop() { @@ -515,26 +555,101 @@ namespace EonaCat.LogStack.Server protected virtual async Task ProcessLogAsync(string logData) { - var root = _options.LogsRootDirectory; - Directory.CreateDirectory(root); + const int maxRetries = 3; + const int retryDelayMs = 100; - var daily = Path.Combine(root, DateTime.Now.ToString("yyyyMMdd")); - Directory.CreateDirectory(daily); - - var basePath = Path.Combine(daily, "EonaCatLogs"); - var filePath = basePath + ".log"; - - int idx = 1; - while (File.Exists(filePath) && new FileInfo(filePath).Length > _options.MaxLogFileSize) + for (int attempt = 0; attempt < maxRetries; attempt++) { - filePath = $"{basePath}_{idx++}.log"; + try + { + var root = _options.LogsRootDirectory; + + // Ensure directory exists with permission handling + if (!EnsureLogDirectory(root)) + { + if (attempt < maxRetries - 1) + { + await Task.Delay(retryDelayMs).ConfigureAwait(false); + continue; + } + // Last attempt failed - log that we couldn't write + Console.WriteLine($"[EonaCat Server] WARNING: Cannot write to log directory. Reason: {Helpers.DirectoryPermissionHelper.GetAccessIssueDiagnosis(root)}"); + Interlocked.Increment(ref Metrics.TotalDropped); + return; + } + + var daily = Path.Combine(root, DateTime.Now.ToString("yyyyMMdd")); + + if (!EnsureLogDirectory(daily)) + { + if (attempt < maxRetries - 1) + { + await Task.Delay(retryDelayMs).ConfigureAwait(false); + continue; + } + Console.WriteLine($"[EonaCat Server] WARNING: Cannot create daily log directory: {daily}"); + Interlocked.Increment(ref Metrics.TotalDropped); + return; + } + + var basePath = Path.Combine(daily, "EonaCatLogs"); + var filePath = basePath + ".log"; + + int idx = 1; + while (File.Exists(filePath) && new FileInfo(filePath).Length > _options.MaxLogFileSize) + { + filePath = $"{basePath}_{idx++}.log"; + } + + File.AppendAllText(filePath, logData + Environment.NewLine); + Interlocked.Increment(ref Metrics.TotalWritten); + LogWritten?.Invoke(logData); + + // Cleanup happens asynchronously to avoid blocking + _ = Task.Run(() => CleanUpOldLogs()); + return; + } + catch (UnauthorizedAccessException ex) when (attempt < maxRetries - 1) + { + Console.WriteLine($"[EonaCat Server] Access denied to log directory (attempt {attempt + 1}/{maxRetries}): {ex.Message}"); + await Task.Delay(retryDelayMs * (attempt + 1)).ConfigureAwait(false); + } + catch (IOException ex) when (attempt < maxRetries - 1) + { + Console.WriteLine($"[EonaCat Server] IO error writing to log (attempt {attempt + 1}/{maxRetries}): {ex.Message}"); + await Task.Delay(retryDelayMs * (attempt + 1)).ConfigureAwait(false); + } + catch (Exception ex) + { + // Any other exception - log and drop the message to prevent crash + Console.WriteLine($"[EonaCat Server] Error writing log: {ex.GetType().Name}: {ex.Message}"); + Interlocked.Increment(ref Metrics.TotalDropped); + LogDropped?.Invoke(logData); + return; + } } - File.AppendAllText(filePath, logData + Environment.NewLine); - Interlocked.Increment(ref Metrics.TotalWritten); - LogWritten?.Invoke(logData); + // All retries exhausted + Console.WriteLine("[EonaCat Server] Max retries exhausted, dropping log message"); + Interlocked.Increment(ref Metrics.TotalDropped); + LogDropped?.Invoke(logData); + } - CleanUpOldLogs(); + /// + /// Ensures a log directory exists and is writable, with permission fixing if needed. + /// Returns true if successful, false otherwise. + /// + private bool EnsureLogDirectory(string dirPath) + { + try + { + return Helpers.DirectoryPermissionHelper.EnsureDirectoryHierarchy(dirPath); + } + catch (Exception ex) + { + Console.WriteLine($"[EonaCat Server] Failed to ensure log directory: {ex.Message}"); + return false; + } } private void CleanUpOldLogs()