日志分析系统/监控系统后端实现

忘是亡心i 2022-03-08 07:22 262阅读 0赞

公司领导想要看到系统按省份、成功率、业务来分类的时时访问监控图像，虽然说是个没啥用的项目，既然老板发话了，还是去做吧，前端显示页面不是自己开发，前端只需要来数据库抽取我时时计算出的结果展示出来就行了，心里一万只草泥马，那就意味着后端所有斗得自己做了。

想了下咋实现呢，监控访问量嘛，最简单通用的方法无非就是读取端口日志，查了下日志，刚好包含了有我所需的字段，所以问题就从0头绪变成了-----日志分析实现监控

有了日志文件，怎么能做到计算后写道数据库呢，还要求几乎是时时（5秒一次更新）的需求，这时候就要用到消息队列了

原理是：

把日志抽取成一条条信息发送到处理代码中计算出结果，最后再写入数据库。

看似一句话，其实需要实现的有：

1、日志抽取

2、消息传递

3、计算逻辑

4、数据库

按以上大方向往下走，慢慢逐个分析

日志抽取：ELK什么的就不谈了，因为我有自己计算代码，所以没用ELK的抽取，用的是filebeat这个超级轻量级日志抽取工具。

filebeat仅是抽取功能，抽到了数据，接下来要传递了，这种批量数据处理更新肯定不能是时时传递处理，所以要用到消息队列

消息队列各种都行，我用的是kafka，优势是有数据持久化和对python友好的对接模块

所以说我的代码是用python来写的，当然为了避免数据库端口压力（避免给自己找坑），数据是在python中批量缓存计算出结果后再写入数据库，这里就要用到pandas模块、json模块等

最后通过pyMySQL写入到MySQL中

内容多又杂，花了一个周实现了后端从0开始搭建（filebeat、kafka、数据库都是新搭建配置）。先在这里纪录下大概，之后会对filebeat搭建与配置、kafka搭建与配置、python-kafka模块使用、pandas模块、json模块、pymysql模块、计算逻辑都会单独纪录

--------------------

2019.3.21更新：

今天突然闲了一下，顺手继续纪录这篇文章。

先从filebeat开始谈起。常用的日志管理套件ELK应该是很出名的了，其中L代表logstash，logstash是ELK的日志抽取工具，所以，日志抽取其实可以采用logstash。我没有采用logstash的原音是，数据不是走到ELK，所以相对logstash来说，基于相同底层开发的filebeat这个工具更加轻量化，系统资源吃的更少。当然如果是做ELK的，无缝对接用logstash也是不错的选择。

filebeat安装很简单，直接yum就可以了，当然也有下载包解压后使用的。yum安装后在/etc/filebeat目录下有相关的配置文件。

[root@localhost ~]# cd /etc/filebeat/
    [root@localhost filebeat]# ls
    fields.yml  filebeat.reference.yml  filebeat.yml  modules.d
    [root@localhost filebeat]#

这里的filebeat.yml文件就是配置文件，修改配置文件，主要加入以下条件：

日志读取文件所在路径、读取条件、发送到kafka的对接信息。

注意yml文件有强校验格式（空格）等行为，启动报错经常会是因为yml配置文件格式不对导致。

我的配置信息：

