[Fleet] add index and task for fleet-synced-integrations (#209762)

## Summary

Closes #206237

Create `fleet-synced-integrations` index in Fleet setup, added async
task that populates the index with a doc that includes remote ES output
data and installed integrations data.

ES change to add `kibana_system` privileges:
elastic/elasticsearch#121753

To test locally:
- run elasticsearch from source to apply the privilege changes, so that
`kibana_system` can create the index.
```
yarn es source -E xpack.security.authc.api_key.enabled=true -E xpack.security.authc.token.enabled=true  --source-path=/Users/juliabardi/elasticsearch  -E path.data=/tmp/es-data -E xpack.ml.enabled=false
```
- enable the feature flag in `kibana.dev.yml`:
`xpack.fleet.enableExperimental: ['enableSyncIntegrationsOnRemote']`
- add a remote ES output with sync enabled
- install some integrations
- wait until Fleet setup and the task runs
- verify that the index is created and contains a doc with the expected
data

```
GET fleet-synced-integrations/_search

 "hits": {
    "total": {
      "value": 1,
      "relation": "eq"
    },
    "max_score": 1,
    "hits": [
      {
        "_index": "fleet-synced-integrations",
        "_id": "fleet-synced-integrations",
        "_score": 1,
        "_source": {
          "remote_es_hosts": [
            {
              "hosts": [
                "http://remote1:80"
              ],
              "name": "remote1",
              "sync_integrations": true
            }
          ],
          "integrations": [
            {
              "package_version": "1.64.1",
              "updated_at": "2025-02-05T11:03:02.226Z",
              "package_name": "system"
            }
          ]
        }
      }
    ]
```



### Checklist

- [x] [Unit or functional
tests](https://www.elastic.co/guide/en/kibana/master/development-tests.html)
were updated or added to match the most common scenarios

---------

Co-authored-by: kibanamachine <[email protected]>

Author: juliaElastic

x-pack/platform/plugins/shared/fleet/server/mocks/index.ts

@@ -140,6 +140,7 @@ export const createAppContextStartContractMock = (
       : {}),
     unenrollInactiveAgentsTask: {} as any,
     deleteUnenrolledAgentsTask: {} as any,
+    syncIntegrationsTask: {} as any,
   };
 };
 

x-pack/platform/plugins/shared/fleet/server/plugin.ts

@@ -147,6 +147,7 @@ import { registerUpgradeManagedPackagePoliciesTask } from './services/setup/mana
 import { registerDeployAgentPoliciesTask } from './services/agent_policies/deploy_agent_policies_task';
 import { DeleteUnenrolledAgentsTask } from './tasks/delete_unenrolled_agents_task';
 import { registerBumpAgentPoliciesTask } from './services/agent_policies/bump_agent_policies_task';
+import { SyncIntegrationsTask } from './tasks/sync_integrations_task';
 
 export interface FleetSetupDeps {
   security: SecurityPluginSetup;
@@ -200,6 +201,7 @@ export interface FleetAppContext {
   deleteUnenrolledAgentsTask: DeleteUnenrolledAgentsTask;
   taskManagerStart?: TaskManagerStartContract;
   fetchUsage?: (abortController: AbortController) => Promise<FleetUsage | undefined>;
+  syncIntegrationsTask: SyncIntegrationsTask;
 }
 
 export type FleetSetupContract = void;
@@ -301,6 +303,7 @@ export class FleetPlugin
   private fleetMetricsTask?: FleetMetricsTask;
   private unenrollInactiveAgentsTask?: UnenrollInactiveAgentsTask;
   private deleteUnenrolledAgentsTask?: DeleteUnenrolledAgentsTask;
+  private syncIntegrationsTask?: SyncIntegrationsTask;
 
   private agentService?: AgentService;
   private packageService?: PackageService;
@@ -647,6 +650,11 @@ export class FleetPlugin
       taskManager: deps.taskManager,
       logFactory: this.initializerContext.logger,
     });
+    this.syncIntegrationsTask = new SyncIntegrationsTask({
+      core,
+      taskManager: deps.taskManager,
+      logFactory: this.initializerContext.logger,
+    });
 
     // Register fields metadata extractors
     registerFieldsMetadataExtractors({ core, fieldsMetadata: deps.fieldsMetadata });
