feat: Cube.js agent

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

@@ -0,0 +1,50 @@
+const fetch = require('node-fetch');
+const crypto = require('crypto');
+
+let flushPromise = null;
+let trackEvents = [];
+
+module.exports = async (event, endpointUrl, logger) => {
+  trackEvents.push({
+    ...event,
+    id: crypto.randomBytes(16).toString('hex'),
+    timestamp: new Date().toJSON()
+  });
+  const flush = async (toFlush, retries) => {
+    if (!toFlush) {
+      toFlush = trackEvents;
+      trackEvents = [];
+    }
+    if (!toFlush.length) {
+      return null;
+    }
+    if (retries == null) {
+      retries = 10;
+    }
+    try {
+      const sentAt = new Date().toJSON();
+      const result = await fetch(endpointUrl, {
+        method: 'post',
+        body: JSON.stringify(toFlush.map(r => ({ ...r, sentAt }))),
+        headers: { 'Content-Type': 'application/json' },
+      });
+      if (result.status !== 200 && retries > 0) {
+        return flush(toFlush, retries - 1);
+      }
+      // console.log(await result.json());
+    } catch (e) {
+      if (retries > 0) {
+        return flush(toFlush, retries - 1);
+      }
+      logger('Agent Error', { error: (e.stack || e).toString() });
+    }
+    return null;
+  };
+  const currentPromise = (flushPromise || Promise.resolve()).then(() => flush()).then(() => {
+    if (currentPromise === flushPromise) {
+      flushPromise = null;
+    }
+  });
+  flushPromise = currentPromise;
+  return flushPromise;
+};

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

@@ -12,6 +12,7 @@ const RefreshScheduler = require('./RefreshScheduler');
 const FileRepository = require('./FileRepository');
 const DevServer = require('./DevServer');
 const track = require('./track');
+const agentCollect = require('./agentCollect');
 
 const DriverDependencies = {
   postgres: '@cubejs-backend/postgres-driver',
@@ -255,6 +256,8 @@ class CubejsServerCore {
       }
     };
 
+    this.initAgent();
+
     if (this.options.devServer) {
       this.devServer = new DevServer(this);
       const oldLogger = this.logger;
@@ -311,6 +314,23 @@ class CubejsServerCore {
     }
   }
 
+  initAgent() {
+    if (process.env.CUBEJS_AGENT_ENDPOINT_URL) {
+      const oldLogger = this.logger;
+      this.logger = (msg, params) => {
+        oldLogger(msg, params);
+        agentCollect(
+          {
+            msg,
+            ...params
+          },
+          process.env.CUBEJS_AGENT_ENDPOINT_URL,
+          oldLogger
+        );
+      };
+    }
+  }
+
   static create(options) {
     return new CubejsServerCore(options);
   }