import expect from '@kbn/expect' ; import type { PutTransformsRequestSchema } from '../../../../plugins/transform/common/api_schemas/transforms' ; import { TransformState , TRANSFORM_STATE } from '../../../../plugins/transform/common/constants' ; import type { TransformStats } from '../../../../plugins/transform/common/types/transform_stats' ; import { FtrProviderContext } from '../../ftr_provider_context' ; import { GetTransformsResponseSchema } from '../../../../plugins/transform/common/api_schemas/transforms' ; import { PostTransformsUpdateRequestSchema } from '../../../../plugins/transform/common/api_schemas/update_transforms' ; import { TransformPivotConfig } from '../../../../plugins/transform/common/types/transform' ; export async function asyncForEach ( array , callback ) { for ( let index = 0 ; index < array . length ; index ++ ) { await callback ( array [ index ] , index , array ) ; } } export function TransformAPIProvider ( { getService } ) { const es = getService ( 'es' ) ; const log = getService ( 'log' ) ; const retry = getService ( 'retry' ) ; const esSupertest = getService ( 'esSupertest' ) ; const esDeleteAllIndices = getService ( 'esDeleteAllIndices' ) ; return { async createIndices ( indices ) { log . debug ( `Creating indices: '${ indices }'...` ) ; if ( ( await es . indices . exists ( { index : indices , allow_no_indices : false } ) ) === true ) { log . debug ( `Indices '${ indices }' already exist. Nothing to create.` ) ; return ; } const createResponse = await es . indices . create ( { index : indices } ) ; expect ( createResponse ) . to . have . property ( 'acknowledged' ) . eql ( true , 'Response for create request indices should be acknowledged.' ) ; await this . waitForIndicesToExist ( indices , `expected ${ indices } to be created` ) ; } , async deleteIndices ( indices , skipWaitForIndicesNotToExist ? ) { log . debug ( `Deleting indices: '${ indices }'...` ) ; if ( ( await es . indices . exists ( { index : indices , allow_no_indices : false } ) ) === false ) { log . debug ( `Indices '${ indices }' don't exist. Nothing to delete.` ) ; return ; } const deleteResponse = await es . indices . delete ( { index : indices , } ) ; expect ( deleteResponse ) . to . have . property ( 'acknowledged' ) . eql ( true , 'Response for delete request should be acknowledged' ) ; if ( ! skipWaitForIndicesNotToExist ) { await this . waitForIndicesNotToExist ( indices , `expected indices '${ indices }' to be deleted` ) ; } } , async waitForIndicesToExist ( indices , errorMsg ? ) { await retry . tryForTime ( 30 * 1000 , async ( ) => { if ( ( await es . indices . exists ( { index : indices , allow_no_indices : false } ) ) === true ) { return true ; } else { throw new Error ( errorMsg || `indices '${ indices }' should exist` ) ; } } ) ; } , async waitForIndicesNotToExist ( indices , errorMsg ? ) { await retry . tryForTime ( 30 * 1000 , async ( ) => { if ( ( await es . indices . exists ( { index : indices , allow_no_indices : false } ) ) === false ) { return true ; } else { throw new Error ( errorMsg || `indices '${ indices }' should not exist` ) ; } } ) ; } , async cleanTransformIndices ( ) { const { body : { transforms } , } = await esSupertest . get ( `/_transform/` ) . expect ( 200 ) ; const transformIds = transforms . map ( ( t ) => t . id ) ; await asyncForEach ( transformIds , async ( transformId ) => { await esSupertest . post ( `/_transform/${ transformId }/_stop?force=true&wait_for_completion` ) . expect ( 200 ) ; await this . waitForTransformState ( transformId , TRANSFORM_STATE . STOPPED ) ; await esSupertest . delete ( `/_transform/${ transformId }` ) . expect ( 200 ) ; await this . waitForTransformNotToExist ( transformId ) ; } ) ; await esDeleteAllIndices ( '.transform-notifications-*' ) ; } , async getTransformStats ( transformId ) { log . debug ( `Fetching transform stats for transform ${ transformId }` ) ; const statsResponse = await esSupertest . get ( `/_transform/${ transformId }/_stats` ) . expect ( 200 ) . then ( ( res ) => res . body ) ; expect ( statsResponse . transforms ) . to . have . length ( 1 , `Expected transform stats to contain exactly 1 object (got '${ statsResponse . transforms . length }')` ) ; return statsResponse . transforms [ 0 ] ; } , async getTransformState ( transformId ) { const stats = await this . getTransformStats ( transformId ) ; return stats . state ; } , async waitForTransformState ( transformId , expectedState ) { await retry . waitForWithTimeout ( `transform state to be ${ expectedState }` , 2 * 60 * 1000 , async ( ) => { const state = await this . getTransformState ( transformId ) ; if ( state === expectedState ) { return true ; } else { throw new Error ( `expected transform state to be ${ expectedState } but got ${ state }` ) ; } } ) ; } , async waitForTransformStateNotToBe ( transformId , notExpectedState ) { await retry . waitForWithTimeout ( `transform state not to be ${ notExpectedState }` , 2 * 60 * 1000 , async ( ) => { const state = await this . getTransformState ( transformId ) ; if ( state !== notExpectedState ) { return true ; } else { throw new Error ( `expected transform state to not be ${ notExpectedState } but got ${ state }` ) ; } } ) ; } , async waitForBatchTransformToComplete ( transformId ) { await retry . waitForWithTimeout ( `batch transform to complete` , 2 * 60 * 1000 , async ( ) => { const stats = await this . getTransformStats ( transformId ) ; if ( stats . state === TRANSFORM_STATE . STOPPED && stats . checkpointing . last . checkpoint === 1 ) { return true ; } else { throw new Error ( `expected batch transform to be stopped with last checkpoint = 1 (got status: '${ stats . state }', checkpoint: '${ stats . checkpointing . last . checkpoint }')` ) ; } } ) ; } , async getTransformList ( size = 10 ) { return ( await esSupertest . get ( `/_transform` ) . expect ( 200 ) . then ( ( response ) => response . body ) ) as GetTransformsResponseSchema ; } , async getTransform ( transformId , expectedCode = 200 ) { return await esSupertest . get ( `/_transform/${ transformId }` ) . expect ( expectedCode ) ; } , async updateTransform ( transformId , updates ) { return await esSupertest . post ( `/_transform/${ transformId }/_update` ) . send ( updates ) . expect ( 200 ) . then ( ( response ) => response . body ) ; } , async createTransform ( transformId , transformConfig ) { log . debug ( `Creating transform with id '${ transformId }'...` ) ; await esSupertest . put ( `/_transform/${ transformId }` ) . send ( transformConfig ) . expect ( 200 ) ; await this . waitForTransformToExist ( transformId , `expected transform '${ transformId }' to be created` ) ; } , async waitForTransformToExist ( transformId , errorMsg ? ) { await retry . waitForWithTimeout ( `'${ transformId }' to exist` , 5 * 1000 , async ( ) => { if ( await this . getTransform ( transformId , 200 ) ) { return true ; } else { throw new Error ( errorMsg || `expected transform '${ transformId }' to exist` ) ; } } ) ; } , async waitForTransformNotToExist ( transformId , errorMsg ? ) { await retry . waitForWithTimeout ( `'${ transformId }' to exist` , 5 * 1000 , async ( ) => { if ( await this . getTransform ( transformId , 404 ) ) { return true ; } else { throw new Error ( errorMsg || `expected transform '${ transformId }' to not exist` ) ; } } ) ; } , async startTransform ( transformId ) { log . debug ( `Starting transform '${ transformId }' ...` ) ; await esSupertest . post ( `/_transform/${ transformId }/_start` ) . expect ( 200 ) ; } , async stopTransform ( transformId ) { log . debug ( `Stopping transform '${ transformId }' ...` ) ; await esSupertest . post ( `/_transform/${ transformId }/_stop` ) . expect ( 200 ) ; } , async createAndRunTransform ( transformId , transformConfig ) { await this . createTransform ( transformId , transformConfig ) ; await this . startTransform ( transformId ) ; if ( transformConfig . sync === undefined ) { await this . waitForBatchTransformToComplete ( transformId ) ; } else { await this . waitForTransformStateNotToBe ( transformId , TRANSFORM_STATE . STOPPED ) ; } } , } ; }