Make implementing Close required by reader.Readers

Closes elastic#19193

filebeat/input/log/harvester.go

@@ -291,7 +291,7 @@ func (h *Harvester) Run() error {
 		}
 
 		h.stop()
-		h.log.Close()
+		h.reader.Close()
 	}(h.state.Source)
 
 	logp.Info("Harvester started for file: %s", h.state.Source)

filebeat/input/log/log.go

@@ -208,7 +208,8 @@ func (f *Log) wait() {
 }
 
 // Close closes the done channel but no th the file handler
-func (f *Log) Close() {
+func (f *Log) Close() error {
 	close(f.done)
 	// Note: File reader is not closed here because that leads to race conditions
+	return nil
 }

libbeat/reader/debug/debug.go

@@ -46,7 +46,7 @@ type CheckFunc func(offset int64, buf []byte) bool
 // Is is useful is you want to detect if you have received garbage from a network volume.
 type Reader struct {
 	log           *logp.Logger
-	reader        io.Reader
+	reader        io.ReadCloser
 	buffer        bytes.Buffer
 	minBufferSize int
 	maxFailures   int
@@ -59,7 +59,7 @@ type Reader struct {
 // NewReader returns a debug reader.
 func NewReader(
 	log *logp.Logger,
-	reader io.Reader,
+	reader io.ReadCloser,
 	minBufferSize int,
 	maxFailures int,
 	predicate CheckFunc,
@@ -115,6 +115,10 @@ func (r *Reader) Read(p []byte) (int, error) {
 	return n, err
 }
 
+func (r *Reader) Close() error {
+	return r.reader.Close()
+}
+
 func makeNullCheck(log *logp.Logger, minSize int) CheckFunc {
 	// create a slice with null bytes to match on the buffer.
 	pattern := make([]byte, minSize, minSize)
@@ -159,7 +163,7 @@ func summarizeBufferInfo(idx int, buf []byte) (int, []byte) {
 
 // AppendReaders look into the current enabled log selector and will add any debug reader that match
 // the selectors.
-func AppendReaders(reader io.Reader) (io.Reader, error) {
+func AppendReaders(reader io.ReadCloser) (io.ReadCloser, error) {
 	var err error
 
 	if logp.HasSelector("detect_null_bytes") || logp.HasSelector("*") {

libbeat/reader/multiline/counter.go

@@ -131,3 +131,7 @@ func (cr *counterReader) resetState() {
 func (cr *counterReader) setState(next func(cr *counterReader) (reader.Message, error)) {
 	cr.state = next
 }
+
+func (cr *counterReader) Close() error {
+	return cr.reader.Close()
+}

libbeat/reader/multiline/pattern.go

@@ -254,6 +254,10 @@ func (pr *patternReader) setState(next func(pr *patternReader) (reader.Message,
 	pr.state = next
 }
 
+func (pr *patternReader) Close() error {
+	return pr.reader.Close()
+}
+
 // matchers
 func afterMatcher(pat match.Matcher) (matcher, error) {
 	return genPatternMatcher(pat, func(last, current []byte) []byte {

libbeat/reader/reader.go

@@ -19,13 +19,15 @@ package reader
 
 import (
 	"errors"
+	"io"
 )
 
 // Reader is the interface that wraps the basic Next method for
 // getting a new message.
 // Next returns the message being read or and error. EOF is returned
 // if reader will not return any new message on subsequent calls.
 type Reader interface {
+	io.Closer
 	Next() (Message, error)
 }
 

libbeat/reader/readfile/encode.go

@@ -43,7 +43,7 @@ type Config struct {
 
 // New creates a new Encode reader from input reader by applying
 // the given codec.
-func NewEncodeReader(r io.Reader, config Config) (EncoderReader, error) {
+func NewEncodeReader(r io.ReadCloser, config Config) (EncoderReader, error) {
 	eReader, err := NewLineReader(r, config)
 	return EncoderReader{eReader}, err
 }
@@ -59,3 +59,7 @@ func (r EncoderReader) Next() (reader.Message, error) {
 		Bytes:   sz,
 	}, err
 }
+
+func (r EncoderReader) Close() error {
+	return r.reader.Close()
+}

libbeat/reader/readfile/limit.go

@@ -49,3 +49,7 @@ func (r *LimitReader) Next() (reader.Message, error) {
 	}
 	return message, err
 }
+
+func (r *LimitReader) Close() error {
+	return r.reader.Close()
+}

libbeat/reader/readfile/line.go

@@ -34,7 +34,7 @@ const unlimited = 0
 // using the configured codec. The reader keeps track of bytes consumed
 // from raw input stream for every decoded line.
 type LineReader struct {
-	reader     io.Reader
+	reader     io.ReadCloser
 	bufferSize int
 	maxBytes   int // max bytes per line limit to avoid OOM with malformatted files
 	nl         []byte
@@ -48,7 +48,7 @@ type LineReader struct {
 }
 
 // New creates a new reader object
-func NewLineReader(input io.Reader, config Config) (*LineReader, error) {
+func NewLineReader(input io.ReadCloser, config Config) (*LineReader, error) {
 	encoder := config.Codec.NewEncoder()
 
 	// Create newline char based on encoding
@@ -271,3 +271,7 @@ func (r *LineReader) decode(end int) (int, error) {
 	r.byteCount += start
 	return start, err
 }
+
+func (r *LineReader) Close() error {
+	return r.reader.Close()
+}

libbeat/reader/readfile/strip_newline.go

@@ -81,3 +81,7 @@ func (p *StripNewline) autoLineEndingChars(l []byte) int {
 	}
 	return 1
 }
+
+func (p *StripNewline) Close() error {
+	return p.reader.Close()
+}

libbeat/reader/readfile/timeout.go

@@ -36,6 +36,7 @@ type TimeoutReader struct {
 	signal  error
 	running bool
 	ch      chan lineMessage
+	done    chan struct{}
 }
 
 type lineMessage struct {
@@ -54,6 +55,7 @@ func NewTimeoutReader(reader reader.Reader, signal error, t time.Duration) *Time
 		signal:  signal,
 		timeout: t,
 		ch:      make(chan lineMessage, 1),
+		done:    make(chan struct{}),
 	}
 }
 
@@ -67,6 +69,11 @@ func (r *TimeoutReader) Next() (reader.Message, error) {
 		r.running = true
 		go func() {
 			for {
+				select {
+				case <-r.done:
+					return
+				default:
+				}
 				message, err := r.reader.Next()
 				r.ch <- lineMessage{message, err}
 				if err != nil {
@@ -85,5 +92,13 @@ func (r *TimeoutReader) Next() (reader.Message, error) {
 		return msg.line, msg.err
 	case <-timer.C:
 		return reader.Message{}, r.signal
+	case <-r.done:
+		return reader.Message{}, nil
 	}
 }
+
+func (r *TimeoutReader) Close() error {
+	close(r.done)
+
+	return r.reader.Close()
+}

libbeat/reader/readjson/docker_json.go

@@ -244,3 +244,7 @@ func stripNewLineWin(msg *reader.Message) {
 		return r == '\n' || r == '\r'
 	})
 }
+
+func (p *DockerJSONReader) Close() error {
+	return p.reader.Close()
+}

libbeat/reader/readjson/json.go

@@ -111,6 +111,10 @@ func (r *JSONReader) Next() (reader.Message, error) {
 	return message, nil
 }
 
+func (r *JSONReader) Close() error {
+	return r.reader.Close()
+}
+
 func createJSONError(message string) common.MapStr {
 	return common.MapStr{"message": message, "type": "json"}
 }