feat(opentelemetry-ingest): add native Syslog ingestion, parsing, queuing, docs & tests

- Add /syslog/v1/logs endpoint and syslog product-type middleware
- Implement SyslogIngestService: normalize/parse messages, build attributes, batch flush to LogService
- Add robust Syslog parser (RFC5424 & RFC3164) and comprehensive unit tests
- Add TelemetryType.Syslog, SyslogQueueService, and queue handling (enqueue + worker processing)
- Expose OPEN_TELEMETRY_INGEST_SYSLOG_FLUSH_BATCH_SIZE config
- Update Otel ingest router, base service helpers, and ProcessTelemetry worker to support Syslog
- Add documentation page and navigation entry for Syslog telemetry

Docs/Content/telemetry/syslog.md

@@ -0,0 +1,106 @@
+# Send Syslog Data to OneUptime
+
+## Overview
+
+The OpenTelemetry Ingest service now accepts native Syslog payloads. You can forward messages from any RFC3164 or RFC5424 compatible source directly to OneUptime over HTTPS. OneUptime parses the syslog priority, facility, severity, structured data, and message body before storing everything as searchable logs.
+
+## Prerequisites
+
+- **Telemetry Ingestion Token** – create one from *Project Settings → Telemetry Ingestion Keys* and copy the `x-oneuptime-token` value.
+- **Syslog forwarder** – any tool capable of sending HTTP POST requests (for example `curl`, `rsyslog` via `omhttp`, or `syslog-ng` with the HTTP destination plugin).
+- **Service name (optional)** – set the `x-oneuptime-service-name` header to group incoming logs under a specific telemetry service. When omitted, OneUptime falls back to the syslog `APP-NAME`, hostname, or `Syslog`.
+
+## Endpoint
+
+```
+POST https://oneuptime.com/syslog/v1/logs
+```
+
+- Replace `oneuptime.com` with your host if you are self hosting OneUptime.
+- Always include the `x-oneuptime-token` header in the request.
+
+## Request Body
+
+Send newline-delimited Syslog strings or a JSON payload with a `messages` array. Both RFC3164 (BSD) and RFC5424 formats are supported.
+
+```json
+{
+  "messages": [
+    "<34>1 2025-03-02T14:48:05.003Z web-01 nginx 7421 ID47 [env@32473 host=\"web-01\"] 502 on /api/login",
+    "<13>Feb  5 17:32:18 db-01 postgres[2419]: connection received from 10.0.0.12"
+  ]
+}
+```
+
+### Supported Content Types
+
+- `application/json` – recommended.
+- `text/plain` – newline separated messages.
+- `application/octet-stream` – raw payloads. Gzip compression (`Content-Encoding: gzip`) is also accepted.
+
+## Quick Test with curl
+
+```bash
+curl \
+  -X POST https://oneuptime.com/syslog/v1/logs \
+  -H "Content-Type: application/json" \
+  -H "x-oneuptime-token: YOUR_TELEMETRY_KEY" \
+  -H "x-oneuptime-service-name: production-web" \
+  -d '{
+    "messages": [
+      "<34>1 2025-03-02T14:48:05.003Z web-01 nginx 7421 ID47 [env@32473 host=\"web-01\"] 502 on /api/login"
+    ]
+  }'
+```
+
+## Forwarding from rsyslog
+
+1. Install the HTTP output module:
+   ```bash
+   sudo apt-get install rsyslog-omhttp
+   ```
+2. Append the destination to `/etc/rsyslog.d/oneuptime.conf`:
+   ```
+   module(load="omhttp")
+
+   template(name="OneUptimeJson" type="list") {
+     constant(value="{\"messages\":[\"")
+     property(name="rawmsg")
+     constant(value="\"]}")
+   }
+
+   action(
+     type="omhttp"
+     server="oneuptime.com"
+     serverport="443"
+     usehttps="on"
+     endpoint="/syslog/v1/logs"
+     header="Content-Type: application/json"
+     header="x-oneuptime-token: YOUR_TELEMETRY_KEY"
+     header="x-oneuptime-service-name: rsyslog-demo"
+     template="OneUptimeJson"
+   )
+   ```
+3. Restart rsyslog:
+   ```bash
+   sudo systemctl restart rsyslog
+   ```
+
+## Parsed Attributes
+
+OneUptime automatically adds the following attributes to each log entry:
+
+- `syslog.priority`, `syslog.facility.code`, `syslog.facility.name`
+- `syslog.severity.code`, `syslog.severity.name`
+- `syslog.hostname`, `syslog.appName`, `syslog.processId`, `syslog.messageId`
+- `syslog.structured.*` (flattened RFC5424 structured data)
+- `syslog.raw` (original message for traceability)
+
+These attributes become searchable inside the Telemetry → Logs explorer.
+
+## Troubleshooting
+
+- **HTTP 401 or empty results** – verify the `x-oneuptime-token` header belongs to the project receiving the logs.
+- **No logs appear** – confirm the request body actually contains syslog lines. Empty bodies are rejected with HTTP 400.
+- **Unexpected service name** – set `x-oneuptime-service-name` to override the default detection logic.
+- **Large bursts** – batching up to 1,000 lines per request is supported. Larger bursts are queued and processed asynchronously.

