How to use multiple csv files in logstash and filebeat

Youssef_SBAI · May 12, 2020, 12:09pm

I want to use multiple csv files in filebeat

my config file beat

filebeat.inputs:

# Each - is an input. Most options can be set at the input level, so
# you can use different inputs for various configurations.
# Below are the input specific configurations.

- type: log

  # Change to true to enable this input configuration.
  enabled: true

  # Paths that should be crawled and fetched. Glob based paths.
  paths:
   - C:\Users\h83710\Desktop\elastic\logstash-7.5.2\test\FICHIER.csv
   - C:\Users\h83710\Desktop\elastic\logstash-7.5.2\test\poca.csv

and my config Logstash

input {
    beats {
    port => "5044"
    }
}    
              
filter {

if [path] == "C:\Users\h83710\Desktop\elastic\logstash-7.5.2\test\FICHIER.csv"{
csv{
separator => ";"
columns => ["chaine", "job", "date_plan", "statut", "date_debut", "date_fin", "serveur", "numero_passage", "application", "sous_application"]
}

date {
match => [ "date_plan" , "dd/MM/YYYY" ]
}

date {
match => [ "date_debut" , "dd/MM/YYYY HH:mm" ]
}

date {
match => [ "date_fin" , "dd/MM/YYYY HH:mm" ] 
}
mutate {
convert => { "numero_passage" => "integer" }
}

}

else if [path] == "C:\Users\h83710\Desktop\elastic\logstash-7.5.2\test\poca.csv" {
csv{
separator => ";"
columns => ["date_fraicheur", "serveur"]
}

mutate {
convert => { "date_fraicheur" => "dd/MM/YYYY" }
}
}

}

output{
        elasticsearch{
            hosts => "http://localhost:9200"
            index => "hello"
        }
stdout {codec => rubydebug}		
    }

Youssef_SBAI · May 12, 2020, 12:11pm

I have this error

