Distinguish between reporting probes and controlling probes

probe/docker/controls.go

@@ -57,7 +57,7 @@ func (r *registry) attachContainer(containerID string, req xfer.Request) xfer.Re
 	hasTTY := c.HasTTY()
 	id, pipe, err := controls.NewPipe(r.pipes, req.AppID)
 	if err != nil {
-		xfer.ResponseError(err)
+		return xfer.ResponseError(err)
 	}
 	local, _ := pipe.Ends()
 	cw, err := r.client.AttachToContainerNonBlocking(docker_client.AttachToContainerOptions{
@@ -103,12 +103,12 @@ func (r *registry) execContainer(containerID string, req xfer.Request) xfer.Resp
 		Container:    containerID,
 	})
 	if err != nil {
-		xfer.ResponseError(err)
+		return xfer.ResponseError(err)
 	}
 
 	id, pipe, err := controls.NewPipe(r.pipes, req.AppID)
 	if err != nil {
-		xfer.ResponseError(err)
+		return xfer.ResponseError(err)
 	}
 	local, _ := pipe.Ends()
 	cw, err := r.client.StartExecNonBlocking(exec.ID, docker_client.StartExecOptions{

probe/docker/reporter.go

@@ -20,14 +20,16 @@ const (
 type Reporter struct {
 	registry Registry
 	hostID   string
+	probeID  string
 	probe    *probe.Probe
 }
 
 // NewReporter makes a new Reporter
-func NewReporter(registry Registry, hostID string, probe *probe.Probe) *Reporter {
+func NewReporter(registry Registry, hostID string, probeID string, probe *probe.Probe) *Reporter {
 	reporter := &Reporter{
 		registry: registry,
 		hostID:   hostID,
+		probeID:  probeID,
 		probe:    probe,
 	}
 	registry.WatchContainerUpdates(reporter.ContainerUpdated)
@@ -103,9 +105,11 @@ func (r *Reporter) containerTopology(localAddrs []net.IP) report.Topology {
 		Icon:  "fa-terminal",
 	})
 
+	metadata := map[string]string{report.ControlProbeID: r.probeID}
+
 	r.registry.WalkContainers(func(c Container) {
 		nodeID := report.MakeContainerNodeID(c.ID())
-		result.AddNode(nodeID, c.GetNode(r.hostID, localAddrs))
+		result.AddNode(nodeID, c.GetNode(r.hostID, localAddrs).WithLatests(metadata))
 	})
 
 	return result

probe/docker/reporter_test.go

@@ -51,7 +51,7 @@ var (
 
 func TestReporter(t *testing.T) {
 	containerImageNodeID := report.MakeContainerImageNodeID("baz")
-	rpt, err := docker.NewReporter(mockRegistryInstance, "host1", nil).Report()
+	rpt, err := docker.NewReporter(mockRegistryInstance, "host1", "probeID", nil).Report()
 	if err != nil {
 		t.Fatal(err)
 	}

prog/probe.go

@@ -155,7 +155,7 @@ func probeMain() {
 		if registry, err := docker.NewRegistry(*dockerInterval, clients); err == nil {
 			defer registry.Stop()
 			p.AddTagger(docker.NewTagger(registry, processCache))
-			p.AddReporter(docker.NewReporter(registry, hostID, p))
+			p.AddReporter(docker.NewReporter(registry, hostID, probeID, p))
 		} else {
 			log.Errorf("Docker: failed to start registry: %v", err)
 		}

render/detailed/node.go

@@ -101,7 +101,10 @@ func controlsFor(topology report.Topology, nodeID string) []ControlInstance {
 
 	for _, id := range node.Controls.Controls {
 		if control, ok := topology.Controls[id]; ok {
-			probeID, _ := node.Latest.Lookup(report.ProbeID)
+			probeID, ok := node.Latest.Lookup(report.ControlProbeID)
+			if !ok {
+				continue
+			}
 			result = append(result, ControlInstance{
 				ProbeID: probeID,
 				NodeID:  nodeID,

report/report.go

@@ -209,4 +209,6 @@ const (
 	HostNodeID = "host_node_id"
 	// ProbeID is the random ID of the probe which generated the specific node.
 	ProbeID = "probe_id"
+	// ControlProbeID is the random ID of the probe which controls the specific node.
+	ControlProbeID = "control_probe_id"
 )