github/weave-scope
1
0
Fork 0
mirror of https://github.com/weaveworks/scope.git synced 2026-08-23 22:36:24 +00:00
Code Issues Packages Projects Releases Wiki Activity

gopacket-based traffic sniffing

Browse Source
This commit is contained in:
Peter Bourgon
2015-08-03 10:55:59 +02:00
parent 8fdbc44680
commit b9afa67ad6
30 changed files with 951 additions and 438 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Makefile
+1 -1
View File
@@ -26,7 +26,7 @@ $(APP_EXE): app/*.go render/*.go report/*.go xfer/*.go
$(PROBE_EXE): probe/*.go probe/docker/*.go probe/endpoint/*.go probe/host/*.go probe/process/*.go probe/overlay/*.go report/*.go xfer/*.go
$(APP_EXE) $(PROBE_EXE):
go get -tags netgo ./$(@D)
go get -d -tags netgo ./$(@D)
go build -ldflags "-extldflags \"-static\" -X main.version $(SCOPE_VERSION)" -tags netgo -o $@ ./$(@D)
@strings $@ | grep cgo_stub\\\.go >/dev/null || { \
rm $@; \
app/api_topology.go
+3 -2
View File
@@ -8,6 +8,7 @@ import (
"github.com/gorilla/websocket"
"github.com/weaveworks/scope/render"
"github.com/weaveworks/scope/report"
)
const (
@@ -27,7 +28,7 @@ type APINode struct {
// APIEdge is returned by the /api/topology/*/*/* handlers.
type APIEdge struct {
Metadata render.AggregateMetadata `json:"metadata"`
Metadata report.EdgeMetadata `json:"metadata"`
}
// Full topology.
@@ -76,7 +77,7 @@ func handleEdge(rep Reporter, t topologyView, w http.ResponseWriter, r *http.Req
localID = vars["local"]
remoteID = vars["remote"]
rpt = rep.Report()
metadata = t.renderer.AggregateMetadata(rpt, localID, remoteID)
metadata = t.renderer.EdgeMetadata(rpt, localID, remoteID)
)
respondWith(w, http.StatusOK, APIEdge{Metadata: metadata})
app/api_topology_test.go
+13 -9
View File
@@ -12,6 +12,7 @@ import (
"github.com/weaveworks/scope/render"
"github.com/weaveworks/scope/render/expected"
"github.com/weaveworks/scope/report"
"github.com/weaveworks/scope/test"
)
@@ -71,6 +72,7 @@ func TestAPITopologyApplications(t *testing.T) {
if err := json.Unmarshal(body, &topo); err != nil {
t.Fatal(err)
}
if want, have := render.OnlyConnected(expected.RenderedProcesses), fixNodeMetadatas(topo.Nodes); !reflect.DeepEqual(want, have) {
t.Error(test.Diff(want, have))
}
@@ -93,9 +95,9 @@ func TestAPITopologyApplications(t *testing.T) {
if err := json.Unmarshal(body, &edge); err != nil {
t.Fatalf("JSON parse error: %s", err)
}
if want, have := (render.AggregateMetadata{
"egress_bytes": 10,
"ingress_bytes": 100,
if want, have := (report.EdgeMetadata{
PacketCount: newu64(100),
ByteCount: newu64(10),
}), edge.Metadata; !reflect.DeepEqual(want, have) {
t.Error(test.Diff(want, have))
}
@@ -112,7 +114,8 @@ func TestAPITopologyHosts(t *testing.T) {
if err := json.Unmarshal(body, &topo); err != nil {
t.Fatal(err)
}
if want, have := (expected.RenderedHosts), fixNodeMetadatas(topo.Nodes); !reflect.DeepEqual(want, have) {
if want, have := expected.RenderedHosts, fixNodeMetadatas(topo.Nodes); !reflect.DeepEqual(want, have) {
t.Error(test.Diff(want, have))
}
}
@@ -134,11 +137,10 @@ func TestAPITopologyHosts(t *testing.T) {
if err := json.Unmarshal(body, &edge); err != nil {
t.Fatalf("JSON parse error: %s", err)
}
want := render.AggregateMetadata{
"max_conn_count_tcp": 3,
}
if !reflect.DeepEqual(want, edge.Metadata) {
t.Errorf("Edge metadata error. Want %v, have %v", want, edge)
if want, have := (report.EdgeMetadata{
MaxConnCountTCP: newu64(3),
}), edge.Metadata; !reflect.DeepEqual(want, have) {
t.Error(test.Diff(want, have))
}
}
}
@@ -176,3 +178,5 @@ func TestAPITopologyWebsocket(t *testing.T) {
equals(t, 0, len(d.Update))
equals(t, 0, len(d.Remove))
}
func newu64(value uint64) *uint64 { return &value }
circle.yml
+2 -3
View File
@@ -19,7 +19,7 @@ dependencies:
cache_directories:
- "~/docker"
override:
- sudo apt-get --only-upgrade install tar
- sudo apt-get --only-upgrade install tar libpcap0.8-dev
- bin/rebuild-ui-build-image
- curl https://sdk.cloud.google.com | bash
- test -z "$SECRET_PASSWORD" || bin/setup-circleci-secrets "$SECRET_PASSWORD"
@@ -40,8 +40,7 @@ test:
- cd $SRCDIR; ./bin/lint .
- cd $SRCDIR; make client-test
- cd $SRCDIR; make static
- cd $SRCDIR; rm -f app/app probe/probe; GOOS=darwin make && make clean
- cd $SRCDIR; rm -f app/app probe/probe; GOOS=linux make
- cd $SRCDIR; rm -f app/scope-app probe/scope-probe; make
- cd $SRCDIR; ./bin/test -slow
- cd $SRCDIR/experimental; make
- test -z "$SECRET_PASSWORD" || (cd $SRCDIR/integration; ./gce.sh setup)
experimental/demoprobe/generate.go
+4 -4
View File
@@ -91,12 +91,10 @@ func DemoReport(nodeCount int) report.Report {
edgeKeyIngress = report.MakeEdgeID(dstPortID, srcPortID)
)
r.Endpoint.EdgeMetadatas[edgeKeyEgress] = report.EdgeMetadata{
WithConnCountTCP: true,
MaxConnCountTCP: uint(rand.Intn(100) + 10),
MaxConnCountTCP: newu64(uint64(rand.Intn(100) + 10)),
}
r.Endpoint.EdgeMetadatas[edgeKeyIngress] = report.EdgeMetadata{
WithConnCountTCP: true,
MaxConnCountTCP: uint(rand.Intn(100) + 10),
MaxConnCountTCP: newu64(uint64(rand.Intn(100) + 10)),
}
// Address topology
@@ -124,3 +122,5 @@ func DemoReport(nodeCount int) report.Report {
return r
}
func newu64(value uint64) *uint64 { return &value }
experimental/genreport/generate.go
+4 -4
View File
@@ -89,12 +89,10 @@ func DemoReport(nodeCount int) report.Report {
edgeKeyIngress = report.MakeEdgeID(dstPortID, srcPortID)
)
r.Endpoint.EdgeMetadatas[edgeKeyEgress] = report.EdgeMetadata{
WithConnCountTCP: true,
MaxConnCountTCP: uint(rand.Intn(100) + 10),
MaxConnCountTCP: newu64(uint64(rand.Intn(100) + 10)),
}
r.Endpoint.EdgeMetadatas[edgeKeyIngress] = report.EdgeMetadata{
WithConnCountTCP: true,
MaxConnCountTCP: uint(rand.Intn(100) + 10),
MaxConnCountTCP: newu64(uint64(rand.Intn(100) + 10)),
}
// Address topology
@@ -122,3 +120,5 @@ func DemoReport(nodeCount int) report.Report {
return r
}
func newu64(value uint64) *uint64 { return &value }
probe/docker/container_test.go → probe/docker/container_linux_test.go
View File
probe/endpoint/reporter.go
+12 -10
View File
@@ -79,15 +79,15 @@ func (r *Reporter) addConnection(rpt *report.Report, c *procspy.Connection) {
var (
localAddressNodeID = report.MakeAddressNodeID(r.hostID, c.LocalAddress.String())
remoteAddressNodeID = report.MakeAddressNodeID(r.hostID, c.RemoteAddress.String())
adjacencyID = report.MakeAdjacencyID(localAddressNodeID)
adjecencyID = report.MakeAdjacencyID(localAddressNodeID)
edgeID = report.MakeEdgeID(localAddressNodeID, remoteAddressNodeID)
)
rpt.Address.Adjacency[adjacencyID] = rpt.Address.Adjacency[adjacencyID].Add(remoteAddressNodeID)
rpt.Address.Adjacency[adjecencyID] = rpt.Address.Adjacency[adjecencyID].Add(remoteAddressNodeID)
if _, ok := rpt.Address.NodeMetadatas[localAddressNodeID]; !ok {
rpt.Address.NodeMetadatas[localAddressNodeID] = report.MakeNodeMetadataWith(map[string]string{
"name": r.hostName, // TODO this is ambiguous, be more specific
"name": r.hostName,
Addr: c.LocalAddress.String(),
})
}
@@ -98,11 +98,11 @@ func (r *Reporter) addConnection(rpt *report.Report, c *procspy.Connection) {
var (
localEndpointNodeID = report.MakeEndpointNodeID(r.hostID, c.LocalAddress.String(), strconv.Itoa(int(c.LocalPort)))
remoteEndpointNodeID = report.MakeEndpointNodeID(r.hostID, c.RemoteAddress.String(), strconv.Itoa(int(c.RemotePort)))
adjacencyID = report.MakeAdjacencyID(localEndpointNodeID)
adjecencyID = report.MakeAdjacencyID(localEndpointNodeID)
edgeID = report.MakeEdgeID(localEndpointNodeID, remoteEndpointNodeID)
)
rpt.Endpoint.Adjacency[adjacencyID] = rpt.Endpoint.Adjacency[adjacencyID].Add(remoteEndpointNodeID)
rpt.Endpoint.Adjacency[adjecencyID] = rpt.Endpoint.Adjacency[adjecencyID].Add(remoteEndpointNodeID)
if _, ok := rpt.Endpoint.NodeMetadatas[localEndpointNodeID]; !ok {
// First hit establishes NodeMetadata for scoped local address + port
@@ -119,9 +119,11 @@ func (r *Reporter) addConnection(rpt *report.Report, c *procspy.Connection) {
}
}
func countTCPConnection(m report.EdgeMetadatas, edgeKey string) {
edgeMeta := m[edgeKey]
edgeMeta.WithConnCountTCP = true
edgeMeta.MaxConnCountTCP++
m[edgeKey] = edgeMeta
func countTCPConnection(mds report.EdgeMetadatas, key string) {
md := mds[key]
if md.MaxConnCountTCP == nil {
md.MaxConnCountTCP = new(uint64)
}
*md.MaxConnCountTCP++
mds[key] = md
}
probe/main.go
+36
View File
@@ -3,11 +3,13 @@ package main
import (
"flag"
"log"
"net"
"net/http"
_ "net/http/pprof"
"os"
"os/signal"
"strconv"
"strings"
"syscall"
"time"
@@ -17,6 +19,7 @@ import (
"github.com/weaveworks/scope/probe/host"
"github.com/weaveworks/scope/probe/overlay"
"github.com/weaveworks/scope/probe/process"
"github.com/weaveworks/scope/probe/sniff"
"github.com/weaveworks/scope/report"
"github.com/weaveworks/scope/xfer"
)
@@ -36,8 +39,13 @@ func main() {
dockerBridge = flag.String("docker.bridge", "docker0", "the docker bridge name")
weaveRouterAddr = flag.String("weave.router.addr", "", "IP address or FQDN of the Weave router")
procRoot = flag.String("proc.root", "/proc", "location of the proc filesystem")
captureEnabled = flag.Bool("capture", false, "perform sampled packet capture")
captureInterfaces = flag.String("capture.interfaces", interfaces(), "packet capture on these interfaces")
captureOn = flag.Duration("capture.on", 1*time.Second, "packet capture duty cycle 'on'")
captureOff = flag.Duration("capture.off", 5*time.Second, "packet capture duty cycle 'off'")
)
flag.Parse()
log.SetFlags(log.Lmicroseconds)
if len(flag.Args()) != 0 {
flag.Usage()
@@ -106,6 +114,21 @@ func main() {
reporters = append(reporters, weave)
}
if *captureEnabled {
if *captureOn > *captureOff {
log.Fatalf("-capture.on (%s) must be <= -capture.off (%s)", *captureOn, *captureOff)
}
for _, iface := range strings.Split(*captureInterfaces, ",") {
source, err := sniff.NewSource(iface)
if err != nil {
log.Printf("warning: %v", err)
continue
}
log.Printf("capturing packets on %s", iface)
reporters = append(reporters, sniff.New(hostID, source, *captureOn, *captureOff))
}
}
log.Printf("listening on %s", *listen)
quit := make(chan struct{})
@@ -153,3 +176,16 @@ func interrupt() chan os.Signal {
signal.Notify(c, syscall.SIGINT, syscall.SIGTERM)
return c
}
func interfaces() string {
ifaces, err := net.Interfaces()
if err != nil {
log.Print(err)
return ""
}
a := make([]string, 0, len(ifaces))
for _, iface := range ifaces {
a = append(a, iface.Name)
}
return strings.Join(a, ",")
}
probe/process/walker_test.go
-1
View File
@@ -9,7 +9,6 @@ import (
)
func TestBasicWalk(t *testing.T) {
// Don't panic or error.
var (
procRoot = "/proc"
procFunc = func(process.Process) {}
probe/sniff/sniffer.go
+251
View File
@@ -0,0 +1,251 @@
package sniff
import (
"io"
"log"
"strconv"
"sync/atomic"
"time"
"github.com/weaveworks/scope/report"
"github.com/google/gopacket"
"github.com/google/gopacket/layers"
)
// Sniffer is a packet-sniffing reporter.
type Sniffer struct {
hostID string
reports chan chan report.Report
parser *gopacket.DecodingLayerParser
decoded []gopacket.LayerType
eth layers.Ethernet
ip4 layers.IPv4
ip6 layers.IPv6
tcp layers.TCP
udp layers.UDP
icmp4 layers.ICMPv4
icmp6 layers.ICMPv6
}
// New returns a new sniffing reporter that samples traffic by turning its
// packet capture facilities on and off. Note that the on and off durations
// represent a way to bound CPU burn. Effective sample rate needs to be
// calculated as (packets decoded / packets observed).
func New(hostID string, src gopacket.ZeroCopyPacketDataSource, on, off time.Duration) *Sniffer {
s := &Sniffer{
hostID: hostID,
reports: make(chan chan report.Report),
}
s.parser = gopacket.NewDecodingLayerParser(
layers.LayerTypeEthernet,
&s.eth, &s.ip4, &s.ip6, &s.tcp, &s.udp, &s.icmp4, &s.icmp6,
)
go s.loop(src, on, off)
return s
}
// Report implements the Reporter interface.
func (s *Sniffer) Report() (report.Report, error) {
c := make(chan report.Report)
s.reports <- c
return <-c, nil
}
func (s *Sniffer) loop(src gopacket.ZeroCopyPacketDataSource, on, off time.Duration) {
var (
process = uint64(1) // initially enabled
total = uint64(0) // total packets seen
count = uint64(0) // count of packets captured
packets = make(chan Packet, 1024) // decoded packets
rpt = report.MakeReport() // the report we build
turnOn = (<-chan time.Time)(nil) // signal to start capture (initially enabled)
turnOff = time.After(on) // signal to stop capture
done = make(chan struct{}) // when src is finished, we're done too
)
go func() {
s.read(src, packets, &process, &total, &count)
close(done)
}()
for {
select {
case p := <-packets:
s.Merge(p, rpt)
case <-turnOn:
atomic.StoreUint64(&process, 1) // enable packet capture
turnOn = nil // disable the on switch
turnOff = time.After(on) // enable the off switch
case <-turnOff:
atomic.StoreUint64(&process, 0) // disable packet capture
turnOn = time.After(off) // enable the on switch
turnOff = nil // disable the off switch
case c := <-s.reports:
rpt.Sampling.Count = atomic.LoadUint64(&count)
rpt.Sampling.Total = atomic.LoadUint64(&total)
interpolateCounts(rpt)
c <- rpt
atomic.StoreUint64(&count, 0)
atomic.StoreUint64(&total, 0)
rpt = report.MakeReport()
case <-done:
return
}
}
}
// interpolateCounts compensates for sampling by artifically inflating counts
// throughout the report. It should be run once for each report, within the
// probe, before it gets emitted into the rest of the system.
func interpolateCounts(r report.Report) {
rate := r.Sampling.Rate()
if rate >= 1.0 {
return
}
factor := 1.0 / rate
for _, topology := range r.Topologies() {
for _, emd := range topology.EdgeMetadatas {
if emd.PacketCount != nil {
*emd.PacketCount = uint64(float64(*emd.PacketCount) * factor)
}
if emd.ByteCount != nil {
*emd.ByteCount = uint64(float64(*emd.ByteCount) * factor)
}
}
}
}
// Packet is an intermediate, decoded form of a packet, with the information
// that the Scope data model cares about. Designed to decouple the packet data
// source loop, which should be as fast as possible, and the process of
// merging the packet information to a report, which may take some time and
// allocations.
type Packet struct {
SrcIP, DstIP string
SrcPort, DstPort string
Network, Transport int // byte counts
}
func (s *Sniffer) read(src gopacket.ZeroCopyPacketDataSource, dst chan Packet, process, total, count *uint64) {
var (
p Packet
data []byte
err error
)
for {
data, _, err = src.ZeroCopyReadPacketData()
if err == io.EOF {
return // done
}
if err != nil {
log.Printf("sniffer: read: %v", err)
continue
}
atomic.AddUint64(total, 1)
if atomic.LoadUint64(process) == 0 {
continue
}
if err := s.parser.DecodeLayers(data, &s.decoded); err != nil {
// We'll always get an error when we encounter a layer type for
// which we haven't configured a decoder.
}
for _, t := range s.decoded {
switch t {
case layers.LayerTypeEthernet:
//
case layers.LayerTypeICMPv4:
p.Network += len(s.icmp4.Payload)
case layers.LayerTypeICMPv6:
p.Network += len(s.icmp6.Payload)
case layers.LayerTypeIPv6:
p.SrcIP = s.ip6.SrcIP.String()
p.DstIP = s.ip6.DstIP.String()
p.Network += len(s.ip6.Payload)
case layers.LayerTypeIPv4:
p.SrcIP = s.ip4.SrcIP.String()
p.DstIP = s.ip4.DstIP.String()
p.Network += len(s.ip4.Payload)
case layers.LayerTypeTCP:
p.SrcPort = strconv.Itoa(int(s.tcp.SrcPort))
p.DstPort = strconv.Itoa(int(s.tcp.DstPort))
p.Transport += len(s.tcp.Payload)
case layers.LayerTypeUDP:
p.SrcPort = strconv.Itoa(int(s.udp.SrcPort))
p.DstPort = strconv.Itoa(int(s.udp.DstPort))
p.Transport += len(s.udp.Payload)
}
}
select {
case dst <- p:
atomic.AddUint64(count, 1)
default:
log.Printf("sniffer dropped packet")
}
}
}
// Merge puts the packet into the report.
func (s *Sniffer) Merge(p Packet, rpt report.Report) {
// With a src and dst IP, we can add to the address topology.
if p.SrcIP != "" && p.DstIP != "" {
var (
srcNodeID = report.MakeAddressNodeID(s.hostID, p.SrcIP)
dstNodeID = report.MakeAddressNodeID(s.hostID, p.DstIP)
edgeID = report.MakeEdgeID(srcNodeID, dstNodeID)
srcAdjacencyID = report.MakeAdjacencyID(srcNodeID)
)
rpt.Address.NodeMetadatas[srcNodeID] = report.MakeNodeMetadata()
rpt.Address.NodeMetadatas[dstNodeID] = report.MakeNodeMetadata()
emd := rpt.Address.EdgeMetadatas[edgeID]
if emd.PacketCount == nil {
emd.PacketCount = new(uint64)
}
*emd.PacketCount++
if emd.ByteCount == nil {
emd.ByteCount = new(uint64)
}
*emd.ByteCount += uint64(p.Network)
rpt.Address.EdgeMetadatas[edgeID] = emd
rpt.Address.Adjacency[srcAdjacencyID] = rpt.Address.Adjacency[srcAdjacencyID].Add(dstNodeID)
}
// With a src and dst IP and port, we can add to the endpoints.
if p.SrcIP != "" && p.DstIP != "" && p.SrcPort != "" && p.DstPort != "" {
var (
srcNodeID = report.MakeEndpointNodeID(s.hostID, p.SrcIP, p.SrcPort)
dstNodeID = report.MakeEndpointNodeID(s.hostID, p.DstIP, p.DstPort)
edgeID = report.MakeEdgeID(srcNodeID, dstNodeID)
srcAdjacencyID = report.MakeAdjacencyID(srcNodeID)
)
rpt.Endpoint.NodeMetadatas[srcNodeID] = report.MakeNodeMetadata()
rpt.Endpoint.NodeMetadatas[dstNodeID] = report.MakeNodeMetadata()
emd := rpt.Endpoint.EdgeMetadatas[edgeID]
if emd.PacketCount == nil {
emd.PacketCount = new(uint64)
}
*emd.PacketCount++
if emd.ByteCount == nil {
emd.ByteCount = new(uint64)
}
*emd.ByteCount += uint64(p.Transport)
rpt.Endpoint.EdgeMetadatas[edgeID] = emd
rpt.Endpoint.Adjacency[srcAdjacencyID] = rpt.Endpoint.Adjacency[srcAdjacencyID].Add(dstNodeID)
}
}
probe/sniff/sniffer_internal_test.go
+45
View File
@@ -0,0 +1,45 @@
package sniff
import (
"testing"
"github.com/weaveworks/scope/report"
)
func TestInterpolateCounts(t *testing.T) {
var (
hostID = "macbook-air"
srcNodeID = report.MakeEndpointNodeID(hostID, "1.2.3.4", "5678")
dstNodeID = report.MakeEndpointNodeID(hostID, "5.6.7.8", "9012")
edgeID = report.MakeEdgeID(srcNodeID, dstNodeID)
samplingCount = uint64(200)
samplingTotal = uint64(2345)
packetCount = uint64(123)
byteCount = uint64(4096)
)
r := report.MakeReport()
r.Sampling.Count = samplingCount
r.Sampling.Total = samplingTotal
r.Endpoint.EdgeMetadatas[edgeID] = report.EdgeMetadata{
PacketCount: newu64(packetCount),
ByteCount: newu64(byteCount),
}
interpolateCounts(r)
var (
rate = float64(samplingCount) / float64(samplingTotal)
factor = 1.0 / rate
apply = func(v uint64) uint64 { return uint64(factor * float64(v)) }
emd = r.Endpoint.EdgeMetadatas[edgeID]
)
if want, have := apply(packetCount), (*emd.PacketCount); want != have {
t.Errorf("want %d packets, have %d", want, have)
}
if want, have := apply(byteCount), (*emd.ByteCount); want != have {
t.Errorf("want %d bytes, have %d", want, have)
}
}
func newu64(value uint64) *uint64 { return &value }
probe/sniff/sniffer_test.go
+137
View File
@@ -0,0 +1,137 @@
package sniff_test
import (
"io"
"reflect"
"sync"
"testing"
"time"
"github.com/google/gopacket"
"github.com/weaveworks/scope/probe/sniff"
"github.com/weaveworks/scope/report"
"github.com/weaveworks/scope/test"
)
func TestSnifferShutdown(t *testing.T) {
var (
hostID = "abcd"
src = newMockSource([]byte{}, nil)
on = time.Millisecond
off = time.Millisecond
s = sniff.New(hostID, src, on, off)
)
// Stopping the source should terminate the sniffer.
src.Close()
time.Sleep(10 * time.Millisecond)
// Try to get a report from the sniffer. It should block forever, as the
// loop goroutine should have exited.
report := make(chan struct{})
go func() { _, _ = s.Report(); close(report) }()
select {
case <-time.After(time.Millisecond):
case <-report:
t.Errorf("shouldn't get report after Close")
}
}
func TestMerge(t *testing.T) {
var (
hostID = "xyz"
src = newMockSource([]byte{}, nil)
on = time.Millisecond
off = time.Millisecond
rpt = report.MakeReport()
p = sniff.Packet{
SrcIP: "1.0.0.0",
SrcPort: "1000",
DstIP: "2.0.0.0",
DstPort: "2000",
Network: 512,
Transport: 256,
}
)
sniff.New(hostID, src, on, off).Merge(p, rpt)
var (
srcEndpointNodeID = report.MakeEndpointNodeID(hostID, p.SrcIP, p.SrcPort)
dstEndpointNodeID = report.MakeEndpointNodeID(hostID, p.DstIP, p.DstPort)
)
if want, have := (report.Topology{
Adjacency: report.Adjacency{
report.MakeAdjacencyID(srcEndpointNodeID): report.MakeIDList(
dstEndpointNodeID,
),
},
EdgeMetadatas: report.EdgeMetadatas{
report.MakeEdgeID(srcEndpointNodeID, dstEndpointNodeID): report.EdgeMetadata{
PacketCount: newu64(1),
ByteCount: newu64(256),
},
},
NodeMetadatas: report.NodeMetadatas{
srcEndpointNodeID: report.MakeNodeMetadata(),
dstEndpointNodeID: report.MakeNodeMetadata(),
},
}), rpt.Endpoint; !reflect.DeepEqual(want, have) {
t.Errorf("%s", test.Diff(want, have))
}
var (
srcAddressNodeID = report.MakeAddressNodeID(hostID, p.SrcIP)
dstAddressNodeID = report.MakeAddressNodeID(hostID, p.DstIP)
)
if want, have := (report.Topology{
Adjacency: report.Adjacency{
report.MakeAdjacencyID(srcAddressNodeID): report.MakeIDList(
dstAddressNodeID,
),
},
EdgeMetadatas: report.EdgeMetadatas{
report.MakeEdgeID(srcAddressNodeID, dstAddressNodeID): report.EdgeMetadata{
PacketCount: newu64(1),
ByteCount: newu64(512),
},
},
NodeMetadatas: report.NodeMetadatas{
srcAddressNodeID: report.MakeNodeMetadata(),
dstAddressNodeID: report.MakeNodeMetadata(),
},
}), rpt.Address; !reflect.DeepEqual(want, have) {
t.Errorf("%s", test.Diff(want, have))
}
}
type mockSource struct {
mtx sync.RWMutex
data []byte
err error
}
func newMockSource(data []byte, err error) *mockSource {
return &mockSource{
data: data,
err: err,
}
}
func (s *mockSource) ZeroCopyReadPacketData() ([]byte, gopacket.CaptureInfo, error) {
s.mtx.RLock()
defer s.mtx.RUnlock()
return s.data, gopacket.CaptureInfo{
Timestamp: time.Now(),
CaptureLength: len(s.data),
Length: len(s.data),
}, s.err
}
func (s *mockSource) Close() {
s.mtx.Lock()
defer s.mtx.Unlock()
s.err = io.EOF
}
func newu64(value uint64) *uint64 { return &value }
probe/sniff/source.go
+24
View File
@@ -0,0 +1,24 @@
package sniff
import (
"github.com/google/gopacket"
"github.com/google/gopacket/pcap"
)
// Source describes a packet data source that can be terminated.
type Source interface {
gopacket.ZeroCopyPacketDataSource
Close()
}
const (
snaplen = 65535
promisc = true
timeout = pcap.BlockForever
)
// NewSource returns a live packet data source via the passed device
// (interface).
func NewSource(device string) (Source, error) {
return pcap.OpenLive(device, snaplen, promisc, timeout)
}
probe/sniff/testclient/main.go
+87
View File
@@ -0,0 +1,87 @@
package main
import (
"flag"
"io"
"log"
"time"
"github.com/google/gopacket"
"github.com/google/gopacket/layers"
"github.com/google/gopacket/pcap"
)
func main() {
var (
device = flag.String("device", "eth0", "device to sniff")
)
flag.Parse()
const (
snaplen = 1024 * 1024
promisc = true
timeout = pcap.BlockForever
)
handle, err := pcap.OpenLive(*device, snaplen, promisc, timeout)
if err != nil {
log.Fatal(err)
}
go func() {
time.Sleep(5 * time.Second)
handle.Close()
}()
var (
eth layers.Ethernet
ip4 layers.IPv4
ip6 layers.IPv6
tcp layers.TCP
udp layers.UDP
icmp4 layers.ICMPv4
icmp6 layers.ICMPv6
)
parser := gopacket.NewDecodingLayerParser(
layers.LayerTypeEthernet,
&eth, &ip4, &ip6, &tcp, &udp, &icmp4, &icmp6,
)
decoded := []gopacket.LayerType{}
done := make(chan struct{})
go func() {
defer close(done)
for {
data, ci, err := handle.ZeroCopyReadPacketData()
if err == io.EOF {
log.Print("read: EOF")
return
}
if err != nil {
log.Printf("read: %v", err)
continue
}
log.Println(ci.Timestamp.String())
err = parser.DecodeLayers(data, &decoded)
for _, t := range decoded {
switch t {
case layers.LayerTypeEthernet:
log.Println(" Ethernet", eth.EthernetType, eth.SrcMAC, eth.DstMAC, eth.Length)
case layers.LayerTypeIPv6:
log.Println(" IP6", ip6.Version, ip6.SrcIP, ip6.DstIP, ip6.Length, ip6.TrafficClass)
case layers.LayerTypeIPv4:
log.Println(" IP4", ip4.Version, ip4.SrcIP, ip4.DstIP, ip4.Length, ip4.TTL, ip4.TOS)
case layers.LayerTypeTCP:
log.Println(" TCP", tcp.SrcPort, tcp.DstPort, tcp.Seq, tcp.Ack, tcp.Window)
case layers.LayerTypeUDP:
log.Println(" UDP", udp.SrcPort, udp.DstPort, udp.Length)
case layers.LayerTypeICMPv4:
log.Println(" ICMP4", icmp4.Id, icmp4.Seq)
case layers.LayerTypeICMPv6:
log.Println(" ICMP6")
}
}
log.Println()
}
}()
<-done
}
render/detailed_node.go
+6 -6
View File
@@ -61,14 +61,14 @@ func MakeDetailedNode(r report.Report, n RenderableNode) DetailedNode {
tables := tables{}
{
rows := []Row{}
if val, ok := n.AggregateMetadata[KeyMaxConnCountTCP]; ok {
rows = append(rows, Row{"TCP connections", strconv.FormatInt(int64(val), 10), ""})
if n.EdgeMetadata.MaxConnCountTCP != nil {
rows = append(rows, Row{"TCP connections", strconv.FormatUint(*n.EdgeMetadata.MaxConnCountTCP, 10), ""})
}
if val, ok := n.AggregateMetadata[KeyBytesIngress]; ok {
rows = append(rows, Row{"Bytes ingress", strconv.FormatInt(int64(val), 10), ""})
if n.EdgeMetadata.PacketCount != nil {
rows = append(rows, Row{"Packets", strconv.FormatUint(*n.EdgeMetadata.PacketCount, 10), ""})
}
if val, ok := n.AggregateMetadata[KeyBytesEgress]; ok {
rows = append(rows, Row{"Bytes egress", strconv.FormatInt(int64(val), 10), ""})
if n.EdgeMetadata.ByteCount != nil {
rows = append(rows, Row{"Bytes", strconv.FormatUint(*n.EdgeMetadata.ByteCount, 10), ""})
}
if len(rows) > 0 {
tables = append(tables, Table{"Connections", true, connectionsRank, rows})
render/detailed_node_test.go
+2 -2
View File
@@ -66,8 +66,8 @@ func TestMakeDetailedNode(t *testing.T) {
Numeric: true,
Rank: 100,
Rows: []render.Row{
{"Bytes ingress", "150", ""},
{"Bytes egress", "1500", ""},
{"Packets", "150", ""},
{"Bytes", "1500", ""},
},
},
{
render/expected/expected.go
+66 -64
View File
@@ -14,25 +14,25 @@ var (
unknownPseudoNode1ID = render.MakePseudoNodeID("10.10.10.10", test.ServerIP, "80")
unknownPseudoNode2ID = render.MakePseudoNodeID("10.10.10.11", test.ServerIP, "80")
unknownPseudoNode1 = render.RenderableNode{
ID: unknownPseudoNode1ID,
LabelMajor: "10.10.10.10",
Pseudo: true,
NodeMetadata: report.MakeNodeMetadata(),
AggregateMetadata: render.AggregateMetadata{},
ID: unknownPseudoNode1ID,
LabelMajor: "10.10.10.10",
Pseudo: true,
NodeMetadata: report.MakeNodeMetadata(),
EdgeMetadata: report.EdgeMetadata{},
}
unknownPseudoNode2 = render.RenderableNode{
ID: unknownPseudoNode2ID,
LabelMajor: "10.10.10.11",
Pseudo: true,
NodeMetadata: report.MakeNodeMetadata(),
AggregateMetadata: render.AggregateMetadata{},
ID: unknownPseudoNode2ID,
LabelMajor: "10.10.10.11",
Pseudo: true,
NodeMetadata: report.MakeNodeMetadata(),
EdgeMetadata: report.EdgeMetadata{},
}
theInternetNode = render.RenderableNode{
ID: render.TheInternetID,
LabelMajor: render.TheInternetMajor,
Pseudo: true,
NodeMetadata: report.MakeNodeMetadata(),
AggregateMetadata: render.AggregateMetadata{},
ID: render.TheInternetID,
LabelMajor: render.TheInternetMajor,
Pseudo: true,
NodeMetadata: report.MakeNodeMetadata(),
EdgeMetadata: report.EdgeMetadata{},
}
ClientProcess1ID = render.MakeProcessID(test.ClientHostID, test.Client1PID)
@@ -54,9 +54,9 @@ var (
test.ClientHostNodeID,
),
NodeMetadata: report.MakeNodeMetadata(),
AggregateMetadata: render.AggregateMetadata{
render.KeyBytesIngress: 100,
render.KeyBytesEgress: 10,
EdgeMetadata: report.EdgeMetadata{
PacketCount: newu64(100),
ByteCount: newu64(10),
},
},
ClientProcess2ID: {
@@ -72,9 +72,9 @@ var (
test.ClientHostNodeID,
),
NodeMetadata: report.MakeNodeMetadata(),
AggregateMetadata: render.AggregateMetadata{
render.KeyBytesIngress: 200,
render.KeyBytesEgress: 20,
EdgeMetadata: report.EdgeMetadata{
PacketCount: newu64(200),
ByteCount: newu64(20),
},
},
ServerProcessID: {
@@ -96,9 +96,9 @@ var (
test.ServerHostNodeID,
),
NodeMetadata: report.MakeNodeMetadata(),
AggregateMetadata: render.AggregateMetadata{
render.KeyBytesIngress: 150,
render.KeyBytesEgress: 1500,
EdgeMetadata: report.EdgeMetadata{
PacketCount: newu64(150),
ByteCount: newu64(1500),
},
},
nonContainerProcessID: {
@@ -112,8 +112,8 @@ var (
test.NonContainerProcessNodeID,
test.ServerHostNodeID,
),
NodeMetadata: report.MakeNodeMetadata(),
AggregateMetadata: render.AggregateMetadata{},
NodeMetadata: report.MakeNodeMetadata(),
EdgeMetadata: report.EdgeMetadata{},
},
unknownPseudoNode1ID: unknownPseudoNode1,
unknownPseudoNode2ID: unknownPseudoNode2,
@@ -136,9 +136,9 @@ var (
test.ClientHostNodeID,
),
NodeMetadata: report.MakeNodeMetadata(),
AggregateMetadata: render.AggregateMetadata{
render.KeyBytesIngress: 300,
render.KeyBytesEgress: 30,
EdgeMetadata: report.EdgeMetadata{
PacketCount: newu64(300),
ByteCount: newu64(30),
},
},
"apache": {
@@ -159,9 +159,9 @@ var (
test.ServerHostNodeID,
),
NodeMetadata: report.MakeNodeMetadata(),
AggregateMetadata: render.AggregateMetadata{
render.KeyBytesIngress: 150,
render.KeyBytesEgress: 1500,
EdgeMetadata: report.EdgeMetadata{
PacketCount: newu64(150),
ByteCount: newu64(1500),
},
},
"bash": {
@@ -174,8 +174,8 @@ var (
test.NonContainerProcessNodeID,
test.ServerHostNodeID,
),
NodeMetadata: report.MakeNodeMetadata(),
AggregateMetadata: render.AggregateMetadata{},
NodeMetadata: report.MakeNodeMetadata(),
EdgeMetadata: report.EdgeMetadata{},
},
unknownPseudoNode1ID: unknownPseudoNode1,
unknownPseudoNode2ID: unknownPseudoNode2,
@@ -199,9 +199,9 @@ var (
test.ClientHostNodeID,
),
NodeMetadata: report.MakeNodeMetadata(),
AggregateMetadata: render.AggregateMetadata{
render.KeyBytesIngress: 300,
render.KeyBytesEgress: 30,
EdgeMetadata: report.EdgeMetadata{
PacketCount: newu64(300),
ByteCount: newu64(30),
},
},
test.ServerContainerID: {
@@ -218,9 +218,9 @@ var (
test.ServerHostNodeID,
),
NodeMetadata: report.MakeNodeMetadata(),
AggregateMetadata: render.AggregateMetadata{
render.KeyBytesIngress: 150,
render.KeyBytesEgress: 1500,
EdgeMetadata: report.EdgeMetadata{
PacketCount: newu64(150),
ByteCount: newu64(1500),
},
},
uncontainedServerID: {
@@ -233,8 +233,8 @@ var (
test.NonContainerProcessNodeID,
test.ServerHostNodeID,
),
NodeMetadata: report.MakeNodeMetadata(),
AggregateMetadata: render.AggregateMetadata{},
NodeMetadata: report.MakeNodeMetadata(),
EdgeMetadata: report.EdgeMetadata{},
},
render.TheInternetID: theInternetNode,
}
@@ -257,9 +257,9 @@ var (
test.ClientHostNodeID,
),
NodeMetadata: report.MakeNodeMetadata(),
AggregateMetadata: render.AggregateMetadata{
render.KeyBytesIngress: 300,
render.KeyBytesEgress: 30,
EdgeMetadata: report.EdgeMetadata{
PacketCount: newu64(300),
ByteCount: newu64(30),
},
},
test.ServerContainerImageName: {
@@ -276,9 +276,9 @@ var (
test.ServerProcessNodeID,
test.ServerHostNodeID),
NodeMetadata: report.MakeNodeMetadata(),
AggregateMetadata: render.AggregateMetadata{
render.KeyBytesIngress: 150,
render.KeyBytesEgress: 1500,
EdgeMetadata: report.EdgeMetadata{
PacketCount: newu64(150),
ByteCount: newu64(1500),
},
},
uncontainedServerID: {
@@ -291,8 +291,8 @@ var (
test.NonContainerProcessNodeID,
test.ServerHostNodeID,
),
NodeMetadata: report.MakeNodeMetadata(),
AggregateMetadata: render.AggregateMetadata{},
NodeMetadata: report.MakeNodeMetadata(),
EdgeMetadata: report.EdgeMetadata{},
},
render.TheInternetID: theInternetNode,
}
@@ -315,8 +315,8 @@ var (
test.ServerAddressNodeID,
),
NodeMetadata: report.MakeNodeMetadata(),
AggregateMetadata: render.AggregateMetadata{
render.KeyMaxConnCountTCP: 3,
EdgeMetadata: report.EdgeMetadata{
MaxConnCountTCP: newu64(3),
},
},
ClientHostRenderedID: {
@@ -331,24 +331,26 @@ var (
test.ClientAddressNodeID,
),
NodeMetadata: report.MakeNodeMetadata(),
AggregateMetadata: render.AggregateMetadata{
render.KeyMaxConnCountTCP: 3,
EdgeMetadata: report.EdgeMetadata{
MaxConnCountTCP: newu64(3),
},
},
pseudoHostID1: {
ID: pseudoHostID1,
LabelMajor: "10.10.10.10",
Pseudo: true,
NodeMetadata: report.MakeNodeMetadata(),
AggregateMetadata: render.AggregateMetadata{},
ID: pseudoHostID1,
LabelMajor: "10.10.10.10",
Pseudo: true,
NodeMetadata: report.MakeNodeMetadata(),
EdgeMetadata: report.EdgeMetadata{},
},
pseudoHostID2: {
ID: pseudoHostID2,
LabelMajor: "10.10.10.11",
Pseudo: true,
NodeMetadata: report.MakeNodeMetadata(),
AggregateMetadata: render.AggregateMetadata{},
ID: pseudoHostID2,
LabelMajor: "10.10.10.11",
Pseudo: true,
NodeMetadata: report.MakeNodeMetadata(),
EdgeMetadata: report.EdgeMetadata{},
},
render.TheInternetID: theInternetNode,
}
)
func newu64(value uint64) *uint64 { return &value }
render/metadata.go
-46
View File
@@ -1,46 +0,0 @@
package render
import (
"github.com/weaveworks/scope/report"
)
// AggregateMetadata is a composable version of an EdgeMetadata. It's used
// when we want to merge nodes/edges for any reason.
//
// Even though we base it on EdgeMetadata, we can apply it to nodes, by
// summing up (merging) all of the {ingress, egress} metadatas of the
// {incoming, outgoing} edges to the node.
type AggregateMetadata map[string]int
const (
// KeyBytesIngress is the aggregate metadata key for the total count of
// ingress bytes.
KeyBytesIngress = "ingress_bytes"
// KeyBytesEgress is the aggregate metadata key for the total count of
// egress bytes.
KeyBytesEgress = "egress_bytes"
// KeyMaxConnCountTCP is the aggregate metadata key for the maximum number
// of simultaneous observed TCP connections in the window.
KeyMaxConnCountTCP = "max_conn_count_tcp"
)
// AggregateMetadataOf calculates a AggregateMetadata from an EdgeMetadata.
func AggregateMetadataOf(md report.EdgeMetadata) AggregateMetadata {
m := AggregateMetadata{}
if md.WithBytes {
m[KeyBytesIngress] = int(md.BytesIngress)
m[KeyBytesEgress] = int(md.BytesEgress)
}
if md.WithConnCountTCP {
// The maximum is the maximum. No need to calculate anything.
m[KeyMaxConnCountTCP] = int(md.MaxConnCountTCP)
}
return m
}
// Merge adds the fields from AggregateMetadata to r. r must be initialized.
func (r *AggregateMetadata) Merge(other AggregateMetadata) {
for k, v := range other {
(*r)[k] += v
}
}
render/metadata_test.go
-87
View File
@@ -1,87 +0,0 @@
package render_test
import (
"reflect"
"testing"
"github.com/weaveworks/scope/render"
"github.com/weaveworks/scope/report"
)
func TestAggregateMetadata(t *testing.T) {
for from, want := range map[report.EdgeMetadata]render.AggregateMetadata{
// Simple connection count
report.EdgeMetadata{
WithConnCountTCP: true,
MaxConnCountTCP: 400,
}: {
render.KeyMaxConnCountTCP: 400,
},
// Connection count rounding
report.EdgeMetadata{
WithConnCountTCP: true,
MaxConnCountTCP: 4,
}: {
render.KeyMaxConnCountTCP: 4,
},
// 0 connections.
report.EdgeMetadata{
WithConnCountTCP: true,
MaxConnCountTCP: 0,
}: {
render.KeyMaxConnCountTCP: 0,
},
// Egress
report.EdgeMetadata{
WithBytes: true,
BytesEgress: 24,
BytesIngress: 0,
}: {
render.KeyBytesEgress: 24,
render.KeyBytesIngress: 0,
},
// Ingress
report.EdgeMetadata{
WithBytes: true,
BytesEgress: 0,
BytesIngress: 1200,
}: {
render.KeyBytesEgress: 0,
render.KeyBytesIngress: 1200,
},
// Nothing there.
report.EdgeMetadata{}: {},
} {
if have := render.AggregateMetadataOf(from); !reflect.DeepEqual(have, want) {
t.Errorf("have: %#v, want %#v", have, want)
}
}
}
func TestAggregateMetadataSum(t *testing.T) {
var (
this = render.AggregateMetadata{
"ingress_bytes": 3,
}
other = render.AggregateMetadata{
"ingress_bytes": 333,
"egress_bytes": 3,
}
want = render.AggregateMetadata{
"ingress_bytes": 336,
"egress_bytes": 3,
}
)
this.Merge(other)
if have := this; !reflect.DeepEqual(have, want) {
t.Errorf("have: %#v, want %#v", have, want)
}
}
render/render.go
+13 -13
View File
@@ -9,7 +9,7 @@ import (
// Renderer is something that can render a report to a set of RenderableNodes.
type Renderer interface {
Render(report.Report) RenderableNodes
AggregateMetadata(rpt report.Report, localID, remoteID string) AggregateMetadata
EdgeMetadata(rpt report.Report, localID, remoteID string) report.EdgeMetadata
}
// Reduce renderer is a Renderer which merges together the output of several
@@ -50,11 +50,11 @@ func (r Reduce) Render(rpt report.Report) RenderableNodes {
return result
}
// AggregateMetadata produces an AggregateMetadata for a given edge.
func (r Reduce) AggregateMetadata(rpt report.Report, localID, remoteID string) AggregateMetadata {
metadata := AggregateMetadata{}
// EdgeMetadata produces an EdgeMetadata for a given edge.
func (r Reduce) EdgeMetadata(rpt report.Report, localID, remoteID string) report.EdgeMetadata {
metadata := report.EdgeMetadata{}
for _, renderer := range r {
metadata.Merge(renderer.AggregateMetadata(rpt, localID, remoteID))
metadata.Merge(renderer.EdgeMetadata(rpt, localID, remoteID))
}
return metadata
}
@@ -107,11 +107,11 @@ func (m Map) render(rpt report.Report) (RenderableNodes, map[string]string) {
return output, mapped
}
// AggregateMetadata gives the metadata of an edge from the perspective of the
// EdgeMetadata gives the metadata of an edge from the perspective of the
// srcRenderableID. Since an edgeID can have multiple edges on the address
// level, it uses the supplied mapping function to translate address IDs to
// renderable node (mapped) IDs.
func (m Map) AggregateMetadata(rpt report.Report, srcRenderableID, dstRenderableID string) AggregateMetadata {
func (m Map) EdgeMetadata(rpt report.Report, srcRenderableID, dstRenderableID string) report.EdgeMetadata {
// First we need to map the ids in this layer into the ids in the underlying layer
_, mapped := m.render(rpt) // this maps from old -> new
inverted := map[string][]string{} // this maps from new -> old(s)
@@ -130,9 +130,9 @@ func (m Map) AggregateMetadata(rpt report.Report, srcRenderableID, dstRenderable
}
// Now recurse for each old edge
output := AggregateMetadata{}
output := report.EdgeMetadata{}
for _, edge := range oldEdges {
metadata := m.Renderer.AggregateMetadata(rpt, edge.src, edge.dst)
metadata := m.Renderer.EdgeMetadata(rpt, edge.src, edge.dst)
output.Merge(metadata)
}
return output
@@ -205,7 +205,7 @@ func (m LeafMap) Render(rpt report.Report) RenderableNodes {
srcRenderableNode.Origins = srcRenderableNode.Origins.Add(srcNodeID)
edgeID := report.MakeEdgeID(srcNodeID, dstNodeID)
if md, ok := t.EdgeMetadatas[edgeID]; ok {
srcRenderableNode.AggregateMetadata.Merge(AggregateMetadataOf(md))
srcRenderableNode.EdgeMetadata.Merge(md)
}
}
@@ -215,11 +215,11 @@ func (m LeafMap) Render(rpt report.Report) RenderableNodes {
return nodes
}
// AggregateMetadata gives the metadata of an edge from the perspective of the
// EdgeMetadata gives the metadata of an edge from the perspective of the
// srcRenderableID. Since an edgeID can have multiple edges on the address
// level, it uses the supplied mapping function to translate address IDs to
// renderable node (mapped) IDs.
func (m LeafMap) AggregateMetadata(rpt report.Report, srcRenderableID, dstRenderableID string) AggregateMetadata {
func (m LeafMap) EdgeMetadata(rpt report.Report, srcRenderableID, dstRenderableID string) report.EdgeMetadata {
t := m.Selector(rpt)
metadata := report.EdgeMetadata{}
for edgeID, edgeMeta := range t.EdgeMetadatas {
@@ -240,7 +240,7 @@ func (m LeafMap) AggregateMetadata(rpt report.Report, srcRenderableID, dstRender
metadata.Flatten(edgeMeta)
}
}
return AggregateMetadataOf(metadata)
return metadata
}
// Render produces a set of RenderableNodes given a Report
render/render_test.go
+15 -15
View File
@@ -10,14 +10,14 @@ import (
type mockRenderer struct {
render.RenderableNodes
aggregateMetadata render.AggregateMetadata
edgeMetadata report.EdgeMetadata
}
func (m mockRenderer) Render(rpt report.Report) render.RenderableNodes {
return m.RenderableNodes
}
func (m mockRenderer) AggregateMetadata(rpt report.Report, localID, remoteID string) render.AggregateMetadata {
return m.aggregateMetadata
func (m mockRenderer) EdgeMetadata(rpt report.Report, localID, remoteID string) report.EdgeMetadata {
return m.edgeMetadata
}
func TestReduceRender(t *testing.T) {
@@ -36,12 +36,12 @@ func TestReduceRender(t *testing.T) {
func TestReduceEdge(t *testing.T) {
renderer := render.Reduce([]render.Renderer{
mockRenderer{aggregateMetadata: render.AggregateMetadata{"foo": 1}},
mockRenderer{aggregateMetadata: render.AggregateMetadata{"bar": 2}},
mockRenderer{edgeMetadata: report.EdgeMetadata{PacketCount: newu64(1)}},
mockRenderer{edgeMetadata: report.EdgeMetadata{PacketCount: newu64(2)}},
})
want := render.AggregateMetadata{"foo": 1, "bar": 2}
have := renderer.AggregateMetadata(report.MakeReport(), "", "")
want := report.EdgeMetadata{PacketCount: newu64(3)}
have := renderer.EdgeMetadata(report.MakeReport(), "", "")
if !reflect.DeepEqual(want, have) {
t.Errorf("want %+v, have %+v", want, have)
@@ -118,8 +118,8 @@ func TestMapEdge(t *testing.T) {
">bar": report.MakeIDList("foo"),
},
EdgeMetadatas: report.EdgeMetadatas{
"foo|bar": report.EdgeMetadata{WithBytes: true, BytesIngress: 1, BytesEgress: 2},
"bar|foo": report.EdgeMetadata{WithBytes: true, BytesIngress: 3, BytesEgress: 4},
"foo|bar": report.EdgeMetadata{PacketCount: newu64(1), ByteCount: newu64(2)},
"bar|foo": report.EdgeMetadata{PacketCount: newu64(3), ByteCount: newu64(4)},
},
}
}
@@ -139,12 +139,10 @@ func TestMapEdge(t *testing.T) {
},
}
want := render.AggregateMetadata{
render.KeyBytesIngress: 1,
render.KeyBytesEgress: 2,
}
have := mapper.AggregateMetadata(report.MakeReport(), "_foo", "_bar")
if !reflect.DeepEqual(want, have) {
if want, have := (report.EdgeMetadata{
PacketCount: newu64(1),
ByteCount: newu64(2),
}), mapper.EdgeMetadata(report.MakeReport(), "_foo", "_bar"); !reflect.DeepEqual(want, have) {
t.Errorf("want %+v, have %+v", want, have)
}
}
@@ -166,3 +164,5 @@ func TestFilterRender(t *testing.T) {
t.Errorf("want %+v, have %+v", want, have)
}
}
func newu64(value uint64) *uint64 { return &value }
render/renderable_node.go
+32 -32
View File
@@ -16,7 +16,7 @@ type RenderableNode struct {
Adjacency report.IDList `json:"adjacency,omitempty"` // Node IDs (in the same topology domain)
Origins report.IDList `json:"origins,omitempty"` // Core node IDs that contributed information
AggregateMetadata `json:"metadata"` // Numeric sums
report.EdgeMetadata `json:"metadata"` // Numeric sums
report.NodeMetadata `json:"-"` // merged NodeMetadata of the nodes used to build this
}
@@ -56,57 +56,57 @@ func (rn *RenderableNode) Merge(other RenderableNode) {
rn.Adjacency = rn.Adjacency.Merge(other.Adjacency)
rn.Origins = rn.Origins.Merge(other.Origins)
rn.AggregateMetadata.Merge(other.AggregateMetadata)
rn.EdgeMetadata.Merge(other.EdgeMetadata)
rn.NodeMetadata.Merge(other.NodeMetadata)
}
// NewRenderableNode makes a new RenderableNode
func NewRenderableNode(id, major, minor, rank string, nmd report.NodeMetadata) RenderableNode {
return RenderableNode{
ID: id,
LabelMajor: major,
LabelMinor: minor,
Rank: rank,
Pseudo: false,
AggregateMetadata: AggregateMetadata{},
NodeMetadata: nmd.Copy(),
ID: id,
LabelMajor: major,
LabelMinor: minor,
Rank: rank,
Pseudo: false,
EdgeMetadata: report.EdgeMetadata{},
NodeMetadata: nmd.Copy(),
}
}
func newDerivedNode(id string, node RenderableNode) RenderableNode {
return RenderableNode{
ID: id,
LabelMajor: "",
LabelMinor: "",
Rank: "",
Pseudo: node.Pseudo,
AggregateMetadata: node.AggregateMetadata,
Origins: node.Origins,
NodeMetadata: report.MakeNodeMetadata(),
ID: id,
LabelMajor: "",
LabelMinor: "",
Rank: "",
Pseudo: node.Pseudo,
EdgeMetadata: node.EdgeMetadata,
Origins: node.Origins,
NodeMetadata: report.MakeNodeMetadata(),
}
}
func newPseudoNode(id, major, minor string) RenderableNode {
return RenderableNode{
ID: id,
LabelMajor: major,
LabelMinor: minor,
Rank: "",
Pseudo: true,
AggregateMetadata: AggregateMetadata{},
NodeMetadata: report.MakeNodeMetadata(),
ID: id,
LabelMajor: major,
LabelMinor: minor,
Rank: "",
Pseudo: true,
EdgeMetadata: report.EdgeMetadata{},
NodeMetadata: report.MakeNodeMetadata(),
}
}
func newDerivedPseudoNode(id, major string, node RenderableNode) RenderableNode {
return RenderableNode{
ID: id,
LabelMajor: major,
LabelMinor: "",
Rank: "",
Pseudo: true,
AggregateMetadata: node.AggregateMetadata,
Origins: node.Origins,
NodeMetadata: report.MakeNodeMetadata(),
ID: id,
LabelMajor: major,
LabelMinor: "",
Rank: "",
Pseudo: true,
EdgeMetadata: node.EdgeMetadata,
Origins: node.Origins,
NodeMetadata: report.MakeNodeMetadata(),
}
}
render/topologies_test.go
+5 -5
View File
@@ -23,7 +23,7 @@ func TestProcessRenderer(t *testing.T) {
have := render.ProcessRenderer.Render(test.Report)
have = trimNodeMetadata(have)
if !reflect.DeepEqual(expected.RenderedProcesses, have) {
t.Error("\n" + test.Diff(expected.RenderedProcesses, have))
t.Error(test.Diff(expected.RenderedProcesses, have))
}
}
@@ -31,7 +31,7 @@ func TestProcessNameRenderer(t *testing.T) {
have := render.ProcessNameRenderer.Render(test.Report)
have = trimNodeMetadata(have)
if !reflect.DeepEqual(expected.RenderedProcessNames, have) {
t.Error("\n" + test.Diff(expected.RenderedProcessNames, have))
t.Error(test.Diff(expected.RenderedProcessNames, have))
}
}
@@ -39,7 +39,7 @@ func TestContainerRenderer(t *testing.T) {
have := render.ContainerRenderer.Render(test.Report)
have = trimNodeMetadata(have)
if !reflect.DeepEqual(expected.RenderedContainers, have) {
t.Error("\n" + test.Diff(expected.RenderedContainers, have))
t.Error(test.Diff(expected.RenderedContainers, have))
}
}
@@ -47,7 +47,7 @@ func TestContainerImageRenderer(t *testing.T) {
have := render.ContainerImageRenderer.Render(test.Report)
have = trimNodeMetadata(have)
if !reflect.DeepEqual(expected.RenderedContainerImages, have) {
t.Error("\n" + test.Diff(expected.RenderedContainerImages, have))
t.Error(test.Diff(expected.RenderedContainerImages, have))
}
}
@@ -55,6 +55,6 @@ func TestHostRenderer(t *testing.T) {
have := render.HostRenderer.Render(test.Report)
have = trimNodeMetadata(have)
if !reflect.DeepEqual(expected.RenderedHosts, have) {
t.Error("\n" + test.Diff(expected.RenderedHosts, have))
t.Error(test.Diff(expected.RenderedHosts, have))
}
}
report/merge.go
+35 -20
View File
@@ -12,6 +12,7 @@ func (r *Report) Merge(other Report) {
r.ContainerImage.Merge(other.ContainerImage)
r.Host.Merge(other.Host)
r.Overlay.Merge(other.Overlay)
r.Sampling.Merge(other.Sampling)
}
// Merge merges another Topology into the receiver.
@@ -62,31 +63,45 @@ func (e *EdgeMetadatas) Merge(other EdgeMetadatas) {
// Merge merges another EdgeMetadata into the receiver. The two edge metadatas
// should represent the same edge on different times.
func (m *EdgeMetadata) Merge(other EdgeMetadata) {
if other.WithBytes {
m.WithBytes = true
m.BytesIngress += other.BytesIngress
m.BytesEgress += other.BytesEgress
}
if other.WithConnCountTCP {
m.WithConnCountTCP = true
if other.MaxConnCountTCP > m.MaxConnCountTCP {
m.MaxConnCountTCP = other.MaxConnCountTCP
}
}
m.PacketCount = merge(m.PacketCount, other.PacketCount, sum)
m.ByteCount = merge(m.ByteCount, other.ByteCount, sum)
m.MaxConnCountTCP = merge(m.MaxConnCountTCP, other.MaxConnCountTCP, max)
}
// Flatten sums two EdgeMetadatas. Their windows should be the same duration;
// they should represent different edges at the same time.
func (m *EdgeMetadata) Flatten(other EdgeMetadata) {
if other.WithBytes {
m.WithBytes = true
m.BytesIngress += other.BytesIngress
m.BytesEgress += other.BytesEgress
m.PacketCount = merge(m.PacketCount, other.PacketCount, sum)
m.ByteCount = merge(m.ByteCount, other.ByteCount, sum)
// Note that summing of two maximums doesn't always give us the true
// maximum. But it's a best effort.
m.MaxConnCountTCP = merge(m.MaxConnCountTCP, other.MaxConnCountTCP, sum)
}
// Merge combines two sampling structures via simple addition.
func (s *Sampling) Merge(other Sampling) {
s.Count += other.Count
s.Total += other.Total
}
func merge(dst, src *uint64, op func(uint64, uint64) uint64) *uint64 {
if src == nil {
return dst
}
if other.WithConnCountTCP {
m.WithConnCountTCP = true
// Note: summing of two maximums doesn't always give the true maximum.
// But it's our Best Effort effort.
m.MaxConnCountTCP += other.MaxConnCountTCP
if dst == nil {
dst = new(uint64)
}
(*dst) = op(*dst, *src)
return dst
}
func sum(dst, src uint64) uint64 {
return dst + src
}
func max(dst, src uint64) uint64 {
if dst > src {
return dst
}
return src
}
report/merge_test.go
+33 -51
View File
@@ -112,102 +112,82 @@ func TestMergeEdgeMetadatas(t *testing.T) {
a: report.EdgeMetadatas{},
b: report.EdgeMetadatas{
"hostA|:192.168.1.1:12345|:192.168.1.2:80": report.EdgeMetadata{
WithBytes: true,
BytesEgress: 12,
BytesIngress: 0,
WithConnCountTCP: true,
MaxConnCountTCP: 2,
PacketCount: newu64(12),
ByteCount: newu64(0),
MaxConnCountTCP: newu64(2),
},
},
want: report.EdgeMetadatas{
"hostA|:192.168.1.1:12345|:192.168.1.2:80": report.EdgeMetadata{
WithBytes: true,
BytesEgress: 12,
BytesIngress: 0,
WithConnCountTCP: true,
MaxConnCountTCP: 2,
PacketCount: newu64(12),
ByteCount: newu64(0),
MaxConnCountTCP: newu64(2),
},
},
},
"Empty b": {
a: report.EdgeMetadatas{
"hostA|:192.168.1.1:12345|:192.168.1.2:80": report.EdgeMetadata{
WithBytes: true,
BytesEgress: 12,
BytesIngress: 0,
PacketCount: newu64(12),
ByteCount: newu64(0),
},
},
b: report.EdgeMetadatas{},
want: report.EdgeMetadatas{
"hostA|:192.168.1.1:12345|:192.168.1.2:80": report.EdgeMetadata{
WithBytes: true,
BytesEgress: 12,
BytesIngress: 0,
PacketCount: newu64(12),
ByteCount: newu64(0),
},
},
},
"Host merge": {
a: report.EdgeMetadatas{
"hostA|:192.168.1.1:12345|:192.168.1.2:80": report.EdgeMetadata{
WithBytes: true,
BytesEgress: 12,
BytesIngress: 0,
WithConnCountTCP: true,
MaxConnCountTCP: 4,
PacketCount: newu64(12),
ByteCount: newu64(0),
MaxConnCountTCP: newu64(4),
},
},
b: report.EdgeMetadatas{
"hostQ|:192.168.1.1:12345|:192.168.1.2:80": report.EdgeMetadata{
WithBytes: true,
BytesEgress: 1,
BytesIngress: 2,
WithConnCountTCP: true,
MaxConnCountTCP: 6,
PacketCount: newu64(1),
ByteCount: newu64(2),
MaxConnCountTCP: newu64(6),
},
},
want: report.EdgeMetadatas{
"hostA|:192.168.1.1:12345|:192.168.1.2:80": report.EdgeMetadata{
WithBytes: true,
BytesEgress: 12,
BytesIngress: 0,
WithConnCountTCP: true,
MaxConnCountTCP: 4,
PacketCount: newu64(12),
ByteCount: newu64(0),
MaxConnCountTCP: newu64(4),
},
"hostQ|:192.168.1.1:12345|:192.168.1.2:80": report.EdgeMetadata{
WithBytes: true,
BytesEgress: 1,
BytesIngress: 2,
WithConnCountTCP: true,
MaxConnCountTCP: 6,
PacketCount: newu64(1),
ByteCount: newu64(2),
MaxConnCountTCP: newu64(6),
},
},
},
"Edge merge": {
a: report.EdgeMetadatas{
"hostA|:192.168.1.1:12345|:192.168.1.2:80": report.EdgeMetadata{
WithBytes: true,
BytesEgress: 12,
BytesIngress: 0,
WithConnCountTCP: true,
MaxConnCountTCP: 7,
PacketCount: newu64(12),
ByteCount: newu64(0),
MaxConnCountTCP: newu64(7),
},
},
b: report.EdgeMetadatas{
"hostA|:192.168.1.1:12345|:192.168.1.2:80": report.EdgeMetadata{
WithBytes: true,
BytesEgress: 1,
BytesIngress: 2,
WithConnCountTCP: true,
MaxConnCountTCP: 9,
PacketCount: newu64(1),
ByteCount: newu64(2),
MaxConnCountTCP: newu64(9),
},
},
want: report.EdgeMetadatas{
"hostA|:192.168.1.1:12345|:192.168.1.2:80": report.EdgeMetadata{
WithBytes: true,
BytesEgress: 13,
BytesIngress: 2,
WithConnCountTCP: true,
MaxConnCountTCP: 9,
PacketCount: newu64(13),
ByteCount: newu64(2),
MaxConnCountTCP: newu64(9),
},
},
},
@@ -319,3 +299,5 @@ func TestMergeNodeMetadatas(t *testing.T) {
}
}
}
func newu64(value uint64) *uint64 { return &value }
report/report.go
+71 -29
View File
@@ -1,5 +1,7 @@
package report
import "fmt"
// Report is the core data type. It's produced by probes, and consumed and
// stored by apps. It's composed of multiple topologies, each representing
// a different (related, but not equivalent) view of the network.
@@ -36,6 +38,75 @@ type Report struct {
// overlaid on the infrastructure. The information is scraped by polling
// their status endpoints. Edges could be present, but aren't currently.
Overlay Topology
// Sampling data for this report.
Sampling
}
// MakeReport makes a clean report, ready to Merge() other reports into.
func MakeReport() Report {
return Report{
Endpoint: NewTopology(),
Address: NewTopology(),
Process: NewTopology(),
Container: NewTopology(),
ContainerImage: NewTopology(),
Host: NewTopology(),
Overlay: NewTopology(),
}
}
// 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.ContainerImage,
r.Host,
r.Overlay,
}
}
// Validate checks the report for various inconsistencies.
func (r Report) Validate() error {
var packets uint64
for _, topology := range r.Topologies() {
if err := topology.Validate(); err != nil {
return err
}
for _, emd := range topology.EdgeMetadatas {
if emd.PacketCount != nil {
packets += *emd.PacketCount
}
}
}
if r.Sampling.Count > r.Sampling.Total {
return fmt.Errorf("sampling count (%d) bigger than total (%d)", r.Sampling.Count, r.Sampling.Total)
}
if packets > 0 && (r.Sampling.Count == 0 || r.Sampling.Total == 0) {
return fmt.Errorf("packets exist in EdgeMetadata, but no sampling count or total in the base report")
}
return nil
}
// Sampling describes how the packet data sources for this report were
// sampled. It can be used to calculate effective sample rates. We can't
// just put the rate here, because that can't be accurately merged. Counts
// in e.g. edge metadata structures have already been adjusted to
// compensate for the sample rate.
type Sampling struct {
Count uint64 // observed and processed
Total uint64 // observed overall
}
// Rate returns the effective sampling rate.
func (s Sampling) Rate() float64 {
if s.Total <= 0 {
return 1.0
}
return float64(s.Count) / float64(s.Total)
}
const (
@@ -77,32 +148,3 @@ func SelectAddress(r Report) Topology {
func SelectHost(r Report) Topology {
return r.Host
}
// MakeReport makes a clean report, ready to Merge() other reports into.
func MakeReport() Report {
return Report{
Endpoint: NewTopology(),
Address: NewTopology(),
Process: NewTopology(),
Container: NewTopology(),
ContainerImage: NewTopology(),
Host: NewTopology(),
Overlay: NewTopology(),
}
}
// 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.ContainerImage, r.Host, r.Overlay}
}
// 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
}
report/report_test.go
+27
View File
@@ -0,0 +1,27 @@
package report_test
import (
"reflect"
"testing"
"github.com/weaveworks/scope/report"
)
// Make sure we don't add a topology and miss it in the Topologies method.
func TestReportTopologies(t *testing.T) {
var (
reportType = reflect.TypeOf(report.MakeReport())
topologyType = reflect.TypeOf(report.NewTopology())
)
var want int
for i := 0; i < reportType.NumField(); i++ {
if reportType.Field(i).Type == topologyType {
want++
}
}
if have := len(report.MakeReport().Topologies()); want != have {
t.Errorf("want %d, have %d", want, have)
}
}
report/topology.go
+5 -9
View File
@@ -28,16 +28,12 @@ type EdgeMetadatas map[string]EdgeMetadata
// IDs.
type NodeMetadatas map[string]NodeMetadata
// EdgeMetadata describes a superset of the metadata that probes can
// conceivably (and usefully) collect about an edge between two nodes in any
// topology.
// EdgeMetadata describes a superset of the metadata that probes can possibly
// collect about a directed edge between two nodes in any topology.
type EdgeMetadata struct {
WithBytes bool `json:"with_bytes,omitempty"`
BytesIngress uint `json:"bytes_ingress,omitempty"` // dst -> src
BytesEgress uint `json:"bytes_egress,omitempty"` // src -> dst
WithConnCountTCP bool `json:"with_conn_count_tcp,omitempty"`
MaxConnCountTCP uint `json:"max_conn_count_tcp,omitempty"`
PacketCount *uint64 `json:"packet_count,omitempty"`
ByteCount *uint64 `json:"byte_count,omitempty"`
MaxConnCountTCP *uint64 `json:"max_conn_count_tcp,omitempty"`
}
// NodeMetadata describes a superset of the metadata that probes can collect
test/report_fixture.go
+22 -25
View File
@@ -108,40 +108,33 @@ var (
},
EdgeMetadatas: report.EdgeMetadatas{
report.MakeEdgeID(Client54001NodeID, Server80NodeID): report.EdgeMetadata{
WithBytes: true,
BytesIngress: 100,
BytesEgress: 10,
PacketCount: newu64(100),
ByteCount: newu64(10),
},
report.MakeEdgeID(Client54002NodeID, Server80NodeID): report.EdgeMetadata{
WithBytes: true,
BytesIngress: 200,
BytesEgress: 20,
PacketCount: newu64(200),
ByteCount: newu64(20),
},
report.MakeEdgeID(Server80NodeID, Client54001NodeID): report.EdgeMetadata{
WithBytes: true,
BytesIngress: 10,
BytesEgress: 100,
PacketCount: newu64(10),
ByteCount: newu64(100),
},
report.MakeEdgeID(Server80NodeID, Client54002NodeID): report.EdgeMetadata{
WithBytes: true,
BytesIngress: 20,
BytesEgress: 200,
PacketCount: newu64(20),
ByteCount: newu64(200),
},
report.MakeEdgeID(Server80NodeID, UnknownClient1NodeID): report.EdgeMetadata{
WithBytes: true,
BytesIngress: 30,
BytesEgress: 300,
PacketCount: newu64(30),
ByteCount: newu64(300),
},
report.MakeEdgeID(Server80NodeID, UnknownClient2NodeID): report.EdgeMetadata{
WithBytes: true,
BytesIngress: 40,
BytesEgress: 400,
PacketCount: newu64(40),
ByteCount: newu64(400),
},
report.MakeEdgeID(Server80NodeID, UnknownClient3NodeID): report.EdgeMetadata{
WithBytes: true,
BytesIngress: 50,
BytesEgress: 500,
PacketCount: newu64(50),
ByteCount: newu64(500),
},
},
},
@@ -222,12 +215,10 @@ var (
},
EdgeMetadatas: report.EdgeMetadatas{
report.MakeEdgeID(ClientAddressNodeID, ServerAddressNodeID): report.EdgeMetadata{
WithConnCountTCP: true,
MaxConnCountTCP: 3,
MaxConnCountTCP: newu64(3),
},
report.MakeEdgeID(ServerAddressNodeID, ClientAddressNodeID): report.EdgeMetadata{
WithConnCountTCP: true,
MaxConnCountTCP: 3,
MaxConnCountTCP: newu64(3),
},
},
},
@@ -251,6 +242,10 @@ var (
},
EdgeMetadatas: report.EdgeMetadatas{},
},
Sampling: report.Sampling{
Count: 1024,
Total: 4096,
},
}
)
@@ -259,3 +254,5 @@ func init() {
panic(err)
}
}
func newu64(value uint64) *uint64 { return &value }