Skip to main content

Data Flow & Topics

System Wide Data Flow​

SMOCS implements a hub and spoke topology with Kafka as the central message bus:

External Sources → Producers → Kafka → Consumers/Agents → Kafka → Storage/Control

Primary Data Flows​

1. Data Ingestion Flow

MQTT Broker → mqtt-kafka-producer → Kafka Topic → Agent Data Ingest Thread → MySQL
                                                  ↓
                                           influxdb-consumer → InfluxDB

2. ML Training Flow

MySQL (accumulated data) → Agent Training Thread → Kafka (training results) → InfluxDB
                                                 ↓
                                           /app/models (model files)

3. ML Inference Flow

Kafka (sensor data) → Agent Inference Thread → Kafka (predictions/anomalies) → InfluxDB
                            ↓
                      /app/models (load latest)

4. RL Control Flow

Gymnasium → Kafka (state) → RL Agent Inference → Kafka (action) → Gymnasium
                          ↓                                      ↓
              Kafka (SARSA) → RL Agent Ingest → MySQL    Execute & publish SARSA
                                              ↓
                                    RL Agent Training (updates policy)

Topic Naming Conventions​

Producer Topics​

MQTT Topics: Sanitized MQTT topic paths

  • Original: sensor/temperature/room1
  • Kafka: sensor.temperature.room1

EPICS Topics: Source identifier from config

  • Config: source: CEBAF
  • Kafka: CEBAF

Gymnasium Topics: Configured explicitly

  • gymnasium-state: Environment states only
  • gymnasium-sarsa: Full SARSA tuples
  • gymnasium-action: Agent actions
  • gymnasium-output: Flattened/decomposed data

Agent Topics​

Input Topics: Source data for processing

  • gymnasium-output (autoencoder input)
  • gymnasium-sarsa (RL agent training input)
  • gymnasium-state (RL agent inference input)

Output Topics: Agent results

  • autoencoder1-anomalies: Anomaly detection results
  • autoencoder1-training-results: Training metrics
  • gymnasium-action: RL agent actions

Topic Characteristics​

Partitions: Default 1 (configurable in kafka section)

  • Single partition ensures message ordering
  • Multiple partitions enable parallel processing
  • SMOCS agents use single-threaded processing, since ordering matters single partition utilized

Retention: Kafka default (7 days)

  • Configure per-topic for different retention needs
  • Critical training data should be persisted to MySQL

Message Flow Patterns​

Fan-Out Pattern​

Single producer, multiple consumers:

gymnasium-kafka-controller
    ↓ (publishes to gymnasium-output)
Kafka Topic: gymnasium-output
    ↓                    ↓
autoencoder-agent1   influxdb-consumer

Each consumer processes the same messages independently.

Processing Pipeline Pattern​

Chained transformations:

Source → Producer → Kafka → Agent → Kafka → Consumer → Storage

Example: MQTT sensor → Kafka → Autoencoder → Kafka → InfluxDB

Request-Response Pattern​

Bidirectional communication utilized for control systems:

Gymnasium                    RL Agent
    ↓ (state)                    ↑
Kafka: gymnasium-state    Kafka: gymnasium-action
    ↓                            ↑
    └──────> (inference) ────────┘

Feedback Loop Pattern​

Closed-loop control utilized in Reinforcement Learning:

Environment → state → Agent → action → Environment
     ↑                                      ↓
     └────────────── (SARSA) ──────────────┘
                        ↓
                 Agent Training (improves policy)

Topic Lifecycle​

Topic Creation​

Topics are created automatically by producers when kafka.auto_create: true:

def send_to_kafka(self, topic, message):
    self.create_topic_if_not_exists(topic)
    self.kafka_producer.send(topic, message)

Topic Discovery​

Consumers use two subscription modes:

Explicit topics:

topics = ['gymnasium-output', 'sensor-data']
consumer.subscribe(topics)

Pattern-based (regex):

pattern = re.compile(r'.*')  # All topics
consumer.subscribe(pattern=pattern)

The influxdb-consumer uses pattern-based subscription to capture all topics.

Topic Deletion​

Topics are not automatically deleted by SMOCS. Manual deletion:

# Using Kafka CLI (inside kafka-broker container)
kafka-topics.sh --delete --topic topic-name --bootstrap-server localhost:9092

# Or using docker compose down with volumes
docker compose down -v  # Deletes all Kafka data by deleting the kafka volume

Data Retention Strategy​

Kafka Retention​

  • Default: 7 days (Kafka default)
  • Purpose: Message replay capability
  • Storage: Lost on docker compose down -v

MySQL Retention​

  • Tables: agent_inferences, agent_replay, agent_information
  • Retention: Indefinite (manual cleanup required)
  • Purpose: Training data, model versioning, agent state
  • Storage: Persistent volumes (mysql-data-*)

InfluxDB Retention​

  • Default: Indefinite
  • Purpose: Long-term metrics storage
  • Configuration: Via InfluxDB retention policies
  • Storage: Persistent volume (influxdb-data)

Model File Retention​

  • Location: /app/models (inside agent containers)
  • Versioning: model_v001.h5, model_v002.h5, etc.
  • Retention: All versions kept (manual cleanup required)
  • Latest: Tracked in latest_model.json

Monitoring Topic Health​

Using Kafka UI​

Enable the UI profile:

COMPOSE_PROFILES=gymnasium,rl1,...,ui
docker compose up

Access at http://localhost:8080:

  • View all topics and message counts
  • Inspect message contents
  • Monitor consumer lag
  • Check partition distribution

Troubleshooting Data Flow​

Verify topic exists:

docker exec kafka-broker kafka-topics.sh --list --bootstrap-server localhost:9092

Check consumer subscription:

docker compose logs agent-name | grep "Subscribed to"