[component] Move component status reporting public API to new componentstatus module

.chloggen/componentstatus-new-module.yaml

@@ -0,0 +1,25 @@
+# Use this changelog template to create an entry for release notes.
+
+# One of 'breaking', 'deprecation', 'new_component', 'enhancement', 'bug_fix'
+change_type: new_component
+
+# The name of the component, or a single word describing the area of concern, (e.g. otlpreceiver)
+component: componentstatus
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Adds new componentstatus module that will soon replace status content in component.
+
+# One or more tracking issues or pull requests related to the change
+issues: [10730]
+
+# (Optional) One or more lines of additional information to render under the primary note.
+# These lines will be padded with 2 spaces and then inserted directly into the document.
+# Use pipe (|) for multiline entries.
+subtext:
+
+# Optional: The change log or logs in which this entry should be included.
+# e.g. '[user]' or '[user, api]'
+# Include 'user' if the change is relevant to end users.
+# Include 'api' if there is a change to a library API.
+# Default: '[user]'
+change_logs: [api]

Makefile

@@ -259,6 +259,7 @@ check-contrib:
 		-replace go.opentelemetry.io/collector=$(CURDIR) \
 		-replace go.opentelemetry.io/collector/client=$(CURDIR)/client \
 		-replace go.opentelemetry.io/collector/component=$(CURDIR)/component  \
+		-replace go.opentelemetry.io/collector/component/componentstatus=$(CURDIR)/component/componentstatus  \
 		-replace go.opentelemetry.io/collector/config/configauth=$(CURDIR)/config/configauth  \
 		-replace go.opentelemetry.io/collector/config/configcompression=$(CURDIR)/config/configcompression  \
 		-replace go.opentelemetry.io/collector/config/configgrpc=$(CURDIR)/config/configgrpc  \
@@ -321,6 +322,7 @@ restore-contrib:
 		-dropreplace go.opentelemetry.io/collector \
 		-dropreplace go.opentelemetry.io/collector/client \
 		-dropreplace go.opentelemetry.io/collector/component \
+		-dropreplace go.opentelemetry.io/collector/component/componentstatus \
 		-dropreplace go.opentelemetry.io/collector/config/configauth  \
 		-dropreplace go.opentelemetry.io/collector/config/configcompression  \
 		-dropreplace go.opentelemetry.io/collector/config/configgrpc  \

component/componentstatus/Makefile

@@ -0,0 +1 @@
+include ../../Makefile.Common

component/componentstatus/go.mod

@@ -0,0 +1,34 @@
+module go.opentelemetry.io/collector/component/componentstatus
+
+go 1.21.0
+
+require (
+	github.com/stretchr/testify v1.9.0
+	go.opentelemetry.io/collector/component v0.105.0
+)
+
+require (
+	github.com/davecgh/go-spew v1.1.1 // indirect
+	github.com/gogo/protobuf v1.3.2 // indirect
+	github.com/pmezard/go-difflib v1.0.0 // indirect
+	go.opentelemetry.io/collector/config/configtelemetry v0.105.0 // indirect
+	go.opentelemetry.io/collector/pdata v1.12.0 // indirect
+	go.opentelemetry.io/otel v1.28.0 // indirect
+	go.opentelemetry.io/otel/metric v1.28.0 // indirect
+	go.opentelemetry.io/otel/trace v1.28.0 // indirect
+	go.uber.org/multierr v1.11.0 // indirect
+	go.uber.org/zap v1.27.0 // indirect
+	golang.org/x/net v0.26.0 // indirect
+	golang.org/x/sys v0.21.0 // indirect
+	golang.org/x/text v0.16.0 // indirect
+	google.golang.org/genproto/googleapis/rpc v0.0.0-20240528184218-531527333157 // indirect
+	google.golang.org/grpc v1.65.0 // indirect
+	google.golang.org/protobuf v1.34.2 // indirect
+	gopkg.in/yaml.v3 v3.0.1 // indirect
+)
+
+replace go.opentelemetry.io/collector/config/configtelemetry => ../../config/configtelemetry
+
+replace go.opentelemetry.io/collector/component => ../
+
+replace go.opentelemetry.io/collector/pdata => ../../pdata

