Add kubelet and CRI-O panic detection invariant test

Signed-off-by: Pannaga Rao Bhoja Ramamanohara

pkg/monitor/monitorapi/types.go

@@ -390,6 +390,9 @@ const (
 	SourceCPUMonitor               IntervalSource = "CPUMonitor"
 	SourceEtcdDiskCommitDuration   IntervalSource = "EtcdDiskCommitDuration"
 	SourceEtcdDiskWalFsyncDuration IntervalSource = "EtcdDiskWalFsyncDuration"
+	KubeletPanic                  IntervalReason = "KubeletPanic"
+	CrioPanic                     IntervalReason = "CrioPanic"
+	SourceCrioLog                 IntervalSource = "CrioLog"
 )
 
 type Interval struct {

pkg/monitortests/node/kubeletlogcollector/monitortest.go

@@ -61,6 +61,7 @@ func (w *kubeletLogCollector) EvaluateTestsFromConstructedIntervals(ctx context.
 	junits = append(junits, nodeFailedLeaseErrorsInRapidSuccession(w.startedAt, finalIntervals)...)
 	junits = append(junits, nodeFailedLeaseErrorsBackOff(w.startedAt, finalIntervals)...)
 	junits = append(junits, testNoSystemdCoreDumps(finalIntervals)...)
+	junits = append(junits, nodeKubeletAndCrioPanicsInvariant(w.startedAt, finalIntervals)...)
 	return junits, nil
 }
 
@@ -173,3 +174,29 @@ func testNoSystemdCoreDumps(events monitorapi.Intervals) []*junitapi.JUnitTestCa
 
 	return []*junitapi.JUnitTestCase{failure}
 }
+
+func nodeKubeletAndCrioPanicsInvariant(startedAt time.Time, finalIntervals monitorapi.Intervals) []*junitapi.JUnitTestCase {
+	const testName = "[sig-node] kubelet-log-collector detects kubelet or CRI-O panics"
+	var failures []string
+
+	intervalsToFailOn := findKubeletAndCrioPanics(finalIntervals)
+	for _, event := range intervalsToFailOn {
+		if event.From.After(startedAt) {
+			failures = append(failures, fmt.Sprintf("%s %v - %v", event.From.Format(time.RFC3339), event.Locator.OldLocator(), event.Message.OldMessage()))
+		}
+	}
+
+	var tests []*junitapi.JUnitTestCase
+	if len(failures) > 0 {
+		tests = append(tests, &junitapi.JUnitTestCase{
+			Name:      testName,
+			SystemOut: strings.Join(failures, "\n"),
+			FailureOutput: &junitapi.FailureOutput{
+				Output: fmt.Sprintf("kubelet-log-collector reports %d kubelet or CRI-O panics.\n\n%v", len(failures), strings.Join(failures, "\n")),
+			},
+		})
+	} else {
+		tests = append(tests, &junitapi.JUnitTestCase{Name: testName})
+	}
+	return tests
+}

pkg/monitortests/node/kubeletlogcollector/node.go

@@ -74,12 +74,21 @@ func intervalsFromNodeLogs(ctx context.Context, kubeClient kubernetes.Interface,
 			}
 			newSystemdCoreDumpIntervals := intervalsFromSystemdCoreDumpLogs(nodeName, systemdCoreDumpLogs)
 
+			crioLogs, err := getNodeLog(ctx, kubeClient, nodeName, "crio")
+			if err != nil {
+				fmt.Fprintf(os.Stderr, "Error getting node crio logs from %s: %s", nodeName, err.Error())
+				errCh <- err
+				return
+			}
+			newCrioLogs := eventsFromCrioLogs(nodeName, crioLogs)
+
 			lock.Lock()
 			defer lock.Unlock()
 			ret = append(ret, newEvents...)
 			ret = append(ret, newOVSEvents...)
 			ret = append(ret, newNetworkManagerIntervals...)
 			ret = append(ret, newSystemdCoreDumpIntervals...)
+			ret = append(ret, newCrioLogs...)
 		}(ctx, node.Name)
 	}
 	wg.Wait()
@@ -118,6 +127,7 @@ func eventsFromKubeletLogs(nodeName string, kubeletLog []byte) monitorapi.Interv
 		ret = append(ret, leaseUpdateError(nodeLocator, currLine)...)
 		ret = append(ret, leaseFailBackOff(nodeLocator, currLine)...)
 		ret = append(ret, parse(nodeName, currLine)...)
+		ret = append(ret, kubeletPanicDetected(nodeName, currLine)...)
 	}
 
 	return ret
@@ -712,3 +722,66 @@ func getNodeLog(ctx context.Context, client kubernetes.Interface, nodeName, syst
 
 	return ioutil.ReadAll(in)
 }
