diff --git a/services/nvpair-node-scanner/daemon.go b/services/nvpair-node-scanner/daemon.go index 117cf76d..68965d15 100644 --- a/services/nvpair-node-scanner/daemon.go +++ b/services/nvpair-node-scanner/daemon.go @@ -14,6 +14,7 @@ import ( "net/http" "os" "path/filepath" + "reflect" "strconv" "strings" "sync" @@ -914,6 +915,12 @@ func (d *daemon) publishSelf() { // publishSelfLocked is publishSelf's body; the caller must hold selfMu (so a // registration burst and a concurrent reloadIdentity can't tear the self entry). func (d *daemon) publishSelfLocked() { + d.publishSelfLockedIf(nil) +} + +// publishSelfLockedIf is publishSelfLocked that, when shouldEmit is set, emits +// only if shouldEmit accepts the stored entry. The caller holds selfMu. +func (d *daemon) publishSelfLockedIf(shouldEmit func(noderec.DirectoryNode) bool) { rec := d.reg.record() if rec.HostUUID == "" { return // no identity resolved yet — nothing to anchor self on @@ -955,7 +962,45 @@ func (d *daemon) publishSelfLocked() { if isNew := d.dir.upsert(node); isNew { method = noderec.NotifyNodeDiscovered } - d.emit(method, d.storedOr(node)) + stored := d.storedOr(node) + if shouldEmit != nil && method == noderec.NotifyNodeUpdated && !shouldEmit(stored) { + return + } + d.emit(method, stored) +} + +// refreshSelfOnce re-enriches this node's own entry on the peer-refresh tick. +// +// Browse events re-enrich every peer, but deliberately never self, and +// publishSelf ran only at startup, on a service (un)registration and after an +// identity or address change. The local node's GPU memory, CPU and memory +// figures in the discovery snapshot — the one the scheduler reads — therefore +// stayed at their startup values while every peer's converged. The live +// telemetry loop refreshes only utilization. +// +// It emits only when the entry changed beyond its LastSeen stamp, so an idle +// node adds no notification per tick. +func (d *daemon) refreshSelfOnce() { + d.selfMu.Lock() + defer d.selfMu.Unlock() + hostUUID := d.reg.record().HostUUID + if hostUUID == "" { + return + } + before, existed := d.dir.get(hostUUID) + if !existed { + d.publishSelfLocked() + return + } + d.publishSelfLockedIf(func(after noderec.DirectoryNode) bool { + return !sameIgnoringLastSeen(before, after) + }) +} + +// sameIgnoringLastSeen reports whether two entries differ only in LastSeen. +func sameIgnoringLastSeen(a, b noderec.DirectoryNode) bool { + a.LastSeen, b.LastSeen = 0, 0 + return reflect.DeepEqual(a, b) } // watchAddress re-derives the advertised ip= whenever the host's network changes @@ -1631,6 +1676,7 @@ func (d *daemon) refreshPeersLoop(ctx context.Context) { } d.refreshClusterIdentityOnce() d.refreshModelsOnce() + d.refreshSelfOnce() } } } diff --git a/services/nvpair-node-scanner/self_refresh_test.go b/services/nvpair-node-scanner/self_refresh_test.go new file mode 100644 index 00000000..b19fcd63 --- /dev/null +++ b/services/nvpair-node-scanner/self_refresh_test.go @@ -0,0 +1,58 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +package main + +import ( + "bytes" + "encoding/json" + "net" + "net/http" + "net/http/httptest" + "strconv" + "strings" + "sync/atomic" + "testing" + + "nvpair-shared/noderec" +) + +// The local node's discovery entry follows its own node-info on the refresh +// tick, as every peer's does, and an unchanged entry emits nothing. +func TestRefreshSelfFollowsTheLocalInventory(t *testing.T) { + var used atomic.Uint64 + used.Store(126877696) + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + v := used.Load() + _ = json.NewEncoder(w).Encode(NodeInfoResponse{GPUs: []GPUInfo{{Name: "GPU", VRAMUsedBytes: &v}}}) + })) + defer srv.Close() + _, portStr, err := net.SplitHostPort(srv.Listener.Addr().String()) + if err != nil { + t.Fatal(err) + } + port, _ := strconv.Atoi(portStr) + + var out bytes.Buffer + d := newSelfTestDaemon("self-uuid", "192.168.1.17") + d.codec = NewCodec(&out) + d.reg.register(noderec.RegisterParams{Service: noderec.ServiceNodeInfo, Port: port}) + d.publishSelf() + updates := func() int { return strings.Count(out.String(), noderec.NotifyNodeUpdated) } + before := updates() + + d.refreshSelfOnce() + if got := updates(); got != before { + t.Fatalf("an unchanged self entry emitted %d update(s)", got-before) + } + + used.Store(9782165504) + d.refreshSelfOnce() + n, _ := d.dir.get("self-uuid") + if len(n.GPUs) != 1 || n.GPUs[0].VRAMUsedBytes == nil || *n.GPUs[0].VRAMUsedBytes != 9782165504 { + t.Fatalf("self GPUs = %+v, want the refreshed VRAM reading", n.GPUs) + } + if got := updates(); got != before+1 { + t.Fatalf("a changed self entry emitted %d update(s), want 1", got-before) + } +}