import { Transform , Readable } from 'stream' ; import { inspect } from 'util' ; import { Client } from 'elasticsearch' ; import { ToolingLog } from '@osd/dev-utils' ; import { Stats } from '../stats' ; import { deleteOpenSearchDashboardsIndices } from './opensearch_dashboards_index' ; import { deleteIndex } from './delete_index' ; interface DocRecord { value : { index : string ; type : string ; settings : Record < string , any > ; mappings : Record < string , any > ; aliases : Record < string , any > ; } ; } export function createCreateIndexStream ( { client , stats , skipExisting = false , log , } ) { const skipDocsFromIndices = new Set ( ) ; let opensearchDashboardsIndexAlreadyDeleted = false ; async function handleDoc ( stream , record ) { if ( skipDocsFromIndices . has ( record . value . index ) ) { return ; } stream . push ( record ) ; } async function handleIndex ( record ) { const { index , settings , mappings , aliases } = record . value ; const isPre7Mapping = ! ! mappings && Object . keys ( mappings ) . length > 0 && ! mappings . properties ; const isOpenSearchDashboards = index . startsWith ( '.kibana' ) ; async function attemptToCreate ( attemptNumber = 1 ) { try { if ( isOpenSearchDashboards && ! opensearchDashboardsIndexAlreadyDeleted ) { await deleteOpenSearchDashboardsIndices ( { client , stats , log } ) ; opensearchDashboardsIndexAlreadyDeleted = true ; } await client . indices . create ( { method : 'PUT' , index , include_type_name : isPre7Mapping , body : { settings , mappings , aliases , } , } as any ) ; stats . createdIndex ( index , { settings } ) ; } catch ( err ) { if ( err ?. body ?. error ?. reason ?. includes ( 'index exists with the same name as the alias' ) && attemptNumber < 3 ) { opensearchDashboardsIndexAlreadyDeleted = false ; const aliasStr = inspect ( aliases ) ; log . info ( `failed to create aliases [${ aliasStr }] because OpenSearch indicated an index/alias already exists, trying again` ) ; await attemptToCreate ( attemptNumber + 1 ) ; return ; } if ( err ?. body ?. error ?. type !== 'resource_already_exists_exception' || attemptNumber >= 3 ) { throw err ; } if ( skipExisting ) { skipDocsFromIndices . add ( index ) ; stats . skippedIndex ( index ) ; return ; } await deleteIndex ( { client , stats , index , log } ) ; await attemptToCreate ( attemptNumber + 1 ) ; return ; } } await attemptToCreate ( ) ; } return new Transform ( { readableObjectMode : true , writableObjectMode : true , async transform ( record , enc , callback ) { try { switch ( record && record . type ) { case 'index' : await handleIndex ( record ) ; break ; case 'doc' : await handleDoc ( this , record ) ; break ; default : this . push ( record ) ; break ; } callback ( ) ; } catch ( err ) { callback ( err ) ; } } , } ) ; }