fix: aggregator should return 413 (Content Too Large) to make ingestor ARC work properly (#461)

500 errors will cause ingestors to slow down due to ARC (https://vector.dev/docs/reference/configuration/sinks/http/#request.adaptive_concurrency), which is good.
However, for case like 413 (due to aggregator payload size limit), ingestors shouldn't slow down and be blocked.
The fix here is to return the original status code so ARC will tune up the concurrency properly

Before:
Events buffer will be queued up and ingestors will be blocked (concurrency stays)

<img width="1170" alt="Screenshot 2024-07-08 at 4 46 10 PM" src="https://github.com/hyperdxio/hyperdx/assets/5959690/a149092a-b062-4508-b62b-a85116ac74b2">


Now:
Events buffer will be cleared out properly (concurrency increases) 

<img width="1237" alt="Screenshot 2024-07-08 at 4 43 22 PM" src="https://github.com/hyperdxio/hyperdx/assets/5959690/7871593b-0e2f-4a1e-960c-c27d90869514">

TODO: we want to revisit the error global handler to make sure the proper status code is returned

Author: wrn14897

.changeset/honest-ears-allow.md

@@ -0,0 +1,7 @@
+---
+'@hyperdx/api': patch
+'@hyperdx/app': patch
+---
+
+fix: aggregator should return 413 (Content Too Large) to make ingestor ARC work
+properly

docker-compose.ci.yml

@@ -81,6 +81,7 @@ services:
     # ports:
     #   - 9000:9000
     environment:
+      AGGREGATOR_PAYLOAD_SIZE_LIMIT: '64mb'
       APP_TYPE: 'api'
       CLICKHOUSE_HOST: http://ch_server:8123
       CLICKHOUSE_PASSWORD: api

packages/api/src/aggregator-app.ts

@@ -1,3 +1,4 @@
+import { recordException } from '@hyperdx/node-opentelemetry';
 import compression from 'compression';
 import type { NextFunction, Request, Response } from 'express';
 import express from 'express';
@@ -10,6 +11,10 @@ import routers from './routers/aggregator';
 import { BaseError, StatusCode } from './utils/errors';
 import logger, { expressLogger } from './utils/logger';
 
+if (!config.AGGREGATOR_PAYLOAD_SIZE_LIMIT) {
+  throw new Error('AGGREGATOR_PAYLOAD_SIZE_LIMIT is not defined');
+}
+
 const app: express.Application = express();
 
 const healthCheckMiddleware = async (
@@ -51,13 +56,35 @@ app.use('/', healthCheckMiddleware, routers.rootRouter);
 // ---------------------------------------------------------
 
 // error handling
-app.use((err: BaseError, _: Request, res: Response, next: NextFunction) => {
-  logger.error({
-    location: 'appErrorHandler',
-    error: serializeError(err),
-  });
-  // WARNING: should always return 500 so the ingestor will queue logs
-  res.status(StatusCode.INTERNAL_SERVER).send('Something broke!');
-});
+app.use(
+  (
+    err: BaseError & { status?: number },
+    _: Request,
+    res: Response,
+    next: NextFunction,
+  ) => {
+    void recordException(err, {
+      mechanism: {
+        type: 'generic',
+        handled: false,
+      },
+    });
+
+    // TODO: REPLACED WITH recordException once we enable tracing SDK
+    logger.error({
+      location: 'appErrorHandler',
+      error: serializeError(err),
+    });
+
+    // TODO: for all non 5xx errors
+    if (err.status === StatusCode.CONTENT_TOO_LARGE) {
+      // WARNING: return origin status code so the ingestor will tune up the adaptive concurrency properly
+      return res.sendStatus(err.status);
+    }
+
+    // WARNING: should always return 500 so the ingestor will queue logs
+    res.status(StatusCode.INTERNAL_SERVER).send('Something broke!');
+  },
+);
 
 export default app;

packages/api/src/utils/errors.ts

@@ -1,6 +1,7 @@
 export enum StatusCode {
   BAD_REQUEST = 400,
   CONFLICT = 409,
+  CONTENT_TOO_LARGE = 413,
   FORBIDDEN = 403,
   INTERNAL_SERVER = 500,
   NOT_FOUND = 404,