Bug fixes: Changed Histograms to Gages and updated the dashboards

.env

@@ -13,7 +13,7 @@ MQTT_PORT=1883
 MQTT_USERNAME=meshdev
 MQTT_PASSWORD=large4cats
 MQTT_KEEPALIVE=60
-MQTT_TOPIC='msh/israel/#'
+MQTT_TOPIC='msh/#'
 MQTT_IS_TLS=false
 
 # MQTT protocol version (default: MQTTv5) the public MQTT server supports MQTTv311

docker/postgres/init.sql

@@ -1,3 +1,5 @@
+CREATE EXTENSION IF NOT EXISTS moddatetime;
+
 CREATE TABLE IF NOT EXISTS messages
 (
     id          TEXT PRIMARY KEY,
@@ -29,13 +31,10 @@ CREATE TABLE IF NOT EXISTS node_details
     role        VARCHAR,
     mqtt_status VARCHAR   default 'none',
 --     Location Data
-    longitude   FLOAT,
-    latitude    FLOAT,
-    altitude    FLOAT,
-    precision   FLOAT,
-    country     VARCHAR,
-    city        VARCHAR,
-    state       VARCHAR,
+    longitude INT,
+    latitude  INT,
+    altitude  INT,
+    precision INT,
 --     SQL Data
     created_at  TIMESTAMP DEFAULT CURRENT_TIMESTAMP NOT NULL,
     updated_at  TIMESTAMP DEFAULT CURRENT_TIMESTAMP NOT NULL

exporter/processor_base.py

@@ -12,7 +12,7 @@ except ImportError:
     from meshtastic.protobuf.mesh_pb2 import MeshPacket, Data, HardwareModel
     from meshtastic.protobuf.portnums_pb2 import PortNum
 
-from prometheus_client import CollectorRegistry, Counter, Histogram, Gauge
+from prometheus_client import CollectorRegistry, Counter, Gauge
 from psycopg_pool import ConnectionPool
 
 from exporter.client_details import ClientDetails
@@ -46,10 +46,14 @@ class MessageProcessor:
             'destination_role'
         ]
 
-        self.message_size_in_bytes = Histogram(
+        reduced_labels = [
+            'source_id', 'destination_id'
+        ]
+
+        self.message_size_in_bytes = Gauge(
             'text_message_app_size_in_bytes',
             'Size of text messages processed by the app in Bytes',
-            common_labels + ['portnum'],
+            reduced_labels + ['portnum'],
             registry=self.registry
         )
 
@@ -74,7 +78,7 @@ class MessageProcessor:
             registry=self.registry
         )
         # Histogram for the rx_time (time in seconds)
