CSV parse/read error when dumped into a folder

[baselai (AiFortress)]

I have the following case: I configured logstash to monitor a specific root and subdirectories if there is any folder created with a csv file in it then it should read it and push it to elasticsearch.

Here is how I configured it:

input {
  file {
    path => "/usr/share/input/**/*.csv"
    start_position => beginning
    sincedb_path => "/dev/null"
    discover_interval => 5
    stat_interval => "1 s"
  }
}
filter {
    csv { 
        source => "message" 
        target => "[@metadata][row]"
        autodetect_column_names => true
    }
    ruby {
        code => '
            r = event.get("[@metadata][row]")
            rn = r["row_number"]
            if rn and r
                a = []
                r.each { |k, v|
                    h = {}
                    h["column_name"] = k
                    h["row_number"] = rn.to_i
                    h["column_value_float"] = v
                    h["column_value_string"] = v
                    a << h                    
                }
                event.set("foo", a)
            end
        '
    }
    split { field => "foo" }
    ruby {
        code => '
            event.get("foo").each { |k, v|
                event.set(k,v)
            }
            event.remove("foo")
        '
    }
}
output {
    stdout { codec => rubydebug }
    file {
        path => "/usr/share/output/output.json"
        codec => "json_lines"
    }
}

But, it works only in this use-case:

• Logstash service is down
• Created a folder and dumper the csv file in it
• Run logstash service again
• Then it will read the file properly and it will read any file in any newly created folder!!

and it doesn't work in this use-case:

• Logstash is up
• Dump a folder with a csv file in it in the root the logstash is listening to.
• it will throw an error and crash

Logs output:

[2019-08-21T19:48:39,973][WARN ][logstash.filters.split   ] Only String and Array types are splittable. field:foo is of type = NilClass
[2019-08-21T19:48:39,977][ERROR][logstash.filters.ruby    ] Ruby exception occurred: undefined method `each' for nil:NilClass
[2019-08-21T19:48:39,978][WARN ][logstash.filters.split   ] Only String and Array types are splittable. field:foo is of type = NilClass
[2019-08-21T19:48:39,977][WARN ][logstash.filters.split   ] Only String and Array types are splittable. field:foo is of type = NilClass
[2019-08-21T19:48:39,981][ERROR][logstash.filters.ruby    ] Ruby exception occurred: undefined method `each' for nil:NilClass
[2019-08-21T19:48:39,982][ERROR][logstash.filters.ruby    ] Ruby exception occurred: undefined method `each' for nil:NilClass

....
{
    "@timestamp" => 2019-08-21T19:23:09.190Z,
          "tags" => [
        [0] "_split_type_failure",
        [1] "_rubyexception"
    ],
          "path" => "/usr/share/input/sentiment-1/sample.csv",
      "@version" => "1",
          "host" => "b61d8fdbca5e",
       "message" => "1,Mike,Hello,11.5\r"
}
...

any ideas @Badger ?

Update:

I changed the time from

discover_interval => 5
stat_interval => "1 s"

to

discover_interval => 3
stat_interval => "2 s"

and waited after the service is up, like around 5 secs and I dumped the folders in the root and it worked, any explanation?