MQTT Data Monitoring app #12
Conversation
dmoore247
left a comment
dmoore247
left a comment
There was a problem hiding this comment.
Looking good. Identified a few items to address please.
| @@ -0,0 +1,6 @@ | |||
| psycopg2-binary==2.9.9 | |||
There was a problem hiding this comment.
I don't see that this is necessary, I could not find Lakebase reference. Please remove if not used.
There was a problem hiding this comment.
This psycopg2-binary dependency still appears in the requirements.txt file.
| self.require_tls = True if str_tls == "true" else False | ||
| self.require_tls = options.get("require_tls", True) | ||
| self.port = int(options.get("port", 8883)) | ||
| self.username = options.get("username", "") |
There was a problem hiding this comment.
Please add input validation logic.
There was a problem hiding this comment.
#High Priority Fixes#
Fix the typo: clean_sesion → clean_session
Implement topic parsing: Complete the _parse_topic() method
Add input validation: Validate broker_address, port ranges, QoS values
Separate frontend code: Extract HTML/CSS/JS to separate files
There was a problem hiding this comment.
Typos fixed
Refactored app code into seperate files
Added a sample mqtt job to be triggered by app to pull data
Input validations for broker, port, qos and clean_session added
[TODO] -topic_parsing would be in v2, have to be a little clever with how we can distribute the topics to pull data in a parallel manner and take advantage of all the compute.
dmoore247
assigned
jefferyann-db and
dmoore247 and unassigned
dmoore247 and
jefferyann-db
mqtt/apps/MQTT Data Monitor/appy.py
Outdated
| # Convert DataFrame to list of dictionaries | ||
| return df.to_dict('records') | ||
| except Exception as e: | ||
| print(f"Database query failed: {e}") |
There was a problem hiding this comment.
Please use the python logger
mqtt/apps/MQTT Data Monitor/appy.py
Outdated
|
|
||
|
|
||
| def create_job(client, notebook_path, cluster_id): | ||
| # cluster_id = ( |
There was a problem hiding this comment.
Please remove dead code
mqtt/apps/MQTT Data Monitor/appy.py
Outdated
| # Callback function for when the client connects | ||
| def on_connect(client, userdata, flags, rc): | ||
| if rc == 0: | ||
| print("Connected successfully to MQTT Broker!") |
There was a problem hiding this comment.
Please use python logger.
mqtt/apps/MQTT Data Monitor/appy.py
Outdated
| connection_result['error'] = f'Failed to connect to {mqtt_server_config["host"]} and {port} and {mqtt_server_config["username"]} are not working' | ||
|
|
||
| # Disconnect | ||
| client.loop_stop() |
There was a problem hiding this comment.
disconnect should be in a finally block in case of exception?
Also, diconnect may additionally generate an exception.
There was a problem hiding this comment.
This is manually handled by the client. What happens is that since the read is stateless we connect session for a brief period read all data and close out the loop which automatically saves our session ids and give us all data that came in during the time we were disconnected. This is why we create a client session ID, everytime we use the class to trigger a new data stream.
mqtt/apps/MQTT Data Monitor/appy.py
Outdated
| 'critical': duplicated, | ||
| 'in_progress': qos2_messages, | ||
| 'resolved_today': unique_topics, | ||
| 'avg_response_time': row_count |
There was a problem hiding this comment.
Doesn't make sense to return row_count for avg_response_time, does it?
mqtt/apps/MQTT Data Monitor/appy.py
Outdated
| query = f"SELECT message, is_duplicate, qos, topic, received_time FROM {catalog}.{schema}.{table} ORDER BY received_time DESC LIMIT 100" | ||
|
|
||
| # Fetch data using get_data function | ||
| curr_data = get_data(query, "4b9b953939869799") |
There was a problem hiding this comment.
What's this literal? Magic values should be handled differently than dropped in the middle of the code.
mqtt/apps/MQTT Data Monitor/appy.py
Outdated
| }), 400 | ||
|
|
||
| # Build the query | ||
| query = f"SELECT message, is_duplicate, qos, topic, received_time FROM {catalog}.{schema}.{table} ORDER BY received_time DESC LIMIT 100" |
There was a problem hiding this comment.
F strings based dynamic sql generation pose a risk of sql injection attack. Try parameterized queries.
mqtt/apps/MQTT Data Monitor/appy.py
Outdated
| """Main MQTT Data Monitor dashboard page""" | ||
| stats = get_mqtt_stats() | ||
|
|
||
| html = ''' |
There was a problem hiding this comment.
There are better ways of serving HTML in an app, e.g. pulling the html code from a file.
| @@ -0,0 +1,6 @@ | |||
| psycopg2-binary==2.9.9 | |||
There was a problem hiding this comment.
This psycopg2-binary dependency still appears in the requirements.txt file.
…code file into individual files for a cleaner read
…l queries, change dict key names for displaying data, removing hard coded warehouse_id, etc
[TODOs]