[root@localhost filebeat]# cat filebeat.yml 
    filebeat.prospectors:
    - input_type: log
      paths:
        - /data/log/testlog/*
      fields:
        log_topic: test01
    
    output.kafka:
      hosts: ["localhost:50071"]
      topic: '%{[fields.log_topic]}'

指定kafka的消息topic，指定机器的ip和kafka的监听端口、指定文件读取的path

这样就能正常使用了，其他相关参数配置可自行google官网详解

启动filebeat很简单，直接输入filebeat即启动，但是是前台启动，需要后台启动的自行加linux参数就行了

--------------------

上面我已经实现的filebeta日志抽取并且发送给kafka，这里纪录下kafka的搭建

kafka安装贼简单，下载tar包，解压就完了。

启动kafka是切换到解压文件夹的bin目录下通过该目录的sh脚本启动。

kafka我理解来看分zk和kfk-service两层

zk负责管理kafka本身的、kfk-service的配置主要针对对接kfk的一些consumer、producer等信息

配置文件没啥要改，我的话就改了下service的启动端口、consumer的groupid等，可以不用改用默认的

启动（在bin文件夹下）：

先要启动zk，才能启动kafka-service，所以启动命令有两个，依次执行：

启动zk：./zookeeper-server-start.sh -daemon ../config/zookeeper.properties  
启动kafka-service：./kafka-server-start.sh ../config/server.properties

由于service默认前台启动，需要后台启动的添加linux后台参数即可。

启动的kafka后，代表filebeat向kafka传递信息kafka已经能够收到了，为了证明能够接受到，打开kafka consumer查看是否接到消息（如果没有后台启动kafka需要新开一个窗口）

./kafka-console-consumer.sh --bootstrap-server localhost:50071 --topic test01  
kafka监听端口我是改了的，按配置文件中的监听端口填参数，后边的topic参数一定要写我们发送消息的topic

只有相同的topic才能接到filebeat过来的消息，测试接到消息代表kafka调通。关于kafka安装报错，晚上有很多安装教程，主要几点就是要改本机名为localhost、配置文件端口、先要启动zk在启动service、如果是前台启动不能关闭了再开consumer/producer等几点。其他的google下就好了

--------------------

调通了kafka，接下来就要考虑到kafka把数据给到python，

kafka和python对接有个python-kafka模块，pip安装模块即可，我用的是python3.5

#!/usr/bin/python
    
    from kafka import  KafkaConsumer
    from kafka import KafkaProducer
    from asyncore import poll
    import pandas as pd
    import time
    from _hashlib import new
    from kafka.consumer.fetcher import ConsumerRecord
    import json
    
    
          
    def accp_data(new_mgs):
    
        for msg in new_mgs:
            byte_msg=msg.value
            string_msg = byte_msg.decode()
            json_msg=json.loads(string_msg)
            msg_get_val=json_msg.get('message')
            msg_list = msg_get_val.split('|')
            msg_list[-1]=msg_list[-1][0:12]
            msg_type=msg_list[1]
            produce=KafkaProducer(bootstrap_servers='45.59.113.211:50071')
            if msg_type =='Res':
                
                meam_res,count_res,count_req=process_data(msg_type,msg_list)
                response1=produce.send('test02',meam_res)
            else:
                meam_res,count_res,count_req=process_data(msg_type,msg_list)
                response2=produce.send('tes03',count_res)
            
    def newframe(dicts):
        
        newdata=pd.DataFrame(dicts,index=[0])
        return newdata
    
    def framedata(datatype,list):
        
        if datatype == 'Res':
            list_res=dict(interface=list[0],type=list[1],province=list[2],status=list[3],processtime=list[4],accepttime=list[5])
            return list_res
        else:
            list_req = dict(interface=list[0],type=list[1],province=list[2],accepttime=list[3])
            return list_req
    
    def process_data(msg_type,msg_list):
        res_frame=pd.DataFrame(columns=['status','province','processtime','type','interface','accepttime'])
        req_frame=pd.DataFrame(columns=['province','type','interface','accepttime'])
        meam_res=None
        count_res=None
        count_req=None
        if msg_type == 'Res':
           
            newlist_res=framedata('Res',msg_list)
            add_res_frame=newframe(newlist_res)
            new_resframe=res_frame.append(add_res_frame,ignore_index=True,sort = False)
            new_resframe['processtime']=new_resframe['processtime'].astype(float)
            meam_res=new_resframe.groupby(['accepttime', 'province','status','interface'], as_index=0).mean()
            count_res=new_resframe.groupby(as_index=0)['accepttime', 'province','status','interface','type'].value_count()
            meam_res_js=meam_res.to_json(orient ="records")
            count_res_js=count_res.to_json(orient ="records")
        elif msg_type == 'Req':
            
            newlist_req=framedata('Req',msg_list)
            
            new_reqframe=req_frame.append(newlist_req,ignore_index=True,sort = False)
            count_req=new_reqframe.groupby(as_index=0)['province','interface','accepttime'].value_count()
            count_req_js=count_req.to_json(orient="records")
        else:
            print('data error')
    
        return meam_res_js,count_res_js,count_req_js
    
    consumer = KafkaConsumer('test01', bootstrap_servers='45.59.113.211:50071',group_id='test01')
    message = consumer.poll(max_records=1000)
    print(message)
    while True:
        if message:
            rec_msg=list(message.values())
            new_msg=rec_msg[0]
            print(new_msg)
            accp_data(new_msg)
        time.sleep(10)

截取了一段我的测试代码，不一定能完全跑通，这里说下思路：

通过kafka模块导入接受函数kafkaconsumer，链接kafka，填写topic和IP等信息，由于要写入数据库，考虑到数据库时时写入的压力，所以最好不要用流式处理，即用for循环后时时写入数据库；通过while定时执行，把数据批量的接入后存在python的字典中，然后进行统计。

python统计有个很好用的模块：pandas，可以生成一个类似于数据库视图的数据列表集合，同时可以对pands视图进行groupby、求和、求平均等运算，所以我选择把数据放到pandas里面，filebeat发送的数据带了一些自己的格式，并且是byte类型的数据，所以python接收到kafka过来的数据时需要转换，由于时byte类型，首先就要转成str类型，之后提取列表、字典、转json等操作，使得最后的数据能够成为pandas的dataframe格式，类似于数据库的单行数据，最后在空的datafame表里面把每一条转换的数据添加到dataframe表形成一个所需的视图，在统计计算。

得到结果后可以直接调用python的数据库模块进行写入，但是我不建议直接写入，因为程序如果阻塞了，那么数据库那边也会同时停止，为了解耦数据库，通过转码成json再次发送到kafka，生成新的消息topic test02在通过python对接test02写入到数据库。

--------------------

数据库搭建按需求进行，没多少难度，这样就实现了接口监控的后端实现

--------------------

总结：

1、为了高可用、解耦等，要选择消息队列中间件

2、尽量选择轻量化工具降低系统开销

3、一定要考虑数据库压力，除非时单实列单服务，否则尽量不要用流处理给到数据库（除非是nosql），而是往数据库写入结果

4、前端解耦的同时也要考虑数据库解耦，保障后期可扩展和易维护特点