[APM] Remove usage of internal client when fetching agent config etag…

…s metrics (elastic#173001)

Closes: elastic#170031
Replaces: elastic/elasticsearch#101467 


**Problem**
We need to know if an agent config has been applied at the edge (by APM
agents). This is determined by comparing the etag (hash) of the config,
with the etag applied at the edges.

Previously the agent config itself contained this information
(`config.applied_by_agent`) but when running with fleet this will
instead be captured in `agent_config` metric documents.

Currently the internal kibana user retrieves the `agent_config` metric
documents from the APM metric index (`metrics-apm-*` by default). This
index is configurable by the end-user so can be changed to something the
internal user doesn't have access to. This is a problem.

**Solution**

This PR replaces the calls made by the internal client with calls made
by the authenticated end user (via `APMEventClient`). This approach
works for requests made from the browser/UI but doesn't work for
background tasks made by fleet. To work around this we only
conditionally query the metric index if the `APMEventClient` is
available.
If `APMEventClient` is not available `applied_by_agent` will be
`undefined`

x-pack/plugins/apm/server/routes/fleet/merge_package_policy_with_apm.ts

@@ -28,7 +28,7 @@ export async function decoratePackagePolicyWithAgentConfigAndSourceMap({
   apmIndices: APMIndices;
 }) {
   const [agentConfigurations, { artifacts }] = await Promise.all([
-    listConfigurations(internalESClient, apmIndices),
+    listConfigurations({ internalESClient, apmIndices }),
     listSourceMapArtifacts({ fleetPluginStart }),
   ]);
 

x-pack/plugins/apm/server/routes/fleet/sync_agent_configs_to_apm_package_policies.ts

@@ -39,7 +39,7 @@ export async function syncAgentConfigsToApmPackagePolicies({
   const [savedObjectsClient, agentConfigurations, packagePolicies] =
     await Promise.all([
       getInternalSavedObjectsClient(coreStart),
-      listConfigurations(internalESClient, apmIndices),
+      listConfigurations({ internalESClient, apmIndices }),
       getApmPackagePolicies({ coreStart, fleetPluginStart }),
     ]);
 

x-pack/plugins/apm/server/routes/settings/agent_configuration/find_exact_configuration.ts

@@ -6,7 +6,7 @@
  */
 
 import type { SearchHit } from '@kbn/es-types';
-import { APMIndices } from '@kbn/apm-data-access-plugin/server';
+import { APMEventClient } from '../../../lib/helpers/create_es_client/create_apm_event_client';
 import { AgentConfiguration } from '../../../../common/agent_configuration/configuration_types';
 import {
   SERVICE_ENVIRONMENT,
@@ -15,16 +15,16 @@ import {
 import { APMInternalESClient } from '../../../lib/helpers/create_es_client/create_internal_es_client';
 import { APM_AGENT_CONFIGURATION_INDEX } from '../apm_indices/apm_system_index_constants';
 import { convertConfigSettingsToString } from './convert_settings_to_string';
-import { getConfigsAppliedToAgentsThroughFleet } from './get_config_applied_to_agent_through_fleet';
+import { getAgentConfigEtagMetrics } from './get_agent_config_etag_metrics';
 
 export async function findExactConfiguration({
   service,
   internalESClient,
-  apmIndices,
+  apmEventClient,
 }: {
   service: AgentConfiguration['service'];
   internalESClient: APMInternalESClient;
-  apmIndices: APMIndices;
+  apmEventClient: APMEventClient;
 }) {
   const serviceNameFilter = service.name
     ? { term: { [SERVICE_NAME]: service.name } }
@@ -43,13 +43,10 @@ export async function findExactConfiguration({
     },
   };
 
-  const [agentConfig, configsAppliedToAgentsThroughFleet] = await Promise.all([
-    internalESClient.search<AgentConfiguration, typeof params>(
-      'find_exact_agent_configuration',
-      params
-    ),
-    getConfigsAppliedToAgentsThroughFleet(internalESClient, apmIndices),
-  ]);
+  const agentConfig = await internalESClient.search<
+    AgentConfiguration,
+    typeof params
+  >('find_exact_agent_configuration', params);
 
   const hit = agentConfig.hits.hits[0] as
     | SearchHit<AgentConfiguration>
@@ -59,11 +56,33 @@ export async function findExactConfiguration({
     return;
   }
 
+  const appliedByAgent = await getIsAppliedByAgent({
+    apmEventClient,
+    agentConfiguration: hit._source,
+  });
+
   return {
     id: hit._id,
     ...convertConfigSettingsToString(hit)._source,
-    applied_by_agent:
-      hit._source.applied_by_agent ||
-      configsAppliedToAgentsThroughFleet.hasOwnProperty(hit._source.etag),
+    applied_by_agent: appliedByAgent,
   };
 }
+
+async function getIsAppliedByAgent({
+  apmEventClient,
+  agentConfiguration,
+}: {
+  apmEventClient: APMEventClient;
+  agentConfiguration: AgentConfiguration;
+}) {
+  if (agentConfiguration.applied_by_agent) {
+    return true;
+  }
+
+  const appliedEtags = await getAgentConfigEtagMetrics(
+    apmEventClient,
+    agentConfiguration.etag
+  );
+
+  return appliedEtags.includes(agentConfiguration.etag);
+}

x-pack/plugins/apm/server/routes/settings/agent_configuration/get_config_applied_to_agent_through_fleet.ts → x-pack/plugins/apm/server/routes/settings/agent_configuration/get_agent_config_etag_metrics.ts

@@ -7,23 +7,26 @@
 
 import { termQuery, rangeQuery } from '@kbn/observability-plugin/server';
 import datemath from '@kbn/datemath';
-import { APMIndices } from '@kbn/apm-data-access-plugin/server';
+import { ProcessorEvent } from '@kbn/observability-plugin/common';
+import { APMEventClient } from '../../../lib/helpers/create_es_client/create_apm_event_client';
 import { METRICSET_NAME } from '../../../../common/es_fields/apm';
-import { APMInternalESClient } from '../../../lib/helpers/create_es_client/create_internal_es_client';
 
-export async function getConfigsAppliedToAgentsThroughFleet(
-  internalESClient: APMInternalESClient,
-  apmIndices: APMIndices
+export async function getAgentConfigEtagMetrics(
+  apmEventClient: APMEventClient,
+  etag?: string
 ) {
   const params = {
-    index: apmIndices.metric,
-    track_total_hits: 0,
-    size: 0,
+    apm: {
+      events: [ProcessorEvent.metric],
+    },
     body: {
+      track_total_hits: 0,
+      size: 0,
       query: {
         bool: {
           filter: [
             ...termQuery(METRICSET_NAME, 'agent_config'),
+            ...termQuery('labels.etag', etag),
             ...rangeQuery(
               datemath.parse('now-15m')!.valueOf(),
               datemath.parse('now')!.valueOf()
@@ -42,18 +45,14 @@ export async function getConfigsAppliedToAgentsThroughFleet(
     },
   };
 
-  const response = await internalESClient.search(
+  const response = await apmEventClient.search(
     'get_config_applied_to_agent_through_fleet',
     params
   );
 
   return (
-    response.aggregations?.config_by_etag.buckets.reduce(
-      (configsAppliedToAgentsThroughFleet, bucket) => {
-        configsAppliedToAgentsThroughFleet[bucket.key as string] = true;
-        return configsAppliedToAgentsThroughFleet;
-      },
-      {} as Record<string, true>
-    ) ?? {}
+    response.aggregations?.config_by_etag.buckets.map(
+      ({ key }) => key as string
+    ) ?? []
   );
 }

x-pack/plugins/apm/server/routes/settings/agent_configuration/list_configurations.ts

@@ -6,27 +6,33 @@
  */
 
 import { APMIndices } from '@kbn/apm-data-access-plugin/server';
+import { APMEventClient } from '../../../lib/helpers/create_es_client/create_apm_event_client';
 import { AgentConfiguration } from '../../../../common/agent_configuration/configuration_types';
 import { convertConfigSettingsToString } from './convert_settings_to_string';
-import { getConfigsAppliedToAgentsThroughFleet } from './get_config_applied_to_agent_through_fleet';
+import { getAgentConfigEtagMetrics } from './get_agent_config_etag_metrics';
 import { APMInternalESClient } from '../../../lib/helpers/create_es_client/create_internal_es_client';
 import { APM_AGENT_CONFIGURATION_INDEX } from '../apm_indices/apm_system_index_constants';
 
-export async function listConfigurations(
-  internalESClient: APMInternalESClient,
-  apmIndices: APMIndices
-) {
+export async function listConfigurations({
+  internalESClient,
+  apmEventClient,
+  apmIndices,
+}: {
+  internalESClient: APMInternalESClient;
+  apmEventClient?: APMEventClient;
+  apmIndices: APMIndices;
+}) {
   const params = {
     index: APM_AGENT_CONFIGURATION_INDEX,
     size: 200,
   };
 
-  const [agentConfigs, configsAppliedToAgentsThroughFleet] = await Promise.all([
+  const [agentConfigs, appliedEtags = []] = await Promise.all([
     internalESClient.search<AgentConfiguration>(
       'list_agent_configuration',
       params
     ),
-    getConfigsAppliedToAgentsThroughFleet(internalESClient, apmIndices),
+    apmEventClient ? getAgentConfigEtagMetrics(apmEventClient) : undefined,
   ]);
 
   return agentConfigs.hits.hits
@@ -36,7 +42,7 @@ export async function listConfigurations(
         ...hit._source,
         applied_by_agent:
           hit._source.applied_by_agent ||
-          configsAppliedToAgentsThroughFleet.hasOwnProperty(hit._source.etag),
+          appliedEtags.includes(hit._source.etag),
       };
     });
 }

x-pack/plugins/apm/server/routes/settings/agent_configuration/queries.test.ts

@@ -55,7 +55,10 @@ describe('agent configuration queries', () => {
     it('fetches configurations', async () => {
       mock = await inspectSearchParams(
         ({ mockInternalESClient, mockIndices }) =>
-          listConfigurations(mockInternalESClient, mockIndices)
+          listConfigurations({
+            internalESClient: mockInternalESClient,
+            apmIndices: mockIndices,
+          })
       );
 
       expect(mock.params).toMatchSnapshot();
@@ -94,11 +97,11 @@ describe('agent configuration queries', () => {
   describe('findExactConfiguration', () => {
     it('find configuration by service.name', async () => {
       mock = await inspectSearchParams(
-        ({ mockInternalESClient, mockIndices }) =>
+        ({ mockInternalESClient, mockApmEventClient }) =>
           findExactConfiguration({
             service: { name: 'foo' },
             internalESClient: mockInternalESClient,
-            apmIndices: mockIndices,
+            apmEventClient: mockApmEventClient,
           })
       );
 
@@ -107,11 +110,11 @@ describe('agent configuration queries', () => {
 
     it('find configuration by service.environment', async () => {
       mock = await inspectSearchParams(
-        ({ mockInternalESClient, mockIndices }) =>
+        ({ mockInternalESClient, mockApmEventClient }) =>
           findExactConfiguration({
             service: { environment: 'bar' },
             internalESClient: mockInternalESClient,
-            apmIndices: mockIndices,
+            apmEventClient: mockApmEventClient,
           })
       );
 
@@ -120,11 +123,11 @@ describe('agent configuration queries', () => {
 
     it('find configuration by service.name and service.environment', async () => {
       mock = await inspectSearchParams(
-        ({ mockInternalESClient, mockIndices }) =>
+        ({ mockInternalESClient, mockApmEventClient }) =>
           findExactConfiguration({
             service: { name: 'foo', environment: 'bar' },
             internalESClient: mockInternalESClient,
-            apmIndices: mockIndices,
+            apmEventClient: mockApmEventClient,
           })
       );
 

x-pack/plugins/apm/server/routes/settings/agent_configuration/route.ts

@@ -50,13 +50,15 @@ const agentConfigurationRoute = createApmServerRoute({
     throwNotFoundIfAgentConfigNotAvailable(resources.featureFlags);
 
     const apmIndices = await resources.getApmIndices();
-    const internalESClient = await createInternalESClientWithResources(
-      resources
-    );
-    const configurations = await listConfigurations(
+    const [internalESClient, apmEventClient] = await Promise.all([
+      createInternalESClientWithResources(resources),
+      getApmEventClient(resources),
+    ]);
+    const configurations = await listConfigurations({
       internalESClient,
-      apmIndices
-    );
+      apmIndices,
+      apmEventClient,
+    });
 
     return { configurations };
   },
@@ -75,14 +77,14 @@ const getSingleAgentConfigurationRoute = createApmServerRoute({
     const { params, logger } = resources;
     const { name, environment } = params.query;
     const service = { name, environment };
-    const apmIndices = await resources.getApmIndices();
-    const internalESClient = await createInternalESClientWithResources(
-      resources
-    );
+    const [internalESClient, apmEventClient] = await Promise.all([
+      createInternalESClientWithResources(resources),
+      getApmEventClient(resources),
+    ]);
     const exactConfig = await findExactConfiguration({
       service,
       internalESClient,
-      apmIndices,
+      apmEventClient,
     });
 
     if (!exactConfig) {
@@ -114,13 +116,14 @@ const deleteAgentConfigurationRoute = createApmServerRoute({
     const { params, logger, core, telemetryUsageCounter } = resources;
     const { service } = params.body;
     const apmIndices = await resources.getApmIndices();
-    const internalESClient = await createInternalESClientWithResources(
-      resources
-    );
+    const [internalESClient, apmEventClient] = await Promise.all([
+      createInternalESClientWithResources(resources),
+      getApmEventClient(resources),
+    ]);
     const exactConfig = await findExactConfiguration({
       service,
       internalESClient,
-      apmIndices,
+      apmEventClient,
     });
     if (!exactConfig) {
       logger.info(
@@ -171,16 +174,17 @@ const createOrUpdateAgentConfigurationRoute = createApmServerRoute({
     const { params, logger, core, telemetryUsageCounter } = resources;
     const { body, query } = params;
     const apmIndices = await resources.getApmIndices();
-    const internalESClient = await createInternalESClientWithResources(
-      resources
-    );
+    const [internalESClient, apmEventClient] = await Promise.all([
+      createInternalESClientWithResources(resources),
+      getApmEventClient(resources),
+    ]);
 
     // if the config already exists, it is fetched and updated
     // this is to avoid creating two configs with identical service params
     const exactConfig = await findExactConfiguration({
       service: body.service,
       internalESClient,
-      apmIndices,
+      apmEventClient,
     });
 
     // if the config exists ?overwrite=true is required

x-pack/test/apm_api_integration/tests/settings/agent_configuration/add_agent_config_metrics.ts

@@ -4,31 +4,19 @@
  * 2.0; you may not use this file except in compliance with the Elastic License
  * 2.0.
  */
-import { timerange, observer } from '@kbn/apm-synthtrace-client';
+import { observer } from '@kbn/apm-synthtrace-client';
 import type { ApmSynthtraceEsClient } from '@kbn/apm-synthtrace';
+import { Readable } from 'stream';
 
-export async function addAgentConfigMetrics({
+export function addAgentConfigEtagMetric({
   synthtraceEsClient,
-  start,
-  end,
+  timestamp,
   etag,
 }: {
   synthtraceEsClient: ApmSynthtraceEsClient;
-  start: number;
-  end: number;
-  etag?: string;
+  timestamp: number;
+  etag: string;
 }) {
-  const agentConfigEvents = [
-    timerange(start, end)
-      .interval('1m')
-      .rate(1)
-      .generator((timestamp) =>
-        observer()
-          .agentConfig()
-          .etag(etag ?? 'test-etag')
-          .timestamp(timestamp)
-      ),
-  ];
-
-  await synthtraceEsClient.index(agentConfigEvents);
+  const agentConfigMetric = observer().agentConfig().etag(etag).timestamp(timestamp);
+  return synthtraceEsClient.index(Readable.from([agentConfigMetric]));
 }