Vector 调研测试同步至 ElasticSearch

前言

继续接着《Vector 调研测试之环境安装》，本文旨在记录对Vector写入Elasticsearch的功能性测试是否能够满足广泛的需求，以期能够为他人提供有用的参考和帮助。

准确工作

• 新建 Kafka Topic

  kafka-topics -–create -–bootstrap-server kafka-1:9092 -–replication-factor 1 -–partitions 6 -–topic bigdata-data-v1

• 查看 Kafka Topic

  	kafka-topics --list --bootstrap-server kafka-1:9092

• 测试数据

  • 测试数据来源

    https://www.kaggle.com/datasets/eliasdabbas/web-server-access-logs

  • 测试数据Demo

    	54.36.149.41 - - [22/Jan/2019:03:56:14 +0330] "GET /filter/27|13%20%D9%85%DA%AF%D8%A7%D9%BE%DB%8C%DA%A9%D8%B3%D9%84,27|%DA%A9%D9%85%D8%AA%D8%B1%20%D8%A7%D8%B2%205%20%D9%85%DA%AF%D8%A7%D9%BE%DB%8C%DA%A9%D8%B3%D9%84,p53 HTTP/1.1" 200 30577 "-" "Mozilla/5.0 (compatible; AhrefsBot/6.1; +http://ahrefs.com/robot/)" "-"
    31.56.96.51 - - [22/Jan/2019:03:56:16 +0330] "GET /image/60844/productModel/200x200 HTTP/1.1" 200 5667 "https://www.zanbil.ir/m/filter/b113" "Mozilla/5.0 (Linux; Android 6.0; ALE-L21 Build/HuaweiALE-L21) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/66.0.3359.158 Mobile Safari/537.36" "-"
    31.56.96.51 - - [22/Jan/2019:03:56:16 +0330] "GET /image/61474/productModel/200x200 HTTP/1.1" 200 5379 "https://www.zanbil.ir/m/filter/b113" "Mozilla/5.0 (Linux; Android 6.0; ALE-L21 Build/HuaweiALE-L21) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/66.0.3359.158 Mobile Safari/537.36" "-"
    40.77.167.129 - - [22/Jan/2019:03:56:17 +0330] "GET /image/14925/productModel/100x100 HTTP/1.1" 200 1696 "-" "Mozilla/5.0 (compatible; bingbot/2.0; +http://www.bing.com/bingbot.htm)" "-"
    91.99.72.15 - - [22/Jan/2019:03:56:17 +0330] "GET /product/31893/62100/%D8%B3%D8%B4%D9%88%D8%A7%D8%B1-%D8%AE%D8%A7%D9%86%DA%AF%DB%8C-%D9%BE%D8%B1%D9%86%D8%B3%D9%84%DB%8C-%D9%85%D8%AF%D9%84-PR257AT HTTP/1.1" 200 41483 "-" "Mozilla/5.0 (Windows NT 6.2; Win64; x64; rv:16.0)Gecko/16.0 Firefox/16.0" "-"
    40.77.167.129 - - [22/Jan/2019:03:56:17 +0330] "GET /image/23488/productModel/150x150 HTTP/1.1" 200 2654 "-" "Mozilla/5.0 (compatible; bingbot/2.0; +http://www.bing.com/bingbot.htm)" "-"

• 写入数据

  	kafka-console-producer --broker-list kafka-1:9092 --topic bigdata-data-v1 < ./access.log

• 消费数据查看

  	kafka-console-consumer --bootstrap-server kakfa-1:9092 --topic bigdata-data-v1 --from-beginning

写入ES

• 编写 Vector 的配置文件

  vim /opt/vector/config/es.toml

  [api]
  enabled = true
  address = "0.0.0.0:8686"
  
  [sources.src_kafka]
  type = "kafka"
  bootstrap_servers = "192.168.31.37:9092"
  auto_offset_reset = "earliest"
  group_id = "consumer-bigdata-es"
  topics = ["bigdata-data-v1"]
  decoding.codec = "bytes"
  
  [transforms.transform_log]
  type = "remap"
  inputs = ["src_kafka"]
  source = """
  . = parse_nginx_log!(string!(.message), "combined")
  .date, err = format_timestamp(.timestamp, "%Y-%m-%d")
  """
  
  [sinks.sink_out]
  inputs = ["transform_log"]
  type = "console"
  encoding.codec = "json"

• 运行测试

  	/opt/vector/bin/vector -c /opt/vector/config/es.toml

  在 console 查看输出的格式是否正确

• 在ES上创建模板

  PUT /_template/access-log_template

  {
    "template": "access-log_*",
    "settings": {
      "number_of_shards": 12
    },
    "mappings": {
      "properties": {
        "agent": {
          "type": "text"
        },
        "client": {
          "type": "text"
        },
        "date": {
          "type": "text"
        },
        "referer": {
          "type": "text"
        },
        "request": {
          "type": "text"
        },
        "size": {
          "type": "integer"
        },
        "status": {
          "type": "integer"
        },
        "timestamp": {
          "type": "text"
        }
      }
    },
    "aliases": {
      "access-log": {}
    }
  }

• 编写配置文件设置保存到ES

  [api]
  enabled = true
  address = "0.0.0.0:8686"
  
  [sources.src_kafka]
  type = "kafka"
  bootstrap_servers = "192.168.31.37:9092"
  auto_offset_reset = "beginning"
  group_id = "consumer-bigdata-es"
  topics = ["bigdata-data-v1"]
  decoding.codec = "bytes"
  
  [transforms.transform_log]
  type = "remap"
  inputs = ["src_kafka"]
  source = """
  . = parse_nginx_log!(string!(.message), "combined")
  .date, err = format_timestamp(.timestamp, "%Y-%m-%d")
  """
  
  [sinks.sink_es]
  type = "elasticsearch"
  inputs = ["transform_log"]
  endpoints = ["http://192.168.31.37:9200"]
  api_version = "v7"
  id_key = "_id"
  mode = "bulk"
  bulk.index = "access-log_{{ date }}"

• 运行

  	/opt/vector/bin/vector -c /opt/vector/config/es.toml

• 查看数据是否正常写入

  如图所示，索引和别名都成功创建。

  抽查索引的数据，发现也是正常的。

结论

经过部署测试，Vector 用 VRL 解析转换再写入 ElasticSearch 在功能上是完全OK的。