Fix agent config indicator when applied through fleet integration

packages/elastic-apm-synthtrace/src/index.ts

@@ -9,6 +9,7 @@
 export { timerange } from './lib/timerange';
 export { apm } from './lib/apm';
 export { stackMonitoring } from './lib/stack_monitoring';
+export { agentConfig } from './lib/agent_config';
 export { cleanWriteTargets } from './lib/utils/clean_write_targets';
 export { createLogger, LogLevel } from './lib/utils/create_logger';
 

packages/elastic-apm-synthtrace/src/lib/agent_config/agent_config.ts

@@ -0,0 +1,29 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0 and the Server Side Public License, v 1; you may not use this file except
+ * in compliance with, at your election, the Elastic License 2.0 or the Server
+ * Side Public License, v 1.
+ */
+
+import { Entity } from '../entity';
+import { AgentConfigFields } from './agent_config_fields';
+import { AgentConfigMetrics } from './agent_config_metrics';
+
+export class AgentConfig extends Entity<AgentConfigFields> {
+  metrics() {
+    return new AgentConfigMetrics({
+      ...this.fields,
+      'processor.event': 'metric',
+      'processor.name': 'metric',
+      'metricset.name': 'agent_config',
+      agent_config_applied: 1,
+    });
+  }
+}
+
+export function agentConfig(etag: string) {
+  return new AgentConfig({
+    'labels.etag': etag,
+  });
+}

packages/elastic-apm-synthtrace/src/lib/agent_config/agent_config_fields.ts

@@ -0,0 +1,30 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0 and the Server Side Public License, v 1; you may not use this file except
+ * in compliance with, at your election, the Elastic License 2.0 or the Server
+ * Side Public License, v 1.
+ */
+
+import { Fields } from '../entity';
+
+export interface Observer {
+  hostname: string;
+  id: string;
+  ephemeral_id: string;
+  type: string;
+  version: string;
+}
+
+export type AgentConfigFields = Fields &
+  Partial<{
+    'processor.event': string;
+    'processor.name': string;
+    'labels.etag': string;
+    'metricset.name': string;
+    observer: Observer;
+    agent_config_applied: number;
+    'ecs.version': string;
+    'event.agent_id_status': string;
+    'event.ingested': string;
+  }>;

packages/elastic-apm-synthtrace/src/lib/agent_config/agent_config_metrics.ts

@@ -0,0 +1,18 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0 and the Server Side Public License, v 1; you may not use this file except
+ * in compliance with, at your election, the Elastic License 2.0 or the Server
+ * Side Public License, v 1.
+ */
+
+import { Serializable } from '../serializable';
+import { AgentConfigFields } from './agent_config_fields';
+
+export class AgentConfigMetrics extends Serializable<AgentConfigFields> {
+  timestamp(timestamp: number): this {
+    super.timestamp(timestamp);
+    this.fields['event.ingested'] = new Date(timestamp).toISOString();
+    return this;
+  }
+}

packages/elastic-apm-synthtrace/src/lib/agent_config/index.ts

@@ -0,0 +1,9 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0 and the Server Side Public License, v 1; you may not use this file except
+ * in compliance with, at your election, the Elastic License 2.0 or the Server
+ * Side Public License, v 1.
+ */
+
+export { agentConfig } from './agent_config';

packages/elastic-apm-synthtrace/src/lib/stream_processor.ts

