Real-Time Multimodal IoT Data Platform

This repository implements a reproducible edge-to-cloud architecture for multimodal IoT ingestion, schema governance, stream/batch processing, high-availability transactional storage, and operational observability. The stack targets real-time livestock and precision-agriculture workloads, but the design is generic enough for other industrial IoT domains.

1. What the platform does

The platform combines six logical planes:

1. Edge and ingestion

   • MQTT ingestion through a 3-node VerneMQ cluster.
   • An MQTT-to-Kafka bridge that forwards telemetry, GPS messages, and media metadata into Kafka topics.

2. Event backbone and schema governance

   • A 3-node Apache Kafka KRaft cluster.
   • Apicurio Registry for JSON schema governance and artifact lifecycle.

3. Object storage

   • SeaweedFS S3 as object storage for raw and derived media objects.
   • Bucket notifications that emit media.object.events to Kafka.

4. Stream and batch processing

   • Apache Flink for stateful stream processing with checkpointing and savepoints on SeaweedFS S3.
   • Apache Beam job server and Python harness for portable pipelines.
   • Apache Airflow 3 for scheduled orchestration and replay/backfill workflows.

5. Operational and analytical storage

   • A 3-node Patroni/etcd-backed TimescaleDB HA cluster.
   • HAProxy for read-write and read-only routing.
   • PgBouncer for client pooling.

6. Observability and administration

   • Prometheus, Grafana, cAdvisor, statsd-exporter, postgres-exporter.
   • pgAdmin for database inspection.

2. High-level architecture

flowchart LR
  subgraph Edge
    D[IoT devices]
    M[Media producers]
  end

  subgraph Ingestion
    V[VerneMQ cluster]
    B[MQTT-Kafka bridge]
  end

  subgraph Backbone
    K[Kafka KRaft cluster]
    R[Apicurio Registry]
    O[SeaweedFS S3]
  end

  subgraph Processing
    F[Flink]
    J[Beam job server]
    A[Airflow 3]
  end

  subgraph Storage
    H[HAProxy]
    P[(Patroni + TimescaleDB HA)]
    G[PgBouncer]
  end

  subgraph Observability
    PR[Prometheus]
    GF[Grafana]
  end

  D --> V --> B --> K
  M --> O --> K
  K --> R
  K --> F
  K --> J
  A --> O
  A --> K
  F --> O
  F --> H --> P
  G --> H
  PR --> GF

Loading

3. Repository layout

airflow/                  Airflow DAGs, logs, plugins
apicurio/bootstrap/       Registry bootstrap artifacts
beam/                     Beam job server and Python harness images
deploy/                   Docker Swarm and Kubernetes production targets
flink/                    Custom PyFlink image and jobs
grafana/                  Dashboards and provisioning
haproxy/                  Read-write / read-only routing
kafka-connect/            Debezium connector definitions
mqtt-kafka-bridge/        MQTT to Kafka bridge
management-console/       Internal architecture management console
orchestration/            Airflow image build
patroni/                  Patroni templates
pgadmin/                  pgAdmin preconfigured servers
pgbouncer/                Explicit pooling configuration
pipelines/                Replay/backfill utilities
prometheus/               Metrics scraping and alert rules
scripts/                  Bootstrap and entrypoint helpers
secrets/                  Docker secrets (local dev only)

4. Critical design decisions finalized in this version

Fixed inconsistencies

This finalized version addresses the following problems found during the audit:

• The Airflow DAG passed --window-start and --window-end, but pipelines/media_backfill.py only accepted --since-minutes.
• The Airflow containers did not receive Kafka and S3/SeaweedFS credentials, so the backfill workflow could not read objects or publish replayed events.
• Kafka Connect internal topics (__connect-configs, __connect-offsets, __connect-status) were missing while Kafka auto-topic creation was disabled.
• The raw.gps topic existed but no matching schema artifact was bootstrapped into Apicurio Registry.
• The MQTT bridge emitted a generic record that did not match the intended telemetry contract.
• PgBouncer shipped dedicated config files and userlists that were not mounted, which made the repository harder to reason about.
• The environment-specific compose overlays were empty.

Final behavior of the backfill pipeline

The Airflow DAG now calls a backfill script that:

1. Reads objects from SeaweedFS S3 in a bounded window.
2. Maps the bucket and media kind to one of:
   • raw.image2d.meta
   • raw.image3d.meta
   • raw.video2d.meta
   • raw.video3d.meta
3. Publishes synthetic metadata events to Kafka so downstream consumers can reprocess missed media arrivals.

This makes the Airflow workflow operational instead of only printing objects to stdout.

5. Topics and contracts

Kafka topics created during bootstrap

