diff --git a/app/controllers/api/edge.php b/app/controllers/api/edge.php index 5edf899589..bd80b91843 100644 --- a/app/controllers/api/edge.php +++ b/app/controllers/api/edge.php @@ -35,21 +35,13 @@ App::post('/v1/edge/sync') ->param('keys', '', new ArrayList(new Text(100), 1000), 'Cache keys. an array containing alphanumerical cache keys') ->inject('request') ->inject('response') - ->inject('register') - ->action(function (array $keys, Request $request, Response $response, Registry $register) { + ->action(function (array $keys, Request $request, Response $response) { - if (empty($keys)) { + //if (empty($keys)) { throw new Exception(Exception::KEY_NOT_FOUND); - } + //} - $pools = $register->get('pools'); - $queue = $pools - ->get('queue') - ->pop() - ->getResource() - ; - - $client = new SyncIn('syncIn', new QueueRedis(fn() => $queue)); + $client = new SyncIn('syncIn', new QueueRedis(App::getEnv('_APP_REDIS_HOST', 'redis'), App::getEnv('_APP_REDIS_PORT', '6379'))); $client->enqueue(['value' => ['keys' => $keys]]); diff --git a/app/init.php b/app/init.php index d0ec741e86..94a60e581e 100644 --- a/app/init.php +++ b/app/init.php @@ -533,12 +533,12 @@ $register->set('pools', function () { 'multiple' => true, 'schemes' => ['mariadb', 'mysql'], ], - 'queue' => [ - 'type' => 'queue', - 'dsns' => App::getEnv('_APP_CONNECTIONS_QUEUE', $fallbackForRedis), - 'multiple' => false, - 'schemes' => ['redis'], - ], +// 'queue' => [ +// 'type' => 'queue', +// 'dsns' => App::getEnv('_APP_CONNECTIONS_QUEUE', $fallbackForRedis), +// 'multiple' => false, +// 'schemes' => ['redis'], +// ], 'pubsub' => [ 'type' => 'pubsub', 'dsns' => App::getEnv('_APP_CONNECTIONS_PUBSUB', $fallbackForRedis), @@ -641,9 +641,11 @@ $register->set('pools', function () { $adapter->setDefaultDatabase($dsn->getDatabase()); break; case 'pubsub': - case 'queue': - $adapter = $resource(); break; + $adapter = $resource(); +// case 'queue': +// $adapter = $resource(); +// break; case 'cache': $adapter = match ($dsn->getScheme()) { 'redis' => new RedisCache($resource()), @@ -1025,12 +1027,11 @@ App::setResource('console', function () { ]); }, []); -App::setResource('queue', function (Group $pools) { - return $pools - ->get('queue') - ->pop() - ->getResource(); -}, ['pools']); +//App::setResource('queue', function (Group $pools) { +// $pools->get('queue') +// ->pop() +// ->getResource(); +//}, ['pools']); App::setResource('dbForProject', function (Group $pools, Database $dbForConsole, Cache $cache, Document $project) { if ($project->isEmpty() || $project->getId() === 'console') { @@ -1062,7 +1063,7 @@ App::setResource('dbForConsole', function (Group $pools, Cache $cache) { return $database; }, ['pools', 'cache']); -App::setResource('cache', function (Group $pools, $queue) { +App::setResource('cache', function (Group $pools) { $list = Config::getParam('pools-cache', []); $adapters = []; @@ -1075,7 +1076,7 @@ App::setResource('cache', function (Group $pools, $queue) { } $cache = new Cache(new Sharding($adapters)); - $client = new SyncOut('syncOut', new QueueRedis(fn() => $queue)); + $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) { $client @@ -1088,7 +1089,7 @@ App::setResource('cache', function (Group $pools, $queue) { }); return $cache; -}, ['pools', 'queue']); +}, ['pools']); App::setResource('deviceLocal', function () { return new Local(); diff --git a/app/tasks/sync-edge.php b/app/tasks/sync-edge.php index c158895747..b9530b2d84 100644 --- a/app/tasks/sync-edge.php +++ b/app/tasks/sync-edge.php @@ -7,7 +7,7 @@ use Utopia\App; use Utopia\CLI\Console; use Utopia\Database\DateTime; use Utopia\Database\Query; -use Utopia\Queue\Client as SyncIn; +use Utopia\Queue\Client as SyncOut; use Utopia\Queue\Connection\Redis as QueueRedis; $cli @@ -15,14 +15,14 @@ $cli ->desc('Schedules edge sync tasks') ->action(function () use ($register) { Console::title('Syncs edges V1'); - Console::success(APP_NAME . ' Syncs cloud process v1 has started'); + Console::success(APP_NAME . ' Syncs Edge process v1 has started'); $interval = (int) App::getEnv('_APP_SYNC_EDGE_INTERVAL', '180'); Console::loop(function () use ($interval, $register) { $database = getConsoleDB(); $time = DateTime::now(); $region = App::getEnv('_APP_REGION', 'nyc1'); - Console::info("[{$time}] Notifying workers with cloud tasks every {$interval} seconds"); + Console::info("[{$time}] Notifying workers with edges tasks every {$interval} seconds"); $time = DateTime::now(); $chunks = $database->find('syncs', [ @@ -31,22 +31,14 @@ $cli ]); if (count($chunks) > 0) { - Console::info("[{$time}] Found " . \count($chunks) . " cache key chunks to purge."); - - $pools = $register->get('pools'); - $queue = $pools - ->get('queue') - ->pop() - ->getResource() - ; - - $client = new SyncIn('syncIn', new QueueRedis(fn() => $queue)); - foreach ($chunks as $chunk) { + $client = new SyncOut('syncOut', new QueueRedis(App::getEnv('_APP_REDIS_HOST', 'redis'), App::getEnv('_APP_REDIS_PORT', '6379'))); + foreach ($chunks as $counter => $chunk) { + Console::info("[{$time}] Sending chunk .($counter+1). ot of " . count($chunks) . " to {$chunk->getAttribute('target')}"); $client ->enqueue([ 'value' => [ - 'region' => $chunk->getAttribute('regionDest'), - 'chunk' => $chunk->getAttribute('keys') + 'region' => $chunk->getAttribute('target'), + 'keys' => $chunk->getAttribute('keys') ] ]); diff --git a/app/worker.php b/app/worker.php index 897c49c029..eef1248f36 100644 --- a/app/worker.php +++ b/app/worker.php @@ -45,3 +45,49 @@ Server::setResource('cache', function (Registry $register) { return new Cache(new Sharding($adapters)); }, ['register']); +/** + * Get console database + * @return Database + */ +function getConsoleDB(): Database +{ + global $register; + + $pools = $register->get('pools'); /** @var \Utopia\Pools\Group $pools */ + + $dbAdapter = $pools + ->get('console') + ->pop() + ->getResource() + ; + + $database = new Database($dbAdapter, getCache()); + + $database->setNamespace('console'); + + return $database; +} + +/** + * 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)); +} diff --git a/app/workers/sync-In.php b/app/workers/sync-In.php index 5ce9518583..0482f92b08 100644 --- a/app/workers/sync-In.php +++ b/app/workers/sync-In.php @@ -10,15 +10,8 @@ use Utopia\Queue\Message; global $register; -$pools = $register->get('pools'); -$queue = $pools - ->get('queue') - ->pop() - ->getResource() -; - -$connection = new Queue\Connection\Redis(fn() => $queue); -$adapter = new Queue\Adapter\Swoole($connection, 1, 'syncIn'); +$connection = new Queue\Connection\Redis(App::getEnv('_APP_REDIS_HOST', 'redis'), App::getEnv('_APP_REDIS_PORT', '6379')); +$adapter = new Queue\Adapter\Swoole($connection, 2, 'syncIn'); $server = new Queue\Server($adapter); $server->job() diff --git a/app/workers/sync-out.php b/app/workers/sync-out.php index b6d34096d2..b8dc9d8e2e 100644 --- a/app/workers/sync-out.php +++ b/app/workers/sync-out.php @@ -7,30 +7,39 @@ use Appwrite\Utopia\Response; use Swoole\Runtime; use Swoole\Timer; use Utopia\App; +use Utopia\CLI\Console; use Utopia\Config\Config; -use Utopia\Database\Database; use Utopia\Database\Document; use Utopia\Database\Exception\Authorization; use Utopia\Database\Exception\Structure; use Utopia\Queue; use Utopia\Queue\Message; -global $register; -$keys = []; -$counter = 0; +$regions = array_filter( + Config::getParam('regions', []), + fn ($region) => App::getEnv('_APP_REGION', 'nyc1') !== $region + && $region !== 'default', + ARRAY_FILTER_USE_KEY +); -const SUBMITION_INTERVAL = 20; -const MAX_KEY_COUNT = 10; +$stack = [ + 'regions' => $regions, + 'keys' => [], +]; + +$failures = []; + +const MAX_KEY_COUNT = 2; const MAX_CURL_SEND_ATTEMPTS = 4; /** * @param string $url * @param string $token - * @param array $keys + * @param array $stack * @return array */ -function send(string $url, string $token, array $keys): array +function send(string $url, string $token, array $stack): array { $payload = []; $ch = curl_init($url); @@ -41,7 +50,7 @@ function send(string $url, string $token, array $keys): array curl_setopt($ch, CURLOPT_RETURNTRANSFER, true); curl_setopt($ch, CURLOPT_TIMEOUT, 5); curl_setopt($ch, CURLOPT_CUSTOMREQUEST, 'POST'); - curl_setopt($ch, CURLOPT_POSTFIELDS, json_encode($keys)); + curl_setopt($ch, CURLOPT_POSTFIELDS, json_encode($stack)); for ($attempts = 0; $attempts < MAX_CURL_SEND_ATTEMPTS; $attempts++) { $response = curl_exec($ch); @@ -68,24 +77,20 @@ function send(string $url, string $token, array $keys): array * @throws Structure * @throws Exception */ -function call($database, $regions, $keys): void +function call($database, $regions, $stack): void { $jwt = new JWT(App::getEnv('_APP_OPENSSL_KEY_V1'), 'HS256', 600, 10); $token = $jwt->encode([]); foreach ($regions as $code => $region) { - var_dump('Sending request to ' . $code . '...............'); - $response = send($region['domain'] . '/v1/edge/sync', $token, ['keys' => $keys]); - var_dump([ - 'keys' => $keys, - 'response' => $response - ]); + Console::info("Sending request to {$code}"); + $response = send($region['domain'] . '/v1/edge/sync', $token, ['keys' => $stack]); if ($response['status'] !== Response::STATUS_CODE_OK) { $database->createDocument('syncs', new Document([ 'region' => App::getEnv('_APP_REGION', 'nyc1'), 'target' => $code, - 'keys' => $keys, + 'keys' => $stack, 'status' => $response['status'], 'payload' => $response['payload'], ])); @@ -93,53 +98,35 @@ function call($database, $regions, $keys): void } } -$pools = $register->get('pools'); -$queue = $pools - ->get('queue') - ->pop() - ->getResource() -; - -$connection = new Queue\Connection\Redis(fn() => $queue); -$adapter = new Queue\Adapter\Swoole($connection, 2, 'syncOut'); +$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'); $server = new Queue\Server($adapter); - - $server->job() ->inject('message') ->inject('dbForConsole') - ->action(function (Message $message, Database $dbForConsole) use (&$keys, &$counter) { + ->action(function (Message $message) use (&$stack, &$failures) { - $payload = $message->getPayload()['value']; - $regions = Config::getParam('regions', true); - $regions = array_filter( - $regions, - fn ($region) => App::getEnv('_APP_REGION', 'nyc1') !== $region - && $region !== 'default', - ARRAY_FILTER_USE_KEY - ); + $payload = $message->getPayload()['value'] ?? []; - if (!empty($payload['region'])) { + if (!empty($payload['keys'])) { $regions = array_filter( - $regions, - fn ($region) => $payload['region'] === $region, + Config::getParam('regions', []), + fn ($region) => $payload['region'] === $region, ARRAY_FILTER_USE_KEY ); + + $failures[] = [ + 'regions' => $regions, + 'keys' => $payload['keys'] + ]; } - if (!empty($payload['chunk'])) { - call($dbForConsole, $regions, $payload['chunk']); - return; + if (!empty($payload['key'])) { + if (!in_array($payload['key'], $stack['keys'] ?? [])) { + $stack['keys'][] = $payload['key']; + } } - - $keys[$payload['key']] = null; - -// if (count($keys) >= MAX_KEY_COUNT || ($counter + SUBMITION_INTERVAL) < time()) { -// call($dbForConsole, $regions, array_keys($keys)); -// $counter = time(); -// $keys = []; -// } }); Runtime::enableCoroutine(SWOOLE_HOOK_ALL); @@ -153,11 +140,28 @@ $server }); $server - ->workerStart(function () { - Timer::tick(1000, function () { - var_dump(date('m/d/Y H:i:s', time())); + ->workerStart(function () use (&$stack, &$failures) { + Timer::tick(10000, function () use (&$stack, &$failures) { + if (empty($stack['keys']) && count($failures) === 0) { + Console::info("Stack is empty"); + return; + } + + if (count($failures) > 0) { + $i = 0; + while ($i < count($failures)) { + $failure = array_shift($failures); + call(getConsoleDB(), $failure['regions'], $failure['keys']); + $i++; + } + return; + } + + $chunk = array_slice($stack['keys'], 0, MAX_KEY_COUNT); + array_splice($stack['keys'], 0, MAX_KEY_COUNT); + call(getConsoleDB(), $stack['regions'], $chunk); + $chunk = []; }); echo "Out region [" . App::getEnv('_APP_REGION', 'nyc1') . "] cache purging worker Started" . PHP_EOL; }) ->start(); - diff --git a/composer.json b/composer.json index 7ca5939359..ccfadca27d 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": "dev-refactor-redis-client as 0.3.0", + "utopia-php/queue": "0.3.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 7ef8f25135..76d15fb9f8 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": "0b5f0fd08db65ca4c2e2f1a6d33cd299", + "content-hash": "d2d51b9a3dcbb3542243d1a68a078ece", "packages": [ { "name": "adhocore/jwt", @@ -693,16 +693,16 @@ }, { "name": "guzzlehttp/psr7", - "version": "2.4.1", + "version": "2.4.2", "source": { "type": "git", "url": "https://github.com/guzzle/psr7.git", - "reference": "69568e4293f4fa993f3b0e51c9723e1e17c41379" + "reference": "3148458748274be1546f8f2809a6c09fe66f44aa" }, "dist": { "type": "zip", - "url": "https://api.github.com/repos/guzzle/psr7/zipball/69568e4293f4fa993f3b0e51c9723e1e17c41379", - "reference": "69568e4293f4fa993f3b0e51c9723e1e17c41379", + "url": "https://api.github.com/repos/guzzle/psr7/zipball/3148458748274be1546f8f2809a6c09fe66f44aa", + "reference": "3148458748274be1546f8f2809a6c09fe66f44aa", "shasum": "" }, "require": { @@ -792,7 +792,7 @@ ], "support": { "issues": "https://github.com/guzzle/psr7/issues", - "source": "https://github.com/guzzle/psr7/tree/2.4.1" + "source": "https://github.com/guzzle/psr7/tree/2.4.2" }, "funding": [ { @@ -808,7 +808,7 @@ "type": "tidelift" } ], - "time": "2022-08-28T14:45:39+00:00" + "time": "2022-10-25T13:49:28+00:00" }, { "name": "influxdb/influxdb-php", @@ -1901,12 +1901,12 @@ "source": { "type": "git", "url": "https://github.com/utopia-php/cache.git", - "reference": "99e7085eb229d0f0159a4f2107ea5ea123f7b32b" + "reference": "d0cd43b679fa23dc720c3bb9dff4c47c0eb4790d" }, "dist": { "type": "zip", - "url": "https://api.github.com/repos/utopia-php/cache/zipball/99e7085eb229d0f0159a4f2107ea5ea123f7b32b", - "reference": "99e7085eb229d0f0159a4f2107ea5ea123f7b32b", + "url": "https://api.github.com/repos/utopia-php/cache/zipball/d0cd43b679fa23dc720c3bb9dff4c47c0eb4790d", + "reference": "d0cd43b679fa23dc720c3bb9dff4c47c0eb4790d", "shasum": "" }, "require": { @@ -1942,7 +1942,7 @@ "issues": "https://github.com/utopia-php/cache/issues", "source": "https://github.com/utopia-php/cache/tree/feat-redis-sync" }, - "time": "2022-10-18T06:58:42+00:00" + "time": "2022-10-26T08:25:43+00:00" }, { "name": "utopia-php/cli", @@ -2537,16 +2537,16 @@ }, { "name": "utopia-php/queue", - "version": "dev-refactor-redis-client", + "version": "0.3.0", "source": { "type": "git", "url": "https://github.com/utopia-php/queue.git", - "reference": "3e49bcbfc343792ea4899d60b24e7b6b334e5910" + "reference": "42b132c6f2431b726c2bc629c386921e4934b863" }, "dist": { "type": "zip", - "url": "https://api.github.com/repos/utopia-php/queue/zipball/3e49bcbfc343792ea4899d60b24e7b6b334e5910", - "reference": "3e49bcbfc343792ea4899d60b24e7b6b334e5910", + "url": "https://api.github.com/repos/utopia-php/queue/zipball/42b132c6f2431b726c2bc629c386921e4934b863", + "reference": "42b132c6f2431b726c2bc629c386921e4934b863", "shasum": "" }, "require": { @@ -2592,9 +2592,9 @@ ], "support": { "issues": "https://github.com/utopia-php/queue/issues", - "source": "https://github.com/utopia-php/queue/tree/refactor-redis-client" + "source": "https://github.com/utopia-php/queue/tree/0.3.0" }, - "time": "2022-10-24T14:29:47+00:00" + "time": "2022-10-19T13:22:07+00:00" }, { "name": "utopia-php/registry", @@ -5469,19 +5469,12 @@ "version": "dev-feat-update-cache-lib", "alias": "0.26.1", "alias_normalized": "0.26.1.0" - }, - { - "package": "utopia-php/queue", - "version": "dev-refactor-redis-client", - "alias": "0.3.0", - "alias_normalized": "0.3.0.0" } ], "minimum-stability": "stable", "stability-flags": { "utopia-php/cache": 20, - "utopia-php/database": 20, - "utopia-php/queue": 20 + "utopia-php/database": 20 }, "prefer-stable": false, "prefer-lowest": false,