Vectorを利用してkafkaからDatadogへログを転送してみた 

k8sベースのアプリケーションログをkafkaにpublishして、Vectorを介してDatadogへ転送する構成を試していた際に一部ハマりそうなポイントがあったため備忘録として記載

## Vectorとは
[公式ドキュメント](https://vector.dev/docs/about/what-is-vector/)にも記載されていますが、ログ/メトリクス/トレースデータの収集、加工、他ツールへの連携を行うためのデータパイプラインツールとなります。

![Screenshot 2023-12-02 at 22-23-04 Data model.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/110280/4164eb89-4c7d-20af-bf6f-dd4162e24312.png)


## やりたいこと
- k8s上で稼働しているアプリログをfluent-bit経由でkafkaへpublish
- kafkaに溜まったログに対して、vectorを経由してDatadogへ転送する

datadog agentを介して直接Datadogへログを転送することも可能ですが、
1\. ログを基盤側でバッファリングしておくことで、Datadogに障害が起きた際のログの可用性を担保したかった
2\. 基盤側で強制的に付与したいDD_TAGSがあった
ため上記の流れでログを転送していました。

## 構成
- application podはEKSクラスタ上に構築
- ログはstdoutへ出力し、fluent bit経由でMSK kafkaへpublishされる
- MSK kafkaに溜まったログをvectorがsubscribe
- Vectorのsink設定として`datadog_logs`を指定し、Datadogへログを転送する

※Amazon MSK Serverlessの利用を試みましたが、[librdkafka](https://github.com/confluentinc/librdkafka/blob/master/CONFIGURATION.md)が[IAMアクセスコントロール](https://docs.aws.amazon.com/ja_jp/msk/latest/developerguide/iam-access-control.html)に対応していないため、断念しました。
![vector.drawio.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/110280/62e3d9ea-6224-8b37-7984-4ae2cb702d20.png)



## kafka構築
terraformを利用して、kafkaクラスタとconfigurationの設定を定義
(プライベートサブネットとセキュリティグループは作成済みの前提)

``` kafka.tf
locals {
    service_name = "vector-log-poc"
}

resource "aws_msk_cluster" "kafka" {
  cluster_name           = "${local.service_name}-cluster"
  kafka_version          = "3.5.1"
  number_of_broker_nodes = 3
  broker_node_group_info {
    instance_type = "kafka.t3.small"
    storage_info {
      ebs_storage_info {
        volume_size = 10 //GB
      }
    }
    client_subnets = [
      aws_subnet.private_ap_northeast_1a.id, //事前に作成しておく
      aws_subnet.private_ap_northeast_1b.id, //事前に作成しておく
      aws_subnet.private_ap_northeast_1d.id, //事前に作成しておく
    ]
    security_groups = [aws_security_group.kafka.id]  //事前に作成しておく
  }

  client_authentication {
    unauthenticated = true

    sasl {
      iam   = false
      scram = false
    }

    tls {}
  }
  encryption_info {
    encryption_in_transit {
      client_broker = "TLS_PLAINTEXT"
    }
  }

  configuration_info {
    arn      = aws_msk_configuration.kafka.arn
    revision = aws_msk_configuration.kafka.latest_revision
  }

  tags = {
    Name = "${local.service_name}-cluster"
  }
}

resource "aws_msk_configuration" "kafka" {
  kafka_versions = ["3.5.1"]
  name           = "${local.service_name}-cluster"

  server_properties = <<PROPERTIES
auto.create.topics.enable = true
delete.topic.enable = true
log.retention.hours = 24
PROPERTIES
}
```

## Vector構築
[Helm chart](https://artifacthub.io/packages/helm/vector/vector)を利用してAggrigatorタイプのVector podをデプロイする

[Quick start](https://artifacthub.io/packages/helm/vector/vector#quick-start)に従ってhelm chartをinstallする
```
helm install --name my-vector vector/vector
```

次に`values.yaml`を作成する

```values.yaml
role: Aggregator

customConfig:
  api:
    enabled: true
    address: 127.0.0.1:8686
    playground: false

  sources:
    kafka:
      type: kafka
      bootstrap_servers: b-1.xxx:9092,b-2.xxx:9092,b-3.xxx:9092
      group_id: log-consumer-group
      topics: 
        - fluent-bit-logs # fluent bitがログを送っているtopicを指定する

    vector_logs:
      type: internal_logs

  transforms:
    parse_kafka:
      type: remap
      inputs: 
        - kafka
      source: >-
        .ddsource = "kafka"

        .ddtags = "env:dev,environment:dev,region:ap-northeast-1"

        .service = "vector-log-poc"

  sinks:
    ddlog:
      type: datadog_logs
      inputs: 
        - parse_kafka
        - vector_logs
      default_api_key: xxx # Datadog側で作成したapi keyを指定する
      site: ap1.datadoghq.com # 利用中のDatadogサイトを指定する https://docs.datadoghq.com/ja/getting_started/site/

autoscaling:
  enabled: true
  minReplicas: 1
  maxReplicas: 3

podDisruptionBudget:
  enabled: true
  minAvailable: 1
```

helm upgradeで`values.yaml`の変更を反映する
```
helm upgrade -f values.yaml my-vector vector/vector
```

## Datadog上でのログの確認
14日間限定で[無料トライアル](https://www.datadoghq.com/ja/free-datadog-trial/)も提供されています。もし、コストをかけずに試したい場合はこちらを利用して確認することも可能です。
以下の通り、ログがDatadogにて可視化できていました。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/110280/4b1de796-c3e2-f90c-1073-b364fb753bbc.png)


## Vectorの設定でハマったポイント
当初transformsの設定が抜けており、[body data](https://docs.datadoghq.com/ja/api/latest/logs/)に必要なmessageやservice、ddtags等が渡せていませんでした。
Vector側で設定関連のエラーが出ていないものの、Datadogへログが転送できていない(ログの表示ができていない)状態となり、切り分けに時間がかかってしまいました。

対応策として、上記の`values.yaml`にある通り、transformsにて.ddsourceの設定を追加してdatadogのlog pipelineにおいてログのパースを行うようにしました。

※参考
https://vector.dev/docs/reference/configuration/sinks/datadog_logs/#attributes

## まとめ
初めてvectorを触ってみましたが、source/transforms/sinkとログ処理の一連の設定を簡易に定義でき、初学者にも扱いやすいツールだなと感じました。
またドキュメントが充実しているため、設定に関しての不明点があれば公式ドキュメントを見れば良さそうでした。