Remote path expression bulk fetch cache

webapp/graphite/remote_storage.py

@@ -102,21 +102,31 @@ def __init__(self, store, metric_path, isLeaf):
     self.fs_path = None
     self.metric_path = metric_path
     self.real_metric = metric_path
-    self.name = metric_path.split('.')[-1]
+    # XXX not nice
+    if isinstance(self.metric_path, list):
+      self.name = "Bulk: %s" % self.metric_path
+    else:
+      self.name = metric_path.split('.')[-1]
     self.__isLeaf = isLeaf
 
 
   def fetch(self, startTime, endTime, now=None):
     if not self.__isLeaf:
       return []
 
+    # XXX not nice
+    if isinstance(self.metric_path, list):
+      targets = [ ('target', v) for v in self.metric_path ]
+    else:
+      targets = [ ('target', self.metric_path) ]
+
     query_params = [
       ('local', '1'),
-      ('target', self.metric_path),
       ('format', 'pickle'),
       ('from', str( int(startTime) )),
       ('until', str( int(endTime) ))
     ]
+    query_params.extend(targets)
     if now is not None:
       query_params.append(('now', str( int(now) )))
     query_string = urlencode(query_params)

webapp/graphite/render/datalib.py

@@ -41,6 +41,7 @@ def __init__(self, name, start, end, step, values, consolidate='average'):
     self.consolidationFunc = consolidate
     self.valuesPerPoint = 1
     self.options = {}
+    self.pathExpression = None
 
 
   def __iter__(self):
@@ -97,6 +98,7 @@ def getInfo(self):
       'end' : self.end,
       'step' : self.step,
       'values' : list(self),
+      'pathExpression' : self.pathExpression
     }
 
 
@@ -254,13 +256,80 @@ def recv_exactly(conn, num_bytes):
 
 if settings.REMOTE_STORE_USE_THREADS:
   threadpool = ThreadPool(settings.REMOTE_STORE_THREADPOOL_SIZE)
+
+def _timebounds(requestContext):
+  startTime = int(epoch(requestContext['startTime']))
+  endTime = int(epoch(requestContext['endTime']))
+  now = int(epoch(requestContext['now']))
+
+  return (startTime, endTime, now)
+
+def _prefetchMetricKey(pathExpression, start, end):
+  return '-'.join([pathExpression, str(start), str(end)])
+
+def prefetchRemoteData(requestContext, pathExpressions):
+  # if required, fetch data from all remote nodes
+  # storing the result in a big hash of the form:
+  # data[node][hash(originalPathExpression, start, end)] = [ matchingSeries, matchingSeries2, ... ]
+
+  prefetchedRemoteData = {}
+  if requestContext['localOnly']:
+    return prefetchedRemoteData
+
+  (startTime, endTime, now) = _timebounds(requestContext)
+  remote_nodes = [ RemoteNode(store, pathExpressions, True) for store in STORE.remote_stores ]
+
+  def _nodeFetch( (node, startTime, endTime, now) ):
+    r = node.fetch(startTime, endTime, now)
+    return (node, r)
+
+  nodesFetchTuples = [ (node, startTime, endTime, now) for node in remote_nodes ]
+  if settings.REMOTE_STORE_USE_THREADS:
+    nodesFetchResults = threadpool.map(_nodeFetch, nodesFetchTuples)
+  else:
+    nodesFetchResults = map(_nodeFetch, nodesFetchTuples)
+
+  for (node, results) in nodesFetchResults:
+    # prefill result with empty list
+    # Needed to be able to detect if a query has already been made
+    prefetchedRemoteData[node.store.host] = {}
+    for pe in pathExpressions:
+      prefetchedRemoteData[node.store.host][_prefetchMetricKey(pe, startTime, endTime)] = []
+
+    for series in results:
+      # series.pathExpression is original target, ie. containing wildcards
+      # XXX Print a useful error and abort. should be nice to disable further prefetch calls to that backend
+      try:
+        k = _prefetchMetricKey(series['pathExpression'], startTime, endTime)
+      except KeyError:
+        log.exception("Remote node %s doesn't support prefetching data... upgrade!" % node.store.host)
+        raise
+
+      if prefetchedRemoteData[node.store.host][k] is not None:
+        # This should not be needed because of above filling with [],
+        # but better be safe than sorry
+        prefetchedRemoteData[node.store.host][k].append(series)
+      else:
+        prefetchedRemoteData[node.store.host][k] = [series]
+
+  return prefetchedRemoteData
+
+def prefetchLookup(requestContext, node):
+  # Returns a seriesList if found in cache
+  # or None is key doesn't exist (aka prefetch didn't cover this pathExpr / timerange
+  (start, end, now) = _timebounds(requestContext)
+  try:
+    cache = requestContext['prefetchedRemoteData'][node.store.host]
+    r = cache[_prefetchMetricKey(node.metric_path, start, end)]
+  except (AttributeError, KeyError):
+    r = None
+
+  return r
 
 # Data retrieval API
 def fetchData(requestContext, pathExpr):
   seriesList = []
