fix(pkg/stanza/input/file/reader): skip building fingerprint in case of configuration change

[sumo-drosiek]

Description:

If customer changes fingerprint_size while offset is behind the value,
it is going to panic as it is unable to rebuild fingerprint

Reproduction:
logs/file.txt

this is some file for testing purposes

use this configuration

extensions:
  file_storage:
    directory: ./
    timeout: 10s
receivers:
  filelog:
    include:
      - logs/file.txt
    fingerprint_size: 16
    start_at: beginning
exporters:
  logging:
    logLevel: debug
service:
  extensions:
    file_storage
  pipelines:
    logs:
      receivers:
        - filelog
      exporters:
        - logging

run again with ioncreased fingerprint_size

extensions:
  file_storage:
    directory: ./
    timeout: 10s
receivers:
  filelog:
    include:
      - logs/file.txt
    fingerprint_size: 17
    start_at: beginning
exporters:
  logging:
    logLevel: debug
service:
  extensions:
    file_storage
  pipelines:
    logs:
      receivers:
        - filelog
      exporters:
        - logging

The error occures:

2022-06-01T17:19:54.955+0200    info    file/file.go:174        Started watching file   {"kind": "receiver", "name": "filelog", "operator_id": "file_input", "operator_type": "file_input", "path": "logs/file.txt"}
panic: runtime error: slice bounds out of range [:39] with capacity 18

goroutine 141 [running]:
github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/operator/input/file.(*Reader).Read(0xc000aa8a20, {0xc000cee000, 0x27?, 0x4000})
        github.com/open-telemetry/opentelemetry-collector-contrib/pkg/[email protected]/operator/input/file/reader.go:234 +0x190
bufio.(*Scanner).Scan(0xc000bc4f80)
        bufio/scan.go:215 +0x865
github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/operator/input/file.(*Reader).ReadToEnd(0xc000aa8a20, {0x53ba918, 0xc000c68680})
        github.com/open-telemetry/opentelemetry-collector-contrib/pkg/[email protected]/operator/input/file/reader.go:134 +0x216
github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/operator/input/file.(*InputOperator).poll.func1(0xc00043cf90?)
        github.com/open-telemetry/opentelemetry-collector-contrib/pkg/[email protected]/operator/input/file/file.go:155 +0x5c
created by github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/operator/input/file.(*InputOperator).poll
        github.com/open-telemetry/opentelemetry-collector-contrib/pkg/[email protected]/operator/input/file/file.go:153 +0x198

Link to tracking Issue: N/A

Testing: ToDo

Documentation: in code comments

[sumo-drosiek]

This is draft as I want to add some unit tests to the PR

[sumo-drosiek]

Not sure, how this change affects this code:

opentelemetry-collector-contrib/pkg/stanza/operator/input/file/fingerprint.go

Lines 59 to 75 in 90efffd

	// StartsWith returns true if the fingerprints are the same
	// or if the new fingerprint starts with the old one
	// This is important functionality for tracking new files,
	// since their initial size is typically less than that of
	// a fingerprint. As the file grows, its fingerprint is updated
	// until it reaches a maximum size, as configured on the operator
	func (f Fingerprint) StartsWith(old *Fingerprint) bool {
	l0 := len(old.FirstBytes)
	if l0 == 0 {
	return false
	}
	l1 := len(f.FirstBytes)
	if l0 > l1 {
	return false
	}
	return bytes.Equal(old.FirstBytes[:l0], f.FirstBytes[:l0])
	}

There shouldn't be an issue for increasing the fingerprint size. I will test it for decreasing the fingerprint size, but I consider PR ready to review

[sumo-drosiek]

for decreasing the fingerprint_size, file is going to be read again, which is not a new behavior and I would rather resolve it in separate issue

@djaglowski WDYT?

[djaglowski]

Am I correct in thinking that this only affects fingerprints that have been loaded from a storage extension? If so, I wonder if we could detect this and normalize the fingerprints during startup. This would seem to be preferable to adding complexity to the ongoing read loop.

