/**
 * Copyright 2020 Opstrace, Inc.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */ import crypto from "crypto" ; import protobuf from "protobufjs" ; import snappy from "snappy" ; import got from "got" ; import Long from "long" ; import * as mathjs from "mathjs" ; import { mtimeDiffSeconds , mtime , sleep } from "../mtime" ; import { log } from "../log" ; import { logHTTPResponseLight , logHTTPResponse } from "../util" ; import { MetricSeries } from "./dummyseries" ; import { SampleBase , FragmentBase , FragmentStatsBase } from "../series" ; export * from "./dummyseries" ; const pbfRoot = protobuf . loadSync ( `${ __dirname }/../resources/promproto_bundle.json` ) ; const pbTypeWriterequest = pbfRoot . lookupType ( "prometheus.WriteRequest" ) ; const pbTypeTimeseries = pbfRoot . lookupType ( "prometheus.TimeSeries" ) ; const pbTypeSample = pbfRoot . lookupType ( "prometheus.Sample" ) ; const pbTypeLabel = pbfRoot . lookupType ( "prometheus.Label" ) ; export interface MetricSeriesFragmentStats extends FragmentStatsBase { timeMillisSinceEpochFirst : number ; timeMillisSinceEpochLast : number ; min : string ; max : string ; var : string ; } /**
 * The metric sample value, i.e. a floating point number The metric sample
 * timestamp which is an int64 in the Prometheus protobuf world -- representing
 * milliseconds since epoch -- here, for the NodeJS runtime we use a `Long`
 * from the `long` library to keep the integer arithmetics. */ export class MetricSample extends SampleBase < number , Long > { } export class MetricSeriesFragment extends FragmentBase < MetricSample , MetricSeries > { public payloadByteCount ( ) { if ( this . stats !== undefined ) { return this . stats . sampleCount * BigInt ( 16 ) ; } return BigInt ( this . samples . length ) * BigInt ( 16 ) ; } public buildStatisticsAndDropData ( ) { if ( ! this . serialized ) { throw new Error ( "not yet serialized" ) ; } const values = [ ] ; for ( const sample of this . samples ) { values . push ( sample . value ) ; } const stats = { min : formatFloatForComp ( mathjs . min ( values ) ) , max : formatFloatForComp ( mathjs . max ( values ) ) , var : formatFloatForComp ( mathjs . variance ( values ) ) , timeMillisSinceEpochFirst : this . samples [ 0 ] . time . toNumber ( ) , timeMillisSinceEpochLast : this . samples . slice ( - 1 ) [ 0 ] . time . toNumber ( ) , sampleCount : BigInt ( values . length ) } ; this . stats = stats ; this . samples . length = 0 ; this . stats = stats ; } protected addSampleHook ( ) { return ; } public serialize ( ) { this . serialized = true ; return new MetricSeriesFragmentPushMessage ( [ this ] ) ; } } export class MetricSeriesFragmentPushMessage { fragments : MetricSeriesFragment [ ] ; datamd5 : string ; data : Buffer ; dataLengthBytes : number ; dataLengthMiB : number ; payloadByteCount : bigint ; serializationTimeSeconds : number ; postHeaders : Record < string , string > ; constructor ( seriesFragments ) { this . fragments = seriesFragments ; this . payloadByteCount = BigInt ( 0 ) ; const [ data , datamd5 , serializationTimeSeconds ] = this . serialize ( ) ; this . datamd5 = datamd5 ; this . data = data ; this . serializationTimeSeconds = serializationTimeSeconds ; this . dataLengthBytes = Buffer . byteLength ( data ) ; this . dataLengthMiB = Buffer . byteLength ( data ) / ( 1024.0 * 1024 ) ; this . postHeaders = { "Content-Type" : "application/x-protobuf" , "Content-Encoding" : "snappy" , "X-Prometheus-Remote-Write-Version" : "0.1.0" } ; } public toString = ( ) : string => { return `PushMessage(md5=${ this . datamd5 })` ; } ; param cortexBaseUrl public async postWithRetryOrError ( cortexBaseUrl , maxRetries = 3 , additionalHeaders ? ) { for ( let attempt = 1 ; attempt <= maxRetries ; attempt ++ ) { let response ; try { response = await this . postToCortex ( `${ cortexBaseUrl }/api/v1/push` , additionalHeaders ) ; } catch ( e ) { if ( e instanceof got . RequestError ) { log . warning ( `POST ${ this }: attempt ${ attempt } failed with ${ e . message }` ) ; await sleep ( 1.0 ) ; continue ; } else { throw e ; } } if ( response . statusCode === 200 ) { logHTTPResponseLight ( response ) ; for ( const fragment of this . fragments ) { fragment . parent ! . nFragmentsSuccessfullySentSinceLastValidate += 1 ; } return ; } if ( response . statusCode === 429 ) { log . info ( `429 resp, sleep 2, body[:300]: ${ response . body . slice ( 0 , 300 ) }` ) ; await sleep ( 3.0 ) ; continue ; } logHTTPResponse ( response ) ; if ( response . statusCode . toString ( ) . startsWith ( "4" ) ) { throw new Error ( "Bad HTTP request (see log above)" ) ; } log . info ( "Treat as transient problem. Wait shortly, and retry" ) ; await sleep ( 3.0 ) ; } throw new Error ( `Failed to POST ${ this } after ${ maxRetries } attempts` ) ; } param url public async postToCortex ( url , additionalHeaders ? ) { let headers = { "Content-Type" : "application/x-protobuf" , "Content-Encoding" : "snappy" , "X-Prometheus-Remote-Write-Version" : "0.1.0" } ; if ( additionalHeaders !== undefined ) { headers = { ... headers , ... additionalHeaders } ; } const response = await got . post ( url , { body : this . data , throwHttpErrors : false , headers : headers , https : { rejectUnauthorized : false } , timeout : { connect : 5000 , request : 60000 } } ) ; return response ; } private serialize ( ) { const t0 = mtime ( ) ; const dataHash = crypto . createHash ( "md5" ) ; const seriesList = [ ] ; for ( const fragment of this . fragments ) { this . payloadByteCount += fragment . payloadByteCount ( ) ; const pblabels = [ ] ; for ( const [ key , val ] of Object . entries ( fragment . labels ) ) { pblabels . push ( pbTypeLabel . create ( { name : key , value : val } ) ) ; } const pbsamples = [ ] ; for ( const sample of fragment . getSamples ( ) ) { pbsamples . push ( pbTypeSample . fromObject ( { timestamp : sample . time , value : sample . value } ) ) ; dataHash . update ( sample . value . toString ( ) ) ; } const series = pbTypeTimeseries . create ( { labels : pblabels , samples : pbsamples } ) ; seriesList . push ( series ) ; fragment . setSerialized ( ) ; } const wr = pbTypeWriterequest . create ( { timeseries : seriesList } ) ; const datamd5 = dataHash . digest ( "hex" ) ; const wrbuffer = pbTypeWriterequest . encode ( wr ) . finish ( ) ; const wrBufferSnapped = snappy . compressSync ( wrbuffer ) ; const serializationTimeSeconds = mtimeDiffSeconds ( t0 ) ; return [ wrBufferSnapped , datamd5 , serializationTimeSeconds ] ; } } export function formatFloatForComp ( input ) { return mathjs . format ( input , { precision : 14 , notation : "fixed" } ) ; }