feat: Add concurrency setting for OpenTelemetry Ingest worker and update related configurations

HelmChart/Public/oneuptime/templates/open-telemetry-ingest.yaml

@@ -93,6 +93,8 @@ spec:
               value: {{ $.Values.openTelemetryIngest.ports.http | quote }} 
             - name: DISABLE_TELEMETRY
               value: {{ $.Values.openTelemetryIngest.disableTelemetryCollection | quote }}
+            - name: OPEN_TELEMETRY_INGEST_CONCURRENCY
+              value: {{ $.Values.openTelemetryIngest.concurrency | squote }}
           ports:
             - containerPort: {{ $.Values.openTelemetryIngest.ports.http }}
               protocol: TCP

HelmChart/Public/oneuptime/values.yaml

@@ -537,6 +537,8 @@ openTelemetryIngest:
   replicaCount: 1
   disableTelemetryCollection: false
   disableAutoscaler: false
+  # Max concurrent telemetry jobs processed by each pod
+  concurrency: 100
   ports:
     http: 3403
   resources:

OpenTelemetryIngest/Config.ts

@@ -0,0 +1,9 @@
+let concurrency: string | number =
+  process.env["OPEN_TELEMETRY_INGEST_CONCURRENCY"] || 100;
+
+if (typeof concurrency === "string") {
+  const parsed: number = parseInt(concurrency, 10);
+  concurrency = !isNaN(parsed) && parsed > 0 ? parsed : 100;
+}
+
+export const OPEN_TELEMETRY_INGEST_CONCURRENCY: number = concurrency as number;

OpenTelemetryIngest/Jobs/TelemetryIngest/ProcessTelemetry.ts

@@ -8,6 +8,7 @@ import logger from "Common/Server/Utils/Logger";
 import { QueueJob, QueueName } from "Common/Server/Infrastructure/Queue";
 import QueueWorker from "Common/Server/Infrastructure/QueueWorker";
 import ObjectID from "Common/Types/ObjectID";
+import { OPEN_TELEMETRY_INGEST_CONCURRENCY } from "../../Config";
 
 // Set up the unified worker for processing telemetry queue
 QueueWorker.getWorker(
@@ -58,7 +59,7 @@ QueueWorker.getWorker(
       throw error;
     }
   },
-  { concurrency: 50 }, // Process up to 50 telemetry jobs concurrently
+  { concurrency: OPEN_TELEMETRY_INGEST_CONCURRENCY },
 );
 
 logger.debug("Unified telemetry worker initialized");

config.example.env

@@ -234,6 +234,11 @@ WORKFLOW_SCRIPT_TIMEOUT_IN_MS=5000
 WORKFLOW_TIMEOUT_IN_MS=5000
 
 
+# Concurrency settings
+# Max number of telemetry jobs processed concurrently by OpenTelemetry Ingest worker
+OPEN_TELEMETRY_INGEST_CONCURRENCY=100
+
+
 
 # Lets encrypt notification email. This email will be used when certs are about to expire
 LETS_ENCRYPT_NOTIFICATION_EMAIL=

docker-compose.base.yml

@@ -482,6 +482,8 @@ services:
             <<: *common-server-variables
             PORT: ${OPEN_TELEMETRY_INGEST_PORT}
             DISABLE_TELEMETRY: ${DISABLE_TELEMETRY_FOR_OPEN_TELEMETRY_INGEST}
+            # Max concurrent telemetry jobs the worker will process
+            OPEN_TELEMETRY_INGEST_CONCURRENCY: ${OPEN_TELEMETRY_INGEST_CONCURRENCY}
         logging:
             driver: "local"
             options: