diff --git a/RemoteExec.Client/LoadBalancingStrategy.cs b/RemoteExec.Client/LoadBalancingStrategy.cs new file mode 100644 index 0000000..a744a07 --- /dev/null +++ b/RemoteExec.Client/LoadBalancingStrategy.cs @@ -0,0 +1,10 @@ +namespace RemoteExec.Client; + +public enum LoadBalancingStrategy +{ + RoundRobin, + Random, + LeastConnections, + LeastActiveTasks, + ResourceAware +} \ No newline at end of file diff --git a/RemoteExec.Client/RemoteExecutor.cs b/RemoteExec.Client/RemoteExecutor.cs index 027d973..be44163 100644 --- a/RemoteExec.Client/RemoteExecutor.cs +++ b/RemoteExec.Client/RemoteExecutor.cs @@ -2,6 +2,7 @@ using RemoteExec.Shared; +using System.Collections.Concurrent; using System.Diagnostics.CodeAnalysis; using System.Reflection; using System.Text.Json; @@ -9,45 +10,101 @@ using System.Threading.Channels; namespace RemoteExec.Client; -public class RemoteExecutor(string url) +public class RemoteExecutor { - private readonly HubConnection _connection = - new HubConnectionBuilder() - .WithUrl(url) - .WithAutomaticReconnect() - .Build(); + private readonly List connections = []; + private readonly ConcurrentDictionary serverMetrics = new(); + private int _currentConnectionIndex = 0; + private readonly Lock @lock = new Lock(); + private readonly LoadBalancingStrategy loadBalancingStrategy; + + public event EventHandler? MetricsUpdated; + + public RemoteExecutor(string url) : this([url], LoadBalancingStrategy.RoundRobin) + { + } + + public RemoteExecutor(string[] urls, LoadBalancingStrategy loadBalancingStrategy = LoadBalancingStrategy.RoundRobin) + { + this.loadBalancingStrategy = loadBalancingStrategy; + + foreach (string url in urls) + { + HubConnection connection = new HubConnectionBuilder() + .WithUrl(url) + .WithAutomaticReconnect() + .Build(); + + connections.Add(connection); + } + } public async Task StartAsync(CancellationToken cancellationToken = default) { - _ = _connection.On($"RequestAssembly", async (string assemblyName, Guid requestId) => + List startTasks = []; + + foreach (HubConnection connection in connections) { - Assembly? assembly = AppDomain.CurrentDomain.GetAssemblies().FirstOrDefault(a => a.GetName().FullName == assemblyName); - - if (assembly == null) + _ = connection.On("MetricsUpdated", metrics => { - assembly = Assembly.Load(new AssemblyName(assemblyName)); - } + serverMetrics[connection] = metrics; + MetricsUpdated?.Invoke(this, new ServerMetricsUpdatedEventArgs(connection, metrics)); + }); - byte[] dllBytes = await File.ReadAllBytesAsync(assembly.Location!); - - Channel channel = Channel.CreateUnbounded(); - - foreach (byte b in dllBytes) + _ = connection.On($"RequestAssembly", async (string assemblyName, Guid requestId) => { - await channel.Writer.WriteAsync(b); - } + Assembly? assembly = AppDomain.CurrentDomain.GetAssemblies().FirstOrDefault(a => a.GetName().FullName == assemblyName) ?? Assembly.Load(new AssemblyName(assemblyName)); + byte[] dllBytes = await File.ReadAllBytesAsync(assembly.Location!); - channel.Writer.Complete(); + Channel channel = Channel.CreateUnbounded(); - await _connection.InvokeAsync("ProvideAssembly", requestId, channel.Reader); - }); + foreach (byte b in dllBytes) + { + await channel.Writer.WriteAsync(b); + } - await _connection.StartAsync(cancellationToken); + channel.Writer.Complete(); + + await connection.InvokeAsync("ProvideAssembly", requestId, channel.Reader); + }); + + startTasks.Add(connection.StartAsync(cancellationToken) + .ContinueWith(async (task, state) => + { + HubConnection conn = (HubConnection)state!; + serverMetrics[conn] = await conn.InvokeAsync("GetMetrics", cancellationToken); + MetricsUpdated?.Invoke(this, new ServerMetricsUpdatedEventArgs(conn, serverMetrics[conn])); + }, connection, TaskScheduler.Default).Unwrap()); + } + + await Task.WhenAll(startTasks); } public async Task StopAsync(CancellationToken cancellationToken = default) { - await _connection.StopAsync(cancellationToken); + List stopTasks = []; + + foreach (HubConnection connection in connections) + { + stopTasks.Add(connection.StopAsync(cancellationToken)); + } + + await Task.WhenAll(stopTasks); + } + + public Dictionary GetCurrentServerMetrics() + { + Dictionary metrics = []; + + foreach (HubConnection connection in connections) + { + if (serverMetrics.TryGetValue(connection, out ServerMetrics? newServerMetrics)) + { + metrics[newServerMetrics.ServerId] = newServerMetrics; + } + } + + return metrics; } public bool TryExecute(TDelegate del, out TResult? result, params object[] args) where TDelegate : Delegate @@ -118,7 +175,9 @@ public class RemoteExecutor(string url) Arguments = args }; - RemoteExecutionResult result = _connection + HubConnection connection = GetNextConnection(); + + RemoteExecutionResult result = connection .InvokeAsync("Execute", request) .GetAwaiter() .GetResult(); @@ -130,4 +189,63 @@ public class RemoteExecutor(string url) return result.Result; } -} \ No newline at end of file + + private HubConnection GetNextConnection() + { + if (connections.Count == 0) + { + throw new InvalidOperationException("No connections available"); + } + + return loadBalancingStrategy switch + { + LoadBalancingStrategy.RoundRobin => GetRoundRobinConnection(), + LoadBalancingStrategy.Random => GetRandomConnection(), + LoadBalancingStrategy.LeastConnections => GetLeastConnections(), + LoadBalancingStrategy.LeastActiveTasks => GetLeastActiveTasksConnections(), + LoadBalancingStrategy.ResourceAware => GetResourceAwareConnections(), + _ => throw new NotSupportedException($"Load balancing strategy {loadBalancingStrategy} is not supported") + }; + } + + private HubConnection GetRoundRobinConnection() + { + lock (@lock) + { + HubConnection connection = connections[_currentConnectionIndex]; + _currentConnectionIndex = (_currentConnectionIndex + 1) % connections.Count; + return connection; + } + } + + private HubConnection GetRandomConnection() + { + int index = Random.Shared.Next(connections.Count); + return connections[index]; + } + + private HubConnection GetLeastConnections() + { + // Purely looks at how many clients are connected to the Hub + return connections.OrderBy(c => serverMetrics.TryGetValue(c, out ServerMetrics? m) ? m.ActiveConnections : 0).First(); + } + + private HubConnection GetLeastActiveTasksConnections() + { + return connections.OrderBy(c => serverMetrics.TryGetValue(c, out ServerMetrics? m) ? m.ActiveTasks : 0).First(); + } + + private HubConnection GetResourceAwareConnections() + { + return connections.OrderBy(c => + { + if (!serverMetrics.TryGetValue(c, out ServerMetrics? m)) + { + return 0; + } + + // Simple heuristic: CPU percentage + (Memory in MB / 1024) + return m.CpuUsage + (m.TotalMemoryUsage / 1024 / 1024 / 100); + }).First(); + } +} diff --git a/RemoteExec.Client/ServerMetricsUpdatedEventArgs.cs b/RemoteExec.Client/ServerMetricsUpdatedEventArgs.cs new file mode 100644 index 0000000..db4ad46 --- /dev/null +++ b/RemoteExec.Client/ServerMetricsUpdatedEventArgs.cs @@ -0,0 +1,11 @@ +using Microsoft.AspNetCore.SignalR.Client; + +using RemoteExec.Shared; + +namespace RemoteExec.Client; + +public class ServerMetricsUpdatedEventArgs(HubConnection connection, ServerMetrics metrics) : EventArgs +{ + public HubConnection Connection { get; } = connection; + public ServerMetrics Metrics { get; } = metrics; +} diff --git a/RemoteExec.Server/Hubs/RemoteExecutionHub.cs b/RemoteExec.Server/Hubs/RemoteExecutionHub.cs index 500f08b..1832e63 100644 --- a/RemoteExec.Server/Hubs/RemoteExecutionHub.cs +++ b/RemoteExec.Server/Hubs/RemoteExecutionHub.cs @@ -3,6 +3,7 @@ using RemoteExec.Shared; using System.Collections.Concurrent; +using System.Diagnostics; using System.Reflection; using System.Text.Json; using System.Threading.Channels; @@ -15,6 +16,11 @@ public class RemoteExecutionHub(ILogger logger) : Hub private static readonly ConcurrentDictionary> pendingAssemblyRequests = new(); + private static DateTime lastMetricsTimestamp; + private static TimeSpan lastTotalProcessorTime; + + private static int activeTasks = 0; + public override Task OnConnectedAsync() { RemoteJobAssemblyLoadContext assemblyLoadContext = new RemoteJobAssemblyLoadContext($"RemoteJob_{Guid.NewGuid()}"); @@ -39,6 +45,8 @@ public class RemoteExecutionHub(ILogger logger) : Hub public async Task Execute(RemoteExecutionRequest req) { + _ = Interlocked.Increment(ref activeTasks); + try { if (!connections.TryGetValue(Context.ConnectionId, out RemoteJobAssemblyLoadContext? assemblyLoadContext)) @@ -110,6 +118,23 @@ public class RemoteExecutionHub(ILogger logger) : Hub object? result = method.Invoke(null, invokeArgs); + if (result is Task taskResult) + { + await taskResult.ConfigureAwait(false); + Type returnType = method.ReturnType; + if (returnType.IsGenericType && returnType.GetGenericTypeDefinition() == typeof(Task<>)) + { + // For Task, get the Result property + PropertyInfo resultProperty = returnType.GetProperty("Result")!; + result = resultProperty.GetValue(taskResult); + } + else + { + // For non-generic Task, result is null + result = null; + } + } + return new RemoteExecutionResult { Result = result @@ -124,6 +149,10 @@ public class RemoteExecutionHub(ILogger logger) : Hub Exception = ex.ToString() }; } + finally + { + _ = Interlocked.Decrement(ref activeTasks); + } } public async Task ProvideAssembly(Guid requestId, ChannelReader stream) @@ -146,6 +175,48 @@ public class RemoteExecutionHub(ILogger logger) : Hub } } + public async Task GetMetrics() + { + return await GetServerMetrics(); + } + + public static async Task BroadcastMetricsAsync(IHubContext hubContext) + { + ServerMetrics metrics = await GetServerMetrics(); + await hubContext.Clients.All.SendAsync("MetricsUpdated", metrics); + } + + private static async Task GetServerMetrics() + { + Process currentProcess = Process.GetCurrentProcess(); + + // Capture current values + DateTime currentTime = DateTime.UtcNow; + TimeSpan currentProcessorTime = currentProcess.TotalProcessorTime; + + // Calculate the difference since the last check + double elapsedMs = (currentTime - lastMetricsTimestamp).TotalMilliseconds; + double cpuMsUsed = (currentProcessorTime - lastTotalProcessorTime).TotalMilliseconds; + + // Calculate percentage: (Time Used / Time Elapsed) / Cores + // We multiply by 100 to get a 0-100 scale + double cpuUsagePercent = cpuMsUsed / elapsedMs / Environment.ProcessorCount * 100; + + // Update static variables for the next call + lastMetricsTimestamp = currentTime; + lastTotalProcessorTime = currentProcessorTime; + + return new ServerMetrics + { + ServerId = Environment.MachineName, + ActiveConnections = connections.Count, + ActiveTasks = activeTasks, + TotalMemoryUsage = currentProcess.WorkingSet64, + CpuUsage = Math.Clamp(Math.Round(cpuUsagePercent, 2), 0, 100), + Timestamp = currentTime + }; + } + private async Task RequestAssemblyAsync(string assemblyName) { try diff --git a/RemoteExec.Server/Program.cs b/RemoteExec.Server/Program.cs index 8348be2..71f0572 100644 --- a/RemoteExec.Server/Program.cs +++ b/RemoteExec.Server/Program.cs @@ -1,4 +1,5 @@ using RemoteExec.Server.Hubs; +using RemoteExec.Server.Services; WebApplicationBuilder builder = WebApplication.CreateBuilder(args); @@ -7,6 +8,10 @@ WebApplicationBuilder builder = WebApplication.CreateBuilder(args); builder.Services.AddControllers(); builder.Services.AddSignalR(); builder.Services.AddOpenApi(); +builder.Services.AddHealthChecks(); + +// Add metrics broadcast background service +builder.Services.AddHostedService(); WebApplication app = builder.Build(); @@ -18,10 +23,12 @@ if (app.Environment.IsDevelopment()) _ = app.MapOpenApi(); } -app.UseHttpsRedirection(); +//app.UseHttpsRedirection(); app.UseAuthorization(); app.MapControllers(); +app.MapHealthChecks("/health"); + await app.RunAsync(); diff --git a/RemoteExec.Server/Services/MetricsBroadcastService.cs b/RemoteExec.Server/Services/MetricsBroadcastService.cs new file mode 100644 index 0000000..f7fd24f --- /dev/null +++ b/RemoteExec.Server/Services/MetricsBroadcastService.cs @@ -0,0 +1,36 @@ +using Microsoft.AspNetCore.SignalR; + +using RemoteExec.Server.Hubs; + +namespace RemoteExec.Server.Services; + +public class MetricsBroadcastService(IHubContext hubContext, ILogger logger) : BackgroundService +{ + private readonly TimeSpan _broadcastInterval = TimeSpan.FromSeconds(2); + + protected override async Task ExecuteAsync(CancellationToken stoppingToken) + { + logger.LogInformation("Metrics broadcast service started"); + + while (!stoppingToken.IsCancellationRequested) + { + try + { + await Task.Delay(_broadcastInterval, stoppingToken); + + await RemoteExecutionHub.BroadcastMetricsAsync(hubContext); + } + catch (OperationCanceledException) + { + // Expected when service is stopping + break; + } + catch (Exception ex) + { + logger.LogError(ex, "Error broadcasting metrics"); + } + } + + logger.LogInformation("Metrics broadcast service stopped"); + } +} \ No newline at end of file diff --git a/RemoteExec.Server/WeatherForecast.cs b/RemoteExec.Server/WeatherForecast.cs deleted file mode 100644 index 88a3d78..0000000 --- a/RemoteExec.Server/WeatherForecast.cs +++ /dev/null @@ -1,12 +0,0 @@ -namespace RemoteExec.Server; - -public class WeatherForecast -{ - public DateOnly Date { get; set; } - - public int TemperatureC { get; set; } - - public int TemperatureF => 32 + (int)(TemperatureC / 0.5556); - - public string? Summary { get; set; } -} diff --git a/RemoteExec.Server/appsettings.Development.json b/RemoteExec.Server/appsettings.Development.json index 0c208ae..1cc9b40 100644 --- a/RemoteExec.Server/appsettings.Development.json +++ b/RemoteExec.Server/appsettings.Development.json @@ -4,5 +4,16 @@ "Default": "Information", "Microsoft.AspNetCore": "Warning" } - } + }, + "Kestrel": { + "Endpoints": { + "Http": { + "Url": "http://*:5000" + }, + "Https": { + "Url": "https://*:5001" + } + } + }, + "AllowedHosts": "*" } diff --git a/RemoteExec.Shared/ServerMetrics.cs b/RemoteExec.Shared/ServerMetrics.cs new file mode 100644 index 0000000..588f905 --- /dev/null +++ b/RemoteExec.Shared/ServerMetrics.cs @@ -0,0 +1,11 @@ +namespace RemoteExec.Shared; + +public sealed class ServerMetrics +{ + public int ActiveConnections { get; set; } + public int ActiveTasks { get; set; } + public long TotalMemoryUsage { get; set; } + public double CpuUsage { get; set; } + public DateTime Timestamp { get; set; } + public string ServerId { get; set; } = string.Empty; +} \ No newline at end of file diff --git a/RemoteExec/Program.cs b/RemoteExec/Program.cs index 00a075b..82b9126 100644 --- a/RemoteExec/Program.cs +++ b/RemoteExec/Program.cs @@ -2,18 +2,31 @@ using RemoteExec.Client; -RemoteExecutor remoteExecutor = new RemoteExecutor("https://localhost:7109/remote"); -await remoteExecutor.StartAsync(); +// Single host example +RemoteExecutor singleHostExecutor = new RemoteExecutor("https://localhost:5001/remote"); -bool success = remoteExecutor.TryExecute(Multiply, out int result, 2, 4); +singleHostExecutor.MetricsUpdated += (sender, e) => +{ + Console.WriteLine($"[METRICS UPDATE] Server: {e.Metrics.ServerId}"); + Console.WriteLine($" Active Connections: {e.Metrics.ActiveConnections}"); + Console.WriteLine($" Pending Requests: {e.Metrics.ActiveTasks}"); + Console.WriteLine($" CPU Usage: {e.Metrics.CpuUsage}%"); + Console.WriteLine($" Memory: {e.Metrics.TotalMemoryUsage / 1024 / 1024} MB"); + Console.WriteLine(); +}; +await singleHostExecutor.StartAsync(); + +bool success = singleHostExecutor.TryExecute(Multiply, out int result, 2, 4); Console.WriteLine($"Success: {success}, Result: {result}"); -result = remoteExecutor.Execute, int>(Multiply, 3, 5); - +result = singleHostExecutor.Execute>, int>(Multiply, 3, 5); Console.WriteLine($"Result: {result}"); -static int Multiply(int x, int y) +await singleHostExecutor.StopAsync(); + +static async Task Multiply(int x, int y) { + await Task.Delay(5000); return x.Multiply(y); } \ No newline at end of file