@@ -211,7 +211,9 @@ export class StreamProcessor<TFields extends Fields = ApmFields> {
     const eventType = d.processor.event as keyof ApmElasticsearchOutputWriteTargets;
     let dataStream = writeTargets[eventType];
     if (eventType === 'metric') {
-      if (!d.service?.name) {
+      if (d.metricset?.name === 'agent_config') {
+        dataStream = 'metrics-apm.internal-default';
+      } else if (!d.service?.name) {
         dataStream = 'metrics-apm.app-default';
       } else {
         if (!d.transaction && !d.span) {

packages/elastic-apm-synthtrace/src/scripts/examples/04_agent_config.ts

@@ -0,0 +1,36 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0 and the Server Side Public License, v 1; you may not use this file except
+ * in compliance with, at your election, the Elastic License 2.0 or the Server
+ * Side Public License, v 1.
+ */
+
+import { agentConfig, timerange } from '../..';
+import { Scenario } from '../scenario';
+import { getLogger } from '../utils/get_common_services';
+import { RunOptions } from '../utils/parse_run_cli_flags';
+import { AgentConfigFields } from '../../lib/agent_config/agent_config_fields';
+
+const scenario: Scenario<AgentConfigFields> = async (runOptions: RunOptions) => {
+  const logger = getLogger(runOptions);
+
+  return {
+    generate: ({ from, to }) => {
+      const agentConfigMetrics = agentConfig('test-etag').metrics();
+
+      const range = timerange(from, to);
+      return range
+        .interval('30s')
+        .rate(1)
+        .generator((timestamp) => {
+          const events = logger.perf('generating_agent_config_events', () => {
+            return agentConfigMetrics.timestamp(timestamp);
+          });
+          return events;
+        });
+    },
+  };
+};
+
+export default scenario;

x-pack/plugins/apm/common/agent_configuration/configuration_types.d.ts

@@ -13,6 +13,7 @@ export type AgentConfigurationIntake = t.TypeOf<
 >;
 
 export type AgentConfiguration = {
+  id: string;
   '@timestamp': number;
   applied_by_agent?: boolean;
   etag?: string;

x-pack/plugins/apm/server/lib/helpers/create_es_client/create_internal_es_client/index.ts

@@ -106,5 +106,14 @@ export async function createInternalESClient({
         params,
       });
     },
+    bulk: (operationName: string, params: estypes.BulkRequest) => {
+      return callEs(operationName, {
+        requestType: 'bulk',
+        cb: (signal) => {
+          return asInternalUser.bulk(params, { signal, meta: true });
+        },
+        params,
+      });
+    },
   };
 }

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

@@ -24,7 +24,7 @@ export function createOrUpdateConfiguration({
 }) {
   const { internalClient, indices } = setup;
 
-  const params: APMIndexDocumentParams<AgentConfiguration> = {
+  const params: APMIndexDocumentParams<Omit<AgentConfiguration, 'id'>> = {
     refresh: true,
     index: indices.apmAgentConfigurationIndex,
     body: {

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

@@ -0,0 +1,55 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+import { termQuery, rangeQuery } from '@kbn/observability-plugin/server';
+import { keyBy, mapValues } from 'lodash';
+import datemath from '@kbn/datemath';
+import { METRICSET_NAME } from '../../../../common/elasticsearch_fieldnames';
+import { Setup } from '../../../lib/helpers/setup_request';
+
+export async function getConfigsAppliedToAgentsThroughFleet({
+  setup,
+}: {
+  setup: Setup;
+}) {
+  const { internalClient, indices } = setup;
+
+  const params = {
+    index: indices.metric,
+    size: 200,
+    body: {
+      query: {
+        bool: {
+          filter: [
+            ...termQuery(METRICSET_NAME, 'agent_config'),
+            ...rangeQuery(
+              datemath.parse('now-15m')!.valueOf(),
+              datemath.parse('now')!.valueOf()
+            ),
+          ],
+        },
+      },
+      aggs: {
+        config_by_etag: {
+          terms: {
+            field: 'labels.etag',
+          },
+        },
+      },
+    },
+  };
+
+  const response = await internalClient.search(
+    'get_config_applied_to_agent_through_fleet',
+    params
+  );
+
+  return mapValues(
+    keyBy(response.aggregations?.config_by_etag.buckets, 'key'),
+    'key'
+  ) as Record<string, string>;
+}

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

@@ -24,5 +24,5 @@ export async function listConfigurations({ setup }: { setup: Setup }) {
 
   return resp.hits.hits
     .map(convertConfigSettingsToString)
-    .map((hit) => hit._source);
+    .map((hit) => ({ ...hit._source, id: hit._id }));
 }

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

@@ -5,6 +5,7 @@
  * 2.0.
  */
 
+import { isEmpty } from 'lodash';
 import { Setup } from '../../../lib/helpers/setup_request';
 import { AgentConfiguration } from '../../../../common/agent_configuration/configuration_types';
 
@@ -34,3 +35,39 @@ export async function markAppliedByAgent({
     params
   );
 }
+
+export async function markAppliedByAgentThroughFleet({
+  configsAppliedToAgentsThroughFleet,
+  configurations,
+  setup,
+}: {
+  configsAppliedToAgentsThroughFleet: Record<string, string>;
+  configurations: AgentConfiguration[];
+  setup: Setup;
+}) {
+  const { internalClient, indices } = setup;
+
+  // Update only the configs that still have applied_by_agent=true
+  // but have been applied by an agent
+  const configsToUpdate = configurations.filter(
+    (config) =>
+      !config.applied_by_agent &&
+      config.etag !== undefined &&
+      configsAppliedToAgentsThroughFleet.hasOwnProperty(config.etag)
+  );
+
+  if (isEmpty(configsToUpdate)) {
+    return;
+  }
+
+  const body = configsToUpdate.flatMap((doc) => [
+    { update: { _id: doc.id } },
+    { doc: { applied_by_agent: true } },
+  ]);
+
+  return internalClient.bulk('mark_config_applied_by_agent_through_fleet', {
+    index: indices.apmAgentConfigurationIndex,
+    body,
+    refresh: 'wait_for',
+  });
+}

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

@@ -18,13 +18,17 @@ import { getEnvironments } from './get_environments';
 import { deleteConfiguration } from './delete_configuration';
 import { createApmServerRoute } from '../../apm_routes/create_apm_server_route';
 import { getAgentNameByService } from './get_agent_name_by_service';
-import { markAppliedByAgent } from './mark_applied_by_agent';
+import {
+  markAppliedByAgent,
+  markAppliedByAgentThroughFleet,
+} from './mark_applied_by_agent';
 import {
   serviceRt,
   agentConfigurationIntakeRt,
 } from '../../../../common/agent_configuration/runtime_types/agent_configuration_intake_rt';
 import { getSearchAggregatedTransactions } from '../../../lib/helpers/transactions';
 import { syncAgentConfigsToApmPackagePolicies } from '../../fleet/sync_agent_configs_to_apm_package_policies';
+import { getConfigsAppliedToAgentsThroughFleet } from './get_config_applied_to_agent_through_fleet';
 
 // get list of configurations
 const agentConfigurationRoute = createApmServerRoute({
@@ -38,8 +42,36 @@ const agentConfigurationRoute = createApmServerRoute({
     >;
   }> => {
     const setup = await setupRequest(resources);
-    const configurations = await listConfigurations({ setup });
-    return { configurations };
+
+    const [configsAppliedToAgentsThroughFleet, configurations] =
+      await Promise.all([
+        getConfigsAppliedToAgentsThroughFleet({ setup }),
+        listConfigurations({ setup }),
+      ]);
+
+    const updatedConfigs = configurations.map(
+      (
+        agentConfig
+      ): import('./../../../../common/agent_configuration/configuration_types').AgentConfiguration => {
+        return {
+          ...agentConfig,
+          applied_by_agent:
+            agentConfig.applied_by_agent ||
+            (agentConfig.etag !== undefined &&
+              configsAppliedToAgentsThroughFleet.hasOwnProperty(
+                agentConfig.etag
+              )),
+        };
+      }
+    );
+
+    markAppliedByAgentThroughFleet({
+      configsAppliedToAgentsThroughFleet,
+      configurations,
+      setup,
+    });
+
+    return { configurations: updatedConfigs };
   },
 });
 

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

@@ -0,0 +1,31 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+import { timerange, agentConfig } from '@elastic/apm-synthtrace';
+import type { ApmSynthtraceEsClient } from '@elastic/apm-synthtrace';
+
+export async function addAgentConfigMetrics({
+  synthtraceEsClient,
+  start,
+  end,
+  etag,
+}: {
+  synthtraceEsClient: ApmSynthtraceEsClient;
+  start: number;
+  end: number;
+  etag?: string;
+}) {
+  const agentConfigMetrics = agentConfig(etag ?? 'test-etag').metrics();
+
+  const agentConfigEvents = [
+    timerange(start, end)
+      .interval('1m')
+      .rate(1)
+      .generator((timestamp) => agentConfigMetrics.timestamp(timestamp)),
+  ];
+
+  await synthtraceEsClient.index(agentConfigEvents);
+}