日志分析

超、凢脫俗 2022-01-30 06:57 258阅读 0赞

### 文章目录 ###

*  日志分析
 *   *  概述
     *  加载日志文件，解析日志
     *  构建消息队列分发器
     *   *  生产者消费者模型
         *  数据处理流程
         *  分发器的实现
     *  注册分析算法,开始分析
     *   *  IP分析器
         *  pv分析器
         *  浏览器分析器
     *  完整代码如下

# 日志分析 #

## 概述 ##

生成中会生成大量的系统日志、应用程序日志、安全日志等等日志，通过对日志的分析可以了解服务器的负载、健 康状况，可以分析客户的分布情况、客户的行为，甚至基于这些分析可以做出预测。  
一般采集流程 日志产出 -> 采集（Logstash、Flume、Scribe） -> 存储 -> 分析 -> 存储（数据库、NoSQL） -> 可视化  
开源实时日志分析ELK平台 Logstash收集日志，并存放到ElasticSearch集群中，Kibana则从ES集群中查询数据生成图表，返回浏览器端

*  半结构化数据  
    日志是半结构化数据，是有组织的，有格式的数据。可以分割成行和列，就可以当做表理解和处理了，当然也可以 分析里面的数据。
 *  文本分析  
    日志是文本文件，需要依赖文件IO、字符串操作、正则表达式等技术。  
    通过这些技术就能够把日志中需要的数据提取出来。  
    [日志文件下载地址][Link 1]

## 加载日志文件，解析日志 ##

## 构建消息队列分发器 ##

### 生产者消费者模型 ###

一个系统的健康运行，需要监控并处理很多数据，包括日志。对其中已有数据进行采集、分析。 被监控对象就是数据的生产者producer，数据的处理程序就是数据的消费者consumer。

*  传统的生产者和消费者模型  
    ![log002][]  
    传统的生产者消费者模型，生产者生产，消费者消费。但这种模型有些问题。 生产者和消费者代码耦合太高，代码实现上要么是生产者产生一个数据就调用消费者，要么就是消费者处理完一个 数据就调用生产者一次。如果生成规模扩大，不易扩展，生产和消费的速度很难匹配等。
 *  带入消息队列的生产者和消费者模型
 *  作用——解耦、缓冲。  
    ![log003][]  
    日志生产者往往会部署好几个程序，日志产生的也很多，而消费者也会有多个程序，去提取日志分析处理  
    数据的生产是不稳定的！可能会造成短时间数据的“潮涌”，需要缓冲。 消费者消费能力不一样，有快有慢，消费者可以自己决定消费缓冲区中的数据。  
    单机时，可以使用标准库queue模块的类来构建进程内的队列，满足多个线程间的生产消费需要。 大型系统可以使用第三方消息中间件——RabbitMQ、RocketMQ、Kafka等。  
    **数据处理所需模块—队列**
 *  queue模块–多队列，提供了一个先进先出的队列Queue
 *  threading模块–线程。帮助队列不断获取数据。  
    为了让生产者的生产数据和消费者的消费数据同时进行，可以使用不同的线程。

### 数据处理流程 ###

*  生产者（数据源）生产数据，缓冲到消息队列中  
    ![log004][]

### 分发器的实现 ###

数据分析的程序有很多，例如PV分析、IP分析、UserAgent分析等。  
同一套数据可能要被多个分析程序并行处理：

*  需要使用多线程来并行处理
 *  多个分析程序又需要同一份数据，这就是一份变多份  
    ![log005][]  
    这是一个典型的分发器
 *  注册统计分析函数，并为其提供一个单独的数据队列
 *  收集日志数据
 *  将一份日志数据发送到多个已注册的分析函数的队列中去
 *  为了并行，每一个分析函数都在一个独立的线程中执行

######消息队列分发
    def dispatchar(src):
        handler = []
        queueler = []
    
        #队列注册
        def reg(fun):
            q = Queue()
            handler.append(threading.Thread(target=fun,args=(q,)))
            queueler.append(q)
    
        #队列分发
        def run():
            #启动线程
            for hd in handler:
                hd.start()
            #开始分发
            for i in src:
                for qu in queueler:
                    qu.put(i)
        return reg,run
    
    ##创建消息队列分发器
    reg,run = dispatchar(load(".",ext="*.log",recursive=True))

## 注册分析算法,开始分析 ##

