Elastic からのエージェント移行

Elastic Stack は、オブザーバビリティデータを収集するためのさまざまなエージェントを提供しています。具体的には、次のとおりです。

• Beats family - Filebeat、Metricbeat、Packetbeat など - は、いずれも libbeat ライブラリをベースにしています。これらの Beats は、Lumberjack プロトコル経由で Elasticsearch、Kafka、Redis、または Logstash にデータを送信する ことをサポートしています。
• Elastic Agent は、ログ、メトリクス、トレース を収集できる統合エージェントです。このエージェントは Elastic Fleet Server を通じて一元管理でき、Elasticsearch、Logstash、Kafka、または Redis への出力をサポートしています。
• Elastic は、OpenTelemetry Collector - EDOT のディストリビューションも提供しています。現時点では Fleet Server によってオーケストレーションすることはできませんが、ClickStack への移行を進める場合には、より柔軟でオープンな選択肢になります。

最適な移行パスは、現在使用しているエージェントによって異なります。以降のセクションでは、主要なエージェントタイプごとの移行オプションを説明します。目標は、移行時の負担を最小限に抑え、可能であれば移行期間中も既存のエージェントを継続して利用できるようにすることです。

可能であれば、ログ、メトリクス、トレースの収集はすべて OpenTelemetry (OTel) Collector に移行し、collector を エッジの エージェント ロール としてデプロイすることを推奨します。これはデータ送信の最も効率的な方法であり、アーキテクチャの複雑化やデータ変換を避けられます。

すべてのデータは、ログ、メトリクス、トレース、セッションデータの主要な受け口として機能する OpenTelemetry (OTel) collector インスタンスを介して ClickStack に取り込まれます。このインスタンスには、ClickStack デプロイメントモデルにすでに組み込まれている 場合を除き、collector の公式な ClickStack ディストリビューション を使用することを推奨します。

ユーザーは、言語別 SDK から、またはインフラストラクチャのメトリクスやログを収集するデータ収集エージェント (エージェント ロールの OTel collectors や、Fluentd や Vector などの他の技術) を介して、この collector にデータを送信します。管理された OpenTelemetry pipeline を必要とするチーム向けに、Bindplane は、ClickStack へのネイティブな宛先を備えた OpenTelemetry ネイティブのソリューションを提供しており、テレメトリーの収集、処理、ルーティングを簡素化します。

この collector は、エージェント へのすべての移行手順で利用可能であることを前提としています。

大規模な Beat デプロイメントを利用しているユーザーは、ClickStack への移行後もそれらを継続して使いたい場合があるでしょう。

現時点でこのオプションの動作確認ができているのは Filebeat のみであるため、ログに対してのみ適しています。

Beats エージェントは Elastic Common Schema (ECS) を使用しています。これは現在、ClickStack で使用される OpenTelemetry 仕様への統合が進められています。ただし、これらのスキーマには依然として大きな差異があるため、現時点では、ClickStack にインジェストする前に ECS フォーマットのイベントを OpenTelemetry フォーマットへ変換するのはユーザーの責任となります。

この変換には、Vector の使用を推奨します。Vector は軽量かつ高性能なオブザーバビリティ向けデータパイプラインで、Vector Remap Language (VRL) と呼ばれる強力な変換言語をサポートしています。

Filebeat エージェントが Kafka にデータを送信するよう設定されている場合 (これは Beats でサポートされている出力先です) 、Vector は Kafka からそれらのイベントを取り込み、VRL を使用してスキーマ変換を適用したうえで、OTLP 経由で ClickStack に同梱されている OpenTelemetry Collector に転送できます。

また、Vector は Logstash で使用される Lumberjack プロトコル経由でイベントを受信することもサポートしています。これにより、Beats エージェントはイベントを直接 Vector に送信でき、そこで同じ変換処理を適用した後、OTLP 経由で ClickStack OpenTelemetry collector に転送できます。

以下では、これら 2 つのアーキテクチャを示します。

