Multiple data sources and destinations

[petermp79]

I wonder whether an instance of vector route data from multiple data sources, e.g. log, app log, metric, ..., to different indices on the same ES cluster? If not, what is the best practice for this use case?

Please advise, appreciated.

[jszwedko]

Hi @petermp79 !

Yes, the bulk.index field on the elasticsearch sink supports Vector's templating feature so that the index can be set to a field from the event. I'd recommend using remap transforms to tag the events with the index you want them to go and referencing that field in the template.

The example on the templating page is for the aws_s3 sink but should give you enough to go on to adapt to the elasticsearch sink.

[petermp79]

HI, Jesse:

Thanks for the hint regarding destination indices.

Another issue is whether vector can pick up data from multiple sources, e.g. log, metric, app logs, etc. I tried the below sample configuration,

[sources.journal]
type = "journald"

[sources.syslog]
type = "file"
include = [ "/home/peter.pham/support/support-2022-10-19T03-58-15/dm-140611-dmn0/syslog/syslog*" ]
read_from = "beginning"

Parse Syslog logs

See the Vector Remap Language reference for more info: https://vrl.dev

[transforms.parse_syslogs]
type = "remap"
inputs = ["syslog"]
source = '''
. = parse_syslog!(string!(.message))
'''

[sinks.es_syslog]
type = "elasticsearch"
inputs = ["parse_syslogs"]
suppress_type_name = true
mode = "bulk"
bulk.action = "index"
bulk.index = "dmn0-syslog-%Y-%m-%d"
tls.ca_file = "/etc/elasticsearch/certs/http_ca.crt"
endpoints = [ "https://elastic:ClX9b*W07OSQ0JuRlf9R@105.133.6.71:9200" ] # local or external host

[sinks.es_journal]
type = "elasticsearch"
inputs = ["journal"]
suppress_type_name = true
mode = "bulk"
bulk.action = "index"
bulk.index = "dmn0-journal-%Y-%m-%d"
tls.ca_file = "/etc/elasticsearch/certs/http_ca.crt"
endpoints = [ "https://elastic:ClX9b*W07OSQ0JuRlf9R@105.133.6.71:9200" ]

But it appears only journal data is shown up on ES cluster. Did I miss something?
Maybe, I should merge the two ES sinks and apply the indexing scheme you suggested.

Is this correct?

[petermp79]

Thanks. I give it a try then.

[petermp79]

Hi, Jesse:

In the below example, I attempted to read data from 2 source files, and output them to console, but it seems only one (syslog) is read and processed:

[sources.journal_file]

type = "journald"

type = "file"
include = [ "./journal-sample" ]
read_from = "beginning"

Add index to journal json data

index = "dmn0-journal"

[transforms.parse_journal]
type = "remap"
inputs = ["journal_file"]
source = '''
'''

Get data from syslog

[sources.syslog_file]
type = "file"
include = [ "./syslog-sample" ]
read_from = "beginning"

Parse Syslog logs

Add syslog index to json data

index = "dmn0-syslog"

[transforms.parse_syslog]
type = "remap"
inputs = ["syslog_file"]
source = '''
. = parse_syslog!(string!(.message))
.source_type = "syslog"
'''

Print parsed logs to stdout

[sinks.print]
type = "console"
inputs = ["parse_syslog", "parse_journal"]
encoding.codec = "json"

