[chore][pkg/stanza] Ensure all start/stop calls in are balanced (open…

…-telemetry#28294)

Follows
open-telemetry#28228

This normalizes calls to `Start` and `Stop` across the test suite. 

In some cases, `poll` is called directly in order to trigger behavior
independently of timing. However, we should _either_ use `poll`
directly, or use both `Start` and `Stop` exactly once. In the future, I
expect `poll` will be exported and tested directly as part of an
internal package.

pkg/stanza/fileconsumer/file_test.go

@@ -79,11 +79,8 @@ See this issue for details: https://github.com/census-instrumentation/opencensus
  operator, _ := buildTestManager(t, cfg)
 
  _ = openTemp(t, tempDir)
- err := operator.Start(testutil.NewUnscopedMockPersister())
- require.NoError(t, err)
- defer func() {
- require.NoError(t, operator.Stop())
- }()
+ require.NoError(t, operator.Start(testutil.NewUnscopedMockPersister()))
+ require.NoError(t, operator.Stop())
 }
 
 // AddFields tests that the `log.file.name` and `log.file.path` fields are included
@@ -491,9 +488,6 @@ func TestReadNewLogs(t *testing.T) {
 
  // Poll once so we know this isn't a new file
  operator.poll(context.Background())
- defer func() {
- require.NoError(t, operator.Stop())
- }()
 
  // Create a new file
  temp := openTemp(t, tempDir)
@@ -1583,28 +1577,25 @@ func TestHeaderPersistance(t *testing.T) {
  writeString(t, temp, "#headerField: headerValue\nlog line\n")
 
  persister := testutil.NewUnscopedMockPersister()
- require.NoError(t, op1.Start(persister))
 
+ require.NoError(t, op1.Start(persister))
  waitForTokenWithAttributes(t, emitCalls1, []byte("log line"), map[string]any{
  "header_key": "headerField",
  "header_value": "headerValue",
  attrs.LogFileName: filepath.Base(temp.Name()),
  })
-
  require.NoError(t, op1.Stop())
 
  writeString(t, temp, "log line 2\n")
 
  op2, emitCalls2 := buildTestManager(t, cfg)
 
  require.NoError(t, op2.Start(persister))
-
  waitForTokenWithAttributes(t, emitCalls2, []byte("log line 2"), map[string]any{
  "header_key": "headerField",
  "header_value": "headerValue",
  attrs.LogFileName: filepath.Base(temp.Name()),
  })
-
  require.NoError(t, op2.Stop())
 }
 
@@ -1626,12 +1617,10 @@ func TestHeaderPersistanceInHeader(t *testing.T) {
  writeString(t, temp, "|headerField1: headerValue1\n")
 
  persister := testutil.NewUnscopedMockPersister()
- require.NoError(t, op1.Start(persister))
-
- // The operator will poll at fixed time intervals, but we just want to make sure at least
- // one poll operation occurs between now and when we stop.
- op1.poll(context.Background())
 
+ // Start and stop the operator, ensuring that at least one poll cycle occurs in between
+ require.NoError(t, op1.Start(persister))
+ time.Sleep(2 * cfg1.PollInterval)
  require.NoError(t, op1.Stop())
 
  writeString(t, temp, "|headerField2: headerValue2\nlog line\n")
@@ -1643,13 +1632,11 @@ func TestHeaderPersistanceInHeader(t *testing.T) {
  op2, emitCalls := buildTestManager(t, cfg2)
 
  require.NoError(t, op2.Start(persister))
-
  waitForTokenWithAttributes(t, emitCalls, []byte("log line"), map[string]any{
  "header_value_1": "headerValue1",
  "header_value_2": "headerValue2",
  attrs.LogFileName: filepath.Base(temp.Name()),
  })
-
  require.NoError(t, op2.Stop())
 }
 

pkg/stanza/fileconsumer/rotation_test.go

@@ -365,10 +365,6 @@ func TestMoveFile(t *testing.T) {
  temp1.Close()
 
  operator.poll(context.Background())
- defer func() {
- require.NoError(t, operator.Stop())
- }()
-
  waitForToken(t, emitCalls, []byte("testlog1"))
 
  // Wait until all goroutines are finished before renaming
@@ -397,10 +393,6 @@ func TestTrackMovedAwayFiles(t *testing.T) {
  temp1.Close()
 
  operator.poll(context.Background())
- defer func() {
- require.NoError(t, operator.Stop())
- }()
-
  waitForToken(t, emitCalls, []byte("testlog1"))
 
  // Wait until all goroutines are finished before renaming
@@ -557,12 +549,7 @@ func TestTruncateThenWrite(t *testing.T) {
  writeString(t, temp1, "testlog1\ntestlog2\n")
 
  operator.poll(context.Background())
- defer func() {
- require.NoError(t, operator.Stop())
- }()
-
- waitForToken(t, emitCalls, []byte("testlog1"))
- waitForToken(t, emitCalls, []byte("testlog2"))
+ waitForTokens(t, emitCalls, []byte("testlog1"), []byte("testlog2"))
 
  require.NoError(t, temp1.Truncate(0))
  _, err := temp1.Seek(0, 0)
@@ -594,12 +581,7 @@ func TestCopyTruncateWriteBoth(t *testing.T) {
  writeString(t, temp1, "testlog1\ntestlog2\n")
 
  operator.poll(context.Background())
- defer func() {
- require.NoError(t, operator.Stop())
- }()
-
- waitForToken(t, emitCalls, []byte("testlog1"))
- waitForToken(t, emitCalls, []byte("testlog2"))
+ waitForTokens(t, emitCalls, []byte("testlog1"), []byte("testlog2"))
  operator.wg.Wait() // wait for all goroutines to finish
 
  // Copy the first file to a new file, and add another log