diff --git a/report/id_list.go b/report/id_list.go index 1c05b582a..9d89d7d08 100644 --- a/report/id_list.go +++ b/report/id_list.go @@ -24,3 +24,9 @@ func (a IDList) Add(ids ...string) IDList { } return a } + +// Contains returns true if id is in the list. +func (a IDList) Contains(id string) bool { + i := sort.Search(len(a), func(i int) bool { return a[i] >= id }) + return i < len(a) && a[i] == id +} diff --git a/report/report.go b/report/report.go index 460e19f4d..5cc5ebae5 100644 --- a/report/report.go +++ b/report/report.go @@ -122,3 +122,18 @@ func (r Report) LocalNetworks() []*net.IPNet { } return ipNets } + +// Topologies returns a slice of Topologies in this report +func (r Report) Topologies() []Topology { + return []Topology{r.Endpoint, r.Address, r.Process, r.Container, r.Host} +} + +// Validate checks the report for various inconsistencies. +func (r Report) Validate() error { + for _, topology := range r.Topologies() { + if err := topology.Validate(); err != nil { + return err + } + } + return nil +} diff --git a/report/topology.go b/report/topology.go index f9abed7c6..ed8a68ed8 100644 --- a/report/topology.go +++ b/report/topology.go @@ -1,6 +1,7 @@ package report import ( + "fmt" "log" "net" "reflect" @@ -266,3 +267,46 @@ type ByID []RenderableNode func (r ByID) Len() int { return len(r) } func (r ByID) Swap(i, j int) { r[i], r[j] = r[j], r[i] } func (r ByID) Less(i, j int) bool { return r[i].ID < r[j].ID } + +// Validate checks the topology for various inconsistencies. +func (t Topology) Validate() error { + // Check all edge metadata keys must have the appropriate entries in adjacencies & node metadata + for edgeID := range t.EdgeMetadatas { + srcNodeID, dstNodeID, ok := ParseEdgeID(edgeID) + if !ok { + return fmt.Errorf("Invalid edge id: %s", edgeID) + } + if _, ok := t.NodeMetadatas[srcNodeID]; !ok { + return fmt.Errorf("Source node missing for edge id: %s", edgeID) + } + + adjs, ok := t.Adjacency[MakeAdjacencyID(srcNodeID)] + if !ok { + return fmt.Errorf("Adjancey entries for missing for node id: %s (from edge %s)", srcNodeID, edgeID) + } + if !adjs.Contains(dstNodeID) { + return fmt.Errorf("Adjancey entry missing for edge id: %s", edgeID) + } + } + + // Check all adjancency keys has entries in NodeMetadata + for adjID := range t.Adjacency { + nodeID, ok := ParseAdjacencyID(adjID) + if !ok { + return fmt.Errorf("Invalid adjacency id: %s", adjID) + } + + if _, ok := t.NodeMetadatas[nodeID]; !ok { + return fmt.Errorf("Source node missing for adjancency id: %s", adjID) + } + } + + // Check all node metadata keys are parse-able (ie, contain a scope) + for nodeID := range t.NodeMetadatas { + if _, _, ok := ParseNodeID(nodeID); !ok { + return fmt.Errorf("Invalid node id: %s", nodeID) + } + } + + return nil +} diff --git a/xfer/collector.go b/xfer/collector.go index a7e49d879..464d8e8fe 100644 --- a/xfer/collector.go +++ b/xfer/collector.go @@ -102,6 +102,10 @@ func (c *realCollector) loop(batchTime time.Duration) { pc <- copy case r := <-c.in: + if err := r.Validate(); err != nil { + log.Printf("Received invalid report from: %v", err) + continue + } current.Merge(r) case ip := <-c.add: diff --git a/xfer/collector_test.go b/xfer/collector_test.go index 97f67f651..5d0d4889c 100644 --- a/xfer/collector_test.go +++ b/xfer/collector_test.go @@ -42,7 +42,7 @@ func TestCollector(t *testing.T) { runtime.Gosched() // make sure it connects // Push a report through everything - reports <- report.Report{Address: report.Topology{NodeMetadatas: report.NodeMetadatas{"a": report.NodeMetadata{}}}} + reports <- report.Report{Address: report.Topology{NodeMetadatas: report.NodeMetadatas{report.MakeAddressNodeID("a", "b"): report.NodeMetadata{}}}} poll(t, 10*time.Millisecond, func() bool { return len(concreteCollector.peek().Address.NodeMetadatas) == 1 }, "missed the report") go func() { publish <- time.Now() }() if want, have := 1, len((<-collector.Reports()).Address.NodeMetadatas); want != have { diff --git a/xfer/merge_test.go b/xfer/merge_test.go index 2bfe4e603..53ca89ec7 100644 --- a/xfer/merge_test.go +++ b/xfer/merge_test.go @@ -37,14 +37,16 @@ func TestMerge(t *testing.T) { defer c.Stop() time.Sleep(batchTime / 10) // connect + k1, k2 := report.MakeHostNodeID("p1"), report.MakeHostNodeID("p2") + { r := report.MakeReport() - r.Host.NodeMetadatas["p1"] = report.NodeMetadata{"host_name": "test1"} + r.Host.NodeMetadatas[k1] = report.NodeMetadata{"host_name": "test1"} p1.Publish(r) } { r := report.MakeReport() - r.Host.NodeMetadatas["p2"] = report.NodeMetadata{"host_name": "test2"} + r.Host.NodeMetadatas[k2] = report.NodeMetadata{"host_name": "test2"} p2.Publish(r) } @@ -52,10 +54,10 @@ func TestMerge(t *testing.T) { go func() { defer close(success) for r := range c.Reports() { - if r.Host.NodeMetadatas["p1"]["host_name"] != "test1" { + if r.Host.NodeMetadatas[k1]["host_name"] != "test1" { continue } - if r.Host.NodeMetadatas["p2"]["host_name"] != "test2" { + if r.Host.NodeMetadatas[k2]["host_name"] != "test2" { continue } return