Detailed Configuration Examples

Prerequisites

SenseOn accepts any valid JSON sent to /log_sink.

That's the only technical requirement. The examples in this guide show common log sources (nginx, syslog, Kubernetes), but the same principles apply to ANY log source:

Format your logs as JSON - Either at the source or using a log processor
Send via HTTPS POST - Using a log forwarder or direct from your application
Include useful metadata - Timestamps, source identifiers, etc. (recommended but not required)

Minimum Requirements

Absolute minimum valid log:

{"event": "test"}

Recommended format with context:

{
  "timestamp": "2025-10-17T10:00:00Z",
  "source": "your-application",
  "level": "info",
  "message": "Your log message here"
}

What You'll Need

Collector endpoint URL - Provided by SenseOn (format: https://<collector-name>.snson.net)
Outbound HTTPS access - Port 443 to the collector endpoint
Log source - Any application, service, or system generating logs
Log forwarder (optional) - FluentBit, rsyslog or similar

How to Use This Guide

Find a similar example - The patterns shown here (nginx, syslog, Kubernetes) demonstrate common scenarios
Adapt to your source - Replace log paths, parsers, and filters with your specific needs
Test incrementally - Start with a simple curl test, then configure your forwarder
Verify end-to-end - Check forwarder logs show successful sends (HTTP 201)

The examples are patterns, not rigid requirements. If your logs are already JSON, you can skip parsing. If they're not, add a parser. The core principle is: send valid JSON logs to the /log_sink endpoint of your SenseOn log collector.

Testing Your Collector

Before configuring any log forwarder, test your collector endpoint:

# Replace <collector-name> with your actual collector code name
curl -X POST https://<collector-name>.snson.net/log_sink \
  -H "Content-Type: application/json" \
  -d '{"timestamp": "2025-10-17T10:00:00Z", "source": "test", "message": "Connection test"}'

Expected response:

HTTP 201 Created - Log accepted successfully
HTTP 200 OK - Also indicates success

If this works, your collector is ready. Proceed to configure your specific log source.

If this fails:

Verify the collector URL is correct
Check outbound HTTPS (443) is allowed in your firewall
Test DNS resolution: nslookup <collector-name>.snson.net

GCE VM Nginx with FluentBit

Configuration Overview

This example shows nginx running on a Google Compute Engine VM, forwarding logs via FluentBit. This same approach works for:

Any Linux VM (AWS EC2, Azure VMs, on-premises)
Physical servers
Any environment where you can install FluentBit as a service

What we're building:

HTTP Request → Nginx (JSON logs) → FluentBit (local service) → SenseOn

Prerequisites

Linux VM with nginx installed
Root/sudo access to install packages
Outbound HTTPS (443) access

Step 1: Configure Nginx JSON Logging

Create or edit /etc/nginx/conf.d/senseon-logging.conf:

log_format senseon_json escape=json '{'
    '"timestamp":"$time_iso8601",'
    '"remote_addr":"$remote_addr",'
    '"remote_user":"$remote_user",'
    '"request":"$request",'
    '"status":"$status",'
    '"body_bytes_sent":$body_bytes_sent,'
    '"request_time":$request_time,'
    '"http_referer":"$http_referer",'
    '"http_user_agent":"$http_user_agent",'
    '"http_x_forwarded_for":"$http_x_forwarded_for",'
    '"upstream_addr":"$upstream_addr",'
    '"upstream_response_time":"$upstream_response_time",'
    '"upstream_status":"$upstream_status"'
'}';

access_log /var/log/nginx/access.log senseon_json;

What each field captures:

timestamp - When the request occurred (ISO8601 format)
remote_addr - Client IP address
request - Full HTTP request line (method, URI, protocol)
status - HTTP response status code
body_bytes_sent - Size of response body
request_time - Total request processing time (seconds)
upstream_* fields - For reverse proxy scenarios (backend server info)

Note: The upstream fields will be empty for direct web serving, but included for reverse proxy scenarios.

Step 2: Restart Nginx

# Test nginx config
sudo nginx -t

# Restart to apply changes
sudo systemctl restart nginx

# Verify logs are JSON
sudo tail -f /var/log/nginx/access.log

# Generate a test request
curl http://localhost/

You should see JSON logs like:

{"timestamp":"2025-10-17T10:00:00+00:00","remote_addr":"127.0.0.1","request":"GET / HTTP/1.1","status":"200"...}

Step 3: Install FluentBit

# For Debian/Ubuntu
curl https://raw.githubusercontent.com/fluent/fluent-bit/master/install.sh | sh

# Or using package manager (recommended for future compatibility)
wget -qO /usr/share/keyrings/fluentbit.gpg https://packages.fluentbit.io/fluentbit.key
echo "deb [signed-by=/usr/share/keyrings/fluentbit.gpg] https://packages.fluentbit.io/ubuntu/focal focal main" | sudo tee /etc/apt/sources.list.d/fluent-bit.list
sudo apt-get update
sudo apt-get install fluent-bit

# Verify installation
fluent-bit --version
# Should show: Fluent Bit v2.2.x

Step 4: Configure FluentBit

Create /etc/fluent-bit/fluent-bit.conf:

[SERVICE]
    Flush        5
    Daemon       Off
    Log_Level    info
    Parsers_File parsers.conf

[INPUT]
    Name              tail
    Path              /var/log/nginx/access.log
    Parser            json
    Tag               nginx.access
    Refresh_Interval  5
    Mem_Buf_Limit     5MB
    Skip_Long_Lines   On

[FILTER]
    Name    modify
    Match   nginx.access
    Add     source nginx-gce-vm
    Add     environment production
    Add     vm_name nginx-log-vm

[OUTPUT]
    Name            http
    Match           nginx.access
    Host            <collector-name>.snson.net
    Port            443
    URI             /log_sink
    Format          json_lines
    json_date_key   false
    tls             On
    tls.verify      On
    Header          Content-Type application/json
    Retry_Limit     3
    net.keepalive   On

What each setting does:

Mem_Buf_Limit 5MB - Limits memory buffer per input (prevents OOM)
Skip_Long_Lines On - Skips lines > 32KB (prevents issues with huge log entries)
vm_name nginx-log-vm - Adds VM identifier (useful when you have multiple VMs)
net.keepalive On - Reuses HTTPS connections (more efficient)

Step 5: Start FluentBit

# Start the service
sudo systemctl start fluent-bit
sudo systemctl enable fluent-bit

# Check status
sudo systemctl status fluent-bit

# Should show: active (running)

Step 6: Verify Log Forwarding

# Check FluentBit is reading nginx logs
sudo journalctl -u fluent-bit -f | grep "HTTP status"

# Should see: HTTP status=201

# Generate test traffic
curl http://localhost/
curl http://localhost/test
curl http://localhost/health

# Check FluentBit logs show forwarding
sudo journalctl -u fluent-bit --since "1 minute ago" | grep "HTTP status"

Example Log Output Sent to SenseOn

{
  "timestamp": "2025-10-14T15:30:45+00:00",
  "remote_addr": "203.0.113.42",
  "remote_user": "",
  "request": "GET /api/users HTTP/1.1",
  "status": "200",
  "body_bytes_sent": 1247,
  "http_referer": "https://example.com/dashboard",
  "http_user_agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36",
  "request_time": 0.234,
  "upstream_response_time": "0.156"
}

Syslog (rsyslog)

Configuration Overview

Forward system logs from rsyslog directly to SenseOn using the HTTP output module.

Step 1: Install rsyslog HTTP Module

# Debian/Ubuntu
sudo apt-get update
sudo apt-get install rsyslog-mmjsonparse rsyslog-omhttp

# RHEL/CentOS
sudo yum install rsyslog-mmjsonparse rsyslog-omhttp

Step 2: Configure rsyslog

Create /etc/rsyslog.d/senseon.conf:

# Load HTTP output module
module(load="omhttp")

# JSON template for system logs
template(name="SenseOnJSONFormat" type="list") {
    constant(value="{")
    constant(value="\"timestamp\":\"")
    property(name="timereported" dateFormat="rfc3339")
    constant(value="\",")
    constant(value="\"hostname\":\"")
    property(name="hostname")
    constant(value="\",")
    constant(value="\"facility\":\"")
    property(name="syslogfacility-text")
    constant(value="\",")
    constant(value="\"severity\":\"")
    property(name="syslogseverity-text")
    constant(value="\",")
    constant(value="\"message\":\"")
    property(name="msg" format="json")
    constant(value="\",")
    constant(value="\"program\":\"")
    property(name="programname")
    constant(value="\"}")
}

# Forward critical system events to SenseOn
# Adjust severity filter as needed (info, notice, warning, err, crit)
:syslogseverity, isequal, "err" action(
    type="omhttp"
    server="<collector-name>.snson.net"
    serverport="443"
    restpath="/log_sink"
    template="SenseOnJSONFormat"
    httpheadercontenttype="application/json"
    errorfile="/var/log/rsyslog-senseon-errors.log"
)

# You can also forward specific facilities
# Example: forward all kernel messages
:syslogfacility-text, isequal, "kern" action(
    type="omhttp"
    server="<collector-name>.snson.net"
    serverport="443"
    restpath="/log_sink"
    template="SenseOnJSONFormat"
    httpheadercontenttype="application/json"
    errorfile="/var/log/rsyslog-senseon-errors.log"
)

Step 3: Restart rsyslog

# Test configuration
sudo rsyslogd -N1

# Restart service
sudo systemctl restart rsyslog

# Check for errors
sudo tail -f /var/log/rsyslog-senseon-errors.log

Example Log Output

{
  "timestamp": "2025-10-14T15:30:45+00:00",
  "hostname": "web-server-01",
  "facility": "kern",
  "severity": "error",
  "message": "Out of memory: Kill process 12345 (apache2) score 512",
  "program": "kernel"
}

FluentBit Log Forwarder

Configuration Overview

FluentBit is a lightweight log forwarder that can collect from multiple sources and forward to SenseOn.

Installation

# Download and install
curl https://raw.githubusercontent.com/fluent/fluent-bit/master/install.sh | sh

# Or using package manager (Ubuntu/Debian)
wget -qO /usr/share/keyrings/fluentbit.gpg https://packages.fluentbit.io/fluentbit.key
echo "deb [signed-by=/usr/share/keyrings/fluentbit.gpg] https://packages.fluentbit.io/ubuntu/focal focal main" | sudo tee
/etc/apt/sources.list.d/fluent-bit.list
sudo apt-get update
sudo apt-get install fluent-bit

Configuration File

Create /etc/fluent-bit/fluent-bit.conf:

[SERVICE]
    Flush        5
    Daemon       Off
    Log_Level    info
    Parsers_File parsers.conf

# Example: Tail multiple log files
[INPUT]
    Name              tail
    Path              /var/log/nginx/access.log
    Parser            json
    Tag               nginx
    Refresh_Interval  5

[INPUT]
    Name              tail
    Path              /var/log/app/application.log
    Parser            json
    Tag               application
    Refresh_Interval  5

# Example: Collect systemd journal logs
[INPUT]
    Name              systemd
    Tag               systemd
    Read_From_Tail    On
    Strip_Underscores On

# Add hostname to all logs
[FILTER]
    Name    modify
    Match   *
    Add     hostname ${HOSTNAME}

# Forward all logs to SenseOn
[OUTPUT]
    Name            http
    Match           *
    Host            <collector-name>.snson.net
    Port            443
    URI             /log_sink
    Format          json_lines
    json_date_key   false
    tls             On
    tls.verify      On
    Header          Content-Type application/json
    Retry_Limit     3

Start FluentBit

sudo systemctl start fluent-bit
sudo systemctl enable fluent-bit

# Check status
sudo systemctl status fluent-bit

# View logs
sudo journalctl -u fluent-bit -f

GKE Nginx with FluentBit

Configuration Overview

This example shows nginx running on Google Kubernetes Engine (GKE), forwarding logs via FluentBit DaemonSet. This same approach works for:

Any Kubernetes cluster (EKS, AKS, self-hosted)
Any application logging JSON to stdout (not just nginx)

What we're building:

HTTP Request → Nginx Pod (JSON logs to stdout) →
Docker wrapper → FluentBit DaemonSet (parse & forward) → SenseOn

The challenge: Kubernetes logs are wrapped in Docker/containerd format, which means your application's JSON logs end up nested and escaped inside a log field. This section shows you how to properly parse Kubernetes logs to send clean JSON to SenseOn.

The Problem: Nested JSON

If you naively forward Kubernetes logs, you'll get this:

{
  "log": "2025-10-17T10:00:00.123456789Z stdout F {\"timestamp\":\"2025-10-17T10:00:00Z\",\"status\":\"200\"}",
  "stream": "stdout",
  "time": "2025-10-17T10:00:00.123456789Z"
}

Your application JSON is escaped and nested inside the log field. This is not what you want.

The Solution: Two-Stage Parsing

You need to parse the log in two stages:

Strip the Docker wrapper (timestamp, stream, tag)
Parse your application JSON from the extracted message

The configuration below implements this correctly.

Prerequisites

Kubernetes cluster (any flavor: GKE, EKS, AKS, self-hosted)
kubectl access with admin permissions
Application containers logging JSON to stdout

Step 1: Create Namespace

kubectl create namespace nginx-logs

Step 2: Create FluentBit DaemonSet

Save this as fluentbit-daemonset.yaml:

apiVersion: v1
kind: ServiceAccount
metadata:
  name: fluent-bit
  namespace: nginx-logs

---
apiVersion: rbac.authorization.k8s.io/v1
kind: ClusterRole
metadata:
  name: fluent-bit
rules:
- apiGroups: [""]
  resources: ["pods", "namespaces"]
  verbs: ["get", "list", "watch"]

---
apiVersion: rbac.authorization.k8s.io/v1
kind: ClusterRoleBinding
metadata:
  name: fluent-bit
roleRef:
  apiGroup: rbac.authorization.k8s.io
  kind: ClusterRole
  name: fluent-bit
subjects:
- kind: ServiceAccount
  name: fluent-bit
  namespace: nginx-logs

---
apiVersion: v1
kind: ConfigMap
metadata:
  name: fluent-bit-config
  namespace: nginx-logs
data:
  fluent-bit.conf: |
    [SERVICE]
        Daemon       Off
        Flush        5
        Log_Level    info
        Parsers_File parsers.conf

    [INPUT]
        Name              tail
        Path              /var/log/containers/*nginx-logs_nginx-*.log
        Tag               nginx
        Refresh_Interval  5
        Mem_Buf_Limit     5MB
        Skip_Long_Lines   On
        Parser            docker

    # Stage 1: Parse Docker wrapper to extract the JSON message
    [FILTER]
        Name    parser
        Match   nginx
        Key_Name log
        Parser  docker_strip
        Reserve_Data On

    # Stage 2: Parse the application JSON from the message field
    [FILTER]
        Name    parser
        Match   nginx
        Key_Name message
        Parser  nginx_json
        Reserve_Data On

    # Remove Docker wrapper fields, keep application data
    [FILTER]
        Name    modify
        Match   nginx
        Remove  log
        Remove  stream
        Remove  time
        Remove  message
        Remove  logtag
        Add     source nginx-gke
        Add     environment production

    [OUTPUT]
        Name            http
        Match           nginx
        Host            <collector-name>.snson.net
        Port            443
        URI             /log_sink
        Format          json_lines
        json_date_key   false
        tls             On
        tls.verify      On
        Header          Content-Type application/json
        Retry_Limit     3
        net.keepalive   On

  parsers.conf: |
    # Parser for Docker JSON format
    [PARSER]
        Name        docker
        Format      json
        Time_Key    time
        Time_Format %Y-%m-%dT%H:%M:%S.%L%z
        Time_Keep   Off

    # Parser to strip Docker wrapper and extract message
    [PARSER]
        Name        docker_strip
        Format      regex
        Regex       ^(?<time>[^ ]+) (?<stream>stdout|stderr) (?<logtag>[^ ]*) (?<message>.*)$
        Time_Key    time
        Time_Format %Y-%m-%dT%H:%M:%S.%L%z
        Time_Keep   Off

    # Parser for nginx JSON logs
    [PARSER]
        Name        nginx_json
        Format      json

---
apiVersion: apps/v1
kind: DaemonSet
metadata:
  name: fluent-bit
  namespace: nginx-logs
spec:
  selector:
    matchLabels:
      app: fluent-bit
  template:
    metadata:
      labels:
        app: fluent-bit
    spec:
      serviceAccountName: fluent-bit
      containers:
      - name: fluent-bit
        image: fluent/fluent-bit:2.2.3
        volumeMounts:
        - name: varlog
          mountPath: /var/log
        - name: varlibdockercontainers
          mountPath: /var/lib/docker/containers
          readOnly: true
        - name: fluent-bit-config
          mountPath: /fluent-bit/etc/
      volumes:
      - name: varlog
        hostPath:
          path: /var/log
      - name: varlibdockercontainers
        hostPath:
          path: /var/lib/docker/containers
      - name: fluent-bit-config
        configMap:
          name: fluent-bit-config

Step 3: Deploy to Kubernetes

# Deploy FluentBit (namespace is already created in Step 1)
kubectl apply -f fluentbit-daemonset.yaml

# Verify deployment
kubectl get pods -n nginx-logs

# Should show one fluent-bit pod per node, all Running
# NAME               READY   STATUS    RESTARTS   AGE
# fluent-bit-xxxxx   1/1     Running   0          30s
# fluent-bit-yyyyy   1/1     Running   0          30s

Step 4: Verify Log Forwarding

# Check FluentBit logs for successful forwarding
kubectl logs -n nginx-logs -l app=fluent-bit --tail=20 | grep "HTTP status"

# Should see: HTTP status=201 (logs accepted by SenseOn)
# Example output:
# [2025/10/17 10:00:05] [ info] [output:http:http.0] <collector-name>.snson.net:443, HTTP status=201

Step 5: Generate Test Logs

If you have nginx pods running in the nginx-logs namespace:

# Generate a test request to your nginx service
kubectl exec -n nginx-logs deployment/nginx -- curl localhost

# Check FluentBit picked up the log
kubectl logs -n nginx-logs -l app=fluent-bit --tail=50 | grep "HTTP status"

How the Parsing Works

Input (from Kubernetes):

2025-10-17T10:00:00.123456789Z stdout F {"timestamp":"2025-10-17T10:00:00+00:00","remote_addr":"10.154.15.237","request":"GET / HTTP/1.1","status":"200"}

After Stage 1 (docker_strip):

Extracts: message = {"timestamp":"2025-10-17T10:00:00+00:00","remote_addr":"10.154.15.237","request":"GET / HTTP/1.1","status":"200"}
Removes: Docker timestamp, stream, tag

After Stage 2 (nginx_json):

Parses the JSON from message field
Lifts fields to top level: timestamp, remote_addr, request, status, etc.

After modify filter:

Removes: log, stream, time, message, logtag (Docker wrapper fields)
Adds: source, environment (for identifying the log source)

Final output sent to SenseOn:

{
  "timestamp": "2025-10-17T10:00:00+00:00",
  "remote_addr": "10.154.15.237",
  "request": "GET / HTTP/1.1",
  "status": "200",
  "body_bytes_sent": 45,
  "request_time": 0.000,
  "http_user_agent": "curl/7.88.1",
  "source": "nginx-gke",
  "environment": "production"
}

Troubleshooting Kubernetes Logs

Problem: Logs still nested/escaped

Check your FluentBit configuration has:

Two parser filters (docker_strip, then nginx_json or app_json)
Reserve_Data On on both parsers
modify filter removing wrapper fields (log, stream, time, message, logtag)
Correct parser names in parsers.conf matching the filter references

Problem: No logs forwarded

# Check FluentBit is reading container logs
kubectl logs -n nginx-logs -l app=fluent-bit --tail=100 | grep "inotify_fs_add"

# Should see: inotify_fs_add(): inode=... name=/var/log/containers/nginx-logs_nginx-...

# Check for errors
kubectl logs -n nginx-logs -l app=fluent-bit --tail=100 | grep -i error

Problem: FluentBit pods not starting

# Check pod status
kubectl describe pod -n nginx-logs -l app=fluent-bit

# Common issues:
# - Missing permissions (check RBAC)
# - Config syntax error (check configmap)
# - Resource limits too low

Important Notes

Critical configuration details:

Use Format json_lines (not json) in OUTPUT - sends newline-delimited JSON
Set json_date_key false - Let SenseOn add timestamps server-side
Use Reserve_Data On - Merges parsed fields instead of replacing
FluentBit version 2.2.x recommended - Stable and well-tested

Cloud Run Nginx with Cloud Logging

Configuration Overview

This example shows nginx running on Cloud Run (serverless), forwarding logs via Cloud Logging. This approach is ideal for:

Serverless nginx (no VM management)
Low-to-medium traffic websites
Dev/test environments
Quick deployments

What we're building:

HTTP Request → Nginx (Cloud Run container) → JSON logs to stdout →
Cloud Logging → Pub/Sub → Cloud Function → SenseOn

Why this is different from VM/GKE:

No FluentBit needed (Cloud Logging handles collection)
Serverless (auto-scaling, no infrastructure)
Pay-per-request pricing
Simpler deployment (just container + function)

Prerequisites

GCP project with billing enabled
gcloud CLI installed and authenticated
Docker installed locally (for building image)

Step 1: Create Nginx Configuration Files

Create nginx.conf (log format definition):

log_format senseon_json escape=json '{'
    '"timestamp":"$time_iso8601",'
    '"remote_addr":"$remote_addr",'
    '"remote_user":"$remote_user",'
    '"request":"$request",'
    '"status":"$status",'
    '"body_bytes_sent":$body_bytes_sent,'
    '"request_time":$request_time,'
    '"http_referer":"$http_referer",'
    '"http_user_agent":"$http_user_agent",'
    '"http_x_forwarded_for":"$http_x_forwarded_for"'
'}';

Create default.conf (server configuration):

server {
    listen 8080;
    server_name localhost;

    access_log /dev/stdout senseon_json;
    error_log /dev/stderr warn;

    location / {
        root   /usr/share/nginx/html;
        index  index.html index.htm;
    }

    error_page   500 502 503 504  /50x.html;
    location = /50x.html {
        root   /usr/share/nginx/html;
    }
}

Create Dockerfile:

FROM nginx:alpine

# Copy log format definition first (loaded alphabetically before default.conf)
COPY nginx.conf /etc/nginx/conf.d/00-log-format.conf

# Copy server configuration
COPY default.conf /etc/nginx/conf.d/default.conf

# Disable default access log in main nginx.conf
RUN sed -i 's/access_log .*;/access_log off;/' /etc/nginx/nginx.conf

# Create a simple index page
RUN echo '<html><body><h1>Cloud Run Nginx</h1><p>Real nginx logs to SenseOn</p></body></html>' > /usr/share/nginx/html/index.html

EXPOSE 8080

Step 2: Deploy to Cloud Run

# Set variables
export PROJECT_ID="your-project-id"
export REGION="europe-west2"
export SERVICE_NAME="nginx-cloudrun"

# Build and submit image
gcloud builds submit --tag gcr.io/$PROJECT_ID/$SERVICE_NAME

# Deploy to Cloud Run
gcloud run deploy $SERVICE_NAME \
  --image gcr.io/$PROJECT_ID/$SERVICE_NAME \
  --region=$REGION \
  --platform=managed \
  --allow-unauthenticated \
  --port=8080

# Get the service URL
SERVICE_URL=$(gcloud run services describe $SERVICE_NAME --region=$REGION --format='value(status.url)')
echo "Nginx URL: $SERVICE_URL"

# Test it
curl $SERVICE_URL

Step 3: Set Up Log Forwarding

Create Pub/Sub topic:

gcloud pubsub topics create senseon-nginx-logs

Create log sink (captures nginx JSON logs):

gcloud logging sinks create senseon-nginx-sink \
  pubsub.googleapis.com/projects/$PROJECT_ID/topics/senseon-nginx-logs \
  --log-filter='resource.type="cloud_run_revision"
                AND resource.labels.service_name="'$SERVICE_NAME'"
                AND jsonPayload.timestamp!=""'

Grant permissions:

SERVICE_ACCOUNT=$(gcloud logging sinks describe senseon-nginx-sink --format='value(writerIdentity)')
gcloud pubsub topics add-iam-policy-binding senseon-nginx-logs \
  --member=$SERVICE_ACCOUNT \
  --role=roles/pubsub.publisher

Step 4: Deploy Log Forwarder Function

Create main.py:

import base64
import json
import requests
import functions_framework
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

COLLECTOR_URL = "https://<collector-name>.snson.net/log_sink"

@functions_framework.cloud_event
def forward_logs(cloud_event):
    """
    Forward real GCP logs to SenseOn collector.

    Forwards RAW JSON logs without wrapping.
    """
    try:
        # Decode Pub/Sub message
        pubsub_message = base64.b64decode(cloud_event.data["message"]["data"]).decode('utf-8')
        log_entry = json.loads(pubsub_message)

        # Extract service name for logging
        resource = log_entry.get("resource", {})
        resource_labels = resource.get("labels", {})
        service_name = (
            resource_labels.get("service_name") or
            resource_labels.get("function_name") or
            "unknown"
        )

        # Forward the raw log data directly (not wrapped)
        if "jsonPayload" in log_entry:
            # This is structured JSON log (like our nginx JSON)
            senseon_log = log_entry["jsonPayload"]

            # Add minimal source tag
            senseon_log["source"] = f"gcp-{resource.get('type', 'unknown')}"
            senseon_log["_gcp_service"] = service_name

        elif "textPayload" in log_entry:
            # Plain text log - wrap minimally
            senseon_log = {
                "message": log_entry["textPayload"],
                "timestamp": log_entry.get("timestamp"),
                "source": f"gcp-{resource.get('type', 'unknown')}",
                "_gcp_service": service_name
            }
        else:
            # Skip logs without payload
            logger.info(f"Skipping log without payload from {service_name}")
            return

        # Forward to SenseOn collector
        response = requests.post(
            COLLECTOR_URL,
            json=senseon_log,
            headers={"Content-Type": "application/json"},
            timeout=10
        )

        if response.status_code == 201:
            logger.info(f"Log forwarded: {service_name} (HTTP {response.status_code})")
        else:
            logger.warning(f"Unexpected status: {response.status_code} for {service_name}")

    except Exception as e:
        logger.error(f"Error forwarding log: {e}")
        raise

Create requirements.txt:

functions-framework==3.*
requests==2.*

Deploy function:

gcloud functions deploy senseon-nginx-forwarder \
  --gen2 \
  --runtime=python311 \
  --region=$REGION \
  --source=. \
  --entry-point=forward_logs \
  --trigger-topic=senseon-nginx-logs \
  --timeout=60s \
  --memory=256MB

Step 5: Verify End-to-End

Generate traffic:

# Multiple requests to generate logs
for i in {1..10}; do curl $SERVICE_URL; done

Check Cloud Run is logging:

gcloud logging read \
  'resource.type="cloud_run_revision"
   AND resource.labels.service_name="'$SERVICE_NAME'"' \
  --limit=5 \
  --format=json

Check function is forwarding:

gcloud functions logs read senseon-nginx-forwarder \
  --region=$REGION \
  --limit=20

# Should see: "Log forwarded (HTTP 201)"

What Gets Sent to SenseOn

Nginx logs this:

{
  "timestamp": "2025-10-17T10:00:00+00:00",
  "remote_addr": "169.254.169.126",
  "request": "GET / HTTP/1.1",
  "status": "200",
  "body_bytes_sent": 84,
  "http_user_agent": "curl/7.88.1",
  "http_x_forwarded_for": "203.0.113.42"
}

SenseOn receives:

{
  "timestamp": "2025-10-17T10:00:00+00:00",
  "remote_addr": "169.254.169.126",
  "request": "GET / HTTP/1.1",
  "status": "200",
  "body_bytes_sent": 84,
  "http_user_agent": "curl/7.88.1",
  "http_x_forwarded_for": "203.0.113.42",
  "source": "gcp-cloud_run_revision",
  "_gcp_service": "nginx-cloudrun"
}

Cloud Run Application Logs

Configuration Overview

This example shows a custom Python application on Cloud Run, forwarding logs via Cloud Logging. This same pattern works for:

Cloud Run services
Cloud Functions
App Engine
Compute Engine (using Cloud Logging agent)
Any GCP service that writes to Cloud Logging

Architecture:

Application → Cloud Logging → Pub/Sub → Cloud Function → SenseOn

Prerequisites

GCP project with billing enabled
gcloud CLI installed and authenticated
Cloud Run service generating JSON logs (or any GCP service)
Permissions to create Pub/Sub topics, sinks, and functions

Step 1: Configure Application to Log JSON

Your application should log JSON to stdout. Here's a Python Flask example:

Create app.py:

"""
Real Data Processing Application

This application:
- Processes data requests
- Validates inputs
- Performs calculations
- Generates authentic application logs
"""

from flask import Flask, request, jsonify
import logging
import json
import time
import random
from datetime import datetime

app = Flask(__name__)

# Configure JSON logging
class JSONFormatter(logging.Formatter):
    def format(self, record):
        log_data = {
            'timestamp': datetime.utcnow().isoformat() + 'Z',
            'level': record.levelname,
            'logger': record.name,
            'message': record.getMessage(),
        }

        # Add extra fields if present
        if hasattr(record, 'duration_ms'):
            log_data['duration_ms'] = record.duration_ms
        if hasattr(record, 'status'):
            log_data['status'] = record.status
        if hasattr(record, 'user_id'):
            log_data['user_id'] = record.user_id
        if hasattr(record, 'request_id'):
            log_data['request_id'] = record.request_id

        return json.dumps(log_data)

# Set up JSON logging to stdout
handler = logging.StreamHandler()
handler.setFormatter(JSONFormatter())
logger = logging.getLogger('data_processor')
logger.addHandler(handler)
logger.setLevel(logging.INFO)

# Remove default Flask logger
log = logging.getLogger('werkzeug')
log.setLevel(logging.ERROR)

def generate_request_id():
    """Generate a unique request ID"""
    return f"req-{int(time.time())}-{random.randint(1000, 9999)}"

@app.route('/')
def home():
    """Health check endpoint"""
    logger.info('Health check requested')
    return {'status': 'healthy', 'service': 'data-processor'}, 200

@app.route('/process', methods=['POST'])
def process_data():
    """
    Generates authentic logs from real processing steps.
    """
    request_id = generate_request_id()
    start_time = time.time()

    try:
        # Log request received (REAL log)
        logger.info(
            'Processing request received',
            extra={
                'request_id': request_id,
                'user_id': request.headers.get('X-User-ID', 'anonymous')
            }
        )

        # Get and validate data (REAL operation)
        data = request.get_json() or {}
        items = data.get('items', [])
        operation = data.get('operation', 'sum')

        if not items:
            # Log validation failure (REAL log)
            logger.warning(
                'Validation failed: no items provided',
                extra={'request_id': request_id, 'status': 'invalid_input'}
            )
            return {'error': 'No items provided'}, 400

        # Log processing start (REAL log)
        logger.info(
            f'Processing {len(items)} items with operation: {operation}',
            extra={'request_id': request_id}
        )

        # Perform actual processing (REAL operation)
        result = 0
        if operation == 'sum':
            result = sum(items)
        elif operation == 'avg':
            result = sum(items) / len(items) if items else 0
        elif operation == 'max':
            result = max(items) if items else 0
        else:
            # Log unsupported operation (REAL log)
            logger.warning(
                f'Unsupported operation: {operation}',
                extra={'request_id': request_id}
            )
            return {'error': f'Unsupported operation: {operation}'}, 400

        # Calculate duration
        duration_ms = int((time.time() - start_time) * 1000)

        # Log successful processing (REAL log)
        logger.info(
            'Processing completed successfully',
            extra={
                'request_id': request_id,
                'status': 'success',
                'duration_ms': duration_ms
            }
        )

        return {
            'request_id': request_id,
            'result': result,
            'items_processed': len(items),
            'operation': operation,
            'duration_ms': duration_ms
        }, 200

    except Exception as e:
        # Log errors (REAL log)
        duration_ms = int((time.time() - start_time) * 1000)
        logger.error(
            f'Processing failed: {str(e)}',
            extra={
                'request_id': request_id,
                'status': 'error',
                'duration_ms': duration_ms
            }
        )
        return {'error': 'Internal server error', 'request_id': request_id}, 500

@app.route('/batch', methods=['POST'])
def batch_process():
    """
    Batch processing endpoint - simulates batch jobs.
    Generates multiple logs as it processes items.
    """
    request_id = generate_request_id()
    start_time = time.time()

    # Log batch start (REAL log)
    logger.info(
        'Batch processing started',
        extra={'request_id': request_id}
    )

    data = request.get_json() or {}
    batches = data.get('batches', 5)

    processed = 0
    for i in range(batches):
        # Simulate processing each batch item
        time.sleep(0.1)  # Simulate work
        processed += 1

        # Log progress (REAL log)
        if i % 2 == 0:  # Log every other item
            logger.info(
                f'Batch progress: {processed}/{batches} items processed',
                extra={'request_id': request_id}
            )

    duration_ms = int((time.time() - start_time) * 1000)

    # Log batch completion (REAL log)
    logger.info(
        'Batch processing completed',
        extra={
            'request_id': request_id,
            'status': 'success',
            'items_processed': processed,
            'duration_ms': duration_ms
        }
    )

    return {
        'request_id': request_id,
        'items_processed': processed,
        'duration_ms': duration_ms
    }, 200

if __name__ == '__main__':
    app.run(host='0.0.0.0', port=8080)

Create requirements.txt:

Flask==3.0.0
gunicorn==21.2.0

Create Dockerfile:

FROM python:3.12-slim

WORKDIR /app

COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY app.py .

CMD ["gunicorn", "--bind", "0.0.0.0:8080", "--workers", "2", "--threads", "4", "--timeout", "60", "--access-logfile", "-", "--error-logfile", "-", "--log-level", "error", "app:app"]

Step 2: Deploy to Cloud Run

# Set your project and region
export PROJECT_ID="your-project-id"
export REGION="europe-west2"
export SERVICE_NAME="data-processor"

gcloud config set project $PROJECT_ID

# Build and deploy
gcloud builds submit --tag gcr.io/$PROJECT_ID/$SERVICE_NAME

gcloud run deploy $SERVICE_NAME \
  --image gcr.io/$PROJECT_ID/$SERVICE_NAME \
  --region=$REGION \
  --platform=managed \
  --allow-unauthenticated \
  --port=8080

# Get the service URL
SERVICE_URL=$(gcloud run services describe $SERVICE_NAME --region=$REGION --format='value(status.url)')
echo "Service URL: $SERVICE_URL"

Step 3: Create Pub/Sub Topic

gcloud pubsub topics create senseon-logs

Step 4: Create Log Sink

# Create sink that captures structured JSON logs from your Cloud Run service
gcloud logging sinks create senseon-log-sink \
  pubsub.googleapis.com/projects/$PROJECT_ID/topics/senseon-logs \
  --log-filter='resource.type="cloud_run_revision"
                AND resource.labels.service_name="'$SERVICE_NAME'"
                AND jsonPayload.level!=""'

Filter explanation: - resource.type="cloud_run_revision" - Only Cloud Run logs - resource.labels.service_name="data-processor" - Only your specific service - jsonPayload.level!="" - Only structured JSON logs (filters out HTTP request logs)

Step 5: Grant Pub/Sub Permissions

# Get the service account from the sink
SERVICE_ACCOUNT=$(gcloud logging sinks describe senseon-log-sink --format='value(writerIdentity)')

# Grant publisher permission
gcloud pubsub topics add-iam-policy-binding senseon-logs \
  --member=$SERVICE_ACCOUNT \
  --role=roles/pubsub.publisher

Step 6: Create Log Forwarder Cloud Function

Create main.py:

import base64
import json
import requests
import functions_framework
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# Replace with your SenseOn collector endpoint
COLLECTOR_URL = "https://<collector-name>.snson.net/log_sink"

@functions_framework.cloud_event
def forward_logs(cloud_event):
    """
    Forward real GCP logs to SenseOn collector.

    Forwards RAW JSON logs without wrapping
    """
    try:
        # Decode Pub/Sub message
        pubsub_message = base64.b64decode(cloud_event.data["message"]["data"]).decode('utf-8')
        log_entry = json.loads(pubsub_message)

        # Extract service name for logging
        resource = log_entry.get("resource", {})
        resource_labels = resource.get("labels", {})
        service_name = (
            resource_labels.get("service_name") or
            resource_labels.get("function_name") or
            "unknown"
        )

        # Forward the raw log data directly (not wrapped)
        if "jsonPayload" in log_entry:
            # This is structured JSON log (like our nginx JSON)
            senseon_log = log_entry["jsonPayload"]

            # Add minimal source tag
            senseon_log["source"] = f"gcp-{resource.get('type', 'unknown')}"
            senseon_log["_gcp_service"] = service_name

        elif "textPayload" in log_entry:
            # Plain text log - wrap minimally
            senseon_log = {
                "message": log_entry["textPayload"],
                "timestamp": log_entry.get("timestamp"),
                "source": f"gcp-{resource.get('type', 'unknown')}",
                "_gcp_service": service_name
            }
        else:
            # Skip logs without payload
            logger.info(f"Skipping log without payload from {service_name}")
            return

        # Forward to SenseOn collector
        response = requests.post(
            COLLECTOR_URL,
            json=senseon_log,
            headers={"Content-Type": "application/json"},
            timeout=10
        )

        if response.status_code == 201:
            logger.info(f"Log forwarded: {service_name} (HTTP {response.status_code})")
        else:
            logger.warning(f"Unexpected status: {response.status_code} for {service_name}")

    except Exception as e:
        logger.error(f"Error forwarding log: {e}")
        raise

Create requirements.txt:

functions-framework==3.*
requests==2.*

Step 7: Deploy Cloud Function

gcloud functions deploy senseon-log-forwarder \
  --gen2 \
  --runtime=python311 \
  --region=$REGION \
  --source=. \
  --entry-point=forward_logs \
  --trigger-topic=senseon-logs \
  --timeout=60s \
  --memory=256MB

Step 8: Verify End-to-End

Generate test logs:

# Test the /process endpoint
curl -X POST $SERVICE_URL/process \
  -H "Content-Type: application/json" \
  -H "X-User-ID: user-123" \
  -d '{"items": [10, 20, 30], "operation": "sum"}'

# Test the /batch endpoint
curl -X POST $SERVICE_URL/batch \
  -H "Content-Type: application/json" \
  -d '{"batches": 5}'

Check Cloud Run logs:

gcloud logging read \
  'resource.type="cloud_run_revision"
   AND resource.labels.service_name="'$SERVICE_NAME'"' \
  --limit=10 \
  --format=json

Check Cloud Function forwarded the logs:

gcloud functions logs read senseon-log-forwarder \
  --region=$REGION \
  --limit=10

# Should see: "Log forwarded: data-processor (HTTP 201)"

What Gets Forwarded

Your application logs this (from a /process request):

{
  "timestamp": "2025-10-17T10:00:00.123456Z",
  "level": "INFO",
  "logger": "data_processor",
  "message": "Processing request received",
  "request_id": "req-1697542800-1234",
  "user_id": "user-123"
}

SenseOn receives this:

{
  "timestamp": "2025-10-17T10:00:00.123456Z",
  "level": "INFO",
  "logger": "data_processor",
  "message": "Processing request received",
  "request_id": "req-1697542800-1234",
  "user_id": "user-123",
  "source": "gcp-cloud-run-revision",
  "_gcp_service": "data-processor"
}

Multiple logs from a single request:

// 1. Request received
{
  "timestamp": "2025-10-17T10:00:00.123456Z",
  "level": "INFO",
  "message": "Processing request received",
  "request_id": "req-1697542800-1234",
  "user_id": "user-123"
}

// 2. Processing
{
  "timestamp": "2025-10-17T10:00:00.234567Z",
  "level": "INFO",
  "message": "Processing 3 items with operation: sum",
  "request_id": "req-1697542800-1234"
}

// 3. Completion
{
  "timestamp": "2025-10-17T10:00:00.345678Z",
  "level": "INFO",
  "message": "Processing completed successfully",
  "request_id": "req-1697542800-1234",
  "status": "success",
  "duration_ms": 2
}

Clean, parsed JSON with minimal metadata added.

Troubleshooting

No logs forwarded:

# Check log sink is active
gcloud logging sinks describe senseon-log-sink

# Check Pub/Sub has messages
gcloud pubsub subscriptions list --filter="topic:senseon-logs"

# Check Cloud Function errors
gcloud functions logs read senseon-log-forwarder --region=$REGION --limit=50

Function returns errors:

Verify COLLECTOR_URL in main.py is correct for your environment
Check function has internet access (Gen2 functions require VPC connector for private networks)
Test collector URL: curl -X POST $COLLECTOR_URL -H "Content-Type: application/json" -d '{"test":"connection"}'

Adapting for Other GCP Services

Cloud Functions:

Same approach, change filter to resource.type="cloud_function"

Compute Engine:

Install Cloud Logging agent on VMs
Change filter to resource.type="gce_instance"

GKE:

GKE automatically forwards to Cloud Logging
Use this pattern OR deploy FluentBit DaemonSet (see Kubernetes section)
Cloud Logging is easier but FluentBit gives more control

Verification and Testing

Test Your Configuration

After configuring your log source, test the integration:

# Test with curl
curl -X POST https://<collector-name>.snson.net/log_sink \
  -H "Content-Type: application/json" \
  -d '{
    "timestamp": "2025-10-14T15:30:45Z",
    "source": "test",
    "message": "Configuration test log"
  }'

Troubleshooting

Common Issues

Logs not being forwarded

Check forwarder service is running
Verify network connectivity to collector
Review forwarder error logs
Confirm JSON format is valid

Connection timeouts

Verify collector endpoint URL is correct
Check firewall rules allow HTTPS (443) outbound
Test network connectivity: curl https://<collector-name>.snson.net/health
Review DNS resolution

Getting Help

If you encounter issues:

Check the troubleshooting Guide for common troubleshooting steps
Review your solutions error logs
Test the collector endpoint with curl
Contact SenseOn support with configuration details and error messages

Detailed Configuration Examples

Prerequisites

Minimum Requirements

What You'll Need

How to Use This Guide

Table of Contents

Examples

Generic Patterns

Testing Your Collector

GCE VM Nginx with FluentBit

Configuration Overview

Prerequisites

Step 1: Configure Nginx JSON Logging

Step 2: Restart Nginx

Step 3: Install FluentBit

Step 4: Configure FluentBit

Step 5: Start FluentBit

Step 6: Verify Log Forwarding

Example Log Output Sent to SenseOn

Syslog (rsyslog)

Configuration Overview

Step 1: Install rsyslog HTTP Module

Step 2: Configure rsyslog

Step 3: Restart rsyslog

Example Log Output

FluentBit Log Forwarder

Configuration Overview

Installation

Configuration File

Start FluentBit

GKE Nginx with FluentBit

Configuration Overview

The Problem: Nested JSON

The Solution: Two-Stage Parsing

Prerequisites

Step 1: Create Namespace

Step 2: Create FluentBit DaemonSet

Step 3: Deploy to Kubernetes

Step 4: Verify Log Forwarding

Step 5: Generate Test Logs

How the Parsing Works

Troubleshooting Kubernetes Logs

Important Notes

Cloud Run Nginx with Cloud Logging

Configuration Overview

Prerequisites

Step 1: Create Nginx Configuration Files

Step 2: Deploy to Cloud Run

Step 3: Set Up Log Forwarding

Step 4: Deploy Log Forwarder Function

Step 5: Verify End-to-End

What Gets Sent to SenseOn

Cloud Run Application Logs

Configuration Overview

Prerequisites

Step 1: Configure Application to Log JSON

Step 2: Deploy to Cloud Run

Step 3: Create Pub/Sub Topic

Step 4: Create Log Sink

Step 5: Grant Pub/Sub Permissions

Step 6: Create Log Forwarder Cloud Function

Step 7: Deploy Cloud Function

Step 8: Verify End-to-End

What Gets Forwarded

Troubleshooting

Adapting for Other GCP Services

Verification and Testing

Test Your Configuration

Troubleshooting

Common Issues

Getting Help