Merge pull request #15112 from smatvienko-tb/feature/mqtts-metrics

MQTTS metrics and client address logging on exceptionCaught

common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/MqttTransportContext.java

@@ -88,20 +88,30 @@ public class MqttTransportContext extends TransportContext {
     @Value("${transport.mqtt.proxy_enabled:false}")
     private boolean proxyEnabled;
 
-    private final AtomicInteger connectionsCounter = new AtomicInteger();
+    private final AtomicInteger connectionsActiveCounterMQTT = new AtomicInteger();
+    private final AtomicInteger connectionsActiveCounterMQTTS = new AtomicInteger();
 
     @PostConstruct
     public void init() {
         super.init();
-        transportService.createGaugeStats("openConnections", connectionsCounter);
+        transportService.createGaugeStats("connections_active", connectionsActiveCounterMQTT, "protocol", "MQTT");
+        transportService.createGaugeStats("connections_active", connectionsActiveCounterMQTTS, "protocol", "MQTTS");
     }
 
-    public void channelRegistered() {
-        connectionsCounter.incrementAndGet();
+    public void channelRegistered(boolean isSSL) {
+        if (isSSL) {
+            connectionsActiveCounterMQTTS.incrementAndGet();
+        } else {
+            connectionsActiveCounterMQTT.incrementAndGet();
+        }
     }
 
-    public void channelUnregistered() {
-        connectionsCounter.decrementAndGet();
+    public void channelUnregistered(boolean isSSL) {
+        if (isSSL) {
+            connectionsActiveCounterMQTTS.decrementAndGet();
+        } else {
+            connectionsActiveCounterMQTT.decrementAndGet();
+        }
     }
 
     public boolean checkAddress(InetSocketAddress address) {

common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/MqttTransportHandler.java

@@ -180,16 +180,20 @@ public class MqttTransportHandler extends ChannelInboundHandlerAdapter implement
         this.rpcAwaitingAck = new ConcurrentHashMap<>();
     }
 
+    boolean isSSL() {
+        return sslHandler != null;
+    }
+
     @Override
     public void channelRegistered(ChannelHandlerContext ctx) throws Exception {
         super.channelRegistered(ctx);
-        context.channelRegistered();
+        context.channelRegistered(isSSL());
     }
 
     @Override
     public void channelUnregistered(ChannelHandlerContext ctx) throws Exception {
         super.channelUnregistered(ctx);
-        context.channelUnregistered();
+        context.channelUnregistered(isSSL());
     }
 
     @Override
@@ -254,6 +258,17 @@ public class MqttTransportHandler extends ChannelInboundHandlerAdapter implement
         }
     }
 
+    String getClientAddr(ChannelHandlerContext ctx) {
+        try {
+            InetSocketAddress remote = getAddress(ctx);
+            if (remote == null) return "unknown";
+            String host = remote.getAddress() != null ? remote.getAddress().getHostAddress() : remote.getHostString();
+            return host + ":" + remote.getPort();
+        } catch (Exception ignored) {
+            return "unknown";
+        }
+    }
+
     InetSocketAddress getAddress(ChannelHandlerContext ctx) {
         var address = ctx.channel().attr(MqttTransportService.ADDRESS).get();
         if (address == null) {
@@ -1150,19 +1165,25 @@ public class MqttTransportHandler extends ChannelInboundHandlerAdapter implement
     public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
         if (cause instanceof IOException) {
             if (log.isDebugEnabled()) {
-                log.debug("[{}][{}][{}] IOException: {}", sessionId,
+                String clientAddr = getClientAddr(ctx);
+                log.debug("[{}][{}][{}][{}] {}: {}", sessionId,
                         Optional.ofNullable(this.deviceSessionCtx.getDeviceInfo()).map(TransportDeviceInfo::getDeviceId).orElse(null),
                         Optional.ofNullable(this.deviceSessionCtx.getDeviceInfo()).map(TransportDeviceInfo::getDeviceName).orElse(""),
+                        clientAddr,
+                        cause.getClass().getSimpleName(),
                         cause.getMessage(),
                         cause);
             } else if (log.isInfoEnabled()) {
-                log.info("[{}][{}][{}] IOException: {}", sessionId,
+                String clientAddr = getClientAddr(ctx);
+                log.info("[{}][{}][{}][{}] {}: {}", sessionId,
                         Optional.ofNullable(this.deviceSessionCtx.getDeviceInfo()).map(TransportDeviceInfo::getDeviceId).orElse(null),
                         Optional.ofNullable(this.deviceSessionCtx.getDeviceInfo()).map(TransportDeviceInfo::getDeviceName).orElse(""),
+                        clientAddr,
+                        cause.getClass().getSimpleName(),
                         cause.getMessage());
             }
         } else {
-            log.error("[{}] Unexpected Exception", sessionId, cause);
+            log.error("[{}][{}] Unexpected Exception", sessionId, getClientAddr(ctx), cause);
         }
 
         closeCtx(ctx, MqttReasonCodes.Disconnect.SERVER_SHUTTING_DOWN);

common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/TransportService.java

@@ -161,5 +161,5 @@ public interface TransportService {
 
     boolean hasSession(SessionInfoProto sessionInfo);
 
-    void createGaugeStats(String openConnections, AtomicInteger connectionsCounter);
+    void createGaugeStats(String statsName, AtomicInteger number, String... tags);
 }

common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/service/DefaultTransportService.java

@@ -1257,9 +1257,21 @@ public class DefaultTransportService extends TransportActivityManager implements
     }
 
     @Override
-    public void createGaugeStats(String statsName, AtomicInteger number) {
-        statsFactory.createGauge(StatsType.TRANSPORT + "." + statsName, number);
-        statsMap.put(statsName, number);
+    public void createGaugeStats(String statsName, AtomicInteger number, String... tags) {
+        String key = "thingsboard" + "." + StatsType.TRANSPORT.getName() + "." + statsName;
+        statsFactory.createGauge(key, number, tags);
+        statsMap.put(statsName + tagsKey(tags), number);
+    }
+
+    private String tagsKey(String... tags) {
+        if (tags == null || tags.length < 2) return "";
+        StringBuilder sb = new StringBuilder("[");
+        for (int i = 0; i < tags.length; i += 2) {
+            if (i > 0) sb.append(',');
+            sb.append(tags[i]).append('=').append(i + 1 < tags.length ? tags[i + 1] : "");
+        }
+        sb.append(']');
+        return sb.toString();
     }
 
     @Scheduled(fixedDelayString = "${transport.stats.print-interval-ms:60000}")

docker/monitoring/grafana/provisioning/dashboards/transport_connections.json

@@ -0,0 +1,458 @@
+{
+  "annotations": {
+    "list": [
+      {
+        "builtIn": 1,
+        "datasource": {
+          "type": "datasource",
+          "uid": "grafana"
+        },
+        "enable": true,
+        "hide": true,
+        "iconColor": "rgba(0, 211, 255, 1)",
+        "name": "Annotations & Alerts",
+        "type": "dashboard"
+      }
+    ]
+  },
+  "editable": true,
+  "fiscalYearStartMonth": 0,
+  "graphTooltip": 0,
+  "id": null,
+  "links": [],
+  "panels": [
+    {
+      "datasource": {
+        "type": "prometheus",
+        "uid": "prometheus"
+      },
+      "description": "",
+      "fieldConfig": {
+        "defaults": {
+          "color": {
+            "mode": "palette-classic"
+          },
+          "custom": {
+            "axisBorderShow": false,
+            "axisCenteredZero": false,
+            "axisColorMode": "text",
+            "axisLabel": "",
+            "axisPlacement": "auto",
+            "barAlignment": 0,
+            "drawStyle": "line",
+            "fillOpacity": 10,
+            "gradientMode": "none",
+            "hideFrom": {
+              "legend": false,
+              "tooltip": false,
+              "viz": false
+            },
+            "insertNulls": false,
+            "lineInterpolation": "linear",
+            "lineWidth": 1,
+            "pointSize": 5,
+            "scaleDistribution": {
+              "type": "linear"
+            },
+            "showPoints": "never",
+            "spanNulls": true,
+            "stacking": {
+              "group": "A",
+              "mode": "none"
+            },
+            "thresholdsStyle": {
+              "mode": "off"
+            }
+          },
+          "mappings": [],
+          "thresholds": {
+            "mode": "absolute",
+            "steps": [
+              {
+                "color": "dark-orange",
+                "value": null
+              }
+            ]
+          },
+          "unit": "short"
+        },
+        "overrides": [
+          {
+            "matcher": {
+              "id": "byName",
+              "options": "Total connections"
+            },
+            "properties": [
+              {
+                "id": "color",
+                "value": {
+                  "fixedColor": "light-orange",
+                  "mode": "fixed"
+                }
+              }
+            ]
+          }
+        ]
+      },
+      "gridPos": {
+        "h": 15,
+        "w": 19,
+        "x": 0,
+        "y": 0
+      },
+      "id": 19,
+      "options": {
+        "legend": {
+          "calcs": [],
+          "displayMode": "list",
+          "placement": "bottom",
+          "showLegend": true
+        },
+        "tooltip": {
+          "mode": "multi",
+          "sort": "none"
+        }
+      },
+      "targets": [
+        {
+          "datasource": {
+            "type": "prometheus",
+            "uid": "prometheus"
+          },
+          "editorMode": "code",
+          "exemplar": false,
+          "expr": "sum by (protocol) (thingsboard_transport_connections_active{})",
+          "format": "time_series",
+          "instant": false,
+          "interval": "",
+          "intervalFactor": 1,
+          "legendFormat": "{{protocol}}",
+          "refId": "A"
+        }
+      ],
+      "title": "Active Connections by Protocol",
+      "type": "timeseries"
+    },
+    {
+      "datasource": {
+        "type": "prometheus",
+        "uid": "prometheus"
+      },
+      "fieldConfig": {
+        "defaults": {
+          "color": {
+            "mode": "thresholds"
+          },
+          "mappings": [],
+          "thresholds": {
+            "mode": "absolute",
+            "steps": [
+              {
+                "color": "rgba(209, 138, 78, 1)",
+                "value": null
+              }
+            ]
+          }
+        },
+        "overrides": []
+      },
+      "gridPos": {
+        "h": 15,
+        "w": 5,
+        "x": 19,
+        "y": 0
+      },
+      "id": 21,
+      "options": {
+        "colorMode": "value",
+        "graphMode": "none",
+        "justifyMode": "auto",
+        "orientation": "auto",
+        "percentChangeColorMode": "standard",
+        "reduceOptions": {
+          "calcs": [
+            "lastNotNull"
+          ],
+          "fields": "",
+          "values": false
+        },
+        "showPercentChange": false,
+        "text": {},
+        "textMode": "auto",
+        "wideLayout": true
+      },
+      "targets": [
+        {
+          "datasource": {
+            "type": "prometheus",
+            "uid": "prometheus"
+          },
+          "editorMode": "code",
+          "exemplar": true,
+          "expr": "sum (thingsboard_transport_connections_active)",
+          "instant": true,
+          "interval": "",
+          "legendFormat": "",
+          "refId": "A"
+        }
+      ],
+      "title": "Total Connections",
+      "type": "stat"
+    },
+    {
+      "datasource": {
+        "type": "prometheus",
+        "uid": "prometheus"
+      },
+      "description": "",
+      "fieldConfig": {
+        "defaults": {
+          "color": {
+            "mode": "palette-classic"
+          },
+          "custom": {
+            "axisBorderShow": false,
+            "axisCenteredZero": false,
+            "axisColorMode": "text",
+            "axisLabel": "",
+            "axisPlacement": "auto",
+            "barAlignment": 0,
+            "drawStyle": "line",
+            "fillOpacity": 10,
+            "gradientMode": "none",
+            "hideFrom": {
+              "legend": false,
+              "tooltip": false,
+              "viz": false
+            },
+            "insertNulls": false,
+            "lineInterpolation": "linear",
+            "lineWidth": 1,
+            "pointSize": 5,
+            "scaleDistribution": {
+              "type": "linear"
+            },
+            "showPoints": "never",
+            "spanNulls": true,
+            "stacking": {
+              "group": "A",
+              "mode": "none"
+            },
+            "thresholdsStyle": {
+              "mode": "off"
+            }
+          },
+          "mappings": [],
+          "thresholds": {
+            "mode": "absolute",
+            "steps": [
+              {
+                "color": "green",
+                "value": null
+              },
+              {
+                "color": "red",
+                "value": 80
+              }
+            ]
+          },
+          "unit": "short"
+        },
+        "overrides": []
+      },
+      "gridPos": {
+        "h": 15,
+        "w": 19,
+        "x": 0,
+        "y": 15
+      },
+      "id": 22,
+      "options": {
+        "legend": {
+          "calcs": [],
+          "displayMode": "list",
+          "placement": "bottom",
+          "showLegend": true
+        },
+        "tooltip": {
+          "mode": "multi",
+          "sort": "none"
+        }
+      },
+      "targets": [
+        {
+          "datasource": {
+            "type": "prometheus",
+            "uid": "prometheus"
+          },
+          "editorMode": "code",
+          "exemplar": true,
+          "expr": "thingsboard_transport_connections_active",
+          "format": "time_series",
+          "interval": "",
+          "intervalFactor": 1,
+          "legendFormat": "{{protocol}} {{pod_name}}",
+          "range": true,
+          "refId": "A"
+        }
+      ],
+      "title": "Active Connections by Pod",
+      "type": "timeseries"
+    },
+    {
+      "datasource": {
+        "type": "prometheus",
+        "uid": "prometheus"
+      },
+      "fieldConfig": {
+        "defaults": {
+          "color": {
+            "mode": "thresholds"
+          },
+          "custom": {
+            "align": "auto",
+            "cellOptions": {
+              "type": "auto"
+            },
+            "filterable": false,
+            "inspect": false
+          },
+          "mappings": [],
+          "min": 0,
+          "thresholds": {
+            "mode": "absolute",
+            "steps": [
+              {
+                "color": "rgba(209, 138, 78, 1)",
+                "value": null
+              }
+            ]
+          }
+        },
+        "overrides": [
+          {
+            "matcher": {
+              "id": "byName",
+              "options": "Connections"
+            },
+            "properties": [
+              {
+                "id": "custom.cellOptions",
+                "value": {
+                  "mode": "gradient",
+                  "type": "gauge"
+                }
+              }
+            ]
+          }
+        ]
+      },
+      "gridPos": {
+        "h": 15,
+        "w": 5,
+        "x": 19,
+        "y": 15
+      },
+      "id": 23,
+      "options": {
+        "cellHeight": "sm",
+        "footer": {
+          "countRows": false,
+          "fields": "",
+          "reducer": [
+            "sum"
+          ],
+          "show": false
+        },
+        "showHeader": true,
+        "sortBy": []
+      },
+      "targets": [
+        {
+          "datasource": {
+            "type": "prometheus",
+            "uid": "prometheus"
+          },
+          "editorMode": "code",
+          "exemplar": true,
+          "expr": "sum by (pod_name) (thingsboard_transport_connections_active)",
+          "format": "table",
+          "instant": true,
+          "interval": "",
+          "legendFormat": "__auto",
+          "queryType": "randomWalk",
+          "refId": "A"
+        }
+      ],
+      "title": "Connection Count by Pod",
+      "transformations": [
+        {
+          "id": "organize",
+          "options": {
+            "excludeByName": {
+              "Time": true,
+              "__name__": true,
+              "app": true,
+              "controller_revision_hash": true,
+              "instance": true,
+              "job": true,
+              "kubernetes_name": true,
+              "kubernetes_namespace": true,
+              "node_name": true,
+              "pod_name": false,
+              "statefulset_kubernetes_io_pod_name": true
+            },
+            "indexByName": {},
+            "renameByName": {
+              "Time": "",
+              "Value": "Connections",
+              "pod_name": "Pod"
+            }
+          }
+        },
+        {
+          "id": "sortBy",
+          "options": {
+            "fields": {},
+            "sort": [
+              {
+                "field": "Pod"
+              }
+            ]
+          }
+        }
+      ],
+      "type": "table"
+    }
+  ],
+  "refresh": "30s",
+  "schemaVersion": 39,
+  "tags": [
+    "thingsboard"
+  ],
+  "templating": {
+    "list": []
+  },
+  "time": {
+    "from": "now-1h",
+    "to": "now"
+  },
+  "timepicker": {
+    "refresh_intervals": [
+      "10s",
+      "30s",
+      "1m",
+      "5m",
+      "15m",
+      "30m",
+      "1h",
+      "2h",
+      "1d"
+    ]
+  },
+  "timezone": "",
+  "title": "Transport Connections",
+  "uid": "lewbrlwjerwkj3",
+  "version": 1,
+  "weekStart": ""
+}