Repository: node-influx/node-influx
Pull request #562 (2021-03-08 23:01:38)

#560 - Expose a "point validation" function
## Proposed Changes
Currently, the logic for validating a point is embedded in the InfluxDB.writePoints function. I am proposing to extract that logic out to a separate function on the InfluxDB instance so it will be used by InfluxDB.writePoints, but is also accessible externally for optional manual schema/coercion validation of a given point *before* being sent to InfluxDB.writePoints. This can be beneficial for "pre-processing" a batch of data to determine which points are valid and which are not, so they can be dealt with accordingly. This gives the user the flexibility to decide how validation is done and what to do if a data point is invalid.

## Checklist

- [x] A test has been added if appropriate
- [x] `npm test` completes successfully
- [ ] Commit messages are in [semantic format](https://seesparkbox.com/foundry/semantic_commit_messages)


bencevans: Fantastic. Cheers! When I get a moment, I'll sort out the CI issue that's popped up recently and get this merged in then.

Patch:
diff --git a/src/index.ts b/src/index.ts
index 7342517..5b26be4 100644
--- a/src/index.ts
+++ b/src/index.ts
@@ -147,6 +147,22 @@ export interface IPoint {
   timestamp?: Date | string | number;
 }
 
+export interface IParsedPoint extends IPoint {
+  /**
+   * Fields Pairs is the list of key/value pairs for each field on the point
+   */
+  fieldsPairs: Array<[string, string]>;
+  /**
+   * Tags Names is the list of tag names in the point
+   */
+  tagsNames: string[];
+  /**
+   * Casted Timestamp is the timestamp value after being casted to the
+   * desired precision. Default 'n'
+   */
+  castedTimestamp?: string;
+}
+
 export interface IWriteOptions {
   /**
    * Precision at which the points are written, defaults to nanoseconds 'n'.
@@ -192,6 +208,18 @@ export interface IQueryOptions {
   placeholders?: Record<string, string>;
 }
 
+export interface IParseOptions {
+  /**
+   * Precision at which the points are written, defaults to nanoseconds 'n'.
+   */
+  precision?: grammar.TimePrecision;
+  /**
+   * Database under which to write the points. This is required if a default
+   * database is not provided in Influx.
+   */
+  database?: string;
+}
+
 /**
  * IRetentionOptions are passed into passed into the {@link
  * InfluxDB#createRetentionPolicy} and {@link InfluxDB#alterRetentionPolicy}.
@@ -1283,14 +1311,13 @@ export class InfluxDB {
 
     let payload = "";
     points.forEach((point) => {
-      const { fields = {}, tags = {}, measurement, timestamp } = point;
-
-      const schema =
-        this._schema[database] && this._schema[database][measurement];
-      const fieldsPairs = schema
-        ? schema.coerceFields(fields)
-        : coerceBadly(fields);
-      const tagsNames = schema ? schema.checkTags(tags) : Object.keys(tags);
+      const {
+        measurement,
+        tags,
+        fieldsPairs,
+        tagsNames,
+        castedTimestamp,
+      } = this.parsePoint(point, { database, precision });
 
       payload += (payload.length > 0 ? "\n" : "") + measurement;
 
@@ -1310,8 +1337,8 @@ export class InfluxDB {
           fieldsPairs[i][1];
       }
 
-      if (timestamp !== undefined) {
-        payload += " " + grammar.castTimestamp(timestamp, precision);
+      if (castedTimestamp !== undefined) {
+        payload += " " + castedTimestamp;
       }
     });
 
@@ -1329,6 +1356,75 @@ export class InfluxDB {
     });
   }
 
+  /**
+   * ParsePoint will perform the coercions/schema checks and return the data
+   * required for writing a point. This will throw an error if a schema check
+   * or coercion fails. This can be useful for flagging or "throwing out" bad
+   * points in a batch write to prevent the entire batch from getting aborted
+   *
+   * ---
+   *
+   * A note when using this function, {@link InfluxDB#writePoints} will still perform
+   * the same checks, so any pre-processed data will be checked for validity twice which
+   * has potential performance implications on large data sets
+   *
+   * @param point
+   * @param [options]
+   * @return
+   * @example
+   * // parse a point as if it is getting written to the default
+   * // databse with the default time precision
+   * influx.parsePoint({
+   *     measurement: 'perf',
+   *     tags: { host: 'box1.example.com' },
+   *     fields: { cpu: getCpuUsage(), mem: getMemUsage() },
+   * })
+   *
+   * // you can manually specify the database and time precision
+   * influx.parsePoint({
+   *     measurement: 'perf',
+   *     tags: { host: 'box1.example.com' },
+   *     fields: { cpu: getCpuUsage(), mem: getMemUsage() },
+   * }, {
+   *   precision: 's',
+   *   database: 'my_db'
+   * })
+   *
+   * // if an error occurs, you can catch the error with try...catch
+   * try {
+   *   influx.parsePoint({
+   *     measurement: 'perf',
+   *     tags: { host: 'box1.example.com', myExtraneousTag: 'value' },
+   *     fields: { cpu: getCpuUsage(), mem: getMemUsage(), myExtraneousField: 'value' },
+   *   })
+   * } catch(err) {
+   *   handleError(err);
+   * }
+   */
+  parsePoint(point: IPoint, options: IParseOptions = {}): IParsedPoint {
+    const { database = this._defaultDB(), precision = "n" } = options;
+
+    const { fields = {}, tags = {}, measurement, timestamp } = point;
+
+    const schema =
+      this._schema[database] && this._schema[database][measurement];
+    const fieldsPairs = schema
+      ? schema.coerceFields(fields)
+      : coerceBadly(fields);
+    const tagsNames = schema ? schema.checkTags(tags) : Object.keys(tags);
+    const castedTimestamp =
+      timestamp && grammar.castTimestamp(timestamp, precision);
+    return {
+      fields,
+      tags,
+      measurement,
+      timestamp,
+      fieldsPairs,
+      tagsNames,
+      castedTimestamp,
+    };
+  }
+
   /**
    * WriteMeasurement functions similarly to {@link InfluxDB#writePoints}, but
    * it automatically fills in the `measurement` value for all points for you.


Test patch:
diff --git a/test/karma.conf.js b/test/karma.conf.js
index d712024..3d597b1 100644
--- a/test/karma.conf.js
+++ b/test/karma.conf.js
@@ -34,10 +34,16 @@ module.exports = function (config) {
      * Webpack and bundling config:
      */
     webpack: require("./webpack.config"),
-    webpackServer: { noInfo: true },
-    webpackMiddleware: { stats: "errors-only" },
+    webpackServer: {
+      noInfo: true,
+    },
+    webpackMiddleware: {
+      stats: "errors-only",
+    },
     files: ["test/karma.shim.js"],
-    preprocessors: { "test/karma.shim.js": ["sourcemap", "webpack"] },
+    preprocessors: {
+      "test/karma.shim.js": ["sourcemap", "webpack"],
+    },
     middleware: ["pool-tests"],
 
     /**
diff --git a/test/unit/influx.test.ts b/test/unit/influx.test.ts
index 02a0f2b..048a52f 100644
--- a/test/unit/influx.test.ts
+++ b/test/unit/influx.test.ts
@@ -640,37 +640,6 @@ describe("influxdb", () => {
         ]);
       });
 
-      it("throws on schema violations", () => {
-        setDefaultDB("my_db");
-
-        expect(() => {
-          influx.writePoints([
-            {
-              measurement: "my_schemed_measure",
-              tags: { not_a_tag: "1" },
-            },
-          ]);
-        }).to.throw(/extraneous tags/i);
-
-        expect(() => {
-          influx.writePoints([
-            {
-              measurement: "my_schemed_measure",
-              fields: { not_a_field: "1" },
-            },
-          ]);
-        }).to.throw(/extraneous fields/i);
-
-        expect(() => {
-          influx.writePoints([
-            {
-              measurement: "my_schemed_measure",
-              fields: { bool: "lol, not a bool" },
-            },
-          ]);
-        }).to.throw(/expected bool/i);
-      });
-
       it("handles lack of tags", () => {
         expectWrite("mymeas myfield=90", {
           precision: "n",
@@ -781,6 +750,119 @@ describe("influxdb", () => {
       });
     });
 
+    describe(".parsePoint()", () => {
+      it("parses a minimal valid point with default options", () => {
+        setDefaultDB("my_db");
+        let point = {
+          measurement: "mymeas",
+          fields: { myfield: 90 },
+        };
+        let parsedPoint = influx.parsePoint(point);
+        expect(parsedPoint.measurement).to.equal(point.measurement);
+        expect(parsedPoint.fields).to.equal(point.fields);
+        expect(parsedPoint.tags).to.deep.equals({});
+        expect(parsedPoint.fieldsPairs).to.deep.equals([["myfield", "90"]]);
+        expect(parsedPoint.tagsNames).to.deep.equals([]);
+        expect(parsedPoint.castedTimestamp).to.be.undefined;
+      });
+
+      it("parses a point with fields, tags, and timestamp", () => {
+        setDefaultDB("my_db");
+        let date = new Date();
+        let point = {
+          measurement: "mymeas",
+          fields: { myfield: 90 },
+          tags: { my_tag: "2" },
+          timestamp: date,
+        };
+        let parsedPoint = influx.parsePoint(point);
+        expect(parsedPoint.measurement).to.equal(point.measurement);
+        expect(parsedPoint.fields).to.equal(point.fields);
+        expect(parsedPoint.tags).to.equals(point.tags);
+        expect(parsedPoint.fieldsPairs).to.deep.equals([["myfield", "90"]]);
+        expect(parsedPoint.tagsNames).to.deep.equals(["my_tag"]);
+        expect(parsedPoint.castedTimestamp).not.to.be.undefined;
+      });
+
+      it("accepts custom precision option", () => {
+        setDefaultDB("my_db");
+        let date = new Date();
+        let point = {
+          measurement: "mymeas",
+          fields: { myfield: 90 },
+          timestamp: date,
+        };
+        let parsedPoint = influx.parsePoint(point, { precision: "ms" });
+        expect(parsedPoint.castedTimestamp).to.equal(date.getTime().toString());
+      });
+
+      it("accepts custom database option", () => {
+        let date = new Date();
+        let point = {
+          measurement: "mymeas",
+          fields: { myfield: 90 },
+          timestamp: date,
+        };
+        let parsedPoint = influx.parsePoint(point, { database: "my_db" });
+        expect(parsedPoint).to.exist;
+      });
+
+      it("uses a schema to coerce", () => {
+        setDefaultDB("my_db");
+        let date = new Date();
+        let point = {
+          measurement: "my_schemed_measure",
+          fields: { bool: true, float: 43, int: 42 },
+          tags: { my_tag: "2" },
+          timestamp: date,
+        };
+        let parsedPoint = influx.parsePoint(point);
+        expect(parsedPoint.measurement).to.equal(point.measurement);
+        expect(parsedPoint.fields).to.equal(point.fields);
+        expect(parsedPoint.tags).to.equals(point.tags);
+        expect(parsedPoint.fieldsPairs).to.deep.equals([
+          ["bool", "T"],
+          ["float", "43"],
+          ["int", "42i"],
+        ]);
+        expect(parsedPoint.tagsNames).to.deep.equals(["my_tag"]);
+        expect(parsedPoint.castedTimestamp).not.to.be.undefined;
+      });
+
+      it("should throw an error if extraneous tags are given", () => {
+        setDefaultDB("my_db");
+        expect(() => {
+          let point = {
+            measurement: "my_schemed_measure",
+            tags: { not_a_tag: "1" },
+          };
+          influx.parsePoint(point);
+        }).to.throw(/extraneous tags/i);
+      });
+
+      it("should throw an error if extraneous fields are given", () => {
+        setDefaultDB("my_db");
+        expect(() => {
+          let point = {
+            measurement: "my_schemed_measure",
+            fields: { not_a_field: "1" },
+          };
+          influx.parsePoint(point);
+        }).to.throw(/extraneous fields/i);
+      });
+
+      it("should throw an error if invalid value for field type given", () => {
+        setDefaultDB("my_db");
+        expect(() => {
+          let point = {
+            measurement: "my_schemed_measure",
+            fields: { bool: "lol, not a bool" },
+          };
+          influx.parsePoint(point);
+        }).to.throw(/expected bool/i);
+      });
+    });
+
     describe(".query", () => {
       beforeEach(() => setDefaultDB("my_db"));