Enable and fix race detector for CI (#3096)

* Stop ignoring RACE_DETECTOR when not on amd64.

The race detector is now supported on arm64 and aarch64.

* Enable the race detector in CI.

* Remove pkg/errors from gotest.go.

* Fix unchecked error return value.

* Fix tests.

---------

Co-authored-by: Craig MacKenzie <[email protected]>
Co-authored-by: Fae Charlton <[email protected]>
(cherry picked from commit f31ceaa)

.ci/Jenkinsfile

@@ -114,7 +114,7 @@ pipeline {
               withGithubNotify(context: "Test-${PLATFORM}") {
                 withMageEnv(){
                   dir("${BASE_DIR}"){
-                    withEnv(["TEST_COVERAGE=${isCodeCoverageEnabled()}"]) {
+                    withEnv(["RACE_DETECTOR=true", "TEST_COVERAGE=${isCodeCoverageEnabled()}"]) {
                       cmd(label: 'Go unitTest', script: 'mage unitTest')
                     }
                   }
@@ -330,7 +330,7 @@ pipeline {
               withGithubNotify(context: "Test-${PLATFORM}") {
                 withMageEnv(){
                   dir("${BASE_DIR}"){
-                    withEnv(["TEST_COVERAGE=${isCodeCoverageEnabled()}"]) {
+                    withEnv(["RACE_DETECTOR=true", "TEST_COVERAGE=${isCodeCoverageEnabled()}"]) {
                       cmd(label: 'Go unitTest', script: 'mage unitTest')
                     }
                   }
@@ -380,7 +380,7 @@ pipeline {
             withGithubNotify(context: "Test-darwin-aarch64") {
               withMageEnv(){
                 dir("${BASE_DIR}"){
-                  withEnv(["TEST_COVERAGE=${isCodeCoverageEnabled()}"]) {
+                  withEnv(["RACE_DETECTOR=true", "TEST_COVERAGE=${isCodeCoverageEnabled()}"]) {
                     cmd(label: 'Go unitTest', script: 'mage unitTest')
                   }
                 }

dev-tools/mage/gotest.go

@@ -7,6 +7,7 @@ package mage
 import (
 	"bytes"
 	"context"
+	"errors"
 	"fmt"
 	"io"
 	"io/ioutil"
@@ -20,7 +21,6 @@ import (
 
 	"github.com/magefile/mage/mg"
 	"github.com/magefile/mage/sh"
-	"github.com/pkg/errors"
 
 	"github.com/elastic/elastic-agent/dev-tools/mage/gotool"
 )
@@ -147,7 +147,7 @@ func GoTestIntegrationForModule(ctx context.Context) error {
 		passThroughEnvs(env, IntegrationTestEnvVars()...)
 		runners, err := NewIntegrationRunners(path.Join("./module", fi.Name()), env)
 		if err != nil {
-			return errors.Wrapf(err, "test setup failed for module %s", fi.Name())
+			return fmt.Errorf("test setup failed for module %s: %w", fi.Name(), err)
 		}
 		err = runners.Test("goIntegTest", func() error {
 			err := GoTest(ctx, GoTestIntegrationArgsForModule(fi.Name()))
@@ -171,8 +171,8 @@ func GoTestIntegrationForModule(ctx context.Context) error {
 }
 
 // InstallGoTestTools installs additional tools that are required to run unit and integration tests.
-func InstallGoTestTools() {
-	gotool.Install(
+func InstallGoTestTools() error {
+	return gotool.Install(
 		gotool.Install.Package("gotest.tools/gotestsum"),
 	)
 }
@@ -214,12 +214,10 @@ func GoTest(ctx context.Context, params GoTestArgs) error {
 
 	var testArgs []string
 
-	// -race is only supported on */amd64
-	if os.Getenv("DEV_ARCH") == "amd64" {
-		if params.Race {
-			testArgs = append(testArgs, "-race")
-		}
+	if params.Race {
+		testArgs = append(testArgs, "-race")
 	}
+
 	if len(params.Tags) > 0 {
 		params := strings.Join(params.Tags, " ")
 		if params != "" {
@@ -253,7 +251,7 @@ func GoTest(ctx context.Context, params GoTestArgs) error {
 	if params.OutputFile != "" {
 		fileOutput, err := os.Create(createDir(params.OutputFile))
 		if err != nil {
-			return errors.Wrap(err, "failed to create go test output file")
+			return fmt.Errorf("failed to create go test output file: %w", err)
 		}
 		defer fileOutput.Close()
 		outputs = append(outputs, fileOutput)
@@ -274,7 +272,7 @@ func GoTest(ctx context.Context, params GoTestArgs) error {
 		// Command ran.
 		var exitErr *exec.ExitError
 		if !errors.As(err, &exitErr) {
-			return errors.Wrap(err, "failed to execute go")
+			return fmt.Errorf("failed to execute go: %w", err)
 		}
 
 		// Command ran but failed. Process the output.
@@ -283,7 +281,7 @@ func GoTest(ctx context.Context, params GoTestArgs) error {
 
 	if goTestErr != nil {
 		// No packages were tested. Probably the code didn't compile.
-		return errors.Wrap(goTestErr, "go test returned a non-zero value")
+		return fmt.Errorf("go test returned a non-zero value: %w", goTestErr)
 	}
 
 	// Generate a HTML code coverage report.
@@ -295,7 +293,7 @@ func GoTest(ctx context.Context, params GoTestArgs) error {
 			"-html="+params.CoverageProfileFile,
 			"-o", htmlCoverReport)
 		if err = coverToHTML(); err != nil {
-			return errors.Wrap(err, "failed to write HTML code coverage report")
+			return fmt.Errorf("failed to write HTML code coverage report: %w", err)
 		}
 	}
 
@@ -308,15 +306,15 @@ func GoTest(ctx context.Context, params GoTestArgs) error {
 		// install pre-requisites
 		installCobertura := sh.RunCmd("go", "install", "github.com/boumenot/gocover-cobertura@latest")
 		if err = installCobertura(); err != nil {
-			return errors.Wrap(err, "failed to install gocover-cobertura")
+			return fmt.Errorf("failed to install gocover-cobertura: %w", err)
 		}
 
 		codecovReport = strings.TrimSuffix(params.CoverageProfileFile,
 			filepath.Ext(params.CoverageProfileFile)) + "-cov.xml"
 
 		coverage, err := ioutil.ReadFile(params.CoverageProfileFile)
 		if err != nil {
-			return errors.Wrap(err, "failed to read code coverage report")
+			return fmt.Errorf("failed to read code coverage report: %w", err)
 		}
 
 		coberturaFile, err := os.Create(codecovReport)
@@ -330,15 +328,15 @@ func GoTest(ctx context.Context, params GoTestArgs) error {
 		coverToXML.Stderr = os.Stderr
 		coverToXML.Stdin = bytes.NewReader(coverage)
 		if err = coverToXML.Run(); err != nil {
-			return errors.Wrap(err, "failed to write XML code coverage report")
+			return fmt.Errorf("failed to write XML code coverage report: %w", err)
 		}
 		fmt.Println(">> go run gocover-cobertura:", params.CoverageProfileFile, "Created")
 	}
 
 	// Return an error indicating that testing failed.
 	if goTestErr != nil {
 		fmt.Println(">> go test:", params.LogName, "Test Failed")
-		return errors.Wrap(goTestErr, "go test returned a non-zero value")
+		return fmt.Errorf("go test returned a non-zero value: %w", goTestErr)
 	}
 
 	fmt.Println(">> go test:", params.LogName, "Test Passed")

internal/pkg/agent/application/dispatcher/dispatcher_test.go

@@ -9,7 +9,6 @@ import (
 	"testing"
 	"time"
 
-	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/mock"
 	"github.com/stretchr/testify/require"
 
@@ -227,7 +226,11 @@ func TestActionDispatcher(t *testing.T) {
 
 	t.Run("Cancel queued action", func(t *testing.T) {
 		def := &mockHandler{}
-		def.On("Handle", mock.Anything, mock.Anything, mock.Anything).Return(nil).Once()
+		calledCh := make(chan bool)
+		call := def.On("Handle", mock.Anything, mock.Anything, mock.Anything).Return(nil).Once()
+		call.RunFn = func(_ mock.Arguments) {
+			calledCh <- true
+		}
 
 		queue := &mockQueue{}
 		queue.On("Save").Return(nil).Once()
@@ -248,10 +251,11 @@ func TestActionDispatcher(t *testing.T) {
 		select {
 		case err := <-d.Errors():
 			t.Fatalf("Unexpected error: %v", err)
-		case <-time.After(200 * time.Microsecond):
-			// we're not expecting any reset,
+		case <-calledCh:
+			// Handle was called, expected
+		case <-time.After(1 * time.Second):
+			t.Fatal("mock Handle never called")
 		}
-		assert.Eventuallyf(t, func() bool { return len(def.Calls) > 0 }, 100*time.Millisecond, 100*time.Microsecond, "mock handler for cancel actions has not been called")
 		def.AssertExpectations(t)
 		queue.AssertExpectations(t)
 	})

internal/pkg/agent/application/upgrade/artifact/download/http/downloader_test.go

@@ -30,19 +30,20 @@ func TestDownloadBodyError(t *testing.T) {
 	// part way through the download, while copying the response body.
 
 	type connKey struct{}
-	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+	srv := httptest.NewUnstartedServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
 		w.WriteHeader(http.StatusOK)
 		w.(http.Flusher).Flush()
 		conn, ok := r.Context().Value(connKey{}).(net.Conn)
 		if ok {
-			conn.Close()
+			_ = conn.Close()
 		}
 	}))
-	defer srv.Close()
-	client := srv.Client()
 	srv.Config.ConnContext = func(ctx context.Context, c net.Conn) context.Context {
 		return context.WithValue(ctx, connKey{}, c)
 	}
+	srv.Start()
+	defer srv.Close()
+	client := srv.Client()
 
 	targetDir, err := ioutil.TempDir(os.TempDir(), "")
 	if err != nil {
@@ -64,6 +65,9 @@ func TestDownloadBodyError(t *testing.T) {
 		t.Fatal("expected Download to return an error")
 	}
 
+	log.lock.RLock()
+	defer log.lock.RUnlock()
+
 	require.GreaterOrEqual(t, len(log.info), 1, "download error not logged at info level")
 	assert.True(t, containsMessage(log.info, "download from %s failed at %s @ %sps: %s"))
 	require.GreaterOrEqual(t, len(log.warn), 1, "download error not logged at warn level")
@@ -113,6 +117,9 @@ func TestDownloadLogProgressWithLength(t *testing.T) {
 	os.Remove(artifactPath)
 	require.NoError(t, err, "Download should not have errored")
 
+	log.lock.RLock()
+	defer log.lock.RUnlock()
+
 	// 2 files are downloaded so 4 log messages are expected in the info level and only the complete is over the warn
 	// window as 2 log messages for warn.
 	require.Len(t, log.info, 4)
@@ -167,6 +174,9 @@ func TestDownloadLogProgressWithoutLength(t *testing.T) {
 	os.Remove(artifactPath)
 	require.NoError(t, err, "Download should not have errored")
 
+	log.lock.RLock()
+	defer log.lock.RUnlock()
+
 	// 2 files are downloaded so 4 log messages are expected in the info level and only the complete is over the warn
 	// window as 2 log messages for warn.
 	require.Len(t, log.info, 4)

pkg/component/runtime/log_writer.go

@@ -36,8 +36,9 @@ type logWriter struct {
 	loggerCore zapcoreWriter
 	logCfg     component.CommandLogSpec
 	logLevel   zap.AtomicLevel
+
+	mx         sync.Mutex
 	unitLevels map[string]zapcore.Level
-	levelMx    sync.RWMutex
 	remainder  []byte
 
 	// inheritLevel is the level that will be used for a log message in the case it doesn't define a log level
@@ -60,9 +61,10 @@ func newLogWriter(core zapcoreWriter, logCfg component.CommandLogSpec, ll zapcor
 }
 
 func (r *logWriter) SetLevels(ll zapcore.Level, unitLevels map[string]zapcore.Level) {
+	// must hold to lock so Write doesn't access the unitLevels
+	r.mx.Lock()
+	defer r.mx.Unlock()
 	r.logLevel.SetLevel(ll)
-	r.levelMx.Lock()
-	defer r.levelMx.Unlock()
 	r.unitLevels = unitLevels
 }
 
@@ -71,6 +73,12 @@ func (r *logWriter) Write(p []byte) (int, error) {
 		// nothing to do
 		return 0, nil
 	}
+
+	// hold the lock so SetLevels and the remainder is not touched
+	// from multiple go routines
+	r.mx.Lock()
+	defer r.mx.Unlock()
+
 	offset := 0
 	for {
 		idx := bytes.IndexByte(p[offset:], '\n')
@@ -127,13 +135,11 @@ func (r *logWriter) handleJSON(line string) bool {
 	allowedLvl := r.logLevel.Level()
 	unitId := getUnitId(evt)
 	if unitId != "" {
-		r.levelMx.RLock()
 		if r.unitLevels != nil {
 			if unitLevel, ok := r.unitLevels[unitId]; ok {
 				allowedLvl = unitLevel
 			}
 		}
-		r.levelMx.RUnlock()
 	}
 	if allowedLvl.Enabled(lvl) {
 		_ = r.loggerCore.Write(zapcore.Entry{