How to use parse_nginx_log in my example

[YangTao0]

Example log:
[
{
"hostname": "myhostname01",
"message": "10.218.15.12 - - [17/Jul/2024:03:40:44 +0800] "POST /udoorkeeper/report/heartbeat HTTP/1.1" 200 2 "-" "remote-zuche-java" "-" 0.009 udoorkeepertest04.xxx.com 10.218.22.11:8080 200 0.009 http - - - -",
"my_ip": "10.218.20.105",
"@timestamp": "2024-07-17T03:40:44+08:00"
},
{
"hostname": "myhostname02",
"message": "10.218.15.192 - - [17/Jul/2024:03:40:44 +0800] "GET /healthCheck HTTP/2.0" 200 19 "-" "Blackbox Exporter/0.23.0" "-" 0.001 zeusnewtest04.xxx.com 10.218.18.139:8080 200 0.003 https - - - -",
"my_ip": "10.218.20.105",
"@timestamp": "2024-07-17T03:40:44+08:00"
}
]
My conf:

data_dir = "/opt/vector/data"
[sources.access_log ]
type = "kafka"
bootstrap_servers = "xxxx”
group_id = "accesslog"
topics = [ "access_log" ]
auto_offset_reset = "largest"
session_timeout_ms = 10000

[transforms.access_parser]
inputs = ["access_log"]
type = "remap"
source = '''
. = parse_json!(.message)
#., err = merge(., parse_nginx_log!(.message, "combined")) ##It doesn't work here
'''
[sinks.print]
type = "console"
inputs = [ "access_parser" ]
encoding.codec = "json"

[jszwedko]

What do you mean by "doesn't work"?

[YangTao0]

The message from kafka like this:

{
    "headers": {},
    "message": "[{\"hostname\":\"lfetools01-test4-ucscompute32-cloudvsp\",\"message\":\"10.223.24.242 - - [17/Jul/2024:22:42:12 +0800] \\\"GET /api/open/check_lfe_config?appname=luckylcpdemo\\u0026envType=PRE HTTP/1.1\\\" 200 38 \\\"-\\\" \\\"Blackbox Exporter/0.23.0\\\" \\\"-\\\" 0.014 lkfetest.example.com 10.218.21.182:80 200 0.014 https\",\"my_ip\":\"10.218.22.244\",\"@timestamp\":\"2024-07-17T22:42:12+08:00\"}]",
    "message_key": "10.223.24.24",
    "offset": 104537455,
    "partition": 3,
    "source_type": "kafka",
    "timestamp": null,
    "topic": "lfeserver-access"
}

When I use:
. = parse_json!(.message)
I get :

2024-07-18T00:35:02.645328Z  INFO vector::app: Loading configs. paths=["vector_lfe.toml"]
2024-07-18T00:35:02.655741Z  INFO vector::topology::running: Running healthchecks.
2024-07-18T00:35:02.656003Z  INFO vector: Vector has started. debug="false" version="0.32.1" arch="x86_64" revision="9965884 2023-08-21 14:52:38.330227446"
2024-07-18T00:35:02.656031Z  INFO vector::app: API is disabled, enable by setting `api.enabled` to `true` and use commands like `vector top`.
2024-07-18T00:35:02.656283Z  INFO vector::topology::builder: Healthcheck passed.
{"@timestamp":"2024-07-17T21:27:00+08:00","hostname":"lfeserver15-test3-ucscompute26-cloudvsp","message":"10.218.40.254 - - [17/Jul/2024:21:26:59 +0800] \"GET /ucarcfcenter-admin/dc/client/ping.json?port=51346&ip
=10.218.40.254&businessLine=lucky&currentBusinessLine=lucky&dataDigest=17211225073913WYc8m5V2jO1XEmMpjS&projectName=luckylbsservice&key=38281eaa16aba8d7505e8eca89c9922a HTTP/1.1\" 200 39 \"-\" \"remote-zuche-java
\" \"-\" 0.006 gaeatest03.example 10.218.21.8:8080 200 0.007 http - - - -","my_ip":"10.218.23.138"}
{"@timestamp":"2024-07-17T21:27:00+08:00","hostname":"lfeserver15-test3-ucscompute26-cloudvsp","message":"10.218.41.19 - - [17/Jul/2024:21:26:59 +0800] \"GET /ucarcfcenter-admin/dc/client/ping.json?port=37850&ip=
10.218.41.19&businessLine=lucky&currentBusinessLine=lucky&dataDigest=1721206675169DpOUTdfrW5LMWznYg8c&projectName=luckydhrrpc&key=lucky_4a2ba3ae4da63c5e867277f6dadc6914 HTTP/1.1\" 200 39 \"-\" \"remote-zuche-java
\" \"-\" 0.005 gaeatest03.example.com 10.218.21.8:8080 200 0.004 http - - - -","my_ip":"10.218.23.138"}

