jmon/JMonAgent/Worker.cs

using System.Runtime.InteropServices;
using Grpc.Net.Client;
using JMon.Protos;

namespace JMonAgent;

public class Worker : BackgroundService
{
    private readonly ILogger<Worker> _logger;
    private readonly IMetricCollector _collector;
    private readonly MetricsQueue _queue;
    private readonly bool _debugMode;
    private GrpcChannel? _channel;
    private MetricsService.MetricsServiceClient? _client;
    private SystemMetrics? _previousMetrics;
    private readonly SemaphoreSlim _executionLock = new(1, 1);
    
    private const double CpuThreshold = 0.5; // Only send if change > 0.5%
    private const long MemoryThreshold = 10 * 1024 * 1024; // 10 MB
    private const long SwapThreshold = 10 * 1024 * 1024; // 10 MB
    private const long DiskThreshold = 100 * 1024 * 1024; // 100 MB

    public Worker(ILogger<Worker> logger, MetricsQueue queue, IConfiguration config)
    {
        _logger = logger;
        _queue = queue;
        _debugMode = config.GetValue<bool>("DEBUG", false);
        
        if (RuntimeInformation.IsOSPlatform(OSPlatform.Linux)) 
            _collector = new LinuxMetricCollector();
        else if (RuntimeInformation.IsOSPlatform(OSPlatform.Windows))
            _collector = new WindowsMetricCollector();
        else
            _collector = new MacMetricCollector();
    }

    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
    {
        _logger.LogInformation("Agent started on {platform}", RuntimeInformation.OSDescription);
        
        if (_debugMode)
        {
            _logger.LogInformation("DEBUG MODE ENABLED - Metrics will be printed, not sent");
            await RunDebugMode(stoppingToken);
        }
        else
        {
            await RunNormalMode(stoppingToken);
        }
    }

    private async Task RunDebugMode(CancellationToken stoppingToken)
    {
        using var timer = new PeriodicTimer(TimeSpan.FromSeconds(5));
        
        try
        {
            while (await timer.WaitForNextTickAsync(stoppingToken))
            {
                // Skip if previous execution is still running (non-blocking lock)
                if (!_executionLock.Wait(0))
                {
                    _logger.LogDebug("Skipping tick: previous execution still running");
                    continue;
                }

                try
                {
                    var metrics = _collector.Collect();
                    var delta = CalculateDelta(metrics);

                    Console.WriteLine("\n=== METRICS DEBUG OUTPUT ===");
                    Console.WriteLine($"Machine: {delta.MachineName}");
                    if (delta.OSDescription != null)
                        Console.WriteLine($"OS: {delta.OSDescription}");
                    if (delta.IpAddress != null)
                        Console.WriteLine($"IP: {delta.IpAddress}");
                    if (delta.HwId != null)
                        Console.WriteLine($"HWID: {delta.HwId}");
                    if (delta.CpuUsagePercentage.HasValue)
                        Console.WriteLine($"CPU: {delta.CpuUsagePercentage:F2}%");
                    if (delta.MemoryUsedBytes.HasValue)
                        Console.WriteLine($"Memory: {delta.MemoryUsedBytes.Value / (1024 * 1024)} MB / {delta.MemoryTotalBytes.GetValueOrDefault() / (1024 * 1024)} MB");
                    if (delta.SwapUsedBytes.HasValue)
                        Console.WriteLine($"Swap: {delta.SwapUsedBytes.Value / (1024 * 1024)} MB / {delta.SwapTotalBytes.GetValueOrDefault() / (1024 * 1024)} MB");
                    if (delta.DiskUsedBytes.HasValue)
                        Console.WriteLine($"Disk: {delta.DiskUsedBytes.Value / (1024 * 1024)} MB / {delta.DiskTotalBytes.GetValueOrDefault() / (1024 * 1024)} MB ({delta.DiskMountPoint})");
                    Console.WriteLine($"Timestamp: {delta.Timestamp:O}");
                    Console.WriteLine("===========================\n");

                    _previousMetrics = metrics;
                }
                catch (Exception ex)
                {
                    _logger.LogError("Error in debug mode: {msg}", ex.Message);
                }
                finally
                {
                    _executionLock.Release();
                }
            }
        }
        catch (OperationCanceledException)
        {
            _logger.LogInformation("Debug mode cancelled");
        }
    }