[sumo-drosiek]

Am I correct in thinking that this only affects fingerprints that have been loaded from a storage extension?

Yes, it affects only fingerprints which have been loaded from storage

If so, I wonder if we could detect this and normalize the fingerprints during startup.

So, as I understand flow correctly, you propose to add logic to this function:

opentelemetry-collector-contrib/pkg/stanza/operator/input/file/file.go

Lines 324 to 361 in c522cef

	func (f *Input) loadLastPollFiles(ctx context.Context) error {
	encoded, err := f.persister.Get(ctx, knownFilesKey)
	if err != nil {
	return err
	}
	if encoded == nil {
	f.knownFiles = make([]*Reader, 0, 10)
	return nil
	}
	dec := json.NewDecoder(bytes.NewReader(encoded))
	// Decode the number of entries
	var knownFileCount int
	if err := dec.Decode(&knownFileCount); err != nil {
	return fmt.Errorf("decoding file count: %w", err)
	}
	// Decode each of the known files
	f.knownFiles = make([]*Reader, 0, knownFileCount)
	for i := 0; i < knownFileCount; i++ {
	splitter, err := f.getMultiline()
	if err != nil {
	return err
	}
	newReader, err := f.NewReader("", nil, nil, splitter)
	if err != nil {
	return err
	}
	if err = dec.Decode(newReader); err != nil {
	return err
	}
	f.knownFiles = append(f.knownFiles, newReader)
	}
	return nil
	}

[djaglowski]

I think in theory this case would be better handled in loadLastPollFiles, but having refreshed myself on the code somewhat, it's going to take some substantial refactoring or duplication to make that possible. I think what you have makes sense.

Comparing benchmarks before and after the shows no notable change:

## main

pkg: github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/operator/input/file
BenchmarkFileInput/Single-10         	  913302	      1138 ns/op	     672 B/op	       9 allocs/op
BenchmarkFileInput/Glob-10           	  245481	      4946 ns/op	    2693 B/op	      36 allocs/op
BenchmarkFileInput/MultiGlob-10      	  245552	      4919 ns/op	    2693 B/op	      36 allocs/op
BenchmarkFileInput/MaxConcurrent-10  	  140121	      7163 ns/op	    2690 B/op	      36 allocs/op
BenchmarkFileInput/FngrPrntLarge-10  	  924232	      1122 ns/op	     672 B/op	       9 allocs/op
BenchmarkFileInput/FngrPrntSmall-10  	  915962	      1138 ns/op	     672 B/op	       9 allocs/op
PASS
coverage: 71.8% of statements
ok  	github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/operator/input/file	52.988s


## This branch

pkg: github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/operator/input/file
BenchmarkFileInput/Single-10         	  913993	      1136 ns/op	     672 B/op	       9 allocs/op
BenchmarkFileInput/Glob-10           	  241573	      4992 ns/op	    2693 B/op	      36 allocs/op
BenchmarkFileInput/MultiGlob-10      	  239896	      4404 ns/op	    2692 B/op	      36 allocs/op
BenchmarkFileInput/MaxConcurrent-10  	  140404	      7155 ns/op	    2690 B/op	      36 allocs/op
BenchmarkFileInput/FngrPrntLarge-10  	  919200	      1134 ns/op	     672 B/op	       9 allocs/op
BenchmarkFileInput/FngrPrntSmall-10  	 1161418	      1087 ns/op	     672 B/op	       9 allocs/op
PASS
coverage: 72.0% of statements
ok  	github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/operator/input/file	52.935s

Only thing I would ask is that new test cases be added rather than adding to the existing "GrowAndStop" test.

[sumo-drosiek]

Only thing I would ask is that new test cases be added rather than adding to the existing "GrowAndStop" test.

Sure, I was considering that, but it will be mostly copy of the GrowAndStop