[Fleet] moving action batching to async - #138870

for testing locally with less than 10k agents enrolled, set a lower number here:

const batchSize = 1000; // options.batchSize ?? SO_SEARCH_LIMIT;

or invoke from API:

POST kbn:/api/fleet/agents/bulk_reassign
{
  "agents": " fleet-agents.policy_id : (\"e57948b0-1d55-11ed-85fe-e34c31cf865b\" or \"26594150-1d56-11ed-85fe-e34c31cf865b\")",
  "policy_id": "e57948b0-1d55-11ed-85fe-e34c31cf865b",
  "batchSize": 1000
}

The default expiration with upgrade Immediately is 30m, this is too short for larger agent selections e.g. 20k. When I tried upgrading immediately, the action still completed for all actions despite expiring about halfway through. I am not sure if this is the expected behavior that Fleet Server continues to execute the action even after expiration.

Yes we could probably have a larger expiration here and minimum expiration duration (this need to be the same) for immediate to happen,

The default expiration with upgrade Immediately is 30m, this is too short for larger agent selections e.g. 20k. When I tried upgrading immediately, the action still completed for all actions despite expiring about halfway through. I am not sure if this is the expected behavior that Fleet Server continues to execute the action even after expiration.

The action should be delivered immediatly to the agent, but Fleet server may have some trouble to get all the agent ack in 30 minutes.

Is there any disadvantage of increasing the minimum expiration time to e.g. 2 hours? updated.

I think increasing the minimum expiration time to 2 hours will mean that every rolling upgrade less than 2hours will not be a rolling upgrade, it's maybe acceptable, or maybe we can just have this to two hours or more for the immediate scenario

The previous logic of the rolling upgrade was to set minimum duration to 30m and expiration to the selected duration.
I updated the minimum to be 2hr (or 1hr if that was selected as the window)
Previously:

  // Perform a rolling upgrade
  if (upgradeDurationSeconds) {
    return {
      start_time: startTime ?? now,
      minimum_execution_duration: MINIMUM_EXECUTION_DURATION_SECONDS,
      expiration: moment(startTime ?? now)
        .add(upgradeDurationSeconds, 'seconds')
        .toISOString(),
    };
  }

Updated:

  // Perform a rolling upgrade
  if (upgradeDurationSeconds) {
    return {
      start_time: startTime ?? now,
      minimum_execution_duration: Math.min(
        MINIMUM_EXECUTION_DURATION_SECONDS,
        upgradeDurationSeconds
      ),
      expiration: moment(startTime ?? now)
        .add(upgradeDurationSeconds, 'seconds')
        .toISOString(),
    };
  }

nbAgentsActionCreated represents how many agents are included in .fleet-actions documents, if this is less than nbAgentsActioned, it indicates something went wrong with kibana batch processing.

For testing the async execution of bulk actions, I had to wait for a Promise, otherwise the FTR shutdown starts and the action doesn't finish.
Added an interval of 1s for 3 times to check whether the action has completed.

Thanks for adding this script to help with testing and local dev setup.

I think removing the expiration condition from this query makes sense.

That method seems a good candidate to be unit tested, and will help to document the expected behavior

Do you mean unit tested?
Yes, I am planning to add more tests after I complete the UI part.

optional nit: this combo of values + reduce could probably simplified to a single flatMap call.

Not quite, the result here is an object, not an array. I reused most of this logic from the existing current_upgrades API.

I'm concerned about the number of queries here. We're fetching all actions, then doing a separate query for action results for every action. Instead I think we should try combining that into a single query and use aggregations to get the count value that we need.

Likely something like this (pseudo-ish code):

let actions = await _getActions(esClient);

let acks = await esClient.search({
  index: AGENT_ACTIONS_RESULTS_INDEX,
  query: {
    bool: {
      // There's some perf/caching advantages to using filter over must
      // See https://www.elastic.co/guide/en/elasticsearch/reference/current/query-filter-context.html#filter-context
      filter: actions.map(a => ({ term: { action_id: a.id } }),
    }
  },
  aggs: {
    ack_counts: { terms: { field: "action_id" } }
  }
})

return actions.map(a => {
  return {
    ...action,
    nbAgentsAck: acks.aggs.ack_counts.buckets.find(b => b.key === a.id).doc_count,
   // other fields
  }
})

Is that okay if I refactor this in next pr about Agent activity UI? I already had to make changes to add more info.

I'd be okay with refactoring this fetch operation all at once in the next PR. Let's try to land this one and move on to the UI.

refactored here https://github.com/elastic/kibana/pull/140510/files#diff-a1a29af3b4d7a9d69539c684d9bbcaf584fd6db8a60eeebfe47da2af803cb621

Do we need to search for all actions? Is there some way to only filter on relevant or recent ones? Similar question on the canceled actions query

We could add a query param to limit the results. Will change this in the next pr.

I wonder if we should be checkpointing the state regardless of whether or not there is an error. We need to consider the case where Kibana crashes or is otherwise shutdown. Ideally in this case there's always a task scheduled to confirm that the process was completed successfully. This may require some significant changes to this PR and if so, we could handle it in a follow up.

yes, I'll tackle this separately.