Docs/Utils/Nav.ts

@@ -189,6 +189,7 @@ const DocsNav: NavGroup[] = [
       { title: "OpenTelemetry", url: "/docs/telemetry/open-telemetry" },
       { title: "FluentBit", url: "/docs/telemetry/fluentbit" },
       { title: "Fluentd", url: "/docs/telemetry/fluentd" },
+      { title: "Syslog", url: "/docs/telemetry/syslog" },
     ],
   },
   {

OpenTelemetryIngest/API/OTelIngest.ts

@@ -1,4 +1,6 @@
-import TelemetryIngest from "Common/Server/Middleware/TelemetryIngest";
+import TelemetryIngest, {
+  TelemetryRequest,
+} from "Common/Server/Middleware/TelemetryIngest";
 import Express, {
   ExpressRequest,
   ExpressResponse,
@@ -10,12 +12,23 @@ import OpenTelemetryRequestMiddleware from "../Middleware/OtelRequestMiddleware"
 import OtelTracesIngestService from "../Services/OtelTracesIngestService";
 import OtelMetricsIngestService from "../Services/OtelMetricsIngestService";
 import OtelLogsIngestService from "../Services/OtelLogsIngestService";
+import SyslogIngestService from "../Services/SyslogIngestService";
 import TelemetryQueueService from "../Services/Queue/TelemetryQueueService";
 import ClusterKeyAuthorization from "Common/Server/Middleware/ClusterKeyAuthorization";
 import { JSONObject } from "Common/Types/JSON";
+import ProductType from "Common/Types/MeteredPlan/ProductType";
 
 const router: ExpressRouter = Express.getRouter();
 
+const syslogProductTypeMiddleware = (
+  req: ExpressRequest,
+  _res: ExpressResponse,
+  next: NextFunction,
+): void => {
+  (req as TelemetryRequest).productType = ProductType.Logs;
+  next();
+};
+
 /**
  *
  *  Otel Middleware
@@ -61,6 +74,19 @@ router.post(
   },
 );
 
+router.post(
+  "/syslog/v1/logs",
+  syslogProductTypeMiddleware,
+  TelemetryIngest.isAuthorizedServiceMiddleware,
+  async (
+    req: ExpressRequest,
+    res: ExpressResponse,
+    next: NextFunction,
+  ): Promise<void> => {
+    return SyslogIngestService.ingestSyslog(req, res, next);
+  },
+);
+
 // Queue stats endpoint
 router.get(
   "/otlp/queue/stats",

OpenTelemetryIngest/Jobs/TelemetryIngest/ProcessTelemetry.ts

@@ -5,6 +5,7 @@ import {
 import OtelLogsIngestService from "../../Services/OtelLogsIngestService";
 import OtelTracesIngestService from "../../Services/OtelTracesIngestService";
 import OtelMetricsIngestService from "../../Services/OtelMetricsIngestService";
+import SyslogIngestService from "../../Services/SyslogIngestService";
 import { TelemetryRequest } from "Common/Server/Middleware/TelemetryIngest";
 import logger from "Common/Server/Utils/Logger";
 import { QueueJob, QueueName } from "Common/Server/Infrastructure/Queue";
@@ -55,6 +56,13 @@ QueueWorker.getWorker(
           );
           break;
 
+        case TelemetryType.Syslog:
+          await SyslogIngestService.processSyslogFromQueue(mockRequest);
+          logger.debug(
+            `Successfully processed syslog payload for project: ${jobData.projectId}`,
+          );
+          break;
+
         default:
           throw new Error(`Unknown telemetry type: ${jobData.type}`);
       }

OpenTelemetryIngest/Services/OtelIngestBaseService.ts

@@ -32,4 +32,30 @@ export default abstract class OtelIngestBaseService {
 
     return "Unknown Service";
   }
+
+  @CaptureSpan()
+  protected static getServiceNameFromHeaders(
+    req: ExpressRequest,
+    defaultName: string = "Unknown Service",
+  ): string {
+    const headerValue: string | string[] | undefined = req.headers[
+      "x-oneuptime-service-name"
+    ];
+
+    if (typeof headerValue === "string" && headerValue.trim()) {
+      return headerValue.trim();
+    }
+
+    if (Array.isArray(headerValue) && headerValue.length > 0) {
+      const value: string = headerValue.find((item: string) => {
+        return item && item.trim();
+      }) as string;
+
+      if (value && value.trim()) {
+        return value.trim();
+      }
+    }
+
+    return defaultName;
+  }
 }

OpenTelemetryIngest/Services/Queue/SyslogQueueService.ts

@@ -0,0 +1,15 @@
+import { TelemetryRequest } from "Common/Server/Middleware/TelemetryIngest";
+import TelemetryQueueService, {
+  TelemetryType,
+} from "./TelemetryQueueService";
+
+export default class SyslogQueueService {
+  public static async addSyslogIngestJob(
+    req: TelemetryRequest,
+  ): Promise<void> {
+    return TelemetryQueueService.addTelemetryIngestJob(
+      req,
+      TelemetryType.Syslog,
+    );
+  }
+}

OpenTelemetryIngest/Services/Queue/TelemetryQueueService.ts

@@ -8,6 +8,7 @@ export enum TelemetryType {
   Logs = "logs",
   Traces = "traces",
   Metrics = "metrics",
+  Syslog = "syslog",
 }
 
 export interface TelemetryIngestJobData {
@@ -31,6 +32,10 @@ export interface MetricsIngestJobData extends TelemetryIngestJobData {
   type: TelemetryType.Metrics;
 }
 
+export interface SyslogIngestJobData extends TelemetryIngestJobData {
+  type: TelemetryType.Syslog;
+}
+
 export default class TelemetryQueueService {
   public static async addTelemetryIngestJob(
     req: TelemetryRequest,

OpenTelemetryIngest/Services/SyslogIngestService.ts

@@ -0,0 +1,506 @@
+import { TelemetryRequest } from "Common/Server/Middleware/TelemetryIngest";
+import BadRequestException from "Common/Types/Exception/BadRequestException";
+import {
+  ExpressRequest,
+  ExpressResponse,
+  NextFunction,
+} from "Common/Server/Utils/Express";
+import Response from "Common/Server/Utils/Response";
+import CaptureSpan from "Common/Server/Utils/Telemetry/CaptureSpan";
+import Dictionary from "Common/Types/Dictionary";
+import { JSONObject } from "Common/Types/JSON";
+import ObjectID from "Common/Types/ObjectID";
+import OneUptimeDate from "Common/Types/Date";
+import LogSeverity from "Common/Types/Log/LogSeverity";
+import TelemetryUtil, {
+  AttributeType,
+} from "Common/Server/Utils/Telemetry/Telemetry";
+import OTelIngestService, {
+  TelemetryServiceMetadata,
+} from "Common/Server/Services/OpenTelemetryIngestService";
+import LogService from "Common/Server/Services/LogService";
+import logger from "Common/Server/Utils/Logger";
+import OtelIngestBaseService from "./OtelIngestBaseService";
+import SyslogQueueService from "./Queue/SyslogQueueService";
+import { OPEN_TELEMETRY_INGEST_LOG_FLUSH_BATCH_SIZE } from "../Config";
+import {
+  ParsedSyslogMessage,
+  ParsedSyslogStructuredData,
+  parseSyslogMessage,
+} from "../Utils/SyslogParser";
+
+export default class SyslogIngestService extends OtelIngestBaseService {
+  private static readonly SYSLOG_FACILITY_LABELS: Array<string> = [
+    "kernel",
+    "user",
+    "mail",
+    "system",
+    "security",
+    "syslogd",
+    "line_printer",
+    "network_news",
+    "uucp",
+    "clock",
+    "security2",
+    "ftp",
+    "ntp",
+    "log_audit",
+    "log_alert",
+    "clock2",
+    "local0",
+    "local1",
+    "local2",
+    "local3",
+    "local4",
+    "local5",
+    "local6",
+    "local7",
+  ];
+
+  private static readonly SYSLOG_SEVERITY_LABELS: Array<string> = [
+    "emergency",
+    "alert",
+    "critical",
+    "error",
+    "warning",
+    "notice",
+    "informational",
+    "debug",
+  ];
+
+  private static readonly DEFAULT_SERVICE_NAME: string = "Syslog";
+
+  private static readonly SYSLOG_TO_OTEL_SEVERITY: Dictionary<{
+    number: number;
+    text: LogSeverity;
+  }> = {
+    "0": { number: 23, text: LogSeverity.Fatal },
+    "1": { number: 23, text: LogSeverity.Fatal },
+    "2": { number: 19, text: LogSeverity.Error },
+    "3": { number: 19, text: LogSeverity.Error },
+    "4": { number: 13, text: LogSeverity.Warning },
+    "5": { number: 9, text: LogSeverity.Information },
+    "6": { number: 9, text: LogSeverity.Information },
+    "7": { number: 5, text: LogSeverity.Debug },
+  };
+
+  @CaptureSpan()
+  public static async ingestSyslog(
+    req: ExpressRequest,
+    res: ExpressResponse,
+    next: NextFunction,
+  ): Promise<void> {
+    try {
+      if (!(req as TelemetryRequest).projectId) {
+        throw new BadRequestException(
+          "Invalid request - projectId not found in request.",
+        );
+      }
+
+      const messages: Array<string> = this.normalizeMessages(req.body);
+
+      if (messages.length === 0) {
+        throw new BadRequestException("No syslog messages found in request.");
+      }
+
+      req.body = {
+        messages,
+      } satisfies JSONObject;
+
+      Response.sendEmptySuccessResponse(req, res);
+
+      await SyslogQueueService.addSyslogIngestJob(req as TelemetryRequest);
+
+      return;
+    } catch (error) {
+      return next(error);
+    }
+  }
+
+  @CaptureSpan()
+  public static async processSyslogFromQueue(
+    req: ExpressRequest,
+  ): Promise<void> {
+    await this.processSyslogAsync(req);
+  }
+
+  @CaptureSpan()
+  private static async processSyslogAsync(req: ExpressRequest): Promise<void> {
+    try {
+      const projectId: ObjectID = (req as TelemetryRequest).projectId;
+      const messages: Array<string> = this.extractMessagesFromRequest(req.body);
+
+      if (messages.length === 0) {
+        logger.warn("Syslog ingest: no messages to process.");
+        return;
+      }
+
+      const dbLogs: Array<JSONObject> = [];
+      const serviceCache: Dictionary<TelemetryServiceMetadata> = {};
+      let processed: number = 0;
+
+      let messageCounter: number = 0;
+
+      for (const rawMessage of messages) {
+        try {
+          if (messageCounter % 500 === 0) {
+            await Promise.resolve();
+          }
+
+          messageCounter++;
+
+          const parsed: ParsedSyslogMessage | null =
+            parseSyslogMessage(rawMessage);
+
+          if (!parsed) {
+            logger.warn(
+              `Syslog ingest: unable to parse message: ${rawMessage}`,
+            );
+            continue;
+          }
+
+          const serviceName: string = this.resolveServiceName(req, parsed);
+
+          if (!serviceCache[serviceName]) {
+            const metadata: {
+              serviceId: ObjectID;
+              dataRententionInDays: number;
+            } = await OTelIngestService.telemetryServiceFromName({
+              serviceName,
+              projectId,
+            });
+
+            serviceCache[serviceName] = {
+              serviceName,
+              serviceId: metadata.serviceId,
+              dataRententionInDays: metadata.dataRententionInDays,
+            } satisfies TelemetryServiceMetadata;
+          }
+
+          const serviceMetadata: TelemetryServiceMetadata =
+            serviceCache[serviceName]!;
+
+          const severityInfo: { number: number; text: LogSeverity } =
+            this.mapSeverity(parsed.severity);
+
+          const timestamp: Date = parsed.timestamp ||
+            OneUptimeDate.getCurrentDate();
+          const ingestionDate: Date = OneUptimeDate.getCurrentDate();
+
+          const attributes: Dictionary<AttributeType | Array<AttributeType>> =
+            this.buildAttributes({
+              parsed,
+              serviceId: serviceMetadata.serviceId,
+              serviceName,
+            });
+
+          const logRow: JSONObject = {
+            _id: ObjectID.generate().toString(),
+            createdAt: OneUptimeDate.toClickhouseDateTime(ingestionDate),
+            updatedAt: OneUptimeDate.toClickhouseDateTime(ingestionDate),
+            projectId: projectId.toString(),
+            serviceId: serviceMetadata.serviceId.toString(),
+            time: OneUptimeDate.toClickhouseDateTime(timestamp),
+            timeUnixNano: Math.trunc(
+              OneUptimeDate.toUnixNano(timestamp),
+            ).toString(),
+            severityNumber: severityInfo.number,
+            severityText: severityInfo.text,
+            attributes,
+            attributeKeys: TelemetryUtil.getAttributeKeys(attributes),
+            traceId: "",
+            spanId: "",
+            body: parsed.message,
+          } satisfies JSONObject;
+
+          dbLogs.push(logRow);
+          processed++;
+
+          if (
+            dbLogs.length >= OPEN_TELEMETRY_INGEST_LOG_FLUSH_BATCH_SIZE
+          ) {
+            await this.flushLogsBuffer(dbLogs);
+          }
+        } catch (processingError) {
+          logger.error("Syslog ingest: error processing message");
+          logger.error(processingError);
+          logger.error(`Syslog message: ${rawMessage}`);
+        }
+      }
+
+      await this.flushLogsBuffer(dbLogs, true);
+
+      if (processed === 0) {
+        logger.warn("Syslog ingest: no valid messages processed");
+      } else {
+        logger.debug(
+          `Syslog ingest: processed ${processed} messages for project ${projectId.toString()}`,
+        );
+      }
+
+      dbLogs.length = 0;
+
+      try {
+        if (req.body) {
+          req.body = null;
+        }
+      } catch (cleanupError) {
+        logger.error("Syslog ingest: error during memory cleanup");
+        logger.error(cleanupError);
+      }
+    } catch (error) {
+      logger.error("Syslog ingest: critical error");
+      logger.error(error);
+      throw error;
+    }
+  }
+
+  private static resolveServiceName(
+    req: ExpressRequest,
+    parsed: ParsedSyslogMessage,
+  ): string {
+    const headerServiceName: string = this.getServiceNameFromHeaders(
+      req,
+      "",
+    );
+
+    if (headerServiceName) {
+      return headerServiceName;
+    }
+
+    if (parsed.appName && parsed.appName.trim()) {
+      return parsed.appName.trim();
+    }
+
+    if (parsed.hostname && parsed.hostname.trim()) {
+      return parsed.hostname.trim();
+    }
+
+    return SyslogIngestService.DEFAULT_SERVICE_NAME;
+  }
+
+  private static buildAttributes(data: {
+    parsed: ParsedSyslogMessage;
+    serviceId: ObjectID;
+    serviceName: string;
+  }): Dictionary<AttributeType | Array<AttributeType>> {
+    const { parsed } = data;
+
+    const attributes: Dictionary<AttributeType | Array<AttributeType>> = {
+      ...TelemetryUtil.getAttributesForServiceIdAndServiceName({
+        serviceId: data.serviceId,
+        serviceName: data.serviceName,
+      }),
+      "syslog.raw": parsed.raw,
+    };
+
+    if (parsed.hostname) {
+      attributes["syslog.hostname"] = parsed.hostname;
+    }
+
+    if (parsed.appName) {
+      attributes["syslog.appName"] = parsed.appName;
+    }
+
+    if (parsed.procId) {
+      attributes["syslog.processId"] = parsed.procId;
+    }
+
+    if (parsed.msgId) {
+      attributes["syslog.messageId"] = parsed.msgId;
+    }
+
+    if (parsed.version !== undefined) {
+      attributes["syslog.version"] = parsed.version;
+    }
+
+    if (parsed.priority !== undefined) {
+      attributes["syslog.priority"] = parsed.priority;
+    }
+
+    if (parsed.severity !== undefined) {
+      attributes["syslog.severity.code"] = parsed.severity;
+      attributes["syslog.severity.name"] = this.getSeverityLabel(
+        parsed.severity,
+      );
+    }
+
+    if (parsed.facility !== undefined) {
+      attributes["syslog.facility.code"] = parsed.facility;
+      attributes["syslog.facility.name"] = this.getFacilityLabel(
+        parsed.facility,
+      );
+    }
+
+    if (parsed.structuredDataRaw) {
+      attributes["syslog.structured.raw"] = parsed.structuredDataRaw;
+    }
+
+    if (parsed.structuredData) {
+      this.appendStructuredDataAttributes(attributes, parsed.structuredData);
+    }
+
+    return attributes;
+  }
+
+  private static appendStructuredDataAttributes(
+    attributes: Dictionary<AttributeType | Array<AttributeType>>,
+    structuredData: ParsedSyslogStructuredData,
+  ): void {
+    for (const [sdId, params] of Object.entries(structuredData)) {
+      for (const [key, value] of Object.entries(params)) {
+        const attributeKey: string = `syslog.structured.${this.sanitizeAttributeKey(sdId)}.${this.sanitizeAttributeKey(key)}`;
+        attributes[attributeKey] = value;
+      }
+    }
+  }
+
+  private static getSeverityLabel(severity: number): string {
+    if (
+      severity >= 0 &&
+      severity < SyslogIngestService.SYSLOG_SEVERITY_LABELS.length
+    ) {
+      return SyslogIngestService.SYSLOG_SEVERITY_LABELS[severity]!;
+    }
+
+    return "unknown";
+  }
+
+  private static getFacilityLabel(facility: number): string {
+    if (
+      facility >= 0 &&
+      facility < SyslogIngestService.SYSLOG_FACILITY_LABELS.length
+    ) {
+      return SyslogIngestService.SYSLOG_FACILITY_LABELS[facility]!;
+    }
+
+    return "unknown";
+  }
+
+  private static mapSeverity(
+    severity?: number | undefined,
+  ): { number: number; text: LogSeverity } {
+    if (severity === undefined || severity === null) {
+      return { number: 0, text: LogSeverity.Unspecified };
+    }
+
+    const key: string = severity.toString();
+
+    if (this.SYSLOG_TO_OTEL_SEVERITY[key]) {
+      return this.SYSLOG_TO_OTEL_SEVERITY[key]!;
+    }
+
+    return { number: 0, text: LogSeverity.Unspecified };
+  }
+
+  private static async flushLogsBuffer(
+    logs: Array<JSONObject>,
+    force: boolean = false,
+  ): Promise<void> {
+    while (
+      logs.length >= OPEN_TELEMETRY_INGEST_LOG_FLUSH_BATCH_SIZE ||
+      (force && logs.length > 0)
+    ) {
+      const batchSize: number = Math.min(
+        logs.length,
+        OPEN_TELEMETRY_INGEST_LOG_FLUSH_BATCH_SIZE,
+      );
+
+      const batch: Array<JSONObject> = logs.splice(0, batchSize);
+
+      if (batch.length === 0) {
+        continue;
+      }
+
+      await LogService.insertJsonRows(batch);
+    }
+  }
+
+  private static extractMessagesFromRequest(body: unknown): Array<string> {
+    if (!body || typeof body !== "object") {
+      return [];
+    }
+
+    const payload: JSONObject = body as JSONObject;
+    const messages: unknown = payload["messages"];
+
+    if (Array.isArray(messages)) {
+      return messages
+        .map((item: unknown) => {
+          if (typeof item === "string") {
+            return item;
+          }
+
+          if (item === null || item === undefined) {
+            return "";
+          }
+
+          return String(item);
+        })
+        .filter((item: string | undefined): item is string => {
+          return Boolean(item && item.trim());
+        });
+    }
+
+    if (typeof messages === "string") {
+      return this.normalizeMessages(messages);
+    }
+
+    return [];
+  }
+
+  private static normalizeMessages(payload: unknown): Array<string> {
+    if (!payload) {
+      return [];
+    }
+
+    if (typeof payload === "string") {
+      return payload
+        .split(/\r?\n/)
+        .map((line: string) => {
+          return line.trim();
+        })
+        .filter((line: string) => {
+          return line.length > 0;
+        });
+    }
+
+    if (Buffer.isBuffer(payload)) {
+      return this.normalizeMessages(payload.toString("utf-8"));
+    }
+
+    if (Array.isArray(payload)) {
+      const results: Array<string> = [];
+      for (const item of payload) {
+        results.push(...this.normalizeMessages(item));
+      }
+      return results;
+    }
+
+    if (typeof payload === "object") {
+      const obj: JSONObject = payload as JSONObject;
+
+      if (Array.isArray(obj["messages"])) {
+        return this.normalizeMessages(obj["messages"]);
+      }
+
+      if (typeof obj["message"] === "string") {
+        return this.normalizeMessages(obj["message"]);
+      }
+
+      if (Array.isArray(obj["syslog"])) {
+        return this.normalizeMessages(obj["syslog"]);
+      }
+
+      if (typeof obj["syslog"] === "string") {
+        return this.normalizeMessages(obj["syslog"]);
+      }
+    }
+
+    return [];
+  }
+
+  private static sanitizeAttributeKey(value: string): string {
+    return value.replace(/[^A-Za-z0-9_.-]/g, "_");
+  }
+}

OpenTelemetryIngest/Tests/Utils/SyslogParser.test.ts

@@ -0,0 +1,58 @@
+import { parseSyslogMessage } from "../../Utils/SyslogParser";
+
+describe("SyslogParser", () => {
+  test("parses RFC5424 message with structured data", () => {
+    const message: string =
+      "<34>1 2025-03-02T14:48:05.003Z mymachine app-name 1234 ID47 " +
+      "[exampleSDID@32473 iut=\"3\" eventSource=\"Application\" eventID=\"1011\"][meta key=\"value\"] " +
+      "BOMAn application event log entry";
+
+    const parsed = parseSyslogMessage(message);
+
+    expect(parsed).not.toBeNull();
+    expect(parsed?.priority).toBe(34);
+    expect(parsed?.severity).toBe(2);
+    expect(parsed?.facility).toBe(4);
+    expect(parsed?.version).toBe(1);
+    expect(parsed?.hostname).toBe("mymachine");
+    expect(parsed?.appName).toBe("app-name");
+    expect(parsed?.procId).toBe("1234");
+    expect(parsed?.msgId).toBe("ID47");
+    expect(parsed?.timestamp?.toISOString()).toBe(
+      "2025-03-02T14:48:05.003Z",
+    );
+    expect(parsed?.structuredData?.["exampleSDID_32473"]?.["iut"]).toBe(
+      "3",
+    );
+    expect(parsed?.structuredData?.["meta"]?.["key"]).toBe("value");
+    expect(parsed?.message).toBe("An application event log entry");
+  });
+
+  test("parses RFC3164 message", () => {
+    const message: string =
+      "<13>Feb  5 17:32:18 mymachine su[12345]: 'su root' failed for lonvick on /dev/pts/8";
+
+    const parsed = parseSyslogMessage(message);
+
+    expect(parsed).not.toBeNull();
+    expect(parsed?.priority).toBe(13);
+    expect(parsed?.severity).toBe(5);
+    expect(parsed?.facility).toBe(1);
+    expect(parsed?.hostname).toBe("mymachine");
+    expect(parsed?.appName).toBe("su");
+    expect(parsed?.procId).toBe("12345");
+    expect(parsed?.message).toBe("'su root' failed for lonvick on /dev/pts/8");
+    expect(parsed?.timestamp).toBeInstanceOf(Date);
+  });
+
+  test("handles message without priority", () => {
+    const message: string = "Simple message without metadata";
+    const parsed = parseSyslogMessage(message);
+
+    expect(parsed).not.toBeNull();
+    expect(parsed?.priority).toBeUndefined();
+    expect(parsed?.severity).toBeUndefined();
+    expect(parsed?.facility).toBeUndefined();
+    expect(parsed?.message).toBe("Simple message without metadata");
+  });
+});

OpenTelemetryIngest/Utils/SyslogParser.ts

@@ -0,0 +1,392 @@
+import OneUptimeDate, { Moment } from "Common/Types/Date";
+
+export interface ParsedSyslogStructuredData {
+  [sdId: string]: {
+    [key: string]: string;
+  };
+}
+
+export interface ParsedSyslogMessage {
+  raw: string;
+  message: string;
+  priority?: number | undefined;
+  severity?: number | undefined;
+  facility?: number | undefined;
+  version?: number | undefined;
+  timestamp?: Date | undefined;
+  hostname?: string | undefined;
+  appName?: string | undefined;
+  procId?: string | undefined;
+  msgId?: string | undefined;
+  structuredDataRaw?: string | undefined;
+  structuredData?: ParsedSyslogStructuredData | undefined;
+}
+
+export function parseSyslogMessage(raw: string): ParsedSyslogMessage | null {
+  if (!raw) {
+    return null;
+  }
+
+  const trimmed: string = raw.trim();
+
+  if (!trimmed) {
+    return null;
+  }
+
+  let remaining: string = trimmed;
+  let priority: number | undefined;
+  let severity: number | undefined;
+  let facility: number | undefined;
+
+  const priorityMatch: RegExpMatchArray | null = remaining.match(/^<(\d{1,3})>/);
+
+  if (priorityMatch) {
+    priority = parseInt(priorityMatch[1]!, 10);
+
+    if (!isNaN(priority)) {
+      severity = priority % 8;
+      facility = Math.floor(priority / 8);
+    }
+
+    remaining = remaining.slice(priorityMatch[0]!.length);
+  }
+
+  const rfc5424Parsed: ParsedSyslogMessage | null = parseRfc5424(remaining);
+
+  if (rfc5424Parsed) {
+    return {
+      raw: trimmed,
+      priority,
+      severity: rfc5424Parsed.severity ?? severity,
+      facility: rfc5424Parsed.facility ?? facility,
+      version: rfc5424Parsed.version,
+      timestamp: rfc5424Parsed.timestamp,
+      hostname: rfc5424Parsed.hostname,
+      appName: rfc5424Parsed.appName,
+      procId: rfc5424Parsed.procId,
+      msgId: rfc5424Parsed.msgId,
+      structuredDataRaw: rfc5424Parsed.structuredDataRaw,
+      structuredData: rfc5424Parsed.structuredData,
+      message: stripBom(rfc5424Parsed.message ?? ""),
+    };
+  }
+
+  const rfc3164Parsed: ParsedSyslogMessage | null = parseRfc3164(remaining);
+
+  if (rfc3164Parsed) {
+    return {
+      raw: trimmed,
+      priority,
+      severity,
+      facility,
+      timestamp: rfc3164Parsed.timestamp,
+      hostname: rfc3164Parsed.hostname,
+      appName: rfc3164Parsed.appName,
+      procId: rfc3164Parsed.procId,
+      message: stripBom(rfc3164Parsed.message ?? ""),
+    };
+  }
+
+  return {
+    raw: trimmed,
+    priority,
+    severity,
+    facility,
+    message: stripBom(remaining.trim()),
+  };
+}
+
+function parseRfc5424(payload: string): ParsedSyslogMessage | null {
+  const tokens: Array<string> = splitTokens(payload, 7);
+
+  if (tokens.length < 7) {
+    return null;
+  }
+
+  const versionToken: string = tokens[0]!;
+
+  if (!/^\d+$/.test(versionToken)) {
+    return null;
+  }
+
+  const version: number = parseInt(versionToken, 10);
+  const timestampToken: string = tokens[1]!
+    .trim()
+    .replace(/^NILVALUE$/i, "-");
+  const hostnameToken: string = tokens[2]!;
+  const appNameToken: string = tokens[3]!;
+  const procIdToken: string = tokens[4]!;
+  const msgIdToken: string = tokens[5]!;
+  const structuredDataAndMessage: string = tokens[6]!;
+
+  const timestamp: Date | undefined =
+    timestampToken && timestampToken !== "-"
+      ? parseRfc5424Timestamp(timestampToken)
+      : undefined;
+
+  const hostname: string | undefined =
+    hostnameToken && hostnameToken !== "-" ? hostnameToken : undefined;
+
+  const appName: string | undefined =
+    appNameToken && appNameToken !== "-" ? appNameToken : undefined;
+
+  const procId: string | undefined =
+    procIdToken && procIdToken !== "-" ? procIdToken : undefined;
+
+  const msgId: string | undefined =
+    msgIdToken && msgIdToken !== "-" ? msgIdToken : undefined;
+
+  const structuredDataParsed: {
+    structuredDataRaw?: string;
+    message?: string;
+    structuredData?: ParsedSyslogStructuredData;
+  } = extractStructuredData(structuredDataAndMessage);
+
+  return {
+    raw: payload,
+    version,
+    timestamp,
+    hostname,
+    appName,
+    procId,
+    msgId,
+    structuredDataRaw: structuredDataParsed.structuredDataRaw,
+    structuredData: structuredDataParsed.structuredData,
+    message: structuredDataParsed.message ?? "",
+  };
+}
+
+function parseRfc3164(payload: string): ParsedSyslogMessage | null {
+  const match: RegExpMatchArray | null = payload.match(
+    /^(\w{3}\s+\d{1,2}\s+\d{2}:\d{2}:\d{2})\s+(\S+)\s+(.*)$/,
+  );
+
+  if (!match) {
+    return null;
+  }
+
+  const timestampToken: string = match[1]!;
+  const hostname: string = match[2]!;
+  const rest: string = match[3] ?? "";
+
+  const timestamp: Date | undefined = parseRfc3164Timestamp(timestampToken);
+
+  let appName: string | undefined;
+  let procId: string | undefined;
+  let message: string = rest.trim();
+
+  const colonIndex: number = rest.indexOf(":");
+
+  if (colonIndex !== -1) {
+    const tag: string = rest.slice(0, colonIndex);
+    message = rest.slice(colonIndex + 1).trim();
+
+    const procMatch: RegExpMatchArray | null = tag.match(/^([^\[]+)\[(.+)\]$/);
+
+    if (procMatch) {
+      appName = procMatch[1]?.trim();
+      procId = procMatch[2]?.trim();
+    } else {
+      appName = tag.trim();
+    }
+  } else {
+    const firstTokenMatch: RegExpMatchArray | null = rest.match(/^(\S+)/);
+
+    if (firstTokenMatch) {
+      const firstToken: string = firstTokenMatch[1]!;
+      const procMatch: RegExpMatchArray | null = firstToken.match(
+        /^([^\[]+)\[(.+)\]$/,
+      );
+
+      if (procMatch) {
+        appName = procMatch[1]?.trim();
+        procId = procMatch[2]?.trim();
+      } else {
+        appName = firstToken.trim();
+      }
+
+      message = rest.slice(firstToken.length).trim();
+    }
+  }
+
+  return {
+    raw: payload,
+    timestamp,
+    hostname,
+    appName,
+    procId,
+    message,
+  };
+}
+
+function splitTokens(source: string, expected: number): Array<string> {
+  const tokens: Array<string> = [];
+  let remaining: string = source.trimStart();
+
+  for (let i: number = 0; i < expected - 1; i++) {
+    if (!remaining) {
+      tokens.push("");
+      continue;
+    }
+
+    const match: RegExpMatchArray | null = remaining.match(/^(\S+)/);
+
+    if (!match) {
+      tokens.push("");
+      remaining = "";
+      continue;
+    }
+
+    const token: string = match[1]!;
+    tokens.push(token);
+    remaining = remaining.slice(token.length).trimStart();
+  }
+
+  tokens.push(remaining);
+
+  return tokens;
+}
+
+function parseRfc5424Timestamp(value: string): Date | undefined {
+  const parsed = Moment(value, Moment.ISO_8601, true);
+
+  if (!parsed.isValid()) {
+    return undefined;
+  }
+
+  return parsed.toDate();
+}
+
+function parseRfc3164Timestamp(value: string): Date | undefined {
+  const currentYear: number = OneUptimeDate.getCurrentYear();
+  const normalized: string = value.replace(/\s+/g, " ");
+  let parsed = Moment(
+    `${normalized} ${currentYear}`,
+    "MMM D HH:mm:ss YYYY",
+    true,
+  );
+
+  if (!parsed.isValid()) {
+    return undefined;
+  }
+
+  const now = Moment();
+
+  if (parsed.isAfter(now.clone().add(1, "months"))) {
+    parsed = parsed.subtract(1, "years");
+  } else if (parsed.isBefore(now.clone().subtract(11, "months"))) {
+    parsed = parsed.add(1, "years");
+  }
+
+  return parsed.toDate();
+}
+
+function extractStructuredData(value: string): {
+  structuredDataRaw?: string;
+  message?: string;
+  structuredData?: ParsedSyslogStructuredData;
+} {
+  const trimmed: string = value.trimStart();
+
+  if (!trimmed) {
+    return { message: "" };
+  }
+
+  if (trimmed.startsWith("-")) {
+    return { message: trimmed.slice(1).trimStart() };
+  }
+
+  if (!trimmed.startsWith("[")) {
+    return { message: trimmed };
+  }
+
+  let depth: number = 0;
+
+  for (let i: number = 0; i < trimmed.length; i++) {
+    const char: string = trimmed[i]!;
+
+    if (char === "[") {
+      depth++;
+    } else if (char === "]") {
+      depth--;
+
+      if (depth === 0) {
+        let peekIndex: number = i + 1;
+
+        while (peekIndex < trimmed.length && trimmed[peekIndex] === " ") {
+          peekIndex++;
+        }
+
+        if (trimmed[peekIndex] === "[") {
+          i = peekIndex - 1;
+          continue;
+        }
+
+        const structuredDataRaw: string = trimmed.slice(0, i + 1).trimEnd();
+        const message: string = trimmed.slice(i + 1).trimStart();
+
+        return {
+          structuredDataRaw,
+          structuredData: parseStructuredData(structuredDataRaw),
+          message,
+        };
+      }
+    }
+  }
+
+  const structuredDataRaw: string = trimmed.trim();
+
+  return {
+    structuredDataRaw,
+    structuredData: parseStructuredData(structuredDataRaw),
+    message: "",
+  };
+}
+
+function parseStructuredData(raw: string): ParsedSyslogStructuredData {
+  const result: ParsedSyslogStructuredData = {};
+  const sdRegex: RegExp = /\[([^\s\]]+)((?:\s+[^\s=]+="[^"]*")*)\]/g;
+  let match: RegExpExecArray | null;
+
+  while ((match = sdRegex.exec(raw)) !== null) {
+    const sdIdRaw: string = match[1]!;
+    const params: string = match[2] ?? "";
+    const sdId: string = sanitizeKey(sdIdRaw);
+
+    if (!result[sdId]) {
+      result[sdId] = {};
+    }
+
+    const paramRegex: RegExp = /([^\s=]+)="([^"]*)"/g;
+    let paramMatch: RegExpExecArray | null;
+
+    while ((paramMatch = paramRegex.exec(params)) !== null) {
+      const keyRaw: string = paramMatch[1]!;
+      const value: string = paramMatch[2] ?? "";
+      const key: string = sanitizeKey(keyRaw);
+      const entry: { [key: string]: string } = result[sdId] ?? {};
+      entry[key] = value;
+      result[sdId] = entry;
+    }
+  }
+
+  return result;
+}
+
+function sanitizeKey(key: string): string {
+  return key.replace(/[^A-Za-z0-9_.-]/g, "_");
+}
+
+function stripBom(value: string): string {
+  if (!value) {
+    return value;
+  }
+
+  let output: string = value.replace(/^\uFEFF/, "");
+
+  if (output.startsWith("BOM")) {
+    output = output.slice(3);
+  }
+
+  return output.trimStart();
+}