Topic	Purpose	Partitions	Replication
raw.gps	GPS events	24	3
raw.sensor	Telemetry events	24	3
raw.image2d.meta	2D image metadata	12	3
raw.image3d.meta	3D image metadata	12	3
raw.video2d.meta	2D video metadata	12	3
raw.video3d.meta	3D video metadata	12	3
media.object.events	S3 object notifications	12	3
features.events	Derived features	24	3
alerts.events	Alerts	12	3
state.latest	Compacted latest state	12	3
dlq.events	Dead-letter events	12	3
governance.data.products	DGA data product catalogue	3	3
governance.access.requests	DGA access requests and decisions	6	3
governance.permission.events	DGA consent/permission lifecycle	6	3
governance.intermediation.log	DGA intermediation activity log	12	3
governance.transfer.notices	DGA unauthorised access/transfer/use notices	6	3
governance.research.projects	Research project register and ethics status	3	3
governance.research.outputs	Research outputs and disclosure review evidence	6	3
governance.dataset.catalog	Dataset catalogue, FAIR metadata and access policy	3	3
governance.data_management_plans	Data Management Plans for research and releases	3	3
governance.repository.exports	Repository, Zenodo and OpenAIRE export evidence	6	3
dataact.product.catalog	Data Act connected-product and related-service catalogue	3	3
dataact.user.access.requests	Data Act user access requests and decisions	6	3
dataact.third_party.sharing	User-authorized third-party sharing evidence	6	3
dataact.user.exports	User export and delivery evidence	6	3
dataact.safeguards	Data Act security and trade-secret safeguards	3	3
dataact.legal_basis.checks	Data Act and personal-data release checks	6	3
security.asset.inventory	Security asset inventory and ownership	3	3
security.incident.events	NIS2/DORA/CRA incident evidence	6	3
security.vulnerability.findings	Vulnerability and remediation register	6	3
security.sbom.attestations	SBOM, provenance and signature evidence	3	3
security.patch.events	Security update and rollback evidence	6	3
resilience.backup.tests	Restore test, RPO and RTO evidence	3	3
resilience.operational.risk	ICT and operational risk register	3	3
resilience.third_party.risk	ICT supplier and exit-plan register	3	3
compliance.scope.decisions	Regulatory scope decision register	3	3
compliance.control.assessments	Compliance control assessment evidence	6	3
compliance.reporting.channels	Regulatory reporting channel register	3	3
compliance.legal.dossier	Legal dossier artefacts and release gates	3	3
cra.product.lifecycle	CRA product support and update lifecycle	3	3
kafkasql-journal-v3	Apicurio KafkaSQL journal	1	3
kafkasql-snapshots-v3	Apicurio KafkaSQL snapshots	1	3
registry-events-v3	Apicurio registry events	1	3
__connect-configs	Kafka Connect internal config topic	1	3
__connect-offsets	Kafka Connect internal offsets topic	12	3
__connect-status	Kafka Connect internal status topic	6	3

Registry artifacts bootstrapped

• raw.sensor
• raw.gps
• raw.image2d.meta
• raw.image3d.meta
• raw.video2d.meta
• raw.video3d.meta
• media.object.events
• dlq.events
• governance.data.products
• governance.access.requests
• governance.permission.events
• governance.intermediation.log
• governance.transfer.notices
• governance.research.projects
• governance.research.outputs
• governance.dataset.catalog
• governance.data_management_plans
• governance.repository.exports
• dataact.product.catalog
• dataact.user.access.requests
• dataact.third_party.sharing
• dataact.user.exports
• dataact.safeguards
• dataact.legal_basis.checks
• security.asset.inventory
• security.incident.events
• security.vulnerability.findings
• security.sbom.attestations
• security.patch.events
• resilience.backup.tests
• resilience.operational.risk
• resilience.third_party.risk
• compliance.scope.decisions
• compliance.control.assessments
• compliance.reporting.channels
• compliance.legal.dossier
• cra.product.lifecycle

WildFi ingestion

The MQTT bridge subscribes to both generic devices and WildFi telemetry:

$share/ingestors/devices/#
$share/ingestors/wildfi/#

Decoded WildFi GPS/GNSS/raw GPS payloads are routed to raw.gps. Decoded IMU, environment, proximity, movement, and metadata payloads are routed to raw.sensor. See docs/runbooks/wildfi-ingestion.md for the expected topic and payload contract.

Native WildFi .bin logs are decoded with the packaged wildfi-decoder image, built from https://github.com/wildlab/WildFiDecoder, not inside the MQTT bridge.

6. Network zones

• ingest_net: internal ingestion plane for MQTT and SeaweedFS producers.
• kafka_net: internal event backbone.
• patroni_backend: internal HA database control plane.
• patroni_frontend: client-facing database access plane.
• orchestration_net: Airflow, Grafana, Apicurio UI, and other control-plane services.

