[Fleet] Use new Fleet Secrets ES APIs instead of reading/writing secrets index (#163075)

Closes https://github.com/elastic/kibana/issues/162915

## Summary
Replace direct calls to Fleet Secrets index with new API calls
introduced with https://github.com/elastic/elasticsearch/pull/97728

### New ES secrets APIs:
```
POST /_fleet/secret/
{
  "value": "<secret value>"
}

// Returns the id of the created secret
{
  "id": "<secret_id>"
}

DELETE /_fleet/secret/<secret_id>

// returns 
{
  "deleted": true
}
```

NOTE: I tried running the secrets integration tests in
https://github.com/elastic/kibana/issues/162732 but there is some ES
error that I'm not sure how to address. I think that the test can be
worked on separately

### Testing

Testing steps are the exact same as
https://github.com/elastic/kibana/pull/157176:
- Start EPR locally loading the `Secrets` test package from Kibana:

```
docker run -p 8080:8080 -v /Users/<YOUR_PATH>/kibana/x-pack/test/fleet_api_integration/apis/fixtures/test_packages:/packages/test-packages -v /Users/<YOUR_PATH>/kibana/x-pack/test/fleet_api_integration/apis/fixtures/package_registry_config.yml:/package-registry/config.yml docker.elastic.co/package-registry/package-registry:main
```
- Point `kibana.dev.yml` to local EPR:
```
  xpack.fleet.registryUrl: http://localhost:8080
```
- Enable the secrets feature flag `secretsStorage`
- Start kibana and navigate to `integrations`, install `Secrets`
package.
- It should create and edit the package policy successfully

<img width="1800" alt="Screenshot 2023-08-08 at 16 26 52"
src="5e2b77d9-71a9-4c5f-8b3b-5fc6546d562f">

- The yml policy should have the redacted secrets and secrets ids:

<img width="771" alt="Screenshot 2023-08-08 at 15 43 22"
src="7db22c6b-b0db-4eb6-bc68-7174374c9c74">

---------

Co-authored-by: Kibana Machine <42973632+kibanamachine@users.noreply.github.com>

x-pack/plugins/fleet/common/constants/secrets.ts

@@ -5,4 +5,4 @@
  * 2.0.
  */
 
-export const SECRETS_INDEX = '.fleet-secrets';
+export const SECRETS_ENDPOINT_PATH = '/_fleet/secret';

x-pack/plugins/fleet/common/types/models/secret.ts

@@ -4,10 +4,9 @@
  * 2.0; you may not use this file except in compliance with the Elastic License
  * 2.0.
  */
-
+import type { PackagePolicyConfigRecordEntry } from '../..';
 export interface Secret {
   id: string;
-  value: string;
 }
 
 export interface SecretElasticDoc {
@@ -18,7 +17,19 @@ export interface VarSecretReference {
   id: string;
   isSecretRef: true;
 }
+export interface SecretPath {
+  path: string;
+  value: PackagePolicyConfigRecordEntry;
+}
 // this is used in the top level secret_refs array on package and agent policies
 export interface PolicySecretReference {
   id: string;
 }
+
+export interface DeletedSecretResponse {
+  deleted: boolean;
+}
+export interface DeletedSecretReference {
+  id: string;
+  deleted: boolean;
+}

x-pack/plugins/fleet/server/constants/index.ts

@@ -78,7 +78,7 @@ export {
   // Message signing service
   MESSAGE_SIGNING_SERVICE_API_ROUTES,
   // secrets
-  SECRETS_INDEX,
+  SECRETS_ENDPOINT_PATH,
 } from '../../common/constants';
 
 export {

x-pack/plugins/fleet/server/services/package_policy.ts

@@ -755,7 +755,6 @@ class PackagePolicyClientImpl implements PackagePolicyClient {
           packageInfo: pkgInfo,
           esClient,
         });
-
         restOfPackagePolicy = secretsRes.packagePolicyUpdate;
         secretReferences = secretsRes.secretReferences;
         secretsToDelete = secretsRes.secretsToDelete;

x-pack/plugins/fleet/server/services/secrets.ts

@@ -6,19 +6,13 @@
  */
 
 import type { ElasticsearchClient, SavedObjectsClientContract } from '@kbn/core/server';
-import type { BulkResponse } from '@elastic/elasticsearch/lib/api/types';
 
-import { keyBy, partition } from 'lodash';
+import { keyBy } from 'lodash';
 import { set } from '@kbn/safer-lodash-set';
 
 import { packageHasNoPolicyTemplates } from '../../common/services/policy_template';
 
-import type {
-  NewPackagePolicy,
-  PackagePolicyConfigRecordEntry,
-  RegistryStream,
-  UpdatePackagePolicy,
-} from '../../common';
+import type { NewPackagePolicy, RegistryStream, UpdatePackagePolicy } from '../../common';
 import { SO_SEARCH_LIMIT } from '../../common';
 
 import {
@@ -34,75 +28,61 @@ import type {
   Secret,
   VarSecretReference,
   PolicySecretReference,
+  SecretPath,
+  DeletedSecretResponse,
+  DeletedSecretReference,
 } from '../types';
 
 import { FleetError } from '../errors';
-import { SECRETS_INDEX } from '../constants';
+import { SECRETS_ENDPOINT_PATH } from '../constants';
+
+import { retryTransientEsErrors } from './epm/elasticsearch/retry';
 
 import { auditLoggingService } from './audit_logging';
 
 import { appContextService } from './app_context';
 import { packagePolicyService } from './package_policy';
 
-interface SecretPath {
-  path: string;
-  value: PackagePolicyConfigRecordEntry;
-}
-
-// This will be removed once the secrets index PR is merged into elasticsearch
-function getSecretsIndex() {
-  const testIndex = appContextService.getConfig()?.developer?.testSecretsIndex;
-  if (testIndex) {
-    return testIndex;
-  }
-  return SECRETS_INDEX;
-}
-
 export async function createSecrets(opts: {
   esClient: ElasticsearchClient;
   values: string[];
 }): Promise<Secret[]> {
   const { esClient, values } = opts;
   const logger = appContextService.getLogger();
-  const body = values.flatMap((value) => [
-    {
-      create: { _index: getSecretsIndex() },
-    },
-    { value },
-  ]);
-  let res: BulkResponse;
-  try {
-    res = await esClient.bulk({
-      body,
+
+  const secretsResponse: Secret[] = await Promise.all(
+    values.map(async (value) => {
+      try {
+        return await retryTransientEsErrors(
+          () =>
+            esClient.transport.request({
+              method: 'POST',
+              path: SECRETS_ENDPOINT_PATH,
+              body: { value },
+            }),
+          { logger }
+        );
+      } catch (err) {
+        const msg = `Error creating secrets: ${err}`;
+        logger.error(msg);
+        throw new FleetError(msg);
+      }
+    })
+  );
+
+  secretsResponse.forEach((item) => {
+    auditLoggingService.writeCustomAuditLog({
+      message: `secret created: ${item.id}`,
+      event: {
+        action: 'secret_create',
+        category: ['database'],
+        type: ['access'],
+        outcome: 'success',
+      },
     });
+  });
 
-    const [errorItems, successItems] = partition(res.items, (a) => a.create?.error);
-
-    successItems.forEach((item) => {
-      auditLoggingService.writeCustomAuditLog({
-        message: `secret created: ${item.create!._id}`,
-        event: {
-          action: 'secret_create',
-          category: ['database'],
-          type: ['access'],
-          outcome: 'success',
-        },
-      });
-    });
-
-    if (errorItems.length) {
-      throw new Error(JSON.stringify(errorItems));
-    }
-
-    return res.items.map((item, i) => ({
-      id: item.create!._id as string,
-      value: values[i],
-    }));
-  } catch (e) {
-    const msg = `Error creating secrets in ${getSecretsIndex()} index: ${e}`;
-    logger.error(msg);
-    throw new FleetError(msg);
-  }
+  return secretsResponse;
 }
 
 export async function deleteSecretsIfNotReferenced(opts: {
@@ -190,24 +170,32 @@ export async function _deleteSecrets(opts: {
 }): Promise<void> {
   const { esClient, ids } = opts;
   const logger = appContextService.getLogger();
-  const body = ids.flatMap((id) => [
-    {
-      delete: { _index: getSecretsIndex(), _id: id },
-    },
-  ]);
 
-  let res: BulkResponse;
+  const deletedRes: DeletedSecretReference[] = await Promise.all(
+    ids.map(async (id) => {
+      try {
+        const getDeleteRes: DeletedSecretResponse = await retryTransientEsErrors(
+          () =>
+            esClient.transport.request({
+              method: 'DELETE',
+              path: `${SECRETS_ENDPOINT_PATH}/${id}`,
+            }),
+          { logger }
+        );
 
-  try {
-    res = await esClient.bulk({
-      body,
-    });
+        return { ...getDeleteRes, id };
+      } catch (err) {
+        const msg = `Error deleting secrets: ${err}`;
+        logger.error(msg);
+        throw new FleetError(msg);
+      }
+    })
+  );
 
-    const [errorItems, successItems] = partition(res.items, (a) => a.delete?.error);
-
-    successItems.forEach((item) => {
+  deletedRes.forEach((item) => {
+    if (item.deleted === true) {
       auditLoggingService.writeCustomAuditLog({
-        message: `secret deleted: ${item.delete!._id}`,
+        message: `secret deleted: ${item.id}`,
         event: {
           action: 'secret_delete',
           category: ['database'],
@@ -215,16 +203,8 @@ export async function _deleteSecrets(opts: {
           outcome: 'success',
         },
       });
-    });
-
-    if (errorItems.length) {
-      throw new Error(JSON.stringify(errorItems));
     }
-  } catch (e) {
-    const msg = `Error deleting secrets from ${getSecretsIndex()} index: ${e}`;
-    logger.error(msg);
-    throw new FleetError(msg);
-  }
+  });
 }
 
 export async function extractAndWriteSecrets(opts: {

x-pack/plugins/fleet/server/types/index.tsx

@@ -85,8 +85,11 @@ export type {
   ExperimentalDataStreamFeature,
   Secret,
   SecretElasticDoc,
+  SecretPath,
   VarSecretReference,
   PolicySecretReference,
+  DeletedSecretResponse,
+  DeletedSecretReference,
   PackageListItem,
   PackageList,
   InstallationInfo,