.\bin\logstash -f configfilebeata.conf
Thread.exclusive is deprecated, use Thread::Mutex
Sending Logstash logs to C:/Users/h83710/Desktop/elastic/logstash-7.5.2/logs which is now configured via log4j2.properties
[2020-05-12T14:04:19,257][WARN ][logstash.config.source.multilocal] Ignoring the 'pipelines.yml' file because modules or command line options are specified
[2020-05-12T14:04:19,415][INFO ][logstash.runner          ] Starting Logstash {"logstash.version"=>"7.5.2"}
[2020-05-12T14:04:24,331][INFO ][org.reflections.Reflections] Reflections took 61 ms to scan 1 urls, producing 20 keys and 40 values
[2020-05-12T14:04:30,687][INFO ][logstash.outputs.elasticsearch][main] Elasticsearch pool URLs updated {:changes=>{:removed=>[], :added=>[http://localhost:9200/]}}
[2020-05-12T14:04:31,048][WARN ][logstash.outputs.elasticsearch][main] Restored connection to ES instance {:url=>"http://localhost:9200/"}
[2020-05-12T14:04:31,133][INFO ][logstash.outputs.elasticsearch][main] ES Output version determined {:es_version=>7}
[2020-05-12T14:04:31,144][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-05-12T14:04:31,259][INFO ][logstash.outputs.elasticsearch][main] New Elasticsearch output {:class=>"LogStash::Outputs::ElasticSearch", :hosts=>["http://localhost:9200"]}
[2020-05-12T14:04:31,355][INFO ][logstash.outputs.elasticsearch][main] Using default mapping template
[2020-05-12T14:04:31,451][ERROR][logstash.javapipeline    ][main] Pipeline aborted due to error {:pipeline_id=>"main", :exception=>#<LogStash::ConfigurationError: translation missing: en.logstash.agent.configuration.invalid_plugin_register>, :backtrace=>["C:/Users/h83710/Desktop/elastic/logstash-7.5.2/vendor/bundle/jruby/2.5.0/gems/logstash-filter-mutate-3.5.0/lib/logstash/filters/mutate.rb:222:in `block in register'", "org/jruby/RubyHash.java:1417:in `each'", "C:/Users/h83710/Desktop/elastic/logstash-7.5.2/vendor/bundle/jruby/2.5.0/gems/logstash-filter-mutate-3.5.0/lib/logstash/filters/mutate.rb:220:in `register'", "org/logstash/config/ir/compiler/AbstractFilterDelegatorExt.java:56:in `register'", "C:/Users/h83710/Desktop/elastic/logstash-7.5.2/logstash-core/lib/logstash/java_pipeline.rb:200:in `block in register_plugins'", "org/jruby/RubyArray.java:1800:in `each'", "C:/Users/h83710/Desktop/elastic/logstash-7.5.2/logstash-core/lib/logstash/java_pipeline.rb:199:in `register_plugins'", "C:/Users/h83710/Desktop/elastic/logstash-7.5.2/logstash-core/lib/logstash/java_pipeline.rb:506:in `maybe_setup_out_plugins'", "C:/Users/h83710/Desktop/elastic/logstash-7.5.2/logstash-core/lib/logstash/java_pipeline.rb:212:in `start_workers'", "C:/Users/h83710/Desktop/elastic/logstash-7.5.2/logstash-core/lib/logstash/java_pipeline.rb:154:in `run'", "C:/Users/h83710/Desktop/elastic/logstash-7.5.2/logstash-core/lib/logstash/java_pipeline.rb:109:in `block in start'"], "pipeline.sources"=>["C:/Users/h83710/Desktop/elastic/logstash-7.5.2/configfilebeata.conf"], :thread=>"#<Thread:0x201527de run>"}
[2020-05-12T14:04:31,463][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}, "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-05-12T14:04:31,499][ERROR][logstash.agent           ] Failed to execute action {:id=>:main, :action_type=>LogStash::ConvergeResult::FailedAction, :message=>"Could not execute action: PipelineAction::Create<main>, action_result: false", :backtrace=>nil}
[2020-05-12T14:04:31,967][INFO ][logstash.agent           ] Successfully started Logstash API endpoint {:port=>9600}
[2020-05-12T14:04:36,687][INFO ][logstash.runner          ] Logstash shut down.

ptamba · May 12, 2020, 12:22pm

use date filter to convert date. mutate convert doesn’t support date

Youssef_SBAI · May 12, 2020, 12:33pm

I change my config Logstsh

input {
    beats {
    port => "5044"
    }
}    
              
filter {

if [path] == "C:\Users\h83710\Desktop\elastic\logstash-7.5.2\test\FICHIER.csv"{
csv{
separator => ";"
columns => ["chaine", "job", "date_plan", "statut", "date_debut", "date_fin", "serveur", "numero_passage", "application", "sous_application"]
}

date {
match => [ "date_plan" , "dd/MM/YYYY" ]
}

date {
match => [ "date_debut" , "dd/MM/YYYY HH:mm" ]
}

date {
match => [ "date_fin" , "dd/MM/YYYY HH:mm" ] 
}
mutate {
convert => { "numero_passage" => "integer" }
}

}

else if [path] == "C:\Users\h83710\Desktop\elastic\logstash-7.5.2\test\poca.csv" {
csv{
separator => ";"
columns => ["date_fraicheur", "serveur"]
}

date {
match => { "date_fraicheur" => "dd/MM/YYYY" }
}

}

}

output{
        elasticsearch{
            hosts => "http://localhost:9200"
            index => "hello"
        }
stdout {codec => rubydebug}		
    }

But i have this error

.\bin\logstash -f configfilebeata.conf
Thread.exclusive is deprecated, use Thread::Mutex
Sending Logstash logs to C:/Users/h83710/Desktop/elastic/logstash-7.5.2/logs which is now configured via log4j2.properties
[2020-05-12T14:29:45,844][WARN ][logstash.config.source.multilocal] Ignoring the 'pipelines.yml' file because modules or command line options are specified
[2020-05-12T14:29:46,058][INFO ][logstash.runner          ] Starting Logstash {"logstash.version"=>"7.5.2"}
[2020-05-12T14:29:53,521][INFO ][org.reflections.Reflections] Reflections took 95 ms to scan 1 urls, producing 20 keys and 40 values
[2020-05-12T14:29:55,184][ERROR][logstash.agent           ] Failed to execute action {:action=>LogStash::PipelineAction::Create/pipeline_id:main, :exception=>"Java::JavaLang::IllegalStateException", :message=>"Unable to configure plugins: (ConfigurationError) translation missing: en.logstash.agent.configuration.invalid_plugin_register", :backtrace=>["org.logstash.config.ir.CompiledPipeline.<init>(CompiledPipeline.java:119)", "org.logstash.execution.JavaBasePipelineExt.initialize(JavaBasePipelineExt.java:60)", "org.logstash.execution.JavaBasePipelineExt$INVOKER$i$1$0$initialize.call(JavaBasePipelineExt$INVOKER$i$1$0$initialize.gen)", "org.jruby.internal.runtime.methods.JavaMethod$JavaMethodN.call(JavaMethod.java:837)", "org.jruby.ir.runtime.IRRuntimeHelpers.instanceSuper(IRRuntimeHelpers.java:1156)", "org.jruby.ir.runtime.IRRuntimeHelpers.instanceSuperSplatArgs(IRRuntimeHelpers.java:1143)", "org.jruby.ir.targets.InstanceSuperInvokeSite.invoke(InstanceSuperInvokeSite.java:39)", "C_3a_.Users.h83710.Desktop.elastic.logstash_minus_7_dot_5_dot_2.logstash_minus_core.lib.logstash.java_pipeline.RUBY$method$initialize$0(C:/Users/h83710/Desktop/elastic/logstash-7.5.2/logstash-core/lib/logstash/java_pipeline.rb:27)", "org.jruby.internal.runtime.methods.CompiledIRMethod.call(CompiledIRMethod.java:91)", "org.jruby.internal.runtime.methods.MixedModeIRMethod.call(MixedModeIRMethod.java:90)", "org.jruby.runtime.callsite.CachingCallSite.cacheAndCall(CachingCallSite.java:332)", "org.jruby.runtime.callsite.CachingCallSite.call(CachingCallSite.java:86)", "org.jruby.RubyClass.newInstance(RubyClass.java:915)", "org.jruby.RubyClass$INVOKER$i$newInstance.call(RubyClass$INVOKER$i$newInstance.gen)", "org.jruby.ir.targets.InvokeSite.invoke(InvokeSite.java:183)", "C_3a_.Users.h83710.Desktop.elastic.logstash_minus_7_dot_5_dot_2.logstash_minus_core.lib.logstash.pipeline_action.create.RUBY$method$execute$0(C:/Users/h83710/Desktop/elastic/logstash-7.5.2/logstash-core/lib/logstash/pipeline_action/create.rb:36)", "C_3a_.Users.h83710.Desktop.elastic.logstash_minus_7_dot_5_dot_2.logstash_minus_core.lib.logstash.pipeline_action.create.RUBY$method$execute$0$__VARARGS__(C:/Users/h83710/Desktop/elastic/logstash-7.5.2/logstash-core/lib/logstash/pipeline_action/create.rb)", "org.jruby.internal.runtime.methods.CompiledIRMethod.call(CompiledIRMethod.java:91)", "org.jruby.internal.runtime.methods.MixedModeIRMethod.call(MixedModeIRMethod.java:90)", "org.jruby.ir.targets.InvokeSite.invoke(InvokeSite.java:183)", "C_3a_.Users.h83710.Desktop.elastic.logstash_minus_7_dot_5_dot_2.logstash_minus_core.lib.logstash.agent.RUBY$block$converge_state$2(C:/Users/h83710/Desktop/elastic/logstash-7.5.2/logstash-core/lib/logstash/agent.rb:326)", "org.jruby.runtime.CompiledIRBlockBody.callDirect(CompiledIRBlockBody.java:136)", "org.jruby.runtime.IRBlockBody.call(IRBlockBody.java:77)", "org.jruby.runtime.Block.call(Block.java:129)", "org.jruby.RubyProc.call(RubyProc.java:295)", "org.jruby.RubyProc.call(RubyProc.java:274)", "org.jruby.RubyProc.call(RubyProc.java:270)", "org.jruby.internal.runtime.RubyRunnable.run(RubyRunnable.java:105)", "java.lang.Thread.run(Unknown Source)"]}
warning: thread "Converge PipelineAction::Create<main>" terminated with exception (report_on_exception is true):
LogStash::Error: Don't know how to handle `Java::JavaLang::IllegalStateException` for `PipelineAction::Create<main>`
          create at org/logstash/execution/ConvergeResultExt.java:109
             add at org/logstash/execution/ConvergeResultExt.java:37
  converge_state at C:/Users/h83710/Desktop/elastic/logstash-7.5.2/logstash-core/lib/logstash/agent.rb:339
[2020-05-12T14:29:55,204][ERROR][logstash.agent           ] An exception happened when converging configuration {:exception=>LogStash::Error, :message=>"Don't know how to handle `Java::JavaLang::IllegalStateException` for `PipelineAction::Create<main>`", :backtrace=>["org/logstash/execution/ConvergeResultExt.java:109:in `create'", "org/logstash/execution/ConvergeResultExt.java:37:in `add'", "C:/Users/h83710/Desktop/elastic/logstash-7.5.2/logstash-core/lib/logstash/agent.rb:339:in `block in converge_state'"]}
[2020-05-12T14:29:55,269][FATAL][logstash.runner          ] An unexpected error occurred! {:error=>#<LogStash::Error: Don't know how to handle `Java::JavaLang::IllegalStateException` for `PipelineAction::Create<main>`>, :backtrace=>["org/logstash/execution/ConvergeResultExt.java:109:in `create'", "org/logstash/execution/ConvergeResultExt.java:37:in `add'", "C:/Users/h83710/Desktop/elastic/logstash-7.5.2/logstash-core/lib/logstash/agent.rb:339:in `block in converge_state'"]}
[2020-05-12T14:29:55,391][ERROR][org.logstash.Logstash    ] java.lang.IllegalStateException: Logstash stopped processing because of an error: (SystemExit) exit

ptamba · May 12, 2020, 1:05pm

date filter accepts array

date { match => ["date_fraicheur" => "dd/MM/YYYY"]}

if it doesn't work, i always find it easier to activate filters one by one to debug which filter causes issue

Youssef_SBAI · May 12, 2020, 3:02pm

i use this date { match => ["date_fraicheur" => "dd/MM/YYYY"]}

but i have this error

.\bin\logstash -f configfilebeata.conf
Thread.exclusive is deprecated, use Thread::Mutex
Sending Logstash logs to C:/Users/h83710/Desktop/elastic/logstash-7.5.2/logs which is now configured via log4j2.properties
[2020-05-12T14:29:45,844][WARN ][logstash.config.source.multilocal] Ignoring the 'pipelines.yml' file because modules or command line options are specified
[2020-05-12T14:29:46,058][INFO ][logstash.runner          ] Starting Logstash {"logstash.version"=>"7.5.2"}
[2020-05-12T14:29:53,521][INFO ][org.reflections.Reflections] Reflections took 95 ms to scan 1 urls, producing 20 keys and 40 values
[2020-05-12T14:29:55,184][ERROR][logstash.agent           ] Failed to execute action {:action=>LogStash::PipelineAction::Create/pipeline_id:main, :exception=>"Java::JavaLang::IllegalStateException", :message=>"Unable to configure plugins: (ConfigurationError) translation missing: en.logstash.agent.configuration.invalid_plugin_register", :backtrace=>["org.logstash.config.ir.CompiledPipeline.<init>(CompiledPipeline.java:119)", "org.logstash.execution.JavaBasePipelineExt.initialize(JavaBasePipelineExt.java:60)", "org.logstash.execution.JavaBasePipelineExt$INVOKER$i$1$0$initialize.call(JavaBasePipelineExt$INVOKER$i$1$0$initialize.gen)", "org.jruby.internal.runtime.methods.JavaMethod$JavaMethodN.call(JavaMethod.java:837)", "org.jruby.ir.runtime.IRRuntimeHelpers.instanceSuper(IRRuntimeHelpers.java:1156)", "org.jruby.ir.runtime.IRRuntimeHelpers.instanceSuperSplatArgs(IRRuntimeHelpers.java:1143)", "org.jruby.ir.targets.InstanceSuperInvokeSite.invoke(InstanceSuperInvokeSite.java:39)", "C_3a_.Users.h83710.Desktop.elastic.logstash_minus_7_dot_5_dot_2.logstash_minus_core.lib.logstash.java_pipeline.RUBY$method$initialize$0(C:/Users/h83710/Desktop/elastic/logstash-7.5.2/logstash-core/lib/logstash/java_pipeline.rb:27)", "org.jruby.internal.runtime.methods.CompiledIRMethod.call(CompiledIRMethod.java:91)", "org.jruby.internal.runtime.methods.MixedModeIRMethod.call(MixedModeIRMethod.java:90)", "org.jruby.runtime.callsite.CachingCallSite.cacheAndCall(CachingCallSite.java:332)", "org.jruby.runtime.callsite.CachingCallSite.call(CachingCallSite.java:86)", "org.jruby.RubyClass.newInstance(RubyClass.java:915)", "org.jruby.RubyClass$INVOKER$i$newInstance.call(RubyClass$INVOKER$i$newInstance.gen)", "org.jruby.ir.targets.InvokeSite.invoke(InvokeSite.java:183)", "C_3a_.Users.h83710.Desktop.elastic.logstash_minus_7_dot_5_dot_2.logstash_minus_core.lib.logstash.pipeline_action.create.RUBY$method$execute$0(C:/Users/h83710/Desktop/elastic/logstash-7.5.2/logstash-core/lib/logstash/pipeline_action/create.rb:36)", "C_3a_.Users.h83710.Desktop.elastic.logstash_minus_7_dot_5_dot_2.logstash_minus_core.lib.logstash.pipeline_action.create.RUBY$method$execute$0$__VARARGS__(C:/Users/h83710/Desktop/elastic/logstash-7.5.2/logstash-core/lib/logstash/pipeline_action/create.rb)", "org.jruby.internal.runtime.methods.CompiledIRMethod.call(CompiledIRMethod.java:91)", "org.jruby.internal.runtime.methods.MixedModeIRMethod.call(MixedModeIRMethod.java:90)", "org.jruby.ir.targets.InvokeSite.invoke(InvokeSite.java:183)", "C_3a_.Users.h83710.Desktop.elastic.logstash_minus_7_dot_5_dot_2.logstash_minus_core.lib.logstash.agent.RUBY$block$converge_state$2(C:/Users/h83710/Desktop/elastic/logstash-7.5.2/logstash-core/lib/logstash/agent.rb:326)", "org.jruby.runtime.CompiledIRBlockBody.callDirect(CompiledIRBlockBody.java:136)", "org.jruby.runtime.IRBlockBody.call(IRBlockBody.java:77)", "org.jruby.runtime.Block.call(Block.java:129)", "org.jruby.RubyProc.call(RubyProc.java:295)", "org.jruby.RubyProc.call(RubyProc.java:274)", "org.jruby.RubyProc.call(RubyProc.java:270)", "org.jruby.internal.runtime.RubyRunnable.run(RubyRunnable.java:105)", "java.lang.Thread.run(Unknown Source)"]}
warning: thread "Converge PipelineAction::Create<main>" terminated with exception (report_on_exception is true):
LogStash::Error: Don't know how to handle `Java::JavaLang::IllegalStateException` for `PipelineAction::Create<main>`
          create at org/logstash/execution/ConvergeResultExt.java:109
             add at org/logstash/execution/ConvergeResultExt.java:37
  converge_state at C:/Users/h83710/Desktop/elastic/logstash-7.5.2/logstash-core/lib/logstash/agent.rb:339
[2020-05-12T14:29:55,204][ERROR][logstash.agent           ] An exception happened when converging configuration {:exception=>LogStash::Error, :message=>"Don't know how to handle `Java::JavaLang::IllegalStateException` for `PipelineAction::Create<main>`", :backtrace=>["org/logstash/execution/ConvergeResultExt.java:109:in `create'", "org/logstash/execution/ConvergeResultExt.java:37:in `add'", "C:/Users/h83710/Desktop/elastic/logstash-7.5.2/logstash-core/lib/logstash/agent.rb:339:in `block in converge_state'"]}
[2020-05-12T14:29:55,269][FATAL][logstash.runner          ] An unexpected error occurred! {:error=>#<LogStash::Error: Don't know how to handle `Java::JavaLang::IllegalStateException` for `PipelineAction::Create<main>`>, :backtrace=>["org/logstash/execution/ConvergeResultExt.java:109:in `create'", "org/logstash/execution/ConvergeResultExt.java:37:in `add'", "C:/Users/h83710/Desktop/elastic/logstash-7.5.2/logstash-core/lib/logstash/agent.rb:339:in `block in converge_state'"]}
[2020-05-12T14:29:55,391][ERROR][org.logstash.Logstash    ] java.lang.IllegalStateException: Logstash stopped processing because of an error: (SystemExit) exit

ptamba · May 12, 2020, 3:59pm

if you remove the entire filter section, what do you get ?

Youssef_SBAI · May 12, 2020, 5:19pm

if i remove the entire filter section Logsth is OK but the ingest in elastic is not OK

ptamba · May 12, 2020, 7:18pm

if the logstash is running , what did you see in the stdout? dis you see any message ?

Youssef_SBAI · May 13, 2020, 8:01am

I seee in the stdout :

{
          "tags" => [
        [0] "poca",
        [1] "beats_input_codec_plain_applied"
    ],
           "log" => {
          "file" => {
            "path" => "C:\\Users\\h83710\\Desktop\\elastic\\logstash-7.5.2\\test\\poca.csv"
        },
        "offset" => 16
    },
      "@version" => "1",
         "input" => {
        "type" => "log"
    },
         "agent" => {
                "type" => "filebeat",
        "ephemeral_id" => "82e32e98-66b5-4489-98f2-fcc7ff224697",
                  "id" => "c19672a9-bf8b-4bf7-8d01-10f24fe51d7a",
             "version" => "7.6.1",
            "hostname" => "C06513361"
    },
       "message" => "12/05/2020",
          "host" => {
        "name" => "C06513361"
    },
           "ecs" => {
        "version" => "1.4.0"
    },
    "@timestamp" => 2020-05-12T18:06:08.480Z
}

ptamba · May 13, 2020, 11:55am

for starters, your output has path wrapped inside [log][file][path] so this part won’t work as it looks for [path]

i would start with one csv, get them parsed correctly, then add the next csv

also since you’re using filebeat to ingest, you can have multiple input with different path for each input. you can apply different tags for different input, then process the event with correct csv filter based on tags.

see here for reference

system · June 10, 2020, 11:55am

This topic was automatically closed 28 days after the last reply. New replies are no longer allowed.