[extension/healthcheckv2] Add event aggregation logic

.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,154 @@
+// 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
+}
+
+// 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,172 @@
+// 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))
+
+	// Clear first error
+	st.ComponentStatusMap["c2"] = &AggregateStatus{
+		Event: component.NewStatusEvent(component.StatusOK),
+	}
+
+	// Returns second error now
+	assert.Equal(t, st.ComponentStatusMap["c3"].Event, aggFunc(st))
+
+	// Clear second error
+	st.ComponentStatusMap["c3"] = &AggregateStatus{
+		Event: component.NewStatusEvent(component.StatusOK),
+	}
+
+	// Returns latest event
+	assert.Equal(t, st.ComponentStatusMap["c3"].Event, aggFunc(st))
+}