import { IndexResponse , UpdateResponse } from '@elastic/elasticsearch/lib/api/typesWithBodyKey' ; import { ElasticsearchClient } from 'src/core/server' ; import { LevelLogger , statuses } from '../' ; import { ReportingCore } from '../../' ; import { ILM_POLICY_NAME , REPORTING_SYSTEM_INDEX } from '../../../common/constants' ; import { JobStatus , ReportOutput , ReportSource } from '../../../common/types' ; import { ReportTaskParams } from '../tasks' ; import { IReport , Report , ReportDocument , SavedReport } from './' ; import { IlmPolicyManager } from './ilm_policy_manager' ; import { indexTimestamp } from './index_timestamp' ; import { mapping } from './mapping' ; import { MIGRATION_VERSION } from './report' ; export type ReportProcessingFields = Required < { kibana_id : Report [ 'kibana_id' ] ; kibana_name : Report [ 'kibana_name' ] ; attempts : Report [ 'attempts' ] ; started_at : Report [ 'started_at' ] ; max_attempts : Report [ 'max_attempts' ] ; timeout : Report [ 'timeout' ] ; process_expiration : Report [ 'process_expiration' ] ; } > ; export type ReportFailedFields = Required < { completed_at : Report [ 'completed_at' ] ; output : ReportOutput | null ; } > ; export type ReportCompletedFields = Required < { completed_at : Report [ 'completed_at' ] ; output : Omit < ReportOutput , 'content' > | null ; } > ; export interface ReportRecordTimeout { _id : string ; _index : string ; _source : { status : JobStatus ; process_expiration ? : string ; } ; } const sourceDoc = ( doc ) : Partial < ReportSource > => { return { ... doc , migration_version : MIGRATION_VERSION , } ; } ; const jobDebugMessage = ( report ) => `${ report . _id } ` + `[_index: ${ report . _index }] ` + `[_seq_no: ${ report . _seq_no }]  ` + `[_primary_term: ${ report . _primary_term }]` + `[attempts: ${ report . attempts }] ` + `[process_expiration: ${ report . process_expiration }]` ; export class ReportingStore { private readonly indexPrefix : string ; private readonly indexInterval : string ; private client ? : ElasticsearchClient ; private ilmPolicyManager ? : IlmPolicyManager ; constructor ( private reportingCore , private logger ) { const config = reportingCore . getConfig ( ) ; this . indexPrefix = REPORTING_SYSTEM_INDEX ; this . indexInterval = config . get ( 'queue' , 'indexInterval' ) ; this . logger = logger . clone ( [ 'store' ] ) ; } private async getClient ( ) { if ( ! this . client ) { ( { asInternalUser : this . client } = await this . reportingCore . getEsClient ( ) ) ; } return this . client ; } private async getIlmPolicyManager ( ) { if ( ! this . ilmPolicyManager ) { const client = await this . getClient ( ) ; this . ilmPolicyManager = IlmPolicyManager . create ( { client } ) ; } return this . ilmPolicyManager ; } private async createIndex ( indexName ) { const client = await this . getClient ( ) ; const { body : exists } = await client . indices . exists ( { index : indexName } ) ; if ( exists ) { return exists ; } try { await client . indices . create ( { index : indexName , body : { settings : { number_of_shards : 1 , auto_expand_replicas : '0-1' , lifecycle : { name : ILM_POLICY_NAME , } , } , mappings : { properties : mapping , } , } , } ) ; return true ; } catch ( error ) { const isIndexExistsError = error . message . match ( /resource_already_exists_exception/ ) ; if ( isIndexExistsError ) { this . logger . warn ( `Automatic index creation failed: index already exists: ${ error }` ) ; return ; } this . logger . error ( error ) ; throw error ; } } private async indexReport ( report ) { const doc = { index : report . _index ! , id : report . _id , refresh : true , body : { ... report . toReportSource ( ) , ... sourceDoc ( { process_expiration : new Date ( 0 ) . toISOString ( ) , attempts : 0 , status : statuses . JOB_STATUS_PENDING , } ) , } , } ; const client = await this . getClient ( ) ; const { body } = await client . index ( doc ) ; return body ; } private async refreshIndex ( index ) { const client = await this . getClient ( ) ; return client . indices . refresh ( { index } ) ; } /**
     * Function to be called during plugin start phase. This ensures the environment is correctly
     * configured for storage of reports.
     */ public async start ( ) { const ilmPolicyManager = await this . getIlmPolicyManager ( ) ; try { if ( await ilmPolicyManager . doesIlmPolicyExist ( ) ) { this . logger . debug ( `Found ILM policy ${ ILM_POLICY_NAME }; skipping creation.` ) ; return ; } this . logger . info ( `Creating ILM policy for managing reporting indices: ${ ILM_POLICY_NAME }` ) ; await ilmPolicyManager . createIlmPolicy ( ) ; } catch ( e ) { this . logger . error ( 'Error in start phase' ) ; this . logger . error ( e . body ?. error ) ; throw e ; } } public async addReport ( report ) { let index = report . _index ; if ( ! index ) { const timestamp = indexTimestamp ( this . indexInterval ) ; index = `${ this . indexPrefix }-${ timestamp }` ; report . _index = index ; } await this . createIndex ( index ) ; try { report . updateWithEsDoc ( await this . indexReport ( report ) ) ; await this . refreshIndex ( index ) ; return report as SavedReport ; } catch ( err ) { this . reportingCore . getEventLogger ( report ) . logError ( err ) ; this . logError ( `Error in adding a report!` , err , report ) ; throw err ; } } public async findReportFromTask ( taskJson ) { if ( ! taskJson . index ) { throw new Error ( 'Task JSON is missing index field!' ) ; } if ( ! taskJson . id || ! taskJson . index ) { const notRetrievable = new Error ( `Unable to retrieve pending report: Invalid report ID!` ) ; this . logger . error ( notRetrievable ) ; throw notRetrievable ; } try { const client = await this . getClient ( ) ; const { body : document } = await client . get < ReportSource > ( { index : taskJson . index , id : taskJson . id , } ) ; return new SavedReport ( { _id : document . _id , _index : document . _index , _seq_no : document . _seq_no , _primary_term : document . _primary_term , jobtype : document . _source ?. jobtype , attempts : document . _source ?. attempts , created_at : document . _source ?. created_at , created_by : document . _source ?. created_by , max_attempts : document . _source ?. max_attempts , meta : document . _source ?. meta , payload : document . _source ?. payload , process_expiration : document . _source ?. process_expiration , status : document . _source ?. status , timeout : document . _source ?. timeout , } ) ; } catch ( err ) { this . logger . error ( `Error in finding the report from the scheduled task info! ` + `[id: ${ taskJson . id }] [index: ${ taskJson . index }]` ) ; this . logger . error ( err ) ; this . reportingCore . getEventLogger ( { _id : taskJson . id } as IReport ) . logError ( err ) ; throw err ; } } public async setReportClaimed ( report , processingInfo ) { const doc = sourceDoc ( { ... processingInfo , status : statuses . JOB_STATUS_PROCESSING , } ) ; let body ; try { const client = await this . getClient ( ) ; body = ( await client . update < ReportDocument > ( { id : report . _id , index : report . _index , if_seq_no : report . _seq_no , if_primary_term : report . _primary_term , refresh : true , body : { doc } , } ) ) . body ; } catch ( err ) { this . logError ( `Error in updating status to processing! Report: ${ jobDebugMessage ( report ) }` , err , report ) ; throw err ; } this . reportingCore . getEventLogger ( report ) . logClaimTask ( ) ; return body ; } private logError ( message , err , report ) { this . logger . error ( message ) ; this . logger . error ( err ) ; this . reportingCore . getEventLogger ( report ) . logError ( err ) ; } public async setReportFailed ( report , failedInfo ) { const doc = sourceDoc ( { ... failedInfo , status : statuses . JOB_STATUS_FAILED , } ) ; let body ; try { const client = await this . getClient ( ) ; body = ( await client . update < ReportDocument > ( { id : report . _id , index : report . _index , if_seq_no : report . _seq_no , if_primary_term : report . _primary_term , refresh : true , body : { doc } , } ) ) . body ; } catch ( err ) { this . logError ( `Error in updating status to failed! Report: ${ jobDebugMessage ( report ) }` , err , report ) ; throw err ; } this . reportingCore . getEventLogger ( report ) . logReportFailure ( ) ; return body ; } public async setReportCompleted ( report , completedInfo ) { const { output } = completedInfo ; const status = output && output . warnings && output . warnings . length > 0 ? statuses . JOB_STATUS_WARNINGS : statuses . JOB_STATUS_COMPLETED ; const doc = sourceDoc ( { ... completedInfo , status , } as ReportSource ) ; let body ; try { const client = await this . getClient ( ) ; body = ( await client . update < ReportDocument > ( { id : report . _id , index : report . _index , if_seq_no : report . _seq_no , if_primary_term : report . _primary_term , refresh : true , body : { doc } , } ) ) . body ; } catch ( err ) { this . logError ( `Error in updating status to complete! Report: ${ jobDebugMessage ( report ) }` , err , report ) ; throw err ; } this . reportingCore . getEventLogger ( report ) . logReportSaved ( ) ; return body ; } public async prepareReportForRetry ( report ) { const doc = sourceDoc ( { status : statuses . JOB_STATUS_PENDING , process_expiration : null , } ) ; let body ; try { const client = await this . getClient ( ) ; body = ( await client . update < ReportDocument > ( { id : report . _id , index : report . _index , if_seq_no : report . _seq_no , if_primary_term : report . _primary_term , refresh : true , body : { doc } , } ) ) . body ; } catch ( err ) { this . logError ( `Error in clearing expiration and status for retry! Report: ${ jobDebugMessage ( report ) }` , err , report ) ; throw err ; } return body ; } public async findStaleReportJob ( ) { const client = await this . getClient ( ) ; const expiredFilter = { bool : { must : [ { range : { process_expiration : { lt : `now` } } } , { terms : { status : [ statuses . JOB_STATUS_PROCESSING ] } } , ] , } , } ; const oldVersionFilter = { bool : { must : [ { terms : { status : [ statuses . JOB_STATUS_PENDING ] } } ] , must_not : [ { exists : { field : 'migration_version' } } ] , } , } ; const { body } = await client . search < ReportRecordTimeout [ '_source' ] > ( { size : 1 , index : this . indexPrefix + '-*' , seq_no_primary_term : true , _source_excludes : [ 'output' ] , body : { sort : { created_at : { order : 'asc' as const } } , query : { bool : { filter : { bool : { should : [ expiredFilter , oldVersionFilter ] } } } } , } , } ) ; return body . hits ?. hits [ 0 ] as ReportRecordTimeout ; } public getReportingIndexPattern ( ) { return `${ this . indexPrefix }-*` ; } }