Jeppe B
2a6a86c9c3
Resolve recommended-profile Critical and High findings, retain narrow analyzer exceptions, and update the edge-broker WebSocket dependency to a non-vulnerable release.
324 lines
12 KiB
PHP
324 lines
12 KiB
PHP
<?php
|
|
|
|
namespace classes;
|
|
|
|
use Throwable;
|
|
|
|
class cron_worker
|
|
{
|
|
private cron_scheduler $scheduler;
|
|
private string $worker_id;
|
|
private string $name;
|
|
private string $source;
|
|
private int $poll_seconds;
|
|
private int $heartbeat_seconds;
|
|
private int $max_runtime_seconds;
|
|
private bool $should_stop = false;
|
|
private int $last_heartbeat = 0;
|
|
|
|
public function __construct(?cron_scheduler $scheduler = null, array $options = [])
|
|
{
|
|
$this->scheduler = $scheduler ?? new cron_scheduler();
|
|
$this->name = $this->stringOption($options, 'name', 'CRON_WORKER_NAME', 'cron-worker');
|
|
$this->worker_id = $this->stringOption($options, 'worker_id', 'CRON_WORKER_ID', $this->name);
|
|
$this->source = $this->stringOption($options, 'source', 'CRON_WORKER_SOURCE', 'coolify_worker');
|
|
$this->poll_seconds = $this->intOption($options, 'poll_seconds', 'CRON_WORKER_POLL_SECONDS', 15, 1, 300);
|
|
$this->heartbeat_seconds = $this->intOption($options, 'heartbeat_seconds', 'CRON_WORKER_HEARTBEAT_SECONDS', 30, 5, 300);
|
|
$this->max_runtime_seconds = $this->intOption($options, 'max_runtime_seconds', 'CRON_WORKER_MAX_RUNTIME_SECONDS', 0, 0, 86400);
|
|
}
|
|
|
|
public function run(): int
|
|
{
|
|
if (!$this->boolOption('CRON_WORKER_ENABLED', true)) {
|
|
$this->heartbeat('disabled', 0, 0, null, true);
|
|
return 0;
|
|
}
|
|
|
|
$this->registerSignalHandlers();
|
|
$started = time();
|
|
$this->heartbeat('starting', 0, 0, null, true);
|
|
|
|
while (!$this->should_stop) {
|
|
$result = $this->tick();
|
|
$this->writeStatusLine($result);
|
|
|
|
if ($this->max_runtime_seconds > 0 && time() - $started >= $this->max_runtime_seconds) {
|
|
$this->should_stop = true;
|
|
break;
|
|
}
|
|
|
|
$this->sleepUntilNextPoll();
|
|
}
|
|
|
|
$this->heartbeat('stopped', 0, 0, null, true, true);
|
|
return 0;
|
|
}
|
|
|
|
public function tick(): array
|
|
{
|
|
$this->heartbeat('running');
|
|
$loopStartedAt = date('Y-m-d H:i:s');
|
|
$staleRuns = 0;
|
|
$ran = ['count' => 0, 'ran' => []];
|
|
$error = null;
|
|
$status = 'running';
|
|
|
|
try {
|
|
$staleRuns = $this->scheduler->markExpiredRunningRuns();
|
|
$ran = $this->scheduler->runDue($this->source);
|
|
} catch (Throwable $throwable) {
|
|
$status = 'failed';
|
|
$error = $throwable->getMessage();
|
|
}
|
|
|
|
$this->heartbeat($status, (int)($ran['count'] ?? 0), $staleRuns, $error, true, false, $loopStartedAt);
|
|
|
|
return [
|
|
'worker_id' => $this->worker_id,
|
|
'status' => $status,
|
|
'ran' => (int)($ran['count'] ?? 0),
|
|
'stale_runs' => $staleRuns,
|
|
'error' => $error,
|
|
];
|
|
}
|
|
|
|
public function listWorkers(): array
|
|
{
|
|
cron_schema_bootstrap::ensureTables();
|
|
$rows = $this->fetchAll('SELECT * FROM cron_worker_state ORDER BY last_heartbeat_at DESC, worker_id');
|
|
$workers = [];
|
|
foreach ($rows as $row) {
|
|
$workers[] = $this->publicWorker($row);
|
|
}
|
|
|
|
return [
|
|
'workers' => $workers,
|
|
'summary' => [
|
|
'total' => count($workers),
|
|
'running' => count(array_filter($workers, static fn(array $worker): bool => ($worker['status'] ?? '') === 'running')),
|
|
'stale' => count(array_filter($workers, static fn(array $worker): bool => (bool)($worker['stale'] ?? false))),
|
|
],
|
|
];
|
|
}
|
|
|
|
private function registerSignalHandlers(): void
|
|
{
|
|
if (!function_exists('pcntl_signal')) {
|
|
return;
|
|
}
|
|
|
|
if (function_exists('pcntl_async_signals')) {
|
|
pcntl_async_signals(true);
|
|
}
|
|
|
|
pcntl_signal(SIGTERM, function (): void {
|
|
$this->should_stop = true;
|
|
});
|
|
pcntl_signal(SIGINT, function (): void {
|
|
$this->should_stop = true;
|
|
});
|
|
}
|
|
|
|
private function sleepUntilNextPoll(): void
|
|
{
|
|
$remaining = $this->poll_seconds;
|
|
while ($remaining > 0 && !$this->should_stop) {
|
|
$sleep = min(1, $remaining);
|
|
sleep($sleep);
|
|
$remaining -= $sleep;
|
|
if (time() - $this->last_heartbeat >= $this->heartbeat_seconds) {
|
|
$this->heartbeat('running');
|
|
}
|
|
}
|
|
}
|
|
|
|
private function heartbeat(
|
|
string $status,
|
|
int $runCount = 0,
|
|
int $staleRunCount = 0,
|
|
?string $error = null,
|
|
bool $force = false,
|
|
bool $stopped = false,
|
|
?string $loopStartedAt = null
|
|
): void {
|
|
if (!$force && time() - $this->last_heartbeat < $this->heartbeat_seconds) {
|
|
return;
|
|
}
|
|
|
|
cron_schema_bootstrap::ensureTables();
|
|
$this->last_heartbeat = time();
|
|
$now = date('Y-m-d H:i:s');
|
|
$workerId = $this->sql($this->worker_id);
|
|
$name = $this->sql($this->name);
|
|
$hostname = $this->nullableSql(gethostname() ?: null);
|
|
$pid = getmypid() ?: 0;
|
|
$source = $this->sql($this->source);
|
|
$statusSql = $this->sql($status);
|
|
$releaseChannelId = $this->nullableInt($this->env('CRON_WORKER_RELEASE_CHANNEL_ID'));
|
|
$releaseTargetId = $this->nullableInt($this->env('CRON_WORKER_RELEASE_TARGET_ID'));
|
|
$resourceUuid = $this->nullableSql($this->env('COOLIFY_RESOURCE_UUID') ?: $this->env('CRON_WORKER_COOLIFY_RESOURCE_UUID'));
|
|
$resourceType = $this->nullableSql($this->env('COOLIFY_RESOURCE_TYPE') ?: $this->env('CRON_WORKER_COOLIFY_RESOURCE_TYPE') ?: 'application');
|
|
$commitSha = $this->nullableSql($this->commitSha());
|
|
$errorSql = $this->nullableSql($error);
|
|
$loopStarted = $this->nullableSql($loopStartedAt);
|
|
$stoppedAt = $stopped ? $this->sql($now) : 'NULL';
|
|
$nowSql = $this->sql($now);
|
|
|
|
$this->query(
|
|
"INSERT INTO cron_worker_state (
|
|
worker_id, name, hostname, pid, source, status, release_channel_id, release_target_id,
|
|
coolify_resource_uuid, coolify_resource_type, commit_sha, poll_seconds, last_run_count,
|
|
last_stale_run_count, last_error, started_at, last_heartbeat_at, last_loop_started_at,
|
|
last_loop_finished_at, stopped_at
|
|
) VALUES (
|
|
$workerId, $name, $hostname, $pid, $source, $statusSql, $releaseChannelId, $releaseTargetId,
|
|
$resourceUuid, $resourceType, $commitSha, $this->poll_seconds, $runCount,
|
|
$staleRunCount, $errorSql, $nowSql, $nowSql, $loopStarted,
|
|
$nowSql, $stoppedAt
|
|
)
|
|
ON DUPLICATE KEY UPDATE
|
|
name = VALUES(name),
|
|
hostname = VALUES(hostname),
|
|
pid = VALUES(pid),
|
|
source = VALUES(source),
|
|
status = VALUES(status),
|
|
release_channel_id = VALUES(release_channel_id),
|
|
release_target_id = VALUES(release_target_id),
|
|
coolify_resource_uuid = VALUES(coolify_resource_uuid),
|
|
coolify_resource_type = VALUES(coolify_resource_type),
|
|
commit_sha = VALUES(commit_sha),
|
|
poll_seconds = VALUES(poll_seconds),
|
|
last_run_count = VALUES(last_run_count),
|
|
last_stale_run_count = VALUES(last_stale_run_count),
|
|
last_error = VALUES(last_error),
|
|
last_heartbeat_at = VALUES(last_heartbeat_at),
|
|
last_loop_started_at = COALESCE(VALUES(last_loop_started_at), last_loop_started_at),
|
|
last_loop_finished_at = VALUES(last_loop_finished_at),
|
|
stopped_at = VALUES(stopped_at)"
|
|
);
|
|
}
|
|
|
|
private function publicWorker(array $row): array
|
|
{
|
|
$heartbeatAt = (string)($row['last_heartbeat_at'] ?? '');
|
|
$heartbeatTs = strtotime($heartbeatAt);
|
|
$threshold = max(60, ((int)($row['poll_seconds'] ?? 15) * 4) + 30);
|
|
$age = $heartbeatTs !== false ? max(0, time() - $heartbeatTs) : null;
|
|
|
|
return [
|
|
'worker_id' => (string)($row['worker_id'] ?? ''),
|
|
'name' => (string)($row['name'] ?? ''),
|
|
'hostname' => $row['hostname'] ?? null,
|
|
'pid' => isset($row['pid']) ? (int)$row['pid'] : null,
|
|
'source' => (string)($row['source'] ?? ''),
|
|
'status' => (string)($row['status'] ?? 'unknown'),
|
|
'release_channel_id' => isset($row['release_channel_id']) ? (int)$row['release_channel_id'] : null,
|
|
'release_target_id' => isset($row['release_target_id']) ? (int)$row['release_target_id'] : null,
|
|
'coolify_resource_uuid' => $row['coolify_resource_uuid'] ?? null,
|
|
'coolify_resource_type' => $row['coolify_resource_type'] ?? null,
|
|
'commit_sha' => $row['commit_sha'] ?? null,
|
|
'poll_seconds' => (int)($row['poll_seconds'] ?? 0),
|
|
'last_run_count' => (int)($row['last_run_count'] ?? 0),
|
|
'last_stale_run_count' => (int)($row['last_stale_run_count'] ?? 0),
|
|
'last_error' => $row['last_error'] ?? null,
|
|
'started_at' => $row['started_at'] ?? null,
|
|
'last_heartbeat_at' => $heartbeatAt !== '' ? $heartbeatAt : null,
|
|
'last_heartbeat_age_seconds' => $age,
|
|
'last_loop_started_at' => $row['last_loop_started_at'] ?? null,
|
|
'last_loop_finished_at' => $row['last_loop_finished_at'] ?? null,
|
|
'stopped_at' => $row['stopped_at'] ?? null,
|
|
'stale' => $age === null || $age > $threshold,
|
|
'stale_after_seconds' => $threshold,
|
|
];
|
|
}
|
|
|
|
private function writeStatusLine(array $result): void
|
|
{
|
|
echo '[' . date('Y-m-d H:i:s') . '][CRON_WORKER] '
|
|
. json_encode($result, JSON_UNESCAPED_UNICODE | JSON_UNESCAPED_SLASHES)
|
|
. PHP_EOL;
|
|
}
|
|
|
|
private function stringOption(array $options, string $key, string $env, string $default): string
|
|
{
|
|
$value = trim((string)($options[$key] ?? $this->env($env) ?? ''));
|
|
return $value !== '' ? $value : $default;
|
|
}
|
|
|
|
private function intOption(array $options, string $key, string $env, int $default, int $min, int $max): int
|
|
{
|
|
$value = (int)($options[$key] ?? $this->env($env) ?? $default);
|
|
return max($min, min($max, $value));
|
|
}
|
|
|
|
private function boolOption(string $env, bool $default): bool
|
|
{
|
|
$value = $this->env($env);
|
|
if ($value === null || trim($value) === '') {
|
|
return $default;
|
|
}
|
|
|
|
return in_array(strtolower(trim($value)), ['1', 'true', 'yes', 'on'], true);
|
|
}
|
|
|
|
private function commitSha(): string
|
|
{
|
|
foreach (['CRON_WORKER_COMMIT_SHA', 'API_COMMIT_SHA', 'RELEASE_COMMIT_SHA', 'COMMIT_SHA', 'GITHUB_SHA'] as $key) {
|
|
$value = trim((string)($this->env($key) ?? ''));
|
|
if ($value !== '') {
|
|
return $value;
|
|
}
|
|
}
|
|
|
|
return '';
|
|
}
|
|
|
|
private function env(string $key): ?string
|
|
{
|
|
$value = getenv($key);
|
|
if ($value !== false) {
|
|
return (string)$value;
|
|
}
|
|
|
|
return isset($_SERVER[$key]) ? (string)$_SERVER[$key] : null;
|
|
}
|
|
|
|
private function nullableInt(?string $value): string
|
|
{
|
|
$value = trim((string)$value);
|
|
if ($value === '' || filter_var($value, FILTER_VALIDATE_INT) === false) {
|
|
return 'NULL';
|
|
}
|
|
|
|
return (string)max(0, (int)$value);
|
|
}
|
|
|
|
private function nullableSql(?string $value): string
|
|
{
|
|
$value = $value !== null ? trim($value) : '';
|
|
return $value === '' ? 'NULL' : $this->sql($value);
|
|
}
|
|
|
|
private function fetchAll(string $sql): array
|
|
{
|
|
$result = $this->query($sql);
|
|
if ($result === false || $result === true) {
|
|
return [];
|
|
}
|
|
|
|
return $result->fetch_all(MYSQLI_ASSOC);
|
|
}
|
|
|
|
private function query(string $sql): \mysqli_result|bool
|
|
{
|
|
global $db;
|
|
return $db->query($sql);
|
|
}
|
|
|
|
private function sql(string $value): string
|
|
{
|
|
global $db;
|
|
return "'" . $db->escape_string($value) . "'";
|
|
}
|
|
}
|