以下の例では、Lumberjack プロトコル経由で Filebeat からログイベントを受信するように Vector を設定するための最初の手順を示します。また、受信した ECS イベントを OTel 仕様にマッピングするための VRL も示し、その後、それらを OTLP 経由で ClickStack OpenTelemetry collector に送信します。Kafka からイベントを取り込むユーザーは、Vector の Logstash source を Kafka source に置き換えることができます。その他の手順はすべて同じです。

Vector をインストールする

公式インストールガイドに従って Vector をインストールします。

Elastic Stack OTel collector と同じインスタンスにインストールできます。

Vector を本番環境に移行する際は、アーキテクチャとセキュリティに関するベストプラクティスに従ってください。

vectorの設定

VectorをLumberjackプロトコル経由でイベントを受信するように設定し、Logstashインスタンスとして動作させます。これは、Vector用にlogstash ログソースを設定することで実現できます。

sources:
  beats:
    type: logstash
    address: 0.0.0.0:5044
    tls:
      enabled: false  # Set to true if you're using TLS
      # The files below are generated from the steps at https://www.elastic.co/docs/reference/fleet/secure-logstash-connections#generate-logstash-certs
      # crt_file: logstash.crt
      # key_file: logstash.key
      # ca_file: ca.crt
      # verify_certificate: true

イベントはECSフォーマットで受信されます。これらはVector Remap Language (VRL) トランスフォーマーを使用してOpenTelemetryスキーマに変換できます。このトランスフォーマーの設定はシンプルで、スクリプトファイルは別ファイルに格納されます：

transforms:
  remap_filebeat:
    inputs: ["beats"]
    type: "remap"
    file: 'beat_to_otel.vrl'

上記のbeatsソースからイベントを受信する点に注意してください。リマップスクリプトを以下に示します。このスクリプトはログイベントのみで検証済みですが、他のフォーマットへの対応の基礎としても活用できます。

VRL - ECS から OTel への移行

# Define keys to ignore at root level
ignored_keys = ["@metadata"]

# Define resource key prefixes
resource_keys = ["host", "cloud", "agent", "service"]

# Create separate objects for resource and log record fields
resource_obj = {}
log_record_obj = {}

# Copy all non-ignored root keys to appropriate objects
root_keys = keys(.)
for_each(root_keys) -> |_index, key| {
    if !includes(ignored_keys, key) {
        val, err = get(., [key])
        if err == null {
            # Check if this is a resource field
            is_resource = false
            if includes(resource_keys, key) {
                is_resource = true
            }

            # Add to appropriate object
            if is_resource {
                resource_obj = set(resource_obj, [key], val) ?? resource_obj
            } else {
                log_record_obj = set(log_record_obj, [key], val) ?? log_record_obj
            }
        }
    }
}

# Flatten both objects separately
flattened_resources = flatten(resource_obj, separator: ".")
flattened_logs = flatten(log_record_obj, separator: ".")

# Process resource attributes
resource_attributes = []
resource_keys_list = keys(flattened_resources)
for_each(resource_keys_list) -> |_index, field_key| {
    field_value, err = get(flattened_resources, [field_key])
    if err == null && field_value != null {
        attribute, err = {
            "key": field_key,
            "value": {
                "stringValue": to_string(field_value)
            }
        }
        if (err == null) {
            resource_attributes = push(resource_attributes, attribute)
        }
    }
}

# Process log record attributes
log_attributes = []
log_keys_list = keys(flattened_logs)
for_each(log_keys_list) -> |_index, field_key| {
    field_value, err = get(flattened_logs, [field_key])
    if err == null && field_value != null {
        attribute, err = {
            "key": field_key,
            "value": {
                "stringValue": to_string(field_value)
            }
        }
        if (err == null) {
            log_attributes = push(log_attributes, attribute)
        }
    }
}

