diff --git a/Dockerfile b/Dockerfile index 1adc4d49e3..73593be925 100755 --- a/Dockerfile +++ b/Dockerfile @@ -332,7 +332,7 @@ RUN mkdir -p /storage/uploads && \ # Executables RUN chmod +x /usr/local/bin/doctor && \ chmod +x /usr/local/bin/maintenance && \ - chmod +x /usr/local/bin/syncs-cloud && \ + chmod +x /usr/local/bin/sync-edge && \ chmod +x /usr/local/bin/usage && \ chmod +x /usr/local/bin/install && \ chmod +x /usr/local/bin/migrate && \ @@ -353,8 +353,8 @@ RUN chmod +x /usr/local/bin/doctor && \ chmod +x /usr/local/bin/worker-mails && \ chmod +x /usr/local/bin/worker-messaging && \ chmod +x /usr/local/bin/worker-webhooks && \ - chmod +x /usr/local/bin/worker-syncs-out && \ - chmod +x /usr/local/bin/worker-syncs-in + chmod +x /usr/local/bin/worker-sync-out && \ + chmod +x /usr/local/bin/worker-sync-in diff --git a/app/cli.php b/app/cli.php index 6fc8e73d3a..ec3d5124f9 100644 --- a/app/cli.php +++ b/app/cli.php @@ -14,7 +14,7 @@ $cli = new CLI(); include 'tasks/doctor.php'; include 'tasks/maintenance.php'; -include 'tasks/syncsCloud.php'; +include 'tasks/sync-edge.php'; include 'tasks/install.php'; include 'tasks/migrate.php'; include 'tasks/sdks.php'; diff --git a/app/config/collections.php b/app/config/collections.php index d148e31b9d..e087b323ab 100644 --- a/app/config/collections.php +++ b/app/config/collections.php @@ -24,7 +24,7 @@ $collections = [ 'name' => 'Syncs', 'attributes' => [ [ - '$id' => ID::custom('regionOrg'), + '$id' => ID::custom('region'), 'type' => Database::VAR_STRING, 'size' => 50, 'required' => true, @@ -33,7 +33,7 @@ $collections = [ 'filters' => [], ], [ - '$id' => ID::custom('regionDest'), + '$id' => ID::custom('target'), 'type' => Database::VAR_STRING, 'size' => 50, 'required' => true, @@ -61,6 +61,17 @@ $collections = [ 'array' => false, 'filters' => [], ], + [ + '$id' => ID::custom('payload'), + 'type' => Database::VAR_STRING, + 'format' => '', + 'size' => 16384, + 'signed' => true, + 'required' => true, + 'default' => [], + 'array' => false, + 'filters' => ['json'], + ], ], 'indexes' => [ [ diff --git a/app/config/regions.php b/app/config/regions.php index 75c7b6df56..dc4f3f92d4 100644 --- a/app/config/regions.php +++ b/app/config/regions.php @@ -2,19 +2,19 @@ return [ 'nyc1' => [ - 'name' => 'North america', + 'name' => 'New york', 'default' => true, 'disabled' => false, 'domain' => '172.17.0.1', ], - 'blr1' => [ - 'name' => 'Asia', + 'spg1' => [ + 'name' => 'Singapore', 'default' => true, 'disabled' => false, 'domain' => '172.17.0.1', ], 'fra1' => [ - 'name' => 'Europe', + 'name' => 'Frankfurt', 'default' => true, 'disabled' => false, 'domain' => '172.17.0.1', diff --git a/app/config/services.php b/app/config/services.php index 9f4a5791f7..88a387a213 100644 --- a/app/config/services.php +++ b/app/config/services.php @@ -187,8 +187,8 @@ return [ 'icon' => '', ], 'syncs' => [ - 'key' => 'syncs', - 'name' => 'syncs', + 'key' => 'edge', + 'name' => 'edge', 'subtitle' => 'Appwrite\'s cloud regions syncs Endpoint', 'description' => 'Cloud edge Endpoint', 'controller' => 'api/edge.php', diff --git a/app/controllers/api/edge.php b/app/controllers/api/edge.php index c8c8c13719..3a02ec2c24 100644 --- a/app/controllers/api/edge.php +++ b/app/controllers/api/edge.php @@ -2,7 +2,6 @@ use Ahc\Jwt\JWT; use Ahc\Jwt\JWTException; -use Appwrite\Event\SyncIn; use Appwrite\Extend\Exception; use Appwrite\Utopia\Request; use Appwrite\Utopia\Response; @@ -11,10 +10,26 @@ use Utopia\Registry\Registry; use Utopia\Validator\ArrayList; use Utopia\Validator\Text; use Utopia\Queue\Client; -use Utopia\Queue\Connection\Redis; -App::post('/v1/edge') +App::init() + ->groups(['edge']) + ->inject('request') + ->action(function (Request $request) { + + $token = $request->getHeader('authorization'); + $token = str_replace(["Bearer"," "], "", $token); + $jwt = new JWT(App::getEnv('_APP_OPENSSL_KEY_V1'), 'HS256', 600, 10); + try { + $payload = $jwt->decode($token); + } catch (JWTException $error) { + throw new Exception(Exception::USER_JWT_INVALID, 'Failed to verify JWT. ' . $error->getMessage()); + } + }); + + +App::post('/v1/edge/sync') ->desc('Purge cache keys') + ->groups(['edge']) ->label('scope', 'public') ->param('keys', '', new ArrayList(new Text(100), 1000), 'Cache keys') ->inject('request') @@ -26,29 +41,12 @@ App::post('/v1/edge') throw new Exception(Exception::KEY_NOT_FOUND); } - $token = $request->getHeader('authorization'); - $token = str_replace(["Bearer"," "], "", $token); - $jwt = new JWT(App::getEnv('_APP_OPENSSL_KEY_V1'), 'HS256', 600, 10); - try { - $payload = $jwt->decode($token); - } catch (JWTException $error) { - throw new Exception(Exception::USER_JWT_INVALID, 'Failed to verify JWT. ' . $error->getMessage()); - } - $connection = $register - ->get('workerRedisConnection'); + ->get('queue'); $client = new Client('syncIn', $connection); - $client->resetStats(); - foreach ($keys as $key) { - $client->enqueue([ - 'type' => 'from endpoint', - 'value' => [ - 'key' => $key - ] - ]); - } + $client->enqueue(['value' => ['keys' => $keys]]); $response ->setStatusCode(Response::STATUS_CODE_OK) diff --git a/app/init.php b/app/init.php index 6cf3ef3953..9d5e2fb05e 100644 --- a/app/init.php +++ b/app/init.php @@ -207,7 +207,6 @@ Config::load('storage-mimes', __DIR__ . '/config/storage/mimes.php'); Config::load('storage-inputs', __DIR__ . '/config/storage/inputs.php'); Config::load('storage-outputs', __DIR__ . '/config/storage/outputs.php'); - $user = App::getEnv('_APP_REDIS_USER', ''); $pass = App::getEnv('_APP_REDIS_PASS', ''); if (!empty($user) || !empty($pass)) { @@ -929,45 +928,29 @@ App::setResource('console', function () { ]); }, []); -$register->set('workerRedisConnection', function () { +$register->set('queue', function () { return new redisQueue('redis', 6379); }); -$register->set('workerSyncOut', function () use ($register) { - return new clientQueue('syncOut', $register->get('workerRedisConnection')); +$register->set('syncOut', function () use ($register) { + return new clientQueue('syncOut', $register->get('queue')); }); - App::setResource('dbForProject', function ($db, $cache, Document $project, $register) { $cache = new Cache(new RedisCache($cache)); $cache->on(cache::EVENT_SAVE, function ($key) use ($register) { $register - ->get('workerSyncOut') - ->resetStats(); - $register - ->get('workerSyncOut') - ->enqueue([ - 'type' => 'saved from init', - 'value' => [ - 'key' => $key - ] - ]); + ->get('syncOut') + ->enqueue(['value' => ['key' => $key]]); }); $cache->on(cache::EVENT_PURGE, function ($key) use ($register) { + $register - ->get('workerSyncOut') - ->resetStats(); - $register - ->get('workerSyncOut') - ->enqueue([ - 'type' => 'purge from init', - 'value' => [ - 'key' => $key - ] - ]); + ->get('syncOut') + ->enqueue(['value' => ['key' => $key]]); }); $database = new Database(new MariaDB($db), $cache); @@ -979,33 +962,17 @@ App::setResource('dbForProject', function ($db, $cache, Document $project, $regi App::setResource('dbForConsole', function ($db, $cache, $register) { $cache = new Cache(new RedisCache($cache)); - $cache->on(cache::EVENT_SAVE, function ($key) use ($register) { + $cache->on(cache::EVENT_SAVE, function ($key) use ($register) { $register - ->get('workerSyncOut') - ->resetStats(); - $register - ->get('workerSyncOut') - ->enqueue([ - 'type' => 'saved from init', - 'value' => [ - 'key' => $key - ] - ]); + ->get('syncOut') + ->enqueue(['value' => ['key' => $key]]); }); $cache->on(cache::EVENT_PURGE, function ($key) use ($register) { $register - ->get('workerSyncOut') - ->resetStats(); - $register - ->get('workerSyncOut') - ->enqueue([ - 'type' => 'purge from init', - 'value' => [ - 'key' => $key - ] - ]); + ->get('syncOut') + ->enqueue(['value' => ['key' => $key]]); }); $database = new Database(new MariaDB($db), $cache); diff --git a/app/realtime.php b/app/realtime.php index 2263f3c396..cb733299cd 100644 --- a/app/realtime.php +++ b/app/realtime.php @@ -188,7 +188,6 @@ $server->onStart(function () use ($stats, $register, $containerId, &$statsDocume Authorization::skip(fn () => $database->updateDocument('realtime', $statsDocument->getId(), $statsDocument)); } catch (\Throwable $th) { - call_user_func($logError, $th, "updateWorkerDocument"); } finally { call_user_func($returnDatabase); diff --git a/app/tasks/syncsCloud.php b/app/tasks/sync-edge.php similarity index 55% rename from app/tasks/syncsCloud.php rename to app/tasks/sync-edge.php index 3f1adef031..efce04f225 100644 --- a/app/tasks/syncsCloud.php +++ b/app/tasks/sync-edge.php @@ -44,54 +44,44 @@ function getConsoleDatabase(): Database return $database; } -function resendFailedRequest($dbForConsole, $regionOrg): void -{ - global $register; - - $time = DateTime::now(); - $chunks = $dbForConsole->find('syncs', [ - Query::equal('regionOrg', [$regionOrg]), - Query::limit(500) - ]); - - if (count($chunks) > 0) { - Console::info("[{$time}] Found " . \count($chunks) . " cache key chunks to purge."); - foreach ($chunks as $chunk) { - $register - ->get('workerSyncOut') - ->resetStats(); - - $register - ->get('workerSyncOut') - ->enqueue([ - 'type' => 'from cloud maintenance', - 'value' => [ - 'region' => $chunk->getAttribute('regionDest'), - 'chunk' => $chunk->getAttribute('keys') - ] - ]); - - $dbForConsole->deleteDocument('syncs', $chunk->getId()); - } - } else { - Console::info("[{$time}] No cache key chunks where found."); - } -} - $cli - ->task('syncsCloud') - ->desc('Schedules cloud sync tasks') + ->task('sync-edge') + ->desc('Schedules edge sync tasks') ->action(function () { - Console::title('Syncs cloud V1'); + Console::title('Syncs edges V1'); Console::success(APP_NAME . ' Syncs cloud process v1 has started'); - $interval = (int) App::getEnv('_APP_SYNCS_CLOUD_INTERVAL', '180'); + $interval = (int) App::getEnv('_APP_SYNC_EDGE_INTERVAL', '180'); Console::loop(function () use ($interval) { $database = getConsoleDatabase(); $time = DateTime::now(); - $currentRegion = App::getEnv('_APP_REGION', 'nyc1'); + $region = App::getEnv('_APP_REGION', 'nyc1'); Console::info("[{$time}] Notifying workers with cloud tasks every {$interval} seconds"); - resendFailedRequest($database, $currentRegion); + global $register; + + $time = DateTime::now(); + $chunks = $database->find('syncs', [ + Query::equal('region', [$region]), + Query::limit(500) + ]); + + if (count($chunks) > 0) { + Console::info("[{$time}] Found " . \count($chunks) . " cache key chunks to purge."); + foreach ($chunks as $chunk) { + $register + ->get('syncOut') + ->enqueue([ + 'value' => [ + 'region' => $chunk->getAttribute('regionDest'), + 'chunk' => $chunk->getAttribute('keys') + ] + ]); + + $database->deleteDocument('syncs', $chunk->getId()); + } + } else { + Console::info("[{$time}] No cache key chunks where found."); + } }, $interval); }); diff --git a/app/workers/syncsIn.php b/app/workers/sync-In.php similarity index 69% rename from app/workers/syncsIn.php rename to app/workers/sync-In.php index 40031ca20e..29eca3eb03 100644 --- a/app/workers/syncsIn.php +++ b/app/workers/sync-In.php @@ -9,8 +9,6 @@ use Utopia\Queue\Message; require_once __DIR__ . '/../init.php'; -define("CURRENT_REGION", App::getEnv('_APP_REGION', 'nyc1')); - /** * @return RedisCache */ @@ -26,12 +24,12 @@ $server = new Queue\Server($adapter); $server->job() ->inject('message') - ->action(function (Message $message) use (&$keys, &$counter) { + ->action(function (Message $message) { $payload = $message->getPayload()['value']; - if (!empty($payload['key'])) { - var_dump('purging ' . $payload['key']); - getCache()->purge($payload['key']); + foreach ($payload['keys'] ?? [] as $key) { + var_dump('purging ' . $key); + var_dump(getCache()->purge($key)); } }); @@ -40,10 +38,11 @@ $server ->inject('error') ->action(function ($error) { echo $error->getMessage() . PHP_EOL; + echo $error->getLine() . PHP_EOL; }); $server ->workerStart(function () { - echo "In region [" . CURRENT_REGION . "] cache purging worker Started" . PHP_EOL; + echo "In region [" . App::getEnv('_APP_REGION', 'nyc1') . "] cache purging worker Started" . PHP_EOL; }) ->start(); diff --git a/app/workers/syncsOut.php b/app/workers/sync-out.php similarity index 76% rename from app/workers/syncsOut.php rename to app/workers/sync-out.php index 326862fa31..ea76098281 100644 --- a/app/workers/syncsOut.php +++ b/app/workers/sync-out.php @@ -12,7 +12,6 @@ use Utopia\Config\Config; use Utopia\Database\Adapter\MariaDB; use Utopia\Database\Database; use Utopia\Database\Document; -use Utopia\Database\DateTime; use Utopia\Database\Exception\Authorization; use Utopia\Database\Exception\Structure; use Utopia\Queue; @@ -29,8 +28,6 @@ const SUBMITION_INTERVAL = 20; const MAX_KEY_COUNT = 10; const MAX_CURL_SEND_ATTEMPTS = 4; -define("CURRENT_REGION", App::getEnv('_APP_REGION', 'nyc1')); - /** * Get console database * @param string $type One of (internal, external, console) @@ -92,7 +89,13 @@ function getDB(string $type, string $projectId = '', string $projectInternalId = return $database; } -function send($url, $token, $keys): int +/** + * @param string $url + * @param string $token + * @param array $keys + * @return array\ + */ +function send(string $url, string $token, array $keys): array { $ch = curl_init($url); @@ -106,17 +109,23 @@ function send($url, $token, $keys): int curl_setopt($ch, CURLOPT_POSTFIELDS, json_encode($keys)); for ($attempts = 0; $attempts < MAX_CURL_SEND_ATTEMPTS; $attempts++) { - curl_exec($ch); - $responseStatus = curl_getinfo($ch, CURLINFO_HTTP_CODE); + $response = curl_exec($ch); + $status = curl_getinfo($ch, CURLINFO_HTTP_CODE); + $payload = [ + 'status' => $status, + 'payload' => json_decode($response, true) + ]; - if ($responseStatus === 200) { - return $responseStatus; + if ($status === 200) { + return $payload; } sleep(2); } + curl_close($ch); - return $responseStatus; + + return $payload; } /** @@ -131,13 +140,16 @@ function call($regions, $keys): void $token = $jwt->encode([]); foreach ($regions as $code => $region) { - $status = send($region['domain'] . '/v1/edge', $token, ['keys' => $keys]); - if ($status !== Response::STATUS_CODE_OK) { + var_dump('Sending request to ' . $code . '...............'); + $payload = send($region['domain'] . '/v1/edge/sync', $token, ['keys' => $keys]); + var_dump($payload); + if ($payload['status'] !== Response::STATUS_CODE_OK) { getDB(DATABASE_CONSOLE)->createDocument('syncs', new Document([ - 'regionOrg' => CURRENT_REGION, - 'regionDest' => $code, + 'region' => App::getEnv('_APP_REGION', 'nyc1'), + 'target' => $code, 'keys' => $keys, - 'status' => $status, + 'status' => $payload['status'], + 'payload' => $payload['payload'], ])); } } @@ -155,7 +167,7 @@ $server->job() $regions = Config::getParam('regions', true); $regions = array_filter( $regions, - fn ($region) => CURRENT_REGION !== $region, + fn ($region) => App::getEnv('_APP_REGION', 'nyc1') !== $region, ARRAY_FILTER_USE_KEY ); @@ -167,14 +179,24 @@ $server->job() ); } - if (!empty($payload['chunk'])) { + var_dump('from chunk'); call($regions, $payload['chunk']); return; } $keys[$payload['key']] = null; if (count($keys) >= MAX_KEY_COUNT || ($counter + SUBMITION_INTERVAL) < time()) { + var_dump('From key'); + var_dump([ + 'regions' => array_keys($regions), + 'because_time' => ($counter + SUBMITION_INTERVAL) < time(), + 'because_count' => count($keys) >= MAX_KEY_COUNT, + 'count' => count($keys), + 'counter' => $counter + SUBMITION_INTERVAL, + 'time' => time(), + 'keys' => array_keys($keys), + ]); call($regions, array_keys($keys)); $counter = time(); $keys = []; @@ -191,6 +213,6 @@ $server $server ->workerStart(function () { - echo "Out region [" . CURRENT_REGION . "] cache purging worker Started" . PHP_EOL; + echo "Out region [" . App::getEnv('_APP_REGION', 'nyc1') . "] cache purging worker Started" . PHP_EOL; }) ->start(); diff --git a/bin/sync-edge b/bin/sync-edge new file mode 100644 index 0000000000..5ff317eff5 --- /dev/null +++ b/bin/sync-edge @@ -0,0 +1,3 @@ +#!/bin/sh + +php /usr/src/code/app/cli.php sync-edge $@ \ No newline at end of file diff --git a/bin/syncs-cloud b/bin/syncs-cloud deleted file mode 100644 index a321c87ed9..0000000000 --- a/bin/syncs-cloud +++ /dev/null @@ -1,3 +0,0 @@ -#!/bin/sh - -php /usr/src/code/app/cli.php syncsCloud $@ \ No newline at end of file diff --git a/bin/worker-sync-in b/bin/worker-sync-in new file mode 100644 index 0000000000..0f38627454 --- /dev/null +++ b/bin/worker-sync-in @@ -0,0 +1,3 @@ +#!/bin/sh + +php /usr/src/code/app/workers/sync-in.php $@ \ No newline at end of file diff --git a/bin/worker-sync-out b/bin/worker-sync-out new file mode 100644 index 0000000000..4c076fd25c --- /dev/null +++ b/bin/worker-sync-out @@ -0,0 +1,3 @@ +#!/bin/sh + +php /usr/src/code/app/workers/sync-out.php $@ \ No newline at end of file diff --git a/bin/worker-syncs-in b/bin/worker-syncs-in deleted file mode 100644 index 6b973c44fe..0000000000 --- a/bin/worker-syncs-in +++ /dev/null @@ -1,3 +0,0 @@ -#!/bin/sh - -php /usr/src/code/app/workers/syncsIn.php $@ \ No newline at end of file diff --git a/bin/worker-syncs-out b/bin/worker-syncs-out deleted file mode 100644 index 8117e84b34..0000000000 --- a/bin/worker-syncs-out +++ /dev/null @@ -1,3 +0,0 @@ -#!/bin/sh - -php /usr/src/code/app/workers/syncsOut.php $@ \ No newline at end of file diff --git a/docker-compose.yml b/docker-compose.yml index 9868cbfe3e..7a77fd5174 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -108,7 +108,6 @@ services: - ./public:/usr/src/code/public - ./src:/usr/src/code/src - ./dev:/usr/local/dev - - ./vendor/utopia-php/cache:/usr/src/code/vendor/utopia-php/cache depends_on: - mariadb - redis @@ -207,7 +206,6 @@ services: volumes: - ./app:/usr/src/code/app - ./src:/usr/src/code/src - - ./vendor/utopia-php/cache:/usr/src/code/vendor/utopia-php/cache depends_on: - mariadb - redis @@ -255,10 +253,10 @@ services: - _APP_LOGGING_PROVIDER - _APP_LOGGING_CONFIG - appwrite-worker-syncs-out: - entrypoint: worker-syncs-out + appwrite-worker-sync-out: + entrypoint: worker-sync-out <<: *x-logging - container_name: appwrite-worker-syncs-out + container_name: appwrite-worker-sync-out image: appwrite-dev build: context: . @@ -282,10 +280,10 @@ services: - _APP_DB_USER - _APP_DB_PASS - appwrite-worker-syncs-in: - entrypoint: worker-syncs-in + appwrite-worker-sync-in: + entrypoint: worker-sync-in <<: *x-logging - container_name: appwrite-worker-syncs-in + container_name: appwrite-worker-sync-in image: appwrite-dev build: context: . @@ -617,10 +615,10 @@ services: - _APP_MAINTENANCE_RETENTION_ABUSE - _APP_MAINTENANCE_RETENTION_AUDIT - appwrite-syncs-cloud: - entrypoint: syncs-cloud + appwrite-sync-edge: + entrypoint: sync-edge <<: *x-logging - container_name: appwrite-syncs-cloud + container_name: appwrite-sync-edge image: appwrite-dev networks: - appwrite @@ -641,6 +639,7 @@ services: - _APP_DB_SCHEMA - _APP_DB_USER - _APP_DB_PASS + - _APP_SYNC_EDGE_INTERVAL appwrite-usage-timeseries: entrypoint: