Merge lp:~drawks/graphite/graphite-msgpack into lp:~graphite-dev/graphite/main

[Dave Rawks]

Added msgpack as a receiver protocol. msgpack is super fast at deserializing and much safer than pickle since it doesn't support serializing things like modules, functions, or class objects. Also added support to enable/disable individual listeners in carbon.conf

New code path can be tested with a simple metric writer like:

import msgpack, socket, time:
s=socket.create_connection(('localhost',2005))
fs=s.makefile()
for second in xrange(int(time.time()) - 86400,int(time.time())):
    msgpack.pack(("foo.bar",(second,1)),fs)
fs.flush()
fs.close()
s.shutdown(socket.SHUT_RDWR)

[Sidnei da Silva (sidnei)]

1. Please move the import of Unpacker from msgpack from module-level to inside connectionMade().
2. (as said on IRC), note that setting the port to 0 effectively disables it, so the 3 new extra settings are not needed. If you do think it's better/more explicit to keep them, then might make sense to remove the 'if port:' from service.py and require setting the ENABLE_* settings instead.

review: Approve

[Dave Rawks (drawks)]

1. Delayed imports are generally frowned upon as you can have a situation where someone setups up the listener and then it doesn't break until the first connection to it is made. Also, if the import is inside connectionMade then this will result in reimport on every new connection to the listener, this seems less than optimal AND doesn't follow the same code conventions used in the pickle listener.

2. I do this that making enabling/disabling explicit is more clear from a managment point of view. Also I can imagine a world where we have more/modular listeners and not all of them will follow the convention of listening on a tcp/udp port. Just for argument sake, say that there is a listener that eats metrics from a serial port... OTOH removing the check for sane non-zero port numbers in the current code would lead to a situation where we'd have an exception deeper in the code when the bogus value is passed into the actual twisted protocol code.

[lp:~drawks/graphite/graphite-msgpack]

688. By Dave Rawks on 2012-02-15

Disable msgpack receiver by default

689. By Dave Rawks on 2012-02-15

changed shebang line to use python from env instead of harcoded interpreter

Unmerged revisions

689. By Dave Rawks on 2012-02-15

changed shebang line to use python from env instead of harcoded interpreter

688. By Dave Rawks on 2012-02-15

Disable msgpack receiver by default

687. By Dave Rawks <email address hidden> on 2012-02-14

Added config options to enable/disable line,pickle, and msgpack receivers.
made msgpack import conditional on the msgpack receiver being enabled
updated check_dependencies to test for msgpack

686. By Dave Rawks <email address hidden> on 2012-02-14

Added msgpack protocol linereceiver

