IT数码 购物 网址 头条 软件 日历 阅读 图书馆
TxT小说阅读器
↓语音阅读,小说下载,古典文学↓
图片批量下载器
↓批量下载图片,美女图库↓
图片自动播放器
↓图片自动播放器↓
一键清除垃圾
↓轻轻一点,清除系统垃圾↓
开发: C++知识库 Java知识库 JavaScript Python PHP知识库 人工智能 区块链 大数据 移动开发 嵌入式 开发工具 数据结构与算法 开发测试 游戏开发 网络协议 系统运维
教程: HTML教程 CSS教程 JavaScript教程 Go语言教程 JQuery教程 VUE教程 VUE3教程 Bootstrap教程 SQL数据库教程 C语言教程 C++教程 Java教程 Python教程 Python3教程 C#教程
数码: 电脑 笔记本 显卡 显示器 固态硬盘 硬盘 耳机 手机 iphone vivo oppo 小米 华为 单反 装机 图拉丁
 
   -> 大数据 -> Python+Celery实现基于Fastnetmon异常流量清洗 -> 正文阅读

[大数据]Python+Celery实现基于Fastnetmon异常流量清洗

背景

FastNetMon+Influxdb+Grafana+GoBGP可搭建一套基于 NetFLOW / sFLOW 的流量统计报告系统,其中:

  • FastNetMon 是一个基于多种抓包引擎(NetFlow, IPFIX, sFLOW, netmap, PF_RING, PCAP)的DoS/DDoS攻击高效分析工具,可以探测和分析网络中的异常流量情况,同时可以通过外部脚本通知或阻断攻击;
  • InfluxDB 是一款开源开源时序型数据库,和FastNetMon集成,用于将数据统计进行存储;
  • Grafana 是一款非常强大且易用的数据可视化工具;
  • GoBGP 是一个开源 BGP 实现,可以提供 BGP 协议的控制平面功能;

清洗需求

查询Influxdb,当某个IP段超过一定的阈值时,我们需要进行以下操作:

  1. 将开始时间、IP段、流量、是否过期、开始时间记录至统计数据库并开始计时;
  2. 执行Gobgp打标命令;
  3. 对于已过期的IP段,不要重复录入Mysql;
  4. 对于未过期的IP段,需保留历史记录,因此按步骤1再次记录至Mysql;
  5. 当计时延迟3600s结束之后,更新IP段的过期字段、结束时间并执行Gobgp去标命令;

需求分析

1.统计数据库选型

统计数据库我们可以复用Influxdb或使用Mysql,但Influxdb作为时序数据库,后续的字段更新必须基于时间戳和tag查询,字段的更新不友好,因此最终我们选择Mysql。

2.异步解耦

需求中bgp打标、去标、延迟执行等操作,在程序运行过程中计时、命令等待、命令返回错误等意外情况,都会导致运行中断,因此我们考虑使用Python + Celery(消息队列工具,可用于处理实时数据以及任务调度),来与以上情况进行异步解耦。

另,在实际应用中Celery可通过队列来调度任务,不用担心并发量高时系统负载过大。

环境准备

1.InfluxDB环境

# 若influxdb环境已存在,可跳过此步骤
docker pull influxdb:1.8

docker run -p 8086:8086 \
	--name influxdb \
	--restart unless-stopped \
	-e DOCKER_INFLUXDB_INIT_USERNAME=admin \
	-e DOCKER_INFLUXDB_INIT_PASSWORD=admin@123 \
    -v /data/influxdb/data:/var/lib/influxdb \
    -v /data/influxdb/config/influxdb.conf:/etc/influxdb/influxdb.conf \
    -v /etc/localtime:/etc/localtime \
    -d influxdb:1.8

数据库记录(若有测试需求,可私聊获取):

在这里插入图片描述

2.Mysql数据库

# 1.安装mysql
docker pull mysql:5.7

docker run -p 3306:3306 --name mysql \
-v /usr/local/docker/mysql/conf:/etc/mysql \
-v /usr/local/docker/mysql/logs:/var/log/mysql \
-v /usr/local/docker/mysql/data:/var/lib/mysql \
-e MYSQL_ROOT_PASSWORD=123456 \
-d mysql:5.7

# 2.运行容器
docker run -p 3306:3306 --name mysql -e MYSQL_ROOT_PASSWORD=123456 -d mysql:5.7

# 3.登录
docker exec -it mysql bash
mysql -uroot -p123456

# 4.建库
create database fastnetmon
# 5.建表
CREATE TABLE `statistic` (
  `id` bigint(20) NOT NULL AUTO_INCREMENT COMMENT '主键',
  `network` varchar(40) DEFAULT NULL COMMENT 'IP段',
  `bits_incoming` int(11) DEFAULT NULL COMMENT '进口流量',
  `starttime` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间',
  `endtime` datetime DEFAULT NULL ON UPDATE CURRENT_TIMESTAMP COMMENT '结束时间',
  isexpire BOOLEAN DEFAULT NULL COMMENT '是否过期',
  PRIMARY KEY (`id`) USING BTREE
) ENGINE=InnoDB AUTO_INCREMENT=1 DEFAULT CHARSET=utf8mb4 COMMENT='统计表';

3.redis环境

yum install redis -y
systemctl start redis

4.python环境

# 1.ananconda虚拟环境安装
# 2.安装python 3.9
conda create -n influx python=3.9
source activate influx
pip install influxdb celery redis pymysql

目录结构

influxdb
├── celery_app
│   ├── celeryconfig.py
│   ├── gobgp.py
│   ├── __init__.py
│   └── record.py
├── celery.log
├── dbconn.py
├── influx.log
├── influx.py
├── __init__.py
├── log.py
└── Readme

代码详解

1.主程序

# 1.日志模块
vim log.py
# -*- coding: utf-8 -*-
import os
import logging

logging.basicConfig(
    level = logging.INFO,
    format = '%(asctime)s, %(filename)s, %(levelname)s, %(message)s',
    datefmt = '%Y-%m-%d %H:%M:%S',
    filename = "influx.log",
    filemode = 'a'
)

# 2.数据库操作封装
vim dbconn.py
# -*- coding: utf-8 -*-
import requests
import json
from log import logging
import pymysql


def QueryInflux(sql):
    try:
        url = "http://192.168.3.243:8086/query?pretty=true&db=fastnetmon&q=" + sql
        # 超时10s
        res = requests.get(url, timeout=10)
        return json.loads(res.text)
    except Exception as e:
        logging.error(e)

def ConnMySql():
    mysql_conn = pymysql.connect(host= '192.168.3.243', port= 3306, user= 'root', password= '123456', db= 'fastnetmon')
    return mysql_conn

def QueryMySql(sql):
    try:
        mysql_conn = ConnMySql()
        with mysql_conn.cursor() as cursor:
            cursor.execute(sql)
            select_result = cursor.fetchone()
            return select_result
    except Exception as e:
        logging.error(e)


def InsertMySql(sql):
    try:
        mysql_conn = ConnMySql()
        with mysql_conn.cursor() as cursor:
            cursor.execute(sql)
        mysql_conn.commit()
    except Exception as e:
        mysql_conn.rollback()
        logging.error(e)


# 3.主程序
vim influx.py
# -*- coding: utf-8 -*-
from celery_app import gobgp,record
from dbconn import QueryInflux,QueryMySql,InsertMySql
from log import logging
import datetime


def WorkHard(*lst):
    try:
        #下发bgp打包任务至celery,立即执行
        #gobgp.cmd.delay("java -version")
        gobgp.cmd.delay("/opt/gobgp/gobgp global rib add -a ipv4 " + lst[0][2] +" community 65100:888")
 
        # 入库
        starttime = datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S') 
        # isexpire字段 1:过期,0:未过期
        sql = "INSERT INTO statistic (network, bits_incoming, isexpire, starttime ) VALUES ('{0}','{1}', '{2}', '{3}')".format(lst[0][2], lst[0][1], 0, starttime)
        logging.info("插入IP段 %s 到 statistic表" % lst[0][2])
        InsertMySql(sql)
        
        #3600秒后记录更新
        record.update.apply_async(args = [lst[0][2]], countdown = 3600)

    except Exception as e:
        logging.error(e)
    

if __name__ == '__main__':
    # 查询原始表符合要求的数据
    #top_sql = "select top(bits_incoming, network, 30),network from networks_traffic where time > now() - 10d tz('Asia/Shanghai')"
    top_sql = "select top(bits_incoming, network, 30),network from networks_traffic where time > now() - 6s"
    top_res = QueryInflux(top_sql)
    try:
        for item in top_res["results"]:
            # 若结果不为空,则判断是否过期
            if "series" in item:
                for series in item["series"]:
                    for value in series["values"]:
                        # 指定阈值
                        if value[1] > 184549376:
                            # statistic表是否为空
                            exist_measure = QueryMySql("select * from statistic limit 1;")
                            # 为空,则插入
                            if  exist_measure is None:
                                WorkHard(value)
                            # 不为空
                            else:
                                # 是否存在过期数据
                                exist_isexp_sql = "select count(*)  from statistic where network = \'" + value[2] + "\' and isexpire = '1'"
                                exist_isexp_ip = QueryMySql(exist_isexp_sql)                  
                                # 不存在过期数据
                                if exist_isexp_ip[0] == 0:
                                    #是否存在未过期数据
                                    exist_noexp_sql = "select count(*)  from statistic where network = \'" + value[2] + "\' and isexpire = '0'"
                                    exist_noexp_ip = QueryMySql(exist_noexp_sql)                  
                                    # 未过期数据不存在
                                    if exist_noexp_ip[0] == 0:
                                        WorkHard(value)
                                    # 未过期数据存在,跳过
                                    else:
                                        logging.info("IP段 %s 已存在且未过期,跳过" % value[2])
                                # 存在过期数据
                                else:
                                    WorkHard(value)
    except Exception as e:
        logging.error(e)

