oexport: Add a file exporter option

Write metrics as InfluxDB line protocol to a local file, enabling
offline collection without a running InfluxDB instance. Repeatable
--tag flag injects extra key=value tags into every point, useful for
stamping metrics with for instance a test run and test name identifier

Signed-off-by: Dimitri Staessens <dimitri@ouroboros.rocks>

5 files changed, 869 insertions, 616 deletions

diff --git a/.gitignore b/.gitignore
index 3065ac3..99bc104 100644
--- a/.gitignore
+++ b/.gitignore
@@ -2,3 +2,6 @@
 *#
 ./config.ini
 ./venv/
+__pycache__/
+*.egg-info/
+.vscode/
diff --git a/README.md b/README.md
index 4cc05eb..394aaf6 100644
--- a/README.md
+++ b/README.md
@@ -3,14 +3,21 @@
 A collection of observability tools for exporting and
 visualising metrics collected from Ouroboros.
 
-Currently has one very simple exporter for InfluxDB, and provides
-additional visualization via grafana.
+Currently has one very simple exporter that can push metrics to
+InfluxDB or write them as InfluxDB line protocol to a local file,
+and provides additional visualization via grafana.
 
 More features will be added over time.
 
 ## Requirements:
 
+The influxdb-client Python package is always required (used for
+point construction and line protocol serialization in both modes).
+It is installed automatically via `pip install .`
+
 InfluxDB OSS 2.0, https://docs.influxdata.com/influxdb/v2.0/
+(only required when pushing metrics directly to InfluxDB, not
+needed for file mode)
 
 ## Optional requirements:
 
