Fix ReadFile handler to consider the value stored in sincedb on plugin restart

[andsel]

Release notes

Fixes read mode when sincedb already stores a reference for the file not completely consumed.

What does this PR do?

Update the file pointer of a read mode file to the max between the read bytes or the sincedb reference for the same file.
This solves a problem, that when a pipeline is restarted, it's able to recover from the last known reference, without restarting from the beginning, and reprocessing already processed lines.

Why is it important/What is the impact to the user?

When a pipeline with file input in read mode is restarted, this let the plugin to recover from where it left if that information is present in the sincedb store.

Checklist

• My code follows the style guidelines of this project
• I have commented my code, particularly in hard-to-understand areas
• [ ] I have made corresponding changes to the documentation
• [ ] I have made corresponding change to the default configuration files (and/or docker env variables)
• I have added tests that prove my fix is effective or that my feature works

Author's Checklist

• verify with the steps used in the bug report Ability to terminate pipeline when EOF reached  #240. I used the following test file:
  sample_fixture.csv.txt

Pipeline definition:

- pipeline.id: SDH_650
  pipeline.workers: 1
  pipeline.batch.size: 5
  config.string: |
    input {
        file {
            path => "/home/andrea/workspace/logstash_configs/file_input_sdh650/sample_fixture.csv"
            sincedb_path => "/home/andrea/workspace/logstash_configs/file_input_sdh650/sincedb"
            mode  => "read"
            start_position => "beginning"
        }
    }

    filter {
        csv {
            separator => ","
            columns => ["id", "host", "fqdn", "IP", "mac", "role", "type", "make", "model", "oid", "fid", "time"]
            remove_field => ["path", "host", "message", "@version" ]   
        }
        sleep {
            time => 1
            every => 10
        }
    }

    output {
        elasticsearch { 
            index => "650" 
            hosts => "http://localhost:9200"
            user => "elastic"
            password => "changeme"
        }
        stdout { codec => dots }
    }

Some curls to configure the ES output index and an aggregation query to verify:

PUT /650
{
  "mappings": {
    "properties": {
      "id":    { "type": "keyword" },  
      "host":  { "type": "text"  }, 
      "fqdn":   { "type": "text"  },
      "IP":   { "type": "text"  },
      "mac":   { "type": "text"  },
      "role":   { "type": "keyword"  },
      "type":   { "type": "keyword"  },
      "make":   { "type": "text"  },
      "model":   { "type": "text"  },
      "oid":   { "type": "text"  },
      "fid":   { "type": "text"  },
      "time":   { "type": "text"  }
    }
  }
}
DELETE 650

GET 650/_search
{
  "aggs": {
    "types": {
      "terms": { "field": "type" }
    }
  }
}

The expectation is to have 2 buckets, equally sized. Without the fix a bucket contains more documents, which means some rows was reprocessed on a pipeline reload.

How to test this PR locally

Follow step steps in #290

Related issues

• Fixes unable to read the whole file when pipeline get reload #290

Use cases

Screenshots

Logs

[jsvd]

Since add_or_update_sincedb_collection ensures the file is in the sincedb and sets the correct position from the sincedb to the watched_file, can't we simplify this a lot to just be:

Suggested change

	if open_file(watched_file)
	add_or_update_sincedb_collection(watched_file) unless sincedb_collection.member?(watched_file.sincedb_key)
	if sincedb_collection.member?(watched_file.sincedb_key)
	previous_pos = sincedb_collection.find(watched_file).position
	watched_file.file_seek([watched_file.bytes_read, previous_pos].max)
	end
	loop do
	if open_file(watched_file)
	add_or_update_sincedb_collection(watched_file)
	watched_file.file_seek(watched_file.bytes_read)
	loop do

[andsel]

@jsvd good point! the semantic remains the same and the flow is simplified

[andsel]

It fails the test https://app.travis-ci.com/github/logstash-plugins/logstash-input-file/jobs/566809162#L974 because without the max(watched_file.bytes_read, sincedb previous_pos) it seems that bytes_read hasn't the same value of position. Need to investigate why.

[andsel]

Without the guard

add_or_update_sincedb_collection(watched_file) unless sincedb_collection.member?

it happens that watched_file.bytes_read is always updated to the incedb_collection.find(watched_file).position.
Now this position comes from the sinceDB and it's the last pin point.

What happens in this test? The objective of the test is to verify stripped-reads from a couple of files:
file1

string1\nstring2

file2

stringA\nstringB

Striped reads means that it reads one line for each one, so the position point \n character.
The test fails because has a file_chunk_size is 10 bytes, so the first run for the first file it reads 10 bytes and put in a buffer, which is:

string1\nst

if seek set filepointer to the filepoistion (8th bytes) then the next chuck grabbed from the file is

string2

which goes in the buffer which contained st creating the string ststring2.
This is the reason to have the max(bytes_read, last_pos).

[kaisecheng]

I have tested it locally. Both solutions work well to fix repeat read in restart. CI has been breaking for some time before this change.

[andsel]

Hold on this PR till the CI is back to green on main, then rebase and ask for review again.

[jsvd]

LGTM