Handle EAGAIN socket error when dropping packets (#515)

datadog/dogstatsd/base.py

@@ -11,6 +11,7 @@
 import logging
 import os
 import socket
+import errno
 from threading import Lock
 
 # datadog
@@ -398,9 +399,15 @@ def _send_to_server(self, packet):
         except socket.timeout:
             # dogstatsd is overflowing, drop the packets (mimicks the UDP behaviour)
             pass
-        except (socket.error, socket.herror, socket.gaierror) as se:
+        except (socket.herror, socket.gaierror) as se:
             log.warning("Error submitting packet: {}, dropping the packet and closing the socket".format(se))
             self.close_socket()
+        except socket.error as se:
+            if se.errno == errno.EAGAIN:
+                log.warning("Socket send would block: {}, dropping the packet".format(se))
+            else:
+                log.warning("Error submitting packet: {}, dropping the packet and closing the socket".format(se))
+                self.close_socket()
         except Exception as e:
             log.error("Unexpected error: %s", str(e))
 

tests/unit/dogstatsd/test_statsd.py

@@ -10,6 +10,7 @@
 from collections import deque
 import os
 import socket
+import errno
 import time
 import unittest
 
@@ -66,6 +67,14 @@ def send(self, payload):
         raise socket.error("Socket error")
 
 
+class OverflownSocket(FakeSocket):
+
+    def send(self, payload):
+        error = socker.error("Socker error")
+        error.errno = errno.EAGAIN
+        raise error
+
+
 def telemetry_metrics(metrics=1, events=0, service_checks=0, bytes_sent=0, bytes_dropped=0, packets_sent=0, packets_dropped=0, transport="udp", tags="", namespace=""):
     version = get_version()
     if tags:
@@ -339,6 +348,11 @@ def test_socket_error(self):
         self.statsd.gauge('no error', 1)
         assert True, 'success'
 
+    def test_socket_overflown(self):
+        self.statsd.socket = OverflownSocket()
+        self.statsd.gauge('no error', 1)
+        assert True, 'success'
+
     def test_timed(self):
         """
         Measure the distribution of a function's run time.