import * as Either from 'fp-ts/lib/Either' ; import * as TaskEither from 'fp-ts/lib/TaskEither' ; import { errors as EsErrors } from '@elastic/elasticsearch' ; import { pipe } from 'fp-ts/lib/pipeable' ; import { ElasticsearchClient } from '../../../elasticsearch' ; import { catchRetryableEsClientErrors , RetryableEsClientError , } from './catch_retryable_es_client_errors' ; import type { IndexNotFound , AcknowledgeResponse } from './' ; import { waitForIndexStatusYellow } from './wait_for_index_status_yellow' ; import { DEFAULT_TIMEOUT , INDEX_AUTO_EXPAND_REPLICAS , INDEX_NUMBER_OF_SHARDS , WAIT_FOR_ALL_SHARDS_TO_BE_ACTIVE , } from './constants' ; export type CloneIndexResponse = AcknowledgeResponse ; internal export interface CloneIndexParams { client : ElasticsearchClient ; source : string ; target : string ; /** only used for testing */ timeout ? : string ; } remarks export const cloneIndex = ( { client , source , target , timeout = DEFAULT_TIMEOUT , } ) : TaskEither . TaskEither < RetryableEsClientError | IndexNotFound , CloneIndexResponse > => { const cloneTask = ( ) => { return client . indices . clone ( { index : source , target , wait_for_active_shards : WAIT_FOR_ALL_SHARDS_TO_BE_ACTIVE , body : { settings : { index : { 'blocks.write' : false , number_of_shards : INDEX_NUMBER_OF_SHARDS , auto_expand_replicas : INDEX_AUTO_EXPAND_REPLICAS , refresh_interval : '1s' , priority : 10 , } , } , } , timeout , } , { maxRetries : 0 } ) . then ( ( res ) => { /**
             * - acknowledged=false, we timed out before the cluster state was
             *   updated with the newly created index, but it probably will be
             *   created sometime soon.
             * - shards_acknowledged=false, we timed out before all shards were
             *   started
             * - acknowledged=true, shards_acknowledged=true, cloning complete
             */ return Either . right ( { acknowledged : res . body . acknowledged , shardsAcknowledged : res . body . shards_acknowledged , } ) ; } ) . catch ( ( error ) => { if ( error ?. body ?. error ?. type === 'index_not_found_exception' ) { return Either . left ( { type : 'index_not_found_exception' as const , index : error . body . error . index , } ) ; } else if ( error ?. body ?. error ?. type === 'resource_already_exists_exception' ) { /**
                 * If the target index already exists it means a previous clone
                 * operation had already been started. However, we can't be sure
                 * that all shards were started so return shardsAcknowledged: false
                 */ return Either . right ( { acknowledged : true , shardsAcknowledged : false , } ) ; } else { throw error ; } } ) . catch ( catchRetryableEsClientErrors ) ; } ; return pipe ( cloneTask , TaskEither . chain ( ( res ) => { if ( res . acknowledged && res . shardsAcknowledged ) { return TaskEither . right ( res ) ; } else { return pipe ( waitForIndexStatusYellow ( { client , index : target , timeout } ) , TaskEither . map ( ( value ) => { /** When the index status is 'yellow' we know that all shards were started */ return { acknowledged : true , shardsAcknowledged : true } ; } ) ) ; } } ) ) ; } ;