Distributed Processing on Edge

A cloud-native edge computing platform for real-time IoT sensor data ingestion, outlier detection, and time-series storage. Built to evaluate the performance and feasibility of running distributed data processing workloads on constrained edge infrastructure using Kubernetes (k3s).

Architecture Overview

flowchart TB
    subgraph Devices["Device Layer"]
        IOT["IoT Sensors\n(temperature)"]
    end

    subgraph Messaging["Message Broker"]
        RMQ[("RabbitMQ\n(AMQP / MQTT)")]
    end

    subgraph Edge["Edge Processing Layer"]
        EP1["Edge Processor\nReplica 1"]
        EP2["Edge Processor\nReplica 2"]
        EP3["Edge Processor\nReplica 3"]
    end

    subgraph Storage["Data & State Layer"]
        REDIS[("Redis\n• Window buffering\n• Distributed locks\n• Hampel history\n• Dead Letter Queue")]
        INFLUX[("InfluxDB\nTime-series storage")]
    end

    subgraph Observability["Observability"]
        PROM["Prometheus\nMetrics scraping"]
        GRAF["Grafana\nDashboards"]
    end

    IOT -->|"MQTT"| RMQ
    RMQ -->|"AMQP\nconsume"| EP1
    RMQ -->|"AMQP\nconsume"| EP2
    RMQ -->|"AMQP\nconsume"| EP3

    EP1 <-->|"buffer / lock / dlq"| REDIS
    EP2 <-->|"buffer / lock / dlq"| REDIS
    EP3 <-->|"buffer / lock / dlq"| REDIS

    EP1 -->|"filtered points"| INFLUX
    EP2 -->|"filtered points"| INFLUX
    EP3 -->|"filtered points"| INFLUX

    EP1 -.->|"/metrics"| PROM
    EP2 -.->|"/metrics"| PROM
    EP3 -.->|"/metrics"| PROM
    PROM -.-> GRAF

Loading

The pipeline processes IoT sensor readings through four stages:

1. Ingest – Sensor data arrives via RabbitMQ (AMQP) from MQTT-connected devices
2. Buffer – Messages are grouped into configurable time windows (e.g., 5 s) in Redis
3. Filter – Closed windows are processed with a Hampel filter (median + MAD) for real-time outlier detection
4. Store – Cleaned data is written to InfluxDB; failed writes are retried through a Dead Letter Queue

Project Structure

├── src/
│   ├── edge/                    # Edge processing microservice (Go)
│   │   ├── main.go              # Entry point — wires all components
│   │   ├── models/              # Data structures (Reading, ProcessedPoint, Config, ...)
│   │   ├── services/
│   │   │   ├── window/          # Time-window buffering + processing orchestration
│   │   │   │   ├── manager.go   # Message handling and window lifecycle
│   │   │   │   ├── processor.go # Window closure, Hampel filtering, InfluxDB writes
│   │   │   │   └── lock.go      # Redis-based distributed lock (prevents duplicate processing)
│   │   │   ├── dlq/             # Dead Letter Queue with exponential backoff retry
│   │   │   ├── redis/           # Redis client (buffers, window index, Hampel history, DLQ)
│   │   │   ├── influxdb/        # InfluxDB client (time-series writes)
│   │   │   ├── metrics/         # Prometheus metric collectors (counters, gauges, histograms)
│   │   │   └── processing/      # Hampel filter implementation
│   │   ├── common/              # Shared utilities (AMQP/RabbitMQ connection)
│   │   ├── Dockerfile           # Multi-stage container build
│   │   ├── go.mod / go.sum
│   │   └── .env                 # Local development config
│   │
│   ├── throughput-test/         # Load-testing CLI (Go)
│   │   ├── main.go              # CLI entry point
│   │   ├── cmd/                 # Cobra commands (test, constant, monitor)
│   │   ├── load/                # Message generator + Prometheus metrics
│   │   └── metrics/             # Test metrics collection
│   │
│   └── dataset/                 # IoT temperature sensor datasets for testing
│       ├── final.csv
│       ├── IOT-temp.csv
│       ├── IOT-temp-transformed.csv
│       ├── generate_iot_dataset.ipynb
│       └── transform_iot_data.ipynb
│
├── k3s/                         # Kubernetes (k3s) deployment manifests
│   ├── data_processing/         # Edge processor Deployment, ConfigMap, Secrets, Service
│   ├── mqtt/                    # RabbitMQ cluster operator + deployment
│   ├── cloud/                   # InfluxDB deployment
│   ├── monitoring/              # Prometheus + Grafana (+ dashboard ConfigMap)
│   └── set-kubeconfig.sh        # Helper to configure kubectl for edge clusters
│
├── ausible/                     # Ansible playbooks for VM provisioning
│   ├── playbook.yml             # Main provisioning playbook
│   ├── playbook-aws.yml         # AWS-specific provisioning
│   ├── requirements.yml         # Ansible Galaxy collections
│   └── roles/                   # Ansible roles
│
├── .github/workflows/           # CI/CD — builds and pushes edge image to ghcr.io
├── slides/                      # Thesis presentation slides (images)
├── LaTex/                       # Thesis LaTeX source
├── pyproject.toml               # Python dev dependencies (ruff, pandas, etc.)
└── README.md

