Logstash ingest and export to elasticsearch files twice

Greninja_San · February 15, 2022, 1:24pm

Hello!

I have a logstash config that gather lines from CSVs and then send them to Elasticsearch. However, for some unknown reason, lines are duplicated in Elasticsearch, taking twice the storage space, and making statistics wrong.

Any ideas of the cause?

Thanks

leandrojmp · February 15, 2022, 1:36pm

You need to share your configuration, use the preformatted button in the forum </>, and paste your configuration.

Greninja_San · February 15, 2022, 1:51pm

Woops sorry

input {
  file {
    path => "/srv/edm/ftp/bcp/*.csv"
    start_position => "beginning"
    ignore_older => 17280000
    sincedb_path => "/dev/null"
    codec => plain {
      charset => "ANSI_X3.4-1968"
    }
  }
}

filter {
     csv {
        separator => ";"
        columns => ["sender", "receiver", "flow_type", "end_traitement", "start_traitement", "size_end_f", "size_start_f", "format_end", "format_start", "platform_end", "platform_start", "transport_end", "transport_start", "prod"]
        skip_header => true
     }
    ruby {
        code => 'event.set("date_start_formated", event.get("start_traitement").ljust(21, "0"))'
    }
    ruby {
        code => 'event.set("date_end_formated", event.get("end_traitement").ljust(21, "0"))'
    }
    ruby {
        code => 'event.set("date_start_formated_cut", ((event.get("date_start_formated"))[0,21]))'
    }
    ruby {
        code => 'event.set("date_end_formated_cut", ((event.get("date_end_formated"))[0,21]))'
    }
   ruby {
        code => 'event.set("date_start", (event.get("date_start_formated_cut").gsub(",",".")))'
    }
    ruby {
        code => 'event.set("date_end", (event.get("date_end_formated_cut").gsub(",",".")))'
    }
    ruby {
    code => 'event.set("size_start", event.get("size_start_f").to_i)'
    }

    ruby {
    code => 'event.set("size_end", event.get("size_end_f").to_i)'
    }
    date {
    match => ["date_start", "dd/MM/yy HH:mm:ss.SSS"]
    target => "start_traitement_true"
    }
    date {
    match => ["date_end", "dd/MM/yy HH:mm:ss.SSS"]
    target => "end_traitement_true"
    }
    ruby {
    code => 'event.set("date_epoch_end", event.get("end_traitement_true").to_i)'
    }
    ruby {
    code => 'event.set("date_epoch_start", event.get("start_traitement_true").to_i)'
    }
    ruby {
    code => 'event.set("time_between", ((event.get("date_epoch_end"))-(event.get("date_epoch_start"))).abs)'
    }
    ruby {
    code => 'event.set("fixedProd", (event.get("prod")).tr("\r", ""))'
    }

   ruby {
    code => 'event.set("cat", "BCP")'
    }

    mutate {
      convert => {
         "sender" => "string"
         "receiver" => "string"
         "flow_type" => "string"
         "size_start" => "integer"
         "size_end" =>  "integer"
         "format_start" => "string"
         "format_end" => "string"
         "platform_start" => "string"
         "platform_end" => "string"
         "date_epoch_end" => "float"
         "date_epoch_start" => "float"
         "transport_start" => "string"
         "transport_end" => "string"
         "prod" => "string"
         }
    }

    prune{
      blacklist_names => ["message","date_end_formated_cut","date_start_formated_cut","date_start_formated","date_end_formated","date_start","date_end"]
    }

}

output {
  stdout{}
  elasticsearch {
    hosts => "http://XXXXXXX:9200"
    index => "index_bcp"
    user => "XXXXXXX"
    password => "XXXXXXX"
    ssl_certificate_verification => false
  }
}

Tomo_M · February 15, 2022, 5:54pm

Have you run the script just once?

Greninja_San · February 15, 2022, 7:14pm

It is launched using systemctl, and just once, yes.

leandrojmp · February 15, 2022, 8:05pm

Any reason to use sincedb_path as /dev/null ? This makes Logstash reads the file again if your service is restarted for some reason.

There is nothing in your configuration that could duplicate the documents unless your logstash is being restarted.

Do you have automatic reload enabled? I think that if you have automatic reload enabled and you change the pipeline configuration file, this will trigger a pipeline reload that could reread the files.

What does your pipelines.yml looks like? Do you have multiple pipelines? Are pointing to a directory with multiple configuration files?

Greninja_San · February 16, 2022, 9:24am

If not /dev/null, I could set any folder?

Greninja_San · February 16, 2022, 9:38am

Also I don't think it is due to a restart, since on duplicate documents, @timestamp is the exact same.

Greninja_San · February 16, 2022, 12:59pm

I just verified, I have just one pipeline, leading to a folder with 2 different config files. And no automatic reload as well.

leandrojmp · February 16, 2022, 1:00pm

