2021-08-10 20:44:31 +12:00
< ? php
2021-08-11 22:15:00 +12:00
global $cli , $register ;
2021-08-10 20:44:31 +12:00
2021-08-13 20:39:46 +12:00
require_once __DIR__ . '/../init.php' ;
2021-08-10 20:44:31 +12:00
use Utopia\App ;
2021-08-13 20:39:46 +12:00
use Utopia\Cache\Adapter\Redis ;
2021-08-11 22:15:00 +12:00
use Utopia\Cache\Cache ;
2021-08-10 20:44:31 +12:00
use Utopia\CLI\Console ;
2021-08-11 22:15:00 +12:00
use Utopia\Database\Adapter\MariaDB ;
use Utopia\Database\Database ;
use Utopia\Database\Document ;
2021-08-17 17:45:07 +12:00
use Utopia\Database\Query ;
2021-08-11 22:15:00 +12:00
use Utopia\Database\Validator\Authorization ;
2021-08-10 20:44:31 +12:00
2021-08-19 18:54:32 +12:00
/**
* Metrics We collect
2021-08-19 20:01:22 +12:00
*
2021-08-19 18:54:32 +12:00
* requests
* network
* executions
* database . collections . create
* database . collections . read
* database . collections . update
* database . collections . delete
* database . documents . create
* database . documents . read
* database . documents . update
* database . documents . delete
* database . collections . { collectionId } . documents . create
* database . collections . { collectionId } . documents . read
* database . collections . { collectionId } . documents . update
* database . collections . { collectionId } . documents . delete
* storage . buckets . { bucketId } . files . create
* storage . buckets . { bucketId } . files . read
* storage . buckets . { bucketId } . files . update
* storage . buckets . { bucketId } . files . delete
* users . create
* users . read
* users . update
* users . delete
* users . sessions . create
* users . sessions . delete
2021-08-19 20:01:22 +12:00
*
2021-08-19 18:54:32 +12:00
* Counters
2021-08-19 20:01:22 +12:00
*
2021-08-19 18:54:32 +12:00
* users . count
2021-08-19 20:01:22 +12:00
* storage . files . count
* database . collections . count
* database . documents . count
* database . collections . { collectionId } . documents . count
*
2021-08-19 20:14:23 +12:00
* Totals
*
* storage . total
*
2021-08-19 18:54:32 +12:00
*/
2021-08-10 20:44:31 +12:00
$cli
-> task ( 'usage' )
-> desc ( 'Schedules syncing data from influxdb to Appwrite console db' )
2021-08-11 22:15:00 +12:00
-> action ( function () use ( $register ) {
2021-08-17 00:31:49 +12:00
Console :: title ( 'Usage Aggregation V1' );
Console :: success ( APP_NAME . ' usage aggregation process v1 has started' );
2021-08-10 20:44:31 +12:00
2021-08-15 20:41:19 +12:00
$interval = ( int ) App :: getEnv ( '_APP_USAGE_AGGREGATION_INTERVAL' , '30' ); //30 seconds
2021-08-16 18:58:34 +12:00
$periods = [
[
'key' => '30m' ,
'startTime' => '-24 hours' ,
],
[
'key' => '1d' ,
'startTime' => '-90 days' ,
],
];
2021-08-17 00:31:49 +12:00
// all the metrics that we are collecting at the moment
2021-08-16 18:58:34 +12:00
$globalMetrics = [
'requests' => [
'table' => 'appwrite_usage_requests_all' ,
],
'network' => [
'table' => 'appwrite_usage_network_all' ,
],
'executions' => [
'table' => 'appwrite_usage_executions_all' ,
],
'database.collections.create' => [
'table' => 'appwrite_usage_database_collections_create' ,
],
'database.collections.read' => [
'table' => 'appwrite_usage_database_collections_read' ,
],
'database.collections.update' => [
'table' => 'appwrite_usage_database_collections_update' ,
],
'database.collections.delete' => [
'table' => 'appwrite_usage_database_collections_delete' ,
],
'database.documents.create' => [
'table' => 'appwrite_usage_database_documents_create' ,
],
'database.documents.read' => [
'table' => 'appwrite_usage_database_documents_read' ,
],
'database.documents.update' => [
'table' => 'appwrite_usage_database_documents_update' ,
],
'database.documents.delete' => [
'table' => 'appwrite_usage_database_documents_delete' ,
],
2021-08-17 00:22:54 +12:00
'database.collections.collectionId.documents.create' => [
2021-08-16 18:58:34 +12:00
'table' => 'appwrite_usage_database_documents_create' ,
'groupBy' => 'collectionId' ,
],
2021-08-17 00:22:54 +12:00
'database.collections.collectionId.documents.read' => [
2021-08-16 18:58:34 +12:00
'table' => 'appwrite_usage_database_documents_read' ,
'groupBy' => 'collectionId' ,
],
2021-08-17 00:22:54 +12:00
'database.collections.collectionId.documents.update' => [
2021-08-16 18:58:34 +12:00
'table' => 'appwrite_usage_database_documents_update' ,
'groupBy' => 'collectionId' ,
],
2021-08-17 00:22:54 +12:00
'database.collections.collectionId.documents.delete' => [
2021-08-16 18:58:34 +12:00
'table' => 'appwrite_usage_database_documents_delete' ,
'groupBy' => 'collectionId' ,
],
2021-08-16 19:25:20 +12:00
'storage.buckets.bucketId.files.create' => [
'table' => 'appwrite_usage_storage_files_create' ,
'groupBy' => 'bucketId' ,
],
'storage.buckets.bucketId.files.read' => [
'table' => 'appwrite_usage_storage_files_read' ,
'groupBy' => 'bucketId' ,
],
'storage.buckets.bucketId.files.update' => [
'table' => 'appwrite_usage_storage_files_update' ,
'groupBy' => 'bucketId' ,
],
'storage.buckets.bucketId.files.delete' => [
'table' => 'appwrite_usage_storage_files_delete' ,
'groupBy' => 'bucketId' ,
],
2021-08-16 20:53:34 +12:00
'users.create' => [
'table' => 'appwrite_usage_users_create' ,
],
'users.read' => [
'table' => 'appwrite_usage_users_read' ,
],
'users.update' => [
'table' => 'appwrite_usage_users_update' ,
],
'users.delete' => [
'table' => 'appwrite_usage_users_delete' ,
],
'users.sessions.create' => [
'table' => 'appwrite_usage_users_sessions_create' ,
'groupBy' => 'provider' ,
],
'users.sessions.delete' => [
'table' => 'appwrite_usage_users_sessions_delete' ,
],
2021-08-16 18:58:34 +12:00
];
2021-08-13 21:45:46 +12:00
$attempts = 0 ;
$max = 10 ;
$sleep = 1 ;
2021-08-16 18:58:34 +12:00
do { // connect to db
2021-08-13 21:45:46 +12:00
try {
$attempts ++ ;
$db = $register -> get ( 'db' );
$redis = $register -> get ( 'cache' );
break ; // leave the do-while if successful
} catch ( \Exception $e ) {
Console :: warning ( " Database not ready. Retrying connection ( { $attempts } )... " );
if ( $attempts >= $max ) {
throw new \Exception ( 'Failed to connect to database: ' . $e -> getMessage ());
}
sleep ( $sleep );
}
} while ( $attempts < $max );
2021-08-10 20:44:31 +12:00
2021-08-15 20:38:31 +12:00
$cacheAdapter = new Cache ( new Redis ( $redis ));
2021-08-13 21:45:46 +12:00
$dbForProject = new Database ( new MariaDB ( $db ), $cacheAdapter );
2021-08-17 18:03:27 +12:00
$dbForConsole = new Database ( new MariaDB ( $db ), $cacheAdapter );
$dbForConsole -> setNamespace ( 'project_console_internal' );
2021-08-10 20:44:31 +12:00
2021-08-16 21:02:35 +12:00
$latestTime = [];
2021-08-11 22:15:00 +12:00
Authorization :: disable ();
2021-08-10 20:44:31 +12:00
2021-08-17 18:03:27 +12:00
$iterations = 0 ;
Console :: loop ( function () use ( $interval , $register , $dbForProject , $dbForConsole , $globalMetrics , $periods , & $latestTime , & $iterations ) {
2021-08-16 21:02:35 +12:00
$now = date ( 'd-m-Y H:i:s' , time ());
Console :: info ( " [ { $now } ] Aggregating usage data every { $interval } seconds " );
2021-08-11 22:15:00 +12:00
2021-08-16 20:53:34 +12:00
$loopStart = microtime ( true );
2021-08-13 20:39:46 +12:00
$client = $register -> get ( 'influxdb' );
2021-08-12 17:53:02 +12:00
if ( $client ) {
2021-08-16 18:58:34 +12:00
$database = $client -> selectDB ( 'telegraf' );
// sync data
2021-08-17 00:31:49 +12:00
foreach ( $globalMetrics as $metric => $options ) { //for each metrics
foreach ( $periods as $period ) { // aggregate data for each period
2021-08-16 18:58:34 +12:00
$start = DateTime :: createFromFormat ( 'U' , \strtotime ( $period [ 'startTime' ])) -> format ( DateTime :: RFC3339 );
2021-08-17 17:45:07 +12:00
if ( ! empty ( $latestTime [ $metric ][ $period [ 'key' ]])) {
2021-08-16 21:02:35 +12:00
$start = DateTime :: createFromFormat ( 'U' , $latestTime [ $metric ][ $period [ 'key' ]]) -> format ( DateTime :: RFC3339 );
}
2021-08-16 18:58:34 +12:00
$end = DateTime :: createFromFormat ( 'U' , \strtotime ( 'now' )) -> format ( DateTime :: RFC3339 );
2021-08-17 00:31:49 +12:00
$table = $options [ 'table' ]; //which influxdb table to query for this metric
$groupBy = empty ( $options [ 'groupBy' ]) ? '' : ', "' . $options [ 'groupBy' ] . '"' ; //some sub level metrics may be grouped by other tags like collectionId, bucketId, etc
2021-08-16 18:58:34 +12:00
2021-08-17 00:31:49 +12:00
$result = $database -> query ( 'SELECT sum(value) AS "value" FROM "' . $table . '" WHERE time > \'' . $start . '\' AND time < \'' . $end . '\' AND "metric_type"=\'counter\' GROUP BY time(' . $period [ 'key' ] . '), "projectId"' . $groupBy . ' FILL(null)' );
2021-08-16 18:58:34 +12:00
$points = $result -> getPoints ();
foreach ( $points as $point ) {
$projectId = $point [ 'projectId' ];
if ( ! empty ( $projectId ) && $projectId != 'console' ) {
$dbForProject -> setNamespace ( 'project_' . $projectId . '_internal' );
if ( ! empty ( $groupBy )) {
2021-08-17 18:03:27 +12:00
$groupedBy = $point [ $groupBy ] ? ? '' ;
2021-08-16 18:58:34 +12:00
if ( empty ( $groupedBy )) {
continue ;
}
$metric = str_replace ( $groupBy , $groupedBy , $metric );
}
$time = \strtotime ( $point [ 'time' ]);
2021-08-17 00:31:49 +12:00
$id = \md5 ( $time . '_' . $period [ 'key' ] . '_' . $metric ); //construct unique id for each metric using time, period and metric
2021-08-16 18:58:34 +12:00
$value = ( ! empty ( $point [ 'value' ])) ? $point [ 'value' ] : 0 ;
try {
$document = $dbForProject -> getDocument ( 'stats' , $id );
if ( $document -> isEmpty ()) {
$dbForProject -> createDocument ( 'stats' , new Document ([
'$id' => $id ,
'period' => $period [ 'key' ],
'time' => $time ,
'metric' => $metric ,
'value' => $value ,
'type' => 0 ,
]));
} else {
$dbForProject -> updateDocument ( 'stats' , $document -> getId (),
$document -> setAttribute ( 'value' , $value ));
}
2021-08-16 21:02:35 +12:00
$latestTime [ $metric ][ $period [ 'key' ]] = $time ;
2021-08-17 17:45:07 +12:00
} catch ( \Exception $e ) {
2021-08-17 00:31:49 +12:00
// if projects are deleted this might fail
2021-08-16 18:58:34 +12:00
Console :: warning ( " Failed to save data for project { $projectId } and metric { $metric } " );
}
}
}
}
2021-08-11 22:15:00 +12:00
}
}
2021-08-16 20:53:34 +12:00
2021-08-19 20:01:22 +12:00
if ( $iterations % 30 == 0 ) { //every 15 minutes
// aggregate number of objects in database
// get count of all the documents per collection -
// buckets will have the same
2021-08-17 18:03:27 +12:00
$latestProject = null ;
do {
$projects = $dbForConsole -> find ( 'projects' , [], 100 , orderAfter : $latestProject );
if ( ! empty ( $projects )) {
$latestProject = $projects [ array_key_last ( $projects )];
foreach ( $projects as $project ) {
$id = $project -> getId ();
2021-08-19 20:14:23 +12:00
// get total storage
$dbForProject -> setNamespace ( 'project_' . $id . '_internal' );
$storageTotal = $dbForProject -> sum ( 'files' , 'sizeOriginal' ) + $dbForProject -> sum ( 'tags' , 'size' );
$dbForProject -> createDocument ( 'stats' , new Document ([
'$id' => $dbForProject -> getId (),
'period' => '15m' ,
'time' => time (),
'metric' => 'storage.total' ,
'value' => $storageTotal ,
'type' => 1 ,
]));
2021-08-19 20:01:22 +12:00
$collections = [
'users' => [
'namespace' => 'internal' ,
],
'collections' => [
'metricPrefix' => 'database' ,
2021-08-19 20:58:10 +12:00
'namespace' => 'internal' ,
2021-08-19 20:01:22 +12:00
'subCollections' => [
'documents' => [
'namespace' => 'external' ,
],
],
],
'files' => [
'metricPrefix' => 'storage' ,
'namespace' => 'internal' ,
],
];
2021-08-17 18:03:27 +12:00
foreach ( $collections as $collection => $options ) {
2021-08-19 21:02:38 +12:00
try {
$dbForProject -> setNamespace ( " project_ { $id } _ { $options [ 'namespace' ] } " );
$count = $dbForProject -> count ( $collection );
$dbForProject -> setNamespace ( " project_ { $id } _internal " );
$metricPrefix = $options [ 'metricPrefix' ] ? ? '' ;
$metric = empty ( $metricPrefix ) ? " { $collection } .count " : " { $metricPrefix } . { $collection } .count " ;
$dbForProject -> createDocument ( 'stats' , new Document ([
'$id' => $dbForProject -> getId (),
'time' => time (),
'period' => '15m' ,
'metric' => $metric ,
'value' => $count ,
'type' => 1 ,
]));
$subCollections = $options [ 'subCollections' ] ? ? [];
if ( ! empty ( $subCollections )) {
$latestParent = null ;
$subCollectionCounts = []; //total project level count of sub collections
do {
$dbForProject -> setNamespace ( " project_ { $id } _ { $options [ 'namespace' ] } " );
$parents = $dbForProject -> find ( $collection , [], 100 , orderAfter : $latestParent );
if ( ! empty ( $parents )) {
$latestParent = $parents [ array_key_last ( $parents )];
foreach ( $parents as $parent ) {
foreach ( $subCollections as $subCollection => $subOptions ) {
$dbForProject -> setNamespace ( " project_ { $id } _ { $subOptions [ 'namespace' ] } " );
$count = $dbForProject -> count ( $parent -> getId ());
$subCollectionsCounts [ $subCollection ] = ( $subCollectionCounts [ $subCollection ] ? ? 0 ) + $count ;
$dbForProject -> setNamespace ( " project_ { $id } _internal " );
$dbForProject -> createDocument ( 'stats' , new Document ([
'$id' => $dbForProject -> getId (),
'time' => time (),
'period' => '15m' ,
'metric' => empty ( $metricPrefix ) ? " { $collection } . { $parent -> getId () } . { $subCollection } .count " : " { $metricPrefix } . { $collection } . { $parent -> getId () } . { $subCollection } .count " ,
'value' => $count ,
'type' => 1 ,
]));
}
2021-08-19 20:01:22 +12:00
}
}
2021-08-19 21:02:38 +12:00
} while ( ! empty ( $parents ));
foreach ( $subCollectionsCounts as $subCollection => $count ) {
$dbForProject -> setNamespace ( " project_ { $id } _internal " );
$dbForProject -> createDocument ( 'stats' , new Document ([
'$id' => $dbForProject -> getId (),
'time' => time (),
'period' => '15m' ,
'metric' => empty ( $metricPrefix ) ? " { $subCollection } .count " : " { $metricPrefix } . { $subCollection } .count " ,
'value' => $count ,
'type' => 1 ,
]));
2021-08-19 20:01:22 +12:00
}
}
2021-08-19 21:02:38 +12:00
} catch ( \Exception $e ) {
Console :: warning ( " Failed to save database counters data for project { $collection } " );
2021-08-19 20:01:22 +12:00
}
2021-08-17 18:03:27 +12:00
}
2021-08-17 17:45:07 +12:00
}
}
2021-08-17 18:03:27 +12:00
} while ( ! empty ( $projects ));
}
$iterations ++ ;
2021-08-17 17:45:07 +12:00
$loopTook = microtime ( true ) - $loopStart ;
$now = date ( 'd-m-Y H:i:s' , time ());
2021-08-17 18:03:27 +12:00
Console :: info ( " [ { $now } ] Aggregation took { $loopTook } seconds " );
}, $interval );
2021-08-10 20:44:31 +12:00
});