import _ from 'lodash' ; import { MigrationOpenSearchClient } from './migration_opensearch_client' ; import { CountResponse , SearchResponse } from '../../../opensearch' ; import { IndexMapping } from '../../mappings' ; import { SavedObjectsMigrationVersion } from '../../types' ; import { AliasAction , RawDoc , ShardsInfo } from './call_cluster' ; import { SavedObjectsRawDocSource } from '../../serialization' ; const settings = { number_of_shards : 1 , auto_expand_replicas : '0-1' } ; export interface FullIndexInfo { aliases : { [ name ] : object ; } ; exists : boolean ; indexName : string ; mappings : IndexMapping ; } /**
 * A slight enhancement to indices.get, that adds indexName, and validates that the
 * index mappings are somewhat what we expect.
 */ export async function fetchInfo ( client , index ) { const { body , statusCode } = await client . indices . get ( { index } , { ignore : [ 404 ] } ) ; if ( statusCode === 404 ) { return { aliases : { } , exists : false , indexName : index , mappings : { dynamic : 'strict' , properties : { } } , } ; } const [ indexName , indexInfo ] = Object . entries ( body ) [ 0 ] ; return assertIsSupportedIndex ( { ... indexInfo , exists : true , indexName } ) ; } param { CallCluster } callCluster param { string } param { opts } prop prop export function reader ( client , index , { batchSize = 10 , scrollDuration = '15m' } ) { const scroll = scrollDuration ; let scrollId ; const nextBatch = ( ) => scrollId !== undefined ? client . scroll < SearchResponse < SavedObjectsRawDocSource > > ( { scroll , scroll_id : scrollId , } ) : client . search < SearchResponse < SavedObjectsRawDocSource > > ( { body : { size : batchSize } , index , scroll , } ) ; const close = async ( ) => scrollId && ( await client . clearScroll ( { scroll_id : scrollId } ) ) ; return async function read ( ) { const result = await nextBatch ( ) ; assertResponseIncludeAllShards ( result . body ) ; scrollId = result . body . _scroll_id ; const docs = result . body . hits . hits ; if ( ! docs . length ) { await close ( ) ; } return docs ; } ; } param { CallCluster } callCluster param { string } index param { RawDoc [ ] } docs export async function write ( client , index , docs ) { const { body } = await client . bulk ( { body : docs . reduce ( ( acc , doc ) => { acc . push ( { index : { _id : doc . _id , _index : index , } , } ) ; acc . push ( doc . _source ) ; return acc ; } , [ ] ) , } ) ; const err = _ . find ( body . items , 'index.error.reason' ) ; if ( ! err ) { return ; } const exception = new Error ( err . index . error ! . reason ) ; exception . detail = err ; throw exception ; } param { CallCluster } callCluster param { string } index param { SavedObjectsMigrationVersion } migrationVersion export async function migrationsUpToDate ( client , index , migrationVersion , retryCount = 10 ) { try { const indexInfo = await fetchInfo ( client , index ) ; if ( ! indexInfo . mappings . properties ?. migrationVersion ) { return false ; } if ( Object . keys ( migrationVersion ) . length <= 0 ) { return true ; } const { body } = await client . count < CountResponse > ( { body : { query : { bool : { should : Object . entries ( migrationVersion ) . map ( ( [ type , latestVersion ] ) => ( { bool : { must : [ { exists : { field : type } } , { bool : { must_not : { term : { [ `migrationVersion.${ type }` ] : latestVersion } } , } , } , ] , } , } ) ) , } , } , } , index , } ) ; assertResponseIncludeAllShards ( body ) ; return body . count === 0 ; } catch ( e ) { if ( e . status !== 503 || retryCount === 0 ) { throw e ; } await new Promise ( ( r ) => setTimeout ( r , 1000 ) ) ; return await migrationsUpToDate ( client , index , migrationVersion , retryCount - 1 ) ; } } export async function createIndex ( client , index , mappings ? ) { await client . indices . create ( { body : { mappings , settings } , index , } ) ; } export async function deleteIndex ( client , index ) { await client . indices . delete ( { index } ) ; } param { CallCluster } callCluster param { FullIndexInfo } info param { string } alias export async function convertToAlias ( client , info , alias , batchSize , script ? ) { await client . indices . create ( { body : { mappings : info . mappings , settings } , index : info . indexName , } ) ; await reindex ( client , alias , info . indexName , batchSize , script ) ; await claimAlias ( client , info . indexName , alias , [ { remove_index : { index : alias } } ] ) ; } param { CallCluster } callCluster param { string } index param { string } alias param { AliasAction [ ] } aliasActions export async function claimAlias ( client , index , alias , aliasActions = [ ] ) { const { body , statusCode } = await client . indices . getAlias ( { name : alias } , { ignore : [ 404 ] } ) ; const aliasInfo = statusCode === 404 ? { } : body ; const removeActions = Object . keys ( aliasInfo ) . map ( ( key ) => ( { remove : { index : key , alias } } ) ) ; await client . indices . updateAliases ( { body : { actions : aliasActions . concat ( removeActions ) . concat ( { add : { index , alias } } ) , } , } ) ; await client . indices . refresh ( { index } ) ; } param { FullIndexInfo } indexInfo function assertIsSupportedIndex ( indexInfo ) { const mappings = indexInfo . mappings as any ; const isV7Index = ! ! mappings . properties ; if ( ! isV7Index ) { throw new Error ( `Index ${ indexInfo . indexName } belongs to a version of OpenSearch Dashboards ` + `that cannot be automatically migrated. Reset it or use the X-Pack upgrade assistant.` ) ; } return indexInfo ; } /**
 * Provides protection against reading/re-indexing against an index with missing
 * shards which could result in data loss. This shouldn't be common, as the Saved
 * Object indices should only ever have a single shard. This is more to handle
 * instances where customers manually expand the shards of an index.
 */ function assertResponseIncludeAllShards ( { _shards } ) { if ( ! _ . has ( _shards , 'total' ) || ! _ . has ( _shards , 'successful' ) ) { return ; } const failed = _shards . total - _shards . successful ; if ( failed > 0 ) { throw new Error ( `Re-index failed :: ${ failed } of ${ _shards . total } shards failed. ` + `Check OpenSearch cluster health for more information.` ) ; } } /**
 * Reindexes from source to dest, polling for the reindex completion.
 */ async function reindex ( client , source , dest , batchSize , script ? ) { const pollInterval = 250 ; const { body : reindexBody } = await client . reindex ( { body : { dest : { index : dest } , source : { index : source , size : batchSize } , script : script ? { source : script , lang : 'painless' , } : undefined , } , refresh : true , wait_for_completion : false , } ) ; const task = reindexBody . task ; let completed = false ; while ( ! completed ) { await new Promise ( ( r ) => setTimeout ( r , pollInterval ) ) ; const { body } = await client . tasks . get ( { task_id : task , } ) ; if ( body . error ) { const e = body . error ; throw new Error ( `Re-index failed [${ e . type }] ${ e . reason } :: ${ JSON . stringify ( e ) }` ) ; } completed = body . completed ; } }