CI: check clockclass after cloud-event-proxy crash with improved readiness

Add crash recovery test for OC, TGM, and dual-follower configs. Fix
PushInitialEvent hang by using context.WithTimeout. Simplify crash
recovery test to use verifyClockClassCurrentState with fresh subscriptions.

WaitForConsumerReady now polls the publisher service health URL
(ptp-event-publisher-service-<node>.openshift-ptp.svc.cluster.local:9043)
from within the consumer pod, then waits for the consumer app to log
"waiting for events" before proceeding, ensuring both the publisher
is reachable and the consumer has finished initialization.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: jzding

test/conformance/serial/ptp.go

@@ -1329,11 +1329,25 @@ var _ = Describe("["+strings.ToLower(DesiredMode.String())+"-serial]", Serial, f
 				if ptphelper.PtpEventEnabled() != 2 {
 					Skip("Skipping: test applies to event API v2 only")
 				}
-				if fullConfig.PtpModeDiscovered != testconfig.BoundaryClock &&
-					fullConfig.PtpModeDiscovered != testconfig.DualNICBoundaryClock &&
-					fullConfig.PtpModeDiscovered != testconfig.DualNICBoundaryClockHA {
-					Skip("Skipping: test applies to boundary clock configurations only")
+
+				// Determine expected clock class based on PTP mode
+				var expectedClockClass fbprotocol.ClockClass
+				switch fullConfig.PtpModeDiscovered {
+				case testconfig.OrdinaryClock, testconfig.DualFollowerClock:
+					// In 4.21+, OC/DualFollower correctly reports its local clock class (255/SlaveOnly).
+					// Before 4.21, OC was incorrectly detected as GM and reported upstream GM's class (6).
+					expectedClockClass = fbprotocol.ClockClass6
+					if ptphelper.IsPTPOperatorVersionAtLeast("4.21") {
+						expectedClockClass = fbprotocol.ClockClassSlaveOnly
+					}
+				case testconfig.BoundaryClock, testconfig.DualNICBoundaryClock, testconfig.DualNICBoundaryClockHA,
+					testconfig.TelcoGrandMasterClock:
+					expectedClockClass = fbprotocol.ClockClass6
+				default:
+					Skip(fmt.Sprintf("Skipping: test does not apply to %s mode", fullConfig.PtpModeDiscovered))
 				}
+				expectedClockClassStr := strconv.Itoa(int(expectedClockClass))
+				logrus.Infof("PtpModeDiscovered: %s, expected clockClass: %s", fullConfig.PtpModeDiscovered, expectedClockClassStr)
 
 				By("Deploying consumer app for event API v2")
 				nodeName := fullConfig.DiscoveredClockUnderTestPod.Spec.NodeName
@@ -1348,28 +1362,29 @@ var _ = Describe("["+strings.ToLower(DesiredMode.String())+"-serial]", Serial, f
 						event.PubSub.Close()
 					}
 				})
-				time.Sleep(10 * time.Second)
+				if waitErr := event.WaitForConsumerReady(nodeName); waitErr != nil {
+					Skip(fmt.Sprintf("Consumer app not ready: %v", waitErr))
+				}
+				event.InitPubSub()
 
-				By("Verifying initial clockClass is 6 via metrics")
-				checkClockClassState(fullConfig, strconv.Itoa(int(fbprotocol.ClockClass6)))
+				By(fmt.Sprintf("Verifying initial clockClass is %s via metrics", expectedClockClassStr))
+				checkClockClassState(fullConfig, expectedClockClassStr)
 
-				By("Verifying initial clockClass is 6 via PMC")
+				// PMC gm.ClockClass always reports the upstream GM's class (6),
+				// regardless of whether this node is BC or OC.
+				By("Verifying initial gm.ClockClass is 6 via PMC")
 				checkClockClassViaPMC(fullConfig, strconv.Itoa(int(fbprotocol.ClockClass6)))
 
-				By("Setting up event monitoring and verifying initial clockClass is 6 via Event API")
-				event.InitPubSub()
-				term, monErr := event.MonitorPodLogsRegex()
-				Expect(monErr).ToNot(HaveOccurred(), "could not start listening to events")
-				DeferCleanup(func() { stopMonitor(term) })
-				verifyClockClassViaEventAPI(int(fbprotocol.ClockClass6), 60*time.Second)
+				By(fmt.Sprintf("Verifying initial clockClass is %s via Event API", expectedClockClassStr))
+				verifyClockClassCurrentState(int(expectedClockClass), 60*time.Second)
 
 				By("Killing cloud-event-proxy process in sidecar container")
 				_, _, killErr := pods.ExecCommand(
 					client.Client,
 					true,
 					fullConfig.DiscoveredClockUnderTestPod,
 					pkg.EventProxyContainerName,
-					[]string{"sh", "-c", "kill -9 $(pgrep -f ^./cloud-event-proxy) || true"},
+					[]string{"sh", "-c", "pkill -9 -f ^./cloud-event-proxy || true"},
 				)
 				Expect(killErr).To(BeNil(), "failed to kill cloud-event-proxy process")
 
