Merge lp:~tmm1/graphite/udp-receiver into lp:~graphite-dev/graphite/main

[Aman Gupta]

[lp:~tmm1/graphite/udp-receiver]

434. By Aman Gupta on 2011-07-21

Only show remote hostname on invalid datagrams

[chrismd (chrismd)]

tmm1 this change looks good except I think the datagram receiver should allow for multiple lines. This will make the data format the same as the current tcp port 2003 receiver, and also the udp listener should use udp port 2003 by default as well since it will use the data format.

review: Needs Fixing

[Aman Gupta (tmm1)]

Good call on both points.

Is it worth having UDP_RECEIVER_* configuration, or should I re-use LINE_RECEIVER_* for both TCP/UDP. I suppose it could be useful to listen for datagrams on a separate ip/port.

[lp:~tmm1/graphite/udp-receiver]

435. By Aman Gupta on 2011-07-21

Listen for datagrams on the same port as the line receiver by default

436. By Aman Gupta on 2011-07-21

UDP Reciever can accept multiple lines per datagram

=== modified file 'carbon/conf/carbon.amqp.conf.example'
--- carbon/conf/carbon.amqp.conf.example	2011-04-01 06:01:35 +0000
+++ carbon/conf/carbon.amqp.conf.example	2011-07-21 03:08:32 +0000
@@ -32,6 +32,9 @@
 LINE_RECEIVER_INTERFACE = 0.0.0.0
 LINE_RECEIVER_PORT = 2003
 
+UDP_RECEIVER_INTERFACE = 0.0.0.0
+UDP_RECEIVER_PORT = 2003
+
 PICKLE_RECEIVER_INTERFACE = 0.0.0.0
 PICKLE_RECEIVER_PORT = 2004
 

=== modified file 'carbon/conf/carbon.conf.example'
--- carbon/conf/carbon.conf.example	2011-04-05 21:42:51 +0000
+++ carbon/conf/carbon.conf.example	2011-07-21 03:08:32 +0000
@@ -30,6 +30,9 @@
 LINE_RECEIVER_INTERFACE = 0.0.0.0
 LINE_RECEIVER_PORT = 2003
 
+UDP_RECEIVER_INTERFACE = 0.0.0.0
+UDP_RECEIVER_PORT = 2003
+
 PICKLE_RECEIVER_INTERFACE = 0.0.0.0
 PICKLE_RECEIVER_PORT = 2004
 

=== modified file 'carbon/lib/carbon/conf.py'
--- carbon/lib/carbon/conf.py	2011-07-20 23:15:06 +0000
+++ carbon/lib/carbon/conf.py	2011-07-21 03:08:32 +0000
@@ -35,6 +35,8 @@
   MAX_CREATES_PER_MINUTE='inf',
   LINE_RECEIVER_INTERFACE='0.0.0.0',
   LINE_RECEIVER_PORT=2003,
+  UDP_RECEIVER_INTERFACE='0.0.0.0',
+  UDP_RECEIVER_PORT=2003,
   PICKLE_RECEIVER_INTERFACE='0.0.0.0',
   PICKLE_RECEIVER_PORT=2004,
   CACHE_QUERY_INTERFACE='0.0.0.0',

=== modified file 'carbon/lib/carbon/listeners.py'
--- carbon/lib/carbon/listeners.py	2010-06-06 16:24:30 +0000
+++ carbon/lib/carbon/listeners.py	2011-07-21 03:08:32 +0000
@@ -1,5 +1,5 @@
 from twisted.internet import reactor
-from twisted.internet.protocol import Factory
+from twisted.internet.protocol import Factory, DatagramProtocol
 from twisted.internet.error import ConnectionDone
 from twisted.protocols.basic import LineOnlyReceiver, Int32StringReceiver
 from carbon.cache import MetricCache
@@ -42,6 +42,19 @@
     metricReceived(metric, datapoint)
 
 
+class MetricDatagramReceiver(LoggingMixin, DatagramProtocol):
+  def datagramReceived(self, data, (host, port)):
+    for line in data.splitlines():
+      try:
+        metric, value, timestamp = line.strip().split()
+        datapoint = ( float(timestamp), float(value) )
+
+        increment('metricsReceived')
+        metricReceived(metric, datapoint)
+      except:
+        log.listener('invalid line received from client %s, ignoring' % host)
+
+
 class MetricPickleReceiver(LoggingMixin, Int32StringReceiver):
   MAX_LENGTH = 2 ** 20
 

=== modified file 'carbon/lib/carbon/service.py'
--- carbon/lib/carbon/service.py	2011-07-08 20:25:00 +0000
+++ carbon/lib/carbon/service.py	2011-07-21 03:08:32 +0000
@@ -16,13 +16,13 @@
 from os.path import exists
 
 from twisted.application.service import MultiService
-from twisted.application.internet import TCPServer, TCPClient
+from twisted.application.internet import TCPServer, TCPClient, UDPServer
 from twisted.internet.protocol import ServerFactory
 
 
 def createBaseService(config):
     from carbon.conf import settings
-    from carbon.listeners import MetricLineReceiver, MetricPickleReceiver
+    from carbon.listeners import MetricLineReceiver, MetricPickleReceiver, MetricDatagramReceiver
 
     root_service = MultiService()
     root_service.setName(settings.program)
@@ -52,6 +52,11 @@
         service = TCPServer(int(port), factory, interface=interface)
         service.setServiceParent(root_service)
 
+    service = UDPServer(int(settings.UDP_RECEIVER_PORT),
+                        MetricDatagramReceiver(),
+                        interface=settings.UDP_RECEIVER_INTERFACE)
+    service.setServiceParent(root_service)
+
     if use_amqp:
         factory = amqp_listener.createAMQPListener(
             amqp_user, amqp_password,