流程說明

在這里插入圖片描述

應用APP生產日志，用來記錄用戶的操作
通過Filebeat讀取日志檔案中的內容，并且將內容發送給Logstash，原因是需要對內容做處理
Logstash接收到內容后，進行處理，如分割操作，然后將內容發送到Elasticsearch中
Kibana會讀取Elasticsearch中的資料，并且在Kibana中進行設計Dashboard，最后進行展示

Docker 部署 elk

先決條件

修改系統記憶體內核引數

vim /etc/sysctl.conf
vm.max_map_count=262144

運行記憶體要大于4G

pull + run

docker pull sebp/elk

docker run -p 5601:5601 -p 9200:9200 -p 5044:5044 -itd --name elk sebp/elk

修改logstash組態檔

 1、進入elk容器
docker exec -it elk /bin/bash
 2、修改組態檔 改為如下
vi /etc/logstash/conf.d/02-beats-input.conf

這里的logstash接收到filebeat發來的日志，將message切割等資料處理操作，之后再將處理好的資料發送給elasticsearch

input {
        beats {
                port => "5044"
        }
}
filter {
        mutate {
                split => {"message"=>"|"}
        }
        mutate {
                add_field => {
                "userId" => "%{[message][1]}"
                "visit" => "%{[message][2]}"
                "date" => "%{[message][3]}"
                }
        }
        mutate {
                convert => {
                "userId" => "integer"
                "visit" => "string"
                "date" => "string"
                }
        }
}
output {
	elasticsearch {
		hosts => [ "192.168.135.10:9200"]
		index => ["logstash-wxf-demo-shopping"]
	}
}

僅保留02-beats-input.conf，別的要么改名要么洗掉，我這里是改名

root@6643ff42f735:/etc/logstash/conf.d# ls
02-beats-input.conf  10-syslog.conf.bak  11-nginx.conf.bak  30-output.conf.bak

退出容器

exit

開啟服務

撰寫服務

我這里用go撰寫的一個程式模擬

package main

import (
	"fmt"
	"math/rand"
	"os"
	"strconv"
	"time"
)

func main() {
	for i := 0; i < 5; i++ {
		go LogCal()
		time.Sleep(time.Second)
	}
	select {}
}

func LogCal() {
	for {
		businessList := []string{"瀏覽頁面", "評論商品", "加入收藏", "加入購物車", "提交訂單", "使用優惠券", "領取優惠券", "搜索", "查看訂單"}

		t := time.Now()
		rand.Seed(time.Now().UnixNano())

		userId := rand.Intn(9000000) + 1000000
		visit := businessList[rand.Intn(9)]
		date := fmt.Sprintf("%d-%d-%d %d:%d:%d", t.Year(), t.Month(), t.Day(), t.Hour(), t.Minute(), t.Second())

		time.Sleep(time.Second * time.Duration(rand.Intn(5)+5))

		result := "DAU|" + strconv.Itoa(userId) + "|" + visit + "|" + date + "\n"
		LogInfo(result)
	}
}

func LogInfo(result string) {
	t := time.Now()
	orderTime := fmt.Sprintf("%d-%d-%d %d:%d:%d", t.Year(), t.Month(), t.Day(), t.Hour(), t.Minute(), t.Second())
	pre := "[INFO] " + orderTime + " [Demo] - "
	result = pre + result

	filePath := "/usr/local/Demo/abc.log"
	file, err := os.OpenFile(filePath, os.O_APPEND|os.O_WRONLY|os.O_CREATE, 0666)
	if err != nil {
		fmt.Printf("open file err=%v\n", err)
		return
	}
	defer file.Close()
	file.WriteString(result)
}

配置filebeat的yml檔案

vi Demo.yml

填寫如下內容，將收集到的日志發送給logstash

filebeat.inputs:
- type: log
  enabled: true
  paths:
    - /usr/local/Demo/*.log
  fields:
    form: Demo-Shopping
  fields_under_root: true
setup.template.settings:
  index.number_of_shards: 1
  index.number_of_replica: 0
output.logstash:
  hosts: ["192.168.135.10:5044"]