component/componentstatus/instance.go

@@ -0,0 +1,13 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package componentstatus // import "go.opentelemetry.io/collector/component/componentstatus"
+
+import "go.opentelemetry.io/collector/component"
+
+// InstanceID uniquely identifies a component instance
+type InstanceID struct {
+	ID          component.ID
+	Kind        component.Kind
+	PipelineIDs map[component.ID]struct{}
+}

component/componentstatus/status.go

@@ -0,0 +1,211 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package componentstatus // import "go.opentelemetry.io/collector/component/componentstatus"
+
+import (
+	"time"
+)
+
+// Watcher is an extra interface for Extension hosted by the OpenTelemetry
+// Collector that is to be implemented by extensions interested in changes to component
+// status.
+type Watcher interface {
+	// ComponentStatusChanged notifies about a change in the source component status.
+	// Extensions that implement this interface must be ready that the ComponentStatusChanged
+	// may be called before, after or concurrently with calls to Component.Start() and Component.Shutdown().
+	// The function may be called concurrently with itself.
+	ComponentStatusChanged(source *InstanceID, event *Event)
+}
+
+type Status int32
+
+// Enumeration of possible component statuses
+const (
+	// StatusNone indicates absence of component status.
+	StatusNone Status = iota
+	// StatusStarting indicates the component is starting.
+	StatusStarting
+	// StatusOK indicates the component is running without issues.
+	StatusOK
+	// StatusRecoverableError indicates that the component has experienced a transient error and may recover.
+	StatusRecoverableError
+	// StatusPermanentError indicates that the component has detected a condition at runtime that will need human intervention to fix. The collector will continue to run in a degraded mode.
+	StatusPermanentError
+	// StatusFatalError indicates that the collector has experienced a fatal runtime error and will shut down.
+	StatusFatalError
+	// StatusStopping indicates that the component is in the process of shutting down.
+	StatusStopping
+	// StatusStopped indicates that the component has completed shutdown.
+	StatusStopped
+)
+
+// String returns a string representation of a Status
+func (s Status) String() string {
+	switch s {
+	case StatusStarting:
+		return "StatusStarting"
+	case StatusOK:
+		return "StatusOK"
+	case StatusRecoverableError:
+		return "StatusRecoverableError"
+	case StatusPermanentError:
+		return "StatusPermanentError"
+	case StatusFatalError:
+		return "StatusFatalError"
+	case StatusStopping:
+		return "StatusStopping"
+	case StatusStopped:
+		return "StatusStopped"
+	}
+	return "StatusNone"
+}
+
+// Event contains a status and timestamp, and can contain an error
+type Event struct {
+	status    Status
+	err       error
+	timestamp time.Time
+}
+
+// Status returns the Status (enum) associated with the Event
+func (ev *Event) Status() Status {
+	return ev.status
+}
+
+// Err returns the error associated with the Event.
+func (ev *Event) Err() error {
+	return ev.err
+}
+
+// Timestamp returns the timestamp associated with the Event
+func (ev *Event) Timestamp() time.Time {
+	return ev.timestamp
+}
+
+// NewEvent creates and returns a Event with the specified status and sets the timestamp
+// time.Now(). To set an error on the event for an error status use one of the dedicated
+// constructors (e.g. NewRecoverableErrorEvent, NewPermanentErrorEvent, NewFatalErrorEvent)
+func NewEvent(status Status) *Event {
+	return &Event{
+		status:    status,
+		timestamp: time.Now(),
+	}
+}
+
+// NewRecoverableErrorEvent wraps a transient error
+// passed as argument as a Event with a status StatusRecoverableError
+// and a timestamp set to time.Now().
+func NewRecoverableErrorEvent(err error) *Event {
+	ev := NewEvent(StatusRecoverableError)
+	ev.err = err
+	return ev
+}
+
+// NewPermanentErrorEvent wraps an error requiring human intervention to fix
+// passed as argument as a Event with a status StatusPermanentError
+// and a timestamp set to time.Now().
+func NewPermanentErrorEvent(err error) *Event {
+	ev := NewEvent(StatusPermanentError)
+	ev.err = err
+	return ev
+}
+
+// NewFatalErrorEvent wraps the fatal runtime error passed as argument as a Event
+// with a status StatusFatalError and a timestamp set to time.Now().
+func NewFatalErrorEvent(err error) *Event {
+	ev := NewEvent(StatusFatalError)
+	ev.err = err
+	return ev
+}
+
+// AggregateStatus will derive a status for the given input using the following rules in order:
+//  1. If all instances have the same status, there is nothing to aggregate, return it.
+//  2. If any instance encounters a fatal error, the component is in a Fatal Error state.
+//  3. If any instance is in a Permanent Error state, the component status is Permanent Error.
+//  4. If any instance is Stopping, the component is in a Stopping state.
+//  5. An instance is Stopped, but not all instances are Stopped, we must be in the process of Stopping the component.
+//  6. If any instance is in a Recoverable Error state, the component status is Recoverable Error.
+//  7. By process of elimination, the only remaining state is starting.
+func AggregateStatus[K comparable](eventMap map[K]*Event) Status {
+	seen := make(map[Status]struct{})
+	for _, ev := range eventMap {
+		seen[ev.Status()] = struct{}{}
+	}
+
+	// All statuses are the same. Note, this will handle StatusOK and StatusStopped as these two
+	// cases require all components be in the same state.
+	if len(seen) == 1 {
+		for st := range seen {
+			return st
+		}
+	}
+
+	// Handle mixed status cases
+	if _, isFatal := seen[StatusFatalError]; isFatal {
+		return StatusFatalError
+	}
+
+	if _, isPermanent := seen[StatusPermanentError]; isPermanent {
+		return StatusPermanentError
+	}
+
+	if _, isStopping := seen[StatusStopping]; isStopping {
+		return StatusStopping
+	}
+
+	if _, isStopped := seen[StatusStopped]; isStopped {
+		return StatusStopping
+	}
+
+	if _, isRecoverable := seen[StatusRecoverableError]; isRecoverable {
+		return StatusRecoverableError
+	}
+
+	// By process of elimination, this is the last possible status; no check necessary.
+	return StatusStarting
+}
+
+// StatusIsError returns true for error statuses (e.g. StatusRecoverableError,
+// StatusPermanentError, or StatusFatalError)
+func StatusIsError(status Status) bool {
+	return status == StatusRecoverableError ||
+		status == StatusPermanentError ||
+		status == StatusFatalError
+}
+
+// AggregateEvent returns a status event where:
+//   - The status is set to the aggregate status of the events in the eventMap
+//   - The timestamp is set to the latest timestamp of the events in the eventMap
+//   - For an error status, the event will have same error as the most current event of the same
+//     error type from the eventMap
+func AggregateEvent[K comparable](eventMap map[K]*Event) *Event {
+	var lastEvent, lastMatchingEvent *Event
+	aggregateStatus := AggregateStatus[K](eventMap)
+
+	for _, ev := range eventMap {
+		if lastEvent == nil || lastEvent.timestamp.Before(ev.timestamp) {
+			lastEvent = ev
+		}
+		if aggregateStatus == ev.Status() &&
+			(lastMatchingEvent == nil || lastMatchingEvent.timestamp.Before(ev.timestamp)) {
+			lastMatchingEvent = ev
+		}
+	}
+
+	// the effective status matches an existing event
+	if lastEvent.Status() == aggregateStatus {
+		return lastEvent
+	}
+
+	// the effective status requires a synthetic event
+	aggregateEvent := &Event{
+		status:    aggregateStatus,
+		timestamp: lastEvent.timestamp,
+	}
+	if StatusIsError(aggregateStatus) {
+		aggregateEvent.err = lastMatchingEvent.err
+	}
+
+	return aggregateEvent
+}