### IP分析器 ###

#注册分析ip算法
    @reg
    def ip_handle(qu:Queue):
        ipdict = { }
        while True:
            try:
                data = qu.get(timeout=3) #阻塞读取
                ip = data["address"]
                if ip:
                    ipdict[ip] = ipdict.get(ip,0)+1
            except :
                # 获取数据失败
                print("ip分析完成(查看前5名ip)：")
                print(sorted(ipdict.items(),key=lambda x:x[1],reverse=True)[:5])
                break

### pv分析器 ###

*  PV指的是Page view，也就是页面浏览量或页面点击量。
 *  PV分析，就是按照URL分析。
 *  可以使用urllib.parse 下面的urlparse对象

#注册pv分析
    @reg
    def pv_handle(qu:Queue):
        pvdict = defaultdict(lambda :[0,defaultdict(int)])
        pvnoset = { "/"}
        while True:
            try:
                data = qu.get(timeout=3)
                url = data["url"]
                ip = data["address"]
                if url and url not in pvnoset:
                    ulpa = urlparse(url)
                    pathname = ulpa.path
                    temp = pvdict[pathname]
                    temp[0] += 1
                    temp[1][ip] += 1
            except :
                #获取数据失败
                arr = sorted(pvdict.items(),key=lambda k:k[1][0],reverse=True)
                print("pv前5名：")
                for i in arr[:5]:
                    print("url:{}\t访问次数:{}\t访问ip前5名：{}".format(i[0],i[1][0],sorted(i[1][1].items(),key=lambda x:x[1],reverse=True)[:5]))
                break

### 浏览器分析器 ###

#客户端分析
    ##需要使用第三方库user_agents中的parse
    ##安装方法：pip install pyyaml ua-parser user-agents
    @reg  #注册客户端分析
    def cilent_handle(qu:Queue):
        # cildict = {}
        cildict = defaultdict(lambda : [0,defaultdict(int)])
        while True:
            try:
                data = qu.get(timeout=3)
                uab:parse= data["useragent"].browser
                if uab:
                    family = uab.family
                    ver = uab.version_string
                    ucname = uab.family,uab.version_string
                    ip = data["address"]
                    temp = cildict[ucname]
                    temp[0] +=1
                    temp[1][ip] += 1
            except :
                # 获取数据失败
                print("客户端分析完成(查看前5名)：")
                arr = sorted(cildict.items(),key=lambda x: x[1][0],reverse=True)
                print("前5名：")
                for k,v in arr[:5]:
                    print("浏览器：{}\t访问次数：{}\tip详细次数：{}".format(k,v[0],sorted(v[1].items(),key=lambda x:x[1],reverse=True)[:5]))
                print("后5名：")
                for k, v in arr[-5:]:
                    print("浏览器：{}\t访问次数：{}\tip详细次数：{}".format(k, v[0],sorted(v[1].items(), key=lambda x: x[1], reverse=True)[:5]))
                break
    
    ##启动分析
    run()

## 完整代码如下 ##

