¶ (c) Technocore - All Rights Reserved.
NOTICE: All information contained herein is, and remains
the property of TechnoCore.
The intellectual and technical concepts contained
herein are proprietary to TechnoCore and dissemination of this information or reproduction of this material
is strictly forbidden unless prior written permission is obtained
from TechnoCore.
¶ ObjMqtt
The ObjMqtt class provides a robust client for connecting to an MQTT broker, subscribing to topics, and processing incoming messages. Its primary function is to listen for event messages, dynamically parse their JSON payloads, and store the data in a relational database in a structured format.
A key feature of this class is its ability to automatically manage the database schema based on the content of the messages it receives.
¶ Dynamic Schema Generation
When an event message is received, the ObjMqtt class performs the following actions to store its data:
-
Table Name Determination: It constructs a table name based on the
event_typefield in the message payload. The format isevent_<event_type>. For example, anevent_typeofuser_loginresults in a table namedevent_user_login. Heartbeat messages land inevent_heartbeat. -
Dynamic Column Creation: The class inspects the top-level keys within the
event_stateJSON object of the payload. For each key, it ensures a corresponding column exists in the target table. -
Data Type Inference: Columns are created as
LONGTEXT;dict/listvalues are serialised to JSON text before storage. The numeric bookkeeping columns are typed explicitly —event_seqandevent_time_msareBIGINT, andlatencyisINT. -
Data Insertion: The message data is then inserted as a new row in the table.
This process ensures that the database schema evolves automatically with the data, requiring no manual intervention when new fields are introduced in the MQTT messages.
¶ Standard Columns
Every table created by the ObjMqtt class includes the following standard columns:
Guid: Primary key — a unique identifier for the stored row.event_guid: The originating event's GUID (from the payloadguid).Package: The software package or service that generated the event.Module: Originating module (populated for heartbeat records).SourceHost: The hostname or IP address of the machine where the event originated.event_state: The rawevent_statevalue (LONGTEXT).event_seq: Per-topic monotonic sequence number (BIGINT).event_time_ms: Event timestamp in epoch milliseconds (BIGINT).latency: Reported processing latency (INT).param1…param9: Generic scalar payload slots (LONGTEXT).Payload: FallbackTEXTcolumn used to storeevent_statewhen it is not a dictionary.created_at: A timestamp indicating when the record was inserted.
¶ Example Workflow
Consider an MQTT message published to the AXION/event/auth_service_user_login topic with the following JSON payload:
{
"guid": "evt-12345-abcde",
"package": "auth_service",
"host": "prod-server-01",
"event_type": "user_login",
"event_state": {
"username": "jdoe",
"ip_address": "192.168.1.100",
"successful": true,
"metadata": {
"client": "WebApp",
"version": "2.5.1"
}
}
}
- The class subscribes to the topic and receives this message.
- It determines the target table name is
event_user_login. - It checks if this table exists. If not, it creates it with the standard columns.
- It inspects the keys in
event_state:username,ip_address,successful, andmetadata. - It ensures columns with these names exist in the
event_user_logintable (allLONGTEXT).metadatais serialised to JSON text because its value is a dictionary.
- It inserts a new row with the corresponding values.
¶ Process Model
Message decoding and all database writes run in a dedicated worker
process (_queue_worker_process, backed by _MqttQueueWorker) so that
each process owns its own database connection — the DB drivers are neither
fork- nor thread-safe. The main process only receives messages on the paho
client thread and hands the raw (topic, payload) tuples to the worker via a
multiprocessing.Queue. A second process (_alert_ticker_process) runs
periodic ObjAlert.check() calls with AlertNotifyMode.NONE to avoid an MQTT
re-publish loop, and a lightweight main-process thread
(_tui_dispatch_loop) forwards TUI events from both workers to the TUI. The
spawn start method is used so workers get a clean interpreter with no
inherited DB file descriptors.
¶ Batch Processing
To optimize database performance, the worker collects messages in memory and
inserts them in bulk per topic, flushing when the batch reaches _BATCH_SIZE
(100) or after _BATCH_TIMEOUT (1.0s) of idle time. If a bulk insert fails
(e.g. an unexpected column), the worker retries with only the columns that
currently exist in the table, and logs any messages it has to drop.
¶ Topic Namespace
All Axion traffic lives under the AXION/ namespace (MQTT_NAMESPACE), and
the consumer subscribes to AXION/# (MQTT_SUBSCRIBE_ALL) by default so it
sees every Axion topic and nothing from co-tenant apps. Topics are routed by
their second segment (the MqttTopicAction):
AXION/heartbeat/<package>/<service>/<instance>→event_heartbeatAXION/event/<package>_<event_type>→event_<event_type>(persisted)AXION/alert/...→ delivered locally viaObjNotify(never re-published)
Non-Axion topics and messages with a missing/invalid event_type are dropped
and reported to the TUI with a MqttDropReason.
¶ Event Registry Catalog (def_mqtt)
Every event the platform emits is a member of the MqttEventType registry
(factory.core/extend.enum/ObjEnumMqtt.py), with a human-readable
description in ObjMqtt.yaml under the top-level events: map. The
seed_event_definitions() method materialises this registry into the
def_mqtt catalog table (EventType, Domain, Description, Severity,
Alert, NotifyCode, Active, Updatedate) so the full event vocabulary is
queryable from the database. Domains are derived from the event-name prefix
(_EVENT_DOMAINS); MQTT events are universal, so the catalog carries no
package column. Seed it with:
python factory.core/ObjMqtt.py seed-events
¶ Alert routing
The top-level alerts: map in ObjMqtt.yaml flags event types that should
page someone — each entry sets severity and a notify code, seeded into
def_mqtt (Alert='Y', Severity, NotifyCode). When the consumer persists
a flagged event it calls ObjNotify.Run(<notify_code>) best-effort
(_maybe_alert), turning the passive event stream into actionable alerts
without any publisher needing to know about alerting. Delivery is local (no
MQTT re-publish), so it cannot loop. Provision the referenced notify codes in
def_notify. Defaults: SCHEDULER_JOB_FAIL, INCIDENT_RAISED, SQL_ERROR.
¶ Retention
The consumer creates one event_<type> table per event type, which grow
unbounded. prune_events(days) deletes rows older than days from every
event_* table (including event_dropped). Run on a schedule:
python factory.core/ObjMqtt.py prune 30 # drop rows older than 30 days
¶ Dropped-message log
Messages the consumer can't route (non-Axion topic, missing event_type,
decode/process error) are reported to the TUI and persisted to
event_dropped (Topic, Reason, created_at) via _drop(), so silent
drops are queryable after the fact.
¶ Signals
AXION/signal/<package> messages are consumed by ObjSignalMqtt's own
subscriber, not this consumer; process_message recognises the signal
action and skips it without flagging an unhandled-prefix drop.
¶ Querying captured events
python ObjMqtt.py stats— per-event_*table row counts + last-seen.python ObjMqtt.py health— event-health dashboard: per-type counts and
last-seen joined with thedef_mqttcatalogue (domain, severity, alert
flag), plus the dropped-message total. The single-glance ops view of the
whole event stream.python ObjMqtt.py tail <EVENT_TYPE> [-n N]— most-recent rows for one type.- Programmatically:
event_counts(),event_health(),recent_events( event_type, limit),list_event_tables().
¶ Emitting events ergonomically
ObjData.event() (in ObjDataEventMixin) is a timing context manager: it
measures the block, emits DONE with latency on success or FAIL on
exception (re-raising), both best-effort so telemetry never masks the block's
own error:
with self.event(MqttEventType.IMPORT_RUN, param1=import_code):
self.run_import()
¶ MQTT Callback API
ObjMqtt uses the paho-mqtt VERSION2 callback API. The _on_connect
method signature is:
def _on_connect(self, client, userdata, connect_flags, reason_code, properties):
if reason_code == 0:
# connected — subscribe to topics
else:
# failed — reason_code gives a human-readable description
reason_code is a paho ReasonCode object; reason_code == 0 tests for
success and str(reason_code) gives a descriptive error string on failure.
¶ Command-Line Usage
The script exposes two Typer commands — main (monitor) and seed-events.
python factory.core/ObjMqtt.py main [TOPIC] [--no-tui]
python factory.core/ObjMqtt.py seed-events
- main TOPIC: The MQTT topic to subscribe to. Defaults to the Axion namespace
AXION/#; pass#to monitor the whole broker. - --no-tui: Disable the TUI dashboard and use plain-text logging.
- seed-events: Create and populate the
def_mqttevent catalog.
Example:
# Monitor the Axion namespace (default) with the TUI
python factory.core/ObjMqtt.py main
# Monitor a specific topic without the TUI
python factory.core/ObjMqtt.py main "AXION/event/auth_service_user_login" --no-tui
# Seed the def_mqtt event catalog
python factory.core/ObjMqtt.py seed-events