cnf ran ptp: make OC cases events-optional

This commit uses the eventmetric package to fallback to metric
assertions whenever PTP events are not enabled.

Additionally, it flips the semantics of CurrentState. By default, the
CurrentState logs are excluded since they don't represent events being
sent to the consumer. They may be explicitly enabled when desired.

There are a few complexities still, such as some assertions needing to
be events-only due to the metric being too transient to capture.
Additionally, the tests will now configure the ServiceMonitor such that
Prometheus scrapes the metrics every second instead of the default 30
seconds. This improves the precision of metric assertions.

Assisted-by: Cursor

Author: klaskosk

tests/cnf/ran/ptp/internal/consumer/bothversions.go

@@ -118,6 +118,8 @@ func getEventAPIVersion(client *clients.Settings) (ptpEventAPIVersion, error) {
 		return eventAPIVersionV1, nil
 	case string(eventAPIVersionV2):
 		return eventAPIVersionV2, nil
+	case "":
+		return eventAPIVersionV2, nil
 	default:
 		return "", fmt.Errorf("unknown event API version %s in PTP operator config", apiVersion)
 	}

tests/cnf/ran/ptp/internal/eventmetric/eventmetric.go

@@ -37,7 +37,9 @@ import (
 	"github.com/rh-ecosystem-edge/eco-gotests/tests/cnf/ran/ptp/internal/consumer"
 	"github.com/rh-ecosystem-edge/eco-gotests/tests/cnf/ran/ptp/internal/events"
 	"github.com/rh-ecosystem-edge/eco-gotests/tests/cnf/ran/ptp/internal/metrics"
+	"github.com/rh-ecosystem-edge/eco-gotests/tests/cnf/ran/ptp/internal/tsparams"
 	"golang.org/x/exp/constraints"
+	"k8s.io/klog/v2"
 )
 
 // AssertConfig is a struct that contains the configuration for the assertion. It combines the possible inputs to
@@ -155,6 +157,17 @@ func (assertConfig *AssertConfig[V]) ExecuteAssertion(ctx context.Context) error
 		return fmt.Errorf("failed to check if events are enabled: %w", err)
 	}
 
+	klog.V(tsparams.LogLevel).Infof(
+		"Starting combined event/metric assertion: node=%s eventsEnabled=%t startTime=%s timeout=%s query=%s "+
+			"expectedMetricValue=%d",
+		assertConfig.NodeName,
+		eventsEnabled,
+		startTime,
+		assertConfig.Timeout,
+		assertConfig.MetricQuery.ToMetricQuery().String(),
+		int64(assertConfig.ExpectedMetricValue),
+	)
+
 	errChan := make(chan error, 2)
 	waitGroup := sync.WaitGroup{}
 
@@ -168,8 +181,14 @@ func (assertConfig *AssertConfig[V]) ExecuteAssertion(ctx context.Context) error
 			err := events.WaitForEvent(
 				eventPod, startTime, assertConfig.Timeout, assertConfig.EventFilter, assertConfig.EventOptions...)
 			if err != nil {
+				klog.V(tsparams.LogLevel).Infof("Event assertion failed for node %s: %v", assertConfig.NodeName, err)
+
 				errChan <- fmt.Errorf("event assertion failed: %w", err)
+
+				return
 			}
+
+			klog.V(tsparams.LogLevel).Infof("Event assertion passed for node %s", assertConfig.NodeName)
 		})
 	}
 
@@ -187,8 +206,19 @@ func (assertConfig *AssertConfig[V]) ExecuteAssertion(ctx context.Context) error
 			metricOptions...,
 		)
 		if err != nil {
+			klog.V(tsparams.LogLevel).Infof("Metric assertion failed for node %s: %v", assertConfig.NodeName, err)
+
 			errChan <- fmt.Errorf("metric assertion failed: %w", err)
+
+			return
 		}
+
+		klog.V(tsparams.LogLevel).Infof(
+			"Metric assertion passed for node %s with query %s and expected value %d",
+			assertConfig.NodeName,
+			assertConfig.MetricQuery.ToMetricQuery().String(),
+			int64(assertConfig.ExpectedMetricValue),
+		)
 	})
 
 	waitGroup.Wait()

