mirror of
https://github.com/tiennm99/coolify.git
synced 2026-10-19 08:15:02 +00:00
chore(v5): archive V5 implementation and remove runtime integration
Move V5 source, migrations, UI, scripts, and tests into documentation, then remove V5 routes, models, jobs, configuration, dependencies, and application hooks.
This commit is contained in:
1 parent
1440d35750
commit
76030a30d4
237 files changed
+288
-7079
No files matched your search
@@ -1,252 +0,0 @@
|
||||
<?php
|
||||
|
||||
namespace App\Services\Flux;
|
||||
|
||||
use App\Models\V5\RevokedAgentToken;
|
||||
use App\Models\V5\Server as V5Server;
|
||||
use Firebase\JWT\JWT;
|
||||
use Illuminate\Support\Facades\File;
|
||||
use Illuminate\Support\Facades\Log;
|
||||
use Illuminate\Support\Str;
|
||||
use RuntimeException;
|
||||
|
||||
/**
|
||||
* Mints the per-host ES256 JWT that authorizes a coold host agent against flux.
|
||||
*
|
||||
* Capability scoping: by default the token carries the EXPLICIT list of
|
||||
* primitive capability strings coold advertises (config('flux.host_capabilities'),
|
||||
* mirroring coold/coold/src/grpc/client.rs:204-231) rather than the
|
||||
* `host-agent:default` wildcard profile. flux intersects the jwt `caps` with
|
||||
* coold's advertised set (flux/src/main.rs:128-141), so the effective power is
|
||||
* unchanged, but the token no longer depends on flux's
|
||||
* `capability_profile_authorizes_all` wildcard bypass (main.rs:124-126).
|
||||
*
|
||||
* @see config/flux.php for the capability list, escape hatch, TTL and kid config.
|
||||
*/
|
||||
class AgentTokenIssuer
|
||||
{
|
||||
public const DEFAULT_PROFILE = 'host-agent:default';
|
||||
|
||||
private const TTL_FLOOR_SECONDS = 60;
|
||||
|
||||
/**
|
||||
* Mint a host JWT.
|
||||
*
|
||||
* @param array<int, string>|null $capabilities Explicit caps; null resolves the configured default set (or escape-hatch profile).
|
||||
* @param int|null $ttl Lifetime in seconds; null resolves config('flux.host_token_ttl'). Clamped to a 60s floor.
|
||||
* @param array<string, mixed> $extraClaims Extra claims merged in (a `jti` here is honored, otherwise one is generated).
|
||||
*/
|
||||
public function issue(string $hostId, ?array $capabilities = null, ?int $ttl = null, array $extraClaims = []): string
|
||||
{
|
||||
if ($hostId === '') {
|
||||
throw new RuntimeException('Flux host id is required.');
|
||||
}
|
||||
|
||||
$privateKeyPath = config('flux.jwt_private_key_path');
|
||||
|
||||
if (! is_string($privateKeyPath) || $privateKeyPath === '' || ! File::isReadable($privateKeyPath)) {
|
||||
throw new RuntimeException("Flux JWT private key not found at {$privateKeyPath}.");
|
||||
}
|
||||
|
||||
$this->assertPrivateKeyPermissions($privateKeyPath);
|
||||
|
||||
$capabilities ??= $this->defaultCapabilities();
|
||||
$ttl ??= (int) config('flux.host_token_ttl', 3600);
|
||||
|
||||
$jti = $extraClaims['jti'] ?? (string) Str::uuid();
|
||||
unset($extraClaims['jti']);
|
||||
|
||||
$now = time();
|
||||
$keyId = (string) config('flux.jwt_kid', 'flux-default');
|
||||
|
||||
return JWT::encode(array_merge($extraClaims, [
|
||||
'sub' => $hostId,
|
||||
'aud' => 'coold',
|
||||
'caps' => $this->normalizeCapabilities($capabilities),
|
||||
'jti' => $jti,
|
||||
'iat' => $now,
|
||||
'exp' => $now + max(self::TTL_FLOOR_SECONDS, $ttl),
|
||||
]), File::get($privateKeyPath), 'ES256', $keyId !== '' ? $keyId : null);
|
||||
}
|
||||
|
||||
public function issueForServer(V5Server $server, ?int $ttl = null): string
|
||||
{
|
||||
$hostId = $server->fluxHostId();
|
||||
|
||||
if ($hostId === '') {
|
||||
throw new RuntimeException('Server is missing a valid Flux host id.');
|
||||
}
|
||||
|
||||
$jti = (string) Str::uuid();
|
||||
$ttl ??= (int) config('flux.host_token_ttl', 3600);
|
||||
|
||||
// team_id/cluster_id/server_id are minted as STRINGS: flux deserializes
|
||||
// the `team_id` claim as a string (coold/flux/src/auth.rs Claims), and
|
||||
// rejects the whole token with a JSON type error if it arrives as a JSON
|
||||
// integer. Keep the sibling ids string-typed for consistency.
|
||||
$token = $this->issue($hostId, $this->defaultCapabilities(), $ttl, [
|
||||
'jti' => $jti,
|
||||
'team_id' => (string) $server->team_id,
|
||||
'cluster_id' => (string) $server->cluster_id,
|
||||
'server_id' => $hostId,
|
||||
'wireguard_management_ip' => (string) $server->wireguard_management_ip,
|
||||
]);
|
||||
|
||||
// Persist the freshly issued jti (so a later destroy/re-home knows which
|
||||
// token to revoke) and its expiry (so the scheduled rotation loop knows
|
||||
// when to re-mint). Use a targeted update keyed by id so this neither
|
||||
// inserts an unsaved model nor flushes unrelated dirty attributes, and
|
||||
// does not depend on the Server model's $fillable.
|
||||
if ($server->exists) {
|
||||
$expiresAt = now()->addSeconds(max(self::TTL_FLOOR_SECONDS, $ttl));
|
||||
|
||||
V5Server::query()->whereKey($server->getKey())->update([
|
||||
'agent_token_jti' => $jti,
|
||||
'agent_token_expires_at' => $expiresAt,
|
||||
]);
|
||||
$server->setAttribute('agent_token_jti', $jti);
|
||||
$server->setAttribute('agent_token_expires_at', $expiresAt);
|
||||
$server->syncOriginalAttribute('agent_token_jti');
|
||||
$server->syncOriginalAttribute('agent_token_expires_at');
|
||||
}
|
||||
|
||||
return $token;
|
||||
}
|
||||
|
||||
/**
|
||||
* Record the server's currently-issued host token jti as revoked AND push
|
||||
* the revocation to flux so it rejects the jti at verify immediately.
|
||||
*
|
||||
* flux now consults a revocation denylist (flux/src/auth.rs `is_revoked`,
|
||||
* fed by `POST /v1/tokens/revoke` on the flux UDS), and Laravel pushes to it
|
||||
* here. The local `RevokedAgentToken` record remains the source of truth
|
||||
* Laravel owns; the flux push is best-effort — if flux is unreachable the
|
||||
* revocation is logged and the local record still stands, with the short TTL
|
||||
* and hourly rotation bounding the exposure until flux is reachable again.
|
||||
*/
|
||||
public function revoke(V5Server $server): void
|
||||
{
|
||||
$jti = $server->agent_token_jti;
|
||||
|
||||
if (! is_string($jti) || $jti === '') {
|
||||
return;
|
||||
}
|
||||
|
||||
$expiresAt = $server->agent_token_expires_at;
|
||||
$expiresAtUnix = $expiresAt instanceof \DateTimeInterface ? $expiresAt->getTimestamp() : null;
|
||||
|
||||
RevokedAgentToken::query()->updateOrCreate(
|
||||
['jti' => $jti],
|
||||
[
|
||||
'server_id' => $server->id,
|
||||
'revoked_at' => now(),
|
||||
'expires_at' => $expiresAt,
|
||||
]
|
||||
);
|
||||
|
||||
// Best-effort: a destroy/teardown must never fail because flux is down.
|
||||
try {
|
||||
app(FluxClient::class)->revokeToken($jti, $expiresAtUnix);
|
||||
} catch (\Throwable $exception) {
|
||||
Log::warning('Failed to push agent token revocation to Flux.', [
|
||||
'server_id' => $server->id,
|
||||
'jti' => $jti,
|
||||
'error' => $exception->getMessage(),
|
||||
]);
|
||||
}
|
||||
|
||||
if ($server->exists) {
|
||||
V5Server::query()->whereKey($server->getKey())->update(['agent_token_jti' => null]);
|
||||
$server->setAttribute('agent_token_jti', null);
|
||||
$server->syncOriginalAttribute('agent_token_jti');
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Revoke the server's currently-issued host token. Alias of {@see revoke()}
|
||||
* kept as the name the team-teardown job resolves via `method_exists`.
|
||||
*/
|
||||
public function revokeForServer(V5Server $server): void
|
||||
{
|
||||
$this->revoke($server);
|
||||
}
|
||||
|
||||
public function isRevoked(string $jti): bool
|
||||
{
|
||||
if ($jti === '') {
|
||||
return false;
|
||||
}
|
||||
|
||||
return RevokedAgentToken::query()->where('jti', $jti)->exists();
|
||||
}
|
||||
|
||||
/**
|
||||
* The default capability set for production host tokens: the explicit
|
||||
* advertised primitive list, unless the emergency escape hatch profile is
|
||||
* configured (then that single profile is minted instead).
|
||||
*
|
||||
* @return array<int, string>
|
||||
*/
|
||||
private function defaultCapabilities(): array
|
||||
{
|
||||
$profile = config('flux.host_capability_profile');
|
||||
|
||||
if (is_string($profile) && trim($profile) !== '') {
|
||||
return [trim($profile)];
|
||||
}
|
||||
|
||||
$configured = config('flux.host_capabilities');
|
||||
|
||||
if (is_array($configured) && $configured !== []) {
|
||||
return array_values($configured);
|
||||
}
|
||||
|
||||
return [self::DEFAULT_PROFILE];
|
||||
}
|
||||
|
||||
/**
|
||||
* Warn (but do not hard-fail — that could break existing installs) when the
|
||||
* private key file is readable by group/other or is not owner-readable. The
|
||||
* key should be generated 0600, e.g.:
|
||||
* openssl genpkey -algorithm EC -pkeyopt ec_paramgen_curve:P-256 \
|
||||
* -out storage/app/flux/jwt.priv && chmod 600 storage/app/flux/jwt.priv
|
||||
*/
|
||||
private function assertPrivateKeyPermissions(string $path): void
|
||||
{
|
||||
$perms = @fileperms($path);
|
||||
|
||||
if ($perms === false) {
|
||||
return;
|
||||
}
|
||||
|
||||
$mode = $perms & 0777;
|
||||
|
||||
if (($mode & 0077) !== 0 || ($mode & 0400) === 0) {
|
||||
Log::warning('Flux JWT private key has insecure permissions.', [
|
||||
'path' => $path,
|
||||
'mode' => sprintf('%04o', $mode),
|
||||
'expected' => '0600',
|
||||
]);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* @param array<int, string> $capabilities
|
||||
* @return array<int, string>
|
||||
*/
|
||||
private function normalizeCapabilities(array $capabilities): array
|
||||
{
|
||||
$normalized = collect($capabilities)
|
||||
->map(fn (string $capability) => trim($capability))
|
||||
->filter()
|
||||
->unique()
|
||||
->values()
|
||||
->all();
|
||||
|
||||
if ($normalized === []) {
|
||||
return [self::DEFAULT_PROFILE];
|
||||
}
|
||||
|
||||
return $normalized;
|
||||
}
|
||||
}
|
||||
@@ -1,400 +0,0 @@
|
||||
<?php
|
||||
|
||||
namespace App\Services\Flux;
|
||||
|
||||
use App\Exceptions\V5\UnsupportedCooldVerb;
|
||||
use Illuminate\Support\Str;
|
||||
use RuntimeException;
|
||||
|
||||
class FluxClient
|
||||
{
|
||||
/**
|
||||
* @return array<int, array{id?: string, name?: string, image?: string, state?: string, networks?: array<int, string>}>
|
||||
*/
|
||||
public function listContainers(string $hostId): array
|
||||
{
|
||||
$payload = $this->dispatch($hostId, [
|
||||
'type' => 'containers.list',
|
||||
]);
|
||||
|
||||
$data = $payload['data'] ?? [];
|
||||
|
||||
return is_array($data) ? $data : [];
|
||||
}
|
||||
|
||||
public function pullImage(string $hostId, string $image): string
|
||||
{
|
||||
$payload = $this->dispatch($hostId, [
|
||||
'type' => 'images.pull',
|
||||
'reference' => $image,
|
||||
]);
|
||||
|
||||
return $this->output($payload, 'Image pulled.');
|
||||
}
|
||||
|
||||
/**
|
||||
* @param array<string, mixed> $spec
|
||||
*/
|
||||
public function createContainer(string $hostId, array $spec): string
|
||||
{
|
||||
$payload = $this->dispatch($hostId, [
|
||||
'type' => 'containers.create',
|
||||
...$spec,
|
||||
]);
|
||||
$data = $payload['data'] ?? [];
|
||||
$id = is_array($data) && is_string($data['id'] ?? null) ? $data['id'] : '';
|
||||
|
||||
if ($id === '') {
|
||||
throw new RuntimeException('Flux did not return a container id.');
|
||||
}
|
||||
|
||||
return $id;
|
||||
}
|
||||
|
||||
public function startContainer(string $hostId, string $id): string
|
||||
{
|
||||
$payload = $this->dispatch($hostId, [
|
||||
'type' => 'containers.start',
|
||||
'id' => $id,
|
||||
]);
|
||||
|
||||
return $this->output($payload, 'Container started.');
|
||||
}
|
||||
|
||||
public function stopContainer(string $hostId, string $id, int $timeoutSeconds = 10): string
|
||||
{
|
||||
$payload = $this->dispatch($hostId, [
|
||||
'type' => 'containers.stop',
|
||||
'id' => $id,
|
||||
'timeout_seconds' => max(0, $timeoutSeconds),
|
||||
]);
|
||||
|
||||
return $this->output($payload, 'Container stopped.');
|
||||
}
|
||||
|
||||
public function removeContainer(string $hostId, string $id, bool $force = false): string
|
||||
{
|
||||
$payload = $this->dispatch($hostId, [
|
||||
'type' => 'containers.delete',
|
||||
'id' => $id,
|
||||
'force' => $force,
|
||||
]);
|
||||
|
||||
return $this->output($payload, 'Container removed.');
|
||||
}
|
||||
|
||||
/**
|
||||
* @return array<string, mixed>
|
||||
*/
|
||||
public function inspectContainer(string $hostId, string $id): array
|
||||
{
|
||||
$payload = $this->dispatch($hostId, [
|
||||
'type' => 'containers.inspect',
|
||||
'id' => $id,
|
||||
]);
|
||||
$data = $payload['data'] ?? [];
|
||||
|
||||
return is_array($data) ? $data : [];
|
||||
}
|
||||
|
||||
/**
|
||||
* @param array<int, array{name: string, config: string}> $apps
|
||||
*/
|
||||
public function applyIngress(string $hostId, string $kind, string $config, array $apps = [], string $meshNetwork = 'coolify-default-mesh'): string
|
||||
{
|
||||
$payload = $this->dispatch($hostId, [
|
||||
'type' => 'ingress.apply',
|
||||
'kind' => $kind,
|
||||
'config' => $config,
|
||||
'apps' => $apps,
|
||||
'mesh_network' => $meshNetwork,
|
||||
]);
|
||||
|
||||
return $this->output($payload, 'Ingress applied.');
|
||||
}
|
||||
|
||||
public function stopIngress(string $hostId, string $kind): string
|
||||
{
|
||||
$payload = $this->dispatch($hostId, [
|
||||
'type' => 'ingress.stop',
|
||||
'kind' => $kind,
|
||||
]);
|
||||
|
||||
return $this->output($payload, 'Ingress stopped.');
|
||||
}
|
||||
|
||||
/**
|
||||
* @param array{id: string, namespace: string, src: string, dst: string, proto: string, port: int} $rule
|
||||
*/
|
||||
public function applyFirewallRule(string $hostId, array $rule): string
|
||||
{
|
||||
$payload = $this->dispatch($hostId, [
|
||||
'type' => 'firewall.allow',
|
||||
...$rule,
|
||||
]);
|
||||
|
||||
$data = $payload['data'] ?? [];
|
||||
$id = is_array($data) && is_string($data['id'] ?? null) ? $data['id'] : '';
|
||||
|
||||
return $id !== '' ? $id : $this->output($payload, 'Firewall rule applied.');
|
||||
}
|
||||
|
||||
public function revokeFirewallRule(string $hostId, string $id): string
|
||||
{
|
||||
$payload = $this->dispatch($hostId, [
|
||||
'type' => 'firewall.revoke',
|
||||
'id' => $id,
|
||||
]);
|
||||
|
||||
return $this->output($payload, 'Firewall rule removed.');
|
||||
}
|
||||
|
||||
/**
|
||||
* @return array<int, array{id?: string, namespace?: string, src?: string, dst?: string, proto?: string, port?: int}>
|
||||
*/
|
||||
public function listFirewallRules(string $hostId, string $namespace = ''): array
|
||||
{
|
||||
$payload = $this->dispatch($hostId, [
|
||||
'type' => 'firewall.list',
|
||||
'namespace' => $namespace,
|
||||
]);
|
||||
|
||||
$data = $payload['data'] ?? [];
|
||||
|
||||
return is_array($data) ? $data : [];
|
||||
}
|
||||
|
||||
public function cooldLogs(string $hostId, int $tail = 200): string
|
||||
{
|
||||
$payload = $this->dispatch($hostId, [
|
||||
'type' => 'coold.logs',
|
||||
'tail' => max(1, min($tail, 1000)),
|
||||
]);
|
||||
|
||||
return $this->output($payload, 'No coold logs returned.');
|
||||
}
|
||||
|
||||
public function containerLogs(string $hostId, string $containerId, int $tail = 200): string
|
||||
{
|
||||
$payload = $this->dispatch($hostId, [
|
||||
'type' => 'containers.logs',
|
||||
'id' => $containerId,
|
||||
'tail' => max(1, min($tail, 1000)),
|
||||
'stdout' => true,
|
||||
'stderr' => true,
|
||||
]);
|
||||
|
||||
return $this->output($payload, 'No container logs returned.');
|
||||
}
|
||||
|
||||
public function corrosionTables(string $hostId, int $limit = 200): string
|
||||
{
|
||||
$payload = $this->dispatch($hostId, [
|
||||
'type' => 'corrosion.tables',
|
||||
'limit' => max(1, min($limit, 1000)),
|
||||
]);
|
||||
|
||||
return $this->output($payload, '{"limit":200,"tables":[]}');
|
||||
}
|
||||
|
||||
/**
|
||||
* Deliver a freshly minted host JWT to the node over the live coold RPC
|
||||
* stream (flux gates the `host.jwt.set` capability; the token must carry
|
||||
* it). Preferred over the SSH push because it reuses the already
|
||||
* authenticated flux<->coold channel and works while the current token is
|
||||
* still valid — exactly the rotation window. Throws like the sibling
|
||||
* dispatch methods (host not connected / UnsupportedCooldVerb / generic
|
||||
* failure) so the caller can catch and fall back to SSH.
|
||||
*/
|
||||
public function pushHostToken(string $hostId, string $token): void
|
||||
{
|
||||
$this->dispatch($hostId, [
|
||||
'type' => 'host.jwt.set',
|
||||
'jwt' => $token,
|
||||
]);
|
||||
}
|
||||
|
||||
/**
|
||||
* Revoke a host token by its `jti` on the flux revocation store so flux
|
||||
* rejects it at verify immediately, instead of waiting for the token's TTL
|
||||
* to lapse (flux/src/unix_bridge.rs `POST /v1/tokens/revoke`,
|
||||
* flux/src/auth.rs `is_revoked`). The optional `expiresAt` (the token `exp`,
|
||||
* unix seconds) lets flux prune the denylist entry once it can no longer
|
||||
* matter.
|
||||
*
|
||||
* Best-effort like the sibling dispatch methods: throws a RuntimeException on
|
||||
* connection failure / timeout / non-2xx so the caller can catch and treat
|
||||
* an unreachable flux as non-fatal (the local revocation record still
|
||||
* stands and the short TTL + rotation bound the exposure).
|
||||
*/
|
||||
public function revokeToken(string $jti, ?int $expiresAt = null): void
|
||||
{
|
||||
if (trim($jti) === '') {
|
||||
return;
|
||||
}
|
||||
|
||||
$requestBody = ['jti' => $jti];
|
||||
|
||||
if ($expiresAt !== null) {
|
||||
$requestBody['expires_at'] = $expiresAt;
|
||||
}
|
||||
|
||||
$body = json_encode($requestBody, JSON_THROW_ON_ERROR);
|
||||
$response = $this->sendOverSocket('/v1/tokens/revoke', $body);
|
||||
$statusCode = $this->statusCode($response);
|
||||
|
||||
if ($statusCode < 200 || $statusCode >= 300) {
|
||||
$responseBody = $this->responseBody($response);
|
||||
$payload = $responseBody === '' ? null : json_decode($responseBody, true);
|
||||
|
||||
throw new RuntimeException(
|
||||
$this->errorMessage($payload, $responseBody) ?? "Flux token revocation returned HTTP {$statusCode}."
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* @param array<string, mixed> $command
|
||||
* @return array<string, mixed>
|
||||
*/
|
||||
private function dispatch(string $hostId, array $command): array
|
||||
{
|
||||
$body = json_encode([
|
||||
'host_id' => $hostId,
|
||||
'request_id' => (string) Str::uuid(),
|
||||
'command' => $command,
|
||||
], JSON_THROW_ON_ERROR);
|
||||
|
||||
$response = $this->sendOverSocket('/v1/coold/dispatch', $body);
|
||||
|
||||
$statusCode = $this->statusCode($response);
|
||||
$responseBody = $this->responseBody($response);
|
||||
$payload = $responseBody === '' ? null : json_decode($responseBody, true);
|
||||
|
||||
if ($statusCode < 200 || $statusCode >= 300) {
|
||||
throw $this->dispatchException(
|
||||
$command,
|
||||
$statusCode,
|
||||
$this->errorMessage($payload, $responseBody) ?? "Flux dispatch returned HTTP {$statusCode}."
|
||||
);
|
||||
}
|
||||
|
||||
if (! is_array($payload)) {
|
||||
throw new RuntimeException('Flux dispatch returned an invalid response.');
|
||||
}
|
||||
|
||||
if (($payload['status'] ?? null) === 'error') {
|
||||
$message = is_string($payload['message'] ?? null) ? $payload['message'] : 'Flux dispatch failed.';
|
||||
|
||||
throw $this->dispatchException($command, $statusCode, $message);
|
||||
}
|
||||
|
||||
return $payload;
|
||||
}
|
||||
|
||||
/**
|
||||
* Send a single HTTP/1.1 request over the flux Unix-domain socket and return
|
||||
* the raw response. Shared by every flux verb (coold dispatch, host token
|
||||
* rotation, token revocation) — only the request path and JSON body differ.
|
||||
*/
|
||||
private function sendOverSocket(string $path, string $body): string
|
||||
{
|
||||
$socketPath = config('flux.unix_socket_path');
|
||||
|
||||
if (! is_string($socketPath) || $socketPath === '') {
|
||||
throw new RuntimeException('Flux socket is not configured.');
|
||||
}
|
||||
|
||||
if (! file_exists($socketPath)) {
|
||||
throw new RuntimeException('Flux socket was not found.');
|
||||
}
|
||||
|
||||
$connectionTimeout = (float) config('flux.connection_timeout_seconds', 1.0);
|
||||
$dispatchTimeout = (float) config('flux.dispatch_timeout_seconds', 35.0);
|
||||
$stream = @stream_socket_client("unix://{$socketPath}", $errorCode, $errorMessage, $connectionTimeout);
|
||||
|
||||
if ($stream === false) {
|
||||
throw new RuntimeException($errorMessage ?: "Could not connect to Flux socket ({$errorCode}).");
|
||||
}
|
||||
|
||||
stream_set_timeout($stream, (int) ceil($dispatchTimeout));
|
||||
|
||||
fwrite($stream, implode("\r\n", [
|
||||
"POST {$path} HTTP/1.1",
|
||||
'Host: flux',
|
||||
'Accept: application/json',
|
||||
'Content-Type: application/json',
|
||||
'Content-Length: '.strlen($body),
|
||||
'Connection: close',
|
||||
'',
|
||||
$body,
|
||||
]));
|
||||
|
||||
$response = stream_get_contents($stream) ?: '';
|
||||
fclose($stream);
|
||||
|
||||
return $response;
|
||||
}
|
||||
|
||||
/**
|
||||
* Flux answers a verb the node's coold did not advertise with HTTP 501 and
|
||||
* the message "primitive <verb> is not supported by host" (coold repo:
|
||||
* flux/src/routing.rs:50-53, flux/src/unix_bridge.rs:227-245). Anything
|
||||
* else — including coold-side command failures relayed with their own
|
||||
* status code — is a generic dispatch failure.
|
||||
*
|
||||
* @param array<string, mixed> $command
|
||||
*/
|
||||
private function dispatchException(array $command, int $statusCode, string $message): RuntimeException
|
||||
{
|
||||
$verb = is_string($command['type'] ?? null) ? $command['type'] : 'unknown';
|
||||
|
||||
if ($statusCode === 501 || preg_match('/primitive .+ is not supported by host/i', $message) === 1) {
|
||||
return new UnsupportedCooldVerb($verb, $message);
|
||||
}
|
||||
|
||||
return new RuntimeException($message);
|
||||
}
|
||||
|
||||
private function statusCode(string $response): int
|
||||
{
|
||||
if ($response === '') {
|
||||
throw new RuntimeException('Flux did not return a response before the timeout. Check that coold is connected to Flux and try again.');
|
||||
}
|
||||
|
||||
if (preg_match('/^HTTP\/\d(?:\.\d)?\s+(\d{3})/', $response, $matches) !== 1) {
|
||||
throw new RuntimeException('Could not talk to Flux. Check that Flux is running in the Coolify container.');
|
||||
}
|
||||
|
||||
return (int) $matches[1];
|
||||
}
|
||||
|
||||
private function responseBody(string $response): string
|
||||
{
|
||||
$position = strpos($response, "\r\n\r\n");
|
||||
|
||||
return $position === false ? '' : substr($response, $position + 4);
|
||||
}
|
||||
|
||||
private function errorMessage(mixed $payload, string $responseBody): ?string
|
||||
{
|
||||
if (is_array($payload) && is_string($payload['message'] ?? null) && $payload['message'] !== '') {
|
||||
return $payload['message'];
|
||||
}
|
||||
|
||||
$message = trim($responseBody);
|
||||
|
||||
return $message === '' ? null : Str::limit($message, 1000);
|
||||
}
|
||||
|
||||
/**
|
||||
* @param array<string, mixed> $payload
|
||||
*/
|
||||
private function output(array $payload, string $fallback): string
|
||||
{
|
||||
$data = $payload['data'] ?? [];
|
||||
$output = is_array($data) && is_string($data['output'] ?? null) ? $data['output'] : '';
|
||||
|
||||
return $output !== '' ? $output : $fallback;
|
||||
}
|
||||
}
|
||||
@@ -1,66 +0,0 @@
|
||||
<?php
|
||||
|
||||
namespace App\Services\Flux;
|
||||
|
||||
class FluxHealth
|
||||
{
|
||||
/**
|
||||
* @return array{available: bool, label: string, message: string, socket: string|null}
|
||||
*/
|
||||
public function check(): array
|
||||
{
|
||||
$socketPath = config('flux.unix_socket_path');
|
||||
|
||||
if (! is_string($socketPath) || $socketPath === '') {
|
||||
return $this->unavailable(null, 'Flux socket is not configured.');
|
||||
}
|
||||
|
||||
if (! file_exists($socketPath)) {
|
||||
return $this->unavailable($socketPath, 'Flux socket was not found.');
|
||||
}
|
||||
|
||||
$timeout = (float) config('flux.health_timeout_seconds', 1.0);
|
||||
$stream = @stream_socket_client("unix://{$socketPath}", $errorCode, $errorMessage, $timeout);
|
||||
|
||||
if ($stream === false) {
|
||||
return $this->unavailable($socketPath, $errorMessage ?: "Could not connect to Flux socket ({$errorCode}).");
|
||||
}
|
||||
|
||||
stream_set_timeout($stream, (int) ceil($timeout));
|
||||
|
||||
fwrite($stream, "GET /v1/health HTTP/1.1\r\nHost: flux\r\nAccept: application/json\r\nConnection: close\r\n\r\n");
|
||||
$response = stream_get_contents($stream) ?: '';
|
||||
fclose($stream);
|
||||
|
||||
if (! str_starts_with($response, 'HTTP/1.1 200') && ! str_starts_with($response, 'HTTP/1.0 200')) {
|
||||
return $this->unavailable($socketPath, 'Flux health endpoint did not return HTTP 200.');
|
||||
}
|
||||
|
||||
$body = str_contains($response, "\r\n\r\n") ? substr($response, strpos($response, "\r\n\r\n") + 4) : '';
|
||||
$payload = json_decode($body, true);
|
||||
|
||||
if (! is_array($payload) || ($payload['ok'] ?? false) !== true) {
|
||||
return $this->unavailable($socketPath, 'Flux health endpoint returned an invalid response.');
|
||||
}
|
||||
|
||||
return [
|
||||
'available' => true,
|
||||
'label' => 'Running',
|
||||
'message' => 'Flux is running.',
|
||||
'socket' => $socketPath,
|
||||
];
|
||||
}
|
||||
|
||||
/**
|
||||
* @return array{available: false, label: string, message: string, socket: string|null}
|
||||
*/
|
||||
private function unavailable(?string $socketPath, string $message): array
|
||||
{
|
||||
return [
|
||||
'available' => false,
|
||||
'label' => 'Unavailable',
|
||||
'message' => $message,
|
||||
'socket' => $socketPath,
|
||||
];
|
||||
}
|
||||
}
|
||||
Reference in new issue
Block a user