[7.x] [Fleet] Support Fleet server system indices (#89372)

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

@@ -130,8 +130,8 @@ interface AgentBase {
   enrolled_at: string;
   unenrolled_at?: string;
   unenrollment_started_at?: string;
-  upgraded_at?: string;
-  upgrade_started_at?: string;
+  upgraded_at?: string | null;
+  upgrade_started_at?: string | null;
   access_api_key_id?: string;
   default_api_key?: string;
   default_api_key_id?: string;
@@ -155,3 +155,163 @@ export interface AgentSOAttributes extends AgentBase {
   current_error_events?: string;
   packages?: string[];
 }
+
+// Generated from FleetServer schema.json
+
+/**
+ * An Elastic Agent that has enrolled into Fleet
+ */
+export interface FleetServerAgent {
+  /**
+   * The version of the document in the index
+   */
+  _version?: number;
+  /**
+   * Shared ID
+   */
+  shared_id?: string;
+  /**
+   * Type
+   */
+  type: AgentType;
+  /**
+   * Active flag
+   */
+  active: boolean;
+  /**
+   * Date/time the Elastic Agent enrolled
+   */
+  enrolled_at: string;
+  /**
+   * Date/time the Elastic Agent unenrolled
+   */
+  unenrolled_at?: string;
+  /**
+   * Date/time the Elastic Agent unenrolled started
+   */
+  unenrollment_started_at?: string;
+  /**
+   * Date/time the Elastic Agent was last upgraded
+   */
+  upgraded_at?: string | null;
+  /**
+   * Date/time the Elastic Agent started the current upgrade
+   */
+  upgrade_started_at?: string | null;
+  /**
+   * ID of the API key the Elastic Agent must used to contact Fleet Server
+   */
+  access_api_key_id?: string;
+  agent?: FleetServerAgentMetadata;
+  /**
+   * User provided metadata information for the Elastic Agent
+   */
+  user_provided_metadata: AgentMetadata;
+  /**
+   * Local metadata information for the Elastic Agent
+   */
+  local_metadata: AgentMetadata;
+  /**
+   * The policy ID for the Elastic Agent
+   */
+  policy_id?: string;
+  /**
+   * The current policy revision_idx for the Elastic Agent
+   */
+  policy_revision_idx?: number | null;
+  /**
+   * The current policy coordinator for the Elastic Agent
+   */
+  policy_coordinator_idx?: number;
+  /**
+   * Date/time the Elastic Agent was last updated
+   */
+  last_updated?: string;
+  /**
+   * Date/time the Elastic Agent checked in last time
+   */
+  last_checkin?: string;
+  /**
+   * Lst checkin status
+   */
+  last_checkin_status?: 'error' | 'online' | 'degraded' | 'updating';
+  /**
+   * ID of the API key the Elastic Agent uses to authenticate with elasticsearch
+   */
+  default_api_key_id?: string;
+  /**
+   * API key the Elastic Agent uses to authenticate with elasticsearch
+   */
+  default_api_key?: string;
+  /**
+   * Date/time the Elastic Agent was last updated
+   */
+  updated_at?: string;
+  /**
+   * Packages array
+   */
+  packages?: string[];
+  /**
+   * The last acknowledged action sequence number for the Elastic Agent
+   */
+  action_seq_no?: number;
+}
+/**
+ * An Elastic Agent metadata
+ */
+export interface FleetServerAgentMetadata {
+  /**
+   * The unique identifier for the Elastic Agent
+   */
+  id: string;
+  /**
+   * The version of the Elastic Agent
+   */
+  version: string;
+  [k: string]: any;
+}
+
+/**
+ * An Elastic Agent action
+ */
+export interface FleetServerAgentAction {
+  /**
+   * The unique identifier for action document
+   */
+  _id?: string;
+  /**
+   * The action sequence number
+   */
+  _seq_no?: number;
+  /**
+   * The unique identifier for the Elastic Agent action. There could be multiple documents with the same action_id if the action is split into two separate documents.
+   */
+  action_id?: string;
+  /**
+   * Date/time the action was created
+   */
+  '@timestamp'?: string;
+  /**
+   * The action expiration date/time
+   */
+  expiration?: string;
+  /**
+   * The action type. APP_ACTION is the value for the actions that suppose to be routed to the endpoints/beats.
+   */
+  type?: string;
+  /**
+   * The input identifier the actions should be routed to.
+   */
+  input_id?: string;
+  /**
+   * The Agent IDs the action is intended for. No support for json.RawMessage with the current generator. Could be useful to lazy parse the agent ids
+   */
+  agents?: string[];
+  /**
+   * The opaque payload.
+   */
+  data?: {
+    [k: string]: unknown;
+  };
+  [k: string]: unknown;
+}

x-pack/plugins/fleet/server/routes/agent/acks_handlers.ts

@@ -20,7 +20,7 @@ export const postAgentAcksHandlerBuilder = function (
     try {
       const soClient = ackService.getSavedObjectsClientContract(request);
       const esClient = ackService.getElasticsearchClientContract();
-      const agent = await ackService.authenticateAgentWithAccessToken(soClient, request);
+      const agent = await ackService.authenticateAgentWithAccessToken(soClient, esClient, request);
       const agentEvents = request.body.events as AgentEvent[];
 
       // validate that all events are for the authorized agent obtained from the api key

x-pack/plugins/fleet/server/routes/agent/actions_handlers.ts

@@ -30,7 +30,7 @@ export const postNewAgentActionHandlerBuilder = function (
 
       const newAgentAction = request.body.action;
 
-      const savedAgentAction = await actionsService.createAgentAction(soClient, {
+      const savedAgentAction = await actionsService.createAgentAction(soClient, esClient, {
         created_at: new Date().toISOString(),
         ...newAgentAction,
         agent_id: agent.id,

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

@@ -132,8 +132,8 @@ export const updateAgentHandler: RequestHandler<
   const esClient = context.core.elasticsearch.client.asCurrentUser;
 
   try {
-    await AgentService.updateAgent(soClient, request.params.agentId, {
-      userProvidedMetatada: request.body.user_provided_metadata,
+    await AgentService.updateAgent(soClient, esClient, request.params.agentId, {
+      user_provided_metadata: request.body.user_provided_metadata,
     });
     const agent = await AgentService.getAgent(soClient, esClient, request.params.agentId);
 
@@ -164,12 +164,13 @@ export const postAgentCheckinHandler: RequestHandler<
   try {
     const soClient = appContextService.getInternalUserSOClient(request);
     const esClient = appContextService.getInternalUserESClient();
-    const agent = await AgentService.authenticateAgentWithAccessToken(soClient, request);
+    const agent = await AgentService.authenticateAgentWithAccessToken(soClient, esClient, request);
     const abortController = new AbortController();
     request.events.aborted$.subscribe(() => {
       abortController.abort();
     });
     const signal = abortController.signal;
+
     const { actions } = await AgentService.agentCheckin(
       soClient,
       esClient,
@@ -205,8 +206,13 @@ export const postAgentEnrollHandler: RequestHandler<
 > = async (context, request, response) => {
   try {
     const soClient = appContextService.getInternalUserSOClient(request);
+    const esClient = context.core.elasticsearch.client.asInternalUser;
     const { apiKeyId } = APIKeyService.parseApiKeyFromHeaders(request.headers);
-    const enrollmentAPIKey = await APIKeyService.getEnrollmentAPIKeyById(soClient, apiKeyId);
+    const enrollmentAPIKey = await APIKeyService.getEnrollmentAPIKeyById(
+      soClient,
+      esClient,
+      apiKeyId
+    );
 
     if (!enrollmentAPIKey || !enrollmentAPIKey.active) {
       return response.unauthorized({
@@ -311,21 +317,16 @@ export const postBulkAgentsReassignHandler: RequestHandler<
   const soClient = context.core.savedObjects.client;
   const esClient = context.core.elasticsearch.client.asInternalUser;
   try {
-    // Reassign by array of IDs
-    const result = Array.isArray(request.body.agents)
-      ? await AgentService.reassignAgents(
-          soClient,
-          esClient,
-          { agentIds: request.body.agents },
-          request.body.policy_id
-        )
-      : await AgentService.reassignAgents(
-          soClient,
-          esClient,
-          { kuery: request.body.agents },
-          request.body.policy_id
-        );
-    const body: PostBulkAgentReassignResponse = result.saved_objects.reduce((acc, so) => {
+    const results = await AgentService.reassignAgents(
+      soClient,
+      esClient,
+      Array.isArray(request.body.agents)
+        ? { agentIds: request.body.agents }
+        : { kuery: request.body.agents },
+      request.body.policy_id
+    );
+
+    const body: PostBulkAgentReassignResponse = results.items.reduce((acc, so) => {
       return {
         ...acc,
         [so.id]: {

x-pack/plugins/fleet/server/routes/agent/upgrade_handler.ts

@@ -8,25 +8,21 @@
 import { RequestHandler } from 'src/core/server';
 import { TypeOf } from '@kbn/config-schema';
 import semverCoerce from 'semver/functions/coerce';
-import {
-  AgentSOAttributes,
-  PostAgentUpgradeResponse,
-  PostBulkAgentUpgradeResponse,
-} from '../../../common/types';
+import { PostAgentUpgradeResponse, PostBulkAgentUpgradeResponse } from '../../../common/types';
 import { PostAgentUpgradeRequestSchema, PostBulkAgentUpgradeRequestSchema } from '../../types';
 import * as AgentService from '../../services/agents';
 import { appContextService } from '../../services';
 import { defaultIngestErrorHandler } from '../../errors';
-import { AGENT_SAVED_OBJECT_TYPE } from '../../constants';
-import { savedObjectToAgent } from '../../services/agents/saved_objects';
 import { isAgentUpgradeable } from '../../../common/services';
+import { getAgent } from '../../services/agents';
 
 export const postAgentUpgradeHandler: RequestHandler<
   TypeOf<typeof PostAgentUpgradeRequestSchema.params>,
   undefined,
   TypeOf<typeof PostAgentUpgradeRequestSchema.body>
 > = async (context, request, response) => {
   const soClient = context.core.savedObjects.client;
+  const esClient = context.core.elasticsearch.client.asInternalUser;
   const { version, source_uri: sourceUri, force } = request.body;
   const kibanaVersion = appContextService.getKibanaVersion();
   try {
@@ -39,12 +35,8 @@ export const postAgentUpgradeHandler: RequestHandler<
       },
     });
   }
-
-  const agentSO = await soClient.get<AgentSOAttributes>(
-    AGENT_SAVED_OBJECT_TYPE,
-    request.params.agentId
-  );
-  if (agentSO.attributes.unenrollment_started_at || agentSO.attributes.unenrolled_at) {
+  const agent = await getAgent(soClient, esClient, request.params.agentId);
+  if (agent.unenrollment_started_at || agent.unenrolled_at) {
     return response.customError({
       statusCode: 400,
       body: {
@@ -53,7 +45,6 @@ export const postAgentUpgradeHandler: RequestHandler<
     });
   }
 
-  const agent = savedObjectToAgent(agentSO);
   if (!force && !isAgentUpgradeable(agent, kibanaVersion)) {
     return response.customError({
       statusCode: 400,
@@ -66,6 +57,7 @@ export const postAgentUpgradeHandler: RequestHandler<
   try {
     await AgentService.sendUpgradeAgentAction({
       soClient,
+      esClient,
       agentId: request.params.agentId,
       version,
       sourceUri,

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

@@ -488,17 +488,17 @@ class AgentPolicyService {
     soClient: SavedObjectsClientContract,
     agentPolicyId: string
   ) {
-    return appContextService.getConfig()?.agents.fleetServerEnabled
-      ? this.createFleetPolicyChangeFleetServer(
-          soClient,
-          appContextService.getInternalUserESClient(),
-          agentPolicyId
-        )
-      : this.createFleetPolicyChangeActionSO(soClient, agentPolicyId);
+    const esClient = appContextService.getInternalUserESClient();
+    if (appContextService.getConfig()?.agents?.fleetServerEnabled) {
+      await this.createFleetPolicyChangeFleetServer(soClient, esClient, agentPolicyId);
+    }
+
+    return this.createFleetPolicyChangeActionSO(soClient, esClient, agentPolicyId);
   }
 
   public async createFleetPolicyChangeActionSO(
     soClient: SavedObjectsClientContract,
+    esClient: ElasticsearchClient,
     agentPolicyId: string
   ) {
     // If Agents is not setup skip the creation of POLICY_CHANGE agent actions
@@ -518,7 +518,7 @@ class AgentPolicyService {
       return acc;
     }, []);
 
-    await createAgentPolicyAction(soClient, {
+    await createAgentPolicyAction(soClient, esClient, {
       type: 'POLICY_CHANGE',
       data: { policy },
       ack_data: { packages },