diff --git a/src/Appwrite/Bus/Listeners/Log.php b/src/Appwrite/Bus/Listeners/Log.php index 076ed5c02d..d52629c21c 100644 --- a/src/Appwrite/Bus/Listeners/Log.php +++ b/src/Appwrite/Bus/Listeners/Log.php @@ -5,6 +5,7 @@ namespace Appwrite\Bus\Listeners; use Appwrite\Bus\Events\ExecutionCompleted; use Appwrite\Event\Message\Execution as ExecutionMessage; use Appwrite\Event\Publisher\Execution as ExecutionPublisher; +use Appwrite\Extend\TraceFunctionExecution; use Utopia\Bus\Listener; use Utopia\Database\Document; @@ -30,9 +31,21 @@ class Log extends Listener public function handle(ExecutionCompleted $event, ExecutionPublisher $publisherForExecutions): void { + $project = new Document($event->project); + $execution = new Document($event->execution); + if ($execution->getAttribute('resourceType', '') === 'functions') { + TraceFunctionExecution::log('v1_executions_enqueue', [ + 'projectId' => $project->getId(), + 'functionId' => $execution->getAttribute('resourceId', ''), + 'executionId' => $execution->getId(), + 'deploymentId' => $execution->getAttribute('deploymentId', ''), + 'status' => $execution->getAttribute('status', ''), + ]); + } + $publisherForExecutions->enqueue(new ExecutionMessage( - project: new Document($event->project), - execution: new Document($event->execution), + project: $project, + execution: $execution, )); } } diff --git a/src/Appwrite/Extend/TraceFunctionExecution.php b/src/Appwrite/Extend/TraceFunctionExecution.php new file mode 100644 index 0000000000..a3731c7afd --- /dev/null +++ b/src/Appwrite/Extend/TraceFunctionExecution.php @@ -0,0 +1,74 @@ + $context Must include projectId and functionId for filtering. + */ + public static function log(string $station, array $context = []): void + { + if (!self::isEnabled()) { + return; + } + + $projectId = $context['projectId'] ?? null; + $functionId = $context['functionId'] ?? null; + if (!self::matches($projectId, $functionId)) { + return; + } + + $payload = \array_merge([ + 'station' => $station, + 'time' => DateTime::now(), + ], $context); + + Console::log('[execution-trace] ' . \json_encode($payload, JSON_UNESCAPED_SLASHES)); + } +} diff --git a/src/Appwrite/Platform/Tasks/ScheduleFunctions.php b/src/Appwrite/Platform/Tasks/ScheduleFunctions.php index 88725a190a..ae8be20303 100644 --- a/src/Appwrite/Platform/Tasks/ScheduleFunctions.php +++ b/src/Appwrite/Platform/Tasks/ScheduleFunctions.php @@ -3,6 +3,7 @@ namespace Appwrite\Platform\Tasks; use Appwrite\Event\Func; +use Appwrite\Extend\TraceFunctionExecution; use Cron\CronExpression; use Utopia\Console; use Utopia\Database\Database; @@ -97,6 +98,16 @@ class ScheduleFunctions extends ScheduleBase $queueForFunctions = new Func($this->publisherFunctions); + $projectDoc = $schedule['project']; + $functionDoc = $schedule['resource']; + TraceFunctionExecution::log('v1_functions_enqueue', [ + 'projectId' => $projectDoc->getId(), + 'functionId' => $functionDoc->getId(), + 'scheduleId' => $schedule['$id'] ?? '', + 'schedule' => $schedule['schedule'] ?? '', + ]); + + $queueForFunctions ->setType('schedule') ->setFunction($schedule['resource']) diff --git a/src/Appwrite/Platform/Workers/Executions.php b/src/Appwrite/Platform/Workers/Executions.php index 673e9de791..56867b5804 100644 --- a/src/Appwrite/Platform/Workers/Executions.php +++ b/src/Appwrite/Platform/Workers/Executions.php @@ -3,6 +3,7 @@ namespace Appwrite\Platform\Workers; use Appwrite\Event\Message\Execution; +use Appwrite\Extend\TraceFunctionExecution; use Exception; use Utopia\Database\Database; use Utopia\Platform\Action; @@ -39,6 +40,14 @@ class Executions extends Action throw new Exception('Missing execution'); } + TraceFunctionExecution::log('executions_worker_upsert', [ + 'projectId' => $executionMessage->project->getId(), + 'functionId' => $execution->getAttribute('resourceId', ''), + 'executionId' => $execution->getId(), + 'deploymentId' => $execution->getAttribute('deploymentId', ''), + 'resourceType' => $execution->getAttribute('resourceType', ''), + ]); + $dbForProject->upsertDocument('executions', $execution); } } diff --git a/src/Appwrite/Platform/Workers/Functions.php b/src/Appwrite/Platform/Workers/Functions.php index bed28dad1c..081c6429a5 100644 --- a/src/Appwrite/Platform/Workers/Functions.php +++ b/src/Appwrite/Platform/Workers/Functions.php @@ -9,6 +9,7 @@ use Appwrite\Event\Func; use Appwrite\Event\Realtime; use Appwrite\Event\Webhook; use Appwrite\Extend\Exception as AppwriteException; +use Appwrite\Extend\TraceFunctionExecution; use Appwrite\Utopia\Response\Model\Execution; use Executor\Executor; use Utopia\Bus\Bus; @@ -115,6 +116,17 @@ class Functions extends Action $log->addTag('projectId', $project->getId()); $log->addTag('type', $type); + if (empty($events) && !$function->isEmpty()) { + TraceFunctionExecution::log('functions_worker_dequeue', [ + 'projectId' => $project->getId(), + 'functionId' => $function->getId(), + 'payloadType' => $type, + 'queuePid' => $message->getPid(), + 'queueName' => $message->getQueue(), + 'messageTimestamp' => $message->getTimestamp(), + ]); + } + if (!empty($events)) { $limit = 100; $sum = 100; @@ -304,6 +316,15 @@ class Functions extends Action 'duration' => 0.0, ]); + TraceFunctionExecution::log('functions_worker_before_execution_completed_bus_fail', [ + 'projectId' => $project->getId(), + 'functionId' => $function->getId(), + 'executionId' => $execution->getId(), + 'deploymentId' => $execution->getAttribute('deploymentId', ''), + 'trigger' => $trigger, + 'status' => $execution->getAttribute('status', ''), + ]); + $bus->dispatch(new ExecutionCompleted( execution: $execution->getArrayCopy(), project: $project->getArrayCopy(), @@ -522,6 +543,13 @@ class Functions extends Action $source = $deployment->getAttribute('buildPath', ''); $extension = str_ends_with($source, '.tar') ? 'tar' : 'tar.gz'; $command = $version === 'v2' ? '' : "cp /tmp/code.$extension /mnt/code/code.$extension && nohup helpers/start.sh \"$command\""; + TraceFunctionExecution::log('functions_worker_before_executor', [ + 'projectId' => $project->getId(), + 'functionId' => $functionId, + 'executionId' => $executionId, + 'deploymentId' => $deployment->getId(), + 'trigger' => $trigger, + ]); $executionResponse = $executor->createExecution( projectId: $project->getId(), deploymentId: $deploymentId, @@ -594,6 +622,14 @@ class Functions extends Action $errorCode = $th->getCode(); } finally { /** Persist final execution status and record usage */ + TraceFunctionExecution::log('functions_worker_before_execution_completed_bus', [ + 'projectId' => $project->getId(), + 'functionId' => $functionId, + 'executionId' => $execution->getId(), + 'deploymentId' => $execution->getAttribute('deploymentId', ''), + 'status' => $execution->getAttribute('status', ''), + 'trigger' => $trigger, + ]); $bus->dispatch(new ExecutionCompleted( execution: $execution->getArrayCopy(), project: $project->getArrayCopy(),