Discovering vector.dev

agent_vector.toml

# Set global options
data_dir = "/var/lib/vector"

[sources.from_file]
  type = "file"
  include = ["/var/log/nginx/*.log"]

[sinks.to_broker]
  type = "kafka"
  inputs = ["from_file"]
  bootstrap_servers = "kafka:29092"
  topic = "events"
  encoding = "json"

docker-compose.yml

version: "3.7"

services:

  zookeeper:
    image: confluentinc/cp-zookeeper:5.4.0
    hostname: zookeeper
    container_name: zookeeper
    ports:
      - "2181:2181"
    environment:
      ZOOKEEPER_CLIENT_PORT: 2181
      ZOOKEEPER_TICK_TIME: 2000

  kafka:
    image: confluentinc/cp-enterprise-kafka:5.4.0
    hostname: kafka
    container_name: kafka
    depends_on:
      - zookeeper
    ports:
      - "29092:29092"
      - "9092:9092"
    environment:
      KAFKA_BROKER_ID: 1
      KAFKA_ZOOKEEPER_CONNECT: 'zookeeper:2181'
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:29092,PLAINTEXT_HOST://localhost:9092
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
      KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0

  elasticsearch:
    image: docker.elastic.co/elasticsearch/elasticsearch:7.6.2
    container_name: elastic
    environment:
      - ES_JAVA_OPTS=-Xms1g -Xmx1g
      - discovery.type=single-node
      - network.host=_site_, _local_
    ulimits:
      memlock:
        soft: -1
        hard: -1
    ports:
      - 9200:9200
      - 9300:9300

  vector:
    image: timberio/vector:0.8.0-alpine
    container_name: vector
    ports:
      - 8888:8888
    volumes: 
      - $PWD/vector.toml:/etc/vector/vector.toml:ro
    depends_on:
      - kafka
      - elasticsearch

  kibana:
    image: docker.elastic.co/kibana/kibana:7.6.2
    container_name: kibana
    ports:
      - 5601:5601
    depends_on:
      - elasticsearch

  webapp:
    build: ./webapp/
    container_name: webapp
    ports:
      - 80:80
      - 9999:9999

vector.toml

# Set global options
data_dir = "/var/lib/vector"

[sources.from_broker]
  type = "kafka"
  bootstrap_servers = "kafka:29092"
  group_id = "vector-consumer"
  topics = ["events"]

[transforms.json_parser]
  type   = "json_parser"
  inputs = ["from_broker"]
  drop_field = true
  field = "message"

[transforms.log_parser]
  type   = "grok_parser"
  inputs = ["json_parser"]
  pattern = '%{IPORHOST:client} - %{USERNAME:user} \[%{HTTPDATE:timestamp}\] \"%{WORD:verb} %{NOTSPACE:path} HTTP/%{NUMBER}\" %{INT:status} %{NUMBER:bytes} \"%{DATA:referer}\" \"%{DATA:user_agent}\"'
  types.status = "int"
  types.bytes = "int"
  types.timestamp = "timestamp|%d/%b/%Y:%H:%M:%S %z"

[sinks.to_indexer]
  type = "elasticsearch"
  inputs = ["log_parser"]
  healthcheck = false
  host = "http://elasticsearch:9200"

[[tests]]
  name = "test_log_parser"

  [[tests.inputs]]
    insert_at = "json_parser"
    type = "raw"
    value = '172.21.0.1 - - [28/Feb/2020:12:38:46 +0000] "GET /path/to/a HTTP/1.1" 200 46459 "http://localhost/path/to/b" "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_3) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/80.0.3987.122 Safari/537.36" "-"'

  [[tests.outputs]]
    extract_from = "log_parser"

    [[tests.outputs.conditions]]
      type = "check_fields"
      "client.equals" = "172.21.0.1"
      "user.equals" = "-"
      "timestamp.equals"= "2020-02-28T12:38:46Z"
      "verb.equals" = "GET"
      "path.equals" = "/path/to/a"
      "status.equals" = 200
      "bytes.equals" = 46459
      "referer.equals" = "http://localhost/path/to/b"
      "user_agent.equals" = "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_3) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/80.0.3987.122 Safari/537.36"