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

Merge branch '1.7.x' into feat-sites

Browse Source
This commit is contained in:
Matej Bačo
2025-03-13 11:41:28 +01:00
parent 74c050ffb6 358e096301
commit 725dff7f0f
137 changed files with 1326 additions and 2802 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/Appwrite/Event/Realtime.php
+46 -11
View File
@@ -7,10 +7,17 @@ use Utopia\Database\Document;
class Realtime extends Event
{
protected array $subscribers = [];
public function __construct()
{
}
/**
* Get Realtime payload for this event.
*
* @return array
*/
public function getRealtimePayload(): array
{
$payload = [];
@@ -24,6 +31,28 @@ class Realtime extends Event
return $payload;
}
/**
* Set subscribers for this realtime event.
*
* @param array $subscribers
* @return array
*/
public function setSubscribers(array $subscribers): self
{
$this->subscribers = $subscribers;
return $this;
}
/**
* Get subscribers for this realtime event.
*
* @return array
*/
public function getSubscribers(): array
{
return $this->subscribers;
}
/**
* Execute Event.
*
@@ -53,17 +82,23 @@ class Realtime extends Event
bucket: $bucket,
);
RealtimeAdapter::send(
projectId: $target['projectId'] ?? $this->getProject()->getId(),
payload: $this->getRealtimePayload(),
events: $allEvents,
channels: $target['channels'],
roles: $target['roles'],
options: [
'permissionsChanged' => $target['permissionsChanged'],
'userId' => $this->getParam('userId')
]
);
$projectIds = !empty($this->getSubscribers())
? $this->getSubscribers()
: [$target['projectId'] ?? $this->getProject()->getId()];
foreach ($projectIds as $projectId) {
RealtimeAdapter::send(
projectId: $projectId,
payload: $this->getRealtimePayload(),
events: $allEvents,
channels: $target['channels'],
roles: $target['roles'],
options: [
'permissionsChanged' => $target['permissionsChanged'],
'userId' => $this->getParam('userId')
]
);
}
return true;
}
src/Appwrite/Event/Usage.php
-78
View File
@@ -1,78 +0,0 @@
<?php
namespace Appwrite\Event;
use Utopia\Database\Document;
use Utopia\Queue\Publisher;
class Usage extends Event
{
protected array $metrics = [];
protected array $reduce = [];
public function __construct(protected Publisher $publisher)
{
parent::__construct($publisher);
$this
->setQueue(Event::USAGE_QUEUE_NAME)
->setClass(Event::USAGE_CLASS_NAME);
}
/**
* Add reduce.
*
* @param Document $document
* @return self
*/
public function addReduce(Document $document): self
{
$this->reduce[] = $document;
return $this;
}
/**
* Add metric.
*
* @param string $key
* @param int $value
* @return self
*/
public function addMetric(string $key, int $value): self
{
$this->metrics[] = [
'key' => $key,
'value' => $value,
];
return $this;
}
/**
* Prepare the payload for the usage event.
*
* @return array
*/
protected function preparePayload(): array
{
return [
'project' => $this->project,
'reduce' => $this->reduce,
'metrics' => $this->metrics,
];
}
/**
* Sends metrics to the usage worker.
*
* @return string|bool
*/
public function trigger(): string|bool
{
parent::trigger();
$this->metrics = [];
return true;
}
}
src/Appwrite/Event/UsageDump.php
-44
View File
@@ -1,44 +0,0 @@
<?php
namespace Appwrite\Event;
use Utopia\Queue\Publisher;
class UsageDump extends Event
{
protected array $stats;
public function __construct(protected Publisher $publisher)
{
parent::__construct($publisher);
$this
->setQueue(Event::USAGE_DUMP_QUEUE_NAME)
->setClass(Event::USAGE_DUMP_CLASS_NAME);
}
/**
* Add Stats.
*
* @param array $stats
* @return self
*/
public function setStats(array $stats): self
{
$this->stats = $stats;
return $this;
}
/**
* Prepare the payload for the usage dump event.
*
* @return array
*/
protected function preparePayload(): array
{
return [
'stats' => $this->stats,
];
}
}
src/Appwrite/Platform/Services/Workers.php
-8
View File
@@ -13,10 +13,6 @@ use Appwrite\Platform\Workers\Migrations;
use Appwrite\Platform\Workers\StatsResources;
use Appwrite\Platform\Workers\StatsUsage;
use Appwrite\Platform\Workers\StatsUsageDump;
/** remove */
use Appwrite\Platform\Workers\Usage;
use Appwrite\Platform\Workers\UsageDump;
/** /remove */
use Appwrite\Platform\Workers\Webhooks;
use Utopia\Platform\Service;
@@ -38,10 +34,6 @@ class Workers extends Service
->addAction(StatsUsage::getName(), new StatsUsage())
->addAction(Migrations::getName(), new Migrations())
->addAction(StatsResources::getName(), new StatsResources())
/** Remove */
->addAction(UsageDump::getName(), new UsageDump())
->addAction(Usage::getName(), new Usage())
/** /remove */
;
}
}
src/Appwrite/Platform/Tasks/Maintenance.php
+9 -3
View File
@@ -24,12 +24,13 @@ class Maintenance extends Action
$this
->desc('Schedules maintenance tasks and publishes them to our queues')
->inject('dbForPlatform')
->inject('console')
->inject('queueForCertificates')
->inject('queueForDeletes')
->callback([$this, 'action']);
}
public function action(Database $dbForPlatform, Certificate $queueForCertificates, Delete $queueForDeletes): void
public function action(Database $dbForPlatform, Document $console, Certificate $queueForCertificates, Delete $queueForDeletes): void
{
Console::title('Maintenance V1');
Console::success(APP_NAME . ' maintenance process v1 has started');
@@ -41,7 +42,7 @@ class Maintenance extends Action
$cacheRetention = (int) System::getEnv('_APP_MAINTENANCE_RETENTION_CACHE', '2592000'); // 30 days
$schedulesDeletionRetention = (int) System::getEnv('_APP_MAINTENANCE_RETENTION_SCHEDULES', '86400'); // 1 Day
Console::loop(function () use ($interval, $cacheRetention, $schedulesDeletionRetention, $usageStatsRetentionHourly, $dbForPlatform, $queueForDeletes, $queueForCertificates) {
Console::loop(function () use ($interval, $cacheRetention, $schedulesDeletionRetention, $usageStatsRetentionHourly, $dbForPlatform, $console, $queueForDeletes, $queueForCertificates) {
$time = DateTime::now();
Console::info("[{$time}] Notifying workers with maintenance tasks every {$interval} seconds");
@@ -52,9 +53,14 @@ class Maintenance extends Action
->setProject($project)
->setUsageRetentionHourlyDateTime(DateTime::addSeconds(new \DateTime(), -1 * $usageStatsRetentionHourly))
->trigger();
});
$queueForDeletes
->setType(DELETE_TYPE_MAINTENANCE)
->setProject($console)
->setUsageRetentionHourlyDateTime(DateTime::addSeconds(new \DateTime(), -1 * $usageStatsRetentionHourly))
->trigger();
$this->notifyDeleteConnections($queueForDeletes);
$this->renewCertificates($dbForPlatform, $queueForCertificates);
$this->notifyDeleteCache($cacheRetention, $queueForDeletes);
src/Appwrite/Platform/Workers/Audits.php
+37 -21
View File
@@ -7,8 +7,6 @@ use Exception;
use Throwable;
use Utopia\Audit\Audit;
use Utopia\CLI\Console;
use Utopia\Database\Database;
use Utopia\Database\DateTime;
use Utopia\Database\Document;
use Utopia\Database\Exception\Authorization;
use Utopia\Database\Exception\Structure;
@@ -18,15 +16,16 @@ use Utopia\System\System;
class Audits extends Action
{
private const BATCH_SIZE_DEVELOPMENT = 1; // smaller batch size for development
private const BATCH_SIZE_PRODUCTION = 5_000;
private const BATCH_AGGREGATION_INTERVAL = 60; // in seconds
protected const BATCH_SIZE_DEVELOPMENT = 1; // smaller batch size for development
protected const BATCH_SIZE_PRODUCTION = 5_000;
protected const BATCH_AGGREGATION_INTERVAL = 60; // in seconds
private int $lastTriggeredTime = 0;
private array $logs = [];
private function getBatchSize(): int
protected function getBatchSize(): int
{
return System::getEnv('_APP_ENV', 'development') === 'development'
? self::BATCH_SIZE_DEVELOPMENT
@@ -46,7 +45,8 @@ class Audits extends Action
$this
->desc('Audits worker')
->inject('message')
->inject('dbForProject')
->inject('getProjectDB')
->inject('project')
->callback([$this, 'action']);
$this->lastTriggeredTime = time();
@@ -55,14 +55,15 @@ class Audits extends Action
/**
* @param Message $message
* @param Database $dbForProject
* @param callable $getProjectDB
* @param Document $project
* @return void
* @throws Throwable
* @throws \Utopia\Database\Exception
* @throws Authorization
* @throws Structure
*/
public function action(Message $message, Database $dbForProject): void
public function action(Message $message, callable $getProjectDB, Document $project): void
{
$payload = $message->getPayload() ?? [];
@@ -100,30 +101,45 @@ class Audits extends Action
'mode' => $mode,
'data' => $auditPayload,
],
'timestamp' => DateTime::formatTz(DateTime::now())
'timestamp' => date("Y-m-d H:i:s", $message->getTimestamp()),
];
$this->logs[] = $eventData;
if (isset($this->logs[$project->getInternalId()])) {
$this->logs[$project->getInternalId()]['logs'][] = $eventData;
} else {
$this->logs[$project->getInternalId()] = [
'project' => new Document([
'$id' => $project->getId(),
'$internalId' => $project->getInternalId(),
'database' => $project->getAttribute('database'),
]),
'logs' => [$eventData]
];
}
// Check if we should process the batch by checking both for the batch size and the elapsed time
$batchSize = $this->getBatchSize();
$shouldProcessBatch = count($this->logs) >= $batchSize;
if (!$shouldProcessBatch && count($this->logs) > 0) {
$shouldProcessBatch = (time() - $this->lastTriggeredTime) >= self::BATCH_AGGREGATION_INTERVAL;
$shouldProcessBatch = \count($this->logs) >= $batchSize;
if (!$shouldProcessBatch && \count($this->logs) > 0) {
$shouldProcessBatch = (\time() - $this->lastTriggeredTime) >= self::BATCH_AGGREGATION_INTERVAL;
}
if ($shouldProcessBatch) {
Console::log('Processing batch with ' . count($this->logs) . ' events');
$audit = new Audit($dbForProject);
try {
$audit->logBatch($this->logs);
Console::success('Audit logs processed successfully');
foreach ($this->logs as $internalId => $projectLogs) {
$dbForProject = $getProjectDB($projectLogs['project']);
Console::log('Processing batch with ' . count($projectLogs['logs']) . ' events');
$audit = new Audit($dbForProject);
$audit->logBatch($projectLogs['logs']);
Console::success('Audit logs processed successfully');
unset($this->logs[$internalId]);
}
} catch (Throwable $e) {
Console::error('Error processing audit logs: ' . $e->getMessage());
} finally {
// Clear the pending events after successful batch processing
$this->logs = [];
$this->lastTriggeredTime = time();
}
}
src/Appwrite/Platform/Workers/Certificates.php
+26 -41
View File
@@ -6,7 +6,8 @@ use Appwrite\Certificates\Adapter as CertificatesAdapter;
use Appwrite\Event\Event;
use Appwrite\Event\Func;
use Appwrite\Event\Mail;
use Appwrite\Messaging\Adapter\Realtime;
use Appwrite\Event\Realtime;
use Appwrite\Event\Webhook;
use Appwrite\Network\Validator\CNAME;
use Appwrite\Template\Template;
use Appwrite\Utopia\Response\Model\Rule;
@@ -46,7 +47,9 @@ class Certificates extends Action
->inject('dbForPlatform')
->inject('queueForMails')
->inject('queueForEvents')
->inject('queueForWebhooks')
->inject('queueForFunctions')
->inject('queueForRealtime')
->inject('log')
->inject('certificates')
->callback([$this, 'action']);
@@ -57,14 +60,16 @@ class Certificates extends Action
* @param Database $dbForPlatform
* @param Mail $queueForMails
* @param Event $queueForEvents
* @param Webhook $queueForWebhooks
* @param Func $queueForFunctions
* @param Realtime $queueForRealtime
* @param Log $log
* @param CertificatesAdapter $certificates
* @return void
* @throws Throwable
* @throws \Utopia\Database\Exception
*/
public function action(Message $message, Database $dbForPlatform, Mail $queueForMails, Event $queueForEvents, Func $queueForFunctions, Log $log, CertificatesAdapter $certificates): void
public function action(Message $message, Database $dbForPlatform, Mail $queueForMails, Event $queueForEvents, Webhook $queueForWebhooks, Func $queueForFunctions, Realtime $queueForRealtime, Log $log, CertificatesAdapter $certificates): void
{
$payload = $message->getPayload() ?? [];
@@ -78,7 +83,7 @@ class Certificates extends Action
$log->addTag('domain', $domain->get());
$this->execute($domain, $dbForPlatform, $queueForMails, $queueForEvents, $queueForFunctions, $log, $certificates, $skipRenewCheck);
$this->execute($domain, $dbForPlatform, $queueForMails, $queueForEvents, $queueForWebhooks, $queueForFunctions, $queueForRealtime, $log, $certificates, $skipRenewCheck);
}
/**
@@ -87,13 +92,14 @@ class Certificates extends Action
* @param Mail $queueForMails
* @param Event $queueForEvents
* @param Func $queueForFunctions
* @param Realtime $queueForRealtime
* @param CertificatesAdapter $certificates
* @param bool $skipRenewCheck
* @return void
* @throws Throwable
* @throws \Utopia\Database\Exception
*/
private function execute(Domain $domain, Database $dbForPlatform, Mail $queueForMails, Event $queueForEvents, Func $queueForFunctions, Log $log, CertificatesAdapter $certificates, bool $skipRenewCheck = false): void
private function execute(Domain $domain, Database $dbForPlatform, Mail $queueForMails, Event $queueForEvents, Webhook $queueForWebhooks, Func $queueForFunctions, Realtime $queueForRealtime, Log $log, CertificatesAdapter $certificates, bool $skipRenewCheck = false): void
{
/**
* 1. Read arguments and validate domain
@@ -183,7 +189,7 @@ class Certificates extends Action
$certificate->setAttribute('updated', DateTime::now());
// Save all changes we made to certificate document into database
$this->saveCertificateDocument($domain->get(), $certificate, $success, $dbForPlatform, $queueForEvents, $queueForFunctions);
$this->saveCertificateDocument($domain->get(), $certificate, $success, $dbForPlatform, $queueForEvents, $queueForWebhooks, $queueForFunctions, $queueForRealtime);
}
}
@@ -196,13 +202,14 @@ class Certificates extends Action
* @param Database $dbForPlatform Database connection for console
* @param Event $queueForEvents
* @param Func $queueForFunctions
* @param Realtime $queueForRealtime
* @return void
* @throws \Utopia\Database\Exception
* @throws Authorization
* @throws Conflict
* @throws Structure
*/
private function saveCertificateDocument(string $domain, Document $certificate, bool $success, Database $dbForPlatform, Event $queueForEvents, Func $queueForFunctions): void
private function saveCertificateDocument(string $domain, Document $certificate, bool $success, Database $dbForPlatform, Event $queueForEvents, Webhook $queueForWebhooks, Func $queueForFunctions, Realtime $queueForRealtime): void
{
// Check if update or insert required
$certificateDocument = $dbForPlatform->findOne('certificates', [Query::equal('domain', [$domain])]);
@@ -216,7 +223,7 @@ class Certificates extends Action
}
$certificateId = $certificate->getId();
$this->updateDomainDocuments($certificateId, $domain, $success, $dbForPlatform, $queueForEvents, $queueForFunctions);
$this->updateDomainDocuments($certificateId, $domain, $success, $dbForPlatform, $queueForEvents, $queueForWebhooks, $queueForFunctions, $queueForRealtime);
}
/**
@@ -335,7 +342,7 @@ class Certificates extends Action
*
* @return void
*/
private function updateDomainDocuments(string $certificateId, string $domain, bool $success, Database $dbForPlatform, Event $queueForEvents, Func $queueForFunctions): void
private function updateDomainDocuments(string $certificateId, string $domain, bool $success, Database $dbForPlatform, Event $queueForEvents, Webhook $queueForWebhooks, Func $queueForFunctions, Realtime $queueForRealtime): void
{
// TODO: @christyjacob remove once we migrate the rules in 1.7.x
if (System::getEnv('_APP_RULES_FORMAT') === 'md5') {
@@ -364,50 +371,28 @@ class Certificates extends Action
return;
}
/** Trigger Webhook */
$ruleModel = new Rule();
$queueForEvents
->setQueue(Event::WEBHOOK_QUEUE_NAME)
->setClass(Event::WEBHOOK_CLASS_NAME)
->setProject($project)
->setEvent('rules.[ruleId].update')
->setParam('ruleId', $rule->getId())
->setPayload($rule->getArrayCopy(array_keys($ruleModel->getRules())))
->trigger();
->setPayload($rule->getArrayCopy(array_keys($ruleModel->getRules())));
/** Trigger Webhook */
$queueForWebhooks
->from($queueForEvents)
->trigger();
/** Trigger Functions */
$queueForFunctions
->setProject($project)
->setEvent('rules.[ruleId].update')
->setParam('ruleId', $rule->getId())
->setPayload($rule->getArrayCopy(array_keys($ruleModel->getRules())))
->from($queueForEvents)
->trigger();
/** Trigger realtime event */
$allEvents = Event::generateEvents('rules.[ruleId].update', [
'ruleId' => $rule->getId(),
]);
$target = Realtime::fromPayload(
// Pass first, most verbose event pattern
event: $allEvents[0],
payload: $rule,
project: $project
);
Realtime::send(
projectId: 'console',
payload: $rule->getArrayCopy(),
events: $allEvents,
channels: $target['channels'],
roles: $target['roles']
);
Realtime::send(
projectId: $project->getId(),
payload: $rule->getArrayCopy(),
events: $allEvents,
channels: $target['channels'],
roles: $target['roles']
);
/** Trigger Realtime Events */
$queueForRealtime
->from($queueForEvents)
->setSubscribers(['console', $projectId])
->trigger();
}
}
}
src/Appwrite/Platform/Workers/Databases.php
+64 -70
View File
@@ -2,8 +2,7 @@
namespace Appwrite\Platform\Workers;
use Appwrite\Event\Event;
use Appwrite\Messaging\Adapter\Realtime;
use Appwrite\Event\Realtime;
use Exception;
use Utopia\CLI\Console;
use Utopia\Database\Database;
@@ -37,6 +36,7 @@ class Databases extends Action
->inject('project')
->inject('dbForPlatform')
->inject('dbForProject')
->inject('queueForRealtime')
->inject('log')
->callback([$this, 'action']);
}
@@ -46,16 +46,17 @@ class Databases extends Action
* @param Document $project
* @param Database $dbForPlatform
* @param Database $dbForProject
* @param Realtime $queueForRealtime
* @param Log $log
* @return void
* @throws \Exception
*/
public function action(Message $message, Document $project, Database $dbForPlatform, Database $dbForProject, Log $log): void
public function action(Message $message, Document $project, Database $dbForPlatform, Database $dbForProject, Realtime $queueForRealtime, Log $log): void
{
$payload = $message->getPayload() ?? [];
if (empty($payload)) {
throw new \Exception('Missing payload');
throw new Exception('Missing payload');
}
$type = $payload['type'];
@@ -75,11 +76,11 @@ class Databases extends Action
match (\strval($type)) {
DATABASE_TYPE_DELETE_DATABASE => $this->deleteDatabase($database, $project, $dbForProject),
DATABASE_TYPE_DELETE_COLLECTION => $this->deleteCollection($database, $collection, $project, $dbForProject),
DATABASE_TYPE_CREATE_ATTRIBUTE => $this->createAttribute($database, $collection, $document, $project, $dbForPlatform, $dbForProject),
DATABASE_TYPE_DELETE_ATTRIBUTE => $this->deleteAttribute($database, $collection, $document, $project, $dbForPlatform, $dbForProject),
DATABASE_TYPE_CREATE_INDEX => $this->createIndex($database, $collection, $document, $project, $dbForPlatform, $dbForProject),
DATABASE_TYPE_DELETE_INDEX => $this->deleteIndex($database, $collection, $document, $project, $dbForPlatform, $dbForProject),
default => throw new \Exception('No database operation for type: ' . \strval($type)),
DATABASE_TYPE_CREATE_ATTRIBUTE => $this->createAttribute($database, $collection, $document, $project, $dbForPlatform, $dbForProject, $queueForRealtime),
DATABASE_TYPE_DELETE_ATTRIBUTE => $this->deleteAttribute($database, $collection, $document, $project, $dbForPlatform, $dbForProject, $queueForRealtime),
DATABASE_TYPE_CREATE_INDEX => $this->createIndex($database, $collection, $document, $project, $dbForPlatform, $dbForProject, $queueForRealtime),
DATABASE_TYPE_DELETE_INDEX => $this->deleteIndex($database, $collection, $document, $project, $dbForPlatform, $dbForProject, $queueForRealtime),
default => throw new Exception('No database operation for type: ' . \strval($type)),
};
}
@@ -90,13 +91,14 @@ class Databases extends Action
* @param Document $project
* @param Database $dbForPlatform
* @param Database $dbForProject
* @param Realtime $queueForRealtime
* @return void
* @throws Authorization
* @throws Conflict
* @throws \Exception
* @throws \Throwable
*/
private function createAttribute(Document $database, Document $collection, Document $attribute, Document $project, Database $dbForPlatform, Database $dbForProject): void
private function createAttribute(Document $database, Document $collection, Document $attribute, Document $project, Database $dbForPlatform, Database $dbForProject, Realtime $queueForRealtime): void
{
if ($collection->isEmpty()) {
throw new Exception('Missing collection');
@@ -106,12 +108,7 @@ class Databases extends Action
}
$projectId = $project->getId();
$events = Event::generateEvents('databases.[databaseId].collections.[collectionId].attributes.[attributeId].update', [
'databaseId' => $database->getId(),
'collectionId' => $collection->getId(),
'attributeId' => $attribute->getId()
]);
$event = "databases.[databaseId].collections.[collectionId].attributes.[attributeId].update";
/**
* TODO @christyjacob4 verify if this is still the case
* Fetch attribute from the database, since with Resque float values are loosing informations.
@@ -169,7 +166,7 @@ class Databases extends Action
break;
default:
if (!$dbForProject->createAttribute('database_' . $database->getInternalId() . '_collection_' . $collection->getInternalId(), $key, $type, $size, $required, $default, $signed, $array, $format, $formatOptions, $filters)) {
throw new \Exception('Failed to create Attribute');
throw new Exception('Failed to create Attribute');
}
}
@@ -200,7 +197,7 @@ class Databases extends Action
throw $e;
} finally {
$this->trigger($database, $collection, $attribute, $project, $projectId, $events);
$this->trigger($database, $collection, $project, $event, $queueForRealtime, $attribute);
if (! $relatedCollection->isEmpty()) {
$dbForProject->purgeCachedDocument('database_' . $database->getInternalId(), $relatedCollection->getId());
@@ -217,13 +214,14 @@ class Databases extends Action
* @param Document $project
* @param Database $dbForPlatform
* @param Database $dbForProject
* @param Realtime $queueForRealtime
* @return void
* @throws Authorization
* @throws Conflict
* @throws \Exception
* @throws \Throwable
**/
private function deleteAttribute(Document $database, Document $collection, Document $attribute, Document $project, Database $dbForPlatform, Database $dbForProject): void
private function deleteAttribute(Document $database, Document $collection, Document $attribute, Document $project, Database $dbForPlatform, Database $dbForProject, Realtime $queueForRealtime): void
{
if ($collection->isEmpty()) {
throw new Exception('Missing collection');
@@ -233,15 +231,9 @@ class Databases extends Action
}
$projectId = $project->getId();
$events = Event::generateEvents('databases.[databaseId].collections.[collectionId].attributes.[attributeId].delete', [
'databaseId' => $database->getId(),
'collectionId' => $collection->getId(),
'attributeId' => $attribute->getId()
]);
$event = 'databases.[databaseId].collections.[collectionId].attributes.[attributeId].delete';
$collectionId = $collection->getId();
$key = $attribute->getAttribute('key', '');
$status = $attribute->getAttribute('status', '');
$type = $attribute->getAttribute('type', '');
$project = $dbForPlatform->getDocument('projects', $projectId);
$options = $attribute->getAttribute('options', []);
@@ -312,7 +304,7 @@ class Databases extends Action
throw $e;
} finally {
$this->trigger($database, $collection, $attribute, $project, $projectId, $events);
$this->trigger($database, $collection, $project, $event, $queueForRealtime, $attribute);
}
// The underlying database removes/rebuilds indexes when attribute is removed
@@ -358,7 +350,7 @@ class Databases extends Action
}
if ($exists) { // Delete the duplicate if created, else update in db
$this->deleteIndex($database, $collection, $index, $project, $dbForPlatform, $dbForProject);
$this->deleteIndex($database, $collection, $index, $project, $dbForPlatform, $dbForProject, $queueForRealtime);
} else {
$dbForProject->updateDocument('indexes', $index->getId(), $index);
}
@@ -381,6 +373,7 @@ class Databases extends Action
* @param Document $project
* @param Database $dbForPlatform
* @param Database $dbForProject
* @param Realtime $queueForRealtime
* @return void
* @throws Authorization
* @throws Conflict
@@ -388,7 +381,7 @@ class Databases extends Action
* @throws DatabaseException
* @throws \Throwable
*/
private function createIndex(Document $database, Document $collection, Document $index, Document $project, Database $dbForPlatform, Database $dbForProject): void
private function createIndex(Document $database, Document $collection, Document $index, Document $project, Database $dbForPlatform, Database $dbForProject, Realtime $queueForRealtime): void
{
if ($collection->isEmpty()) {
throw new Exception('Missing collection');
@@ -398,12 +391,7 @@ class Databases extends Action
}
$projectId = $project->getId();
$events = Event::generateEvents('databases.[databaseId].collections.[collectionId].indexes.[indexId].update', [
'databaseId' => $database->getId(),
'collectionId' => $collection->getId(),
'indexId' => $index->getId()
]);
$event = 'databases.[databaseId].collections.[collectionId].indexes.[indexId].update';
$collectionId = $collection->getId();
$key = $index->getAttribute('key', '');
$type = $index->getAttribute('type', '');
@@ -430,7 +418,7 @@ class Databases extends Action
throw $e;
} finally {
$this->trigger($database, $collection, $index, $project, $projectId, $events);
$this->trigger($database, $collection, $project, $event, $queueForRealtime, null, $index);
$dbForProject->purgeCachedDocument('database_' . $database->getInternalId(), $collectionId);
}
}
@@ -442,6 +430,7 @@ class Databases extends Action
* @param Document $project
* @param Database $dbForPlatform
* @param Database $dbForProject
* @param Realtime $queueForRealtime
* @return void
* @throws Authorization
* @throws Conflict
@@ -449,7 +438,7 @@ class Databases extends Action
* @throws DatabaseException
* @throws \Throwable
*/
private function deleteIndex(Document $database, Document $collection, Document $index, Document $project, Database $dbForPlatform, Database $dbForProject): void
private function deleteIndex(Document $database, Document $collection, Document $index, Document $project, Database $dbForPlatform, Database $dbForProject, Realtime $queueForRealtime): void
{
if ($collection->isEmpty()) {
throw new Exception('Missing collection');
@@ -459,12 +448,7 @@ class Databases extends Action
}
$projectId = $project->getId();
$events = Event::generateEvents('databases.[databaseId].collections.[collectionId].indexes.[indexId].delete', [
'databaseId' => $database->getId(),
'collectionId' => $collection->getId(),
'indexId' => $index->getId()
]);
$event = 'databases.[databaseId].collections.[collectionId].indexes.[indexId].delete';
$key = $index->getAttribute('key');
$status = $index->getAttribute('status', '');
$project = $dbForPlatform->getDocument('projects', $projectId);
@@ -490,7 +474,7 @@ class Databases extends Action
throw $e;
} finally {
$this->trigger($database, $collection, $index, $project, $projectId, $events);
$this->trigger($database, $collection, $project, $event, $queueForRealtime, null, $index);
$dbForProject->purgeCachedDocument('database_' . $database->getInternalId(), $collection->getId());
}
}
@@ -532,7 +516,6 @@ class Databases extends Action
$collectionId = $collection->getId();
$collectionInternalId = $collection->getInternalId();
$databaseId = $database->getId();
$databaseInternalId = $database->getInternalId();
$dbForProject->deleteCollection('database_' . $databaseInternalId . '_collection_' . $collection->getInternalId());
@@ -568,21 +551,21 @@ class Databases extends Action
/**
* @param string $collection collectionID
* @param string $collectionId
* @param array $queries
* @param Database $database
* @param callable|null $callback
* @return void
* @throws Exception
*/
protected function deleteByGroup(string $collection, array $queries, Database $database, callable $callback = null): void
protected function deleteByGroup(string $collectionId, array $queries, Database $database, callable $callback = null): void
{
$start = \microtime(true);
try {
$documents = $database->deleteDocuments($collection, $queries);
$documents = $database->deleteDocuments($collectionId, $queries);
} catch (\Throwable $th) {
Console::error('Failed to delete documents for collection ' . $collection . ': ' . $th->getMessage());
Console::error('Failed to delete documents for collection ' . $collectionId . ': ' . $th->getMessage());
return;
}
@@ -598,31 +581,42 @@ class Databases extends Action
Console::info("Deleted {$count} documents by group in " . ($end - $start) . " seconds");
}
/**
* @param Document $database
* @param Document $collection
* @param Document $project
* @param Realtime $queueForRealtime
* @param Document|null $attribute
* @param Document|null $index
* @return void
*/
protected function trigger(
Document $database,
Document $collection,
Document $attribute,
Document $project,
string $projectId,
array $events
string $event,
Realtime $queueForRealtime,
Document|null $attribute = null,
Document|null $index = null,
): void {
$target = Realtime::fromPayload(
// Pass first, most verbose event pattern
event: $events[0],
payload: $attribute,
project: $project,
);
Realtime::send(
projectId: 'console',
payload: $attribute->getArrayCopy(),
events: $events,
channels: $target['channels'],
roles: $target['roles'],
options: [
'projectId' => $projectId,
'databaseId' => $database->getId(),
'collectionId' => $collection->getId()
]
);
$queueForRealtime
->setProject($project)
->setSubscribers(['console'])
->setEvent($event)
->setParam('databaseId', $database->getId())
->setParam('collectionId', $collection->getId());
if ($attribute !== null && !empty($attribute)) {
$queueForRealtime
->setParam('attributeId', $attribute->getId())
->setPayload($attribute->getArrayCopy());
}
if ($index !== null && !empty($index)) {
$queueForRealtime
->setParam('indexId', $index->getId())
->setPayload($index->getArrayCopy());
}
$queueForRealtime->trigger();
}
}
src/Appwrite/Platform/Workers/Deletes.php
+27 -12
View File
@@ -48,7 +48,7 @@ class Deletes extends Action
->inject('project')
->inject('dbForPlatform')
->inject('getProjectDB')
->inject('timelimit')
->inject('getLogsDB')
->inject('deviceForFiles')
->inject('deviceForFunctions')
->inject('deviceForSites')
@@ -65,7 +65,7 @@ class Deletes extends Action
* @throws Exception
* @throws Throwable
*/
public function action(Message $message, Document $project, Database $dbForPlatform, callable $getProjectDB, callable $timelimit, Device $deviceForFiles, Device $deviceForFunctions, Device $deviceForSites, Device $deviceForBuilds, Device $deviceForCache, CertificatesAdapter $certificates, string $executionRetention, string $auditRetention, Log $log): void
public function action(Message $message, Document $project, Database $dbForPlatform, callable $getProjectDB, callable $getLogsDB, Device $deviceForFiles, Device $deviceForFunctions, Device $deviceForSites, Device $deviceForBuilds, Device $deviceForCache, CertificatesAdapter $certificates, string $executionRetention, string $auditRetention, Log $log): void
{
$payload = $message->getPayload() ?? [];
@@ -133,7 +133,7 @@ class Deletes extends Action
$this->deleteExpiredSessions($project, $getProjectDB);
break;
case DELETE_TYPE_USAGE:
$this->deleteUsageStats($project, $getProjectDB, $hourlyUsageRetentionDatetime);
$this->deleteUsageStats($project, $getProjectDB, $getLogsDB, $hourlyUsageRetentionDatetime);
break;
case DELETE_TYPE_CACHE_BY_RESOURCE:
$this->deleteCacheByResource($project, $getProjectDB, $resource, $resourceType);
@@ -160,7 +160,7 @@ class Deletes extends Action
$this->deleteExpiredTargets($project, $getProjectDB);
$this->deleteExecutionLogs($project, $getProjectDB, $executionRetention);
$this->deleteAuditLogs($project, $getProjectDB, $auditRetention);
$this->deleteUsageStats($project, $getProjectDB, $hourlyUsageRetentionDatetime);
$this->deleteUsageStats($project, $getProjectDB, $getLogsDB, $hourlyUsageRetentionDatetime);
$this->deleteExpiredSessions($project, $getProjectDB);
break;
default:
@@ -414,14 +414,27 @@ class Deletes extends Action
* @return void
* @throws Exception
*/
private function deleteUsageStats(Document $project, callable $getProjectDB, string $hourlyUsageRetentionDatetime): void
private function deleteUsageStats(Document $project, callable $getProjectDB, callable $getLogsDB, string $hourlyUsageRetentionDatetime): void
{
/** @var \Utopia\Database\Database $dbForProject*/
$dbForProject = $getProjectDB($project);
// Delete Usage stats
// Delete Usage stats from projectDB
$this->deleteByGroup('stats', [
Query::lessThan('time', $hourlyUsageRetentionDatetime),
Query::equal('period', ['1h']),
], $dbForProject);
if ($project->getId() !== 'console') {
/** @var \Utopia\Database\Database $dbForLogs*/
$dbForLogs = call_user_func($getLogsDB, $project);
// Delete Usage stats from logsDB
$this->deleteByGroup('stats', [
Query::lessThan('time', $hourlyUsageRetentionDatetime),
Query::equal('period', ['1h']),
], $dbForLogs);
}
}
/**
@@ -728,10 +741,12 @@ class Deletes extends Action
{
$projectId = $project->getId();
$dbForProject = $getProjectDB($project);
$audit = new Audit($dbForProject);
try {
$audit->cleanup($auditRetention);
$this->deleteByGroup(Audit::COLLECTION, [
Query::lessThan('time', $auditRetention),
Query::orderDesc('time'),
], $dbForProject);
} catch (DatabaseException $e) {
Console::error('Failed to delete audit logs for project ' . $projectId . ': ' . $e->getMessage());
}
@@ -1002,7 +1017,7 @@ class Deletes extends Action
} else {
Console::error('Failed to delete deployment files: ' . $deploymentPath);
}
} catch (\Throwable $th) {
} catch (Throwable $th) {
Console::error('Failed to delete deployment files: ' . $deploymentPath);
Console::error('[Error] Type: ' . get_class($th));
Console::error('[Error] Message: ' . $th->getMessage());
@@ -1032,7 +1047,7 @@ class Deletes extends Action
} else {
Console::error('Failed to delete build files: ' . $buildPath);
}
} catch (\Throwable $th) {
} catch (Throwable $th) {
Console::error('Failed to delete deployment files: ' . $buildPath);
Console::error('[Error] Type: ' . get_class($th));
Console::error('[Error] Message: ' . $th->getMessage());
@@ -1102,7 +1117,7 @@ class Deletes extends Action
* @param Database $database
* @param ?callable $callback
* @return void
* @throws Exception
* @throws DatabaseException
*/
protected function deleteByGroup(
string $collection,
@@ -1114,7 +1129,7 @@ class Deletes extends Action
try {
$documents = $database->deleteDocuments($collection, $queries);
} catch (\Throwable $th) {
} catch (Throwable $th) {
Console::error('Failed to delete documents for collection ' . $collection . ': ' . $th->getMessage());
return;
}
src/Appwrite/Platform/Workers/Functions.php
+24 -36
View File
@@ -5,8 +5,9 @@ namespace Appwrite\Platform\Workers;
use Ahc\Jwt\JWT;
use Appwrite\Event\Event;
use Appwrite\Event\Func;
use Appwrite\Event\Realtime;
use Appwrite\Event\StatsUsage;
use Appwrite\Messaging\Adapter\Realtime;
use Appwrite\Event\Webhook;
use Appwrite\Utopia\Response\Model\Execution;
use Exception;
use Executor\Executor;
@@ -44,7 +45,9 @@ class Functions extends Action
->inject('project')
->inject('message')
->inject('dbForProject')
->inject('queueForWebhooks')
->inject('queueForFunctions')
->inject('queueForRealtime')
->inject('queueForEvents')
->inject('queueForStatsUsage')
->inject('log')
@@ -52,7 +55,7 @@ class Functions extends Action
->callback([$this, 'action']);
}
public function action(Document $project, Message $message, Database $dbForProject, Func $queueForFunctions, Event $queueForEvents, StatsUsage $queueForStatsUsage, Log $log, callable $isResourceBlocked): void
public function action(Document $project, Message $message, Database $dbForProject, Webhook $queueForWebhooks, Func $queueForFunctions, Realtime $queueForRealtime, Event $queueForEvents, StatsUsage $queueForStatsUsage, Log $log, callable $isResourceBlocked): void
{
$payload = $message->getPayload() ?? [];
@@ -136,7 +139,9 @@ class Functions extends Action
$this->execute(
log: $log,
dbForProject: $dbForProject,
queueForWebhooks: $queueForWebhooks,
queueForFunctions: $queueForFunctions,
queueForRealtime: $queueForRealtime,
queueForStatsUsage: $queueForStatsUsage,
queueForEvents: $queueForEvents,
project: $project,
@@ -176,7 +181,9 @@ class Functions extends Action
$this->execute(
log: $log,
dbForProject: $dbForProject,
queueForWebhooks: $queueForWebhooks,
queueForFunctions: $queueForFunctions,
queueForRealtime: $queueForRealtime,
queueForStatsUsage: $queueForStatsUsage,
queueForEvents: $queueForEvents,
project: $project,
@@ -198,7 +205,9 @@ class Functions extends Action
$this->execute(
log: $log,
dbForProject: $dbForProject,
queueForWebhooks: $queueForWebhooks,
queueForFunctions: $queueForFunctions,
queueForRealtime: $queueForRealtime,
queueForStatsUsage: $queueForStatsUsage,
queueForEvents: $queueForEvents,
project: $project,
@@ -284,6 +293,7 @@ class Functions extends Action
* @param Log $log
* @param Database $dbForProject
* @param Func $queueForFunctions
* @param Realtime $queueForRealtime
* @param StatsUsage $queueForStatsUsage
* @param Event $queueForEvents
* @param Document $project
@@ -307,7 +317,9 @@ class Functions extends Action
private function execute(
Log $log,
Database $dbForProject,
Webhook $queueForWebhooks,
Func $queueForFunctions,
Realtime $queueForRealtime,
StatsUsage $queueForStatsUsage,
Event $queueForEvents,
Document $project,
@@ -552,20 +564,20 @@ class Functions extends Action
;
}
$execution = $dbForProject->updateDocument('executions', $executionId, $execution);
/** Trigger Webhook */
$executionModel = new Execution();
$queueForEvents
->setQueue(Event::WEBHOOK_QUEUE_NAME)
->setClass(Event::WEBHOOK_CLASS_NAME)
->setProject($project)
->setUser($user)
->setEvent('functions.[functionId].executions.[executionId].update')
->setParam('functionId', $function->getId())
->setParam('executionId', $execution->getId())
->setPayload($execution->getArrayCopy(array_keys($executionModel->getRules())))
->setPayload($execution->getArrayCopy(array_keys($executionModel->getRules())));
/** Trigger Webhook */
$queueForWebhooks
->from($queueForEvents)
->trigger();
/** Trigger Functions */
@@ -573,35 +585,11 @@ class Functions extends Action
->from($queueForEvents)
->trigger();
/** Trigger realtime event */
$allEvents = Event::generateEvents('functions.[functionId].executions.[executionId].update', [
'functionId' => $function->getId(),
'executionId' => $execution->getId()
]);
// Ensure all placeholder got related attribute
$execution = $execution->setAttribute('functionId', $execution->getAttribute('resourceId'));
$target = Realtime::fromPayload(
// Pass first, most verbose event pattern
event: $allEvents[0],
payload: $execution,
project: $project
);
Realtime::send(
projectId: 'console',
payload: $execution->getArrayCopy(),
events: $allEvents,
channels: $target['channels'],
roles: $target['roles']
);
Realtime::send(
projectId: $project->getId(),
payload: $execution->getArrayCopy(),
events: $allEvents,
channels: $target['channels'],
roles: $target['roles']
);
/** Trigger Realtime Events */
$queueForRealtime
->from($queueForEvents)
->setSubscribers(['console', $project->getId()])
->trigger();
if (!empty($error)) {
throw new Exception($error, $errorCode);
src/Appwrite/Platform/Workers/Migrations.php
+20 -38
View File
@@ -3,8 +3,7 @@
namespace Appwrite\Platform\Workers;
use Ahc\Jwt\JWT;
use Appwrite\Event\Event;
use Appwrite\Messaging\Adapter\Realtime;
use Appwrite\Event\Realtime;
use Exception;
use Utopia\CLI\Console;
use Utopia\Config\Config;
@@ -54,13 +53,14 @@ class Migrations extends Action
->inject('dbForProject')
->inject('dbForPlatform')
->inject('logError')
->inject('queueForRealtime')
->callback([$this, 'action']);
}
/**
* @throws Exception
*/
public function action(Message $message, Document $project, Database $dbForProject, Database $dbForPlatform, callable $logError): void
public function action(Message $message, Document $project, Database $dbForProject, Database $dbForPlatform, callable $logError, Realtime $queueForRealtime): void
{
$payload = $message->getPayload() ?? [];
@@ -87,7 +87,7 @@ class Migrations extends Action
return;
}
$this->processMigration($migration);
$this->processMigration($migration, $queueForRealtime);
}
/**
@@ -155,34 +155,16 @@ class Migrations extends Action
* @throws \Utopia\Database\Exception
* @throws Exception
*/
protected function updateMigrationDocument(Document $migration, Document $project): Document
protected function updateMigrationDocument(Document $migration, Document $project, Realtime $queueForRealtime): Document
{
/** Trigger Realtime */
$allEvents = Event::generateEvents('migrations.[migrationId].update', [
'migrationId' => $migration->getId(),
]);
$target = Realtime::fromPayload(
event: $allEvents[0],
payload: $migration,
project: $project
);
Realtime::send(
projectId: 'console',
payload: $migration->getArrayCopy(),
events: $allEvents,
channels: $target['channels'],
roles: $target['roles'],
);
Realtime::send(
projectId: $project->getId(),
payload: $migration->getArrayCopy(),
events: $allEvents,
channels: $target['channels'],
roles: $target['roles'],
);
/** Trigger Realtime Events */
$queueForRealtime
->setProject($project)
->setSubscribers(['console', $project->getId()])
->setEvent('migrations.[migrationId].update')
->setParam('migrationId', $migration->getId())
->setPayload($migration->getArrayCopy())
->trigger();
return $this->dbForProject->updateDocument('migrations', $migration->getId(), $migration);
}
@@ -231,7 +213,7 @@ class Migrations extends Action
* @throws \Utopia\Database\Exception
* @throws Exception
*/
protected function processMigration(Document $migration): void
protected function processMigration(Document $migration, Realtime $queueForRealtime): void
{
$project = $this->project;
$projectDocument = $this->dbForPlatform->getDocument('projects', $project->getId());
@@ -255,7 +237,7 @@ class Migrations extends Action
$migration->setAttribute('stage', 'processing');
$migration->setAttribute('status', 'processing');
$this->updateMigrationDocument($migration, $projectDocument);
$this->updateMigrationDocument($migration, $projectDocument, $queueForRealtime);
$source = $this->processSource($migration);
$destination = $this->processDestination($migration, $tempAPIKey);
@@ -269,14 +251,14 @@ class Migrations extends Action
/** Start Transfer */
$migration->setAttribute('stage', 'migrating');
$this->updateMigrationDocument($migration, $projectDocument);
$this->updateMigrationDocument($migration, $projectDocument, $queueForRealtime);
$transfer->run(
$migration->getAttribute('resources'),
function () use ($migration, $transfer, $projectDocument) {
function () use ($migration, $transfer, $projectDocument, $queueForRealtime) {
$migration->setAttribute('resourceData', json_encode($transfer->getCache()));
$migration->setAttribute('statusCounters', json_encode($transfer->getStatusCounters()));
$this->updateMigrationDocument($migration, $projectDocument);
$this->updateMigrationDocument($migration, $projectDocument, $queueForRealtime);
},
$migration->getAttribute('resourceId'),
$migration->getAttribute('resourceType')
@@ -313,7 +295,7 @@ class Migrations extends Action
}
$migration->setAttribute('errors', $errorMessages);
$this->updateMigrationDocument($migration, $projectDocument);
$this->updateMigrationDocument($migration, $projectDocument, $queueForRealtime);
return;
}
@@ -354,7 +336,7 @@ class Migrations extends Action
$migration->setAttribute('errors', $errorMessages);
}
} finally {
$this->updateMigrationDocument($migration, $projectDocument);
$this->updateMigrationDocument($migration, $projectDocument, $queueForRealtime);
if ($migration->getAttribute('status', '') === 'failed') {
Console::error('Migration('.$migration->getInternalId().':'.$migration->getId().') failed, Project('.$this->project->getInternalId().':'.$this->project->getId().')');
src/Appwrite/Platform/Workers/StatsResources.php
+40 -10
View File
@@ -77,10 +77,13 @@ class StatsResources extends Action
// Reset documents for each job
$this->documents = [];
$startTime = microtime(true);
$this->countForProject($dbForPlatform, $getLogsDB, $getProjectDB, $project);
$endTime = microtime(true);
$executionTime = $endTime - $startTime;
Console::info('Project: ' . $project->getId() . '(' . $project->getInternalId() . ') aggregated in ' . $executionTime .' seconds');
}
protected function countForProject(Database $dbForPlatform, callable $getLogsDB, callable $getProjectDB, Document $project): void
{
Console::info('Begining count for: ' . $project->getId());
@@ -129,9 +132,20 @@ class StatsResources extends Action
]);
$teams = $dbForProject->count('teams');
$functions = $dbForProject->count('functions');
$messages = $dbForProject->count('messages');
$providers = $dbForProject->count('providers');
$topics = $dbForProject->count('topics');
$targets = $dbForProject->count('targets');
$emailTargets = $dbForProject->count('targets', [
Query::equal('providerType', [MESSAGE_TYPE_EMAIL])
]);
$pushTargets = $dbForProject->count('targets', [
Query::equal('providerType', [MESSAGE_TYPE_PUSH])
]);
$smsTargets = $dbForProject->count('targets', [
Query::equal('providerType', [MESSAGE_TYPE_SMS])
]);
$metrics = [
METRIC_DATABASES => $databases,
@@ -148,6 +162,10 @@ class StatsResources extends Action
METRIC_PROVIDERS => $providers,
METRIC_TOPICS => $topics,
METRIC_KEYS => $keys,
METRIC_TARGETS => $targets,
str_replace('{providerType}', MESSAGE_TYPE_EMAIL, METRIC_PROVIDER_TYPE_TARGETS) => $emailTargets,
str_replace('{providerType}', MESSAGE_TYPE_PUSH, METRIC_PROVIDER_TYPE_TARGETS) => $pushTargets,
str_replace('{providerType}', MESSAGE_TYPE_SMS, METRIC_PROVIDER_TYPE_TARGETS) => $smsTargets,
];
foreach ($metrics as $metric => $value) {
@@ -167,7 +185,7 @@ class StatsResources extends Action
}
try {
$this->countForDatabase($dbForProject, $dbForLogs, $region);
$this->countForDatabase($dbForProject, $region);
} catch (Throwable $th) {
call_user_func_array($this->logError, [$th, "StatsResources", "count_for_database_{$project->getId()}"]);
}
@@ -227,42 +245,54 @@ class StatsResources extends Action
$this->createStatsDocuments($region, METRIC_FILES_IMAGES_TRANSFORMED, $totalImageTransformations);
}
protected function countForDatabase(Database $dbForProject, Database $dbForLogs, string $region)
protected function countForDatabase(Database $dbForProject, string $region)
{
$totalCollections = 0;
$totalDocuments = 0;
$this->foreachDocument($dbForProject, 'databases', [], function ($database) use ($dbForProject, $dbForLogs, $region, &$totalCollections, &$totalDocuments) {
$totalDatabaseStorage = 0;
$this->foreachDocument($dbForProject, 'databases', [], function ($database) use ($dbForProject, $region, &$totalCollections, &$totalDocuments, &$totalDatabaseStorage) {
$collections = $dbForProject->count('database_' . $database->getInternalId());
$metric = str_replace('{databaseInternalId}', $database->getInternalId(), METRIC_DATABASE_ID_COLLECTIONS);
$this->createStatsDocuments($region, $metric, $collections);
$documents = $this->countForCollections($dbForProject, $dbForLogs, $database, $region);
[$documents, $storage] = $this->countForCollections($dbForProject, $database, $region);
$totalDatabaseStorage += $storage;
$totalDocuments += $documents;
$totalCollections += $collections;
});
$this->createStatsDocuments($region, METRIC_COLLECTIONS, $totalCollections);
$this->createStatsDocuments($region, METRIC_DOCUMENTS, $totalDocuments);
$this->createStatsDocuments($region, METRIC_DATABASES_STORAGE, $totalDatabaseStorage);
}
protected function countForCollections(Database $dbForProject, Database $dbForLogs, Document $database, string $region): int
protected function countForCollections(Database $dbForProject, Document $database, string $region): array
{
$databaseDocuments = 0;
$this->foreachDocument($dbForProject, 'database_' . $database->getInternalId(), [], function ($collection) use ($dbForProject, $dbForLogs, $database, $region, &$totalCollections, &$databaseDocuments) {
$databaseStorage = 0;
$this->foreachDocument($dbForProject, 'database_' . $database->getInternalId(), [], function ($collection) use ($dbForProject, $database, $region, &$databaseStorage, &$databaseDocuments) {
$documents = $dbForProject->count('database_' . $database->getInternalId() . '_collection_' . $collection->getInternalId());
$metric = str_replace(['{databaseInternalId}', '{collectionInternalId}'], [$database->getInternalId(), $collection->getInternalId()], METRIC_DATABASE_ID_COLLECTION_ID_DOCUMENTS);
$this->createStatsDocuments($region, $metric, $documents);
$databaseDocuments += $documents;
$collectionStorage = $dbForProject->getSizeOfCollection('database_' . $database->getInternalId() . '_collection_' . $collection->getInternalId());
$metric = str_replace(['{databaseInternalId}', '{collectionInternalId}'], [$database->getInternalId(), $collection->getInternalId()], METRIC_DATABASE_ID_COLLECTION_ID_STORAGE);
$this->createStatsDocuments($region, $metric, $collectionStorage);
$databaseStorage += $collectionStorage;
});
$metric = str_replace(['{databaseInternalId}'], [$database->getInternalId()], METRIC_DATABASE_ID_DOCUMENTS);
$this->createStatsDocuments($region, $metric, $databaseDocuments);
return $databaseDocuments;
$metric = str_replace(['{databaseInternalId}'], [$database->getInternalId()], METRIC_DATABASE_ID_STORAGE);
$this->createStatsDocuments($region, $metric, $databaseStorage);
return [$databaseDocuments, $databaseStorage];
}
protected function countForFunctions(Database $dbForProject, Database $dbForLogs, string $region)
src/Appwrite/Platform/Workers/StatsUsage.php
+9 -2
View File
@@ -17,13 +17,20 @@ class StatsUsage extends Action
private int $lastTriggeredTime = 0;
private int $keys = 0;
private const INFINITY_PERIOD = '_inf_';
private const KEYS_THRESHOLD = 10000;
private const BATCH_SIZE_DEVELOPMENT = 1;
private const BATCH_SIZE_PRODUCTION = 10_000;
public static function getName(): string
{
return 'stats-usage';
}
private function getBatchSize(): int
{
return System::getEnv('_APP_ENV', 'development') === 'development'
? self::BATCH_SIZE_DEVELOPMENT
: self::BATCH_SIZE_PRODUCTION;
}
/**
* @throws Exception
*/
@@ -86,7 +93,7 @@ class StatsUsage extends Action
// If keys crossed threshold or X time passed since the last send and there are some keys in the array ($this->stats)
if (
$this->keys >= self::KEYS_THRESHOLD ||
$this->keys >= $this->getBatchSize() ||
(time() - $this->lastTriggeredTime > $aggregationInterval && $this->keys > 0)
) {
Console::warning('[' . DateTime::now() . '] Aggregated ' . $this->keys . ' keys');
src/Appwrite/Platform/Workers/StatsUsageDump.php
+17 -7
View File
@@ -48,6 +48,7 @@ class StatsUsageDump extends Action
METRIC_BUILDS => true,
METRIC_COLLECTIONS => true,
METRIC_DOCUMENTS => true,
METRIC_DATABASES_STORAGE => true,
];
/**
@@ -63,6 +64,7 @@ class StatsUsageDump extends Action
'.deployments.storage',
'.builds',
'.builds.storage',
'.databases.storage'
];
/**
@@ -117,7 +119,7 @@ class StatsUsageDump extends Action
$project = new Document($stats['project'] ?? []);
$numberOfKeys = !empty($stats['keys']) ? count($stats['keys']) : 0;
$receivedAt = $stats['receivedAt'] ?? 'NONE';
$receivedAt = $stats['receivedAt'] ?? null;
if ($numberOfKeys === 0) {
continue;
}
@@ -134,7 +136,7 @@ class StatsUsageDump extends Action
if (str_contains($key, METRIC_DATABASES_STORAGE)) {
try {
$this->handleDatabaseStorage($key, $dbForProject, $project);
$this->handleDatabaseStorage($key, $dbForProject, $project, $receivedAt);
} catch (\Exception $e) {
console::error('[' . DateTime::now() . '] failed to calculate database storage for key [' . $key . '] ' . $e->getMessage());
}
@@ -142,7 +144,11 @@ class StatsUsageDump extends Action
}
foreach ($this->periods as $period => $format) {
$time = 'inf' === $period ? null : date($format, time());
$time = null;
if ($period !== 'inf') {
$time = !empty($receivedAt) ? (new \DateTime($receivedAt))->format($format) : date($format, time());
}
$id = \md5("{$time}_{$period}_{$key}");
$document = new Document([
@@ -171,12 +177,12 @@ class StatsUsageDump extends Action
}
}
private function handleDatabaseStorage(string $key, Database $dbForProject, Document $project): void
private function handleDatabaseStorage(string $key, Database $dbForProject, Document $project, string $receivedAt): void
{
$data = explode('.', $key);
$start = microtime(true);
$updateMetric = function (Database $dbForProject, Document $project, int $value, string $key, string $period, string|null $time) {
$updateMetric = function (Database $dbForProject, Document $project, int $value, string $key, string $period, string|null $time) use ($receivedAt) {
$id = \md5("{$time}_{$period}_{$key}");
$document = new Document([
@@ -197,7 +203,11 @@ class StatsUsageDump extends Action
};
foreach ($this->periods as $period => $format) {
$time = 'inf' === $period ? null : date($format, time());
$time = null;
if ($period !== 'inf') {
$time = !empty($receivedAt) ? (new \DateTime($receivedAt))->format($format) : date($format, time());
}
$id = \md5("{$time}_{$period}_{$key}");
$value = 0;
@@ -320,7 +330,7 @@ class StatsUsageDump extends Action
protected function writeToLogsDB(Document $project, Document $document): void
{
if (!System::getEnv('_APP_STATS_USAGE_DUAL_WRITING', false)) {
if (System::getEnv('_APP_STATS_USAGE_DUAL_WRITING', 'disabled') === 'disabled') {
Console::log('Dual Writing is disabled. Skipping...');
return;
}
src/Appwrite/Platform/Workers/Usage.php
-290
View File
@@ -1,290 +0,0 @@
<?php
namespace Appwrite\Platform\Workers;
use Appwrite\Event\UsageDump;
use Exception;
use Utopia\CLI\Console;
use Utopia\Database\DateTime;
use Utopia\Database\Document;
use Utopia\Platform\Action;
use Utopia\Queue\Message;
use Utopia\System\System;
class Usage extends Action
{
private array $stats = [];
private int $lastTriggeredTime = 0;
private int $aggregationInterval = 20;
private int $keys = 0;
private const INFINITY_PERIOD = '_inf_';
private const KEYS_THRESHOLD = 20000;
public static function getName(): string
{
return 'usage';
}
/**
* @throws Exception
*/
public function __construct()
{
$this
->desc('Usage worker')
->inject('message')
->inject('project')
->inject('getProjectDB')
->inject('queueForUsageDump')
->callback([$this, 'action']);
$this->aggregationInterval = (int) System::getEnv('_APP_USAGE_AGGREGATION_INTERVAL', '20');
$this->lastTriggeredTime = time();
}
/**
* @param Message $message
* @param Document $project
* @param callable $getProjectDB
* @param UsageDump $queueForUsageDump
* @return void
* @throws \Utopia\Database\Exception
* @throws Exception
*/
public function action(Message $message, Document $project, callable $getProjectDB, UsageDump $queueForUsageDump): void
{
$payload = $message->getPayload() ?? [];
if (empty($payload)) {
throw new Exception('Missing payload');
}
if (empty($project->getAttribute('database'))) {
var_dump($payload);
return;
}
$projectId = $project->getInternalId();
foreach ($payload['reduce'] ?? [] as $document) {
if (empty($document)) {
continue;
}
$this->reduce(
project: $project,
document: new Document($document),
metrics: $payload['metrics'],
getProjectDB: $getProjectDB
);
}
$this->stats[$projectId]['project'] = [
'$id' => $project->getId(),
'$internalId' => $project->getInternalId(),
'database' => $project->getAttribute('database'),
];
$this->stats[$projectId]['receivedAt'] = DateTime::now();
foreach ($payload['metrics'] ?? [] as $metric) {
$this->keys++;
if (!isset($this->stats[$projectId]['keys'][$metric['key']])) {
$this->stats[$projectId]['keys'][$metric['key']] = $metric['value'];
continue;
}
$this->stats[$projectId]['keys'][$metric['key']] += $metric['value'];
}
// If keys crossed threshold or X time passed since the last send and there are some keys in the array ($this->stats)
if (
$this->keys >= self::KEYS_THRESHOLD ||
(time() - $this->lastTriggeredTime > $this->aggregationInterval && $this->keys > 0)
) {
Console::warning('[' . DateTime::now() . '] Aggregated ' . $this->keys . ' keys');
$queueForUsageDump
->setStats($this->stats)
->trigger();
$this->stats = [];
$this->keys = 0;
$this->lastTriggeredTime = time();
}
}
/**
* On Documents that tied by relations like functions>deployments>build || documents>collection>database || buckets>files.
* When we remove a parent document we need to deduct his children aggregation from the project scope.
* @param Document $project
* @param Document $document
* @param array $metrics
* @param callable $getProjectDB
* @return void
*/
private function reduce(Document $project, Document $document, array &$metrics, callable $getProjectDB): void
{
$dbForProject = $getProjectDB($project);
try {
switch (true) {
case $document->getCollection() === 'users': // users
$sessions = count($document->getAttribute(METRIC_SESSIONS, 0));
if (!empty($sessions)) {
$metrics[] = [
'key' => METRIC_SESSIONS,
'value' => ($sessions * -1),
];
}
break;
case $document->getCollection() === 'databases': // databases
$collections = $dbForProject->getDocument('stats', md5(self::INFINITY_PERIOD . str_replace('{databaseInternalId}', $document->getInternalId(), METRIC_DATABASE_ID_COLLECTIONS)));
$documents = $dbForProject->getDocument('stats', md5(self::INFINITY_PERIOD . str_replace('{databaseInternalId}', $document->getInternalId(), METRIC_DATABASE_ID_DOCUMENTS)));
if (!empty($collections['value'])) {
$metrics[] = [
'key' => METRIC_COLLECTIONS,
'value' => ($collections['value'] * -1),
];
}
if (!empty($documents['value'])) {
$metrics[] = [
'key' => METRIC_DOCUMENTS,
'value' => ($documents['value'] * -1),
];
}
break;
case str_starts_with($document->getCollection(), 'database_') && !str_contains($document->getCollection(), 'collection'): //collections
$parts = explode('_', $document->getCollection());
$databaseInternalId = $parts[1] ?? 0;
$documents = $dbForProject->getDocument('stats', md5(self::INFINITY_PERIOD . str_replace(['{databaseInternalId}', '{collectionInternalId}'], [$databaseInternalId, $document->getInternalId()], METRIC_DATABASE_ID_COLLECTION_ID_DOCUMENTS)));
if (!empty($documents['value'])) {
$metrics[] = [
'key' => METRIC_DOCUMENTS,
'value' => ($documents['value'] * -1),
];
$metrics[] = [
'key' => str_replace('{databaseInternalId}', $databaseInternalId, METRIC_DATABASE_ID_DOCUMENTS),
'value' => ($documents['value'] * -1),
];
}
break;
case $document->getCollection() === 'buckets':
$files = $dbForProject->getDocument('stats', md5(self::INFINITY_PERIOD . str_replace('{bucketInternalId}', $document->getInternalId(), METRIC_BUCKET_ID_FILES)));
$storage = $dbForProject->getDocument('stats', md5(self::INFINITY_PERIOD . str_replace('{bucketInternalId}', $document->getInternalId(), METRIC_BUCKET_ID_FILES_STORAGE)));
if (!empty($files['value'])) {
$metrics[] = [
'key' => METRIC_FILES,
'value' => ($files['value'] * -1),
];
}
if (!empty($storage['value'])) {
$metrics[] = [
'key' => METRIC_FILES_STORAGE,
'value' => ($storage['value'] * -1),
];
}
break;
case $document->getCollection() === 'functions':
$deployments = $dbForProject->getDocument('stats', md5(self::INFINITY_PERIOD . str_replace(['{resourceType}', '{resourceInternalId}'], ['functions', $document->getInternalId()], METRIC_FUNCTION_ID_DEPLOYMENTS)));
$deploymentsStorage = $dbForProject->getDocument('stats', md5(self::INFINITY_PERIOD . str_replace(['{resourceType}', '{resourceInternalId}'], ['functions', $document->getInternalId()], METRIC_FUNCTION_ID_DEPLOYMENTS_STORAGE)));
$builds = $dbForProject->getDocument('stats', md5(self::INFINITY_PERIOD . str_replace('{functionInternalId}', $document->getInternalId(), METRIC_FUNCTION_ID_BUILDS)));
$buildsSuccess = $dbForProject->getDocument('stats', md5(self::INFINITY_PERIOD . str_replace('{functionInternalId}', $document->getInternalId(), METRIC_FUNCTION_ID_BUILDS_SUCCESS)));
$buildsFailed = $dbForProject->getDocument('stats', md5(self::INFINITY_PERIOD . str_replace('{functionInternalId}', $document->getInternalId(), METRIC_FUNCTION_ID_BUILDS_FAILED)));
$buildsStorage = $dbForProject->getDocument('stats', md5(self::INFINITY_PERIOD . str_replace('{functionInternalId}', $document->getInternalId(), METRIC_FUNCTION_ID_BUILDS_STORAGE)));
$buildsCompute = $dbForProject->getDocument('stats', md5(self::INFINITY_PERIOD . str_replace('{functionInternalId}', $document->getInternalId(), METRIC_FUNCTION_ID_BUILDS_COMPUTE)));
$buildsComputeSuccess = $dbForProject->getDocument('stats', md5(self::INFINITY_PERIOD . str_replace('{functionInternalId}', $document->getInternalId(), METRIC_FUNCTION_ID_BUILDS_COMPUTE_SUCCESS)));
$buildsComputeFailed = $dbForProject->getDocument('stats', md5(self::INFINITY_PERIOD . str_replace('{functionInternalId}', $document->getInternalId(), METRIC_FUNCTION_ID_BUILDS_COMPUTE_FAILED)));
$executions = $dbForProject->getDocument('stats', md5(self::INFINITY_PERIOD . str_replace('{functionInternalId}', $document->getInternalId(), METRIC_FUNCTION_ID_EXECUTIONS)));
$executionsCompute = $dbForProject->getDocument('stats', md5(self::INFINITY_PERIOD . str_replace('{functionInternalId}', $document->getInternalId(), METRIC_FUNCTION_ID_EXECUTIONS_COMPUTE)));
if (!empty($deployments['value'])) {
$metrics[] = [
'key' => METRIC_DEPLOYMENTS,
'value' => ($deployments['value'] * -1),
];
}
if (!empty($deploymentsStorage['value'])) {
$metrics[] = [
'key' => METRIC_DEPLOYMENTS_STORAGE,
'value' => ($deploymentsStorage['value'] * -1),
];
}
if (!empty($builds['value'])) {
$metrics[] = [
'key' => METRIC_BUILDS,
'value' => ($builds['value'] * -1),
];
}
if (!empty($buildsSuccess['value'])) {
$metrics[] = [
'key' => METRIC_BUILDS_SUCCESS,
'value' => ($buildsSuccess['value'] * -1),
];
}
if (!empty($buildsFailed['value'])) {
$metrics[] = [
'key' => METRIC_BUILDS_FAILED,
'value' => ($buildsFailed['value'] * -1),
];
}
if (!empty($buildsStorage['value'])) {
$metrics[] = [
'key' => METRIC_BUILDS_STORAGE,
'value' => ($buildsStorage['value'] * -1),
];
}
if (!empty($buildsCompute['value'])) {
$metrics[] = [
'key' => METRIC_BUILDS_COMPUTE,
'value' => ($buildsCompute['value'] * -1),
];
}
if (!empty($buildsComputeSuccess['value'])) {
$metrics[] = [
'key' => METRIC_BUILDS_COMPUTE_SUCCESS,
'value' => ($buildsComputeSuccess['value'] * -1),
];
}
if (!empty($buildsComputeFailed['value'])) {
$metrics[] = [
'key' => METRIC_BUILDS_COMPUTE_FAILED,
'value' => ($buildsComputeFailed['value'] * -1),
];
}
if (!empty($executions['value'])) {
$metrics[] = [
'key' => METRIC_EXECUTIONS,
'value' => ($executions['value'] * -1),
];
}
if (!empty($executionsCompute['value'])) {
$metrics[] = [
'key' => METRIC_EXECUTIONS_COMPUTE,
'value' => ($executionsCompute['value'] * -1),
];
}
break;
default:
break;
}
} catch (\Throwable $e) {
console::error("[reducer] " . " {DateTime::now()} " . " {$project->getInternalId()} " . " {$e->getMessage()}");
}
}
}
src/Appwrite/Platform/Workers/UsageDump.php
-284
View File
@@ -1,284 +0,0 @@
<?php
namespace Appwrite\Platform\Workers;
use Appwrite\Extend\Exception;
use Utopia\CLI\Console;
use Utopia\Database\Database;
use Utopia\Database\DateTime;
use Utopia\Database\Document;
use Utopia\Database\Exception\Duplicate;
use Utopia\Platform\Action;
use Utopia\Queue\Message;
use Utopia\System\System;
const METRIC_COLLECTION_LEVEL_STORAGE = 4;
const METRIC_DATABASE_LEVEL_STORAGE = 3;
const METRIC_PROJECT_LEVEL_STORAGE = 2;
class UsageDump extends Action
{
protected array $stats = [];
protected array $periods = [
'1h' => 'Y-m-d H:00',
'1d' => 'Y-m-d 00:00',
'inf' => '0000-00-00 00:00'
];
public static function getName(): string
{
return 'usage-dump';
}
/**
* @throws \Exception
*/
public function __construct()
{
$this
->inject('message')
->inject('getProjectDB')
->callback([$this, 'action']);
}
/**
* @param Message $message
* @param callable $getProjectDB
* @return void
* @throws Exception
* @throws \Utopia\Database\Exception
*/
public function action(Message $message, callable $getProjectDB): void
{
$payload = $message->getPayload() ?? [];
if (empty($payload)) {
throw new Exception('Missing payload');
}
foreach ($payload['stats'] ?? [] as $stats) {
$project = new Document($stats['project'] ?? []);
/**
* End temp bug fallback
*/
$numberOfKeys = !empty($stats['keys']) ? count($stats['keys']) : 0;
$receivedAt = $stats['receivedAt'] ?? 'NONE';
if ($numberOfKeys === 0) {
continue;
}
console::log('['.DateTime::now().'] Id: '.$project->getId(). ' InternalId: '.$project->getInternalId(). ' Db: '.$project->getAttribute('database').' ReceivedAt: '.$receivedAt. ' Keys: '.$numberOfKeys);
try {
$dbForProject = $getProjectDB($project);
foreach ($stats['keys'] ?? [] as $key => $value) {
if ($value == 0) {
continue;
}
if (str_contains($key, METRIC_DATABASES_STORAGE)) {
try {
$this->handleDatabaseStorage($key, $dbForProject);
} catch (\Exception $e) {
console::error('[' . DateTime::now() . '] failed to calculate database storage for key [' . $key . '] ' . $e->getMessage());
}
continue;
}
foreach ($this->periods as $period => $format) {
$time = 'inf' === $period ? null : date($format, time());
$id = \md5("{$time}_{$period}_{$key}");
try {
$dbForProject->createDocument('stats', new Document([
'$id' => $id,
'period' => $period,
'time' => $time,
'metric' => $key,
'value' => $value,
'region' => System::getEnv('_APP_REGION', 'default'),
]));
} catch (Duplicate $th) {
if ($value < 0) {
$dbForProject->decreaseDocumentAttribute(
'stats',
$id,
'value',
abs($value)
);
} else {
$dbForProject->increaseDocumentAttribute(
'stats',
$id,
'value',
$value
);
}
}
}
}
} catch (\Exception $e) {
console::error('[' . DateTime::now() . '] project [' . $project->getInternalId() . '] database [' . $project['database'] . '] ' . ' ' . $e->getMessage());
}
}
}
private function handleDatabaseStorage(string $key, Database $dbForProject): void
{
$data = explode('.', $key);
$start = microtime(true);
$updateMetric = function (Database $dbForProject, int $value, string $key, string $period, string|null $time) {
$id = \md5("{$time}_{$period}_{$key}");
try {
$dbForProject->createDocument('stats', new Document([
'$id' => $id,
'period' => $period,
'time' => $time,
'metric' => $key,
'value' => $value,
'region' => System::getEnv('_APP_REGION', 'default'),
]));
} catch (Duplicate $th) {
if ($value < 0) {
$dbForProject->decreaseDocumentAttribute(
'stats',
$id,
'value',
abs($value)
);
} else {
$dbForProject->increaseDocumentAttribute(
'stats',
$id,
'value',
$value
);
}
}
};
foreach ($this->periods as $period => $format) {
$time = 'inf' === $period ? null : date($format, time());
$id = \md5("{$time}_{$period}_{$key}");
$value = 0;
$previousValue = 0;
try {
$previousValue = ($dbForProject->getDocument('stats', $id))->getAttribute('value', 0);
} catch (\Exception $e) {
// No previous value
}
switch (count($data)) {
// Collection Level
case METRIC_COLLECTION_LEVEL_STORAGE:
Console::log('[' . DateTime::now() . '] Collection Level Storage Calculation [' . $key . ']');
$databaseInternalId = $data[0];
$collectionInternalId = $data[1];
try {
$value = $dbForProject->getSizeOfCollection('database_' . $databaseInternalId . '_collection_' . $collectionInternalId);
} catch (\Exception $e) {
// Collection not found
if ($e->getMessage() !== 'Collection not found') {
throw $e;
}
}
// Compare with previous value
$diff = $value - $previousValue;
if ($diff === 0) {
break;
}
// Update Collection
$updateMetric($dbForProject, $diff, $key, $period, $time);
// Update Database
$databaseKey = str_replace(['{databaseInternalId}'], [$data[0]], METRIC_DATABASE_ID_STORAGE);
$updateMetric($dbForProject, $diff, $databaseKey, $period, $time);
// Update Project
$projectKey = METRIC_DATABASES_STORAGE;
$updateMetric($dbForProject, $diff, $projectKey, $period, $time);
break;
// Database Level
case METRIC_DATABASE_LEVEL_STORAGE:
Console::log('[' . DateTime::now() . '] Database Level Storage Calculation [' . $key . ']');
$databaseInternalId = $data[0];
$collections = [];
try {
$collections = $dbForProject->find('database_' . $databaseInternalId);
} catch (\Exception $e) {
// Database not found
if ($e->getMessage() !== 'Collection not found') {
throw $e;
}
}
foreach ($collections as $collection) {
try {
$value += $dbForProject->getSizeOfCollection('database_' . $databaseInternalId . '_collection_' . $collection->getInternalId());
} catch (\Exception $e) {
// Collection not found
if ($e->getMessage() !== 'Collection not found') {
throw $e;
}
}
}
$diff = $value - $previousValue;
if ($diff === 0) {
break;
}
// Update Database
$databaseKey = str_replace(['{databaseInternalId}'], [$data[0]], METRIC_DATABASE_ID_STORAGE);
$updateMetric($dbForProject, $diff, $databaseKey, $period, $time);
// Update Project
$projectKey = METRIC_DATABASES_STORAGE;
$updateMetric($dbForProject, $diff, $projectKey, $period, $time);
break;
// Project Level
case METRIC_PROJECT_LEVEL_STORAGE:
Console::log('[' . DateTime::now() . '] Project Level Storage Calculation [' . $key . ']');
// Get all project databases
$databases = $dbForProject->find('database');
// Recalculate all databases
foreach ($databases as $database) {
$collections = $dbForProject->find('database_' . $database->getInternalId());
foreach ($collections as $collection) {
try {
$value += $dbForProject->getSizeOfCollection('database_' . $database->getInternalId() . '_collection_' . $collection->getInternalId());
} catch (\Exception $e) {
// Collection not found
if ($e->getMessage() !== 'Collection not found') {
throw $e;
}
}
}
}
$diff = $value - $previousValue;
// Update Project
$projectKey = METRIC_DATABASES_STORAGE;
$updateMetric($dbForProject, $diff, $projectKey, $period, $time);
break;
}
}
$end = microtime(true);
console::log('[' . DateTime::now() . '] DB Storage Calculation [' . $key . '] took ' . (($end - $start) * 1000) . ' milliseconds');
}
}
src/Appwrite/Utopia/Database/Validator/Queries/Base.php
+10 -1
View File
@@ -24,7 +24,15 @@ class Base extends Queries
public function __construct(string $collection, array $allowedAttributes)
{
$config = Config::getParam('collections', []);
$collections = array_merge($config['projects'], $config['buckets'], $config['databases'], $config['console']);
$collections = array_merge(
$config['projects'],
$config['buckets'],
$config['databases'],
$config['console'],
$config['logs']
);
$collection = $collections[$collection];
// array for constant lookup time
$allowedAttributesLookup = [];
@@ -35,6 +43,7 @@ class Base extends Queries
$attributes = [];
foreach ($collection['attributes'] as $attribute) {
$key = $attribute['$id'];
if (!isset($allowedAttributesLookup[$key])) {
continue;
}
src/Appwrite/Utopia/Response/Model/UsageBuckets.php
+13
View File
@@ -42,6 +42,19 @@ class UsageBuckets extends Model
'example' => [],
'array' => true
])
->addRule('imageTransformations', [
'type' => Response::MODEL_METRIC,
'description' => 'Aggregated number of files transformations per period.',
'default' => [],
'example' => [],
'array' => true
])
->addRule('imageTransformationsTotal', [
'type' => self::TYPE_INTEGER,
'description' => 'Total aggregated number of files transformations.',
'default' => 0,
'example' => 0,
])
;
}
src/Appwrite/Utopia/Response/Model/UsageProject.php
+13
View File
@@ -197,6 +197,19 @@ class UsageProject extends Model
'example' => [],
'array' => true
])
->addRule('imageTransformations', [
'type' => Response::MODEL_METRIC,
'description' => 'An array of aggregated number of image transformations.',
'default' => [],
'example' => [],
'array' => true
])
->addRule('imageTransformationsTotal', [
'type' => self::TYPE_INTEGER,
'description' => 'Total aggregated number of image transformations.',
'default' => 0,
'example' => 0,
])
;
}