Fix integration test timeouts on Windows (#3949)

* Revert "Skip APM propagation test on Windows. (#3932)"

This reverts commit a76f4ba.

* Upgrade gRPC to fix serializer goroutine leak.

* Only run Windows tests.

* Only wait on proc if it was killed

* Fix wait goroutine leak in fixture.

* Revert "Only run Windows tests."

This reverts commit 677b79c.

pkg/core/process/process.go

@@ -119,7 +119,9 @@ func (i *Info) StopWait() error {
 	return err
 }
 
-// Wait returns a channel that will send process state once it exits.
+// Wait returns a channel that will send process state once it exits. Each
+// call to Wait() creates a goroutine. Failure to read from the returned
+// channel will leak this goroutine.
 func (i *Info) Wait() <-chan *os.ProcessState {
 	ch := make(chan *os.ProcessState)
 

pkg/testing/fixture.go

@@ -312,9 +312,10 @@ func (f *Fixture) RunBeat(ctx context.Context) error {
 		return fmt.Errorf("failed to spawn %s: %w", f.binaryName, err)
 	}
 
+	procWaitCh := proc.Wait()
 	killProc := func() {
 		_ = proc.Kill()
-		<-proc.Wait()
+		<-procWaitCh
 	}
 
 	var doneChan <-chan time.Time
@@ -328,7 +329,7 @@ func (f *Fixture) RunBeat(ctx context.Context) error {
 		case <-ctx.Done():
 			killProc()
 			return ctx.Err()
-		case ps := <-proc.Wait():
+		case ps := <-procWaitCh:
 			if stopping {
 				return nil
 			}
@@ -384,9 +385,10 @@ func RunProcess(t *testing.T,
 		return fmt.Errorf("failed to spawn %q: %w", processPath, err)
 	}
 
+	procWaitCh := proc.Wait()
 	killProc := func() {
 		_ = proc.Kill()
-		<-proc.Wait()
+		<-procWaitCh
 	}
 
 	var doneChan <-chan time.Time
@@ -400,7 +402,7 @@ func RunProcess(t *testing.T,
 		case <-ctx.Done():
 			killProc()
 			return ctx.Err()
-		case ps := <-proc.Wait():
+		case ps := <-procWaitCh:
 			if stopping {
 				return nil
 			}
@@ -503,11 +505,6 @@ func (f *Fixture) RunWithClient(ctx context.Context, shouldWatchState bool, stat
 		return fmt.Errorf("failed to spawn %s: %w", f.binaryName, err)
 	}
 
-	killProc := func() {
-		_ = proc.Kill()
-		<-proc.Wait()
-	}
-
 	if shouldWatchState {
 		agentClient = client.New(client.WithAddress(cAddr))
 		f.setClient(agentClient)
@@ -520,13 +517,19 @@ func (f *Fixture) RunWithClient(ctx context.Context, shouldWatchState bool, stat
 		doneChan = time.After(f.runLength)
 	}
 
+	procWaitCh := proc.Wait()
+	killProc := func() {
+		_ = proc.Kill()
+		<-procWaitCh
+	}
+
 	stopping := false
 	for {
 		select {
 		case <-ctx.Done():
 			killProc()
 			return ctx.Err()
-		case ps := <-proc.Wait():
+		case ps := <-procWaitCh:
 			if stopping {
 				return nil
 			}

testing/integration/apm_propagation_test.go

@@ -13,7 +13,6 @@ import (
 	"fmt"
 	"io"
 	"net/http"
-	"runtime"
 	"strings"
 	"testing"
 	"text/template"
@@ -57,12 +56,6 @@ func TestAPMConfig(t *testing.T) {
 		Group: Default,
 		Stack: &define.Stack{},
 	})
-
-	if runtime.GOOS == "windows" {
-		// This test hangs indefinitely on Windows. Root cause is TBD.
-		t.Skip("Flaky test: https://github.com/elastic/ingest-dev/issues/2668")
-	}
-
 	f, err := define.NewFixture(t, define.Version())
 	require.NoError(t, err)