[extension/healthcheckv2] Add event aggregation logic (#32695)

**Description:** 
This PR is the second in a series to decompose
#30673
into more manageable pieces for review.

**Aggregator**

This PR introduces an aggregator data structure and event aggregation
logic for status events. The extension implements the `StatusWatcher`
optional interface, which the collector will call with a
`component.StatusEvent` for each change in component status. These
events will be aggregated by an aggregation function, and stored in the
aggregator.

The aggregator is a recursive data structure. At the top it contains the
overall status of the collector. At the next level, it contains the
statuses for each pipeline, and at the level below that, it contains the
statuses for each component in a pipeline. Each node in the data
structure is an aggregation the status events in the level below. The
overall collector status is the aggregation of the pipeline statuses,
and at the next level, the pipeline statuses are the aggregations of the
component statuses. The data structure allows you to query the status of
the collector overall, or for individual pipelines by name. There is
also a pub/sub mechanism used for streaming aggregated statuses.

**Aggregation Function**

The purpose of the aggregator is to aggregate events so that the most
relevant status event bubbles to the top. This allows us to get the
status of the collector overall or a pipeline through a simple lookup.
There is an aggregation function that determines the priority of events
and how they should be aggregated. In many cases, the result will be an
existing status event. In some cases a new event will be synthesized. In
order to match the behavior existing healthcheck extension, lifecycle
events (e.g. starting, stopping, etc) are prioritized over runtime
events. Next, error statuses are prioritized with PermanentErrors as
higher priority than RecoverableErrors, but this can vary based on user
provided configuration. If PermanentErrors are ignored by configuration,
but RecoverableErrors are included, then RecoverableErrors will take
priority over PermanentErrors.

The StatusWatcher interface receives immutable events of type
`component.StatusEvent`. Since we sometimes need to synthesize new
events during aggregation, an `Event` interface was introduced so that
the aggregator can use `component.StatusEvent` instances or instances of
events synthesized by the status package.

It's worth mentioning that there is [existing status event aggregation
logic](https://github.com/open-telemetry/opentelemetry-collector/blob/main/component/status.go#L101-L190)
in collector core, but it did not meet the needs of this extension. It
does not prioritize lifecycle events over error events, and it will
always prioritize permanent errors over recoverable. By prioritizing
lifecycle events over error events we can return a 503 when restarting a
collector rather than a 500 when a collector in a final state, such as
PermanentError. This is necessary to match the behavior of the existing
extension. Since users have the option to include or ignore recoverable
and permanent errors, we need the ability to prioritize them
accordingly. We can discuss what the fate of the aggregation code in
core should be.

**Examples**
Below are examples of overall collector and pipeline status that are
based on the aggregator data structure. The rendering of the examples
will come in a later PR. You can also look at the parent PR to see how
all of this fits together. Note that the pipeline status example is a
subtree of the overall collector status.

Overall collector status:

```json
{
    "start_time": "2024-01-18T17:27:12.570394-08:00",
    "healthy": true,
    "status": "StatusRecoverableError",
    "error": "rpc error: code = ResourceExhausted desc = resource exhausted",
    "status_time": "2024-01-18T17:27:32.572301-08:00",
    "components": {
        "extensions": {
            "healthy": true,
            "status": "StatusOK",
            "status_time": "2024-01-18T17:27:12.570428-08:00",
            "components": {
                "extension:healthcheckv2": {
                    "healthy": true,
                    "status": "StatusOK",
                    "status_time": "2024-01-18T17:27:12.570428-08:00"
                }
            }
        },
        "pipeline:metrics/grpc": {
            "healthy": true,
            "status": "StatusRecoverableError",
            "error": "rpc error: code = ResourceExhausted desc = resource exhausted",
            "status_time": "2024-01-18T17:27:32.572301-08:00",
            "components": {
                "exporter:otlp/staging": {
                    "healthy": true,
                    "status": "StatusRecoverableError",
                    "error": "rpc error: code = ResourceExhausted desc = resource exhausted",
                    "status_time": "2024-01-18T17:27:32.572301-08:00"
                },
                "processor:batch": {
                    "healthy": true,
                    "status": "StatusOK",
                    "status_time": "2024-01-18T17:27:12.571132-08:00"
                },
                "receiver:otlp": {
                    "healthy": true,
                    "status": "StatusOK",
                    "status_time": "2024-01-18T17:27:12.571576-08:00"
                }
            }
        },
        "pipeline:traces/http": {
            "healthy": true,
            "status": "StatusOK",
            "status_time": "2024-01-18T17:27:12.571625-08:00",
            "components": {
                "exporter:otlphttp/staging": {
                    "healthy": true,
                    "status": "StatusOK",
                    "status_time": "2024-01-18T17:27:12.571615-08:00"
                },
                "processor:batch": {
                    "healthy": true,
                    "status": "StatusOK",
                    "status_time": "2024-01-18T17:27:12.571621-08:00"
                },
                "receiver:otlp": {
                    "healthy": true,
                    "status": "StatusOK",
                    "status_time": "2024-01-18T17:27:12.571625-08:00"
                }
            }
        }
    }
}
```
Status for pipeline `traces/http`:

```json
{
    "start_time": "2024-01-18T17:27:12.570394-08:00",
    "healthy": true,
    "status": "StatusOK",
    "status_time": "2024-01-18T17:27:12.571625-08:00",
    "components": {
        "exporter:otlphttp/staging": {
            "healthy": true,
            "status": "StatusOK",
            "status_time": "2024-01-18T17:27:12.571615-08:00"
        },
        "processor:batch": {
            "healthy": true,
            "status": "StatusOK",
            "status_time": "2024-01-18T17:27:12.571621-08:00"
        },
        "receiver:otlp": {
            "healthy": true,
            "status": "StatusOK",
            "status_time": "2024-01-18T17:27:12.571625-08:00"
        }
    }
}
```




**Link to tracking Issue:** #26661

**Testing:** Units / manual

**Documentation:** Comments, etc

---------

Co-authored-by: Evan Bradley <11745660+evan-bradley@users.noreply.github.com>

.chloggen/healthcheck-agg.yaml

@@ -0,0 +1,27 @@
+# Use this changelog template to create an entry for release notes.
+
+# One of 'breaking', 'deprecation', 'new_component', 'enhancement', 'bug_fix'
+change_type: 'enhancement'
+
+# The name of the component, or a single word describing the area of concern, (e.g. filelogreceiver)
+component: healthcheckv2extension
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Add shared aggregation logic for status events.
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [26661]
+
+# (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:
+
+# If your change doesn't affect end users or the exported elements of any package,
+# you should instead start your pull request title with [chore] or use the "Skip Changelog" label.
+# 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: []

extension/healthcheckv2extension/internal/status/aggregation.go

@@ -0,0 +1,156 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package status // import "github.com/open-telemetry/opentelemetry-collector-contrib/extension/healthcheckv2extension/internal/status"
+
+import (
+	"time"
+
+	"go.opentelemetry.io/collector/component"
+)
+
+// statusEvent contains a status and timestamp, and can contain an error. Note:
+// this is duplicated from core because we need to be able to "rewrite" the
+// timestamps of some events during aggregation.
+type statusEvent struct {
+	status    component.Status
+	err       error
+	timestamp time.Time
+}
+
+var _ Event = (*statusEvent)(nil)
+
+// Status returns the Status (enum) associated with the StatusEvent
+func (ev *statusEvent) Status() component.Status {
+	return ev.status
+}
+
+// Err returns the error associated with the StatusEvent.
+func (ev *statusEvent) Err() error {
+	return ev.err
+}
+
+// Timestamp returns the timestamp associated with the StatusEvent
+func (ev *statusEvent) Timestamp() time.Time {
+	return ev.timestamp
+}
+
+type ErrorPriority int
+
+const (
+	PriorityPermanent ErrorPriority = iota
+	PriorityRecoverable
+)
+
+type aggregationFunc func(*AggregateStatus) Event
+
+// The purpose of aggregation is to ensure that the most relevant status bubbles
+// upwards in the aggregate status. This aggregation func prioritizes lifecycle
+// events (including FatalError) over PermanentError and RecoverableError
+// events. The priority argument determines the priority of PermanentError
+// events vs RecoverableError events. Lifecycle events will have the timestamp
+// of the most recent event and error events will have the timestamp of the
+// first occurrence. We use the first occurrence of an error event as this marks
+// the beginning of a possible failure. This is important for two reasons:
+// recovery duration and causality. We expect a RecoverableError to recover
+// before the RecoveryDuration elapses. We need to use the earliest timestamp so
+// that a later RecoverableError does not shadow an earlier event in the
+// aggregate status. Additionally, this makes sense in the case where a
+// RecoverableError in one component cascades to other components; the earliest
+// error event is likely to be correlated with the cause. For non-error stauses
+// we use the latest event as it represents the last time a successful status was
+// reported.
+func newAggregationFunc(priority ErrorPriority) aggregationFunc {
+	statusFunc := func(st *AggregateStatus) component.Status {
+		seen := make(map[component.Status]struct{})
+		for _, cs := range st.ComponentStatusMap {
+			seen[cs.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[component.StatusFatalError]; isFatal {
+			return component.StatusFatalError
+		}
+
+		if _, isStarting := seen[component.StatusStarting]; isStarting {
+			return component.StatusStarting
+		}
+
+		if _, isStopping := seen[component.StatusStopping]; isStopping {
+			return component.StatusStopping
+		}
+
+		if _, isStopped := seen[component.StatusStopped]; isStopped {
+			return component.StatusStopping
+		}
+
+		if priority == PriorityPermanent {
+			if _, isPermanent := seen[component.StatusPermanentError]; isPermanent {
+				return component.StatusPermanentError
+			}
+			if _, isRecoverable := seen[component.StatusRecoverableError]; isRecoverable {
+				return component.StatusRecoverableError
+			}
+		} else {
+			if _, isRecoverable := seen[component.StatusRecoverableError]; isRecoverable {
+				return component.StatusRecoverableError
+			}
+			if _, isPermanent := seen[component.StatusPermanentError]; isPermanent {
+				return component.StatusPermanentError
+			}
+		}
+
+		return component.StatusNone
+	}
+
+	return func(st *AggregateStatus) Event {
+		var ev, lastEvent, matchingEvent Event
+		status := statusFunc(st)
+		isError := component.StatusIsError(status)
+
+		for _, cs := range st.ComponentStatusMap {
+			ev = cs.Event
+			if lastEvent == nil || lastEvent.Timestamp().Before(ev.Timestamp()) {
+				lastEvent = ev
+			}
+			if status == ev.Status() {
+				switch {
+				case matchingEvent == nil:
+					matchingEvent = ev
+				case isError:
+					// Use earliest to mark beginning of a failure
+					if ev.Timestamp().Before(matchingEvent.Timestamp()) {
+						matchingEvent = ev
+					}
+				case ev.Timestamp().After(matchingEvent.Timestamp()):
+					// Use most recent for last successful status
+					matchingEvent = ev
+				}
+			}
+		}
+
+		// the error status will be the first matching event
+		if isError {
+			return matchingEvent
+		}
+
+		// the aggregate status matches an existing event
+		if lastEvent.Status() == status {
+			return lastEvent
+		}
+
+		// the aggregate status requires a synthetic event
+		return &statusEvent{
+			status:    status,
+			timestamp: lastEvent.Timestamp(),
+		}
+	}
+}

extension/healthcheckv2extension/internal/status/aggregation_test.go

@@ -0,0 +1,175 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package status
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"go.opentelemetry.io/collector/component"
+)
+
+func TestAggregationFuncs(t *testing.T) {
+	aggRecoverable := newAggregationFunc(PriorityRecoverable)
+	aggPermanent := newAggregationFunc(PriorityPermanent)
+
+	type statusExpectation struct {
+		priorityPermanent   component.Status
+		priorityRecoverable component.Status
+	}
+
+	for _, tc := range []struct {
+		name            string
+		aggregateStatus *AggregateStatus
+		expectedStatus  *statusExpectation
+	}{
+		{
+			name: "FatalError takes precedence over all",
+			aggregateStatus: &AggregateStatus{
+				ComponentStatusMap: map[string]*AggregateStatus{
+					"c1": {
+						Event: component.NewStatusEvent(component.StatusFatalError),
+					},
+					"c2": {
+						Event: component.NewStatusEvent(component.StatusStarting),
+					},
+					"c3": {
+						Event: component.NewStatusEvent(component.StatusOK),
+					},
+					"c4": {
+						Event: component.NewStatusEvent(component.StatusRecoverableError),
+					},
+					"c5": {
+						Event: component.NewStatusEvent(component.StatusPermanentError),
+					},
+					"c6": {
+						Event: component.NewStatusEvent(component.StatusStopping),
+					},
+					"c7": {
+						Event: component.NewStatusEvent(component.StatusStopped),
+					},
+				},
+			},
+			expectedStatus: &statusExpectation{
+				priorityPermanent:   component.StatusFatalError,
+				priorityRecoverable: component.StatusFatalError,
+			},
+		},
+		{
+			name: "Lifecycle: Starting takes precedence over non-fatal errors",
+			aggregateStatus: &AggregateStatus{
+				ComponentStatusMap: map[string]*AggregateStatus{
+					"c1": {
+						Event: component.NewStatusEvent(component.StatusStarting),
+					},
+					"c2": {
+						Event: component.NewStatusEvent(component.StatusRecoverableError),
+					},
+					"c3": {
+						Event: component.NewStatusEvent(component.StatusPermanentError),
+					},
+				},
+			},
+			expectedStatus: &statusExpectation{
+				priorityPermanent:   component.StatusStarting,
+				priorityRecoverable: component.StatusStarting,
+			},
+		},
+		{
+			name: "Lifecycle: Stopping takes precedence over non-fatal errors",
+			aggregateStatus: &AggregateStatus{
+				ComponentStatusMap: map[string]*AggregateStatus{
+					"c1": {
+						Event: component.NewStatusEvent(component.StatusStopping),
+					},
+					"c2": {
+						Event: component.NewStatusEvent(component.StatusRecoverableError),
+					},
+					"c3": {
+						Event: component.NewStatusEvent(component.StatusPermanentError),
+					},
+				},
+			},
+			expectedStatus: &statusExpectation{
+				priorityPermanent:   component.StatusStopping,
+				priorityRecoverable: component.StatusStopping,
+			},
+		},
+		{
+			name: "Prioritized error takes priority over OK",
+			aggregateStatus: &AggregateStatus{
+				ComponentStatusMap: map[string]*AggregateStatus{
+					"c1": {
+						Event: component.NewStatusEvent(component.StatusOK),
+					},
+					"c2": {
+						Event: component.NewStatusEvent(component.StatusRecoverableError),
+					},
+					"c3": {
+						Event: component.NewStatusEvent(component.StatusPermanentError),
+					},
+				},
+			},
+			expectedStatus: &statusExpectation{
+				priorityPermanent:   component.StatusPermanentError,
+				priorityRecoverable: component.StatusRecoverableError,
+			},
+		},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			assert.Equal(t, tc.expectedStatus.priorityPermanent,
+				aggPermanent(tc.aggregateStatus).Status())
+			assert.Equal(t, tc.expectedStatus.priorityRecoverable,
+				aggRecoverable(tc.aggregateStatus).Status())
+		})
+	}
+}
+
+func TestEventTemporalOrder(t *testing.T) {
+	// Note: ErrorPriority does not affect temporal ordering
+	aggFunc := newAggregationFunc(PriorityPermanent)
+	st := &AggregateStatus{
+		ComponentStatusMap: map[string]*AggregateStatus{
+			"c1": {
+				Event: component.NewStatusEvent(component.StatusOK),
+			},
+		},
+	}
+	assert.Equal(t, st.ComponentStatusMap["c1"].Event, aggFunc(st))
+
+	// Record first error
+	st.ComponentStatusMap["c2"] = &AggregateStatus{
+		Event: component.NewRecoverableErrorEvent(assert.AnError),
+	}
+
+	// Returns first error
+	assert.Equal(t, st.ComponentStatusMap["c2"].Event, aggFunc(st))
+
+	// Record second error
+	st.ComponentStatusMap["c3"] = &AggregateStatus{
+		Event: component.NewRecoverableErrorEvent(assert.AnError),
+	}
+
+	// Still returns first error
+	assert.Equal(t, st.ComponentStatusMap["c2"].Event, aggFunc(st))
+
+	// Replace first error with later error
+	st.ComponentStatusMap["c2"] = &AggregateStatus{
+		Event: component.NewRecoverableErrorEvent(assert.AnError),
+	}
+
+	// Returns second error now
+	assert.Equal(t, st.ComponentStatusMap["c3"].Event, aggFunc(st))
+
+	// Clear errors
+	st.ComponentStatusMap["c2"] = &AggregateStatus{
+		Event: component.NewStatusEvent(component.StatusOK),
+	}
+	st.ComponentStatusMap["c3"] = &AggregateStatus{
+		Event: component.NewStatusEvent(component.StatusOK),
+	}
+
+	// Returns latest event
+	assert.Equal(t, st.ComponentStatusMap["c3"].Event, aggFunc(st))
+}