=== modified file 'carbon/bin/validate-storage-schemas.py'
--- carbon/bin/validate-storage-schemas.py	2011-09-11 06:43:44 +0000
+++ carbon/bin/validate-storage-schemas.py	2012-02-15 23:21:21 +0000
@@ -1,4 +1,4 @@
-#!/usr/bin/python
+#!/usr/bin/env python
 """Copyright 2009 Chris Davis
 
 Licensed under the Apache License, Version 2.0 (the "License");

=== modified file 'carbon/conf/carbon.conf.example'
--- carbon/conf/carbon.conf.example	2011-12-14 16:22:17 +0000
+++ carbon/conf/carbon.conf.example	2012-02-15 23:21:21 +0000
@@ -56,6 +56,7 @@
 # the files quickly but at the risk of slowing I/O down considerably for a while.
 MAX_CREATES_PER_MINUTE = 50
 
+ENABLE_LINE_RECEIVER = True
 LINE_RECEIVER_INTERFACE = 0.0.0.0
 LINE_RECEIVER_PORT = 2003
 
@@ -66,6 +67,7 @@
 UDP_RECEIVER_INTERFACE = 0.0.0.0
 UDP_RECEIVER_PORT = 2003
 
+ENABLE_PICKLE_RECEIVER = True
 PICKLE_RECEIVER_INTERFACE = 0.0.0.0
 PICKLE_RECEIVER_PORT = 2004
 
@@ -74,6 +76,10 @@
 # Set this to True to revert to the old-fashioned insecure unpickler.
 USE_INSECURE_UNPICKLER = False
 
+ENABLE_MSGPACK_RECEIVER = False
+MSGPACK_RECEIVER_INTERFACE = 0.0.0.0
+MSGPACK_RECEIVER_PORT = 2005
+
 CACHE_QUERY_INTERFACE = 0.0.0.0
 CACHE_QUERY_PORT = 7002
 

=== modified file 'carbon/lib/carbon/conf.py'
--- carbon/lib/carbon/conf.py	2012-02-09 21:38:55 +0000
+++ carbon/lib/carbon/conf.py	2012-02-15 23:21:21 +0000
@@ -32,11 +32,16 @@
   MAX_CACHE_SIZE=float('inf'),
   MAX_UPDATES_PER_SECOND=500,
   MAX_CREATES_PER_MINUTE=float('inf'),
+  ENABLE_LINE_RECEIVER=True,
   LINE_RECEIVER_INTERFACE='0.0.0.0',
   LINE_RECEIVER_PORT=2003,
   ENABLE_UDP_LISTENER=False,
   UDP_RECEIVER_INTERFACE='0.0.0.0',
   UDP_RECEIVER_PORT=2003,
+  ENABLE_MSGPACK_RECEIVER=False,
+  MSGPACK_RECEIVER_INTERFACE='0.0.0.0',
+  MSGPACK_RECEIVER_PORT=2005,
+  ENABLE_PICKLE_RECEIVER=True,
   PICKLE_RECEIVER_INTERFACE='0.0.0.0',
   PICKLE_RECEIVER_PORT=2004,
   CACHE_QUERY_INTERFACE='0.0.0.0',

=== modified file 'carbon/lib/carbon/protocols.py'
--- carbon/lib/carbon/protocols.py	2012-02-11 06:26:28 +0000
+++ carbon/lib/carbon/protocols.py	2012-02-15 23:21:21 +0000
@@ -1,11 +1,13 @@
 from twisted.internet import reactor
-from twisted.internet.protocol import DatagramProtocol
+from twisted.internet.protocol import DatagramProtocol, Protocol
 from twisted.internet.error import ConnectionDone
 from twisted.protocols.basic import LineOnlyReceiver, Int32StringReceiver
 from carbon import log, events, state, management
 from carbon.conf import settings
 from carbon.regexlist import WhiteList, BlackList
 from carbon.util import pickle, get_unpickler
+if settings.ENABLE_MSGPACK_RECEIVER:
+  from msgpack import Unpacker
 
 
 class MetricReceiver:
@@ -106,6 +108,33 @@
       self.metricReceived(metric, datapoint)
 
 
+class MetricMessagePackReceiver(MetricReceiver, Protocol):
+
+  def connectionMade(self):
+    MetricReceiver.connectionMade(self)
+    self.unpacker = Unpacker()
+
+  def dataReceived(self, data):
+    if len(data) <= 0:
+      log.listener('msgpack short read from %s' % self.peerName)
+      return
+    try:
+      self.unpacker.feed(data)
+      for (metric, datapoint) in self.unpacker:
+        if not isinstance(metric,str):
+          log.listener('invalid metric name/type %r/%r received from %s' % ( metric, type(metric), self.peerName))
+          continue
+        try:
+          datapoint = ( float(datapoint[0]), float(datapoint[1]) )
+        except:
+          continue
+        self.metricReceived(metric, datapoint)
+    except:
+      log.listener('invalid msgpack received from %s, ignoring' % self.peerName)
+      return
+
+
+
 class CacheManagementHandler(Int32StringReceiver):
   def connectionMade(self):
     peer = self.transport.getPeer()

=== modified file 'carbon/lib/carbon/service.py'
--- carbon/lib/carbon/service.py	2011-12-14 16:22:17 +0000
+++ carbon/lib/carbon/service.py	2012-02-15 23:21:21 +0000
@@ -40,7 +40,7 @@
 def createBaseService(config):
     from carbon.conf import settings
     from carbon.protocols import (MetricLineReceiver, MetricPickleReceiver,
-                                  MetricDatagramReceiver)
+                                  MetricDatagramReceiver, MetricMessagePackReceiver)
 
     root_service = CarbonRootService()
     root_service.setName(settings.program)
@@ -59,12 +59,20 @@
         amqp_exchange_name = settings.get("AMQP_EXCHANGE", "graphite")
 
 
-    for interface, port, protocol in ((settings.LINE_RECEIVER_INTERFACE,
-                                       settings.LINE_RECEIVER_PORT,
-                                       MetricLineReceiver),
-                                      (settings.PICKLE_RECEIVER_INTERFACE,
-                                       settings.PICKLE_RECEIVER_PORT,
-                                       MetricPickleReceiver)):
+    receivers = []
+    if settings.ENABLE_LINE_RECEIVER:
+       receivers.append((settings.LINE_RECEIVER_INTERFACE,
+                         settings.LINE_RECEIVER_PORT,
+                         MetricLineReceiver))
+    if settings.ENABLE_PICKLE_RECEIVER:
+       receivers.append((settings.PICKLE_RECEIVER_INTERFACE,
+                         settings.PICKLE_RECEIVER_PORT,
+                         MetricPickleReceiver))
+    if settings.ENABLE_MSGPACK_RECEIVER:
+       receivers.append((settings.MSGPACK_RECEIVER_INTERFACE,
+                         settings.MSGPACK_RECEIVER_PORT,
+                         MetricMessagePackReceiver))
+    for (interface, port, protocol) in receivers:
         if port:
             factory = ServerFactory()
             factory.protocol = protocol

=== modified file 'check-dependencies.py'
--- check-dependencies.py	2012-02-10 05:14:15 +0000
+++ check-dependencies.py	2012-02-15 23:21:21 +0000
@@ -168,6 +168,14 @@
   print "Note that txamqp requires python 2.5 or greater."
   warning += 1
 
+# Test for msgpack
+try:
+  import msgpack
+except:
+  print "[WARNING]"
+  print "Unable to import the 'msgpack' module, this is required if you want to use msgpack receiver."
+  warning += 1
+
 
 if fatal:
   print "%d necessary dependencies not met. Graphite will not function until these dependencies are fulfilled." % fatal