Merge pull request #111 from craigerl/ratelimit

Add ratelimiting for acks and other packets

aprsd/client.py

@@ -84,6 +84,8 @@ class Client:
 
 class APRSISClient(Client):
 
+    _client = None
+
     @staticmethod
     def is_enabled():
         # Defaults to True if the enabled flag is non existent
@@ -115,6 +117,12 @@ class APRSISClient(Client):
             return True
         return True
 
+    def is_alive(self):
+        if self._client:
+            return self._client.is_alive()
+        else:
+            return False
+
     @staticmethod
     def transport():
         return TRANSPORT_APRSIS
@@ -157,6 +165,8 @@ class APRSISClient(Client):
 
 class KISSClient(Client):
 
+    _client = None
+
     @staticmethod
     def is_enabled():
         """Return if tcp or serial KISS is enabled."""
@@ -189,6 +199,12 @@ class KISSClient(Client):
             return True
         return False
 
+    def is_alive(self):
+        if self._client:
+            return self._client.is_alive()
+        else:
+            return False
+
     @staticmethod
     def transport():
         if CONF.kiss_serial.enabled:
@@ -214,8 +230,8 @@ class KISSClient(Client):
 
     @trace.trace
     def setup_connection(self):
-        client = kiss.KISS3Client()
+        self._client = kiss.KISS3Client()
-        return client
+        return self._client
 
 
 class ClientFactory:
@@ -241,7 +257,6 @@ class ClientFactory:
             elif KISSClient.is_enabled():
                 key = KISSClient.transport()
 
-        LOG.debug(f"GET client '{key}'")
         builder = self._builders.get(key)
         if not builder:
             raise ValueError(key)

aprsd/clients/aprsis.py

@@ -37,6 +37,10 @@ class Aprsdis(aprslib.IS):
         """Send an APRS Message object."""
         self.sendall(packet.raw)
 
+    def is_alive(self):
+        """If the connection is alive or not."""
+        return self._connected
+
     def _socket_readlines(self, blocking=False):
         """
         Generator for complete lines, received from the server

aprsd/clients/kiss.py

@@ -17,6 +17,9 @@ class KISS3Client:
     def __init__(self):
         self.setup()
 
+    def is_alive(self):
+        return True
+
     def setup(self):
         # we can be TCP kiss or Serial kiss
         if CONF.kiss_serial.enabled:

aprsd/conf/common.py

@@ -47,6 +47,20 @@ aprsd_opts = [
         default="imperial",
         help="Units for display, imperial or metric",
     ),
+    cfg.IntOpt(
+        "ack_rate_limit_period",
+        default=1,
+        help="The wait period in seconds per Ack packet being sent."
+             "1 means 1 ack packet per second allowed."
+             "2 means 1 pack packet every 2 seconds allowed",
+    ),
+    cfg.IntOpt(
+        "msg_rate_limit_period",
+        default=2,
+        help="Wait period in seconds per non AckPacket being sent."
+             "2 means 1 packet every 2 seconds allowed."
+             "5 means 1 pack packet every 5 seconds allowed",
+    ),
 ]
 
 watch_list_opts = [

aprsd/rpc/server.py

@@ -62,6 +62,7 @@ class APRSDService(rpyc.Service):
     def get_stats(self):
         stat = stats.APRSDStats()
         stats_dict = stat.stats()
+        LOG.debug(stats_dict)
         return json.dumps(stats_dict, indent=4, sort_keys=True, default=str)
 
     @rpyc.exposed

aprsd/stats.py

@@ -233,6 +233,7 @@ class APRSDStats:
             },
             "plugins": plugin_stats,
         }
+        LOG.debug(f"STATS = {stats}")
         return stats
 
     def __str__(self):

aprsd/threads/aprsd.py

@@ -66,6 +66,10 @@ class APRSDThread(threading.Thread, metaclass=abc.ABCMeta):
     def _cleanup(self):
         """Add code to subclass to do any cleanup"""
 
+    def __str__(self):
+        out = f"Thread <{self.__class__.__name__}({self.name}) Alive? {self.is_alive()}>"
+        return out
+
     def run(self):
         LOG.debug("Starting")
         while not self._should_quit():

aprsd/threads/keep_alive.py

@@ -64,22 +64,35 @@ class KeepAliveThread(APRSDThread):
                 len(thread_list),
             )
             LOG.info(keepalive)
+            thread_out = []
+            for thread in thread_list.threads_list:
+                alive = thread.is_alive()
+                thread_out.append(f"{thread.__class__.__name__}:{alive}")
+                if not alive:
+                    LOG.error(f"Thread {thread}")
+            LOG.info(",".join(thread_out))
 
-            # See if we should reset the aprs-is client
+            # check the APRS connection
-            # Due to losing a keepalive from them
+            cl = client.factory.create()
-            delta_dict = utils.parse_delta_str(last_msg_time)
+            if not cl.is_alive():
-            delta = datetime.timedelta(**delta_dict)
+                LOG.error(f"{cl.__class__.__name__} is not alive!!! Resetting")
+                client.factory.create().reset()
+            else:
+                # See if we should reset the aprs-is client
+                # Due to losing a keepalive from them
+                delta_dict = utils.parse_delta_str(last_msg_time)
+                delta = datetime.timedelta(**delta_dict)
 
-            if delta > self.max_delta:
+                if delta > self.max_delta:
-                #  We haven't gotten a keepalive from aprs-is in a while
+                    #  We haven't gotten a keepalive from aprs-is in a while
-                # reset the connection.a
+                    # reset the connection.a
-                if not client.KISSClient.is_enabled():
+                    if not client.KISSClient.is_enabled():
-                    LOG.warning(f"Resetting connection to APRS-IS {delta}")
+                        LOG.warning(f"Resetting connection to APRS-IS {delta}")
-                    client.factory.create().reset()
+                        client.factory.create().reset()
 
-            # Check version every hour
+            # Check version every day
             delta = now - self.checker_time
-            if delta > datetime.timedelta(hours=1):
+            if delta > datetime.timedelta(hours=24):
                 self.checker_time = now
                 level, msg = utils._check_version()
                 if level:

aprsd/threads/rx.py

@@ -65,6 +65,7 @@ class APRSDPluginRXThread(APRSDRXThread):
     receives packets from APRIS and then sends them for
     processing in the PluginProcessPacketThread.
     """
+
     def process_packet(self, *args, **kwargs):
         packet = self._client.decode_packet(*args, **kwargs)
         # LOG.debug(raw)

aprsd/threads/tx.py

@@ -2,34 +2,61 @@ import datetime
 import logging
 import time
 
+from oslo_config import cfg
+from ratelimiter import RateLimiter
+
 from aprsd import client
 from aprsd import threads as aprsd_threads
 from aprsd.packets import core, tracker
 
 
+CONF = cfg.CONF
 LOG = logging.getLogger("APRSD")
 
 
+def limited(until):
+    duration = int(round(until - time.time()))
+    LOG.debug(f"Rate limited, sleeping for {duration:d} seconds")
+
+
 def send(packet: core.Packet, direct=False, aprs_client=None):
     """Send a packet either in a thread or directly to the client."""
     # prepare the packet for sending.
     # This constructs the packet.raw
     packet.prepare()
+    if isinstance(packet, core.AckPacket):
+        _send_ack(packet, direct=direct, aprs_client=aprs_client)
+    else:
+        _send_packet(packet, direct=direct, aprs_client=aprs_client)
+
+
+@RateLimiter(max_calls=1, period=CONF.msg_rate_limit_period, callback=limited)
+def _send_packet(packet: core.Packet, direct=False, aprs_client=None):
     if not direct:
-        if isinstance(packet, core.AckPacket):
+        thread = SendPacketThread(packet=packet)
-            thread = SendAckThread(packet=packet)
-        else:
-            thread = SendPacketThread(packet=packet)
         thread.start()
     else:
-        if aprs_client:
+        _send_direct(packet, aprs_client=aprs_client)
-            cl = aprs_client
-        else:
-            cl = client.factory.create()
 
-        packet.update_timestamp()
+
-        packet.log(header="TX")
+@RateLimiter(max_calls=1, period=CONF.ack_rate_limit_period, callback=limited)
-        cl.send(packet)
+def _send_ack(packet: core.AckPacket, direct=False, aprs_client=None):
+    if not direct:
+        thread = SendAckThread(packet=packet)
+        thread.start()
+    else:
+        _send_direct(packet, aprs_client=aprs_client)
+
+
+def _send_direct(packet, aprs_client=None):
+    if aprs_client:
+        cl = aprs_client
+    else:
+        cl = client.factory.create()
+
+    packet.update_timestamp()
+    packet.log(header="TX")
+    cl.send(packet)
 
 
 class SendPacketThread(aprsd_threads.APRSDThread):

dev-requirements.txt

@@ -5,22 +5,22 @@
 #    pip-compile --annotation-style=line --resolver=backtracking dev-requirements.in
 #
 add-trailing-comma==2.4.0  # via gray
-alabaster==0.7.12         # via sphinx
+alabaster==0.7.13         # via sphinx
 attrs==22.2.0             # via jsonschema, pytest
 autoflake==1.5.3          # via gray
 babel==2.11.0             # via sphinx
 black==22.12.0            # via gray
-build==0.9.0              # via pip-tools
+build==0.10.0             # via pip-tools
-cachetools==5.2.0         # via tox
+cachetools==5.2.1         # via tox
 certifi==2022.12.7        # via requests
 cfgv==3.3.1               # via pre-commit
 chardet==5.1.0            # via tox
-charset-normalizer==2.1.1  # via requests
+charset-normalizer==3.0.1  # via requests
 click==8.1.3              # via black, pip-tools
 colorama==0.4.6           # via tox
 commonmark==0.9.1         # via rich
 configargparse==1.5.3     # via gray
-coverage[toml]==7.0.3     # via pytest-cov
+coverage[toml]==7.0.5     # via pytest-cov
 distlib==0.3.6            # via virtualenv
 docutils==0.19            # via sphinx
 exceptiongroup==1.1.0     # via pytest
@@ -28,7 +28,7 @@ filelock==3.9.0           # via tox, virtualenv
 fixit==0.1.4              # via gray
 flake8==6.0.0             # via -r dev-requirements.in, fixit, pep8-naming
 gray==0.13.0              # via -r dev-requirements.in
-identify==2.5.12          # via pre-commit
+identify==2.5.13          # via pre-commit
 idna==3.4                 # via requests
 imagesize==1.4.1          # via sphinx
 importlib-metadata==6.0.0  # via sphinx
@@ -38,14 +38,13 @@ isort==5.11.4             # via -r dev-requirements.in, gray
 jinja2==3.1.2             # via sphinx
 jsonschema==4.17.3        # via fixit
 libcst==0.4.9             # via fixit
-markupsafe==2.1.1         # via jinja2
+markupsafe==2.1.2         # via jinja2
 mccabe==0.7.0             # via flake8
 mypy==0.991               # via -r dev-requirements.in
 mypy-extensions==0.4.3    # via black, mypy, typing-inspect
 nodeenv==1.7.0            # via pre-commit
-packaging==22.0           # via build, pyproject-api, pytest, sphinx, tox
+packaging==23.0           # via build, pyproject-api, pytest, sphinx, tox
 pathspec==0.10.3          # via black
-pep517==0.13.0            # via build
 pep8-naming==0.13.3       # via -r dev-requirements.in
 pip-tools==6.12.1         # via -r dev-requirements.in
 platformdirs==2.6.2       # via black, tox, virtualenv
@@ -54,18 +53,19 @@ pre-commit==2.21.0        # via -r dev-requirements.in
 pycodestyle==2.10.0       # via flake8
 pyflakes==3.0.1           # via autoflake, flake8
 pygments==2.14.0          # via rich, sphinx
-pyproject-api==1.4.0      # via tox
+pyproject-api==1.5.0      # via tox
+pyproject-hooks==1.0.0    # via build
 pyrsistent==0.19.3        # via jsonschema
-pytest==7.2.0             # via -r dev-requirements.in, pytest-cov
+pytest==7.2.1             # via -r dev-requirements.in, pytest-cov
 pytest-cov==4.0.0         # via -r dev-requirements.in
-pytz==2022.7              # via babel
+pytz==2022.7.1            # via babel
 pyupgrade==3.3.1          # via gray
 pyyaml==6.0               # via fixit, libcst, pre-commit
-requests==2.28.1          # via sphinx
+requests==2.28.2          # via sphinx
 rich==12.6.0              # via gray
 snowballstemmer==2.2.0    # via sphinx
-sphinx==6.1.2             # via -r dev-requirements.in
+sphinx==6.1.3             # via -r dev-requirements.in
-sphinxcontrib-applehelp==1.0.2  # via sphinx
+sphinxcontrib-applehelp==1.0.3  # via sphinx
 sphinxcontrib-devhelp==1.0.2  # via sphinx
 sphinxcontrib-htmlhelp==2.0.0  # via sphinx
 sphinxcontrib-jsmath==1.0.1  # via sphinx
@@ -73,13 +73,13 @@ sphinxcontrib-qthelp==1.0.3  # via sphinx
 sphinxcontrib-serializinghtml==1.1.5  # via sphinx
 tokenize-rt==5.0.0        # via add-trailing-comma, pyupgrade
 toml==0.10.2              # via autoflake
-tomli==2.0.1              # via black, build, coverage, mypy, pep517, pyproject-api, pytest, tox
+tomli==2.0.1              # via black, build, coverage, mypy, pyproject-api, pyproject-hooks, pytest, tox
-tox==4.2.6                # via -r dev-requirements.in
+tox==4.3.5                # via -r dev-requirements.in
 typing-extensions==4.4.0  # via black, libcst, mypy, typing-inspect
 typing-inspect==0.8.0     # via libcst
 unify==0.5                # via gray
 untokenize==0.1.1         # via unify
-urllib3==1.26.13          # via requests
+urllib3==1.26.14          # via requests
 virtualenv==20.17.1       # via pre-commit, tox
 wheel==0.38.4             # via pip-tools
 zipp==3.11.0              # via importlib-metadata, importlib-resources

requirements.in

@@ -36,3 +36,4 @@ rpyc
 # raspi
 cryptography==38.0.1
 shellingham==1.5.0.post1
+ratelimiter

requirements.txt

@@ -12,7 +12,7 @@ bidict==0.22.1            # via python-socketio
 bitarray==2.6.2           # via ax253, kiss3
 certifi==2022.12.7        # via requests
 cffi==1.15.1              # via cryptography
-charset-normalizer==2.1.1  # via requests
+charset-normalizer==3.0.1  # via requests
 click==8.1.3              # via -r requirements.in, click-completion, flask
 click-completion==0.5.2   # via -r requirements.in
 commonmark==0.9.1         # via rich
@@ -20,8 +20,8 @@ cryptography==38.0.1      # via -r requirements.in, pyopenssl
 dacite2==2.0.0            # via -r requirements.in
 dataclasses==0.6          # via -r requirements.in
 debtcollector==2.5.0      # via oslo-config
-dnspython==2.2.1          # via eventlet
+dnspython==2.3.0          # via eventlet
-eventlet==0.33.2          # via -r requirements.in
+eventlet==0.33.3          # via -r requirements.in
 flask==2.1.2              # via -r requirements.in, flask-classful, flask-httpauth, flask-socketio
 flask-classful==0.14.2    # via -r requirements.in
 flask-httpauth==4.7.0     # via -r requirements.in
@@ -33,11 +33,11 @@ importlib-metadata==6.0.0  # via ax253, flask, kiss3
 itsdangerous==2.1.2       # via flask
 jinja2==3.1.2             # via click-completion, flask
 kiss3==8.0.0              # via -r requirements.in
-markupsafe==2.1.1         # via jinja2
+markupsafe==2.1.2         # via jinja2
 netaddr==0.8.0            # via oslo-config
 oslo-config==9.1.0        # via -r requirements.in
 oslo-i18n==5.1.0          # via oslo-config
-pbr==5.11.0               # via -r requirements.in, oslo-i18n, stevedore
+pbr==5.11.1               # via -r requirements.in, oslo-i18n, stevedore
 pluggy==1.0.0             # via -r requirements.in
 plumbum==1.8.1            # via rpyc
 pycparser==2.21           # via cffi
@@ -47,9 +47,10 @@ pyserial==3.5             # via pyserial-asyncio
 pyserial-asyncio==0.6     # via kiss3
 python-engineio==4.3.4    # via python-socketio
 python-socketio==5.7.2    # via flask-socketio
-pytz==2022.7              # via -r requirements.in
+pytz==2022.7.1            # via -r requirements.in
 pyyaml==6.0               # via -r requirements.in, oslo-config
-requests==2.28.1          # via -r requirements.in, oslo-config, update-checker
+ratelimiter==1.2.0.post0  # via -r requirements.in
+requests==2.28.2          # via -r requirements.in, oslo-config, update-checker
 rfc3986==2.0.0            # via oslo-config
 rich==12.6.0              # via -r requirements.in
 rpyc==5.3.0               # via -r requirements.in
@@ -61,7 +62,7 @@ tabulate==0.9.0           # via -r requirements.in
 thesmuggler==1.0.1        # via -r requirements.in
 ua-parser==0.16.1         # via user-agents
 update-checker==0.18.0    # via -r requirements.in
-urllib3==1.26.13          # via requests
+urllib3==1.26.14          # via requests
 user-agents==2.2.0        # via -r requirements.in
 werkzeug==2.1.2           # via -r requirements.in, flask
 wrapt==1.14.1             # via -r requirements.in, debtcollector