Logstash découpage avec split et ingestion ne fonctionne pas

[romainfoulono]

Bonjour tout le monde,

Je souhaite ingérer dans Elasticsearch via Logstash des CSV et des textes. Sauf que Logstash envoie un champs unique "message" avec la ligne de texte à l'intérieur. J'aimerai découper cette ligne pour avoir plusieurs colonnes donc j'ai voulu utiliser un filtre, sauf que ça ne fonctionne pas : L'ingestion ne se fait pas.

Avant que je ne mette en place le filtre, l'ingestion avait fonctionné au moins une fois.

Voici le code de la configuration Logstash :

input {
  file {
		path => "C:/logs/*.txt"
		start_position => "beginning"
		sincedb_path => "NULL"
	}
}
filter {
	mutate {
		split => { "message" => ";" }
		add_field => { "col" => "%{message[0]}" }
	}
}
output {
 elasticsearch { hosts => ["localhost:9200"] }
  stdout { codec => rubydebug }
}

[ahmed_charafouddine]

Peux-tu nous donner un peu plus de détails notamment un event log afin de te proposer un pattern grok.

En fait, c'est plutôt le plugin grok qui te permettra de parser ton message.
Merci de trouver ci-apres un exemple d'un grok pattern. ça pourrait beaucoup t'aider.
Lien : https://www.elastic.co/guide/en/logstash/current/plugins-filters-grok.html

[romainfoulono]

J'ai voulu suivre le conseil. J'ai vu que la syntaxe est la suivante : %{SYNTAX:SEMANTIC}
Il existe des mots-clés pour le nom SYNTAX mais sont-ils listés quelque part ?
Etant donné la structure du fichier de test, un simple split ne suffirait pas ?

Ca c'est le contenu du fichier de test

1234;unmot

Sachant que je voudrais essayer sur plus de types de données (comme l'exemple ci-dessous) mais j'ai déjà un problème pour les deux premières valeurs.

1234;un mot;;UN AUTRE MOT;PA ;;14/02567/AB/167;

Ca c'est la configuration Logstash

input {
  file {
		path => "C:/logs/*.txt"
		start_position => "beginning"
		sincedb_path => "NULL"
	}
}
filter {
	grok {
		match => {
			"message" => "%{NUMBER:col0};%{WORD:col1}" 
		}
	}
}
output {
 elasticsearch { hosts => ["localhost:9200"] }
  stdout { codec => rubydebug }
}

Et voilà le log

[2020-02-13T14:35:46,738][WARN ][logstash.config.source.multilocal] Ignoring the 'pipelines.yml' file because modules or command line options are specified
[2020-02-13T14:35:46,845][INFO ][logstash.runner          ] Starting Logstash {"logstash.version"=>"7.5.1"}
[2020-02-13T14:35:48,795][INFO ][org.reflections.Reflections] Reflections took 38 ms to scan 1 urls, producing 20 keys and 40 values 
[2020-02-13T14:35:50,239][INFO ][logstash.outputs.elasticsearch][main] Elasticsearch pool URLs updated {:changes=>{:removed=>[], :added=>[http://localhost:9200/]}}
[2020-02-13T14:35:50,423][WARN ][logstash.outputs.elasticsearch][main] Restored connection to ES instance {:url=>"http://localhost:9200/"}
[2020-02-13T14:35:50,471][INFO ][logstash.outputs.elasticsearch][main] ES Output version determined {:es_version=>7}
[2020-02-13T14:35:50,479][WARN ][logstash.outputs.elasticsearch][main] Detected a 6.x and above cluster: the `type` event field won't be used to determine the document _type {:es_version=>7}
[2020-02-13T14:35:50,547][INFO ][logstash.outputs.elasticsearch][main] New Elasticsearch output {:class=>"LogStash::Outputs::ElasticSearch", :hosts=>["//localhost:9200"]}
[2020-02-13T14:35:50,607][INFO ][logstash.outputs.elasticsearch][main] Using default mapping template
[2020-02-13T14:35:50,703][INFO ][logstash.outputs.elasticsearch][main] Attempting to install template {:manage_template=>{"index_patterns"=>"logstash-*", "version"=>60001, "settings"=>{"index.refresh_interval"=>"5s", "number_of_shards"=>1, "index.lifecycle.name"=>"logstash-policy", "index.lifecycle.rollover_alias"=>"logstash"}, "mappings"=>{"dynamic_templates"=>[{"message_field"=>{"path_match"=>"message", "match_mapping_type"=>"string", "mapping"=>{"type"=>"text", "norms"=>false}}}, {"string_fields"=>{"match"=>"*", "match_mapping_type"=>"string", "mapping"=>{"type"=>"text", "norms"=>false, "fields"=>{"keyword"=>{"type"=>"keyword", "ignore_above"=>256}}}}}], "properties"=>{"@timestamp"=>{"type"=>"date"}, "@version"=>{"type"=>"keyword"}, "geoip"=>{"dynamic"=>true, "properties"=>{"ip"=>{"type"=>"ip"}, "location"=>{"type"=>"geo_point"}, "latitude"=>{"type"=>"half_float"}, "longitude"=>{"type"=>"half_float"}}}}}}}
[2020-02-13T14:35:50,835][WARN ][org.logstash.instrument.metrics.gauge.LazyDelegatingGauge][main] A gauge metric of an unknown type (org.jruby.specialized.RubyArrayOneObject) has been create for key: cluster_uuids. This may result in invalid serialization.  It is recommended to log an issue to the responsible developer/development team.
[2020-02-13T14:35:50,843][INFO ][logstash.javapipeline    ][main] Starting pipeline {:pipeline_id=>"main", "pipeline.workers"=>8, "pipeline.batch.size"=>125, "pipeline.batch.delay"=>50, "pipeline.max_inflight"=>1000, "pipeline.sources"=>["C:/logstash-file-read.conf"], :thread=>"#<Thread:0x154d1065 run>"}
[2020-02-13T14:35:51,507][INFO ][logstash.javapipeline    ][main] Pipeline started {"pipeline.id"=>"main"}
[2020-02-13T14:35:51,559][INFO ][filewatch.observingtail  ][main] START, creating Discoverer, Watch with file and sincedb collections
[2020-02-13T14:35:51,567][INFO ][logstash.agent           ] Pipelines running {:count=>1, :running_pipelines=>[:main], :non_running_pipelines=>[]}
[2020-02-13T14:35:52,227][INFO ][logstash.agent           ] Successfully started Logstash API endpoint {:port=>9600}

[romainfoulono]

Bonjour,

Comme vous semblez avoir peu de temps, j'ai reposté le sujet sur le forum anglophone.

Voici le lien : Logstash : ingestion via Grok pattern does not work