Merge pull request #3752 from weaveworks/report-window

Set timestamp and window on each report

app/collector_test.go

@@ -50,8 +50,8 @@ func TestCollector(t *testing.T) {
 
 	c.Add(ctx, r2, nil)
 	merged := report.MakeReport()
-	merged = merged.Merge(r1)
-	merged = merged.Merge(r2)
+	merged.UnsafeMerge(r1)
+	merged.UnsafeMerge(r2)
 	have, err = c.Report(ctx, mtime.Now())
 	if err != nil {
 		t.Error(err)

probe/plugins/registry.go

@@ -222,7 +222,7 @@ func (r *Registry) Report() (report.Report, error) {
 		if plugin.Implements("controller") {
 			r.updateAndRegisterControlsInReport(&pluginReport)
 		}
-		rpt = rpt.Merge(pluginReport)
+		rpt.UnsafeMerge(pluginReport)
 	})
 	return rpt, nil
 }

probe/probe.go

@@ -7,6 +7,7 @@ import (
 
 	"github.com/armon/go-metrics"
 	log "github.com/sirupsen/logrus"
+	"github.com/weaveworks/common/mtime"
 	"golang.org/x/time/rate"
 
 	"github.com/weaveworks/scope/report"
@@ -185,6 +186,7 @@ func (p *Probe) report() report.Report {
 	}
 
 	result := report.MakeReport()
+	result.TS = mtime.Now()
 	for i := 0; i < cap(reports); i++ {
 		result.UnsafeMerge(<-reports)
 	}
@@ -234,6 +236,7 @@ ForLoop:
 
 func (p *Probe) publishLoop() {
 	defer p.done.Done()
+	startTime := mtime.Now()
 	pubTick := time.Tick(p.publishInterval)
 	publishCount := 0
 	var lastFullReport report.Report
@@ -247,6 +250,8 @@ func (p *Probe) publishLoop() {
 			if !fullReport {
 				rpt.UnsafeUnMerge(lastFullReport)
 			}
+			rpt.Window = mtime.Now().Sub(startTime)
+			startTime = mtime.Now()
 			err = p.publisher.Publish(rpt)
 			if err == nil {
 				if fullReport {

probe/probe_internal_test.go

@@ -69,6 +69,7 @@ func TestProbe(t *testing.T) {
 	node := report.MakeNodeWith("a", map[string]string{"b": "c"})
 
 	want.Endpoint.AddNode(node)
+	want.TS = now
 
 	pub := mockPublisher{make(chan report.Report, 10)}
 

render/theinternet_test.go

@@ -10,7 +10,7 @@ import (
 )
 
 func TestReportLocalNetworks(t *testing.T) {
-	r := report.MakeReport().Merge(report.Report{
+	r := report.Report{
 		Host: report.Topology{
 			Nodes: report.Nodes{
 				"nonets": report.MakeNode("nonets"),
@@ -27,7 +27,7 @@ func TestReportLocalNetworks(t *testing.T) {
 				),
 			},
 		},
-	})
+	}.Copy()
 	want := report.MakeNetworks()
 	for _, cidr := range []string{"10.0.0.1/8", "192.168.1.1/24", "10.32.0.1/12"} {
 		if err := want.AddCIDR(cidr); err != nil {

report/report.go

@@ -85,6 +85,9 @@ var topologyNames = []string{
 // stored by apps. It's composed of multiple topologies, each representing
 // a different (related, but not equivalent) view of the network.
 type Report struct {
+	// TS is the time this report was generated
+	TS time.Time
+
 	// Endpoint nodes are individual (address, port) tuples on each host.
 	// They come from inspecting active connections and can (theoretically)
 	// be traced back to a process. Edges are present.
@@ -321,6 +324,7 @@ func MakeReport() Report {
 // Copy returns a value copy of the report.
 func (r Report) Copy() Report {
 	newReport := Report{
+		TS:       r.TS,
 		DNS:      r.DNS.Copy(),
 		Sampling: r.Sampling,
 		Window:   r.Window,
@@ -334,16 +338,12 @@ func (r Report) Copy() Report {
 	return newReport
 }
 
-// Merge merges another Report into the receiver and returns the result. The
-// original is not modified.
-func (r Report) Merge(other Report) Report {
-	newReport := r.Copy()
-	newReport.UnsafeMerge(other)
-	return newReport
-}
-
 // UnsafeMerge merges another Report into the receiver. The original is modified.
 func (r *Report) UnsafeMerge(other Report) {
+	// Merged report has the earliest non-zero timestamp
+	if !other.TS.IsZero() && (r.TS.IsZero() || other.TS.Before(r.TS)) {
+		r.TS = other.TS
+	}
 	r.DNS = r.DNS.Merge(other.DNS)
 	r.Sampling = r.Sampling.Merge(other.Sampling)
 	r.Window = r.Window + other.Window