Below is the console output:
peter.pham@sjcb9esx1vm1:/vector-sample$ vi vector-all.toml
peter.pham@sjcb9esx1vm1:/vector-sample$ sudo vector --config ./vector-all.toml
2022-12-08T18:53:30.654251Z INFO vector::app: Internal log rate limit configured. internal_log_rate_secs=10
2022-12-08T18:53:30.654471Z INFO vector::app: Log level is enabled. level="vector=info,codec=info,vrl=info,file_source=info,tower_limit=trace,rdkafka=info,buffers=info,lapin=info,kube=info"
2022-12-08T18:53:30.654627Z INFO vector::app: Loading configs. paths=["vector-all.toml"]
2022-12-08T18:53:30.663518Z INFO vector::topology::running: Running healthchecks.
2022-12-08T18:53:30.664149Z INFO vector: Vector has started. debug="false" version="0.25.1" arch="x86_64" revision="9125a99 2022-11-07"
2022-12-08T18:53:30.664193Z INFO vector::app: API is disabled, enable by setting api.enabled to true and use commands like vector top.
2022-12-08T18:53:30.666200Z INFO vector::topology::builder: Healthcheck: Passed.
2022-12-08T18:53:30.667782Z INFO source{component_kind="source" component_id=syslog_file component_type=file component_name=syslog_file}: vector::sources::file: Starting file server. include=["./syslog-sample"] exclude=[]
2022-12-08T18:53:30.671281Z INFO source{component_kind="source" component_id=syslog_file component_type=file component_name=syslog_file}:file_server: file_source::checkpointer: Attempting to read legacy checkpoint files.
2022-12-08T18:53:30.671994Z INFO source{component_kind="source" component_id=journal_file component_type=file component_name=journal_file}: vector::sources::file: Starting file server. include=["./journal-sample"] exclude=[]
2022-12-08T18:53:30.674954Z INFO source{component_kind="source" component_id=syslog_file component_type=file component_name=syslog_file}:file_server: vector::internal_events::file::source: Found new file to watch. file=syslog-sample
2022-12-08T18:53:30.676531Z INFO source{component_kind="source" component_id=journal_file component_type=file component_name=journal_file}:file_server: file_source::checkpointer: Attempting to read legacy checkpoint files.
{"appname":"rsyslogd","hostname":"dm-140611-dmn0","message":"rsyslogd was HUPed","origin":{"software":"rsyslogd","swVersion":"8.2001.0","x-info":"https://www.rsyslog.com","x-pid":"1014"},"source_type":"syslog","timestamp":"2022-10-19T07:00:01Z"}
{"appname":"systemd","hostname":"dm-140611-dmn0","message":"logrotate.service: Succeeded.","procid":1,"source_type":"syslog","timestamp":"2022-10-19T07:00:01Z"}
{"appname":"systemd","hostname":"dm-140611-dmn0","message":"Finished Rotate log files.","procid":1,"source_type":"syslog","timestamp":"2022-10-19T07:00:02Z"}
{"appname":"systemd","hostname":"dm-140611-dmn0","message":"man-db.service: Succeeded.","procid":1,"source_type":"syslog","timestamp":"2022-10-19T07:00:03Z"}
{"appname":"systemd","hostname":"dm-140611-dmn0","message":"Finished Daily man-db regeneration.","procid":1,"source_type":"syslog","timestamp":"2022-10-19T07:00:04Z"}
{"appname":"PackageKit","hostname":"dm-140611-dmn0","message":"daemon quit","source_type":"syslog","timestamp":"2022-10-19T07:00:05Z"}
{"appname":"systemd","hostname":"dm-140611-dmn0","message":"packagekit.service: Succeeded.","procid":1,"source_type":"syslog","timestamp":"2022-10-19T07:00:06Z"}
{"appname":"systemd","hostname":"dm-140611-dmn0","message":"Created slice User Slice of UID 1000.","procid":1,"source_type":"syslog","timestamp":"2022-10-19T07:00:07Z"}
{"appname":"systemd","hostname":"dm-140611-dmn0","message":"Starting User Runtime Directory /run/user/1000...","procid":1,"source_type":"syslog","timestamp":"2022-10-19T07:00:08Z"}
{"appname":"systemd","hostname":"dm-140611-dmn0","message":"Finished User Runtime Directory /run/user/1000.","procid":1,"source_type":"syslog","timestamp":"2022-10-19T07:00:09Z"}
^C2022-12-08T18:53:54.551856Z INFO vector: Vector has stopped.

Did I miss anything? I thought input from multiple sources should be alright.

Please advise.