Make connection causality detection more reliable.

experimental/tracer/ptrace/ptracer.go

@@ -301,6 +301,11 @@ func pidForTid(tid int) (pid int, err error) {
 }
 
 func (t *PTracer) thread(tid int) (*thread, error) {
+	// unfortunately we can't propage fd affinitiy as we
+	// can reliably determin who clone'd new threads, as
+	// the pids returned by the ptrace calls are in the process'
+	// namespace, not ours.
+
 	thread, ok := t.threads[tid]
 	if !ok {
 		pid, err := pidForTid(tid)

experimental/tracer/ptrace/thread.go

@@ -43,14 +43,17 @@ type thread struct {
 	callRegs   syscall.PtraceRegs
 	resultRegs syscall.PtraceRegs
 
-	currentConnection *Fd // current incoming connection
+	currentIncoming map[int]*Fd
+	currentOutgoing map[int]*Fd
 }
 
 func newThread(pid int, process *process, tracer *PTracer) *thread {
 	t := &thread{
-		tid:     pid,
-		process: process,
-		tracer:  tracer,
+		tid:             pid,
+		process:         process,
+		tracer:          tracer,
+		currentIncoming: map[int]*Fd{},
+		currentOutgoing: map[int]*Fd{},
 	}
 	return t
 }
@@ -187,9 +190,6 @@ func (t *thread) handleConnect(call, result *syscall.PtraceRegs) {
 	}
 
 	t.process.newFd(connection)
-	if t.currentConnection != nil {
-		t.currentConnection.Children = append(t.currentConnection.Children, connection)
-	}
 
 	t.logf("Made connection from %s:%d -> %s:%d on fd %d",
 		connection.ToAddr, connection.ToPort, connection.FromAddr,
@@ -205,16 +205,27 @@ func (t *thread) handleClose(call, result *syscall.PtraceRegs) {
 	}
 
 	fd.close()
-	delete(t.process.fds, fdNum)
-
-	// clear current connection if we just closed it!
-	if t.currentConnection != nil && t.currentConnection.fd == fdNum {
-		t.currentConnection = nil
-	}
 
 	// if this connection was incoming, add it to 'the registry'
 	if fd.direction == incoming {
+		for _, outgoing := range t.currentOutgoing {
+			t.logf("Fd %d caused %d", fd.fd, outgoing.fd)
+			fd.Children = append(fd.Children, outgoing)
+		}
+
 		t.tracer.store.RecordConnection(t.process.pid, fd)
+	} else {
+		for _, incoming := range t.currentIncoming {
+			t.logf("Fd %d caused %d", incoming.fd, fd.fd)
+			incoming.Children = append(incoming.Children, fd)
+		}
+	}
+
+	// now make sure we've remove it from everywhere
+	delete(t.process.fds, fdNum)
+	for _, thread := range t.process.threads {
+		delete(thread.currentIncoming, fdNum)
+		delete(t.currentOutgoing, fdNum)
 	}
 }
 
@@ -228,7 +239,9 @@ func (t *thread) handleIO(call, result *syscall.PtraceRegs) {
 
 	if fd.direction == incoming {
 		t.logf("IO on incoming connection %d; setting affinity", fdNum)
-		t.currentConnection = fd
+		t.currentIncoming[fdNum] = fd
+	} else {
+		t.currentOutgoing[fdNum] = fd
 	}
 }
 

experimental/tracer/ui/index.html

@@ -32,12 +32,16 @@
       Handlebars.registerHelper('duration', function(input) {
         var ms = input.Stop - input.Start
         if (ms < 60000) {
-          return sprintf("%0.2fs", ms / 1000)
+          return new Handlebars.SafeString(sprintf("%0.2fs", ms / 1000));
         }
         var ds = moment.duration(ms).humanize();
         return new Handlebars.SafeString(ds);
       });
 
+      Handlebars.registerHelper('count', function(input) {
+        return new Handlebars.SafeString(input === null ? '0' : sprintf("%d", input.len));
+      });
+
       function render() {
         var template = $('script#process-template').text();
         template = Handlebars.compile(template);
@@ -138,7 +142,7 @@
               {{#traces}}
               <tr><td>{{ts Root.Start}}</td><td>{{duration Root}}</td>
                 <td>{{PID}}</td><td>{{Root.FromAddr}}:{{Root.FromPort}}</td>
-                <td>{{Root.ToAddr}}:{{Root.ToPort}}</td></tr>
+                <td>{{Root.ToAddr}}:{{Root.ToPort}}</td><td>{{count Root.Children}}</tr>
               {{/traces}}
             </tbody>
           </table>