Appwrite/appwrite
1
0
Fork 0
mirror of https://github.com/appwrite/appwrite.git synced 2026-05-26 13:51:13 +00:00
Code Issues Packages Projects Releases Wiki Activity

cloud sync maintenance worker

Browse Source
This commit is contained in:
shimon
2022-10-19 12:36:45 +03:00
parent 462b30b721
commit 70f1a3eaf8
15 changed files with 207 additions and 94 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Dockerfile
+3 -1
View File
@@ -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/
app/cli.php
+1
View File
@@ -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';
app/config/collections.php
+11 -2
View File
@@ -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,
app/config/regions.php
+22
View File
@@ -0,0 +1,22 @@
<?php
return [
'nyc1' => [
'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',
],
];
app/config/services.php
+2 -2
View File
@@ -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' => '',
app/controllers/api/syncs.php → app/controllers/api/edge.php
+31 -3
View File
@@ -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();
});
app/init.php
+2 -19
View File
@@ -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');
app/tasks/maintenance.php
+2 -27
View File
@@ -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);
});
app/tasks/syncsCloud.php
+88
View File
@@ -0,0 +1,88 @@
<?php
global $cli;
global $register;
use Appwrite\Event\SyncOut;
use Utopia\App;
use Utopia\Cache\Cache;
use Utopia\CLI\Console;
use Utopia\Database\Adapter\MariaDB;
use Utopia\Database\Database;
use Utopia\Database\DateTime;
use Utopia\Cache\Adapter\Redis as RedisCache;
use Utopia\Database\Query;
function getConsoleDatabase(): Database
{
global $register;
$attempts = 0;
do {
try {
$attempts++;
$cache = new Cache(new RedisCache($register->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);
});
app/workers/deletes.php
-10
View File
@@ -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);
}
}
app/workers/syncsIn.php
-5
View File
@@ -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']);
}
}
app/workers/syncsOut.php
+17 -15
View File
@@ -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'
bin/syncs-cloud
+3
View File
@@ -0,0 +1,3 @@
#!/bin/sh
php /usr/src/code/app/cli.php syncsCloud $@
docker-compose.yml
+25
View File
@@ -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
src/Appwrite/Event/Delete.php
-10
View File
@@ -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.
*