import { argv } from 'yargs' ; import pLimit from 'p-limit' ; import pRetry from 'p-retry' ; import { set } from '@elastic/safer-lodash-set' ; import { uniq , without , merge , flatten } from 'lodash' ; import * as histogram from 'hdr-histogram-js' ; import { HOST_NAME , SERVICE_NAME , TRANSACTION_NAME , TRANSACTION_TYPE , AGENT_NAME , SERVICE_ENVIRONMENT , POD_NAME , CONTAINER_ID , SERVICE_VERSION , TRANSACTION_RESULT , PROCESSOR_EVENT , } from '../../common/elasticsearch_fieldnames' ; import { createOrUpdateIndex } from '../shared/create_or_update_index' ; import { parseIndexUrl } from '../shared/parse_index_url' ; import { ESClient , getEsClient } from '../shared/get_es_client' ; export async function aggregateLatencyMetrics ( ) { const interval = parseInt ( String ( argv . interval ) , 10 ) || 1 ; const concurrency = parseInt ( String ( argv . concurrency ) , 10 ) || 3 ; const numSigFigures = ( parseInt ( String ( argv . sigfig ) , 10 ) || 2 ) as 1 | 2 | 3 | 4 | 5 ; const from = new Date ( String ( argv . from ) ) . getTime ( ) ; const to = new Date ( String ( argv . to ) ) . getTime ( ) ; if ( isNaN ( from ) || isNaN ( to ) ) { throw new Error ( `from and to are not valid dates - please supply valid ISO timestamps` ) ; } if ( to <= from ) { throw new Error ( 'to cannot be earlier than from' ) ; } const limit = pLimit ( concurrency ) ; const retry = ( fn ) => { return ( ) => pRetry ( fn , { factor : 1 , retries : 3 , minTimeout : 2500 , } ) ; } ; const tasks = [ ] ; const defaultFields = [ SERVICE_NAME , SERVICE_VERSION , SERVICE_ENVIRONMENT , AGENT_NAME , HOST_NAME , POD_NAME , CONTAINER_ID , TRANSACTION_NAME , TRANSACTION_RESULT , TRANSACTION_TYPE , ] ; const include = String ( argv . include ?? '' ) . split ( ',' ) . filter ( Boolean ) as string [ ] ; const exclude = String ( argv . exclude ?? '' ) . split ( ',' ) . filter ( Boolean ) as string [ ] ; const only = String ( argv . only ?? '' ) . split ( ',' ) . filter ( Boolean ) as string [ ] ; const fields = only . length ? uniq ( only ) : without ( uniq ( [ ... include , ... defaultFields ] ) , ... exclude ) ; const globalFilter = argv . filter ? JSON . parse ( String ( argv . filter ) ) : { } ; console . log ( 'Aggregating on' , fields . join ( ',' ) ) ; const source = String ( argv . source ?? '' ) ; const dest = String ( argv . dest ?? '' ) ; const sourceOptions = parseIndexUrl ( source ) ; const sourceClient = getEsClient ( { node : sourceOptions . node } ) ; let destClient ; let destOptions ; const uploadMetrics = ! ! dest ; if ( uploadMetrics ) { destOptions = parseIndexUrl ( dest ) ; destClient = getEsClient ( { node : destOptions . node } ) ; const mappings = await sourceClient . indices . getMapping ( { index : sourceOptions . index , } ) ; const lastMapping = mappings [ Object . keys ( mappings ) [ 0 ] ] ; const newMapping = merge ( { } , lastMapping , { mappings : { properties : { transaction : { properties : { duration : { properties : { histogram : { type : 'histogram' , } , } , } , } , } , } , } , } ) ; await createOrUpdateIndex ( { client : destClient , indexName : destOptions . index , clear : false , template : newMapping , } ) ; } else { console . log ( 'No destination was defined, not uploading aggregated documents' ) ; } let at = to ; while ( at > from ) { const end = at ; const start = Math . max ( from , at - interval * 60 * 1000 ) ; tasks . push ( limit ( retry ( async ( ) => { const filter = [ { term : { [ PROCESSOR_EVENT ] : 'transaction' , } , } , { range : { '@timestamp' : { gte : start , lt : end , } , } , } , ] ; const query = { ... globalFilter , query : { ... ( globalFilter ?. query ?? { } ) , bool : { ... ( globalFilter ?. query ?. bool ?? { } ) , filter : [ ... Object . values ( globalFilter ?. query ?. bool ?. filter ?? { } ) , ... filter , ] , } , } , } ; async function paginateThroughBuckets ( buckets , after ? ) { const params = { index : sourceOptions . index , body : { ... query , aggs : { transactionGroups : { composite : { ... ( after ? { after } : { } ) , size : 10000 , sources : fields . map ( ( field ) => ( { [ field ] : { terms : { field , missing_bucket : true , } , } , } ) ) , } , ... ( dest ? { aggs : { recorded_values : { scripted_metric : { init_script : 'state.values = new ArrayList()' , map_script : `
                            if (!doc['transaction.duration.us'].empty) {
                              state.values.add(doc['transaction.duration.us'].value);
                            }
                          ` , combine_script : 'return state.values' , reduce_script : `
                            return states.stream().flatMap(l -> l.stream()).collect(Collectors.toList())
                          ` , } , } , } , } : { } ) , } , } , } , } ; const response = await sourceClient . search ( params ) ; const { aggregations } = response . body ; if ( ! aggregations ) { return buckets ; } const { transactionGroups } = aggregations ; const nextBuckets = buckets . concat ( transactionGroups . buckets ) ; if ( ! transactionGroups . after_key ) { return nextBuckets ; } return nextBuckets . concat ( await paginateThroughBuckets ( buckets , transactionGroups . after_key ) ) ; } async function getNumberOfTransactionDocuments ( ) { const params = { index : sourceOptions . index , body : { query : { bool : { filter , } , } , track_total_hits : true , } , } ; const response = await sourceClient . search ( params ) ; return response . body . hits . total . value ; } const [ buckets , numberOfTransactionDocuments ] = await Promise . all ( [ paginateThroughBuckets ( [ ] ) , getNumberOfTransactionDocuments ( ) , ] ) ; const rangeLabel = `${ new Date ( start ) . toISOString ( ) }-${ new Date ( end ) . toISOString ( ) }` ; console . log ( `${ rangeLabel }: Compression: ${ buckets . length }/${ numberOfTransactionDocuments } (${ ( ( buckets . length / numberOfTransactionDocuments ) * 100 ) . toPrecision ( 2 ) }%)` ) ; const docs = [ ] ; if ( uploadMetrics ) { buckets . forEach ( ( bucket ) => { const values = ( bucket . recorded_values ?. value ?? [ ] ) as number [ ] ; const h = histogram . build ( { numberOfSignificantValueDigits : numSigFigures , } ) ; values . forEach ( ( value ) => { h . recordValue ( value ) ; } ) ; const iterator = h . recordedValuesIterator ; const distribution = { values : [ ] as number [ ] , counts : [ ] as number [ ] , } ; iterator . reset ( ) ; while ( iterator . hasNext ( ) ) { const value = iterator . next ( ) ; distribution . values . push ( value . valueIteratedTo ) ; distribution . counts . push ( value . countAtValueIteratedTo ) ; } const structured = Object . keys ( bucket . key ) . reduce ( ( prev , key ) => { set ( prev , key , bucket . key [ key ] ) ; return prev ; } , { } ) ; const doc = merge ( { } , structured , { '@timestamp' : new Date ( start ) . toISOString ( ) , timestamp : { us : start * 1000 , } , processor : { name : 'metric' , event : 'metric' , } , transaction : { duration : { histogram : distribution , } , } , } ) ; docs . push ( doc ) ; } ) ; if ( ! docs . length ) { console . log ( `${ rangeLabel }: No docs to upload` ) ; return ; } const response = await ( destClient as any ) ?. bulk ( { refresh : 'wait_for' , body : flatten ( docs . map ( ( doc ) => [ { index : { _index : destOptions ?. index } } , doc , ] ) ) , } ) ; if ( response ?. body . errors ) { throw new Error ( `${ rangeLabel }: Could not upload all metric documents` ) ; } console . log ( `${ rangeLabel }: Uploaded ${ docs . length } metric documents` ) ; } } ) ) ) ; at = start ; } await Promise . all ( tasks ) ; }