CF gorouterアクセスログからElastic APMトレースデータ生成 (Nginxなどにも応用可能)

Cloud Foundry, VMware Tanzu Application Service の gorouter ( https://github.com/cloudfoundry/gorouter ) では、分散トレースのB3・W3C TraceContextヘッダをサポートしているが、トレースデータを外部送信する機能はない。

Elastic APMで分散トレーシングをするとき、gorouterが生成したトレースIDを各アプリケーションが引き継いだ時に、親スパンが存在しないことにより期待通り表示されないことがあった。解決策として、gorouterのアクセスログからトレースデータを生成してElasticsearch(APMインデックス)に送信する。

なお、未検証だが、Nginx, Apacheなど標準機能ではトレースデータを送信できないWebサーバでも同様にアクセスログを元にトレースデータを生成することができると考えられる。

アクセスログ形式

gorouterで記録されるアクセスログは以下の通り。 (TAS 2.11)

<Request Host> - [<Start Date>] "<Request Method> <Request URL> <Request Protocol>" <Status Code> <Bytes Received> <Bytes Sent> "<Referrer>" "<User-Agent>" <Remote Address> <Backend Address> x_forwarded_for:"<X-Forwarded-For>" x_forwarded_proto:"<X-Forwarded-Proto>" vcap_request_id:<X-Vcap-Request-ID> response_time:<Response Time> gorouter_time:<Gorouter Time> app_id:<Application ID> app_index:<Application Index> x_cf_routererror:<X-Cf-RouterError> <Extra Headers>

Troubleshooting router error responses in Cloud Foundry | Cloud Foundry Docs

トレース情報はExtra Headersに含まれる。Zipkinトレーシングを有効にしていると x_b3_traceid, x_b3_spanid, x_b3_parentspanid, b3 が追加されるので、Logstash grokで以下の通りパースする。

match => { "@message" => '%{HOSTNAME:url.domain} - \[%{TIMESTAMP_ISO8601:timestamp}\] "%{NOTSPACE:method} %{NOTSPACE:path} HTTP/%{NUMBER:httpversion}" %{NUMBER:status:int} %{NUMBER:bytes:int} \d+ "%{GREEDYDATA:referer}" "%{GREEDYDATA:user_agent}" ".*" ".*" x_forwarded_for:"%{GREEDYDATA:x_forwarded_for}" .* vcap_request_id:"%{NOTSPACE:vcap_request_id}" response_time:%{NUMBER:response_time:float} gorouter_time:%{NUMBER:gorouter_time:float} app_id:"%{NOTSPACE:app_id}" app_index:"%{NOTSPACE:app_index}" instance_id:"%{NOTSPACE:instance_id}" x_cf_routererror:"%{GREEDYDATA:x_cf_routererror}" x_b3_traceid:"%{NOTSPACE:trace_id}" x_b3_spanid:"%{NOTSPACE:span_id}" x_b3_parentspanid:"%{NOTSPACE:parent_span_id}" b3:"%{NOTSPACE:b3}"' }

Elastic APMトレースindex

Elastic APMでトレースデータを参照可能にするためには、最低限以下のindexでデータを投入する必要がある。(namespace: default の場合)(Elastic Stack ver. 8.8)

data stream	(index)	description
metrics-apm.internal-default	.ds-metrics-apm.internal-default-YYYY.MM.dd-n	apm
traces-apm-default	.ds-traces-apm-default-YYYY.MM.dd-n	apm

それぞれgorouterのアクセスログからdocumentを生成する。

document複製

アクセスログからTraceを生成するために、documentをcloneする。

if ('-' in [parent_span_id]) {
  clone {
    clones => ["clone_parent_span", "clone_parent_metrics"]
    ecs_compatibility => "v1"
  }
}

metrics生成

closeしたdocumentをAPM metrics indexに投入できるよう整形する。

if ('clone_parent_metrics' in [tags]) {
  ruby { code => 'event.set("response_time_us", (event.get("response_time") * 1_000_000).to_i)' }
  ruby { code => 'event.set("transaction.duration.histogram", {"counts" => [1], "values" => [event.get("response_time_us")]})' }
  mutate {
    remove_field => [ "timestamp" ]
  }
  mutate {
    id => "%{span_id}-gorouter-metric"
    add_field => {
      "[agent][name]" => "otlp"
      "[processor][name]" => "metric"
      "[processor][event]" => "metric"
      "[observer][hostname]" => "%{instance_name}"
      "[observer][type]" => "apm-server"
      "[observer][version]" => "8.5.1"
      "[ecs][version]" => "1.12.0"
      "[service][name]" => "gorouter"
      "[transaction][name]" => "%{method} %{url.domain} %{path}"
      "[transaction][type]" => "request"
    }
  }
  prune {
    whitelist_names => [
      "^@timestamp$",
      "^tags$",
      "^agent$",
      "^processor$",
      "^observer$",
      "^ecs$",
      "^service$",
      "^transaction$",
      "^transaction.duration.histogram$"
    ]
  }
}

span生成

同様にAPM traces indexに投入できるよう整形する。

if ('clone_parent_span' in [tags]) {
  ruby { code => 'event.set("timestamp_us", (Time.iso8601(event.get("timestamp")).to_f * 1_000_000 ).round(0))' }
  ruby { code => 'event.set("response_time_us", (event.get("response_time") * 1_000_000).to_i)' }
  mutate {
    remove_field => [ "timestamp" ]
  }
  mutate {
    id => "%{span_id}-gorouter-span"
    add_field => {
      "[agent][name]" => "otlp"
      "[agent][version]" => "unknown"
      "[processor][name]" => "transaction"
      "[processor][event]" => "transaction"
      "[labels][path]" => "%{path}"
      "[observer][hostname]" => "%{instance_name}"
      "[observer][type]" => "apm-server"
      "[observer][version]" => "8.5.1"
      "[trace][id]" => "%{trace_id}"
      "[ecs][version]" => "1.12.0"
      "[service][name]" => "gorouter"
      "[timestamp][us]" => "%{timestamp_us}"
      "[transaction][duration][us]" => "%{response_time_us}"
      "[transaction][name]" => "%{method} %{url.domain} %{path}"
      "[transaction][id]" => "%{span_id}"
      "[transaction][type]" => "request"
      "[transaction][sampled]" => "true"
    }
  }
  mutate {
    convert => {
      "[timestamp][us]" => "integer"
      "[transaction][duration][us]" => "integer"
      "[transaction][sampled]" => "boolean"
    }
  }
  prune {
    whitelist_names => [
      "^@timestamp$",
      "^tags$",
      "^agent$",
      "^processor$",
      "^labels$",
      "^observer$",
      "^trace$",
      "^ecs$",
      "^service$",
      "^timestamp$",
      "^transaction$"
    ]
  }
}

output

それぞれtagを元にelasticseachにoutputする。

output {
  if 'clone_parent_span' in [tags] {
    elasticsearch {
      hosts => ["${ELASTICSEARCH_HOST01}","${ELASTICSEARCH_HOST02}","${ELASTICSEARCH_HOST03}"]
      user => "${ELASTICSEARCH_USER}"
      password => "${ELASTICSEARCH_PASSWORD}"
      ssl_certificate_verification => false
      data_stream => "true"
      data_stream_type => "traces"
      data_stream_dataset => "apm"
      data_stream_namespace => "default"
    }
  } else if 'clone_parent_metrics' in [tags] {
    elasticsearch {
      hosts => ["${ELASTICSEARCH_HOST01}","${ELASTICSEARCH_HOST02}","${ELASTICSEARCH_HOST03}"]
      user => "${ELASTICSEARCH_USER}"
      password => "${ELASTICSEARCH_PASSWORD}"
      ssl_certificate_verification => false
      data_stream => "true"
      data_stream_type => "metrics"
      data_stream_dataset => "apm.internal"
      data_stream_namespace => "default"
    }
  }
}