From 70f1a3eaf8f279ff4b4568c134f70c75fb107bec Mon Sep 17 00:00:00 2001 From: shimon Date: Wed, 19 Oct 2022 12:36:45 +0300 Subject: [PATCH] cloud sync maintenance worker --- Dockerfile | 4 +- app/cli.php | 1 + app/config/collections.php | 13 ++- app/config/regions.php | 22 ++++++ app/config/services.php | 4 +- app/controllers/api/{syncs.php => edge.php} | 34 +++++++- app/init.php | 21 +---- app/tasks/maintenance.php | 29 +------ app/tasks/syncsCloud.php | 88 +++++++++++++++++++++ app/workers/deletes.php | 10 --- app/workers/syncsIn.php | 5 -- app/workers/syncsOut.php | 32 ++++---- bin/syncs-cloud | 3 + docker-compose.yml | 25 ++++++ src/Appwrite/Event/Delete.php | 10 --- 15 files changed, 207 insertions(+), 94 deletions(-) create mode 100644 app/config/regions.php rename app/controllers/api/{syncs.php => edge.php} (54%) create mode 100644 app/tasks/syncsCloud.php create mode 100644 bin/syncs-cloud diff --git a/Dockerfile b/Dockerfile index b2345f984e..1adc4d49e3 100755 --- a/Dockerfile +++ b/Dockerfile @@ -331,7 +331,8 @@ RUN mkdir -p /storage/uploads && \ # Executables RUN chmod +x /usr/local/bin/doctor && \ - chmod +x /usr/local/bin/maintenance && \ + chmod +x /usr/local/bin/maintenance && \ + chmod +x /usr/local/bin/syncs-cloud && \ chmod +x /usr/local/bin/usage && \ chmod +x /usr/local/bin/install && \ chmod +x /usr/local/bin/migrate && \ @@ -356,6 +357,7 @@ RUN chmod +x /usr/local/bin/doctor && \ chmod +x /usr/local/bin/worker-syncs-in + # Letsencrypt Permissions RUN mkdir -p /etc/letsencrypt/live/ && chmod -Rf 755 /etc/letsencrypt/live/ diff --git a/app/cli.php b/app/cli.php index 09b3dc9413..6fc8e73d3a 100644 --- a/app/cli.php +++ b/app/cli.php @@ -14,6 +14,7 @@ $cli = new CLI(); include 'tasks/doctor.php'; include 'tasks/maintenance.php'; +include 'tasks/syncsCloud.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 58d00c85aa..5cfb09633a 100644 --- a/app/config/collections.php +++ b/app/config/collections.php @@ -24,9 +24,18 @@ $collections = [ 'name' => 'Syncs', 'attributes' => [ [ - '$id' => ID::custom('region'), + '$id' => ID::custom('regionOrg'), 'type' => Database::VAR_STRING, - 'size' => 256, + 'size' => 50, + 'required' => true, + 'signed' => true, + 'array' => false, + 'filters' => [], + ], + [ + '$id' => ID::custom('regionDest'), + 'type' => Database::VAR_STRING, + 'size' => 50, 'required' => true, 'signed' => true, 'array' => false, diff --git a/app/config/regions.php b/app/config/regions.php new file mode 100644 index 0000000000..75c7b6df56 --- /dev/null +++ b/app/config/regions.php @@ -0,0 +1,22 @@ + [ + 'name' => 'North america', + 'default' => true, + 'disabled' => false, + 'domain' => '172.17.0.1', + ], + 'blr1' => [ + 'name' => 'Asia', + 'default' => true, + 'disabled' => false, + 'domain' => '172.17.0.1', + ], + 'fra1' => [ + 'name' => 'Europe', + 'default' => true, + 'disabled' => false, + 'domain' => '172.17.0.1', + ], +]; diff --git a/app/config/services.php b/app/config/services.php index a1af27aa6e..9f4a5791f7 100644 --- a/app/config/services.php +++ b/app/config/services.php @@ -190,8 +190,8 @@ return [ 'key' => 'syncs', 'name' => 'syncs', 'subtitle' => 'Appwrite\'s cloud regions syncs Endpoint', - 'description' => 'Syncs Endpoint', - 'controller' => 'api/syncs.php', + 'description' => 'Cloud edge Endpoint', + 'controller' => 'api/edge.php', 'sdk' => false, 'docs' => false, 'docsUrl' => '', diff --git a/app/controllers/api/syncs.php b/app/controllers/api/edge.php similarity index 54% rename from app/controllers/api/syncs.php rename to app/controllers/api/edge.php index ccd6f82c8f..61dc5f2916 100644 --- a/app/controllers/api/syncs.php +++ b/app/controllers/api/edge.php @@ -2,6 +2,7 @@ use Ahc\Jwt\JWT; use Ahc\Jwt\JWTException; +use Appwrite\Event\Delete; use Appwrite\Event\SyncIn; use Appwrite\Extend\Exception; use Appwrite\Utopia\Request; @@ -9,8 +10,9 @@ use Appwrite\Utopia\Response; use Utopia\App; use Utopia\Validator\ArrayList; use Utopia\Validator\Text; +use Utopia\Validator\WhiteList; -App::post('/v1/syncs') +App::post('/v1/edge') ->desc('Purge cache keys') ->label('scope', 'public') ->param('keys', '', new ArrayList(new Text(100), 1000), 'Cache keys') @@ -18,9 +20,9 @@ App::post('/v1/syncs') ->inject('response') ->action(function (array $keys, Request $request, Response $response) { - //if (empty($keys)) { + if (empty($keys)) { throw new Exception(Exception::KEY_NOT_FOUND); - //} + } $token = $request->getHeader('authorization'); $token = str_replace(["Bearer"," "], "", $token); @@ -42,3 +44,29 @@ App::post('/v1/syncs') ->setStatusCode(Response::STATUS_CODE_OK) ->send(); }); + +App::post('/v1/edge/notify') + ->desc('Flush notification') + ->label('scope', 'public') + ->param('region', '', new WhiteList(['nyc1', 'blr1', 'fra1']), 'Cloud regions') + ->inject('request') + ->inject('response') + ->action(function (string $region, Request $request, Response $response) { + + $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()); + } + + (new Delete()) + ->setRegion($region) + ->trigger(); + + $response + ->setStatusCode(Response::STATUS_CODE_OK) + ->send(); + }); diff --git a/app/init.php b/app/init.php index efb6554c54..d0763c6295 100644 --- a/app/init.php +++ b/app/init.php @@ -37,7 +37,6 @@ use Appwrite\Event\Delete; use Appwrite\Event\Event; use Appwrite\Event\Mail; use Appwrite\Event\Phone; -use Appwrite\Event\SyncIn; use Appwrite\Network\Validator\Email; use Appwrite\Network\Validator\IP; use Appwrite\Network\Validator\URL; @@ -154,7 +153,6 @@ const DELETE_TYPE_BUCKETS = 'buckets'; const DELETE_TYPE_SESSIONS = 'sessions'; const DELETE_TYPE_CACHE_BY_TIMESTAMP = 'cacheByTimeStamp'; const DELETE_TYPE_CACHE_BY_RESOURCE = 'cacheByResource'; -const DELETE_TYPE_SYNCS = 'syncs'; // Compression type const COMPRESSION_TYPE_NONE = 'none'; const COMPRESSION_TYPE_GZIP = 'gzip'; @@ -191,6 +189,7 @@ Config::load('roles', __DIR__ . '/config/roles.php'); // User roles and scopes Config::load('scopes', __DIR__ . '/config/scopes.php'); // User roles and scopes Config::load('services', __DIR__ . '/config/services.php'); // List of services Config::load('variables', __DIR__ . '/config/variables.php'); // List of env variables +Config::load('regions', __DIR__ . '/config/regions.php'); // List of cloud regions Config::load('avatar-browsers', __DIR__ . '/config/avatars/browsers.php'); Config::load('avatar-credit-cards', __DIR__ . '/config/avatars/credit-cards.php'); Config::load('avatar-flags', __DIR__ . '/config/avatars/flags.php'); @@ -206,6 +205,7 @@ 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)) { @@ -931,10 +931,6 @@ $register->set('syncOut', function () { return new SyncOut(); }); -$register->set('deletes', function () { - return new Delete(); -}); - App::setResource('dbForProject', function ($db, $cache, Document $project, $register) { $cache = new Cache(new RedisCache($cache)); @@ -952,12 +948,6 @@ App::setResource('dbForProject', function ($db, $cache, Document $project, $regi ->trigger(); }); - $cache->on(cache::EVENT_FLUSH, function ($region) use ($register) { - $register - ->get('deletes') - ->setRegion($region) - ->trigger(); - }); $database = new Database(new MariaDB($db), $cache); $database->setDefaultDatabase(App::getEnv('_APP_DB_SCHEMA', 'appwrite')); @@ -984,13 +974,6 @@ App::setResource('dbForConsole', function ($db, $cache, $register) { ->trigger(); }); - $cache->on(cache::EVENT_FLUSH, function ($region) use ($register) { - $register - ->get('deletes') - ->setRegion($region) - ->trigger(); - }); - $database = new Database(new MariaDB($db), $cache); $database->setDefaultDatabase(App::getEnv('_APP_DB_SCHEMA', 'appwrite')); $database->setNamespace('_console'); diff --git a/app/tasks/maintenance.php b/app/tasks/maintenance.php index c60a7f24dd..fd111ca761 100644 --- a/app/tasks/maintenance.php +++ b/app/tasks/maintenance.php @@ -6,7 +6,6 @@ global $register; use Appwrite\Auth\Auth; use Appwrite\Event\Certificate; use Appwrite\Event\Delete; -use Appwrite\Event\SyncOut; use Utopia\App; use Utopia\Cache\Cache; use Utopia\CLI\Console; @@ -140,30 +139,8 @@ $cli ->trigger(); } - function syncRegionalCache($dbForConsole): void - { - $time = DateTime::now(); - $chunks = $dbForConsole->find('syncs', [ - Query::notEqual('status', 200), - Query::limit(300) - ]); - - if (\count($chunks) > 0) { - Console::info("[{$time}] Found " . \count($chunks) . " cache chunks to purge."); - foreach ($chunks as $chunk) { - $keys = $chunk->getAttribute('keys'); -// (new SyncOut()) - // ->setRegion($chunk->getAttribute('region')) -// ->addKey($key) -// ->trigger(); - } - } else { - Console::info("[{$time}] No certificates for renewal."); - } - } - - // # of days in seconds (1 day = 86400s) + // # of days in seconds (1 day = 86400s) $interval = (int) App::getEnv('_APP_MAINTENANCE_INTERVAL', '86400'); $executionLogsRetention = (int) App::getEnv('_APP_MAINTENANCE_RETENTION_EXECUTION', '1209600'); $auditLogRetention = (int) App::getEnv('_APP_MAINTENANCE_RETENTION_AUDIT', '1209600'); @@ -171,9 +148,8 @@ $cli $usageStatsRetention30m = (int) App::getEnv('_APP_MAINTENANCE_RETENTION_USAGE_30M', '129600'); //36 hours $usageStatsRetention1d = (int) App::getEnv('_APP_MAINTENANCE_RETENTION_USAGE_1D', '8640000'); // 100 days $cacheRetention = (int) App::getEnv('_APP_MAINTENANCE_RETENTION_CACHE', '2592000'); // 30 days - $regionalCacheSyncRetention = (int) App::getEnv('_APP_MAINTENANCE_CACHE_SYNC', '300'); // 5 minutes - Console::loop(function () use ($interval, $executionLogsRetention, $abuseLogsRetention, $auditLogRetention, $usageStatsRetention30m, $usageStatsRetention1d, $cacheRetention, $regionalCacheSyncRetention) { + Console::loop(function () use ($interval, $executionLogsRetention, $abuseLogsRetention, $auditLogRetention, $usageStatsRetention30m, $usageStatsRetention1d, $cacheRetention) { $database = getConsoleDB(); $time = DateTime::now(); @@ -187,6 +163,5 @@ $cli notifyDeleteExpiredSessions(); renewCertificates($database); notifyDeleteCache($cacheRetention); - syncRegionalCache($database); }, $interval); }); diff --git a/app/tasks/syncsCloud.php b/app/tasks/syncsCloud.php new file mode 100644 index 0000000000..276e20930f --- /dev/null +++ b/app/tasks/syncsCloud.php @@ -0,0 +1,88 @@ +get('cache'))); + $database = new Database(new MariaDB($register->get('db')), $cache); + $database->setDefaultDatabase(App::getEnv('_APP_DB_SCHEMA', 'appwrite')); + $database->setNamespace('_console'); // Main DB + + if (!$database->exists($database->getDefaultDatabase(), 'certificates')) { + throw new \Exception('Console project not ready'); + } + + break; // leave loop if successful + } catch (\Exception $e) { + Console::warning("Database not ready. Retrying connection ({$attempts})..."); + if ($attempts >= DATABASE_RECONNECT_MAX_ATTEMPTS) { + throw new \Exception('Failed to connect to database: ' . $e->getMessage()); + } + sleep(DATABASE_RECONNECT_SLEEP); + } + } while ($attempts < DATABASE_RECONNECT_MAX_ATTEMPTS); + + return $database; +} + +$cli + ->task('syncsCloud') + ->desc('Schedules cloud sync tasks') + ->action(function () { + Console::title('Syncs cloud V1'); + Console::success(APP_NAME . ' Syncs cloud process v1 has started'); + + function syncRegionalCache($dbForConsole, $regionOrg): void + { + $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) { + $keys = $chunk->getAttribute('keys'); + foreach ($keys['keys'] ?? [] as $key) { + (new SyncOut()) + ->setRegion($chunk->getAttribute('region')) + ->addKey($key) + ->trigger(); + } + $dbForConsole->deleteDocument('syncs', $chunk->getId()); + } + } else { + Console::info("[{$time}] No cache key chunks where found."); + } + } + + $interval = (int) App::getEnv('_APP_SYNCS_CLOUD_INTERVAL', '180'); + + Console::loop(function () use ($interval) { + $database = getConsoleDatabase(); + $time = DateTime::now(); + $currentRegion = App::getEnv('_APP_REGION', 'nyc1'); + Console::info("[{$time}] Notifying workers with cloud tasks every {$interval} seconds"); + syncRegionalCache($database, $currentRegion); + }, $interval); + }); diff --git a/app/workers/deletes.php b/app/workers/deletes.php index 06205a961f..b015043b1d 100644 --- a/app/workers/deletes.php +++ b/app/workers/deletes.php @@ -114,9 +114,6 @@ class DeletesV1 extends Worker case DELETE_TYPE_CACHE_BY_TIMESTAMP: $this->deleteCacheByDate(); break; - case DELETE_TYPE_SYNCS: - $this->deleteRegionalCache(); - break; default: Console::error('No delete operation for type: ' . $type); break; @@ -678,11 +675,4 @@ class DeletesV1 extends Worker $device->deletePath($document->getId()); } - - protected function deleteRegionalCache() - { - $this->deleteByGroup('syncs', [ - Query::equal('region', [$this->args['region']]) - ], $this->getConsoleDB); - } } diff --git a/app/workers/syncsIn.php b/app/workers/syncsIn.php index 77841f6eb5..00c426ff10 100644 --- a/app/workers/syncsIn.php +++ b/app/workers/syncsIn.php @@ -2,7 +2,6 @@ use Appwrite\Resque\Worker; use Utopia\Cache\Adapter\Redis as RedisCache; -use Utopia\Cache\Cache; use Utopia\CLI\Console; require_once __DIR__ . '/../init.php'; @@ -12,8 +11,6 @@ Console::success(APP_NAME . ' syncs in worker v1 has started'); class SyncsInV1 extends Worker { - protected array $errors = []; - public function getName(): string { return "syncs-in"; @@ -26,8 +23,6 @@ class SyncsInV1 extends Worker public function run(): void { if (!empty($this->args['key'])) { - //var_dump('Purging -> ' . $this->args['key'] . ' from Redis cache'); - //$this->getCache()->purge($this->args['key']); $this->getCache()->purge($this->args['key']); } } diff --git a/app/workers/syncsOut.php b/app/workers/syncsOut.php index 2498d9ebf6..a622941768 100644 --- a/app/workers/syncsOut.php +++ b/app/workers/syncsOut.php @@ -2,10 +2,13 @@ use Ahc\Jwt\JWT; use Appwrite\Resque\Worker; +use Appwrite\Utopia\Response; use Utopia\App; use Utopia\CLI\Console; +use Utopia\Config\Config; use Utopia\Database\DateTime; use Utopia\Database\Document; +use Utopia\Response as ResponseAlias; require_once __DIR__ . '/../init.php'; @@ -14,11 +17,7 @@ Console::success(APP_NAME . ' syncs out worker v1 has started'); class SyncsOutV1 extends Worker { - private array $regions = [ - 'fra1' => '172.17.0.1', - 'nyc1' => '172.17.0.1', - 'blr1' => '172.17.0.1', - ]; + private array $regions; public function getName(): string { @@ -27,14 +26,15 @@ class SyncsOutV1 extends Worker public function init(): void { + $this->regions = Config::getParam('regions', []); } public function run(): void { - //TODO current region env implementation - $currentRegion = 'nyc1'; - $data['keys'][] = $this->args['key']; + $currentRegion = App::getEnv('_APP_REGION', 'nyc1'); + + $data[] = $this->args['key']; $jwt = new JWT(App::getEnv('_APP_OPENSSL_KEY_V1'), 'HS256', 600, 10); $token = $jwt->encode($data); @@ -42,16 +42,18 @@ class SyncsOutV1 extends Worker $this->regions = $this->regions[$this->args['region']]; } - foreach ($this->regions as $region => $host) { - if ($currentRegion === $region) { + foreach ($this->regions as $code => $region) { + if ($currentRegion === $code) { continue; } - $status = $this->send($host, $token, $data); - if ($status !== 200) { + $status = $this->send($region['domain'] . '/v1/edge', $token, ['keys' => $data]); + + if ($status !== Response::STATUS_CODE_OK) { $this->getConsoleDB()->createDocument('syncs', new Document([ 'requestedAt' => DateTime::now(), - 'region' => $region, + 'regionOrg' => $currentRegion, + 'regionDest' => $code, 'keys' => $data, 'status' => $status, ])); @@ -59,10 +61,10 @@ class SyncsOutV1 extends Worker } } - private function send($host, $token, $data): int + private function send($url, $token, $data): int { - $ch = curl_init($host . '/v1/syncs'); + $ch = curl_init($url); curl_setopt($ch, CURLOPT_HTTPHEADER, [ 'Authorization: Bearer ' . $token, 'Content-Type: application/json' diff --git a/bin/syncs-cloud b/bin/syncs-cloud new file mode 100644 index 0000000000..a321c87ed9 --- /dev/null +++ b/bin/syncs-cloud @@ -0,0 +1,3 @@ +#!/bin/sh + +php /usr/src/code/app/cli.php syncsCloud $@ \ No newline at end of file diff --git a/docker-compose.yml b/docker-compose.yml index cb34d788cc..9868cbfe3e 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -617,6 +617,31 @@ services: - _APP_MAINTENANCE_RETENTION_ABUSE - _APP_MAINTENANCE_RETENTION_AUDIT + appwrite-syncs-cloud: + entrypoint: syncs-cloud + <<: *x-logging + container_name: appwrite-syncs-cloud + image: appwrite-dev + networks: + - appwrite + volumes: + - ./app:/usr/src/code/app + - ./src:/usr/src/code/src + depends_on: + - mariadb + - redis + environment: + - _APP_ENV + - _APP_REDIS_HOST + - _APP_REDIS_PORT + - _APP_REDIS_USER + - _APP_REDIS_PASS + - _APP_DB_HOST + - _APP_DB_PORT + - _APP_DB_SCHEMA + - _APP_DB_USER + - _APP_DB_PASS + appwrite-usage-timeseries: entrypoint: - usage diff --git a/src/Appwrite/Event/Delete.php b/src/Appwrite/Event/Delete.php index 81de8ee939..c7792f344f 100644 --- a/src/Appwrite/Event/Delete.php +++ b/src/Appwrite/Event/Delete.php @@ -93,16 +93,6 @@ class Delete extends Event return $this; } - /** - * Sets cloud region. - * - * @param string $region - */ - public function setRegion($region): void - { - $this->region = $region; - } - /** * Returns the resource for the delete event. *