Add intervals and a test for systemd-coredumps

Author: dgoodwin

pkg/monitor/monitorapi/types.go

@@ -272,7 +272,8 @@ const (
 	ReasonHighGeneration    IntervalReason = "HighGeneration"
 	ReasonInvalidGeneration IntervalReason = "GenerationViolation"
 
-	ReasonEtcdBootstrap IntervalReason = "EtcdBootstrap"
+	ReasonEtcdBootstrap     IntervalReason = "EtcdBootstrap"
+	ReasonProcessDumpedCore IntervalReason = "ProcessDumpedCore"
 )
 
 type AnnotationKey string
@@ -358,6 +359,7 @@ const (
 	SourceUnexpectedReady           IntervalSource = "NodeUnexpectedNotReady"
 	SourceUnreachable               IntervalSource = "NodeUnreachable"
 	SourceKubeletLog                IntervalSource = "KubeletLog"
+	SourceSystemdCoreDumpLog        IntervalSource = "SystemdCoreDumpLog"
 	SourcePodLog                    IntervalSource = "PodLog"
 	SourceEtcdLog                   IntervalSource = "EtcdLog"
 	SourceEtcdLeadership            IntervalSource = "EtcdLeadership"

pkg/monitortests/node/kubeletlogcollector/monitortest.go

@@ -60,6 +60,7 @@ func (w *kubeletLogCollector) EvaluateTestsFromConstructedIntervals(ctx context.
 	junits := []*junitapi.JUnitTestCase{}
 	junits = append(junits, nodeFailedLeaseErrorsInRapidSuccession(w.startedAt, finalIntervals)...)
 	junits = append(junits, nodeFailedLeaseErrorsBackOff(w.startedAt, finalIntervals)...)
+	junits = append(junits, testNoSystemdCoreDumps(finalIntervals)...)
 	return junits, nil
 }
 
@@ -125,3 +126,51 @@ func nodeFailedLeaseErrorsBackOff(startedAt time.Time, finalIntervals monitorapi
 	tests = append(tests, &junitapi.JUnitTestCase{Name: testName})
 	return tests
 }
+
+func testNoSystemdCoreDumps(events monitorapi.Intervals) []*junitapi.JUnitTestCase {
+	const testName = "[Jira:\"Test Framework\"] should not find any systemd-coredump logs in system journal"
+	success := &junitapi.JUnitTestCase{Name: testName}
+
+	var failures []string
+	processCount := make(map[string]int)
+
+	for _, event := range events {
+		if event.Source != monitorapi.SourceSystemdCoreDumpLog {
+			continue
+		}
+		if strings.Contains(event.Message.HumanMessage, "dumped core") {
+			processName := "unknown"
+			if event.Message.Annotations != nil {
+				if proc, exists := event.Message.Annotations["process"]; exists {
+					processName = proc
+				}
+			}
+
+			processCount[processName]++
+			msg := fmt.Sprintf("%v - Process: %s - %v", event.Locator.OldLocator(), processName, event.Message.OldMessage())
+			failures = append(failures, msg)
+		}
+	}
+
+	if len(failures) == 0 {
+		return []*junitapi.JUnitTestCase{success}
+	}
+
+	// Create summary of process failures
+	processSummary := make([]string, 0, len(processCount))
+	for process, count := range processCount {
+		processSummary = append(processSummary, fmt.Sprintf("%s: %d occurrences", process, count))
+	}
+
+	failure := &junitapi.JUnitTestCase{
+		Name:      testName,
+		SystemOut: strings.Join(failures, "\n"),
+		FailureOutput: &junitapi.FailureOutput{
+			Output: fmt.Sprintf("Found %d core dumps from %d different processes. Process breakdown:\n%s\n\nDetailed events:\n%v",
+				len(failures), len(processCount), strings.Join(processSummary, "\n"), strings.Join(failures, "\n")),
+		},
+	}
+
+	// Core dumps are serious issues, but treating as flake initially to gather data
+	return []*junitapi.JUnitTestCase{failure, success}
+}

pkg/monitortests/node/kubeletlogcollector/node.go

@@ -56,7 +56,7 @@ func intervalsFromNodeLogs(ctx context.Context, kubeClient kubernetes.Interface,
 				errCh <- err
 				return
 			}
-			newOVSEvents := eventsFromOVSVswitchdLogs(nodeName, ovsVswitchdLogs)
+			newOVSEvents := intervalsFromOVSVswitchdLogs(nodeName, ovsVswitchdLogs)
 
 			networkManagerLogs, err := getNodeLog(ctx, kubeClient, nodeName, "NetworkManager")
 			if err != nil {
@@ -66,11 +66,20 @@ func intervalsFromNodeLogs(ctx context.Context, kubeClient kubernetes.Interface,
 			}
 			newNetworkManagerIntervals := intervalsFromNetworkManagerLogs(nodeName, networkManagerLogs)
 
+			systemdCoreDumpLogs, err := getNodeLog(ctx, kubeClient, nodeName, "systemd-coredump")
+			if err != nil {
+				fmt.Fprintf(os.Stderr, "Error getting node systemd-coredump logs from %s: %s", nodeName, err.Error())
+				errCh <- err
+				return
+			}
+			newSystemdCoreDumpIntervals := intervalsFromSystemdCoreDumpLogs(nodeName, systemdCoreDumpLogs)
+
 			lock.Lock()
 			defer lock.Unlock()
 			ret = append(ret, newEvents...)
 			ret = append(ret, newOVSEvents...)
 			ret = append(ret, newNetworkManagerIntervals...)
+			ret = append(ret, newSystemdCoreDumpIntervals...)
 		}(ctx, node.Name)
 	}
 	wg.Wait()