When I use:
. = parse_json!(.message)
., err = merge(., parse_nginx_log!(.message, "combined"))
I get:

2024-07-18T00:39:02.023822Z  INFO vector::app: Loading configs. paths=["vector_lfe.toml"]
2024-07-18T00:39:02.036887Z  INFO vector::topology::running: Running healthchecks.
2024-07-18T00:39:02.037176Z  INFO vector::topology::builder: Healthcheck passed.
2024-07-18T00:39:02.037270Z  INFO vector: Vector has started. debug="false" version="0.32.1" arch="x86_64" revision="9965884 2023-08-21 14:52:38.330227446"
2024-07-18T00:39:02.037304Z  INFO vector::app: API is disabled, enable by setting `api.enabled` to `true` and use commands like `vector top`.
{}
{}
{}
{}
{}
{}

When I use:
. = parse_json!(.message)
. = parse_nginx_log!(.message, "combined")
I get:

2024-07-18T01:37:03.492474Z  INFO vector::app: Loading configs. paths=["vector_lfe.toml"]
2024-07-18T01:37:03.503296Z  INFO vector::topology::running: Running healthchecks.
2024-07-18T01:37:03.503434Z  INFO vector::topology::builder: Healthcheck passed.
2024-07-18T01:37:03.503546Z  INFO vector: Vector has started. debug="false" version="0.32.1" arch="x86_64" revision="9965884 2023-08-21 14:52:38.330227446"
2024-07-18T01:37:03.503571Z  INFO vector::app: API is disabled, enable by setting `api.enabled` to `true` and use commands like `vector top`.
2024-07-18T01:37:03.578869Z ERROR transform{component_kind="transform" component_id=access_parser component_type=remap component_name=access_parser}: vector::internal_events::remap: Mapping failed with event. error="function call error for \"parse_nginx_log\" at (30:68): expected string, got null" error_type="conversion_failed" stage="processing" internal_log_rate_limit=true
2024-07-18T01:37:03.579300Z ERROR transform{component_kind="transform" component_id=access_parser component_type=remap component_name=access_parser}: vector::internal_events::remap: Internal log [Mapping failed with event.] is being suppressed to avoid flooding.

[jszwedko]

I think the issue here is that the message field is an array of messages. I think what you'll want to do is use two remap transforms:

In the first, do:

.message = parse_json!(.message)
. = unnest!(.message)

This will result in events like:

	{
		"headers": {},
		"message": {
			"@timestamp": "2024-07-17T22:42:12+08:00",
			"hostname": "lfetools01-test4-ucscompute32-cloudvsp",
			"message": "10.223.24.242 - - [17/Jul/2024:22:42:12 +0800] \"GET /api/open/check_lfe_config?appname=luckylcpdemo&envType=PRE HTTP/1.1\" 200 38 \"-\" \"Blackbox Exporter/0.23.0\" \"-\" 0.014 lkfetest.example.com 10.218.21.182:80 200 0.014 https",
			"my_ip": "10.218.22.244"
		},
		"message_key": "10.223.24.24",
		"offset": 104537455,
		"partition": 3,
		"source_type": "kafka",
		"timestamp": null,
		"topic": "lfeserver-access"
	}