Edge Processor

The core microservice (src/edge/) is a Go application that consumes IoT sensor messages and transforms them into clean, time-series data.

Configuration

All settings are passed via environment variables or a .env file:

Variable	Default	Description
AMQP_URL	amqp://guest:guest@localhost:5672/	RabbitMQ connection URL
REDIS_URL	redis://localhost:6379	Redis connection URL
INFLUXDB_URL	http://localhost:8888	InfluxDB connection URL
INFLUXDB_TOKEN	—	InfluxDB authentication token
INFLUXDB_DBNAME	lab	InfluxDB bucket/database name
WINDOW_SIZE_SECONDS	5	Time window duration for grouping readings
GRACE_PERIOD_SECONDS	10	Extra time to wait for late-arriving messages
HAMPEL_WINDOW_SIZE	7	Number of historical values for the Hampel filter
HAMPEL_THRESHOLD	3.0	MAD multiplier for outlier detection
DLQ_MAX_RETRIES	10	Maximum retry attempts for failed InfluxDB writes
DLQ_BASE_DELAY_SECONDS	30	Base delay for exponential backoff
DLQ_MAX_DELAY_SECONDS	3600	Maximum delay between retries (1 h)
DLQ_PROCESSOR_INTERVAL_SECONDS	30	DLQ polling interval

Data Flow

flowchart TB
    A["📨 RabbitMQ\nmessage arrives"] --> B["🧮 Assign to time window\n window:{start_ts}:{device_id}"]
    B --> C["💾 Store in Redis\n(sorted set index)"]
    C --> D{"⏰ Window closed?\n(now > end + grace period)"}
    D -->|"no"| C
    D -->|"yes"| E["🔒 Acquire distributed lock\n lock:window:{key}"]
    E --> F{"Lock acquired?"}
    F -->|"no — another replica\nis processing it"| G["⏭️ Skip window"]
    F -->|"yes"| H["📊 Read readings from Redis"]
    H --> I["🔬 Apply Hampel filter\n(median + MAD)"]
    I --> J{"Outlier detected?"}
    J -->|"yes"| K["✏️ Replace with median\nflag as filtered"]
    J -->|"no"| L["✅ Keep original value"]
    K --> M["📝 Write processed point\nto InfluxDB"]
    L --> M
    M --> N{"Write succeeded?"}
    N -->|"yes"| O["🧹 Cleanup Redis window data"]
    N -->|"no"| P["📥 Store in Dead Letter Queue\n(exponential backoff)"]
    P --> Q["🔄 Retry DLQ:\nbase × 2^retry\n(capped at 1 h)"]
    Q --> M
    O --> R["✅ Window processed"]

Loading

Step-by-step:

1. Message consumption – The processor listens for JSON sensor readings on RabbitMQ:

   {
     "device_id": "sensor-01",
     "timestamp": 1718000000,
     "temperature": 22.5,
     "direction": "in",
     "room_id": "room-a"
   }

