日志分析大致流程

簡單概述：
生產(chǎn)過程中會(huì)生成大量的系統(tǒng)日志，應(yīng)用程序日志，安全日志等等，通過對(duì)日志的分析可以了解服務(wù)器的負(fù)載，健康狀況，可以分析客戶的分布情況，客戶的行為，甚至基于這些分析可以做出預(yù)測(cè)。

一般采集流程：
日志產(chǎn)出 ——>采集——>儲(chǔ)存——>分析——>儲(chǔ)存——>可視化

數(shù)據(jù)提?。?br/>由于日志文件基本都以文本形式產(chǎn)出，所以對(duì)日志的分析基本就是對(duì)文本的字符串進(jìn)行分析。所以我們需將文本中有用的信息通過一些設(shè)定條件將其提取出來以方便后面操作。

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

     # 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) # 遍歷當(dāng)前目錄
for log in logs: # path對(duì)象
yield from fir_load(str(log.absolute()), encoding=encoding)
elif path.is_file():
yield from fir_load(str(path),encoding='utf-8')

以上是遍歷路徑提取所有相關(guān)日志文件并按行輸出，并調(diào)用函數(shù)exent，從而對(duì)日志文件完成提取操作。

日志文件的有用文件提取出來后，后面也就明了了主要就是對(duì)自己需要的信息進(jìn)行分析操作了，這里我們需先進(jìn)行時(shí)間管理分析，通過這個(gè)代碼可以將日志文件按照文件本身的產(chǎn)生時(shí)間，通過分組輸出處理，優(yōu)化了分析。
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)]

那么這里完了接下來就是將相應(yīng)的分析代碼寫出來傳個(gè)形參handler就可以得到分析結(jié)果了，比如，先寫個(gè)狀態(tài)碼分析：
#狀態(tài)碼分析

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()}

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

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()         #啟動(dòng)線程處理數(shù)據(jù)

    for item in src:       #將數(shù)據(jù)源取到的數(shù)據(jù)分發(fā)到所有隊(duì)列中
        for q in queues:
            q.put(item)
return reg,run
    這樣的話，window函數(shù)也要進(jìn)行小小的修改，以能get到隊(duì)列。
    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)]

這樣話再調(diào)用的話就可以了。這里再加個(gè)瀏覽器分析
#瀏覽器分析
allbrowsers ={}#所有瀏覽器的統(tǒng)計(jì)
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

再對(duì)數(shù)據(jù)提取的條件進(jìn)行適量改變?nèi)缓笳{(diào)用運(yùn)行即可。
數(shù)據(jù)提取:
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)}
最終調(diào)用：
if name == 'main':
path = 'G:/'

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