日志分析大致流程

簡單概述：
生產過程中會生成大量的系統日志，應用程序日志，安全日志等等，通過對日志的分析可以了解服務器的負載，健康狀況，可以分析客戶的分布情況，客戶的行為，甚至基于這些分析可以做出預測。

一般采集流程：
日志產出 ——>采集——>儲存——>分析——>儲存——>可視化

數據提取：
由于日志文件基本都以文本形式產出，所以對日志的分析基本就是對文本的字符串進行分析。所以我們需將文本中有用的信息通過一些設定條件將其提取出來以方便后面操作。

     所以我們的思路，通過遍歷路徑將所有的相關日志文件全部按行輸出，再通過正則表達式寫出每行信息相對的提取規則，再加個字典文件對提取出來的有用信息進一步提升。

     # line = '''114.249.235.230 - - [11/Apr/2017:10:45:51 +0800] "GET / HTTP/1.1" 200 7488 "-" "Mozilla/5.0 (iPhone; CPU iPhone OS 10_2_1 like Mac OS X) AppleWebKit/602.4.6 (KHTML, like Gecko) Version/10.0 Mobile/14D27 Safari/602.1"''
    以上是這篇日志的主要格式。

import re

patten = '''(?P<name1>[\d.]{7,15}) - - [(?P<datetime>[/\w +:]+)] "(?P<method>\w+) (?P<url>\S+) (?P<protocol>[\w/\d.]+)" (?P<status>\d+) (?P<length>\d+) .+ "(?P<name5>.+)"'''

opms = {'datetime':lambda strt:datetime.datetime.strptime(strt,'%d/%b/%Y:%H:%M:%S %z'),'status':int, 'length':int}

gexdx = re.compile(patten)

def exent(line:str):
matcher = gexdx.match(line)
return {k:opms.get(k,lambda x:x)(v) for k,v in matcher.groupdict().items()}

以上就是最終代碼的提取每行有用信息的代碼。

from pathlib import Path
def fir_load(filename ,encoding='utf-8'):
with open(filename,encoding='utf-8') as f:
for line in f:
fields = exent(line)
if fields:
yield fields
else:
pass

def load(paths,encoding='utf-8',ext='.log',r=False):
for p in paths:
path = Path(p)
if path.is_dir():
if isinstance(ext,str):
ext = [ext]
for e in ext:
logs = path.rglob(e) if r else path.glob(e) # 遍歷當前目錄
for log in logs: # path對象
yield from fir_load(str(log.absolute()), encoding=encoding)
elif path.is_file():
yield from fir_load(str(path),encoding='utf-8')

以上是遍歷路徑提取所有相關日志文件并按行輸出，并調用函數exent，從而對日志文件完成提取操作。

日志文件的有用文件提取出來后，后面也就明了了主要就是對自己需要的信息進行分析操作了，這里我們需先進行時間管理分析，通過這個代碼可以將日志文件按照文件本身的產生時間，通過分組輸出處理，優化了分析。
import datetime
def window(handler,width:int,interval:int):
buf = []
start = datetime.datetime.strptime('1970/01/01 00:00:01 +0800','%Y/%m/%d %H:%M:%S %z')
current = datetime.datetime.strptime('1970/01/01 00:00:02 +0800', '%Y/%m/%d %H:%M:%S %z')
delta =datetime.timedelta(seconds=width-interval)
while True:
for date in s:
if date:
buf.append(date)
current = date['datetime']

        if (current-start).total_seconds() > interval:
            ret = handler(buf)
            print(ret)
            start = current
            buf = [x for x in buf if x['datetime'] > (current - delta)]

那么這里完了接下來就是將相應的分析代碼寫出來傳個形參handler就可以得到分析結果了，比如，先寫個狀態碼分析：
#狀態碼分析

def status_handler(iterable): #列表包字典
state = {}
for item in iterable:
ss = item['status']
state[ss] =state.get(ss,0)+1
length = len(iterable)
return {k:v/length for k,v in state.items()}

    這樣一個簡單的日志分析就完了，但是如果想同時進行多個分析怎辦呢？在平常工作過程中難免會要進行多次分析的，這時就需用到分發。
    #分發器

def dispatcher(src):
handlers = []
queues = []

def reg(handler,width:int,interval:int):
    q =Queue()
    queues.append(q)

    h = threading.Thread(target = window,args=(q,handler,width,interval))
    handlers.append(h)

def run():
    for t in handlers:
        t.start()         #啟動線程處理數據

    for item in src:       #將數據源取到的數據分發到所有隊列中
        for q in queues:
            q.put(item)
return reg,run
    這樣的話，window函數也要進行小小的修改，以能get到隊列。
    def window(src:Queue,handler,width:int,interval:int):
buf = []
start = datetime.datetime.strptime('1970/01/01 00:00:01 +0800','%Y/%m/%d %H:%M:%S %z')
current = datetime.datetime.strptime('1970/01/01 00:00:02 +0800', '%Y/%m/%d %H:%M:%S %z')
delta =datetime.timedelta(seconds=width-interval)
while True:
    date = src.get()
    if date:
        buf.append(date)
        current = date['datetime']

        if (current-start).total_seconds() > interval:
            ret = handler(buf)
            print(ret)
            start = current
            buf = [x for x in buf if x['datetime'] > (current - delta)]

這樣話再調用的話就可以了。這里再加個瀏覽器分析
#瀏覽器分析
allbrowsers ={}#所有瀏覽器的統計
def browser_handler(iterable):
browsers = {}
for item in iterable:
ua = item['useragent']

    key = (ua.browser.family,ua.browser.version_string)
    browsers[key] = browsers.get(key,0)+1
    allbrowsers[key] = allbrowsers.get(key,0)+1

print(sort(allbrowsers.items(),key=lambda x:x[1],reverse=True)[:10])
return browsers

再對數據提取的條件進行適量改變然后調用運行即可。
數據提取:
from user_agents import parse

patten = '''(?P<name1>[\d.]{7,15}) - - [(?P<datetime>[/\w +:]+)] "(?P<method>\w+) (?P<url>\S+) (?P<protocol>[\w/\d.]+)" (?P<status>\d+) (?P<length>\d+) .+ "(?P<useragent>[^"]+)"'''

opms = {'datetime':lambda strt:datetime.datetime.strptime(strt,'%d/%b/%Y:%H:%M:%S %z'),'status':int, 'length':int,'useragent':lambda ua:parse(ua)}
最終調用：
if name == 'main':
path = 'G:/'

reg,run = dispatcher(load(path))
reg(status_handler,10,5)
reg(browser_handler,5,5)
run()