Detailed Configuration Examples

Prerequisites

SenseOn accepts any valid JSON.

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:

1. Format your logs as JSON - Either at the source or using a log processor
2. Send to your collector URL - Using a log forwarder or direct from your application
3. Include useful metadata - 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://<code-name>-collector.snson.net)
• Outbound HTTPS access - Port 443 to the collector endpoint
• Log source - Any application, service, or system generating logs
• Log forwarder (optional) - Such as FluentBit

---

How to Use This Guide

1. Find a similar example - The patterns shown here (nginx, syslog, Kubernetes) demonstrate common scenarios
2. Adapt to your source - Replace log paths, parsers, and filters with your specific needs
3. Test incrementally - Start with a simple curl test, then configure your forwarder
4. 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: get valid JSON to the /log_sink endpoint.

---

Table of Contents

Production-Tested Examples

1. GKE Nginx - Nginx on Google Kubernetes Engine
2. GCE VM Nginx - Nginx on Google Compute Engine VM
3. Cloud Run Nginx - Nginx on Cloud Run (serverless)
4. Cloud Run Application - Custom application on Cloud Run

Generic Patterns

1. Syslog (rsyslog)
2. FluentBit Generic

---

Testing Your Collector

Before configuring any log forwarder, test your collector endpoint:

# Replace <code-name> with your actual collector code name
curl -X POST https://<code-name>-collector.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 <code-name>-collector.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 (secure method for Ubuntu 20.04+)
wget -qO /usr/share/keyrings/fluentbit-archive-keyring.gpg https://packages.fluentbit.io/fluentbit.key
echo "deb [signed-by=/usr/share/keyrings/fluentbit-archive-keyring.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            <code-name>-collector.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

Notable settings functionalities: - 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)

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/

# 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="<code-name>-collector.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="<code-name>-collector.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-archive-keyring.gpg https://packages.fluentbit.io/fluentbit.key
echo "deb [signed-by=/usr/share/keyrings/fluentbit-archive-keyring.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            <code-name>-collector.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 K8s 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

Parse the log in two stages: 1. Strip the Docker wrapper (timestamp, stream, tag) 2. Parse your application JSON from the extracted message

The configuration below implements this.

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            <code-name>-collector.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] <code-name>-collector.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: 1. Two parser filters (docker_strip, then nginx_json or app_json) 2. Reserve_Data On on both parsers 3. modify filter removing wrapper fields (log, stream, time, message, logtag) 4. 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

Notes

Best practice configuration details: - Use Format json_lines (preferred over 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

---

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://<code-name>-collector.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 - exactly as customers would.
    """
    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 (customers would do this)
            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 is a realistic customer application that:
- 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 (what customers would do)
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():
    """
    Process data endpoint - this is what a real customer app does.
    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://<code-name>-collector.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 - exactly as customers would.
    """
    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 (customers would do this)
            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 - 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

Why This Pattern Is Realistic

For GCP customers: - Uses native GCP services (no additional infrastructure) - Serverless (fully managed, no VMs to maintain) - Auto-scaling (handles any log volume) - Cost-effective for low-to-medium volumes - Simple code (< 50 lines of Python)

Common in production: - Microservices on Cloud Run - Serverless applications - Cloud-native organisations - Teams wanting minimal operational overhead

---

Verification and Testing

Test Your Configuration

After configuring your log source, test the integration:

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

Monitor Log Flow

1. Check your log forwarder logs for errors
2. Verify logs are being read from source files
3. Monitor network connectivity to the collector
4. Contact SenseOn support to confirm log reception

---

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 - Review DNS resolution

---

Getting Help

If you encounter issues:

1. Check the main ingestion guide for common troubleshooting steps
2. Review your forwarder's error logs
3. Test the collector endpoint with curl
4. Contact SenseOn support with configuration details and error messages

Need help?

Contact SenseOn support at support@senseon.io

When contacting support, include:

• Your organisation name
• Collector endpoint URL
• Example log entries (sanitised)
• Relevant error messages (if applicable)
• Log source type
• Configuration files (if applicable)