/**
 * 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 { strict as assert } from "assert" ; import { ZonedDateTime } from "@js-joda/core" ; import got , { Response as GotResponse } from "got" ; import Long from "long" ; import { logqlLabelString } from "../logs" ; import { mtimeDiffSeconds , mtime , sleep } from "../mtime" ; import { log } from "../log" ; import { logHTTPResponse , httpTimeoutSettings } from "../util" ; import * as mathjs from "mathjs" ; import { TimeseriesBase , LabelSet , WalltimeCouplingOptions } from "../series" ; import { MetricSample , MetricSeriesFragment , MetricSeriesFragmentStats , formatFloatForComp } from "./index" ; export interface MetricSeriesOpts { metricName : string ; starttime : ZonedDateTime ; uniqueName : string ; labelset : LabelSet | undefined ; sample_time_increment_ns : number ; n_samples_per_series_fragment : number ; wtopts ? : WalltimeCouplingOptions ; counterForwardLeap ? : any ; } export interface MetricSeriesFetchAndValidateOpts { querierBaseUrl : string ; additionalHeaders ? : Record < string , string > ; customHTTPGetFunc ? : ( url : string , gotRequestOptions : any ) => Promise < GotResponse < string > > ; } export class MetricSeries extends TimeseriesBase < MetricSeriesFragment > { private millisSinceEpochOfLastGeneratedSample : Long ; private metrics_time_increment_ms : Long ; private fragmentWidthSecondsForQuery : BigInt ; private lastValue : number ; metricName : string ; nFragmentsSuccessfullySentSinceLastValidate : number ; constructor ( opts ) { super ( opts ) ; if ( this . walltimeCouplingOptions === undefined ) { log . info ( "MetricSeries requires walltimeCouplingOptions to be defined. " + "Fall back to using default parameters." ) ; this . walltimeCouplingOptions = { maxLagSeconds : 30 * 60 , minLagSeconds : 2 * 60 } ; this . validateWtOpts ( this . walltimeCouplingOptions ) ; } this . metricName = opts . metricName ; if ( opts . starttime . nano ( ) != 0 ) { throw new Error ( "start time must not have fraction of seconds" ) ; } if ( opts . sample_time_increment_ns % 1000 !== 0 ) { throw new Error ( "for metrics, sample_time_increment_ns must be integer multiple of 1000" ) ; } const sampleTimeIncrementMs = opts . sample_time_increment_ns / 1000 ; if ( sampleTimeIncrementMs > 1000 ) { if ( sampleTimeIncrementMs % 1000 !== 0 ) { throw new Error ( "the time increment between metric samples must be an integer  " + "multiple of one second if it is larger than one second" ) ; } } if ( sampleTimeIncrementMs < 1000 ) { if ( ( opts . n_samples_per_series_fragment * sampleTimeIncrementMs ) % 1000 !== 0 ) { throw new Error ( "with metrics_time_increment_ms < 1000 choose " + "n_samples_per_series_fragment so that " + "n_samples_per_series_fragment * metrics_time_increment_ms = multiple of 1000" ) ; } } const fragmentWidthSeconds = ( ( opts . n_samples_per_series_fragment - 1 ) * sampleTimeIncrementMs ) / 1000.0 ; if ( opts . n_samples_per_series_fragment > 1 ) { this . fragmentWidthSecondsForQuery = BigInt ( Math . ceil ( fragmentWidthSeconds ) ) ; } else { this . fragmentWidthSecondsForQuery = BigInt ( 1 ) ; } this . metrics_time_increment_ms = Long . fromInt ( sampleTimeIncrementMs ) ; const starttime_fractional_part_as_ms = Long . fromInt ( Math . floor ( opts . starttime . nano ( ) / 10 ** 6 ) ) ; const starttime_seconds_part_as_ms = Long . fromInt ( opts . starttime . toEpochSecond ( ) ) . multiply ( Long . fromInt ( 1000 ) ) ; this . millisSinceEpochOfLastGeneratedSample = starttime_seconds_part_as_ms . add ( starttime_fractional_part_as_ms ) . subtract ( this . metrics_time_increment_ms ) ; this . nFragmentsSuccessfullySentSinceLastValidate = 0 ; this . nSamplesValidatedSoFar = BigInt ( 0 ) ; this . lastValue = Number ( ( ( Math . random ( ) - 0.5 ) * 10.0 ) . toFixed ( 1 ) ) ; } protected buildLabelSetFromOpts ( opts ) { let ls ; if ( opts . labelset !== undefined ) { ls = opts . labelset ; ls . uniquename = opts . uniqueName ; ls . __name__ = opts . metricName ; } else { ls = { uniquename : opts . uniqueName , __name__ : opts . metricName } ; } return ls ; } public promQueryString ( ) { return `${ this . metricName }{uniquename="${ this . uniqueName }"}` ; } private nextValue ( ) { if ( Math . random ( ) >= 0.5 ) { this . lastValue = this . lastValue + 0.1 ; } else { this . lastValue = this . lastValue - 0.1 ; } return this . lastValue ; } protected nextSample ( ) { this . millisSinceEpochOfLastGeneratedSample = this . millisSinceEpochOfLastGeneratedSample . add ( this . metrics_time_increment_ms ) ; return new MetricSample ( this . nextValue ( ) , this . millisSinceEpochOfLastGeneratedSample ) ; } public currentTimeRFC3339Nano ( ) { return this . millisSinceEpochOfLastGeneratedSample . toString ( ) ; } protected lastSampleSecondsSinceEpoch ( ) { return this . millisSinceEpochOfLastGeneratedSample . divide ( 1000 ) . toNumber ( ) ; } protected leapForward ( n ) { assert ( this . walltimeCouplingOptions ) ; this . millisSinceEpochOfLastGeneratedSample = this . millisSinceEpochOfLastGeneratedSample . add ( Number ( n ) * 1000 ) ; } protected generateNextFragment ( ) { const f = new MetricSeriesFragment ( this . labels , this . nFragmentsConsumed + 1 , this ) ; for ( let i = 0 ; i < this . n_samples_per_series_fragment ; i ++ ) { f . addSample ( this . nextSample ( ) ) ; } this . nFragmentsConsumed += 1 ; return f ; } /**
     * Unbuffered POST to Cortex/Prometheus remote_write API:
     * generate/post/generate/post sequentially,  in that order ("buffered" would
     * be to pre-generate fragment and put them into a buffer so that when one
     * POST is done the next could start _immediately_)
     */ public async postFragmentsToCortex ( nFragments , cortexBaseUrl , additionalHeaders ? ) { for ( let i = 1 ; i <= nFragments ; i ++ ) { let fragment ; while ( true ) { const [ shiftIntoPastSeconds , f ] = this . generateNextFragmentOrSkip ( ) ; if ( f !== undefined ) { fragment = f ; break ; } log . debug ( `${ this }: current lag compared to wall time ` + `(${ shiftIntoPastSeconds . toFixed ( 1 ) } s)` + "is too small. Delay fragment generation." ) ; await sleep ( 5 ) ; } const t0 = mtime ( ) ; const pm = fragment . serialize ( ) ; const genduration = mtimeDiffSeconds ( t0 ) ; if ( fragment . index < 5 || fragment . index % 10 === 0 ) { log . info ( "Generated push message for series %s in %s s, push %s MiB (%s entries)" , this . uniqueName , genduration . toFixed ( 2 ) , pm . dataLengthMiB . toFixed ( 4 ) , pm . fragments [ 0 ] . sampleCount ( ) ) ; } await pm . postWithRetryOrError ( cortexBaseUrl , 3 , additionalHeaders ) ; fragment . buildStatisticsAndDropData ( ) ; if ( this . shouldBeValidated ( ) ) { assert ( this . postedFragmentsSinceLastValidate ) ; this . postedFragmentsSinceLastValidate . push ( fragment ) ; } } } private queryParamsForFragment ( fragment ) { assert ( fragment . stats ) ; const stats = fragment . stats as MetricSeriesFragmentStats ; if ( ! Number . isInteger ( stats . timeMillisSinceEpochLast ) ) { log . error ( "not integer: %s" , stats . timeMillisSinceEpochLast ) ; throw new Error ( "fragment.stats.timeMillisSinceEpochLast must be an integer value" ) ; } const timeRightBoundarySeconds = Math . ceil ( stats . timeMillisSinceEpochLast / 1000.0 ) ; log . debug ( "timeRightBoundarySeconds: %s" , timeRightBoundarySeconds ) ; const params = { query : `${ logqlLabelString ( this . labels ) }[${ this . fragmentWidthSecondsForQuery }s]` , time : timeRightBoundarySeconds } ; log . debug ( "query params: %s" , params ) ; return params ; } protected async fetchAndValidateFragment ( fragment , opts ) { const url = `${ opts . querierBaseUrl }/api/v1/query` ; const qparams = this . queryParamsForFragment ( fragment ) ; let headers = { } ; if ( opts . additionalHeaders !== undefined ) { headers = { ... headers , ... opts . additionalHeaders } ; } const gotRequestOptions = { retry : 0 , throwHttpErrors : false , searchParams : qparams , headers : headers , timeout : httpTimeoutSettings , https : { rejectUnauthorized : false } } ; let response ; if ( opts . customHTTPGetFunc !== undefined ) { response = await opts . customHTTPGetFunc ( url , gotRequestOptions ) ; } else { response = await got ( url , gotRequestOptions ) ; if ( response . statusCode != 200 ) logHTTPResponse ( response ) ; } const data = JSON . parse ( response . body ) ; const resultArray = data [ "data" ] [ "result" ] ; log . debug ( "number of series in query result: %s" , resultArray . length ) ; if ( resultArray . length != 1 ) { log . error ( "unexpected response data: %s" , JSON . stringify ( data , null , 2 ) ) ; throw Error ( "not precisely one series in result" ) ; } const values = [ ] ; const timestamps = [ ] ; for ( const sample of resultArray [ 0 ] . values ) { timestamps . push ( sample [ 0 ] ) ; values . push ( Number ( sample [ 1 ] ) ) ; } if ( this . metrics_time_increment_ms . toNumber ( ) < 1000 ) { if ( values . length === this . n_samples_per_series_fragment + 1 ) { log . debug ( "last sample in query result belongs to next fragment, ignore" ) ; timestamps . pop ( ) ; values . pop ( ) ; } } const stats = { min : formatFloatForComp ( mathjs . min ( values ) ) , max : formatFloatForComp ( mathjs . max ( values ) ) , var : formatFloatForComp ( mathjs . variance ( values ) ) , timeMillisSinceEpochFirst : Math . trunc ( timestamps [ 0 ] * 1000 ) , timeMillisSinceEpochLast : Math . trunc ( timestamps . slice ( - 1 ) [ 0 ] * 1000 ) , sampleCount : BigInt ( values . length ) } ; log . debug ( "fragmentStats reference: %s" , fragment . stats ) ; log . debug ( "for comparison: %s" , stats ) ; assert . deepEqual ( fragment . stats , stats ) ; this . nSamplesValidatedSoFar += BigInt ( values . length ) ; return values . length ; } }