diff --git a/probe/endpoint/conntrack.go b/probe/endpoint/conntrack.go new file mode 100644 index 000000000..7daf5fa2a --- /dev/null +++ b/probe/endpoint/conntrack.go @@ -0,0 +1,225 @@ +package endpoint + +import ( + "bufio" + "encoding/xml" + "fmt" + "log" + "os" + "strings" + "sync" + + "github.com/weaveworks/scope/test/exec" +) + +// Constants exported for testing +const ( + modules = "/proc/modules" + conntrackModule = "nf_conntrack" + XMLHeader = "\n" + ConntrackOpenTag = "\n" + TimeWait = "TIME_WAIT" + TCP = "tcp" + New = "new" + Update = "update" + Destroy = "destroy" +) + +// Layer3 - these structs are for the parsed conntrack output +type Layer3 struct { + XMLName xml.Name `xml:"layer3"` + SrcIP string `xml:"src"` + DstIP string `xml:"dst"` +} + +// Layer4 - these structs are for the parsed conntrack output +type Layer4 struct { + XMLName xml.Name `xml:"layer4"` + SrcPort int `xml:"sport"` + DstPort int `xml:"dport"` + Proto string `xml:"protoname,attr"` +} + +// Meta - these structs are for the parsed conntrack output +type Meta struct { + XMLName xml.Name `xml:"meta"` + Direction string `xml:"direction,attr"` + Layer3 Layer3 `xml:"layer3"` + Layer4 Layer4 `xml:"layer4"` + ID int64 `xml:"id"` + State string `xml:"state"` +} + +// Flow - these structs are for the parsed conntrack output +type Flow struct { + XMLName xml.Name `xml:"flow"` + Metas []Meta `xml:"meta"` + Type string `xml:"type,attr"` + + Original, Reply, Independent *Meta `xml:"-"` +} + +// Conntracker uses the conntrack command to track network connections +type Conntracker struct { + sync.Mutex + cmd exec.Cmd + activeFlows map[int64]Flow // active flows in state != TIME_WAIT + bufferedFlows []Flow // flows coming out of activeFlows spend 1 walk cycle here +} + +// NewConntracker creates and starts a new Conntracter +func NewConntracker(args ...string) (*Conntracker, error) { + if !ConntrackModulePresent() { + return nil, fmt.Errorf("No conntrack module") + } + result := &Conntracker{ + activeFlows: map[int64]Flow{}, + } + go result.run(args...) + return result, nil +} + +// ConntrackModulePresent returns true if the kernel has the conntrack module +// present. It is made public for mocking. +var ConntrackModulePresent = func() bool { + f, err := os.Open(modules) + if err != nil { + return false + } + defer f.Close() + + scanner := bufio.NewScanner(f) + for scanner.Scan() { + line := scanner.Text() + if strings.HasPrefix(line, conntrackModule) { + return true + } + } + if err := scanner.Err(); err != nil { + log.Printf("conntrack error: %v", err) + } + + log.Printf("conntrack: failed to find module %s", conntrackModule) + return false +} + +// NB this is not re-entrant! +func (c *Conntracker) run(args ...string) { + args = append([]string{"-E", "-o", "xml"}, args...) + cmd := exec.Command("conntrack", args...) + stdout, err := cmd.StdoutPipe() + if err != nil { + log.Printf("conntrack error: %v", err) + return + } + if err := cmd.Start(); err != nil { + log.Printf("conntrack error: %v", err) + return + } + + c.Lock() + c.cmd = cmd + c.Unlock() + defer func() { + if err := cmd.Wait(); err != nil { + log.Printf("conntrack error: %v", err) + } + }() + + // Swallow the first two lines + reader := bufio.NewReader(stdout) + if line, err := reader.ReadString('\n'); err != nil { + log.Printf("conntrack error: %v", err) + return + } else if line != XMLHeader { + log.Printf("conntrack invalid output: '%s'", line) + return + } + if line, err := reader.ReadString('\n'); err != nil { + log.Printf("conntrack error: %v", err) + return + } else if line != ConntrackOpenTag { + log.Printf("conntrack invalid output: '%s'", line) + return + } + + // Now loop on the output stream + decoder := xml.NewDecoder(reader) + for { + var f Flow + if err := decoder.Decode(&f); err != nil { + log.Printf("conntrack error: %v", err) + } + c.handleFlow(f) + } +} + +// Stop stop stop +func (c *Conntracker) Stop() { + c.Lock() + defer c.Unlock() + if c.cmd == nil { + return + } + + if p := c.cmd.Process(); p != nil { + p.Kill() + } +} + +func (c *Conntracker) handleFlow(f Flow) { + // A flow consists of 3 'metas' - the 'original' 4 tuple (as seen by this + // host) and the 'reply' 4 tuple, which is what it has been rewritten to. + // This code finds those metas, which are identified by a Direction + // attribute. + for i := range f.Metas { + meta := &f.Metas[i] + switch meta.Direction { + case "original": + f.Original = meta + case "reply": + f.Reply = meta + case "independent": + f.Independent = meta + } + } + + // For not, I'm only interested in tcp connections - there is too much udp + // traffic going on (every container talking to weave dns, for example) to + // render nicely. TODO: revisit this. + if f.Original.Layer4.Proto != TCP { + return + } + + c.Lock() + defer c.Unlock() + + switch f.Type { + case New, Update: + if f.Independent.State != TimeWait { + c.activeFlows[f.Independent.ID] = f + } else if _, ok := c.activeFlows[f.Independent.ID]; ok { + delete(c.activeFlows, f.Independent.ID) + c.bufferedFlows = append(c.bufferedFlows, f) + } + case Destroy: + if _, ok := c.activeFlows[f.Independent.ID]; ok { + delete(c.activeFlows, f.Independent.ID) + c.bufferedFlows = append(c.bufferedFlows, f) + } + } +} + +// WalkFlows calls f with all active flows and flows that have come and gone +// since the last call to WalkFlows +func (c *Conntracker) WalkFlows(f func(Flow)) { + c.Lock() + defer c.Unlock() + for _, flow := range c.activeFlows { + f(flow) + } + for _, flow := range c.bufferedFlows { + f(flow) + } + c.bufferedFlows = c.bufferedFlows[:0] +} diff --git a/probe/endpoint/conntrack_test.go b/probe/endpoint/conntrack_test.go new file mode 100644 index 000000000..41f00dbc3 --- /dev/null +++ b/probe/endpoint/conntrack_test.go @@ -0,0 +1,142 @@ +package endpoint_test + +import ( + "bufio" + "encoding/xml" + "io" + "testing" + "time" + + . "github.com/weaveworks/scope/probe/endpoint" + "github.com/weaveworks/scope/test" + "github.com/weaveworks/scope/test/exec" +) + +func makeFlow(id int64, srcIP, dstIP string, srcPort, dstPort int, ty, state string) Flow { + return Flow{ + XMLName: xml.Name{ + Local: "flow", + }, + Type: ty, + Metas: []Meta{ + { + XMLName: xml.Name{ + Local: "meta", + }, + Direction: "original", + Layer3: Layer3{ + XMLName: xml.Name{ + Local: "layer3", + }, + SrcIP: srcIP, + DstIP: dstIP, + }, + Layer4: Layer4{ + XMLName: xml.Name{ + Local: "layer4", + }, + SrcPort: srcPort, + DstPort: dstPort, + Proto: TCP, + }, + }, + { + XMLName: xml.Name{ + Local: "meta", + }, + Direction: "independent", + ID: id, + State: state, + Layer3: Layer3{ + XMLName: xml.Name{ + Local: "layer3", + }, + }, + Layer4: Layer4{ + XMLName: xml.Name{ + Local: "layer4", + }, + }, + }, + }, + } +} + +func TestConntracker(t *testing.T) { + oldExecCmd, oldConntrackPresent := exec.Command, ConntrackModulePresent + defer func() { exec.Command, ConntrackModulePresent = oldExecCmd, oldConntrackPresent }() + + ConntrackModulePresent = func() bool { + return true + } + + reader, writer := io.Pipe() + exec.Command = func(name string, args ...string) exec.Cmd { + return exec.NewMockCmd(reader) + } + + conntracker, err := NewConntracker() + if err != nil { + t.Fatal(err) + } + + bw := bufio.NewWriter(writer) + if _, err := bw.WriteString(XMLHeader); err != nil { + t.Fatal(err) + } + if _, err := bw.WriteString(ConntrackOpenTag); err != nil { + t.Fatal(err) + } + if err := bw.Flush(); err != nil { + t.Fatal(err) + } + + have := func() interface{} { + result := []Flow{} + conntracker.WalkFlows(func(f Flow) { + f.Original = nil + f.Reply = nil + f.Independent = nil + result = append(result, f) + }) + return result + } + ts := 100 * time.Millisecond + + // First, assert we have no flows + test.Poll(t, ts, []Flow{}, have) + + // Now add some flows + xmlEncoder := xml.NewEncoder(bw) + writeFlow := func(f Flow) { + if err := xmlEncoder.Encode(f); err != nil { + t.Fatal(err) + } + if _, err := bw.WriteString("\n"); err != nil { + t.Fatal(err) + } + if err := bw.Flush(); err != nil { + t.Fatal(err) + } + } + + flow1 := makeFlow(1, "1.2.3.4", "2.3.4.5", 2, 3, New, "") + writeFlow(flow1) + test.Poll(t, ts, []Flow{flow1}, have) + + // Now check when we remove the flow, we still get it in the next Walk + flow1.Type = Destroy + writeFlow(flow1) + test.Poll(t, ts, []Flow{flow1}, have) + test.Poll(t, ts, []Flow{}, have) + + // This time we're not going to remove it, but put it in state TIME_WAIT + flow1.Type = New + writeFlow(flow1) + test.Poll(t, ts, []Flow{flow1}, have) + + flow1.Metas[1].State = TimeWait + writeFlow(flow1) + test.Poll(t, ts, []Flow{flow1}, have) + test.Poll(t, ts, []Flow{}, have) +} diff --git a/probe/endpoint/nat.go b/probe/endpoint/nat.go index 2639c88fb..2548ab2d9 100644 --- a/probe/endpoint/nat.go +++ b/probe/endpoint/nat.go @@ -1,54 +1,11 @@ package endpoint import ( - "bufio" - "encoding/xml" - "io" - "log" - "os" - "os/exec" "strconv" - "strings" "github.com/weaveworks/scope/report" ) -const ( - modules = "/proc/modules" - conntrackModule = "nf_conntrack" -) - -// these structs are for the parsed conntrack output -type layer3 struct { - XMLName xml.Name `xml:"layer3"` - SrcIP string `xml:"src"` - DstIP string `xml:"dst"` -} - -type layer4 struct { - XMLName xml.Name `xml:"layer4"` - SrcPort int `xml:"sport"` - DstPort int `xml:"dport"` - Proto string `xml:"protoname,attr"` -} - -type meta struct { - XMLName xml.Name `xml:"meta"` - Direction string `xml:"direction,attr"` - Layer3 layer3 `xml:"layer3"` - Layer4 layer4 `xml:"layer4"` -} - -type flow struct { - XMLName xml.Name `xml:"flow"` - Metas []meta `xml:"meta"` -} - -type conntrack struct { - XMLName xml.Name `xml:"conntrack"` - Flows []flow `xml:"flow"` -} - // This is our 'abstraction' of the endpoint that have been rewritten by NAT. // Original is the private IP that has been rewritten. type endpointMapping struct { @@ -59,111 +16,51 @@ type endpointMapping struct { rewrittenPort int } -// natTable returns a list of endpoints that have been remapped by NAT. -func natTable() ([]endpointMapping, error) { - var conntrack conntrack - cmd := exec.Command("conntrack", "-L", "--any-nat", "-o", "xml") - stdout, err := cmd.StdoutPipe() +type natmapper struct { + *Conntracker +} + +func newNATMapper() (*natmapper, error) { + ct, err := NewConntracker("--any-nat") if err != nil { return nil, err } - if err := cmd.Start(); err != nil { - return nil, err - } - defer func() { - if err := cmd.Wait(); err != nil { - log.Printf("conntrack error: %v", err) + return &natmapper{ct}, nil +} + +func toMapping(f Flow) *endpointMapping { + var mapping endpointMapping + if f.Original.Layer3.SrcIP == f.Reply.Layer3.DstIP { + mapping = endpointMapping{ + originalIP: f.Reply.Layer3.SrcIP, + originalPort: f.Reply.Layer4.SrcPort, + rewrittenIP: f.Original.Layer3.DstIP, + rewrittenPort: f.Original.Layer4.DstPort, } - }() - if err := xml.NewDecoder(stdout).Decode(&conntrack); err != nil { - if err == io.EOF { - return []endpointMapping{}, nil + } else { + mapping = endpointMapping{ + originalIP: f.Original.Layer3.SrcIP, + originalPort: f.Original.Layer4.SrcPort, + rewrittenIP: f.Reply.Layer3.DstIP, + rewrittenPort: f.Reply.Layer4.DstPort, } - return nil, err } - output := []endpointMapping{} - for _, flow := range conntrack.Flows { - // A flow consists of 3 'metas' - the 'original' 4 tuple (as seen by this - // host) and the 'reply' 4 tuple, which is what it has been rewritten to. - // This code finds those metas, which are identified by a Direction - // attribute. - original, reply := meta{}, meta{} - for _, meta := range flow.Metas { - if meta.Direction == "original" { - original = meta - } else if meta.Direction == "reply" { - reply = meta - } - } - - if original.Layer4.Proto != "tcp" { - continue - } - - var conn endpointMapping - if original.Layer3.SrcIP == reply.Layer3.DstIP { - conn = endpointMapping{ - originalIP: reply.Layer3.SrcIP, - originalPort: reply.Layer4.SrcPort, - rewrittenIP: original.Layer3.DstIP, - rewrittenPort: original.Layer4.DstPort, - } - } else { - conn = endpointMapping{ - originalIP: original.Layer3.SrcIP, - originalPort: original.Layer4.SrcPort, - rewrittenIP: reply.Layer3.DstIP, - rewrittenPort: reply.Layer4.DstPort, - } - } - - output = append(output, conn) - } - - return output, nil + return &mapping } // applyNAT duplicates NodeMetadatas in the endpoint topology of a // report, based on the NAT table as returns by natTable. -func applyNAT(rpt report.Report, scope string) error { - mappings, err := natTable() - if err != nil { - return err - } - - for _, mapping := range mappings { +func (n *natmapper) applyNAT(rpt report.Report, scope string) { + n.WalkFlows(func(f Flow) { + mapping := toMapping(f) realEndpointID := report.MakeEndpointNodeID(scope, mapping.originalIP, strconv.Itoa(mapping.originalPort)) copyEndpointID := report.MakeEndpointNodeID(scope, mapping.rewrittenIP, strconv.Itoa(mapping.rewrittenPort)) nmd, ok := rpt.Endpoint.NodeMetadatas[realEndpointID] if !ok { - continue + return } rpt.Endpoint.NodeMetadatas[copyEndpointID] = nmd.Copy() - } - - return nil -} - -func conntrackModulePresent() bool { - f, err := os.Open(modules) - if err != nil { - return false - } - defer f.Close() - - scanner := bufio.NewScanner(f) - for scanner.Scan() { - line := scanner.Text() - if strings.HasPrefix(line, conntrackModule) { - return true - } - } - if err := scanner.Err(); err != nil { - log.Printf("conntrack error: %v", err) - } - - log.Printf("conntrack: failed to find module %s", conntrackModule) - return false + }) } diff --git a/probe/endpoint/reporter.go b/probe/endpoint/reporter.go index 0c7c5bfad..3a758f9ee 100644 --- a/probe/endpoint/reporter.go +++ b/probe/endpoint/reporter.go @@ -1,7 +1,7 @@ package endpoint import ( - "fmt" + "log" "strconv" "time" @@ -24,6 +24,8 @@ type Reporter struct { hostName string includeProcesses bool includeNAT bool + conntracker *Conntracker + natmapper *natmapper } // SpyDuration is an exported prometheus metric @@ -43,12 +45,41 @@ var SpyDuration = prometheus.NewSummaryVec( // on the host machine, at the granularity of host and port. That information // is stored in the Endpoint topology. It optionally enriches that topology // with process (PID) information. -func NewReporter(hostID, hostName string, includeProcesses bool) *Reporter { +func NewReporter(hostID, hostName string, includeProcesses bool, useConntrack bool) *Reporter { + var ( + conntrackModulePresent = ConntrackModulePresent() + conntracker *Conntracker + natmapper *natmapper + err error + ) + if conntrackModulePresent && useConntrack { + conntracker, err = NewConntracker() + if err != nil { + log.Printf("Failed to start conntracker: %v", err) + } + } + if conntrackModulePresent { + natmapper, err = newNATMapper() + if err != nil { + log.Printf("Failed to start natMapper: %v", err) + } + } return &Reporter{ hostID: hostID, hostName: hostName, includeProcesses: includeProcesses, - includeNAT: conntrackModulePresent(), + conntracker: conntracker, + natmapper: natmapper, + } +} + +// Stop stop stop +func (r *Reporter) Stop() { + if r.conntracker != nil { + r.conntracker.Stop() + } + if r.natmapper != nil { + r.natmapper.Stop() } } @@ -65,50 +96,73 @@ func (r *Reporter) Report() (report.Report, error) { } for conn := conns.Next(); conn != nil; conn = conns.Next() { - r.addConnection(&rpt, conn) + var ( + localPort = conn.LocalPort + remotePort = conn.RemotePort + localAddr = conn.LocalAddress.String() + remoteAddr = conn.RemoteAddress.String() + ) + r.addConnection(&rpt, localAddr, remoteAddr, localPort, remotePort, &conn.Proc) } - if r.includeNAT { - err = applyNAT(rpt, r.hostID) + if r.conntracker != nil { + r.conntracker.WalkFlows(func(f Flow) { + var ( + localPort = f.Original.Layer4.SrcPort + remotePort = f.Original.Layer4.DstPort + localAddr = f.Original.Layer3.SrcIP + remoteAddr = f.Original.Layer3.DstIP + ) + r.addConnection(&rpt, localAddr, remoteAddr, uint16(localPort), uint16(remotePort), nil) + }) + } + + if r.natmapper != nil { + r.natmapper.applyNAT(rpt, r.hostID) } return rpt, err } -func (r *Reporter) addConnection(rpt *report.Report, c *procspy.Connection) { - var ( - localIsClient = int(c.LocalPort) > int(c.RemotePort) - localAddressNodeID = report.MakeAddressNodeID(r.hostID, c.LocalAddress.String()) - remoteAddressNodeID = report.MakeAddressNodeID(r.hostID, c.RemoteAddress.String()) - adjacencyID = "" - edgeID = "" - ) +func (r *Reporter) addConnection(rpt *report.Report, localAddr, remoteAddr string, localPort, remotePort uint16, proc *procspy.Proc) { + localIsClient := int(localPort) > int(remotePort) - if localIsClient { - adjacencyID = report.MakeAdjacencyID(localAddressNodeID) - rpt.Address.Adjacency[adjacencyID] = rpt.Address.Adjacency[adjacencyID].Add(remoteAddressNodeID) - - edgeID = report.MakeEdgeID(localAddressNodeID, remoteAddressNodeID) - } else { - adjacencyID = report.MakeAdjacencyID(remoteAddressNodeID) - rpt.Address.Adjacency[adjacencyID] = rpt.Address.Adjacency[adjacencyID].Add(localAddressNodeID) - - edgeID = report.MakeEdgeID(remoteAddressNodeID, localAddressNodeID) - } - - if _, ok := rpt.Address.NodeMetadatas[localAddressNodeID]; !ok { - rpt.Address.NodeMetadatas[localAddressNodeID] = report.MakeNodeMetadataWith(map[string]string{ - "name": r.hostName, - Addr: c.LocalAddress.String(), - }) - } - - countTCPConnection(rpt.Address.EdgeMetadatas, edgeID) - - if c.Proc.PID > 0 { + // Update address topology + { 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))) + localAddressNodeID = report.MakeAddressNodeID(r.hostID, localAddr) + remoteAddressNodeID = report.MakeAddressNodeID(r.hostID, remoteAddr) + adjacencyID = "" + edgeID = "" + ) + + if localIsClient { + adjacencyID = report.MakeAdjacencyID(localAddressNodeID) + rpt.Address.Adjacency[adjacencyID] = rpt.Address.Adjacency[adjacencyID].Add(remoteAddressNodeID) + + edgeID = report.MakeEdgeID(localAddressNodeID, remoteAddressNodeID) + } else { + adjacencyID = report.MakeAdjacencyID(remoteAddressNodeID) + rpt.Address.Adjacency[adjacencyID] = rpt.Address.Adjacency[adjacencyID].Add(localAddressNodeID) + + edgeID = report.MakeEdgeID(remoteAddressNodeID, localAddressNodeID) + } + + countTCPConnection(rpt.Address.EdgeMetadatas, edgeID) + + if _, ok := rpt.Address.NodeMetadatas[localAddressNodeID]; !ok { + rpt.Address.NodeMetadatas[localAddressNodeID] = report.MakeNodeMetadataWith(map[string]string{ + "name": r.hostName, + Addr: localAddr, + }) + } + } + + // Update endpoint topology + if r.includeProcesses { + var ( + localEndpointNodeID = report.MakeEndpointNodeID(r.hostID, localAddr, strconv.Itoa(int(localPort))) + remoteEndpointNodeID = report.MakeEndpointNodeID(r.hostID, remoteAddr, strconv.Itoa(int(remotePort))) adjacencyID = "" edgeID = "" ) @@ -125,18 +179,24 @@ func (r *Reporter) addConnection(rpt *report.Report, c *procspy.Connection) { edgeID = report.MakeEdgeID(remoteEndpointNodeID, localEndpointNodeID) } - if _, ok := rpt.Endpoint.NodeMetadatas[localEndpointNodeID]; !ok { - // First hit establishes NodeMetadata for scoped local address + port - md := report.MakeNodeMetadataWith(map[string]string{ - Addr: c.LocalAddress.String(), - Port: strconv.Itoa(int(c.LocalPort)), - process.PID: fmt.Sprint(c.Proc.PID), - }) + countTCPConnection(rpt.Endpoint.EdgeMetadatas, edgeID) + md, ok := rpt.Endpoint.NodeMetadatas[localEndpointNodeID] + updated := !ok + if !ok { + md = report.MakeNodeMetadataWith(map[string]string{ + Addr: localAddr, + Port: strconv.Itoa(int(localPort)), + }) + } + if proc != nil && proc.PID > 0 { + pid := strconv.FormatUint(uint64(proc.PID), 10) + updated = updated || md.Metadata[process.PID] != pid + md.Metadata[process.PID] = pid + } + if updated { rpt.Endpoint.NodeMetadatas[localEndpointNodeID] = md } - - countTCPConnection(rpt.Endpoint.EdgeMetadatas, edgeID) } } diff --git a/probe/endpoint/reporter_test.go b/probe/endpoint/reporter_test.go index 455d18ada..0258c1636 100644 --- a/probe/endpoint/reporter_test.go +++ b/probe/endpoint/reporter_test.go @@ -19,7 +19,6 @@ var ( fixRemotePortB = uint16(12346) fixProcessPID = uint(4242) fixProcessName = "nginx" - fixProcessPIDB = uint(4243) fixConnections = []procspy.Connection{ { @@ -55,9 +54,9 @@ var ( LocalAddress: fixLocalAddress, LocalPort: fixLocalPort, RemoteAddress: fixRemoteAddress, - RemotePort: fixRemotePort, + RemotePort: fixRemotePortB, Proc: procspy.Proc{ - PID: fixProcessPIDB, + PID: fixProcessPID, Name: fixProcessName, }, }, @@ -72,7 +71,7 @@ func TestSpyNoProcesses(t *testing.T) { nodeName = "frenchs-since-1904" // TODO rename to hostNmae ) - reporter := endpoint.NewReporter(nodeID, nodeName, false) + reporter := endpoint.NewReporter(nodeID, nodeName, false, false) r, _ := reporter.Report() //buf, _ := json.MarshalIndent(r, "", " ") //t.Logf("\n%s\n", buf) @@ -109,7 +108,7 @@ func TestSpyWithProcesses(t *testing.T) { nodeName = "fishermans-friend" // TODO rename to hostNmae ) - reporter := endpoint.NewReporter(nodeID, nodeName, false) + reporter := endpoint.NewReporter(nodeID, nodeName, true, false) r, _ := reporter.Report() // buf, _ := json.MarshalIndent(r, "", " ") ; t.Logf("\n%s\n", buf) diff --git a/probe/main.go b/probe/main.go index 02274459f..db0c24971 100644 --- a/probe/main.go +++ b/probe/main.go @@ -46,6 +46,7 @@ func main() { 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'") printVersion = flag.Bool("version", false, "print version number and exit") + useConntrack = flag.Bool("conntrack", true, "also use conntrack to track connections") ) flag.Parse() @@ -103,13 +104,16 @@ func main() { } var ( - taggers = []Tagger{newTopologyTagger(), host.NewTagger(hostID)} - reporters = []Reporter{host.NewReporter(hostID, hostName, localNets), endpoint.NewReporter(hostID, hostName, *spyProcs)} - processCache *process.CachingWalker + endpointReporter = endpoint.NewReporter(hostID, hostName, *spyProcs, *useConntrack) + processCache = process.NewCachingWalker(process.NewWalker(*procRoot)) + reporters = []Reporter{ + endpointReporter, + host.NewReporter(hostID, hostName, localNets), + process.NewReporter(processCache, hostID), + } + taggers = []Tagger{newTopologyTagger(), host.NewTagger(hostID)} ) - - processCache = process.NewCachingWalker(process.NewWalker(*procRoot)) - reporters = append(reporters, process.NewReporter(processCache, hostID)) + defer endpointReporter.Stop() if *dockerEnabled { if err := report.AddLocalBridge(*dockerBridge); err != nil { diff --git a/probe/overlay/weave.go b/probe/overlay/weave.go index 38c580853..314458b77 100644 --- a/probe/overlay/weave.go +++ b/probe/overlay/weave.go @@ -4,17 +4,16 @@ import ( "bufio" "encoding/json" "fmt" - "io" "log" "net" "net/http" "net/url" - "os/exec" "regexp" "strings" "github.com/weaveworks/scope/probe/docker" "github.com/weaveworks/scope/report" + "github.com/weaveworks/scope/test/exec" ) const ( @@ -36,16 +35,6 @@ const ( var weavePsMatch = regexp.MustCompile(`^([0-9a-f]{12}) ((?:[0-9a-f][0-9a-f]\:){5}(?:[0-9a-f][0-9a-f]))(.*)$`) var ipMatch = regexp.MustCompile(`([0-9]{1,3}\.[0-9]{1,3}\.[0-9]{1,3}\.[0-9]{1,3})(/[0-9]+)`) -// Cmd is a hook for mocking -type Cmd interface { - StdoutPipe() (io.ReadCloser, error) - Start() error - Wait() error -} - -// ExecCommand is a hook for mocking -var ExecCommand = func(name string, args ...string) Cmd { return exec.Command(name, args...) } - // Weave represents a single Weave router, presumably on the same host // as the probe. It is both a Reporter and a Tagger: it produces an Overlay // topology, and (in theory) can tag existing topologies with foreign keys to @@ -114,7 +103,7 @@ type psEntry struct { func (w Weave) ps() ([]psEntry, error) { var result []psEntry - cmd := ExecCommand("weave", "--local", "ps") + cmd := exec.Command("weave", "--local", "ps") out, err := cmd.StdoutPipe() if err != nil { return result, err diff --git a/probe/overlay/weave_test.go b/probe/overlay/weave_test.go index 679c45095..23ae17e59 100644 --- a/probe/overlay/weave_test.go +++ b/probe/overlay/weave_test.go @@ -1,10 +1,7 @@ package overlay_test import ( - "bytes" "fmt" - "io" - "io/ioutil" "net/http" "net/http/httptest" "reflect" @@ -14,37 +11,14 @@ import ( "github.com/weaveworks/scope/probe/overlay" "github.com/weaveworks/scope/report" "github.com/weaveworks/scope/test" + "github.com/weaveworks/scope/test/exec" ) -type mockCmd struct { - *bytes.Buffer -} - -func (c *mockCmd) Start() error { - return nil -} - -func (c *mockCmd) Wait() error { - return nil -} - -func (c *mockCmd) StdoutPipe() (io.ReadCloser, error) { - return struct { - io.Reader - io.Closer - }{ - c.Buffer, - ioutil.NopCloser(nil), - }, nil -} - func TestWeaveTaggerOverlayTopology(t *testing.T) { - oldExecCmd := overlay.ExecCommand - defer func() { overlay.ExecCommand = oldExecCmd }() - overlay.ExecCommand = func(name string, args ...string) overlay.Cmd { - return &mockCmd{ - bytes.NewBufferString(fmt.Sprintf("%s %s %s/24\n", mockContainerID, mockContainerMAC, mockContainerIP)), - } + oldExecCmd := exec.Command + defer func() { exec.Command = oldExecCmd }() + exec.Command = func(name string, args ...string) exec.Cmd { + return exec.NewMockCmdString(fmt.Sprintf("%s %s %s/24\n", mockContainerID, mockContainerMAC, mockContainerIP)) } s := httptest.NewServer(http.HandlerFunc(mockWeaveRouter)) diff --git a/test/exec/exec.go b/test/exec/exec.go new file mode 100644 index 000000000..19ebd7d04 --- /dev/null +++ b/test/exec/exec.go @@ -0,0 +1,68 @@ +package exec + +import ( + "bytes" + "io" + "io/ioutil" + "os" + "os/exec" +) + +// Cmd is a hook for mocking +type Cmd interface { + StdoutPipe() (io.ReadCloser, error) + Start() error + Wait() error + Process() *os.Process +} + +// Command is a hook for mocking +var Command = func(name string, args ...string) Cmd { + return &realCmd{exec.Command(name, args...)} +} + +type realCmd struct { + *exec.Cmd +} + +func (c *realCmd) Process() *os.Process { + return c.Cmd.Process +} + +type mockCmd struct { + io.ReadCloser +} + +// NewMockCmdString creates a new mock Cmd which has s on its stdout pipe +func NewMockCmdString(s string) Cmd { + return &mockCmd{ + struct { + io.Reader + io.Closer + }{ + bytes.NewBufferString(s), + ioutil.NopCloser(nil), + }, + } +} + +// NewMockCmd creates a new mock Cmd with rc as its stdout pipe +func NewMockCmd(rc io.ReadCloser) Cmd { + return &mockCmd{rc} +} + +func (c *mockCmd) Start() error { + return nil +} + +func (c *mockCmd) Wait() error { + return nil +} + +func (c *mockCmd) StdoutPipe() (io.ReadCloser, error) { + return c.ReadCloser, nil +} + +func (c *mockCmd) Process() *os.Process { + return nil +} diff --git a/test/poll.go b/test/poll.go index 864043f25..da20f15e3 100644 --- a/test/poll.go +++ b/test/poll.go @@ -2,6 +2,7 @@ package test import ( "reflect" + "runtime" "testing" "time" ) @@ -20,6 +21,7 @@ func Poll(t *testing.T, d time.Duration, want interface{}, have func() interface } h := have() if !reflect.DeepEqual(want, h) { - t.Fatal(Diff(want, h)) + _, file, line, _ := runtime.Caller(1) + t.Fatalf("%s:%d: %s", file, line, Diff(want, h)) } }