@@ -114,9 +123,9 @@ func eventsFromKubeletLogs(nodeName string, kubeletLog []byte) monitorapi.Interv
 	return ret
 }
 
-// eventsFromOVSVswitchdLogs returns the produced intervals.  Any errors during this creation are logged, but
+// intervalsFromOVSVswitchdLogs returns the produced intervals.  Any errors during this creation are logged, but
 // not returned because this is a best effort step
-func eventsFromOVSVswitchdLogs(nodeName string, ovsLogs []byte) monitorapi.Intervals {
+func intervalsFromOVSVswitchdLogs(nodeName string, ovsLogs []byte) monitorapi.Intervals {
 	nodeLocator := monitorapi.NewLocator().NodeFromName(nodeName)
 	ret := monitorapi.Intervals{}
 
@@ -164,6 +173,55 @@ func unreasonablyLongPollInterval(logLine string, nodeLocator monitorapi.Locator
 
 var unreasonablyLongPollIntervalRE = regexp.MustCompile(`Unreasonably long (\d+)ms poll interval`)
 
+// intervalsFromSystemdCoreDumpLogs returns the produced intervals.  Any errors during this creation are logged, but
+// not returned because this is a best effort step
+func intervalsFromSystemdCoreDumpLogs(nodeName string, coreDumpLogs []byte) monitorapi.Intervals {
+	nodeLocator := monitorapi.NewLocator().NodeFromName(nodeName)
+	ret := monitorapi.Intervals{}
+
+	scanner := bufio.NewScanner(bytes.NewBuffer(coreDumpLogs))
+	for scanner.Scan() {
+		currLine := scanner.Text()
+		ret = append(ret, processCoreDump(currLine, nodeLocator)...)
+	}
+
+	return ret
+}
+
+// processCoreDump searches for core dump events with process information
+//
+// Process 7798 (haproxy) of user 1000680000 dumped core.
+func processCoreDump(logLine string, nodeLocator monitorapi.Locator) monitorapi.Intervals {
+	if !strings.Contains(logLine, "dumped core") {
+		return nil
+	}
+
+	logTime := utility.SystemdJournalLogTime(logLine, time.Now().Year())
+
+	// Extract the process name from within parentheses
+	var processName string
+	match := coreDumpProcessRE.FindStringSubmatch(logLine)
+	if match != nil && len(match) > 1 {
+		processName = match[1]
+	}
+
+	message := logLine[strings.Index(logLine, "Process"):]
+
+	// Build the message with process annotation if we extracted it
+	messageBuilder := monitorapi.NewMessage().HumanMessage(message).Reason(monitorapi.ReasonProcessDumpedCore)
+	if processName != "" {
+		messageBuilder = messageBuilder.WithAnnotation("process", processName)
+	}
+
+	interval := monitorapi.NewInterval(monitorapi.SourceSystemdCoreDumpLog, monitorapi.Warning).Locator(
+		nodeLocator).Message(messageBuilder).
+		Display().Build(logTime, logTime.Add(1*time.Second))
+
+	return monitorapi.Intervals{interval}
+}
+
+var coreDumpProcessRE = regexp.MustCompile(`Process \d+ \(([^)]+)\) of user \d+ dumped core`)
+
 // intervalsFromNetworkManagerLogs returns the produced intervals.  Any errors during this creation are logged, but
 // not returned because this is a best effort step
 func intervalsFromNetworkManagerLogs(nodeName string, ovsLogs []byte) monitorapi.Intervals {