elk stack部署自动化日志收集分析平台-阿里云开发者社区

elk stack部署自动化日志收集分析平台

2024-01-11 80

版权

本文内容由阿里云实名注册用户自发贡献，版权归原作者所有，阿里云开发者社区不拥有其著作权，亦不承担相应法律责任。具体规则请查看《阿里云开发者社区用户服务协议》和《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容，填写侵权投诉表单进行举报，一经查实，本社区将立刻删除涉嫌侵权内容。

本文涉及的产品

检索分析服务 Elasticsearch 版，2核4GB开发者规格 1个月

简介： elk stack部署自动化日志收集分析平台

流程说明

应用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"]