[Ingest Manager] add upgrade action (#77412)

* at agents/agent-id/upgrade endpoint

* handle upgrade ack

* let upgrade endpoint accept url and version

* wrap action data in data prop

* decrypt data of actions

* type upgrade action and decrypt data in ack

* error if trying to update to diff version of kibana

* add some integration tests

* untype

* fix test

* update integration test

* reset upgraded_at when upgrading

* use defaultIngestErrorHandler

* use ack_data instead of data

* copy data to ack_data

x-pack/plugins/ingest_manager/common/constants/routes.ts

@@ -90,6 +90,7 @@ export const AGENT_API_ROUTES = {
   REASSIGN_PATTERN: `${FLEET_API_ROOT}/agents/{agentId}/reassign`,
   BULK_REASSIGN_PATTERN: `${FLEET_API_ROOT}/agents/bulk_reassign`,
   STATUS_PATTERN: `${FLEET_API_ROOT}/agent-status`,
+  UPGRADE_PATTERN: `${FLEET_API_ROOT}/agents/{agentId}/upgrade`,
 };
 
 export const ENROLLMENT_API_KEY_ROUTES = {

x-pack/plugins/ingest_manager/common/services/agent_status.ts

@@ -19,6 +19,9 @@ export function getAgentStatus(agent: Agent, now: number = Date.now()): AgentSta
   if (!agent.last_checkin) {
     return 'enrolling';
   }
+  if (agent.upgrade_started_at && !agent.upgraded_at) {
+    return 'upgrading';
+  }
 
   const msLastCheckIn = new Date(lastCheckIn || 0).getTime();
   const msSinceLastCheckIn = new Date().getTime() - msLastCheckIn;

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

@@ -19,10 +19,10 @@ export type AgentStatus =
   | 'warning'
   | 'enrolling'
   | 'unenrolling'
+  | 'upgrading'
   | 'degraded';
 
-export type AgentActionType = 'CONFIG_CHANGE' | 'UNENROLL';
-
+export type AgentActionType = 'CONFIG_CHANGE' | 'UNENROLL' | 'UPGRADE';
 export interface NewAgentAction {
   type: AgentActionType;
   data?: any;
@@ -65,7 +65,6 @@ export type AgentPolicyActionSOAttributes = CommonAgentActionSOAttributes & {
   policy_id: string;
   policy_revision: number;
 };
-
 export type BaseAgentActionSOAttributes = AgentActionSOAttributes | AgentPolicyActionSOAttributes;
 
 export interface NewAgentEvent {
@@ -110,6 +109,8 @@ interface AgentBase {
   enrolled_at: string;
   unenrolled_at?: string;
   unenrollment_started_at?: string;
+  upgraded_at?: string;
+  upgrade_started_at?: string;
   shared_id?: string;
   access_api_key_id?: string;
   default_api_key?: string;

x-pack/plugins/ingest_manager/common/types/rest_spec/agent.ts

@@ -113,13 +113,20 @@ export interface PostAgentUnenrollRequest {
 // eslint-disable-next-line @typescript-eslint/no-empty-interface
 export interface PostAgentUnenrollResponse {}
 
+export interface PostAgentUpgradeRequest {
+  params: {
+    agentId: string;
+  };
+}
 export interface PostBulkAgentUnenrollRequest {
   body: {
     agents: string[] | string;
     force?: boolean;
   };
 }
 
+// eslint-disable-next-line @typescript-eslint/no-empty-interface
+export interface PostAgentUpgradeResponse {}
 // eslint-disable-next-line @typescript-eslint/no-empty-interface
 export interface PostBulkAgentUnenrollResponse {}
 

x-pack/plugins/ingest_manager/server/routes/agent/index.ts

@@ -29,6 +29,7 @@ import {
   PutAgentReassignRequestSchema,
   PostBulkAgentReassignRequestSchema,
   PostAgentEnrollRequestBodyJSONSchema,
+  PostAgentUpgradeRequestSchema,
 } from '../../types';
 import {
   getAgentsHandler,
@@ -48,6 +49,7 @@ import { postNewAgentActionHandlerBuilder } from './actions_handlers';
 import { appContextService } from '../../services';
 import { postAgentUnenrollHandler, postBulkAgentsUnenrollHandler } from './unenroll_handler';
 import { IngestManagerConfigType } from '../..';
+import { postAgentUpgradeHandler } from './upgrade_handler';
 
 const ajv = new Ajv({
   coerceTypes: true,
@@ -215,7 +217,15 @@ export const registerRoutes = (router: IRouter, config: IngestManagerConfigType)
     },
     getAgentStatusForAgentPolicyHandler
   );
-
+  // upgrade agent
+  router.post(
+    {
+      path: AGENT_API_ROUTES.UPGRADE_PATTERN,
+      validate: PostAgentUpgradeRequestSchema,
+      options: { tags: [`access:${PLUGIN_ID}-all`] },
+    },
+    postAgentUpgradeHandler
+  );
   // Bulk reassign
   router.post(
     {

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

@@ -0,0 +1,47 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License;
+ * you may not use this file except in compliance with the Elastic License.
+ */
+
+import { RequestHandler } from 'src/core/server';
+import { TypeOf } from '@kbn/config-schema';
+import { PostAgentUpgradeResponse } from '../../../common/types';
+import { PostAgentUpgradeRequestSchema } from '../../types';
+import * as AgentService from '../../services/agents';
+import { appContextService } from '../../services';
+import { defaultIngestErrorHandler } from '../../errors';
+
+export const postAgentUpgradeHandler: RequestHandler<
+  TypeOf<typeof PostAgentUpgradeRequestSchema.params>,
+  undefined,
+  TypeOf<typeof PostAgentUpgradeRequestSchema.body>
+> = async (context, request, response) => {
+  const soClient = context.core.savedObjects.client;
+  const { version, source_uri: sourceUri } = request.body;
+
+  // temporarily only allow upgrading to the same version as the installed kibana version
+  const kibanaVersion = appContextService.getKibanaVersion();
+  if (kibanaVersion !== version) {
+    return response.customError({
+      statusCode: 400,
+      body: {
+        message: `cannot upgrade agent to ${version} because it is different than the installed kibana version ${kibanaVersion}`,
+      },
+    });
+  }
+
+  try {
+    await AgentService.sendUpgradeAgentAction({
+      soClient,
+      agentId: request.params.agentId,
+      version,
+      sourceUri,
+    });
+
+    const body: PostAgentUpgradeResponse = {};
+    return response.ok({ body });
+  } catch (error) {
+    return defaultIngestErrorHandler({ error, response });
+  }
+};

x-pack/plugins/ingest_manager/server/saved_objects/index.ts

@@ -68,6 +68,8 @@ const savedObjectTypes: { [key: string]: SavedObjectsType } = {
         enrolled_at: { type: 'date' },
         unenrolled_at: { type: 'date' },
         unenrollment_started_at: { type: 'date' },
+        upgraded_at: { type: 'date' },
+        upgrade_started_at: { type: 'date' },
         access_api_key_id: { type: 'keyword' },
         version: { type: 'keyword' },
         user_provided_metadata: { type: 'flattened' },

x-pack/plugins/ingest_manager/server/services/agents/acks.ts

@@ -28,6 +28,7 @@ import {
 } from '../../constants';
 import { getAgentActionByIds } from './actions';
 import { forceUnenrollAgent } from './unenroll';
+import { ackAgentUpgraded } from './upgrade';
 
 const ALLOWED_ACKNOWLEDGEMENT_TYPE: string[] = ['ACTION_RESULT'];
 
@@ -80,6 +81,11 @@ export async function acknowledgeAgentActions(
     await forceUnenrollAgent(soClient, agent.id);
   }
 
+  const upgradeAction = actions.find((action) => action.type === 'UPGRADE');
+  if (upgradeAction) {
+    await ackAgentUpgraded(soClient, upgradeAction);
+  }
+
   const configChangeAction = getLatestConfigChangePolicyActionIfUpdated(agent, actions);
 
   await soClient.bulkUpdate<AgentSOAttributes | AgentActionSOAttributes>([

x-pack/plugins/ingest_manager/server/services/agents/actions.ts

@@ -225,7 +225,11 @@ export async function getAgentPolicyActionByIds(
   );
 }
 
-export async function getNewActionsSince(soClient: SavedObjectsClientContract, timestamp: string) {
+export async function getNewActionsSince(
+  soClient: SavedObjectsClientContract,
+  timestamp: string,
+  decryptData: boolean = true
+) {
   const filter = nodeTypes.function.buildNode('and', [
     nodeTypes.function.buildNode(
       'not',
@@ -243,14 +247,33 @@ export async function getNewActionsSince(soClient: SavedObjectsClientContract, t
       }
     ),
   ]);
-  const res = await soClient.find<AgentActionSOAttributes>({
-    type: AGENT_ACTION_SAVED_OBJECT_TYPE,
-    filter,
-  });
 
-  return res.saved_objects
+  const actions = (
+    await soClient.find<AgentActionSOAttributes>({
+      type: AGENT_ACTION_SAVED_OBJECT_TYPE,
+      filter,
+    })
+  ).saved_objects
     .filter(isAgentActionSavedObject)
     .map((so) => savedObjectToAgentAction(so));
+
+  if (!decryptData) {
+    return actions;
+  }
+
+  return await Promise.all(
+    actions.map(async (action) => {
+      // Get decrypted actions
+      return savedObjectToAgentAction(
+        await appContextService
+          .getEncryptedSavedObjects()
+          .getDecryptedAsInternalUser<AgentActionSOAttributes>(
+            AGENT_ACTION_SAVED_OBJECT_TYPE,
+            action.id
+          )
+      );
+    })
+  );
 }
 
 export async function getLatestConfigChangeAction(

x-pack/plugins/ingest_manager/server/services/agents/index.ts

@@ -9,6 +9,7 @@ export * from './events';
 export * from './checkin';
 export * from './enroll';
 export * from './unenroll';
+export * from './upgrade';
 export * from './status';
 export * from './crud';
 export * from './update';

x-pack/plugins/ingest_manager/server/services/agents/upgrade.ts

@@ -0,0 +1,61 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License;
+ * you may not use this file except in compliance with the Elastic License.
+ */
+
+import { SavedObjectsClientContract } from 'src/core/server';
+import { AgentSOAttributes, AgentAction, AgentActionSOAttributes } from '../../types';
+import { AGENT_ACTION_SAVED_OBJECT_TYPE, AGENT_SAVED_OBJECT_TYPE } from '../../constants';
+import { createAgentAction } from './actions';
+
+export async function sendUpgradeAgentAction({
+  soClient,
+  agentId,
+  version,
+  sourceUri,
+}: {
+  soClient: SavedObjectsClientContract;
+  agentId: string;
+  version: string;
+  sourceUri: string;
+}) {
+  const now = new Date().toISOString();
+  const data = {
+    version,
+    source_uri: sourceUri,
+  };
+  await createAgentAction(soClient, {
+    agent_id: agentId,
+    created_at: now,
+    data,
+    ack_data: data,
+    type: 'UPGRADE',
+  });
+  await soClient.update<AgentSOAttributes>(AGENT_SAVED_OBJECT_TYPE, agentId, {
+    upgraded_at: undefined,
+    upgrade_started_at: now,
+  });
+}
+
+export async function ackAgentUpgraded(
+  soClient: SavedObjectsClientContract,
+  agentAction: AgentAction
+) {
+  const {
+    attributes: { ack_data: ackData },
+  } = await soClient.get<AgentActionSOAttributes>(AGENT_ACTION_SAVED_OBJECT_TYPE, agentAction.id);
+  if (!ackData) throw new Error('data missing from UPGRADE action');
+  const { version } = JSON.parse(ackData);
+  if (!version) throw new Error('version missing from UPGRADE action');
+  await soClient.update<AgentSOAttributes>(AGENT_SAVED_OBJECT_TYPE, agentAction.agent_id, {
+    upgraded_at: new Date().toISOString(),
+    local_metadata: {
+      elastic: {
+        agent: {
+          version,
+        },
+      },
+    },
+  });
+}

x-pack/plugins/ingest_manager/server/types/models/agent.ts

@@ -62,7 +62,12 @@ export const AgentEventSchema = schema.object({
 });
 
 export const NewAgentActionSchema = schema.object({
-  type: schema.oneOf([schema.literal('CONFIG_CHANGE'), schema.literal('UNENROLL')]),
+  type: schema.oneOf([
+    schema.literal('CONFIG_CHANGE'),
+    schema.literal('UNENROLL'),
+    schema.literal('UPGRADE'),
+  ]),
   data: schema.maybe(schema.any()),
+  ack_data: schema.maybe(schema.any()),
   sent_at: schema.maybe(schema.string()),
 });

x-pack/plugins/ingest_manager/server/types/rest_spec/agent.ts

@@ -172,6 +172,16 @@ export const PostAgentUnenrollRequestSchema = {
   ),
 };
 
+export const PostAgentUpgradeRequestSchema = {
+  params: schema.object({
+    agentId: schema.string(),
+  }),
+  body: schema.object({
+    source_uri: schema.string(),
+    version: schema.string(),
+  }),
+};
+
 export const PostBulkAgentUnenrollRequestSchema = {
   body: schema.object({
     agents: schema.oneOf([schema.arrayOf(schema.string()), schema.string()]),

x-pack/test/ingest_manager_api_integration/apis/fleet/agents/acks.ts

@@ -13,6 +13,7 @@ export default function (providerContext: FtrProviderContext) {
   const { getService } = providerContext;
   const esArchiver = getService('esArchiver');
   const esClient = getService('es');
+  const kibanaServer = getService('kibanaServer');
 
   const supertest = getSupertestWithoutAuth(providerContext);
   let apiKey: { id: string; api_key: string };
@@ -205,5 +206,50 @@ export default function (providerContext: FtrProviderContext) {
         'ACTION not allowed for acknowledgment only ACTION_RESULT'
       );
     });
+
+    it('ack upgrade should update fleet-agent SO', async () => {
+      const { body: actionRes } = await supertest
+        .post(`/api/ingest_manager/fleet/agents/agent1/actions`)
+        .set('kbn-xsrf', 'xx')
+        .set(
+          'Authorization',
+          `ApiKey ${Buffer.from(`${apiKey.id}:${apiKey.api_key}`).toString('base64')}`
+        )
+        .send({
+          action: {
+            type: 'UPGRADE',
+            ack_data: { version: '8.0.0', source_uri: 'http://localhost:8000' },
+          },
+        })
+        .expect(200);
+      const actionId = actionRes.item.id;
+      await supertest
+        .post(`/api/ingest_manager/fleet/agents/agent1/acks`)
+        .set('kbn-xsrf', 'xx')
+        .set(
+          'Authorization',
+          `ApiKey ${Buffer.from(`${apiKey.id}:${apiKey.api_key}`).toString('base64')}`
+        )
+        .send({
+          events: [
+            {
+              type: 'ACTION_RESULT',
+              subtype: 'ACKNOWLEDGED',
+              timestamp: '2020-09-21T13:25:29.02838-04:00',
+              action_id: actionId,
+              agent_id: 'agent1',
+              message:
+                "Action '70d97288-ffd9-4549-8c49-2423a844f67f' of type 'UPGRADE' acknowledged.",
+            },
+          ],
+        })
+        .expect(200);
+
+      const res = await kibanaServer.savedObjects.get({
+        type: 'fleet-agents',
+        id: 'agent1',
+      });
+      expect(res.attributes.upgraded_at).to.be.ok();
+    });
   });
 }