# Get timestamp for timeUnixNano (convert to nanoseconds)
timestamp_nano = if exists(.@timestamp) {
    to_unix_timestamp!(parse_timestamp!(.@timestamp, format: "%Y-%m-%dT%H:%M:%S%.3fZ"), unit: "nanoseconds")
} else {
    to_unix_timestamp(now(), unit: "nanoseconds")
}

# Get message/body field
body_value = if exists(.message) {
    to_string!(.message)
} else if exists(.body) {
    to_string!(.body)
} else {
    ""
}

# Create the OpenTelemetry structure
. = {
    "resourceLogs": [
        {
            "resource": {
                "attributes": resource_attributes
            },
            "scopeLogs": [
                {
                    "scope": {},
                    "logRecords": [
                        {
                            "timeUnixNano": to_string(timestamp_nano),
                            "severityNumber": 9,
                            "severityText": "info",
                            "body": {
                                "stringValue": body_value
                            },
                            "attributes": log_attributes
                        }
                    ]
                }
            ]
        }
    ]
}

最後に、変換されたイベントをOTLP経由でOpenTelemetry CollectorからClickStackに送信できます。これには、VectorにOTLP sinkを設定する必要があります。このsinkは、remap_filebeatトランスフォームからのイベントを入力として受け取ります。

sinks:
  otlp:
    type: opentelemetry
    inputs: [remap_filebeat] # receives events from a remap transform - see below
    protocol:
      type: http  # Use "grpc" for port 4317
      uri: http://localhost:4318/v1/logs # logs endpoint for the OTel collector 
      method: post
      encoding:
        codec: json
      framing:
        method: newline_delimited
      headers:
        content-type: application/json
        authorization: ${YOUR_INGESTION_API_KEY}

ここで使用する YOUR_INGESTION_API_KEY は ClickStack によって生成されます。このキーは ClickStack の UI (HyperDX) の Team Settings → API Keys から確認できます。

最終的な完全なconfigurationを以下に示します：

sources:
  beats:
    type: logstash
    address: 0.0.0.0:5044
    tls:
      enabled: false  # Set to true if you're using TLS
        #crt_file: /data/elasticsearch-9.0.1/logstash/logstash.crt
        #key_file: /data/elasticsearch-9.0.1/logstash/logstash.key
        #ca_file: /data/elasticsearch-9.0.1/ca/ca.crt
        #verify_certificate: true

transforms:
  remap_filebeat:
    inputs: ["beats"]
    type: "remap"
    file: 'beat_to_otel.vrl'

sinks:
  otlp:
    type: opentelemetry
    inputs: [remap_filebeat]
    protocol:
      type: http  # Use "grpc" for port 4317
      uri: http://localhost:4318/v1/logs
      method: post
      encoding:
        codec: json
      framing:
        method: newline_delimited
      headers:
        content-type: application/json

Filebeat を設定する

既存の Filebeat インストールでは、イベントの送信先が Vector になるよう設定を変更するだけで済みます。そのためには Logstash 出力を設定する必要があります。ここでも、必要に応じて TLS を設定できます。

# ------------------------------ Logstash Output -------------------------------
output.logstash:
  # The Logstash hosts
  hosts: ["localhost:5044"]

  # Optional SSL. By default is off.
  # List of root certificates for HTTPS server verifications
  #ssl.certificate_authorities: ["/etc/pki/root/ca.pem"]

  # Certificate for SSL client authentication
  #ssl.certificate: "/etc/pki/client/cert.pem"

  # Client Certificate Key
  #ssl.key: "/etc/pki/client/cert.key"

Elastic Agent は、複数の Elastic Beats を 1 つのパッケージに統合したものです。この エージェント は Elastic Fleet と統合されており、一元的にオーケストレーションおよび設定を行えます。

Elastic Agent をデプロイしているユーザーには、いくつかの移行方法があります。