-        self.rx_time_histogram = Histogram(
+        self.rx_time_histogram = Gauge(
             'mesh_packet_rx_time',
             'Receive time of mesh packets (seconds since 1970)',
             common_labels,
@@ -205,10 +209,15 @@ class MessageProcessor:
             'destination_role': destination_client_details.role,
         }
 
+        reduced_labels = {
+            'source_id': source_client_details.node_id,
+            'destination_id': destination_client_details.node_id
+        }
+
         self.message_size_in_bytes.labels(
-            **common_labels,
+            **reduced_labels,
             portnum=self.get_port_name_from_portnum(port_num)
-        ).observe(sys.getsizeof(mesh_packet))
+        ).set(sys.getsizeof(mesh_packet))
 
         self.source_message_type_counter.labels(
             **common_labels,
@@ -226,7 +235,7 @@ class MessageProcessor:
 
         self.rx_time_histogram.labels(
             **common_labels
-        ).observe(mesh_packet.rx_time)
+        ).set(mesh_packet.rx_time)
 
         self.rx_snr_gauge.labels(
             **common_labels

exporter/processors.py

@@ -167,7 +167,7 @@ class NodeInfoAppProcessor(Processor):
 
             conn.commit()
 
-        self.metrics.db.execute_db_operation(db_operation)
+        self.metrics.get_db().execute_db_operation(db_operation)
 
 
 @ProcessorRegistry.register_processor(PortNum.ROUTING_APP)
@@ -499,24 +499,8 @@ class NeighborInfoAppProcessor(Processor):
         except Exception as e:
             logger.error(f"Failed to parse NEIGHBORINFO_APP packet: {e}")
             return
-        self.update_node_graph(neighbor_info, client_details)
         self.update_node_neighbors(neighbor_info, client_details)
 
-    def update_node_graph(self, neighbor_info: NeighborInfo, client_details: ClientDetails):
-        def operation(cur, conn):
-            cur.execute("""
-                INSERT INTO node_graph (node_id, last_sent_by_node_id, broadcast_interval_secs)
-                VALUES (%s, %s, %s)
-                ON CONFLICT (node_id) 
-                DO UPDATE SET 
-                    last_sent_by_node_id = EXCLUDED.last_sent_by_node_id,
-                    broadcast_interval_secs = EXCLUDED.broadcast_interval_secs,
-                    last_sent_at = CURRENT_TIMESTAMP
-            """, (client_details.node_id, neighbor_info.last_sent_by_id, neighbor_info.node_broadcast_interval_secs))
-            conn.commit()
-
-        self.metrics.db.execute_db_operation(operation)
-
     def update_node_neighbors(self, neighbor_info: NeighborInfo, client_details: ClientDetails):
         def operation(cur, conn):
             new_neighbor_ids = [str(neighbor.node_id) for neighbor in neighbor_info.neighbors]
@@ -549,7 +533,7 @@ class NeighborInfoAppProcessor(Processor):
 
             conn.commit()
 
-        self.metrics.db.execute_db_operation(operation)
+        self.metrics.get_db().execute_db_operation(operation)
 
 
 @ProcessorRegistry.register_processor(PortNum.ATAK_PLUGIN)

exporter/registry.py

@@ -1,4 +1,5 @@
-import geopy.point
+import uuid
+
 from geopy.geocoders import Nominatim
 from prometheus_client import CollectorRegistry, Counter, Gauge
 
@@ -19,8 +20,11 @@ class _Metrics:
             self._registry = registry
             self._init_metrics()
             self.initialized = True  # Attribute to indicate initialization
-            self.geolocator = Nominatim()
-            self.db = db
+        self.geolocator = Nominatim(user_agent=f"meshtastic-prometheus-exporter-{str(uuid.uuid4())}")
+        self.db = db
+
+    def get_db(self):
+        return self.db
 
     @staticmethod
     def _get_common_labels():
@@ -36,22 +40,24 @@ class _Metrics:
         self._init_route_discovery_metrics()
 
     def update_metrics_position(self, latitude, longitude, altitude, precision, client_details: ClientDetails):
-        point = geopy.point.Point(latitude, longitude, altitude)
-        location = self.geolocator.reverse(point, language='en')
+        # Could be used to calculate more complex data (Like distances etc..)
+        # point = geopy.point.Point(latitude, longitude, altitude) # Not used for now
 
-        country = location.raw.get('address', {}).get('country', 'Unknown')
-        city = location.raw.get('address', {}).get('city', 'Unknown')
-        state = location.raw.get('address', {}).get('state', 'Unknown')
+        if latitude != 0 and longitude != 0:
+            # location = RateLimiter(self.geolocator.reverse, min_delay_seconds=10, swallow_exceptions=False)((latitude, longitude), language='en', timeout=10)
+            # country = location.raw.get('address', {}).get('country', 'Unknown')
+            # city = location.raw.get('address', {}).get('city', 'Unknown')
+            # state = location.raw.get('address', {}).get('state', 'Unknown')
 
-        def db_operation(cur, conn):
-            cur.execute("""
-            UPDATE node_details
-            SET latitude = %s, longitude = %s, altitude = %s, precision = %s, country = %s, city = %s, state = %s
-            WHERE node_id = %s
-            """, (latitude, longitude, altitude, precision, country, city, state, client_details.node_id))
-            conn.commit()
+            def db_operation(cur, conn):
+                cur.execute("""
+                        UPDATE node_details
+                        SET latitude = %s, longitude = %s, altitude = %s, precision = %s
+                        WHERE node_id = %s
+                        """, (latitude, longitude, altitude, precision, client_details.node_id))
+                conn.commit()
 
-        self.db.execute_db_operation(db_operation)
+            self.db.execute_db_operation(db_operation)
 
     def _init_metrics_telemetry_power(self):
         self.ch1_voltage_gauge = Gauge(

requirements.txt

@@ -7,3 +7,4 @@ psycopg~=3.1.19
 psycopg_pool~=3.2.2
 meshtastic~=2.3.13
 psycopg-binary~=3.1.20
+geopy>=2.4.1