updates

2024-06-29 11:40:45 +12:00 · 2022-12-04 19:06:23 +02:00 · 2022-12-04 19:06:23 +02:00 · b71bba2a9c
parent d3bc4da15c
commit b71bba2a9c
10 changed files with 142 additions and 63 deletions
--- a/3
+++ b/3
@ -323,7 +323,8 @@ RUN chmod +x /usr/local/bin/doctor && \
    chmod +x /usr/local/bin/worker-builds && \
    chmod +x /usr/local/bin/worker-mails && \
    chmod +x /usr/local/bin/worker-messaging && \
-    chmod +x /usr/local/bin/worker-webhooks
+    chmod +x /usr/local/bin/worker-webhooks && \
+    chmod +x /usr/local/bin/worker-usage

 # Letsencrypt Permissions
 RUN mkdir -p /etc/letsencrypt/live/ && chmod -Rf 755 /etc/letsencrypt/live/
--- a/app/controllers/general.php
+++ b/app/controllers/general.php
@ -632,7 +632,6 @@ App::get('/.well-known/acme-challenge')

 include_once __DIR__ . '/shared/api.php';
 include_once __DIR__ . '/shared/api/auth.php';
-include_once __DIR__ . '/shared/api/cache.php';

 foreach (Config::getParam('services', []) as $service) {
    include_once $service['controller'];
--- a/app/controllers/shared/api.php
+++ b/app/controllers/shared/api.php
@ -48,64 +48,64 @@ $parseLabel = function (string $label, array $responsePayload, array $requestPar
    return $label;
 };

-$databaseListener = function (string $event, Document $document, Document $project, Usage $usage) {
+$databaseListener = function (string $event, Document $document, Document $project, Usage $queueForUsage) {
    $value = 1;

    if ($event === Database::EVENT_DOCUMENT_DELETE) {
        $value = -1;
    }
-
-    switch ($document->getCollection()) {
-        case 'users':
-            $usage->addMetric("{$project->getId()}", "users", $value); // per project
+    var_dump($document->getCollection());
+    switch (true) {
+        case $document->getCollection() === 'users':
+            $queueForUsage->addMetric("{$project->getId()}", "users", $value); // per project
            break;
-        case 'teams':
-            $usage->addMetric("{$project->getId()}", "teams", $value); // per project
+        case $document->getCollection() === 'teams':
+            $queueForUsage->addMetric("{$project->getId()}", "teams", $value); // per project
            break;
-        case 'sessions':
-            $usage->addMetric("{$project->getId()}", "sessions", $value); // per project
+        case $document->getCollection() === 'sessions':
+            $queueForUsage->addMetric("{$project->getId()}", "sessions", $value); // per project
            break;
-        case 'databases':
-            $usage->addMetric("{$project->getId()}", "databases", $value); // per project
+        case $document->getCollection() === 'databases':
+            $queueForUsage->addMetric("{$project->getId()}", "databases", $value); // per project
            break;
-        case 'collections':
-            $usage->addMetric("{$project->getId()}.[DATABASE_ID]", "collections", $value); // per database
-            $usage->addMetric("{$project->getId()}", "collections", $value); // per project
+        case str_starts_with($document->getCollection(), 'database'): // collections
+            $queueForUsage->addMetric("{$project->getId()}.{$document['databaseId']}", "collections", $value); // per database
+            $queueForUsage->addMetric("{$project->getId()}", "collections", $value); // per project
            break;
-        case 'documents':
-            $usage->addMetric("{$project->getId()}.[DATABASE_ID].[COLLECTION_ID]", "documents", $value);  // per collection
-            $usage->addMetric("{$project->getId()}.[DATABASE_ID]", "documents", $value);  // per database
-            $usage->addMetric("{$project->getId()}", "documents", $value); // per project
+        case $document->getCollection() === 'documents':
+            $queueForUsage->addMetric("{$project->getId()}.{$document['databaseId']}.{$document['collectionId']}", "documents", $value);  // per collection
+            $queueForUsage->addMetric("{$project->getId()}.{$document['databaseId']}", "documents", $value);  // per database
+            $queueForUsage->addMetric("{$project->getId()}", "documents", $value); // per project
            break;
-        case 'buckets':
-            $usage->addMetric("{$project->getId()}", "buckets", $value); // per project
+        case $document->getCollection() === 'buckets':
+            $queueForUsage->addMetric("{$project->getId()}", "buckets", $value); // per project
            break;
-        case 'files':
-            $usage->addMetric("{$project->getId()}.[BUCKET_ID]", "files", $value); // per bucket
-            $usage->addMetric("{$project->getId()}.[BUCKET_ID]", "files.storage", $document->getAttribute('sizeOriginal') * $value); // per bucket
-            $usage->addMetric("{$project->getId()}", "files", $value); // per project
-            $usage->addMetric("{$project->getId()}", "files.storage", $document->getAttribute('sizeOriginal') * $value); // per project
+        case $document->getCollection() === 'files':
+            $queueForUsage->addMetric("{$project->getId()}.{$document['bucketId']}", "files", $value); // per bucket
+            $queueForUsage->addMetric("{$project->getId()}.{$document['bucketId']}", "files.storage", $document->getAttribute('sizeOriginal') * $value); // per bucket
+            $queueForUsage->addMetric("{$project->getId()}", "files", $value); // per project
+            $queueForUsage->addMetric("{$project->getId()}", "files.storage", $document->getAttribute('sizeOriginal') * $value); // per project
            break;
-        case 'functions':
-            $usage->addMetric("{$project->getId()}", "functions", $value); // per project
+        case $document->getCollection() === 'functions':
+            $queueForUsage->addMetric("{$project->getId()}", "functions", $value); // per project
            break;
-        case 'deployments':
-            $usage->addMetric("{$project->getId()}.[FUNCTION_ID]", "deployments", $value); // per function
-            $usage->addMetric("{$project->getId()}.[FUNCTION_ID]", "deployments.storage", $document->getAttribute('size') * $value); // per function
-            $usage->addMetric("{$project->getId()}", "deployments", $value); // per project
-            $usage->addMetric("{$project->getId()}", "deployments.storage", $document->getAttribute('size') * $value); // per project
+        case $document->getCollection() === 'deployments':
+            $queueForUsage->addMetric("{$project->getId()}.{$document['functionId']}", "deployments", $value); // per function
+            $queueForUsage->addMetric("{$project->getId()}.{$document['functionId']}", "deployments.storage", $document->getAttribute('size') * $value); // per function
+            $queueForUsage->addMetric("{$project->getId()}", "deployments", $value); // per project
+            $queueForUsage->addMetric("{$project->getId()}", "deployments.storage", $document->getAttribute('size') * $value); // per project
            break;
-        case 'builds':
-            $usage->addMetric("{$project->getId()}.[FUNCTION_ID]", "builds", $value); // per function
-            $usage->addMetric("{$project->getId()}.[FUNCTION_ID]", "builds.storage", $document->getAttribute('size') * $value); // per function
-            $usage->addMetric("{$project->getId()}", "builds", $value); // per project
-            $usage->addMetric("{$project->getId()}", "builds.storage", $document->getAttribute('size') * $value); // per project
+        case $document->getCollection() === 'builds':
+            $queueForUsage->addMetric("{$project->getId()}.{$document['functionId']}", "builds", $value); // per function
+            $queueForUsage->addMetric("{$project->getId()}.{$document['functionId']}", "builds.storage", $document->getAttribute('size') * $value); // per function
+            $queueForUsage->addMetric("{$project->getId()}", "builds", $value); // per project
+            $queueForUsage->addMetric("{$project->getId()}", "builds.storage", $document->getAttribute('size') * $value); // per project
            break;
-        case 'executions':
-            $usage->addMetric("{$project->getId()}.[FUNCTION_ID]", "executions", $value); // per function
-            $usage->addMetric("{$project->getId()}.[FUNCTION_ID]", "executions.compute", $document->getAttribute('duration') * $value); // per function
-            $usage->addMetric("{$project->getId()}", "executions", $value); // per project
-            $usage->addMetric("{$project->getId()}", "executions.compute", $document->getAttribute('duration') * $value); // per project
+        case $document->getCollection() === 'executions':
+            $queueForUsage->addMetric("{$project->getId()}.{$document['functionId']}", "executions", $value); // per function
+            $queueForUsage->addMetric("{$project->getId()}.{$document['functionId']}", "executions.compute", $document->getAttribute('duration') * $value); // per function
+            $queueForUsage->addMetric("{$project->getId()}", "executions", $value); // per project
+            $queueForUsage->addMetric("{$project->getId()}", "executions.compute", $document->getAttribute('duration') * $value); // per project
            break;
        default:
            // if (strpos($collection, 'bucket_') === 0) {
@ -141,8 +141,9 @@ App::init()
    ->inject('deletes')
    ->inject('database')
    ->inject('dbForProject')
+    ->inject('queueForUsage')
    ->inject('mode')
-    ->action(function (App $utopia, Request $request, Response $response, Document $project, Document $user, Event $events, Audit $audits, Mail $mails, Delete $deletes, EventDatabase $database, Database $dbForProject, string $mode) use ($databaseListener) {
+    ->action(function (App $utopia, Request $request, Response $response, Document $project, Document $user, Event $events, Audit $audits, Mail $mails, Delete $deletes, EventDatabase $database, Database $dbForProject, Usage $queueForUsage, string $mode) use ($databaseListener) {

        $route = $utopia->match($request);

@ -240,8 +241,8 @@ App::init()
        $database->setProject($project);

        $dbForProject
-            ->on(Database::EVENT_DOCUMENT_CREATE, fn ($event, Document $document) => $databaseListener($event, $document))
-            ->on(Database::EVENT_DOCUMENT_DELETE, fn ($event, Document $document) => $databaseListener($event, $document))
+            ->on(Database::EVENT_DOCUMENT_CREATE, fn ($event, Document $document) => $databaseListener($event, $document, $project, $queueForUsage))
+            ->on(Database::EVENT_DOCUMENT_DELETE, fn ($event, Document $document) => $databaseListener($event, $document, $project, $queueForUsage))
        ;

        $useCache = $route->getLabel('cache', false);
@ -315,7 +316,8 @@ App::shutdown()
    ->inject('database')
    ->inject('dbForProject')
    ->inject('queueForFunctions')
-    ->action(function (App $utopia, Request $request, Response $response, Document $project, Event $events, Audit $audits, Delete $deletes, EventDatabase $database, Database $dbForProject, Func $queueForFunctions) use ($parseLabel) {
+    ->inject('queueForUsage')
+    ->action(function (App $utopia, Request $request, Response $response, Document $project, Event $events, Audit $audits, Delete $deletes, EventDatabase $database, Database $dbForProject, Func $queueForFunctions, Usage $queueForUsage) use ($parseLabel) {

        $responsePayload = $response->getPayload();

@ -472,18 +474,17 @@ App::shutdown()
            }
        }

-        if (
-            App::getEnv('_APP_USAGE_STATS', 'enabled') == 'enabled'
-            && $project->getId()
-            && !empty($route->getLabel('sdk.namespace', null))
-        ) { // Don't calculate console usage on admin mode
-            
+        if ($project->getId() && !empty($route->getLabel('sdk.namespace', null))) {
            $fileSize = 0;
            $file = $request->getFiles('file');
            if (!empty($file)) {
                $fileSize = (\is_array($file['size']) && isset($file['size'][0])) ? $file['size'][0] : $file['size'];
            }

+            $queueForUsage->addMetric("{$project->getId()}", "network.inbound", $request->getSize() + $fileSize);
+            $queueForUsage->addMetric("{$project->getId()}", "network.outbound", $response->getSize());
+            $queueForUsage->trigger();
+
            // $usage
            //     ->setParam('project.{scope}.network.inbound', $request->getSize() + $fileSize)
            //     ->setParam('project.{scope}.network.outbound', $response->getSize())
--- a/app/init.php
+++ b/app/init.php
@ -18,6 +18,7 @@ ini_set('display_startup_errors', 1);
 ini_set('default_socket_timeout', -1);
 error_reporting(E_ALL);

+use Appwrite\Event\Usage;
 use Appwrite\Extend\Exception;
 use Appwrite\Auth\Auth;
 use Appwrite\SMS\Adapter\Mock;
@ -835,6 +836,9 @@ App::setResource('queue', function (Group $pools) {
 App::setResource('queueForFunctions', function (Connection $queue) {
    return new Func($queue);
 }, ['queue']);
+App::setResource('queueForUsage', function (Connection $queue) {
+    return new Usage($queue);
+}, ['queue']);
 App::setResource('clients', function ($request, $console, $project) {
    $console->setAttribute('platforms', [ // Always allow current host
        '$collection' => ID::custom('platforms'),
--- a/app/worker.php
+++ b/app/worker.php
@ -100,7 +100,7 @@ Server::setResource('pools', function ($register) {
 $pools = $register->get('pools');
 $connection = $pools->get('queue')->pop()->getResource();
 $workerNumber = swoole_cpu_num() * intval(App::getEnv('_APP_WORKER_PER_CORE', 6));
-
+$workerNumber = 1;
 if (empty(App::getEnv('QUEUE'))) {
    throw new Exception('Please configure "QUEUE" environemnt variable.');
 }
--- a/app/workers/usage.php
+++ b/app/workers/usage.php
@ -0,0 +1,34 @@
+<?php
+
+require_once __DIR__ . '/../worker.php';
+
+use Swoole\Timer;
+use Utopia\Queue\Message;
+
+$stack = [];
+
+$server->job()
+    ->inject('message')
+    ->action(function (Message $message) use (&$stack) {
+        $payload = $message->getPayload() ?? [];
+        foreach ($payload['metrics'] ?? [] as $metric) {
+            if (!isset($stack[$metric['namespace']][$metric['key']])) {
+                $stack[$metric['namespace']][$metric['key']] = $metric['value'];
+                continue;
+            }
+            $stack[$metric['namespace']][$metric['key']] += $metric['value'];
+        }
+    });
+
+$server
+    ->workerStart()
+    ->action(function () use (&$stack) {
+        Timer::tick(30000, function () use (&$stack) {
+            $chunk = array_slice($stack, 0, count($stack));
+            array_splice($stack, 0, count($stack));
+            var_dump($chunk);
+
+        });
+    });
+
+$server->start();
--- a/bin/worker-usage
+++ b/bin/worker-usage
@ -0,0 +1,3 @@
+#!/bin/sh
+
+QUEUE=v1-usage php /usr/src/code/app/workers/usage.php $@
--- a/docker-compose.yml
+++ b/docker-compose.yml
@ -93,6 +93,7 @@ services:
      - ./public:/usr/src/code/public
      - ./src:/usr/src/code/src
      - ./dev:/usr/local/dev
+      - ./vendor/utopia-php/database:/usr/src/code/vendor/utopia-php/database

    depends_on:
      - mariadb
@ -550,6 +551,43 @@ services:
      - _APP_LOGGING_PROVIDER
      - _APP_LOGGING_CONFIG

+  appwrite-worker-usage:
+    entrypoint: worker-usage
+    <<: *x-logging
+    container_name: appwrite-worker-usage
+    image: appwrite-dev
+    networks:
+      - appwrite
+    volumes:
+      - ./app:/usr/src/code/app
+      - ./src:/usr/src/code/src
+      - ./vendor/utopia-php/database:/usr/src/code/vendor/utopia-php/database
+    depends_on:
+      - redis
+      - mariadb
+    environment:
+      - _APP_ENV
+      - _APP_WORKER_PER_CORE
+      - _APP_CONNECTIONS_MAX
+      - _APP_POOL_CLIENTS
+      - _APP_OPENSSL_KEY_V1
+      - _APP_DB_HOST
+      - _APP_DB_PORT
+      - _APP_DB_SCHEMA
+      - _APP_DB_USER
+      - _APP_DB_PASS
+      - _APP_REDIS_HOST
+      - _APP_REDIS_PORT
+      - _APP_REDIS_USER
+      - _APP_REDIS_PASS
+      - _APP_CONNECTIONS_DB_CONSOLE
+      - _APP_CONNECTIONS_DB_PROJECT
+      - _APP_CONNECTIONS_CACHE
+      - _APP_CONNECTIONS_QUEUE
+      - _APP_USAGE_STATS
+      - DOCKERHUB_PULL_USERNAME
+      - DOCKERHUB_PULL_PASSWORD
+
  appwrite-maintenance:
    entrypoint: maintenance
    <<: *x-logging
--- a/src/Appwrite/Event/Event.php
+++ b/src/Appwrite/Event/Event.php
@ -23,6 +23,9 @@ class Event
    public const FUNCTIONS_QUEUE_NAME = 'v1-functions';
    public const FUNCTIONS_CLASS_NAME = 'FunctionsV1';

+    public const USAGE_QUEUE_NAME = 'v1-usage';
+    public const USAGE_CLASS_NAME = 'UsageV1';
+
    public const WEBHOOK_QUEUE_NAME = 'v1-webhooks';
    public const WEBHOOK_CLASS_NAME = 'WebhooksV1';

--- a/src/Appwrite/Event/Usage.php
+++ b/src/Appwrite/Event/Usage.php
@ -11,11 +11,11 @@ class Usage extends Event

    public function __construct(protected Connection $connection)
    {
-        parent::__construct(Event::FUNCTIONS_QUEUE_NAME, Event::FUNCTIONS_CLASS_NAME);
+        parent::__construct(Event::USAGE_QUEUE_NAME, Event::USAGE_CLASS_NAME);
    }

    /**
-     * Sets function document for the function event.
+     * Add metric.
     *
     * @param string $namespace
     * @param string $key
@ -34,19 +34,15 @@ class Usage extends Event
    }

    /**
-     * Executes the function event and sends it to the functions worker.
+     * Sends metrics to the usage worker.
     *
-     * @return bool
+     * @return string|bool
     */
    public function trigger(): string|bool
    {
        $client = new Client($this->queue, $this->connection);

        return $client->enqueue([
-            'project' => $this->project,
-            'user' => $this->user,
-            'type' => $this->type,
-            'payload' => $this->payload,
            'metrics' => $this->metrics,
        ]);
    }