importing 14.4 release

Author: ivmfnal

CacheLogParser.py

@@ -0,0 +1,255 @@
+from datetime import datetime, timedelta
+import sys, os, stat, re, time
+from timelib import ShiftTZ, timestamp
+from GraphiteInterface import GraphiteInterface
+
+class CacheLogParser:
+
+    def __init__(self, files, last_file_inode, last_file_position, last_t):
+        self.Files = files[:]
+        self.TMax = None
+        self.TStart = last_t
+        self.LastFileInode = last_file_inode
+        self.LastFilePosition = last_file_position
+        
+    def parseLine(self, l):
+        out = {}
+        while l:
+            l = l.strip()
+            if not l:   break
+            words = l.split("=", 1)
+            if len(words) != 2:
+                break
+            n, rest = tuple(words)
+            if rest[0] == '[':
+                words = rest[1:].split("]",1)
+            else:
+                words = rest.split(None, 1)
+            val = words[0]
+            rest = words[1] if len(words) > 1 else ""
+            if n == 'time':
+                words = val.split()
+                dt = datetime.strptime(words[0], "%d/%b/%Y:%H:%M:%S")
+                zone = None
+                if len(words) > 1:
+                    zone = words[1]
+                    minus = 1.0
+                    if zone[0] == '-':
+                        minus = -1.0
+                        zone = zone[1:]
+                    elif zone[0] == '+':
+                        zone = zone[1:]
+                    zone = zone[:4]
+                    h = float(zone[:2])
+                    m = float(zone[2:])/60.0
+                    shift = (h+m)*minus
+                    zone = ShiftTZ(shift)
+                    dt = dt.replace(tzinfo=zone)
+                out["clock"] = timestamp(dt)
+                #print val, " -> ", dt, out["clock"]
+            elif n == 'request':
+                req, url, proto = val.split(None, 2)
+                out["req"] = req
+                out["url"] = url
+                out["proto"] = proto
+            out[n] = val
+            l = rest
+        #print "out=", out
+        return out
+        
+    def validateData(self, data):
+        for k in ['req','url','clock','cached','bytes_sent']:
+            if k not in data: return False
+        return True
+                
+    def parse(self):
+        out = []        # [(t, data),...]
+        t_max = last_t
+        
+        files = []
+        for f in self.Files:
+            try:    s = os.stat(f)
+            except:
+                print(sys.exc_info())
+            mtime = s.st_mtime
+            if last_t == None or mtime >= self.TStart:
+                files.append((mtime, f))
+            else:
+                print("File %s was not updated since last scan time(%s). File updated at %s" % (
+                    f, time.ctime(mtime), time.ctime(last_t)))
+                
+        files.sort()
+        
+        t_begin = None
+        
+        for mt, fn in files:
+            try:
+                f = open(fn, 'r')
+            except:
+                continue
+                
+            s = os.fstat(f.fileno())
+            size = s.st_size
+            inode = s.st_ino
+            
+            if inode == self.LastFileInode:
+                f.seek(self.LastFilePosition)
+            
+            self.LastFileInode = inode
+            self.LastFilePosition = size
+            
+            for l in f.readlines():
+                try:
+                    data = self.parseLine(l)
+                except:
+                    continue
+                if self.validateData(data):
+                    t = data["clock"]
+                    if last_t == None or t > last_t:
+                        yield (t, data)
+                        if t_max == None:    
+                            t_begin = t
+                            t_max = t
+                        t_max = max(t_max, t)
+                else:
+                    #print "data validation failed:", l
+                    pass
+        self.TMax = t_max
+        
+class ServiceData:
+    def __init__(self, suffix, interval):
+        self.Suffix = suffix
+        self.Interval = interval
+        self.RequestsMiss = 0
+        self.RequestsHit = 0
+        self.BytesMiss = 0
+        self.BytesHit = 0
+        
+    def frequencies(self):
+        return (
+            float(self.RequestsHit)/self.Interval, 
+            float(self.RequestsMiss)/self.Interval, 
+            float(self.BytesHit)/self.Interval, 
+            float(self.BytesMiss)/self.Interval   
+        )
+        
+    def __str__(self):
+        return "[Service %s: requests: %d/%d  bytes: %d/%d]" % (self.Suffix, 
+                self.RequestsHit, self.RequestsMiss, self.BytesHit, self.BytesMiss)
+        
+class CacheLogSummarizer:
+
+    def __init__(self, parser, map):
+        self.Files = files
+        self.URLMap = map            # (graphite_suffix, url_head)
+        self.Parser = parser
+        
+    def interval(self, dt, interval):
+        dt = int(dt)
+        t0 = (dt//interval)*interval
+        t1 = t0 + interval
+        return t0, t1
+        
+    def summarize(self, interval):
+        # interval - seconds
+        lst = self.Parser.parse()
+        t0, t1 = None, None
+        int_data = {}
+        for t, data in lst:
+            if t0 == None or t >= t1:
+                if int_data:    yield (t0, t1, int_data)
+                t0, t1 = self.interval(t, interval)
+                int_data = {}   # {service_name -> ServiceData}
+            # find service
+            svc = None
+            suffix = None
+            for url_head, sfx in self.URLMap:
+                if data["url"].startswith(url_head):
+                    suffix = sfx
+                    break
+            else:
+                continue
+                
+            svc_data = int_data.get(sfx)
+            if svc_data == None:
+                svc_data = ServiceData(sfx, interval)
+                int_data[sfx] = svc_data
+            cache_sts = data["cached"]
+            try:
+                n = int(data["bytes_sent"])
+            except:
+                n = 0
+                
+            if cache_sts == 'HIT':
+                svc_data.RequestsHit += 1
+                svc_data.BytesHit += n
+            else:
+                svc_data.RequestsMiss += 1
+                svc_data.BytesMiss += n
+                
+        if int_data:
+            yield (t0, t1, int_data)
+
+if __name__ == '__main__':
+
+    import yaml, getopt, sys, glob
+    
+    config_file = None
+    
+    opts, args = getopt.getopt(sys.argv[1:], "c:")
+
+    for opt, val in opts:
+        if opt == '-c': config_file = val
+
+    config = yaml.load(open(config_file, 'r').read())
+    map = config.get("mapping")
+    
+    map_lst = []
+    for d in map:
+        map_lst.append((d["url"],d["suffix"]))
+    #print map_lst
+    
+    
+    state_file = config.get("state_file")
+    interval = config.get("aggregation_interval", 60)
+    files = config.get("files")
+    
+    GI = GraphiteInterface(config.get("Graphite"))
+    
+    last_t = None
+    
+    try:
+        sf = open(state_file, 'r')
+        words = sf.read().split()
+        last_file_inode = int(words[0])
+        last_file_position = int(words[1])
+        last_t = int(words[2])
+    except:
+        last_file_inode = None
+        last_file_position = None
+        last_t = None
+    
+    parser = CacheLogParser(glob.glob(files), last_file_inode, last_file_position, last_t)
+    s = CacheLogSummarizer(parser, map_lst)
+    
+    lst = s.summarize(interval)
+    for t0, t1, data in lst:
+        for svc, svc_data in sorted(data.items()):
+            rh, rm, bh, bm = svc_data.frequencies()
+            GI.feedData(t1, svc_data.Suffix+".requests.hit", rh)
+            GI.feedData(t1, svc_data.Suffix+".requests.miss", rm)
+            GI.feedData(t1, svc_data.Suffix+".bytes.hit", bh)
+            GI.feedData(t1, svc_data.Suffix+".bytes.miss", bm)
+            #print t0, t1, svc, svc_data.frequencies()
+        last_t = t0
+    GI.flushData()
+            
+    if last_t:
+        open(state_file, 'w').write("%s %s %d" % (parser.LastFileInode, parser.LastFilePosition, last_t))
+            
+        
+        
+        
+        
+        
+