SmartConveyor-ICS

Distributed industrial control system simulation on Kubernetes: replicated gRPC servers with leader-gated writes, Kafka telemetry flowing into InfluxDB, and a CrewAI agent service running on a local Ollama model.

Demo video

What it does

Three simulated conveyor clients generate sensor readings (load, voltage, and values derived from them: speed, RPM, current, power, torque, efficiency) once per second. Each reading goes two ways: published to the Kafka topic sensor_topic for the telemetry pipeline, and sent as a gRPC GetOptimizedSettings call to a replicated server deployment, which answers with adjusted speed, voltage, and PID gains.

The part worth reading closely is the leader-gating protocol between client and server. Every server pod runs a RaftNode; only the pod that considers itself leader serves optimization requests. A follower aborts the RPC with UNAVAILABLE and embeds the leader's address in the error message. The client parses that address out of the gRPC error string, redirects, and retries (3 attempts, then reset to the default service address). Clients also stamp each reading with a nanosecond lamport_time for ordering.

Honest caveat: the Raft implementation in server/raft.py is a teaching-scale stub, not full consensus. It keeps terms, votes, and an append_entries log, but start_election skips the voting round and declares the candidate leader immediately. What the project actually demonstrates is the leader-redirect protocol and the failure handling around it, on top of Kubernetes replica sets.

Architecture

flowchart LR
    subgraph Kubernetes
        C[client x3] -->|gRPC :50051| S[server x3, HPA 1-10]
        S -.->|UNAVAILABLE + leader address| C
        C -->|JSON, topic sensor_topic| K[Kafka + ZooKeeper]
        K --> T[Telegraf kafka_consumer]
        T --> I[(InfluxDB)]
        I --> G[Grafana]
        S -->|:8000 metrics| P[Prometheus counter]
    end
    subgraph Agent service
        A[Flask api.py :5000] --> W[CrewAI crew]
        W --> O[Ollama tinyllama]
        A -->|/influx-data| I
    end

Loading

Repository layout

protobufs/conveyor.proto      SensorData (9 fields) -> OptimizedSettings (speed, voltage, PID)
server/                       gRPC ConveyorService, RaftNode, Prometheus counter,
                              Kafka consumer thread
client/                       Sensor simulator, Kafka producer, leader-following gRPC client
kubernetes/                   Deployments and services: server (3 replicas + HPA),
                              client (3 replicas), Kafka, ZooKeeper, Telegraf, InfluxDB,
                              Grafana, Ollama, Open WebUI, crewai, ingress, PVCs
src/icsagents/                CrewAI service: Flask job API + two crew definitions
distributedML/data/           Scripts that generate input/output training pairs from the
                              simulator (optimization_data.json, safety_data.json)
monitoring/prometheus.yml     Prometheus scrape config
scripts/load_test.py          50 concurrent gRPC client threads
deliverables.md               kubectl validation and troubleshooting commands
docker-compose.yml            Compose alternative for the agent stack
                              (Ollama + crewai app + Open WebUI)
old_dumps/                    Earlier iterations, kept for reference

The simulation model

ConveyorBelt.generate_data() draws load (1 to 50 kg) and motor voltage (10 to 24 V) at random and derives the rest: speed = 0.1 x voltage, RPM = 100 x speed, current = voltage / (load + 1), power = V x I, and an efficiency figure clamped to [0.5, 1.0]. The server's optimize_system answer is deliberately simple: speed +5 %, voltage -5 %, fixed PID gains (Kp 1.2, Ki 0.6, Kd 0.15). The distributedML/data scripts reuse the same model to emit 10,000 input/output text pairs per file as fine-tuning data.

The agent service

src/icsagents is a CrewAI project exposed through a small Flask job API:

• POST /run accepts JSON, creates a job id, and kicks off the crew in a background thread (returns 202 immediately)
• GET /status/<job_id> and GET /jobs report job state from a lock-guarded in-memory store with hourly cleanup
• GET /influx-data pulls the last hour of telemetry from InfluxDB
• GET /health for probes

Two crews live side by side: crew.py (wired to main.run()) drives two agents from config/agents.yaml, which still carries a math-assignment template, while icscrew.py defines the industrial trio this project is about: a sensor analyst, an optimization engineer, and a validation expert, with inline role prompts. Both run ollama/tinyllama through a local Ollama server.

Getting started

Kubernetes path (Minikube):

git clone https://github.com/hsn07pk/SmartConveyor-ICS.git
cd SmartConveyor-ICS

minikube start
eval $(minikube docker-env)
docker build -t server:latest -f server/Dockerfile server
docker build -t client:latest -f client/Dockerfile client

kubectl apply -f kubernetes/

Compose path for just the agent stack:

docker compose up --build   # Ollama :11434, crewai API :5001, Open WebUI :3000

Configuration lives in .env.example (copy to .env): Ollama base URLs for dev vs Docker, MODEL=ollama/tinyllama, and a MODE flag. Server pods learn their identity from the POD_IP environment variable injected by the deployment manifest.

Verifying the system

deliverables.md collects the working validation commands. The short version:

# Kafka is receiving sensor data
kubectl logs deployment/server | grep "Kafka received"

# a leader was elected
kubectl logs deployment/server | grep "Elected leader"
# -> Elected leader at 10.244.0.5:50051 (Term 1)

# clients follow redirects
kubectl logs deployment/client | grep "Redirecting"

# load test: 50 concurrent gRPC clients
python scripts/load_test.py

The server HPA scales between 1 and 10 replicas at 80 % average CPU; request counts are exposed as the Prometheus counter grpc_requests_total on port 8000 of each server pod.

Limitations

• Leader election is self-election, as described above; there is no log replication between pods.
• The Kafka consumer on the server only logs messages; persistence into InfluxDB happens through Telegraf.
• The optimization rule is a fixed formula, and the CrewAI crews are not yet in the gRPC request path.