-  startTime = int(epoch(requestContext['startTime']))
-  endTime = int(epoch(requestContext['endTime']))
-  now = int(epoch(requestContext['now']))
+  (startTime, endTime, now) = _timebounds(requestContext)
 
   dbFiles = [dbFile for dbFile in LOCAL_STORE.find(pathExpr)]
 
@@ -295,9 +364,19 @@ def fetchData(requestContext, pathExpr):
     remote_nodes = [ RemoteNode(store, pathExpr, True) for store in STORE.remote_stores ]
 
     def _nodeFetch( (node, startTime, endTime, now) ):
-      r = node.fetch(startTime, endTime, now)
+      if settings.PREFETCH_REMOTE_DATA:
+        r = prefetchLookup(requestContext, node)
+        # Will be either:
+        #   []: prefetch done, returned no data. Do not fetch
+        #   seriesList: prefetch done, returned data, do not fetch
+        #   None: prefetch not done, FETCH
+        if r is None:
+          r = node.fetch(startTime, endTime, now)
+      else:
+        r = node.fetch(startTime, endTime, now)
       return (node, r)
 
+    # This thread stuff is not needed if PREFETCH_REMOTE_DATA is in effect
     nodesFetchTuples = [ (node, startTime, endTime, now) for node in remote_nodes ]
     if settings.REMOTE_STORE_USE_THREADS:
       nodesFetchResults = threadpool.map(_nodeFetch, nodesFetchTuples)

webapp/graphite/render/evaluator.py

@@ -42,6 +42,27 @@ def evaluateTokens(requestContext, tokens):
   elif tokens.boolean:
     return tokens.boolean[0] == 'true'
 
+def extractPathExpressions(targets):
+  # Returns a list of unique pathExpressions found in the targets list
+
+  pathExpressions = []
+
+  def extractPathExpression(tokens):
+    if tokens.expression:
+      return extractPathExpression(tokens.expression)
+    elif tokens.pathExpression:
+      pathExpressions.append(tokens.pathExpression)
+    elif tokens.call:
+      [extractPathExpression(arg) for arg in tokens.call.args]
+
+  for target in targets:
+    tokens = grammar.parseString(target)
+    extractPathExpression(tokens)
+
+  s = set(pathExpressions)
+  pathExpressions = list(s)
+  return pathExpressions
+
 
 #Avoid import circularities
 from graphite.render.functions import SeriesFunctions

webapp/graphite/render/views.py

@@ -34,7 +34,8 @@
 from graphite.util import getProfileByUsername, json, unpickle
 from graphite.remote_storage import HTTPConnectionWithTimeout
 from graphite.logger import log
-from graphite.render.evaluator import evaluateTarget
+from graphite.render.evaluator import evaluateTarget, extractPathExpressions
+from graphite.render.datalib import prefetchRemoteData
 from graphite.render.attime import parseATTime
 from graphite.render.functions import PieFunctions
 from graphite.render.hashing import hashRequest, hashData
@@ -59,6 +60,7 @@ def renderView(request):
     'endTime' : requestOptions['endTime'],
     'now': requestOptions['now'],
     'localOnly' : requestOptions['localOnly'],
+    'prefetchedRemoteData' : {},
     'data' : []
   }
   data = requestContext['data']
@@ -109,7 +111,13 @@ def renderView(request):
     if cachedData is not None:
       requestContext['data'] = data = cachedData
     else: # Have to actually retrieve the data now
-      for target in requestOptions['targets']:
+      targets = requestOptions['targets']
+      if settings.PREFETCH_REMOTE_DATA:
+        t = time()
+        pathExpressions = extractPathExpressions(targets)
+        requestContext['prefetchedRemoteData'] = prefetchRemoteData(requestContext, pathExpressions)
+        log.rendering("Prefetching remote data took %.6f" % (time() - t))
+      for target in targets:
         t = time()
         seriesList = evaluateTarget(requestContext, target)
         log.rendering("Retrieval of %s took %.6f" % (target, time() - t))

webapp/graphite/settings.py

@@ -139,6 +139,7 @@
 WHISPER_FETCH_STRATEGY_COARSE_THRESHOLD = float("inf")
 REMOTE_STORE_USE_THREADS = False
 REMOTE_STORE_THREADPOOL_SIZE = 8
+PREFETCH_REMOTE_DATA = False
 
 ## Load our local_settings
 try: