IPv6-safe connection metrics + opt-in per-connection traffic + IPv4/IPv6 totals

README.md

@@ -90,6 +90,7 @@ The default configuration file comes with a sample configuration, making it easy
 
     connections = True              # IP connections metrics
     connection_stats = False        # Open IP connections metrics
+    connection_traffic = False      # Open IP connections traffic metrics (high cardinality)
 
     interface = True                # Interfaces traffic metrics
     wireguard_peers = False         # Wireguard peers metrics
@@ -512,7 +513,7 @@ With many connected devices everywhere, one can often only guess where do they g
 ```
 connection_stats = False        # Open IP connections metrics 
 ```
-Setting this to `True` obviously enables the feature and allows to see something like that:
+Setting this to `True` obviously enables the feature and allows to see something like that. Both IPv4 and IPv6 connections are included:
 
 <img width="2346" alt="conns" src="https://user-images.githubusercontent.com/5028474/217042107-bffa0a81-a6a0-4474-87d4-1597cdd80735.png">
 
@@ -529,6 +530,24 @@ Let's go check on that in the dashboard, or just get the info right from the com
 ```
 *A few quick checks show all of the destination IPs relate to AWS instances, so supposedly it's legit... but let's remain vigilant, to know better :)*
 
+The exporter reports these connection totals:
+
+- `mktxp_ip_connections_total` for combined IPv4 and IPv6 totals
+- `mktxp_ipv4_connections_total` for IPv4-only totals
+- `mktxp_ipv6_connections_total` for IPv6-only totals
+
+If you need per-connection byte counters for currently active src/dst/protocol tuples, enable the following option:
+
+```
+connection_traffic = False      # Open IP connections traffic metrics (high cardinality)
+```
+
+This produces the following metrics for both IPv4 and IPv6 connection tracking tables:
+
+- `mktxp_connection_upload_bytes`
+- `mktxp_connection_download_bytes`
+- `mktxp_connection_total_bytes`
+
 
 ### Parallel routers fetch
 Concurrent exports across multiple devices can considerably speed up things for slow network connections. This feature can be turned on and configured with the following [system options](https://github.com/akpw/mktxp/blob/main/README.md#mktxp-system-configuration):

deploy/kubernetes/secret.yaml

@@ -87,6 +87,7 @@ stringData:
 
             connections = True              # IP connections metrics
             connection_stats = False        # Open IP connections metrics
+            connection_traffic = False      # Open IP connections traffic metrics (high cardinality)
 
             interface = True                # Interfaces traffic metrics
             wireguard_peers = False         # Wireguard peers metrics

mktxp/cli/config/config.py

@@ -90,6 +90,7 @@ class MKTXPConfigKeys:
     FE_DHCP_LEASE_KEY = 'dhcp_lease'
     FE_IP_CONNECTIONS_KEY = 'connections'
     FE_CONNECTION_STATS_KEY = 'connection_stats'
+    FE_CONNECTION_TRAFFIC_KEY = 'connection_traffic'
     FE_INTERFACE_KEY = 'interface'
     FE_WG_PEER_KEY = 'wireguard_peers'
 
@@ -213,7 +214,7 @@ class MKTXPConfigKeys:
 
 
     BOOLEAN_KEYS_NO = {ENABLED_KEY, SSL_KEY, NO_SSL_CERTIFICATE, FE_CHECK_FOR_UPDATES, FE_KID_CONTROL_DEVICE, FE_KID_CONTROL_DYNAMIC,FE_WG_PEER_KEY,
-                       SSL_CERTIFICATE_VERIFY, FE_IPV6_ROUTE_KEY, FE_IPV6_DHCP_POOL_KEY, FE_IPV6_FIREWALL_KEY, FE_IPV6_NEIGHBOR_KEY, FE_CONNECTION_STATS_KEY, FE_BFD_KEY, FE_BGP_KEY,
+                       SSL_CERTIFICATE_VERIFY, FE_IPV6_ROUTE_KEY, FE_IPV6_DHCP_POOL_KEY, FE_IPV6_FIREWALL_KEY, FE_IPV6_NEIGHBOR_KEY, FE_CONNECTION_STATS_KEY, FE_CONNECTION_TRAFFIC_KEY, FE_BFD_KEY, FE_BGP_KEY,
                        FE_EOIP_KEY, FE_GRE_KEY, FE_IPIP_KEY, FE_IPSEC_KEY, FE_LTE_KEY, FE_SWITCH_PORT_KEY, FE_ROUTING_STATS_KEY, FE_CERTIFICATE_KEY, FE_DNS_KEY, FE_CONTAINER_KEY, FE_W60G_KEY, FE_MODULE_ONLY_KEY, FE_BRIDGE_VLAN_KEY}
 
     # Feature keys enabled by default
@@ -248,7 +249,7 @@ class ConfigEntry:
                                                        MKTXPConfigKeys.SSL_KEY, MKTXPConfigKeys.NO_SSL_CERTIFICATE, MKTXPConfigKeys.SSL_CERTIFICATE_VERIFY, MKTXPConfigKeys.SSL_CHECK_HOSTNAME, MKTXPConfigKeys.SSL_CA_FILE, MKTXPConfigKeys.PLAINTEXT_LOGIN_KEY,
                                                        MKTXPConfigKeys.FE_DHCP_KEY, MKTXPConfigKeys.FE_HEALTH_KEY, MKTXPConfigKeys.FE_PACKAGE_KEY, MKTXPConfigKeys.FE_DHCP_LEASE_KEY, MKTXPConfigKeys.FE_INTERFACE_KEY,MKTXPConfigKeys.FE_WG_PEER_KEY,
                                                        MKTXPConfigKeys.FE_MONITOR_KEY, MKTXPConfigKeys.FE_W60G_KEY, MKTXPConfigKeys.FE_WIRELESS_KEY, MKTXPConfigKeys.FE_WIRELESS_CLIENTS_KEY,
-                                                       MKTXPConfigKeys.FE_IP_CONNECTIONS_KEY, MKTXPConfigKeys.FE_CONNECTION_STATS_KEY, MKTXPConfigKeys.FE_CAPSMAN_KEY, MKTXPConfigKeys.FE_CAPSMAN_CLIENTS_KEY, MKTXPConfigKeys.FE_POE_KEY,
+                                                       MKTXPConfigKeys.FE_IP_CONNECTIONS_KEY, MKTXPConfigKeys.FE_CONNECTION_STATS_KEY, MKTXPConfigKeys.FE_CONNECTION_TRAFFIC_KEY, MKTXPConfigKeys.FE_CAPSMAN_KEY, MKTXPConfigKeys.FE_CAPSMAN_CLIENTS_KEY, MKTXPConfigKeys.FE_POE_KEY,
                                                        MKTXPConfigKeys.FE_NETWATCH_KEY, MKTXPConfigKeys.FE_INTERFACE_NAME_FORMAT, MKTXPConfigKeys.FE_PUBLIC_IP_KEY,
                                                        MKTXPConfigKeys.FE_ROUTE_KEY, MKTXPConfigKeys.FE_DHCP_POOL_KEY, MKTXPConfigKeys.FE_FIREWALL_KEY, MKTXPConfigKeys.FE_ADDRESS_LIST_KEY, MKTXPConfigKeys.FE_NEIGHBOR_KEY, MKTXPConfigKeys.FE_DNS_KEY,
                                                        MKTXPConfigKeys.FE_IPV6_ROUTE_KEY, MKTXPConfigKeys.FE_IPV6_DHCP_POOL_KEY, MKTXPConfigKeys.FE_IPV6_FIREWALL_KEY, MKTXPConfigKeys.FE_IPV6_ADDRESS_LIST_KEY, MKTXPConfigKeys.FE_IPV6_NEIGHBOR_KEY,

mktxp/cli/config/mktxp.conf

@@ -43,6 +43,7 @@
 
     connections = True              # IP connections metrics
     connection_stats = False        # Open IP connections metrics
+    connection_traffic = False      # Open IP connections traffic metrics (high cardinality)
 
     interface = True                # Interfaces traffic metrics
     wireguard_peers = False         # Wireguard peers metrics

mktxp/collector/connection_collector.py

@@ -14,7 +14,7 @@
 
 from mktxp.collector.base_collector import BaseCollector
 from mktxp.flow.processor.output import BaseOutputProcessor
-from mktxp.datasource.connection_ds import IPConnectionDatasource, IPConnectionStatsDatasource
+from mktxp.datasource.connection_ds import IPConnectionDatasource, IPConnectionStatsDatasource, IPConnectionTrafficDatasource
 
 
 class IPConnectionCollector(BaseCollector):
@@ -23,10 +23,16 @@ class IPConnectionCollector(BaseCollector):
     @staticmethod
     def collect(router_entry):
         if router_entry.config_entry.connections:       
-            connection_records = IPConnectionDatasource.metric_records(router_entry)
+            connection_records = IPConnectionDatasource.metric_records(router_entry, include_stack_counts = True)
             if connection_records:
-                connection_metrics = BaseCollector.gauge_collector('ip_connections_total', 'Number of IP connections', connection_records, 'count',)
-                yield connection_metrics
+                connection_metrics = (
+                    ('ip_connections_total', 'Number of IP connections', 'count'),
+                    ('ipv4_connections_total', 'Number of IPv4 connections', 'ipv4_count'),
+                    ('ipv6_connections_total', 'Number of IPv6 connections', 'ipv6_count'),
+                )
+
+                for metric_name, metric_desc, metric_key in connection_metrics:
+                    yield BaseCollector.gauge_collector(metric_name, metric_desc, connection_records, metric_key,)
 
         if router_entry.config_entry.connection_stats:
             connection_stats_records = IPConnectionStatsDatasource.metric_records(router_entry)
@@ -39,3 +45,19 @@ def collect(router_entry):
                                                                     connection_stats_records, 'connection_count', connection_stats_labels)
                 yield connection_stats_metrics_gauge
 
+        if router_entry.config_entry.connection_traffic:
+            connection_traffic_records = IPConnectionTrafficDatasource.metric_records(router_entry)
+            if connection_traffic_records:
+                for connection_traffic_record in connection_traffic_records:
+                    BaseOutputProcessor.augment_record(router_entry, connection_traffic_record, id_key = 'src_address')
+
+                connection_traffic_labels = ['src_address', 'dst_address', 'protocol', 'dhcp_name']
+                connection_traffic_metrics = (
+                    ('connection_upload_bytes', 'Observed uploaded bytes on active src to dst connections', 'upload_bytes'),
+                    ('connection_download_bytes', 'Observed downloaded bytes on active src to dst connections', 'download_bytes'),
+                    ('connection_total_bytes', 'Observed total bytes on active src to dst connections', 'total_bytes'),
+                )
+
+                for metric_name, metric_desc, metric_key in connection_traffic_metrics:
+                    yield BaseCollector.gauge_collector(metric_name, metric_desc,
+                                                        connection_traffic_records, metric_key, connection_traffic_labels)

mktxp/datasource/connection_ds.py

@@ -18,46 +18,120 @@
 
 class IPConnectionDatasource:
     ''' IP connections data provider
-    '''             
+    '''
+    _RESOURCE_PATHS = (('ipv4', '/ip/firewall/connection/'), ('ipv6', '/ipv6/firewall/connection/'))
+
     @staticmethod
-    def metric_records(router_entry, *, metric_labels = None):
+    def metric_records(router_entry, *, metric_labels = None, include_stack_counts = False):
         if metric_labels is None:
-            metric_labels = []        
+            metric_labels = []
         try:
-            res = router_entry.api_connection.router_api().get_resource('/ip/firewall/connection/').call('print', {'count-only': ''})
-            # result processing as described at: https://github.com/socialwifi/RouterOS-api/issues/79#issuecomment-2089744809
-            cnt_str = res.done_message.get('ret')
-            try:
-                count = int(cnt_str)
-            except (ValueError, TypeError):
-                cnt_str = '0'
-            records = [{'count': cnt_str}]            
+            count_by_family = IPConnectionDatasource._count_connection_records_by_family(router_entry)
+            records = [{'count': str(sum(count_by_family.values()))}]
+            if include_stack_counts:
+                records[0]['ipv4_count'] = str(count_by_family['ipv4'])
+                records[0]['ipv6_count'] = str(count_by_family['ipv6'])
             return BaseDSProcessor.trimmed_records(router_entry, router_records = records, metric_labels = metric_labels)
         except Exception as exc:
             print(f'Error getting IP connection info from router {router_entry.router_name}@{router_entry.config_entry.hostname}: {exc}')
             return None
 
+    @staticmethod
+    def _count_connection_records_by_family(router_entry):
+        router_api = router_entry.api_connection.router_api()
+        count_by_family = {'ipv4': 0, 'ipv6': 0}
+        last_exc = None
+        has_success = False
+
+        for family, resource_path in IPConnectionDatasource._RESOURCE_PATHS:
+            try:
+                res = router_api.get_resource(resource_path).call('print', {'count-only': ''})
+                cnt_str = res.done_message.get('ret')
+                try:
+                    count_by_family[family] = int(cnt_str)
+                except (ValueError, TypeError):
+                    pass
+                has_success = True
+            except Exception as exc:
+                last_exc = exc
+
+        if not has_success:
+            raise last_exc if last_exc else RuntimeError('Unable to read connection counters')
+
+        return count_by_family
+
+    @staticmethod
+    def _read_connection_records(router_entry, *, proplist):
+        router_api = router_entry.api_connection.router_api()
+        records = []
+        last_exc = None
+        has_success = False
+
+        for _, resource_path in IPConnectionDatasource._RESOURCE_PATHS:
+            try:
+                connection_records = router_api.get_resource(resource_path).call('print', {'proplist': proplist})
+                records.extend(connection_records)
+                has_success = True
+            except Exception as exc:
+                last_exc = exc
+
+        if not has_success:
+            raise last_exc if last_exc else RuntimeError('Unable to read connection records')
+
+        return records
+
+    @staticmethod
+    def _normalize_address(address, port = None):
+        if not address:
+            return ''
+
+        if port:
+            bracketed_suffix = f']:{port}'
+            plain_suffix = f':{port}'
+
+            if address.startswith('[') and address.endswith(bracketed_suffix):
+                address = address[1:-len(bracketed_suffix)]
+            elif address.endswith(plain_suffix):
+                address = address[:-len(plain_suffix)]
+        elif address.startswith('[') and address.endswith(']'):
+            address = address[1:-1]
+
+        return address
+
+    @staticmethod
+    def _format_address(address, port = None):
+        address = IPConnectionDatasource._normalize_address(address, port)
+        if not address or not port:
+            return address
+
+        if ':' in address:
+            return f'[{address}]:{port}'
+
+        return f'{address}:{port}'
+
 
 class IPConnectionStatsDatasource:
     ''' IP connections stats data provider
-    '''             
+    '''
     @staticmethod
     def metric_records(router_entry, *, metric_labels = None, add_router_id = True):
         if metric_labels is None:
-            metric_labels = []        
+            metric_labels = []
         try:
             # First, check if there are any connections
             count_records = IPConnectionDatasource.metric_records(router_entry)
             if count_records[0].get('count', 0) == '0':
                 return []
 
-            connection_records = router_entry.api_connection.router_api().get_resource('/ip/firewall/connection/').call('print', \
-                                                                                            {'proplist':'src-address,dst-address,protocol'})
-             # calculate number of connections per src-address
+            connection_records = IPConnectionDatasource._read_connection_records(
+                router_entry, proplist = 'src-address,src-port,dst-address,dst-port,protocol')
+            # calculate number of connections per src-address
             connections_per_src_address = {}
             for connection_record in connection_records:
-                address = connection_record['src-address'].split(':')[0]
-                destination = f"{connection_record.get('dst-address')}({connection_record.get('protocol')})"
+                address = IPConnectionDatasource._normalize_address(
+                    connection_record.get('src-address'), connection_record.get('src-port'))
+                destination = f"{IPConnectionDatasource._format_address(connection_record.get('dst-address'), connection_record.get('dst-port'))}" \
+                              f"({connection_record.get('protocol')})"
 
                 count, destinations = 0, set()
                 if connections_per_src_address.get(address):
@@ -70,14 +144,65 @@ def metric_records(router_entry, *, metric_labels = None, add_router_id = True):
             records = []
             for key, entry in connections_per_src_address.items():
                 record = {'src_address': key, 'connection_count': entry.count, 'dst_addresses': ', '.join(entry.destinations)}
-                if add_router_id: 
+                if add_router_id:
                     for router_key, router_value in router_entry.router_id.items():
                         record[router_key] = router_value
                 records.append(record)
-            return records 
+            return records
         except Exception as exc:
             print(f'Error getting IP connection stats info from router {router_entry.router_name}@{router_entry.config_entry.hostname}: {exc}')
             return None
 
 
-ConnStatsEntry = namedtuple('ConnStatsEntry', ['count', 'destinations'])
+class IPConnectionTrafficDatasource:
+    ''' IP connection traffic data provider
+    '''
+    @staticmethod
+    def metric_records(router_entry, *, metric_labels = None, add_router_id = True):
+        if metric_labels is None:
+            metric_labels = []
+        try:
+            count_records = IPConnectionDatasource.metric_records(router_entry)
+            if count_records[0].get('count', 0) == '0':
+                return []
+
+            connection_records = IPConnectionDatasource._read_connection_records(
+                router_entry, proplist = 'src-address,src-port,dst-address,dst-port,protocol,orig-bytes,repl-bytes')
+            traffic_per_connection = {}
+            for connection_record in connection_records:
+                src_address = IPConnectionDatasource._normalize_address(
+                    connection_record.get('src-address'), connection_record.get('src-port'))
+                dst_address = IPConnectionDatasource._normalize_address(
+                    connection_record.get('dst-address'), connection_record.get('dst-port'))
+                protocol = connection_record.get('protocol', '')
+                key = (src_address, dst_address, protocol)
+
+                upload_bytes = int(connection_record.get('orig-bytes') or 0)
+                download_bytes = int(connection_record.get('repl-bytes') or 0)
+
+                entry = traffic_per_connection.get(key, ConnTrafficEntry(0, 0))
+                traffic_per_connection[key] = ConnTrafficEntry(
+                    entry.upload_bytes + upload_bytes,
+                    entry.download_bytes + download_bytes,
+                )
+
+            records = []
+            for (src_address, dst_address, protocol), entry in traffic_per_connection.items():
+                record = {'src_address': src_address,
+                          'dst_address': dst_address,
+                          'protocol': protocol,
+                          'upload_bytes': entry.upload_bytes,
+                          'download_bytes': entry.download_bytes,
+                          'total_bytes': entry.upload_bytes + entry.download_bytes}
+                if add_router_id:
+                    for router_key, router_value in router_entry.router_id.items():
+                        record[router_key] = router_value
+                records.append(record)
+            return BaseDSProcessor.trimmed_records(router_entry, router_records = records, metric_labels = metric_labels)
+        except Exception as exc:
+            print(f'Error getting IP connection traffic info from router {router_entry.router_name}@{router_entry.config_entry.hostname}: {exc}')
+            return None
+
+
+ConnStatsEntry = namedtuple('ConnStatsEntry', ['count', 'destinations'])
+ConnTrafficEntry = namedtuple('ConnTrafficEntry', ['upload_bytes', 'download_bytes'])