Feat-batch writes in stats usage dump

.env

@@ -87,7 +87,7 @@ _APP_MAINTENANCE_RETENTION_EXECUTION=1209600
 _APP_MAINTENANCE_RETENTION_ABUSE=86400
 _APP_MAINTENANCE_RETENTION_AUDIT=1209600
 _APP_USAGE_AGGREGATION_INTERVAL=30
-_APP_STATS_RESOURCES_INTERVAL=3600
+_APP_STATS_RESOURCES_INTERVAL=30
 _APP_MAINTENANCE_RETENTION_USAGE_HOURLY=8640000
 _APP_MAINTENANCE_RETENTION_SCHEDULES=86400
 _APP_USAGE_STATS=enabled

app/controllers/api/databases.php

@@ -534,7 +534,6 @@ App::post('/v1/databases')
         }
 
         $queueForEvents->setParam('databaseId', $database->getId());
-        $queueForStatsUsage->addMetric(str_replace(['{databaseInternalId}'], [$database->getInternalId()], METRIC_DATABASE_ID_STORAGE), 1); // per database
 
         $response
             ->setStatusCode(Response::STATUS_CODE_CREATED)
@@ -830,9 +829,6 @@ App::delete('/v1/databases/:databaseId')
             ->setParam('databaseId', $database->getId())
             ->setPayload($response->output($database, Response::MODEL_DATABASE));
 
-        $queueForStatsUsage
-            ->addMetric(METRIC_DATABASES_STORAGE, 1); // Global, deletion forces full recalculation
-
         $response->noContent();
     });
 
@@ -2732,9 +2728,6 @@ App::delete('/v1/databases/:databaseId/collections/:collectionId/attributes/:key
             ->setContext('database', $db)
             ->setPayload($response->output($attribute, $model));
 
-        $queueForStatsUsage
-            ->addMetric(str_replace(['{databaseInternalId}', '{collectionInternalId}'], [$db->getInternalId(), $collection->getInternalId()], METRIC_DATABASE_ID_COLLECTION_ID_STORAGE), 1); // per collection
-
         $response->noContent();
     });
 
@@ -3356,8 +3349,7 @@ App::post('/v1/databases/:databaseId/collections/:collectionId/documents')
 
         $queueForStatsUsage
             ->addMetric(METRIC_DATABASES_OPERATIONS_WRITES, $operations)
-            ->addMetric(str_replace('{databaseInternalId}', $database->getInternalId(), METRIC_DATABASE_ID_OPERATIONS_WRITES), $operations)
-            ->addMetric(str_replace(['{databaseInternalId}', '{collectionInternalId}'], [$database->getInternalId(), $collection->getInternalId()], METRIC_DATABASE_ID_COLLECTION_ID_STORAGE), 1); // per collection
+            ->addMetric(str_replace('{databaseInternalId}', $database->getInternalId(), METRIC_DATABASE_ID_OPERATIONS_WRITES), $operations);
 
         $response->addHeader('X-Debug-Operations', $operations);
 