2. Time-window buffering – Each message is assigned to an aligned time window (e.g., every 5 s on the 0, 5, 10, ... boundary) and stored in Redis under window:{start_ts}:{device_id}. A sorted set index tracks all open windows.

3. Window closure check – A periodic ticker identifies windows whose end time + grace period has passed. Only closed windows advance to processing.

4. Distributed locking – Before processing, the replica acquires a Redis lock (lock:window:{key}). In a multi-replica deployment, this guarantees exactly one instance processes each window.

5. Hampel outlier filtering – Each temperature reading is evaluated against a sliding history window:

   • Computes the median and Median Absolute Deviation (MAD) of recent values
   • If |value - median| > threshold × MAD, the value is replaced with the median
   • Filtered values are flagged in the output

6. Storage – Processed points are written to InfluxDB with device, room, direction, temperature, and a filtering flag.

7. Dead Letter Queue – If an InfluxDB write fails, the point is stored in a Redis DLQ with exponential backoff retries (base × 2^retry, capped at 1 h). After exhausting retries, the item remains in the DLQ for manual inspection.

Endpoints

Port	Endpoint	Description
8080	/health	Health check (liveness probe)
9090	/metrics	Prometheus metrics

Metrics (Prometheus)

The service exposes detailed operational metrics:

• Counters: Total messages received, processed, errors by type
• Gauges: Node up/down, health status, memory usage, CPU (goroutine count), queue depth, active windows, Redis buffer sizes, Hampel outlier count
• Histograms: Message processing duration, window processing duration, Hampel filter duration, database write duration
• DLQ metrics: Retry attempts and successes per device

Throughput Testing

The throughput-test CLI tool evaluates edge processor performance under controlled load.

Usage:
  throughput-test test [dataset]     Progressive load test (ramping rate)
  throughput-test constant [dataset] Constant-rate load test
  throughput-test monitor            Live metrics monitoring

Progressive Test

The default test mode ramps the message rate from 0 to --max-rate in steps of --step-duration seconds each. This helps identify the saturation point of the edge processor.

throughput-test test dataset.csv \
  --rabbitmq-url "amqp://..." \
  --max-rate 1000 \
  --step-duration 120 \
  --output results.json

Constant Test

Runs at a fixed message rate for a specified duration — useful for sustained-load characterization.

throughput-test constant dataset.csv \
  --rate 200 \
  --duration 300 \
  --output results.json

Monitor Mode

Starts a live monitoring endpoint that scrapes the edge service's Prometheus metrics and exposes them on a local port.

throughput-test monitor \
  --prometheus-url "http://edge:9090/metrics" \
  --monitor-port 8080

Both test modes produce JSON output files containing per-step metrics, latency distributions, error counts, and resource utilization snapshots.

Deployment

1. VM Provisioning (Ansible)

Provision VMs with the required dependencies:

ansible-galaxy collection install -r ansible/requirements.yml
ansible-playbook -v -K ansible/playbook.yml

Prerequisites:

• An SSH key pair (id_ed25519_lab) placed in ./.ssh/
• The key added to your SSH agent

2. Cluster Setup (k3s)

Configure kubectl for your edge cluster:

./k3s/set-kubeconfig.sh <CLUSTER_NAME> <MASTER_IPV4>

Install the required system components:

On the MQTT cluster:

kubectl apply -f "https://github.com/rabbitmq/cluster-operator/releases/latest/download/cluster-operator.yml"
kubectl apply -f ./k3s/mqtt/k8s-deployment.yml

On the main cluster:

kubectl apply -f ./k3s/cloud/k8s-influxdb.yaml
kubectl apply -f ./k3s/data_processing/k8s-redis.yaml
kubectl apply -f ./k3s/monitoring/k8s-prometheus.yaml
kubectl apply -f ./k3s/data_processing/k8s-edge.yaml
kubectl apply -f ./k3s/monitoring/k8s-grafana.yaml

3. Edge Processor Configuration

The edge deployment uses a ConfigMap for tunable parameters and a Secret for credentials. Update k3s/data_processing/k8s-edge.yaml with your environment-specific values.

