Metering connector actions request body bytes (elastic#186804)

Towards: elastic/response-ops-team#209

This PR collects body-bytes from the requests made by the connectors to
the 3rd parties and saves them in the event-log.

There is a new metric collector: `ConnectorMetricsCollector`
Action TaskRunner, creates a new instance of it on each action execution
and passes it to the actionType.executor.
Then the actionType.executor passes it to the request function provided
by the actions plugin.
Request function passes the response (or the error) from axios to
`addRequestBodyBytes` method of the `ConnectorMetricsCollector`.
Since axios always returns `request.headers['Content-Length']` either in
success result or error, metric collector uses its value to get the
request body bytes.

In case there is no `Content-Length` header, `addRequestBodyBytes`
method fallbacks to the body object that we pass as the second param. So
It calculates the body bytes by using `Buffer.byteLength(body,
'utf8');`, which is also used by axios to populate
`request.headers['Content-Length']`

For the connectors or the subActions that we don't use the request
function or axios:
addRequestBodyBytes method is called just before making the request only
with the body param in order to force it to use the fallback.

Note: If there are more than one requests in an execution, the bytes are
summed.

## To verify:
Create a rule with a connector that you would like to test.
Let the rule run and check the event-log of your connector, request body
bytes should be saved in:
`kibana.action.execution.metrics.request_body_bytes`

Alternatively:
You can create a connector and run it on its test tab.

You can use the below query to check the event-log:
```
 {
    "query": {
      "bool": { 
        "must": [
          { "match": { "event.provider":"actions"}},
          { "match": { "kibana.action.type_id":"{**your-action-type-id**}"}} 
        ],
        "filter": [ 
          { "term":  { "event.action": "execute" }}
        ]
      }
    },
    "size" : 100
}
```

x-pack/plugins/actions/server/lib/action_executor.test.ts

@@ -4,7 +4,6 @@
  * 2.0; you may not use this file except in compliance with the Elastic License
  * 2.0.
  */
-
 import { KibanaRequest } from '@kbn/core/server';
 import { schema } from '@kbn/config-schema';
 import { ActionExecutor } from './action_executor';
@@ -18,7 +17,7 @@ import {
 } from '@kbn/core/server/mocks';
 import { eventLoggerMock } from '@kbn/event-log-plugin/server/mocks';
 import { spacesServiceMock } from '@kbn/spaces-plugin/server/spaces_service/spaces_service.mock';
-import { ActionType as ConnectorType } from '../types';
+import { ActionType as ConnectorType, ConnectorUsageCollector } from '../types';
 import { actionsAuthorizationMock, actionsMock } from '../mocks';
 import {
   asBackgroundTaskExecutionSource,
@@ -150,6 +149,10 @@ const connectorSavedObject = {
   references: [],
 };
 
+interface ActionUsage {
+  request_body_bytes: number;
+}
+
 const getBaseExecuteStartEventLogDoc = (unsecured: boolean) => {
   return {
     event: {
@@ -163,6 +166,7 @@ const getBaseExecuteStartEventLogDoc = (unsecured: boolean) => {
         },
         id: CONNECTOR_ID,
         name: '1',
+        type_id: 'test',
       },
       ...(unsecured
         ? {}
@@ -190,10 +194,23 @@ const getBaseExecuteStartEventLogDoc = (unsecured: boolean) => {
   };
 };
 
-const getBaseExecuteEventLogDoc = (unsecured: boolean) => {
+const getBaseExecuteEventLogDoc = (
+  unsecured: boolean,
+  actionUsage: ActionUsage = { request_body_bytes: 0 }
+) => {
   const base = getBaseExecuteStartEventLogDoc(unsecured);
   return {
     ...base,
+    kibana: {
+      ...base.kibana,
+      action: {
+        ...base.kibana.action,
+        execution: {
+          ...base.kibana.action.execution,
+          usage: actionUsage,
+        },
+      },
+    },
     event: {
       ...base.event,
       action: 'execute',
@@ -211,9 +228,12 @@ const getBaseExecuteEventLogDoc = (unsecured: boolean) => {
   };
 };
 
+const mockGetRequestBodyByte = jest.spyOn(ConnectorUsageCollector.prototype, 'getRequestBodyByte');
+
 beforeEach(() => {
   jest.resetAllMocks();
   jest.clearAllMocks();
+  mockGetRequestBodyByte.mockReturnValue(0);
   spacesMock.getSpaceId.mockReturnValue('some-namespace');
   loggerMock.get.mockImplementation(() => loggerMock);
   const mockRealm = { name: 'default_native', type: 'native' };
@@ -237,6 +257,7 @@ describe('Action Executor', () => {
     const label = executeUnsecure ? 'executes unsecured' : 'executes';
 
     test(`successfully ${label}`, async () => {
+      mockGetRequestBodyByte.mockReturnValue(300);
       encryptedSavedObjectsClient.getDecryptedAsInternalUser.mockResolvedValueOnce(
         connectorSavedObject
       );
@@ -280,13 +301,15 @@ describe('Action Executor', () => {
         },
         params: { foo: true },
         logger: loggerMock,
+        connectorUsageCollector: expect.any(ConnectorUsageCollector),
       });
 
       expect(loggerMock.debug).toBeCalledWith('executing action test:1: 1');
       expect(eventLogger.logEvent).toHaveBeenCalledTimes(2);
 
       const execStartDoc = getBaseExecuteStartEventLogDoc(executeUnsecure);
-      const execDoc = getBaseExecuteEventLogDoc(executeUnsecure);
+      const execDoc = getBaseExecuteEventLogDoc(executeUnsecure, { request_body_bytes: 300 });
+
       expect(eventLogger.logEvent).toHaveBeenNthCalledWith(1, execStartDoc);
       expect(eventLogger.logEvent).toHaveBeenNthCalledWith(2, execDoc);
     });
@@ -353,13 +376,15 @@ describe('Action Executor', () => {
           params: { foo: true },
           logger: loggerMock,
           source: executionSource.source,
+          connectorUsageCollector: expect.any(ConnectorUsageCollector),
         });
 
         expect(loggerMock.debug).toBeCalledWith('executing action test:1: 1');
         expect(eventLogger.logEvent).toHaveBeenCalledTimes(2);
 
         const execStartDoc = getBaseExecuteStartEventLogDoc(executeUnsecure);
         const execDoc = getBaseExecuteEventLogDoc(executeUnsecure);
+
         expect(eventLogger.logEvent).toHaveBeenNthCalledWith(1, {
           ...execStartDoc,
           kibana: {
@@ -431,13 +456,15 @@ describe('Action Executor', () => {
         },
         params: { foo: true },
         logger: loggerMock,
+        connectorUsageCollector: expect.any(ConnectorUsageCollector),
       });
 
       expect(loggerMock.debug).toBeCalledWith('executing action test:preconfigured: Preconfigured');
       expect(eventLogger.logEvent).toHaveBeenCalledTimes(2);
 
       const execStartDoc = getBaseExecuteStartEventLogDoc(executeUnsecure);
       const execDoc = getBaseExecuteEventLogDoc(executeUnsecure);
+
       expect(eventLogger.logEvent).toHaveBeenNthCalledWith(1, {
         ...execStartDoc,
         kibana: {
@@ -513,6 +540,7 @@ describe('Action Executor', () => {
           params: { foo: true },
           logger: loggerMock,
           request: {},
+          connectorUsageCollector: expect.any(ConnectorUsageCollector),
         });
       }
 
@@ -532,6 +560,7 @@ describe('Action Executor', () => {
 
       const execStartDoc = getBaseExecuteStartEventLogDoc(executeUnsecure);
       const execDoc = getBaseExecuteEventLogDoc(executeUnsecure);
+
       expect(eventLogger.logEvent).toHaveBeenNthCalledWith(1, {
         ...execStartDoc,
         kibana: {
@@ -540,6 +569,7 @@ describe('Action Executor', () => {
             ...execStartDoc.kibana.action,
             id: 'system-connector-.cases',
             name: 'System action: .cases',
+            type_id: '.cases',
           },
           saved_objects: [
             {
@@ -569,6 +599,7 @@ describe('Action Executor', () => {
             ...execDoc.kibana.action,
             id: 'system-connector-.cases',
             name: 'System action: .cases',
+            type_id: '.cases',
           },
           saved_objects: [
             {
@@ -890,6 +921,7 @@ describe('Action Executor', () => {
         },
         params: { foo: true },
         logger: loggerMock,
+        connectorUsageCollector: expect.any(ConnectorUsageCollector),
       });
     });
 
@@ -921,6 +953,7 @@ describe('Action Executor', () => {
         params: { foo: true },
         logger: loggerMock,
         request: {},
+        connectorUsageCollector: expect.any(ConnectorUsageCollector),
       });
     });
 
@@ -989,13 +1022,15 @@ describe('Action Executor', () => {
         },
         params: { foo: true },
         logger: loggerMock,
+        connectorUsageCollector: expect.any(ConnectorUsageCollector),
       });
 
       expect(loggerMock.debug).toBeCalledWith('executing action test:preconfigured: Preconfigured');
       expect(eventLogger.logEvent).toHaveBeenCalledTimes(2);
 
       const execStartDoc = getBaseExecuteStartEventLogDoc(executeUnsecure);
       const execDoc = getBaseExecuteEventLogDoc(executeUnsecure);
+
       expect(eventLogger.logEvent).toHaveBeenNthCalledWith(1, {
         ...execStartDoc,
         kibana: {
@@ -1026,6 +1061,12 @@ describe('Action Executor', () => {
             ...execDoc.kibana.action,
             id: 'preconfigured',
             name: 'Preconfigured',
+            execution: {
+              ...execStartDoc.kibana.action.execution,
+              usage: {
+                request_body_bytes: 0,
+              },
+            },
           },
           saved_objects: [
             {
@@ -1074,6 +1115,7 @@ describe('Action Executor', () => {
         params: { foo: true },
         logger: loggerMock,
         request: {},
+        connectorUsageCollector: expect.any(ConnectorUsageCollector),
       });
 
       expect(loggerMock.debug).toBeCalledWith(
@@ -1083,6 +1125,7 @@ describe('Action Executor', () => {
 
       const execStartDoc = getBaseExecuteStartEventLogDoc(executeUnsecure);
       const execDoc = getBaseExecuteEventLogDoc(executeUnsecure);
+
       expect(eventLogger.logEvent).toHaveBeenNthCalledWith(1, {
         ...execStartDoc,
         kibana: {
@@ -1091,6 +1134,7 @@ describe('Action Executor', () => {
             ...execStartDoc.kibana.action,
             id: 'system-connector-.cases',
             name: 'System action: .cases',
+            type_id: '.cases',
           },
           saved_objects: [
             {
@@ -1120,6 +1164,7 @@ describe('Action Executor', () => {
             ...execDoc.kibana.action,
             id: 'system-connector-.cases',
             name: 'System action: .cases',
+            type_id: '.cases',
           },
           saved_objects: [
             {
@@ -1290,6 +1335,7 @@ describe('Action Executor', () => {
           },
           params: { foo: true },
           logger: loggerMock,
+          connectorUsageCollector: expect.any(ConnectorUsageCollector),
         });
       }
     });
@@ -1385,6 +1431,7 @@ describe('Event log', () => {
           },
           name: undefined,
           id: 'action1',
+          type_id: 'test',
         },
         alert: {
           rule: {
@@ -1430,6 +1477,7 @@ describe('Event log', () => {
           },
           name: 'action-1',
           id: '1',
+          type_id: 'test',
         },
         alert: {
           rule: {
@@ -1483,6 +1531,7 @@ describe('Event log', () => {
           },
           name: 'action-1',
           id: '1',
+          type_id: 'test',
         },
         alert: {
           rule: {
@@ -1559,9 +1608,13 @@ describe('Event log', () => {
             gen_ai: {
               usage: mockGenAi.usage,
             },
+            usage: {
+              request_body_bytes: 0,
+            },
           },
           name: 'action-1',
           id: '1',
+          type_id: '.gen-ai',
         },
         alert: {
           rule: {
@@ -1655,9 +1708,13 @@ describe('Event log', () => {
                 total_tokens: 35,
               },
             },
+            usage: {
+              request_body_bytes: 0,
+            },
           },
           name: 'action-1',
           id: '1',
+          type_id: '.gen-ai',
         },
         alert: {
           rule: {

x-pack/plugins/actions/server/lib/action_executor.ts

@@ -23,6 +23,7 @@ import { IEventLogger, SAVED_OBJECT_REL_PRIMARY } from '@kbn/event-log-plugin/se
 import { createTaskRunError, TaskErrorSource } from '@kbn/task-manager-plugin/server';
 import { getErrorSource } from '@kbn/task-manager-plugin/server/task_running';
 import { GEN_AI_TOKEN_COUNT_EVENT } from './event_based_telemetry';
+import { ConnectorUsageCollector } from '../usage/connector_usage_collector';
 import { getGenAiTokenTracking, shouldTrackGenAiToken } from './gen_ai_token_tracking';
 import {
   validateConfig,
@@ -293,6 +294,7 @@ export class ActionExecutor {
       actionExecutionId,
       isInMemory: this.actionInfo.isInMemory,
       ...(source ? { source } : {}),
+      actionTypeId: this.actionInfo.actionTypeId,
     });
 
     eventLogger.logEvent(event);
@@ -394,6 +396,14 @@ export class ActionExecutor {
 
         const { actionTypeId, name, config, secrets } = actionInfo;
 
+        const loggerId = actionTypeId.startsWith('.') ? actionTypeId.substring(1) : actionTypeId;
+        const logger = this.actionExecutorContext!.logger.get(loggerId);
+
+        const connectorUsageCollector = new ConnectorUsageCollector({
+          logger,
+          connectorId: actionId,
+        });
+
         if (!this.actionInfo || this.actionInfo.actionId !== actionId) {
           this.actionInfo = actionInfo;
         }
@@ -434,9 +444,6 @@ export class ActionExecutor {
           return err.result;
         }
 
-        const loggerId = actionTypeId.startsWith('.') ? actionTypeId.substring(1) : actionTypeId;
-        const logger = this.actionExecutorContext!.logger.get(loggerId);
-
         if (span) {
           span.name = `${executeLabel} ${actionTypeId}`;
           span.addLabels({
@@ -477,6 +484,7 @@ export class ActionExecutor {
           actionExecutionId,
           isInMemory: this.actionInfo.isInMemory,
           ...(source ? { source } : {}),
+          actionTypeId,
         });
 
         eventLogger.startTiming(event);
@@ -510,6 +518,7 @@ export class ActionExecutor {
             logger,
             source,
             ...(actionType.isSystemActionType ? { request } : {}),
+            connectorUsageCollector,
           });
 
           if (rawResult && rawResult.status === 'error') {
@@ -548,6 +557,11 @@ export class ActionExecutor {
           event.user = event.user || {};
           event.user.name = currentUser?.username;
           event.user.id = currentUser?.profile_uid;
+          set(
+            event,
+            'kibana.action.execution.usage.request_body_bytes',
+            connectorUsageCollector.getRequestBodyByte()
+          );
 
           if (result.status === 'ok') {
             span?.setOutcome('success');