import { RequestOptions } from "https" ; import * as url from "url" ; import * as b from "./builder" ; import * as grammar from "./grammar" ; import { IPingStats , IPoolOptions , Pool } from "./pool" ; import { assertNoErrors , IResults , parse , parseSingle } from "./results" ; import { coerceBadly , ISchemaOptions , Schema } from "./schema" ; const defaultHost = Object . freeze ( { host : "127.0.0.1" , port : 8086 , path : "" , protocol : "http" as const , } ) ; const defaultOptions = Object . freeze ( { database : null , hosts : [ ] , password : "root" , schema : [ ] , username : "root" , } ) ; export * from "./builder" ; export { INanoDate , FieldType , Precision , Raw , TimePrecision , escape , toNanoDate , } from "./grammar" ; export { ISchemaOptions } from "./schema" ; export { IPingStats , IPoolOptions } from "./pool" ; export { IResults , IResponse , ResultError } from "./results" ; export interface IHostConfig { /**
     * Influx host to connect to, defaults to 127.0.0.1.
     */ host ? : string ; /**
     * Influx port to connect to, defaults to 8086.
     */ port ? : number ; /**
     * Path for Influx within the host, defaults to ''.
     * May be used if Influx is behind a reverse proxy or load balancer.
     */ path ? : string ; /**
     * Protocol to connect over, defaults to 'http'.
     */ protocol ? : "http" | "https" ; /**
     * Optional request option overrides.
     */ options ? : RequestOptions ; } export interface ISingleHostConfig extends IHostConfig { /**
     * Username for connecting to the database. Defaults to 'root'.
     */ username ? : string ; /**
     * Password for connecting to the database. Defaults to 'root'.
     */ password ? : string ; /**
     * Default database to write information to.
     */ database ? : string ; /**
     * Settings for the connection pool.
     */ pool ? : IPoolOptions ; /**
     * A list of schema for measurements in the database.
     */ schema ? : ISchemaOptions [ ] ; } export interface IClusterConfig { /**
     * Username for connecting to the database. Defaults to 'root'.
     */ username ? : string ; /**
     * Password for connecting to the database. Defaults to 'root'.
     */ password ? : string ; /**
     * Default database to write information to.
     */ database ? : string ; /**
     * A list of cluster hosts to connect to.
     */ hosts : IHostConfig [ ] ; /**
     * Settings for the connection pool.
     */ pool ? : IPoolOptions ; /**
     * A list of schema for measurements in the database.
     */ schema ? : ISchemaOptions [ ] ; } export interface IPoint { /**
     * Measurement is the Influx measurement name.
     */ measurement ? : string ; /**
     * Tags is the list of tag values to insert.
     */ tags ? : { [ name ] : string ; } ; /**
     * Fields is the list of field values to insert.
     */ fields ? : { [ name ] : any ; } ; /**
     * Timestamp tags this measurement with a date. This can be a Date object,
     * in which case we'll adjust it to the desired precision, or a numeric
     * string or number, in which case it gets passed directly to Influx.
     */ 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'.
     */ precision ? : grammar . TimePrecision ; /**
     * Retention policy to write the points under, defaults to the DEFAULT
     * database policy.
     */ retentionPolicy ? : string ; /**
     * Database under which to write the points. This is required if a default
     * database is not provided in Influx.
     */ database ? : string ; } export interface IQueryOptions { /**
     * Defines the precision at which to query points. When left blank, it will
     * query in nanosecond precision.
     */ precision ? : grammar . TimePrecision ; /**
     * Retention policy to query from, defaults to the DEFAULT
     * database policy.
     */ retentionPolicy ? : string ; /**
     * Database under which to query the points. This is required if a default
     * database is not provided in Influx.
     */ database ? : string ; /**
     * Any placeholders used by the query. Using these is strongly recommended
     * to avoid injection attacks.
     */ placeholders ? : Record < string , string | number > ; } 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  InfluxDB .
 * See the [Downsampling and Retention page](https://docs.influxdata.com/
 * influxdb/v1.0/guides/downsampling_and_retention/) on the Influx docs for
 * more information.
  export interface IRetentionOptions { database ? : string ; duration : string ; replication : number ; isDefault ? : boolean ; } /**
 * Parses the URL out into into a IClusterConfig object
 */ function parseOptionsUrl ( addr ) { const parsed = url . parse ( addr ) ; const options = { host : parsed . hostname , port : Number ( parsed . port ) , protocol : parsed . protocol . slice ( 0 , - 1 ) as "http" | "https" , } ; if ( parsed . auth ) { [ options . username , options . password ] = parsed . auth . split ( ":" ) ; } if ( parsed . pathname . length > 1 ) { options . database = parsed . pathname . slice ( 1 ) ; } return options ; } /**
 * Works similarly to Object.assign, but only overwrites
 * properties that resolve to undefined.
 */ function defaults < T > ( target , ... srcs ) { srcs . forEach ( ( src ) => { Object . keys ( src ) . forEach ( ( key ) => { if ( target [ key ] === undefined ) { target [ key ] = src [ key ] ; } } ) ; } ) ; return target ; } example example export class InfluxDB { private private readonly _pool : Pool ; private private readonly _options : IClusterConfig ; private private _schema : { [ db ] : { [ measurement ] : Schema ; } ; } = Object . create ( null ) ; constructor ( options ) ; /**
     * Connect to an InfluxDB cluster by specifying a
     * set of connection options.
     */ constructor ( options ) ; example constructor ( url ) ; /**
     * Connects to a local, default Influx instance.
     */ constructor ( ) ; param options example example example constructor ( options ? ) { if ( typeof options === "string" ) { options = parseOptionsUrl ( options ) ; } else if ( ! options ) { options = defaultHost ; } if ( ! options . hasOwnProperty ( "hosts" ) ) { options = { database : options . database , hosts : [ options ] , password : options . password , pool : options . pool , schema : options . schema , username : options . username , } ; } const resolved = options as IClusterConfig ; resolved . hosts = resolved . hosts . map ( ( host ) => { return defaults ( { host : host . host , port : host . port , path : host . path , protocol : host . protocol , options : host . options , } , defaultHost ) ; } ) ; this . _pool = new Pool ( resolved . pool ) ; this . _options = defaults ( resolved , defaultOptions ) ; resolved . hosts . forEach ( ( host ) => { this . _pool . addHost ( `${ host . protocol }://${ host . host }:${ host . port }${ host . path }` , host . options ) ; } ) ; this . _options . schema . forEach ( ( schema ) => this . _createSchema ( schema ) ) ; } param { ISchemaOptions } schema memberof public addSchema ( schema ) { this . _createSchema ( schema ) ; } param databaseName return example public createDatabase ( databaseName ) { return this . _pool . json ( this . _getQueryOpts ( { q : `create database ${ grammar . escape . quoted ( databaseName ) }` , } , "POST" ) ) . then ( assertNoErrors ) . then ( ( ) => undefined ) ; } param databaseName return example public dropDatabase ( databaseName ) { return this . _pool . json ( this . _getQueryOpts ( { q : `drop database ${ grammar . escape . quoted ( databaseName ) }` , } , "POST" ) ) . then ( assertNoErrors ) . then ( ( ) => undefined ) ; } returns example public getDatabaseNames ( ) { return this . _pool . json ( this . _getQueryOpts ( { q : "show databases" } ) ) . then ( ( res ) => parseSingle < { name : string ; } > ( res ) . map ( ( r ) => r . name ) ) ; } returns param database example public getMeasurements ( database = this . _defaultDB ( ) ) { return this . _pool . json ( this . _getQueryOpts ( { db : database , q : "show measurements" , } ) ) . then ( ( res ) => parseSingle < { name : string ; } > ( res ) . map ( ( r ) => r . name ) ) ; } param options param options . measurement param options . database returns example public getSeries ( options = { } ) { const { database = this . _defaultDB ( ) , measurement } = options ; let query = "show series" ; if ( measurement ) { query += ` from ${ grammar . escape . quoted ( measurement ) }` ; } return this . _pool . json ( this . _getQueryOpts ( { db : database , q : query , } ) ) . then ( ( res ) => parseSingle < { key : string ; } > ( res ) . map ( ( r ) => r . key ) ) ; } param measurement param database return example public dropMeasurement ( measurement , database = this . _defaultDB ( ) ) { return this . _pool . json ( this . _getQueryOpts ( { db : database , q : `drop measurement ${ grammar . escape . quoted ( measurement ) }` , } , "POST" ) ) . then ( assertNoErrors ) . then ( ( ) => undefined ) ; } returns example public dropSeries ( options ) { const db = "database" in options ? ( options as any ) . database : this . _defaultDB ( ) ; let q = "drop series" ; if ( "measurement" in options ) { q += " from " + b . parseMeasurement ( options ) ; } if ( "where" in options ) { q += " where " + b . parseWhere ( options ) ; } return this . _pool . json ( this . _getQueryOpts ( { db , q } , "POST" ) ) . then ( assertNoErrors ) . then ( ( ) => undefined ) ; } return example public getUsers ( ) { return this . _pool . json ( this . _getQueryOpts ( { q : "show users" } ) ) . then ( ( result ) => parseSingle < { user : string ; admin : boolean ; } > ( result ) ) ; } param username param password param admin return example public createUser ( username , password , admin = false ) { return this . _pool . json ( this . _getQueryOpts ( { q : `create user ${ grammar . escape . quoted ( username ) } with password ` + grammar . escape . stringLit ( password ) + ( admin ? " with all privileges" : "" ) , } , "POST" ) ) . then ( assertNoErrors ) . then ( ( ) => undefined ) ; } param username param password return example public setPassword ( username , password ) { return this . _pool . json ( this . _getQueryOpts ( { q : `set password for ${ grammar . escape . quoted ( username ) } = ` + grammar . escape . stringLit ( password ) , } , "POST" ) ) . then ( assertNoErrors ) . then ( ( ) => undefined ) ; } param username param privilege param database return example public grantPrivilege ( username , privilege , database = this . _defaultDB ( ) ) { return this . _pool . json ( this . _getQueryOpts ( { q : `grant ${ privilege } on ${ grammar . escape . quoted ( database ) } ` + `to ${ grammar . escape . quoted ( username ) }` , } , "POST" ) ) . then ( assertNoErrors ) . then ( ( ) => undefined ) ; } param username param privilege param database return example public revokePrivilege ( username , privilege , database = this . _defaultDB ( ) ) { return this . _pool . json ( this . _getQueryOpts ( { q : `revoke ${ privilege } on ${ grammar . escape . quoted ( database ) } from ` + grammar . escape . quoted ( username ) , } , "POST" ) ) . then ( assertNoErrors ) . then ( ( ) => undefined ) ; } param username return example public grantAdminPrivilege ( username ) { return this . _pool . json ( this . _getQueryOpts ( { q : `grant all to ${ grammar . escape . quoted ( username ) }` , } , "POST" ) ) . then ( assertNoErrors ) . then ( ( ) => undefined ) ; } param username return example public revokeAdminPrivilege ( username ) { return this . _pool . json ( this . _getQueryOpts ( { q : `revoke all from ${ grammar . escape . quoted ( username ) }` , } , "POST" ) ) . then ( assertNoErrors ) . then ( ( ) => undefined ) ; } param username return example public dropUser ( username ) { return this . _pool . json ( this . _getQueryOpts ( { q : `drop user ${ grammar . escape . quoted ( username ) }` , } , "POST" ) ) . then ( assertNoErrors ) . then ( ( ) => undefined ) ; } param name param query param database param resample return example public createContinuousQuery ( name , query , database = this . _defaultDB ( ) , resample = "" ) { return this . _pool . json ( this . _getQueryOpts ( { q : `create continuous query ${ grammar . escape . quoted ( name ) }` + ` on ${ grammar . escape . quoted ( database ) } ${ resample } begin ${ query } end` , } , "POST" ) ) . then ( assertNoErrors ) . then ( ( ) => undefined ) ; } param database return example public showContinousQueries ( database = this . _defaultDB ( ) ) { return this . _pool . json ( this . _getQueryOpts ( { db : database , q : "show continuous queries" , } ) ) . then ( ( result ) => parseSingle < { name : string ; query : string ; } > ( result ) ) ; } param name param database return example public dropContinuousQuery ( name , database = this . _defaultDB ( ) ) { return this . _pool . json ( this . _getQueryOpts ( { q : `drop continuous query ${ grammar . escape . quoted ( name ) }` + ` on ${ grammar . escape . quoted ( database ) }` , } , "POST" ) ) . then ( assertNoErrors ) . then ( ( ) => undefined ) ; } param name param options param options . database param options . duration param options . replication param options . isDefault return example public createRetentionPolicy ( name , options ) { const q = `create retention policy ${ grammar . escape . quoted ( name ) } on ` + grammar . escape . quoted ( options . database || this . _defaultDB ( ) ) + ` duration ${ options . duration } replication ${ options . replication }` + ( options . isDefault ? " default" : "" ) ; return this . _pool . json ( this . _getQueryOpts ( { q } , "POST" ) ) . then ( assertNoErrors ) . then ( ( ) => undefined ) ; } param name param options param options . database param options . duration param options . replication param options . default return example public alterRetentionPolicy ( name , options ) { const q = `alter retention policy ${ grammar . escape . quoted ( name ) } on ` + grammar . escape . quoted ( options . database || this . _defaultDB ( ) ) + ` duration ${ options . duration } replication ${ options . replication }` + ( options . isDefault ? " default" : "" ) ; return this . _pool . json ( this . _getQueryOpts ( { q } , "POST" ) ) . then ( assertNoErrors ) . then ( ( ) => undefined ) ; } param name param database return example public dropRetentionPolicy ( name , database = this . _defaultDB ( ) ) { return this . _pool . json ( this . _getQueryOpts ( { q : `drop retention policy ${ grammar . escape . quoted ( name ) } ` + `on ${ grammar . escape . quoted ( database ) }` , } , "POST" ) ) . then ( assertNoErrors ) . then ( ( ) => undefined ) ; } param database return example public showRetentionPolicies ( database = this . _defaultDB ( ) ) { return this . _pool . json ( this . _getQueryOpts ( { q : `show retention policies on ${ grammar . escape . quoted ( database ) }` , } , "GET" ) ) . then ( ( result ) => parseSingle < { default : boolean ; duration : string ; name : string ; replicaN : number ; shardGroupDuration : string ; } > ( result ) ) ; } param database return example public showShards ( database = this . _defaultDB ( ) ) { return this . _pool . json ( this . _getQueryOpts ( { q : "show shards " , } , "GET" ) ) . then ( ( result ) => parseSingle < { id : number ; database : string ; retention_policy : string ; shard_group : number ; start_time : string ; end_time : string ; expiry_time : string ; owners : string ; } > ( result ) . filter ( function ( i ) { return i . database === database ; } ) ) ; } param shard_id return example public dropShard ( shard_id ) { return this . _pool . json ( this . _getQueryOpts ( { q : `drop shard ${ shard_id }` , } , "POST" ) ) . then ( assertNoErrors ) . then ( ( ) => undefined ) ; } /**
     * WritePoints sends a list of points together in a batch to InfluxDB. In
     * each point you must specify the measurement name to write into as well
     * as a list of tag and field values. Optionally, you can specify the
     * time to tag that point at, defaulting to the current time.
     *
     * If you defined a schema for the measurement in the options you passed
     * to `new Influx(options)`, we'll use that to make sure that types get
     * cast correctly and that there are no extraneous fields or columns.
     *
     * For best performance, it's recommended that you batch your data into
     * sets of a couple thousand records before writing it. In the future we'll
     * have some utilities within node-influx to make this easier.
     *
     * ---
     *
     * A note when using manually-specified times and precisions: by default
     * we write using the `ms` precision since that's what JavaScript gives us.
     * You can adjust this. However, there is some special behaviour if you
     * manually specify a timestamp in your points:
     *  - if you specify the timestamp as a Date object, we'll convert it to
     *    milliseconds and manipulate it as needed to get the right precision
     *  - if provide a INanoDate as returned from  toNanoTime or the
     *    results from an Influx query, we'll be able to pull the precise
     *    nanosecond timestamp and manipulate it to get the right precision
     *  - if you provide a string or number as the timestamp, we'll pass it
     *    straight into Influx.
     *
     * Please see the IPoint and IWriteOptions types for a
     * full list of possible options.
     *
     *  param points param options return example public writePoints ( points , options = { } ) { const { database = this . _defaultDB ( ) , precision = "n" as grammar . TimePrecision , retentionPolicy , } = options ; let payload = "" ; points . forEach ( ( point ) => { const { measurement , tags , fieldsPairs , tagsNames , castedTimestamp } = this . parsePoint ( point , { database , precision } ) ; payload += ( payload . length > 0 ? "\n" : "" ) + measurement ; for ( let tagsName of tagsNames ) { payload += "," + grammar . escape . tag ( tagsName ) + "=" + grammar . escape . tag ( tags [ tagsName ] ) ; } for ( let i = 0 ; i < fieldsPairs . length ; i += 1 ) { payload += ( i === 0 ? " " : "," ) + grammar . escape . tag ( fieldsPairs [ i ] [ 0 ] ) + "=" + fieldsPairs [ i ] [ 1 ] ; } if ( castedTimestamp !== undefined ) { payload += " " + castedTimestamp ; } } ) ; return this . _pool . discard ( { body : payload , method : "POST" , path : "/write" , query : { db : database , p : this . _options . password , precision , rp : retentionPolicy , u : this . _options . username , } , } ) ; } /**
     * 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,  InfluxDB 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 parsePoint ( point , options = { } ) { 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  InfluxDB , but
     * it automatically fills in the `measurement` value for all points for you.
     *
     *  param measurement param points param options return example public writeMeasurement ( measurement , points , options = { } ) { points = points . map ( ( p ) => ( { measurement , ... p } ) ) ; return this . writePoints ( points , options ) ; } public query < T > ( query , options ? ) ; public query < T > ( query , options ? ) ; /**
     * .query() runs a query (or list of queries), and returns the results in a
     * friendly format,  IResults . If you run multiple queries, an array of results
     * will be returned, otherwise a single result (array of objects) will be returned.
     *
     *  param query param options return example public query < T > ( query , options = { } ) { if ( Array . isArray ( query ) ) { query = query . join ( ";" ) ; } if ( options . precision === "n" ) { options = { ... options } ; delete options . precision ; } return this . queryRaw ( query , options ) . then ( ( res ) => parse < T > ( res , options . precision ) ) ; } param query param options return example public queryRaw ( query , options = { } ) { const { database = this . _defaultDB ( ) , retentionPolicy , placeholders = { } , } = options ; if ( query instanceof Array ) { query = query . join ( ";" ) ; } return this . _pool . json ( this . _getQueryOpts ( { db : database , epoch : options . precision , q : query , rp : retentionPolicy , params : JSON . stringify ( placeholders ) , } ) ) ; } param timeout return example public ping ( timeout ) { let auth = undefined ; if ( typeof this . _options . username === "string" ) { auth = `${ this . _options . username }:${ this . _options . password || "" }` ; } return this . _pool . ping ( timeout , "/ping" , auth ) ; } private private _defaultDB ( ) { if ( ! this . _options . database ) { throw new Error ( "Attempted to run an influx query without a default" + " database specified or an explicit database provided." ) ; } return this . _options . database ; } private private _getQueryOpts ( params , method = "GET" ) { return { method , path : "/query" , query : { p : this . _options . password , u : this . _options . username , ... params , } , } ; } private param { ISchemaOptions } schema memberof private _createSchema ( schema ) { schema . database = schema . database || this . _options . database ; if ( ! schema . database ) { throw new Error ( `Schema ${ schema . measurement } doesn't have a database specified,` + "and no default database is provided!" ) ; } if ( ! this . _schema [ schema . database ] ) { this . _schema [ schema . database ] = Object . create ( null ) ; } this . _schema [ schema . database ] [ schema . measurement ] = new Schema ( schema ) ; } }