insight/src/Agent/Insight.Agent/Services/EventService.cs

using Insight.Agent.Messages;
using Insight.Agent.Network;
using Microsoft.Extensions.Hosting;
using Microsoft.Extensions.Logging;
using System.Diagnostics.Eventing.Reader;
using System.Runtime.Versioning;
using System.Threading.Channels;
using Vaitr.Network;
using static Insight.Agent.Messages.Event;
using EventLevel = System.Diagnostics.Tracing.EventLevel;

namespace Insight.Agent.Services
{
    [SupportedOSPlatform("windows")]
    internal class EventService : BackgroundService
    {
        private readonly Channel<Event> _queue;
        private readonly TcpSessionPool<AgentSession, MemPackSerializer, IAgentMessage> _pool;
        private readonly ILogger<EventService> _logger;

        public EventService(TcpSessionPool<AgentSession, MemPackSerializer, IAgentMessage> pool, ILogger<EventService> logger)
        {
            _pool = pool;
            _logger = logger;

            _queue = Channel.CreateBounded<Event>(new BoundedChannelOptions(1000)
            {
                SingleReader = true,
                SingleWriter = true,
                AllowSynchronousContinuations = false,
                FullMode = BoundedChannelFullMode.DropOldest
            });
        }

        protected override async Task ExecuteAsync(CancellationToken cancellationToken)
        {
            _logger.LogTrace("ExecuteAsync");

            while (cancellationToken.IsCancellationRequested is false)
            {
                try
                {
                    var tasks = new List<Task>
                    {
                        HandleQueueAsync(cancellationToken),
                        WatchAsync("Application", "*", cancellationToken),
                        WatchAsync("Security", "*", cancellationToken),
                        WatchAsync("System", "*", cancellationToken),
                        WatchAsync("Microsoft-Windows-PrintService/Admin", "*", cancellationToken),
                        WatchAsync("Microsoft-Windows-TaskScheduler/Operational", "*", cancellationToken),
                        WatchAsync("Microsoft-Windows-TerminalServices-RemoteConnectionManager/Operational", "*", cancellationToken),
                        WatchAsync("Microsoft-Windows-TerminalServices-LocalSessionManager/Operational", "*", cancellationToken),
                        WatchAsync("Microsoft-Windows-TerminalServices-RDPClient/Operational", "*", cancellationToken),
                        WatchAsync("Microsoft-Windows-SmbClient/Connectivity", "*", cancellationToken),
                        WatchAsync("Microsoft-Windows-SmbClient/Security", "*", cancellationToken),
                        WatchAsync("Microsoft-Windows-SMBServer/Security", "*", cancellationToken),
                        WatchAsync("Microsoft-Windows-StorageSpaces-Driver/Operational", "*", cancellationToken),
                        WatchAsync("Microsoft-Windows-Diagnostics-Performance/Operational", "*", cancellationToken)
                    };

                    await Task.WhenAll(tasks);
                }
                catch (OperationCanceledException) { }
                catch (Exception ex)
                {
                    _logger.LogError("{ex}", ex);
                }
            }
        }

        private async Task WatchAsync(string path, string query, CancellationToken cancellationToken)
        {
            var config = new EventLogConfiguration(path);
            if (config is null) return;

            if (config.IsEnabled is false)
            {
                config.IsEnabled = true;
                config.SaveChanges();
            }

            var watcher = new EventLogWatcher(new EventLogQuery(path, PathType.LogName, query)
            {
                TolerateQueryErrors = true,
                Session = EventLogSession.GlobalSession
            });

            try
            {
                watcher.EventRecordWritten += new EventHandler<EventRecordWrittenEventArgs>(OnEvent);
                watcher.Enabled = true;

                using var semaphore = new SemaphoreSlim(0, 1);
                await semaphore.WaitAsync(cancellationToken);
            }
            catch (Exception) { }
            finally
            {
                watcher.EventRecordWritten -= new EventHandler<EventRecordWrittenEventArgs>(OnEvent);
                watcher.Enabled = false;

                watcher?.Dispose();
            }
        }

        private async Task HandleQueueAsync(CancellationToken cancellationToken)
        {
            while (await _queue.Reader.WaitToReadAsync(cancellationToken))
            {
                var session = _pool.FirstOrDefault();

                if (session.Value is null)
                {
                    await Task.Delay(10000, cancellationToken);
                    continue;
                }

                if (_queue.Reader.TryRead(out var item) is false)
                {
                    await Task.Delay(1000, cancellationToken);
                    continue;
                }

                try
                {
                    await session.Value.SendAsync(item, cancellationToken);
                }
                catch (OperationCanceledException) { }
                catch (Exception ex)
                {
                    _logger.LogError("{ex}", ex);
                    await Task.Delay(10000, cancellationToken);
                }
            }
        }

        private void OnEvent(object? sender, EventRecordWrittenEventArgs e)
        {
            if (e is null || e.EventRecord is null) return;

            try
            {
                var @event = new Event
                {
                    Timestamp = e?.EventRecord?.TimeCreated,
                    EventId = e?.EventRecord?.Id,
                    Source = e?.EventRecord?.ProviderName,
                    Category = e?.EventRecord?.LogName,
                    Task = e?.EventRecord?.TaskDisplayName,
                    Message = e?.EventRecord?.FormatDescription(),
                };

                if (e?.EventRecord?.Level is not null)
                {
                    @event.Status = (EventLevel)Convert.ToInt32(e?.EventRecord?.Level) switch
                    {
                        EventLevel.Informational => StatusType.Information,
                        EventLevel.Warning => StatusType.Warning,
                        EventLevel.Error => StatusType.Error,
                        EventLevel.Critical => StatusType.Critical,
                        _ => StatusType.Information,
                    };
                }

                _queue.Writer.WriteAsync(@event, default);
            }
            catch (Exception) { } // app crash
        }
    }
}