@@ -4141,8 +4133,7 @@ App::delete('/v1/databases/:databaseId/collections/:collectionId/documents/:docu
 
         $queueForStatsUsage
             ->addMetric(METRIC_DATABASES_OPERATIONS_WRITES, 1)
-            ->addMetric(str_replace('{databaseInternalId}', $database->getInternalId(), METRIC_DATABASE_ID_OPERATIONS_WRITES), 1)
-            ->addMetric(str_replace(['{databaseInternalId}', '{collectionInternalId}'], [$database->getInternalId(), $collection->getInternalId()], METRIC_DATABASE_ID_COLLECTION_ID_STORAGE), 1); // per collection
+            ->addMetric(str_replace('{databaseInternalId}', $database->getInternalId(), METRIC_DATABASE_ID_OPERATIONS_WRITES), 1);
 
         $response->addHeader('X-Debug-Operations', 1);
 

app/controllers/shared/api.php

@@ -109,66 +109,6 @@ $usageDatabaseListener = function (string $event, Document $document, StatsUsage
         case $document->getCollection() === 'sessions': // sessions
             $queueForStatsUsage->addMetric(METRIC_SESSIONS, $value); //per project
             break;
-        case $document->getCollection() === 'databases': // databases
-            $queueForStatsUsage->addMetric(METRIC_DATABASES, $value); // per project
-
-            if ($event === Database::EVENT_DOCUMENT_DELETE) {
-                $queueForStatsUsage->addReduce($document);
-            }
-            break;
-        case str_starts_with($document->getCollection(), 'database_') && !str_contains($document->getCollection(), 'collection'): //collections
-            $parts = explode('_', $document->getCollection());
-            $databaseInternalId = $parts[1] ?? 0;
-            $queueForStatsUsage
-                ->addMetric(METRIC_COLLECTIONS, $value) // per project
-                ->addMetric(str_replace('{databaseInternalId}', $databaseInternalId, METRIC_DATABASE_ID_COLLECTIONS), $value);
-
-            if ($event === Database::EVENT_DOCUMENT_DELETE) {
-                $queueForStatsUsage->addReduce($document);
-            }
-            break;
-        case str_starts_with($document->getCollection(), 'database_') && str_contains($document->getCollection(), '_collection_'): //documents
-            $parts = explode('_', $document->getCollection());
-            $databaseInternalId   = $parts[1] ?? 0;
-            $collectionInternalId = $parts[3] ?? 0;
-            $queueForStatsUsage
-                ->addMetric(METRIC_DOCUMENTS, $value)  // per project
-                ->addMetric(str_replace('{databaseInternalId}', $databaseInternalId, METRIC_DATABASE_ID_DOCUMENTS), $value) // per database
-                ->addMetric(str_replace(['{databaseInternalId}', '{collectionInternalId}'], [$databaseInternalId, $collectionInternalId], METRIC_DATABASE_ID_COLLECTION_ID_DOCUMENTS), $value);  // per collection
-            break;
-        case $document->getCollection() === 'buckets': //buckets
-            $queueForStatsUsage
-                ->addMetric(METRIC_BUCKETS, $value); // per project
-            if ($event === Database::EVENT_DOCUMENT_DELETE) {
-                $queueForStatsUsage
-                    ->addReduce($document);
-            }
-            break;
-        case str_starts_with($document->getCollection(), 'bucket_'): // files
-            $parts = explode('_', $document->getCollection());
-            $bucketInternalId  = $parts[1];
-            $queueForStatsUsage
-                ->addMetric(METRIC_FILES, $value) // per project
-                ->addMetric(METRIC_FILES_STORAGE, $document->getAttribute('sizeOriginal') * $value) // per project
-                ->addMetric(str_replace('{bucketInternalId}', $bucketInternalId, METRIC_BUCKET_ID_FILES), $value) // per bucket
-                ->addMetric(str_replace('{bucketInternalId}', $bucketInternalId, METRIC_BUCKET_ID_FILES_STORAGE), $document->getAttribute('sizeOriginal') * $value); // per bucket
-            break;
-        case $document->getCollection() === 'functions':
-            $queueForStatsUsage
-                ->addMetric(METRIC_FUNCTIONS, $value); // per project
-
-            if ($event === Database::EVENT_DOCUMENT_DELETE) {
-                $queueForStatsUsage
-                    ->addReduce($document);
-            }
-            break;
-        case $document->getCollection() === 'deployments':
-            $queueForStatsUsage
-                ->addMetric(METRIC_DEPLOYMENTS, $value) // per project
-                ->addMetric(METRIC_DEPLOYMENTS_STORAGE, $document->getAttribute('size') * $value) // per project
-                ->addMetric(str_replace(['{resourceType}', '{resourceInternalId}'], [$document->getAttribute('resourceType'), $document->getAttribute('resourceInternalId')], METRIC_FUNCTION_ID_DEPLOYMENTS), $value) // per function
-                ->addMetric(str_replace(['{resourceType}', '{resourceInternalId}'], [$document->getAttribute('resourceType'), $document->getAttribute('resourceInternalId')], METRIC_FUNCTION_ID_DEPLOYMENTS_STORAGE), $document->getAttribute('size') * $value);
-            break;
         default:
             break;
     }

src/Appwrite/Platform/Workers/Builds.php

@@ -709,12 +709,8 @@ class Builds extends Action
             }
 
             $queueForStatsUsage
-                ->addMetric(METRIC_BUILDS, 1) // per project
-                ->addMetric(METRIC_BUILDS_STORAGE, $build->getAttribute('size', 0))
                 ->addMetric(METRIC_BUILDS_COMPUTE, (int)$build->getAttribute('duration', 0) * 1000)
                 ->addMetric(METRIC_BUILDS_MB_SECONDS, (int)(($spec['memory'] ?? APP_FUNCTION_MEMORY_DEFAULT) * $build->getAttribute('duration', 0) * ($spec['cpus'] ?? APP_FUNCTION_CPUS_DEFAULT)))