The sincedb_path needs to point to a file where logstash will keep track of the position already read from the file, you can remove this option of your config and let logstash set the sincedb path itself.

Using /dev/null means that logstash does not keep track of the position in the file already read and it can read the file again on a restart.

If the @timestamp is the same, then you are right, it may not be related to a restart as this would change the @timestamp when logstash reread the file.

You didn't share your pipelines.yml, are you using multiple pipelines or just one main pipeline pointing to /etc/logstash/conf.d/*.conf? Share your pipelines.yml.

The only other way that I can think of that could create duplicate documents is if you are pointing to a folder with multiple configurations and have another Elasticsearch output indexing to the same index name.

Greninja_San · February 16, 2022, 1:01pm

leandrojmp · February 16, 2022, 1:04pm

Ok, and do you have any other *.conf files in the /etc/logstash/conf.d directory?

If there is other files, please share them as well.

Greninja_San · February 16, 2022, 1:08pm

These 2 configs :

input {
  file {
    path => "/srv/edm/ftp/not_reversed/*.csv"
    start_position => "beginning"
    ignore_older => 17280000
    sincedb_path => "/dev/null"
    codec => plain {
      charset => "ANSI_X3.4-1968"
    }
  }
}

filter {
     csv {
        separator => ";"
        columns => ["sender", "receiver", "flow_type", "start_traitement", "end_traitement", "size_start_f", "size_end_f", "format_start", "format_end", "platform_start", "platform_end", "transport_start", "transport_end", "prod"]
        skip_header => true
     }
    ruby {
        code => 'event.set("date_start_formated", event.get("start_traitement").ljust(21, "0"))'
    }
    ruby {
        code => 'event.set("date_end_formated", event.get("end_traitement").ljust(21, "0"))'
    }
    ruby {
        code => 'event.set("date_start_formated_cut", ((event.get("date_start_formated"))[0,21]))'
    }
    ruby {
        code => 'event.set("date_end_formated_cut", ((event.get("date_end_formated"))[0,21]))'
    }
   ruby {
        code => 'event.set("date_start", (event.get("date_start_formated_cut").gsub(",",".")))'
    }
    ruby {
        code => 'event.set("date_end", (event.get("date_end_formated_cut").gsub(",",".")))'
    }
    ruby {
    code => 'event.set("size_start", event.get("size_start_f").to_i)'
    }

    ruby {
    code => 'event.set("size_end", event.get("size_end_f").to_i)'
    }

    date {
    match => ["date_start", "dd/MM/yy HH:mm:ss.SSS"]
    target => "start_traitement_true"
    }
    date {
    match => ["date_end", "dd/MM/yy HH:mm:ss.SSS"]
    target => "end_traitement_true"
    }
    ruby {
    code => 'event.set("date_epoch_end", event.get("end_traitement_true").to_i)'
    }
    ruby {
    code => 'event.set("date_epoch_start", event.get("start_traitement_true").to_i)'
    }
    ruby {
    code => 'event.set("time_between", ((event.get("date_epoch_end"))-(event.get("date_epoch_start"))).abs)'
    }
    ruby {
    code => 'event.set("fixedProd", (event.get("prod")).tr("\r", ""))'
    }

   ruby {
    code => 'event.set("cat", "BCP")'
    }

    mutate {
      convert => {
         "sender" => "string"
         "receiver" => "string"
         "flow_type" => "string"
         "size_start" => "integer"
         "size_end" =>  "integer"
         "format_start" => "string"
         "format_end" => "string"
         "platform_start" => "string"
         "platform_end" => "string"
         "date_epoch_end" => "float"
         "date_epoch_start" => "float"
         "transport_start" => "string"
         "transport_end" => "string"
         "prod" => "string"
         }
    }

    prune{
      blacklist_names => ["message","date_end_formated_cut","date_start_formated_cut","date_start_formated","date_end_formated","date_start","date_end"]
    }

}

output {
  stdout{}
  elasticsearch {
    hosts => "http://XXXXXXX:9200"
    index => "index_bcp"
    user => "XXXXX"
    password => "XXXXXXX"
    ssl_certificate_verification => false
  }
}

AND

input {
  file {
    path => "/srv/edm/ftp/bcp/*.csv"
    start_position => "beginning"
    ignore_older => 17280000
    sincedb_path => "/dev/null"
    codec => plain {
      charset => "ANSI_X3.4-1968"
    }
  }
}

