import { Observable , Subject } from 'rxjs' ; import { first , map , shareReplay , takeUntil } from 'rxjs/operators' ; import { merge } from '@osd/std' ; import { CoreService } from '../../types' ; import { CoreContext } from '../core_context' ; import { Logger } from '../logging' ; import { LegacyClusterClient , ILegacyCustomClusterClient , LegacyOpenSearchClientConfig , } from './legacy' ; import { ClusterClient , ICustomClusterClient , OpenSearchClientConfig } from './client' ; import { OpenSearchConfig , OpenSearchConfigType } from './opensearch_config' ; import { InternalHttpServiceSetup , GetAuthHeaders } from '../http/' ; import { AuditTrailStart , AuditorFactory } from '../audit_trail' ; import { InternalOpenSearchServiceSetup , InternalOpenSearchServiceStart } from './types' ; import { pollOpenSearchNodesVersion } from './version_check/ensure_opensearch_version' ; import { calculateStatus$ } from './status' ; interface SetupDeps { http : InternalHttpServiceSetup ; } interface StartDeps { auditTrail : AuditTrailStart ; } internal export class OpenSearchService implements CoreService < InternalOpenSearchServiceSetup , InternalOpenSearchServiceStart > { private readonly log : Logger ; private readonly config$ : Observable < OpenSearchConfig > ; private auditorFactory ? : AuditorFactory ; private stop$ = new Subject ( ) ; private opensearchDashboardsVersion : string ; private getAuthHeaders ? : GetAuthHeaders ; private createLegacyCustomClient ? : ( type : string , clientConfig ? : Partial < LegacyOpenSearchClientConfig > ) => ILegacyCustomClusterClient ; private legacyClient ? : LegacyClusterClient ; private client ? : ClusterClient ; constructor ( private readonly coreContext ) { this . opensearchDashboardsVersion = coreContext . env . packageInfo . version ; this . log = coreContext . logger . get ( 'opensearch-service' ) ; this . config$ = coreContext . configService . atPath < OpenSearchConfigType > ( 'opensearch' ) . pipe ( map ( ( rawConfig ) => new OpenSearchConfig ( rawConfig ) ) ) ; } public async setup ( deps ) { this . log . debug ( 'Setting up opensearch service' ) ; const config = await this . config$ . pipe ( first ( ) ) . toPromise ( ) ; this . getAuthHeaders = deps . http . getAuthHeaders ; this . legacyClient = this . createLegacyClusterClient ( 'data' , config ) ; this . client = this . createClusterClient ( 'data' , config ) ; const opensearchNodesCompatibility$ = pollOpenSearchNodesVersion ( { internalClient : this . client . asInternalUser , optimizedHealthcheckId : config . optimizedHealthcheckId , log : this . log , ignoreVersionMismatch : config . ignoreVersionMismatch , opensearchVersionCheckInterval : config . healthCheckDelay . asMilliseconds ( ) , opensearchDashboardsVersion : this . opensearchDashboardsVersion , } ) . pipe ( takeUntil ( this . stop$ ) , shareReplay ( { refCount : true , bufferSize : 1 } ) ) ; this . createLegacyCustomClient = ( type , clientConfig = { } ) => { const finalConfig = merge ( { } , config , clientConfig ) ; return this . createLegacyClusterClient ( type , finalConfig ) ; } ; return { legacy : { config$ : this . config$ , client : this . legacyClient , createClient : this . createLegacyCustomClient , } , opensearchNodesCompatibility$ , status$ : calculateStatus$ ( opensearchNodesCompatibility$ ) , } ; } public async start ( { auditTrail } ) { this . auditorFactory = auditTrail ; if ( ! this . legacyClient || ! this . createLegacyCustomClient ) { throw new Error ( 'OpenSearchService needs to be setup before calling start' ) ; } const config = await this . config$ . pipe ( first ( ) ) . toPromise ( ) ; const createClient = ( type , clientConfig = { } ) : ICustomClusterClient => { const finalConfig = merge ( { } , config , clientConfig ) ; return this . createClusterClient ( type , finalConfig ) ; } ; return { client : this . client ! , createClient , legacy : { config$ : this . config$ , client : this . legacyClient , createClient : this . createLegacyCustomClient , } , } ; } public async stop ( ) { this . log . debug ( 'Stopping opensearch service' ) ; this . stop$ . next ( ) ; if ( this . client ) { await this . client . close ( ) ; } if ( this . legacyClient ) { this . legacyClient . close ( ) ; } } private createClusterClient ( type , config ) { return new ClusterClient ( config , this . coreContext . logger . get ( 'opensearch' , type ) , this . getAuthHeaders ) ; } private createLegacyClusterClient ( type , config ) { return new LegacyClusterClient ( config , this . coreContext . logger . get ( 'opensearch' , type ) , this . getAuditorFactory , this . getAuthHeaders ) ; } private getAuditorFactory = ( ) => { if ( ! this . auditorFactory ) { throw new Error ( 'auditTrail has not been initialized' ) ; } return this . auditorFactory ; } ; }