-                ->addMetric(str_replace('{functionInternalId}', $function->getInternalId(), METRIC_FUNCTION_ID_BUILDS), 1) // per function
-                ->addMetric(str_replace('{functionInternalId}', $function->getInternalId(), METRIC_FUNCTION_ID_BUILDS_STORAGE), $build->getAttribute('size', 0))
                 ->addMetric(str_replace('{functionInternalId}', $function->getInternalId(), METRIC_FUNCTION_ID_BUILDS_COMPUTE), (int)$build->getAttribute('duration', 0) * 1000)
                 ->addMetric(str_replace('{functionInternalId}', $function->getInternalId(), METRIC_FUNCTION_ID_BUILDS_MB_SECONDS), (int)(($spec['memory'] ?? APP_FUNCTION_MEMORY_DEFAULT) * $build->getAttribute('duration', 0) * ($spec['cpus'] ?? APP_FUNCTION_CPUS_DEFAULT)))
                 ->setProject($project)

src/Appwrite/Platform/Workers/Messaging.php

@@ -300,7 +300,6 @@ class Messaging extends Action
                             $errorTotal = \count($deliveryErrors);
                             $queueForStatsUsage
                                 ->setProject($project)
-                                ->addMetric(METRIC_MESSAGES, ($deliveredTotal + $errorTotal))
                                 ->addMetric(METRIC_MESSAGES_SENT, $deliveredTotal)
                                 ->addMetric(METRIC_MESSAGES_FAILED, $errorTotal)
                                 ->addMetric(str_replace('{type}', $provider->getAttribute('type'), METRIC_MESSAGES_TYPE), ($deliveredTotal + $errorTotal))

src/Appwrite/Platform/Workers/StatsUsageDump.php

@@ -4,7 +4,6 @@ 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\Platform\Action;
@@ -14,57 +13,10 @@ use Utopia\System\System;
 
 class StatsUsageDump extends Action
 {
-    public const METRIC_COLLECTION_LEVEL_STORAGE = 4;
-    public const METRIC_DATABASE_LEVEL_STORAGE = 3;
-    public const METRIC_PROJECT_LEVEL_STORAGE = 2;
     protected array $stats = [];
 
     protected Registry $register;
 
-    /**
-     * Metrics to skip writing to logsDB
-     * As these metrics are calculated separately
-     * by logs DB
-     * @var array
-     */
-    protected array $skipBaseMetrics = [
-        METRIC_DATABASES => true,
-        METRIC_BUCKETS => true,
-        METRIC_USERS => true,
-        METRIC_FUNCTIONS => true,
-        METRIC_TEAMS => true,
-        METRIC_MESSAGES => true,
-        METRIC_MAU => true,
-        METRIC_WEBHOOKS => true,
-        METRIC_PLATFORMS => true,
-        METRIC_PROVIDERS => true,
-        METRIC_TOPICS => true,
-        METRIC_KEYS => true,
-        METRIC_FILES => true,
-        METRIC_FILES_STORAGE => true,
-        METRIC_DEPLOYMENTS_STORAGE => true,
-        METRIC_BUILDS_STORAGE => true,
-        METRIC_DEPLOYMENTS => true,
-        METRIC_BUILDS => true,
-        METRIC_COLLECTIONS => true,
-        METRIC_DOCUMENTS => true,
-    ];
-
-    /**
-     * Skip metrics associated with parent IDs
-     * these need to be checked individually with `str_ends_with`
-     */
-    protected array $skipParentIdMetrics = [
-        '.files',
-        '.files.storage',
-        '.collections',
-        '.documents',
-        '.deployments',
-        '.deployments.storage',
-        '.builds',
-        '.builds.storage',
-    ];
-
     /**
      * @var callable
      */
@@ -127,20 +79,13 @@ class StatsUsageDump extends Action
             try {
                 /** @var \Utopia\Database\Database $dbForProject */
                 $dbForProject = $getProjectDB($project);
+                $documents = [];
+                $documentsClone = [];
                 foreach ($stats['keys'] ?? [] as $key => $value) {
                     if ($value == 0) {
                         continue;
                     }
 
-                    if (str_contains($key, METRIC_DATABASES_STORAGE)) {
-                        try {
-                            $this->handleDatabaseStorage($key, $dbForProject, $project, $receivedAt);
-                        } 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 = null;
 
@@ -157,191 +102,39 @@ class StatsUsageDump extends Action
                             'value' => $value,
                             'region' => System::getEnv('_APP_REGION', 'default'),
                         ]);
+                        $documents[] = $document;
 
                         $documentClone = new Document($document->getArrayCopy());
-
-                        $dbForProject->createOrUpdateDocumentsWithIncrease(
-                            'stats',
-                            'value',
-                            [$document]
-                        );
-
-                        $this->writeToLogsDB($project, $documentClone);
+                        $documentsClone[] = $documentClone;
                     }
                 }
