救救家長：疫情封控下packetbeat+kafka+ES套件監控青少年上網行為
緣起
疫情，不少孩子封控在家，需要上網課，但是老是抑制不住地去打遊戲或看影片。 朋友圈裡面，某位技術大牛這麼描述疫情封控期間，他與孩子的居家“戰爭”：
孩子上網課已經一個多月了，孩子因為愛玩遊戲愛看B站，與我鬥智鬥勇好幾回，目前戰鬥情況如下： 上課時間玩手機遊戲 ~ 沒收手機 在電腦上裝手機模擬器 繼續玩手機遊戲 ~ 解除安裝模擬器 在電腦上看B站 ~ 設定host檔案遮蔽B站域名 在電腦上看芒果TV ~ 繼續設定遮蔽域名 繼續安裝手機模擬器、找到host檔案刪除遮蔽，看B站玩遊戲 ~ 被打，被解除安裝各種軟體，被警告再發現就換Linux作業系統 解封后，先買個企業級路由器管控起來... 或者再裝個攝像頭再加上AI人體姿態識別？😭
因本文是技術文章，在這裡咱暫先不討論教育之道，先提供一個技術方案，解決家長的燃眉之急，不用企業級路由器，也不用AI人體姿態識別。 我的方案是：packetbeat+kafka+ES套件，以大資料視覺化方式監控孩子。 這個方案的特點： + 一是抓住了矛盾的主要方面，疫情封控期間，孩子因網課有正當理由使用電腦等裝置，監控重點是上網行為，尤其在上課時間段內不能打遊戲看影片 + 二是最大程度利用了現有裝置，不需要額外添置監控裝置
架構設計
架構設計上有個難點，我的方案中考慮了異地監控的情況，比如孩子在杭州上課，我在新加坡工作，考慮跨國網路的不穩定性，我在常規的packetbeat和ES叢集之間，加了一個準備放在阿里雲上的kafka訊息佇列，這樣就不用穿透內網，也不怕網路不穩丟訊息了，遠在新加坡Shopee雲的監控主機可以用非同步方式讀取kafka。
效果展示
監控實圖1：
監控實圖2：
孩子訪問了啥網站，啥時候訪問，網路地址、流量情況一應俱全：） 雖然和教育的目的相背，但是國內就好這一口不是😄 IT前輩大牛目前還在用物理手段，我已經升級到 雲+大資料了，我要把這個專案開源，造福中國的父母，嘿嘿
安裝執行
注意版本問題！如下未特別說明的，都採用7.8.1版本，版本不匹配會有坑。
packetbeat
總體過程是：官網下載，unzip解壓到本地目錄，配置yml檔案，後啟動執行
10360* sudo ./packetbeat -e -c packetbeat.yml
10362 cd packetbeat-7.8.1-darwin-x86_64
10363* sudo ./packetbeat setup --dashboards
10364* sudo ./packetbeat -e -c packetbeat.yml
配置packetbeat.yml 輸出到Kafka
```
---------------------------- Kafka -------------------------------------------
output.kafka: hosts: ["localhost:9092"] topic: packetbeat required_acks: 1 ```
Kafka
採用docker-compose方式安裝
version: '3'
services:
zookeeper:
image: wurstmeister/zookeeper
container_name: zookeeper
ports:
- 2181:2181
environment:
ZOO_MY_ID: 1
kafka:
image: wurstmeister/kafka
container_name: kafka
ports:
- 9092:9092
environment:
KAFKA_ADVERTISED_HOST_NAME: 192.168.18.37
KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
volumes:
- /var/run/docker.sock:/var/run/docker.sock
kafka_manager:
image: sheepkiller/kafka-manager
container_name: kafka_manager
ports:
- 9000:9000
environment:
ZK_HOSTS: "zookeeper:2181"
APPLICATION_SECRET: "random-secret"
command: -Dpidfile.path=/dev/null
消費端程式碼如下供參考，也可以用logstash直接拉取
```
package main
import ( "fmt" "github.com/Shopify/sarama" )
const TOPIC = "packetbeat" func main() { consumer, err := sarama.NewConsumer([]string{"127.0.0.1:9092"}, nil) if err != nil { fmt.Printf("fail to start consumer, err:%v\n", err) return } partitionList, err := consumer.Partitions(TOPIC) // 根據topic取到所有的分割槽 if err != nil { fmt.Printf("fail to get list of partition:err%v\n", err) return } fmt.Println(partitionList) for partition := range partitionList { // 遍歷所有的分割槽 // 針對每個分割槽建立一個對應的分割槽消費者 pc, err := consumer.ConsumePartition(TOPIC, int32(partition), sarama.OffsetOldest) fmt.Printf("---%+v\n", pc) if err != nil { fmt.Printf("failed to start consumer for partition %d,err:%v\n", partition, err) return } defer pc.AsyncClose() // 非同步從每個分割槽消費資訊 i := 0 for msg := range pc.Messages() { fmt.Printf("Partition:%d Offset:%d Key:%v Value:%v\n", msg.Partition, msg.Offset, string(msg.Key), string(msg.Value)) i++ if i > 5 { return } } } } ```
logstash
安裝過程參考官網說明，配置 ``` input { kafka { bootstrap_servers => "localhost:59471" topics => ["packetbeat"] } }
output { elasticsearch { hosts => ["localhost:9200"] index => "packetbeat-7.8.1-2022.05.01-000001" document_type => "_doc" } stdout { codec => rubydebug } } ```
ES叢集+kibana
也是docker-compose安裝 ``` services:
elasticsearch01: build: context: elasticsearch/ args: ELK_VERSION: ${ELK_VERSION:-7.8.1} volumes: - type: bind source: ./elasticsearch/elasticsearch.yml target: /usr/share/elasticsearch/config/elasticsearch.yml read_only: true - ./data_elasticsearch01:/usr/share/elasticsearch/data ports: - "9200:9200" environment: - node.name=elasticsearch01 - discovery.seed_hosts=elasticsearch02 - cluster.initial_master_nodes=elasticsearch01,elasticsearch02 - bootstrap.memory_lock=true #- "ES_JAVA_OPTS=-Xms${ES_HEAP_SIZE:-2g} -Xmx${ES_HEAP_SIZE:-2g}" ulimits: memlock: soft: -1 hard: -1
elasticsearch02: build: context: elasticsearch/ args: ELK_VERSION: ${ELK_VERSION:-7.8.1} volumes: - type: bind source: ./elasticsearch/elasticsearch.yml target: /usr/share/elasticsearch/config/elasticsearch.yml read_only: true - ./data_elasticsearch02:/usr/share/elasticsearch/data environment: - node.name=elasticsearch02 - discovery.seed_hosts=elasticsearch01 - cluster.initial_master_nodes=elasticsearch01,elasticsearch02 - bootstrap.memory_lock=true #- "ES_JAVA_OPTS=-Xms${ES_HEAP_SIZE:-2g} -Xmx${ES_HEAP_SIZE:-2g}" ulimits: memlock: soft: -1 hard: -1
kibana: build: context: kibana/ args: ELK_VERSION: ${ELK_VERSION:-7.8.1} volumes: - type: bind source: ./kibana/kibana.yml target: /usr/share/kibana/config/kibana.yml read_only: true ports: - "5601:5601" environment: - elasticsearch.hosts=["http://elasticsearch01:9200"] ```