from pathlib import Path
    import re,datetime,threading
    from queue import Queue
    from collections import defaultdict
    from user_agents import parse
    from urllib.parse import urlparse
    
    expdict = { 
        "datetime":lambda x:datetime.datetime.strptime(x,"%d/%b/%Y:%H:%M:%S %z"),
        "length":int,
        "status":int,
        "useragent":lambda ua: parse(ua)
    }
    
    #解析日志
    def loganalysis(file:str,encoding=None):
        restr = '''^(?P<address>[\d.]{7,}) - . \[(?P<datetime>[^\]]*)\] "(?P<method>[^ ]+) (?P<url>[^ ]+) (?P<protocol>[^ ]+)" (?P<status>\d{3}) (?P<length>\d)+ "[^ ]*" "(?P<useragent>[^"]+)"'''
        req = re.compile(restr)
        with Path(file).open(encoding=encoding) as f:
            for line in f:
                regex = req.match(line)
                if regex: #日志行解析成功
                    yield { k:expdict.get(k,lambda x:x)(v) for k,v in regex.groupdict().items()}
                else: #解析失败
                    raise Exception("No match. {}".format(line))
    
    #加载日志文件
    def load(*paths,ext:str="*.log",recursive:bool=False,encoding:str="utf8"):
        for filepath in paths:
            path = Path(filepath)
            ext = [ext] if isinstance(ext,str) else list(ext)
            if path.is_dir(): #是目录
                for et in ext:
                    for f in (path.rglob if recursive else path.rglob)(et):
                        yield from loganalysis(f.absolute(),encoding)
            else: #是文件，直接读取
                yield from loganalysis(path.absolute(),encoding)
    
    ######消息队列分发
    def dispatchar(src):
        handler = []
        queueler = []
    
        #队列注册
        def reg(fun):
            q = Queue()
            handler.append(threading.Thread(target=fun,args=(q,)))
            queueler.append(q)
    
        #队列分发
        def run():
            #启动线程
            for hd in handler:
                hd.start()
            #开始分发
            for i in src:
                for qu in queueler:
                    qu.put(i)
        return reg,run
    
    ##创建消息队列分发器
    reg,run = dispatchar(load(".",ext="*.log",recursive=True))
    
    #注册分析ip算法
    @reg
    def ip_handle(qu:Queue):
        ipdict = { }
        while True:
            try:
                data = qu.get(timeout=3) #阻塞读取
                ip = data["address"]
                if ip:
                    ipdict[ip] = ipdict.get(ip,0)+1
            except :
                # 获取数据失败
                print("ip分析完成(查看前5名ip)：")
                print(sorted(ipdict.items(),key=lambda x:x[1],reverse=True)[:5])
                break
    
    #注册pv分析
    @reg
    def pv_handle(qu:Queue):
        pvdict = defaultdict(lambda :[0,defaultdict(int)])
        pvnoset = { "/"}
        while True:
            try:
                data = qu.get(timeout=3)
                url = data["url"]
                ip = data["address"]
                if url and url not in pvnoset:
                    ulpa = urlparse(url)
                    pathname = ulpa.path
                    temp = pvdict[pathname]
                    temp[0] += 1
                    temp[1][ip] += 1
            except :
                #获取数据失败
                arr = sorted(pvdict.items(),key=lambda k:k[1][0],reverse=True)
                print("pv前5名：")
                for i in arr[:5]:
                    print("url:{}\t访问次数:{}\t访问ip前5名：{}".format(i[0],i[1][0],sorted(i[1][1].items(),key=lambda x:x[1],reverse=True)[:5]))
                break
    
    #客户端分析
    ##需要使用第三方库user_agents中的parse
    ##安装方法：pip install pyyaml ua-parser user-agents
    @reg  #注册客户端分析
    def cilent_handle(qu:Queue):
        # cildict = {}
        cildict = defaultdict(lambda : [0,defaultdict(int)])
        while True:
            try:
                data = qu.get(timeout=3)
                uab:parse= data["useragent"].browser
                if uab:
                    family = uab.family
                    ver = uab.version_string
                    ucname = uab.family,uab.version_string
                    ip = data["address"]
                    temp = cildict[ucname]
                    temp[0] +=1
                    temp[1][ip] += 1
            except :
                # 获取数据失败
                print("客户端分析完成(查看前5名)：")
                arr = sorted(cildict.items(),key=lambda x: x[1][0],reverse=True)
                print("前5名：")
                for k,v in arr[:5]:
                    print("浏览器：{}\t访问次数：{}\tip详细次数：{}".format(k,v[0],sorted(v[1].items(),key=lambda x:x[1],reverse=True)[:5]))
                print("后5名：")
                for k, v in arr[-5:]:
                    print("浏览器：{}\t访问次数：{}\tip详细次数：{}".format(k, v[0],sorted(v[1].items(), key=lambda x: x[1], reverse=True)[:5]))
                break
    
    ##启动分析
    run()

运行结果如下：  
![log001][]

[Link 1]: https://github.com/1263351411/xdd.github.io/blob/master/Python/%E9%A1%B9%E7%9B%AE%E5%AE%9E%E6%88%98/%E6%97%A5%E5%BF%97%E5%88%86%E6%9E%90/logs/test.log
[log002]: /images/20220130/4fc02b49f4e84ba491675c233c773786.png
[log003]: /images/20220130/20d9c4a711ab43a0be18f3df0d089239.png
[log004]: /images/20220130/edef7fa18b564db8af7ee134b9fd9487.png
[log005]: /images/20220130/db916bab19e24b8193df465041ec6fb4.png
[log001]: /images/20220130/8d378ae41af8433dbbb0d902f2bc2c16.png