971 lines
32 KiB
C#
971 lines
32 KiB
C#
using System.Net;
|
|
using System.Net.Sockets;
|
|
using System.Text.Json;
|
|
using Microsoft.Extensions.Logging;
|
|
using QemuVmManager.Models;
|
|
|
|
namespace QemuVmManager.Core;
|
|
|
|
public class P2PNode : IDisposable
|
|
{
|
|
private readonly string _nodeId;
|
|
private readonly int _port;
|
|
private readonly IUPnPManager _upnpManager;
|
|
private readonly QemuProcessManager _qemuManager;
|
|
private readonly ILogger<P2PNode>? _logger;
|
|
|
|
private readonly Dictionary<string, NodeInfo> _knownNodes = new();
|
|
private readonly Dictionary<string, VmInstance> _localVms = new();
|
|
private readonly object _clusterLock = new();
|
|
|
|
private UdpClient? _udpClient;
|
|
private TcpListener? _tcpListener;
|
|
private CancellationTokenSource? _cancellationTokenSource;
|
|
private Task? _heartbeatTask;
|
|
private Task? _electionTask;
|
|
private Task? _discoveryTask;
|
|
|
|
private NodeRole _currentRole = NodeRole.Follower;
|
|
private long _currentTerm = 0;
|
|
private string? _votedFor;
|
|
private DateTime _lastHeartbeat = DateTime.UtcNow;
|
|
private bool _isRunning = false;
|
|
|
|
public event EventHandler<NodeRole>? RoleChanged;
|
|
public event EventHandler<NodeInfo>? NodeJoined;
|
|
public event EventHandler<NodeInfo>? NodeLeft;
|
|
public event EventHandler<VmInstance>? VmStarted;
|
|
public event EventHandler<VmInstance>? VmStopped;
|
|
|
|
public NodeInfo CurrentNode { get; private set; }
|
|
public ClusterState ClusterState { get; private set; } = new();
|
|
public bool IsMaster => _currentRole == NodeRole.Master;
|
|
|
|
public P2PNode(string nodeId, int port = 8080, IUPnPManager? upnpManager = null,
|
|
QemuProcessManager? qemuManager = null, ILogger<P2PNode>? logger = null)
|
|
{
|
|
_nodeId = nodeId;
|
|
_port = port;
|
|
_upnpManager = upnpManager ?? new UPnPManager();
|
|
_qemuManager = qemuManager ?? new QemuProcessManager();
|
|
_logger = logger;
|
|
|
|
CurrentNode = new NodeInfo
|
|
{
|
|
NodeId = _nodeId,
|
|
Hostname = Environment.MachineName,
|
|
IpAddress = GetLocalIpAddress(),
|
|
Port = _port,
|
|
Role = NodeRole.Follower,
|
|
State = NodeState.Stopped,
|
|
SystemInfo = GetSystemInfo()
|
|
};
|
|
}
|
|
|
|
public async Task StartAsync()
|
|
{
|
|
if (_isRunning)
|
|
return;
|
|
|
|
_logger?.LogInformation("Starting P2P node {NodeId} on port {Port}", _nodeId, _port);
|
|
|
|
_cancellationTokenSource = new CancellationTokenSource();
|
|
_isRunning = true;
|
|
CurrentNode.State = NodeState.Starting;
|
|
|
|
try
|
|
{
|
|
// Initialize network components
|
|
await InitializeNetworkAsync();
|
|
|
|
// Start background tasks
|
|
_heartbeatTask = Task.Run(() => HeartbeatLoopAsync(_cancellationTokenSource.Token));
|
|
_electionTask = Task.Run(() => ElectionLoopAsync(_cancellationTokenSource.Token));
|
|
_discoveryTask = Task.Run(() => DiscoveryLoopAsync(_cancellationTokenSource.Token));
|
|
|
|
CurrentNode.State = NodeState.Running;
|
|
_logger?.LogInformation("P2P node {NodeId} started successfully", _nodeId);
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
_logger?.LogError(ex, "Failed to start P2P node {NodeId}", _nodeId);
|
|
CurrentNode.State = NodeState.Error;
|
|
throw;
|
|
}
|
|
}
|
|
|
|
public async Task StopAsync()
|
|
{
|
|
if (!_isRunning)
|
|
return;
|
|
|
|
_logger?.LogInformation("Stopping P2P node {NodeId}", _nodeId);
|
|
|
|
_isRunning = false;
|
|
CurrentNode.State = NodeState.Stopping;
|
|
|
|
_cancellationTokenSource?.Cancel();
|
|
|
|
// Stop all local VMs
|
|
foreach (var vm in _localVms.Values.ToList())
|
|
{
|
|
await StopLocalVmAsync(vm.VmId);
|
|
}
|
|
|
|
// Clean up network resources
|
|
_udpClient?.Close();
|
|
_tcpListener?.Stop();
|
|
|
|
CurrentNode.State = NodeState.Stopped;
|
|
_logger?.LogInformation("P2P node {NodeId} stopped", _nodeId);
|
|
}
|
|
|
|
public async Task<VmInstance> StartVmAsync(VmConfiguration config, string? targetNodeId = null)
|
|
{
|
|
var vmId = Guid.NewGuid().ToString();
|
|
var targetNode = targetNodeId ?? _nodeId;
|
|
|
|
if (targetNode == _nodeId)
|
|
{
|
|
// Start VM locally
|
|
return await StartLocalVmAsync(vmId, config);
|
|
}
|
|
else
|
|
{
|
|
// Request remote node to start VM
|
|
return await RequestRemoteVmStartAsync(vmId, config, targetNode);
|
|
}
|
|
}
|
|
|
|
public async Task<bool> StopVmAsync(string vmId)
|
|
{
|
|
if (_localVms.ContainsKey(vmId))
|
|
{
|
|
return await StopLocalVmAsync(vmId);
|
|
}
|
|
else
|
|
{
|
|
// Find which node has this VM and request stop
|
|
var vm = ClusterState.DistributedVms.GetValueOrDefault(vmId);
|
|
if (vm != null)
|
|
{
|
|
return await RequestRemoteVmStopAsync(vmId, vm.NodeId);
|
|
}
|
|
}
|
|
|
|
return false;
|
|
}
|
|
|
|
public async Task<PortForwardingResponse> RequestPortForwardingAsync(string vmId, int privatePort, int? publicPort = null)
|
|
{
|
|
if (!IsMaster)
|
|
{
|
|
throw new InvalidOperationException("Only the master node can request port forwarding");
|
|
}
|
|
|
|
var vm = ClusterState.DistributedVms.GetValueOrDefault(vmId);
|
|
if (vm == null)
|
|
{
|
|
return new PortForwardingResponse
|
|
{
|
|
VmId = vmId,
|
|
Success = false,
|
|
ErrorMessage = "VM not found"
|
|
};
|
|
}
|
|
|
|
try
|
|
{
|
|
var actualPublicPort = publicPort ?? await GetAvailablePortAsync();
|
|
var success = await _upnpManager.AddPortMappingAsync(actualPublicPort, privatePort, $"QEMU VM {vmId}");
|
|
|
|
if (success)
|
|
{
|
|
var externalIp = await _upnpManager.GetExternalIpAddressAsync();
|
|
vm.PublicEndpoint = new NetworkEndpoint
|
|
{
|
|
PublicIp = externalIp ?? IPAddress.Any,
|
|
PublicPort = actualPublicPort,
|
|
PrivateIp = vm.Configuration.Network.Interfaces.FirstOrDefault()?.Mac != null ?
|
|
IPAddress.Parse("192.168.1.100") : IPAddress.Any, // Simplified
|
|
PrivatePort = privatePort,
|
|
Protocol = "TCP",
|
|
Description = $"QEMU VM {vmId}"
|
|
};
|
|
|
|
return new PortForwardingResponse
|
|
{
|
|
VmId = vmId,
|
|
Success = true,
|
|
PublicIp = externalIp,
|
|
PublicPort = actualPublicPort
|
|
};
|
|
}
|
|
|
|
return new PortForwardingResponse
|
|
{
|
|
VmId = vmId,
|
|
Success = false,
|
|
ErrorMessage = "Failed to create port mapping"
|
|
};
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
_logger?.LogError(ex, "Failed to request port forwarding for VM {VmId}", vmId);
|
|
return new PortForwardingResponse
|
|
{
|
|
VmId = vmId,
|
|
Success = false,
|
|
ErrorMessage = ex.Message
|
|
};
|
|
}
|
|
}
|
|
|
|
public async Task<VmMigrationResponse> MigrateVmAsync(string vmId, string targetNodeId)
|
|
{
|
|
var vm = ClusterState.DistributedVms.GetValueOrDefault(vmId);
|
|
if (vm == null)
|
|
{
|
|
return new VmMigrationResponse
|
|
{
|
|
VmId = vmId,
|
|
Success = false,
|
|
ErrorMessage = "VM not found"
|
|
};
|
|
}
|
|
|
|
if (vm.NodeId == targetNodeId)
|
|
{
|
|
return new VmMigrationResponse
|
|
{
|
|
VmId = vmId,
|
|
Success = true,
|
|
ErrorMessage = "VM is already on target node"
|
|
};
|
|
}
|
|
|
|
try
|
|
{
|
|
// Stop VM on source node
|
|
await StopVmAsync(vmId);
|
|
|
|
// Start VM on target node
|
|
var newVm = await StartVmAsync(vm.Configuration, targetNodeId);
|
|
|
|
return new VmMigrationResponse
|
|
{
|
|
VmId = vmId,
|
|
Success = true
|
|
};
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
_logger?.LogError(ex, "Failed to migrate VM {VmId} to node {TargetNodeId}", vmId, targetNodeId);
|
|
return new VmMigrationResponse
|
|
{
|
|
VmId = vmId,
|
|
Success = false,
|
|
ErrorMessage = ex.Message
|
|
};
|
|
}
|
|
}
|
|
|
|
private async Task InitializeNetworkAsync()
|
|
{
|
|
// Initialize UDP client for discovery and heartbeats
|
|
_udpClient = new UdpClient();
|
|
_udpClient.Client.SetSocketOption(SocketOptionLevel.Socket, SocketOptionName.ReuseAddress, true);
|
|
_udpClient.Client.SetSocketOption(SocketOptionLevel.Socket, SocketOptionName.Broadcast, true);
|
|
_udpClient.Client.Bind(new IPEndPoint(IPAddress.Any, _port));
|
|
|
|
// Initialize TCP listener for direct communication on a different port
|
|
var tcpPort = _port + 1; // Use next port for TCP
|
|
_tcpListener = new TcpListener(IPAddress.Any, tcpPort);
|
|
_tcpListener.Start();
|
|
|
|
_logger?.LogInformation("Network initialized - UDP: {UdpPort}, TCP: {TcpPort}", _port, tcpPort);
|
|
|
|
// Start listening for incoming connections
|
|
_ = Task.Run(() => ListenForConnectionsAsync(_cancellationTokenSource!.Token));
|
|
|
|
// Start listening for UDP messages
|
|
_ = Task.Run(() => ListenForUdpMessagesAsync(_cancellationTokenSource!.Token));
|
|
}
|
|
|
|
private async Task HeartbeatLoopAsync(CancellationToken cancellationToken)
|
|
{
|
|
while (!cancellationToken.IsCancellationRequested)
|
|
{
|
|
try
|
|
{
|
|
var heartbeat = new HeartbeatMessage
|
|
{
|
|
NodeId = _nodeId,
|
|
Role = _currentRole,
|
|
Term = _currentTerm,
|
|
Timestamp = DateTime.UtcNow,
|
|
Metadata = new Dictionary<string, object>
|
|
{
|
|
["running_vms"] = _localVms.Count,
|
|
["cpu_usage"] = await GetCpuUsageAsync(),
|
|
["memory_usage"] = await GetMemoryUsageAsync()
|
|
}
|
|
};
|
|
|
|
await BroadcastHeartbeatAsync(heartbeat);
|
|
|
|
// Check for stale nodes
|
|
await CleanupStaleNodesAsync();
|
|
|
|
await Task.Delay(5000, cancellationToken); // Send heartbeat every 5 seconds
|
|
}
|
|
catch (OperationCanceledException)
|
|
{
|
|
break;
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
_logger?.LogError(ex, "Error in heartbeat loop");
|
|
await Task.Delay(1000, cancellationToken);
|
|
}
|
|
}
|
|
}
|
|
|
|
private async Task ElectionLoopAsync(CancellationToken cancellationToken)
|
|
{
|
|
var electionTimeout = TimeSpan.FromSeconds(10 + Random.Shared.Next(10)); // 10-20 seconds
|
|
|
|
while (!cancellationToken.IsCancellationRequested)
|
|
{
|
|
try
|
|
{
|
|
if (_currentRole == NodeRole.Follower)
|
|
{
|
|
// Wait for heartbeat from master
|
|
var timeout = DateTime.UtcNow.Add(electionTimeout);
|
|
while (DateTime.UtcNow < timeout && !cancellationToken.IsCancellationRequested)
|
|
{
|
|
if (_lastHeartbeat > DateTime.UtcNow.AddSeconds(-5))
|
|
{
|
|
await Task.Delay(1000, cancellationToken);
|
|
continue;
|
|
}
|
|
break;
|
|
}
|
|
|
|
if (!cancellationToken.IsCancellationRequested)
|
|
{
|
|
// No heartbeat received, start election
|
|
await StartElectionAsync();
|
|
}
|
|
}
|
|
else if (_currentRole == NodeRole.Candidate)
|
|
{
|
|
// Wait for election results
|
|
await Task.Delay(5000, cancellationToken);
|
|
}
|
|
else if (_currentRole == NodeRole.Master)
|
|
{
|
|
// Master continues to send heartbeats
|
|
await Task.Delay(1000, cancellationToken);
|
|
}
|
|
}
|
|
catch (OperationCanceledException)
|
|
{
|
|
break;
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
_logger?.LogError(ex, "Error in election loop");
|
|
await Task.Delay(1000, cancellationToken);
|
|
}
|
|
}
|
|
}
|
|
|
|
private async Task DiscoveryLoopAsync(CancellationToken cancellationToken)
|
|
{
|
|
while (!cancellationToken.IsCancellationRequested)
|
|
{
|
|
try
|
|
{
|
|
// Send discovery message
|
|
var discoveryMessage = JsonSerializer.Serialize(new
|
|
{
|
|
type = "discovery",
|
|
nodeId = _nodeId,
|
|
timestamp = DateTime.UtcNow
|
|
});
|
|
|
|
var data = System.Text.Encoding.UTF8.GetBytes(discoveryMessage);
|
|
|
|
// Try multiple broadcast addresses
|
|
var broadcastAddresses = new[]
|
|
{
|
|
IPAddress.Broadcast,
|
|
IPAddress.Parse("255.255.255.255"),
|
|
GetLocalBroadcastAddress()
|
|
};
|
|
|
|
foreach (var broadcastAddr in broadcastAddresses)
|
|
{
|
|
try
|
|
{
|
|
await _udpClient!.SendAsync(data, data.Length, new IPEndPoint(broadcastAddr, _port));
|
|
_logger?.LogDebug("Sent discovery message to {BroadcastAddress}", broadcastAddr);
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
_logger?.LogDebug(ex, "Failed to send discovery to {BroadcastAddress}", broadcastAddr);
|
|
}
|
|
}
|
|
|
|
_logger?.LogInformation("Discovery cycle completed. Known nodes: {NodeCount}", _knownNodes.Count);
|
|
await Task.Delay(30000, cancellationToken); // Send discovery every 30 seconds
|
|
}
|
|
catch (OperationCanceledException)
|
|
{
|
|
break;
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
_logger?.LogError(ex, "Error in discovery loop");
|
|
await Task.Delay(5000, cancellationToken);
|
|
}
|
|
}
|
|
}
|
|
|
|
private async Task StartElectionAsync()
|
|
{
|
|
_logger?.LogInformation("Starting election for term {Term}", _currentTerm + 1);
|
|
|
|
_currentTerm++;
|
|
_currentRole = NodeRole.Candidate;
|
|
_votedFor = _nodeId;
|
|
|
|
var electionRequest = new ElectionRequest
|
|
{
|
|
CandidateId = _nodeId,
|
|
Term = _currentTerm,
|
|
Timestamp = DateTime.UtcNow
|
|
};
|
|
|
|
var votes = 1; // Vote for self
|
|
var requiredVotes = (_knownNodes.Count + 1) / 2 + 1; // Majority
|
|
|
|
// Request votes from other nodes
|
|
foreach (var node in _knownNodes.Values)
|
|
{
|
|
try
|
|
{
|
|
var response = await RequestVoteAsync(node, electionRequest);
|
|
if (response.VoteGranted && response.Term == _currentTerm)
|
|
{
|
|
votes++;
|
|
}
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
_logger?.LogWarning(ex, "Failed to request vote from node {NodeId}", node.NodeId);
|
|
}
|
|
}
|
|
|
|
if (votes >= requiredVotes)
|
|
{
|
|
await BecomeMasterAsync();
|
|
}
|
|
else
|
|
{
|
|
_currentRole = NodeRole.Follower;
|
|
}
|
|
}
|
|
|
|
private async Task BecomeMasterAsync()
|
|
{
|
|
_logger?.LogInformation("Becoming master for term {Term}", _currentTerm);
|
|
|
|
_currentRole = NodeRole.Master;
|
|
CurrentNode.Role = NodeRole.Master;
|
|
|
|
// Update cluster state
|
|
lock (_clusterLock)
|
|
{
|
|
ClusterState.MasterNode = CurrentNode;
|
|
ClusterState.LastUpdated = DateTime.UtcNow;
|
|
}
|
|
|
|
RoleChanged?.Invoke(this, NodeRole.Master);
|
|
|
|
// Start master-specific tasks
|
|
if (IsMaster)
|
|
{
|
|
await StartMasterTasksAsync();
|
|
}
|
|
}
|
|
|
|
private async Task StartMasterTasksAsync()
|
|
{
|
|
// Master-specific initialization
|
|
_logger?.LogInformation("Starting master tasks");
|
|
|
|
// Check UPnP availability
|
|
var upnpAvailable = await _upnpManager.IsUPnPAvailableAsync();
|
|
_logger?.LogInformation("UPnP available: {Available}", upnpAvailable);
|
|
}
|
|
|
|
private async Task<VmInstance> StartLocalVmAsync(string vmId, VmConfiguration config)
|
|
{
|
|
try
|
|
{
|
|
var success = await _qemuManager.StartVmAsync(config);
|
|
if (!success)
|
|
{
|
|
throw new InvalidOperationException("Failed to start QEMU VM");
|
|
}
|
|
|
|
var vmInstance = new VmInstance
|
|
{
|
|
VmId = vmId,
|
|
VmName = config.Name,
|
|
NodeId = _nodeId,
|
|
State = VmState.Running,
|
|
StartedAt = DateTime.UtcNow,
|
|
Configuration = config
|
|
};
|
|
|
|
_localVms[vmId] = vmInstance;
|
|
|
|
lock (_clusterLock)
|
|
{
|
|
ClusterState.DistributedVms[vmId] = vmInstance;
|
|
ClusterState.LastUpdated = DateTime.UtcNow;
|
|
}
|
|
|
|
VmStarted?.Invoke(this, vmInstance);
|
|
_logger?.LogInformation("Started local VM {VmId} ({VmName})", vmId, config.Name);
|
|
|
|
return vmInstance;
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
_logger?.LogError(ex, "Failed to start local VM {VmId}", vmId);
|
|
throw;
|
|
}
|
|
}
|
|
|
|
private async Task<bool> StopLocalVmAsync(string vmId)
|
|
{
|
|
if (!_localVms.TryGetValue(vmId, out var vm))
|
|
{
|
|
return false;
|
|
}
|
|
|
|
try
|
|
{
|
|
var success = await _qemuManager.StopVmAsync(vm.VmName, true);
|
|
if (success)
|
|
{
|
|
_localVms.Remove(vmId);
|
|
|
|
lock (_clusterLock)
|
|
{
|
|
ClusterState.DistributedVms.Remove(vmId);
|
|
ClusterState.LastUpdated = DateTime.UtcNow;
|
|
}
|
|
|
|
VmStopped?.Invoke(this, vm);
|
|
_logger?.LogInformation("Stopped local VM {VmId}", vmId);
|
|
}
|
|
|
|
return success;
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
_logger?.LogError(ex, "Failed to stop local VM {VmId}", vmId);
|
|
return false;
|
|
}
|
|
}
|
|
|
|
private async Task<VmInstance> RequestRemoteVmStartAsync(string vmId, VmConfiguration config, string targetNodeId)
|
|
{
|
|
// This would implement RPC to remote node
|
|
throw new NotImplementedException("Remote VM start not yet implemented");
|
|
}
|
|
|
|
private async Task<bool> RequestRemoteVmStopAsync(string vmId, string targetNodeId)
|
|
{
|
|
// This would implement RPC to remote node
|
|
throw new NotImplementedException("Remote VM stop not yet implemented");
|
|
}
|
|
|
|
private async Task<ElectionResponse> RequestVoteAsync(NodeInfo node, ElectionRequest request)
|
|
{
|
|
// This would implement RPC to request vote
|
|
throw new NotImplementedException("Vote request not yet implemented");
|
|
}
|
|
|
|
private async Task BroadcastHeartbeatAsync(HeartbeatMessage heartbeat)
|
|
{
|
|
var message = JsonSerializer.Serialize(heartbeat);
|
|
var data = System.Text.Encoding.UTF8.GetBytes(message);
|
|
|
|
// Send to known nodes
|
|
foreach (var node in _knownNodes.Values)
|
|
{
|
|
try
|
|
{
|
|
await _udpClient!.SendAsync(data, data.Length, new IPEndPoint(node.IpAddress, node.Port));
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
_logger?.LogWarning(ex, "Failed to send heartbeat to node {NodeId}", node.NodeId);
|
|
}
|
|
}
|
|
|
|
// Also broadcast to discover new nodes
|
|
var broadcastAddresses = new[]
|
|
{
|
|
IPAddress.Broadcast,
|
|
IPAddress.Parse("255.255.255.255"),
|
|
GetLocalBroadcastAddress()
|
|
};
|
|
|
|
foreach (var broadcastAddr in broadcastAddresses)
|
|
{
|
|
try
|
|
{
|
|
await _udpClient!.SendAsync(data, data.Length, new IPEndPoint(broadcastAddr, _port));
|
|
_logger?.LogDebug("Broadcasted heartbeat to {BroadcastAddress}", broadcastAddr);
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
_logger?.LogDebug(ex, "Failed to broadcast heartbeat to {BroadcastAddress}", broadcastAddr);
|
|
}
|
|
}
|
|
}
|
|
|
|
private async Task CleanupStaleNodesAsync()
|
|
{
|
|
var staleThreshold = DateTime.UtcNow.AddSeconds(-30);
|
|
var staleNodes = _knownNodes.Values.Where(n => n.LastSeen < staleThreshold).ToList();
|
|
|
|
foreach (var node in staleNodes)
|
|
{
|
|
_knownNodes.Remove(node.NodeId);
|
|
NodeLeft?.Invoke(this, node);
|
|
_logger?.LogInformation("Removed stale node {NodeId}", node.NodeId);
|
|
}
|
|
}
|
|
|
|
private async Task ListenForConnectionsAsync(CancellationToken cancellationToken)
|
|
{
|
|
while (!cancellationToken.IsCancellationRequested)
|
|
{
|
|
try
|
|
{
|
|
var client = await _tcpListener!.AcceptTcpClientAsync(cancellationToken);
|
|
_ = Task.Run(() => HandleClientAsync(client, cancellationToken));
|
|
}
|
|
catch (OperationCanceledException)
|
|
{
|
|
break;
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
_logger?.LogError(ex, "Error accepting client connection");
|
|
}
|
|
}
|
|
}
|
|
|
|
private async Task ListenForUdpMessagesAsync(CancellationToken cancellationToken)
|
|
{
|
|
while (!cancellationToken.IsCancellationRequested)
|
|
{
|
|
try
|
|
{
|
|
var result = await _udpClient!.ReceiveAsync(cancellationToken);
|
|
var message = System.Text.Encoding.UTF8.GetString(result.Buffer);
|
|
|
|
_logger?.LogDebug("Received UDP message from {Endpoint}: {Message}",
|
|
result.RemoteEndPoint, message);
|
|
|
|
await ProcessUdpMessageAsync(message, result.RemoteEndPoint);
|
|
}
|
|
catch (OperationCanceledException)
|
|
{
|
|
break;
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
_logger?.LogError(ex, "Error receiving UDP message");
|
|
}
|
|
}
|
|
}
|
|
|
|
private async Task ProcessUdpMessageAsync(string message, IPEndPoint remoteEndpoint)
|
|
{
|
|
try
|
|
{
|
|
var data = JsonSerializer.Deserialize<JsonElement>(message);
|
|
var messageType = data.GetProperty("type").GetString();
|
|
|
|
switch (messageType)
|
|
{
|
|
case "heartbeat":
|
|
await ProcessHeartbeatAsync(data);
|
|
break;
|
|
case "discovery":
|
|
await ProcessDiscoveryAsync(data);
|
|
break;
|
|
default:
|
|
_logger?.LogWarning("Unknown UDP message type: {MessageType}", messageType);
|
|
break;
|
|
}
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
_logger?.LogError(ex, "Error processing UDP message: {Message}", message);
|
|
}
|
|
}
|
|
|
|
private async Task HandleClientAsync(TcpClient client, CancellationToken cancellationToken)
|
|
{
|
|
try
|
|
{
|
|
using var stream = client.GetStream();
|
|
using var reader = new StreamReader(stream);
|
|
using var writer = new StreamWriter(stream);
|
|
|
|
var message = await reader.ReadLineAsync();
|
|
if (message != null)
|
|
{
|
|
await ProcessMessageAsync(message, writer);
|
|
}
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
_logger?.LogError(ex, "Error handling client connection");
|
|
}
|
|
finally
|
|
{
|
|
client.Close();
|
|
}
|
|
}
|
|
|
|
private async Task ProcessMessageAsync(string message, StreamWriter writer)
|
|
{
|
|
try
|
|
{
|
|
var data = JsonSerializer.Deserialize<JsonElement>(message);
|
|
var messageType = data.GetProperty("type").GetString();
|
|
|
|
switch (messageType)
|
|
{
|
|
case "heartbeat":
|
|
await ProcessHeartbeatAsync(data);
|
|
break;
|
|
case "discovery":
|
|
await ProcessDiscoveryAsync(data);
|
|
break;
|
|
case "election_request":
|
|
await ProcessElectionRequestAsync(data, writer);
|
|
break;
|
|
case "election_response":
|
|
await ProcessElectionResponseAsync(data);
|
|
break;
|
|
default:
|
|
_logger?.LogWarning("Unknown message type: {MessageType}", messageType);
|
|
break;
|
|
}
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
_logger?.LogError(ex, "Error processing message: {Message}", message);
|
|
}
|
|
}
|
|
|
|
private async Task ProcessHeartbeatAsync(JsonElement data)
|
|
{
|
|
var nodeId = data.GetProperty("nodeId").GetString()!;
|
|
var role = Enum.Parse<NodeRole>(data.GetProperty("role").GetString()!);
|
|
var term = data.GetProperty("term").GetInt64();
|
|
var timestamp = data.GetProperty("timestamp").GetDateTime();
|
|
|
|
if (term > _currentTerm)
|
|
{
|
|
_currentTerm = term;
|
|
_currentRole = NodeRole.Follower;
|
|
_votedFor = null;
|
|
}
|
|
|
|
if (role == NodeRole.Master && term >= _currentTerm)
|
|
{
|
|
_currentRole = NodeRole.Follower;
|
|
_lastHeartbeat = timestamp;
|
|
}
|
|
|
|
// Update node info
|
|
if (_knownNodes.TryGetValue(nodeId, out var node))
|
|
{
|
|
node.LastSeen = DateTime.UtcNow;
|
|
node.Role = role;
|
|
}
|
|
}
|
|
|
|
private async Task ProcessDiscoveryAsync(JsonElement data)
|
|
{
|
|
var nodeId = data.GetProperty("nodeId").GetString()!;
|
|
|
|
if (nodeId != _nodeId && !_knownNodes.ContainsKey(nodeId))
|
|
{
|
|
var newNode = new NodeInfo
|
|
{
|
|
NodeId = nodeId,
|
|
Hostname = Environment.MachineName, // We'll get this from heartbeat
|
|
IpAddress = IPAddress.Any, // We'll get this from heartbeat
|
|
Port = _port,
|
|
Role = NodeRole.Follower,
|
|
State = NodeState.Running,
|
|
LastSeen = DateTime.UtcNow,
|
|
SystemInfo = new SystemInfo() // We'll get this from heartbeat
|
|
};
|
|
|
|
_knownNodes[nodeId] = newNode;
|
|
NodeJoined?.Invoke(this, newNode);
|
|
_logger?.LogInformation("Discovered new node {NodeId}", nodeId);
|
|
}
|
|
else if (nodeId != _nodeId && _knownNodes.ContainsKey(nodeId))
|
|
{
|
|
// Update last seen for existing node
|
|
_knownNodes[nodeId].LastSeen = DateTime.UtcNow;
|
|
}
|
|
}
|
|
|
|
private async Task ProcessElectionRequestAsync(JsonElement data, StreamWriter writer)
|
|
{
|
|
var candidateId = data.GetProperty("candidateId").GetString()!;
|
|
var term = data.GetProperty("term").GetInt64();
|
|
|
|
var response = new ElectionResponse
|
|
{
|
|
VoterId = _nodeId,
|
|
Term = _currentTerm,
|
|
VoteGranted = false
|
|
};
|
|
|
|
if (term > _currentTerm)
|
|
{
|
|
_currentTerm = term;
|
|
_currentRole = NodeRole.Follower;
|
|
_votedFor = null;
|
|
}
|
|
|
|
if (term == _currentTerm && (_votedFor == null || _votedFor == candidateId))
|
|
{
|
|
_votedFor = candidateId;
|
|
response.VoteGranted = true;
|
|
}
|
|
|
|
var responseJson = JsonSerializer.Serialize(response);
|
|
await writer.WriteLineAsync(responseJson);
|
|
await writer.FlushAsync();
|
|
}
|
|
|
|
private async Task ProcessElectionResponseAsync(JsonElement data)
|
|
{
|
|
// Handle election response
|
|
var voterId = data.GetProperty("voterId").GetString()!;
|
|
var voteGranted = data.GetProperty("voteGranted").GetBoolean();
|
|
var term = data.GetProperty("term").GetInt64();
|
|
|
|
if (voteGranted && term == _currentTerm && _currentRole == NodeRole.Candidate)
|
|
{
|
|
// Count votes and potentially become master
|
|
// This is simplified - in a real implementation you'd track votes
|
|
}
|
|
}
|
|
|
|
private async Task<int> GetAvailablePortAsync()
|
|
{
|
|
// Find an available port for UPnP mapping
|
|
using var listener = new TcpListener(IPAddress.Any, 0);
|
|
listener.Start();
|
|
var port = ((IPEndPoint)listener.LocalEndpoint).Port;
|
|
listener.Stop();
|
|
return port;
|
|
}
|
|
|
|
private async Task<double> GetCpuUsageAsync()
|
|
{
|
|
// Simplified CPU usage calculation
|
|
return Environment.ProcessorCount * 0.1; // 10% per core
|
|
}
|
|
|
|
private async Task<double> GetMemoryUsageAsync()
|
|
{
|
|
// Simplified memory usage calculation
|
|
return GC.GetTotalMemory(false) / (1024.0 * 1024.0); // MB
|
|
}
|
|
|
|
private IPAddress GetLocalIpAddress()
|
|
{
|
|
try
|
|
{
|
|
var host = Dns.GetHostEntry(Dns.GetHostName());
|
|
return host.AddressList.FirstOrDefault(ip =>
|
|
ip.AddressFamily == AddressFamily.InterNetwork &&
|
|
!IPAddress.IsLoopback(ip)) ?? IPAddress.Any;
|
|
}
|
|
catch
|
|
{
|
|
return IPAddress.Any;
|
|
}
|
|
}
|
|
|
|
private IPAddress GetLocalBroadcastAddress()
|
|
{
|
|
try
|
|
{
|
|
var localIp = GetLocalIpAddress();
|
|
if (localIp != null && localIp != IPAddress.Any)
|
|
{
|
|
var ipBytes = localIp.GetAddressBytes();
|
|
// Set all host bits to 1 for broadcast
|
|
for (int i = 3; i >= 0; i--)
|
|
{
|
|
ipBytes[i] = 255;
|
|
}
|
|
return new IPAddress(ipBytes);
|
|
}
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
_logger?.LogWarning(ex, "Failed to get local broadcast address");
|
|
}
|
|
return IPAddress.Broadcast;
|
|
}
|
|
|
|
private SystemInfo GetSystemInfo()
|
|
{
|
|
return new SystemInfo
|
|
{
|
|
OsName = Environment.OSVersion.Platform.ToString(),
|
|
OsVersion = Environment.OSVersion.VersionString,
|
|
Architecture = Environment.GetEnvironmentVariable("PROCESSOR_ARCHITECTURE") ?? "Unknown",
|
|
CpuCores = Environment.ProcessorCount,
|
|
TotalMemory = GC.GetGCMemoryInfo().TotalAvailableMemoryBytes / (1024 * 1024), // MB
|
|
AvailableMemory = GC.GetTotalMemory(false) / (1024 * 1024), // MB
|
|
AvailableVirtualization = _qemuManager.GetAvailableVirtualization(),
|
|
QemuInstalled = _qemuManager.IsQemuInstalled(),
|
|
QemuVersion = _qemuManager.GetQemuVersion()
|
|
};
|
|
}
|
|
|
|
public void Dispose()
|
|
{
|
|
StopAsync().Wait();
|
|
_udpClient?.Dispose();
|
|
_tcpListener?.Stop();
|
|
_cancellationTokenSource?.Dispose();
|
|
}
|
|
}
|