Merge pull request #1828 from weaveworks/1826-conntrack-into-kernel

Improve conntrack kernel-support detection for in-kernel compilations

probe/endpoint/conntrack.go

@@ -5,7 +5,7 @@ import (
 	"encoding/xml"
 	"io"
 	"os"
-	"strings"
+	"path/filepath"
 	"sync"
 	"time"
 
@@ -15,8 +15,9 @@ import (
 )
 
 const (
-	modules          = "/proc/modules"
-	conntrackModule  = "nf_conntrack"
+	// From https://www.kernel.org/doc/Documentation/networking/nf_conntrack-sysctl.txt
+	// Check a tcp-related file for existence since we need tcp tracking
+	procFileToCheck  = "sys/net/netfilter/nf_conntrack_tcp_timeout_close"
 	xmlHeader        = "<?xml version=\"1.0\" encoding=\"utf-8\"?>\n"
 	conntrackOpenTag = "<conntrack>\n"
 	timeWait         = "TIME_WAIT"
@@ -85,11 +86,11 @@ type conntrackWalker struct {
 }
 
 // newConntracker creates and starts a new conntracker.
-func newConntrackFlowWalker(useConntrack bool, args ...string) flowWalker {
-	if !ConntrackModulePresent() {
-		log.Info("Not using conntrack: module not present")
+func newConntrackFlowWalker(useConntrack bool, procRoot string, args ...string) flowWalker {
+	if !useConntrack {
 		return nilFlowWalker{}
-	} else if !useConntrack {
+	} else if err := IsConntrackSupported(procRoot); err != nil {
+		log.Warnf("Not using conntrack: not supported by the kernel: %s", err)
 		return nilFlowWalker{}
 	}
 	result := &conntrackWalker{
@@ -101,28 +102,11 @@ func newConntrackFlowWalker(useConntrack bool, args ...string) flowWalker {
 	return result
 }
 
-// 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.Errorf("conntrack error: %v", err)
-	}
-
-	log.Errorf("conntrack: failed to find module %s", conntrackModule)
-	return false
+// IsConntrackSupported returns true if conntrack is suppported by the kernel
+var IsConntrackSupported = func(procRoot string) error {
+	procFile := filepath.Join(procRoot, procFileToCheck)
+	_, err := os.Stat(procFile)
+	return err
 }
 
 func (c *conntrackWalker) loop() {

probe/endpoint/conntrack_internal_test.go

@@ -73,11 +73,11 @@ func addIndependant(f *flow, id int64, state string) *meta {
 }
 
 func TestConntracker(t *testing.T) {
-	oldExecCmd, oldConntrackPresent := exec.Command, ConntrackModulePresent
-	defer func() { exec.Command, ConntrackModulePresent = oldExecCmd, oldConntrackPresent }()
+	oldExecCmd, oldIsConntrackSupported := exec.Command, IsConntrackSupported
+	defer func() { exec.Command, IsConntrackSupported = oldExecCmd, oldIsConntrackSupported }()
 
-	ConntrackModulePresent = func() bool {
-		return true
+	IsConntrackSupported = func(_ string) error {
+		return nil
 	}
 
 	first := true
@@ -91,7 +91,7 @@ func TestConntracker(t *testing.T) {
 		return testexec.NewMockCmd(reader)
 	}
 
-	flowWalker := newConntrackFlowWalker(true)
+	flowWalker := newConntrackFlowWalker(true, "")
 	defer flowWalker.stop()
 
 	// First write out some empty xml for the existing connections

probe/endpoint/reporter.go

@@ -52,14 +52,14 @@ 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, spyProcs, useConntrack, walkProc bool, scanner procspy.ConnectionScanner) *Reporter {
+func NewReporter(hostID, hostName string, spyProcs, useConntrack, walkProc bool, procRoot string, scanner procspy.ConnectionScanner) *Reporter {
 	return &Reporter{
 		hostID:          hostID,
 		hostName:        hostName,
 		spyProcs:        spyProcs,
 		walkProc:        walkProc,
-		flowWalker:      newConntrackFlowWalker(useConntrack),
-		natMapper:       makeNATMapper(newConntrackFlowWalker(useConntrack, "--any-nat")),
+		flowWalker:      newConntrackFlowWalker(useConntrack, procRoot),
+		natMapper:       makeNATMapper(newConntrackFlowWalker(useConntrack, procRoot, "--any-nat")),
 		reverseResolver: newReverseResolver(),
 		scanner:         scanner,
 	}

probe/endpoint/reporter_test.go

@@ -69,7 +69,7 @@ func TestSpyNoProcesses(t *testing.T) {
 	)
 
 	scanner := procspy.FixedScanner(fixConnections)
-	reporter := endpoint.NewReporter(nodeID, nodeName, false, false, false, scanner)
+	reporter := endpoint.NewReporter(nodeID, nodeName, false, false, false, "", scanner)
 	r, _ := reporter.Report()
 	//buf, _ := json.MarshalIndent(r, "", "    ")
 	//t.Logf("\n%s\n", buf)
@@ -86,7 +86,7 @@ func TestSpyWithProcesses(t *testing.T) {
 	)
 
 	scanner := procspy.FixedScanner(fixConnectionsWithProcesses)
-	reporter := endpoint.NewReporter(nodeID, nodeName, true, false, true, scanner)
+	reporter := endpoint.NewReporter(nodeID, nodeName, true, false, true, "", scanner)
 	r, _ := reporter.Report()
 	// buf, _ := json.MarshalIndent(r, "", "    ") ; t.Logf("\n%s\n", buf)
 

prog/probe.go

@@ -145,7 +145,7 @@ func probeMain(flags probeFlags) {
 		p.AddReporter(process.NewReporter(processCache, hostID, process.GetDeltaTotalJiffies))
 	}
 
-	endpointReporter := endpoint.NewReporter(hostID, hostName, flags.spyProcs, flags.useConntrack, flags.procEnabled, scanner)
+	endpointReporter := endpoint.NewReporter(hostID, hostName, flags.spyProcs, flags.useConntrack, flags.procEnabled, flags.procRoot, scanner)
 	defer endpointReporter.Stop()
 	p.AddReporter(endpointReporter)