[LEP-4139] feat(nvlink): dedup link failures, report one-shot kmsg scans (#1272)

## Summary

- detect both post-RX failure signatures from
`knvlinkUpdatePostRxDetectLinkMask_*` and
`knvlinkDiscoverPostRxDetLinks_*`
- normalize paired and repeated messages into one event per 5-minute
bucket for daemon and one-shot scan paths, consistent with the other
NVIDIA kmsg components
- report one-shot kmsg scan results and return an unhealthy state with a
reboot recommendation when the failure is found

Author: gyuho

components/accelerator/nvidia/nvlink/component.go

@@ -32,9 +32,10 @@ import (
 const Name = "accelerator-nvidia-nvlink"
 
 const (
-	defaultCheckInterval       = time.Minute
-	defaultCheckStaleAfter     = 2 * defaultCheckInterval
-	defaultStateUpdateInterval = 30 * time.Second
+	defaultCheckInterval        = time.Minute
+	defaultCheckStaleAfter      = 2 * defaultCheckInterval
+	defaultStateUpdateInterval  = 30 * time.Second
+	defaultKmsgEventDedupWindow = 5 * time.Minute
 )
 
 var _ components.Component = &component{}
@@ -102,6 +103,7 @@ func New(gpudInstance *components.GPUdInstance) (components.Component, error) {
 				cctx,
 				func(line string) (string, string) { return matchWithBootID(line, bootID) },
 				c.eventBucket,
+				kmsg.WithCacheKeyTruncateSeconds(int(defaultKmsgEventDedupWindow.Seconds())),
 			)
 			if err != nil {
 				c.eventBucket.Close()
@@ -274,10 +276,29 @@ func (c *component) Check() components.CheckResult {
 			log.Logger.Warnw(cr.reason, "error", cr.err)
 			return cr
 		}
+		cr.KmsgScanned = true
+		type kmsgEventWindow struct {
+			name, message string
+			window        int64
+		}
+		seen := make(map[kmsgEventWindow]struct{})
 		for _, message := range kmsgs {
-			if !HasPostRxDetectFailure(message.Message) {
+			eventName, eventMessage := Match(message.Message)
+			if eventName == "" {
 				continue
 			}
+			key := kmsgEventWindow{
+				name:    eventName,
+				message: eventMessage,
+				window:  message.Timestamp.Unix() / int64(defaultKmsgEventDedupWindow/time.Second),
+			}
+			if _, ok := seen[key]; ok {
+				continue
+			}
+			seen[key] = struct{}{}
+			cr.MatchedKmsgs = append(cr.MatchedKmsgs, message)
+		}
+		if len(cr.MatchedKmsgs) > 0 {
 			state := unhealthyRebootState(cr.ts, postRxDetectFailureMessage)
 			cr.health = state.Health
 			cr.reason = state.Reason
@@ -439,6 +460,9 @@ func (c *component) Check() components.CheckResult {
 var _ components.CheckResult = &checkResult{}
 
 type checkResult struct {
+	KmsgScanned  bool           `json:"-"`
+	MatchedKmsgs []kmsg.Message `json:"-"`
+
 	// NVLinks contains detailed NVLink information for all GPUs checked
 	NVLinks []NVLink `json:"nvlinks,omitempty"`
 
@@ -548,7 +572,14 @@ func (cr *checkResult) String() string {
 	if cr == nil {
 		return ""
 	}
+	kmsgSummary := ""
+	if cr.KmsgScanned {
+		kmsgSummary = fmt.Sprintf("matched %d kmsg(s)", len(cr.MatchedKmsgs))
+	}
 	if len(cr.NVLinks) == 0 {
+		if kmsgSummary != "" {
+			return kmsgSummary
+		}
 		return "no data"
 	}
 
@@ -581,13 +612,19 @@ func (cr *checkResult) String() string {
 	}
 	table.Render()
 
+	if kmsgSummary != "" {
+		return kmsgSummary + "\n\n" + buf.String()
+	}
 	return buf.String()
 }
 
 func (cr *checkResult) Summary() string {
 	if cr == nil {
 		return ""
 	}
+	if cr.KmsgScanned {
+		return "scanned kmsg(s); " + cr.reason
+	}
 	return cr.reason
 }
 

components/accelerator/nvidia/nvlink/component_test.go

@@ -613,6 +613,15 @@ func TestData_String(t *testing.T) {
 			}
 		})
 	}
+
+	t.Run("with kmsg summary", func(t *testing.T) {
+		got := (&checkResult{
+			KmsgScanned: true,
+			NVLinks:     []NVLink{{UUID: "gpu-uuid-123"}},
+		}).String()
+		assert.Contains(t, got, "matched 0 kmsg(s)")
+		assert.Contains(t, got, "gpu-uuid-123")
+	})
 }
 
 func TestData_Summary(t *testing.T) {

components/accelerator/nvidia/nvlink/kmsg.go

@@ -7,13 +7,12 @@ const (
 	// update an NVLink post-RX-detection link mask.
 	EventNamePostRxDetectFailure = "nvlink_post_rx_detect_failure"
 
-	// NVIDIA's published GH100 kernel source describes
-	// knvlinkDiscoverPostRxDetLinks_GH100 as discovering links that are training
-	// or trained on both GPUs. It emits this exact LEVEL_ERROR and returns
-	// NV_ERR_INVALID_STATE when knvlinkUpdatePostRxDetectLinkMask fails. The
-	// source does not classify this error as a driver hang.
+	// NVIDIA's published GH100 discovery path calls
+	// knvlinkUpdatePostRxDetectLinkMask and reports the peer-mask failure when
+	// that update fails. Some driver builds also emit the paired update failure.
+	// The source does not classify this error as a driver hang.
 	// ref. https://github.com/NVIDIA/open-gpu-kernel-modules/blob/452cec62d827034798072827d3866d1881662b77/src/nvidia/src/kernel/gpu/nvlink/arch/hopper/kernel_nvlink_gh100.c#L173-L249
-	RegexPostRxDetectFailureKMessage = `NVRM: knvlinkDiscoverPostRxDetLinks_[A-Za-z0-9_]+: Getting peer[0-9]+(?:'s)? postRxDetLinkMask failed!`
+	RegexPostRxDetectFailureKMessage = `NVRM: (?:knvlinkUpdatePostRxDetectLinkMask_[A-Za-z0-9_]+: Failed to update Rx Detect Link mask!|knvlinkDiscoverPostRxDetLinks_[A-Za-z0-9_]+: Getting peer[0-9]+(?:'s)? postRxDetLinkMask failed!)`
 
 	postRxDetectFailureMessage = "NVIDIA driver failed to update an NVLink post-RX-detection link mask"
 )

components/accelerator/nvidia/nvlink/kmsg_test.go

@@ -17,6 +17,11 @@ func TestMatch(t *testing.T) {
 			line:  "NVRM: knvlinkDiscoverPostRxDetLinks_GH100: Getting peer0's postRxDetLinkMask failed!",
 			match: true,
 		},
+		{
+			name:  "incident link-mask update signature",
+			line:  "NVRM: knvlinkUpdatePostRxDetectLinkMask_IMPL: Failed to update Rx Detect Link mask!",
+			match: true,
+		},
 		{
 			name:  "other GPU architecture and peer",
 			line:  "kernel: NVRM: knvlinkDiscoverPostRxDetLinks_GB100_REV2: Getting peer17 postRxDetLinkMask failed!",

components/accelerator/nvidia/nvlink/mock_component_test.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"errors"
 	"os"
+	"reflect"
 	"runtime"
 	"testing"
 	"time"
@@ -27,13 +28,15 @@ func TestNew_WithEventStoreAsRoot_KmsgSyncer_WithMockey(t *testing.T) {
 
 		bucket := &memoryEventBucket{}
 		var match kmsg.MatchFunc
+		var syncerOpts []kmsg.OpOption
 		mockey.Mock(kmsg.NewSyncer).To(func(
 			_ context.Context,
 			matchFunc kmsg.MatchFunc,
 			eventBucket eventstore.Bucket,
-			_ ...kmsg.OpOption,
+			opts ...kmsg.OpOption,
 		) (*kmsg.Syncer, error) {
 			match = matchFunc
+			syncerOpts = opts
 			assert.Same(t, bucket, eventBucket)
 			return &kmsg.Syncer{}, nil
 		}).Build()
@@ -48,6 +51,8 @@ func TestNew_WithEventStoreAsRoot_KmsgSyncer_WithMockey(t *testing.T) {
 		require.NotNil(t, c.kmsgSyncer)
 		assert.Nil(t, c.readAllKmsg, "daemon components should rely on the kmsg syncer")
 		require.NotNil(t, match)
+		require.Len(t, syncerOpts, 1)
+		assert.Equal(t, int(defaultKmsgEventDedupWindow.Seconds()), getKmsgCacheKeyTruncateSeconds(t, syncerOpts[0]))
 
 		name, message := match("NVRM: knvlinkDiscoverPostRxDetLinks_GH100: Getting peer0's postRxDetLinkMask failed!")
 		assert.Equal(t, EventNamePostRxDetectFailure, name)
@@ -63,6 +68,17 @@ func TestNew_WithEventStoreAsRoot_KmsgSyncer_WithMockey(t *testing.T) {
 	})
 }
 
+func getKmsgCacheKeyTruncateSeconds(t *testing.T, opt kmsg.OpOption) int {
+	t.Helper()
+	op := &kmsg.Op{}
+	opt(op)
+
+	v := reflect.ValueOf(op).Elem().FieldByName("cacheKeyTruncateSeconds")
+	require.True(t, v.IsValid(), "cacheKeyTruncateSeconds field must exist")
+	require.Equal(t, reflect.Int, v.Kind(), "cacheKeyTruncateSeconds must be int")
+	return int(v.Int())
+}
+
 func TestNew_WithEventStoreAsRoot_KmsgSyncerError_WithMockey(t *testing.T) {
 	mockey.PatchConvey("New cleans up when the kmsg syncer cannot be created", t, func() {
 		mockey.Mock(os.Geteuid).Return(0).Build()

components/accelerator/nvidia/nvlink/post_rx_detect_failure_test.go

@@ -10,6 +10,7 @@ import (
 
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 
 	apiv1 "github.com/leptonai/gpud/api/v1"
 	"github.com/leptonai/gpud/pkg/eventstore"
@@ -192,7 +193,7 @@ func TestHealthStateFailsWhenNoCheckCompletesAfterMonitoringStarts(t *testing.T)
 	assert.Contains(t, state.Reason, "has not refreshed")
 }
 
-func TestCheckDetectsPostRxDetectFailureForOneShotScan(t *testing.T) {
+func TestCheckDetectsAndDeduplicatesPostRxDetectFailureForOneShotScan(t *testing.T) {
 	getNVLinkCalled := false
 	c := mustComponent(t, MockNVLinkComponent(
 		context.Background(),
@@ -202,19 +203,39 @@ func TestCheckDetectsPostRxDetectFailureForOneShotScan(t *testing.T) {
 			return NVLink{}, nil
 		},
 	))
+	baseTime := time.Date(2026, 2, 3, 3, 0, 0, 0, time.UTC)
+	reportedKmsgs := []kmsg.Message{
+		{Timestamp: metav1.NewTime(baseTime), Message: "[6619715.535712] NVRM: knvlinkUpdatePostRxDetectLinkMask_IMPL: Failed to update Rx Detect Link mask!"},
+		{Timestamp: metav1.NewTime(baseTime.Add(18 * time.Microsecond)), Message: "[6619715.535730] NVRM: knvlinkDiscoverPostRxDetLinks_GH100: Getting peer0's postRxDetLinkMask failed!"},
+		{Timestamp: metav1.NewTime(baseTime.Add(4*time.Second + 2771*time.Microsecond)), Message: "[6619719.538483] NVRM: knvlinkUpdatePostRxDetectLinkMask_IMPL: Failed to update Rx Detect Link mask!"},
+		{Timestamp: metav1.NewTime(baseTime.Add(4*time.Second + 2787*time.Microsecond)), Message: "[6619719.538499] NVRM: knvlinkDiscoverPostRxDetLinks_GH100: Getting peer0's postRxDetLinkMask failed!"},
+		{Timestamp: metav1.NewTime(baseTime.Add(8*time.Second + 3808*time.Microsecond)), Message: "[6619723.539520] NVRM: knvlinkUpdatePostRxDetectLinkMask_IMPL: Failed to update Rx Detect Link mask!"},
+		{Timestamp: metav1.NewTime(baseTime.Add(8*time.Second + 3827*time.Microsecond)), Message: "[6619723.539539] NVRM: knvlinkDiscoverPostRxDetLinks_GH100: Getting peer0's postRxDetLinkMask failed!"},
+	}
+	for _, message := range reportedKmsgs {
+		eventName, eventMessage := Match(message.Message)
+		assert.Equal(t, EventNamePostRxDetectFailure, eventName, message.Message)
+		assert.Equal(t, postRxDetectFailureMessage, eventMessage, message.Message)
+	}
 	c.readAllKmsg = func(context.Context) ([]kmsg.Message, error) {
-		return []kmsg.Message{{
-			Message: "NVRM: knvlinkDiscoverPostRxDetLinks_GH100: Getting peer0's postRxDetLinkMask failed!",
-		}}, nil
+		return reportedKmsgs, nil
 	}
 
 	result := c.Check()
 	assert.Equal(t, apiv1.HealthStateTypeUnhealthy, result.HealthStateType())
+	assert.Contains(t, result.Summary(), "scanned kmsg(s)")
 	assert.Contains(t, result.Summary(), postRxDetectFailureMessage)
+	assert.Equal(t, "matched 1 kmsg(s)", result.String())
 	assert.False(t, getNVLinkCalled, "kmsg failure should short-circuit NVML probing")
 	state := result.HealthStates()[0]
 	require.NotNil(t, state.SuggestedActions)
 	assert.Equal(t, []apiv1.RepairActionType{apiv1.RepairActionTypeRebootSystem}, state.SuggestedActions.RepairActions)
+
+	reportedKmsgs = append(reportedKmsgs, kmsg.Message{
+		Timestamp: metav1.NewTime(baseTime.Add(defaultKmsgEventDedupWindow)),
+		Message:   "NVRM: knvlinkUpdatePostRxDetectLinkMask_IMPL: Failed to update Rx Detect Link mask!",
+	})
+	assert.Equal(t, "matched 2 kmsg(s)", c.Check().String(), "a later dedup window should report the failure again")
 }
 
 func TestCheckHandlesOneShotKmsgReadResult(t *testing.T) {
@@ -223,6 +244,7 @@ func TestCheckHandlesOneShotKmsgReadResult(t *testing.T) {
 		readAll    func(context.Context) ([]kmsg.Message, error)
 		wantHealth apiv1.HealthStateType
 		wantReason string
+		wantString string
 	}{
 		{
 			name: "read failure",
@@ -231,6 +253,7 @@ func TestCheckHandlesOneShotKmsgReadResult(t *testing.T) {
 			},
 			wantHealth: apiv1.HealthStateTypeUnhealthy,
 			wantReason: "failed to read kmsg",
+			wantString: "no data",
 		},
 		{
 			name: "no matching message",
@@ -239,6 +262,7 @@ func TestCheckHandlesOneShotKmsgReadResult(t *testing.T) {
 			},
 			wantHealth: apiv1.HealthStateTypeHealthy,
 			wantReason: "all 0 GPU(s) were checked, no nvlink issue found",
+			wantString: "matched 0 kmsg(s)",
 		},
 	}
 
@@ -254,6 +278,7 @@ func TestCheckHandlesOneShotKmsgReadResult(t *testing.T) {
 			result := c.Check()
 			assert.Equal(t, tt.wantHealth, result.HealthStateType())
 			assert.Contains(t, result.Summary(), tt.wantReason)
+			assert.Equal(t, tt.wantString, result.String())
 		})
 	}
 }