/**
 * 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 { strict as assert } from "assert" ; import { ZonedDateTime , ZoneOffset } from "@js-joda/core" ; import { LogSeriesOpts , LogSeriesFragment , LogSample , LogSeriesFetchAndValidateOpts , LogSeriesFragmentStats } from "./logs" ; import { MetricSeriesOpts , MetricSeriesFragment , MetricSample , MetricSeriesFetchAndValidateOpts , MetricSeriesFragmentStats } from "./metrics" ; import { log } from "./log" ; export interface LabelSet { [ key ] : string ; } export interface WalltimeCouplingOptions { maxLagSeconds : number ; minLagSeconds : number ; } export interface FragmentStatsBase { sampleCount : bigint ; } export abstract class SampleBase < ValueType , TimeType > { public value : ValueType ; public time : TimeType ; constructor ( value , time ) { this . value = value ; this . time = time ; } } export abstract class FragmentBase < SampleType , ParentType > { /** The label set defining the time series this fragment is part of. */ public labels : LabelSet ; /** The individual samples in this time series fragment */ protected samples : Array < SampleType > ; /** Sequential number for locating fragment in the time series. Set by caller. */ public index : number ; /** The time series that this fragment is part of (the "parent"). */ public parent : ParentType | undefined ; /**
     * An object representing the samples in this fragment -- is built upon
     * serialization, i.e. when this.serialized is `true` -- in the
     * implementation, make sure this is built once and not changed afterwards
     */ public stats : LogSeriesFragmentStats | MetricSeriesFragmentStats | undefined ; /**
     * For internal book-keeping: is this 'closed' (has been serialized, no more
     * samples can be added) or can samples still be added?
     */ protected serialized : boolean ; /**
     * Return number of payload bytes in this fragment.
     *
     * For a Prometheus metric sample, that's a double precision float (8 bytes)
     * for sample value, and an int64 per sample timestamp, i.e. 16 bytes per
     * sample.
     *
     * For a Loki log sample, that's a protobuf timestamp (int64 + int32) per
     * sample (12 bytes), and the size of the sample value text encoded as UTF-8.
     */ abstract payloadByteCount ( ) ; constructor ( labels , index = 0 , parentSeries = undefined ) { this . labels = labels ; this . samples = new Array < SampleType > ( ) ; this . index = index ; this . parent = parentSeries ; this . serialized = false ; } /**
     * Return shallow copy so that mutation of the returned array does not have
     * side effects in here. However, if individual samples were to be mutated
     * this would take effect here, too.*/ public getSamples ( ) { return [ ... this . samples ] ; } /**
     * Return the current number of samples in this fragment.
     * Type `bigint` to stress that this is never fractional.
     */ public sampleCount ( ) { if ( this . stats !== undefined ) { return this . stats . sampleCount ; } return BigInt ( this . samples . length ) ; } /** Return stringified and zero-padded index. */ public indexString ( length ) { const is = this . index . toString ( ) ; return is . padStart ( length , "0" ) ; } public addSample ( s ) { if ( this . serialized ) { throw new Error ( "cannot mutate fragment anymore" ) ; } this . samples . push ( s ) ; this . addSampleHook ( s ) ; } /**
     * Use this to indicate that this fragment was serialized (into a binary
     * msg) out-of-band, i..e not with the `serialize()` method.
     */ public setSerialized ( ) { this . serialized = true ; } /**
     * Can be used in the child to execute custom functionality when adding a
     * sample.
     */ protected abstract addSampleHook ( s ) ; } export abstract class TimeseriesBase < FragmentType > { /** The label set (set of key/value pairs) which uniquely defines this time
     * series*/ protected labels : LabelSet ; /** The start time of this synthetic time series (note(jp): does first sample
     * have this value? might depend on log vs. metrics implementation) */ protected starttime : ZonedDateTime ; /** The set of options configuring this time series -- serialized into a
     * string */ protected optionstring : string ; /** The number of fragments consumed from this time series -- for
     * book-keeping purposes. */ protected nFragmentsConsumed : number ; /** For keeing track of how many entries were validated (from the start of the
     *stream). Used by fetchAndValidate(). */ protected nSamplesValidatedSoFar : bigint ; /**
     * Configure 'walltime coupling' where the synthetic time source is loosely
     * coupled to the wall time.
     *
     * `undefined`: disable this mechanism. can fall back into the past
     * arbitrarily far, and go into the future arbitrarily far. Not expected to
     * work for metrics (cortex, blocks storage) -- may work for logs (loki, also
     * see reject_old_samples option and discussion in
     * https://github.com/opstrace/opstrace/pull/1140#discussion_r679837228)
     */ protected walltimeCouplingOptions : WalltimeCouplingOptions | undefined ; protected counterForwardLeap : any | undefined ; /**
     * Fragment time leap, defined as the time difference between the last sample
     *  of a fragment and the last sample of the previous fragment.
     */ fragmentTimeLeapSeconds : number ; postedFragmentsSinceLastValidate : Array < LogSeriesFragment | MetricSeriesFragment > | undefined ; sample_time_increment_ns : number ; n_samples_per_series_fragment : number ; uniqueName : string ; lastFragmentConsumed : LogSeriesFragment | MetricSeriesFragment | undefined ; constructor ( opts ) { this . nFragmentsConsumed = 0 ; this . starttime = opts . starttime ; this . uniqueName = opts . uniqueName ; this . optionstring = `${ JSON . stringify ( opts ) }` ; this . labels = this . buildLabelSetFromOpts ( opts ) ; this . n_samples_per_series_fragment = opts . n_samples_per_series_fragment ; this . sample_time_increment_ns = opts . sample_time_increment_ns ; this . nSamplesValidatedSoFar = BigInt ( 0 ) ; this . walltimeCouplingOptions = opts . wtopts ; this . postedFragmentsSinceLastValidate = undefined ; if ( ! Number . isInteger ( opts . sample_time_increment_ns ) ) { throw new Error ( "sample_time_increment_ns must be an integer value" ) ; } this . fragmentTimeLeapSeconds = this . n_samples_per_series_fragment * ( this . sample_time_increment_ns / 10 ** 6 ) ; this . validateWtOpts ( opts . wtopts ) ; if ( opts . counterForwardLeap !== undefined ) { this . counterForwardLeap = opts . counterForwardLeap ; if ( opts . counterForwardLeap . inc === undefined ) { throw new Error ( "the counterForwardLeap obj needs to have an `inc()` method" ) ; } } } protected abstract buildLabelSetFromOpts ( opts ) ; abstract currentTimeRFC3339Nano ( ) ; abstract promQueryString ( ) ; protected abstract nextSample ( ) ; protected abstract generateNextFragment ( ) ; /**
     * Return floating point number representing the timestamp of the last sample
     * in "seconds since epoch". It is OK to potentially have less absolute
     * precision as compared to the internal time-keeping variables (can't fit
     * nanosecond resolution for a wide date range into double precision float,
     * which is why for log series the prometheus ecosystem tracks seconds and
     * the fractional part in nanoseconds in two separate variables).
     */ protected abstract lastSampleSecondsSinceEpoch ( ) ; /**
     * Leap forward by N seconds.
     *
     * Must only be called after just having completed a fragment (right?).
     */ protected abstract leapForward ( n ) ; protected abstract fetchAndValidateFragment ( fragment , opts ) ; public toString ( ) { return `${ this . constructor . name }(${ this . promQueryString ( ) })` ; } protected validateWtOpts ( o ) { if ( o === undefined ) { return ; } assert ( Number . isInteger ( o [ "minLagSeconds" ] ) ) ; assert ( Number . isInteger ( o [ "maxLagSeconds" ] ) ) ; if ( o . maxLagSeconds <= this . fragmentTimeLeapSeconds ) { throw new Error ( "maxLagSeconds <= fragmentTimeLeapSeconds" ) ; } } /**
     * When the time of the last sample in the last generated fragment has fallen
     * behind too far compared to the current wall time, this method can be used
     * to make the first sample of the next fragment to be generated be closer to
     * the current wall time again.
     *
     * If this method ever needs to be called externally, it should be called
     * between two calls to `generateAndGetNextFragment()`. For now it is a
     * private method and is called at the beginning of the implementation of
     * `generateNextFragment()`.
     *
     * ## Background
     *
     * With the Cortex Blocks Storage engine, we cannot push samples that go into
     * the future (compared to wall time in the ingest system), and we also
     * cannot push samples that are older than 1 hour compared to the wall time
     * in the ingest system. In both cases, Cortex rejects these samples.
     *
     * Context: https://github.com/opstrace/opstrace-corp/issues/147 and
     * https://github.com/cortexproject/cortex/issues/2366.
     *
     * The challenge: for deep read validation we need to _know_ the timestamps
     * for individual metric samples that were written into the remote storage
     * system. Keeping them all in memory is not an option towards validation,
     * i.e. they need to follow a predictable pattern. That is, they can't be
     * consumed from a clock source of this machine, but they need to be
     * generated synthetically, following said pattern. The simple pattern being
     * used here: adjacent timestamps in the synthetically generated samples have
     * a fixed time distance between them (`this.metrics_time_increment_ms`).
     *
     * Now, during execution, this synthetic generation of timestamps, is either
     * faster or slower than the evolution of the actual time. This depends on
     * many factors, but certainly on the user-given choice of
     * `this.metrics_time_increment_ms`.
     *
     * This method here provides a correction for one of both cases: when the
     * synthetic clock source evolves _slower_ than wall time, i.e. when it
     * slowly falls behind. When it has fallen behind by a specific amount, it
     * leaps the synthetic time source forward by a particular correction amount.
     *
     * Also note: this strategy makes use of the fact that fragments are
     * validated _individually_ based on their first and last sample's
     * timestamps, i.e. the leap forward is done _between_ fragments, and not
     * _within_ an individual fragment.
     */ protected bringCloserToWalltimeIfFallenBehind ( ) { assert ( this . walltimeCouplingOptions ) ; const lastSampleSecondsSinceEpoch = this . lastSampleSecondsSinceEpoch ( ) ; const nowSecondsSinceEpoch = ZonedDateTime . now ( ZoneOffset . UTC ) . toEpochSecond ( ) ; const shiftIntoPastSeconds = nowSecondsSinceEpoch - lastSampleSecondsSinceEpoch ; if ( shiftIntoPastSeconds < 0 ) { throw new Error ( `${ this }: shiftIntoPastSeconds < 0: ${ shiftIntoPastSeconds . toFixed ( 4 ) } -- lastSampleSecondsSinceEpoch: ${ lastSampleSecondsSinceEpoch . toFixed ( 4 ) } -- nowSecondsSinceEpoch: ${ nowSecondsSinceEpoch . toFixed ( 4 ) }` ) ; } if ( shiftIntoPastSeconds > this . walltimeCouplingOptions . maxLagSeconds ) { const secondsToLeapForward = BigInt ( Math . floor ( shiftIntoPastSeconds - this . walltimeCouplingOptions . minLagSeconds - 5 ) ) ; const lfm = Number ( secondsToLeapForward ) / 60 ; log . debug ( `${ this }: leap forward by ${ lfm . toFixed ( 2 ) } minutes` ) ; this . leapForward ( secondsToLeapForward ) ; if ( this . counterForwardLeap !== undefined ) { this . counterForwardLeap . inc ( ) ; } return shiftIntoPastSeconds - lfm * 60 ; } if ( this . nFragmentsConsumed > 0 && this . nFragmentsConsumed % 10000 === 0 ) { const m = shiftIntoPastSeconds / 60.0 ; log . debug ( "%s: lag behind walltime: %s minutes" , this , m . toFixed ( 1 ) ) ; } return shiftIntoPastSeconds ; } public generateNextFragmentOrSkip ( ) { if ( this . walltimeCouplingOptions === undefined ) { return [ - 1 , this . generateNextFragment ( ) ] ; } const shiftIntoPastSeconds = this . bringCloserToWalltimeIfFallenBehind ( ) ; if ( shiftIntoPastSeconds - this . fragmentTimeLeapSeconds > this . walltimeCouplingOptions . minLagSeconds ) { return [ shiftIntoPastSeconds + this . fragmentTimeLeapSeconds , this . generateNextFragment ( ) ] ; } return [ shiftIntoPastSeconds , undefined ] ; } public async fetchAndValidate ( opts ) { log . debug ( "%s fetchAndValidate()" , this ) ; let samplesValidated = 0 ; let fragmentsValidated = 0 ; if ( this . postedFragmentsSinceLastValidate === undefined ) { return 0 ; } for ( const fragment of this . postedFragmentsSinceLastValidate ) { const validated = await this . fetchAndValidateFragment ( fragment , opts ) ; samplesValidated += validated ; fragmentsValidated += 1 ; if ( fragmentsValidated % 20 === 0 ) { log . debug ( "%s fetchAndValidate(): %s fragments validated (%s samples)" , this . uniqueName , fragmentsValidated , samplesValidated ) ; } } this . postedFragmentsSinceLastValidate = [ ] ; return samplesValidated ; } public disableValidation ( ) { this . postedFragmentsSinceLastValidate = undefined ; } public enableValidation ( ) { this . postedFragmentsSinceLastValidate = [ ] ; } public shouldBeValidated ( ) { if ( this . postedFragmentsSinceLastValidate === undefined ) { return false ; } return true ; } public rememberFragmentForValidation ( f ) { if ( this . postedFragmentsSinceLastValidate !== undefined ) { this . postedFragmentsSinceLastValidate . push ( f ) ; } } public dropValidationInfo ( ) { this . postedFragmentsSinceLastValidate = [ ] ; } }