@@ -1399,14 +1414,14 @@ var _ = Describe("["+strings.ToLower(DesiredMode.String())+"-serial]", Serial, f
 				}, 2*time.Minute, 2*time.Second).Should(ContainSubstring("OK"),
 					"cloud-event-proxy health endpoint did not recover after restart")
 
-				By("Verifying clockClass remains 6 via metrics after cloud-event-proxy restart")
-				checkClockClassState(fullConfig, strconv.Itoa(int(fbprotocol.ClockClass6)))
+				By(fmt.Sprintf("Verifying clockClass remains %s via metrics after cloud-event-proxy restart", expectedClockClassStr))
+				checkClockClassState(fullConfig, expectedClockClassStr)
 
-				By("Verifying clockClass remains 6 via PMC after cloud-event-proxy restart")
+				By("Verifying gm.ClockClass remains 6 via PMC after cloud-event-proxy restart")
 				checkClockClassViaPMC(fullConfig, strconv.Itoa(int(fbprotocol.ClockClass6)))
 
-				By("Verifying clockClass is 6 via Event API after cloud-event-proxy restart")
-				verifyClockClassViaEventAPI(int(fbprotocol.ClockClass6), 90*time.Second)
+				By(fmt.Sprintf("Verifying clockClass is %s via Event API after cloud-event-proxy restart", expectedClockClassStr))
+				verifyClockClassCurrentState(int(expectedClockClass), 90*time.Second)
 			})
 
 			Context("Event API version validation", func() {
@@ -3692,7 +3707,10 @@ func setupBCClockClassEvents(nodeName string) bcEventContext {
 		logrus.Warnf("PTP events not available: %s; skipping event checks", createErr)
 		return ctx
 	}
-	time.Sleep(10 * time.Second)
+	if waitErr := event.WaitForConsumerReady(nodeName); waitErr != nil {
+		logrus.Warnf("Consumer app not ready: %s; skipping event checks", waitErr)
+		return ctx
+	}
 	event.InitPubSub()
 	var eventCleanup func()
 	ctx.subs, eventCleanup = event.SubscribeToGMChangeEvents(100, true, 60*time.Second)
@@ -3725,8 +3743,12 @@ func verifyClockClassViaEvent(evCtx bcEventContext, expectedClass int) {
 	verifyMetric(events[ptpEvent.PtpClockClassChange], float64(expectedClass))
 }
 
-// verifyClockClassViaEventAPI verifies clock class via Event Fast Notification API getCurrentState
-func verifyClockClassViaEventAPI(expectedClockClass int, timeout time.Duration) {
+// verifyClockClassCurrentState creates a fresh event subscription with pushInitial=true
+// to query the current clock class state from the Event API's getCurrentState endpoint.
+// Unlike verifyClockClassViaEvent (which drains an existing long-lived subscription),
+// this function is safe to call after disruptions such as a cloud-event-proxy restart,
+// where the previous subscription may be stale or disconnected.
+func verifyClockClassCurrentState(expectedClockClass int, timeout time.Duration) {
 	const incomingEventsBuffer = 100
 	subs, cleanup := event.SubscribeToGMChangeEvents(incomingEventsBuffer, true, timeout)
 	defer cleanup()

test/pkg/event/event.go

@@ -38,6 +38,7 @@ const (
 	ConsumerSaName                = "consumer-sa"
 	ConsumerServiceName           = "consumer-events-subscription-service"
 	TestSidecarSuccessLogString   = "rest service returned healthy status"
+	ConsumerReadyLogString        = "waiting for events"
 	ConsumerContainerName         = "consumer"
 	sidecarNamespaceDeleteTimeout = time.Minute * 2
 	ApiBaseV1                     = "127.0.0.1:9085/api/ocloudNotifications/v1/"
@@ -322,6 +323,50 @@ func CreateConsumerApp(nodeNameFull string) (err error) {
 	return nil
 }
 
+// WaitForConsumerReady polls the publisher's health endpoint from within the
+// consumer pod until it responds OK, then waits for the consumer app to log
+// "waiting for events" confirming it has finished initialization.
+func WaitForConsumerReady(nodeNameFull string) error {
+	nodeName := nodeNameFull
+	if nodeNameFull != "" && strings.Contains(nodeNameFull, ".") {
+		nodeName = strings.Split(nodeName, ".")[0]
+	}
+	publisherHealthURL := fmt.Sprintf("http://ptp-event-publisher-service-%s.%s.svc.cluster.local:9043/api/ocloudNotifications/v2/health", nodeName, ProducerNamespace)
+	logrus.Infof("Waiting for publisher to become ready at %s", publisherHealthURL)
+
+	consumerPod, err := testclient.Client.Pods(ConsumerNamespace).Get(context.TODO(), ConsumerPodName, metav1.GetOptions{})
+	if err != nil {
+		return fmt.Errorf("could not get consumer pod: %s", err)
+	}
+
+	deadline := time.Now().Add(1 * time.Minute)
+	for time.Now().Before(deadline) {
+		buf, _, execErr := pods.ExecCommand(
+			testclient.Client,
+			false,
+			consumerPod,
+			ConsumerContainerName,
+			[]string{"curl", "-s", publisherHealthURL},
+		)
+		if execErr == nil && strings.Contains(buf.String(), "OK") {
+			logrus.Info("Publisher health endpoint is ready")
+			break
+		}
+		time.Sleep(2 * time.Second)
+	}
+	if time.Now().After(deadline) {
+		return fmt.Errorf("publisher health endpoint %s did not return OK within 1 minute", publisherHealthURL)
+	}
+
+	logrus.Info("Waiting for consumer app to finish initialization")
+	_, err = pods.GetPodLogsRegex(ConsumerNamespace, ConsumerPodName, ConsumerContainerName, ConsumerReadyLogString, true, 1*time.Minute)
+	if err != nil {
+		return fmt.Errorf("consumer app did not become ready: %s", err)
+	}
+	logrus.Info("Consumer app is ready")
+	return nil
+}
+
 const roleName = "use-privileged"
 const roleBindingName = "sidecar-rolebinding"
 
@@ -523,39 +568,35 @@ func PushInitialEvent(eventType string, timeout time.Duration) (err error) {
 		TailLines: &count,
 	}
 
+	ctx, cancel := context.WithTimeout(context.Background(), timeout)
+	defer cancel()
+
 	podLogRequest := testclient.Client.CoreV1().Pods(namespace).GetLogs(podName, &podLogOptions)
-	stream, err := podLogRequest.Stream(context.TODO())
+	stream, err := podLogRequest.Stream(ctx)
 	if err != nil {
 		return fmt.Errorf("could not retrieve log in ns=%s pod=%s, err=%s", namespace, podName, err)
 	}
 	defer stream.Close()
-	start := time.Now()
-	for {
-		scanner := bufio.NewScanner(stream)
-		for scanner.Scan() {
-			t := time.Now()
-			elapsed := t.Sub(start)
-			if elapsed > timeout {
-				return fmt.Errorf("timedout PushInitialValue, waiting for log in ns=%s pod=%s, looking for = %s", namespace, podName, regex)
+
+	r := regexp.MustCompile(regex)
+	scanner := bufio.NewScanner(stream)
+	for scanner.Scan() {
+		line := scanner.Text()
+		logrus.Trace(line)
+
+		matches := r.FindAllStringSubmatch(line, -1)
+		if len(matches) > 0 {
+			aStoredEvent, eType, err := createStoredEvent([]byte(matches[0][1]))
+			if err != nil {
+				return err
 			}
-			line := scanner.Text()
-			logrus.Trace(line)
-
-			r := regexp.MustCompile(regex)
-			matches := r.FindAllStringSubmatch(line, -1)
-			if len(matches) > 0 {
-				aStoredEvent, eType, err := createStoredEvent([]byte(matches[0][1]))
-				if err != nil {
-					return err
-				}
-				if eType == eventType {
-					PubSub.Publish(eType, aStoredEvent)
-					return nil
-				}
+			if eType == eventType {
+				PubSub.Publish(eType, aStoredEvent)
+				return nil
 			}
 		}
-
 	}
+	return fmt.Errorf("timed out PushInitialEvent, waiting for log in ns=%s pod=%s, looking for %s", namespace, podName, regex)
 }
 func createStoredEvent(data []byte) (aStoredEvent exports.StoredEvent, aType string, err error) {
 	apiVersion := ptphelper.PtpEventEnabled()