    private async Task RunNormalMode(CancellationToken stoppingToken)
    {
        // Create gRPC channel (ignore self-signed certs for development)
        var handler = new HttpClientHandler();
        handler.ServerCertificateCustomValidationCallback = (msg, cert, chain, errors) => true;
        
        _channel = GrpcChannel.ForAddress("https://localhost:5001", new GrpcChannelOptions
        {
            HttpHandler = handler
        });
        
        _client = new MetricsService.MetricsServiceClient(_channel);
        
        using var timer = new PeriodicTimer(TimeSpan.FromSeconds(5));
        
        try
        {
            while (await timer.WaitForNextTickAsync(stoppingToken))
            {
                // Skip if previous execution is still running (non-blocking lock)
                if (!_executionLock.Wait(0))
                {
                    _logger.LogWarning("Skipping tick: previous execution still running (>5s)");
                    continue;
                }

                try
                {
                    // Try to flush queued metrics first
                    if (_queue.Count > 0)
                    {
                        var queuedMetrics = _queue.DequeueAll();
                        _logger.LogInformation("Flushing {count} queued metrics", queuedMetrics.Count);
                        
                        foreach (var queued in queuedMetrics)
                        {
                            try
                            {
                                await _client.ReportMetricsAsync(queued, cancellationToken: stoppingToken);
                            }
                            catch (Exception ex)
                            {
                                _logger.LogError("Failed to flush queued metric: {msg}", ex.Message);
                                _queue.Enqueue(queued); // Re-queue on failure
                                // Continue trying other queued metrics instead of breaking
                            }
                        }
                    }

                    var metrics = _collector.Collect();
                    _logger.LogDebug("Collected metrics: CPU={cpu}%, Memory={mem}MB/{total}MB", 
                        metrics.CpuUsagePercentage.ToString("F2"),
                        (metrics.MemoryUsedBytes / (1024 * 1024)),
                        (metrics.MemoryTotalBytes / (1024 * 1024)));

                    var delta = CalculateDelta(metrics);
                    
                    // Convert to gRPC message using object initializer
                    var grpcDelta = new JMon.Protos.DeltaMetrics
                    {
                        MachineName = delta.MachineName,
                        Hwid = delta.HwId,
                        TimestampUnixMs = new DateTimeOffset(delta.Timestamp).ToUnixTimeMilliseconds(),
                        ProcessId = delta.ProcessId
                    };

                    if (!string.IsNullOrEmpty(delta.OSDescription))
                        grpcDelta.OsDescription = delta.OSDescription;
                    if (delta.CpuUsagePercentage.HasValue)
                        grpcDelta.CpuUsagePercentage = delta.CpuUsagePercentage.Value;
                    if (delta.MemoryUsedBytes.HasValue)
                        grpcDelta.MemoryUsedBytes = delta.MemoryUsedBytes.Value;
                    if (delta.MemoryTotalBytes.HasValue)
                        grpcDelta.MemoryTotalBytes = delta.MemoryTotalBytes.Value;
                    if (delta.SwapUsedBytes.HasValue)
                        grpcDelta.SwapUsedBytes = delta.SwapUsedBytes.Value;
                    if (delta.SwapTotalBytes.HasValue)
                        grpcDelta.SwapTotalBytes = delta.SwapTotalBytes.Value;
                    if (delta.DiskUsedBytes.HasValue)
                        grpcDelta.DiskUsedBytes = delta.DiskUsedBytes.Value;
                    if (delta.DiskTotalBytes.HasValue)
                        grpcDelta.DiskTotalBytes = delta.DiskTotalBytes.Value;
                    if (!string.IsNullOrEmpty(delta.DiskMountPoint))
                        grpcDelta.DiskMountPoint = delta.DiskMountPoint;
                    if (!string.IsNullOrEmpty(delta.IpAddress))
                        grpcDelta.IpAddress = delta.IpAddress;

                    var response = await _client.ReportMetricsAsync(grpcDelta, cancellationToken: stoppingToken);
                    
                    _logger.LogInformation("Metrics reported: {response}", response.Command);
                    
                    // Handle server commands
                    if (response.Command == MetricsResponse.Types.CommandType.Sync)
                    {
                        _logger.LogInformation("Server requested full resync: {msg}", response.Message);
                        _previousMetrics = null; // Reset to send full update next time
                    }
                    else if (response.Command == MetricsResponse.Types.CommandType.Sigterm)
                    {
                        _logger.LogInformation("Server terminating process: {msg}", response.Message);
                        Environment.Exit(0);
                    }
                    
                    _previousMetrics = metrics;
                }
                catch (Exception ex)
                {
                    _logger.LogError("Error reporting metrics: {msg}", ex.Message);
                    
                    // On failure, queue the FULL metrics for retry (not delta)
                    // This ensures the server can reconstruct properly even if it was restarted
                    if (ex is not OperationCanceledException)
                    {
                        var metrics = _collector.Collect();
                        
                        var grpcDelta = new JMon.Protos.DeltaMetrics
                        {
                            MachineName = metrics.MachineName,
                            Hwid = metrics.HwId,
                            TimestampUnixMs = new DateTimeOffset(metrics.Timestamp).ToUnixTimeMilliseconds(),
                            Retried = true,
                            ProcessId = metrics.ProcessId,
                            // Send FULL metrics (all fields) on retry
                            OsDescription = metrics.OSDescription,
                            CpuUsagePercentage = metrics.CpuUsagePercentage,
                            MemoryUsedBytes = metrics.MemoryUsedBytes,
                            MemoryTotalBytes = metrics.MemoryTotalBytes,
                            SwapUsedBytes = metrics.SwapUsedBytes,
                            SwapTotalBytes = metrics.SwapTotalBytes,
                            DiskUsedBytes = metrics.DiskUsedBytes,
                            DiskTotalBytes = metrics.DiskTotalBytes,
                            DiskMountPoint = metrics.DiskMountPoint,
                            IpAddress = metrics.IpAddress
                        };

                        _queue.Enqueue(grpcDelta);
                    }
                }
                finally
                {
                    _executionLock.Release();
                }
            }
        }
        catch (OperationCanceledException)
        {
            _logger.LogInformation("Normal mode cancelled");
        }
    }

