Merge remote-tracking branch 'upstream/master' into feature/macosx-pr…

…-basis-when

* upstream/master:
  [CI] run everything for branches/tags (elastic#20057)
  Limit the number of bytes read by LineReader in Filebeat (elastic#19552)
  Prefer testify.Error/NoError over NotNil/Nil (elastic#20044)

CHANGELOG.next.asciidoc

@@ -217,6 +217,7 @@ https://github.com/elastic/beats/compare/v7.0.0-alpha2...master[Check the HEAD d
 - Fix bug with empty filter values in system/service {pull}19812[19812]
 - Fix S3 input to trim delimiter /n from each log line. {pull}19972[19972]
 - Ignore missing in Zeek module when dropping unecessary fields. {pull}19984[19984]
+- Fix Filebeat OOMs on very long lines {issue}19500[19500], {pull}19552[19552]
 
 *Heartbeat*
 

Jenkinsfile

@@ -1141,11 +1141,18 @@ def reportCoverage(){
   }
 }
 
-// isChanged treats the patterns as regular expressions. In order to check if
-// any file in a directoy is modified use `^<path to dir>/.*`.
+/**
+*  isChanged treats the patterns as regular expressions. In order to check if
+*  any file in a directoy is modified use `^<path to dir>/.*`.
+*
+*  In addition, there are another two alternatives to report that there are
+*  changes, when `runAllStages` parameter is set to true or when running on a
+*  branch/tag basis.
+*/
 def isChanged(patterns){
   return (
-    params.runAllStages
+    params.runAllStages   // when runAllStages UI parameter is set to true
+    || !isPR()            // when running on a branch/tag
     || isGitRegionMatch(patterns: patterns, comparator: 'regexp')
   )
 }

filebeat/config/config_test.go

@@ -34,24 +34,24 @@ func TestReadConfig2(t *testing.T) {
 	absPath, err := filepath.Abs("../tests/files/")
 
 	assert.NotNil(t, absPath)
-	assert.Nil(t, err)
+	assert.NoError(t, err)
 
 	config := &Config{}
 
 	// Reads second config file
 	err = cfgfile.Read(config, absPath+"/config2.yml")
-	assert.Nil(t, err)
+	assert.NoError(t, err)
 }
 
 func TestGetConfigFiles_File(t *testing.T) {
 	absPath, err := filepath.Abs("../tests/files/")
 
 	assert.NotNil(t, absPath)
-	assert.Nil(t, err)
+	assert.NoError(t, err)
 
 	files, err := getConfigFiles(absPath + "/config.yml")
 
-	assert.Nil(t, err)
+	assert.NoError(t, err)
 	assert.Equal(t, 1, len(files))
 
 	assert.Equal(t, absPath+"/config.yml", files[0])
@@ -61,11 +61,11 @@ func TestGetConfigFiles_Dir(t *testing.T) {
 	absPath, err := filepath.Abs("../tests/files/")
 
 	assert.NotNil(t, absPath)
-	assert.Nil(t, err)
+	assert.NoError(t, err)
 
 	files, err := getConfigFiles(absPath)
 
-	assert.Nil(t, err)
+	assert.NoError(t, err)
 	assert.Equal(t, 2, len(files))
 
 	assert.Equal(t, filepath.Join(absPath, "/config.yml"), files[0])
@@ -76,36 +76,36 @@ func TestGetConfigFiles_EmptyDir(t *testing.T) {
 	absPath, err := filepath.Abs("../tests/files/")
 
 	assert.NotNil(t, absPath)
-	assert.Nil(t, err)
+	assert.NoError(t, err)
 
 	files, err := getConfigFiles(absPath + "/logs")
 
-	assert.Nil(t, err)
+	assert.NoError(t, err)
 	assert.Equal(t, 0, len(files))
 }
 
 func TestGetConfigFiles_Invalid(t *testing.T) {
 	absPath, err := filepath.Abs("../tests/files/")
 
 	assert.NotNil(t, absPath)
-	assert.Nil(t, err)
+	assert.NoError(t, err)
 
 	// Invalid directory
 	files, err := getConfigFiles(absPath + "/qwerwer")
 
-	assert.NotNil(t, err)
+	assert.Error(t, err)
 	assert.Nil(t, files)
 }
 
 func TestMergeConfigFiles(t *testing.T) {
 	absPath, err := filepath.Abs("../tests/files/")
 
 	assert.NotNil(t, absPath)
-	assert.Nil(t, err)
+	assert.NoError(t, err)
 
 	files, err := getConfigFiles(absPath)
 
-	assert.Nil(t, err)
+	assert.NoError(t, err)
 	assert.Equal(t, 2, len(files))
 
 	config := &Config{}

filebeat/fileset/modules_integration_test.go

@@ -142,7 +142,7 @@ func TestAvailableProcessors(t *testing.T) {
 	}
 
 	err = checkAvailableProcessors(client, requiredProcessors)
-	assert.NotNil(t, err)
+	assert.Error(t, err)
 	assert.Contains(t, err.Error(), "ingest-test")
 	assert.Contains(t, err.Error(), "ingest-hello")
 }
@@ -235,7 +235,7 @@ func TestLoadMultiplePipelinesWithRollback(t *testing.T) {
 	}
 
 	err = reg.LoadPipelines(client, false)
-	assert.NotNil(t, err)
+	assert.Error(t, err)
 	assert.Contains(t, err.Error(), "invalid_processor")
 
 	status, _, _ := client.Request("GET", "/_ingest/pipeline/filebeat-6.6.0-foo-multibad-pipeline", "", nil, nil)

filebeat/harvester/util_test.go

@@ -45,27 +45,27 @@ func InitMatchers(exprs ...string) ([]match.Matcher, error) {
 
 func TestMatchAnyRegexps(t *testing.T) {
 	matchers, err := InitMatchers("\\.gz$")
-	assert.Nil(t, err)
+	assert.NoError(t, err)
 	assert.Equal(t, MatchAny(matchers, "/var/log/log.gz"), true)
 }
 
 func TestExcludeLine(t *testing.T) {
 	regexp, err := InitMatchers("^DBG")
-	assert.Nil(t, err)
+	assert.NoError(t, err)
 	assert.True(t, MatchAny(regexp, "DBG: a debug message"))
 	assert.False(t, MatchAny(regexp, "ERR: an error message"))
 }
 
 func TestIncludeLine(t *testing.T) {
 	regexp, err := InitMatchers("^ERR", "^WARN")
 
-	assert.Nil(t, err)
+	assert.NoError(t, err)
 	assert.False(t, MatchAny(regexp, "DBG: a debug message"))
 	assert.True(t, MatchAny(regexp, "ERR: an error message"))
 	assert.True(t, MatchAny(regexp, "WARNING: a simple warning message"))
 }
 
 func TestInitRegexp(t *testing.T) {
 	_, err := InitMatchers("(((((")
-	assert.NotNil(t, err)
+	assert.Error(t, err)
 }

filebeat/input/log/harvester.go

@@ -631,6 +631,8 @@ func (h *Harvester) newLogFileReader() (reader.Reader, error) {
 	var r reader.Reader
 	var err error
 
+	logp.Debug("harvester", "newLogFileReader with config.MaxBytes: %d", h.config.MaxBytes)
+
 	// TODO: NewLineReader uses additional buffering to deal with encoding and testing
 	//       for new lines in input stream. Simple 8-bit based encodings, or plain
 	//       don't require 'complicated' logic.
@@ -644,10 +646,17 @@ func (h *Harvester) newLogFileReader() (reader.Reader, error) {
 		return nil, err
 	}
 
+	// Configure MaxBytes limit for EncodeReader as multiplied by 4
+	// for the worst case scenario where incoming UTF32 charchers are decoded to the single byte UTF-8 characters.
+	// This limit serves primarily to avoid memory bload or potential OOM with expectedly long lines in the file.
+	// The further size limiting is performed by LimitReader at the end of the readers pipeline as needed.
+	encReaderMaxBytes := h.config.MaxBytes * 4
+
 	r, err = readfile.NewEncodeReader(reader, readfile.Config{
 		Codec:      h.encoding,
 		BufferSize: h.config.BufferSize,
 		Terminator: h.config.LineTerminator,
+		MaxBytes:   encReaderMaxBytes,
 	})
 	if err != nil {
 		return nil, err

filebeat/input/log/harvester_test.go

@@ -41,7 +41,7 @@ func TestReadLine(t *testing.T) {
 	logFile := absPath + "/tmp" + strconv.Itoa(rand.Int()) + ".log"
 
 	assert.NotNil(t, absPath)
-	assert.Nil(t, err)
+	assert.NoError(t, err)
 
 	if err != nil {
 		t.Fatalf("Error creating the absolute path: %s", absPath)
@@ -51,26 +51,26 @@ func TestReadLine(t *testing.T) {
 	defer file.Close()
 	defer os.Remove(logFile)
 
-	assert.Nil(t, err)
+	assert.NoError(t, err)
 	assert.NotNil(t, file)
 
 	firstLineString := "9Characte\n"
 	secondLineString := "This is line 2\n"
 
 	length, err := file.WriteString(firstLineString)
-	assert.Nil(t, err)
+	assert.NoError(t, err)
 	assert.NotNil(t, length)
 
 	length, err = file.WriteString(secondLineString)
-	assert.Nil(t, err)
+	assert.NoError(t, err)
 	assert.NotNil(t, length)
 
 	file.Sync()
 
 	// Open file for reading
 	readFile, err := os.Open(logFile)
 	defer readFile.Close()
-	assert.Nil(t, err)
+	assert.NoError(t, err)
 
 	source := File{File: readFile}
 
@@ -102,7 +102,7 @@ func TestReadLine(t *testing.T) {
 	// Read third line
 	_, text, bytesread, _, err := readLine(r)
 	t.Logf("received line: '%s'\n", text)
-	assert.Nil(t, err)
+	assert.NoError(t, err)
 	assert.Equal(t, text, firstLineString[0:len(firstLineString)-1])
 	assert.Equal(t, bytesread, len(firstLineString))
 
@@ -111,7 +111,7 @@ func TestReadLine(t *testing.T) {
 	t.Logf("received line: '%s'\n", text)
 	assert.Equal(t, text, secondLineString[0:len(secondLineString)-1])
 	assert.Equal(t, bytesread, len(secondLineString))
-	assert.Nil(t, err)
+	assert.NoError(t, err)
 
 	// Read third line, which doesn't exist
 	_, text, bytesread, _, err = readLine(r)

filebeat/processor/add_kubernetes_metadata/matchers_test.go

@@ -132,7 +132,7 @@ func executeTestWithResourceType(t *testing.T, cfgLogsPath string, cfgResourceTy
 	}
 
 	logMatcher, err := newLogsPathMatcher(*testConfig)
-	assert.Nil(t, err)
+	assert.NoError(t, err)
 
 	input := common.MapStr{
 		"log": common.MapStr{

heartbeat/look/look_test.go

@@ -32,7 +32,7 @@ import (
 // helper
 func testRTT(t *testing.T, expected time.Duration, provided time.Duration) {
 	actual, err := RTT(provided).GetValue("us")
-	assert.Nil(t, err)
+	assert.NoError(t, err)
 	assert.Equal(t, expected, actual)
 }
 

heartbeat/monitors/active/http/config_test.go

@@ -88,10 +88,10 @@ func TestConfigValidate(t *testing.T) {
 
 			err := config.Validate()
 			if test.result {
-				assert.Nil(t, err)
+				assert.NoError(t, err)
 				assert.Equal(t, test.convertedHost, config.Hosts[0])
 			} else {
-				assert.NotNil(t, err)
+				assert.Error(t, err)
 			}
 
 		})

heartbeat/monitors/active/http/respbody_test.go

@@ -246,7 +246,7 @@ func Test_readPrefixAndHash(t *testing.T) {
 			expectedHash := sha256.Sum256([]byte(tt.body))
 			assert.Equal(t, hex.EncodeToString(expectedHash[:]), gotHashStr)
 
-			assert.Nil(t, err)
+			assert.NoError(t, err)
 		})
 	}
 }

heartbeat/monitors/active/http/task_test.go

@@ -131,7 +131,7 @@ func TestSplitHostnamePort(t *testing.T) {
 
 func makeTestHTTPRequest(t *testing.T) *http.Request {
 	req, err := http.NewRequest("GET", "http://example.net", nil)
-	assert.Nil(t, err)
+	assert.NoError(t, err)
 	return req
 }
 
@@ -168,7 +168,7 @@ func TestRequestBuildingWithCustomHost(t *testing.T) {
 
 	request, err := buildRequest("localhost", &config, encoder)
 
-	if assert.Nil(t, err) {
+	if assert.NoError(t, err) {
 		assert.Equal(t, "custom-host", request.Host)
 		assert.Equal(t, "custom-host", request.Header.Get("Host"))
 	}
@@ -177,7 +177,7 @@ func TestRequestBuildingWithCustomHost(t *testing.T) {
 func TestRequestBuildingWithNoUserAgent(t *testing.T) {
 	request, err := buildRequest("localhost", &Config{}, nilEncoder{})
 
-	require.Nil(t, err)
+	require.NoError(t, err)
 	assert.Equal(t, useragent.UserAgent("Heartbeat"), request.Header.Get("User-Agent"))
 }
 
@@ -196,6 +196,6 @@ func TestRequestBuildingWithExplicitUserAgent(t *testing.T) {
 
 	request, err := buildRequest("localhost", &config, nilEncoder{})
 
-	require.Nil(t, err)
+	require.NoError(t, err)
 	assert.Equal(t, expectedUserAgent, request.Header.Get("User-Agent"))
 }

libbeat/autodiscover/appender_test.go

@@ -53,27 +53,27 @@ func TestAppenderRegistry(t *testing.T) {
 	cfg, err := common.NewConfigFrom(&config)
 
 	// Make sure that config building doesn't fail
-	assert.Nil(t, err)
+	assert.NoError(t, err)
 	appender, err := reg.BuildAppender(cfg)
-	assert.Nil(t, err)
+	assert.NoError(t, err)
 	assert.NotNil(t, appender)
 
 	// Attempt to build using an array of configs
 	Registry.AddAppender("fake", newFakeAppender)
 	cfgs := []*common.Config{cfg}
 	appenders, err := NewAppenders(cfgs)
-	assert.Nil(t, err)
+	assert.NoError(t, err)
 	assert.Equal(t, len(appenders), 1)
 
 	// Attempt to build using an incorrect config
 	incorrectConfig := AppenderConfig{
 		Type: "wrong",
 	}
 	icfg, err := common.NewConfigFrom(&incorrectConfig)
-	assert.Nil(t, err)
+	assert.NoError(t, err)
 	cfgs = append(cfgs, icfg)
 	appenders, err = NewAppenders(cfgs)
-	assert.NotNil(t, err)
+	assert.Error(t, err)
 	assert.Nil(t, appenders)
 
 	// Try to append onto an event using fakeAppender and the result should have one item

libbeat/autodiscover/appenders/config/config_test.go

@@ -92,11 +92,11 @@ condition.equals:
 			}
 
 			appender, err := NewConfigAppender(config)
-			assert.Nil(t, err)
+			assert.NoError(t, err)
 			assert.NotNil(t, appender)
 
 			eveConfig, err := common.NewConfigFrom(&test.eventConfig)
-			assert.Nil(t, err)
+			assert.NoError(t, err)
 
 			test.event["config"] = []*common.Config{eveConfig}
 			appender.Append(test.event)

libbeat/autodiscover/builder_test.go

@@ -55,10 +55,10 @@ func TestBuilderRegistry(t *testing.T) {
 	cfg, err := common.NewConfigFrom(&config)
 
 	// Make sure that config building doesn't fail
-	assert.Nil(t, err)
+	assert.NoError(t, err)
 
 	builder, err := reg.BuildBuilder(cfg)
-	assert.Nil(t, err)
+	assert.NoError(t, err)
 	assert.NotNil(t, builder)
 
 	// Try to create a config with fake builder and assert length