@@ -696,6 +704,7 @@ export class FleetPlugin
       deleteUnenrolledAgentsTask: this.deleteUnenrolledAgentsTask!,
       taskManagerStart: plugins.taskManager,
       fetchUsage: this.fetchUsage,
+      syncIntegrationsTask: this.syncIntegrationsTask!,
     });
     licenseService.start(plugins.licensing.license$);
     this.telemetryEventsSender.start(plugins.telemetry, core).catch(() => {});
@@ -708,6 +717,7 @@ export class FleetPlugin
     this.fleetMetricsTask
       ?.start(plugins.taskManager, core.elasticsearch.client.asInternalUser)
       .catch(() => {});
+    this.syncIntegrationsTask?.start({ taskManager: plugins.taskManager }).catch(() => {});
 
     const logger = appContextService.getLogger();
 

x-pack/platform/plugins/shared/fleet/server/services/epm/packages/get.ts

@@ -311,7 +311,7 @@ export async function getPackageSavedObjects(
   return result;
 }
 
-async function getInstalledPackageSavedObjects(
+export async function getInstalledPackageSavedObjects(
   savedObjectsClient: SavedObjectsClientContract,
   options: Omit<GetInstalledPackagesOptions, 'savedObjectsClient' | 'esClient'>
 ) {

x-pack/platform/plugins/shared/fleet/server/services/setup.ts

@@ -64,6 +64,7 @@ import {
 } from './preconfiguration/delete_unenrolled_agent_setting';
 import { backfillPackagePolicySupportsAgentless } from './backfill_agentless';
 import { updateDeprecatedComponentTemplates } from './setup/update_deprecated_component_templates';
+import { createOrUpdateFleetSyncedIntegrationsIndex } from './setup/fleet_synced_integrations';
 
 export interface SetupStatus {
   isInitialized: boolean;
@@ -313,6 +314,9 @@ async function createSetupSideEffects(
   logger.debug('Update deprecated _source.mode in component templates');
   await updateDeprecatedComponentTemplates(esClient);
 
+  logger.debug('Create or update fleet-synced-integrations index');
+  await createOrUpdateFleetSyncedIntegrationsIndex(esClient);
+
   const nonFatalErrors = [
     ...preconfiguredPackagesNonFatalErrors,
     ...(messageSigningServiceNonFatalError ? [messageSigningServiceNonFatalError] : []),

x-pack/platform/plugins/shared/fleet/server/services/setup/fleet_synced_integrations.test.ts

@@ -0,0 +1,73 @@
+/*
+ * 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 { createOrUpdateFleetSyncedIntegrationsIndex } from './fleet_synced_integrations';
+
+jest.mock('../app_context', () => ({
+  appContextService: {
+    getExperimentalFeatures: jest.fn().mockReturnValue({ enableSyncIntegrationsOnRemote: true }),
+  },
+}));
+
+describe('fleet_synced_integrations', () => {
+  let esClientMock: any;
+  const mockExists = jest.fn();
+  const mockGetMapping = jest.fn();
+
+  beforeEach(() => {
+    esClientMock = {
+      indices: {
+        create: jest.fn(),
+        exists: mockExists,
+        getMapping: mockGetMapping,
+        putMapping: jest.fn(),
+      },
+    };
+  });
+
+  it('should create index if not exists', async () => {
+    mockExists.mockResolvedValue(false);
+
+    await createOrUpdateFleetSyncedIntegrationsIndex(esClientMock);
+
+    expect(esClientMock.indices.create).toHaveBeenCalled();
+  });
+
+  it('should update index if older version exists', async () => {
+    mockExists.mockResolvedValue(true);
+    mockGetMapping.mockResolvedValue({
+      'fleet-synced-integrations': {
+        mappings: {
+          _meta: {
+            version: '0.0',
+          },
+        },
+      },
+    });
+
+    await createOrUpdateFleetSyncedIntegrationsIndex(esClientMock);
+
+    expect(esClientMock.indices.putMapping).toHaveBeenCalled();
+  });
+
+  it('should not update index if same version exists', async () => {
+    mockExists.mockResolvedValue(true);
+    mockGetMapping.mockResolvedValue({
+      'fleet-synced-integrations': {
+        mappings: {
+          _meta: {
+            version: '1.0',
+          },
+        },
+      },
+    });
+
+    await createOrUpdateFleetSyncedIntegrationsIndex(esClientMock);
+
+    expect(esClientMock.indices.putMapping).not.toHaveBeenCalled();
+  });
+});

x-pack/platform/plugins/shared/fleet/server/services/setup/fleet_synced_integrations.ts

@@ -0,0 +1,117 @@
+/*
+ * 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 type { ElasticsearchClient } from '@kbn/core/server';
+
+import { FleetSetupError } from '../../errors';
+import { appContextService } from '../app_context';
+
+export const FLEET_SYNCED_INTEGRATIONS_INDEX_NAME = 'fleet-synced-integrations';
+
+export const FLEET_SYNCED_INTEGRATIONS_INDEX_CONFIG = {
+  settings: {
+    auto_expand_replicas: '0-1',
+  },
+  mappings: {
+    dynamic: false,
+    _meta: {
+      version: '1.0',
+    },
+    properties: {
+      remote_es_hosts: {
+        properties: {
+          name: {
+            type: 'keyword',
+          },
+          hosts: {
+            type: 'keyword',
+          },
+          sync_integrations: {
+            type: 'boolean',
+          },
+        },
+      },
+      integrations: {
+        properties: {
+          package_name: {
+            type: 'keyword',
+          },
+          package_version: {
+            type: 'keyword',
+          },
+          updated_at: {
+            type: 'date',
+          },
+        },
+      },
+    },
+  },
+};
+
+export async function createOrUpdateFleetSyncedIntegrationsIndex(esClient: ElasticsearchClient) {
+  const { enableSyncIntegrationsOnRemote } = appContextService.getExperimentalFeatures();
+
+  if (!enableSyncIntegrationsOnRemote) {
+    return;
+  }
+
+  await createOrUpdateIndex(
+    esClient,
+    FLEET_SYNCED_INTEGRATIONS_INDEX_NAME,
+    FLEET_SYNCED_INTEGRATIONS_INDEX_CONFIG
+  );
+}
+
+async function createOrUpdateIndex(
+  esClient: ElasticsearchClient,
+  indexName: string,
+  indexData: any
+) {
+  const resExists = await esClient.indices.exists({
+    index: indexName,
+  });
+
+  if (resExists) {
+    return updateIndex(esClient, indexName, indexData);
+  }
+
+  return createIndex(esClient, indexName, indexData);
+}
+
+async function updateIndex(esClient: ElasticsearchClient, indexName: string, indexData: any) {
+  try {
+    const res = await esClient.indices.getMapping({
+      index: indexName,
+    });
+
+    const versionChanged =
+      res[indexName].mappings?._meta?.version !== indexData.mappings._meta.version;
+    if (versionChanged) {
+      await esClient.indices.putMapping({
+        index: indexName,
+        body: Object.assign({
+          ...indexData.mappings,
+        }),
+      });
+    }
+  } catch (err) {
+    throw new FleetSetupError(`update of index [${indexName}] failed ${err}`);
+  }
+}
+
+async function createIndex(esClient: ElasticsearchClient, indexName: string, indexData: any) {
+  try {
+    await esClient.indices.create({
+      index: indexName,
+      body: indexData,
+    });
+  } catch (err) {
+    if (err?.body?.error?.type !== 'resource_already_exists_exception') {
+      throw new FleetSetupError(`create of index [${indexName}] failed ${err}`);
+    }
+  }
+}

x-pack/platform/plugins/shared/fleet/server/services/setup/index.ts

@@ -9,3 +9,4 @@ export { upgradePackageInstallVersion } from './upgrade_package_install_version'
 export { upgradeAgentPolicySchemaVersion } from './upgrade_agent_policy_schema_version';
 export { ensureAgentPoliciesFleetServerKeysAndPolicies } from './fleet_server_policies_enrollment_keys';
 export { updateDeprecatedComponentTemplates } from './update_deprecated_component_templates';
+export { createOrUpdateFleetSyncedIntegrationsIndex } from './fleet_synced_integrations';