+
+                $dbForProject->createOrUpdateDocumentsWithIncrease(
+                    'stats',
+                    'value',
+                    $documents
+                );
+                $this->writeToLogsDB($project, $documentsClone);
             } catch (\Exception $e) {
                 console::error('[' . DateTime::now() . '] project [' . $project->getInternalId() . '] database [' . $project['database'] . '] ' . ' ' . $e->getMessage());
             }
         }
     }
 
-    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) use ($receivedAt) {
-            $id = \md5("{$time}_{$period}_{$key}");
-
-            $document = new Document([
-                '$id' => $id,
-                'period' => $period,
-                'time' => $time,
-                'metric' => $key,
-                'value' => $value,
-                'region' => System::getEnv('_APP_REGION', 'default'),
-            ]);
-            $documentClone = new Document($document->getArrayCopy());
-            $dbForProject->createOrUpdateDocumentsWithIncrease(
-                'stats',
-                'value',
-                [$document]
-            );
-            $this->writeToLogsDB($project, $documentClone);
-        };
-
-        foreach ($this->periods as $period => $format) {
-            $time = null;
-
-            if ($period !== 'inf') {
-                $time = !empty($receivedAt) ? (new \DateTime($receivedAt))->format($format) : 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 self::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, $project, $diff, $key, $period, $time);
-
-                    // Update Database
-                    $databaseKey = str_replace(['{databaseInternalId}'], [$data[0]], METRIC_DATABASE_ID_STORAGE);
-                    $updateMetric($dbForProject, $project, $diff, $databaseKey, $period, $time);
-
-                    // Update Project
-                    $projectKey = METRIC_DATABASES_STORAGE;
-                    $updateMetric($dbForProject, $project, $diff, $projectKey, $period, $time);
-                    break;
-                    // Database Level
-                case self::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, $project, $diff, $databaseKey, $period, $time);
-
-                    // Update Project
-                    $projectKey = METRIC_DATABASES_STORAGE;
-                    $updateMetric($dbForProject, $project, $diff, $projectKey, $period, $time);
-                    break;
-                    // Project Level
-                case self::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, $project, $diff, $projectKey, $period, $time);
-                    break;
-            }
-        }
-
-        $end = microtime(true);
-
-        console::log('[' . DateTime::now() . '] DB Storage Calculation [' . $key . '] took ' . (($end - $start) * 1000) . ' milliseconds');
-    }
-
-    protected function writeToLogsDB(Document $project, Document $document): void
+    /**
+     * Write to logs DB
+     *
+     * @param Document $project
+     * @param array<Document> $documents
+     * @return void
+     */
+    protected function writeToLogsDB(Document $project, array $documents): void
     {
         if (System::getEnv('_APP_STATS_USAGE_DUAL_WRITING', 'disabled') === 'disabled') {
             Console::log('Dual Writing is disabled. Skipping...');
             return;
         }
 
-        if (array_key_exists($document->getAttribute('metric'), $this->skipBaseMetrics)) {
-            return;
-        }
-        foreach ($this->skipParentIdMetrics as $skipMetric) {
-            if (str_ends_with($document->getAttribute('metric'), $skipMetric)) {
-                return;
-            }
-        }
-
         /** @var \Utopia\Database\Database $dbForLogs*/
         $dbForLogs = call_user_func($this->getLogsDB, $project);
 
@@ -349,7 +142,7 @@ class StatsUsageDump extends Action
             $dbForLogs->createOrUpdateDocumentsWithIncrease(
                 'stats',
                 'value',
-                [$document]
+                $documents
             );
             Console::success('Usage logs pushed to Logs DB');
         } catch (\Throwable $th) {