/**
 * Copyright 2021 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 fs from "fs" ; import { strict as assert } from "assert" ; import { ZonedDateTime , ZoneOffset } from "@js-joda/core" ; import got , { Response as GotResponse } from "got" ; import { Semaphore } from "await-semaphore" ; import Denque from "denque" ; import { LogSeries , LogSeriesFragmentPushRequest , LogSeriesFragment , LogSeriesFetchAndValidateOpts , LokiQueryResult } from "./logs" ; import { MetricSeries , MetricSeriesFragment , MetricSeriesFragmentPushMessage , MetricSeriesFetchAndValidateOpts } from "./metrics" ; import { LabelSet , WalltimeCouplingOptions } from "./series" ; import { sleep , mtimeDiffSeconds , mtimeDeadlineInSeconds , mtime } from "./mtime" ; import { log } from "./log" ; import * as util from "./util" ; export const START_TIME_JODA = ZonedDateTime . now ( ZoneOffset . UTC ) ; const START_TIME_MONOTONIC = mtime ( ) ; import { parseCmdlineArgs , CFG , BEARER_TOKEN } from "./args" ; import * as pm from "./prommetrics" ; interface WriteStats { nEntriesSent : number ; nPayloadBytesSent : number ; entriesSentPerSec : number ; megaPayloadBytesSentPerSec : number ; durationSeconds : number ; } interface ReadStats { nEntriesRead : number ; nPayloadBytesRead : number ; entriesReadPerSec : number ; megaPayloadBytesReadPerSec : number ; durationSeconds : number ; } export const DEFAULT_LOG_LEVEL_STDERR = "info" ; export let WALLTIME_COUPLING_PARAMS ; let tooFastMsgLoggedLastTime = BigInt ( 0 ) ; let CYCLE_START_TIME_MONOTONIC ; let CYCLE_STOP_WRITE_AFTER_SECONDS ; let COUNTER_STREAM_FRAGMENTS_PUSHED = BigInt ( 0 ) ; let COUNTER_PUSHREQUEST_STATS_LOG_THROTTLE = 0 ; let PER_STREAM_FRAGMENTS_CONSUMED_IN_CURRENT_CYCLE = { } ; function setUptimeGauge ( ) { const uptimeSeconds = mtimeDiffSeconds ( START_TIME_MONOTONIC ) ; pm . gauge_uptime . set ( Math . round ( uptimeSeconds ) ) ; } async function main ( ) { parseCmdlineArgs ( ) ; const httpServerTerminator = pm . setupPromExporter ( ) ; WALLTIME_COUPLING_PARAMS = calcWalltimeCouplingOptions ( ) ; log . info ( "cycle 1: create a fresh set of time series" ) ; let series = await createNewSeries ( cycleId ( 1 ) ) ; let cyclenum = 0 ; while ( true ) { cyclenum ++ ; if ( cyclenum > 1 ) { process . stderr . write ( "\n\n" ) ; } log . info ( "enter write/read cycle %s" , cyclenum ) ; setUptimeGauge ( ) ; const cycleid = cycleId ( cyclenum ) ; if ( cyclenum > 1 ) { if ( CFG . change_series_every_n_cycles > 0 && ( cyclenum - 1 ) % CFG . change_series_every_n_cycles === 0 ) { log . info ( `cycle ${ cyclenum }: fresh time series object(s) as ` + "of CFG.change_series_every_n_cycles" ) ; series = await createNewSeries ( cycleid ) ; } else { log . info ( "cycle %s: continue to use time series of previous cycle" , cyclenum ) ; } } pm . counter_rw_cycles . inc ( ) ; try { await performWriteReadCycle ( cyclenum , series , cycleid ) ; } catch ( err ) { log . crit ( "err during write/read cycle: %s" , err ) ; process . exit ( 1 ) ; } if ( CFG . n_cycles !== 0 ) { if ( cyclenum === CFG . n_cycles ) { log . info ( `cycles completed: ${ CFG . n_cycles }, terminate` ) ; } break ; } } log . debug ( "shutting down http server" ) ; httpServerTerminator . terminate ( ) ; } param n function cycleId ( n ) { return `${ CFG . invocation_id }-${ n }` ; } function calcFragmentTimeLeapSeconds ( ) { let sample_time_increment_ns ; if ( CFG . metrics_mode ) { sample_time_increment_ns = CFG . metrics_time_increment_ms * 1000 ; } else { sample_time_increment_ns = CFG . log_time_increment_ns ; } const fragmentTimeLeapSeconds = CFG . n_samples_per_series_fragment * ( sample_time_increment_ns / 10 ** 6 ) ; return fragmentTimeLeapSeconds ; } function calcWalltimeCouplingOptions ( ) { const fragmentTimeLeapSeconds = calcFragmentTimeLeapSeconds ( ) ; log . info ( "walltime coupling: fragmentTimeLeapSeconds: " + `${ fragmentTimeLeapSeconds . toFixed ( 2 ) } s` ) ; const minimalMaxLagSeconds = 2 * 60 ; const dynamicMaxLagSeconds = Math . ceil ( fragmentTimeLeapSeconds * 5 ) ; let actualMaxLagSeconds ; if ( dynamicMaxLagSeconds < minimalMaxLagSeconds ) { log . info ( `walltime coupling: use minimal maxLagSeconds: ${ minimalMaxLagSeconds }` ) ; actualMaxLagSeconds = minimalMaxLagSeconds ; } else { log . info ( `walltime coupling: use dynamic maxLagSeconds: ${ dynamicMaxLagSeconds } ` + `(based on fragmentTimeLeapSeconds: ${ fragmentTimeLeapSeconds . toFixed ( 2 ) } s)` ) ; actualMaxLagSeconds = dynamicMaxLagSeconds ; } return { maxLagSeconds : actualMaxLagSeconds , minLagSeconds : 0.5 * 60 } ; } async function createNewSeries ( invocationCycleId ) { const series = [ ] ; log . info ( `create ${ CFG . n_series } time series objects` ) ; if ( CFG . additional_labels !== undefined ) { log . info ( "adding additional labels: %s" , JSON . stringify ( CFG . additional_labels ) ) ; } const now = ZonedDateTime . now ( ) ; for ( let i = 1 ; i < CFG . n_series + 1 ; i ++ ) { const seriesname = `${ invocationCycleId }-${ i . toString ( ) }` ; const labelset = { } ; if ( CFG . additional_labels !== undefined ) { for ( const kvpair of CFG . additional_labels ) { labelset [ kvpair [ 0 ] ] = kvpair [ 1 ] ; } } let s ; const startTimeOffsetIntoPast = util . rndFloatFromInterval ( WALLTIME_COUPLING_PARAMS . minLagSeconds + 1.5 * calcFragmentTimeLeapSeconds ( ) , WALLTIME_COUPLING_PARAMS . maxLagSeconds - 1.5 * calcFragmentTimeLeapSeconds ( ) ) ; const starttime = now . minusSeconds ( startTimeOffsetIntoPast ) . withNano ( 0 ) ; if ( CFG . metrics_mode ) { s = new MetricSeries ( { metricName : seriesname . replace ( /-/g , "_" ) , uniqueName : seriesname , n_samples_per_series_fragment : CFG . n_samples_per_series_fragment , starttime : starttime , sample_time_increment_ns : CFG . metrics_time_increment_ms * 1000 , labelset : labelset , wtopts : WALLTIME_COUPLING_PARAMS , counterForwardLeap : pm . counter_forward_leap } ) ; } else { let wtopts = WALLTIME_COUPLING_PARAMS ; let logstarttime = starttime ; if ( CFG . log_start_time !== "" ) { log . info ( "wall time coupling disabled as of --log-start-time" ) ; wtopts = undefined ; logstarttime = ZonedDateTime . parse ( CFG . log_start_time ) ; } s = new LogSeries ( { n_samples_per_series_fragment : CFG . n_samples_per_series_fragment , n_chars_per_msg : CFG . n_chars_per_msg , starttime : logstarttime , uniqueName : seriesname , sample_time_increment_ns : CFG . log_time_increment_ns , includeTimeInMsg : true , labelset : labelset , compressability : CFG . compressability , wtopts : wtopts } ) ; } const msg = `Initialized series: ${ s }. Time of first sample: ${ s . currentTimeRFC3339Nano ( ) }` ; const logEveryN = util . logEveryNcalc ( CFG . n_series ) ; if ( i % logEveryN == 0 ) { log . info ( msg + ` (${ logEveryN - 1 } msgs like this are/will be hidden)` ) ; await sleep ( 0.0001 ) ; } else { log . debug ( msg ) ; } series . push ( s ) ; } log . info ( "time series initialization finished" ) ; return series ; } async function performWriteReadCycle ( cyclenum , series , invocationCycleId ) { CYCLE_START_TIME_MONOTONIC = mtime ( ) ; CYCLE_STOP_WRITE_AFTER_SECONDS = CFG . cycle_stop_write_after_n_seconds ; if ( CFG . cycle_stop_write_after_n_seconds_jitter !== 0 ) { const jitter = util . rndFloatFromInterval ( - CFG . cycle_stop_write_after_n_seconds_jitter , CFG . cycle_stop_write_after_n_seconds_jitter ) ; CYCLE_STOP_WRITE_AFTER_SECONDS += jitter ; log . info ( "CYCLE_STOP_WRITE_AFTER_SECONDS: (%s + %s) s" , CFG . cycle_stop_write_after_n_seconds , jitter . toFixed ( 2 ) ) ; } const arrayLengthBeforeWrite = series . length ; log . info ( "cycle %s: entering write phase" , cyclenum ) ; const writestats = await writePhase ( series ) ; assert ( arrayLengthBeforeWrite === series . length ) ; log . info ( "cycle %s: entering read phase" , cyclenum ) ; await sleep ( 1 ) ; const readstats = await readPhase ( series ) ; const report = { argv : process . argv , renderedConfig : CFG , invocationTime : util . timestampToRFC3339Nano ( START_TIME_JODA ) , cycleStats : { cycleNum : cyclenum , invocationCycleId : invocationCycleId , write : writestats , read : readstats } } ; log . debug ( "Report:\n%s" , JSON . stringify ( report , null , 2 ) ) ; const reportFilePath = `${ invocationCycleId }.report.json` ; fs . writeFileSync ( reportFilePath , JSON . stringify ( report , null , 2 ) ) ; log . info ( "wrote report to %s" , reportFilePath ) ; } async function writePhase ( streams ) { const fragmentsPushedBefore = COUNTER_STREAM_FRAGMENTS_PUSHED ; log . debug ( "reset per-series fragment push counter" ) ; PER_STREAM_FRAGMENTS_CONSUMED_IN_CURRENT_CYCLE = { } ; for ( const s of streams ) { PER_STREAM_FRAGMENTS_CONSUMED_IN_CURRENT_CYCLE [ s . uniqueName ] = 0 ; } const lt0 = mtime ( ) ; if ( CFG . skip_read ) { log . info ( "mark all streams to not keep track of validation info: --skip-read" ) ; for ( const s of streams ) { s . disableValidation ( ) ; } } else if ( CFG . read_n_series_only === 0 ) { log . info ( "mark all streams to keep track of validation info: no --skip-read, and no --read-n-streams-only" ) ; for ( const s of streams ) { s . enableValidation ( ) ; } } if ( CFG . read_n_series_only !== 0 ) { log . info ( "mark all streams to not keep track of validation info: --read-n-streams-only is set" ) ; for ( const s of streams ) { s . disableValidation ( ) ; } log . info ( "randomly marking %s/%s streams for validation" , CFG . read_n_series_only , CFG . n_series ) ; const streamsToValidate = util . randomSampleFromArray ( streams , CFG . read_n_series_only ) ; if ( CFG . read_n_series_only < 20 ) { const names = streamsToValidate . map ( s => s . uniqueName ) . join ( ", " ) ; log . info ( "selected: %s" , names ) ; } for ( const s of streamsToValidate ) { s . enableValidation ( ) ; } } const labelForValidationDurationSeconds = mtimeDiffSeconds ( lt0 ) ; log . debug ( "re-labeling series for validation info collection took %s s" , labelForValidationDurationSeconds . toFixed ( 2 ) ) ; const wt0 = mtime ( ) ; await generateAndPostFragments ( streams ) ; const writeDurationSeconds = mtimeDiffSeconds ( wt0 ) ; const nStreamFragmentsSent = Number ( COUNTER_STREAM_FRAGMENTS_PUSHED - fragmentsPushedBefore ) ; log . info ( `fragments POSTed in write phase: ${ nStreamFragmentsSent }, across ` + `${ streams . length } series` ) ; const nEntriesSent = nStreamFragmentsSent * CFG . n_samples_per_series_fragment ; let nPayloadBytesSent = nEntriesSent * CFG . n_chars_per_msg ; if ( CFG . metrics_mode ) { nPayloadBytesSent = nEntriesSent * 8 ; } const megaPayloadBytesSent = nPayloadBytesSent / 10 ** 6 ; const megaPayloadBytesSentPerSec = megaPayloadBytesSent / writeDurationSeconds ; const stats = { nEntriesSent : nEntriesSent , nPayloadBytesSent : nPayloadBytesSent , entriesSentPerSec : nEntriesSent / writeDurationSeconds , megaPayloadBytesSentPerSec : megaPayloadBytesSentPerSec , durationSeconds : writeDurationSeconds } ; log . info ( "End of write phase. Log entries / metric samples sent: %s, Payload bytes sent: %s million" , nEntriesSent , megaPayloadBytesSent . toFixed ( 2 ) ) ; log . info ( "Payload write net throughput (mean): " + megaPayloadBytesSentPerSec . toFixed ( 5 ) + " million bytes per second (assumes utf8 for logs and 12+8 " + "bytes per sample for metrics)" ) ; await sleep ( 1 ) ; return stats ; } async function throttledFetchAndValidate ( semaphore , stream ) { let fetchresult ; const st0 = mtime ( ) ; const release = await semaphore . acquire ( ) ; log . debug ( "fetchAndValidate held back by semaphore for %s s" , mtimeDiffSeconds ( st0 ) . toFixed ( 2 ) ) ; try { fetchresult = await unthrottledFetchAndValidate ( stream ) ; } catch ( err ) { log . info ( "err in sem-protected, release, re-throw" ) ; release ( ) ; throw err ; } release ( ) ; return fetchresult ; } async function unthrottledFetchAndValidate ( stream ) { if ( CFG . metrics_mode ) { const opts = { querierBaseUrl : CFG . apibaseurl , customHTTPGetFunc : httpGETRetryUntil200OrError } ; return await stream . fetchAndValidate ( opts ) ; } const opts = { querierBaseUrl : CFG . apibaseurl , additionalHeaders : { } , customLokiQueryFunc : queryLokiWithRetryOrError } ; return await stream . fetchAndValidate ( opts ) ; } async function readPhase ( streams ) { log . debug ( "entering read / validation phase" ) ; const validators = [ ] ; const vt0 = mtime ( ) ; if ( CFG . skip_read ) { log . info ( "skipping readout as of --skip-read" ) ; } else { if ( CFG . max_concurrent_reads === 0 ) { for ( const s of streams ) { if ( s . shouldBeValidated ( ) ) { validators . push ( unthrottledFetchAndValidate ( s ) ) ; } } } else { const semaphore = new Semaphore ( CFG . max_concurrent_reads ) ; for ( const s of streams ) { if ( s . shouldBeValidated ( ) ) { validators . push ( throttledFetchAndValidate ( semaphore , s ) ) ; } } } } log . info ( `created ${ validators . length } actor(s) for validation/readout` ) ; await sleep ( 1 ) ; let nSamplesReadArr ; try { nSamplesReadArr = await Promise . all ( validators ) ; } catch ( err ) { log . crit ( "error during validation: %s" , err ) ; process . exit ( 1 ) ; } if ( CFG . read_n_series_only !== 0 ) { log . debug ( "drop 'validation info' for all streams" ) ; for ( const s of streams ) { s . dropValidationInfo ( ) ; } } const nSamplesRead = nSamplesReadArr . reduce ( ( a , b ) => a + b , 0 ) ; const readDurationSeconds = mtimeDiffSeconds ( vt0 ) ; log . info ( `validation (read) took ${ readDurationSeconds . toFixed ( 1 ) } s overall. ` + `${ nSamplesRead } samples read across ${ validators . length } series (` + `${ nSamplesRead / validators . length } per series).` ) ; let nPayloadBytesRead = nSamplesRead * CFG . n_chars_per_msg ; if ( CFG . metrics_mode ) { nPayloadBytesRead = nSamplesRead * 8 ; } const megaPayloadBytesRead = nPayloadBytesRead / 10 ** 6 ; const stats = { nEntriesRead : nSamplesRead , nPayloadBytesRead : nPayloadBytesRead , entriesReadPerSec : nSamplesRead / readDurationSeconds , megaPayloadBytesReadPerSec : megaPayloadBytesRead / readDurationSeconds , durationSeconds : readDurationSeconds } ; return stats ; } export async function generateAndPostFragments ( series ) { const actors = [ ] ; const seriespool = new Denque ( series ) ; const actorCount = CFG . max_concurrent_writes ; for ( let i = 1 ; i <= actorCount ; i ++ ) { actors . push ( produceAndPOSTpushrequestsUntilCycleStopCriterion ( seriespool , CFG . n_fragments_per_push_message , actorCount , i ) ) ; } await Promise . all ( actors ) ; log . info ( "all write actors terminated" ) ; await sleep ( 1 ) ; } param seriespool param nf param actorCount param actorIndex returns async function tryToGetNFragmentsFromSeriesPool ( seriespool , nf , actorCount , actorIndex ) { const fragments = [ ] ; let candidatesPoppedSinceLastFragmentGenerated = 0 ; while ( true ) { const s = seriespool . pop ( ) ; candidatesPoppedSinceLastFragmentGenerated += 1 ; if ( s === undefined ) { log . info ( `write actor ${ actorIndex }: series pool is empty. ` + `generated ${ fragments . length } fragments (desired: ${ nf })` ) ; return fragments ; } let fragment = undefined ; let histobserved = false ; while ( true ) { const [ shiftIntoPastSeconds , f ] = s . generateNextFragmentOrSkip ( ) ; if ( ! histobserved ) { pm . hist_lag_compared_to_wall_time . observe ( shiftIntoPastSeconds ) ; histobserved = true ; } if ( f !== undefined ) { fragment = f ; candidatesPoppedSinceLastFragmentGenerated = 0 ; break ; } if ( mtimeDiffSeconds ( tooFastMsgLoggedLastTime ) > 10 ) { const shiftIntoPastMinutes = shiftIntoPastSeconds / 60 ; log . info ( `write actor ${ actorIndex }: ${ s }: current lag compared to wall time is ${ shiftIntoPastMinutes . toFixed ( 1 ) } minutes. Sample generation is too fast. Delay generating ` + "and pushing the next fragment. This may take up to " + `${ ( s . fragmentTimeLeapSeconds / 60.0 ) . toFixed ( 1 ) } minutes, ` + "the time width of a series fragment. " + "(not logged for every case)" ) ; tooFastMsgLoggedLastTime = mtime ( ) ; } pm . counter_fragment_generation_delayed . inc ( 1 ) ; if ( candidatesPoppedSinceLastFragmentGenerated >= CFG . n_series ) { await sleep ( 0.5 ) ; } const ql = seriespool . length ; if ( ql <= actorCount - 1 ) { log . debug ( `write actor ${ actorIndex }: actors are running out of work ` + `(queue length: ${ ql }): idle-watch candidate` ) ; await sleep ( 5 ) ; continue ; } seriespool . unshift ( s ) ; break ; } if ( fragment === undefined ) { continue ; } fragments . push ( fragment ) ; s . lastFragmentConsumed = fragment ; PER_STREAM_FRAGMENTS_CONSUMED_IN_CURRENT_CYCLE [ s . uniqueName ] += 1 ; if ( fragments . length === nf ) { log . debug ( `seriespool.length: ${ seriespool . length }` ) ; log . debug ( `write actor ${ actorIndex }: collected ${ nf } fragments` ) ; return fragments ; } } } async function produceAndPOSTpushrequestsUntilCycleStopCriterion ( seriespool , nfppm , actorCount , actorIndex ) { while ( true ) { const fragments = await tryToGetNFragmentsFromSeriesPool ( seriespool , nfppm , actorCount , actorIndex ) ; if ( fragments . length === 0 ) { log . info ( `write actor ${ actorIndex }: tryToGetNFragmentsFromSeriesPool() ` + "return 0 fragments: work done, exit actor" ) ; return ; } await _postFragments ( fragments , actorIndex ) ; if ( CYCLE_STOP_WRITE_AFTER_SECONDS !== 0 ) { const secondsSinceCycleStart = mtimeDiffSeconds ( CYCLE_START_TIME_MONOTONIC ) ; if ( secondsSinceCycleStart > CYCLE_STOP_WRITE_AFTER_SECONDS ) { log . info ( `write actor ${ actorIndex }: ${ secondsSinceCycleStart . toFixed ( 2 ) } seconds passed: stop` ) ; return ; } } for ( const f of fragments ) { const s = f . parent ! ; if ( CFG . cycle_stop_write_after_n_fragments !== 0 ) { if ( PER_STREAM_FRAGMENTS_CONSUMED_IN_CURRENT_CYCLE [ s . uniqueName ] === CFG . cycle_stop_write_after_n_fragments ) { log . debug ( `write actor ${ actorIndex }:  CFG.cycle_stop_write_after_n_fragments ` + `(${ CFG . cycle_stop_write_after_n_fragments }) ` + `pushed for ${ s . uniqueName }, do not put back into work queue` ) ; continue ; } } seriespool . unshift ( s ) ; log . debug ( `put ${ s . uniqueName } back into work queue` ) ; } } } async function _postFragments ( fragments , actorIndex ) { const t0 = mtime ( ) ; let pr ; if ( CFG . metrics_mode ) { pr = new MetricSeriesFragmentPushMessage ( fragments as MetricSeriesFragment [ ] ) ; } else { pr = new LogSeriesFragmentPushRequest ( fragments as LogSeriesFragment [ ] ) ; } const genduration = mtimeDiffSeconds ( t0 ) ; const firstseries = fragments [ 0 ] . parent as LogSeries | MetricSeries ; const fcount = fragments . length ; const name = `write actor ${ actorIndex }: _postFragments((nstreams=${ fcount }, ` + `first=${ firstseries . promQueryString ( ) })` ; if ( COUNTER_PUSHREQUEST_STATS_LOG_THROTTLE < 1 || COUNTER_PUSHREQUEST_STATS_LOG_THROTTLE % 200 == 0 ) { const firstFragment = pr . fragments [ 0 ] ; const lastFragment = pr . fragments . slice ( - 1 ) [ 0 ] ; log . info ( `${ name }: generated pushrequest msg in ${ genduration . toFixed ( 2 ) } s with ` + `${ fcount } series ` + `(first: ${ firstFragment . parent ?. uniqueName }, index ${ firstFragment . indexString ( 3 ) } -- ` + `last: ${ lastFragment . parent ?. uniqueName }, index ${ lastFragment . indexString ( 3 ) }), ` + `size: ${ pr . dataLengthMiB . toFixed ( 4 ) } MiB). ` + "POST it (not logged for every case)." ) ; } COUNTER_PUSHREQUEST_STATS_LOG_THROTTLE ++ ; const postT0 = mtime ( ) ; try { await customPostWithRetryOrError ( pr , CFG . apibaseurl ) ; } catch ( err ) { log . crit ( "consider critical: %s" , err ) ; process . exit ( 1 ) ; } const postDurationSeconds = mtimeDiffSeconds ( postT0 ) ; pm . hist_duration_post_with_retry_seconds . observe ( postDurationSeconds ) ; COUNTER_STREAM_FRAGMENTS_PUSHED = COUNTER_STREAM_FRAGMENTS_PUSHED + BigInt ( fcount ) ; pm . counter_fragments_pushed . inc ( fcount ) ; pm . counter_log_entries_pushed . inc ( CFG . n_samples_per_series_fragment * fcount ) ; pm . counter_payload_bytes_pushed . inc ( Number ( pr . payloadByteCount ) ) ; pm . counter_serialized_fragments_bytes_pushed . inc ( pr . dataLengthBytes ) ; pm . gauge_last_http_request_body_size_bytes . set ( pr . dataLengthBytes ) ; } function calcDelayBetweenPostRetries ( attempt ) { if ( attempt === 1 ) { return 0 ; } const expBackoff = 1.6 ** attempt / 2.56 ; const unjitteredDelay = expBackoff * CFG . retry_post_min_delay_seconds ; const jitter = util . rndFloatFromInterval ( - CFG . retry_post_jitter * unjitteredDelay , CFG . retry_post_jitter * unjitteredDelay ) ; const backoffPlusJitter = unjitteredDelay + jitter ; log . debug ( `calcDelayBetweenPostRetries: expBackoff: ${ expBackoff . toFixed ( 2 ) }, ` + `unjitteredDelay: ${ unjitteredDelay . toFixed ( 2 ) }, jitter: ${ jitter . toFixed ( 2 ) }  ` + `-> ${ backoffPlusJitter . toFixed ( 2 ) }` ) ; if ( backoffPlusJitter > CFG . retry_post_max_delay_seconds ) { log . debug ( "calcDelayBetweenPostRetries: return cap: %s" , CFG . retry_post_max_delay_seconds ) ; return CFG . retry_post_max_delay_seconds ; } return backoffPlusJitter ; } async function customPostWithRetryOrError ( pr , baseUrl ) { const deadline = mtimeDeadlineInSeconds ( CFG . retry_post_deadline_seconds ) ; let attempt = 0 ; let previousPushSuccessAmbiguous = false ; while ( true ) { attempt ++ ; const delay = calcDelayBetweenPostRetries ( attempt ) ; if ( attempt > 1 ) { log . info ( "POST %s: schedule to perform attempt %s in %s s" , pr , attempt , delay . toFixed ( 2 ) ) ; await sleep ( delay ) ; } const overDeadline = mtimeDiffSeconds ( deadline ) ; if ( overDeadline > 0 ) { throw new Error ( `Failed to POST ${ pr }: attempt ${ attempt } would trigger ` + `${ overDeadline . toFixed ( 2 ) } s over the deadline (${ CFG . retry_post_deadline_seconds } s).` ) ; } setUptimeGauge ( ) ; let pushpath = "/loki/api/v1/push" ; if ( CFG . metrics_mode ) { pushpath = "/api/v1/push" ; } let response ; try { response = await httpPostProtobuf ( `${ baseUrl }${ pushpath }` , pr . data , pr . postHeaders ) ; } catch ( e ) { if ( e instanceof got . RequestError ) { pm . counter_post_non_http_errors . inc ( ) ; log . warning ( `POST ${ pr }: attempt ${ attempt }: transient problem: ${ e . message }` ) ; if ( e . message . includes ( "Timeout awaiting 'request'" ) || e . message . includes ( "Timeout awaiting 'response'" ) ) { previousPushSuccessAmbiguous = true ; log . warning ( `POST ${ pr }: previousPushSuccessAmbiguous: set as of a ` + "got http client timeout for the 'request' or 'response' " + "phase -> next 'out or order' error not fatal" ) ; } continue ; } else { throw e ; } } pm . counter_post_responses . inc ( { statuscode : response . statusCode } ) ; if ( [ 200 , 204 ] . includes ( response . statusCode ) ) { util . logHTTPResponseLight ( response , `${ pr }` ) ; markPushMessageAsSuccessfullySent ( pr ) ; return ; } if ( response . statusCode === 429 ) { log . info ( `POST ${ pr }: 429 resp, sleep 2, body[:200]: ${ response . body . slice ( 0 , 200 ) }` ) ; continue ; } util . logHTTPResponse ( response , `${ pr }` ) ; if ( response . statusCode . toString ( ) . startsWith ( "4" ) ) { if ( response . statusCode === 400 ) { if ( response . body . includes ( "out of order sample" ) ) { if ( previousPushSuccessAmbiguous ) { log . warning ( "POST %s: saw 'out of order' error but do not treat fatal as of previous ambiguous push result" , pr ) ; markPushMessageAsSuccessfullySent ( pr ) ; return ; } } } throw new Error ( `POST ${ pr }: Bad HTTP request (see log above)` ) ; } if ( response . statusCode === 500 ) { const needles = [ "DeadlineExceeded" , "code = Unavailable desc = transport is closing" ] ; for ( const n in needles ) { if ( response . body . includes ( n ) ) { previousPushSuccessAmbiguous = true ; log . warning ( `POST ${ pr }: previousPushSuccessAmbiguous: set as of a` + `500(${ n }) response, next 'out or order' error not fatal` ) ; } } } log . info ( "POST %s: Treat as transient problem, retry after sleep" , pr ) ; } } function markPushMessageAsSuccessfullySent ( pr ) { for ( const f of pr . fragments ) { if ( f . parent ! . shouldBeValidated ( ) ) { f . buildStatisticsAndDropData ( ) ; f . parent ! . rememberFragmentForValidation ( f ) ; } } } async function httpGETRetryUntil200OrError ( url , gotRequestOptions ) { const maxRetries = 10 ; for ( let attempt = 1 ; attempt <= maxRetries ; attempt ++ ) { setUptimeGauge ( ) ; let response ; if ( BEARER_TOKEN ) { if ( gotRequestOptions . headers === undefined ) { gotRequestOptions . headers = { } ; } gotRequestOptions . headers [ "Authorization" ] = `Bearer ${ BEARER_TOKEN }` ; } try { response = await got ( url , gotRequestOptions ) ; } catch ( e ) { if ( e instanceof got . RequestError ) { pm . counter_get_non_http_errors . inc ( ) ; log . warning ( `Query: attempt ${ attempt }/${ maxRetries } failed with ${ e . message }` ) ; await sleep ( 3.0 ) ; continue ; } else { throw e ; } } pm . counter_get_responses . inc ( { statuscode : response . statusCode } ) ; if ( response . statusCode === 200 ) { util . logHTTPResponseLight ( response ) ; return response ; } if ( response . statusCode === 429 ) { log . info ( `429 resp, sleep 2, body[:200]: ${ response . body . slice ( 0 , 200 ) }` ) ; await sleep ( 3.0 ) ; continue ; } util . logHTTPResponse ( response ) ; if ( response . statusCode . toString ( ) . startsWith ( "4" ) ) { throw new Error ( "Bad HTTP request (see log above)" ) ; } log . info ( "Treat as transient problem, retry" ) ; } throw new Error ( `Failed to GET after ${ maxRetries } attempts. Request options: ${ JSON . stringify ( gotRequestOptions , null , 2 ) }` ) ; } async function queryLokiWithRetryOrError ( lokiQuerierBaseUrl , additionalHeaders , queryParams , expectedEntryCount , chunkIndex , stream ) { log . debug ( "looker-specific query func" ) ; async function _queryAndCountEntries ( lokiQuerierBaseUrl , queryParams ) { const url = `${ CFG . apibaseurl }/loki/api/v1/query_range` ; const gotRequestOptions = { retry : 0 , throwHttpErrors : false , searchParams : new URLSearchParams ( queryParams ) , timeout : util . httpTimeoutSettings , https : { rejectUnauthorized : false } } ; const response = await httpGETRetryUntil200OrError ( url , gotRequestOptions ) ; const data = JSON . parse ( response . body ) ; if ( data . data === undefined || data . data . result === undefined ) { throw new Error ( `unexpected JSON doc: ${ JSON . stringify ( data , null , 2 ) . slice ( 0 , 400 ) }}` ) ; } if ( data . data . result . length !== 1 ) { throw new Error ( `data.data.result.length: expected 1, but got ${ data . data . result . length }}` ) ; } const entrycount = data . data . result [ 0 ] . values . length ; if ( entrycount === expectedEntryCount ) { const labels = data . data . result [ 0 ] . stream ; const result = { entries : data . data . result [ 0 ] . values , labels : labels , textmd5 : "disabled" } ; return result ; } throw new Error ( `unexpected entry count returned in query result: ${ entrycount } ` + `(expected: ${ expectedEntryCount }). stream: ${ stream . uniqueName } chunkIndex: ${ chunkIndex }` ) ; } const maxRetries = 3 ; for ( let attempt = 1 ; attempt <= maxRetries ; attempt ++ ) { try { return await _queryAndCountEntries ( lokiQuerierBaseUrl , queryParams ) ; } catch ( err ) { pm . counter_unexpected_query_results . inc ( ) ; log . warning ( `_queryAndCountEntries() failed with \`${ err . message }\`, retry soon` ) ; await sleep ( 5.0 ) ; } } throw new Error ( `_queryAndCountEntries() failed ${ maxRetries } times. ` + `stream: ${ stream . uniqueName } chunkIndex: ${ chunkIndex }` ) ; } async function httpPostProtobuf ( url , data , headers ) { if ( BEARER_TOKEN ) { headers = { ... headers , Authorization : `Bearer ${ BEARER_TOKEN }` } ; } const response = await got . post ( url , { body : data , throwHttpErrors : false , headers : headers , https : { rejectUnauthorized : false } , timeout : { connect : 12000 , secureConnect : 5000 , send : 20000 , response : 60000 , request : 80000 } } ) ; return response ; } if ( require . main === module ) { process . on ( "SIGINT" , function ( ) { log . info ( "Received SIGINT, exiting" ) ; process . exit ( 1 ) ; } ) ; process . on ( "unhandledRejection" , err => { throw err ; } ) ; main ( ) ; }