Log event data model validation at source

[bmallya]

I have started using some of the Vector's components and while I am still new to it, I think that they are really great and easy to setup (for the initial configuration).
While I continue to read the documentation and examples provided, I wanted to check/understand if there is a way that we can enforce a data model/schema on the events that are coming in. Basically in the log event coming in from a particular source(s), I want to ensure that the required/mandatory fields are present. If not we drop them (with a log if possible?), and if present we continue to work on the log event either by transforming it or sinking it.

Thanks in Advance.

[spencergilbert]

Hey - we've started planning the work required for this here: #9388

[spencergilbert]

The above . |= parse_json(.message) also seems to throw an error

┌─ :6:7
  │
6 │ .  |= parse_json(.message)
  │       ^^^^^^^^^^^^^^^^^^^^ this expression must resolve to an object
  │
  = see language documentation at https://vrl.dev

Is this because message is a string and merge is expecting an object (from where it is merging)?

As you can see from the function definition, parse_json doesn't guarantee an object being returned as parse_json!("0") is valid and would return 0. This comes up frequently and we've discussed here some syntactic sugar to make this easier.

[JeanMertz]

D'oh 🤦, I made the same mistake I documented 9 days ago in the issue @spencergilbert linked.

So yeah, one solution as described in that issue is to use . |= object!(parse_json!(.message)).

A question here and probably something that might have come up earlier (and answered?) is will this have an overhead if we are doing this for every event that comes in from multiple sources

Any additional computation you perform on incoming events will add "overhead" to your pipelines. Sending events from a source to a sink, without any transforms in-between, is the most efficient way to use Vector, but it's also very limited, and probably isn't sufficient for your use-case.

If you need validation or any other form of event processing, then the above method is currently the most performant way to do it. It's up to you to measure the impact it has on the throughput, and to then determine if that impact is worth the extra safety you get from validating your events.

For most operators, the performance impact is relatively minimal, and the value add is way higher, but I can't answer that for your specific use-case.

[bmallya]

Thanks for the quick response @JeanMertz. Do we need to be doing a merge? Does it not duplicate the contents? In my instance when I send the event through the transform and then to sink (file as ndjson), I seem to be having duplicate

curl -XPOST -d '{"timestamp": 1586960586000,"severity_text": "Error","severity_number": "17","name": "Event 1","body": "Network error","flags":"A","trace_id": "tracexx-001","span_id": "spanyy-001","tenant_id": "0001a","error_code": "EC0001"}' '127.0.0.1:9000'

cat output.txt 
{"body":"Network error","error_code":"EC0001","flags":"A","message":"{\"timestamp\": 1586960586000,\"severity_text\": \"Error\",\"severity_number\": \"17\",\"name\": \"Event 1\",\"body\": \"Network error\",\"flags\":\"A\",\"trace_id\": \"tracexx-001\",\"span_id\": \"spanyy-001\",\"tenant_id\": \"0001a\",\"error_code\": \"EC0001\"}","name":"Event 1","path":"/","severity_number":"17","severity_text":"Error","source_type":"http","span_id":"spanyy-001","tenant_id":"0001a","timestamp":1586960586000,"trace_id":"tracexx-001"}

And so, I was replacing the root of the event with the JSON object using the below,

#Parse the message field as JSON, or error at runtime, merge the resulting
#JSON object into the root of the event
structured, err =  parse_json(.message)
if err != null {
  log("Log event is not in JSON format", level: "error")
  abort
}else {
  #. = merge!(., structured)
  . =  structured
}

curl -XPOST -d '{"timestamp": 1586960586000,"severity_text": "Error","severity_number": "17","name": "Event 1","body": "Network error","flags":"A","trace_id": "tracexx-001","span_id": "spanyy-001","tenant_id": "0001a","error_code": "EC0001"}' '127.0.0.1:9000'

cat output.txt 
{"body":"Network error","error_code":"EC0001","flags":"A","name":"Event 1","severity_number":"17","severity_text":"Error","span_id":"spanyy-001","tenant_id":"0001a","timestamp":1586960586000,"timestamp_unix_nano":"2021-11-16T18:59:14.587984Z","trace_id":"tracexx-001"}

With this remap, I did not see the duplication. But in both cases I did notice that the ordering was changed to be alphabetical. Is there a way to preserve the original order?

[JeanMertz]

Would you be willing to share your Vector config? That way, I can get a sense of what's going on.

[bmallya]

Here the config that I am working with @JeanMertz

[sources.dummy_logs]
type = "http"
address = "0.0.0.0:9000"
encoding = "text"

# Parse Syslog logs
# See the Vector Remap Language reference for more info: https://vrl.dev
[transforms.validator]
type = "remap"
inputs = ["dummy_logs"]
file = "./logEventValidator.vrl"
#drop_on_error = true
#drop_on_abort = true
#reroute_dropped = true


# Print parsed logs to stdout
[sinks.print]
type = "file"
inputs = ["validator"]
compression = "none"
path = "output.txt"
encoding.codec = "ndjson"

#[sinks.con]
#type = "console"
#inputs = ["validator.dropped"]
#encoding.codec = "json"

and the logEventValidator is based on what you had given me

#Validate that the .message  field is a string, or an error at runtime
.message = string!(.message)

#Parse the message field as JSON, or error at runtime, merge the resulting 
#JSON object into the root of the event
structured, err =  parse_json(.message)
if err != null {
  log("Log event is not in JSON format", level: "error")
  abort
}else {
  #. = merge!(., structured)
  . =  structured
}
#This is only for debug.
sev = exists(.severity_number)
log(sev)

#Validate the schema of the event
if !exists(.severity_number) {
	log("missing 'severity_number' field", level: "error")
	abort
}

if !exists(.error_code) {
	log("missing 'error_code' field", level: "error")
	abort
} #Check format of errorcode, ex: starts with

Since I am using
vector 0.17.3 (x86_64-apple-darwin d72c6e7 2021-10-21) now, I have disabled the reroute_dropped flag.