diff --git a/app/api_topologies.go b/app/api_topologies.go index 595241b17..c1e1d100d 100644 --- a/app/api_topologies.go +++ b/app/api_topologies.go @@ -479,6 +479,7 @@ func (r *Registry) makeTopologyList(rep Reporter) CtxHandlerFunc { respondWith(ctx, w, http.StatusInternalServerError, err) return } + report.UnsafeRemovePartMergedNodes(ctx) respondWith(ctx, w, http.StatusOK, r.renderTopologies(ctx, report, req)) } } @@ -579,6 +580,7 @@ func (r *Registry) captureRenderer(rep Reporter, f rendererHandler) CtxHandlerFu respondWith(ctx, w, http.StatusInternalServerError, err) return } + rpt.UnsafeRemovePartMergedNodes(ctx) req.ParseForm() renderer, filter, err := r.RendererForTopology(topologyID, req.Form, rpt) if err != nil { diff --git a/app/api_topology.go b/app/api_topology.go index b6a941ce2..255976216 100644 --- a/app/api_topology.go +++ b/app/api_topology.go @@ -182,6 +182,7 @@ func (wc *websocketState) update(ctx context.Context) error { if err != nil { return errors.Wrap(err, "Error generating report") } + re.UnsafeRemovePartMergedNodes(ctx) renderer, filter, err := topologyRegistry.RendererForTopology(wc.topologyID, wc.values, re) if err != nil { return errors.Wrap(err, "Error generating report") diff --git a/app/benchmark_internal_test.go b/app/benchmark_internal_test.go index 499ce1ed8..da87079d5 100644 --- a/app/benchmark_internal_test.go +++ b/app/benchmark_internal_test.go @@ -87,6 +87,7 @@ func getReport(b *testing.B) report.Report { func benchmarkRender(b *testing.B, f func(report.Report)) { r := getReport(b) + r.UnsafeRemovePartMergedNodes(context.Background()) b.ResetTimer() for i := 0; i < b.N; i++ { b.StopTimer() diff --git a/app/collector_test.go b/app/collector_test.go index 22af88167..f33d4c21e 100644 --- a/app/collector_test.go +++ b/app/collector_test.go @@ -23,10 +23,10 @@ func TestCollector(t *testing.T) { defer mtime.NowReset() r1 := report.MakeReport() - r1.Endpoint.AddNode(report.MakeNode("foo")) + r1.Endpoint.AddNode(report.MakeNode("foo").WithTopology("bar")) r2 := report.MakeReport() - r2.Endpoint.AddNode(report.MakeNode("foo")) + r2.Endpoint.AddNode(report.MakeNode("foo").WithTopology("bar")) have, err := c.Report(ctx, mtime.Now()) if err != nil { @@ -52,6 +52,7 @@ func TestCollector(t *testing.T) { merged := report.MakeReport() merged.UnsafeMerge(r1) merged.UnsafeMerge(r2) + merged.UnsafeRemovePartMergedNodes(context.Background()) have, err = c.Report(ctx, mtime.Now()) if err != nil { t.Error(err) diff --git a/report/node.go b/report/node.go index 859398be0..10f523dfd 100644 --- a/report/node.go +++ b/report/node.go @@ -236,3 +236,10 @@ func (n *Node) UnsafeUnMerge(other Node) bool { // metrics don't overlap so just check if we have any return remove && len(n.Metrics) == 0 } + +// If a node is removed from source between two full reports, then we +// might only have a delta of its last state. Detect that from a blank topology, +// which cannot arise in a properly-formed Node. +func (n *Node) isPartMerged() bool { + return n.Topology == "" +} diff --git a/report/report.go b/report/report.go index 1534c6803..611f84e98 100644 --- a/report/report.go +++ b/report/report.go @@ -1,11 +1,14 @@ package report import ( + "context" "fmt" "math/rand" "strings" "time" + opentracing "github.com/opentracing/opentracing-go" + "github.com/weaveworks/scope/common/xfer" ) @@ -363,6 +366,29 @@ func (r *Report) UnsafeUnMerge(other Report) { }) } +// UnsafeRemovePartMergedNodes removes nodes that have not fully re-merged. +// E.g. if a node is removed from source between two full reports, then we +// might only have a delta of its last state. Remove that from the set. +// The original is modified. +func (r *Report) UnsafeRemovePartMergedNodes(ctx context.Context) { + dropped := map[string]int{} + r.WalkNamedTopologies(func(name string, t *Topology) { + for k, v := range t.Nodes { + if v.isPartMerged() { + delete(t.Nodes, k) + dropped[name]++ + } + } + }) + if span := opentracing.SpanFromContext(ctx); span != nil && len(dropped) > 0 { + msg := "" + for name, count := range dropped { + msg += fmt.Sprintf("%s: %d, ", name, count) + } + span.LogKV("dropped-part-merged", msg) + } +} + // WalkTopologies iterates through the Topologies of the report, // potentially modifying them func (r *Report) WalkTopologies(f func(*Topology)) {