@@ -56,11 +63,73 @@ source venv/bin/activate
 pip install .
 ```
 
+### InfluxDB mode (default)
+
 Edit the config.ini.example file and fill out the InfluxDB
 information (token, org). Save it as config.ini.
 
-and run oexport.py
+Run oexport.py:
 
 ```
 python oexport.py
 ```
+
+Extra tags can be added to every metric:
+
+```
+python oexport.py --tag env=production --tag region=eu
+```
+
+### File mode
+
+Write metrics as InfluxDB line protocol to a local file instead of
+pushing to a running InfluxDB instance. This requires no InfluxDB
+connection and no config.ini:
+
+```
+python oexport.py --output-file metrics.lp
+```
+
+Tags can be injected in file mode as well, which is useful for
+stamping metrics with a test name and run identifier:
+
+```
+python oexport.py --output-file metrics.lp \
+    --tag test=oping_unimpaired \
+    --tag run=20260401_143022
+```
+
+The resulting file can be bulk-imported into InfluxDB later:
+
+```
+influx write --bucket ouroboros-metrics --file metrics.lp
+```
+
+### CLI options
+
+| Flag                | Description                                     |
+|---------------------|-------------------------------------------------|
+| `-i, --interval`    | Collection interval in milliseconds (default: 1000) |
+| `-b, --bucket`      | InfluxDB bucket name (default: ouroboros-metrics) |
+| `-o, --output-file` | Write line protocol to file instead of InfluxDB |
+| `-t, --tag`         | Extra tag as key=value (repeatable)             |
+| `-c, --config`      | Path to config.ini (default: ./config.ini)      |
+| `--url`             | InfluxDB server URL                             |
+| `--org`             | InfluxDB organization                           |
+| `--token`           | InfluxDB authentication token                   |
+
+### Config file
+
+All settings can also be specified in config.ini. CLI flags take
+precedence over config file values. See config.ini.example for
+details.
+
+| Section      | Key        | Equivalent CLI flag |
+|--------------|------------|---------------------|
+| `[influx2]`  | `url`      | `--url`             |
+| `[influx2]`  | `org`      | `--org`             |
+| `[influx2]`  | `token`    | `--token`           |
+| `[exporter]` | `interval` | `--interval`        |
+| `[exporter]` | `bucket`   | `--bucket`          |
+| `[output]`   | `file`     | `--output-file`     |
+| `[output]`   | `tags`     | `--tag`             |
diff --git a/config.ini.example b/config.ini.example
index f2656f7..056adce 100644
--- a/config.ini.example
+++ b/config.ini.example
@@ -3,4 +3,16 @@ url=http://localhost:8086
 org=<your-org>
 token=<your-token>
 timeout=6000
-verify_ssl=False
\ No newline at end of file
+verify_ssl=False
+
+[exporter]
+; Collection interval in milliseconds (default: 1000)
+;interval = 1000
+; InfluxDB bucket name (default: ouroboros-metrics)
+;bucket = ouroboros-metrics
+
+[output]
+; Uncomment to write line protocol to file instead of InfluxDB
+;file = /path/to/metrics.lp
+; Extra tags added to every point (comma-separated key=value pairs)
+;tags = test=mytest,run=12345
diff --git a/oexport.py b/oexport.py
index 2f38e2e..f2dc3af 100755
--- a/oexport.py
+++ b/oexport.py
@@ -33,552 +33,95 @@ OF THE POSSIBILITY OF SUCH DAMAGE.
 """
 
 import os
-import re
 import socket
+import sys
 import time
 import argparse
+import configparser
+from abc import ABC, abstractmethod
 from datetime import datetime
-from typing import Optional
 
 from influxdb_client import InfluxDBClient, Point
 from influxdb_client.client.write_api import WriteOptions, PointSettings
 from influxdb_client.rest import ApiException
 from pytz import utc
 
-IPCP_TYPE_UNICAST = 'unicast'
-IPCP_TYPE_BROADCAST = 'broadcast'
-IPCP_TYPE_UDP = 'udp'
-IPCP_TYPE_ETH_DIX = 'eth-dix'
-IPCP_TYPE_ETH_LLC = 'eth-llc'
-IPCP_TYPE_LOCAL = 'local'
-IPCP_TYPE_UNKNOWN = 'unknown'
-
-IPCP_TYPES = [IPCP_TYPE_UNICAST,
-              IPCP_TYPE_BROADCAST,
-              IPCP_TYPE_UDP,
-              IPCP_TYPE_ETH_DIX,
-              IPCP_TYPE_ETH_LLC,
-              IPCP_TYPE_LOCAL,
-              IPCP_TYPE_UNKNOWN]
-
-IPCP_STATE_NULL = 'null'
-IPCP_STATE_INIT = 'init'
-IPCP_STATE_OPERATIONAL = 'operational'
-IPCP_STATE_SHUTDOWN = 'shutdown'
-
-IPCP_STATES = [IPCP_STATE_NULL,
-               IPCP_STATE_INIT,
-               IPCP_STATE_OPERATIONAL,
-               IPCP_STATE_SHUTDOWN]
-
-
-class OuroborosRIBReader:
-    """
-    Class for reading stuff from the Ouroboros system
-    Resource Information Base (RIB)
-    """
-    def __init__(self,
-                 rib_path: str):
-
-        self.rib_path = rib_path
-
-    def _get_dir_for_ipcp(self,
-                          ipcp_name: str) -> str:
-
-        return os.path.join(self.rib_path, ipcp_name)
-
-    def _get_dir_for_process(self,
-                             process_name: str) -> str:
-
-        return os.path.join(self.rib_path, process_name)
-
-    def _get_dt_dir_for_ipcp(self,
-                             ipcp_name: str) -> Optional[str]:
-
-        path = self._get_dir_for_ipcp(ipcp_name)
-        try:
-            _subdirs = [f.name for f in os.scandir(path)]
-        except IOError as _:
-            return None
-
-        for _dir in _subdirs:
-            if len(_dir) > 3 and _dir[:3] == 'dt.':
-                return os.path.join(path, _dir)
-
-        return None
-
-    def _get_path_for_ipcp_flow_n_plus_1_info(self,
-                                              ipcp_name: str,
-                                              fd: str):
-
-        return os.path.join(self.rib_path, ipcp_name, 'flow-allocator', fd)
-
-    def _get_path_for_ipcp_flow_n_minus_1_info(self,
-                                               ipcp_name: str,
-                                               fd: str) -> str:
-
-        dt_dir = self._get_dt_dir_for_ipcp(ipcp_name)
-        return os.path.join(dt_dir, fd)
-
-    def _get_path_for_frct_flow_info(self,
-                                     process: str,
-                                     fd: str) -> str:
+from ribreader import OuroborosRIBReader, IPCP_TYPES
 
-        process_dir = self._get_dir_for_process(process)
-        return os.path.join(process_dir, str(fd), 'frct')
 
-    def _get_ipcp_type_for_ipcp(self,
-                                ipcp_name: str) -> str:
+class MetricsWriter(ABC):
+    """Abstract base class for metrics output writers."""
 
-        _dir = self._get_dir_for_ipcp(ipcp_name)
-        path = '{}/info/_type'.format(_dir)
-        if not os.path.exists(path):
-            return IPCP_TYPE_UNKNOWN
+    @abstractmethod
+    def write(self, bucket: str, record: Point):
+        """Write a single metric point."""
 
-        try:
-            with open(path) as f:
-                return f.readline()[:-1]
-        except IOError as _:
-            return IPCP_TYPE_UNKNOWN
-
-    def _get_layer_name_for_ipcp(self,
-                                 ipcp_name: str) -> str:
-
-        _dir = self._get_dir_for_ipcp(ipcp_name)
-        path = '{}/info/_layer'.format(_dir)
-        if not os.path.exists(path):
-            return '(error)'
-
-        try:
-            with open(path) as f:
-                return f.readline()[:-1]
-        except IOError as _:
-            return '(error)'
+    @abstractmethod
+    def close(self):
+        """Release resources held by the writer."""
 
-    def _get_ipcp_state_for_ipcp(self,
-                                 ipcp_name: str) -> str:
 
-        _dir = self._get_dir_for_ipcp(ipcp_name)
-        path = '{}/info/_state'.format(_dir)
-        if not os.path.exists(path):
-            return IPCP_TYPE_UNKNOWN
+class InfluxDBWriter(MetricsWriter):
+    """Write metrics to an InfluxDB instance."""
 
-        try:
-            with open(path) as f:
-                return f.readline()[:-1]
-        except IOError as e:
-            print(e)
-            return IPCP_TYPE_UNKNOWN
-
-    def _get_n_plus_1_flows_for_ipcp(self,
-                                     ipcp_name: str) -> list[str]:
+    def __init__(self,  # pylint: disable=too-many-arguments,too-many-positional-arguments
+                 cfg_path: str = './config.ini',
+                 tags: dict = None,
+                 url: str = None,
+                 org: str = None,
+                 token: str = None):
 
-        path = os.path.join(self._get_dir_for_ipcp(ipcp_name), 'flow-allocator')
-
-        if not os.path.exists(path):
-            return []
-
-        try:
-            return [f.name for f in os.scandir(path)]
-        except IOError as e:
-            print(e)
-
-        return []
-
-    def _get_n_minus_1_flows_for_ipcp(self,
-                                      ipcp_name: str) -> list[str]:
-
-        path = self._get_dt_dir_for_ipcp(ipcp_name)
-        if path is None:
-            return []
-
-        if not os.path.exists(path):
-            return []
-
-        try:
-            return [f.name for f in os.scandir(path)]
-        except IOError as e:
-            print(e)
-            return []
-
-    def _get_address_for_ipcp(self,
-                              ipcp_name):
-
-        path = self._get_dir_for_ipcp(ipcp_name)
-        try:
-            _subdirs = [f.name for f in os.scandir(path)]
-        except IOError as _:
-            return None
-
-        for _dir in _subdirs:
-            if len(_dir) > 3 and _dir[:3] == 'dt.':
-                return _dir[3:]
-
-        return None
-
-    def get_lsdb_stats_for_ipcp(self,
-                                ipcp_name: str) -> dict:
-        """
-        Get statistics for the link state database of an IPCP
-        :param ipcp_name: name of the IPCP
-        :return: statistics in a dict
-        """
+        point_settings = PointSettings()
+        point_settings.add_default_tag('system', socket.gethostname())
 
-        address = self._get_address_for_ipcp(ipcp_name)
-        if address is None:
-            return {}
+        if tags:
+            for key, value in tags.items():
+                point_settings.add_default_tag(key, value)
 
-        path = os.path.join(self._get_dir_for_ipcp(ipcp_name), 'lsdb/')
-        if not os.path.exists(path):
-            return {}
+        write_options = WriteOptions(batch_size=500,
+                                     flush_interval=10_000,
+                                     jitter_interval=1_000,
+                                     retry_interval=1_000,
+                                     max_retries=5,
+                                     max_retry_delay=30_000,
+                                     exponential_base=2)
 
-        nodes = []
-        neighbors = 0
-        links = 0
+        if url:
+            self.client = InfluxDBClient(url=url, org=org, token=token)
+        else:
+            self.client = InfluxDBClient.from_config_file(cfg_path)
+        self._write_api = self.client.write_api(write_options=write_options,
+                                                point_settings=point_settings).write
 
-        lsdb_entries = []
+    def write(self, bucket: str, record: Point):
         try:
-            lsdb_entries = [f.path for f in os.scandir(path)]
-        except IOError as _:
-            pass
-
-        for lsdb_entry in lsdb_entries:
-            try:
-                with open(lsdb_entry) as e:
-                    for line in e.readlines():
-                        if 'src' in line:
-                            src = line.split()[-1]
-                            if src not in nodes:
-                                nodes += [src]
-                            if src == address:
-                                neighbors += 1
-                        if 'dst' in line:
-                            dst = line.split()[-1]
-                            if dst not in nodes:
-                                nodes += [dst]
-                links += 1
-            except IOError as _:
-                continue
-
-        stats = {'neighbors': neighbors,
-                 'nodes': len(nodes),
-                 'links': links}
-
-        return stats
-
-    @staticmethod
-    def _get_trailing_number(s: str) -> int:
-        m = re.search(r'\d+$', s)
-        return int(m.group()) if m else None
-
-
-    def get_dht_stats_for_ipcp(self,
-                               ipcp_name: str) -> dict:
-        """
-        Get statistics for the DHT directory of an IPCP
-        :param ipcp_name: name of the IPCP
-        :return: statistics in a dict
-        """
-
-        str_to_metric = {
-        '  Number of keys': 'keys',
-        '  Number of local values': 'local_values',
-        '  Number of non-local values': 'non_local_values'
-        }
-
-        path = os.path.join(self._get_dir_for_ipcp(ipcp_name), 'dht/stats')
-        if not os.path.exists(path):
-            return {}
-
-        ret = dict()
-
-        with open(path) as f:
-            for line in f.readlines():
-                split_line = line.split(':')
-                phrase = split_line[0]
-                if phrase not in str_to_metric.keys():
-                    continue
-                metric = str_to_metric[phrase]
-                value = self._get_trailing_number(split_line[1])
-                ret[metric] = value
-
-        return ret
-
-    def _get_flows_for_process(self,
-                               process_name: str) -> list[str]:
-        path = self._get_dir_for_process(process_name)
-
-        if not os.path.exists(path):
-            return []
-
-        try:
-            return [f.name for f in os.scandir(path) if f.is_dir()]
-        except IOError as e:
+            self._write_api(bucket=bucket, record=record)
+        except ApiException as e:
             print(e)
 
-        return []
-
-    def _get_flow_info_for_n_plus_1_flow(self,
-                                         ipcp_name: str,
-                                         fd: str) -> dict:
-
-        str_to_metric = {
-            #'Flow established at': None,
-            #'Remote address': None,
-            'Local endpoint ID': 'endpoint_id',
-            #'Remote endpoint ID': None,
-            'Sent (packets)': 'sent_pkts_total',
-            'Sent (bytes)': 'sent_bytes_total',
-            'Send failed (packets)': 'send_failed_packets_total',
-            'Send failed (bytes)': 'send_failed_bytes_total',
-            'Received (packets)': 'recv_pkts_total',
-            'Received (bytes)': 'recv_bytes_total',
-            'Receive failed (packets)': 'recv_failed_pkts_total',
-            'Receive failed (bytes)': 'recv_failed_bytes_total',
-            'Sent flow updates (packets)': 'sent_flow_updates_total',
-            'Received flow updates (packets)': 'recv_flow_updates_total',
-            #'Congestion avoidance algorithm': None,
-            'Upstream congestion level': 'up_cong_lvl',
-            'Downstream congestion level': 'down_cong_lvl',
-            'Upstream packet counter': 'up_pkt_ctr',
-            'Downstream packet counter': 'down_pkt_ctr',
-            'Congestion window size (ns)': 'cong_wnd_width_ns',
-            'Packets in this window': 'cong_wnd_current_pkts',
-            'Bytes in this window': 'cong_wnd_current_bytes',
-            'Max bytes in this window': 'cong_wnd_size_bytes',
-            #'Current congestion regime': None
-        }
-
-        ret = dict()
-
-        path = self._get_path_for_ipcp_flow_n_plus_1_info(ipcp_name, fd)
-
-        if not os.path.exists(path):
-            return dict()
-
-        with open(path) as f:
-            for line in f.readlines():
-                split_line = line.split(':')
-                phrase = split_line[0]
-                if phrase not in str_to_metric.keys():
-                    continue
-                metric = str_to_metric[phrase]
-                value = self._get_trailing_number(split_line[1])
-                ret[metric] = value
-
-        return ret
-
-    def _get_frct_info_for_process_flow(self,
-                                        process: str,
-                                        fd: str) -> dict:
-
-        str_to_metric = {
-            'Maximum packet lifetime (ns)': 'mpl_timer_ns',
-            'Max time to Ack (ns)': 'a_timer_ns',
-            'Max time to Retransmit (ns)': 'r_timer_ns',
-            'Smoothed rtt (ns)': 'srtt_ns',
-            'RTT standard deviation (ns)': 'mdev_ns',
-            'Retransmit timeout RTO (ns)': 'rto_ns',
-            'Sender left window edge': 'snd_lwe',
-            'Sender right window edge': 'snd_rwe',
-            'Sender inactive (ns)': 'snd_inact',
-            'Sender current sequence number': 'snd_seqno',
-            'Receiver left window edge': 'rcv_lwe',
-            'Receiver right window edge': 'rcv_rwe',
-            'Receiver inactive (ns)': 'rcv_inact',
-            'Receiver last ack': 'rcv_seqno',
-            'Number of pkt retransmissions': 'n_rxm',
-            'Number of rtt probes': 'n_prb',
-            'Number of rtt estimates': 'n_rtt',
-            'Number of duplicates received': 'n_dup',
-            'Number of delayed acks received': 'n_dak',
-            'Number of rendez-vous sent': 'n_rdv',
-            'Number of packets out of window': 'n_out',
-            'Number of packets out of rqueue': 'n_rqo'
-        }
-
-        ret = dict()
-
-        path = self._get_path_for_frct_flow_info(process, fd)
-
-        if not os.path.exists(path):
-            return dict()
-
-        ret['fd'] = fd
-
-        with open(path) as f:
-            for line in f.readlines():
-                split_line = line.split(':')
-                phrase = split_line[0]
-                if phrase not in str_to_metric.keys():
-                    continue
-                metric = str_to_metric[phrase]
-                value = self._get_trailing_number(split_line[1])
-                ret[metric] = value
-
-        return ret
-
-    def get_flow_allocator_flow_info_for_ipcp(self,
-                                              ipcp_name: str) -> list[dict]:
-        """
-        Get the flow intformation for all N+1 flows in a certain IPCP
-        :param ipcp_name: name of the IPCP
-        :return: dict with flow information
-        """
-        flow_info = []
-
-        flow_descriptors = self._get_n_plus_1_flows_for_ipcp(ipcp_name)
-        for flow in flow_descriptors:
-            info = self._get_flow_info_for_n_plus_1_flow(ipcp_name, flow)
-            flow_info += [info]
-
-        return flow_info
-
-    def _get_flow_info_for_n_minus_1_flow(self,
-                                          ipcp_name: str,
-                                          fd: str) -> dict:
-
-        ret = dict()
-
-        path = self._get_path_for_ipcp_flow_n_minus_1_info(ipcp_name, fd)
-
-        str_to_qos_metric = {
-         #'Flow established at': None,
-         ' sent (packets)': 'sent_packets_total',
-         ' sent (bytes)': 'sent_bytes_total',
-         ' rcvd (packets)': 'recv_packets_total',
-         ' rcvd (bytes)': 'recv_bytes_total',
-         ' local sent (packets)': 'local_sent_packets_total',
-         ' local sent (bytes)': 'local_sent_bytes_total',
-         ' local rcvd (packets)': 'local_recv_packets_total',
-         ' local rcvd (bytes)': 'local_recv_bytes_total',
-         ' dropped ttl (packets)': 'ttl_packets_dropped_total',
-         ' dropped ttl (bytes)': 'ttl_bytes_dropped_total',
-         ' failed writes (packets)': 'write_packets_dropped_total',
-         ' failed writes (bytes)': 'write_bytes_dropped_total',
-         ' failed nhop (packets)': 'nhop_packets_dropped_total',
-         ' failed nhop (bytes)': 'nhop_bytes_dropped_total'
-        }
-
-        if not os.path.exists(path):
-            return dict()
-
-        with open(path) as f:
-            _current_cube = ''
-            ret['fd'] = fd
-            for line in [_line for _line in f.readlines() if _line != '\n']:
-                if 'Endpoint address' in line:
-                    ret['endpoint'] = line.split(':')[-1].replace(' ', '')[:-1]
-                elif 'Queued packets (rx)' in line:
-                    ret['queued_packets_rx'] = self._get_trailing_number(line)
-                elif 'Queued packets (tx)' in line:
-                    ret['queued_packets_tx'] = self._get_trailing_number(line)
-                elif 'Qos cube' in line:
-                    _cube = self._get_trailing_number(line[:-2])
-                    _current_cube = 'QoS cube ' + str(_cube)
-                    ret[_current_cube] = dict()
-                else:
-                    split_line = line.split(':')
-                    if split_line[0] not in str_to_qos_metric.keys():
-                        continue
-                    metric = str_to_qos_metric[split_line[0]]
-                    value = self._get_trailing_number(split_line[1])
-                    ret[_current_cube][metric] = value
-
-        return ret
-
-    def get_data_transfer_flow_info_for_ipcp(self,
-                                             ipcp_name: str) -> list[dict]:
-        """
-        Get the flow information for all Data Transfer (N-1) flows in a certain IPCP
-        :param ipcp_name: name of the IPCP
-        :return: flow information for the data transfer flows
-        """
-
-        flow_info = []
-
-        flow_descriptors = self._get_n_minus_1_flows_for_ipcp(ipcp_name)
-        for flow in flow_descriptors:
-            info = self._get_flow_info_for_n_minus_1_flow(ipcp_name, flow)
-            flow_info += [info]
-
-        return flow_info
-
-    def get_frct_info_for_process(self,
-                                  process_name: str) -> list[dict]:
-        """
-        Get the frct information for all flows for a certain process
-        :param process_name: name of the process
-        :return: flow information for the N-1 flows
-        """
-
-        frct_info = []
-
-        flow_descriptors = self._get_flows_for_process(process_name)
-
-        for flow in flow_descriptors:
-            info = self._get_frct_info_for_process_flow(process_name, flow)
-            frct_info += [info]
-
-        return frct_info
-
-    # pylint: disable-msg=too-many-arguments
-    def get_ipcp_list(self,
-                      names_only: bool = False,  # return name and layer name
-                      types: bool = True,
-                      states: bool = True,
-                      layers: bool = True,
-                      flows: bool = True) -> list[dict]:
-        """
-        Get a list of all IPCPs
-        :param names_only: only return IPCP names and layer names
-        :param types: return IPCP type
-        :param states: return IPCP state
-        :param layers: return layer in which the IPCP is enrolled
-        :param flows: return the number of allocated (N+1) flows for this IPCP
-        :return: list of dicts containing IPCP info
-        """
+    def close(self):
+        self.client.close()
 
-        ipcp_list = []
 
-        if not os.path.exists(self.rib_path):
-            return []
+class FileWriter(MetricsWriter):
+    """Write metrics as InfluxDB line protocol to a file."""
 
-        for ipcp_dir in [f.path for f in os.scandir(self.rib_path)
-                         if f.is_dir() and not f.name.startswith('proc.')]:
-            ipcp_name = os.path.split(ipcp_dir)[-1]
-            ipcp_type = None
-            ipcp_state = None
-            ipcp_layer = self._get_layer_name_for_ipcp(ipcp_name) if layers else None
-            ipcp_flows = None
-            if not names_only:
-                ipcp_type = self._get_ipcp_type_for_ipcp(ipcp_name) if types else None
-                ipcp_state = self._get_ipcp_state_for_ipcp(ipcp_name) if states else None
-                ipcp_flows = self._get_n_plus_1_flows_for_ipcp(ipcp_name) if flows else None
-
-            ipcp_list += [{'name': ipcp_name,
-                           'type': ipcp_type,
-                           'state': ipcp_state,
-                           'layer': ipcp_layer,
-                           'flows': len(ipcp_flows) if ipcp_flows else None}]
-        return ipcp_list
-        # pylint: enable-msg=too-many-arguments
+    def __init__(self,
+                 path: str,
+                 tags: dict = None):
 
-    def get_process_list(self) -> list[str]:
-        """
-        Get a list of all the Ouroboros applications that may be exposing frct stats
-        :return: list of process names ("proc.<pid>")
-        """
-        proc_list = []
+        self._tags = tags or {}
+        self._file = open(path, 'a', encoding='utf-8')  # pylint: disable=consider-using-with
 
-        if not os.path.exists(self.rib_path):
-            return []
+    def write(self, bucket: str, record: Point):
+        for key, value in self._tags.items():
+            record = record.tag(key, value)
+        record = record.tag('system', socket.gethostname())
+        self._file.write(record.to_line_protocol() + '\n')
+        self._file.flush()
 
-        for proc in [f.name for f in os.scandir(self.rib_path)
-                     if f.is_dir() and f.name.startswith('proc.')]:
-            proc_list += [proc]
-
-        return proc_list
+    def close(self):
+        self._file.close()
 
 
 class OuroborosExporter:
@@ -587,30 +130,16 @@ class OuroborosExporter:
     """
 
     def __init__(self,
+                 metrics_writer: MetricsWriter,
                  bucket='ouroboros-metrics',
-                 config='./config.ini',
                  rib_path='/tmp/ouroboros/'):
 
-        point_settings = PointSettings()
-        point_settings.add_default_tag('system', socket.gethostname())
-
-        write_options = WriteOptions(batch_size=500,
-                                     flush_interval=10_000,
-                                     jitter_interval=1_000,
-                                     retry_interval=1_000,
-                                     max_retries=5,
-                                     max_retry_delay=30_000,
-                                     exponential_base=2)
-
         self.bucket = bucket
-        self.client = InfluxDBClient.from_config_file(config)
-        self.write_api = self.client.write_api(write_options=write_options,
-                                               point_settings=point_settings).write
-        self.query_api = self.client.query_api()
+        self.writer = metrics_writer
         self.ribreader = OuroborosRIBReader(rib_path=rib_path)
 
     def __exit__(self, _type, value, traceback):
-        self.client.close()
+        self.writer.close()
 
     def _write_ouroboros_ipcps_total(self,
                                      now,
@@ -618,7 +147,7 @@ class OuroborosExporter:
                                      n_ipcps):
 
         point = {
-            'measurement': 'ouroboros_{}_ipcps_total'.format(ipcp_type),
+            'measurement': f'ouroboros_{ipcp_type}_ipcps_total',
             'tags': {
                 'type': ipcp_type,
             },
@@ -628,8 +157,8 @@ class OuroborosExporter:
             }
         }
 
-        self.write_api(bucket=self.bucket,
-                       record=Point.from_dict(point))
+        self.writer.write(bucket=self.bucket,
+                          record=Point.from_dict(point))
 
     def _write_ouroboros_flow_allocator_flows_total(self,
                                                     now,
@@ -648,10 +177,10 @@ class OuroborosExporter:
             }
         }
 
-        self.write_api(bucket=self.bucket,
-                       record=Point.from_dict(point))
+        self.writer.write(bucket=self.bucket,
+                          record=Point.from_dict(point))
 
-    # pylint: disable-msg=too-many-arguments
+    # pylint: disable-msg=too-many-arguments,too-many-positional-arguments
     def _write_ouroboros_fa_congestion_metric(self,
                                               metric: str,
                                               now: str,
@@ -661,7 +190,7 @@ class OuroborosExporter:
                                               value):
 
         point = {
-            'measurement': 'ouroboros_flow_allocator_' + metric,
+            'measurement': f'ouroboros_flow_allocator_{metric}',
             'tags': {
                 'ipcp': ipcp_name,
                 'layer': layer,
@@ -673,11 +202,8 @@ class OuroborosExporter:
             }
         }
 
-        try:
-            self.write_api(bucket=self.bucket,
-                           record=Point.from_dict(point))
-        except ApiException as e:
-            print(e, point)
+        self.writer.write(bucket=self.bucket,
+                          record=Point.from_dict(point))
 
     def _write_ouroboros_lsdb_node_metric(self,
                                           metric: str,
@@ -687,7 +213,7 @@ class OuroborosExporter:
                                           value):
 
         point = {
-            'measurement': 'ouroboros_lsdb_' + metric + '_total',
+            'measurement': f'ouroboros_lsdb_{metric}_total',
             'tags': {
                 'ipcp': ipcp_name,
                 'layer': layer
@@ -698,11 +224,8 @@ class OuroborosExporter:
             }
         }
 
-        try:
-            self.write_api(bucket=self.bucket,
-                           record=Point.from_dict(point))
-        except ApiException as e:
-            print(e, point)
+        self.writer.write(bucket=self.bucket,
+                          record=Point.from_dict(point))
 
     def _write_ouroboros_dht_node_metric(self,
                                          metric: str,
@@ -712,7 +235,7 @@ class OuroborosExporter:
                                          value):
 
         point = {
-            'measurement': 'ouroboros_dht_' + metric + '_total',
+            'measurement': f'ouroboros_dht_{metric}_total',
             'tags': {
                 'ipcp': ipcp_name,
                 'layer': layer
@@ -723,11 +246,8 @@ class OuroborosExporter:
             }
         }
 
-        try:
-            self.write_api(bucket=self.bucket,
-                           record=Point.from_dict(point))
-        except ApiException as e:
-            print(e, point)
+        self.writer.write(bucket=self.bucket,
+                          record=Point.from_dict(point))
 
     def _write_ouroboros_data_transfer_metric(self,
                                               metric: str,
@@ -740,7 +260,7 @@ class OuroborosExporter:
                                               value):
 
         point = {
-            'measurement': 'ouroboros_data_transfer_' + metric,
+            'measurement': f'ouroboros_data_transfer_{metric}',
             'tags': {
                 'ipcp': ipcp_name,
                 'layer': layer,
@@ -754,11 +274,8 @@ class OuroborosExporter:
             }
         }
 
-        try:
-            self.write_api(bucket=self.bucket,
-                           record=Point.from_dict(point))
-        except ApiException as e:
-            print(e, point)
+        self.writer.write(bucket=self.bucket,
+                          record=Point.from_dict(point))
 
     def _write_ouroboros_data_transfer_queued(self,
                                               now,
@@ -766,10 +283,10 @@ class OuroborosExporter:
                                               ipcp_name,
                                               layer,
                                               metrics) -> None:
-        point = dict()
+        point = {}
         for metric in metrics:
             point = {
-                'measurement': 'ouroboros_data_transfer_' + metric,
+                'measurement': f'ouroboros_data_transfer_{metric}',
                 'tags': {
                     'ipcp': ipcp_name,
                     'layer': layer,
@@ -781,11 +298,8 @@ class OuroborosExporter:
                 }
             }
 
-        try:
-            self.write_api(bucket=self.bucket,
-                           record=Point.from_dict(point))
-        except ApiException as e:
-            print(e, point)
+        self.writer.write(bucket=self.bucket,
+                          record=Point.from_dict(point))
 
     def _write_ouroboros_process_frct_metric(self,
                                              now,
@@ -794,7 +308,7 @@ class OuroborosExporter:
                                              process,
                                              value):
         point = {
-            'measurement': 'ouroboros_process_frct_' + metric,
+            'measurement': f'ouroboros_process_frct_{metric}',
             'tags': {
                 'process': process,
                 'flow_descriptor': fd,
@@ -805,12 +319,9 @@ class OuroborosExporter:
             }
         }
 
-        try:
-            self.write_api(bucket=self.bucket,
-                           record=Point.from_dict(point))
-        except ApiException as e:
-            print(e, point)
-    # pylint: enable-msg=too-many-arguments
+        self.writer.write(bucket=self.bucket,
+                          record=Point.from_dict(point))
+    # pylint: enable-msg=too-many-arguments,too-many-positional-arguments
 
     @staticmethod
     def _filter_ipcp_list(ipcp_list: list[dict],
@@ -819,7 +330,7 @@ class OuroborosExporter:
                           layer: str = None) -> list[dict]:
 
         if ipcp_type not in IPCP_TYPES:
-            print("Unknown IPCP TYPE: %s" % ipcp_type)
+            print(f"Unknown IPCP TYPE: {ipcp_type}")
             return []
 
         if ipcp_type:
@@ -837,7 +348,7 @@ class OuroborosExporter:
 
         ipcps = self.ribreader.get_ipcp_list()
 
-        ipcps_total = dict()
+        ipcps_total = {}
 
         for _type in IPCP_TYPES:
             ipcps_total[_type] = len(self._filter_ipcp_list(ipcps, ipcp_type=_type))
@@ -854,7 +365,8 @@ class OuroborosExporter:
         now = datetime.now(utc)
 
         for ipcp in [ipcp for ipcp in ipcps if ipcp['flows'] is not None]:
-            self._write_ouroboros_flow_allocator_flows_total(now, ipcp['name'], ipcp['layer'], ipcp['flows'])
+            self._write_ouroboros_flow_allocator_flows_total(
+                now, ipcp['name'], ipcp['layer'], ipcp['flows'])
 
     def _export_ouroboros_fa_congestion_metrics(self):
 
@@ -865,16 +377,13 @@ class OuroborosExporter:
         for ipcp in ipcps:
             flows = self.ribreader.get_flow_allocator_flow_info_for_ipcp(ipcp['name'])
             for flow in flows:
-                for metric in flow:
+                for metric, value in flow.items():
                     if metric == 'endpoint_id':
                         continue
 
-                    self._write_ouroboros_fa_congestion_metric(metric,
-                                                               str(now),
-                                                               ipcp['name'],
-                                                               flow['endpoint_id'],
-                                                               ipcp['layer'],
-                                                               flow[metric])
+                    self._write_ouroboros_fa_congestion_metric(
+                        metric, str(now), ipcp['name'],
+                        flow['endpoint_id'], ipcp['layer'], value)
 
     def _export_ouroboros_lsdb_metrics(self):
 
@@ -891,7 +400,7 @@ class OuroborosExporter:
                                                        ipcp['layer'],
                                                        value)
 
-    def _expoert_ouroboros_dht_metrics(self):
+    def _export_ouroboros_dht_metrics(self):
 
         ipcps = self.ribreader.get_ipcp_list(names_only=True)
 
@@ -917,20 +426,18 @@ class OuroborosExporter:
                 qoscubes = [_field for _field in flow if str(_field).startswith('QoS cube')]
                 for qoscube in qoscubes:
                     for metric in flow[qoscube]:
-                        self._write_ouroboros_data_transfer_metric(metric,
-                                                                   str(now),
-                                                                   qoscube,
-                                                                   flow['fd'],
-                                                                   flow['endpoint'],
-                                                                   ipcp['name'],
-                                                                   ipcp['layer'],
-                                                                   flow[qoscube][metric])
-                self._write_ouroboros_data_transfer_queued(str(now),
-                                                           flow['fd'],
-                                                           ipcp['name'],
-                                                           ipcp['layer'],
-                                                           {'queued_packets_rx': flow['queued_packets_rx'],
-                                                            'queued_packets_tx': flow['queued_packets_tx']})
+                        self._write_ouroboros_data_transfer_metric(
+                            metric, str(now), qoscube,
+                            flow['fd'], flow['endpoint'],
+                            ipcp['name'], ipcp['layer'],
+                            flow[qoscube][metric])
+                queued = {
+                    'queued_packets_rx': flow['queued_packets_rx'],
+                    'queued_packets_tx': flow['queued_packets_tx']
+                }
+                self._write_ouroboros_data_transfer_queued(
+                    str(now), flow['fd'], ipcp['name'],
+                    ipcp['layer'], queued)
 
     def _export_ouroboros_process_frct_metrics(self):
         processes = self.ribreader.get_process_list()
@@ -939,12 +446,10 @@ class OuroborosExporter:
 
         for process in processes:
             for frct_info in self.ribreader.get_frct_info_for_process(process):
-                for metric in frct_info:
-                    self._write_ouroboros_process_frct_metric(str(now),
-                                                              metric,
-                                                              frct_info['fd'],
-                                                              process,
-                                                              frct_info[metric])
+                for metric, value in frct_info.items():
+                    self._write_ouroboros_process_frct_metric(
+                        str(now), metric, frct_info['fd'],
+                        process, value)
 
     def export(self):
         """
@@ -956,7 +461,7 @@ class OuroborosExporter:
         self._export_ouroboros_flow_allocator_flows_total()
         self._export_ouroboros_fa_congestion_metrics()
         self._export_ouroboros_lsdb_metrics()
-        self._expoert_ouroboros_dht_metrics()
+        self._export_ouroboros_dht_metrics()
         self._export_ouroboros_data_transfer_metrics()
         self._export_ouroboros_process_frct_metrics()
 
@@ -975,12 +480,78 @@ class OuroborosExporter:
 
 
 if __name__ == '__main__':
-    argparser = argparse.ArgumentParser(description="Ouroboros InfluxDB metrics exporter")
+    argparser = argparse.ArgumentParser(description="Ouroboros metrics exporter")
     argparser.add_argument('-i', '--interval', type=int, default='1000',
                            help="Interval at which to collect metrics (milliseconds)")
     argparser.add_argument('-b', '--bucket', type=str, default='ouroboros-metrics',
                            help="InfluxDB bucket to write to")
+    argparser.add_argument('-o', '--output-file', type=str, default=None,
+                           help="Write line protocol to file instead of InfluxDB")
+    argparser.add_argument('-t', '--tag', action='append', default=[],
+                           help="Extra tag as key=value (repeatable)")
+    argparser.add_argument('-c', '--config', type=str, default='./config.ini',
+                           help="Path to config.ini")
+    argparser.add_argument('--url', type=str, default=None,
+                           help="InfluxDB server URL")
+    argparser.add_argument('--org', type=str, default=None,
+                           help="InfluxDB organization")
+    argparser.add_argument('--token', type=str, default=None,
+                           help="InfluxDB authentication token")
     args = argparser.parse_args()
-    interval_ms = args.interval
-    exporter = OuroborosExporter(bucket=args.bucket)
-    exporter.run(interval_ms=interval_ms)
+
+    # Parse extra tags from CLI
+    extra_tags = {}
+    for tag_str in args.tag:
+        if '=' not in tag_str:
+            print(f"Error: --tag must be in key=value format, got: '{tag_str}'",
+                  file=sys.stderr)
+            sys.exit(1)
+        k, v = tag_str.split('=', 1)
+        extra_tags[k] = v
+
+    # Read config file for defaults
+    output_file = args.output_file
+    interval = args.interval
+    export_bucket = args.bucket
+    influx_url = args.url
+    influx_org = args.org
+    influx_token = args.token
+
+    config = configparser.ConfigParser()
+    if os.path.exists(args.config):
+        config.read(args.config)
+        if config.has_section('output'):
+            if output_file is None and config.has_option('output', 'file'):
+                output_file = config.get('output', 'file')
+            if config.has_option('output', 'tags'):
+                for tag_str in config.get('output', 'tags').split(','):
+                    tag_str = tag_str.strip()
+                    if tag_str and '=' in tag_str:
+                        k, v = tag_str.split('=', 1)
+                        extra_tags.setdefault(k, v)
+        if config.has_section('exporter'):
+            if interval == 1000 and config.has_option('exporter', 'interval'):
+                interval = config.getint('exporter', 'interval')
+            if export_bucket == 'ouroboros-metrics' and config.has_option('exporter', 'bucket'):
+                export_bucket = config.get('exporter', 'bucket')
+        if config.has_section('influx2'):
+            if influx_url is None and config.has_option('influx2', 'url'):
+                influx_url = config.get('influx2', 'url')
+            if influx_org is None and config.has_option('influx2', 'org'):
+                influx_org = config.get('influx2', 'org')
+            if influx_token is None and config.has_option('influx2', 'token'):
+                influx_token = config.get('influx2', 'token')
+
+    # Instantiate the appropriate writer
+    if output_file:
+        writer = FileWriter(path=output_file, tags=extra_tags)
+    else:
+        writer = InfluxDBWriter(cfg_path=args.config, tags=extra_tags,
+                                url=influx_url, org=influx_org,
+                                token=influx_token)
+
+    exporter = OuroborosExporter(metrics_writer=writer, bucket=export_bucket)
+    try:
+        exporter.run(interval_ms=interval)
+    finally:
+        writer.close()
diff --git a/ribreader.py b/ribreader.py
new file mode 100644
index 0000000..30a2f57
--- /dev/null
+++ b/ribreader.py
@@ -0,0 +1,598 @@
+"""
+Ouroboros RIB filesystem reader
+
+Redistribution and use in source and binary forms, with or without
+modification, are permitted provided that the following conditions
+are met:
+
+1. Redistributions of source code must retain the above copyright
+notice, this list of conditions and the following disclaimer.
+
+2. Redistributions in binary form must reproduce the above
+copyright notice, this list of conditions and the following
+disclaimer in the documentation and/or other materials provided
+with the distribution.
+
+3. Neither the name of the copyright holder nor the names of its
+contributors may be used to endorse or promote products derived
+from this software without specific prior written permission.
+
+THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+"AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
+LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS
+FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE
+COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT,
+INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES
+(INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR
+SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION)
+HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT,
+STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
+ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED
+OF THE POSSIBILITY OF SUCH DAMAGE.
+"""
+
+import os
+import re
+from typing import Optional
+
+IPCP_TYPE_UNICAST = 'unicast'
+IPCP_TYPE_BROADCAST = 'broadcast'
+IPCP_TYPE_UDP = 'udp'
+IPCP_TYPE_ETH_DIX = 'eth-dix'
+IPCP_TYPE_ETH_LLC = 'eth-llc'
+IPCP_TYPE_LOCAL = 'local'
+IPCP_TYPE_UNKNOWN = 'unknown'
+
+IPCP_TYPES = [IPCP_TYPE_UNICAST,
+              IPCP_TYPE_BROADCAST,
+              IPCP_TYPE_UDP,
+              IPCP_TYPE_ETH_DIX,
+              IPCP_TYPE_ETH_LLC,
+              IPCP_TYPE_LOCAL,
+              IPCP_TYPE_UNKNOWN]
+
+IPCP_STATE_NULL = 'null'
+IPCP_STATE_INIT = 'init'
+IPCP_STATE_OPERATIONAL = 'operational'
+IPCP_STATE_SHUTDOWN = 'shutdown'
+
+IPCP_STATES = [IPCP_STATE_NULL,
+               IPCP_STATE_INIT,
+               IPCP_STATE_OPERATIONAL,
+               IPCP_STATE_SHUTDOWN]
+
+
+class OuroborosRIBReader:
+    """
+    Class for reading stuff from the Ouroboros system
+    Resource Information Base (RIB)
+    """
+    def __init__(self,
+                 rib_path: str):
+
+        self.rib_path = rib_path
+
+    def _get_dir_for_ipcp(self,
+                          ipcp_name: str) -> str:
+
+        return os.path.join(self.rib_path, ipcp_name)
+
+    def _get_dir_for_process(self,
+                             process_name: str) -> str:
+
+        return os.path.join(self.rib_path, process_name)
+
+    def _get_dt_dir_for_ipcp(self,
+                             ipcp_name: str) -> Optional[str]:
+
+        path = self._get_dir_for_ipcp(ipcp_name)
+        try:
+            _subdirs = [f.name for f in os.scandir(path)]
+        except IOError as _:
+            return None
+
+        for _dir in _subdirs:
+            if len(_dir) > 3 and _dir[:3] == 'dt.':
+                return os.path.join(path, _dir)
+
+        return None
+
+    def _get_path_for_ipcp_flow_n_plus_1_info(self,
+                                              ipcp_name: str,
+                                              fd: str):
+
+        return os.path.join(self.rib_path, ipcp_name,
+                            'flow-allocator', fd)
+
+    def _get_path_for_ipcp_flow_n_minus_1_info(self,
+                                               ipcp_name: str,
+                                               fd: str) -> str:
+
+        dt_dir = self._get_dt_dir_for_ipcp(ipcp_name)
+        return os.path.join(dt_dir, fd)
+
+    def _get_path_for_frct_flow_info(self,
+                                     process: str,
+                                     fd: str) -> str:
+
+        process_dir = self._get_dir_for_process(process)
+        return os.path.join(process_dir, str(fd), 'frct')
+
+    def _get_ipcp_type_for_ipcp(self,
+                                ipcp_name: str) -> str:
+
+        _dir = self._get_dir_for_ipcp(ipcp_name)
+        path = f'{_dir}/info/_type'
+        if not os.path.exists(path):
+            return IPCP_TYPE_UNKNOWN
+
+        try:
+            with open(path, encoding='utf-8') as f:
+                return f.readline()[:-1]
+        except IOError as _:
+            return IPCP_TYPE_UNKNOWN
+
+    def _get_layer_name_for_ipcp(self,
+                                 ipcp_name: str) -> str:
+
+        _dir = self._get_dir_for_ipcp(ipcp_name)
+        path = f'{_dir}/info/_layer'
+        if not os.path.exists(path):
+            return '(error)'
+
+        try:
+            with open(path, encoding='utf-8') as f:
+                return f.readline()[:-1]
+        except IOError as _:
+            return '(error)'
+
+    def _get_ipcp_state_for_ipcp(self,
+                                 ipcp_name: str) -> str:
+
+        _dir = self._get_dir_for_ipcp(ipcp_name)
+        path = f'{_dir}/info/_state'
+        if not os.path.exists(path):
+            return IPCP_TYPE_UNKNOWN
+
+        try:
+            with open(path, encoding='utf-8') as f:
+                return f.readline()[:-1]
+        except IOError as e:
+            print(e)
+            return IPCP_TYPE_UNKNOWN
+
+    def _get_n_plus_1_flows_for_ipcp(self,
+                                     ipcp_name: str) -> list[str]:
+
+        path = os.path.join(self._get_dir_for_ipcp(ipcp_name),
+                            'flow-allocator')
+
+        if not os.path.exists(path):
+            return []
+
+        try:
+            return [f.name for f in os.scandir(path)]
+        except IOError as e:
+            print(e)
+
+        return []
+
+    def _get_n_minus_1_flows_for_ipcp(self,
+                                      ipcp_name: str) -> list[str]:
+
+        path = self._get_dt_dir_for_ipcp(ipcp_name)
+        if path is None:
+            return []
+
+        if not os.path.exists(path):
+            return []
+
+        try:
+            return [f.name for f in os.scandir(path)]
+        except IOError as e:
+            print(e)
+            return []
+
+    def _get_address_for_ipcp(self,
+                              ipcp_name):
+
+        path = self._get_dir_for_ipcp(ipcp_name)
+        try:
+            _subdirs = [f.name for f in os.scandir(path)]
+        except IOError as _:
+            return None
+
+        for _dir in _subdirs:
+            if len(_dir) > 3 and _dir[:3] == 'dt.':
+                return _dir[3:]
+
+        return None
+
+    def get_lsdb_stats_for_ipcp(self,
+                                ipcp_name: str) -> dict:
+        """
+        Get statistics for the link state database of an IPCP
+        :param ipcp_name: name of the IPCP
+        :return: statistics in a dict
+        """
+
+        address = self._get_address_for_ipcp(ipcp_name)
+        if address is None:
+            return {}
+
+        path = os.path.join(self._get_dir_for_ipcp(ipcp_name),
+                            'lsdb/')
+        if not os.path.exists(path):
+            return {}
+
+        nodes = []
+        neighbors = 0
+        links = 0
+
+        lsdb_entries = []
+        try:
+            lsdb_entries = [f.path for f in os.scandir(path)]
+        except IOError as _:
+            pass
+
+        for lsdb_entry in lsdb_entries:
+            try:
+                with open(lsdb_entry, encoding='utf-8') as e:
+                    for line in e.readlines():
+                        if 'src' in line:
+                            src = line.split()[-1]
+                            if src not in nodes:
+                                nodes += [src]
+                            if src == address:
+                                neighbors += 1
+                        if 'dst' in line:
+                            dst = line.split()[-1]
+                            if dst not in nodes:
+                                nodes += [dst]
+                links += 1
+            except IOError as _:
+                continue
+
+        stats = {'neighbors': neighbors,
+                 'nodes': len(nodes),
+                 'links': links}
+
+        return stats
+
+    @staticmethod
+    def _get_trailing_number(s: str) -> int:
+        """Extract trailing integer from a string."""
+        m = re.search(r'\d+$', s)
+        return int(m.group()) if m else None
+
+    def get_dht_stats_for_ipcp(self,
+                               ipcp_name: str) -> dict:
+        """
+        Get statistics for the DHT directory of an IPCP
+        :param ipcp_name: name of the IPCP
+        :return: statistics in a dict
+        """
+
+        str_to_metric = {
+            '  Number of keys': 'keys',
+            '  Number of local values': 'local_values',
+            '  Number of non-local values': 'non_local_values'
+        }
+
+        path = os.path.join(self._get_dir_for_ipcp(ipcp_name),
+                            'dht/stats')
+        if not os.path.exists(path):
+            return {}
+
+        ret = {}
+
+        with open(path, encoding='utf-8') as f:
+            for line in f.readlines():
+                split_line = line.split(':')
+                phrase = split_line[0]
+                if phrase not in str_to_metric:
+                    continue
+                metric = str_to_metric[phrase]
+                value = self._get_trailing_number(split_line[1])
+                ret[metric] = value
+
+        return ret
+
+    def _get_flows_for_process(self,
+                               process_name: str) -> list[str]:
+        """Get flow descriptors for a process."""
+        path = self._get_dir_for_process(process_name)
+
+        if not os.path.exists(path):
+            return []
+
+        try:
+            return [f.name for f in os.scandir(path) if f.is_dir()]
+        except IOError as e:
+            print(e)
+
+        return []
+
+    def _get_flow_info_for_n_plus_1_flow(self,
+                                         ipcp_name: str,
+                                         fd: str) -> dict:
+        """Get info for a single N+1 flow."""
+
+        str_to_metric = {
+            'Local endpoint ID': 'endpoint_id',
+            'Sent (packets)': 'sent_pkts_total',
+            'Sent (bytes)': 'sent_bytes_total',
+            'Send failed (packets)': 'send_failed_packets_total',
+            'Send failed (bytes)': 'send_failed_bytes_total',
+            'Received (packets)': 'recv_pkts_total',
+            'Received (bytes)': 'recv_bytes_total',
+            'Receive failed (packets)': 'recv_failed_pkts_total',
+            'Receive failed (bytes)': 'recv_failed_bytes_total',
+            'Sent flow updates (packets)': 'sent_flow_updates_total',
+            'Received flow updates (packets)': 'recv_flow_updates_total',
+            'Upstream congestion level': 'up_cong_lvl',
+            'Downstream congestion level': 'down_cong_lvl',
+            'Upstream packet counter': 'up_pkt_ctr',
+            'Downstream packet counter': 'down_pkt_ctr',
+            'Congestion window size (ns)': 'cong_wnd_width_ns',
+            'Packets in this window': 'cong_wnd_current_pkts',
+            'Bytes in this window': 'cong_wnd_current_bytes',
+            'Max bytes in this window': 'cong_wnd_size_bytes',
+        }
+
+        ret = {}
+
+        path = self._get_path_for_ipcp_flow_n_plus_1_info(
+            ipcp_name, fd)
+
+        if not os.path.exists(path):
+            return {}
+
+        with open(path, encoding='utf-8') as f:
+            for line in f.readlines():
+                split_line = line.split(':')
+                phrase = split_line[0]
+                if phrase not in str_to_metric:
+                    continue
+                metric = str_to_metric[phrase]
+                value = self._get_trailing_number(split_line[1])
+                ret[metric] = value
+
+        return ret
+
+    def _get_frct_info_for_process_flow(self,
+                                        process: str,
+                                        fd: str) -> dict:
+        """Get FRCT info for a single process flow."""
+
+        str_to_metric = {
+            'Maximum packet lifetime (ns)': 'mpl_timer_ns',
+            'Max time to Ack (ns)': 'a_timer_ns',
+            'Max time to Retransmit (ns)': 'r_timer_ns',
+            'Smoothed rtt (ns)': 'srtt_ns',
+            'RTT standard deviation (ns)': 'mdev_ns',
+            'Retransmit timeout RTO (ns)': 'rto_ns',
+            'Sender left window edge': 'snd_lwe',
+            'Sender right window edge': 'snd_rwe',
+            'Sender inactive (ns)': 'snd_inact',
+            'Sender current sequence number': 'snd_seqno',
+            'Receiver left window edge': 'rcv_lwe',
+            'Receiver right window edge': 'rcv_rwe',
+            'Receiver inactive (ns)': 'rcv_inact',
+            'Receiver last ack': 'rcv_seqno',
+            'Number of pkt retransmissions': 'n_rxm',
+            'Number of rtt probes': 'n_prb',
+            'Number of rtt estimates': 'n_rtt',
+            'Number of duplicates received': 'n_dup',
+            'Number of delayed acks received': 'n_dak',
+            'Number of rendez-vous sent': 'n_rdv',
+            'Number of packets out of window': 'n_out',
+            'Number of packets out of rqueue': 'n_rqo'
+        }
+
+        ret = {}
+
+        path = self._get_path_for_frct_flow_info(process, fd)
+
+        if not os.path.exists(path):
+            return {}
+
+        ret['fd'] = fd
+
+        with open(path, encoding='utf-8') as f:
+            for line in f.readlines():
+                split_line = line.split(':')
+                phrase = split_line[0]
+                if phrase not in str_to_metric:
+                    continue
+                metric = str_to_metric[phrase]
+                value = self._get_trailing_number(split_line[1])
+                ret[metric] = value
+
+        return ret
+
+    def get_flow_allocator_flow_info_for_ipcp(self,
+                                              ipcp_name: str
+                                              ) -> list[dict]:
+        """
+        Get the flow information for all N+1 flows in an IPCP.
+        :param ipcp_name: name of the IPCP
+        :return: dict with flow information
+        """
+        flow_info = []
+
+        flow_descriptors = self._get_n_plus_1_flows_for_ipcp(
+            ipcp_name)
+        for flow in flow_descriptors:
+            info = self._get_flow_info_for_n_plus_1_flow(
+                ipcp_name, flow)
+            flow_info += [info]
+
+        return flow_info
+
+    def _get_flow_info_for_n_minus_1_flow(self,
+                                          ipcp_name: str,
+                                          fd: str) -> dict:
+        """Get info for a single N-1 (data transfer) flow."""
+
+        ret = {}
+
+        path = self._get_path_for_ipcp_flow_n_minus_1_info(
+            ipcp_name, fd)
+
+        str_to_qos_metric = {
+            ' sent (packets)': 'sent_packets_total',
+            ' sent (bytes)': 'sent_bytes_total',
+            ' rcvd (packets)': 'recv_packets_total',
+            ' rcvd (bytes)': 'recv_bytes_total',
+            ' local sent (packets)': 'local_sent_packets_total',
+            ' local sent (bytes)': 'local_sent_bytes_total',
+            ' local rcvd (packets)': 'local_recv_packets_total',
+            ' local rcvd (bytes)': 'local_recv_bytes_total',
+            ' dropped ttl (packets)': 'ttl_packets_dropped_total',
+            ' dropped ttl (bytes)': 'ttl_bytes_dropped_total',
+            ' failed writes (packets)': 'write_packets_dropped_total',
+            ' failed writes (bytes)': 'write_bytes_dropped_total',
+            ' failed nhop (packets)': 'nhop_packets_dropped_total',
+            ' failed nhop (bytes)': 'nhop_bytes_dropped_total'
+        }
+
+        if not os.path.exists(path):
+            return {}
+
+        with open(path, encoding='utf-8') as f:
+            _current_cube = ''
+            ret['fd'] = fd
+            for line in [_l for _l in f.readlines() if _l != '\n']:
+                if 'Endpoint address' in line:
+                    ret['endpoint'] = (
+                        line.split(':')[-1].replace(' ', '')[:-1])
+                elif 'Queued packets (rx)' in line:
+                    ret['queued_packets_rx'] = (
+                        self._get_trailing_number(line))
+                elif 'Queued packets (tx)' in line:
+                    ret['queued_packets_tx'] = (
+                        self._get_trailing_number(line))
+                elif 'Qos cube' in line:
+                    _cube = self._get_trailing_number(line[:-2])
+                    _current_cube = f'QoS cube {_cube}'
+                    ret[_current_cube] = {}
+                else:
+                    split_line = line.split(':')
+                    if split_line[0] not in str_to_qos_metric:
+                        continue
+                    metric = str_to_qos_metric[split_line[0]]
+                    value = self._get_trailing_number(
+                        split_line[1])
+                    ret[_current_cube][metric] = value
+
+        return ret
+
+    def get_data_transfer_flow_info_for_ipcp(self,
+                                             ipcp_name: str
+                                             ) -> list[dict]:
+        """
+        Get flow info for all Data Transfer (N-1) flows in an IPCP.
+        :param ipcp_name: name of the IPCP
+        :return: flow information for the data transfer flows
+        """
+
+        flow_info = []
+
+        flow_descriptors = self._get_n_minus_1_flows_for_ipcp(
+            ipcp_name)
+        for flow in flow_descriptors:
+            info = self._get_flow_info_for_n_minus_1_flow(
+                ipcp_name, flow)
+            flow_info += [info]
+
+        return flow_info
+
+    def get_frct_info_for_process(self,
+                                  process_name: str) -> list[dict]:
+        """
+        Get the frct information for all flows for a process.
+        :param process_name: name of the process
+        :return: flow information for the N-1 flows
+        """
+
+        frct_info = []
+
+        flow_descriptors = self._get_flows_for_process(
+            process_name)
+
+        for flow in flow_descriptors:
+            info = self._get_frct_info_for_process_flow(
+                process_name, flow)
+            frct_info += [info]
+
+        return frct_info
+
+    # pylint: disable-msg=too-many-arguments,too-many-positional-arguments
+    def get_ipcp_list(self,
+                      names_only: bool = False,
+                      types: bool = True,
+                      states: bool = True,
+                      layers: bool = True,
+                      flows: bool = True) -> list[dict]:
+        """
+        Get a list of all IPCPs.
+        :param names_only: only return IPCP names and layer names
+        :param types: return IPCP type
+        :param states: return IPCP state
+        :param layers: return layer the IPCP is enrolled in
+        :param flows: return number of allocated (N+1) flows
+        :return: list of dicts containing IPCP info
+        """
+
+        ipcp_list = []
+
+        if not os.path.exists(self.rib_path):
+            return []
+
+        for ipcp_dir in [f.path for f in os.scandir(self.rib_path)
+                         if f.is_dir()
+                         and not f.name.startswith('proc.')]:
+            ipcp_name = os.path.split(ipcp_dir)[-1]
+            ipcp_type = None
+            ipcp_state = None
+            ipcp_layer = (self._get_layer_name_for_ipcp(ipcp_name)
+                          if layers else None)
+            ipcp_flows = None
+            if not names_only:
+                ipcp_type = (
+                    self._get_ipcp_type_for_ipcp(ipcp_name)
+                    if types else None)
+                ipcp_state = (
+                    self._get_ipcp_state_for_ipcp(ipcp_name)
+                    if states else None)
+                ipcp_flows = (
+                    self._get_n_plus_1_flows_for_ipcp(ipcp_name)
+                    if flows else None)
+
+            ipcp_list += [{
+                'name': ipcp_name,
+                'type': ipcp_type,
+                'state': ipcp_state,
+                'layer': ipcp_layer,
+                'flows': (len(ipcp_flows)
+                          if ipcp_flows else None)}]
+        return ipcp_list
+    # pylint: enable-msg=too-many-arguments,too-many-positional-arguments
+
+    def get_process_list(self) -> list[str]:
+        """
+        Get a list of all Ouroboros applications exposing frct stats.
+        :return: list of process names ("proc.<pid>")
+        """
+        proc_list = []
+
+        if not os.path.exists(self.rib_path):
+            return []
+
+        for proc in [f.name for f in os.scandir(self.rib_path)
+                     if f.is_dir()
+                     and f.name.startswith('proc.')]:
+            proc_list += [proc]
+
+        return proc_list