    private DeltaMetrics CalculateDelta(SystemMetrics current)
    {
        // First time - send everything
        if (_previousMetrics == null)
        {
            return new DeltaMetrics(
                current.MachineName,
                current.OSDescription,
                current.CpuUsagePercentage,
                current.MemoryUsedBytes,
                current.MemoryTotalBytes,
                current.SwapUsedBytes,
                current.SwapTotalBytes,
                current.DiskUsedBytes,
                current.DiskTotalBytes,
                current.DiskMountPoint,
                current.Timestamp,
                current.IpAddress,
                current.HwId
            );
        }

        // Calculate deltas and only include changed values
        return new DeltaMetrics(
            current.MachineName,
            current.OSDescription != _previousMetrics.OSDescription ? current.OSDescription : null,
            Math.Abs(current.CpuUsagePercentage - _previousMetrics.CpuUsagePercentage) > CpuThreshold ? current.CpuUsagePercentage : null,
            Math.Abs(current.MemoryUsedBytes - _previousMetrics.MemoryUsedBytes) > MemoryThreshold ? current.MemoryUsedBytes : null,
            current.MemoryTotalBytes != _previousMetrics.MemoryTotalBytes ? current.MemoryTotalBytes : null,
            Math.Abs(current.SwapUsedBytes - _previousMetrics.SwapUsedBytes) > SwapThreshold ? current.SwapUsedBytes : null,
            current.SwapTotalBytes != _previousMetrics.SwapTotalBytes ? current.SwapTotalBytes : null,
            Math.Abs(current.DiskUsedBytes - _previousMetrics.DiskUsedBytes) > DiskThreshold ? current.DiskUsedBytes : null,
            current.DiskTotalBytes != _previousMetrics.DiskTotalBytes ? current.DiskTotalBytes : null,
            current.DiskMountPoint != _previousMetrics.DiskMountPoint ? current.DiskMountPoint : null,
            current.Timestamp,
            current.IpAddress != _previousMetrics.IpAddress ? current.IpAddress : null,
            current.HwId  // Always send HWID as it's the unique identifier
        );
    }

    public override async Task StopAsync(CancellationToken cancellationToken)
    {
        if (_channel != null)
        {
            await _channel.ShutdownAsync();
            _channel.Dispose();
        }
        await base.StopAsync(cancellationToken);
    }
}