Merge pull request #1043 from weaveworks/1040-docker-exec-crash

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

@@ -50,8 +50,10 @@ var (
 )
 
 func TestReporter(t *testing.T) {
+	var controlProbeID = "a1b2c3d4"
+
 	containerImageNodeID := report.MakeContainerImageNodeID("baz")
-	rpt, err := docker.NewReporter(mockRegistryInstance, "host1", nil).Report()
+	rpt, err := docker.NewReporter(mockRegistryInstance, "host1", controlProbeID, nil).Report()
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -65,9 +67,10 @@ func TestReporter(t *testing.T) {
 		}
 
 		for k, want := range map[string]string{
-			docker.ContainerID:   "ping",
-			docker.ContainerName: "pong",
-			docker.ImageID:       "baz",
+			docker.ContainerID:    "ping",
+			docker.ContainerName:  "pong",
+			docker.ImageID:        "baz",
+			report.ControlProbeID: controlProbeID,
 		} {
 			if have, ok := node.Latest.Lookup(k); !ok || have != want {
 				t.Errorf("Expected container %s latest %q: %q, got %q", containerNodeID, k, want, have)

probe/host/tagger.go

@@ -9,15 +9,13 @@ import (
 // in every topology to an origin host node in the host topology.
 type Tagger struct {
 	hostNodeID string
-	probeID    string
 }
 
 // NewTagger tags each node with a foreign key linking it to its origin host
 // in the host topology.
-func NewTagger(hostID, probeID string) Tagger {
+func NewTagger(hostID string) Tagger {
 	return Tagger{
 		hostNodeID: report.MakeHostNodeID(hostID),
-		probeID:    probeID,
 	}
 }
 
@@ -27,11 +25,8 @@ func (Tagger) Name() string { return "Host" }
 // Tag implements Tagger.
 func (t Tagger) Tag(r report.Report) (report.Report, error) {
 	var (
-		metadata = map[string]string{
-			report.HostNodeID: t.hostNodeID,
-			report.ProbeID:    t.probeID,
-		}
-		parents = report.EmptySets.Add(report.Host, report.MakeStringSet(t.hostNodeID))
+		metadata = map[string]string{report.HostNodeID: t.hostNodeID}
+		parents  = report.EmptySets.Add(report.Host, report.MakeStringSet(t.hostNodeID))
 	)
 
 	// Explicity don't tag Endpoints and Addresses - These topologies include pseudo nodes,

probe/host/tagger_test.go

@@ -10,14 +10,13 @@ import (
 func TestTagger(t *testing.T) {
 	var (
 		hostID         = "foo"
-		probeID        = "a1b2c3d4"
 		endpointNodeID = report.MakeEndpointNodeID(hostID, "1.2.3.4", "56789") // hostID ignored
 		node           = report.MakeNodeWith(map[string]string{"foo": "bar"})
 	)
 
 	r := report.MakeReport()
 	r.Process.AddNode(endpointNodeID, node)
-	rpt, _ := host.NewTagger(hostID, probeID).Tag(r)
+	rpt, _ := host.NewTagger(hostID).Tag(r)
 	have := rpt.Process.Nodes[endpointNodeID].Copy()
 
 	// It should now have the host ID
@@ -26,11 +25,6 @@ func TestTagger(t *testing.T) {
 		t.Errorf("Expected %q got %q", wantHostID, report.MakeHostNodeID(hostID))
 	}
 
-	// It should now have the probe ID
-	if haveProbeID, ok := have.Latest.Lookup(report.ProbeID); !ok || haveProbeID != probeID {
-		t.Errorf("Expected %q got %q", probeID, haveProbeID)
-	}
-
 	// It should still have the other keys
 	want := "bar"
 	if have, ok := have.Latest.Lookup("foo"); !ok || have != want {

prog/probe.go

@@ -146,7 +146,7 @@ func probeMain() {
 		host.NewReporter(hostID, hostName),
 		process.NewReporter(processCache, hostID, process.GetDeltaTotalJiffies),
 	)
-	p.AddTagger(probe.NewTopologyTagger(), host.NewTagger(hostID, probeID))
+	p.AddTagger(probe.NewTopologyTagger(), host.NewTagger(hostID))
 
 	if *dockerEnabled {
 		if err := report.AddLocalBridge(*dockerBridge); err != nil {
@@ -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

@@ -207,6 +207,6 @@ const (
 	// a node in the host topology. That host node is the origin host, where
 	// the node was originally detected.
 	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"
 )