4. Grafana Dashboard

The Grafana dashboard definition is at k3s/monitoring/dashboards/dashboard.json. To update it:

kubectl create configmap grafana-dashboard \
  --from-file=dashboard.json=k3s/monitoring/dashboards/dashboard.json \
  --dry-run=client -o yaml | kubectl apply -f -
kubectl apply -f ./k3s/monitoring/k8s-grafana.yaml

CI/CD

A GitHub Actions workflow (.github/workflows/build-and-push-edge.yml) builds the edge processor Docker image and pushes it to ghcr.io. Trigger it manually via workflow_dispatch.

Local container build:

docker build -t edge:latest ./src/edge

For air-gapped k3s clusters, import directly:

podman save localhost/edge:latest | k3s ctr images import -

Dataset

The src/dataset/ directory contains a synthetic IoT temperature sensor dataset created to benchmark the edge processor under controlled conditions. Because real-world sensor data is often sparse, irregular, or privacy-sensitive, we generated a synthetic dataset with known characteristics — making it possible to precisely measure throughput, outlier detection accuracy, and window-processing behavior at scale.

File	Description
final.csv	Curated synthetic dataset used in throughput tests
IOT-temp.csv	Raw synthetic temperature readings
IOT-temp-transformed.csv	Transformed/cleaned version
generate_iot_dataset.ipynb	Jupyter notebook that generates the synthetic dataset
transform_iot_data.ipynb	Jupyter notebook for data transformation and analysis

Key Design Decisions

• Time-window alignment – Windows are aligned to the wall clock (e.g., seconds: 0, 5, 10, ...), making it easy to reason about window boundaries and enabling downstream consumers to correlate windows across replicas.
• Distributed locking with Redis – SET NX EX + Lua scripts ensure exactly-once window processing semantics in a multi-replica deployment without external coordination.
• Exponential backoff for DLQ – Failed writes retry with exponentially increasing delays, preventing thundering-herd problems against InfluxDB during transient outages.
• Hampel filter instead of simple threshold – The Hampel identifier adapts to local data variability using median and MAD, making it robust against both sudden spikes and gradual trend shifts without manually tuning per-device thresholds.
• Prometheus-native metrics – All operational telemetry is exposed via a standard /metrics endpoint, compatible with the broader observability ecosystem (Prometheus + Grafana).

Future Work

Based on the experimental evaluation of the platform, the following areas have been identified for future development:

1. Larger-scale testing – Validate the infrastructure with hundreds of IoT simulators and loads in the order of hundreds of msg/s to identify the extreme scalability limits of the architecture.

2. Time-series write buffer – Introduce an intermediate buffering layer (e.g., Kafka or an additional Redis queue) to absorb write spikes and reduce pressure on InfluxDB, which was identified as the main bottleneck under high load.

3. Predictive monitoring – Leverage Prometheus metrics to implement predictive alerting (e.g., forecasting DLQ saturation, Redis performance degradation, or memory exhaustion before they become critical).

4. Network latency experiments – Add test scenarios with variable network latency to simulate real-world Fog–Cloud connectivity conditions, including intermittent link failures.

5. Comparative evaluation – Benchmark against alternative technology stacks (e.g., Kafka instead of RabbitMQ, traditional SQL databases instead of InfluxDB) to further validate architectural decisions.

6. Redis sharding for HA – Address the single point of failure represented by the shared Redis state by introducing Redis Cluster sharding, allowing the system to survive the loss of multiple nodes without data loss.

References

• Kleppmann, M. Designing Data-Intensive Applications
• Newman, S. Building Microservices
• Jeff Geerling. Ansible for DevOps
• Roos-Hoefgeest Toribio, M. et al. "A Novel Approach to Speed Up Hampel Filter for Outlier Detection." Sensors, 2025
• Oliveira, F. B. et al. "IoTDeploy: Deployment of IoT Smart Applications over the Computing Continuum." Internet of Things, 2024
• Vaño, R. et al. "Cloud-Native Workload Orchestration at the Edge: A Deployment Review and Future Directions." Sensors, 2023

Slides