CI: fix PushInitialEvent timeout and simplify crash recovery test

PushInitialEvent used Follow:true with scanner.Scan() but only
checked the timeout after receiving a log line. If no lines arrived,
it blocked indefinitely. Fix by passing context.WithTimeout to the
log stream so scanner.Scan returns when the deadline expires.

Also in the crash recovery test:
- Remove redundant MonitorPodLogsRegex (verifyClockClassCurrentState
  uses PushInitialEvent which reads logs directly)
- Rename verifyClockClassViaEventAPI to verifyClockClassCurrentState
  to reflect that it queries current state via fresh subscription
  with pushInitial=true, unlike verifyClockClassViaEvent which
  drains an existing long-lived subscription

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
Signed-off-by: Jack Ding <jackding@gmail.com>

Author: jzding

test/conformance/serial/ptp.go

@@ -1363,6 +1363,7 @@ var _ = Describe("["+strings.ToLower(DesiredMode.String())+"-serial]", Serial, f
 					}
 				})
 				time.Sleep(10 * time.Second)
+				event.InitPubSub()
 
 				By(fmt.Sprintf("Verifying initial clockClass is %s via metrics", expectedClockClassStr))
 				checkClockClassState(fullConfig, expectedClockClassStr)
@@ -1372,12 +1373,8 @@ var _ = Describe("["+strings.ToLower(DesiredMode.String())+"-serial]", Serial, f
 				By("Verifying initial gm.ClockClass is 6 via PMC")
 				checkClockClassViaPMC(fullConfig, strconv.Itoa(int(fbprotocol.ClockClass6)))
 
-				By(fmt.Sprintf("Setting up event monitoring and verifying initial clockClass is %s via Event API", expectedClockClassStr))
-				event.InitPubSub()
-				term, monErr := event.MonitorPodLogsRegex()
-				Expect(monErr).ToNot(HaveOccurred(), "could not start listening to events")
-				DeferCleanup(func() { stopMonitor(term) })
-				verifyClockClassViaEventAPI(int(expectedClockClass), 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(
@@ -1422,7 +1419,7 @@ var _ = Describe("["+strings.ToLower(DesiredMode.String())+"-serial]", Serial, f
 				checkClockClassViaPMC(fullConfig, strconv.Itoa(int(fbprotocol.ClockClass6)))
 
 				By(fmt.Sprintf("Verifying clockClass is %s via Event API after cloud-event-proxy restart", expectedClockClassStr))
-				verifyClockClassViaEventAPI(int(expectedClockClass), 90*time.Second)
+				verifyClockClassCurrentState(int(expectedClockClass), 90*time.Second)
 			})
 
 			Context("Event API version validation", func() {
@@ -3741,8 +3738,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

@@ -523,39 +523,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()