/**
 * 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 crypto from "crypto" ; import got from "got" ; import { ZonedDateTime , LocalDateTime , ZoneOffset } from "@js-joda/core" ; import { log } from "../log" ; import { mtimeDiffSeconds , mtimeDeadlineInSeconds , mtime , sleep } from "../mtime" ; import { timestampToRFC3339Nano , rndstringFast , rndstringFastBoringFill , httpTimeoutSettings , logHTTPResponse } from "../util" ; import { LogSample , LogSampleTimestamp , LogSeriesFragment , LogSeriesFragmentStats , logqlLabelString } from "./index" ; import { TimeseriesBase , LabelSet , WalltimeCouplingOptions } from "../series" ; export interface LogSeriesOpts { n_samples_per_series_fragment : number ; n_chars_per_msg : number ; starttime : ZonedDateTime ; sample_time_increment_ns : number ; includeTimeInMsg : boolean ; uniqueName : string ; labelset : LabelSet | undefined ; compressability : string ; wtopts ? : WalltimeCouplingOptions ; counterForwardLeap ? : any ; } type TypeHttpHeaderDict = Record < string , string > ; type TypeQueryParamDict = Record < string , string > ; type CustomQueryFuncSigType = ( arg0 : string , arg1 : TypeHttpHeaderDict , arg2 : TypeQueryParamDict , arg3 : number , arg4 : number , arg5 : LogSeries ) => Promise < LokiQueryResult > ; export interface LogSeriesFetchAndValidateOpts { querierBaseUrl : string ; additionalHeaders : Record < string , string > ; customLokiQueryFunc ? : CustomQueryFuncSigType ; } export class LogSeries extends TimeseriesBase < LogSeriesFragment > { private currentSeconds : number ; private currentNanos : number ; private includeTimeInMsg : boolean ; private firstEntryGenerated : boolean ; private genChars : ( n : number ) => string ; n_chars_per_msg : number ; nFragmentsSuccessfullySentSinceLastValidate : number ; constructor ( opts ) { super ( opts ) ; this . currentSeconds = opts . starttime . toEpochSecond ( ) ; this . currentNanos = opts . starttime . nano ( ) ; this . firstEntryGenerated = false ; this . n_chars_per_msg = opts . n_chars_per_msg ; this . nFragmentsConsumed = 0 ; this . includeTimeInMsg = opts . includeTimeInMsg ; this . nFragmentsSuccessfullySentSinceLastValidate = 0 ; if ( this . sample_time_increment_ns > 999999999 ) throw Error ( "sample_time_increment_ns must be smaller than 1 s" ) ; if ( this . includeTimeInMsg && this . n_chars_per_msg < 19 ) { throw Error ( "timestamp consumes 18+1 characters" ) ; } switch ( opts . compressability ) { case "min" : this . genChars = rndstringFast ; break ; case "max" : this . genChars = ( n ) => { return "a" . repeat ( n ) ; } ; break ; case "medium" : this . genChars = ( n ) => { return rndstringFastBoringFill ( ( n / 2 ) >> 0 , ( ( n / 2 ) >> 0 ) + ( n % 2 ) ) ; } ; break ; default : throw new Error ( `bad compressability value: ${ opts . compressability }` ) ; } } protected buildLabelSetFromOpts ( opts ) { let ls ; if ( opts . labelset !== undefined ) { ls = opts . labelset ; ls . looker_uniquename = opts . uniqueName ; } else { ls = { looker_uniquename : opts . uniqueName } ; } return ls ; } public promQueryString ( ) { return `{looker_uniquename="${ this . uniqueName }"}` ; } private buildMsgText ( ) { let text ; if ( ! this . includeTimeInMsg ) text = this . genChars ( this . n_chars_per_msg ) ; else { const timestring = `${ this . currentSeconds }${ this . currentNanos . toString ( ) . padStart ( 9 , "0" ) }` ; text = `${ timestring }:${ this . genChars ( this . n_chars_per_msg - 19 ) }` ; } return text ; } protected lastSampleSecondsSinceEpoch ( ) { return this . currentSeconds + this . currentNanos / 10 ** 9 ; } protected nextSample ( ) { if ( this . firstEntryGenerated ) { this . currentNanos += this . sample_time_increment_ns ; if ( this . currentNanos > 999999999 ) { this . currentNanos = this . currentNanos - 10 ** 9 ; this . currentSeconds += 1 ; } } const ts = { seconds : this . currentSeconds , nanos : this . currentNanos } ; this . firstEntryGenerated = true ; return new LogSample ( this . buildMsgText ( ) , ts ) ; } protected generateNextFragment ( ) { const f = new LogSeriesFragment ( 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 ; } param lokiBaseUrl public async postFragmentsToLoki ( nFragments , lokiBaseUrl , 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 pushrequest = fragment . serialize ( ) ; const genduration = mtimeDiffSeconds ( t0 ) ; if ( fragment . index < 5 || fragment . index % 10 === 0 ) { log . info ( "Generated PR for stream %s in %s s, push %s MiB (%s entries)" , this . uniqueName , genduration . toFixed ( 2 ) , pushrequest . dataLengthMiB . toFixed ( 4 ) , pushrequest . fragments [ 0 ] . sampleCount ( ) ) ; } await pushrequest . postWithRetryOrError ( lokiBaseUrl , 3 , additionalHeaders ) ; if ( this . shouldBeValidated ( ) ) { assert ( this . postedFragmentsSinceLastValidate ) ; this . postedFragmentsSinceLastValidate . push ( fragment ) ; } } } protected leapForward ( n ) { assert ( this . walltimeCouplingOptions ) ; this . currentSeconds += Number ( n ) ; } public currentTime ( ) { const tsSecondResolution = LocalDateTime . ofEpochSecond ( this . currentSeconds , ZoneOffset . UTC ) ; const tsNanoSecondResolution = LocalDateTime . of ( tsSecondResolution . year ( ) , tsSecondResolution . month ( ) , tsSecondResolution . dayOfMonth ( ) , tsSecondResolution . hour ( ) , tsSecondResolution . minute ( ) , tsSecondResolution . second ( ) , this . currentNanos ) ; return tsNanoSecondResolution . atZone ( ZoneOffset . UTC ) ; } public currentTimeRFC3339Nano ( ) { return timestampToRFC3339Nano ( this . currentTime ( ) ) ; } private queryParamsForFragment ( fragment ) { assert ( fragment . stats ) ; const stats = fragment . stats as LogSeriesFragmentStats ; if ( stats . sampleCount > 60000 ) { throw new Error ( "too many samples to fetch in one go -- needs feature: N queries per fragment" ) ; } const qparams = { query : logqlLabelString ( this . labels ) , direction : "FORWARD" , limit : stats . sampleCount . toString ( ) , start : stats . timeOfFirstEntry . toString ( ) , end : ( stats . timeOfLastEntry + BigInt ( 1 ) ) . toString ( ) } ; log . debug ( "query params: %s" , qparams ) ; return qparams ; } protected async fetchAndValidateFragment ( fragment , opts ) { assert ( fragment . stats ) ; const stats = fragment . stats as LogSeriesFragmentStats ; const qparams = this . queryParamsForFragment ( fragment ) ; let result ; if ( opts . customLokiQueryFunc !== undefined ) { result = await opts . customLokiQueryFunc ( opts . querierBaseUrl , opts . additionalHeaders , qparams , Number ( stats . sampleCount ) , fragment . index , this ) ; } else { result = await waitForLokiQueryResult ( opts . querierBaseUrl , opts . additionalHeaders , qparams , Number ( stats . sampleCount ) , false , 1 , false ) ; } const logTextHash = crypto . createHash ( "md5" ) ; for ( const e of result . entries ) { logTextHash . update ( e [ 1 ] ) ; } const textmd5fromq = logTextHash . digest ( "hex" ) ; if ( stats . textmd5 !== textmd5fromq ) { throw new Error ( "log time series fragment text checksum mismatch: " + `series: ${ fragment . parent ! . promQueryString } ` + `fragment: ${ fragment . index } + \n fragment stats: ` + JSON . stringify ( stats , null , 2 ) + `\nchecksum from query result: ${ textmd5fromq }` ) ; } return result . entries . length ; } } export interface LokiQueryResult { entries : Array < [ string , string ] > ; labels : LabelSet ; textmd5 : string ; } /**
 * Expected to throw got.RequestError, handle in caller if desired.
 */ async function queryLoki ( baseUrl , queryParams , additionalHeaders ) { const url = `${ baseUrl }/loki/api/v1/query_range` ; const options = { throwHttpErrors : false , searchParams : queryParams , timeout : httpTimeoutSettings , headers : additionalHeaders , https : { rejectUnauthorized : false } } ; const response = await got ( url , options ) ; if ( response . statusCode !== 200 ) logHTTPResponse ( response ) ; return JSON . parse ( response . body ) ; } async function waitForLokiQueryResult ( lokiQuerierBaseUrl , additionalHeaders , queryParams , expectedSampleCount , logDetails = true , expectedStreamCount = 1 , buildhash = true , maxWaitSeconds = 30 ) { const deadline = mtimeDeadlineInSeconds ( maxWaitSeconds ) ; if ( logDetails ) { log . info ( `Enter Loki query loop, wait for expected result, deadline ${ maxWaitSeconds } s.
Query parameters: ${ JSON . stringify ( queryParams , Object . keys ( queryParams ) . sort ( ) , 2 ) }` ) ; } const qparms = new URLSearchParams ( queryParams ) ; let queryCount = 0 ; const t0 = mtime ( ) ; while ( true ) { if ( mtime ( ) > deadline ) { log . error ( "query deadline hit" ) ; break ; } queryCount += 1 ; let result ; try { result = await queryLoki ( lokiQuerierBaseUrl , qparms , additionalHeaders ) ; } catch ( e ) { if ( e instanceof got . RequestError ) { log . info ( `waitForLokiQueryResult() loop: http request failed: ${ e . message } -- ignore, proceed with next iteration` ) ; continue ; } else if ( e instanceof SyntaxError ) { log . warning ( `waitForLokiQueryResult() loop: http parse response failed: ${ e . message } -- ignore, proceed with next iteration` ) ; continue ; } else { throw e ; } } if ( result . status === undefined ) { log . warning ( "no `status` property in response doc: %s" , JSON . stringify ( result ) ) ; await sleep ( 1 ) ; continue ; } if ( result . status !== "success" ) { log . warning ( "status property is not `success`: %s" , JSON . stringify ( result . status ) ) ; await sleep ( 1 ) ; continue ; } const streams = result . data . result ; if ( streams . length === 0 ) { if ( queryCount % 10 === 0 ) { log . info ( "queried %s times, no log entries seen yet" , queryCount ) ; } await sleep ( 0.5 ) ; continue ; } if ( logDetails ) log . info ( "query %s response data:\n%s" , queryCount , JSON . stringify ( result , null , 2 ) ) ; if ( expectedStreamCount !== 0 ) { assert . equal ( streams . length , expectedStreamCount ) ; } assert ( "values" in streams [ 0 ] ) ; const sampleCount = streams [ 0 ] [ "values" ] . length ; log . info ( "expected nbr of query results: %s, got %s" , expectedSampleCount , sampleCount ) ; if ( expectedSampleCount === undefined || sampleCount === expectedSampleCount ) { log . info ( "got expected result in query %s after %s s" , queryCount , mtimeDiffSeconds ( t0 ) . toFixed ( 2 ) ) ; const labels = streams [ 0 ] . stream ; let textmd5 = "disabled" ; if ( buildhash ) { const logTextHash = crypto . createHash ( "md5" ) ; for ( const entry of streams [ 0 ] [ "values" ] ) { logTextHash . update ( entry [ 1 ] ) ; } textmd5 = logTextHash . digest ( "hex" ) ; } const result = { entries : streams [ 0 ] [ "values" ] , labels : labels , textmd5 : textmd5 } ; return result ; } if ( sampleCount < expectedSampleCount ) { log . info ( "not enough entries returned yet, waiting" ) ; await sleep ( 1 ) ; continue ; } else throw new Error ( "too many entries returned in query result" ) ; } throw new Error ( `Expectation not fulfilled within ${ maxWaitSeconds } s` ) ; }