Being emitted (arrays in remap are emitted as individual events per https://vector.dev/highlights/2021-07-16-remap-multiple/). See https://playground.vrl.dev/?state=eyJwcm9ncmFtIjoiLm1lc3NhZ2UgPSBwYXJzZV9qc29uISgubWVzc2FnZSlcbi4gPSB1bm5lc3QhKC5tZXNzYWdlKSIsImV2ZW50Ijp7ImhlYWRlcnMiOnt9LCJtZXNzYWdlIjoiW3tcImhvc3RuYW1lXCI6XCJsZmV0b29sczAxLXRlc3Q0LXVjc2NvbXB1dGUzMi1jbG91ZHZzcFwiLFwibWVzc2FnZVwiOlwiMTAuMjIzLjI0LjI0MiAtIC0gWzE3L0p1bC8yMDI0OjIyOjQyOjEyICswODAwXSBcXFwiR0VUIC9hcGkvb3Blbi9jaGVja19sZmVfY29uZmlnP2FwcG5hbWU9bHVja3lsY3BkZW1vXFx1MDAyNmVudlR5cGU9UFJFIEhUVFAvMS4xXFxcIiAyMDAgMzggXFxcIi1cXFwiIFxcXCJCbGFja2JveCBFeHBvcnRlci8wLjIzLjBcXFwiIFxcXCItXFxcIiAwLjAxNCBsa2ZldGVzdC5leGFtcGxlLmNvbSAxMC4yMTguMjEuMTgyOjgwIDIwMCAwLjAxNCBodHRwc1wiLFwibXlfaXBcIjpcIjEwLjIxOC4yMi4yNDRcIixcIkB0aW1lc3RhbXBcIjpcIjIwMjQtMDctMTdUMjI6NDI6MTIrMDg6MDBcIn1dIiwibWVzc2FnZV9rZXkiOiIxMC4yMjMuMjQuMjQiLCJvZmZzZXQiOjEwNDUzNzQ1NSwicGFydGl0aW9uIjozLCJzb3VyY2VfdHlwZSI6ImthZmthIiwidGltZXN0YW1wIjpudWxsLCJ0b3BpYyI6ImxmZXNlcnZlci1hY2Nlc3MifSwiaXNfanNvbmwiOmZhbHNlLCJlcnJvciI6bnVsbH0%3D

Then, in a second remap transform do:

. = merge!(., del(.message))
. = merge(., parse_nginx_log!(.message, "combined"))

See https://playground.vrl.dev/?state=eyJwcm9ncmFtIjoiLiA9IG1lcmdlISguLCBkZWwoLm1lc3NhZ2UpKVxuLiA9IG1lcmdlKC4sIHBhcnNlX25naW54X2xvZyEoLm1lc3NhZ2UsIFwiY29tYmluZWRcIikpIiwiZXZlbnQiOnsiaGVhZGVycyI6e30sIm1lc3NhZ2UiOnsiQHRpbWVzdGFtcCI6IjIwMjQtMDctMTdUMjI6NDI6MTIrMDg6MDAiLCJob3N0bmFtZSI6ImxmZXRvb2xzMDEtdGVzdDQtdWNzY29tcHV0ZTMyLWNsb3VkdnNwIiwibWVzc2FnZSI6IjEwLjIyMy4yNC4yNDIgLSAtIFsxNy9KdWwvMjAyNDoyMjo0MjoxMiArMDgwMF0gXCJHRVQgL2FwaS9vcGVuL2NoZWNrX2xmZV9jb25maWc%2FYXBwbmFtZT1sdWNreWxjcGRlbW8mZW52VHlwZT1QUkUgSFRUUC8xLjFcIiAyMDAgMzggXCItXCIgXCJCbGFja2JveCBFeHBvcnRlci8wLjIzLjBcIiBcIi1cIiAwLjAxNCBsa2ZldGVzdC5leGFtcGxlLmNvbSAxMC4yMTguMjEuMTgyOjgwIDIwMCAwLjAxNCBodHRwcyIsIm15X2lwIjoiMTAuMjE4LjIyLjI0NCJ9LCJtZXNzYWdlX2tleSI6IjEwLjIyMy4yNC4yNCIsIm9mZnNldCI6MTA0NTM3NDU1LCJwYXJ0aXRpb24iOjMsInNvdXJjZV90eXBlIjoia2Fma2EiLCJ0aW1lc3RhbXAiOm51bGwsInRvcGljIjoibGZlc2VydmVyLWFjY2VzcyJ9LCJpc19qc29ubCI6ZmFsc2UsImVycm9yIjpudWxsfQ%3D%3D

However, I noticed your input message doesn't parse correctly as an nginx log. Did you customize the format at all in nginx? Otherwise maybe it is a bug in Vector's parser.

[YangTao0]

That's great！thank you for your help.
. = unnest!(.message) it is useful
Yes,we customize the format at all in nginx,so in a second remap transform we will try:

. = merge!(., del(.message))
.msg,err = parse_regex(.message, r'^(?P<client_ip>\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}) - (?P<remote_user>[^ ]*) \[(?P<time_local>[^\]]+)\] "(?P<method>\S+) (?P<request>[^\s]+) HTTP/(?P<http_version>\d+\.\d+)" (?P<status>\d{3}) (?P<body_bytes_sent>\d+) "(?P<http_referer>[^"]*)" "(?P<http_user_agent>[^"]*)" "(?P<http_x_forwarded_for>[^"]*)" (?P<request_time>[\d\.]+) (?P<host>[^\s]+) (?P<upstream_addr>[^\s]+) (?P<upstream_status>\d{3}) (?P<upstream_response_time>[\d\.]+) (?P<scheme>\S+)')
. = merge(., .msg)
del(.msg)
del(.message)

https://playground.vrl.dev/?state=eyJwcm9ncmFtIjoiLiA9IG1lcmdlISguLCBkZWwoLm1lc3NhZ2UpKVxuLm1zZyxlcnIgPSBwYXJzZV9yZWdleCgubWVzc2FnZSwgcideKD9QPGNsaWVudF9pcD5cXGR7MSwzfVxcLlxcZHsxLDN9XFwuXFxkezEsM31cXC5cXGR7MSwzfSkgLSAoP1A8cmVtb3RlX3VzZXI%2BW14gXSopIFxcWyg%2FUDx0aW1lX2xvY2FsPlteXFxdXSspXFxdIFwiKD9QPG1ldGhvZD5cXFMrKSAoP1A8cmVxdWVzdD5bXlxcc10rKSBIVFRQLyg%2FUDxodHRwX3ZlcnNpb24%2BXFxkK1xcLlxcZCspXCIgKD9QPHN0YXR1cz5cXGR7M30pICg%2FUDxib2R5X2J5dGVzX3NlbnQ%2BXFxkKykgXCIoP1A8aHR0cF9yZWZlcmVyPlteXCJdKilcIiBcIig%2FUDxodHRwX3VzZXJfYWdlbnQ%2BW15cIl0qKVwiIFwiKD9QPGh0dHBfeF9mb3J3YXJkZWRfZm9yPlteXCJdKilcIiAoP1A8cmVxdWVzdF90aW1lPltcXGRcXC5dKykgKD9QPGhvc3Q%2BW15cXHNdKykgKD9QPHVwc3RyZWFtX2FkZHI%2BW15cXHNdKykgKD9QPHVwc3RyZWFtX3N0YXR1cz5cXGR7M30pICg%2FUDx1cHN0cmVhbV9yZXNwb25zZV90aW1lPltcXGRcXC5dKykgKD9QPHNjaGVtZT5cXFMrKScpXG4uID0gbWVyZ2UoLiwgLm1zZylcbmRlbCgubXNnKVxuZGVsKC5tZXNzYWdlKVxuIiwiZXZlbnQiOnsiaGVhZGVycyI6e30sIm1lc3NhZ2UiOnsiQHRpbWVzdGFtcCI6IjIwMjQtMDctMTdUMjI6NDI6MTIrMDg6MDAiLCJob3N0bmFtZSI6ImxmZXRvb2xzMDEtdGVzdDQtdWNzY29tcHV0ZTMyLWNsb3VkdnNwIiwibWVzc2FnZSI6IjEwLjIyMy4yNC4yNDIgLSAtIFsxNy9KdWwvMjAyNDoyMjo0MjoxMiArMDgwMF0gXCJHRVQgL2FwaS9vcGVuL2NoZWNrX2xmZV9jb25maWc%2FYXBwbmFtZT1sdWNreWxjcGRlbW8mZW52VHlwZT1QUkUgSFRUUC8xLjFcIiAyMDAgMzggXCItXCIgXCJCbGFja2JveCBFeHBvcnRlci8wLjIzLjBcIiBcIi1cIiAwLjAxNCBsa2ZldGVzdC5leGFtcGxlLmNvbSAxMC4yMTguMjEuMTgyOjgwIDIwMCAwLjAxNCBodHRwcyIsIm15X2lwIjoiMTAuMjE4LjIyLjI0NCJ9LCJtZXNzYWdlX2tleSI6IjEwLjIyMy4yNC4yNCIsIm9mZnNldCI6MTA0NTM3NDU1LCJwYXJ0aXRpb24iOjMsInNvdXJjZV90eXBlIjoia2Fma2EiLCJ0aW1lc3RhbXAiOm51bGwsInRvcGljIjoibGZlc2VydmVyLWFjY2VzcyJ9LCJpc19qc29ubCI6ZmFsc2UsImVycm9yIjpudWxsfQ%3D%3D

[jszwedko]

Makes sense, if the log format is customized you'll need to use regex to parse it. parse_nginx_log only works with the standard formats. Glad to see it seems like you figured it out!