MQTT

You are not logged in. Any changes you make will be marked as anonymous. You may want to Log In if you already have an account.

<h1 style="text-align: left;">MQTT Protocol</h1><h3 style="text-align: left;">1. Project Overview</h3>Purpose and Scope: This system is an MQTT subscriber designed for Freeport FMI. It connects to a local Mosquitto MQTT broker, listens for sensor data on a specific topic pattern, and performs two main actions:<ul style="text-align: left;"><li>Publishes received sensor data directly to a Google Cloud Pub/Sub topic for downstream processing.</li><li>Saves the data to a Google BigQuery table for audit and historical analysis.</li></ul>Primary Use Cases:<ul style="text-align: left;"><li>Real-time ingestion of sensor readings from industrial equipment or IoT devices.</li><li>Forwarding sensor data to cloud-based event processing pipelines (via Pub/Sub).</li><li>Persisting sensor data for compliance, auditing, or analytics (via BigQuery).</li></ul>Explicitly Not Handled:<ul style="text-align: left;"><li>Data transformation or enrichment beyond basic field extraction.</li><li>Downstream processing or analytics (handled elsewhere).</li><li>MQTT message publishing (this is a subscriber only).</li><li>Authentication/authorization for MQTT broker (assumes local/trusted broker).</li></ul><hr style="text-align: left;" /><h3 style="text-align: left;">2. System Architecture</h3>Core Components:<ul style="text-align: left;"><li>MQTT Client: Subscribes to a topic pattern and receives messages.</li><li>Pub/Sub Publisher: Publishes validated sensor events to a Google Pub/Sub topic.</li><li>BigQuery Client: Inserts validated sensor events into a BigQuery table.</li><li>Logging: Uses Python logging for local and cloud-based logs.</li></ul>Data and Control Flow:<ol style="text-align: left;"><li>MQTT client connects to broker and subscribes to a topic.</li><li>On message receipt:<ul><li>Parses topic and payload.</li><li>Validates required fields.</li><li>Publishes event to Pub/Sub (if enabled).</li><li>Inserts event into BigQuery (if enabled).</li><li>Logs all actions and errors.</li></ul></li></ol>External Services and Dependencies:<ul style="text-align: left;"><li>Mosquitto MQTT broker (local or remote).</li><li>Google Cloud Pub/Sub (for event publishing).</li><li>Google BigQuery (for data persistence).</li><li>Google Cloud Logging (optional, not actively used in this script).</li></ul><hr style="text-align: left;" /><h3 style="text-align: left;">3. Core Concepts &amp; Domain Logic</h3>Key Abstractions and Domain Terms:<ul style="text-align: left;"><li>Sensor Reading: A JSON payload containing <code class="undefined">site_code</code>, <code class="undefined">sensor_id</code>, <code class="undefined">timestamp</code>, and <code class="undefined">value</code>.</li><li>KPI Event: The structure sent to Pub/Sub, mapping sensor readings to KPI semantics.</li><li>Topic Parsing: MQTT topics are expected in the form <code class="undefined">FCTS/FAE/&lt;site&gt;/SolvExtract/&lt;pi_tag&gt;</code>.</li></ul>Business/Technical Invariants:<ul style="text-align: left;"><li>Only messages with all required fields are processed.</li><li>Data is published to Pub/Sub and/or BigQuery only in certain environments (DEV, PPRD, PRD).</li><li>Ordering key for Pub/Sub is the site code, ensuring per-site event ordering.</li></ul>Mental Model:<ul style="text-align: left;"><li>The system is a stateless, real-time bridge between MQTT, Pub/Sub, and BigQuery, with environment-driven feature toggling.</li></ul><hr style="text-align: left;" /><h3 style="text-align: left;">4. Codebase Structure</h3>Single-file Script:<ul style="text-align: left;"><li>All logic is contained in one Python file.</li><li>Major sections: configuration, client setup, message handling, main loop.</li></ul>Responsibility Boundaries:<ul style="text-align: left;"><li>MQTT handling (connection, subscription, message receipt).</li><li>Data validation and transformation.</li><li>Pub/Sub publishing.</li><li>BigQuery insertion.</li><li>Logging and error handling.</li></ul>What Changes Together:<ul style="text-align: left;"><li>Topic parsing and payload validation logic.</li><li>Pub/Sub and BigQuery schema changes.</li></ul><hr style="text-align: left;" /><h3 style="text-align: left;">5. Configuration &amp; Environment</h3>Environment Variables:<ul style="text-align: left;"><li><code class="undefined">APP_ENV</code>: Controls environment (LOCAL, DEV, PPRD, PRD).</li><li><code class="undefined">GCP_PROJECT_ID</code>: Google Cloud project.</li><li><code class="undefined">BIGQUERY_DATASET</code>, <code class="undefined">BIGQUERY_TABLE</code>: BigQuery dataset/table.</li><li><code class="undefined">PUBSUB_TOPIC</code>: Pub/Sub topic name.</li><li><code class="undefined">MQTT_BROKER</code>, <code class="undefined">MQTT_PORT</code>, <code class="undefined">MQTT_TOPIC</code>: MQTT connection details.</li></ul>Configuration Files:<ul style="text-align: left;"><li>None; all configuration is via environment variables.</li></ul>Environment Differences:<ul style="text-align: left;"><li>In LOCAL, only logs are written (no Pub/Sub or BigQuery).</li><li>In DEV/PPRD/PRD, Pub/Sub and BigQuery are enabled.</li></ul><hr style="text-align: left;" /><h3 style="text-align: left;">6. Runtime Behavior</h3>Startup Sequence:<ul style="text-align: left;"><li>Loads configuration from environment.</li><li>Initializes BigQuery and Pub/Sub clients if enabled.</li><li>Connects to MQTT broker and subscribes to topic.</li></ul>Normal Execution Flow:<ul style="text-align: left;"><li>Waits for MQTT messages.</li><li>On message:<ul><li>Parses and validates payload.</li><li>Publishes to Pub/Sub (if enabled).</li><li>Inserts into BigQuery (if enabled).</li><li>Logs all actions.</li></ul></li></ul>Error Handling and Logging:<ul style="text-align: left;"><li>Uses Python logging for all actions and errors.</li><li>Catches and logs JSON decode errors, Pub/Sub/BigQuery errors, and connection issues.</li><li>Continues running on recoverable errors.</li></ul><hr style="text-align: left;" /><h3 style="text-align: left;">7. Deployment &amp; Operations</h3>Build Process:<ul style="text-align: left;"><li>No explicit build; run as a Python script.</li><li>Requires Python 3 and dependencies (<code class="undefined">paho-mqtt</code>, <code class="undefined">google-cloud-bigquery</code>, <code class="undefined">google-cloud-pubsub</code>).</li></ul>Deployment Method:<ul style="text-align: left;"><li>Typically deployed as a container or systemd service on a VM or cloud instance.</li><li>Environment variables must be set appropriately.</li></ul>Runtime Dependencies:<ul style="text-align: left;"><li>Access to MQTT broker.</li><li>Google Cloud credentials (for BigQuery and Pub/Sub).</li></ul>Scaling and Rollback:<ul style="text-align: left;"><li>Stateless; can run multiple instances for higher throughput (ensure unique MQTT client IDs).</li><li>Rollback is as simple as redeploying a previous version.</li></ul><hr style="text-align: left;" /><h3 style="text-align: left;">8. Extending the System</h3>Where and How to Add New Features:<ul style="text-align: left;"><li>Add new message processing logic in <code class="undefined">on_message</code>.</li><li>Extend Pub/Sub or BigQuery schemas as needed (update both code and cloud resources).</li><li>Add new output sinks (e.g., HTTP, file) by extending the message handler.</li></ul>Recommended Patterns:<ul style="text-align: left;"><li>Keep message validation strict.</li><li>Use environment variables for all configuration.</li><li>Modularize new output integrations.</li></ul>Anti-patterns and Risk Areas:<ul style="text-align: left;"><li>Avoid hardcoding credentials or configuration.</li><li>Do not relax payload validation (could lead to data quality issues).</li><li>Avoid blocking operations in the message handler (could delay message processing).</li></ul>Testing Strategy:<ul style="text-align: left;"><li>Unit test message parsing and validation logic.</li><li>Use mocks for Pub/Sub and BigQuery in tests.</li><li>Integration test with test MQTT broker and cloud resources.</li></ul><hr style="text-align: left;" /><h3 style="text-align: left;">9. Security &amp; Compliance</h3>Authentication and Authorization:<ul style="text-align: left;"><li>Assumes trusted access to MQTT broker (no explicit auth in code).</li><li>Uses Google Cloud credentials for Pub/Sub and BigQuery (should be managed via service accounts).</li></ul>Secrets Handling:<ul style="text-align: left;"><li>No secrets in code; relies on environment and Google Cloud IAM.</li></ul>Data Sensitivity:<ul style="text-align: left;"><li>Sensor data may be sensitive; ensure BigQuery and Pub/Sub permissions are tightly scoped.</li></ul><hr style="text-align: left;" /><h3 style="text-align: left;">10. Common Pitfalls &amp; Gotchas</h3><ul style="text-align: left;"><li>Topic Structure: Expects exactly 5 parts in MQTT topic; malformed topics are ignored.</li><li>Environment Toggling: Features are enabled/disabled strictly by <code class="undefined">APP_ENV</code> variable; misconfiguration can lead to data loss or test data in production.</li><li>BigQuery Schema Drift: Changes to table schema require code and cloud resource updates.</li><li>Pub/Sub Ordering: Uses site code as ordering key; if site codes are not unique or consistent, ordering may break.</li><li>Error Handling: Some errors are only logged, not retried; transient failures may result in data loss unless handled externally.</li><li>Single-threaded: Blocking operations in message handler can delay processing of subsequent messages.</li></ul><h3>11. Other components of the protocol.</h3><ol><li><img class="editor-inline-macro" src="/plugins/servlet/confluence/placeholder/macro?definition=e2FkZC1wYWdlOm5hbWU9TW9zcXVpdHRvIGNvbmZpZ3VyYXRpb258bGlua1RleHQ9TW9zcXVpdHRvIGNvbmZpZ3VyYXRpb259&amp;locale=en_GB&amp;version=2" data-macro-name="add-page" data-macro-id="7ce2df84-677f-4c53-9cdf-bb1420c5057a" role="button" tabindex="0" aria-haspopup="true" aria-label="add-page macro" data-macro-parameters="linkText=Mosquitto configuration|name=Mosquitto configuration" data-macro-schema-version="1"></li><li><img class="editor-inline-macro" src="/plugins/servlet/confluence/placeholder/macro?definition=e2FkZC1wYWdlOm5hbWU9RGVwbG95bWVudHxsaW5rVGV4dD1EZXBsb3ltZW50fQ&amp;locale=en_GB&amp;version=2" data-macro-name="add-page" data-macro-id="e5fd81c6-d361-4b40-8b13-be08a62fc057" role="button" tabindex="0" aria-haspopup="true" aria-label="add-page macro" data-macro-parameters="linkText=Deployment|name=Deployment" data-macro-schema-version="1"></li></ol>