tests/cnf/ran/ptp/internal/events/README.md

@@ -33,9 +33,9 @@ The `WaitForEvent` function accepts several optional parameters:
 
 Specifies the container name within the pod from which to retrieve logs. If not specified, logs are retrieved from the default container. This is particularly useful when monitoring PTP events from specific containers like `"cloud-event-proxy"`.
 
-#### `WithoutCurrentState(ignoreCurrentState bool)`
+#### `WithCurrentState(includeCurrentState bool)`
 
-Controls whether to ignore messages about the current state of events. When set to `true`, only events received as subscriptions are considered, filtering out initial state reports. This is useful when you want to wait for new events rather than existing state information.
+Controls whether to include messages about the current state of events. By default, current state messages are not included.
 
 ### Event Filtering
 
@@ -112,14 +112,13 @@ func main() {
 
     fmt.Printf("Waiting for PTP Sync State Locked event on interface %s...\n", targetInterface)
 
-    // Use both WithContainer and WithoutCurrentState options
+    // Use the container option.
     err := events.WaitForEvent(
         myPtpPod,
         startTime,
         timeout,
         eventFilter,
         events.WithContainer("cloud-event-proxy"),
-        events.WithoutCurrentState(true), // Only wait for new events, not current state
     )
     if err != nil {
         fmt.Printf("Error waiting for event: %v\n", err)

tests/cnf/ran/ptp/internal/events/events.go

@@ -31,8 +31,8 @@ var (
 // waitForEventOptions is a struct that holds options for the WaitForEvent function. Options will update this struct and
 // the final result is used to configure the WaitForEvent function.
 type waitForEventOptions struct {
-	container          string
-	ignoreCurrentState bool
+	container           string
+	includeCurrentState bool
 }
 
 // WaitForEventOption is a function that modifies the waitForEventOptions struct. It is used to set options for the
@@ -47,11 +47,11 @@ func WithContainer(container string) WaitForEventOption {
 	}
 }
 
-// WithoutCurrentState is an option for the WaitForEvent function that specifies whether to ignore messages about the
-// current state of events. This allows for checking only events that are received as a subscription.
-func WithoutCurrentState(ignoreCurrentState bool) WaitForEventOption {
+// WithCurrentState is an option for the WaitForEvent function that specifies whether to include current state messages
+// while extracting events from logs.
+func WithCurrentState(includeCurrentState bool) WaitForEventOption {
 	return func(options *waitForEventOptions) {
-		options.ignoreCurrentState = ignoreCurrentState
+		options.includeCurrentState = includeCurrentState
 	}
 }
 
@@ -71,6 +71,15 @@ func WaitForEvent(
 		option(&combinedOptions)
 	}
 
+	klog.V(tsparams.LogLevel).Infof(
+		"Waiting for event: startTime=%s timeout=%s container=%q includeCurrentState=%t filter=%#v",
+		startTime,
+		timeout,
+		combinedOptions.container,
+		combinedOptions.includeCurrentState,
+		filter,
+	)
+
 	return wait.PollUntilContextTimeout(
 		context.TODO(), 5*time.Second, timeout, true, func(ctx context.Context) (bool, error) {
 			// Each loop we save the previous start time and set the new start time to the current time.
@@ -91,24 +100,42 @@ func WaitForEvent(
 
 			klog.V(tsparams.LogLevel).Infof("Logs: %s", string(logs))
 
-			extractedEvents := extractEventsFromLogs(logs, combinedOptions.ignoreCurrentState)
+			totalEventLikeLines := len(containsEventRegexp.FindAll(logs, -1))
+
+			nonCurrentStateEventLines := len(containsEventNotStateRegexp.FindAll(logs, -1))
+			if !combinedOptions.includeCurrentState && totalEventLikeLines > nonCurrentStateEventLines {
+				klog.V(tsparams.LogLevel).Infof(
+					"Ignored %d CurrentState event log lines (includeCurrentState=%t)",
+					totalEventLikeLines-nonCurrentStateEventLines,
+					combinedOptions.includeCurrentState,
+				)
+			}
+
+			extractedEvents := extractEventsFromLogs(logs, combinedOptions.includeCurrentState)
 
 			klog.V(tsparams.LogLevel).Infof("Extracted events: %#v\nFilter: %#v", extractedEvents, filter)
 
-			return slices.ContainsFunc(extractedEvents, filter.Filter), nil
+			matchingEventIndex := slices.IndexFunc(extractedEvents, filter.Filter)
+			if matchingEventIndex != -1 {
+				klog.V(tsparams.LogLevel).Infof("Matched event: %#v", extractedEvents[matchingEventIndex])
+
+				return true, nil
+			}
+
+			return false, nil
 		})
 }
 
 // extractEventsFromLogs extracts events from the logs of either the cloud event consumer or the cloud event proxy
 // containers. Rather than return errors, this function logs them and ignores the line. All lines that were able to be
 // parsed into events are returned.
-func extractEventsFromLogs(logs []byte, ignoreCurrentState bool) []event.Event {
+func extractEventsFromLogs(logs []byte, includeCurrentState bool) []event.Event {
 	var extractedEvents []event.Event
 
 	for line := range bytes.Lines(logs) {
-		matcher := containsEventRegexp
-		if ignoreCurrentState {
-			matcher = containsEventNotStateRegexp
+		matcher := containsEventNotStateRegexp
+		if includeCurrentState {
+			matcher = containsEventRegexp
 		}
 
 		if !matcher.Match(line) {

tests/cnf/ran/ptp/internal/metrics/assert.go

@@ -139,6 +139,17 @@ func AssertQuery[V constraints.Integer](
 		option(opts)
 	}
 
+	metricQuery := query.ToMetricQuery()
+	klog.V(tsparams.LogLevel).Infof(
+		"Starting query assertion: query=%s expected=%d startTime=%s timeout=%s pollInterval=%s stableDuration=%s",
+		metricQuery.String(),
+		int64(expected),
+		opts.startTime,
+		opts.timeout,
+		opts.pollInterval,
+		opts.stableDuration,
+	)
+
 	// queryTime is the time at which each query is executed. It begins as the start time and will be incremented by
 	// the poll interval until the timeout is reached.
 	queryTime := opts.startTime
@@ -189,6 +200,16 @@ func AssertQuery[V constraints.Integer](
 		}
 	}
 
+	klog.V(tsparams.LogLevel).Infof(
+		"Query assertion timed out: query=%s expected=%d startTime=%s timeout=%s pollInterval=%s stableDuration=%s",
+		metricQuery.String(),
+		int64(expected),
+		opts.startTime,
+		opts.timeout,
+		opts.pollInterval,
+		opts.stableDuration,
+	)
+
 	return fmt.Errorf("failed to assert query eventually: timeout of %s exceeded", opts.timeout)
 }
 

tests/cnf/ran/ptp/ptp_suite_test.go

@@ -63,7 +63,6 @@ var _ = AfterSuite(func() {
 	By("cleaning up Prometheus API client resources")
 	err = querier.CleanupQuerierResources(RANConfig.Spoke1APIClient)
 	Expect(err).ToNot(HaveOccurred(), "Failed to cleanup Prometheus API client resources")
-
 })
 
 var _ = JustAfterEach(func() {

tests/cnf/ran/ptp/tests/ptp-event-consumer.go

@@ -17,6 +17,7 @@ import (
 	"github.com/rh-ecosystem-edge/eco-gotests/tests/cnf/ran/internal/ranparam"
 	"github.com/rh-ecosystem-edge/eco-gotests/tests/cnf/ran/internal/version"
 	"github.com/rh-ecosystem-edge/eco-gotests/tests/cnf/ran/ptp/internal/consumer"
+	"github.com/rh-ecosystem-edge/eco-gotests/tests/cnf/ran/ptp/internal/eventmetric"
 	"github.com/rh-ecosystem-edge/eco-gotests/tests/cnf/ran/ptp/internal/events"
 	"github.com/rh-ecosystem-edge/eco-gotests/tests/cnf/ran/ptp/internal/iface"
 	"github.com/rh-ecosystem-edge/eco-gotests/tests/cnf/ran/ptp/internal/metrics"
@@ -107,15 +108,19 @@ var _ = Describe("PTP Event Consumer", Label(tsparams.LabelEventConsumer), func(
 				metrics.AssertWithTimeout(10*time.Minute))
 			Expect(err).ToNot(HaveOccurred(), "Failed to assert clock state is locked and stable after pod restart")
 
-			By("waiting up to 10 minutes since startTime for the locked event on the node")
-			eventPod, err := consumer.GetConsumerPodforNode(RANConfig.Spoke1APIClient, nodeInfo.Name)
-			Expect(err).ToNot(HaveOccurred(), "Failed to get event pod for node %s", nodeInfo.Name)
-
-			filter := events.All(
+			By("waiting up to 10 minutes since startTime for the locked event and metric on the node")
+			lockedFilter := events.All(
 				events.IsType(eventptp.PtpStateChange),
 				events.HasValue(events.WithSyncState(eventptp.LOCKED)),
 			)
-			err = events.WaitForEvent(eventPod, startTime, 10*time.Minute, filter)
+			err = eventmetric.NewAssertion(prometheusAPI,
+				metrics.ClockStateQuery{Process: metrics.DoesNotEqual(metrics.ProcessChronyd)},
+				metrics.ClockStateLocked, lockedFilter).
+				ForNode(RANConfig.Spoke1APIClient, nodeInfo.Name).
+				WithStartTime(startTime).
+				WithTimeout(10 * time.Minute).
+				WithMetricOptions(metrics.AssertWithStableDuration(10 * time.Second)).
+				ExecuteAssertion(context.TODO())
 			Expect(err).ToNot(HaveOccurred(), "Failed to wait for locked event on node %s", nodeInfo.Name)
 		}
 
@@ -126,6 +131,14 @@ var _ = Describe("PTP Event Consumer", Label(tsparams.LabelEventConsumer), func(
 
 	// 82218 - Validates the consumer events after ptpoperatorconfig api version is modified
 	It("validates the consumer events after ptpoperatorconfig api version is modified", reportxml.ID("82218"), func() {
+		By("checking if events are enabled")
+		eventsEnabled, err := consumer.AreEventsEnabled(RANConfig.Spoke1APIClient)
+		Expect(err).ToNot(HaveOccurred(), "Failed to check if events are enabled")
+
+		if !eventsEnabled {
+			Skip("Events are not enabled, skipping event consumer API version test")
+		}
+
 		By("checking if the PTP version is within the 4.16-4.18 range")
 		inRange, err := version.IsVersionStringInRange(RANConfig.Spoke1OperatorVersions[ranparam.PTP], "4.16", "4.18")
 		Expect(err).ToNot(HaveOccurred(), "Failed to check PTP version range")

tests/cnf/ran/ptp/tests/ptp-events-and-metrics.go

@@ -14,7 +14,7 @@ import (
 	"github.com/rh-ecosystem-edge/eco-gotests/tests/cnf/internal/nicinfo"
 	"github.com/rh-ecosystem-edge/eco-gotests/tests/cnf/ran/internal/querier"
 	. "github.com/rh-ecosystem-edge/eco-gotests/tests/cnf/ran/internal/raninittools"
-	"github.com/rh-ecosystem-edge/eco-gotests/tests/cnf/ran/ptp/internal/consumer"
+	"github.com/rh-ecosystem-edge/eco-gotests/tests/cnf/ran/ptp/internal/eventmetric"
 	"github.com/rh-ecosystem-edge/eco-gotests/tests/cnf/ran/ptp/internal/events"
 	"github.com/rh-ecosystem-edge/eco-gotests/tests/cnf/ran/ptp/internal/iface"
 	"github.com/rh-ecosystem-edge/eco-gotests/tests/cnf/ran/ptp/internal/metrics"
@@ -100,30 +100,47 @@ var _ = Describe("PTP Events and Metrics", Label(tsparams.LabelEventsAndMetrics)
 				continue
 			}
 
-			testRanAtLeastOnce = true
 			ifaceGroups := iface.GroupInterfacesByNIC(profiles.GetInterfacesNames(clientInterfaces))
 
-			By("getting the event pod for the node")
-			eventPod, err := consumer.GetConsumerPodforNode(RANConfig.Spoke1APIClient, nodeInfo.Name)
-			Expect(err).ToNot(HaveOccurred(), "Failed to get event pod for node %s", nodeInfo.Name)
+			By("getting the egress interface for the node")
+			egressInterface, err := iface.GetEgressInterfaceName(RANConfig.Spoke1APIClient, nodeInfo.Name)
+			Expect(err).ToNot(HaveOccurred(), "Failed to get egress interface name for node %s", nodeInfo.Name)
 
 			for nic, ifaces := range ifaceGroups {
+				if nic == egressInterface.GetNIC() {
+					klog.V(tsparams.LogLevel).Infof("Skipping egress NIC %s", nic)
+
+					continue
+				}
+
+				testRanAtLeastOnce = true
+
 				// Include this interface in the interface information report for this suite.
 				nicinfo.Node(nodeInfo.Name).MarkTested(string(ifaces[0]))
 
 				startTime := time.Now()
 
-				By("adjusting the PHC by 5 ms for NIC " + string(nic))
-				err := iface.AdjustPTPHardwareClock(RANConfig.Spoke1APIClient, nodeInfo.Name, ifaces[0], 0.005)
+				By("adjusting the PHC by 100 ms for NIC " + string(nic))
+				err := iface.AdjustPTPHardwareClock(RANConfig.Spoke1APIClient, nodeInfo.Name, ifaces[0], 0.1)
 				Expect(err).ToNot(HaveOccurred(),
 					"Failed to adjust PTP hardware clock for interface %s on node %s", ifaces[0], nodeInfo.Name)
 
-				By("waiting to receive a FREERUN event")
-				filter := events.All(
+				By("waiting to receive a FREERUN event and metric")
+				freerunFilter := events.All(
 					events.IsType(eventptp.PtpStateChange),
-					events.HasValue(events.WithSyncState(eventptp.FREERUN)),
+					events.HasValue(events.WithSyncState(eventptp.FREERUN), events.OnInterface(nic)),
 				)
-				err = events.WaitForEvent(eventPod, startTime, 5*time.Minute, filter, events.WithoutCurrentState(true))
+				ptp4lClockStateQuery := metrics.ClockStateQuery{
+					Node:      metrics.Equals(nodeInfo.Name),
+					Interface: metrics.Equals(nic),
+					Process:   metrics.Equals(metrics.ProcessPTP4L),
+				}
+				err = eventmetric.NewAssertion(prometheusAPI, ptp4lClockStateQuery, metrics.ClockStateFreerun, freerunFilter).
+					ForNode(RANConfig.Spoke1APIClient, nodeInfo.Name).
+					WithStartTime(startTime).
+					WithTimeout(5 * time.Minute).
+					WithMetricOptions(metrics.AssertWithPollInterval(1 * time.Second)).
+					ExecuteAssertion(context.TODO())
 				Expect(err).ToNot(HaveOccurred(),
 					"Failed to wait for free run event on interface %s on node %s", ifaces[0], nodeInfo.Name)
 
@@ -137,12 +154,16 @@ var _ = Describe("PTP Events and Metrics", Label(tsparams.LabelEventsAndMetrics)
 				// The locked event should happen much sooner than in 15 minutes, except for GM
 				// profiles. This is since the RDS's ts2phc settings for the servo do not allow the PHC
 				// to be updated as quickly. The ptp4l settings allow it to be updated much faster.
-				By("waiting to receive a locked event")
-				filter = events.All(
+				By("waiting to receive a locked event and metric")
+				lockedFilter := events.All(
 					events.IsType(eventptp.PtpStateChange),
-					events.HasValue(events.WithSyncState(eventptp.LOCKED)),
+					events.HasValue(events.WithSyncState(eventptp.LOCKED), events.OnInterface(nic)),
 				)
-				err = events.WaitForEvent(eventPod, startTime, 15*time.Minute, filter, events.WithoutCurrentState(true))
+				err = eventmetric.NewAssertion(prometheusAPI, ptp4lClockStateQuery, metrics.ClockStateLocked, lockedFilter).
+					ForNode(RANConfig.Spoke1APIClient, nodeInfo.Name).
+					WithStartTime(startTime).
+					WithTimeout(15 * time.Minute).
+					ExecuteAssertion(context.TODO())
 				Expect(err).ToNot(HaveOccurred(),
 					"Failed to wait for locked event on interface %s on node %s", ifaces[0], nodeInfo.Name)
 			}