注意:当主程序第一次运行时,需要首先考虑到数据库为空的情况,然后再判断是否存在过期数据。

  • 日志模块,实现对日志格式的定义

  • 数据库操作封装,实现对Influxdb、Mysql的连接、查询、插入等操作的封装;

    其中Influxdb我们选择使用http api,而是不是influx-client,因为大数据量情况下client端操作会卡死库。

  • 主程序,主流程逻辑判断与调用;

2.Celery异步操作

# 1.Celery实例初始化
vim __init__.py
# -*- coding: utf-8 -*-
from celery import Celery

# 创建 Celery 实例
app = Celery('tasks')

# 通过 Celery 实例加载配置模块
app.config_from_object('celery_app.celeryconfig')

# 2.Celery配置
vim celeryconfig.py
# -*- coding: utf-8 -*-
# 指定Broker
BROKER_URL = 'redis://127.0.0.1:6379'
# 指定Backend
CELERY_RESULT_BACKEND = 'redis://127.0.0.1:6379/0'
# 指定时区,默认是 UTC
CELERY_TIMEZONE='Asia/Shanghai'
# 指定导入的任务模块
CELERY_IMPORTS = (
    'celery_app.gobgp',
    'celery_app.record'
)

# 3.命令执行模块
vim gobgp.py
# -*- coding: utf-8 -*-
# bgp打包

import sys
import subprocess
from celery_app import app
sys.path.append("..")

@app.task
def cmd(command):
    try:
        subp = subprocess.Popen(command, shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
        #等待执行时间
        subp.wait(3)
        if subp.poll() == 0:
            #执行命令成功,返回命令结果
            res = subp.communicate()
        else:
            #命令执行不成功,返回报错
            res = subp.stderr.read()
   
        print(res)
    except Exception as e:
        print(e)

# 4.数据库更新
vim record.py
# -*- coding: utf-8 -*-
# 更新过期记录

from celery_app import app,gobgp
import sys 
sys.path.append("..") 
from dbconn import QueryMySql,InsertMySql,QueryInflux
import datetime

@app.task
def update(network):
    try:
        # 去标
        #gobgp.cmd("pwd")
        gobgp.cmd("/opt/gobgp/gobgp globa rib del " + network + " -a ipv4")

        # 过期且更新结束时间
        endtime = datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')
        InsertMySql("update statistic set endtime = '{0}', isexpire = '{1}' where network = '{2}' and isexpire = '{3}'".format(endtime, 1, network, 0))

        print('设置IP段 %s 过期' % network)
    except Exception as e:
        print(e)

Celery任务主要分为:

  • 命令执行模块,执行具体的命令操作,捕获命令返回结果、命令异常捕获;
  • 数据库记录模块,命令执行及数据库记录更新等操作;

程序执行

# 1.运行celery
# 首先在influxdb目录下运行celery
cd influxdb
nohup celery -A celery_app worker --loglevel=info -c 4  >> celery.log  2>&1 &
# 建议使用supervisor托管celery,实现celery的自启动管理
# 2.运行主程序
python influx.py

# 3.日志查看
cd influxdb
celery任务在此目录下查看celery.log
主程序日志在此目录下查看influx.log

Mysql数据库新记录、未过期记录、过期记录效果如下:
在这里插入图片描述
在这里插入图片描述

总结

通过这篇文章,如果你想快搭建一套基于 NetFLOW / sFLOW 的流量统计报告系统,你可以体验下FastNetMon+Influxdb+Grafana+GoBGP的解决方案;如果你想学习Python + Celery 的具体使用,也可参考清洗需求来进行实践。

  大数据 最新文章
实现Kafka至少消费一次
亚马逊云科技:还在苦于ETL?Zero ETL的时代
初探MapReduce
【SpringBoot框架篇】32.基于注解+redis实现
Elasticsearch:如何减少 Elasticsearch 集
Go redis操作
Redis面试题
专题五 Redis高并发场景
基于GBase8s和Calcite的多数据源查询
Redis——底层数据结构原理
上一篇文章      下一篇文章      查看所有文章
加:2022-03-13 21:52:47  更:2022-03-13 21:54:14 
 
开发: C++知识库 Java知识库 JavaScript Python PHP知识库 人工智能 区块链 大数据 移动开发 嵌入式 开发工具 数据结构与算法 开发测试 游戏开发 网络协议 系统运维
教程: HTML教程 CSS教程 JavaScript教程 Go语言教程 JQuery教程 VUE教程 VUE3教程 Bootstrap教程 SQL数据库教程 C语言教程 C++教程 Java教程 Python教程 Python3教程 C#教程
数码: 电脑 笔记本 显卡 显示器 固态硬盘 硬盘 耳机 手机 iphone vivo oppo 小米 华为 单反 装机 图拉丁

360图书馆 购物 三丰科技 阅读网 日历 万年历 2024年11日历 -2024/11/24 8:58:01-

图片自动播放器
↓图片自动播放器↓
TxT小说阅读器
↓语音阅读,小说下载,古典文学↓
一键清除垃圾
↓轻轻一点,清除系统垃圾↓
图片批量下载器
↓批量下载图片,美女图库↓
  网站联系: qq:121756557 email:121756557@qq.com  IT数码