From ca87c9732d78f96dfeeba61f992c2c9cc67f2c4b Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Matej=20Ba=C4=8Do?= Date: Tue, 19 Jul 2022 12:07:02 +0000 Subject: [PATCH] Fix proxy race conditions --- .env | 2 +- app/executor.php | 133 ++++++++++++++++++++++---------------- app/functionsProxy.php | 8 ++- src/Executor/Executor.php | 5 +- 4 files changed, 87 insertions(+), 61 deletions(-) diff --git a/.env b/.env index dc121a0d53..35b2abefa2 100644 --- a/.env +++ b/.env @@ -84,4 +84,4 @@ DOCKERHUB_PULL_USERNAME= DOCKERHUB_PULL_PASSWORD= DOCKERHUB_PULL_EMAIL= #_APP_EXECUTORS=executor_01=appwrite-executor,executor_02=appwrite-executor2 -_APP_EXECUTORS=executor_001=104.248.31.61:3000 \ No newline at end of file +_APP_EXECUTORS=executor_001=appwrite-executor \ No newline at end of file diff --git a/app/executor.php b/app/executor.php index 6b5b41903b..bd6d848253 100644 --- a/app/executor.php +++ b/app/executor.php @@ -472,9 +472,9 @@ App::post('/v1/execution') function (string $runtimeId, array $vars, string $data, $timeout, string $projectId, string $deploymentId, string $source, string $destination, array $commands, string $runtime, string $baseImage, string $entrypoint, $activeRuntimes, Response $response) { // Prepare runtime if (!$activeRuntimes->exists($runtimeId)) { - // TODO: Try multiple times? - $executor = new Executor('http://localhost:3000/v1'); - $response = $executor->createRuntime( + // TODO: Try multiple times? Try/catch? + $executor = new Executor('http://localhost/v1'); + $runtimeResponse = $executor->createRuntime( deploymentId: $deploymentId, projectId: $projectId, source: $source, @@ -482,16 +482,16 @@ App::post('/v1/execution') baseImage: $baseImage, vars: $vars, entrypoint: $entrypoint, - commands: [] + commands: [], + key: App::getEnv('_APP_EXECUTOR_SECRET', '') ); - \var_dump($response); } // Ensure runtime started for ($i = 0; $i < 5; $i++) { if ($activeRuntimes->get($runtimeId)['status'] === 'pending') { Console::info('Waiting for runtime to be ready...'); - sleep(1); + \sleep(1); } else { break; } @@ -501,6 +501,7 @@ App::post('/v1/execution') } } + // Ensure we have secret $runtime = $activeRuntimes->get($runtimeId); $secret = $runtime['key']; if (empty($secret)) { @@ -509,59 +510,76 @@ App::post('/v1/execution') Console::info('Executing Runtime: ' . $runtimeId); - $execution = []; $executionStart = \microtime(true); + + // Prepare request to executor + $sendExecuteRequest = function() use ($vars, $data, $runtimeId, $secret) { + $statusCode = 0; + $errNo = -1; + $executorResponse = ''; + + $timeout ??= (int) App::getEnv('_APP_FUNCTIONS_TIMEOUT', 900); + + $ch = \curl_init(); + $body = \json_encode([ + 'env' => $vars, + 'payload' => $data, + 'timeout' => $timeout + ]); + \curl_setopt($ch, CURLOPT_URL, "http://" . $runtimeId . ":3000/"); + \curl_setopt($ch, CURLOPT_POST, true); + \curl_setopt($ch, CURLOPT_POSTFIELDS, $body); + \curl_setopt($ch, CURLOPT_RETURNTRANSFER, true); + \curl_setopt($ch, CURLOPT_TIMEOUT, $timeout); + \curl_setopt($ch, CURLOPT_CONNECTTIMEOUT, 10); + + \curl_setopt($ch, CURLOPT_HTTPHEADER, [ + 'Content-Type: application/json', + 'Content-Length: ' . \strlen($body), + 'x-internal-challenge: ' . $secret, + 'host: null' + ]); + + $executorResponse = \curl_exec($ch); + + $statusCode = \curl_getinfo($ch, CURLINFO_HTTP_CODE); + + $error = \curl_error($ch); + + $errNo = \curl_errno($ch); + + \curl_close($ch); + + return [ + 'errNo' => $errNo, + 'error' => $error, + 'statusCode' => $statusCode, + 'executorResponse' => $executorResponse + ]; + }; + + // Execute function + for ($i = 0; $i < 5; $i++) { + [ $errNo, $error, $statusCode, $executorResponse ] = \call_user_func($sendExecuteRequest); + + // No error + if($errNo === 0) { + break; + } + + Console::info('Waiting for runtime to respond...'); + + \sleep(1); + + if ($i === 4) { + throw new Exception('Runtime failed to respond in allocated time: ' . $error, 500); + } + } + + // Extract response + $execution = []; $stdout = ''; $stderr = ''; - $statusCode = 0; - $errNo = -1; - $executorResponse = ''; - - $timeout ??= (int) App::getEnv('_APP_FUNCTIONS_TIMEOUT', 900); - - $ch = \curl_init(); - $body = \json_encode([ - 'env' => $vars, - 'payload' => $data, - 'timeout' => $timeout - ]); - \curl_setopt($ch, CURLOPT_URL, "http://" . $runtimeId . ":3000/"); - \curl_setopt($ch, CURLOPT_POST, true); - \curl_setopt($ch, CURLOPT_POSTFIELDS, $body); - \curl_setopt($ch, CURLOPT_RETURNTRANSFER, true); - \curl_setopt($ch, CURLOPT_TIMEOUT, $timeout); - \curl_setopt($ch, CURLOPT_CONNECTTIMEOUT, 10); - - \curl_setopt($ch, CURLOPT_HTTPHEADER, [ - 'Content-Type: application/json', - 'Content-Length: ' . \strlen($body), - 'x-internal-challenge: ' . $secret, - 'host: null' - ]); - - $executorResponse = \curl_exec($ch); - - $statusCode = \curl_getinfo($ch, CURLINFO_HTTP_CODE); - - $error = \curl_error($ch); - - $errNo = \curl_errno($ch); - - \curl_close($ch); - - // TODO: Re-run in case of 406 - - switch (true) { - /** No Error. */ - case $errNo === 0: - break; - /** Runtime not ready for requests yet. 111 is the swoole error code for Connection Refused - see https://openswoole.com/docs/swoole-error-code */ - case $errNo === 111: - throw new Exception('An internal curl error has occurred within the executor! Error Msg: ' . $error, 406); - /** Any other CURL error */ - default: - throw new Exception('An internal curl error has occurred within the executor! Error Msg: ' . $error, 500); - } switch (true) { case $statusCode >= 500: @@ -589,10 +607,11 @@ App::post('/v1/execution') 'time' => $executionTime, ]; - /** Update swoole table */ + // Update swoole table $runtime['updated'] = \time(); $activeRuntimes->set($runtimeId, $runtime); + // Finish request $response ->setStatusCode(Response::STATUS_CODE_OK) ->json($execution); diff --git a/app/functionsProxy.php b/app/functionsProxy.php index c3ee8ce96a..ffcd1b8354 100644 --- a/app/functionsProxy.php +++ b/app/functionsProxy.php @@ -66,10 +66,13 @@ function markOnline(cache $cache, string $executorId, bool $forceShowError = fal // Fetch info about executors function fetchExecutorsState(RedisPool $redisPool, bool $forceShowError = false) { + \var_dump("Now"); $executors = \explode(',', App::getEnv('_APP_EXECUTORS', '')); foreach ($executors as $executor) { + \var_dump("Now2"); go(function () use ($redisPool, $executor, $forceShowError) { + \var_dump("Now3"); $redis = $redisPool->get(); $cache = new Cache(new Redis($redis)); @@ -226,7 +229,10 @@ $run = function (SwooleRequest $request, SwooleResponse $response) use ($adapter }; $http->on('start', function () use ($redisPool) { - Timer::tick(30000, fn (int $timerId, array $params) => fetchExecutorsState($params[0]), [$redisPool]); + Timer::tick(3000, function (int $timerId) use ($redisPool) { + \var_dump("OK"); + // fetchExecutorsState($redisPool, false); + }); }); $http->on('request', function (SwooleRequest $swooleRequest, SwooleResponse $swooleResponse) use ($run) { diff --git a/src/Executor/Executor.php b/src/Executor/Executor.php index 4b3a9e4fb4..e5d3b3c477 100644 --- a/src/Executor/Executor.php +++ b/src/Executor/Executor.php @@ -63,12 +63,13 @@ class Executor string $workdir = '', string $destination = '', array $vars = [], - array $commands = [] + array $commands = [], + string $key = null ) { $route = "/runtimes"; $headers = [ 'content-type' => 'application/json', - 'x-appwrite-executor-key' => App::getEnv('_APP_FUNCTIONS_PROXY_SECRET', '') + 'x-appwrite-executor-key' => $key ?? App::getEnv('_APP_FUNCTIONS_PROXY_SECRET', '') ]; $params = [ 'runtimeId' => "$projectId-$deploymentId",