• エージェント を設定し、Lumberjack プロトコル経由で Vector エンドポイントに送信します。現時点では、この方法は Elastic Agent のみを使用してログデータを収集しているユーザーについてのみテストされています。 これは Kibana の Fleet UI から一元的に設定できます。
• エージェント を Elastic OpenTelemetry Collector (EDOT) として実行します。Elastic Agent には EDOT Collector が組み込まれており、アプリケーションとインフラストラクチャを一度インストルメントするだけで、複数のベンダーやバックエンドにデータを送信できます。この構成では、EDOT collector がイベントを OTLP 経由で ClickStack OTel collector に転送するように設定するだけです。この方法はすべてのイベントタイプをサポートします。

以下では、これら 2 つのオプションを紹介します。

Vector をインストールして設定する

Filebeat からの移行について記載されている同じ手順で、Vector をインストールして設定します。

Elastic Agent を設定する

Elastic Agent は、Logstash プロトコルの Lumberjack 経由でデータを送信するよう設定する必要があります。これはサポートされているデプロイパターンであり、一元的に設定することも、Fleet を使用せずにデプロイする場合は エージェント の設定ファイル elastic-agent.yaml を使って設定することもできます。

Kibana から一元的に設定するには、Fleet に Output を追加します。

この Output は、エージェント policy で使用できます。これにより、そのポリシーを使用するすべての エージェント は自動的に Vector にデータを送信するようになります。

これには TLS による安全な通信の設定が必要なため、Vector インスタンスを Logstash の役割を担うものとして、ガイド「Configure SSL/TLS for the Logstash output」に従うことを推奨します。

また、Vector の Logstash ソース側でも相互 TLS を設定する必要がある点に注意してください。入力を適切に設定するため、このガイドで生成した秘密鍵と証明書を使用してください。

sources:
  beats:
    type: logstash
    address: 0.0.0.0:5044
    tls:
      enabled: true  # TLS を使用する場合は true に設定します。 
      # 以下のファイルは https://www.elastic.co/docs/reference/fleet/secure-logstash-connections#generate-logstash-certs の手順で生成されます
      crt_file: logstash.crt
      key_file: logstash.key
      ca_file: ca.crt
      verify_certificate: true

Elastic Agent には EDOT Collector が組み込まれており、アプリケーションとインフラストラクチャを一度インストルメントするだけで、複数のベンダーやバックエンドにデータを送信できます。

EDOT collector とともに Elastic Agent を実行するには、Elastic の公式ガイドを参照してください。ガイドにあるように Elastic の endpoint を設定する代わりに、既存の exporters を削除して OTLP 出力を設定し、ClickStack OpenTelemetry collector にデータを送信します。たとえば、exporters の設定は次のようになります。

exporters:
  # Exporter to send logs and metrics to Elasticsearch Managed OTLP Input
  otlp:
    endpoint: localhost:4317
    headers:
      authorization: ${YOUR_INGESTION_API_KEY}
    tls:
      insecure: true

ここでの YOUR_INGESTION_API_KEY は ClickStack で生成されるものです。キーは ClickStack の UI の Team Settings → API Keys で確認できます。

Vector が相互 TLS を使用するように設定されており、証明書と秘密鍵がガイド "Logstash 出力用に SSL/TLS を設定する" の手順で生成されている場合は、otlp エクスポーターもそれに合わせて設定する必要があります。たとえば次のようになります。

exporters:
  # Exporter to send logs and metrics to Elasticsearch Managed OTLP Input
  otlp:
    endpoint: localhost:4317
    headers:
      authorization: ${YOUR_INGESTION_API_KEY}
    tls:
      insecure: false
      ca_file: /path/to/ca.crt
      cert_file: /path/to/client.crt
      key_file: /path/to/client.key

すでに Elastic OpenTelemetry Collector (EDOT) を実行している場合は、OTLP 経由で ClickStack OpenTelemetry collector に送信するようエージェントを再設定するだけです。必要な手順は、上記の Elastic Agent を OpenTelemetry collector として実行する で説明した手順と同じです。この方法は、あらゆるデータ型に使用できます。