import { resolve } from 'path' ; import { createReadStream } from 'fs' ; import { Readable } from 'stream' ; import { ToolingLog , OsdClient } from '@osd/dev-utils' ; import { Client } from 'elasticsearch' ; import { createPromiseFromStreams , concatStreamProviders } from '../lib/streams' ; import { isGzip , createStats , prioritizeMappings , readDirectory , createParseArchiveStreams , createCreateIndexStream , createIndexDocRecordsStream , migrateOpenSearchDashboardsIndex , Progress , createDefaultSpace , } from '../lib' ; const pipeline = ( ... streams ) => streams . reduce ( ( source , dest ) => source . once ( 'error' , ( error ) => dest . emit ( 'error' , error ) ) . pipe ( dest as any ) ) ; export async function loadAction ( { name , skipExisting , useCreate , client , dataDir , log , osdClient , } ) { const inputDir = resolve ( dataDir , name ) ; const stats = createStats ( name , log ) ; const files = prioritizeMappings ( await readDirectory ( inputDir ) ) ; const opensearchDashboardsPluginIds = await osdClient . plugins . getEnabledIds ( ) ; const recordStream = concatStreamProviders ( files . map ( ( filename ) => ( ) => { log . info ( '[%s] Loading %j' , name , filename ) ; return pipeline ( createReadStream ( resolve ( inputDir , filename ) ) , ... createParseArchiveStreams ( { gzip : isGzip ( filename ) } ) ) ; } ) , { objectMode : true } ) ; const progress = new Progress ( ) ; progress . activate ( log ) ; await createPromiseFromStreams ( [ recordStream , createCreateIndexStream ( { client , stats , skipExisting , log } ) , createIndexDocRecordsStream ( client , stats , progress , useCreate ) , ] ) ; progress . deactivate ( ) ; const result = stats . toJSON ( ) ; for ( const [ index , { docs } ] of Object . entries ( result ) ) { if ( docs && docs . indexed > 0 ) { log . info ( '[%s] Indexed %d docs into %j' , name , docs . indexed , index ) ; } } await client . indices . refresh ( { index : '_all' , allowNoIndices : true , } ) ; if ( Object . keys ( result ) . some ( ( k ) => k . startsWith ( '.kibana' ) ) ) { await migrateOpenSearchDashboardsIndex ( { client , osdClient } ) ; if ( opensearchDashboardsPluginIds . includes ( 'spaces' ) ) { await createDefaultSpace ( { client , index : '.kibana' } ) ; } } return result ; }