openrvdas.logger.writers.udp_writer

  1#!/usr/bin/env python3
  2
  3import errno
  4import logging
  5import socket
  6import struct
  7
  8from typing import Union
  9
 10from logger.utils.das_record import DASRecord  # noqa E402
 11from logger.writers.writer import Writer  # noqa E402
 12
 13# So that we can write the user's record no matter how silly big it is, we
 14# autodetect the system's maximum datagram size and write() fragments the
 15# record into smaller packets if needed.  Each fragmented packet is marked with
 16# FRAGMENT_MARKER so that UDPReader can notice and reassemble the record.
 17FRAGMENT_MARKER = b'\xff\xffTOOBIG\xff\xff'
 18
 19# Maximum allowable size of a UDP datagram on this system, autodetect once at
 20# module load
 21#
 22# NOTE: You can test/debug the fragmentation code by loading the udp_writer
 23#       module and than manually setting udp_writer.MAXSIZE after autodetection
 24#       has happened.
 25#
 26MAXSIZE = None
 27
 28
 29def __detect_maxsize():
 30    """Autodetect the maximum datagram size we can send"""
 31    global MAXSIZE
 32    if MAXSIZE:
 33        return
 34
 35    s = socket.socket(family=socket.AF_INET, type=socket.SOCK_DGRAM,
 36                      proto=socket.IPPROTO_UDP)
 37
 38    # start with the maximum allowable size of a UDP packet, and work our way
 39    # down one byte at a time.
 40    #
 41    # FIXME: I know, I know.  Lame, slow, etc, a binary search would be "most
 42    #        correct", but I don't feel like making this go faster right now.
 43    #        And it's only done once, so whatever.
 44    #
 45    trysize = 65535
 46    while trysize > 0:
 47        logging.debug("__detect_maxsize: trying %d", trysize)
 48        try:
 49            s.sendto(b'a'*trysize, ('127.0.0.1', 9999))
 50        except OSError as e:
 51            if e.errno == errno.EMSGSIZE:
 52                trysize -= 1
 53                continue
 54            else:
 55                # For whatever reason, this won't work... print a warning
 56                logging.warning("__detect_maxsize: send() failed: %s", str(e))
 57                break
 58        # outstanding!
 59        break
 60
 61    if not trysize:
 62        logging.warning("Failed to autodetect maximum UDP datagram size. "
 63                        "Record fragmentation disabled")
 64    else:
 65        logging.info("Detected maximum UDP datagram size %d", trysize)
 66        MAXSIZE = trysize
 67
 68
 69# attempt to detect maximum datagram size when module is loaded
 70__detect_maxsize()
 71
 72
 73class UDPWriter(Writer):
 74    """Write UDP packets to network."""
 75
 76    def __init__(self, destination=None, port=None,
 77                 mc_interface=None, mc_ttl=3, num_retry=2, warning_limit=5, eol='',
 78                 reuseaddr=False, reuseport=False, **kwargs):
 79        """Write records to a UDP network socket.
 80        ```
 81        destination  The destination to send UDP packets to. If '' or None,
 82                     the UDPWriter will broadcast to 255.255.255.255.  On a
 83                     system connected to more than one subnet, you'll want to
 84                     specify the broadcast address of the network you're trying
 85                     to send to (e.g., 192.168.1.255).
 86
 87        port         Port to which packets should be sent.  REQUIRED
 88
 89        mc_interface REQUIRED for multicast, the interface to send from.  Can be
 90                     specified as either IP or a resolvable hostname.
 91
 92        mc_ttl       For multicast, how many network hops to allow.
 93
 94        num_retry    Number of times to retry if write fails. If writer exceeds
 95                     this number, it will give up on writing the message and
 96                     move on.
 97
 98        warning_limit  Number of times the writer gives up on writing a message
 99                     (without any intervening successes) before it gives up complaining
100                     about failures.
101
102        eol          If specified, an end of line string to append to record
103                     before sending.
104
105        reuseaddr    Specifies wether to set SO_REUSEADDR on the created socket.  If
106                     you don't know you need this, don't enable it.
107
108        reuseport    Specifies wether to set SO_REUSEPORT on the created socket.  If
109                     you don't know you need this, don't enable it.
110
111        encoding - 'utf-8' by default. If empty or None, do not attempt any
112                decoding and return raw bytes. Other possible encodings are
113                listed in online documentation here:
114                https://docs.python.org/3/library/codecs.html#standard-encodings
115
116        encoding_errors - 'ignore' by default. Other error strategies are
117                'strict', 'replace', and 'backslashreplace', described here:
118                https://docs.python.org/3/howto/unicode.html#encodings
119        ```
120
121        """
122        # Initialize type checking
123        super().__init__(**kwargs)  # processes 'quiet', encodings and type hints
124
125        self.num_retry = num_retry
126        self.warning_limit = warning_limit
127        self.num_warnings = 0
128        self.good_writes = 0  # consecutive good writes, for detecting UDP errors
129
130        # 'eol' comes in as a (probably escaped) string. We need to
131        # unescape it, which means converting to bytes and back.
132        if eol is not None and self.encoding:
133            eol = self._unescape_str(eol)
134        self.eol = eol
135
136        self.target_str = 'destination: %s, port: %d' % (destination, port)
137
138        # do name resolution once in the constructor
139        #
140        # NOTE: This means the hostname must be valid when we start, otherwise
141        #       the config_check code will puke.  That's fine.  The alternative
142        #       is we let name resolution happen while we're running, but then
143        #       each failed lookup is going to block our write() routine for a
144        #       few seconds - not good.
145        #
146        # NOTE: This also catches specifying impropperly formatted IP
147        #       addresses.  The only way through gethostbyname() w/out throwing
148        #       an exception is to provide a valid hostname or IP address.
149        #       Propperly formatted IPs just get returned.
150        #
151        if destination:
152            destination = socket.gethostbyname(destination)
153        else:
154            # If no destination, it's a broadcast; set dest to special string
155            destination = '<broadcast>'
156
157        self.destination = destination
158
159        # make sure user passed in `port`
160        #
161        # NOTE: We want the order of the arguments to consistently be (ip,
162        #       port, ...) across all the network readers/writers... but we
163        #       want `destination` to be optional.  All kwargs need to come
164        #       after all regular args, so we've assigned a default value of
165        #       None to `port`.  But don't be confused, it is REQUIRED.
166        #
167        if not port:
168            raise TypeError('must specify `port`')
169        # make sure port gets stored as an int, even if passed in as a string
170        self.port = int(port)
171
172        # multicast options
173        if mc_interface:
174            # resolve once in constructor
175            mc_interface = socket.gethostbyname(mc_interface)
176        self.mc_interface = mc_interface
177        self.mc_ttl = mc_ttl
178
179        self.reuseaddr = reuseaddr
180        self.reuseport = reuseport
181
182        # socket gets initialized on-demand in write()
183        self.socket = None
184
185    ############################
186    def __del__(self):
187        if self.socket:
188            self.socket.close()
189
190    ############################
191    def _open_socket(self):
192        """Do socket prep so we're ready to write().  Returns socket object or None on
193        failure.
194        """
195        udp_socket = socket.socket(family=socket.AF_INET,
196                                   type=socket.SOCK_DGRAM,
197                                   proto=socket.IPPROTO_UDP)
198        if self.reuseaddr:
199            udp_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, True)
200        if self.reuseport:
201            try:  # Raspbian doesn't recognize SO_REUSEPORT
202                udp_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEPORT, True)
203            except AttributeError:
204                logging.warning('Unable to set socket REUSEPORT; may be unsupported')
205
206        # set multicast/broadcast options
207        if self.mc_interface:
208            # set the time-to-live for messages
209            udp_socket.setsockopt(socket.IPPROTO_IP, socket.IP_MULTICAST_TTL,
210                                  struct.pack('b', self.mc_ttl))
211            # set outgoing multicast interface
212            udp_socket.setsockopt(socket.IPPROTO_IP, socket.IP_MULTICAST_IF,
213                                  socket.inet_aton(self.mc_interface))
214        else:
215            # maybe broadcast, but very non-trivial to detect broadcast IP, so
216            # we set the broadcast flag anytime we're not doing multicast
217            udp_socket.setsockopt(socket.SOL_SOCKET, socket.SO_BROADCAST, True)
218
219        try:
220            udp_socket.connect((self.destination, self.port))
221            return udp_socket
222        except OSError as e:
223            logging.error('Unable to connect to %s:%d - %s', self.destination, self.port, e)
224            return None
225
226    ############################
227    def write(self, record: Union[str, bytes, DASRecord]):
228        """Write the record to the network."""
229
230        # See if it's something we can process, and if not, try digesting
231        if not self.can_process_record(record):  # inherited from BaseModule()
232            self.digest_record(record)  # inherited from BaseModule()
233            return
234
235        if isinstance(record, DASRecord):
236            record = record.as_json()
237
238        # Append eol if configured
239        if self.eol:
240            record += self.eol
241
242        # Encode the record, so we're dealing with bytes from here on out
243        record = self._encode_str(record)
244
245        # Fragment record if needed, and recurse.
246        if len(record) > MAXSIZE:
247            record_list = []
248            max_fragment_size = MAXSIZE - len(FRAGMENT_MARKER)
249            fragment_sizes = []
250            while len(record) > max_fragment_size:
251                r = record[:max_fragment_size]+FRAGMENT_MARKER
252                record_list.append(r)
253                fragment_sizes.append(str(len(r)))
254                record = record[max_fragment_size:]
255            # last record doesn't get FRAGMENT_MARKER
256            record_list.append(record)
257            fragment_sizes.append("{} bytes".format(len(record)))
258            fragment_sizes = ', '.join(fragment_sizes)
259            logging.info("write: fragmented record into %d datagrams: %s",
260                         len(record_list), fragment_sizes)
261            logging.debug(str(record_list))
262
263            # change our encoding to binary temporarily, because we've already
264            # encoded to binary and added our marker (which has non-utf chars
265            # in it)
266            old_encoding = self.encoding
267            self.encoding = None
268            self.write(record_list)
269            # restore old encoding
270            self.encoding = old_encoding
271            return
272
273        # If socket isn't connected, try reconnecting. If we can't
274        # reconnect, complain and return without writing.
275        if not self.socket:
276            self.socket = self._open_socket()
277        if not self.socket:
278            logging.error('Unable to write record to %s:%d',
279                          self.destination, self.port)
280            return
281
282        num_tries = bytes_sent = 0
283        rec_len = len(record)
284        while num_tries <= self.num_retry and bytes_sent < rec_len:
285            try:
286                bytes_sent = self.socket.send(record)
287
288                # If here, write succeeded. Reset warnings
289                #
290                # NOTE: If the host is unreachable, every other send will fail.
291                #       Since UDP doesn't actually know it failed, the initial
292                #       send() cannot fail.  However, the network stack will
293                #       see the ICMP host unreachable message and will store
294                #       THAT as the the error message for next write, then the
295                #       next send fails and clears the error...  Then the next
296                #       "succeeds" and the next fails, etc, etc
297                #
298                #       So we look for 2 consecutive "successful" writes before
299                #       resetting num_warnings.
300                #
301                self.good_writes += 1
302                if self.good_writes >= 2:
303                    if self.num_warnings == self.warning_limit:
304                        logging.info('UDPWriter.write() succeeded in writing after series of '
305                                     'failures; resetting warnings.')
306                    self.num_warnings = 0  # we've succeeded
307
308            except (OSError, ConnectionRefusedError) as e:
309                # If we failed, complain, unless we've already complained too much
310                self.good_writes = 0
311                if self.num_warnings < self.warning_limit:
312                    logging.error(f'UDPWriter: send() error: {self.target_str}: {str(e)}')
313                    if 'Message too long' in str(e):
314                        logging.error(f'Message length is {rec_len}')
315                    self.num_warnings += 1
316                    if self.num_warnings == self.warning_limit:
317                        logging.error('UDPWriter.write() - muting errors')
318            num_tries += 1
319
320        logging.debug('UDPWriter.write() wrote %d/%d bytes after %d tries',
321                      bytes_sent, rec_len, num_tries)