filter {
     csv {
        separator => ";"
        columns => ["sender", "receiver", "flow_type", "end_traitement", "start_traitement", "size_end_f", "size_start_f", "format_end", "format_start", "platform_end", "platform_start", "transport_end", "transport_start", "prod"]
        skip_header => true
     }
    ruby {
        code => 'event.set("date_start_formated", event.get("start_traitement").ljust(21, "0"))'
    }
    ruby {
        code => 'event.set("date_end_formated", event.get("end_traitement").ljust(21, "0"))'
    }
    ruby {
        code => 'event.set("date_start_formated_cut", ((event.get("date_start_formated"))[0,21]))'
    }
    ruby {
        code => 'event.set("date_end_formated_cut", ((event.get("date_end_formated"))[0,21]))'
    }
   ruby {
        code => 'event.set("date_start", (event.get("date_start_formated_cut").gsub(",",".")))'
    }
    ruby {
        code => 'event.set("date_end", (event.get("date_end_formated_cut").gsub(",",".")))'
    }
    ruby {
    code => 'event.set("size_start", event.get("size_start_f").to_i)'
    }

    ruby {
    code => 'event.set("size_end", event.get("size_end_f").to_i)'
    }
    date {
    match => ["date_start", "dd/MM/yy HH:mm:ss.SSS"]
    target => "start_traitement_true"
    }
    date {
    match => ["date_end", "dd/MM/yy HH:mm:ss.SSS"]
    target => "end_traitement_true"
    }
    ruby {
    code => 'event.set("date_epoch_end", event.get("end_traitement_true").to_i)'
    }
    ruby {
    code => 'event.set("date_epoch_start", event.get("start_traitement_true").to_i)'
    }
    ruby {
    code => 'event.set("time_between", ((event.get("date_epoch_end"))-(event.get("date_epoch_start"))).abs)'
    }
    ruby {
    code => 'event.set("fixedProd", (event.get("prod")).tr("\r", ""))'
    }

   ruby {
    code => 'event.set("cat", "BCP")'
    }

    mutate {
      convert => {
         "sender" => "string"
         "receiver" => "string"
         "flow_type" => "string"
         "size_start" => "integer"
         "size_end" =>  "integer"
         "format_start" => "string"
         "format_end" => "string"
         "platform_start" => "string"
         "platform_end" => "string"
         "date_epoch_end" => "float"
         "date_epoch_start" => "float"
         "transport_start" => "string"
         "transport_end" => "string"
         "prod" => "string"
         }
    }

    prune{
      blacklist_names => ["message","date_end_formated_cut","date_start_formated_cut","date_start_formated","date_end_formated","date_start","date_end"]
    }

}

output {
  stdout{}
  elasticsearch {
    hosts => "http://XXXXXX:9200"
    index => "index_bcp"
    user => "XXXXXXX"
    password => "XXXXXXX"
    ssl_certificate_verification => false
  }
}

Both hosts, user and password are the same.

Greninja_San · February 16, 2022, 1:09pm

The 2nd one is the one I already gave.

leandrojmp · February 16, 2022, 1:25pm

That's the issue.

Your pipeline.yml has only one pipeline pointing to /etc/logstash/conf.d/*.conf, so when Logstash starts all the files in this directory will be merged and the events will pass through every filter and output.

In reality what you have is something like this:

input {
  file {
    path => "/srv/edm/ftp/not_reversed/*.csv"
    start_position => "beginning"
    ignore_older => 17280000
    sincedb_path => "/dev/null"
    codec => plain {
      charset => "ANSI_X3.4-1968"
    }
  }
  file {
    path => "/srv/edm/ftp/bcp/*.csv"
    start_position => "beginning"
    ignore_older => 17280000
    sincedb_path => "/dev/null"
    codec => plain {
      charset => "ANSI_X3.4-1968"
    }
  }
}
filter {
    all your filters
}
output {
  stdout{}
  elasticsearch {
    hosts => "http://XXXXXXX:9200"
    index => "index_bcp"
    user => "XXXXX"
    password => "XXXXXXX"
    ssl_certificate_verification => false
  }
  stdout{}
  elasticsearch {
    hosts => "http://XXXXXX:9200"
    index => "index_bcp"
    user => "XXXXXXX"
    password => "XXXXXXX"
    ssl_certificate_verification => false
  }
}

Since you use the same index, you have your output twice, this is what get your documents duplicated.

You should isolate your pipelines changing your pipelines.yml to something similar as this:

- pipeline.id: "pipeline-one"
  path.config: "/etc/logstash/conf.d/file-one.conf"

- pipeline.id: "pipeline-two"
  path.config: "/etc/logstash/conf.d/file-two.conf

This will isolate your pipelines and you will have just one output per pipeline.

Greninja_San · February 16, 2022, 2:06pm

That was the solution! Thank you very much <3

system · March 16, 2022, 2:06pm

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

Topic		Replies	Views
Import CSV into Elasticsearch with Logstash issue Logstash	6	1626	May 2, 2018
How can i send duplicate lines (from logstash to elasticsearch) in log file Logstash	6	1092	April 13, 2020
Problem beetween elastic and logstash: csv files are saved twice Logstash	6	552	July 20, 2020
Duplicate log entries Elasticsearch	18	3697	January 20, 2021
Allow duplicate entires into logstash Logstash	3	450	February 6, 2018

Logstash ingest and export to elasticsearch files twice

Related topics