[Fleet] Fix issue of agent sometimes not getting inputs using a new a…

…gent policy with system integration (#177594)

## Summary

Closes #177372

When creating an agent policy with a package policy immediately (e.g.
system integration), the `deployPolicy` logic was called once, creating
a doc in `.fleet-policies` with `revision:1` without `inputs`, and then
updating the doc with `inputs`, still on `revision:1`.
This is causing an intermittent issue on the agents, if Fleet-server
picks up the first document, and delivers to agent without `inputs`.
As a fix, added an option to skip `deploPolicy` when called from the
`createAgentPolicyWithPackages` function, as the policy will be deployed
after creating the package policies.

To verify:
- create an agent policy with system monitoring (default option)
- check that the created documents in `.fleet-policies` are correct:
there should be one doc with `revision_idx:1` and `coordinator_idx:0`
(created by Fleet API), and one doc with `revision_idx:1` and
`coordinator_idx:1` (created by fleet-server)
- verify that both documents have `data.inputs` field populated

Used this query to verify:
```
POST .fleet-policies/_search
{
  "query": {
    "bool": {
      "must": [
        {
          "term": {"coordinator_idx": 0}
        }
      ],
    "filter": {
      "term": {
      "policy_id": "<agent policy id>"
      }
    }
    }
  }, 
  "_source": [
    "revision_idx","coordinator_idx", "policy_id",  "@timestamp", "data.inputs"
  ],
  "sort": [
    {
      "revision_idx": {
        "order": "desc"
      }
    }
  ]
}
```


### 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

x-pack/plugins/fleet/server/routes/agent_policy/handlers.ts

@@ -172,8 +172,7 @@ export const createAgentPolicyHandler: FleetRequestHandler<
   const user = (await appContextService.getSecurity()?.authc.getCurrentUser(request)) || undefined;
   const withSysMonitoring = request.query.sys_monitoring ?? false;
   const monitoringEnabled = request.body.monitoring_enabled;
-  const force = request.body.force;
-  const { has_fleet_server: hasFleetServer, ...newPolicy } = request.body;
+  const { has_fleet_server: hasFleetServer, force, ...newPolicy } = request.body;
   const spaceId = fleetContext.spaceId;
   const authorizationHeader = HTTPAuthorizationHeader.parseFromRequest(request, user?.username);
 

x-pack/plugins/fleet/server/services/agent_policy.test.ts

@@ -865,7 +865,7 @@ describe('agent policy', () => {
       expect(esClient.bulk).toBeCalledWith(
         expect.objectContaining({
           index: AGENT_POLICY_INDEX,
-          body: [
+          operations: [
             expect.objectContaining({
               index: {
                 _id: expect.anything(),

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

@@ -112,9 +112,10 @@ class AgentPolicyService {
     soClient: SavedObjectsClientContract,
     esClient: ElasticsearchClient,
     action: 'created' | 'updated' | 'deleted',
-    agentPolicyId: string
+    agentPolicyId: string,
+    options?: { skipDeploy?: boolean }
   ) => {
-    return agentPolicyUpdateEventHandler(soClient, esClient, action, agentPolicyId);
+    return agentPolicyUpdateEventHandler(soClient, esClient, action, agentPolicyId, options);
   };
 
   private async _update(
@@ -286,6 +287,7 @@ class AgentPolicyService {
       id?: string;
       user?: AuthenticatedUser;
       authorizationHeader?: HTTPAuthorizationHeader | null;
+      skipDeploy?: boolean;
     } = {}
   ): Promise<AgentPolicy> {
     // Ensure an ID is provided, so we can include it in the audit logs below
@@ -330,7 +332,9 @@ class AgentPolicyService {
     );
 
     await appContextService.getUninstallTokenService()?.generateTokenForPolicyId(newSo.id);
-    await this.triggerAgentPolicyUpdatedEvent(soClient, esClient, 'created', newSo.id);
+    await this.triggerAgentPolicyUpdatedEvent(soClient, esClient, 'created', newSo.id, {
+      skipDeploy: options.skipDeploy,
+    });
     logger.debug(`Created new agent policy with id ${newSo.id}`);
     return { id: newSo.id, ...newSo.attributes };
   }
@@ -1034,7 +1038,7 @@ class AgentPolicyService {
 
     const bulkResponse = await esClient.bulk({
       index: AGENT_POLICY_INDEX,
-      body: fleetServerPoliciesBulkBody,
+      operations: fleetServerPoliciesBulkBody,
       refresh: 'wait_for',
     });
 

x-pack/plugins/fleet/server/services/agent_policy_create.test.ts

@@ -190,6 +190,35 @@ describe('createAgentPolicyWithPackages', () => {
     );
   });
 
+  it('should call deploy policy once when create policy with system package', async () => {
+    mockedAgentPolicyService.deployPolicy.mockClear();
+    mockedAgentPolicyService.create.mockImplementation((soClient, esClient, newPolicy, options) => {
+      if (!options?.skipDeploy) {
+        mockedAgentPolicyService.deployPolicy(soClientMock, 'new_id');
+      }
+      return Promise.resolve({
+        ...newPolicy,
+        id: options?.id || 'new_id',
+      } as AgentPolicy);
+    });
+    const response = await createAgentPolicyWithPackages({
+      esClient: esClientMock,
+      soClient: soClientMock,
+      newPolicy: { name: 'Agent policy 1', namespace: 'default' },
+      withSysMonitoring: true,
+      spaceId: 'default',
+    });
+
+    expect(response.id).toEqual('new_id');
+    expect(mockedBulkInstallPackages).toHaveBeenCalledWith({
+      savedObjectsClient: soClientMock,
+      esClient: esClientMock,
+      packagesToInstall: ['system'],
+      spaceId: 'default',
+    });
+    expect(mockedAgentPolicyService.deployPolicy).toHaveBeenCalledTimes(1);
+  });
+
   it('should create policy with system and elastic_agent package', async () => {
     const response = await createAgentPolicyWithPackages({
       esClient: esClientMock,

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

@@ -54,6 +54,7 @@ async function createPackagePolicy(
     spaceId: string;
     user: AuthenticatedUser | undefined;
     authorizationHeader?: HTTPAuthorizationHeader | null;
+    force?: boolean;
   }
 ) {
   const newPackagePolicy = await packagePolicyService
@@ -78,6 +79,7 @@ async function createPackagePolicy(
     user: options.user,
     bumpRevision: false,
     authorizationHeader: options.authorizationHeader,
+    force: options.force,
   });
 }
 
@@ -140,6 +142,7 @@ export async function createAgentPolicyWithPackages({
     user,
     id: agentPolicyId,
     authorizationHeader,
+    skipDeploy: true, // skip deploying the policy until package policies are added
   });
 
   // Create the fleet server package policy and add it to agent policy.
@@ -148,6 +151,7 @@ export async function createAgentPolicyWithPackages({
       spaceId,
       user,
       authorizationHeader,
+      force,
     });
   }
 
@@ -157,6 +161,7 @@ export async function createAgentPolicyWithPackages({
       spaceId,
       user,
       authorizationHeader,
+      force,
     });
   }
 

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

@@ -32,7 +32,8 @@ export async function agentPolicyUpdateEventHandler(
   soClient: SavedObjectsClientContract,
   esClient: ElasticsearchClient,
   action: string,
-  agentPolicyId: string
+  agentPolicyId: string,
+  options?: { skipDeploy?: boolean }
 ) {
   // `soClient` from ingest `appContextService` is used to create policy change actions
   // to ensure encrypted SOs are handled correctly
@@ -44,7 +45,9 @@ export async function agentPolicyUpdateEventHandler(
       agentPolicyId,
       forceRecreate: true,
     });
-    await agentPolicyService.deployPolicy(internalSoClient, agentPolicyId);
+    if (!options?.skipDeploy) {
+      await agentPolicyService.deployPolicy(internalSoClient, agentPolicyId);
+    }
   }
 
   if (action === 'updated') {

x-pack/test/fleet_api_integration/apis/agent_policy/agent_policy.ts

@@ -232,6 +232,32 @@ export default function (providerContext: FtrProviderContext) {
         });
       });
 
+      it('should create .fleet-policies document with inputs', async () => {
+        const res = await supertest
+          .post(`/api/fleet/agent_policies?sys_monitoring=true`)
+          .set('kbn-xsrf', 'xxxx')
+          .send({
+            name: 'test-policy-with-system',
+            namespace: 'default',
+            force: true, // using force to bypass package verification error
+          })
+          .expect(200);
+
+        const policyDocRes = await es.search({
+          index: '.fleet-policies',
+          query: {
+            term: {
+              policy_id: res.body.item.id,
+            },
+          },
+        });
+
+        expect(policyDocRes?.hits?.hits.length).to.eql(1);
+        const source = policyDocRes?.hits?.hits[0]?._source as any;
+        expect(source?.revision_idx).to.eql(1);
+        expect(source?.data?.inputs.length).to.eql(3);
+      });
+
       it('should return a 400 with an empty namespace', async () => {
         await supertest
           .post(`/api/fleet/agent_policies`)