[component helper] Adding lifecycle state management to *helper.

exporter/exporterhelper/common.go

@@ -21,6 +21,7 @@ import (
 	"go.opentelemetry.io/collector/component"
 	"go.opentelemetry.io/collector/consumer"
 	"go.opentelemetry.io/collector/exporter/exporterhelper/internal"
+	"go.opentelemetry.io/collector/internal/lifecycle"
 	"go.opentelemetry.io/collector/obsreport"
 )
 
@@ -153,10 +154,13 @@ type baseExporter struct {
 	obsrep   *obsExporter
 	sender   requestSender
 	qrSender *queuedRetrySender
+	state    lifecycle.State
 }
 
 func newBaseExporter(cfg component.ExporterConfig, set component.ExporterCreateSettings, bs *baseSettings, signal component.DataType, reqUnmarshaler internal.RequestUnmarshaler) (*baseExporter, error) {
-	be := &baseExporter{}
+	be := &baseExporter{
+		state: lifecycle.NewState(),
+	}
 
 	var err error
 	be.obsrep, err = newObsExporter(obsreport.ExporterSettings{ExporterID: cfg.ID(), ExporterCreateSettings: set}, globalInstruments)
@@ -167,6 +171,10 @@ func newBaseExporter(cfg component.ExporterConfig, set component.ExporterCreateS
 	be.qrSender = newQueuedRetrySender(cfg.ID(), signal, bs.QueueSettings, bs.RetrySettings, reqUnmarshaler, &timeoutSender{cfg: bs.TimeoutSettings}, set.Logger)
 	be.sender = be.qrSender
 	be.StartFunc = func(ctx context.Context, host component.Host) error {
+		// Checking if the exporter has been started already.
+		if !be.state.Start() {
+			return nil
+		}
 		// First start the wrapped exporter.
 		if err := bs.StartFunc.Start(ctx, host); err != nil {
 			return err
@@ -176,6 +184,10 @@ func newBaseExporter(cfg component.ExporterConfig, set component.ExporterCreateS
 		return be.qrSender.start(ctx, host)
 	}
 	be.ShutdownFunc = func(ctx context.Context) error {
+		// Checking if the exporter has been shutdown already.
+		if !be.state.Stop() {
+			return nil
+		}
 		// First shutdown the queued retry sender
 		be.qrSender.shutdown()
 		// Last shutdown the wrapped exporter itself.

internal/lifecycle/state.go

@@ -0,0 +1,55 @@
+// Copyright  The OpenTelemetry Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package lifecycle // import "go.opentelemetry.io/collector/internal/lifecycle"
+
+import "go.uber.org/atomic"
+
+// Lifecycle represents the current
+// running state of the component
+// that is intended to be concurrency safe.
+type State interface {
+	// Start atomically updates the state and returns true if
+	// the previous state was not running.
+	Start() (started bool)
+
+	// Stop atomically updates the state and returns true if
+	// the previous state was not shutdown.
+	Stop() (stopped bool)
+}
+
+type state struct {
+	current *atomic.Int64
+}
+
+const (
+	shutdown = iota
+	running
+)
+
+var (
+	_ State = (*state)(nil)
+)
+
+func NewState() State {
+	return &state{current: atomic.NewInt64(shutdown)}
+}
+
+func (s *state) Start() (started bool) {
+	return s.current.Swap(running) == shutdown
+}
+
+func (s *state) Stop() (stopped bool) {
+	return s.current.Swap(shutdown) == running
+}

internal/lifecycle/state_test.go

@@ -0,0 +1,75 @@
+// Copyright  The OpenTelemetry Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package lifecycle
+
+import (
+	"sync"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestLifecycleState(t *testing.T) {
+	t.Parallel()
+
+	state := NewState()
+	assert.True(t, state.Start(), "Must confirm the state has been updated")
+	assert.False(t, state.Start(), "Must confirm the state no state change")
+
+	assert.True(t, state.Stop(), "Must confirm the state has been updated")
+	assert.False(t, state.Stop(), "Must confirm the state no state change")
+}
+
+func TestLifecycleConcurrency(t *testing.T) {
+	t.Parallel()
+
+	state := NewState()
+
+	var (
+		updated int
+		start   = make(chan struct{})
+		wg      sync.WaitGroup
+	)
+
+	for i := 0; i < 4; i++ {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			<-start
+			if state.Start() {
+				updated++
+			}
+		}()
+	}
+	close(start)
+	wg.Wait()
+	assert.Equal(t, 1, updated, "Must have only been updated once")
+
+	start = make(chan struct{})
+
+	for i := 0; i < 4; i++ {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			<-start
+			if state.Stop() {
+				updated--
+			}
+		}()
+	}
+	close(start)
+	wg.Wait()
+	assert.Equal(t, 0, updated, "Must have only been updated once")
+}

receiver/scraperhelper/scrapercontroller.go

@@ -25,6 +25,7 @@ import (
 	"go.opentelemetry.io/collector/component"
 	"go.opentelemetry.io/collector/config"
 	"go.opentelemetry.io/collector/consumer"
+	"go.opentelemetry.io/collector/internal/lifecycle"
 	"go.opentelemetry.io/collector/obsreport"
 	"go.opentelemetry.io/collector/pdata/pmetric"
 	"go.opentelemetry.io/collector/receiver/scrapererror"
@@ -81,6 +82,7 @@ type controller struct {
 
 	tickerCh <-chan time.Time
 
+	state       lifecycle.State
 	initialized bool
 	done        chan struct{}
 	terminated  chan struct{}
@@ -148,6 +150,10 @@ func NewScraperControllerReceiver(
 
 // Start the receiver, invoked during service start.
 func (sc *controller) Start(ctx context.Context, host component.Host) error {
+	if !sc.state.Start() {
+		return nil
+	}
+
 	for _, scraper := range sc.scrapers {
 		if err := scraper.Start(ctx, host); err != nil {
 			return err
@@ -161,6 +167,10 @@ func (sc *controller) Start(ctx context.Context, host component.Host) error {
 
 // Shutdown the receiver, invoked during service shutdown.
 func (sc *controller) Shutdown(ctx context.Context) error {
+	if !sc.state.Stop() {
+		return nil
+	}
+
 	sc.stopScraping()
 
 	// wait until scraping ticker has terminated