not discard the log when the parsing rules cannot be matched

[uglyliu]

How does the vector log collection tool not discard the log when the parsing rules cannot be matched, for example, write to another elasticsearch index when the rules do not match

ERROR

2023-05-22T14:23:39.145659Z ERROR transform{component_kind="transform" component_id=parse_logs component_type=remap component_name=parse_logs}: vector::internal_events::remap: Mapping failed with event. error="function call error for \"parse_json\" at (40:70): unable to parse json: expected value at line 1 column 1" error_type="conversion_failed" stage="processing" internal_log_rate_limit=true

[jszwedko]

Hi @uglyliu !

You can use https://vector.dev/docs/reference/configuration/transforms/remap/#reroute_dropped to route errored events to a dropped output in the transform that you can use in subsequent transforms like <my transform>.dropped.

[uglyliu]

Hi @uglyliu !

You can use https://vector.dev/docs/reference/configuration/transforms/remap/#reroute_dropped to route errored events to a dropped output in the transform that you can use in subsequent transforms like <my transform>.dropped.

Could you please give me some examples，thank u very much

[uglyliu]

Hi @uglyliu !
You can use https://vector.dev/docs/reference/configuration/transforms/remap/#reroute_dropped to route errored events to a dropped output in the transform that you can use in subsequent transforms like <my transform>.dropped.

Could you please give me some examples，thank u very much
Is this configuration right？

data_dir = "/var/lib/vector"
[sources.dummy_logs]
type = "kafka" # required
bootstrap_servers = "127.0.0.1:9092" # required
group_id = "test-vector" # required
key_field = "vector_key" # optional, no default
auto_offset_reset = "latest"
commit_interval_ms = 2000
topics = ["filter-java"] # required

[transforms.filter_ingress_error_log]
type = "filter"
inputs = ["dummy_logs"]
condition = 'contains(string!(.message), "[error]")'

[transforms.parse_ingress_error_log]
type = "remap"
inputs = ["filter_ingress_error_log"]
drop_on_error = false
reroute_dropped = true
  source        = '''
     . = parse_json!(.message)
     .@timestamp = now()
     .timestamp = now()
     .unix_time,error = to_int(format_timestamp!(.timestamp, format: "%s"))
     .utc8_time = .unix_time + 28800
     .currentdate = format_timestamp!(to_timestamp!(.utc8_time), format: "%Y.%m.%d")
     .sink = "vector"
     del(.@metadata);del(.ecs);del(.input);del(.log);del(.prospector);del(.source);del(.fields);del(.input);del(.agent);del(.host);del(.timestamp);del(.unix_time);del(.utc8_time);del(.beat);del(.meta)
'''
[transforms.logger]
type = "remap"
inputs = [ "*.dropped" ]
drop_on_error = true
reroute_dropped = false
source = """
.message = string!(.message)
log(.message, level: "warn")
"""

[sinks.elasticsearch]
 type = "elasticsearch" # required
 inputs = ["parse_ingress_error_log"] # required
 api_version = "auto"
 compression = "none" # optional, default
 endpoints = ["http://10.0.20.8:9200"] # required
 mode = "bulk"
 bulk.action = "index"
 bulk.index = "vector-filternginxerror-{{ .currentdate }}" # optional, default

[sinks.elasticsearch_failed]
 type = "elasticsearch" # required
 inputs = ["logger"] # required
 api_version = "auto"
 compression = "none" # optional, default
 endpoints = ["http://10.0.20.8:9200"] # required
 mode = "bulk"
 bulk.action = "index"
 bulk.index = "vector-notmatch-{{ .currentdate }}" # optional, default