add auto option to get the default line terminator behaviour

filebeat/_meta/common.reference.inputs.yml

@@ -80,9 +80,9 @@ filebeat.inputs:
   # This is especially useful for multiline log messages which can get large.
   #max_bytes: 10485760
 
-  # Characters which separate the lines. Valid values: line_feed, vertical_tab, form_feed,
+  # Characters which separate the lines. Valid values: auto, line_feed, vertical_tab, form_feed,
   # carriage_return, carriage_return_line_feed, next_line, line_separator, paragraph_separator.
-  #line_terminator: line_feed
+  #line_terminator: auto
 
   ### Recursive glob configuration
 

filebeat/filebeat.reference.yml

@@ -485,9 +485,9 @@ filebeat.inputs:
   # This is especially useful for multiline log messages which can get large.
   #max_bytes: 10485760
 
-  # Characters which separate the lines. Valid values: line_feed, vertical_tab, form_feed,
+  # Characters which separate the lines. Valid values: auto, line_feed, vertical_tab, form_feed,
   # carriage_return, carriage_return_line_feed, next_line, line_separator, paragraph_separator.
-  #line_terminator: line_feed
+  #line_terminator: auto
 
   ### Recursive glob configuration
 

filebeat/input/log/config.go

@@ -58,7 +58,7 @@ var (
 		// Harvester
 		BufferSize:     16 * humanize.KiByte,
 		MaxBytes:       10 * humanize.MiByte,
-		LineTerminator: readfile.LineFeed,
+		LineTerminator: readfile.AutoLineTerminator,
 		LogConfig: LogConfig{
 			Backoff:       1 * time.Second,
 			BackoffFactor: 2,

filebeat/input/log/harvester.go

@@ -34,7 +34,6 @@ import (
 	"fmt"
 	"io"
 	"os"
-	"runtime"
 	"sync"
 	"time"
 
@@ -120,13 +119,8 @@ func NewHarvester(
 		return nil, err
 	}
 
-	cfg := defaultConfig
-	if runtime.GOOS == "windows" {
-		cfg.LineTerminator = readfile.CarriageReturnLineFeed
-	}
-
 	h := &Harvester{
-		config:        cfg,
+		config:        defaultConfig,
 		state:         state,
 		states:        states,
 		publishState:  publishState,

filebeat/tests/system/config/filebeat_inputs.yml.j2

@@ -4,7 +4,6 @@ filebeat.inputs:
     - {{input.path}}
   scan_frequency: 0.5s
   encoding: {{input.encoding | default("plain") }}
-  line_terminator: {{ input.line_terminator }}
 {% endfor %}
 
 {% if not skip_registry_config %}

filebeat/tests/system/test_crawler.py

@@ -504,8 +504,7 @@ def test_utf8(self):
 
         self.render_config_template(
             path=os.path.abspath(self.working_dir) + "/log/*",
-            encoding="utf-8",
-            line_terminator="line_feed",
+            encoding="utf-8"
         )
         os.mkdir(self.working_dir + "/log/")
 
@@ -579,12 +578,11 @@ def test_encodings(self):
         for enc_go, enc_py, _ in encodings:
             inputs.append({
                 "path": self.working_dir + "/log/test-{}".format(enc_py),
-                "encoding": enc_go,
-                "line_terminator": "line_feed",
+                "encoding": enc_go
             })
         self.render_config_template(
             template_name="filebeat_inputs",
-            inputs=inputs,
+            inputs=inputs
         )
 
         # run filebeat

filebeat/tests/system/test_harvester.py

@@ -221,7 +221,6 @@ def test_empty_lines_only(self):
         """
         Checks that no empty events are sent for a file with only empty lines
         """
-
         self.render_config_template(
             path=os.path.abspath(self.working_dir) + "/log/test.log",
         )
@@ -261,7 +260,6 @@ def test_exceed_buffer(self):
         """
         Checks that also full line is sent if lines exceeds buffer
         """
-
         self.render_config_template(
             path=os.path.abspath(self.working_dir) + "/log/test.log",
             harvester_buffer_size=10,
@@ -442,7 +440,6 @@ def test_bom_utf8(self):
         """
         self.render_config_template(
             path=os.path.abspath(self.working_dir) + "/log/*",
-            line_terminator="carriage_return",
         )
 
         os.mkdir(self.working_dir + "/log/")
@@ -480,7 +477,6 @@ def test_boms(self, fb_encoding, py_encoding, bom):
             path=os.path.abspath(self.working_dir) + "/log/" + fb_encoding + "*",
             encoding=fb_encoding,
             output_file_filename=fb_encoding,
-            line_terminator="line_feed",
         )
 
         logfile = self.working_dir + "/log/" + fb_encoding + "test.log"
@@ -790,19 +786,18 @@ def test_decode_error(self):
         self.render_config_template(
             path=os.path.abspath(self.working_dir) + "/log/*",
             encoding="utf-16be",
-            line_terminator="line_feed",
         )
 
         os.mkdir(self.working_dir + "/log/")
 
         logfile = self.working_dir + "/log/test.log"
 
-        with io.open(logfile, 'w', encoding="utf-16le", newline="\n") as file:
+        with io.open(logfile, 'w', encoding="utf-16le") as file:
             file.write(u'hello world1')
             file.write(u"\n")
-        with io.open(logfile, 'a', encoding="utf-16le", newline="\n") as file:
-            file.write(u"\u00012345=Ra")
-        with io.open(logfile, 'a', encoding="utf-16le", newline="\n") as file:
+        with io.open(logfile, 'a', encoding="utf-16le") as file:
+            file.write(u"\U00012345=Ra")
+        with io.open(logfile, 'a', encoding="utf-16le") as file:
             file.write(u"\n")
             file.write(u"hello world2")
             file.write(u"\n")
@@ -861,25 +856,3 @@ def test_debug_reader(self):
             'Matching null byte found at offset (13|14)')), max_timeout=5)
 
         filebeat.check_kill_and_wait()
-
-    def test_line_terminator(self):
-        """
-        Test if line_terminator can be configured to vertical tab
-        """
-
-        self.render_config_template(
-            path=os.path.abspath(self.beat_path) + "/tests/system/input/test-newline.log",
-            line_terminator="vertical_tab",
-        )
-
-        filebeat = self.start_beat()
-
-        messages = [
-            "\"message\": \"hello world\"",
-            "\"message\": \"goodbye world\"",
-        ]
-
-        for message in messages:
-            self.wait_until(lambda: self.log_contains(message))
-
-        filebeat.check_kill_and_wait()

filebeat/tests/system/test_input.py

@@ -609,7 +609,6 @@ def test_restart_recursive_glob(self):
         """
         Check that file reading via recursive glob patterns continues after restart
         """
-
         self.render_config_template(
             path=os.path.abspath(self.working_dir) + "/log/**",
             scan_frequency="1s"

filebeat/tests/system/test_json.py

@@ -15,8 +15,7 @@ def test_docker_logs(self):
         """
         self.render_config_template(
             path=os.path.abspath(self.working_dir) + "/log/*",
-            json=dict(message_key="log"),
-            line_terminator="line_feed",
+            json=dict(message_key="log")
         )
 
         os.mkdir(self.working_dir + "/log/")
@@ -43,8 +42,7 @@ def test_docker_logs_filtering(self):
         self.render_config_template(
             path=os.path.abspath(self.working_dir) + "/log/*",
             json=dict(message_key="log", keys_under_root=True),
-            exclude_lines=["windows"],
-            line_terminator="line_feed",
+            exclude_lines=["windows"]
         )
 
         os.mkdir(self.working_dir + "/log/")
@@ -74,7 +72,6 @@ def test_simple_json_overwrite(self):
                 message_key="message",
                 keys_under_root=True,
                 overwrite_keys=True),
-            line_terminator="line_feed",
             exclude_lines=["windows"]
         )
 
@@ -99,7 +96,6 @@ def test_json_add_tags(self):
             json=dict(
                 keys_under_root=True,
             ),
-            line_terminator="line_feed",
             agent_tags=["tag3", "tag4"]
         )
         os.mkdir(self.working_dir + "/log/")
@@ -168,7 +164,6 @@ def test_timestamp_in_message(self):
                 keys_under_root=True,
                 overwrite_keys=True
             ),
-            line_terminator="line_feed",
         )
         os.mkdir(self.working_dir + "/log/")
         self.copy_files(["logs/json_timestamp.log"],
@@ -210,7 +205,6 @@ def test_type_in_message(self):
                 keys_under_root=True,
                 overwrite_keys=True
             ),
