991 lines
37 KiB
PHP
991 lines
37 KiB
PHP
<?php
|
|
|
|
namespace classes;
|
|
|
|
use Exception;
|
|
use objects\edge_gateway_operation_events_o;
|
|
use objects\edge_gateway_operations_o;
|
|
use objects\edge_gateways_o;
|
|
|
|
class edge_gateway_operation_service
|
|
{
|
|
public const TYPE_DISCOVERY = 'DISCOVERY';
|
|
public const TYPE_UPDATE = 'UPDATE';
|
|
public const TYPE_UNINSTALL = 'UNINSTALL';
|
|
|
|
public const STATUS_PENDING = 'PENDING';
|
|
public const STATUS_IN_PROGRESS = 'IN_PROGRESS';
|
|
public const STATUS_CANCEL_REQUESTED = 'CANCEL_REQUESTED';
|
|
public const STATUS_CANCELLED = 'CANCELLED';
|
|
public const STATUS_COMPLETED = 'COMPLETED';
|
|
public const STATUS_FAILED = 'FAILED';
|
|
|
|
public const LEVEL_INFO = 'INFO';
|
|
public const LEVEL_WARNING = 'WARNING';
|
|
public const LEVEL_ERROR = 'ERROR';
|
|
|
|
public const ERROR_OFFLINE = 'EDGE_GATEWAY_OFFLINE';
|
|
public const ERROR_STALE_HEARTBEAT = 'EDGE_GATEWAY_STALE_HEARTBEAT';
|
|
public const ERROR_INVALID_TOKEN = 'EDGE_GATEWAY_INVALID_TOKEN';
|
|
public const ERROR_OPERATION_TIMEOUT = 'EDGE_GATEWAY_OPERATION_TIMEOUT';
|
|
public const ERROR_UNSUPPORTED_VERSION = 'EDGE_GATEWAY_UNSUPPORTED_VERSION';
|
|
public const ERROR_CONFLICT = 'EDGE_GATEWAY_CONFLICT';
|
|
public const ERROR_VALIDATION = 'EDGE_GATEWAY_VALIDATION_FAILED';
|
|
public const ERROR_CANCELLED = 'EDGE_GATEWAY_CANCELLED';
|
|
|
|
public const POLL_INTERVAL_MICROSECONDS = 250000;
|
|
public const OPERATION_TIMEOUT_SECONDS = 900;
|
|
public const OPERATION_LEASE_SECONDS = 45;
|
|
|
|
public function __construct(private readonly ?edge_gateway_manager $manager = null)
|
|
{
|
|
edge_gateway_schema_bootstrap::ensureTables();
|
|
}
|
|
|
|
/**
|
|
* @return array<int,array<string,mixed>>
|
|
* @throws Exception
|
|
*/
|
|
public function listOperations(int $gatewayId, int $limit = 20, bool $includeEvents = true): array
|
|
{
|
|
$this->requireGateway($gatewayId);
|
|
$this->failTimedOutOperations($gatewayId);
|
|
$rows = (new edge_gateway_operations_o())->getFieldsWhere([
|
|
'gateway_id' => $gatewayId,
|
|
'deleted_at' => null,
|
|
], ['id']);
|
|
|
|
$ids = array_map(static fn(array $row): int => (int)$row['id'], $rows);
|
|
rsort($ids);
|
|
$ids = array_slice($ids, 0, max(1, $limit));
|
|
|
|
$operations = [];
|
|
foreach ($ids as $id) {
|
|
$operations[] = $this->serializeOperation((new edge_gateway_operations_o())->select($id), $includeEvents);
|
|
}
|
|
|
|
return $operations;
|
|
}
|
|
|
|
/**
|
|
* @return array<int,array<string,mixed>>
|
|
* @throws Exception
|
|
*/
|
|
public function listOperationEvents(int $gatewayId, int $operationId, int $limit = 100): array
|
|
{
|
|
$this->requireOperation($gatewayId, $operationId);
|
|
|
|
$rows = (new edge_gateway_operation_events_o())->getFieldsWhere([
|
|
'gateway_id' => $gatewayId,
|
|
'operation_id' => $operationId,
|
|
], ['id']);
|
|
|
|
$ids = array_map(static fn(array $row): int => (int)$row['id'], $rows);
|
|
sort($ids);
|
|
$ids = array_slice($ids, max(0, count($ids) - max(1, $limit)));
|
|
|
|
$events = [];
|
|
foreach ($ids as $id) {
|
|
$events[] = (new edge_gateway_operation_events_o())->select($id)->asArray();
|
|
}
|
|
|
|
return $events;
|
|
}
|
|
|
|
/**
|
|
* @throws Exception
|
|
*/
|
|
public function getActiveOperation(int $gatewayId, bool $includeEvents = true): ?array
|
|
{
|
|
$this->requireGateway($gatewayId);
|
|
$this->failTimedOutOperations($gatewayId);
|
|
|
|
$statement = db::getPDO()->prepare(
|
|
"SELECT id
|
|
FROM edge_gateway_operations
|
|
WHERE gateway_id = :gateway_id
|
|
AND deleted_at IS NULL
|
|
AND status IN ('PENDING', 'IN_PROGRESS', 'CANCEL_REQUESTED')
|
|
ORDER BY FIELD(status, 'IN_PROGRESS', 'CANCEL_REQUESTED', 'PENDING'), id ASC
|
|
LIMIT 1"
|
|
);
|
|
$statement->execute([':gateway_id' => $gatewayId]);
|
|
$row = $statement->fetch();
|
|
|
|
if (!is_array($row) || empty($row['id'])) {
|
|
return null;
|
|
}
|
|
|
|
return $this->serializeOperation((new edge_gateway_operations_o())->select((int)$row['id']), $includeEvents);
|
|
}
|
|
|
|
/**
|
|
* @throws Exception
|
|
*/
|
|
public function buildRecentOperationsSummary(int $gatewayId): array
|
|
{
|
|
$operations = $this->listOperations($gatewayId, 25, false);
|
|
$summary = [
|
|
'total' => count($operations),
|
|
'pending' => 0,
|
|
'in_progress' => 0,
|
|
'cancel_requested' => 0,
|
|
'cancelled' => 0,
|
|
'completed' => 0,
|
|
'failed' => 0,
|
|
'latest_cancelled_at' => null,
|
|
'latest_completed_at' => null,
|
|
'latest_failed_at' => null,
|
|
'latest_type' => $operations[0]['type'] ?? null,
|
|
'latest_status' => $operations[0]['status'] ?? null,
|
|
];
|
|
|
|
foreach ($operations as $operation) {
|
|
$status = (string)($operation['status'] ?? self::STATUS_PENDING);
|
|
if ($status === self::STATUS_PENDING) {
|
|
$summary['pending'] += 1;
|
|
} elseif ($status === self::STATUS_IN_PROGRESS) {
|
|
$summary['in_progress'] += 1;
|
|
} elseif ($status === self::STATUS_CANCEL_REQUESTED) {
|
|
$summary['cancel_requested'] += 1;
|
|
} elseif ($status === self::STATUS_CANCELLED) {
|
|
$summary['cancelled'] += 1;
|
|
$summary['latest_cancelled_at'] ??= $operation['completed_at'] ?? null;
|
|
} elseif ($status === self::STATUS_COMPLETED) {
|
|
$summary['completed'] += 1;
|
|
$summary['latest_completed_at'] ??= $operation['completed_at'] ?? null;
|
|
} elseif ($status === self::STATUS_FAILED) {
|
|
$summary['failed'] += 1;
|
|
$summary['latest_failed_at'] ??= $operation['completed_at'] ?? null;
|
|
}
|
|
}
|
|
|
|
return $summary;
|
|
}
|
|
|
|
/**
|
|
* @throws Exception
|
|
*/
|
|
public function queueOperation(int $gatewayId, string $type, array $request = [], ?int $requestedBy = null): array
|
|
{
|
|
$gateway = $this->requireGateway($gatewayId);
|
|
$type = self::normalizeOperationType($type);
|
|
$request = $this->validateOperationRequest($gateway, $type, $request);
|
|
|
|
if ($this->getActiveOperation($gatewayId) !== null) {
|
|
throw new edge_gateway_operation_exception(
|
|
'Another gateway operation is already active',
|
|
self::ERROR_CONFLICT,
|
|
409
|
|
);
|
|
}
|
|
|
|
if ($type === self::TYPE_DISCOVERY) {
|
|
$gateway->discovery_status->set('PENDING');
|
|
} elseif ($type === self::TYPE_UPDATE) {
|
|
$targetVersion = trim((string)($request['target_version'] ?? ''));
|
|
if ($targetVersion !== '') {
|
|
$gateway->target_version->set($targetVersion);
|
|
}
|
|
}
|
|
|
|
$summary = [
|
|
'label' => match ($type) {
|
|
self::TYPE_DISCOVERY => 'Discovery queued',
|
|
self::TYPE_UPDATE => 'Update queued',
|
|
self::TYPE_UNINSTALL => 'Uninstall queued',
|
|
},
|
|
'progress' => 0,
|
|
'retryable' => true,
|
|
];
|
|
|
|
$operationId = (new edge_gateway_operations_o())->add_object([
|
|
'gateway_id' => $gatewayId,
|
|
'type' => $type,
|
|
'operation_type' => $type,
|
|
'status' => self::STATUS_PENDING,
|
|
'request_json' => $request,
|
|
'summary_json' => $summary,
|
|
'result_json' => [],
|
|
'error_code' => null,
|
|
'error_message' => null,
|
|
'correlation_id' => bin2hex(random_bytes(16)),
|
|
'agent_instance_id' => null,
|
|
'lease_expires_at' => null,
|
|
'last_progress_at' => null,
|
|
'attempt_count' => 0,
|
|
'requested_by' => $requestedBy,
|
|
'requested_at' => $this->now(),
|
|
'started_at' => null,
|
|
'completed_at' => null,
|
|
]);
|
|
|
|
$this->appendEventRecord(
|
|
$gatewayId,
|
|
$operationId,
|
|
self::LEVEL_INFO,
|
|
'OPERATION_QUEUED',
|
|
'Operation queued for gateway execution',
|
|
['type' => $type, 'request' => $request]
|
|
);
|
|
|
|
$this->manager()->logGatewayAudit(
|
|
$gatewayId,
|
|
(int)$gateway->department_id->value(),
|
|
'GATEWAY_OPERATION_QUEUED',
|
|
$requestedBy,
|
|
['operation_id' => $operationId, 'type' => $type]
|
|
);
|
|
|
|
$this->refreshGatewayViewCache($gatewayId);
|
|
|
|
return $this->serializeOperation((new edge_gateway_operations_o())->select($operationId), true);
|
|
}
|
|
|
|
/**
|
|
* @throws Exception
|
|
*/
|
|
public function queueDiscoveryOperation(int $gatewayId, ?int $requestedBy = null): array
|
|
{
|
|
return $this->queueOperation($gatewayId, self::TYPE_DISCOVERY, [], $requestedBy);
|
|
}
|
|
|
|
/**
|
|
* @throws Exception
|
|
*/
|
|
public function cancelOperation(int $gatewayId, int $operationId, ?int $requestedBy = null): array
|
|
{
|
|
$gateway = $this->requireGateway($gatewayId);
|
|
$operation = $this->requireOperation($gatewayId, $operationId);
|
|
$status = (string)$operation->status->value();
|
|
|
|
if (in_array($status, [self::STATUS_COMPLETED, self::STATUS_FAILED, self::STATUS_CANCELLED], true)) {
|
|
return $this->serializeOperation($operation, true);
|
|
}
|
|
|
|
$message = 'Operation cancelled by operator';
|
|
if ($status === self::STATUS_PENDING) {
|
|
$this->markOperationCancelled(
|
|
$gatewayId,
|
|
$operation,
|
|
$message,
|
|
'OPERATION_CANCELLED',
|
|
['requested_by' => $requestedBy]
|
|
);
|
|
$this->manager()->logGatewayAudit(
|
|
$gatewayId,
|
|
(int)$gateway->department_id->value(),
|
|
'GATEWAY_OPERATION_CANCELLED',
|
|
$requestedBy,
|
|
['operation_id' => $operationId, 'type' => (string)$operation->type->value()]
|
|
);
|
|
} elseif ($status === self::STATUS_IN_PROGRESS) {
|
|
$operation->status->set(self::STATUS_CANCEL_REQUESTED);
|
|
$operation->error_code->set(self::ERROR_CANCELLED);
|
|
$operation->error_message->set('Operation cancellation requested by operator');
|
|
$operation->last_progress_at->set($this->now());
|
|
$summary = (array)($operation->summary_json->value() ?? []);
|
|
$summary['label'] = 'Cancellation requested';
|
|
$summary['retryable'] = true;
|
|
$operation->summary_json->set($summary);
|
|
$this->appendEventRecord(
|
|
$gatewayId,
|
|
$operationId,
|
|
self::LEVEL_WARNING,
|
|
'OPERATION_CANCEL_REQUESTED',
|
|
'Operation cancellation requested by operator',
|
|
['requested_by' => $requestedBy]
|
|
);
|
|
$this->manager()->logGatewayAudit(
|
|
$gatewayId,
|
|
(int)$gateway->department_id->value(),
|
|
'GATEWAY_OPERATION_CANCEL_REQUESTED',
|
|
$requestedBy,
|
|
['operation_id' => $operationId, 'type' => (string)$operation->type->value()]
|
|
);
|
|
}
|
|
|
|
$this->refreshGatewayViewCache($gatewayId);
|
|
|
|
return $this->serializeOperation($operation, true);
|
|
}
|
|
|
|
/**
|
|
* @throws Exception
|
|
*/
|
|
public function rotateCredentials(int $gatewayId, ?int $userId = null): array
|
|
{
|
|
return $this->manager()->rotateGatewayCredentials($gatewayId, $userId);
|
|
}
|
|
|
|
/**
|
|
* @throws Exception
|
|
*/
|
|
public function claimNextOperation(
|
|
int $gatewayId,
|
|
string $plainToken,
|
|
int $waitSeconds = edge_gateway_manager::COMMAND_POLL_TIMEOUT_SECONDS,
|
|
?string $agentInstanceId = null
|
|
): ?array {
|
|
try {
|
|
$this->manager()->authenticateGateway($gatewayId, $plainToken);
|
|
} catch (Exception) {
|
|
throw new edge_gateway_operation_exception('Invalid edge gateway token', self::ERROR_INVALID_TOKEN, 401);
|
|
}
|
|
|
|
$deadline = microtime(true) + max(0, $waitSeconds);
|
|
do {
|
|
$this->failTimedOutOperations($gatewayId);
|
|
$operation = $this->claimPendingOperation($gatewayId, $this->normalizeAgentInstanceId($agentInstanceId, $gatewayId));
|
|
if ($operation !== null) {
|
|
return $operation;
|
|
}
|
|
|
|
if (microtime(true) >= $deadline) {
|
|
break;
|
|
}
|
|
|
|
usleep(self::POLL_INTERVAL_MICROSECONDS);
|
|
} while (true);
|
|
|
|
return null;
|
|
}
|
|
|
|
/**
|
|
* @throws Exception
|
|
*/
|
|
public function appendAgentOperationEvent(int $gatewayId, int $operationId, string $plainToken, array $payload): array
|
|
{
|
|
try {
|
|
$this->manager()->authenticateGateway($gatewayId, $plainToken);
|
|
} catch (Exception) {
|
|
throw new edge_gateway_operation_exception('Invalid edge gateway token', self::ERROR_INVALID_TOKEN, 401);
|
|
}
|
|
|
|
$operation = $this->requireOperation($gatewayId, $operationId);
|
|
if (in_array((string)$operation->status->value(), [
|
|
self::STATUS_COMPLETED,
|
|
self::STATUS_FAILED,
|
|
self::STATUS_CANCEL_REQUESTED,
|
|
self::STATUS_CANCELLED,
|
|
], true)) {
|
|
return $this->serializeOperation($operation, true);
|
|
}
|
|
|
|
$level = strtoupper(trim((string)($payload['level'] ?? self::LEVEL_INFO)));
|
|
if (!in_array($level, [self::LEVEL_INFO, self::LEVEL_WARNING, self::LEVEL_ERROR], true)) {
|
|
$level = self::LEVEL_INFO;
|
|
}
|
|
|
|
$message = trim((string)($payload['message'] ?? 'Operation event received'));
|
|
if ($message === '') {
|
|
$message = 'Operation event received';
|
|
}
|
|
|
|
$code = isset($payload['code']) ? trim((string)$payload['code']) : null;
|
|
$context = isset($payload['context']) && is_array($payload['context']) ? (array)$payload['context'] : [];
|
|
$this->appendEventRecord($gatewayId, $operationId, $level, $code, $message, $context);
|
|
|
|
$summary = (array)($operation->summary_json->value() ?? []);
|
|
$summary['last_event_at'] = $this->now();
|
|
$summary['last_event_message'] = $message;
|
|
if (isset($context['progress'])) {
|
|
$summary['progress'] = max(0, min(100, (int)$context['progress']));
|
|
}
|
|
if (isset($context['label']) && trim((string)$context['label']) !== '') {
|
|
$summary['label'] = trim((string)$context['label']);
|
|
}
|
|
$operation->summary_json->set($summary);
|
|
$this->refreshOperationLease($operation);
|
|
|
|
if ($level === self::LEVEL_ERROR) {
|
|
$this->markOperationFailedFromEvent($gatewayId, $operation, $code, $message, $context);
|
|
}
|
|
|
|
$this->refreshGatewayViewCache($gatewayId);
|
|
|
|
return $this->serializeOperation($operation, true);
|
|
}
|
|
|
|
/**
|
|
* @throws Exception
|
|
*/
|
|
public function completeAgentOperation(int $gatewayId, int $operationId, string $plainToken, array $payload): array
|
|
{
|
|
try {
|
|
$this->manager()->authenticateGateway($gatewayId, $plainToken);
|
|
} catch (Exception) {
|
|
throw new edge_gateway_operation_exception('Invalid edge gateway token', self::ERROR_INVALID_TOKEN, 401);
|
|
}
|
|
|
|
$operation = $this->requireOperation($gatewayId, $operationId);
|
|
if (in_array((string)$operation->status->value(), [
|
|
self::STATUS_COMPLETED,
|
|
self::STATUS_FAILED,
|
|
self::STATUS_CANCELLED,
|
|
], true)) {
|
|
return $this->serializeOperation($operation, true);
|
|
}
|
|
|
|
$ok = (bool)($payload['ok'] ?? false);
|
|
$result = isset($payload['result']) && is_array($payload['result']) ? (array)$payload['result'] : [];
|
|
$errorMessage = trim((string)($payload['error_message'] ?? $payload['error'] ?? ''));
|
|
$errorCode = trim((string)($payload['error_code'] ?? ''));
|
|
$status = (string)$operation->status->value();
|
|
if ($status === self::STATUS_CANCEL_REQUESTED && !$ok && $errorCode === '') {
|
|
$errorCode = self::ERROR_CANCELLED;
|
|
}
|
|
if (!$ok && $errorCode === '') {
|
|
$errorCode = $this->classifyCompletionError($gatewayId, $errorMessage);
|
|
}
|
|
if (!$ok && $errorMessage === '') {
|
|
$errorMessage = $errorCode === self::ERROR_CANCELLED
|
|
? 'Gateway operation cancelled'
|
|
: 'Gateway operation failed';
|
|
}
|
|
|
|
$finalStatus = $ok
|
|
? self::STATUS_COMPLETED
|
|
: (($status === self::STATUS_CANCEL_REQUESTED && $errorCode === self::ERROR_CANCELLED)
|
|
? self::STATUS_CANCELLED
|
|
: self::STATUS_FAILED);
|
|
$operation->status->set($finalStatus);
|
|
$operation->result_json->set($result);
|
|
$operation->error_code->set($ok ? null : ($finalStatus === self::STATUS_CANCELLED ? self::ERROR_CANCELLED : $errorCode));
|
|
$operation->error_message->set($ok ? null : $errorMessage);
|
|
$operation->completed_at->set($this->now());
|
|
$operation->lease_expires_at->set(null);
|
|
$operation->last_progress_at->set($this->now());
|
|
|
|
$summary = (array)($operation->summary_json->value() ?? []);
|
|
$summary['label'] = match ($finalStatus) {
|
|
self::STATUS_COMPLETED => 'Completed',
|
|
self::STATUS_CANCELLED => 'Cancelled',
|
|
default => 'Failed',
|
|
};
|
|
$summary['progress'] = $finalStatus === self::STATUS_CANCELLED
|
|
? max(0, min(100, (int)($summary['progress'] ?? 0)))
|
|
: 100;
|
|
$summary['retryable'] = $finalStatus === self::STATUS_CANCELLED
|
|
? true
|
|
: (!$ok && $errorCode !== self::ERROR_UNSUPPORTED_VERSION);
|
|
$operation->summary_json->set($summary);
|
|
|
|
$this->appendEventRecord(
|
|
$gatewayId,
|
|
$operationId,
|
|
$finalStatus === self::STATUS_COMPLETED
|
|
? self::LEVEL_INFO
|
|
: ($finalStatus === self::STATUS_CANCELLED ? self::LEVEL_WARNING : self::LEVEL_ERROR),
|
|
$finalStatus === self::STATUS_COMPLETED
|
|
? 'OPERATION_COMPLETED'
|
|
: ($finalStatus === self::STATUS_CANCELLED ? 'OPERATION_CANCELLED' : $errorCode),
|
|
$finalStatus === self::STATUS_COMPLETED
|
|
? 'Operation completed successfully'
|
|
: $errorMessage,
|
|
$result
|
|
);
|
|
|
|
if ($finalStatus !== self::STATUS_CANCELLED) {
|
|
$this->applyCompletionSideEffects(
|
|
$gatewayId,
|
|
$operation,
|
|
$ok,
|
|
$result,
|
|
$errorCode,
|
|
$errorMessage
|
|
);
|
|
}
|
|
$this->refreshGatewayViewCache($gatewayId);
|
|
|
|
return $this->serializeOperation($operation, true);
|
|
}
|
|
|
|
private static function normalizeOperationType(string $type): string
|
|
{
|
|
$normalized = strtoupper(trim($type));
|
|
if (in_array($normalized, [self::TYPE_DISCOVERY, self::TYPE_UPDATE, self::TYPE_UNINSTALL], true)) {
|
|
return $normalized;
|
|
}
|
|
|
|
throw new edge_gateway_operation_exception(
|
|
'Unsupported gateway operation type',
|
|
self::ERROR_VALIDATION,
|
|
422
|
|
);
|
|
}
|
|
|
|
/**
|
|
* @throws Exception
|
|
*/
|
|
private function validateOperationRequest(edge_gateways_o $gateway, string $type, array $request): array
|
|
{
|
|
if ($type === self::TYPE_DISCOVERY) {
|
|
return $request;
|
|
}
|
|
|
|
if ($type === self::TYPE_UPDATE) {
|
|
$targetVersion = trim((string)($request['target_version'] ?? ''));
|
|
if ($targetVersion === '') {
|
|
throw new edge_gateway_operation_exception(
|
|
'Update operations require target_version',
|
|
self::ERROR_VALIDATION,
|
|
422
|
|
);
|
|
}
|
|
|
|
$releaseChannel = trim((string)($request['release_channel'] ?? $gateway->release_channel->value() ?? edge_gateway_manager::DEFAULT_RELEASE_CHANNEL));
|
|
if ($releaseChannel === '') {
|
|
$releaseChannel = edge_gateway_manager::DEFAULT_RELEASE_CHANNEL;
|
|
}
|
|
|
|
return array_merge(
|
|
$this->manager()->buildUpdateOperationRequest($targetVersion, $releaseChannel),
|
|
$request,
|
|
[
|
|
'target_version' => $targetVersion,
|
|
'release_channel' => $releaseChannel,
|
|
]
|
|
);
|
|
}
|
|
|
|
if ($type === self::TYPE_UNINSTALL) {
|
|
return array_merge([
|
|
'service_name' => edge_gateway_manager::DEFAULT_STACK_SERVICE_NAME,
|
|
'install_dir' => edge_gateway_manager::DEFAULT_INSTALL_DIR,
|
|
], $request);
|
|
}
|
|
|
|
return $request;
|
|
}
|
|
|
|
/**
|
|
* @throws Exception
|
|
*/
|
|
private function claimPendingOperation(int $gatewayId, string $agentInstanceId): ?array
|
|
{
|
|
$pdo = db::getPDO();
|
|
$pdo->beginTransaction();
|
|
|
|
try {
|
|
$statement = $pdo->prepare(
|
|
"SELECT id
|
|
FROM edge_gateway_operations
|
|
WHERE gateway_id = :gateway_id
|
|
AND deleted_at IS NULL
|
|
AND status = 'PENDING'
|
|
ORDER BY id ASC
|
|
LIMIT 1
|
|
FOR UPDATE"
|
|
);
|
|
$statement->execute([':gateway_id' => $gatewayId]);
|
|
$row = $statement->fetch();
|
|
|
|
if (!is_array($row) || empty($row['id'])) {
|
|
$pdo->commit();
|
|
return null;
|
|
}
|
|
|
|
$operation = (new edge_gateway_operations_o())->select((int)$row['id']);
|
|
$operation->status->set(self::STATUS_IN_PROGRESS);
|
|
$operation->started_at->set($this->now());
|
|
$operation->agent_instance_id->set($agentInstanceId);
|
|
$operation->last_progress_at->set($this->now());
|
|
$operation->lease_expires_at->set($this->leaseExpiry());
|
|
$operation->attempt_count->set(((int)($operation->attempt_count->value() ?? 0)) + 1);
|
|
$summary = (array)($operation->summary_json->value() ?? []);
|
|
$summary['label'] = 'Gateway is processing the operation';
|
|
$summary['progress'] = max(5, (int)($summary['progress'] ?? 0));
|
|
$summary['claimed_by'] = $agentInstanceId;
|
|
$operation->summary_json->set($summary);
|
|
$pdo->commit();
|
|
|
|
$this->appendEventRecord(
|
|
$gatewayId,
|
|
(int)$operation->id,
|
|
self::LEVEL_INFO,
|
|
'OPERATION_STARTED',
|
|
'Gateway started processing the operation',
|
|
[
|
|
'type' => (string)$operation->type->value(),
|
|
'agent_instance_id' => $agentInstanceId,
|
|
'attempt_count' => (int)($operation->attempt_count->value() ?? 1),
|
|
]
|
|
);
|
|
|
|
$this->refreshGatewayViewCache($gatewayId);
|
|
|
|
return $this->serializeOperation($operation, true);
|
|
} catch (\Throwable $throwable) {
|
|
if ($pdo->inTransaction()) {
|
|
$pdo->rollBack();
|
|
}
|
|
throw $throwable;
|
|
}
|
|
}
|
|
|
|
/**
|
|
* @throws Exception
|
|
*/
|
|
private function failTimedOutOperations(int $gatewayId): void
|
|
{
|
|
$statement = db::getPDO()->prepare(
|
|
"SELECT id
|
|
FROM edge_gateway_operations
|
|
WHERE gateway_id = :gateway_id
|
|
AND deleted_at IS NULL
|
|
AND status IN ('IN_PROGRESS', 'CANCEL_REQUESTED')"
|
|
);
|
|
$statement->execute([':gateway_id' => $gatewayId]);
|
|
$rows = $statement->fetchAll();
|
|
|
|
$now = time();
|
|
foreach ($rows as $row) {
|
|
$operation = (new edge_gateway_operations_o())->select((int)$row['id']);
|
|
$status = (string)$operation->status->value();
|
|
$startedAt = $operation->started_at->value() === null ? null : strtotime((string)$operation->started_at->value());
|
|
$leaseExpiresAt = $operation->lease_expires_at->value() === null ? null : strtotime((string)$operation->lease_expires_at->value());
|
|
$timedOut = $startedAt !== false
|
|
&& $startedAt !== null
|
|
&& ($now - $startedAt) >= self::OPERATION_TIMEOUT_SECONDS;
|
|
$leaseExpired = $leaseExpiresAt !== false
|
|
&& $leaseExpiresAt !== null
|
|
&& $leaseExpiresAt <= $now;
|
|
|
|
if (!$timedOut && !$leaseExpired) {
|
|
continue;
|
|
}
|
|
|
|
if ($status === self::STATUS_CANCEL_REQUESTED) {
|
|
$this->markOperationCancelled(
|
|
$gatewayId,
|
|
$operation,
|
|
'Gateway did not acknowledge cancellation before the operation lease expired',
|
|
'OPERATION_CANCELLED',
|
|
[
|
|
'agent_instance_id' => $operation->agent_instance_id->value(),
|
|
'last_progress_at' => $operation->last_progress_at->value(),
|
|
]
|
|
);
|
|
$this->refreshGatewayViewCache($gatewayId);
|
|
continue;
|
|
}
|
|
|
|
$errorMessage = $leaseExpired
|
|
? 'Gateway stopped reporting operation progress before the lease expired'
|
|
: 'Gateway operation timed out';
|
|
$summaryLabel = $leaseExpired ? 'Lease expired' : 'Timed out';
|
|
$operation->status->set(self::STATUS_FAILED);
|
|
$operation->completed_at->set($this->now());
|
|
$operation->error_code->set(self::ERROR_OPERATION_TIMEOUT);
|
|
$operation->error_message->set($errorMessage);
|
|
$operation->lease_expires_at->set(null);
|
|
$summary = (array)($operation->summary_json->value() ?? []);
|
|
$summary['label'] = $summaryLabel;
|
|
$summary['retryable'] = true;
|
|
$operation->summary_json->set($summary);
|
|
|
|
$this->appendEventRecord(
|
|
$gatewayId,
|
|
(int)$operation->id,
|
|
self::LEVEL_ERROR,
|
|
self::ERROR_OPERATION_TIMEOUT,
|
|
$errorMessage,
|
|
[
|
|
'agent_instance_id' => $operation->agent_instance_id->value(),
|
|
'last_progress_at' => $operation->last_progress_at->value(),
|
|
]
|
|
);
|
|
|
|
$this->refreshGatewayViewCache($gatewayId);
|
|
}
|
|
}
|
|
|
|
/**
|
|
* @throws Exception
|
|
*/
|
|
private function applyCompletionSideEffects(
|
|
int $gatewayId,
|
|
edge_gateway_operations_o $operation,
|
|
bool $ok,
|
|
array $result,
|
|
string $errorCode,
|
|
string $errorMessage
|
|
): void {
|
|
$gateway = $this->requireGateway($gatewayId);
|
|
$metadata = (array)($gateway->metadata_json->value() ?? []);
|
|
$type = (string)$operation->type->value();
|
|
$now = $this->now();
|
|
|
|
$metadata['last_operation'] = [
|
|
'id' => (int)$operation->id,
|
|
'type' => $type,
|
|
'status' => $ok ? self::STATUS_COMPLETED : self::STATUS_FAILED,
|
|
'completed_at' => $now,
|
|
'error_code' => $ok ? null : $errorCode,
|
|
];
|
|
|
|
if ($type === self::TYPE_DISCOVERY) {
|
|
if ($ok) {
|
|
$inventory = isset($result['inventory']) && is_array($result['inventory']) ? (array)$result['inventory'] : [];
|
|
$this->manager()->syncGatewayInventory($gatewayId, $inventory);
|
|
$gateway->discovery_status->set('READY');
|
|
$metadata['last_discovery_completed_at'] = $now;
|
|
$metadata['last_discovery_error'] = null;
|
|
} else {
|
|
$gateway->discovery_status->set('FAILED');
|
|
$metadata['last_discovery_error'] = [
|
|
'code' => $errorCode,
|
|
'message' => $errorMessage,
|
|
'at' => $now,
|
|
];
|
|
}
|
|
} elseif ($type === self::TYPE_UPDATE) {
|
|
$requestedUpdate = is_array($operation->request_json->value()) ? (array)$operation->request_json->value() : [];
|
|
$stagedVersion = trim((string)($result['staged_version'] ?? $result['target_version'] ?? $requestedUpdate['target_version'] ?? ''));
|
|
$applied = array_key_exists('applied', $result)
|
|
? (bool)$result['applied']
|
|
: trim((string)($result['installed_version'] ?? '')) !== '';
|
|
if ($ok) {
|
|
$installedVersion = trim((string)($result['installed_version'] ?? ($applied ? $stagedVersion : '')));
|
|
if ($applied && $installedVersion !== '') {
|
|
$gateway->installed_version->set($installedVersion);
|
|
$gateway->target_version->set($installedVersion);
|
|
}
|
|
$metadata['last_update_completed_at'] = $now;
|
|
$metadata['last_update_error'] = null;
|
|
if ($stagedVersion !== '') {
|
|
$metadata['staged_version'] = [
|
|
'target_version' => $stagedVersion,
|
|
'staged_at' => $result['staged_at'] ?? $now,
|
|
'apply_after' => $result['apply_after'] ?? null,
|
|
'status' => $applied ? 'APPLIED' : 'STAGED',
|
|
];
|
|
}
|
|
if (isset($result['update_window'])) {
|
|
$metadata['update_window'] = (string)$result['update_window'];
|
|
}
|
|
if (isset($result['rollback_status']) && is_array($result['rollback_status'])) {
|
|
$metadata['rollback_status'] = (array)$result['rollback_status'];
|
|
}
|
|
} else {
|
|
$metadata['last_update_error'] = [
|
|
'code' => $errorCode,
|
|
'message' => $errorMessage,
|
|
'at' => $now,
|
|
];
|
|
}
|
|
} elseif ($type === self::TYPE_UNINSTALL) {
|
|
if ($ok) {
|
|
$gateway->status->set(edge_gateway_manager::STATUS_OFFLINE);
|
|
$metadata['uninstalled_at'] = $now;
|
|
$metadata['uninstall_error'] = null;
|
|
} else {
|
|
$metadata['uninstall_error'] = [
|
|
'code' => $errorCode,
|
|
'message' => $errorMessage,
|
|
'at' => $now,
|
|
];
|
|
}
|
|
}
|
|
|
|
$gateway->metadata_json->set($metadata);
|
|
|
|
$this->manager()->logGatewayAudit(
|
|
$gatewayId,
|
|
(int)$gateway->department_id->value(),
|
|
$ok ? 'GATEWAY_OPERATION_COMPLETED' : 'GATEWAY_OPERATION_FAILED',
|
|
null,
|
|
[
|
|
'operation_id' => (int)$operation->id,
|
|
'type' => $type,
|
|
'status' => $ok ? self::STATUS_COMPLETED : self::STATUS_FAILED,
|
|
'error_code' => $ok ? null : $errorCode,
|
|
]
|
|
);
|
|
}
|
|
|
|
/**
|
|
* @throws Exception
|
|
*/
|
|
private function serializeOperation(edge_gateway_operations_o $operation, bool $includeEvents = true): array
|
|
{
|
|
$operationArray = $operation->asArray();
|
|
if ($includeEvents) {
|
|
$operationArray['events'] = $this->listOperationEvents(
|
|
(int)$operation->gateway_id->value(),
|
|
(int)$operation->id,
|
|
20
|
|
);
|
|
}
|
|
|
|
return $operationArray;
|
|
}
|
|
|
|
/**
|
|
* @throws Exception
|
|
*/
|
|
private function requireGateway(int $gatewayId): edge_gateways_o
|
|
{
|
|
$gateway = (new edge_gateways_o())->select($gatewayId);
|
|
if (!$gateway->exists() || $gateway->deleted_at->value() !== null) {
|
|
throw new Exception('Edge gateway not found');
|
|
}
|
|
|
|
return $gateway;
|
|
}
|
|
|
|
/**
|
|
* @throws Exception
|
|
*/
|
|
private function requireOperation(int $gatewayId, int $operationId): edge_gateway_operations_o
|
|
{
|
|
$operation = (new edge_gateway_operations_o())->select($operationId);
|
|
if (!$operation->exists() || $operation->deleted_at->value() !== null) {
|
|
throw new Exception('Edge gateway operation not found');
|
|
}
|
|
if ((int)$operation->gateway_id->value() !== $gatewayId) {
|
|
throw new Exception('Edge gateway operation does not belong to this gateway');
|
|
}
|
|
|
|
return $operation;
|
|
}
|
|
|
|
private function appendEventRecord(
|
|
int $gatewayId,
|
|
int $operationId,
|
|
string $level,
|
|
?string $code,
|
|
string $message,
|
|
array $context
|
|
): array {
|
|
$eventId = (new edge_gateway_operation_events_o())->add_object([
|
|
'operation_id' => $operationId,
|
|
'gateway_id' => $gatewayId,
|
|
'level' => $level,
|
|
'code' => $code,
|
|
'message' => $message,
|
|
'context_json' => $context,
|
|
]);
|
|
|
|
return (new edge_gateway_operation_events_o())->select($eventId)->asArray();
|
|
}
|
|
|
|
private function refreshOperationLease(edge_gateway_operations_o $operation): void
|
|
{
|
|
$operation->last_progress_at->set($this->now());
|
|
$operation->lease_expires_at->set($this->leaseExpiry());
|
|
}
|
|
|
|
private function refreshGatewayViewCache(int $gatewayId): void
|
|
{
|
|
edge_gateway_view_cache::syncGateway($this->manager()->getGateway($gatewayId));
|
|
}
|
|
|
|
/**
|
|
* @throws Exception
|
|
*/
|
|
private function classifyCompletionError(int $gatewayId, string $errorMessage): string
|
|
{
|
|
$gateway = $this->requireGateway($gatewayId);
|
|
$gatewayStatus = edge_gateway_manager::resolveGatewayStatus(
|
|
(string)$gateway->status->value(),
|
|
$gateway->last_heartbeat_at->value() === null ? null : (string)$gateway->last_heartbeat_at->value()
|
|
);
|
|
|
|
if ($gatewayStatus === edge_gateway_manager::STATUS_OFFLINE) {
|
|
return self::ERROR_OFFLINE;
|
|
}
|
|
if ($gatewayStatus === edge_gateway_manager::STATUS_DEGRADED) {
|
|
return self::ERROR_STALE_HEARTBEAT;
|
|
}
|
|
|
|
$normalizedMessage = strtolower(trim($errorMessage));
|
|
if (str_contains($normalizedMessage, 'cancel')) {
|
|
return self::ERROR_CANCELLED;
|
|
}
|
|
if (str_contains($normalizedMessage, 'version')) {
|
|
return self::ERROR_UNSUPPORTED_VERSION;
|
|
}
|
|
if (str_contains($normalizedMessage, 'validation')) {
|
|
return self::ERROR_VALIDATION;
|
|
}
|
|
|
|
return self::ERROR_VALIDATION;
|
|
}
|
|
|
|
private function manager(): edge_gateway_manager
|
|
{
|
|
return $this->manager ?? new edge_gateway_manager();
|
|
}
|
|
|
|
private function now(): string
|
|
{
|
|
return date('Y-m-d H:i:s');
|
|
}
|
|
|
|
private function leaseExpiry(): string
|
|
{
|
|
return date('Y-m-d H:i:s', time() + self::OPERATION_LEASE_SECONDS);
|
|
}
|
|
|
|
private function normalizeAgentInstanceId(?string $agentInstanceId, int $gatewayId): string
|
|
{
|
|
$candidate = trim((string)$agentInstanceId);
|
|
if ($candidate === '') {
|
|
return 'gateway-' . $gatewayId;
|
|
}
|
|
|
|
return substr($candidate, 0, 128);
|
|
}
|
|
|
|
private function markOperationFailedFromEvent(
|
|
int $gatewayId,
|
|
edge_gateway_operations_o $operation,
|
|
?string $code,
|
|
string $message,
|
|
array $context
|
|
): void {
|
|
$operation->status->set(self::STATUS_FAILED);
|
|
$operation->error_code->set($code !== null && trim($code) !== '' ? trim($code) : self::ERROR_VALIDATION);
|
|
$operation->error_message->set($message);
|
|
$operation->completed_at->set($this->now());
|
|
$operation->lease_expires_at->set(null);
|
|
$summary = (array)($operation->summary_json->value() ?? []);
|
|
$summary['label'] = 'Failed';
|
|
$summary['retryable'] = ((string)$operation->error_code->value()) !== self::ERROR_UNSUPPORTED_VERSION;
|
|
if (isset($context['progress'])) {
|
|
$summary['progress'] = max(0, min(100, (int)$context['progress']));
|
|
}
|
|
$operation->summary_json->set($summary);
|
|
}
|
|
|
|
private function markOperationCancelled(
|
|
int $gatewayId,
|
|
edge_gateway_operations_o $operation,
|
|
string $message,
|
|
string $eventCode,
|
|
array $context = []
|
|
): void {
|
|
$operation->status->set(self::STATUS_CANCELLED);
|
|
$operation->error_code->set(self::ERROR_CANCELLED);
|
|
$operation->error_message->set($message);
|
|
$operation->completed_at->set($this->now());
|
|
$operation->lease_expires_at->set(null);
|
|
$operation->last_progress_at->set($this->now());
|
|
$summary = (array)($operation->summary_json->value() ?? []);
|
|
$summary['label'] = 'Cancelled';
|
|
$summary['retryable'] = true;
|
|
$summary['progress'] = max(0, min(100, (int)($summary['progress'] ?? 0)));
|
|
$operation->summary_json->set($summary);
|
|
$this->appendEventRecord(
|
|
$gatewayId,
|
|
(int)$operation->id,
|
|
self::LEVEL_WARNING,
|
|
$eventCode,
|
|
$message,
|
|
$context
|
|
);
|
|
}
|
|
}
|