1302 lines
50 KiB
PHP
1302 lines
50 KiB
PHP
#!/usr/bin/env php
|
|
<?php
|
|
|
|
if (PHP_SAPI !== 'cli') {
|
|
fwrite(STDERR, "This agent must run from the CLI.\n");
|
|
exit(1);
|
|
}
|
|
|
|
final class AgentConfig
|
|
{
|
|
public function __construct(public array $data, public string $path)
|
|
{
|
|
}
|
|
|
|
public static function load(string $path): self
|
|
{
|
|
if (!is_file($path)) {
|
|
throw new RuntimeException('Missing config file: ' . $path);
|
|
}
|
|
|
|
$decoded = json_decode((string)file_get_contents($path), true);
|
|
if (!is_array($decoded)) {
|
|
throw new RuntimeException('Invalid config file: ' . $path);
|
|
}
|
|
|
|
return new self($decoded, $path);
|
|
}
|
|
|
|
public function get(string $key, mixed $default = null): mixed
|
|
{
|
|
return array_key_exists($key, $this->data) ? $this->data[$key] : $default;
|
|
}
|
|
|
|
public function set(string $key, mixed $value): void
|
|
{
|
|
$this->data[$key] = $value;
|
|
}
|
|
|
|
public function save(): void
|
|
{
|
|
file_put_contents($this->path, json_encode($this->data, JSON_PRETTY_PRINT | JSON_UNESCAPED_SLASHES) . PHP_EOL);
|
|
}
|
|
}
|
|
|
|
final class HttpJsonClient
|
|
{
|
|
public function __construct(private readonly string $baseUrl)
|
|
{
|
|
}
|
|
|
|
public function post(string $path, array $payload, int $timeoutSeconds = 30): ?array
|
|
{
|
|
$response = $this->requestJson(
|
|
'POST',
|
|
rtrim($this->baseUrl, '/') . '/' . ltrim($path, '/'),
|
|
$payload,
|
|
$timeoutSeconds
|
|
);
|
|
|
|
return $response['decoded'];
|
|
}
|
|
|
|
public function getJson(string $url, int $timeoutSeconds = 10): array
|
|
{
|
|
$response = $this->requestJson('GET', $url, null, $timeoutSeconds);
|
|
return $response['decoded'] ?? [];
|
|
}
|
|
|
|
public function download(string $url, int $timeoutSeconds = 60): string
|
|
{
|
|
$response = $this->requestRaw('GET', $url, null, $timeoutSeconds, ['Accept: */*']);
|
|
return $response['body'];
|
|
}
|
|
|
|
private function requestJson(string $method, string $url, ?array $payload, int $timeoutSeconds): array
|
|
{
|
|
$headers = ['Accept: application/json'];
|
|
if ($payload !== null) {
|
|
$headers[] = 'Content-Type: application/json';
|
|
}
|
|
|
|
$response = $this->requestRaw($method, $url, $payload, $timeoutSeconds, $headers);
|
|
$decoded = json_decode($response['body'], true);
|
|
if (!is_array($decoded)) {
|
|
throw new RuntimeException('Invalid JSON response from ' . $method . ' ' . $url);
|
|
}
|
|
|
|
$response['decoded'] = $decoded;
|
|
return $response;
|
|
}
|
|
|
|
private function requestRaw(
|
|
string $method,
|
|
string $url,
|
|
?array $payload,
|
|
int $timeoutSeconds,
|
|
array $headers
|
|
): array {
|
|
$responseHeaders = [];
|
|
$ch = curl_init($url);
|
|
curl_setopt_array($ch, [
|
|
CURLOPT_CUSTOMREQUEST => $method,
|
|
CURLOPT_RETURNTRANSFER => true,
|
|
CURLOPT_TIMEOUT => $timeoutSeconds,
|
|
CURLOPT_CONNECTTIMEOUT => min(10, $timeoutSeconds),
|
|
CURLOPT_FOLLOWLOCATION => true,
|
|
CURLOPT_HTTPHEADER => $headers,
|
|
CURLOPT_HEADERFUNCTION => static function ($curl, string $headerLine) use (&$responseHeaders): int {
|
|
$trimmed = trim($headerLine);
|
|
if ($trimmed !== '') {
|
|
$responseHeaders[] = $trimmed;
|
|
}
|
|
return strlen($headerLine);
|
|
},
|
|
]);
|
|
|
|
if ($payload !== null) {
|
|
curl_setopt($ch, CURLOPT_POSTFIELDS, json_encode($payload, JSON_UNESCAPED_SLASHES));
|
|
}
|
|
|
|
$raw = curl_exec($ch);
|
|
$status = (int)curl_getinfo($ch, CURLINFO_RESPONSE_CODE);
|
|
$error = curl_error($ch);
|
|
curl_close($ch);
|
|
|
|
if ($raw === false) {
|
|
throw new RuntimeException(sprintf('%s %s failed: %s', $method, $url, $error !== '' ? $error : 'unknown curl error'));
|
|
}
|
|
|
|
if ($status >= 400) {
|
|
$preview = substr(preg_replace('/\s+/', ' ', (string)$raw) ?? '', 0, 240);
|
|
$headerPreview = implode('; ', array_slice($responseHeaders, 0, 8));
|
|
throw new RuntimeException(sprintf(
|
|
'%s %s returned HTTP %d. Headers: %s. Body preview: %s',
|
|
$method,
|
|
$url,
|
|
$status,
|
|
$headerPreview !== '' ? $headerPreview : 'none',
|
|
$preview !== '' ? $preview : 'empty'
|
|
));
|
|
}
|
|
|
|
return [
|
|
'status' => $status,
|
|
'body' => (string)$raw,
|
|
'headers' => $responseHeaders,
|
|
];
|
|
}
|
|
}
|
|
|
|
final class Logger
|
|
{
|
|
public function __construct(private readonly string $logFile)
|
|
{
|
|
}
|
|
|
|
public function info(string $message): void
|
|
{
|
|
$this->write('INFO', $message);
|
|
}
|
|
|
|
public function warning(string $message): void
|
|
{
|
|
$this->write('WARNING', $message);
|
|
}
|
|
|
|
public function error(string $message): void
|
|
{
|
|
$this->write('ERROR', $message);
|
|
}
|
|
|
|
private function write(string $level, string $message): void
|
|
{
|
|
$line = '[' . date('c') . "] {$level} {$message}\n";
|
|
file_put_contents($this->logFile, $line, FILE_APPEND);
|
|
fwrite($level === 'ERROR' ? STDERR : STDOUT, $line);
|
|
}
|
|
}
|
|
|
|
final class LocalStateStore
|
|
{
|
|
private SQLite3 $db;
|
|
|
|
public function __construct(string $path)
|
|
{
|
|
if (!class_exists('SQLite3')) {
|
|
throw new RuntimeException('The edge-agent container requires sqlite3 support.');
|
|
}
|
|
|
|
$directory = dirname($path);
|
|
if (!is_dir($directory)) {
|
|
@mkdir($directory, 0777, true);
|
|
}
|
|
|
|
$this->db = new SQLite3($path);
|
|
$this->db->busyTimeout(5000);
|
|
$this->db->exec('PRAGMA journal_mode = WAL;');
|
|
$this->db->exec('PRAGMA synchronous = NORMAL;');
|
|
$this->db->exec(
|
|
'CREATE TABLE IF NOT EXISTS kv (
|
|
key TEXT PRIMARY KEY,
|
|
value_json TEXT NOT NULL,
|
|
updated_at TEXT NOT NULL
|
|
)'
|
|
);
|
|
$this->db->exec(
|
|
'CREATE TABLE IF NOT EXISTS outbox (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
item_type TEXT NOT NULL,
|
|
endpoint TEXT NOT NULL,
|
|
payload_json TEXT NOT NULL,
|
|
created_at TEXT NOT NULL
|
|
)'
|
|
);
|
|
}
|
|
|
|
public function getJson(string $key, mixed $default = null): mixed
|
|
{
|
|
$statement = $this->db->prepare('SELECT value_json FROM kv WHERE key = :key LIMIT 1');
|
|
$statement->bindValue(':key', $key, SQLITE3_TEXT);
|
|
$result = $statement->execute();
|
|
$row = $result instanceof SQLite3Result ? $result->fetchArray(SQLITE3_ASSOC) : false;
|
|
if (!is_array($row) || !array_key_exists('value_json', $row)) {
|
|
return $default;
|
|
}
|
|
|
|
$decoded = json_decode((string)$row['value_json'], true);
|
|
return json_last_error() === JSON_ERROR_NONE ? $decoded : $default;
|
|
}
|
|
|
|
public function setJson(string $key, mixed $value): void
|
|
{
|
|
$statement = $this->db->prepare(
|
|
'INSERT INTO kv (key, value_json, updated_at)
|
|
VALUES (:key, :value_json, :updated_at)
|
|
ON CONFLICT(key) DO UPDATE SET value_json = excluded.value_json, updated_at = excluded.updated_at'
|
|
);
|
|
$statement->bindValue(':key', $key, SQLITE3_TEXT);
|
|
$statement->bindValue(':value_json', json_encode($value, JSON_UNESCAPED_SLASHES), SQLITE3_TEXT);
|
|
$statement->bindValue(':updated_at', date('c'), SQLITE3_TEXT);
|
|
$statement->execute();
|
|
}
|
|
|
|
public function enqueue(string $type, string $endpoint, array $payload): void
|
|
{
|
|
$statement = $this->db->prepare(
|
|
'INSERT INTO outbox (item_type, endpoint, payload_json, created_at)
|
|
VALUES (:item_type, :endpoint, :payload_json, :created_at)'
|
|
);
|
|
$statement->bindValue(':item_type', $type, SQLITE3_TEXT);
|
|
$statement->bindValue(':endpoint', $endpoint, SQLITE3_TEXT);
|
|
$statement->bindValue(':payload_json', json_encode($payload, JSON_UNESCAPED_SLASHES), SQLITE3_TEXT);
|
|
$statement->bindValue(':created_at', date('c'), SQLITE3_TEXT);
|
|
$statement->execute();
|
|
}
|
|
|
|
/**
|
|
* @return array<int,array<string,mixed>>
|
|
*/
|
|
public function queuedItems(int $limit = 25): array
|
|
{
|
|
$statement = $this->db->prepare(
|
|
'SELECT id, item_type, endpoint, payload_json, created_at
|
|
FROM outbox
|
|
ORDER BY id ASC
|
|
LIMIT :limit'
|
|
);
|
|
$statement->bindValue(':limit', max(1, $limit), SQLITE3_INTEGER);
|
|
$result = $statement->execute();
|
|
$items = [];
|
|
if (!$result instanceof SQLite3Result) {
|
|
return $items;
|
|
}
|
|
|
|
while (($row = $result->fetchArray(SQLITE3_ASSOC)) !== false) {
|
|
$items[] = [
|
|
'id' => (int)$row['id'],
|
|
'type' => (string)$row['item_type'],
|
|
'endpoint' => (string)$row['endpoint'],
|
|
'payload' => json_decode((string)$row['payload_json'], true) ?: [],
|
|
'created_at' => (string)$row['created_at'],
|
|
];
|
|
}
|
|
|
|
return $items;
|
|
}
|
|
|
|
public function removeOutboxItem(int $id): void
|
|
{
|
|
$statement = $this->db->prepare('DELETE FROM outbox WHERE id = :id');
|
|
$statement->bindValue(':id', $id, SQLITE3_INTEGER);
|
|
$statement->execute();
|
|
}
|
|
|
|
public function outboxSummary(): array
|
|
{
|
|
$countResult = $this->db->querySingle('SELECT COUNT(*) FROM outbox');
|
|
$oldestResult = $this->db->querySingle('SELECT created_at FROM outbox ORDER BY id ASC LIMIT 1');
|
|
$lastSyncAt = $this->getJson('last_sync_at');
|
|
|
|
return [
|
|
'queued' => (int)$countResult,
|
|
'oldest_queued_at' => is_string($oldestResult) && $oldestResult !== '' ? $oldestResult : null,
|
|
'last_replayed_at' => is_string($lastSyncAt) && $lastSyncAt !== '' ? $lastSyncAt : null,
|
|
];
|
|
}
|
|
}
|
|
|
|
final class TruckwashEdgeAgent
|
|
{
|
|
private const DEFAULT_INSTALL_DIR = '/opt/truckwash-edge-agent';
|
|
private const DEFAULT_RUNTIME_DIR = '/opt/truckwash-edge-agent/runtime';
|
|
private const DEFAULT_STATE_DATABASE = '/opt/truckwash-edge-agent/runtime/gateway-state.sqlite';
|
|
private const DEFAULT_WORKER_BASE_URL = 'http://lan-worker:8090';
|
|
private const DEFAULT_UPDATE_WINDOW = '02:00-04:00';
|
|
|
|
private AgentConfig $config;
|
|
private HttpJsonClient $http;
|
|
private HttpJsonClient $workerHttp;
|
|
private Logger $logger;
|
|
private LocalStateStore $stateStore;
|
|
private string $installDir;
|
|
private string $runtimeDir;
|
|
private string $statePath;
|
|
private string $lastOperationSnapshotPath;
|
|
private string $lastHeartbeatMarkerPath;
|
|
private string $stagedUpdatePath;
|
|
private int $lastHeartbeatAt = 0;
|
|
private string $agentInstanceId;
|
|
|
|
public function __construct(string $configPath)
|
|
{
|
|
$this->config = AgentConfig::load($configPath);
|
|
$this->installDir = rtrim((string)$this->config->get('installDir', self::DEFAULT_INSTALL_DIR), DIRECTORY_SEPARATOR);
|
|
$this->runtimeDir = rtrim((string)$this->config->get('runtimeDir', self::DEFAULT_RUNTIME_DIR), DIRECTORY_SEPARATOR);
|
|
if (!is_dir($this->runtimeDir)) {
|
|
@mkdir($this->runtimeDir, 0777, true);
|
|
}
|
|
|
|
$this->http = new HttpJsonClient((string)$this->config->get('apiUrl'));
|
|
$this->workerHttp = new HttpJsonClient((string)$this->config->get('workerBaseUrl', self::DEFAULT_WORKER_BASE_URL));
|
|
$this->logger = new Logger($this->runtimeDir . DIRECTORY_SEPARATOR . 'agent.log');
|
|
$this->stateStore = new LocalStateStore((string)$this->config->get('stateDatabasePath', self::DEFAULT_STATE_DATABASE));
|
|
$this->statePath = $this->runtimeDir . DIRECTORY_SEPARATOR . 'current-operation.json';
|
|
$this->lastOperationSnapshotPath = $this->runtimeDir . DIRECTORY_SEPARATOR . 'last-operation.json';
|
|
$this->lastHeartbeatMarkerPath = $this->runtimeDir . DIRECTORY_SEPARATOR . 'last-heartbeat-ok.txt';
|
|
$this->stagedUpdatePath = $this->runtimeDir . DIRECTORY_SEPARATOR . 'staged-update.json';
|
|
$this->agentInstanceId = $this->ensureAgentInstanceId();
|
|
}
|
|
|
|
public function run(): void
|
|
{
|
|
$this->logger->info('Truckwash compose edge-agent starting.');
|
|
$this->recoverPreviousOperationState();
|
|
|
|
while (true) {
|
|
try {
|
|
$this->reloadConfigFromDisk();
|
|
$this->ensureClaimed();
|
|
$this->flushOutbox();
|
|
$this->heartbeat();
|
|
$processedManagementOperation = $this->processManagementOperation();
|
|
if (!$processedManagementOperation) {
|
|
$this->processCommandQueue();
|
|
}
|
|
$this->flushOutbox();
|
|
} catch (Throwable $throwable) {
|
|
$this->logger->error($throwable->getMessage());
|
|
sleep(2);
|
|
}
|
|
|
|
usleep(250000);
|
|
}
|
|
}
|
|
|
|
private function ensureClaimed(): void
|
|
{
|
|
if ($this->config->get('gatewayId') && $this->config->get('agentToken')) {
|
|
return;
|
|
}
|
|
|
|
$installedVersion = (string)$this->config->get('installedVersion', 'compose-php-agent-v2');
|
|
$response = $this->http->post('/edge-agent/claim', [
|
|
'token' => (string)$this->config->get('installToken'),
|
|
'hostname' => gethostname() ?: 'truckwash-edge',
|
|
'installed_version' => $installedVersion,
|
|
'metadata' => [
|
|
'runtime' => 'compose-php',
|
|
'runtime_mode' => 'compose',
|
|
'php_version' => PHP_VERSION,
|
|
'agent_instance_id' => $this->agentInstanceId,
|
|
'update_window' => (string)$this->config->get('updateWindow', self::DEFAULT_UPDATE_WINDOW),
|
|
'container_health' => $this->buildContainerHealth(),
|
|
'outbox_status' => $this->buildOutboxStatus(),
|
|
'last_sync_at' => $this->stateStore->getJson('last_sync_at'),
|
|
'rollback_status' => $this->readRollbackStatus(),
|
|
'staged_version' => $this->currentStagedUpdate(),
|
|
],
|
|
]);
|
|
|
|
$payload = $response['data'] ?? [];
|
|
$gateway = is_array($payload['gateway'] ?? null) ? (array)$payload['gateway'] : [];
|
|
$this->config->set('gatewayId', $gateway['id'] ?? null);
|
|
$this->config->set('agentToken', $payload['agent_token'] ?? null);
|
|
$this->config->set('installedVersion', $installedVersion);
|
|
$this->config->set('targetVersion', (string)($gateway['target_version'] ?? $installedVersion));
|
|
$this->config->set('agentInstanceId', $this->agentInstanceId);
|
|
$this->config->save();
|
|
$this->stateStore->setJson('last_sync_at', date('c'));
|
|
$this->logger->info('Claimed gateway ' . (string)($gateway['id'] ?? 'unknown') . '.');
|
|
}
|
|
|
|
private function heartbeat(bool $force = false, array $metadata = []): void
|
|
{
|
|
$interval = (int)$this->config->get('heartbeatIntervalSeconds', 15);
|
|
if (!$force && (time() - $this->lastHeartbeatAt) < max(5, $interval)) {
|
|
return;
|
|
}
|
|
|
|
$gatewayId = (int)$this->config->get('gatewayId');
|
|
if ($gatewayId <= 0) {
|
|
return;
|
|
}
|
|
|
|
$operationState = $this->readOperationState();
|
|
$payload = [
|
|
'agent_token' => (string)$this->config->get('agentToken'),
|
|
'status' => 'ONLINE',
|
|
'hostname' => gethostname() ?: 'truckwash-edge',
|
|
'installed_version' => (string)$this->config->get('installedVersion', 'compose-php-agent-v2'),
|
|
'target_version' => (string)$this->config->get(
|
|
'targetVersion',
|
|
$this->config->get('installedVersion', 'compose-php-agent-v2')
|
|
),
|
|
'metadata' => array_merge([
|
|
'agent_instance_id' => $this->agentInstanceId,
|
|
'runtime' => 'compose-php',
|
|
'runtime_mode' => 'compose',
|
|
'current_operation' => $operationState,
|
|
'system_metrics' => $this->buildSystemMetrics(),
|
|
'container_health' => $this->buildContainerHealth(),
|
|
'outbox_status' => $this->buildOutboxStatus(),
|
|
'last_sync_at' => $this->stateStore->getJson('last_sync_at'),
|
|
'update_window' => (string)$this->config->get('updateWindow', self::DEFAULT_UPDATE_WINDOW),
|
|
'staged_version' => $this->currentStagedUpdate(),
|
|
'rollback_status' => $this->readRollbackStatus(),
|
|
], $metadata),
|
|
];
|
|
|
|
$posted = $this->sendControlPlaneEvent(
|
|
'/edge-agent/gateways/' . $gatewayId . '/heartbeat',
|
|
$payload,
|
|
'heartbeat'
|
|
);
|
|
if (!$posted) {
|
|
return;
|
|
}
|
|
|
|
$this->lastHeartbeatAt = time();
|
|
file_put_contents($this->lastHeartbeatMarkerPath, json_encode([
|
|
'gateway_id' => $gatewayId,
|
|
'at' => date('c'),
|
|
'agent_instance_id' => $this->agentInstanceId,
|
|
], JSON_UNESCAPED_SLASHES) . PHP_EOL);
|
|
}
|
|
|
|
private function processCommandQueue(): void
|
|
{
|
|
$gatewayId = (int)$this->config->get('gatewayId');
|
|
if ($gatewayId <= 0) {
|
|
return;
|
|
}
|
|
|
|
try {
|
|
$response = $this->http->post('/edge-agent/gateways/' . $gatewayId . '/commands/poll', [
|
|
'agent_token' => (string)$this->config->get('agentToken'),
|
|
'wait_seconds' => (int)$this->config->get('operationPollTimeoutSeconds', 20),
|
|
]);
|
|
} catch (Throwable $throwable) {
|
|
$this->logger->warning('Command polling failed: ' . $throwable->getMessage());
|
|
return;
|
|
}
|
|
|
|
$command = $response['data'] ?? null;
|
|
if (!is_array($command) || empty($command['id'])) {
|
|
return;
|
|
}
|
|
|
|
$jobId = (int)$command['id'];
|
|
$type = (string)($command['command_type'] ?? $command['commandType'] ?? '');
|
|
$request = is_array($command['payload'] ?? null) ? (array)$command['payload'] : [];
|
|
|
|
try {
|
|
$result = match ($type) {
|
|
'DISCOVER_SHELLY' => $this->runDiscovery(),
|
|
'GET_RELAY_STATUS' => $this->readRelayStatus($request),
|
|
'SET_RELAY_STATE' => $this->switchRelay($request),
|
|
default => throw new RuntimeException('Unsupported command type: ' . $type),
|
|
};
|
|
|
|
$this->sendControlPlaneEvent(
|
|
'/edge-agent/gateways/' . $gatewayId . '/commands/' . $jobId . '/result',
|
|
[
|
|
'agent_token' => (string)$this->config->get('agentToken'),
|
|
'ok' => true,
|
|
'result' => $result,
|
|
],
|
|
'command_result'
|
|
);
|
|
} catch (Throwable $throwable) {
|
|
$this->sendControlPlaneEvent(
|
|
'/edge-agent/gateways/' . $gatewayId . '/commands/' . $jobId . '/result',
|
|
[
|
|
'agent_token' => (string)$this->config->get('agentToken'),
|
|
'ok' => false,
|
|
'error' => $throwable->getMessage(),
|
|
],
|
|
'command_result'
|
|
);
|
|
}
|
|
}
|
|
|
|
private function processManagementOperation(): bool
|
|
{
|
|
$gatewayId = (int)$this->config->get('gatewayId');
|
|
if ($gatewayId <= 0) {
|
|
return false;
|
|
}
|
|
|
|
try {
|
|
$response = $this->http->post('/edge-agent/gateways/' . $gatewayId . '/operations/next', [
|
|
'agent_token' => (string)$this->config->get('agentToken'),
|
|
'wait_seconds' => (int)$this->config->get('operationPollTimeoutSeconds', 20),
|
|
'agent_instance_id' => $this->agentInstanceId,
|
|
]);
|
|
} catch (Throwable $throwable) {
|
|
$this->logger->warning('Operation polling failed: ' . $throwable->getMessage());
|
|
return false;
|
|
}
|
|
|
|
$operation = $response['data'] ?? null;
|
|
if (!is_array($operation) || empty($operation['id'])) {
|
|
return false;
|
|
}
|
|
|
|
$operationId = (int)$operation['id'];
|
|
$type = (string)($operation['type'] ?? '');
|
|
$request = is_array($operation['request'] ?? null) ? (array)$operation['request'] : [];
|
|
$this->persistOperationState([
|
|
'gateway_id' => $gatewayId,
|
|
'operation_id' => $operationId,
|
|
'type' => $type,
|
|
'status' => 'IN_PROGRESS',
|
|
'stage' => 'claimed',
|
|
'progress' => 5,
|
|
'agent_instance_id' => $this->agentInstanceId,
|
|
'started_at' => date('c'),
|
|
'request' => $request,
|
|
]);
|
|
|
|
$this->emitOperationProgress(
|
|
$gatewayId,
|
|
$operationId,
|
|
'OPERATION_AGENT_STARTED',
|
|
'Compose edge-agent started processing the operation',
|
|
10,
|
|
['type' => $type, 'agent_instance_id' => $this->agentInstanceId],
|
|
'starting'
|
|
);
|
|
|
|
try {
|
|
$result = match ($type) {
|
|
'DISCOVERY' => $this->runManagementDiscovery($gatewayId, $operationId, $request),
|
|
'UPDATE' => $this->runUpdate($gatewayId, $operationId, $request),
|
|
'UNINSTALL' => $this->runUninstall($gatewayId, $operationId, $request),
|
|
default => throw new RuntimeException('Unsupported operation type: ' . $type),
|
|
};
|
|
|
|
$this->sendControlPlaneEvent(
|
|
'/edge-agent/gateways/' . $gatewayId . '/operations/' . $operationId . '/complete',
|
|
[
|
|
'agent_token' => (string)$this->config->get('agentToken'),
|
|
'ok' => true,
|
|
'result' => $result,
|
|
],
|
|
'operation_complete'
|
|
);
|
|
$this->snapshotOperationState('COMPLETED', ['result' => $result]);
|
|
$this->clearOperationState();
|
|
} catch (Throwable $throwable) {
|
|
$errorCode = $this->classifyManagementError($throwable, $type);
|
|
$this->postOperationEvent($gatewayId, $operationId, [
|
|
'level' => 'ERROR',
|
|
'code' => $errorCode,
|
|
'message' => $throwable->getMessage(),
|
|
'context' => ['type' => $type, 'progress' => 100, 'agent_instance_id' => $this->agentInstanceId],
|
|
]);
|
|
|
|
$this->sendControlPlaneEvent(
|
|
'/edge-agent/gateways/' . $gatewayId . '/operations/' . $operationId . '/complete',
|
|
[
|
|
'agent_token' => (string)$this->config->get('agentToken'),
|
|
'ok' => false,
|
|
'error_code' => $errorCode,
|
|
'error_message' => $throwable->getMessage(),
|
|
],
|
|
'operation_complete'
|
|
);
|
|
$this->snapshotOperationState('FAILED', [
|
|
'error_code' => $errorCode,
|
|
'error_message' => $throwable->getMessage(),
|
|
]);
|
|
$this->clearOperationState();
|
|
}
|
|
|
|
return true;
|
|
}
|
|
|
|
private function postOperationEvent(int $gatewayId, int $operationId, array $payload): void
|
|
{
|
|
$this->sendControlPlaneEvent(
|
|
'/edge-agent/gateways/' . $gatewayId . '/operations/' . $operationId . '/events',
|
|
[
|
|
'agent_token' => (string)$this->config->get('agentToken'),
|
|
'level' => (string)($payload['level'] ?? 'INFO'),
|
|
'code' => $payload['code'] ?? null,
|
|
'message' => (string)($payload['message'] ?? 'Operation event'),
|
|
'context' => is_array($payload['context'] ?? null) ? (array)$payload['context'] : [],
|
|
],
|
|
'operation_event'
|
|
);
|
|
}
|
|
|
|
private function emitOperationProgress(
|
|
int $gatewayId,
|
|
int $operationId,
|
|
string $code,
|
|
string $message,
|
|
int $progress,
|
|
array $context = [],
|
|
?string $stage = null
|
|
): void {
|
|
$operationState = $this->readOperationState();
|
|
if ($operationState !== null) {
|
|
$operationState['progress'] = max(0, min(100, $progress));
|
|
$operationState['stage'] = $stage ?? ($operationState['stage'] ?? 'running');
|
|
$operationState['updated_at'] = date('c');
|
|
$this->persistOperationState($operationState);
|
|
}
|
|
|
|
$this->heartbeat(true, [
|
|
'current_operation' => $this->readOperationState(),
|
|
]);
|
|
|
|
$this->postOperationEvent($gatewayId, $operationId, [
|
|
'level' => 'INFO',
|
|
'code' => $code,
|
|
'message' => $message,
|
|
'context' => array_merge($context, [
|
|
'progress' => max(0, min(100, $progress)),
|
|
'agent_instance_id' => $this->agentInstanceId,
|
|
]),
|
|
]);
|
|
}
|
|
|
|
private function runDiscovery(): array
|
|
{
|
|
try {
|
|
$response = $this->workerHttp->post('/discover', [
|
|
'hostname' => gethostname() ?: 'truckwash-edge',
|
|
'gateway_id' => (int)$this->config->get('gatewayId', 0),
|
|
], 20);
|
|
$inventory = isset($response['inventory']) && is_array($response['inventory']) ? (array)$response['inventory'] : [];
|
|
return ['inventory' => $inventory];
|
|
} catch (Throwable) {
|
|
$hostname = gethostname() ?: 'truckwash-edge';
|
|
return ['inventory' => [[
|
|
'device_id' => 'gateway-runtime-' . substr(sha1($hostname), 0, 10),
|
|
'local_ip' => gethostbyname($hostname),
|
|
'model' => 'TruckWash Edge Gateway',
|
|
'channel_count' => 1,
|
|
'online' => true,
|
|
'capabilities' => [
|
|
'local_discovery' => true,
|
|
'relay_commands' => true,
|
|
'gateway_management_v2' => true,
|
|
],
|
|
'metadata' => [
|
|
'hostname' => $hostname,
|
|
'runtime' => 'compose-php',
|
|
'php_version' => PHP_VERSION,
|
|
],
|
|
]]];
|
|
}
|
|
}
|
|
|
|
private function runManagementDiscovery(int $gatewayId, int $operationId, array $request): array
|
|
{
|
|
$this->emitOperationProgress(
|
|
$gatewayId,
|
|
$operationId,
|
|
'DISCOVERY_COLLECTING',
|
|
'Collecting local gateway inventory from lan-worker',
|
|
35,
|
|
[],
|
|
'discovering'
|
|
);
|
|
$inventory = isset($request['inventory']) && is_array($request['inventory']) && $request['inventory'] !== []
|
|
? (array)$request['inventory']
|
|
: (array)($this->runDiscovery()['inventory'] ?? []);
|
|
|
|
$this->emitOperationProgress($gatewayId, $operationId, 'DISCOVERY_COMPLETED', 'Inventory collected', 90, [
|
|
'device_count' => count($inventory),
|
|
], 'finishing');
|
|
|
|
return ['inventory' => $inventory];
|
|
}
|
|
|
|
private function runUpdate(int $gatewayId, int $operationId, array $request): array
|
|
{
|
|
$targetVersion = trim((string)($request['target_version'] ?? $request['targetVersion'] ?? ''));
|
|
if ($targetVersion === '') {
|
|
throw new RuntimeException('Update request is missing target version');
|
|
}
|
|
|
|
$updateWindow = trim((string)($request['updateWindow'] ?? $this->config->get('updateWindow', self::DEFAULT_UPDATE_WINDOW)));
|
|
if ($updateWindow === '') {
|
|
$updateWindow = self::DEFAULT_UPDATE_WINDOW;
|
|
}
|
|
|
|
$requiredArtifacts = [
|
|
[
|
|
'url' => (string)($request['artifactUrl'] ?? ''),
|
|
'sha256' => (string)($request['artifactSha256'] ?? ''),
|
|
'path' => $this->installDir . DIRECTORY_SEPARATOR . 'agent.php',
|
|
'label' => 'edge-agent runtime',
|
|
'progress' => 35,
|
|
'code' => 'UPDATE_DOWNLOAD_AGENT',
|
|
],
|
|
[
|
|
'url' => (string)($request['lanWorkerArtifactUrl'] ?? ''),
|
|
'sha256' => (string)($request['lanWorkerArtifactSha256'] ?? ''),
|
|
'path' => $this->installDir . DIRECTORY_SEPARATOR . 'lan-worker.php',
|
|
'label' => 'lan-worker runtime',
|
|
'progress' => 45,
|
|
'code' => 'UPDATE_DOWNLOAD_WORKER',
|
|
],
|
|
[
|
|
'url' => (string)($request['composeFileUrl'] ?? ''),
|
|
'sha256' => (string)($request['composeFileSha256'] ?? ''),
|
|
'path' => $this->installDir . DIRECTORY_SEPARATOR . 'docker-compose.gateway.yml',
|
|
'label' => 'compose stack',
|
|
'progress' => 55,
|
|
'code' => 'UPDATE_DOWNLOAD_COMPOSE',
|
|
],
|
|
[
|
|
'url' => (string)($request['edgeAgentDockerfileUrl'] ?? ''),
|
|
'sha256' => (string)($request['edgeAgentDockerfileSha256'] ?? ''),
|
|
'path' => $this->installDir . DIRECTORY_SEPARATOR . 'Dockerfile.edge-agent',
|
|
'label' => 'edge-agent Dockerfile',
|
|
'progress' => 60,
|
|
'code' => 'UPDATE_DOWNLOAD_EDGE_DOCKERFILE',
|
|
],
|
|
[
|
|
'url' => (string)($request['lanWorkerDockerfileUrl'] ?? ''),
|
|
'sha256' => (string)($request['lanWorkerDockerfileSha256'] ?? ''),
|
|
'path' => $this->installDir . DIRECTORY_SEPARATOR . 'Dockerfile.lan-worker',
|
|
'label' => 'lan-worker Dockerfile',
|
|
'progress' => 65,
|
|
'code' => 'UPDATE_DOWNLOAD_WORKER_DOCKERFILE',
|
|
],
|
|
[
|
|
'url' => (string)($request['launcherScriptUrl'] ?? ''),
|
|
'sha256' => (string)($request['launcherScriptSha256'] ?? ''),
|
|
'path' => $this->installDir . DIRECTORY_SEPARATOR . 'gateway-launcher.sh',
|
|
'label' => 'gateway launcher',
|
|
'progress' => 72,
|
|
'code' => 'UPDATE_DOWNLOAD_LAUNCHER',
|
|
],
|
|
[
|
|
'url' => (string)($request['stackServiceUnitUrl'] ?? ''),
|
|
'sha256' => (string)($request['stackServiceUnitSha256'] ?? ''),
|
|
'path' => $this->installDir . DIRECTORY_SEPARATOR . 'truckwash-edge-gateway-stack.service',
|
|
'label' => 'compose systemd unit',
|
|
'progress' => 80,
|
|
'code' => 'UPDATE_DOWNLOAD_STACK_SERVICE',
|
|
],
|
|
[
|
|
'url' => (string)($request['serviceUnitUrl'] ?? ''),
|
|
'sha256' => (string)($request['serviceUnitSha256'] ?? ''),
|
|
'path' => $this->installDir . DIRECTORY_SEPARATOR . 'truckwash-edge-agent.service',
|
|
'label' => 'compatibility systemd unit',
|
|
'progress' => 84,
|
|
'code' => 'UPDATE_DOWNLOAD_LEGACY_SERVICE',
|
|
],
|
|
];
|
|
|
|
foreach (['composeFileUrl', 'launcherScriptUrl', 'stackServiceUnitUrl'] as $requiredKey) {
|
|
if (trim((string)($request[$requiredKey] ?? '')) === '') {
|
|
throw new RuntimeException('Update request is missing compose artifact ' . $requiredKey);
|
|
}
|
|
}
|
|
|
|
$this->emitOperationProgress($gatewayId, $operationId, 'UPDATE_VALIDATED', 'Validated compose update request', 20, [
|
|
'target_version' => $targetVersion,
|
|
'update_window' => $updateWindow,
|
|
], 'validating');
|
|
|
|
$changedFiles = [];
|
|
foreach ($requiredArtifacts as $artifact) {
|
|
if ($artifact['url'] === '') {
|
|
continue;
|
|
}
|
|
|
|
$this->emitOperationProgress(
|
|
$gatewayId,
|
|
$operationId,
|
|
(string)$artifact['code'],
|
|
'Downloading ' . (string)$artifact['label'],
|
|
(int)$artifact['progress'],
|
|
['url' => (string)$artifact['url']],
|
|
'downloading'
|
|
);
|
|
$changedFiles[] = $this->downloadToFileAtomic(
|
|
(string)$artifact['url'],
|
|
(string)$artifact['path'],
|
|
(string)$artifact['sha256']
|
|
);
|
|
}
|
|
|
|
$this->ensureExecutable($this->installDir . DIRECTORY_SEPARATOR . 'agent.php');
|
|
$this->ensureExecutable($this->installDir . DIRECTORY_SEPARATOR . 'lan-worker.php');
|
|
$this->ensureExecutable($this->installDir . DIRECTORY_SEPARATOR . 'gateway-launcher.sh');
|
|
|
|
$stagedAt = date('c');
|
|
$stagedUpdate = [
|
|
'target_version' => $targetVersion,
|
|
'staged_at' => $stagedAt,
|
|
'apply_after' => $this->nextUpdateWindowStartIso($updateWindow),
|
|
'status' => 'STAGED',
|
|
'update_window' => $updateWindow,
|
|
'compose_project_name' => (string)($request['composeProjectName'] ?? $this->config->get('composeProjectName', 'truckwash-edge-gateway')),
|
|
'stack_service_name' => (string)($request['stackServiceName'] ?? $this->config->get('stackServiceName', 'truckwash-edge-gateway-stack.service')),
|
|
'changed_files' => $changedFiles,
|
|
];
|
|
|
|
$this->emitOperationProgress($gatewayId, $operationId, 'UPDATE_STAGE_METADATA', 'Writing staged update metadata', 90, [], 'writing-config');
|
|
$this->config->set('targetVersion', $targetVersion);
|
|
$this->config->set('lastStagedUpdate', $stagedUpdate);
|
|
$this->config->save();
|
|
$this->stateStore->setJson('staged_update', $stagedUpdate);
|
|
$this->stateStore->setJson('rollback_status', [
|
|
'state' => 'IDLE',
|
|
'reason' => null,
|
|
'rolled_back_to' => null,
|
|
'at' => null,
|
|
]);
|
|
file_put_contents($this->stagedUpdatePath, json_encode($stagedUpdate, JSON_PRETTY_PRINT | JSON_UNESCAPED_SLASHES) . PHP_EOL);
|
|
|
|
$this->emitOperationProgress($gatewayId, $operationId, 'UPDATE_STAGED', 'Compose rollout staged for the maintenance window', 96, [
|
|
'target_version' => $targetVersion,
|
|
'apply_after' => $stagedUpdate['apply_after'],
|
|
], 'staged');
|
|
|
|
return [
|
|
'applied' => false,
|
|
'installed_version' => (string)$this->config->get('installedVersion', 'compose-php-agent-v2'),
|
|
'staged_version' => $targetVersion,
|
|
'target_version' => $targetVersion,
|
|
'staged_at' => $stagedAt,
|
|
'apply_after' => $stagedUpdate['apply_after'],
|
|
'update_window' => $updateWindow,
|
|
'restart_required' => true,
|
|
'service_name' => (string)($request['stackServiceName'] ?? $this->config->get('stackServiceName', 'truckwash-edge-gateway-stack.service')),
|
|
'changed_files' => $changedFiles,
|
|
'agent_instance_id' => $this->agentInstanceId,
|
|
'rollback_status' => $this->readRollbackStatus(),
|
|
];
|
|
}
|
|
|
|
private function runUninstall(int $gatewayId, int $operationId, array $request): array
|
|
{
|
|
$this->emitOperationProgress($gatewayId, $operationId, 'UNINSTALL_PREPARE', 'Preparing uninstall manifest', 40, [], 'preparing-uninstall');
|
|
$manifestPath = $this->runtimeDir . DIRECTORY_SEPARATOR . 'uninstall-plan.json';
|
|
$manifest = [
|
|
'service_name' => (string)($request['service_name'] ?? $this->config->get('stackServiceName', 'truckwash-edge-gateway-stack.service')),
|
|
'install_dir' => (string)($request['install_dir'] ?? $this->installDir),
|
|
'generated_at' => date('c'),
|
|
'agent_instance_id' => $this->agentInstanceId,
|
|
];
|
|
file_put_contents($manifestPath, json_encode($manifest, JSON_PRETTY_PRINT | JSON_UNESCAPED_SLASHES) . PHP_EOL);
|
|
file_put_contents($this->runtimeDir . DIRECTORY_SEPARATOR . 'uninstall.flag', date('c') . PHP_EOL);
|
|
$this->config->set('pendingUninstall', $manifest);
|
|
$this->config->save();
|
|
|
|
$this->emitOperationProgress($gatewayId, $operationId, 'UNINSTALL_READY', 'Gateway marked for controlled uninstall', 90, [
|
|
'manifest_path' => $manifestPath,
|
|
], 'uninstall-ready');
|
|
|
|
return [
|
|
'uninstalled' => true,
|
|
'service_name' => $manifest['service_name'],
|
|
'install_dir' => $manifest['install_dir'],
|
|
'manual_cleanup_required' => true,
|
|
'cleanup_manifest_path' => $manifestPath,
|
|
];
|
|
}
|
|
|
|
private function readRelayStatus(array $request): array
|
|
{
|
|
$localIp = (string)($request['localIp'] ?? $request['local_ip'] ?? '');
|
|
$channel = (int)($request['channel'] ?? 0);
|
|
|
|
try {
|
|
return $this->workerHttp->post('/relay/status', [
|
|
'local_ip' => $localIp,
|
|
'channel' => $channel,
|
|
], 8) ?? [];
|
|
} catch (Throwable) {
|
|
return $this->fetchShellyState($localIp, $channel);
|
|
}
|
|
}
|
|
|
|
private function switchRelay(array $request): array
|
|
{
|
|
$localIp = (string)($request['localIp'] ?? $request['local_ip'] ?? '');
|
|
$channel = (int)($request['channel'] ?? 0);
|
|
$on = (bool)($request['on'] ?? false);
|
|
|
|
try {
|
|
return $this->workerHttp->post('/relay/switch', [
|
|
'local_ip' => $localIp,
|
|
'channel' => $channel,
|
|
'on' => $on,
|
|
], 8) ?? [];
|
|
} catch (Throwable) {
|
|
$rpcUrl = sprintf('http://%s/rpc/Switch.Set?id=%d&on=%s', $localIp, $channel, $on ? 'true' : 'false');
|
|
try {
|
|
$this->http->getJson($rpcUrl, 8);
|
|
} catch (Throwable) {
|
|
$legacyUrl = sprintf('http://%s/relay/%d?turn=%s', $localIp, $channel, $on ? 'on' : 'off');
|
|
$this->http->getJson($legacyUrl, 8);
|
|
}
|
|
|
|
return $this->fetchShellyState($localIp, $channel);
|
|
}
|
|
}
|
|
|
|
private function downloadToFileAtomic(string $url, string $path, string $expectedSha256 = ''): array
|
|
{
|
|
$directory = dirname($path);
|
|
if (!is_dir($directory)) {
|
|
@mkdir($directory, 0777, true);
|
|
}
|
|
|
|
$temporaryPath = $path . '.download';
|
|
$backupPath = $path . '.bak';
|
|
$raw = $this->http->download($url, 60);
|
|
file_put_contents($temporaryPath, $raw);
|
|
|
|
if ($expectedSha256 !== '') {
|
|
$actualSha = hash_file('sha256', $temporaryPath);
|
|
if (!is_string($actualSha) || !hash_equals(strtolower($expectedSha256), strtolower($actualSha))) {
|
|
@unlink($temporaryPath);
|
|
throw new RuntimeException('Artifact checksum mismatch for ' . basename($path));
|
|
}
|
|
}
|
|
|
|
if (is_file($backupPath)) {
|
|
@unlink($backupPath);
|
|
}
|
|
|
|
if (is_file($path) && !@rename($path, $backupPath)) {
|
|
@unlink($temporaryPath);
|
|
throw new RuntimeException('Could not create backup for ' . $path);
|
|
}
|
|
|
|
if (!@rename($temporaryPath, $path)) {
|
|
if (is_file($backupPath)) {
|
|
@rename($backupPath, $path);
|
|
}
|
|
@unlink($temporaryPath);
|
|
throw new RuntimeException('Could not promote downloaded artifact for ' . $path);
|
|
}
|
|
|
|
return [
|
|
'path' => $path,
|
|
'url' => $url,
|
|
'backup_path' => is_file($backupPath) ? $backupPath : null,
|
|
'sha256' => hash_file('sha256', $path),
|
|
];
|
|
}
|
|
|
|
private function classifyManagementError(Throwable $throwable, string $type): string
|
|
{
|
|
$message = strtolower(trim($throwable->getMessage()));
|
|
if (str_contains($message, 'token') || str_contains($message, 'credential')) {
|
|
return 'EDGE_GATEWAY_INVALID_TOKEN';
|
|
}
|
|
if (str_contains($message, 'timeout') || str_contains($message, 'http 5')) {
|
|
return 'EDGE_GATEWAY_OPERATION_TIMEOUT';
|
|
}
|
|
if ($type === 'UPDATE' && (str_contains($message, 'version') || str_contains($message, 'checksum'))) {
|
|
return 'EDGE_GATEWAY_UNSUPPORTED_VERSION';
|
|
}
|
|
|
|
return 'EDGE_GATEWAY_VALIDATION_FAILED';
|
|
}
|
|
|
|
private function fetchShellyState(string $localIp, int $channel): array
|
|
{
|
|
if ($localIp === '') {
|
|
throw new RuntimeException('Missing Shelly IP address');
|
|
}
|
|
|
|
try {
|
|
$payload = $this->http->getJson(sprintf('http://%s/rpc/Switch.GetStatus?id=%d', $localIp, $channel), 8);
|
|
return [
|
|
'online' => true,
|
|
'on' => (bool)($payload['output'] ?? false),
|
|
'output' => (bool)($payload['output'] ?? false),
|
|
'raw' => $payload,
|
|
];
|
|
} catch (Throwable) {
|
|
$payload = $this->http->getJson(sprintf('http://%s/relay/%d', $localIp, $channel), 8);
|
|
return [
|
|
'online' => true,
|
|
'on' => (bool)($payload['ison'] ?? false),
|
|
'output' => (bool)($payload['ison'] ?? false),
|
|
'raw' => $payload,
|
|
];
|
|
}
|
|
}
|
|
|
|
private function buildSystemMetrics(): array
|
|
{
|
|
$diskTotal = @disk_total_space($this->installDir);
|
|
$diskFree = @disk_free_space($this->installDir);
|
|
$diskUsed = (is_numeric($diskTotal) && is_numeric($diskFree)) ? ((float)$diskTotal - (float)$diskFree) : null;
|
|
$diskUsagePct = ($diskTotal && $diskUsed !== null && $diskTotal > 0)
|
|
? (int)round(($diskUsed / (float)$diskTotal) * 100)
|
|
: null;
|
|
$memoryLimitBytes = $this->iniBytes((string)ini_get('memory_limit'));
|
|
$memoryUsage = memory_get_usage(true);
|
|
$memoryUsagePct = ($memoryLimitBytes !== null && $memoryLimitBytes > 0)
|
|
? (int)round(($memoryUsage / $memoryLimitBytes) * 100)
|
|
: null;
|
|
$loadAverage = function_exists('sys_getloadavg') ? sys_getloadavg() : [];
|
|
|
|
return [
|
|
'memory_usage_bytes' => $memoryUsage,
|
|
'memory_peak_bytes' => memory_get_peak_usage(true),
|
|
'memory_usage_pct' => $memoryUsagePct,
|
|
'cpu_usage_pct' => is_array($loadAverage) && isset($loadAverage[0]) ? max(0, (int)round((float)$loadAverage[0] * 100)) : null,
|
|
'load_average' => $loadAverage,
|
|
'disk_usage_pct' => $diskUsagePct,
|
|
'disk_used_bytes' => $diskUsed,
|
|
'disk_total_bytes' => is_numeric($diskTotal) ? (int)$diskTotal : null,
|
|
'disk_mount' => $this->installDir,
|
|
'latency_ms' => null,
|
|
];
|
|
}
|
|
|
|
private function buildContainerHealth(): array
|
|
{
|
|
$services = [[
|
|
'name' => 'edge-agent',
|
|
'status' => 'healthy',
|
|
'updated_at' => date('c'),
|
|
]];
|
|
|
|
try {
|
|
$workerHealth = $this->workerHttp->getJson(rtrim((string)$this->config->get('workerBaseUrl', self::DEFAULT_WORKER_BASE_URL), '/') . '/health', 5);
|
|
$services[] = [
|
|
'name' => 'lan-worker',
|
|
'status' => (string)($workerHealth['status'] ?? 'healthy'),
|
|
'updated_at' => (string)($workerHealth['timestamp'] ?? date('c')),
|
|
];
|
|
} catch (Throwable $throwable) {
|
|
$services[] = [
|
|
'name' => 'lan-worker',
|
|
'status' => 'degraded',
|
|
'updated_at' => date('c'),
|
|
'error' => $throwable->getMessage(),
|
|
];
|
|
}
|
|
|
|
$healthyCount = count(array_filter($services, static fn(array $service): bool => (string)($service['status'] ?? '') === 'healthy'));
|
|
$state = $healthyCount === count($services) ? 'ONLINE' : 'DEGRADED';
|
|
|
|
return [
|
|
'state' => $state,
|
|
'summary' => sprintf('%d/%d containers healthy', $healthyCount, count($services)),
|
|
'services' => $services,
|
|
];
|
|
}
|
|
|
|
private function buildOutboxStatus(): array
|
|
{
|
|
$summary = $this->stateStore->outboxSummary();
|
|
$queued = (int)($summary['queued'] ?? 0);
|
|
return [
|
|
'state' => $queued > 0 ? 'QUEUED' : 'IN_SYNC',
|
|
'queued' => $queued,
|
|
'oldest_queued_at' => $summary['oldest_queued_at'] ?? null,
|
|
'last_replayed_at' => $summary['last_replayed_at'] ?? null,
|
|
'summary' => $queued > 0 ? sprintf('%d outbound items queued', $queued) : 'Outbox is empty',
|
|
];
|
|
}
|
|
|
|
private function flushOutbox(): void
|
|
{
|
|
$items = $this->stateStore->queuedItems(25);
|
|
foreach ($items as $item) {
|
|
try {
|
|
$this->http->post((string)$item['endpoint'], is_array($item['payload'] ?? null) ? (array)$item['payload'] : [], 20);
|
|
$this->stateStore->removeOutboxItem((int)$item['id']);
|
|
$this->stateStore->setJson('last_sync_at', date('c'));
|
|
} catch (Throwable $throwable) {
|
|
$this->logger->warning('Outbox replay blocked on ' . (string)$item['type'] . ': ' . $throwable->getMessage());
|
|
break;
|
|
}
|
|
}
|
|
}
|
|
|
|
private function sendControlPlaneEvent(string $endpoint, array $payload, string $type): bool
|
|
{
|
|
try {
|
|
$this->http->post($endpoint, $payload, 20);
|
|
$this->stateStore->setJson('last_sync_at', date('c'));
|
|
return true;
|
|
} catch (Throwable $throwable) {
|
|
$this->stateStore->enqueue($type, $endpoint, $payload);
|
|
$this->logger->warning('Queued ' . $type . ' to local outbox after transport failure: ' . $throwable->getMessage());
|
|
return false;
|
|
}
|
|
}
|
|
|
|
private function readRollbackStatus(): array
|
|
{
|
|
$rollbackPath = $this->runtimeDir . DIRECTORY_SEPARATOR . 'rollback-status.json';
|
|
if (is_file($rollbackPath)) {
|
|
$decoded = json_decode((string)file_get_contents($rollbackPath), true);
|
|
if (is_array($decoded)) {
|
|
return $decoded;
|
|
}
|
|
}
|
|
|
|
$rollback = $this->stateStore->getJson('rollback_status', null);
|
|
if (is_array($rollback)) {
|
|
return $rollback;
|
|
}
|
|
|
|
return [
|
|
'state' => 'IDLE',
|
|
'reason' => null,
|
|
'rolled_back_to' => null,
|
|
'at' => null,
|
|
];
|
|
}
|
|
|
|
private function currentStagedUpdate(): ?array
|
|
{
|
|
if (is_file($this->stagedUpdatePath)) {
|
|
$decoded = json_decode((string)file_get_contents($this->stagedUpdatePath), true);
|
|
if (is_array($decoded)) {
|
|
return $decoded;
|
|
}
|
|
}
|
|
|
|
$staged = $this->stateStore->getJson('staged_update', null);
|
|
return is_array($staged) ? $staged : null;
|
|
}
|
|
|
|
private function ensureExecutable(string $path): void
|
|
{
|
|
if (is_file($path)) {
|
|
@chmod($path, 0755);
|
|
}
|
|
}
|
|
|
|
private function ensureAgentInstanceId(): string
|
|
{
|
|
$configured = trim((string)$this->config->get('agentInstanceId', ''));
|
|
if ($configured !== '') {
|
|
return substr($configured, 0, 128);
|
|
}
|
|
|
|
$generated = sprintf(
|
|
'%s-%s',
|
|
preg_replace('/[^A-Za-z0-9\-]+/', '-', gethostname() ?: 'truckwash-edge') ?: 'truckwash-edge',
|
|
substr(bin2hex(random_bytes(6)), 0, 12)
|
|
);
|
|
$this->config->set('agentInstanceId', $generated);
|
|
$this->config->save();
|
|
return $generated;
|
|
}
|
|
|
|
private function recoverPreviousOperationState(): void
|
|
{
|
|
$state = $this->readOperationState();
|
|
if ($state === null) {
|
|
return;
|
|
}
|
|
|
|
$this->logger->warning(sprintf(
|
|
'Recovered previous unfinished operation snapshot: operation %s (%s) last stage %s.',
|
|
(string)($state['operation_id'] ?? 'unknown'),
|
|
(string)($state['type'] ?? 'unknown'),
|
|
(string)($state['stage'] ?? 'unknown')
|
|
));
|
|
}
|
|
|
|
private function persistOperationState(array $state): void
|
|
{
|
|
file_put_contents($this->statePath, json_encode($state, JSON_PRETTY_PRINT | JSON_UNESCAPED_SLASHES) . PHP_EOL);
|
|
}
|
|
|
|
private function snapshotOperationState(string $status, array $extra = []): void
|
|
{
|
|
$state = $this->readOperationState() ?? [];
|
|
$state['status'] = $status;
|
|
$state['finished_at'] = date('c');
|
|
$state = array_merge($state, $extra);
|
|
file_put_contents($this->lastOperationSnapshotPath, json_encode($state, JSON_PRETTY_PRINT | JSON_UNESCAPED_SLASHES) . PHP_EOL);
|
|
}
|
|
|
|
private function clearOperationState(): void
|
|
{
|
|
if (is_file($this->statePath)) {
|
|
@unlink($this->statePath);
|
|
}
|
|
}
|
|
|
|
private function readOperationState(): ?array
|
|
{
|
|
if (!is_file($this->statePath)) {
|
|
return null;
|
|
}
|
|
|
|
$decoded = json_decode((string)file_get_contents($this->statePath), true);
|
|
return is_array($decoded) ? $decoded : null;
|
|
}
|
|
|
|
private function nextUpdateWindowStartIso(string $window): string
|
|
{
|
|
$parts = explode('-', $window, 2);
|
|
$start = trim((string)($parts[0] ?? '02:00'));
|
|
if (!preg_match('/^\d{2}:\d{2}$/', $start)) {
|
|
$start = '02:00';
|
|
}
|
|
|
|
$now = new DateTimeImmutable('now');
|
|
[$hour, $minute] = array_map('intval', explode(':', $start));
|
|
$candidate = $now->setTime($hour, $minute, 0);
|
|
if ($candidate <= $now) {
|
|
$candidate = $candidate->modify('+1 day');
|
|
}
|
|
|
|
return $candidate->format(DateTimeInterface::ATOM);
|
|
}
|
|
|
|
private function reloadConfigFromDisk(): void
|
|
{
|
|
$reloaded = AgentConfig::load($this->config->path);
|
|
$this->config = $reloaded;
|
|
$this->workerHttp = new HttpJsonClient((string)$this->config->get('workerBaseUrl', self::DEFAULT_WORKER_BASE_URL));
|
|
}
|
|
|
|
private function iniBytes(string $value): ?int
|
|
{
|
|
$normalized = trim(strtolower($value));
|
|
if ($normalized === '' || $normalized === '-1') {
|
|
return null;
|
|
}
|
|
|
|
$unit = substr($normalized, -1);
|
|
$number = (float)$normalized;
|
|
return match ($unit) {
|
|
'g' => (int)round($number * 1024 * 1024 * 1024),
|
|
'm' => (int)round($number * 1024 * 1024),
|
|
'k' => (int)round($number * 1024),
|
|
default => is_numeric($normalized) ? (int)$normalized : null,
|
|
};
|
|
}
|
|
}
|
|
|
|
$configPath = null;
|
|
foreach ($argv as $index => $argument) {
|
|
if ($argument === '--config' && isset($argv[$index + 1])) {
|
|
$configPath = $argv[$index + 1];
|
|
}
|
|
}
|
|
|
|
if ($configPath === null) {
|
|
fwrite(STDERR, "Usage: php agent.php --config /path/to/config.json\n");
|
|
exit(1);
|
|
}
|
|
|
|
$agent = new TruckwashEdgeAgent($configPath);
|
|
$agent->run();
|