exposing source id (#15)

* exposing source id * eslint fixes
Azure · Feb 21, 2019 · c93a1e1 · c93a1e1
1 parent a9ae8ca
commit c93a1e1
Show file tree

Hide file tree

Showing 5 changed files with 37 additions and 12 deletions.
diff --git a/azure-kusto-ingest/package-lock.json b/azure-kusto-ingest/package-lock.json
diff --git a/azure-kusto-ingest/package.json b/azure-kusto-ingest/package.json
@@ -29,7 +29,8 @@
     "azure-storage": "^2.10.2",
     "moment": "^2.22.2",
     "request": "^2.88.0",
-    "uuid": "^3.3.2"
+    "uuid": "^3.3.2",
+    "uuid-validate": "0.0.3"
   },
   "devDependencies": {
     "mocha": "^5.2.0",

diff --git a/azure-kusto-ingest/source/descriptors.js b/azure-kusto-ingest/source/descriptors.js
@@ -2,7 +2,13 @@ const fs = require("fs");
 const path = require("path");
 const zlib = require("zlib");
 const Transform = require("stream").Transform;
+const uuidValidate = require("uuid-validate");
 
+function assertUuid4(maybeUuid, errorMessage) {
+    if (!!maybeUuid && !uuidValidate(maybeUuid, 4)) {
+        throw Error(errorMessage);
+    }
+}
 
 class BytesCounter extends Transform {
     constructor() {
@@ -20,12 +26,15 @@ class BytesCounter extends Transform {
 }
 
 class FileDescriptor {
-    constructor(filePath) {
+    constructor(filePath, sourceId = null) {
         this.filePath = filePath;
         this.name = path.basename(this.filePath);
         this.extension = path.extname(this.filePath).toLowerCase();
         this.size = null;
         this.zipped = this.extension === ".gz";
+
+        assertUuid4(sourceId, "sourceId is not a valid uuid/v4");
+        this.sourceId = sourceId;
     }
 
     _gzip(callback) {
@@ -52,12 +61,15 @@ class FileDescriptor {
 
 
 class StreamDescriptor {
-    constructor(stream) {
+    constructor(stream, sourceId = null) {
         this._stream = stream;
 
         this.stream = null;
         this.name = "stream";
         this.size = null;
+
+        assertUuid4(sourceId, "sourceId is not a valid uuid/v4");
+        this.sourceId = sourceId;
     }
 
     pipe(dest) {
@@ -70,9 +82,12 @@ class StreamDescriptor {
 }
 
 class BlobDescriptor {
-    constructor(path, size = null) {
+    constructor(path, size = null, sourceId = null) {
         this.path = path;
         this.size = size;
+
+        assertUuid4(sourceId, "sourceId is not a valid uuid/v4");
+        this.sourceId = sourceId;
     }
 }
 

diff --git a/azure-kusto-ingest/source/ingestClient.js b/azure-kusto-ingest/source/ingestClient.js
@@ -89,7 +89,7 @@ module.exports = class KustoIngestClient {
                 blobService.createBlockBlobFromLocalFile(containerDetails.objectName, blobName, fileToUpload, (err) => {
                     if (err) return callback(err);
                     let blobUri = `${containerDetails.toURI({ withSas: false })}/${blobName}?${containerDetails.sas}`;
-                    return this.ingestFromBlob(new BlobDescriptor(blobUri, descriptor.size), props, callback);
+                    return this.ingestFromBlob(new BlobDescriptor(blobUri, descriptor.size, descriptor.sourceId), props, callback);
                 });
             });
 
@@ -100,6 +100,10 @@ module.exports = class KustoIngestClient {
         const props = this._mergeProps(ingestionProperties);
         props.validate();
 
+        if (typeof (blob) === "string") {
+            blob = new BlobDescriptor(blob);
+        }
+
         return this.resourceManager.getIngestionQueues((err, queues) => {
             if (err) return callback(err);
 

diff --git a/azure-kusto-ingest/source/ingestionBlobInfo.js b/azure-kusto-ingest/source/ingestionBlobInfo.js
@@ -2,9 +2,9 @@ const uuidv4 = require("uuid/v4");
 const moment = require("moment");
 
 module.exports = class IngestionBlobInfo {
-    constructor(blob, ingestionProperties, authContext) {
-        this.BlobPath = blob.path;
-        this.RawDataSize = blob.size;
+    constructor(blobDescriptor, ingestionProperties, authContext) {
+        this.BlobPath = blobDescriptor.path;
+        this.RawDataSize = blobDescriptor.size;
         this.DatabaseName = ingestionProperties.database;
         this.TableName = ingestionProperties.table;
         this.RetainBlobOnSuccess = true;
@@ -13,7 +13,7 @@ module.exports = class IngestionBlobInfo {
         this.ReportLevel = ingestionProperties.reportLevel;
         this.ReportMethod = ingestionProperties.reportMethod;
         this.SourceMessageCreationTime = moment.utc();
-        this.Id = uuidv4();
+        this.Id = blobDescriptor.sourceId || uuidv4();
 
         let additionalProperties = ingestionProperties.additionalProperties || {};
         additionalProperties.authorizationContext = authContext;