diff --git a/backend/src/lib/search.ts b/backend/src/lib/search.ts index c4b0ef01d08..988e15c8f07 100644 --- a/backend/src/lib/search.ts +++ b/backend/src/lib/search.ts @@ -1,7 +1,10 @@ /* Copyright Contributors to the Open Cluster Management project */ +import type { IncomingMessage } from 'node:http' import type { OutgoingHttpHeaders } from 'node:http2' import type { RequestOptions } from 'node:https' import { request } from 'node:https' +import { pipeline } from 'node:stream/promises' +import { Writable } from 'node:stream' import { URL } from 'node:url' import { getMultiClusterHub } from '../lib/multi-cluster-hub' import { getNamespace, getServiceAccountToken } from '../lib/serviceAccountToken' @@ -26,6 +29,21 @@ export type ISearchResult = { message?: string } +function collectResponseBody(res: IncomingMessage): Promise { + return new Promise((resolve, reject) => { + let body = '' + const collector = new Writable({ + write(chunk: Buffer, _encoding, callback) { + body += chunk.toString() + callback() + }, + }) + pipeline(res, collector) + .then(() => resolve(body)) + .catch(reject) + }) +} + export async function getServiceAccountSearchRequestOptions() { const serviceAccountToken = getServiceAccountToken() const headers: OutgoingHttpHeaders = { @@ -38,9 +56,10 @@ export async function getServiceAccountSearchRequestOptions() { } export async function getSearchRequestOptions(headers: OutgoingHttpHeaders): Promise { - const mch = await getMultiClusterHub() + const multiClusterHub = await getMultiClusterHub() const namespace = getNamespace() - const machineNs = process.env.NODE_ENV === 'test' ? 'undefined' : `${mch?.metadata?.namespace || namespace}` + const machineNs = + process.env.NODE_ENV === 'test' ? 'undefined' : `${multiClusterHub?.metadata?.namespace || namespace}` const searchService = `https://search-search-api.${machineNs}.svc.cluster.local:4010` const searchUrl = process.env.SEARCH_API_URL || searchService const endpoint = process.env.globalSearchFeatureFlag === 'enabled' ? '/federated' : '/searchapi/graphql' @@ -62,41 +81,51 @@ export async function getSearchResults(query: IQuery) { const options = await getServiceAccountSearchRequestOptions() const requestTimeout = 2 * 60 * 1000 return new Promise((resolve, reject) => { - let body = '' - const id = setTimeout(() => { - logger.error(`getSearchResults request timeout`) - reject(new Error('request timeout')) - }, requestTimeout) - const req = request(options, (res) => { - res.on('data', (data) => { - body += data - }) - res.on('end', () => { - try { - const result = JSON.parse(body) as ISearchResult - const message = typeof result === 'string' ? result : result.message - if (message) { - logger.error(`getSearchResults return error ${message}`) - reject(new Error(result.message)) + let settled = false + const timeout = { requestTimeoutId: undefined as NodeJS.Timeout | undefined } + const finish = (fn: () => void) => { + if (settled) return + settled = true + clearTimeout(timeout.requestTimeoutId) + fn() + } + const clientRequest = request(options, (res) => { + void collectResponseBody(res) + .then((body) => { + try { + const result = JSON.parse(body) as ISearchResult + const message = typeof result === 'string' ? result : result.message + if (message) { + logger.error(`getSearchResults return error ${message}`) + finish(() => reject(new Error(result.message))) + return + } + finish(() => resolve(result)) + } catch (e) { + // search might be overwhelmed + // pause before next request + logger.error(`getSearchResults parse error ${e} ${body}`) + clearTimeout(timeout.requestTimeoutId) + setTimeout(() => { + finish(() => reject(new Error(body))) + }, requestTimeout) } - resolve(result) - } catch (e) { - // search might be overwhelmed - // pause before next request - logger.error(`getSearchResults parse error ${e} ${body}`) - setTimeout(() => { - reject(new Error(body)) - }, requestTimeout) - } - clearTimeout(id) - }) + }) + .catch((e: Error) => { + finish(() => reject(e)) + }) }) - req.on('error', (e) => { + timeout.requestTimeoutId = setTimeout(() => { + logger.error(`getSearchResults request timeout`) + clientRequest.destroy() + finish(() => reject(new Error('request timeout'))) + }, requestTimeout) + clientRequest.on('error', (e) => { logger.error(`getSearchResults request error ${e.message}`) - reject(e) + finish(() => reject(e)) }) - req.write(JSON.stringify(query)) - req.end() + clientRequest.write(JSON.stringify(query)) + clientRequest.end() }) } @@ -125,37 +154,45 @@ const ping = { export async function pingSearchAPI() { const options = await getServiceAccountSearchRequestOptions() return new Promise((resolve, reject) => { - let body = '' - const id = setTimeout( + let settled = false + const timeout = { requestTimeoutId: undefined as NodeJS.Timeout | undefined } + const finish = (fn: () => void) => { + if (settled) return + settled = true + clearTimeout(timeout.requestTimeoutId) + fn() + } + const clientRequest = request(options, (res) => { + void collectResponseBody(res) + .then((body) => { + try { + const result = JSON.parse(body) as { data: unknown } + if (result.data) { + finish(() => resolve(true)) + } else { + finish(() => reject(new Error('no data'))) + } + } catch (e) { + logger.error(`pingSearchAPI parse error ${e} ${body}`) + finish(() => reject(new Error(String(e).valueOf()))) + } + }) + .catch((e: Error) => { + finish(() => reject(e)) + }) + }) + timeout.requestTimeoutId = setTimeout( () => { logger.error(`ping searchAPI timeout`) - reject(new Error('request timeout')) + clientRequest.destroy() + finish(() => reject(new Error('request timeout'))) }, 4 * 60 * 1000 ) - const req = request(options, (res) => { - res.on('data', (data) => { - body += data - }) - res.on('end', () => { - try { - const result = JSON.parse(body) as { data: unknown } - if (result.data) { - resolve(true) - } else { - reject(new Error('no data')) - } - } catch (e) { - logger.error(`pingSearchAPI parse error ${e} ${body}`) - reject(new Error(String(e).valueOf())) - } - clearTimeout(id) - }) - }) - req.on('error', (e) => { - reject(e) + clientRequest.on('error', (e) => { + finish(() => reject(e)) }) - req.write(JSON.stringify(ping)) - req.end() + clientRequest.write(JSON.stringify(ping)) + clientRequest.end() }) } diff --git a/backend/src/routes/aggregators/applications.ts b/backend/src/routes/aggregators/applications.ts index 11fd1860723..11291ca08b3 100644 --- a/backend/src/routes/aggregators/applications.ts +++ b/backend/src/routes/aggregators/applications.ts @@ -4,6 +4,7 @@ import { addOCPQueryInputs, addSystemQueryInputs, cacheOCPApplications } from '. import { ApplicationSetKind, type IApplicationSet, type IResource, type SearchResult } from '../../resources/resource' import type { FilterSelections, ISortBy } from '../../lib/pagination' import { logger } from '../../lib/logger' +import { getMultiClusterHub } from '../../lib/multi-cluster-hub' import { discoverSystemAppNamespacePrefixes, getApplicationsHelper, @@ -196,12 +197,30 @@ export const promiseTimeout = (promise: Promise, delay: number) => { // ////////////////////////////////////////////////////////////////////////////////// export async function startAggregatingApplications() { await discoverSystemAppNamespacePrefixes() - void searchLoop() + await searchLoop() } let stopping = false +let cancelPendingWait: (() => void) | undefined + +function waitWhileRunning(ms: number): Promise { + if (stopping) return Promise.resolve() + return new Promise((resolve) => { + const timeoutId = setTimeout(() => { + cancelPendingWait = undefined + resolve() + }, ms) + cancelPendingWait = () => { + clearTimeout(timeoutId) + cancelPendingWait = undefined + resolve() + } + }) +} + export function stopAggregatingApplications(): void { stopping = true + cancelPendingWait?.() } /** Reset aggregation stopping flag. Used for test isolation. */ @@ -353,7 +372,22 @@ export async function addUIData(items: ITransformedResource[]) { export async function searchLoop() { let pass = 1 let searchAPIMissing = false + let multiClusterHubMissing = false while (!stopping) { + const multiClusterHub = await getMultiClusterHub(true) + if (!multiClusterHub) { + if (!multiClusterHubMissing) { + logger.info('MultiClusterHub not found; waiting before search aggregation') + multiClusterHubMissing = true + } + await waitWhileRunning(5 * 60 * 1000) + continue + } + if (multiClusterHubMissing) { + logger.info('MultiClusterHub found') + multiClusterHubMissing = false + } + // make sure there's an active search api // otherwise there's no point let exists @@ -370,7 +404,7 @@ export async function searchLoop() { logger.error('search API missing') searchAPIMissing = true } - await new Promise((r) => setTimeout(r, 5 * 60 * 1000)) + await waitWhileRunning(5 * 60 * 1000) } } while (!exists) /* istanbul ignore if */ @@ -393,7 +427,7 @@ export async function searchLoop() { // process every APP_SEARCH_INTERVAL seconds /* istanbul ignore if */ if (process.env.NODE_ENV !== 'test') { - await new Promise((r) => setTimeout(r, pass <= 3 ? 15000 : Number(process.env.APP_SEARCH_INTERVAL) || 60000)) + await waitWhileRunning(pass <= 3 ? 15000 : Number(process.env.APP_SEARCH_INTERVAL) || 60000) } else { stopping = true } diff --git a/backend/test/lib/search.test.ts b/backend/test/lib/search.test.ts new file mode 100644 index 00000000000..f3395f1de0c --- /dev/null +++ b/backend/test/lib/search.test.ts @@ -0,0 +1,151 @@ +/* Copyright Contributors to the Open Cluster Management project */ +import EventEmitter from 'node:events' +import type { IncomingMessage } from 'node:http' +import { Readable } from 'node:stream' +import { jest, describe, it, expect, beforeEach, afterEach } from '@jest/globals' +import { request } from 'node:https' +import type { IQuery } from '../../src/routes/aggregators/applications' +import { getSearchResults, pingSearchAPI } from '../../src/lib/search' + +jest.mock('node:https', () => ({ + request: jest.fn(), +})) + +jest.mock('../../src/lib/multi-cluster-hub', () => ({ + getMultiClusterHub: jest.fn<() => Promise<{ metadata: { namespace: string } }>>().mockResolvedValue({ + metadata: { namespace: 'ocm' }, + }), +})) + +jest.mock('../../src/lib/serviceAccountToken', () => ({ + getServiceAccountToken: jest.fn(() => 'token'), + getNamespace: jest.fn(() => 'ocm'), +})) + +jest.mock('../../src/lib/agent', () => ({ + getServiceAgent: jest.fn(() => ({})), +})) + +jest.mock('../../src/lib/logger', () => ({ + logger: { + error: jest.fn(), + info: jest.fn(), + }, +})) + +const mockRequest = request as jest.MockedFunction + +type MockClientRequest = EventEmitter & { + write: jest.Mock + end: jest.Mock + destroy: jest.Mock +} + +function createMockResponse(body: string): IncomingMessage { + return Readable.from([body]) as unknown as IncomingMessage +} + +function createMockClientRequest(onEnd?: () => void, error?: Error): MockClientRequest { + const clientRequest = new EventEmitter() as MockClientRequest + clientRequest.write = jest.fn() + clientRequest.end = jest.fn(() => { + if (error) { + process.nextTick(() => clientRequest.emit('error', error)) + return + } + if (onEnd) { + process.nextTick(onEnd) + } + }) + clientRequest.destroy = jest.fn(() => { + clientRequest.emit('close') + }) + return clientRequest +} + +const emptySearchQuery: IQuery = { + operationName: 'searchResult', + variables: { + input: [], + }, + query: 'query searchResult($input: [SearchInput]) { searchResult: search(input: $input) { items } }', +} + +describe('search lib', () => { + beforeEach(() => { + jest.clearAllMocks() + }) + + afterEach(() => { + jest.useRealTimers() + jest.restoreAllMocks() + }) + + describe('pingSearchAPI', () => { + it('clears the timeout when the request fails', async () => { + const clearTimeoutSpy = jest.spyOn(global, 'clearTimeout') + const requestError = new Error('getaddrinfo ENOTFOUND search-search-api.undefined.svc.cluster.local') + mockRequest.mockImplementation( + () => createMockClientRequest(undefined, requestError) as unknown as ReturnType + ) + + await expect(pingSearchAPI()).rejects.toThrow('getaddrinfo ENOTFOUND') + expect(clearTimeoutSpy).toHaveBeenCalled() + }) + + it('destroys the request when the ping times out', async () => { + jest.useFakeTimers() + mockRequest.mockImplementation(() => createMockClientRequest() as unknown as ReturnType) + + const promise = pingSearchAPI() + const expectation = expect(promise).rejects.toThrow('request timeout') + await jest.advanceTimersByTimeAsync(4 * 60 * 1000) + await expectation + + const clientRequest = mockRequest.mock.results[0].value as MockClientRequest + expect(clientRequest.destroy).toHaveBeenCalled() + }) + }) + + describe('getSearchResults', () => { + it('clears the timeout when the request fails', async () => { + const clearTimeoutSpy = jest.spyOn(global, 'clearTimeout') + const requestError = new Error('getaddrinfo ENOTFOUND search-search-api.undefined.svc.cluster.local') + mockRequest.mockImplementation( + () => createMockClientRequest(undefined, requestError) as unknown as ReturnType + ) + + await expect(getSearchResults(emptySearchQuery)).rejects.toThrow('getaddrinfo ENOTFOUND') + expect(clearTimeoutSpy).toHaveBeenCalled() + }) + + it('destroys the request when the search request times out', async () => { + jest.useFakeTimers() + mockRequest.mockImplementation(() => createMockClientRequest() as unknown as ReturnType) + + const promise = getSearchResults(emptySearchQuery) + const expectation = expect(promise).rejects.toThrow('request timeout') + await jest.advanceTimersByTimeAsync(2 * 60 * 1000) + await expectation + + const clientRequest = mockRequest.mock.results[0].value as MockClientRequest + expect(clientRequest.destroy).toHaveBeenCalled() + }) + + it('rejects with malformed response data without request timeout', async () => { + jest.useFakeTimers() + mockRequest.mockImplementation((_options, callback) => { + const clientRequest = createMockClientRequest() + if (typeof callback === 'function') { + callback(createMockResponse('not-json')) + } + return clientRequest as unknown as ReturnType + }) + + const promise = getSearchResults(emptySearchQuery) + const expectation = expect(promise).rejects.toThrow('not-json') + await jest.advanceTimersByTimeAsync(2 * 60 * 1000) + await expectation + }) + }) +}) diff --git a/backend/test/routes/aggregator.test.ts b/backend/test/routes/aggregator.test.ts index cc71224147f..516867f565b 100644 --- a/backend/test/routes/aggregator.test.ts +++ b/backend/test/routes/aggregator.test.ts @@ -1,5 +1,6 @@ /* Copyright Contributors to the Open Cluster Management project */ import { parseResponseJsonBody } from '../../src/lib/body-parser' +import { logger } from '../../src/lib/logger' import { aggregateLocalApplications, aggregateRemoteApplications, @@ -14,6 +15,7 @@ import { request } from '../mock-request' import nock from 'nock' import { discoverSystemAppNamespacePrefixes, resetSystemAppNamespacePrefixes } from '../../src/routes/aggregators/utils' import { resetMultiClusterHubCache } from '../../src/lib/multi-cluster-hub' +import * as multiClusterHub from '../../src/lib/multi-cluster-hub' import { resetMultiClusterEngineCache } from '../../src/lib/multi-cluster-engine' import { ServerSideEvents } from '../../src/lib/server-side-events' import { polledAggregation } from '../../src/routes/aggregator' @@ -197,6 +199,77 @@ describe(`aggregator Route`, function () { expect(res.statusCode).toEqual(200) expect(await parseResponseJsonBody(res)).toEqual(uidata) }) + + describe('startAggregatingApplications', () => { + const pingSearchApiBody = + '{"operationName":"searchResult","variables":{"input":[{"filters":[{"property":"kind","values":["Pod"]},{"property":"name","values":["search-api*"]}],"limit":1}]},"query":"query searchResult($input: [SearchInput]) {\\n searchResult: search(input: $input) {\\n items\\n }\\n}"}' + + afterEach(() => { + jest.useRealTimers() + }) + + it('should not ping search API when MultiClusterHub is missing', async function () { + jest.useFakeTimers() + resetMultiClusterHubCache() + jest.spyOn(multiClusterHub, 'getMultiClusterHub').mockResolvedValue(undefined) + const searchScope = nock('https://search-search-api.undefined.svc.cluster.local:4010') + .post('/searchapi/graphql') + .reply(200, { data: { searchResult: [] } }) + const infoSpy = jest.spyOn(logger, 'info').mockImplementation(() => undefined) + + const promise = searchLoop() + await Promise.resolve() + expect(infoSpy).toHaveBeenCalledWith('MultiClusterHub not found; waiting before search aggregation') + expect(searchScope.isDone()).toBe(false) + stopAggregatingApplications() + await promise + infoSpy.mockRestore() + jest.restoreAllMocks() + }) + + it('should start searchLoop when MultiClusterHub is present', async function () { + resetMultiClusterHubCache() + nock(process.env.CLUSTER_API_URL) + .persist() + .get('/apis/operator.open-cluster-management.io/v1/multiclusterhubs') + .reply(200, { + items: [ + { + metadata: { + namespace: 'ocm', + }, + status: { + currentVersion: '2.5.1', + }, + }, + ], + }) + + const pingScope = nock('https://search-search-api.undefined.svc.cluster.local:4010') + .post('/searchapi/graphql', pingSearchApiBody) + .reply(200, { + data: { + searchResult: [{ items: [{ status: 'Running' }] }], + }, + }) + nock('https://search-search-api.undefined.svc.cluster.local:4010') + .persist() + .post('/searchapi/graphql') + .reply(200, { + data: { + searchResult: [ + { items: [], related: [] }, + { items: [], related: [] }, + { items: [], related: [] }, + ], + }, + }) + + await searchLoop() + + expect(pingScope.isDone()).toBe(true) + }) + }) }) const systemPrefixes = ['openshift', 'hive', 'open-cluster-management', 'multicluster-engine'] @@ -554,6 +627,20 @@ const responseFiltered = { } /// to get exact nock request body, put bp at line 303 in /backend/node_modules/nock/lib/intercepted_request_router.js function setupNocks(prefixes?: boolean) { + if (!prefixes) { + nock(process.env.CLUSTER_API_URL) + .persist() + .get('/apis/operator.open-cluster-management.io/v1/multiclusterhubs') + .reply(200, { + items: [ + { + metadata: { namespace: 'ocm' }, + status: { currentVersion: '2.5.1' }, + }, + ], + }) + } + // // PING SEARCHAPI nock('https://search-search-api.undefined.svc.cluster.local:4010')