feat: Scheduled Refresh REST API

Author: paveltiunov

packages/cubejs-api-gateway/index.js

@@ -212,6 +212,7 @@ class ApiGateway {
     this.apiSecret = apiSecret;
     this.compilerApi = compilerApi;
     this.adapterApi = adapterApi;
+    this.refreshScheduler = options.refreshScheduler;
     this.logger = logger;
     this.checkAuthMiddleware = options.checkAuthMiddleware || this.checkAuth.bind(this);
     this.checkAuthFn = options.checkAuth || this.defaultCheckAuth.bind(this);
@@ -254,6 +255,14 @@ class ApiGateway {
         res: this.resToResultFn(res)
       });
     }));
+
+    app.get(`${this.basePath}/v1/run-scheduled-refresh`, this.checkAuthMiddleware, (async (req, res) => {
+      await this.runScheduledRefresh({
+        queryingOptions: req.query.queryingOptions,
+        context: await this.contextByReq(req, req.authInfo, this.requestIdByReq(req)),
+        res: this.resToResultFn(res)
+      });
+    }));
   }
 
   initSubscriptionServer(sendMessage) {
@@ -264,6 +273,19 @@ class ApiGateway {
     return requestStarted && (new Date().getTime() - requestStarted.getTime());
   }
 
+  async runScheduledRefresh({ context, res, queryingOptions }) {
+    const requestStarted = new Date();
+    try {
+      const refreshScheduler = this.refreshScheduler();
+      await refreshScheduler.runScheduledRefresh(context, this.parseQueryParam(queryingOptions || {}));
+      res({}); // TODO status
+    } catch (e) {
+      this.handleError({
+        e, context, res, requestStarted
+      });
+    }
+  }
+
   async meta({ context, res }) {
     const requestStarted = new Date();
     try {

packages/cubejs-query-orchestrator/orchestrator/PreAggregations.js

@@ -109,7 +109,7 @@ class PreAggregationLoadCache {
     return this.versionEntries;
   }
 
-  async keyQueryResult(keyQuery, waitForRenew) {
+  async keyQueryResult(keyQuery, waitForRenew, priority) {
     if (!this.queryResults[this.queryCache.queryRedisKey(keyQuery)]) {
       this.queryResults[this.queryCache.queryRedisKey(keyQuery)] = await this.queryCache.cacheQueryResult(
         Array.isArray(keyQuery) ? keyQuery[0] : keyQuery,
@@ -119,7 +119,8 @@ class PreAggregationLoadCache {
         {
           renewalThreshold: this.queryCache.options.refreshKeyRenewalThreshold || 2 * 60,
           renewalKey: keyQuery,
-          waitForRenew
+          waitForRenew,
+          priority
         }
       );
     }
@@ -308,7 +309,7 @@ class PreAggregationLoader {
   getInvalidationKeyValues() {
     return Promise.all(
       (this.preAggregation.invalidateKeyQueries || [])
-        .map(keyQuery => this.loadCache.keyQueryResult(keyQuery, this.waitForRenew))
+        .map(keyQuery => this.loadCache.keyQueryResult(keyQuery, this.waitForRenew, this.priority(10)))
     );
   }
 

packages/cubejs-server-core/core/RefreshScheduler.js

@@ -73,7 +73,7 @@ class RefreshScheduler {
     }
   }
 
-  async ensurePreAggregationsRefreshed(context, queryingOptions) {
+  async runScheduledRefresh(context, queryingOptions) {
     queryingOptions = { timezone: 'UTC', ...queryingOptions };
     context = { requestId: `scheduler-${uuid()}`, ...context };
     this.serverCore.logger('Refresh Scheduler Run', {

packages/cubejs-server-core/core/index.js

@@ -281,7 +281,8 @@ class CubejsServerCore {
           checkAuthMiddleware: this.options.checkAuthMiddleware,
           checkAuth: this.options.checkAuth,
           queryTransformer: this.options.queryTransformer,
-          extendContext: this.options.extendContext
+          extendContext: this.options.extendContext,
+          refreshScheduler: () => new RefreshScheduler(this)
         }
       );
     }
@@ -368,7 +369,7 @@ class CubejsServerCore {
 
   async runScheduledRefresh(context, queryingOptions) {
     const scheduler = new RefreshScheduler(this);
-    await scheduler.ensurePreAggregationsRefreshed(context, queryingOptions);
+    await scheduler.runScheduledRefresh(context, queryingOptions);
   }
 
   async getDriver() {