/**
 * 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 logfmt from "logfmt" ; import protobuf from "protobufjs" ; import snappy from "snappy" ; import got from "got" ; import { mtimeDiffSeconds , mtime , sleep } from "../mtime" ; import { log } from "../log" ; import { logHTTPResponseLight , logHTTPResponse } from "../util" ; import { LogSeries } from "./dummystream" ; import { SampleBase , FragmentBase , LabelSet , FragmentStatsBase } from "../series" ; export * from "./dummystream" ; const logPbfRoot = protobuf . loadSync ( `${ __dirname }/../resources/logproto_bundle.json` ) ; const pbTypePushrequest = logPbfRoot . lookupType ( "logproto.PushRequest" ) ; const pbTypeStream = logPbfRoot . lookupType ( "logproto.Stream" ) ; const pbTypeEntry = logPbfRoot . lookupType ( "logproto.Entry" ) ; const pbTypeTimestamp = logPbfRoot . lookupType ( "google.protobuf.Timestamp" ) ; /** The log sample timestamp which is comprised of two components: seconds
 * since epoch, and a fractional part: nanoseconds
 */ export interface LogSampleTimestamp { seconds : number ; nanos : number ; } export class LogSample extends SampleBase < string , LogSampleTimestamp > { } export interface LogSeriesFragmentStats extends FragmentStatsBase { timeOfFirstEntry : bigint ; timeOfLastEntry : bigint ; textmd5 : string ; } export class LogSeriesFragment extends FragmentBase < LogSample , LogSeries > { private payloadbytecounter : number ; constructor ( labels , index = 0 , dummystream = undefined ) { super ( labels , index , dummystream ) ; this . payloadbytecounter = 0 ; } public payloadByteCount ( ) { return BigInt ( this . payloadbytecounter ) ; } protected addSampleHook ( s ) { this . payloadbytecounter += 12 + Buffer . from ( s . value , "utf8" ) . length ; } public buildStatisticsAndDropData ( ) { if ( ! this . serialized ) { throw new Error ( "not yet serialized" ) ; } const tsfirst = this . samples [ 0 ] . time ; const tslast = this . samples . slice ( - 1 ) [ 0 ] . time ; const logTextHash = crypto . createHash ( "md5" ) ; for ( const s of this . samples ) { logTextHash . update ( s . value ) ; } const stats = { sampleCount : BigInt ( this . samples . length ) , timeOfFirstEntry : BigInt ( logSampleTimeStampToString ( tsfirst ) ) , timeOfLastEntry : BigInt ( logSampleTimeStampToString ( tslast ) ) , textmd5 : logTextHash . digest ( "hex" ) } ; this . stats = stats ; this . samples . length = 0 ; this . stats = stats ; } public serialize ( ) { this . serialized = true ; return new LogSeriesFragmentPushRequest ( [ this ] ) ; } } export class LogSeriesFragmentPushRequest { fragments : LogSeriesFragment [ ] ; textmd5 : string ; data : Buffer ; dataLengthBytes : number ; dataLengthMiB : number ; payloadByteCount : bigint ; serializationTimeSeconds : number ; postHeaders : Record < string , string > ; constructor ( LogSeriesFragment ) { this . fragments = LogSeriesFragment ; this . payloadByteCount = BigInt ( 0 ) ; const [ data , textmd5 , serializationTimeSeconds ] = this . toPushrequestBuffer ( ) ; this . textmd5 = textmd5 ; 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" } ; } public toString = ( ) : string => { return `PushRequest(md5=${ this . textmd5 })` ; } ; param lokiBaseUrl public async postWithRetryOrError ( lokiBaseUrl , maxRetries = 3 , additionalHeaders ? ) { for ( let attempt = 1 ; attempt <= maxRetries ; attempt ++ ) { let response ; try { response = await this . postToLoki ( `${ lokiBaseUrl }/loki/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 === 204 ) { 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, retry" ) ; } throw new Error ( `Failed to POST ${ this } after ${ maxRetries } attempts` ) ; } param url public async postToLoki ( url , additionalHeaders ? ) { let headers = { "Content-Type" : "application/x-protobuf" } ; 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 toPushrequestBuffer ( ) { const t0 = mtime ( ) ; const logTextHash = crypto . createHash ( "md5" ) ; const streamsList = [ ] ; for ( const fragment of this . fragments ) { this . payloadByteCount += fragment . payloadByteCount ( ) ; const labelsEncodedAsString = logqlLabelString ( fragment . labels ) ; const pbsamples = [ ] ; for ( const entry of fragment . getSamples ( ) ) { pbsamples . push ( pbTypeEntry . create ( { timestamp : pbTypeTimestamp . create ( entry . time ) , line : entry . value } ) ) ; logTextHash . update ( entry . value ) ; } const stream = pbTypeStream . create ( { labels : labelsEncodedAsString , entries : pbsamples } ) ; streamsList . push ( stream ) ; fragment . setSerialized ( ) ; } const pr = pbTypePushrequest . create ( { streams : streamsList } ) ; const logtextmd5 = logTextHash . digest ( "hex" ) ; const prbuffer = pbTypePushrequest . encode ( pr ) . finish ( ) ; const prBufferSnapped = snappy . compressSync ( prbuffer ) ; const serializationTimeSeconds = mtimeDiffSeconds ( t0 ) ; return [ prBufferSnapped , logtextmd5 , serializationTimeSeconds ] ; } } export function logSampleTimeStampToString ( t ) { return t . seconds . toString ( ) + t . nanos . toString ( ) . toString ( ) . padStart ( 9 , "0" ) ; } export function logqlKvPairTextToObj ( lql ) { const intermediate = logfmt . parse ( lql . slice ( 1 , - 1 ) . replace ( /,/g , "" ) ) ; const labels = Object . fromEntries ( Object . entries ( intermediate ) . map ( ( [ k , v ] ) => [ k , v . toString ( ) ] ) ) ; return labels ; } export function logqlLabelString ( labels ) { const core = Object . entries ( labels ) . map ( ( [ k , v ] ) => `${ k }="${ v }"` ) . join ( ", " ) ; return `{${ core }}` ; }