Merge b88fe81 into f1b68a3

kube-HPC · Apr 5, 2020 · 8b27194 · 8b27194
2 parents f1b68a3 + b88fe81
commit 8b27194
Show file tree

Hide file tree

Showing 13 changed files with 543 additions and 468 deletions.
diff --git a/lib/storage-manager.js b/lib/storage-manager.js
@@ -1,3 +1,4 @@
+const { Encoding } = require('@hkube/encoding');
 const { STORAGE_PREFIX } = require('../consts/storage-prefix');
 const StorageBase = require('./storage/storage-base');
 const Hkube = require('./storage/hkube');
@@ -23,15 +24,16 @@ class StorageManager {
             this.prefixesTypes = Object.values(STORAGE_PREFIX).map(x => `${config.clusterName}-${x}`);
             await this._adapter.init(storage.connection, this.prefixesTypes, bootstrap);
             this._wasInit = true;
-            this.storage = new StorageBase(this._adapter, log, config);
-            this.hkube = new Hkube(this._adapter, log, config);
-            this.hkubeResults = new HkubeResults(this._adapter, log, config);
-            this.hkubeStore = new HkubeStore(this._adapter, log, config);
-            this.hkubeMetadata = new HkubeMetadata(this._adapter, log, config);
-            this.hkubeExecutions = new HkubeExecutions(this._adapter, log, config);
-            this.hkubeIndex = new HkubeIndex(this._adapter, log, config);
-            this.hkubeBuilds = new HkubeBuilds(this._adapter, log, config);
-            this.hkubeAlgoMetrics = new HkubeAlgoMetrics(this._adapter, log, config);
+            const encoding = new Encoding({ type: storage.encoding });
+            this.storage = new StorageBase(this._adapter, log, encoding);
+            this.hkube = new Hkube(this._adapter, log, config, encoding);
+            this.hkubeResults = new HkubeResults(this._adapter, log, config, encoding);
+            this.hkubeStore = new HkubeStore(this._adapter, log, config, encoding);
+            this.hkubeMetadata = new HkubeMetadata(this._adapter, log, config, encoding);
+            this.hkubeExecutions = new HkubeExecutions(this._adapter, log, config, encoding);
+            this.hkubeIndex = new HkubeIndex(this._adapter, log, config, encoding);
+            this.hkubeBuilds = new HkubeBuilds(this._adapter, log, config, encoding);
+            this.hkubeAlgoMetrics = new HkubeAlgoMetrics(this._adapter, log, config, encoding);
         }
     }
 

