From 545d86d994d3a302cfa2d3a41a1fbd3652d87c01 Mon Sep 17 00:00:00 2001 From: shimon Date: Mon, 31 Oct 2022 15:38:50 +0200 Subject: [PATCH] Timer::tick --- .env | 5 +- app/controllers/api/edge.php | 25 +++++++- app/init.php | 9 ++- app/tasks/sync-edge.php | 40 +++++++++++-- app/worker.php | 112 ++++++++++++++++++++++------------- app/workers/sync-In.php | 28 +++++---- app/workers/sync-out.php | 57 ++++++++++-------- composer.json | 2 +- composer.lock | 14 ++--- docker-compose.yml | 8 +-- 10 files changed, 200 insertions(+), 100 deletions(-) diff --git a/.env b/.env index 950fd44821..900a443080 100644 --- a/.env +++ b/.env @@ -84,8 +84,9 @@ _APP_MAINTENANCE_RETENTION_AUDIT=1209600 _APP_USAGE_TIMESERIES_INTERVAL=2 _APP_USAGE_DATABASE_INTERVAL=15 _APP_USAGE_STATS=enabled -_APP_LOGGING_PROVIDER= -_APP_LOGGING_CONFIG= +_APP_LOGGING_PROVIDER=sentry +_APP_LOGGING_CONFIG=4190483bb7d14d659e526aa61b0c7b5e;4504077395230720 +_APP_REGION=nyc1 DOCKERHUB_PULL_USERNAME= DOCKERHUB_PULL_PASSWORD= DOCKERHUB_PULL_EMAIL= \ No newline at end of file diff --git a/app/controllers/api/edge.php b/app/controllers/api/edge.php index 22c845a557..858a39c87d 100644 --- a/app/controllers/api/edge.php +++ b/app/controllers/api/edge.php @@ -2,14 +2,16 @@ use Ahc\Jwt\JWT; use Ahc\Jwt\JWTException; +use Appwrite\DSN\DSN; use Appwrite\Extend\Exception; +use Appwrite\URL\URL as AppwriteURL; use Appwrite\Utopia\Request; use Appwrite\Utopia\Response; use Utopia\App; +use Utopia\CLI\Console; use Utopia\Database\Document; use Utopia\Queue\Client as SyncIn; use Utopia\Queue\Connection\Redis as QueueRedis; -use Utopia\Registry\Registry; use Utopia\Validator\ArrayList; use Utopia\Validator\Text; @@ -41,7 +43,26 @@ App::post('/v1/edge/sync') throw new Exception(Exception::KEY_NOT_FOUND); } - $client = new SyncIn('syncIn', new QueueRedis(App::getEnv('_APP_REDIS_HOST', 'redis'), App::getEnv('_APP_REDIS_PORT', '6379'))); + $fallbackForRedis = AppwriteURL::unparse([ + 'scheme' => 'redis', + 'host' => App::getEnv('_APP_REDIS_HOST', 'redis'), + 'port' => App::getEnv('_APP_REDIS_PORT', '6379'), + 'user' => App::getEnv('_APP_REDIS_USER', ''), + 'pass' => App::getEnv('_APP_REDIS_PASS', ''), + ]); + + $connection = App::getEnv('_APP_CONNECTIONS_QUEUE', $fallbackForRedis); + $dsns = explode(',', $connection ?? ''); + + if (empty($dsns)) { + Console::error("No Dsn found"); + } + + $dsn = explode('=', $dsns[0]); + $dsn = $dsn[1] ?? ''; + $dsn = new DSN($dsn); + + $client = new SyncIn('syncIn', new QueueRedis($dsn->getHost(), $dsn->getPort())); $client->enqueue(['value' => ['keys' => $keys]]); diff --git a/app/init.php b/app/init.php index 8a427e72ee..4673ccb59f 100644 --- a/app/init.php +++ b/app/init.php @@ -931,7 +931,7 @@ App::setResource('user', function ($mode, $project, $console, $request, $respons if ($project->isEmpty()) { $user = new Document(['$id' => ID::custom(''), '$collection' => 'users']); } else { - var_dump('Save cache invoked'); + //Todo fix cache save re-invoked $user = $dbForProject->getDocument('users', Auth::$unique); } } else { @@ -1080,11 +1080,18 @@ App::setResource('cache', function (Group $pools) { $client = new SyncOut('syncOut', new QueueRedis(App::getEnv('_APP_REDIS_HOST', 'redis'), App::getEnv('_APP_REDIS_PORT', '6379'))); $cache->on(cache::EVENT_SAVE, function ($key) use ($client) { + //Todo fix cache re-invoked + if ($key === 'cache-console:_metadata:users') { + return; + } $client ->enqueue(['value' => ['key' => $key]]); }); $cache->on(cache::EVENT_PURGE, function ($key) use ($client) { + if ($key === 'cache-console:_metadata:users') { + return; + } $client ->enqueue(['value' => ['key' => $key]]); }); diff --git a/app/tasks/sync-edge.php b/app/tasks/sync-edge.php index a3564a3685..18ad595f02 100644 --- a/app/tasks/sync-edge.php +++ b/app/tasks/sync-edge.php @@ -3,6 +3,8 @@ global $cli; global $register; +use Appwrite\DSN\DSN; +use Appwrite\URL\URL as AppwriteURL; use Utopia\App; use Utopia\CLI\Console; use Utopia\Database\DateTime; @@ -15,13 +17,39 @@ $cli ->desc('Schedules edge sync tasks') ->action(function () use ($register) { Console::title('Syncs edges V1'); - Console::success(APP_NAME . ' Syncs Edge process v1 has started'); - sleep(3); + Console::success(APP_NAME . ' Sync Edge process v1 has started'); + + $fallbackForRedis = AppwriteURL::unparse([ + 'scheme' => 'redis', + 'host' => App::getEnv('_APP_REDIS_HOST', 'redis'), + 'port' => App::getEnv('_APP_REDIS_PORT', '6379'), + 'user' => App::getEnv('_APP_REDIS_USER', ''), + 'pass' => App::getEnv('_APP_REDIS_PASS', ''), + ]); + + $connection = App::getEnv('_APP_CONNECTIONS_QUEUE', $fallbackForRedis); + $dsns = explode(',', $connection ?? ''); + + if (empty($dsns)) { + Console::error("No Dsn found"); + } + + $dsn = explode('=', $dsns[0]); + $dsn = $dsn[1] ?? ''; + $dsn = new DSN($dsn); + + // Todo fix pdo PDOException + //Table 'appwrite.console__metadata' doesn't exist + sleep(4); $interval = (int) App::getEnv('_APP_SYNC_EDGE_INTERVAL', '180'); - Console::loop(function () use ($interval, $register) { + Console::loop(function () use ($interval, $register, $dsn) { $database = getConsoleDB(); $time = DateTime::now(); - $region = App::getEnv('_APP_REGION', 'nyc1'); + $region = App::getEnv('_APP_REGION', 'default'); + if (App::getEnv('_APP_REGION', 'default') === 'default') { + return; + } + Console::info("[{$time}] Notifying workers with edges tasks every {$interval} seconds"); $time = DateTime::now(); @@ -31,9 +59,9 @@ $cli ]); if (count($chunks) > 0) { - $client = new SyncOut('syncOut', new QueueRedis(App::getEnv('_APP_REDIS_HOST', 'redis'), App::getEnv('_APP_REDIS_PORT', '6379'))); + $client = new SyncOut('syncOut', new QueueRedis($dsn->getHost(), $dsn->getPort())); foreach ($chunks as $counter => $chunk) { - Console::info("[{$time}] Sending chunk .($counter+1). ot of " . count($chunks) . " to {$chunk->getAttribute('target')}"); + Console::info("[{$time}] Sending chunk .$counter. ot of " . count($chunks) . " to {$chunk->getAttribute('target')}"); $client ->enqueue([ 'value' => [ diff --git a/app/worker.php b/app/worker.php index eef1248f36..d1a68ea278 100644 --- a/app/worker.php +++ b/app/worker.php @@ -2,19 +2,28 @@ require_once __DIR__ . '/init.php'; +use Appwrite\DSN\DSN; +use Appwrite\URL\URL as AppwriteURL; +use Swoole\Runtime; +use Utopia\App; use Utopia\Cache\Adapter\Sharding; use Utopia\Cache\Cache; +use Utopia\CLI\Console; use Utopia\Config\Config; use Utopia\Database\Database; +use Utopia\Database\Validator\Authorization; +use Utopia\Logger\Log; +use Utopia\Logger\Logger; use Utopia\Queue\Server; use Utopia\Registry\Registry; global $register; +Runtime::enableCoroutine(SWOOLE_HOOK_ALL); + Server::setResource('register', fn() => $register); Server::setResource('dbForConsole', function (Cache $cache, Registry $register) { - $pools = $register->get('pools'); $dbAdapter = $pools ->get('console') @@ -29,7 +38,6 @@ Server::setResource('dbForConsole', function (Cache $cache, Registry $register) }, ['cache', 'register']); Server::setResource('cache', function (Registry $register) { - $pools = $register->get('pools'); $list = Config::getParam('pools-cache', []); $adapters = []; @@ -45,49 +53,71 @@ Server::setResource('cache', function (Registry $register) { return new Cache(new Sharding($adapters)); }, ['register']); -/** - * Get console database - * @return Database - */ -function getConsoleDB(): Database -{ - global $register; +Server::setResource('ErrorLog', function (Logger $logger) { + return function (Throwable $error, $action) use ($logger) { - $pools = $register->get('pools'); /** @var \Utopia\Pools\Group $pools */ + if ($logger) { + $version = App::getEnv('_APP_VERSION', 'UNKNOWN'); - $dbAdapter = $pools - ->get('console') - ->pop() - ->getResource() - ; + if ($error->getCode() >= 500 || $error->getCode() === 0) { + $log = new Log(); - $database = new Database($dbAdapter, getCache()); + $log->setNamespace("http"); + $log->setServer(\gethostname()); + $log->setVersion($version); + $log->setType(Log::TYPE_ERROR); + $log->setMessage($error->getMessage()); - $database->setNamespace('console'); + $log->setAction($action); - return $database; + $log->addTag('verboseType', get_class($error)); + $log->addTag('code', $error->getCode()); + + $log->addExtra('file', $error->getFile()); + $log->addExtra('line', $error->getLine()); + $log->addExtra('trace', $error->getTraceAsString()); + $log->addExtra('detailedTrace', $error->getTrace()); + $log->addExtra('roles', Authorization::$roles); + + $isProduction = App::getEnv('_APP_ENV', 'development') === 'production'; + $log->setEnvironment($isProduction ? Log::ENVIRONMENT_PRODUCTION : Log::ENVIRONMENT_STAGING); + + $responseCode = $logger->addLog($log); + Console::info('Log pushed with status code: ' . $responseCode); + } + } + + $code = $error->getCode(); + $message = $error->getMessage(); + $file = $error->getFile(); + $line = $error->getLine(); + $trace = $error->getTrace(); + + Console::error('[Error] Timestamp: ' . date('c', time())); + Console::error('[Error] Type: ' . get_class($error)); + Console::error('[Error] Message: ' . $message); + Console::error('[Error] File: ' . $file); + Console::error('[Error] Line: ' . $line); + Console::error('[Error] Code: ' . $code); + Console::error('[Error] Trace: ' . $trace); + }; +}); + +$fallbackForRedis = AppwriteURL::unparse([ + 'scheme' => 'redis', + 'host' => App::getEnv('_APP_REDIS_HOST', 'redis'), + 'port' => App::getEnv('_APP_REDIS_PORT', '6379'), + 'user' => App::getEnv('_APP_REDIS_USER', ''), + 'pass' => App::getEnv('_APP_REDIS_PASS', ''), +]); + +$connection = App::getEnv('_APP_CONNECTIONS_QUEUE', $fallbackForRedis); +$dsns = explode(',', $connection ?? ''); + +if (empty($dsns)) { + Console::error("Dsn not found"); } -/** - * Get Cache - * @return Cache - */ -function getCache(): Cache -{ - global $register; - - $pools = $register->get('pools'); /** @var \Utopia\Pools\Group $pools */ - - $list = Config::getParam('pools-cache', []); - $adapters = []; - - foreach ($list as $value) { - $adapters[] = $pools - ->get($value) - ->pop() - ->getResource() - ; - } - - return new Cache(new Sharding($adapters)); -} +$dsn = explode('=', $dsns[0]); +$dsn = $dsn[1] ?? ''; +$dsn = new DSN($dsn); diff --git a/app/workers/sync-In.php b/app/workers/sync-In.php index 5c52df08df..86f957a0e3 100644 --- a/app/workers/sync-In.php +++ b/app/workers/sync-In.php @@ -2,7 +2,7 @@ require_once __DIR__ . '/../worker.php'; - +use Appwrite\Extend\Exception; use Utopia\App; use Utopia\Cache\Cache; use Utopia\CLI\Console; @@ -10,9 +10,14 @@ use Utopia\Database\DateTime; use Utopia\Queue; use Utopia\Queue\Message; -global $register; +if (App::getEnv('_APP_REGION', 'default') === 'default') { + throw new Exception(Exception::GENERAL_SERVER_ERROR); +} -$connection = new Queue\Connection\Redis(App::getEnv('_APP_REDIS_HOST', 'redis'), App::getEnv('_APP_REDIS_PORT', '6379')); +global $register; +global $dsn; + +$connection = new Queue\Connection\Redis($dsn->getHost(), $dsn->getPort()); $adapter = new Queue\Adapter\Swoole($connection, 2, 'syncIn'); $server = new Queue\Server($adapter); @@ -33,13 +38,16 @@ $server->job() $server ->error() ->inject('error') - ->action(function ($error) { - echo $error->getMessage() . PHP_EOL; - echo $error->getLine() . PHP_EOL; + ->inject('logError') + ->action(function ($error, $logError) { + Console::error($error->getMessage() . ' ' . $error->getFile() . ' ' . $error->getLine()); + call_user_func($logError, $error, 'sync-in-worker'); }); $server - ->workerStart(function () { - echo "In [" . App::getEnv('_APP_REGION', 'nyc1') . "] edge cache purging worker Started" . PHP_EOL; - }) - ->start(); + ->workerStart() + ->action(function () { + Console::success("In [" . App::getEnv('_APP_REGION', 'nyc1') . "] edge cache purging worker Started"); + }); + +$server->start(); diff --git a/app/workers/sync-out.php b/app/workers/sync-out.php index 00d7f4bfcf..ee89c4bf0b 100644 --- a/app/workers/sync-out.php +++ b/app/workers/sync-out.php @@ -3,8 +3,8 @@ require_once __DIR__ . '/../worker.php'; use Ahc\Jwt\JWT; +use Appwrite\Extend\Exception; use Appwrite\Utopia\Response; -use Swoole\Runtime; use Swoole\Timer; use Utopia\App; use Utopia\CLI\Console; @@ -16,9 +16,15 @@ use Utopia\Database\Exception\Structure; use Utopia\Queue; use Utopia\Queue\Message; +if (App::getEnv('_APP_REGION', 'default') === 'default') { + throw new Exception(Exception::GENERAL_SERVER_ERROR); +} + +global $dsn; + $regions = array_filter( Config::getParam('regions', []), - fn ($region) => App::getEnv('_APP_REGION', 'nyc1') !== $region + fn ($region) => App::getEnv('_APP_REGION') !== $region && $region !== 'default', ARRAY_FILTER_USE_KEY ); @@ -38,7 +44,7 @@ const MAX_CURL_SEND_ATTEMPTS = 4; * @param array $stack * @return array */ -function send(string $url, string $token, array $stack): array +function call(string $url, string $token, array $stack): array { $payload = []; $ch = curl_init($url); @@ -76,7 +82,7 @@ function send(string $url, string $token, array $stack): array * @throws Structure * @throws Exception */ -function call($regions, $stack): void +function handle($dbForConsole, $regions, $stack): void { global $register; @@ -86,13 +92,12 @@ function call($regions, $stack): void foreach ($regions as $code => $region) { $time = DateTime::now(); - $response = send($region['domain'] . '/v1/edge/sync', $token, ['keys' => $stack]); + $response = call($region['domain'] . '/v1/edge/sync', $token, ['keys' => $stack]); if ($response['status'] !== Response::STATUS_CODE_OK) { Console::error("[{$time}] Request to {$code} has failed"); try { - $database = getConsoleDB(); - $database->createDocument('syncs', new Document([ - 'region' => App::getEnv('_APP_REGION', 'nyc1'), + $dbForConsole->createDocument('syncs', new Document([ + 'region' => App::getEnv('_APP_REGION'), 'target' => $code, 'keys' => $stack, 'status' => $response['status'], @@ -105,13 +110,11 @@ function call($regions, $stack): void } } -$connection = new Queue\Connection\Redis(App::getEnv('_APP_REDIS_HOST', 'redis'), App::getEnv('_APP_REDIS_PORT', '6379')); -$adapter = new Queue\Adapter\Swoole($connection, 1, 'syncOut'); +$connection = new Queue\Connection\Redis($dsn->getHost(), $dsn->getPort()); +$adapter = new Queue\Adapter\Swoole($connection, 2, 'syncOut'); $server = new Queue\Server($adapter); - $server->job() ->inject('message') - ->inject('dbForConsole') ->action(function (Message $message) use (&$stack, &$failures) { $payload = $message->getPayload()['value'] ?? []; @@ -136,20 +139,24 @@ $server->job() } }); -Runtime::enableCoroutine(SWOOLE_HOOK_ALL); - $server ->error() ->inject('error') - ->action(function ($error) { - echo $error->getMessage() . PHP_EOL; - echo $error->getLine() . PHP_EOL; + ->inject('errorLog') + ->action(function ($error, $errorLog) { + var_dump($error); + Console::error($error->getMessage() . ' ' . $error->getFile() . ' ' . $error->getLine()); + call_user_func($errorLog, $error, 'sync-out-worker'); }); $server - ->workerStart(function () use (&$stack, &$failures) { - Timer::tick(20000, function () use (&$stack, &$failures) { + ->workerStart() + ->inject('dbForConsole') + ->action(function ($dbForConsole) use (&$stack, &$failures) { + + Timer::tick(5000, function () use ($dbForConsole, &$stack, &$failures) { $time = DateTime::now(); + if (empty($stack['keys']) && count($failures) === 0) { Console::info("[{$time}] Stack is empty"); return; @@ -160,7 +167,7 @@ $server while ($i < count($failures)) { $failure = array_shift($failures); Console::info("[{$time}] ReSending " . count($failure['keys']) . " to " . key($failure['regions'])); - call($failure['regions'], $failure['keys']); + handle($dbForConsole, $failure['regions'], $failure['keys']); $i++; } return; @@ -169,10 +176,10 @@ $server $chunk = array_slice($stack['keys'], 0, CHUNK_MAX_KEYS); array_splice($stack['keys'], 0, CHUNK_MAX_KEYS); Console::info("[{$time}] Sending " . count($chunk) . " remains " . count($stack['keys'])); - call($stack['regions'], $chunk); - //var_dump($stack['keys']); + handle($dbForConsole, $stack['regions'], $chunk); $chunk = []; }); - echo "Out [" . App::getEnv('_APP_REGION', 'nyc1') . "] edge cache purging worker Started" . PHP_EOL; - }) - ->start(); + Console::success("Out [" . App::getEnv('_APP_REGION') . "] edge cache purging worker Started"); + }); + + $server->start(); diff --git a/composer.json b/composer.json index ccfadca27d..110e44ca9d 100644 --- a/composer.json +++ b/composer.json @@ -61,7 +61,7 @@ "utopia-php/websocket": "0.1.0", "utopia-php/image": "0.5.*", "utopia-php/orchestration": "0.6.*", - "utopia-php/queue": "0.3.0", + "utopia-php/queue": "0.4.0", "utopia-php/pools": "0.1.*", "resque/php-resque": "1.3.6", "matomo/device-detector": "6.0.0", diff --git a/composer.lock b/composer.lock index e57e277ad2..2aa790ead4 100644 --- a/composer.lock +++ b/composer.lock @@ -4,7 +4,7 @@ "Read more about it at https://getcomposer.org/doc/01-basic-usage.md#installing-dependencies", "This file is @generated automatically" ], - "content-hash": "d2d51b9a3dcbb3542243d1a68a078ece", + "content-hash": "a02c3502dca5a3a9f0f283e06e11e30e", "packages": [ { "name": "adhocore/jwt", @@ -2537,16 +2537,16 @@ }, { "name": "utopia-php/queue", - "version": "0.3.0", + "version": "0.4.0", "source": { "type": "git", "url": "https://github.com/utopia-php/queue.git", - "reference": "42b132c6f2431b726c2bc629c386921e4934b863" + "reference": "0cad4cf4231377aa6c67956b51ba1954e0d02166" }, "dist": { "type": "zip", - "url": "https://api.github.com/repos/utopia-php/queue/zipball/42b132c6f2431b726c2bc629c386921e4934b863", - "reference": "42b132c6f2431b726c2bc629c386921e4934b863", + "url": "https://api.github.com/repos/utopia-php/queue/zipball/0cad4cf4231377aa6c67956b51ba1954e0d02166", + "reference": "0cad4cf4231377aa6c67956b51ba1954e0d02166", "shasum": "" }, "require": { @@ -2592,9 +2592,9 @@ ], "support": { "issues": "https://github.com/utopia-php/queue/issues", - "source": "https://github.com/utopia-php/queue/tree/0.3.0" + "source": "https://github.com/utopia-php/queue/tree/0.4.0" }, - "time": "2022-10-19T13:22:07+00:00" + "time": "2022-10-31T06:23:08+00:00" }, { "name": "utopia-php/registry", diff --git a/docker-compose.yml b/docker-compose.yml index c5bf54a205..612b9d7b60 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -280,9 +280,6 @@ services: volumes: - ./app:/usr/src/code/app - ./src:/usr/src/code/src - - ./vendor/utopia-php/queue:/usr/src/code/vendor/utopia-php/queue - - ./vendor/utopia-php/cache:/usr/src/code/vendor/utopia-php/cache - - ./vendor/utopia-php/database:/usr/src/code/vendor/utopia-php/database depends_on: - mariadb - redis @@ -301,6 +298,7 @@ services: - _APP_CONNECTIONS_DB_CONSOLE - _APP_CONNECTIONS_CACHE - _APP_CONNECTIONS_QUEUE + - _APP_REGION appwrite-worker-sync-in: entrypoint: worker-sync-in @@ -312,8 +310,6 @@ services: volumes: - ./app:/usr/src/code/app - ./src:/usr/src/code/src - - ./vendor/utopia-php/cache:/usr/src/code/vendor/utopia-php/cache - - ./vendor/utopia-php/database:/usr/src/code/vendor/utopia-php/database depends_on: - redis environment: @@ -326,6 +322,7 @@ services: - _APP_CONNECTIONS_DB_CONSOLE - _APP_CONNECTIONS_CACHE - _APP_CONNECTIONS_QUEUE + - _APP_REGION appwrite-worker-webhooks: entrypoint: worker-webhooks @@ -696,6 +693,7 @@ services: - _APP_CONNECTIONS_DB_CONSOLE - _APP_CONNECTIONS_CACHE - _APP_CONNECTIONS_QUEUE + - _APP_REGION appwrite-usage-timeseries: entrypoint: