Add: application performance monitoring via Honeycomb (#20)

If you don't have "honeycomb-beeline" package installed, it doesn't
do anything.
If you do have it installed, it ensure all important functions
are traced.

By default, this library creates a new trace for each new incoming
packet. For all other cases it is up to the application to make
good traces out of this.

Author: TrueBrain

openttd_protocol/protocol/content.py

@@ -2,6 +2,7 @@
 import logging
 import struct
 
+from .. import tracer
 from ..wire.exceptions import PacketInvalidData
 from ..wire.read import (
     read_string,
@@ -71,6 +72,7 @@ class ContentProtocol(TCPProtocol):
     PACKET_END = PacketContentType.PACKET_CONTENT_END
 
     @staticmethod
+    @tracer.traced("content")
     def receive_PACKET_CONTENT_CLIENT_INFO_LIST(source, data):
         content_type, data = read_uint8(data)
         openttd_version, data = read_uint32(data)
@@ -141,6 +143,7 @@ def _receive_client_info(data, count, has_content_id=False, has_content_type_and
         return content_infos, data
 
     @classmethod
+    @tracer.traced("content")
     def receive_PACKET_CONTENT_CLIENT_INFO_ID(cls, source, data):
         count, data = read_uint16(data)
 
@@ -152,6 +155,7 @@ def receive_PACKET_CONTENT_CLIENT_INFO_ID(cls, source, data):
         return {"content_infos": content_infos}
 
     @classmethod
+    @tracer.traced("content")
     def receive_PACKET_CONTENT_CLIENT_INFO_EXTID(cls, source, data):
         count, data = read_uint8(data)
 
@@ -163,6 +167,7 @@ def receive_PACKET_CONTENT_CLIENT_INFO_EXTID(cls, source, data):
         return {"content_infos": content_infos}
 
     @classmethod
+    @tracer.traced("content")
     def receive_PACKET_CONTENT_CLIENT_INFO_EXTID_MD5(cls, source, data):
         count, data = read_uint8(data)
 
@@ -176,6 +181,7 @@ def receive_PACKET_CONTENT_CLIENT_INFO_EXTID_MD5(cls, source, data):
         return {"content_infos": content_infos}
 
     @classmethod
+    @tracer.traced("content")
     def receive_PACKET_CONTENT_CLIENT_CONTENT(cls, source, data):
         count, data = read_uint16(data)
 
@@ -186,6 +192,7 @@ def receive_PACKET_CONTENT_CLIENT_CONTENT(cls, source, data):
 
         return {"content_infos": content_infos}
 
+    @tracer.traced("content")
     async def send_PACKET_CONTENT_SERVER_INFO(
         self, content_type, content_id, filesize, name, version, url, description, unique_id, md5sum, dependencies, tags
     ):
@@ -227,6 +234,7 @@ async def send_PACKET_CONTENT_SERVER_INFO(
         write_presend(data, SEND_TCP_COMPAT_MTU)
         await self.send_packet(data)
 
+    @tracer.traced("content")
     async def send_PACKET_CONTENT_SERVER_CONTENT(self, content_type, content_id, filesize, filename, stream):
         # First, send a packet to tell the client it will be receiving a file
         data = write_init(PacketContentType.PACKET_CONTENT_SERVER_CONTENT)

openttd_protocol/protocol/coordinator.py

@@ -1,6 +1,7 @@
 import enum
 import logging
 
+from .. import tracer
 from ..wire.exceptions import PacketInvalidData
 from ..wire.read import (
     read_bytes,
@@ -94,6 +95,7 @@ class CoordinatorProtocol(TCPProtocol):
     PACKET_END = PacketCoordinatorType.PACKET_COORDINATOR_END
 
     @staticmethod
+    @tracer.traced("coordinator")
     def receive_PACKET_COORDINATOR_SERVER_REGISTER(source, data):
         protocol_version, data = read_uint8(data)
 
@@ -128,6 +130,7 @@ def receive_PACKET_COORDINATOR_SERVER_REGISTER(source, data):
         }
 
     @staticmethod
+    @tracer.traced("coordinator")
     def receive_PACKET_COORDINATOR_SERVER_UPDATE(source, data):
         protocol_version, data = read_uint8(data)
 
@@ -240,6 +243,7 @@ def receive_PACKET_COORDINATOR_SERVER_UPDATE(source, data):
         }
 
     @staticmethod
+    @tracer.traced("coordinator")
     def receive_PACKET_COORDINATOR_CLIENT_LISTING(source, data):
         protocol_version, data = read_uint8(data)
 
@@ -268,6 +272,7 @@ def receive_PACKET_COORDINATOR_CLIENT_LISTING(source, data):
         }
 
     @staticmethod
+    @tracer.traced("coordinator")
     def receive_PACKET_COORDINATOR_CLIENT_CONNECT(source, data):
         protocol_version, data = read_uint8(data)
 
@@ -285,6 +290,7 @@ def receive_PACKET_COORDINATOR_CLIENT_CONNECT(source, data):
         }
 
     @staticmethod
+    @tracer.traced("coordinator")
     def receive_PACKET_COORDINATOR_SERCLI_CONNECT_FAILED(source, data):
         protocol_version, data = read_uint8(data)
 
@@ -304,6 +310,7 @@ def receive_PACKET_COORDINATOR_SERCLI_CONNECT_FAILED(source, data):
         }
 
     @staticmethod
+    @tracer.traced("coordinator")
     def receive_PACKET_COORDINATOR_CLIENT_CONNECTED(source, data):
         protocol_version, data = read_uint8(data)
 
@@ -321,6 +328,7 @@ def receive_PACKET_COORDINATOR_CLIENT_CONNECTED(source, data):
         }
 
     @staticmethod
+    @tracer.traced("coordinator")
     def receive_PACKET_COORDINATOR_SERCLI_STUN_RESULT(source, data):
         protocol_version, data = read_uint8(data)
 
@@ -341,6 +349,7 @@ def receive_PACKET_COORDINATOR_SERCLI_STUN_RESULT(source, data):
             "result": result,
         }
 
+    @tracer.traced("coordinator")
     async def send_PACKET_COORDINATOR_GC_ERROR(self, protocol_version, error_no, error_detail):
         data = write_init(PacketCoordinatorType.PACKET_COORDINATOR_GC_ERROR)
 
@@ -358,6 +367,7 @@ async def send_PACKET_COORDINATOR_GC_ERROR(self, protocol_version, error_no, err
         write_presend(data, SEND_TCP_MTU)
         await self.send_packet(data)
 
+    @tracer.traced("coordinator")
     async def send_PACKET_COORDINATOR_GC_REGISTER_ACK(
         self, protocol_version, connection_type, invite_code, invite_code_secret
     ):
@@ -396,6 +406,7 @@ def _fill_NEWGRF_LOOKUP_PACKET(self, newgrf_lookup_table_cursor, newgrf_lookup_t
         if count != 0:
             yield count, data
 
+    @tracer.traced("coordinator")
     async def send_PACKET_COORDINATOR_GC_NEWGRF_LOOKUP(
         self, protocol_version, newgrf_lookup_table_cursor, newgrf_lookup_table
     ):
@@ -413,6 +424,7 @@ async def send_PACKET_COORDINATOR_GC_NEWGRF_LOOKUP(
             write_presend(data, SEND_TCP_MTU)
             await self.send_packet(data)
 
+    @tracer.traced("coordinator")
     async def send_PACKET_COORDINATOR_GC_LISTING(
         self, protocol_version, game_info_version, servers, newgrf_lookup_table
     ):
@@ -491,6 +503,7 @@ async def send_PACKET_COORDINATOR_GC_LISTING(
         write_presend(data, SEND_TCP_MTU)
         await self.send_packet(data)
 
+    @tracer.traced("coordinator")
     async def send_PACKET_COORDINATOR_GC_CONNECTING(self, protocol_version, token, invite_code):
         data = write_init(PacketCoordinatorType.PACKET_COORDINATOR_GC_CONNECTING)
 
@@ -500,6 +513,7 @@ async def send_PACKET_COORDINATOR_GC_CONNECTING(self, protocol_version, token, i
         write_presend(data, SEND_TCP_MTU)
         await self.send_packet(data)
 
+    @tracer.traced("coordinator")
     async def send_PACKET_COORDINATOR_GC_CONNECT_FAILED(self, protocol_version, token):
         data = write_init(PacketCoordinatorType.PACKET_COORDINATOR_GC_CONNECT_FAILED)
 
@@ -508,6 +522,7 @@ async def send_PACKET_COORDINATOR_GC_CONNECT_FAILED(self, protocol_version, toke
         write_presend(data, SEND_TCP_MTU)
         await self.send_packet(data)
 
+    @tracer.traced("coordinator")
     async def send_PACKET_COORDINATOR_GC_DIRECT_CONNECT(self, protocol_version, token, tracking_number, hostname, port):
         data = write_init(PacketCoordinatorType.PACKET_COORDINATOR_GC_DIRECT_CONNECT)
 
@@ -519,6 +534,7 @@ async def send_PACKET_COORDINATOR_GC_DIRECT_CONNECT(self, protocol_version, toke
         write_presend(data, SEND_TCP_MTU)
         await self.send_packet(data)
 
+    @tracer.traced("coordinator")
     async def send_PACKET_COORDINATOR_GC_STUN_REQUEST(self, protocol_version, token):
         data = write_init(PacketCoordinatorType.PACKET_COORDINATOR_GC_STUN_REQUEST)
 
@@ -527,6 +543,7 @@ async def send_PACKET_COORDINATOR_GC_STUN_REQUEST(self, protocol_version, token)
         write_presend(data, SEND_TCP_MTU)
         await self.send_packet(data)
 
+    @tracer.traced("coordinator")
     async def send_PACKET_COORDINATOR_GC_STUN_CONNECT(
         self, protocol_version, token, tracking_number, interface_number, hostname, port
     ):
@@ -541,6 +558,7 @@ async def send_PACKET_COORDINATOR_GC_STUN_CONNECT(
         write_presend(data, SEND_TCP_MTU)
         await self.send_packet(data)
 
+    @tracer.traced("coordinator")
     async def send_PACKET_COORDINATOR_GC_TURN_CONNECT(
         self, protocol_version, token, tracking_number, ticket, connection_string
     ):

openttd_protocol/protocol/game.py

@@ -1,6 +1,7 @@
 import enum
 import logging
 
+from .. import tracer
 from ..wire.exceptions import PacketInvalidData
 from ..wire.read import (
     read_bytes,
@@ -53,6 +54,7 @@ class GameProtocol(TCPProtocol):
     PACKET_END = PacketGameType.PACKET_END
 
     @staticmethod
+    @tracer.traced("game")
     def receive_PACKET_SERVER_GAME_INFO(source, data):
         game_info_version, data = read_uint8(data)
 
@@ -158,12 +160,14 @@ def receive_PACKET_SERVER_GAME_INFO(source, data):
         }
 
     @staticmethod
+    @tracer.traced("game")
     def receive_PACKET_SERVER_SHUTDOWN(source, data):
         if len(data) != 0:
             raise PacketInvalidData("more bytes than expected in SERVER_SHUTDOWN; remaining: ", len(data))
 
         return {}
 
+    @tracer.traced("game")
     async def send_PACKET_CLIENT_GAME_INFO(self):
         data = write_init(PacketGameType.PACKET_CLIENT_GAME_INFO)
         write_presend(data, SEND_TCP_MTU)

openttd_protocol/protocol/stun.py

@@ -1,6 +1,7 @@
 import enum
 import logging
 
+from .. import tracer
 from ..wire.exceptions import PacketInvalidData
 from ..wire.read import (
     read_string,
@@ -21,6 +22,7 @@ class StunProtocol(TCPProtocol):
     PACKET_END = PacketStunType.PACKET_STUN_END
 
     @staticmethod
+    @tracer.traced("stun")
     def receive_PACKET_STUN_SERCLI_STUN(source, data):
         protocol_version, data = read_uint8(data)
 

openttd_protocol/protocol/turn.py

@@ -1,6 +1,7 @@
 import enum
 import logging
 
+from .. import tracer
 from ..wire.exceptions import PacketInvalidData
 from ..wire.read import (
     read_string,
@@ -29,6 +30,7 @@ class TurnProtocol(TCPProtocol):
     PACKET_END = PacketTurnType.PACKET_TURN_END
 
     @staticmethod
+    @tracer.traced("turn")
     def receive_PACKET_TURN_SERCLI_CONNECT(source, data):
         protocol_version, data = read_uint8(data)
 
@@ -42,6 +44,7 @@ def receive_PACKET_TURN_SERCLI_CONNECT(source, data):
 
         return {"protocol_version": protocol_version, "ticket": ticket}
 
+    @tracer.traced("turn")
     async def send_PACKET_TURN_TURN_CONNECTED(self, protocol_version, hostname):
         data = write_init(PacketTurnType.PACKET_TURN_TURN_CONNECTED)
 

openttd_protocol/tracer.py

@@ -0,0 +1,38 @@
+try:
+    import beeline
+
+    def traced(prefix):
+        # The common case is to use the function-name, so automate this.
+        def wrapper(func):
+            return beeline.traced_impl(beeline.tracer, f"{prefix}.{func.__name__}", None, None)(func)
+
+        return wrapper
+
+    tracer = beeline.tracer
+    untraced = beeline.untraced
+    add_context = beeline.add_context
+
+except ImportError:
+    # Honeycomb Beeline package is not installed. Mock the tracer functions.
+
+    def traced(prefix):
+        def wrapper(func):
+            return func
+
+        return wrapper
+
+    class tracer:
+        def __init__(self, name):
+            pass
+
+        def __enter__(self):
+            pass
+
+        def __exit__(self, type, value, traceback):
+            pass
+
+    def untraced(func):
+        return func
+
+    def add_context(payload):
+        pass

openttd_protocol/wire/tcp.py

@@ -14,6 +14,7 @@
     read_uint16,
 )
 from .source import Source
+from .. import tracer
 
 log = logging.getLogger(__name__)
 
@@ -173,14 +174,19 @@ async def _process_queue(self):
             if await self._callback.receive_raw(self.source, data):
                 return
 
-        try:
-            packet_type, kwargs = self.receive_packet(self.source, data)
-        except PacketInvalid as err:
-            log.info("Dropping invalid packet from %s:%d: %r", self.source.ip, self.source.port, err)
-            raise SocketClosed
+        with tracer.tracer("tcp.receive-packet"):
+            try:
+                packet_type, kwargs = self.receive_packet(self.source, data)
+            except PacketInvalid as err:
+                log.info("Dropping invalid packet from %s:%d: %r", self.source.ip, self.source.port, err)
+                raise SocketClosed
+
+            tracer.add_context({"command": f"receive.{packet_type.name}"})
 
-        await getattr(self._callback, f"receive_{packet_type.name}")(self.source, **kwargs)
+            with tracer.tracer("tcp.packet-handler"):
+                await getattr(self._callback, f"receive_{packet_type.name}")(self.source, **kwargs)
 
+    @tracer.traced("tcp")
     def receive_packet(self, source, data):
         # Check length of packet
         length, data = read_uint16(data)
@@ -202,6 +208,7 @@ def receive_packet(self, source, data):
         kwargs = func(source, data)
         return type, kwargs
 
+    @tracer.traced("tcp")
     async def send_packet(self, data):
         await self._can_write.wait()