MES-ETL/MesETL.App/HostedServices/TaskMonitorService.cs

using System.Diagnostics;
using System.Text;
using MesETL.App.Services;
using MesETL.App.Services.Loggers;
using Microsoft.Extensions.DependencyInjection;

namespace MesETL.App.HostedServices;

/// <summary>
/// 任务监控
/// </summary>
public class TaskMonitorService
{
    private readonly IEnumerable<ITaskMonitorLogger> _monitorLoggers;
    private readonly ProcessContext _context;
    private readonly DataRecordQueue _producerQueue;
    private readonly RecordQueuePool _queuePool;
    
    private string _outputPath = Path.Combine(AppDomain.CurrentDomain.BaseDirectory, "Log/progress.txt");

    public TaskMonitorService(ProcessContext context,
        [FromKeyedServices(Const.ConstVar.Producer)]
        DataRecordQueue producerQueue,
        RecordQueuePool queuePool,
        IEnumerable<ITaskMonitorLogger> monitorLoggers)
    {
        _context = context;
        _producerQueue = producerQueue;
        _queuePool = queuePool;
        _monitorLoggers = monitorLoggers;
    }

    public async Task Monitor(CancellationToken stoppingToken)
    {
        var sw = Stopwatch.StartNew();
        var lastTime = sw.ElapsedMilliseconds;
        var lastInputCount = _context.InputCount;
        var lastTransformCount = _context.TransformCount;
        var lastOutputCount = _context.OutputCount;

        bool endCheck = false;
        while (!stoppingToken.IsCancellationRequested)
        {
            EndCheck:
            // var running = 0;
            // var error = 0;
            // var completed = 0;
            // var canceled = 0;
            // foreach (var task in _taskManager.Tasks)
            // {
            //     switch (task.Status)
            //     {
            //         case TaskStatus.Canceled:
            //             canceled++;
            //             break;
            //         case TaskStatus.Faulted:
            //             error++;
            //             break;
            //         case TaskStatus.RanToCompletion:
            //             completed++;
            //             break;
            //         default:
            //             running++;
            //             break;
            //     }
            // }

            var time = sw.ElapsedMilliseconds;
            var inputCount = _context.InputCount;
            var transformCount = _context.TransformCount;
            var outputCount = _context.OutputCount;
            
            var elapseTime = (time - lastTime) / 1000f;
            var inputSpeed = (inputCount - lastInputCount) / elapseTime;
            var transformSpeed = (transformCount - lastTransformCount) / elapseTime;
            var outputSpeed = (outputCount - lastOutputCount) / elapseTime;
            
            // _logger.LogInformation(
            //     "Task monitor: running: {Running}, error: {Error}, completed: {Completed}, canceled: {Canceled}, outputSpeed: {Speed} records/s",
            //     running, error, completed, canceled, outputSpeed);
            foreach (var logger in _monitorLoggers)
            {
                var memory = GC.GetTotalMemory(false) / 1024 / 1024;
                _context.MaxMemoryUsage = Math.Max(_context.MaxMemoryUsage, memory);
                logger.LogStatus("系统监控", new Dictionary<string, string>
                {
                    { "输入速度", _context.IsInputCompleted ? "OK" : $"{inputSpeed:F2}/s" },
                    { "转换速度", _context.IsTransformCompleted ? "OK" : $"{transformSpeed:F2}/s" },
                    { "输出速度", _context.IsOutputCompleted ? "OK" : $"{outputSpeed:F2}/s" },

                    { "| 输入队列长度", _producerQueue.Count.ToString() },
                    { "输出队列长度", _queuePool.Queues.Values.Sum(queue => queue.Count).ToString() },
                    { "内存使用", $"{memory} MiB" },
                });

                var dict = _context.TableProgress
                    .ToDictionary(kv => kv.Key, kv => $"{kv.Value.input}:{kv.Value.output}");
                logger.LogStatus("系统监控: 表处理进度(I:O)", dict, ITaskMonitorLogger.LogLevel.Progress);
                logger.LogStatus("系统监控：输出队列状态", 
                    _queuePool.Queues.ToDictionary(q => q.Key, q => q.Value.Count.ToString()), 
                    ITaskMonitorLogger.LogLevel.Progress);
                var sb = new StringBuilder("表处理进度：\n");
                foreach (var kv in dict)
                {
                    sb.Append(kv.Key).AppendLine(kv.Value);
                }

                sb.AppendLine($"数据记录字段的最大长度：{_producerQueue.LongestFieldCharCount}");
                await File.WriteAllTextAsync(_outputPath, sb.ToString(), CancellationToken.None);
                
            
                // logger.LogStatus("Monitor: Process count", new Dictionary<string, string>
                // {
                //     {"Input", inputCount.ToString()},
                //     {"Transform", transformCount.ToString()},
                //     {"Output", outputCount.ToString()}
                // }, ITaskMonitorLogger.LogLevel.Progress);
            }

            
            await Task.Delay(5000, stoppingToken);

            lastTime = time;
            lastInputCount = inputCount;
            lastTransformCount = transformCount;
            lastOutputCount = outputCount;

            if (_context is { IsInputCompleted: true, IsTransformCompleted: true, IsOutputCompleted: true })
            {
                if (!endCheck)
                {
                    endCheck = true;
                    goto EndCheck;
                }
                break;
            }
        }
    }

}