import Boom from '@hapi/boom' ; import { IScopedClusterClient } from 'kibana/server' ; import { duration } from 'moment' ; import { parseInterval } from '../../../common/util/parse_interval' ; import { initCardinalityFieldsCache } from './fields_aggs_cache' ; import { AggCardinality } from '../../../common/types/fields' ; import { isValidAggregationField } from '../../../common/util/validation_utils' ; import { getDatafeedAggregations } from '../../../common/util/datafeed_utils' ; import { Datafeed , IndicesOptions } from '../../../common/types/anomaly_detection_jobs' ; import { RuntimeMappings } from '../../../common/types/fields' ; import { isPopulatedObject } from '../../../common/util/object_utils' ; /**
 * Service for carrying out queries to obtain data
 * specific to fields in Elasticsearch indices.
 */ export function fieldsServiceProvider ( { asCurrentUser } ) { const fieldsAggsCache = initCardinalityFieldsCache ( ) ; /**
     * Caps the time range to the last 90 days if necessary
     */ function getSafeTimeRange ( earliestMs , latestMs ) { const capOffsetMs = duration ( 3 , 'months' ) . asMilliseconds ( ) ; const capRangeStart = latestMs - capOffsetMs ; return { start : Math . max ( earliestMs , capRangeStart ) , end : latestMs , } ; } /**
     * Gets aggregatable fields.
     */ async function getAggregatableFields ( index , fieldNames , datafeedConfig ? ) { const { body } = await asCurrentUser . fieldCaps ( { index , fields : fieldNames , } ) ; const aggregatableFields = [ ] ; const datafeedAggregations = getDatafeedAggregations ( datafeedConfig ) ; fieldNames . forEach ( ( fieldName ) => { if ( typeof datafeedConfig ?. script_fields === 'object' && datafeedConfig . script_fields . hasOwnProperty ( fieldName ) ) { aggregatableFields . push ( fieldName ) ; } if ( typeof datafeedConfig ?. runtime_mappings === 'object' && datafeedConfig . runtime_mappings . hasOwnProperty ( fieldName ) ) { aggregatableFields . push ( fieldName ) ; } if ( datafeedAggregations !== undefined && isValidAggregationField ( datafeedAggregations , fieldName ) ) { aggregatableFields . push ( fieldName ) ; } const fieldInfo = body . fields [ fieldName ] ; const typeKeys = fieldInfo !== undefined ? Object . keys ( fieldInfo ) : [ ] ; if ( typeKeys . length > 0 ) { const fieldType = typeKeys [ 0 ] ; const isFieldAggregatable = fieldInfo [ fieldType ] . aggregatable ; if ( isFieldAggregatable === true ) { aggregatableFields . push ( fieldName ) ; } } } ) ; return aggregatableFields ; } async function getCardinalityOfFields ( index , fieldNames , query , timeFieldName , earliestMs , latestMs , datafeedConfig ? ) { const aggregatableFields = await getAggregatableFields ( index , fieldNames , datafeedConfig ) ; if ( aggregatableFields . length === 0 ) { return { } ; } const { start , end } = getSafeTimeRange ( earliestMs , latestMs ) ; const cachedValues = fieldsAggsCache . getValues ( index , timeFieldName , start , end , 'overallCardinality' , fieldNames ) ?? { } ; const fieldsToAgg = aggregatableFields . filter ( ( field ) => ! cachedValues . hasOwnProperty ( field ) ) ; if ( fieldsToAgg . length === 0 ) { return cachedValues ; } const mustCriteria = [ { range : { [ timeFieldName ] : { gte : start , lte : end , format : 'epoch_millis' , } , } , } , ] ; if ( query ) { mustCriteria . push ( query ) ; } const runtimeMappings = { } ; const aggs = fieldsToAgg . reduce ( ( obj , field ) => { if ( typeof datafeedConfig ?. script_fields === 'object' && datafeedConfig . script_fields . hasOwnProperty ( field ) ) { obj [ field ] = { cardinality : { script : datafeedConfig . script_fields [ field ] . script } } ; } else if ( typeof datafeedConfig ?. runtime_mappings === 'object' && datafeedConfig . runtime_mappings . hasOwnProperty ( field ) ) { obj [ field ] = { cardinality : { field } } ; runtimeMappings . runtime_mappings = datafeedConfig . runtime_mappings ; } else { obj [ field ] = { cardinality : { field } } ; } return obj ; } , { } as { [ field ] : AggCardinality ; } ) ; const body = { query : { bool : { must : mustCriteria , } , } , size : 0 , _source : { excludes : [ ] , } , aggs , ... runtimeMappings , } ; const { body : { aggregations } , } = await asCurrentUser . search ( { index , body , ... ( datafeedConfig ?. indices_options ?? { } ) , } ) ; if ( ! aggregations ) { return { } ; } const aggResult = fieldsToAgg . reduce ( ( obj , field ) => { obj [ field ] = ( aggregations [ field ] || { value : 0 } ) . value ; return obj ; } , { } as { [ field ] : number ; } ) ; fieldsAggsCache . updateValues ( index , timeFieldName , start , end , { overallCardinality : aggResult , } ) ; return { ... cachedValues , ... aggResult , } ; } /**
     * Gets time boundaries of the index data based on the provided time field.
     */ async function getTimeFieldRange ( index , timeFieldName , query , runtimeMappings ? , indicesOptions ? ) { const obj = { success : true , start : { epoch : 0 , string : '' } , end : { epoch : 0 , string : '' } } ; const { body : { aggregations } , } = await asCurrentUser . search ( { index , size : 0 , body : { ... ( query ? { query } : { } ) , aggs : { earliest : { min : { field : timeFieldName , } , } , latest : { max : { field : timeFieldName , } , } , } , ... ( isPopulatedObject ( runtimeMappings ) ? { runtime_mappings : runtimeMappings } : { } ) , } , ... ( indicesOptions ?? { } ) , } ) ; if ( aggregations && aggregations . earliest && aggregations . latest ) { obj . start . epoch = aggregations . earliest . value ; obj . start . string = aggregations . earliest . value_as_string ; obj . end . epoch = aggregations . latest . value ; obj . end . string = aggregations . latest . value_as_string ; } return obj ; } /**
     * Caps provided time boundaries based on the interval
     */ function getSafeTimeRangeForInterval ( interval , ... timeRange ) ; function getSafeTimeRangeForInterval ( interval , earliestMs , latestMs ) { const maxNumberOfBuckets = 1000 ; const end = latestMs ; const intervalDuration = parseInterval ( interval ) ; if ( intervalDuration === null ) { throw Boom . badRequest ( 'Interval is invalid' ) ; } const start = Math . max ( earliestMs , latestMs - maxNumberOfBuckets * intervalDuration . asMilliseconds ( ) ) ; return { start , end } ; } param index param fieldNames param query param timeFieldName param earliestMs param latestMs param interval async function getMaxBucketCardinalities ( index , fieldNames , query , timeFieldName , earliestMs , latestMs , interval , datafeedConfig ? ) { if ( ! interval ) { throw Boom . badRequest ( 'Interval is required to retrieve max bucket cardinalities.' ) ; } const aggregatableFields = await getAggregatableFields ( index , fieldNames , datafeedConfig ) ; if ( aggregatableFields . length === 0 ) { return { } ; } const { start , end } = getSafeTimeRangeForInterval ( interval , ... Object . values ( getSafeTimeRange ( earliestMs , latestMs ) ) ) ; const cachedValues = fieldsAggsCache . getValues ( index , timeFieldName , start , end , 'maxBucketCardinality' , fieldNames ) ?? { } ; const fieldsToAgg = aggregatableFields . filter ( ( field ) => ! cachedValues . hasOwnProperty ( field ) ) ; if ( fieldsToAgg . length === 0 ) { return cachedValues ; } const mustCriteria = [ { range : { [ timeFieldName ] : { gte : start , lte : end , format : 'epoch_millis' , } , } , } , ] ; if ( query ) { mustCriteria . push ( query ) ; } const dateHistogramAggKey = 'bucket_span_buckets' ; /**
         * Replace any non-word characters
         */ const getSafeAggName = ( field ) => field . replace ( /\W/g , '' ) ; const getMaxBucketAggKey = ( field ) => `max_bucket_${ field }` ; const fieldsCardinalityAggs = fieldsToAgg . reduce ( ( obj , field ) => { obj [ getSafeAggName ( field ) ] = { cardinality : { field } } ; return obj ; } , { } as { [ field ] : { cardinality : { field : string ; } ; } ; } ) ; const maxBucketCardinalitiesAggs = Object . keys ( fieldsCardinalityAggs ) . reduce ( ( acc , field ) => { acc [ getMaxBucketAggKey ( field ) ] = { max_bucket : { buckets_path : `${ dateHistogramAggKey }>${ field }` , } , } ; return acc ; } , { } as { [ key ] : { max_bucket : { buckets_path : string ; } ; } ; } ) ; const body = { query : { bool : { filter : mustCriteria , } , } , size : 0 , aggs : { [ dateHistogramAggKey ] : { date_histogram : { field : timeFieldName , fixed_interval : interval , } , aggs : fieldsCardinalityAggs , } , ... maxBucketCardinalitiesAggs , } , } ; const { body : { aggregations } , } = await asCurrentUser . search ( { index , body , ... ( datafeedConfig ?. indices_options ?? { } ) , } ) ; if ( ! aggregations ) { return cachedValues ; } const aggResult = fieldsToAgg . reduce ( ( obj , field ) => { obj [ field ] = ( aggregations [ getMaxBucketAggKey ( field ) ] || { value : 0 } ) . value ?? 0 ; return obj ; } , { } as { [ field ] : number ; } ) ; fieldsAggsCache . updateValues ( index , timeFieldName , start , end , { maxBucketCardinality : aggResult , } ) ; return { ... cachedValues , ... aggResult , } ; } return { getCardinalityOfFields , getTimeFieldRange , getMaxBucketCardinalities , } ; }