diff --git a/lib/storage/hkube-algo-metrics.js b/lib/storage/hkube-algo-metrics.js
@@ -3,8 +3,8 @@ const StorageBase = require('./storage-base');
 const { STORAGE_PREFIX } = require('../../consts/storage-prefix');
 
 class HkubeAlgoMetrics extends StorageBase {
-    constructor(adapter, log, config) {
-        super(adapter, log);
+    constructor(adapter, log, config, encoding) {
+        super(adapter, log, encoding);
         this.clusterName = config.clusterName;
     }
 

diff --git a/lib/storage/hkube-builds.js b/lib/storage/hkube-builds.js
@@ -3,48 +3,52 @@ const StorageBase = require('./storage-base');
 const { STORAGE_PREFIX } = require('../../consts/storage-prefix');
 
 class HkubeBuilds extends StorageBase {
-    constructor(adapter, log, config) {
-        super(adapter, log);
+    constructor(adapter, log, config, encoding) {
+        super(adapter, log, encoding);
         this.clusterName = config.clusterName;
     }
 
     async put({ buildId, data }, tracer) {
         return super.put({
-            path: path.join(this.clusterName + '-' + STORAGE_PREFIX.HKUBE_BUILDS, buildId),
+            path: this.createPath({ buildId }),
             data
         }, tracer);
     }
 
     async get({ buildId }, tracer) {
         return super.get({
-            path: path.join(this.clusterName + '-' + STORAGE_PREFIX.HKUBE_BUILDS, buildId)
+            path: this.createPath({ buildId })
         }, tracer);
     }
 
     async list({ buildId }) {
         return super.list({
-            path: path.join(this.clusterName + '-' + STORAGE_PREFIX.HKUBE_BUILDS, buildId)
+            path: this.createPath({ buildId })
         });
     }
 
     async delete({ buildId }) {
         return super.delete({
-            path: path.join(this.clusterName + '-' + STORAGE_PREFIX.HKUBE_BUILDS, buildId)
+            path: this.createPath({ buildId })
         });
     }
 
     async getStream({ buildId }) {
         return super.getStream({
-            path: path.join(this.clusterName + '-' + STORAGE_PREFIX.HKUBE_BUILDS, buildId)
+            path: this.createPath({ buildId })
         });
     }
 
     async putStream({ buildId, data }) {
         return super.putStream({
-            path: path.join(this.clusterName + '-' + STORAGE_PREFIX.HKUBE_BUILDS, buildId),
+            path: this.createPath({ buildId }),
             data
         });
     }
+
+    createPath({ buildId }) {
+        return path.join(this.clusterName + '-' + STORAGE_PREFIX.HKUBE_BUILDS, buildId);
+    }
 }
 
 module.exports = HkubeBuilds;
diff --git a/lib/storage/hkube-execution.js b/lib/storage/hkube-execution.js
@@ -4,35 +4,39 @@ const { STORAGE_PREFIX } = require('../../consts/storage-prefix');
 const DESCRIPTOR_JSON = 'descriptor.json';
 
 class StorageHkubeExecution extends StorageBase {
-    constructor(adapter, log, config) {
-        super(adapter, log);
+    constructor(adapter, log, config, encoding) {
+        super(adapter, log, encoding);
         this.clusterName = config.clusterName;
     }
 
     async put({ jobId, data }, tracer) {
         return super.put({
-            path: path.join(this.clusterName + '-' + STORAGE_PREFIX.HKUBE_EXECUTION, jobId, DESCRIPTOR_JSON),
+            path: this.createPath({ jobId, name: DESCRIPTOR_JSON }),
             data
         }, tracer);
     }
 
     async get({ jobId }, tracer) {
         return super.get({
-            path: path.join(this.clusterName + '-' + STORAGE_PREFIX.HKUBE_EXECUTION, jobId, DESCRIPTOR_JSON)
+            path: this.createPath({ jobId, name: DESCRIPTOR_JSON })
         }, tracer);
     }
 
     async list({ jobId }) {
         return super.list({
-            path: path.join(this.clusterName + '-' + STORAGE_PREFIX.HKUBE_EXECUTION, jobId)
+            path: this.createPath({ jobId })
         });
     }
 
     async delete({ jobId }) {
         return super.delete({
-            path: path.join(this.clusterName + '-' + STORAGE_PREFIX.HKUBE_EXECUTION, jobId)
+            path: this.createPath({ jobId })
         });
     }
+
+    createPath({ jobId, name = '' }) {
+        return path.join(this.clusterName + '-' + STORAGE_PREFIX.HKUBE_EXECUTION, jobId, name);
+    }
 }
 
 module.exports = StorageHkubeExecution;
diff --git a/lib/storage/hkube-index.js b/lib/storage/hkube-index.js
@@ -4,28 +4,28 @@ const StorageBase = require('./storage-base');
 const { STORAGE_PREFIX } = require('../../consts/storage-prefix');
 
 class StorageHkubeIndex extends StorageBase {
-    constructor(adapter, log, config) {
-        super(adapter, log);
+    constructor(adapter, log, config, encoding) {
+        super(adapter, log, encoding);
         this.clusterName = config.clusterName;
     }
 
     async put({ jobId }, tracer) {
         return super.put({
-            path: path.join(this.clusterName + '-' + STORAGE_PREFIX.HKUBE_INDEX, moment().format(super.DateFormat), jobId),
+            path: this.createPath({ jobId }),
             data: [],
         }, tracer);
     }
 
     async get({ date, jobId }, tracer) {
         return super.get({
-            path: path.join(this.clusterName + '-' + STORAGE_PREFIX.HKUBE_INDEX, moment(date).format(super.DateFormat), jobId)
+            path: this.createPath({ date, jobId })
         }, tracer);
     }
 
     async list({ date }) {
         if (date) {
             return super.list({
-                path: path.join(this.clusterName + '-' + STORAGE_PREFIX.HKUBE_INDEX, moment(date).format(super.DateFormat))
+                path: this.createPath({ date })
             });
         }
         return super.list({ path: this.clusterName + '-' + STORAGE_PREFIX.HKUBE_INDEX });
@@ -39,9 +39,14 @@ class StorageHkubeIndex extends StorageBase {
 
     async delete({ date, jobId = '' }) {
         return super.delete({
-            path: path.join(this.clusterName + '-' + STORAGE_PREFIX.HKUBE_INDEX, moment(date).format(super.DateFormat), jobId)
+            path: this.createPath({ date, jobId })
         });
     }
+
+    createPath({ date, jobId = '' }) {
+        const dateFormat = moment(date).format(super.DateFormat);
+        return path.join(this.clusterName + '-' + STORAGE_PREFIX.HKUBE_INDEX, dateFormat, jobId);
+    }
 }
 
 module.exports = StorageHkubeIndex;
diff --git a/lib/storage/hkube-metadata.js b/lib/storage/hkube-metadata.js
@@ -3,8 +3,8 @@ const StorageBase = require('./storage-base');
 const { STORAGE_PREFIX } = require('../../consts/storage-prefix');
 
 class StorageHkubeMetadata extends StorageBase {
-    constructor(adapter, log, config) {
-        super(adapter, log);
+    constructor(adapter, log, config, encoding) {
+        super(adapter, log, encoding);
         this.clusterName = config.clusterName;
     }
     async put({ jobId, taskId, data }, tracer) { // eslint-disable-line
@@ -31,6 +31,10 @@ class StorageHkubeMetadata extends StorageBase {
             path: path.join(this.clusterName + '-' + STORAGE_PREFIX.HKUBE_METADATA, jobId, taskId)
         });
     }
+
+    createPath({ jobId, name = '' }) {
+        return path.join(this.clusterName + '-' + STORAGE_PREFIX.HKUBE_RESULTS, jobId, name);
+    }
 }
 
 module.exports = StorageHkubeMetadata;
diff --git a/lib/storage/hkube-results.js b/lib/storage/hkube-results.js
@@ -4,35 +4,39 @@ const { STORAGE_PREFIX } = require('../../consts/storage-prefix');
 const RESULT_JSON = 'result.json';
 
 class StorageHkubeResults extends StorageBase {
-    constructor(adapter, log, config) {
-        super(adapter, log);
+    constructor(adapter, log, config, encoding) {
+        super(adapter, log, encoding);
         this.clusterName = config.clusterName;
     }
 
     async put({ jobId, data }, tracer) {
         return super.put({
-            path: path.join(this.clusterName + '-' + STORAGE_PREFIX.HKUBE_RESULTS, jobId, RESULT_JSON),
+            path: this.createPath({ jobId, name: RESULT_JSON }),
             data
         }, tracer);
     }
 
     async get({ jobId }, tracer) {
         return super.get({
-            path: path.join(this.clusterName + '-' + STORAGE_PREFIX.HKUBE_RESULTS, jobId, RESULT_JSON)
+            path: this.createPath({ jobId, name: RESULT_JSON })
         }, tracer);
     }
 
     async list({ jobId }) {
         return super.list({
-            path: path.join(this.clusterName + '-' + STORAGE_PREFIX.HKUBE_RESULTS, jobId)
+            path: this.createPath({ jobId })
         });
     }
 
     async delete({ jobId }) {
         return super.delete({
-            path: path.join(this.clusterName + '-' + STORAGE_PREFIX.HKUBE_RESULTS, jobId)
+            path: this.createPath({ jobId })
         });
     }
+
+    createPath({ jobId, name = '' }) {
+        return path.join(this.clusterName + '-' + STORAGE_PREFIX.HKUBE_RESULTS, jobId, name);
+    }
 }
 
 module.exports = StorageHkubeResults;
diff --git a/lib/storage/hkube-store.js b/lib/storage/hkube-store.js
@@ -3,35 +3,40 @@ const StorageBase = require('./storage-base');
 const { STORAGE_PREFIX } = require('../../consts/storage-prefix');
 
 class StorageHkubeStore extends StorageBase {
-    constructor(adapter, log, config) {
-        super(adapter, log);
+    constructor(adapter, log, config, encoding) {
+        super(adapter, log, encoding);
         this.clusterName = config.clusterName;
     }
 
     async put({ type, name, data }, tracer) {
         return super.put({
-            path: path.join(this.clusterName + '-' + STORAGE_PREFIX.HKUBE_STORE, type, `${name}.json`),
+            path: this.createPath({ type, name }),
             data,
         }, tracer);
     }
 
     async get({ type, name }, tracer) {
         return super.get({
-            path: path.join(this.clusterName + '-' + STORAGE_PREFIX.HKUBE_STORE, type, `${name}.json`)
+            path: this.createPath({ type, name })
         }, tracer);
     }
 
     async list({ type }) {
         return super.list({
-            path: path.join(this.clusterName + '-' + STORAGE_PREFIX.HKUBE_STORE, type)
+            path: this.createPath({ type })
         });
     }
 
     async delete({ type, name }) {
         return super.delete({
-            path: path.join(this.clusterName + '-' + STORAGE_PREFIX.HKUBE_STORE, type, `${name}.json`)
+            path: this.createPath({ type, name })
         });
     }
+
+    createPath({ type, name }) {
+        const newName = (name && `${name}.json`) || '';
+        return path.join(this.clusterName + '-' + STORAGE_PREFIX.HKUBE_STORE, type, newName);
+    }
 }
 
 module.exports = StorageHkubeStore;
diff --git a/lib/storage/hkube.js b/lib/storage/hkube.js
@@ -3,35 +3,39 @@ const StorageBase = require('./storage-base');
 const { STORAGE_PREFIX } = require('../../consts/storage-prefix');
 
 class StorageHkube extends StorageBase {
-    constructor(adapter, log, config) {
-        super(adapter, log);
+    constructor(adapter, log, config, encoding) {
+        super(adapter, log, encoding);
         this.clusterName = config.clusterName;
     }
 
     async put({ jobId, taskId, data }, tracer) {
         return super.put({
-            path: path.join(this.clusterName + '-' + STORAGE_PREFIX.HKUBE, jobId, taskId),
+            path: this.createPath({ jobId, taskId }),
             data
         }, tracer);
     }
 
     async get({ jobId, taskId }, tracer) {
         return super.get({
-            path: path.join(this.clusterName + '-' + STORAGE_PREFIX.HKUBE, jobId, taskId)
+            path: this.createPath({ jobId, taskId })
         }, tracer);
     }
 
     async list({ jobId }) {
         return super.list({
-            path: path.join(this.clusterName + '-' + STORAGE_PREFIX.HKUBE, jobId)
+            path: this.createPath({ jobId })
         });
     }
 
     async delete({ jobId, taskId = '' }) {
         return super.delete({
-            path: path.join(this.clusterName + '-' + STORAGE_PREFIX.HKUBE, jobId, taskId)
+            path: this.createPath({ jobId, taskId })
         });
     }
+
+    createPath({ jobId, taskId = '' }) {
+        return path.join(this.clusterName + '-' + STORAGE_PREFIX.HKUBE, jobId, taskId);
+    }
 }
 
 module.exports = StorageHkube;
diff --git a/lib/storage/storage-base.js b/lib/storage/storage-base.js
@@ -1,13 +1,15 @@
 
 const now = require('performance-now');
+
 const component = 'StorageManager';
 class StorageBase {
     get DateFormat() {
         return 'YYYY-MM-DD';
     }
 
-    constructor(adapter, log) {
+    constructor(adapter, log, encoding) {
         this._adapter = adapter;
+        this._encoding = encoding;
         this._log = log;
     }
 
@@ -18,7 +20,8 @@ class StorageBase {
             if (tracer) {
                 span = tracer();
             }
-            const result = await this._adapter.put(options);
+            const data = this._encoding.encode(options.data);
+            const result = await this._adapter.put({ ...options, data });
             if (span) span.finish();
             const end = now();
             const diff = (end - start).toFixed(3);
@@ -42,7 +45,8 @@ class StorageBase {
             if (tracer) {
                 span = tracer(options);
             }
-            const result = await this._adapter.get(options);
+            const data = await this._adapter.get(options);
+            const result = this._encoding.decode(data);
             if (span) span.finish();
             const end = now();
             const diff = (end - start).toFixed(3);