-            line_terminator="line_feed",
         )
         os.mkdir(self.working_dir + "/log/")
         self.copy_files(["logs/json_type.log"],
@@ -249,7 +243,6 @@ def test_with_generic_filtering(self):
                 overwrite_keys=True,
                 add_error_key=True
             ),
-            line_terminator="line_feed",
             processors=[{
                 "drop_fields": {
                     "fields": ["headers.request-id"],
@@ -283,7 +276,6 @@ def test_json_decoding_error_true(self):
         """
         Test if json_decoding_error is set to true, that no errors are logged.
         """
-
         self.render_config_template(
             path=os.path.abspath(self.working_dir) + "/log/*",
             json=dict(
@@ -317,7 +309,6 @@ def test_json_decoding_error_false(self):
         """
         Test if json_decoding_error is set to false, that an errors is logged.
         """
-
         self.render_config_template(
             path=os.path.abspath(self.working_dir) + "/log/*",
             json=dict(
@@ -361,7 +352,6 @@ def test_with_generic_filtering_remove_headers(self):
                 overwrite_keys=True,
                 add_error_key=True
             ),
-            line_terminator="line_feed",
             processors=[{
                 "drop_fields": {
                     "fields": ["headers", "res"],
@@ -401,7 +391,6 @@ def test_integer_condition(self):
             json=dict(
                 keys_under_root=True,
             ),
-            line_terminator="line_feed",
             processors=[{
                 "drop_event": {
                     "when": "equals.status: 200",

filebeat/tests/system/test_multiline.py

@@ -19,8 +19,7 @@ def test_java_elasticsearch_log(self):
             multiline=True,
             pattern="^\[",
             negate="true",
-            match="after",
-            line_terminator="line_feed",
+            match="after"
         )
 
         os.mkdir(self.working_dir + "/log/")
@@ -50,8 +49,7 @@ def test_c_style_log(self):
             path=os.path.abspath(self.working_dir) + "/log/*",
             multiline=True,
             pattern="\\\\$",
-            match="before",
-            line_terminator="line_feed",
+            match="before"
         )
 
         os.mkdir(self.working_dir + "/log/")
@@ -127,8 +125,7 @@ def test_max_lines(self):
             pattern="^\[",
             negate="true",
             match="after",
-            max_lines=3,
-            line_terminator="line_feed",
+            max_lines=3
         )
 
         os.mkdir(self.working_dir + "/log/")
@@ -210,8 +207,7 @@ def test_max_bytes(self):
             pattern="^\[",
             negate="true",
             match="after",
-            max_bytes=60,
-            line_terminator="line_feed",
+            max_bytes=60
         )
 
         os.mkdir(self.working_dir + "/log/")

filebeat/tests/system/test_processors.py

@@ -172,7 +172,6 @@ def test_dissect_bad_tokenizer(self):
         """
         Check dissect with a bad tokenizer
         """
-
         self.render_config_template(
             path=os.path.abspath(self.working_dir) + "/test.log",
             processors=[{

filebeat/tests/system/test_registrar.py

@@ -1445,13 +1445,13 @@ def test_registrar_meta(self):
     ids:
       - container_id
 - type: docker
-  line_terminator: line_feed
   containers:
     path: {path}
     stream: stderr
     ids:
       - container_id
-            '''.format(path=os.path.abspath(self.working_dir) + "/log/", line_terminator="line_feed"))
+            '''.format(path=os.path.abspath(self.working_dir) + "/log/")
+        )
         os.mkdir(self.working_dir + "/log/")
         os.mkdir(self.working_dir + "/log/container_id")
         testfile_path1 = self.working_dir + "/log/container_id/test.log"

filebeat/tests/system/test_stdin.py

@@ -17,8 +17,7 @@ def test_stdin(self):
         Test stdin input. Checks if reading is continued after the first read.
         """
         self.render_config_template(
-            type="stdin",
-            line_terminator="line_feed",
+            type="stdin"
         )
 
         proc = self.start_beat()
@@ -56,7 +55,6 @@ def test_stdin_eof(self):
         self.render_config_template(
             type="stdin",
             close_eof="true",
-            line_terminator="line_feed",
         )
 
         args = [self.test_binary, "-systemTest"]

libbeat/reader/readfile/line_terminator.go

@@ -26,6 +26,8 @@ type LineTerminator uint8
 const (
 	// InvalidTerminator is the invalid terminator
 	InvalidTerminator LineTerminator = iota
+	// AutoLineTerminator accepts both LF and CR+LF
+	AutoLineTerminator
 	// LineFeed is the unicode char LF
 	LineFeed
 	// VerticalTab is the unicode char VT
@@ -46,6 +48,7 @@ const (
 
 var (
 	lineTerminators = map[string]LineTerminator{
+		"auto":                      AutoLineTerminator,
 		"line_feed":                 LineFeed,
 		"vertical_tab":              VerticalTab,
 		"form_feed":                 FormFeed,
@@ -57,6 +60,7 @@ var (
 	}
 
 	lineTerminatorCharacters = map[LineTerminator][]byte{
+		AutoLineTerminator:     []byte{'\u000A'},
 		LineFeed:               []byte{'\u000A'},
 		VerticalTab:            []byte{'\u000B'},
 		FormFeed:               []byte{'\u000C'},