This segmentation limits accidental coupling between components and makes troubleshooting easier.

SonarQube setup

This repository is configured for GitHub Actions-based SonarQube analysis:

• Workflow: .github/workflows/sonarqube.yml
• Project scanner config: sonar-project.properties
• Required repository secrets:
  • SONAR_TOKEN
  • SONAR_HOST_URL (optional; defaults to https://sonarcloud.io when unset, set it for self-hosted SonarQube)
  • CODACY_PROJECT_TOKEN (optional; when set, coverage is also uploaded to Codacy)

The workflow runs on pushes to main/master, on pull requests, and manually via workflow_dispatch.

7. Configuration model

The stack uses two complementary configuration channels:

1. .env / .env.example

   • Non-file environment variables.
   • Public URLs.
   • Build pins.
   • SeaweedFS S3 credentials used by Flink/Airflow/Beam.

2. ./secrets/*.txt

   • Database passwords.
   • SeaweedFS S3 credentials.
   • PgAdmin password.
   • PgBouncer userlists.

Minimum .env variables

cp .env.example .env

Then set at least:

• AIRFLOW_FERNET_KEY
• AIRFLOW_API_SECRET_KEY
• AIRFLOW_JWT_SECRET
• AIRFLOW_ADMIN_USER
• AIRFLOW_ADMIN_PASSWORD
• AIRFLOW_DB_PASSWORD
• VERNEMQ_DISTRIBUTED_COOKIE
• VERNEMQ_ADMIN_PASSWORD
• SEAWEEDFS_S3_ACCESS_KEY
• SEAWEEDFS_S3_SECRET_KEY
• APICURIO_PUBLIC_URL
• GRAFANA_PUBLIC_URL

Docker secret files required

secrets/app_user_password.txt
secrets/debezium_password.txt
secrets/seaweedfs_s3_secret_key.txt
secrets/seaweedfs_s3_access_key.txt
secrets/patroni_repl_password.txt
secrets/patroni_rewind_password.txt

## 8. Quality checks and bug-fix workflow

Before committing changes, run the local automated checks:

```bash
pytest -q

Recent reliability hardening:

• The MQTT→Kafka bridge now logs asynchronous Kafka producer failures with the actual exception details, which makes transient broker/network issues diagnosable during operations. secrets/pgadmin_default_password.txt secrets/pgbouncer_ro_userlist.txt secrets/pgbouncer_rw_userlist.txt secrets/postgres_exporter_password.txt secrets/postgres_superuser_password.txt

### Generate new secrets locally

```bash
python - <<'PY'
import secrets
from pathlib import Path

target = Path("secrets")
target.mkdir(exist_ok=True)

files = {
    "app_user_password.txt": secrets.token_urlsafe(24),
    "debezium_password.txt": secrets.token_urlsafe(24),
    "seaweedfs_s3_access_key.txt": secrets.token_urlsafe(24),
    "seaweedfs_s3_secret_key.txt": secrets.token_urlsafe(32),
    "patroni_repl_password.txt": secrets.token_urlsafe(24),
    "patroni_rewind_password.txt": secrets.token_urlsafe(24),
    "pgadmin_default_password.txt": secrets.token_urlsafe(24),
    "postgres_exporter_password.txt": secrets.token_urlsafe(24),
    "postgres_superuser_password.txt": secrets.token_urlsafe(24),
}

for name, value in files.items():
    (target / name).write_text(value + "\n", encoding="utf-8")
PY

8. Startup sequence

Use the base compose file plus an environment-specific overlay. The base file describes the internal topology and does not publish host ports. Development ports live in docker-compose.dev.yml; production-like local validation exposes only the explicit edge service.

Development

docker compose -f docker-compose.yml -f docker-compose.dev.yml up -d --build
# or
./install_dealiot.sh dev up -d --build

Staging

docker compose -f docker-compose.yml -f docker-compose.staging.yml up -d --build
# or
./install_dealiot.sh staging up -d --build

Production-like local validation

docker compose -f docker-compose.yml -f docker-compose.prod.yml up -d --build
# or
./install_dealiot.sh prod up -d --build

Docker Swarm production

The Swarm stack lives in deploy/swarm/dealiot-stack.yml. It deploys the application/runtime plane and expects production-grade stateful dependencies to be provisioned outside the stack: Kafka, MQTT, S3-compatible storage, Airflow PostgreSQL, and Airflow Redis.

docker stack deploy -c deploy/swarm/dealiot-stack.yml dealiot

The CI-only stack deploy/swarm/dealiot-smoke-stack.yml verifies Swarm deployment mechanics with the locally built bridge image.

Kubernetes production

The Kubernetes target is a Kustomize base under deploy/kubernetes/base.

kubectl apply -k deploy/kubernetes/base

Create environment-specific overlays for real clusters to patch image tags, external endpoints, ingress, storage classes, and secret references. The CI overlay deploy/kubernetes/overlays/ci-smoke is intentionally minimal and validates rollout wiring on kind.

9. Service endpoints

These endpoints are available when the development overlay is active.

Service	URL / Port
Airflow API/UI	http://localhost:8088
Flink UI	http://localhost:8081
Kafka Connect	http://localhost:8083
Apicurio Registry API	http://localhost:8082/apis/registry/v3
Apicurio Registry UI	http://localhost:8888
SeaweedFS S3 API	http://localhost:8333
SeaweedFS Filer UI	http://localhost:8889
pgAdmin	http://localhost:5050
Grafana	http://localhost:3000
Management Console	http://localhost:8090
Prometheus	http://localhost:9090
HAProxy stats	http://localhost:7000
PostgreSQL RW	localhost:5432
PostgreSQL RO	localhost:5433
PgBouncer RW	localhost:6432
PgBouncer RO	localhost:6433

10. End-to-end smoke test

After rendering Compose and before accepting a platform change, run:

bash scripts/smoke-e2e.sh

The smoke test starts the event-flow services, submits the minimal Flink job, publishes MQTT fixtures, and verifies raw.sensor, dlq.events, features.events, state.latest, and Apicurio artifacts.

11. Validation checklist

After startup, validate in this order:

Kafka

docker compose exec kafka1 /opt/kafka/bin/kafka-topics.sh   --bootstrap-server kafka1:9092 --list

Patroni / PostgreSQL

docker compose exec pg1 curl -s http://127.0.0.1:8008/patroni
docker compose exec haproxy sh -lc "nc -zv 127.0.0.1 5432"

SeaweedFS buckets

docker compose exec seaweedfs-init sh -lc 'echo "SeaweedFS bootstrap completed"'

Apicurio artifacts

Open the registry UI and verify the expected groups/artifacts are present.

Airflow DAG import

Open the Airflow UI and verify that media_backfill is visible and import-error free.

12. Operational notes

Event contract enforcement

Python producers validate critical event contracts before publishing to Kafka:

• valid records go to their intended raw.* topic
• invalid records go to dlq.events
• timestamp is the event time; ingested_at is when the platform received or replayed it

This local validation is a guardrail. Apicurio remains the source of schema documentation and consumer compatibility governance.

Airflow 3

The stack uses the Airflow 3 split model:

• API/UI server (airflow-apiserver)
• scheduler
• worker
• triggerer
• standalone DAG processor

This separation is intentional and aligns with the Airflow 3 deployment model.

Flink state management

Flink stores checkpoints and savepoints in SeaweedFS S3:

• s3://flink-checkpoints/streaming
• s3://flink-savepoints/streaming

Debezium / Kafka Connect

The timescaledb-source connector captures changes from appdb via a named publication:

• publication: dbz_publication
• slot: debezium_appdb

Database routing

• RW traffic: HAProxy 5432
• RO traffic: HAProxy 5433
• pooled RW: PgBouncer 6432
• pooled RO: PgBouncer 6433

13. Runbooks

• Operations
• Backup and restore
• Security hardening
• Security resilience compliance
• Legal applicability
• Legal compliance dossier
• Legal finalization report
• Data Governance Act
• Data Act
• Dataset catalogue and Data Management Plan
• Zenodo dataset export
• OpenAIRE metadata export
• Legal readiness review

14. Recommended next steps

1. Add TLS and authentication for Kafka, MQTT, object storage, and public UIs before any non-local deployment.
2. Add backup and restore runbooks for Kafka metadata, TimescaleDB, SeaweedFS, Grafana, and Airflow metadata.
3. Finalize the production stateful layer with managed services or dedicated operators for Kafka, PostgreSQL, Redis, MQTT, and object storage before go-live.
4. Add downstream consumers for features.events, alerts.events, and state.latest.
5. Add one or more domain-specific Flink jobs under flink/jobs/.

15. Included artifacts in this finalized package

This finalized package contains:

• a patched docker-compose.yml
• functional environment overlays for dev/staging/prod
• a shared lightweight event contract module used by Python producers
• a dlq.events schema and DLQ routing for invalid producer events
• an E2E smoke script and runbooks for operations, backup/restore, and security hardening
• a new README.md
• a new .env.example
• a corrected MQTT bridge
• a management console for architecture health, data catalogue, runbooks, and compliance controls
• a corrected media backfill utility
• a new raw.gps Apicurio schema
• Docker Swarm and Kubernetes production deployment targets with CI smoke workflows