[APM] Replace ML index queries with searching via mlAnomalySearch API (#69099)
* Closes #69092 by replacing direct queries on ml indices with seaching via the `mlAnomalySearch` client API + job_id filters. Also removes `getMlIndex` since it is no longer relevant. * Use the mlCapabilities API to ensure the required license is active for ml queries Co-authored-by: Elastic Machine <elasticmachine@users.noreply.github.com>
This commit is contained in:
parent
abdc0f17a9
commit
0ed7597822
|
@ -5,7 +5,6 @@
|
|||
*/
|
||||
|
||||
import {
|
||||
getMlIndex,
|
||||
getMlJobId,
|
||||
getMlPrefix,
|
||||
getMlJobServiceName,
|
||||
|
@ -36,16 +35,6 @@ describe('ml_job_constants', () => {
|
|||
);
|
||||
});
|
||||
|
||||
it('getMlIndex', () => {
|
||||
expect(getMlIndex('myServiceName')).toBe(
|
||||
'.ml-anomalies-myservicename-high_mean_response_time'
|
||||
);
|
||||
|
||||
expect(getMlIndex('myServiceName', 'myTransactionType')).toBe(
|
||||
'.ml-anomalies-myservicename-mytransactiontype-high_mean_response_time'
|
||||
);
|
||||
});
|
||||
|
||||
describe('getMlJobServiceName', () => {
|
||||
it('extracts the service name from a job id', () => {
|
||||
expect(
|
||||
|
|
|
@ -26,10 +26,6 @@ export function getMlJobServiceName(jobId: string) {
|
|||
return jobId.split('-').slice(0, -2).join('-');
|
||||
}
|
||||
|
||||
export function getMlIndex(serviceName: string, transactionType?: string) {
|
||||
return `.ml-anomalies-${getMlJobId(serviceName, transactionType)}`;
|
||||
}
|
||||
|
||||
export function encodeForMlApi(value: string) {
|
||||
return value.replace(/\s+/g, '_').toLowerCase();
|
||||
}
|
||||
|
|
|
@ -38,6 +38,11 @@ Array [
|
|||
"query": Object {
|
||||
"bool": Object {
|
||||
"filter": Array [
|
||||
Object {
|
||||
"term": Object {
|
||||
"job_id": "myservicename-mytransactiontype-high_mean_response_time",
|
||||
},
|
||||
},
|
||||
Object {
|
||||
"exists": Object {
|
||||
"field": "bucket_span",
|
||||
|
@ -57,7 +62,6 @@ Array [
|
|||
},
|
||||
"size": 0,
|
||||
},
|
||||
"index": ".ml-anomalies-myservicename-mytransactiontype-high_mean_response_time",
|
||||
},
|
||||
],
|
||||
]
|
||||
|
|
|
@ -19,7 +19,11 @@ describe('anomalyAggsFetcher', () => {
|
|||
intervalString: 'myInterval',
|
||||
mlBucketSize: 10,
|
||||
setup: {
|
||||
client: { search: clientSpy },
|
||||
ml: {
|
||||
mlSystem: {
|
||||
mlAnomalySearch: clientSpy,
|
||||
},
|
||||
} as any,
|
||||
start: 100000,
|
||||
end: 200000,
|
||||
} as any,
|
||||
|
@ -42,7 +46,13 @@ describe('anomalyAggsFetcher', () => {
|
|||
|
||||
return expect(
|
||||
anomalySeriesFetcher({
|
||||
setup: { client: { search: failedRequestSpy } },
|
||||
setup: {
|
||||
ml: {
|
||||
mlSystem: {
|
||||
mlAnomalySearch: failedRequestSpy,
|
||||
},
|
||||
} as any,
|
||||
},
|
||||
} as any)
|
||||
).resolves.toEqual(undefined);
|
||||
});
|
||||
|
@ -53,7 +63,13 @@ describe('anomalyAggsFetcher', () => {
|
|||
|
||||
return expect(
|
||||
anomalySeriesFetcher({
|
||||
setup: { client: { search: failedRequestSpy } },
|
||||
setup: {
|
||||
ml: {
|
||||
mlSystem: {
|
||||
mlAnomalySearch: failedRequestSpy,
|
||||
},
|
||||
} as any,
|
||||
},
|
||||
} as any)
|
||||
).rejects.toThrow(otherError);
|
||||
});
|
||||
|
|
|
@ -4,7 +4,7 @@
|
|||
* you may not use this file except in compliance with the Elastic License.
|
||||
*/
|
||||
|
||||
import { getMlIndex } from '../../../../../common/ml_job_constants';
|
||||
import { getMlJobId } from '../../../../../common/ml_job_constants';
|
||||
import { PromiseReturnType } from '../../../../../../observability/typings/common';
|
||||
import { Setup, SetupTimeRange } from '../../../helpers/setup_request';
|
||||
|
||||
|
@ -26,19 +26,23 @@ export async function anomalySeriesFetcher({
|
|||
mlBucketSize: number;
|
||||
setup: Setup & SetupTimeRange;
|
||||
}) {
|
||||
const { client, start, end } = setup;
|
||||
const { ml, start, end } = setup;
|
||||
if (!ml) {
|
||||
return;
|
||||
}
|
||||
|
||||
// move the start back with one bucket size, to ensure to get anomaly data in the beginning
|
||||
// this is required because ML has a minimum bucket size (default is 900s) so if our buckets are smaller, we might have several null buckets in the beginning
|
||||
const newStart = start - mlBucketSize * 1000;
|
||||
const jobId = getMlJobId(serviceName, transactionType);
|
||||
|
||||
const params = {
|
||||
index: getMlIndex(serviceName, transactionType),
|
||||
body: {
|
||||
size: 0,
|
||||
query: {
|
||||
bool: {
|
||||
filter: [
|
||||
{ term: { job_id: jobId } },
|
||||
{ exists: { field: 'bucket_span' } },
|
||||
{
|
||||
range: {
|
||||
|
@ -74,7 +78,7 @@ export async function anomalySeriesFetcher({
|
|||
};
|
||||
|
||||
try {
|
||||
const response = await client.search(params);
|
||||
const response = await ml.mlSystem.mlAnomalySearch(params);
|
||||
return response;
|
||||
} catch (err) {
|
||||
const isHttpError = 'statusCode' in err;
|
||||
|
|
|
@ -4,7 +4,7 @@
|
|||
* you may not use this file except in compliance with the Elastic License.
|
||||
*/
|
||||
|
||||
import { getMlIndex } from '../../../../../common/ml_job_constants';
|
||||
import { getMlJobId } from '../../../../../common/ml_job_constants';
|
||||
import { Setup, SetupTimeRange } from '../../../helpers/setup_request';
|
||||
|
||||
interface IOptions {
|
||||
|
@ -22,15 +22,20 @@ export async function getMlBucketSize({
|
|||
transactionType,
|
||||
setup,
|
||||
}: IOptions): Promise<number> {
|
||||
const { client, start, end } = setup;
|
||||
const { ml, start, end } = setup;
|
||||
if (!ml) {
|
||||
return 0;
|
||||
}
|
||||
const jobId = getMlJobId(serviceName, transactionType);
|
||||
|
||||
const params = {
|
||||
index: getMlIndex(serviceName, transactionType),
|
||||
body: {
|
||||
_source: 'bucket_span',
|
||||
size: 1,
|
||||
query: {
|
||||
bool: {
|
||||
filter: [
|
||||
{ term: { job_id: jobId } },
|
||||
{ exists: { field: 'bucket_span' } },
|
||||
{
|
||||
range: {
|
||||
|
@ -48,7 +53,7 @@ export async function getMlBucketSize({
|
|||
};
|
||||
|
||||
try {
|
||||
const resp = await client.search<ESResponse, typeof params>(params);
|
||||
const resp = await ml.mlSystem.mlAnomalySearch<ESResponse>(params);
|
||||
return resp.hits.hits[0]?._source.bucket_span || 0;
|
||||
} catch (err) {
|
||||
const isHttpError = 'statusCode' in err;
|
||||
|
|
|
@ -26,8 +26,8 @@ describe('getAnomalySeries', () => {
|
|||
setup: {
|
||||
start: 0,
|
||||
end: 500000,
|
||||
client: { search: clientSpy } as any,
|
||||
internalClient: { search: clientSpy } as any,
|
||||
client: { search: () => {} } as any,
|
||||
internalClient: { search: () => {} } as any,
|
||||
config: new Proxy(
|
||||
{},
|
||||
{
|
||||
|
@ -46,6 +46,12 @@ describe('getAnomalySeries', () => {
|
|||
apmCustomLinkIndex: 'myIndex',
|
||||
},
|
||||
dynamicIndexPattern: null as any,
|
||||
ml: {
|
||||
mlSystem: {
|
||||
mlAnomalySearch: clientSpy,
|
||||
mlCapabilities: async () => ({ isPlatinumOrTrialLicense: true }),
|
||||
},
|
||||
} as any,
|
||||
},
|
||||
});
|
||||
});
|
||||
|
|
|
@ -42,6 +42,17 @@ export async function getAnomalySeries({
|
|||
return;
|
||||
}
|
||||
|
||||
// don't fetch anomalies if the ML plugin is not setup
|
||||
if (!setup.ml) {
|
||||
return;
|
||||
}
|
||||
|
||||
// don't fetch anomalies if required license is not satisfied
|
||||
const mlCapabilities = await setup.ml.mlSystem.mlCapabilities();
|
||||
if (!mlCapabilities.isPlatinumOrTrialLicense) {
|
||||
return;
|
||||
}
|
||||
|
||||
const mlBucketSize = await getMlBucketSize({
|
||||
serviceName,
|
||||
transactionType,
|
||||
|
|
Loading…
Reference in a new issue