[Security Solution][Endpoint] Add step to the security solution plugi…

…n `start` phase (non-blocking) to check endpoint policy indices (#198089)

## Summary

- adds a step to the plugin `start` phase to retrieve all Endpoint
policies from Fleet and check to ensure they have backing DOT indices.
    - This is a follow up to PR #196953 
- this check will be removed once it is deployed to Serverless, since it
only needs to run once in that flavor of kibana

x-pack/plugins/security_solution/server/endpoint/migrations/ensure_indices_exists_for_policies.test.ts

@@ -0,0 +1,47 @@
+/*
+ * 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 { createMockEndpointAppContextService } from '../mocks';
+import { ensureIndicesExistsForPolicies } from './ensure_indices_exists_for_policies';
+import { createPolicyDataStreamsIfNeeded as _createPolicyDataStreamsIfNeeded } from '../../fleet_integration/handlers/create_policy_datastreams';
+
+jest.mock('../../fleet_integration/handlers/create_policy_datastreams');
+const createPolicyDataStreamsIfNeededMock =
+  _createPolicyDataStreamsIfNeeded as unknown as jest.Mock;
+
+describe('Ensure indices exists for policies migration', () => {
+  let endpointAppContextServicesMock: ReturnType<typeof createMockEndpointAppContextService>;
+
+  beforeEach(() => {
+    endpointAppContextServicesMock = createMockEndpointAppContextService();
+
+    (
+      endpointAppContextServicesMock.getInternalFleetServices().packagePolicy.listIds as jest.Mock
+    ).mockResolvedValue({
+      items: ['foo-1', 'foo-2', 'foo-3'],
+    });
+  });
+
+  it('should query fleet looking for all endpoint integration policies', async () => {
+    const fleetServicesMock = endpointAppContextServicesMock.getInternalFleetServices();
+    await ensureIndicesExistsForPolicies(endpointAppContextServicesMock);
+
+    expect(fleetServicesMock.packagePolicy.listIds).toHaveBeenCalledWith(expect.anything(), {
+      kuery: fleetServicesMock.endpointPolicyKuery,
+      perPage: 10000,
+    });
+  });
+
+  it('should call createPolicyDataStreamsIfNeeded() with list of existing policies', async () => {
+    await ensureIndicesExistsForPolicies(endpointAppContextServicesMock);
+
+    expect(createPolicyDataStreamsIfNeededMock).toHaveBeenCalledWith({
+      endpointServices: endpointAppContextServicesMock,
+      endpointPolicyIds: ['foo-1', 'foo-2', 'foo-3'],
+    });
+  });
+});

x-pack/plugins/security_solution/server/endpoint/migrations/ensure_indices_exists_for_policies.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 { createPolicyDataStreamsIfNeeded } from '../../fleet_integration/handlers/create_policy_datastreams';
+import type { EndpointAppContextService } from '../endpoint_app_context_services';
+
+export const ensureIndicesExistsForPolicies = async (
+  endpointServices: EndpointAppContextService
+): Promise<void> => {
+  const logger = endpointServices.createLogger('startupPolicyIndicesChecker');
+
+  const fleetServices = endpointServices.getInternalFleetServices();
+  const soClient = fleetServices.savedObjects.createInternalUnscopedSoClient();
+  const endpointPoliciesIds = await fleetServices.packagePolicy.listIds(soClient, {
+    kuery: fleetServices.endpointPolicyKuery,
+    perPage: 10000,
+  });
+
+  logger.info(
+    `Checking to ensure [${endpointPoliciesIds.items.length}] endpoint policies have backing indices`
+  );
+
+  await createPolicyDataStreamsIfNeeded({
+    endpointServices,
+    endpointPolicyIds: endpointPoliciesIds.items,
+  });
+};

x-pack/plugins/security_solution/server/fleet_integration/handlers/create_policy_datastreams.ts

@@ -78,22 +78,20 @@ export const createPolicyDataStreamsIfNeeded: PolicyDataStreamsCreator = async (
   });
   const indexesCreated: string[] = [];
   const createErrors: string[] = [];
-  const indicesToCreate: string[] = Object.values(policyNamespaces.integrationPolicy).reduce<
-    string[]
-  >((acc, namespaceList) => {
-    for (const namespace of namespaceList) {
-      acc.push(
-        buildIndexNameWithNamespace(DEFAULT_DIAGNOSTIC_INDEX, namespace),
-        buildIndexNameWithNamespace(ENDPOINT_ACTION_RESPONSES_DS, namespace)
-      );
-
-      if (endpointServices.isServerless()) {
-        acc.push(buildIndexNameWithNamespace(ENDPOINT_HEARTBEAT_INDEX_PATTERN, namespace));
+  const indicesToCreate: string[] = Array.from(
+    Object.values(policyNamespaces.integrationPolicy).reduce<Set<string>>((acc, namespaceList) => {
+      for (const namespace of namespaceList) {
+        acc.add(buildIndexNameWithNamespace(DEFAULT_DIAGNOSTIC_INDEX, namespace));
+        acc.add(buildIndexNameWithNamespace(ENDPOINT_ACTION_RESPONSES_DS, namespace));
+
+        if (endpointServices.isServerless()) {
+          acc.add(buildIndexNameWithNamespace(ENDPOINT_HEARTBEAT_INDEX_PATTERN, namespace));
+        }
       }
-    }
 
-    return acc;
-  }, []);
+      return acc;
+    }, new Set<string>())
+  );
 
   const processesDatastreamIndex = async (datastreamIndexName: string): Promise<void> => {
     if (cache.get(datastreamIndexName)) {

x-pack/plugins/security_solution/server/plugin.ts

@@ -19,6 +19,7 @@ import type { ILicense } from '@kbn/licensing-plugin/server';
 import type { NewPackagePolicy, UpdatePackagePolicy } from '@kbn/fleet-plugin/common';
 import { FLEET_ENDPOINT_PACKAGE } from '@kbn/fleet-plugin/common';
 
+import { ensureIndicesExistsForPolicies } from './endpoint/migrations/ensure_indices_exists_for_policies';
 import { CompleteExternalResponseActionsTask } from './endpoint/lib/response_actions';
 import { registerAgentRoutes } from './endpoint/routes/agent';
 import { endpointPackagePoliciesStatsSearchStrategyProvider } from './search_strategy/endpoint_package_policies_stats';
@@ -606,8 +607,10 @@ export class Plugin implements ISecuritySolutionPlugin {
       plugins.fleet
         .fleetSetupCompleted()
         .then(async () => {
+          logger.info('Dependent plugin setup complete');
+
           if (this.manifestTask) {
-            logger.info('Dependent plugin setup complete - Starting ManifestTask');
+            logger.info('Starting ManifestTask');
             await this.manifestTask.start({
               taskManager,
             });
@@ -625,6 +628,10 @@ export class Plugin implements ISecuritySolutionPlugin {
           );
 
           await turnOffAgentPolicyFeatures(fleetServices, productFeaturesService, logger);
+
+          // Ensure policies have backing DOT indices (We don't need to `await` this.
+          // It can run in the background)
+          ensureIndicesExistsForPolicies(this.endpointAppContextService).catch(() => {});
         })
         .catch(() => {});