Add possibility to force flush logs after certain period of time

operator/builtin/input/file/config.go

@@ -117,7 +117,8 @@ func (c InputConfig) Build(context operator.BuildContext) ([]operator.Operator,
 		return nil, err
 	}
 
-	splitFunc, err := c.Multiline.Build(context, encoding.Encoding, false)
+	// Ensure that multiline is buildable
+	_, err = c.Multiline.Build(encoding.Encoding, false)
 	if err != nil {
 		return nil, err
 	}
@@ -156,7 +157,7 @@ func (c InputConfig) Build(context operator.BuildContext) ([]operator.Operator,
 		InputOperator:         inputOperator,
 		Include:               c.Include,
 		Exclude:               c.Exclude,
-		SplitFunc:             splitFunc,
+		Multiline:             c.Multiline,
 		PollInterval:          c.PollInterval.Raw(),
 		FilePathField:         filePathField,
 		FileNameField:         fileNameField,

operator/builtin/input/file/file.go

@@ -15,7 +15,6 @@
 package file
 
 import (
-	"bufio"
 	"bytes"
 	"context"
 	"encoding/json"
@@ -44,7 +43,7 @@ type InputOperator struct {
 	FilePathResolvedField entry.Field
 	FileNameResolvedField entry.Field
 	PollInterval          time.Duration
-	SplitFunc             bufio.SplitFunc
+	Multiline             helper.MultilineConfig
 	MaxLogSize            int
 	MaxConcurrentFiles    int
 	SeenPaths             map[string]struct{}
@@ -323,7 +322,11 @@ func (f *InputOperator) newReader(file *os.File, fp *Fingerprint, firstCheck boo
 	}
 
 	// If we don't match any previously known files, create a new reader from scratch
-	newReader, err := f.NewReader(file.Name(), file, fp)
+	multiline, err := f.getMultiline()
+	if err != nil {
+		return nil, err
+	}
+	newReader, err := f.NewReader(file.Name(), file, fp, multiline)
 	if err != nil {
 		return nil, err
 	}
@@ -393,7 +396,11 @@ func (f *InputOperator) loadLastPollFiles(ctx context.Context) error {
 	// Decode each of the known files
 	f.knownFiles = make([]*Reader, 0, knownFileCount)
 	for i := 0; i < knownFileCount; i++ {
-		newReader, err := f.NewReader("", nil, nil)
+		multiline, err := f.getMultiline()
+		if err != nil {
+			return err
+		}
+		newReader, err := f.NewReader("", nil, nil, multiline)
 		if err != nil {
 			return err
 		}
@@ -405,3 +412,7 @@ func (f *InputOperator) loadLastPollFiles(ctx context.Context) error {
 
 	return nil
 }
+
+func (f *InputOperator) getMultiline() (*helper.Multiline, error) {
+	return f.Multiline.Build(f.encoding.Encoding, false)
+}

operator/builtin/input/file/file_test.go

@@ -625,7 +625,11 @@ func TestFileReader_FingerprintUpdated(t *testing.T) {
 	tempCopy := openFile(t, temp.Name())
 	fp, err := operator.NewFingerprint(temp)
 	require.NoError(t, err)
-	reader, err := operator.NewReader(temp.Name(), tempCopy, fp)
+
+	multiline, err := operator.getMultiline()
+	require.NoError(t, err)
+
+	reader, err := operator.NewReader(temp.Name(), tempCopy, fp, multiline)
 	require.NoError(t, err)
 	defer reader.Close()
 
@@ -666,7 +670,10 @@ func TestFingerprintGrowsAndStops(t *testing.T) {
 			require.NoError(t, err)
 			require.Equal(t, []byte(""), fp.FirstBytes)
 
-			reader, err := operator.NewReader(temp.Name(), tempCopy, fp)
+			multiline, err := operator.getMultiline()
+			require.NoError(t, err)
+
+			reader, err := operator.NewReader(temp.Name(), tempCopy, fp, multiline)
 			require.NoError(t, err)
 			defer reader.Close()
 

operator/builtin/input/file/reader.go

@@ -26,6 +26,7 @@ import (
 	"golang.org/x/text/transform"
 
 	"github.com/open-telemetry/opentelemetry-log-collection/errors"
+	"github.com/open-telemetry/opentelemetry-log-collection/operator/helper"
 )
 
 // File attributes contains information about file paths
@@ -70,11 +71,13 @@ type Reader struct {
 	decoder      *encoding.Decoder
 	decodeBuffer []byte
 
+	multiline *helper.Multiline
+
 	*zap.SugaredLogger `json:"-"`
 }
 
 // NewReader creates a new file reader
-func (f *InputOperator) NewReader(path string, file *os.File, fp *Fingerprint) (*Reader, error) {
+func (f *InputOperator) NewReader(path string, file *os.File, fp *Fingerprint, multiline *helper.Multiline) (*Reader, error) {
 	r := &Reader{
 		Fingerprint:    fp,
 		file:           file,
@@ -83,13 +86,14 @@ func (f *InputOperator) NewReader(path string, file *os.File, fp *Fingerprint) (
 		decoder:        f.encoding.Encoding.NewDecoder(),
 		decodeBuffer:   make([]byte, 1<<12),
 		fileAttributes: f.resolveFileAttributes(path),
+		multiline:      multiline,
 	}
 	return r, nil
 }
 
 // Copy creates a deep copy of a Reader
 func (r *Reader) Copy(file *os.File) (*Reader, error) {
-	reader, err := r.fileInput.NewReader(r.fileAttributes.Path, file, r.Fingerprint.Copy())
+	reader, err := r.fileInput.NewReader(r.fileAttributes.Path, file, r.Fingerprint.Copy(), r.multiline)
 	if err != nil {
 		return nil, err
 	}
@@ -116,7 +120,7 @@ func (r *Reader) ReadToEnd(ctx context.Context) {
 		return
 	}
 
-	scanner := NewPositionalScanner(r, r.fileInput.MaxLogSize, r.Offset, r.fileInput.SplitFunc)
+	scanner := NewPositionalScanner(r, r.fileInput.MaxLogSize, r.Offset, r.multiline.SplitFunc)
 
 	// Iterate over the tokenized file, emitting entries as we go
 	for {
@@ -131,8 +135,13 @@ func (r *Reader) ReadToEnd(ctx context.Context) {
 			if err := getScannerError(scanner); err != nil {
 				r.Errorw("Failed during scan", zap.Error(err))
 			}
+
+			// Force flush eventually in next iteration
+			r.multiline.CheckAndFlush()
 			break
 		}
+		// Update information about last flush time
+		r.multiline.Flushed()
 
 		if err := r.emit(ctx, scanner.Bytes()); err != nil {
 			r.Error("Failed to emit entry", zap.Error(err))

operator/builtin/input/tcp/tcp.go

@@ -98,7 +98,7 @@ func (c TCPInputConfig) Build(context operator.BuildContext) ([]operator.Operato
 		return nil, err
 	}
 
-	splitFunc, err := c.Multiline.Build(context, encoding.Encoding, true)
+	multiline, err := c.Multiline.Build(encoding.Encoding, true)
 	if err != nil {
 		return nil, err
 	}
@@ -114,7 +114,7 @@ func (c TCPInputConfig) Build(context operator.BuildContext) ([]operator.Operato
 		MaxLogSize:    int(c.MaxLogSize),
 		addAttributes: c.AddAttributes,
 		encoding:      encoding,
-		splitFunc:     splitFunc,
+		splitFunc:     multiline.SplitFunc,
 		backoff: backoff.Backoff{
 			Max: 3 * time.Second,
 		},

operator/builtin/input/udp/udp.go

@@ -81,7 +81,7 @@ func (c UDPInputConfig) Build(context operator.BuildContext) ([]operator.Operato
 		return nil, err
 	}
 
-	splitFunc, err := c.Multiline.Build(context, encoding.Encoding, true)
+	multiline, err := c.Multiline.Build(encoding.Encoding, true)
 	if err != nil {
 		return nil, err
 	}
@@ -97,7 +97,7 @@ func (c UDPInputConfig) Build(context operator.BuildContext) ([]operator.Operato
 		buffer:        make([]byte, MaxUDPSize),
 		addAttributes: c.AddAttributes,
 		encoding:      encoding,
-		splitFunc:     splitFunc,
+		splitFunc:     multiline.SplitFunc,
 		resolver:      resolver,
 	}
 	return []operator.Operator{udpInput}, nil

operator/helper/multiline.go

@@ -19,62 +19,125 @@ import (
 	"bytes"
 	"fmt"
 	"regexp"
+	"time"
 
 	"golang.org/x/text/encoding"
-
-	"github.com/open-telemetry/opentelemetry-log-collection/operator"
 )
 
+type ForceFlush struct {
+	Force     bool
+	LastFlush time.Time
+}
+
+func NewForceFlush() *ForceFlush {
+	return &ForceFlush{
+		Force:     false,
+		LastFlush: time.Now(),
+	}
+}
+
+type Multiline struct {
+	SplitFunc   bufio.SplitFunc
+	force       *ForceFlush
+	forcePeriod time.Duration
+
+	// lastFlush > force.LastFlush => we can force flush if no logs are incoming for forcePeriod
+	// lastFlush = force.LastFlush => last flush was forced, so we do cannot force, we can update lastFlush
+	// lastFlush < force.LastFlush =>we just forced flush, set lastFlush to force.LastFlush
+	lastFlush time.Time
+}
+
+// Flushed update lastFlush with current timestamp
+func (m *Multiline) Flushed() {
+	if m.lastFlush.Sub(m.force.LastFlush) < 0 {
+		m.lastFlush = m.force.LastFlush
+	} else {
+		m.lastFlush = time.Now()
+	}
+}
+
+// CheckAndFlush returns true if data is going to be force flushed
+func (m *Multiline) CheckAndFlush() {
+	if m.forcePeriod > 0 && time.Since(m.lastFlush) > m.forcePeriod && m.lastFlush.Sub(m.force.LastFlush) > 0 {
+		m.force.Force = true
+	}
+}
+
 // NewBasicConfig creates a new Multiline config
 func NewMultilineConfig() MultilineConfig {
 	return MultilineConfig{
 		LineStartPattern: "",
 		LineEndPattern:   "",
+		ForceFlushPeriod: "0s",
 	}
 }
 
 // MultilineConfig is the configuration of a multiline helper
 type MultilineConfig struct {
 	LineStartPattern string `mapstructure:"line_start_pattern"  json:"line_start_pattern" yaml:"line_start_pattern"`
 	LineEndPattern   string `mapstructure:"line_end_pattern"    json:"line_end_pattern"   yaml:"line_end_pattern"`
+	ForceFlushPeriod string `mapstructure:"force_flush_period"  json:"force_flush_period" yaml:"force_flush_period"`
 }
 
 // Build will build a Multiline operator.
-func (c MultilineConfig) Build(context operator.BuildContext, encoding encoding.Encoding, flushAtEOF bool) (bufio.SplitFunc, error) {
-	return c.getSplitFunc(encoding, flushAtEOF)
+func (c MultilineConfig) Build(encoding encoding.Encoding, flushAtEOF bool) (*Multiline, error) {
+	force := NewForceFlush()
+	splitFunc, err := c.getSplitFunc(encoding, flushAtEOF, force)
+	if err != nil {
+		return nil, err
+	}
+
+	duration, err := time.ParseDuration(c.ForceFlushPeriod)
+	if err != nil {
+		return nil, err
+	}
+
+	return &Multiline{
+		SplitFunc:   splitFunc,
+		force:       force,
+		forcePeriod: duration,
+	}, nil
 }
 
 // getSplitFunc returns split function for bufio.Scanner basing on configured pattern
-func (c MultilineConfig) getSplitFunc(encoding encoding.Encoding, flushAtEOF bool) (bufio.SplitFunc, error) {
+func (c MultilineConfig) getSplitFunc(encoding encoding.Encoding, flushAtEOF bool, force *ForceFlush) (bufio.SplitFunc, error) {
 	endPattern := c.LineEndPattern
 	startPattern := c.LineStartPattern
 
 	switch {
 	case endPattern != "" && startPattern != "":
 		return nil, fmt.Errorf("only one of line_start_pattern or line_end_pattern can be set")
 	case endPattern == "" && startPattern == "":
-		return NewNewlineSplitFunc(encoding, flushAtEOF)
+		return NewNewlineSplitFunc(encoding, flushAtEOF, force)
 	case endPattern != "":
 		re, err := regexp.Compile("(?m)" + c.LineEndPattern)
 		if err != nil {
 			return nil, fmt.Errorf("compile line end regex: %s", err)
 		}
-		return NewLineEndSplitFunc(re, flushAtEOF), nil
+		return NewLineEndSplitFunc(re, flushAtEOF, force), nil
 	case startPattern != "":
 		re, err := regexp.Compile("(?m)" + c.LineStartPattern)
 		if err != nil {
 			return nil, fmt.Errorf("compile line start regex: %s", err)
 		}
-		return NewLineStartSplitFunc(re, flushAtEOF), nil
+		return NewLineStartSplitFunc(re, flushAtEOF, force), nil
 	default:
 		return nil, fmt.Errorf("unreachable")
 	}
 }
 
 // NewLineStartSplitFunc creates a bufio.SplitFunc that splits an incoming stream into
 // tokens that start with a match to the regex pattern provided
-func NewLineStartSplitFunc(re *regexp.Regexp, flushAtEOF bool) bufio.SplitFunc {
+func NewLineStartSplitFunc(re *regexp.Regexp, flushAtEOF bool, force *ForceFlush) bufio.SplitFunc {
 	return func(data []byte, atEOF bool) (advance int, token []byte, err error) {
+		if force.Force {
+			force.Force = false
+			force.LastFlush = time.Now()
+			token = trimWhitespaces(data)
+			advance = len(data)
+			return
+		}
+
 		firstLoc := re.FindIndex(data)
 		if firstLoc == nil {
 			// Flush if no more data is expected
@@ -123,8 +186,15 @@ func NewLineStartSplitFunc(re *regexp.Regexp, flushAtEOF bool) bufio.SplitFunc {
 
 // NewLineEndSplitFunc creates a bufio.SplitFunc that splits an incoming stream into
 // tokens that end with a match to the regex pattern provided
-func NewLineEndSplitFunc(re *regexp.Regexp, flushAtEOF bool) bufio.SplitFunc {
+func NewLineEndSplitFunc(re *regexp.Regexp, flushAtEOF bool, force *ForceFlush) bufio.SplitFunc {
 	return func(data []byte, atEOF bool) (advance int, token []byte, err error) {
+		if force.Force {
+			force.Force = false
+			force.LastFlush = time.Now()
+			token = trimWhitespaces(data)
+			advance = len(data)
+			return
+		}
 		loc := re.FindIndex(data)
 		if loc == nil {
 			// Flush if no more data is expected
@@ -151,7 +221,7 @@ func NewLineEndSplitFunc(re *regexp.Regexp, flushAtEOF bool) bufio.SplitFunc {
 
 // NewNewlineSplitFunc splits log lines by newline, just as bufio.ScanLines, but
 // never returning an token using EOF as a terminator
-func NewNewlineSplitFunc(encoding encoding.Encoding, flushAtEOF bool) (bufio.SplitFunc, error) {
+func NewNewlineSplitFunc(encoding encoding.Encoding, flushAtEOF bool, force *ForceFlush) (bufio.SplitFunc, error) {
 	newline, err := encodedNewline(encoding)
 	if err != nil {
 		return nil, err
@@ -163,6 +233,13 @@ func NewNewlineSplitFunc(encoding encoding.Encoding, flushAtEOF bool) (bufio.Spl
 	}
 
 	return func(data []byte, atEOF bool) (advance int, token []byte, err error) {
+		if force.Force {
+			force.Force = false
+			force.LastFlush = time.Now()
+			token = trimWhitespaces(data)
+			advance = len(data)
+			return
+		}
 		if atEOF && len(data) == 0 {
 			return 0, nil, nil
 		}