import type * as estypes from '@elastic/elasticsearch/lib/api/typesWithBodyKey' ; import { i18n } from '@kbn/i18n' ; import type { IScopedClusterClient , IUiSettingsClient } from 'src/core/server' ; import type { IScopedSearchClient } from 'src/plugins/data/server' ; import type { Datatable } from 'src/plugins/expressions/server' ; import type { Writable } from 'stream' ; import type { ReportingConfig } from '../../..' ; import type { IndexPattern , ISearchSource , ISearchStartSearchSource , SearchFieldValue , SearchSourceFields , } from '../../../../../../../src/plugins/data/common' ; import { cellHasFormulas , ES_SEARCH_STRATEGY , tabifyDocs , } from '../../../../../../../src/plugins/data/common' ; import type { FieldFormat , FieldFormatConfig , IFieldFormatsRegistry , } from '../../../../../../../src/plugins/field_formats/common' ; import { KbnServerError } from '../../../../../../../src/plugins/kibana_utils/server' ; import type { CancellationToken } from '../../../../common/cancellation_token' ; import { CONTENT_TYPE_CSV } from '../../../../common/constants' ; import { byteSizeValueToNumber } from '../../../../common/schema_utils' ; import type { LevelLogger } from '../../../lib' ; import type { TaskRunResult } from '../../../lib/tasks' ; import type { JobParamsCSV } from '../types' ; import { CsvExportSettings , getExportSettings } from './get_export_settings' ; import { MaxSizeStringBuilder } from './max_size_string_builder' ; interface Clients { es : IScopedClusterClient ; data : IScopedSearchClient ; uiSettings : IUiSettingsClient ; } interface Dependencies { searchSourceStart : ISearchStartSearchSource ; fieldFormatsRegistry : IFieldFormatsRegistry ; } function isPlainStringArray ( fields ) { let result = true ; if ( Array . isArray ( fields ) ) { fields . forEach ( ( field ) => { if ( typeof field !== 'string' || field === '*' || field === '_source' ) { result = false ; } } ) ; } return result ; } export class CsvGenerator { private _columns ? : string [ ] ; private csvContainsFormulas = false ; private maxSizeReached = false ; private csvRowCount = 0 ; constructor ( private job , private config , private clients , private dependencies , private cancellationToken , private logger , private stream ) { } private async scan ( index , searchSource , settings ) { const { scroll : scrollSettings , includeFrozen } = settings ; const searchBody = searchSource . getSearchRequestBody ( ) ; this . logger . debug ( `executing search request` ) ; const searchParams = { params : { body : searchBody , index : index . title , scroll : scrollSettings . duration , size : scrollSettings . size , ignore_throttled : includeFrozen ? false : undefined , } , } ; const results = ( await this . clients . data . search ( searchParams , { strategy : ES_SEARCH_STRATEGY } ) . toPromise ( ) ) . rawResponse as estypes . SearchResponse < unknown > ; return results ; } private async scroll ( scrollId , scrollSettings ) { this . logger . debug ( `executing scroll request` ) ; const results = ( await this . clients . es . asCurrentUser . scroll ( { scroll : scrollSettings . duration , scroll_id : scrollId , } ) ) . body ; return results ; } private getFormatters ( table ) { const formatters = { } ; table . columns . forEach ( ( c ) => { const fieldFormat = this . dependencies . fieldFormatsRegistry . deserialize ( c . meta . params ) ; formatters [ c . id ] = fieldFormat ; } ) ; return formatters ; } private escapeValues ( settings ) { return ( value ) => { if ( settings . checkForFormulas && cellHasFormulas ( value ) ) { this . csvContainsFormulas = true ; } return settings . escapeValue ( value ) ; } ; } private getColumns ( searchSource , table ) { if ( this . _columns != null ) { return this . _columns ; } const getFromSearchSource = ( ) : string [ ] => { const fieldValues = { fields : searchSource . getField ( 'fields' ) , fieldsFromSource : searchSource . getField ( 'fieldsFromSource' ) , } ; const fieldSource = fieldValues . fieldsFromSource ? 'fieldsFromSource' : 'fields' ; this . logger . debug ( `Getting columns from '${ fieldSource }' in search source.` ) ; const fields = fieldValues [ fieldSource ] ; if ( isPlainStringArray ( fields ) ) { return fields ; } const columnIds = table . columns . map ( ( c ) => c . id ) ; columnIds . sort ( ) ; return columnIds ; } ; this . _columns = this . job . columns ?. length ? this . job . columns : getFromSearchSource ( ) ; return this . _columns ; } private formatCellValues ( formatters ) { return ( { column : tableColumn , data : dataTableCell , } ) : string => { let cell ; if ( tableColumn && dataTableCell ) { try { cell = formatters [ tableColumn ] . convert ( dataTableCell ) ; } catch ( err ) { this . logger . error ( err ) ; cell = '-' ; } try { cell = JSON . parse ( cell ) ; } catch ( e ) { } if ( Array . isArray ( cell ) ) { cell = cell . map ( ( c ) => ( typeof c === 'object' ? JSON . stringify ( c ) : c ) ) . join ( ', ' ) ; } if ( typeof cell === 'object' ) { cell = JSON . stringify ( cell ) ; } return cell ; } return '-' ; } ; } private generateHeader ( columns , builder , settings ) { this . logger . debug ( `Building CSV header row...` ) ; const header = columns . map ( this . escapeValues ( settings ) ) . join ( settings . separator ) + '\n' ; if ( ! builder . tryAppend ( header ) ) { return { content : '' , maxSizeReached : true , warnings : [ ] , } ; } } private async generateRows ( columns , table , builder , formatters , settings ) { this . logger . debug ( `Building ${ table . rows . length } CSV data rows...` ) ; for ( const dataTableRow of table . rows ) { if ( this . cancellationToken . isCancelled ( ) ) { break ; } await new Promise ( setImmediate ) ; const rowDefinition = [ ] ; const format = this . formatCellValues ( formatters ) ; const escape = this . escapeValues ( settings ) ; for ( const column of columns ) { rowDefinition . push ( escape ( format ( { column , data : dataTableRow [ column ] } ) ) ) ; } if ( ! builder . tryAppend ( rowDefinition . join ( settings . separator ) + '\n' ) ) { this . logger . warn ( `Max Size Reached after ${ this . csvRowCount } rows.` ) ; this . maxSizeReached = true ; if ( this . cancellationToken ) { this . cancellationToken . cancel ( ) ; } break ; } this . csvRowCount ++ ; } } public async generateData ( ) { const [ settings , searchSource ] = await Promise . all ( [ getExportSettings ( this . clients . uiSettings , this . config , this . job . browserTimezone , this . logger ) , this . dependencies . searchSourceStart . create ( this . job . searchSource ) , ] ) ; const index = searchSource . getField ( 'index' ) ; if ( ! index ) { throw new Error ( `The search must have a reference to an index pattern!` ) ; } const { maxSizeBytes , bom , escapeFormulaValues , scroll : scrollSettings } = settings ; const builder = new MaxSizeStringBuilder ( this . stream , byteSizeValueToNumber ( maxSizeBytes ) , bom ) ; const warnings = [ ] ; let first = true ; let currentRecord = - 1 ; let totalRecords = 0 ; let scrollId ; try { index . fields . getByType ( 'date' ) . forEach ( ( { name } ) => { this . logger . debug ( `setting timezone on ${ name }` ) ; const format = { ... index . fieldFormatMap [ name ] , id : index . fieldFormatMap [ name ] ?. id || 'date' , params : { ... index . fieldFormatMap [ name ] ?. params , timezone : settings . timezone , } , } ; index . setFieldFormat ( name , format ) ; } ) ; } catch ( err ) { this . logger . error ( err ) ; } try { do { if ( this . cancellationToken . isCancelled ( ) ) { break ; } let results ; if ( scrollId == null ) { results = await this . scan ( index , searchSource , settings ) ; scrollId = results ?. _scroll_id ; if ( results . hits ?. total != null ) { totalRecords = results . hits . total as number ; this . logger . debug ( `Total search results: ${ totalRecords }` ) ; } } else { results = await this . scroll ( scrollId , scrollSettings ) ; } if ( ! results ) { this . logger . warning ( `Search results are undefined!` ) ; break ; } { const { hits : { hits , ... hitsMeta } , ... header } = results ; this . logger . debug ( 'Results metadata: ' + JSON . stringify ( { header , hitsMeta } ) ) ; } let table ; try { table = tabifyDocs ( results , index , { shallow : true , includeIgnoredValues : true } ) ; } catch ( err ) { this . logger . error ( err ) ; } if ( ! table ) { break ; } const columns = this . getColumns ( searchSource , table ) || [ ] ; if ( first ) { first = false ; this . generateHeader ( columns , builder , settings ) ; } if ( table . rows . length < 1 ) { break ; } const formatters = this . getFormatters ( table ) ; await this . generateRows ( columns , table , builder , formatters , settings ) ; currentRecord += table . rows . length ; } while ( currentRecord < totalRecords - 1 ) ; if ( this . csvContainsFormulas && escapeFormulaValues ) { warnings . push ( i18n . translate ( 'xpack.reporting.exportTypes.csv.generateCsv.escapedFormulaValues' , { defaultMessage : 'CSV may contain formulas whose values have been escaped' , } ) ) ; } } catch ( err ) { this . logger . error ( err ) ; if ( err instanceof KbnServerError && err . errBody ) { throw JSON . stringify ( err . errBody . error ) ; } } finally { if ( scrollId ) { this . logger . debug ( `executing clearScroll request` ) ; try { await this . clients . es . asCurrentUser . clearScroll ( { scroll_id : [ scrollId ] } ) ; } catch ( err ) { this . logger . error ( err ) ; } } else { this . logger . warn ( `No scrollId to clear!` ) ; } } this . logger . debug ( `Finished generating. Row count: ${ this . csvRowCount }.` ) ; if ( ! this . maxSizeReached && this . csvRowCount !== totalRecords ) { this . logger . warning ( `ES scroll returned fewer total hits than expected! ` + `Search result total hits: ${ totalRecords }. Row count: ${ this . csvRowCount }.` ) ; } return { content_type : CONTENT_TYPE_CSV , csv_contains_formulas : this . csvContainsFormulas && ! escapeFormulaValues , max_size_reached : this . maxSizeReached , warnings , } ; } }