vector.toml

# User info table (all fields are string, thus no schema)
[enrichment_tables.users]
type = "file"

[enrichment_tables.users.file]
path = "/var/lib/vector/data/users.csv"
encoding = { type = "csv" }

# Error codes table (with specified schema)
[enrichment_tables.codes]
type = "file"

[enrichment_tables.codes.file]
path = "/var/lib/vector/data/codes.csv"
encoding = { type = "csv" }

[enrichment_tables.codes.schema]
code = "integer"
tag = "string"
message = "string"

# Generate user info messages
[sources.user_info_messages]
type = "demo_logs"
format = "shuffle"
lines = [
  '{"first_name": "Karolann", "last_name": "Wiza", "message": "Hello World"}',
  '{"first_name": "Nick", "last_name": "Adams", "message": "Hello World"}',
  '{"first_name": "Hilbert", "last_name": "Stark", "message": "Hello World"}',
  '{"first_name": "Irwin", "last_name": "Green", "message": "Hello World"}',
  '{"first_name": "Katlyn", "last_name": "Steuber", "message": "Hello World"}',
]
interval = 2

# Generate coded error messages
[sources.coded_error_messages]
type = "demo_logs"
format = "shuffle"
lines = [
  '{"code":1,"device_id":"e5ad503d","timestamp":"2021-10-18T15:35:09.158139Z"}',
  '{"code":2,"device_id":"a5b2401e","timestamp":"2021-10-18T15:35:28.517210Z"}',
  '{"code":3,"device_id":"b48f41aa","timestamp":"2021-10-18T15:35:37.846783Z"}'
]
interval = 2

[transforms.remap_user_info]
type = "remap"
inputs = ["user_info_messages"]
source = """
. = object!(parse_json(.message) ?? {})
. |= get_enrichment_table_record("users", { "last_name": .last_name, "first_name": .first_name }) ?? {}
"""

[transforms.remap_coded_errors]
type = "remap"
inputs = ["coded_error_messages"]
source = """
. = object!(parse_json(.message) ?? {})

row, err = get_enrichment_table_record("codes", { "code": del(.code) })

if err != null {
  log(err, level: "error")
} else {
  . |= {"message": row.message, "tag": row.tag}
}
"""

[sinks.console]
type = "console"
inputs = ["remap_*"]
encoding = { codec = "json" }