Log Collection and Ingestion

Overview

SecuSiem collects logs from client applications and infrastructure through the ingestion-api service, which acts as a Loki-compatible proxy with multi-tenant isolation.

Ingestion Architecture

Client Application / Agent
        |
        | HTTPS + Bearer Token (sk_live_xxx)
        v
+-------------------+
|  ingestion-api    |
|  (Port 8000)      |
+-------------------+
        |
        | Internal HTTP + X-Scope-OrgID header
        v
+-------------------+
|  Grafana Loki 3.0 |
|  (Port 3100)      |
+-------------------+
        |
        v
  Filesystem / S3 Storage

Supported Log Sources

Source	Protocol	Description
Application SDK	HTTPS POST	Direct API calls with Loki-format JSON
Syslog	Syslog -> Promtail -> Loki	Traditional syslog forwarding via Promtail
Kubernetes	Promtail DaemonSet	Automatic pod log collection
Custom agents	HTTPS POST	Any HTTP client sending Loki-format data

Sending Logs (SDK / API)

Authentication

Every request must include a valid API key:

Authorization: Bearer sk_live_AbCdEfGh...

API keys are generated per tenant via the tenant-manager API.

Log Format (Loki Push API)

{
  "streams": [
    {
      "stream": {
        "job": "web-api",
        "app": "conformvault",
        "environment": "production",
        "level": "error"
      },
      "values": [
        ["1738000000000000000", "Error processing request: connection timeout"],
        ["1738000001000000000", "Retrying connection to database"]
      ]
    }
  ]
}

stream: Label key-value pairs for indexing and querying
values: Array of [timestamp_nanoseconds, log_line] pairs
Timestamps must be in nanoseconds (Unix epoch * 10^9)

Example: Python Client

import httpx
import time

SECUSIEM_URL = "https://ingestion.secusiem.secuaas.dev"
API_KEY = "sk_live_AbCdEfGh..."

async def send_logs(logs: list[str]):
    timestamp_ns = str(int(time.time() * 1e9))
    
    payload = {
        "streams": [{
            "stream": {
                "job": "my-app",
                "app": "web-api",
                "environment": "production",
                "level": "info"
            },
            "values": [[timestamp_ns, log] for log in logs]
        }]
    }
    
    async with httpx.AsyncClient() as client:
        response = await client.post(
            f"{SECUSIEM_URL}/loki/api/v1/push",
            headers={
                "Authorization": f"Bearer {API_KEY}",
                "Content-Type": "application/json"
            },
            json=payload
        )
        assert response.status_code == 204

Example: cURL

curl -X POST https://ingestion.secusiem.secuaas.dev/loki/api/v1/push \
  -H "Authorization: Bearer sk_live_AbCdEfGh..." \
  -H "Content-Type: application/json" \
  -d '{
    "streams": [{
      "stream": {"job": "syslog", "level": "warning"},
      "values": [["'$(date +%s)000000000'", "Failed login attempt from 10.0.0.1"]]
    }]
  }'

Multi-tenant Flow

Client sends logs with Bearer token
ingestion-api validates the API key
Tenant is identified from the API key
Loki X-Scope-OrgID header is set to tenant-{slug}
Logs are forwarded to Loki with tenant isolation
Each tenant's logs are stored in separate Loki streams

Rate Limiting

Rate limiting is configured per API key via the rate_limit_rpm field:

Plan	Default Rate Limit
Starter	100 requests/min
Pro	500 requests/min
Business	2,000 requests/min
Enterprise	10,000 requests/min

Rate limiting is enforced via Redis counters in the ingestion-api.

Loki Storage Configuration

Development

Mode: SingleBinary
Storage: Filesystem (local PVC)
Retention: Default Loki retention
Replication: 1 replica

Production (planned)

Mode: Distributed (read/write/backend)
Storage: OVH S3 Object Storage
Retention: Per-tenant lifecycle policies
Replication: 3 replicas for high availability

Log Collection and Ingestion

On this page