[EDR Workflows][Investigation] Telemetry config watcher fix (elastic#210406)

## Summary

To update the `global_telemetry_config` flag in Defend package policies,
we subscribe to the Telemetry plugin's `isOptedIn$` observable during
Kibana's `start()` phase, and receive the initial value immediately.
This feature is used for 'migrating' existing package policies: after
stack upgrade, when Kibana starts up, this subscription mechanism makes
sure that existing policies are backfilled with the new field.

But not on cloud and serverless instances.

It turned out, that while this works on local instances, on cloud and
serverless instances, at the very moment we receive the value during
`start()`, some mechanisms are not yet green, and this resulted in
`security_exception: missing authentication credentials for REST
request` when trying to read Saved Objects.

As subscribing to`core.status.core$`, and waiting until `ServiceStatus`
for `elasticsearch` and `savedObjects` is `available` didn't solve the
issue, I simply added a retry mechanism, which, at least, protects
against other temporary issues as well.

Some additional logging is added as well.

### Checklist

Check the PR satisfies following conditions. 

Reviewers should verify this PR satisfies this list as well.

- [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: Elastic Machine <[email protected]>

Author: gergoabraham

x-pack/solutions/security/plugins/security_solution/server/endpoint/lib/policy/telemetry_watch.test.ts

@@ -6,43 +6,49 @@
  */
 
 import { Subject } from 'rxjs';
-import { elasticsearchServiceMock, savedObjectsServiceMock } from '@kbn/core/server/mocks';
+import { elasticsearchServiceMock } from '@kbn/core/server/mocks';
 import { createPackagePolicyServiceMock } from '@kbn/fleet-plugin/server/mocks';
 import type { PackagePolicyClient } from '@kbn/fleet-plugin/server';
-import type { PackagePolicy, UpdatePackagePolicy } from '@kbn/fleet-plugin/common';
+import {
+  PACKAGE_POLICY_SAVED_OBJECT_TYPE,
+  type PackagePolicy,
+  type UpdatePackagePolicy,
+} from '@kbn/fleet-plugin/common';
 import { createPackagePolicyMock } from '@kbn/fleet-plugin/common/mocks';
 import { policyFactory } from '../../../../common/endpoint/models/policy_config';
 import type { PolicyConfig } from '../../../../common/endpoint/types';
 import { TelemetryConfigWatcher } from './telemetry_watch';
 import { TelemetryConfigProvider } from '../../../../common/telemetry_config/telemetry_config_provider';
 import { createMockEndpointAppContextService } from '../../mocks';
+import type { MockedLogger } from '@kbn/logging-mocks';
+import { loggerMock } from '@kbn/logging-mocks';
+import type { NewPackagePolicyWithId } from '@kbn/fleet-plugin/server/services/package_policy';
 
 const MockPackagePolicyWithEndpointPolicy = (
   cb?: (p: PolicyConfig) => PolicyConfig
 ): PackagePolicy => {
   const packagePolicy = createPackagePolicyMock();
-  if (!cb) {
-    // eslint-disable-next-line no-param-reassign
-    cb = (p) => p;
-  }
-  const policyConfig = cb(policyFactory());
+
+  const policyConfig = cb?.(policyFactory()) ?? policyFactory();
+
   packagePolicy.inputs[0].config = { policy: { value: policyConfig } };
 
   return packagePolicy;
 };
 
 describe('Telemetry config watcher', () => {
-  const soStartMock = savedObjectsServiceMock.createStartContract();
   const esStartMock = elasticsearchServiceMock.createStart();
-  let packagePolicySvcMock: jest.Mocked<PackagePolicyClient>;
+
+  let mockedLogger: MockedLogger;
+  let packagePolicyServiceMock: jest.Mocked<PackagePolicyClient>;
   let telemetryWatcher: TelemetryConfigWatcher;
 
   const preparePackagePolicyMock = ({
     isGlobalTelemetryEnabled,
   }: {
     isGlobalTelemetryEnabled: boolean;
   }) => {
-    packagePolicySvcMock.list.mockResolvedValueOnce({
+    packagePolicyServiceMock.list.mockResolvedValueOnce({
       items: [
         MockPackagePolicyWithEndpointPolicy((pc: PolicyConfig): PolicyConfig => {
           pc.global_telemetry_enabled = isGlobalTelemetryEnabled;
@@ -56,13 +62,21 @@ describe('Telemetry config watcher', () => {
   };
 
   beforeEach(() => {
-    packagePolicySvcMock = createPackagePolicyServiceMock();
+    packagePolicyServiceMock = createPackagePolicyServiceMock();
+    packagePolicyServiceMock.bulkUpdate.mockResolvedValue({
+      updatedPolicies: [],
+      failedPolicies: [],
+    });
+
+    mockedLogger = loggerMock.create();
+    const endpointAppContextServiceMock = createMockEndpointAppContextService();
+    endpointAppContextServiceMock.createLogger = jest.fn().mockReturnValue(mockedLogger);
 
     telemetryWatcher = new TelemetryConfigWatcher(
-      packagePolicySvcMock,
-      soStartMock,
+      packagePolicyServiceMock,
       esStartMock,
-      createMockEndpointAppContextService()
+      endpointAppContextServiceMock,
+      { immediateRetry: true }
     );
   });
 
@@ -90,7 +104,7 @@ describe('Telemetry config watcher', () => {
     const TOTAL = 247;
 
     // set up the mocked package policy service to return and do what we want
-    packagePolicySvcMock.list
+    packagePolicyServiceMock.list
       .mockResolvedValueOnce({
         items: Array.from({ length: 100 }, () => MockPackagePolicyWithEndpointPolicy()),
         total: TOTAL,
@@ -112,12 +126,101 @@ describe('Telemetry config watcher', () => {
 
     await telemetryWatcher.watch(true); // manual trigger
 
-    expect(packagePolicySvcMock.list).toBeCalledTimes(3);
+    expect(packagePolicyServiceMock.list).toBeCalledTimes(3);
 
     // Assert: on the first call to packagePolicy.list, we asked for page 1
-    expect(packagePolicySvcMock.list.mock.calls[0][1].page).toBe(1);
-    expect(packagePolicySvcMock.list.mock.calls[1][1].page).toBe(2); // second call, asked for page 2
-    expect(packagePolicySvcMock.list.mock.calls[2][1].page).toBe(3); // etc
+    expect(packagePolicyServiceMock.list.mock.calls[0][1].page).toBe(1);
+    expect(packagePolicyServiceMock.list.mock.calls[1][1].page).toBe(2); // second call, asked for page 2
+    expect(packagePolicyServiceMock.list.mock.calls[2][1].page).toBe(3); // etc
+
+    expect(mockedLogger.warn).not.toHaveBeenCalled();
+    expect(mockedLogger.error).not.toHaveBeenCalled();
+  });
+
+  describe('error handling', () => {
+    it('retries fetching package policies', async () => {
+      packagePolicyServiceMock.list.mockRejectedValueOnce(new Error());
+
+      packagePolicyServiceMock.list.mockResolvedValueOnce({
+        items: Array.from({ length: 6 }, () => MockPackagePolicyWithEndpointPolicy()),
+        total: 6,
+        page: 1,
+        perPage: 100,
+      });
+
+      await telemetryWatcher.watch(true);
+
+      expect(packagePolicyServiceMock.list).toBeCalledTimes(2);
+      const expectedParams = {
+        page: 1,
+        perPage: 100,
+        kuery: `${PACKAGE_POLICY_SAVED_OBJECT_TYPE}.package.name: endpoint`,
+      };
+      expect(packagePolicyServiceMock.list.mock.calls[0][1]).toStrictEqual(expectedParams);
+      expect(packagePolicyServiceMock.list.mock.calls[1][1]).toStrictEqual(expectedParams);
+
+      expect(mockedLogger.warn).not.toHaveBeenCalled();
+      expect(mockedLogger.error).not.toHaveBeenCalled();
+    });
+
+    it('retries fetching package policies maximum 5 times', async () => {
+      packagePolicyServiceMock.list.mockRejectedValue(new Error());
+
+      await telemetryWatcher.watch(true);
+
+      expect(packagePolicyServiceMock.list).toBeCalledTimes(5);
+      expect(mockedLogger.warn).toHaveBeenCalledTimes(1);
+      expect(mockedLogger.error).not.toHaveBeenCalled();
+    });
+
+    it('retries bulk updating package policies', async () => {
+      preparePackagePolicyMock({ isGlobalTelemetryEnabled: true });
+      packagePolicyServiceMock.bulkUpdate.mockRejectedValueOnce(new Error());
+
+      await telemetryWatcher.watch(false);
+
+      expect(packagePolicyServiceMock.bulkUpdate).toHaveBeenCalledTimes(2);
+      expect(mockedLogger.warn).not.toHaveBeenCalled();
+      expect(mockedLogger.error).not.toHaveBeenCalled();
+    });
+
+    it('retries bulk updating package policies maximum 5 times', async () => {
+      preparePackagePolicyMock({ isGlobalTelemetryEnabled: true });
+      packagePolicyServiceMock.bulkUpdate.mockRejectedValue(new Error());
+
+      await telemetryWatcher.watch(false);
+
+      expect(packagePolicyServiceMock.bulkUpdate).toHaveBeenCalledTimes(5);
+      expect(mockedLogger.warn).toHaveBeenCalledTimes(1);
+      expect(mockedLogger.error).not.toHaveBeenCalled();
+    });
+
+    it('logs the ids of package policies that are failed to be updated', async () => {
+      preparePackagePolicyMock({ isGlobalTelemetryEnabled: true });
+      packagePolicyServiceMock.bulkUpdate.mockResolvedValueOnce({
+        updatedPolicies: [],
+        failedPolicies: [
+          {
+            error: new Error('error message 1'),
+            packagePolicy: { id: 'policy-id-1' } as NewPackagePolicyWithId,
+          },
+          {
+            error: new Error('error message 2'),
+            packagePolicy: { id: 'policy-id-2' } as NewPackagePolicyWithId,
+          },
+        ],
+      });
+
+      await telemetryWatcher.watch(false);
+
+      expect(packagePolicyServiceMock.bulkUpdate).toHaveBeenCalledTimes(1);
+      expect(mockedLogger.warn).toHaveBeenCalledTimes(1);
+      expect(mockedLogger.error).not.toHaveBeenCalled();
+
+      const logMessage = mockedLogger.warn.mock.calls[0][0] as string;
+      expect(logMessage).toMatch(/- id: policy-id-1, error:.+error message 1/);
+      expect(logMessage).toMatch(/- id: policy-id-2, error:.+error message 2/);
+    });
   });
 
   it.each([true, false])(
@@ -127,17 +230,18 @@ describe('Telemetry config watcher', () => {
 
       await telemetryWatcher.watch(value);
 
-      expect(packagePolicySvcMock.bulkUpdate).not.toHaveBeenCalled();
+      expect(packagePolicyServiceMock.bulkUpdate).not.toHaveBeenCalled();
     }
   );
 
-  it.each([true, false])('updates `global_telemetry_config` field to %s', async (value) => {
+  it.each([true, false])('updates `global_telemetry_enabled` field to %s', async (value) => {
     preparePackagePolicyMock({ isGlobalTelemetryEnabled: !value });
 
     await telemetryWatcher.watch(value);
 
-    expect(packagePolicySvcMock.bulkUpdate).toHaveBeenCalled();
-    const policyUpdates: UpdatePackagePolicy[] = packagePolicySvcMock.bulkUpdate.mock.calls[0][2];
+    expect(packagePolicyServiceMock.bulkUpdate).toHaveBeenCalled();
+    const policyUpdates: UpdatePackagePolicy[] =
+      packagePolicyServiceMock.bulkUpdate.mock.calls[0][2];
     expect(policyUpdates.length).toBe(1);
     const updatedPolicyConfigs: PolicyConfig = policyUpdates[0].inputs[0].config?.policy.value;
     expect(updatedPolicyConfigs.global_telemetry_enabled).toBe(value);