<?phpnamespace App\Xapi\Fake;use App\Xapi\Xva;/** * The demo pool's HTTP handlers for XVA files (part of FakeXapi): GET /export?session_id=…&ref=<vm>&task_id=…&use_compression=… * streams a small XVA (a tar: ova.xml + the disks in 1 MiB members), PUT /import?session_id=…&sr_id=…&task_id=… reads one and * makes a halted VM of it, its ref in the task's result as XAPI writes it. task.create makes the task the client follows. */trait XvaOps{ /** @var list<FakeHttpStream> */ private array $httpStreams = []; /** the size of each fake disk in an export (tests make it smaller or bigger) */ public int $xvaDiskBytes = 2097152; /** @return array{0: \Closure(): object, 1: string} how to open a stream of one's own to this server (HttpStreamTransfer), and its Host */ public function connector(): array { return [function (): object { $stream = new FakeHttpStream(fn (FakeHttpStream $s, string $what) => $this->onHttp($s, $what)); $this->httpStreams[] = $stream; return $stream; }, $this->address]; } private function pollXva(float $dt): void { foreach ($this->httpStreams as $i => $stream) { $stream->poll(); $task = (string)($stream->context['task'] ?? ''); if ($task !== '' && $stream->method === 'GET' && isset($stream->context['size']) && ($this->db['task'][$task]['status'] ?? '') === 'pending') { $progress = round(min(0.99, $stream->sent / max(1.0, (float)$stream->context['size'])), 2); if (abs($progress - (float)$this->db['task'][$task]['progress']) >= 0.1) { $this->set('task', $task, ['progress' => $progress]); } if ($stream->isClosed()) { $this->set('task', $task, ['status' => 'success', 'progress' => 1.0, 'finished' => $this->date()]); } } if ($stream->isClosed()) { unset($this->httpStreams[$i]); } } $this->httpStreams = array_values($this->httpStreams); } /** @return array{0: float, 1: \Closure, 2: string, 3: string}|null */ private function xvaOperation(string $class, string $method, array $p): ?array { if ($class === 'task' && $method === 'create') { return [0.0, fn (): string => $this->add('task', ['name_label' => (string)($p[0] ?? ''), 'name_description' => (string)($p[1] ?? ''), 'status' => 'pending', 'progress' => 0.0, 'created' => $this->date(), 'finished' => '19700101T00:00:00Z', 'result' => '', 'error_info' => [], 'resident_on' => array_key_first($this->db['host']), 'allowed_operations' => ['cancel'], 'other_config' => []]), 'Creating a task', '']; } return null; } /** task.cancel of a task an HTTP handler works for: the handler stops and closes its connection, as XAPI's do. */ private function cancelHttp(string $task): void { foreach ($this->httpStreams as $stream) { if (($stream->context['task'] ?? '') === $task && !$stream->isClosed()) { $this->set('task', $task, ['status' => 'cancelled', 'finished' => $this->date()]); $stream->close(); } } } private function onHttp(FakeHttpStream $s, string $what): void { parse_str((string)parse_url($s->path, PHP_URL_QUERY), $q); $route = (string)parse_url($s->path, PHP_URL_PATH); $task = (string)($q['task_id'] ?? ''); if ($what === 'head') { $s->context['task'] = isset($this->db['task'][$task]) ? $task : ''; if (!isset($this->sessions[(string)($q['session_id'] ?? '')])) { $s->reply(403, 'Forbidden', ['Content-Length' => '0']); return; } if ($route === '/export' && $s->method === 'GET') { $this->httpExport($s, (string)($q['ref'] ?? ''), ($q['use_compression'] ?? '') === 'true'); } elseif ($route === '/import' && $s->method === 'PUT') { $sr = (string)($q['sr_id'] ?? ''); if (!isset($this->db['sr'][$sr])) { $this->failTask($s, ['HANDLE_INVALID', 'SR', $sr]); return; } $s->context['sr'] = $sr; $s->context['length'] = (float)($s->headers['content-length'] ?? 0); if ($s->context['length'] <= 0) { $this->failTask($s, ['IMPORT_ERROR', 'No Content-Length']); } } else { $s->reply(404, 'Not Found', ['Content-Length' => '0']); } return; } // a part of an upload arrived if ($route !== '/import' || !isset($s->context['sr']) || !empty($s->context['answered'])) { return; } $length = (float)$s->context['length']; $task = (string)$s->context['task']; if ($task !== '') { $progress = round(min(0.99, $s->received / max(1.0, $length)), 2); if (abs($progress - (float)$this->db['task'][$task]['progress']) >= 0.1) { $this->set('task', $task, ['progress' => $progress]); } } if ($s->received < $length) { return; } $s->context['answered'] = true; $info = Xva::readHeader($s->bodyHead); if ($info === null) { $this->failTask($s, ['IMPORT_ERROR', 'The file is not an XVA export (no ova.xml at its start).']); return; } $disk = 0; foreach ($info['disks'] as [, $size]) { $disk = max($disk, (int)ceil($size / self::GIB)); } $network = (string)array_key_first($this->db['network']); $vm = $this->addVm(['name_label' => (string)$info['name_label'], 'description' => (string)$info['name_description'], 'power_state' => 'Halted', 'memory' => max(256 * self::MIB, (float)$info['memory_static_max']), 'vcpus' => max(1, (int)$info['VCPUs_max']), 'disk' => max(1, $disk), 'sr' => (string)$s->context['sr'], 'networks' => [$network], 'index' => count($this->db['vm'])]); if ($task !== '') { $this->set('task', $task, ['status' => 'success', 'progress' => 1.0, 'finished' => $this->date(), 'result' => '<value><array><data><value>' . $vm . '</value></data></array></value>']); } $s->reply(200, 'OK', ['Content-Length' => '0']); } private function httpExport(FakeHttpStream $s, string $vm, bool $compress): void { $record = $this->db['vm'][$vm] ?? null; if ($record === null) { $this->failTask($s, ['HANDLE_INVALID', 'VM', $vm]); return; } if ($record['power_state'] !== 'Halted' && empty($record['is_a_snapshot'])) { $this->failTask($s, ['VM_BAD_POWER_STATE', $vm, 'halted', strtolower((string)$record['power_state'])]); return; } $disks = []; foreach ((array)$record['VBDs'] as $vbd) { $vdi = (string)($this->db['vbd'][$vbd]['VDI'] ?? ''); if (($this->db['vbd'][$vbd]['type'] ?? '') === 'Disk' && isset($this->db['vdi'][$vdi])) { $disks[] = ['id' => 'Ref:' . (count($disks) + 1), 'name' => (string)$this->db['vdi'][$vdi]['name_label'], 'size' => (float)$this->db['vdi'][$vdi]['virtual_size']]; } } $parts = [Xva::member('ova.xml', Xva::ovaXml($record, $disks), $this->now)]; foreach ($disks as $disk) { // the disks' blocks: 1 MiB members and their checksums (a pattern here, not the VM's data) for ($at = 0, $n = 0; $at < $this->xvaDiskBytes; $at += 1048576, $n++) { $block = str_repeat(substr(md5($disk['id'] . $n), 0, 16), (int)(min(1048576, $this->xvaDiskBytes - $at) / 16)); $name = $disk['id'] . '/' . sprintf('%08d', $n); $parts[] = Xva::member($name, $block, $this->now); $parts[] = Xva::member($name . '.checksum', sha1($block), $this->now); } } $parts[] = Xva::end(); if ($compress && function_exists('gzencode')) { // XAPI's use_compression: the whole tar gzipped, sent until the connection closes (no length) $parts = [(string)gzencode(implode('', $parts), 1)]; $s->context['size'] = (float)strlen($parts[0]); $s->reply(200, 'OK', ['Content-Type' => 'application/octet-stream'], str_split($parts[0], 262144)); return; } $size = 0; foreach ($parts as $part) { $size += strlen($part); } $s->context['size'] = (float)$size; $s->reply(200, 'OK', ['Content-Type' => 'application/octet-stream', 'Content-Length' => (string)$size], $parts); } /** @param list<string> $failure the task fails with it, the HTTP answer is a 500 */ private function failTask(FakeHttpStream $s, array $failure): void { $task = (string)($s->context['task'] ?? ''); if ($task !== '') { $this->set('task', $task, ['status' => 'failure', 'progress' => 1.0, 'finished' => $this->date(), 'error_info' => $failure]); } $s->context['answered'] = true; $s->reply(500, 'Internal Error', ['Content-Type' => 'text/plain', 'Content-Length' => (string)strlen(implode(' ', $failure))], implode(' ', $failure)); }}