+
+var panicHeadlineRegex = regexp.MustCompile(`(panic:|fatal error:)`)
+
+func kubeletPanicDetected(nodeName, logLine string) monitorapi.Intervals {
+	if !panicHeadlineRegex.MatchString(logLine) {
+		return nil
+	}
+
+	failureTime := utility.SystemdJournalLogTime(logLine, time.Now().Year())
+	nodeLocator := monitorapi.NewLocator().NodeFromName(nodeName)
+
+	return monitorapi.Intervals{
+		monitorapi.NewInterval(monitorapi.SourceKubeletLog, monitorapi.Error).
+			Locator(nodeLocator).
+			Message(monitorapi.NewMessage().Reason(monitorapi.KubeletPanic).
+				HumanMessage("kubelet panic detected, check logs for details")).
+			Display().
+			Build(failureTime, failureTime.Add(1*time.Second)),
+	}
+}
+
+// eventsFromCrioLogs returns the produced intervals from CRI-O logs.
+// Right now it only detects panics, but more detectors can be added as needed.
+func eventsFromCrioLogs(nodeName string, crioLog []byte) monitorapi.Intervals {
+	ret := monitorapi.Intervals{}
+
+	scanner := bufio.NewScanner(bytes.NewBuffer(crioLog))
+	for scanner.Scan() {
+		currLine := scanner.Text()
+		ret = append(ret, crioPanicDetected(nodeName, currLine)...)
+	}
+
+	return ret
+}
+
+func crioPanicDetected(nodeName, logLine string) monitorapi.Intervals {
+	if !panicHeadlineRegex.MatchString(logLine) {
+		return nil
+	}
+
+	failureTime := utility.SystemdJournalLogTime(logLine, time.Now().Year())
+	nodeLocator := monitorapi.NewLocator().NodeFromName(nodeName)
+
+	return monitorapi.Intervals{
+		monitorapi.NewInterval(monitorapi.SourceCrioLog, monitorapi.Error).
+			Locator(nodeLocator).
+			Message(monitorapi.NewMessage().Reason(monitorapi.CrioPanic).
+				HumanMessage("CRI-O panic detected, check logs for details")).
+			Display().
+			Build(failureTime, failureTime.Add(1*time.Second)),
+	}
+}
+
+// findKubeletAndCrioPanics returns all intervals with Reason KubeletPanic or CrioPanic.
+func findKubeletAndCrioPanics(intervals monitorapi.Intervals) monitorapi.Intervals {
+	var panics monitorapi.Intervals
+	for _, interval := range intervals {
+		if interval.Message.Reason == monitorapi.KubeletPanic || interval.Message.Reason == monitorapi.CrioPanic {
+			panics = append(panics, interval)
+		}
+	}
+	return panics
+}

pkg/monitortests/node/kubeletlogcollector/node_test.go

@@ -1243,3 +1243,41 @@ func Test_processCoreDump(t *testing.T) {
 		})
 	}
 }
+
+func TestKubeletPanicDetected(t *testing.T) {
+	nodeName := "test-node"
+	panicLog := "panic: runtime error: invalid memory address or nil pointer dereference"
+	nonPanicLog := "normal log line"
+
+	intervals := kubeletPanicDetected(nodeName, panicLog)
+	if intervals == nil || len(intervals) == 0 {
+		t.Errorf("Expected interval for panic log, got none")
+	}
+	if intervals[0].Condition.Message.HumanMessage != "kubelet panic detected, check logs for details" {
+		t.Errorf("Unexpected message: %v", intervals[0].Condition.Message.HumanMessage)
+	}
+
+	intervals = kubeletPanicDetected(nodeName, nonPanicLog)
+	if intervals != nil {
+		t.Errorf("Expected nil for non-panic log, got: %v", intervals)
+	}
+}
+
+func TestCrioPanicDetected(t *testing.T) {
+	nodeName := "test-node"
+	panicLog := "panic: runtime error: invalid memory address or nil pointer dereference"
+	nonPanicLog := "normal log line"
+
+	intervals := crioPanicDetected(nodeName, panicLog)
+	if intervals == nil || len(intervals) == 0 {
+		t.Errorf("Expected interval for CRI-O panic log, got none")
+	}
+	if intervals[0].Condition.Message.HumanMessage != "CRI-O panic detected, check logs for details" {
+		t.Errorf("Unexpected message: %v", intervals[0].Condition.Message.HumanMessage)
+	}
+
+	intervals = crioPanicDetected(nodeName, nonPanicLog)
+	if intervals != nil {
+		t.Errorf("Expected nil for non-panic log, got: %v", intervals)
+	}
+}