<?phpnamespace App\Model;use App\Xapi\FakeXapi;use App\Xapi\StreamTransport;use App\Xapi\TlsStream;use App\Xapi\Transport;use App\Xapi\XapiClient;use App\Xapi\XapiError;use JumPHP\Async\Task;use JumPHP\Kernel\Engine;use JumPHP\Kernel\Platform;/** * One connected server (a pool, or a host alone): its XAPI client, its cache filled by event.from, the actions run on it. * State: disconnected, connecting, connected, failed (with the reason). A pool member answering HOST_IS_SLAVE is left for its * master, as XenCenter does. */final class Connection{ public const DISCONNECTED = 'disconnected'; public const CONNECTING = 'connecting'; public const CONNECTED = 'connected'; public const FAILED = 'failed'; /** the classes event.from follows */ public const CLASSES = ['pool', 'host', 'host_metrics', 'vm', 'vm_metrics', 'vm_guest_metrics', 'sr', 'pbd', 'vdi', 'vbd', 'vif', 'network', 'pif', 'pif_metrics', 'task', 'message', 'console', 'bond']; public string $state = self::DISCONNECTED; public string $error = ''; public string $phase = ''; public Cache $cache; public ?XapiClient $client = null; private string $token = ''; private float $retryIn = -1; private int $generation = 0; /** @var array<string, array{0: Task, 1: LogEntry}> task ref => what waits for it */ private array $tracked = []; private ?FakeXapi $fake = null; /** @var list<\App\Xapi\HttpStreamTransfer> big HTTP transfers running (XVA export / import, src/Feature/Xva): polled each frame */ public array $transfers = []; public function __construct(public ServerProfile $profile, private Session $session) { $this->cache = new Cache(); } /** The name in the tree: the pool's, or the master host's, or the address while connecting. */ public function name(): string { $pool = $this->cache->poolRef(); if ($this->state === self::CONNECTED && $pool !== null) { $name = trim((string)$this->cache->field($pool, 'name_label')); if ($name !== '') { return $name; } $master = $this->cache->master(); if ($master !== null && $this->cache->get($master) !== null) { return $this->cache->name($master); } } return $this->profile->friendlyName !== '' ? $this->profile->friendlyName : $this->profile->host; } public function isConnected(): bool { return $this->state === self::CONNECTED; } public function fake(): ?FakeXapi { return $this->fake; } /** The roads to the other hosts of the pool (their performance figures): src/Feature/Pools/HostLinks.php. */ private ?\App\Feature\Pools\HostLinks $links = null; public function hostLinks(): \App\Feature\Pools\HostLinks { return $this->links ??= new \App\Feature\Pools\HostLinks(); } // connecting -------------------------------------------------------------------------------------------------------- public function connect(): Task { $done = new Task(); $this->disconnect(false); $generation = ++$this->generation; $this->state = self::CONNECTING; $this->error = ''; $this->phase = 'Connecting to ' . $this->profile->host . '…'; $this->cache = new Cache(); $this->token = ''; // a local address from a browser without the local network: the server's bridge would refuse it with a cryptic error if (($problem = LocalNetwork::check($this->profile->host, $this->profile->method, false)) !== null) { $this->failed($problem, $done); return $done; } try { $this->client = new XapiClient($this->transport()); } catch (\Throwable $e) { $this->failed($e->getMessage(), $done); return $done; } $this->phase = 'Logging in…'; $this->client->login($this->profile->user, $this->profile->password)->then( function () use ($generation, $done): void { if ($generation !== $this->generation) { return; } $this->phase = 'Synchronizing…'; $this->pollEvents($generation, $done); }, function (\Throwable $e) use ($generation, $done): void { if ($generation !== $this->generation) { return; } if ($e instanceof XapiError && $e->code() === 'HOST_IS_SLAVE' && isset($e->failure[1])) { // a pool member: its master is the one to talk to $this->session->log('Redirected to the pool master ' . $e->failure[1], $this->profile->host); $this->profile->host = (string)$e->failure[1]; $this->connect()->then(fn () => $done->settle(true), fn (\Throwable $e) => $done->settle(null, $e)); return; } $this->failed($e->getMessage(), $done, $e); }, ); return $done; } private function failed(string $message, Task $done, ?\Throwable $e = null): void { $this->state = self::FAILED; $this->error = $message; $this->phase = ''; $this->client?->transport->close(); $this->client = null; $done->settle(null, $e ?? new \RuntimeException($message)); } public function disconnect(bool $logout = true): void { $this->generation++; if ($this->client !== null) { if ($logout) { $this->client->logout(); } $client = $this->client; // the logout leaves before the streams close: the transport keeps being polled meanwhile (TLS writes in poll()) $this->closing[] = $client->transport; Engine::get()->gameLoop->addTimer(0.5, function () use ($client): void { $client->transport->close(); $this->closing = array_values(array_filter($this->closing, fn ($t) => $t !== $client->transport)); }); } foreach ($this->tracked as [$task, $entry]) { $entry->status = LogEntry::FAILED; $entry->detail = 'Disconnected'; $task->settle(null, new XapiError(['TRANSPORT', 'Disconnected'])); } $this->tracked = []; foreach ($this->transfers as $transfer) { $transfer->cancel('Disconnected.'); } $this->transfers = []; foreach ($this->polled as $stream) { $stream->close(); // the consoles' TLS streams } $this->polled = []; $this->links?->close(); $this->client = null; $this->state = self::DISCONNECTED; $this->phase = ''; $this->retryIn = -1; } private function transport(): Transport { $p = $this->profile; $engine = Engine::get(); return match ($p->method) { 'demo' => $this->fake = FakeXapi::open($p->host), 'https' => TlsStream::available() ? StreamTransport::tls($p->host, $p->port()) : throw new \RuntimeException('HTTPS is not possible here (no TLS sockets in a browser): use HTTP or the SSH tunnel.'), 'ssh' => StreamTransport::tcp($engine->ssh()->via(array_filter(['sshHost' => $p->host, 'sshPort' => $p->sshPort, 'sshUser' => $p->sshUser, 'sshKey' => $p->sshKey !== '' ? $p->sshKey : null, 'sshPassword' => $p->sshKey === '' ? $p->password : null, 'sshKeys' => $engine->sshKeyring()->all() ?: null], fn ($v) => $v !== null && $v !== '')), '127.0.0.1', $p->port(), "SSH tunnel {$p->sshUser}@{$p->host}" . (Platform::isWasm() ? ' (OpenSSH in the browser)' : '')), default => StreamTransport::tcp($engine->tcp(), $p->host, $p->port(), "HTTP {$p->host}:{$p->port()}" . (Platform::isWasm() ? ' (' . (LocalNetwork::inUse() ? 'JumPHP runtime, your network' : 'server bridge') . ')' : '')), }; } // consoles --------------------------------------------------------------------------------------------------------- /** @var list<TlsStream> streams of consoles this connection polls itself */ private array $polled = []; /** * The VNC console of a running VM: CONNECT to the console's location (the host the VM runs on), through the same road as the * API (a socket or the JumPHP runtime, TLS, or the SSH tunnel: `ssh -W <host>:80` through the pool master). */ /** * The addresses a console can be reached at, in the order to try. XAPI's location names the management address of the host the VM * runs on, which the user's computer may not reach: a pool managed through a VPN or a NAT is reached at other addresses than the * ones the hosts know (here: host01.example.prod -> 192.168.x.11 for a master whose XAPI name is host01.example.net, address a public * IP). And the master does not hand the console of another host's VM over (Console.Failure). So: * 1. the VM runs on the master: the address the user typed (it works: the API uses it); * 2. the user's naming applied to the VM's host: typed "host01.example.prod" for the master "host01…" -> "host02.example.prod"; * 3. XAPI's location; * 4. the other IP addresses of the VM's host (its other networks); * 5. the address the user typed. * * @return list<string> */ public function consoleTargets(string $vm): array { $location = (string)$this->cache->field($this->rfbConsole($vm) ?? '', 'location', ''); $at = trim((string)(parse_url($location, PHP_URL_HOST) ?: ''), '[]'); $typed = $this->profile->host; $master = (string)$this->cache->master(); $resident = (string)$this->cache->field($vm, 'resident_on', ''); if ($resident === '' || $this->cache->get($resident) === null) { foreach ($this->cache->hosts() as $ref => $h) { if ($at !== '' && $h['address'] === $at) { $resident = $ref; } } } // 0. the address the user set for that host (Connection Addresses…) $targets = [\App\Feature\Pools\HostAddresses::userSet($this, $resident)]; if ($resident === $master || ($at !== '' && $at === (string)$this->cache->field($master, 'address', ''))) { $targets[] = $typed; } $targets[] = self::sameNaming($typed, $this->shortNames($master), $this->shortNames($resident)); $targets[] = $at; foreach ($this->cache->pifsOf($resident) as $pif) { $targets[] = (string)($pif['IP'] ?? ''); } $targets[] = $typed; return array_values(array_unique(array_filter($targets, fn ($h) => $h !== null && $h !== ''))); } /** @return list<string> the short names of a host (its XAPI name and hostname, before the first dot), lower case */ private function shortNames(string $host): array { $r = $this->cache->get($host) ?? []; $out = []; foreach ([(string)($r['name_label'] ?? ''), (string)($r['hostname'] ?? '')] as $name) { if ($name !== '' && !Address::isIp($name)) { $out[] = strtolower(explode('.', $name)[0]); } } return array_values(array_unique($out)); } /** * The name the user would give another host: the typed name's first label is the master's short name -> the other host's short * name with the same domain. sameNaming('host01.example.prod', ['host01'], ['host02']) = 'host02.example.prod'. * * @param list<string> $master @param list<string> $other */ public static function sameNaming(string $typed, array $master, array $other): ?string { if ($typed === '' || Address::isIp($typed) || $other === []) { return null; } $parts = explode('.', strtolower($typed), 2); if (!in_array($parts[0], $master, true) || in_array($other[0], $master, true)) { return null; } return $other[0] . (isset($parts[1]) ? '.' . $parts[1] : ''); } private function rfbConsole(string $vm): ?string { foreach ((array)$this->cache->field($vm, 'consoles', []) as $ref) { if (($this->cache->get((string)$ref)['protocol'] ?? '') === 'rfb') { return (string)$ref; } } return null; } /** The console of a VM through one of consoleTargets() ($attempt: its index). */ public function openConsole(string $vm, int $attempt = 0): \App\Vnc\RfbClient { $console = null; foreach ((array)$this->cache->field($vm, 'consoles', []) as $ref) { if (($this->cache->get((string)$ref)['protocol'] ?? '') === 'rfb') { $console = (string)$ref; } } if ($console === null || $this->client?->session === null) { throw new \RuntimeException('This VM has no graphical console.'); } $location = (string)$this->cache->field($console, 'location', ''); $host = $this->consoleTargets($vm)[$attempt] ?? throw new \RuntimeException('No more address to try for this console.'); $p = $this->profile; $engine = Engine::get(); $stream = match ($p->method) { 'demo' => $this->fake?->openConsole($console) ?? throw new \RuntimeException('No demo pool.'), 'https' => $this->polled[] = new TlsStream($host, $p->port(), 6.0), 'ssh' => $engine->ssh()->via(array_filter(['sshHost' => $p->host, 'sshPort' => $p->sshPort, 'sshUser' => $p->sshUser, 'sshKey' => $p->sshKey !== '' ? $p->sshKey : null, 'sshPassword' => $p->sshKey === '' ? $p->password : null, 'sshKeys' => $engine->sshKeyring()->all() ?: null], fn ($v) => $v !== null && $v !== ''))->connect($host === $p->host ? '127.0.0.1' : $host, $p->port()), default => $engine->tcp()->connect($host, $p->port(), 6.0), }; $rfb = new \App\Vnc\RfbClient($stream, \App\Vnc\RfbClient::connectRequest($location, $console, (string)$this->client->session)); $rfb->target = ($p->method === 'ssh' ? "SSH {$p->host} → " : '') . $host . ':' . $p->port(); return $rfb; } // events ---------------------------------------------------------------------------------------------------------- private function pollEvents(int $generation, ?Task $first = null): void { $client = $this->client; if ($client === null) { return; } $client->events(self::CLASSES, $this->token, 30.5)->then( function (mixed $batch) use ($generation, $first): void { if ($generation !== $this->generation) { return; } $this->apply((array)$batch); if ($this->state !== self::CONNECTED) { $this->state = self::CONNECTED; $this->phase = ''; $this->session->log('Connected to ' . $this->name(), $this->name()); $first?->settle(true); } $this->pollEvents($generation); }, function (\Throwable $e) use ($generation, $first): void { if ($generation !== $this->generation) { return; } if ($first !== null) { $this->failed($e->getMessage(), $first, $e); return; } if ($e instanceof XapiError && $e->code() === 'SESSION_INVALID') { $this->state = self::FAILED; $this->error = 'The session ended: reconnect.'; return; } if ($e instanceof XapiError && $e->code() !== 'TRANSPORT') { // EVENTS_LOST, EVENT_FROM_TOKEN_PARSE_FAILURE... (a long outage, a toolstack restart): everything again from scratch $this->token = ''; $this->cache = new Cache(); } // the network went away: try again $this->phase = 'Connection lost, retrying…'; $this->retryIn = 5; }, ); } /** @param array<string, mixed> $batch */ public function apply(array $batch): void { foreach ((array)($batch['events'] ?? []) as $event) { $class = strtolower((string)($event['class'] ?? '')); $ref = (string)($event['ref'] ?? ''); if (($event['operation'] ?? '') === 'del') { $this->cache->remove($class, $ref); } elseif (is_array($event['snapshot'] ?? null)) { $this->cache->put($class, $ref, $event['snapshot']); if ($class === 'task') { $this->taskChanged($ref, $event['snapshot']); } } } $this->token = (string)($batch['token'] ?? $this->token); } /** @var list<Transport> transports closing (their logout leaving) */ private array $closing = []; public function tick(float $dt): void { $this->client?->transport->poll($dt); foreach ($this->closing as $transport) { $transport->poll($dt); } $this->links?->poll($dt); foreach ($this->polled as $i => $stream) { $stream->poll(); if ($stream->isClosed()) { unset($this->polled[$i]); } } $this->polled = array_values($this->polled); foreach ($this->transfers as $transfer) { $transfer->poll($dt); } $this->transfers = array_values(array_filter($this->transfers, fn ($t) => !$t->isDone())); if ($this->retryIn > 0) { $this->retryIn -= $dt; if ($this->retryIn <= 0) { $this->retryIn = -1; $this->phase = ''; $this->pollEvents($this->generation); } } } // actions --------------------------------------------------------------------------------------------------------- /** * An operation as XAPI's own task (Async.<method>): its progress shows in Events, the Task settles when it ends. * run('Starting VM web-01', 'VM.start', [$vm, false, false], $vm) */ public function run(string $title, string $method, array $params, ?string $target = null): Task { $done = new Task(); $entry = $this->session->log($title, $this->name(), $target, LogEntry::RUNNING); if ($this->client === null || !$this->isConnected()) { $entry->status = LogEntry::FAILED; $entry->detail = 'Not connected'; $done->settle(null, new XapiError(['TRANSPORT', 'Not connected'])); return $done; } $generation = $this->generation; $this->client->call('Async.' . $method, ...$params)->then( function (mixed $taskRef) use ($done, $entry, $generation): void { if ($generation !== $this->generation) { $entry->status = LogEntry::FAILED; $entry->detail = 'Disconnected'; $done->settle(null, new XapiError(['TRANSPORT', 'Disconnected'])); return; } $entry->taskRef = (string)$taskRef; $this->tracked[(string)$taskRef] = [$done, $entry]; if (($record = $this->cache->get((string)$taskRef)) !== null) { $this->taskChanged((string)$taskRef, $record); } }, function (\Throwable $e) use ($done, $entry): void { $entry->status = LogEntry::FAILED; $entry->detail = $e->getMessage(); $entry->finished = $this->session->clock; $done->settle(null, $e); }, ); return $done; } /** Waits for a task made with task.create (the one an HTTP handler, /export or /import, works for) as run() does for its own. */ public function follow(string $taskRef, LogEntry $entry): Task { $done = new Task(); $entry->taskRef = $taskRef; $this->tracked[$taskRef] = [$done, $entry]; if (($record = $this->cache->get($taskRef)) !== null) { $this->taskChanged($taskRef, $record); } return $done; } /** A quick call (a name, a description...): logged when it fails. */ public function call(string $method, mixed ...$params): Task { $done = new Task(); if ($this->client === null) { $done->settle(null, new XapiError(['TRANSPORT', 'Not connected'])); return $done; } $this->client->call($method, ...$params)->then(fn ($r) => $done->settle($r), function (\Throwable $e) use ($done, $method): void { $this->session->log($method . ' failed', $this->name(), null, LogEntry::FAILED, $e->getMessage()); $done->settle(null, $e); }); return $done; } /** @param array<string, mixed> $task */ private function taskChanged(string $ref, array $task): void { if (!isset($this->tracked[$ref])) { return; } [$done, $entry] = $this->tracked[$ref]; $entry->progress = (float)($task['progress'] ?? 0); $status = (string)($task['status'] ?? 'pending'); if ($status === 'pending') { return; } unset($this->tracked[$ref]); $entry->finished = $this->session->clock; $entry->progress = 1; if ($status === 'success') { $entry->status = LogEntry::DONE; $result = (string)($task['result'] ?? ''); // the result of a task is an XML-RPC value: <value>OpaqueRef:...</value> if (preg_match('#(OpaqueRef:[0-9a-f-]+)#', $result, $m)) { $result = $m[1]; } $done->settle($result); } else { $entry->status = $status === 'cancelled' ? LogEntry::CANCELLED : LogEntry::FAILED; $failure = array_map('strval', (array)($task['error_info'] ?? [])); $entry->detail = $status === 'cancelled' ? 'Cancelled' : XapiError::explain($failure); $done->settle(null, new XapiError($failure ?: ['CANCELLED'])); } $this->client?->call('task.destroy', $ref); } public function cancel(LogEntry $entry): void { if ($entry->taskRef !== null && $entry->status === LogEntry::RUNNING) { $this->client?->call('task.cancel', $entry->taskRef); } }}