Merge pull request #1208 from weaveworks/1016-host-level-ttys

host-level ttys

common/xfer/pipes.go

@@ -27,7 +27,16 @@ type pipe struct {
 	onClose         func()
 }
 
-// NewPipe makes a new... pipe.
+// NewPipeFromEnds makes a new pipe specifying its ends
+func NewPipeFromEnds(local io.ReadWriter, remote io.ReadWriter) Pipe {
+	return &pipe{
+		port:      local,
+		starboard: remote,
+		quit:      make(chan struct{}),
+	}
+}
+
+// NewPipe makes a new pipe
 func NewPipe() Pipe {
 	r1, w1 := io.Pipe()
 	r2, w2 := io.Pipe()

integration/410_container_control_test.sh

@@ -13,8 +13,6 @@ wait_for_containers $HOST1 60 alpine
 
 assert "docker_on $HOST1 inspect --format='{{.State.Running}}' alpine" "true"
 PROBEID=$(docker_on $HOST1 logs weavescope 2>&1 | grep "probe starting" | sed -n 's/^.*ID \([0-9a-f]*\)$/\1/p')
-HOSTID=$(echo $HOST1 | cut -d"." -f1)
-
 
 # Execute 'echo foo' in a container tty and check its output
 PIPEID=$(curl -s -f -X POST  "http://$HOST1:4040/api/control/$PROBEID/$CID;<container>/docker_exec_container" | jq -r '.pipe' )

integration/420_host_control_test.sh

@@ -0,0 +1,19 @@
+#! /bin/bash
+
+. ./config.sh
+
+start_suite "Test host controls"
+
+weave_on $HOST1 launch
+scope_on $HOST1 launch
+
+sleep 10
+
+PROBEID=$(docker_on $HOST1 logs weavescope 2>&1 | grep "probe starting" | sed -n 's/^.*ID \([0-9a-f]*\)$/\1/p')
+HOSTID=$($SSH $HOST1 hostname)
+
+# Execute 'echo foo' in the host tty and check its output
+PIPEID=$(curl -s -f -X POST  "http://$HOST1:4040/api/control/$PROBEID/$HOSTID;<host>/host_exec" | jq -r '.pipe' )
+assert "(sleep 1 && echo \"PS1=''; echo foo\" && sleep 1) | wscat -b 'ws://$HOST1:4040/api/pipe/$PIPEID' | col -pb | tail -n 1" "foo\n"
+
+scope_end_suite

probe/controls/pipes.go

@@ -2,6 +2,7 @@ package controls
 
 import (
 	"fmt"
+	"io"
 	"math/rand"
 
 	"github.com/weaveworks/scope/common/xfer"
@@ -21,11 +22,10 @@ type pipe struct {
 	client    PipeClient
 }
 
-// NewPipe creats a new pipe and connects it to the app.
-var NewPipe = func(c PipeClient, appID string) (string, xfer.Pipe, error) {
+func newPipe(p xfer.Pipe, c PipeClient, appID string) (string, xfer.Pipe, error) {
 	pipeID := fmt.Sprintf("pipe-%d", rand.Int63())
 	pipe := &pipe{
-		Pipe:   xfer.NewPipe(),
+		Pipe:   p,
 		appID:  appID,
 		id:     pipeID,
 		client: c,
@@ -36,6 +36,16 @@ var NewPipe = func(c PipeClient, appID string) (string, xfer.Pipe, error) {
 	return pipeID, pipe, nil
 }
 
+// NewPipe creates a new pipe and connects it to the app.
+var NewPipe = func(c PipeClient, appID string) (string, xfer.Pipe, error) {
+	return newPipe(xfer.NewPipe(), c, appID)
+}
+
+// NewPipeFromEnds creates a new pipe from its ends and connects it to the app.
+func NewPipeFromEnds(local, remote io.ReadWriter, c PipeClient, appID string) (string, xfer.Pipe, error) {
+	return newPipe(xfer.NewPipeFromEnds(local, remote), c, appID)
+}
+
 func (p *pipe) Close() error {
 	err1 := p.Pipe.Close()
 	err2 := p.client.PipeClose(p.appID, p.id)

probe/docker/controls.go

@@ -10,7 +10,7 @@ import (
 	"github.com/weaveworks/scope/report"
 )
 
-// Control IDs used by the docker intergation.
+// Control IDs used by the docker integration.
 const (
 	StopContainer    = "docker_stop_container"
 	StartContainer   = "docker_start_container"

probe/docker/reporter.go

@@ -102,7 +102,7 @@ func (r *Reporter) containerTopology(localAddrs []net.IP) report.Topology {
 	})
 	result.Controls.AddControl(report.Control{
 		ID:    ExecContainer,
-		Human: "Exec /bin/sh",
+		Human: "Exec shell",
 		Icon:  "fa-terminal",
 	})
 

probe/host/controls.go

@@ -0,0 +1,58 @@
+package host
+
+import (
+	"os/exec"
+
+	log "github.com/Sirupsen/logrus"
+	"github.com/kr/pty"
+
+	"github.com/weaveworks/scope/common/xfer"
+	"github.com/weaveworks/scope/probe/controls"
+)
+
+// Control IDs used by the host integration.
+const (
+	ExecHost = "host_exec"
+)
+
+func (r *Reporter) registerControls() {
+	controls.Register(ExecHost, r.execHost)
+}
+
+func (*Reporter) deregisterControls() {
+	controls.Rm(ExecHost)
+}
+
+func (r *Reporter) execHost(req xfer.Request) xfer.Response {
+	cmd := exec.Command(r.hostShellCmd[0], r.hostShellCmd[1:]...)
+	cmd.Env = []string{"TERM=xterm"}
+	ptyPipe, err := pty.Start(cmd)
+	if err != nil {
+		return xfer.ResponseError(err)
+	}
+
+	id, pipe, err := controls.NewPipeFromEnds(nil, ptyPipe, r.pipes, req.AppID)
+	if err != nil {
+		return xfer.ResponseError(err)
+	}
+	pipe.OnClose(func() {
+		if err := cmd.Process.Kill(); err != nil {
+			log.Errorf("Error stopping host shell: %v", err)
+		}
+		if err := ptyPipe.Close(); err != nil {
+			log.Errorf("Error closing host shell's pty: %v", err)
+		}
+		log.Info("Host shell closed.")
+	})
+	go func() {
+		if err := cmd.Wait(); err != nil {
+			log.Errorf("Error waiting on host shell: %v", err)
+		}
+		pipe.Close()
+	}()
+
+	return xfer.Response{
+		Pipe:   id,
+		RawTTY: true,
+	}
+}

probe/host/controls_darwin.go

@@ -0,0 +1,5 @@
+package host
+
+func getHostShellCmd() []string {
+	return []string{"/bin/bash"}
+}

probe/host/controls_linux.go

@@ -0,0 +1,88 @@
+package host
+
+import (
+	"bytes"
+	"os/exec"
+	"strings"
+	"syscall"
+
+	log "github.com/Sirupsen/logrus"
+	"github.com/willdonnelly/passwd"
+)
+
+func getHostShellCmd() []string {
+	if isProbeContainerized() {
+		// Escape the container namespaces and jump into the ones from
+		// the host's init process.
+		// Note: There should be no need to enter into the host network
+		// and PID namespace because we should already already be there
+		// but it doesn't hurt.
+		readPasswdCmd := []string{"/usr/bin/nsenter", "-t1", "-m", "--no-fork", "cat", "/etc/passwd"}
+		uid, gid, shell := getRootUserDetails(readPasswdCmd)
+		return []string{
+			"/usr/bin/nsenter", "-t1", "-m", "-i", "-n", "-p", "--no-fork",
+			"--setuid", uid,
+			"--setgid", gid,
+			shell,
+		}
+	}
+
+	_, _, shell := getRootUserDetails([]string{"cat", "/etc/passwd"})
+	return []string{shell}
+}
+
+func getRootUserDetails(readPasswdCmd []string) (uid, gid, shell string) {
+	uid = "0"
+	gid = "0"
+	shell = "/bin/sh"
+
+	cmd := exec.Command(readPasswdCmd[0], readPasswdCmd[1:]...)
+	cmdBuffer := &bytes.Buffer{}
+	cmd.Stdout = cmdBuffer
+	if err := cmd.Run(); err != nil {
+		log.Warnf(
+			"getRootUserDetails(): error running read passwd command %q: %s",
+			strings.Join(readPasswdCmd, " "),
+			err,
+		)
+		return
+	}
+
+	entries, err := passwd.ParseReader(cmdBuffer)
+	if err != nil {
+		log.Warnf("getRootUserDetails(): error parsing passwd: %s", err)
+		return
+	}
+
+	entry, ok := entries["root"]
+	if !ok {
+		log.Warnf("getRootUserDetails(): no root entry in passwd")
+		return
+	}
+
+	return entry.Uid, entry.Gid, entry.Shell
+}
+
+func isProbeContainerized() bool {
+	// Figure out whether we are running in a container by checking if our
+	// mount namespace matches the one from init process. This works
+	// because, when containerized, the Scope probes run in the host's PID
+	// namespace (and if they weren't due to a configuration problem, we
+	// wouldn't have a way to escape the container anyhow).
+	var statT syscall.Stat_t
+
+	path := "/proc/self/ns/mnt"
+	if err := syscall.Stat(path, &statT); err != nil {
+		log.Warnf("isProbeContainerized(): stat() error on %q: %s", path, err)
+		return false
+	}
+	selfMountNamespaceID := statT.Ino
+
+	path = "/proc/1/ns/mnt"
+	if err := syscall.Stat(path, &statT); err != nil {
+		log.Warnf("isProbeContainerized(): stat() error on %q: %s", path, err)
+		return false
+	}
+
+	return selfMountNamespaceID != statT.Ino
+}

probe/host/reporter.go

@@ -6,6 +6,7 @@ import (
 	"time"
 
 	"github.com/weaveworks/scope/common/mtime"
+	"github.com/weaveworks/scope/probe/controls"
 	"github.com/weaveworks/scope/report"
 )
 
@@ -34,17 +35,25 @@ const (
 
 // Reporter generates Reports containing the host topology.
 type Reporter struct {
-	hostID   string
-	hostName string
+	hostID       string
+	hostName     string
+	probeID      string
+	pipes        controls.PipeClient
+	hostShellCmd []string
 }
 
 // NewReporter returns a Reporter which produces a report containing host
 // topology for this host.
-func NewReporter(hostID, hostName string) *Reporter {
-	return &Reporter{
-		hostID:   hostID,
-		hostName: hostName,
+func NewReporter(hostID, hostName, probeID string, pipes controls.PipeClient) *Reporter {
+	r := &Reporter{
+		hostID:       hostID,
+		hostName:     hostName,
+		probeID:      probeID,
+		pipes:        pipes,
+		hostShellCmd: getHostShellCmd(),
 	}
+	r.registerControls()
+	return r
 }
 
 // Name of this reporter, for metrics gathering
@@ -98,6 +107,7 @@ func (r *Reporter) Report() (report.Report, error) {
 	memoryUsage, max := GetMemoryUsageBytes()
 	metrics[MemoryUsage] = report.MakeMetric().Add(now, memoryUsage).WithMax(max)
 
+	metadata := map[string]string{report.ControlProbeID: r.probeID}
 	rep.Host.AddNode(report.MakeHostNodeID(r.hostID), report.MakeNodeWith(map[string]string{
 		Timestamp:     mtime.Now().UTC().Format(time.RFC3339Nano),
 		HostName:      r.hostName,
@@ -106,7 +116,18 @@ func (r *Reporter) Report() (report.Report, error) {
 		Uptime:        uptime.String(),
 	}).WithSets(report.EmptySets.
 		Add(LocalNetworks, report.MakeStringSet(localCIDRs...)),
-	).WithMetrics(metrics))
+	).WithMetrics(metrics).WithControls(ExecHost).WithLatests(metadata))
+
+	rep.Host.Controls.AddControl(report.Control{
+		ID:    ExecHost,
+		Human: "Exec shell",
+		Icon:  "fa-terminal",
+	})
 
 	return rep, nil
 }
+
+// Stop stops the reporter.
+func (r *Reporter) Stop() {
+	r.deregisterControls()
+}

probe/host/reporter_test.go

@@ -18,6 +18,7 @@ func TestReporter(t *testing.T) {
 		network   = "192.168.0.0/16"
 		hostID    = "hostid"
 		hostname  = "hostname"
+		probeID   = "abcdeadbeef"
 		timestamp = time.Now()
 		metrics   = report.Metrics{
 			host.Load1:       report.MakeMetric().Add(timestamp, 1.0),
@@ -57,7 +58,7 @@ func TestReporter(t *testing.T) {
 	host.GetMemoryUsageBytes = func() (float64, float64) { return 60.0, 100.0 }
 	host.GetLocalNetworks = func() ([]*net.IPNet, error) { return []*net.IPNet{ipnet}, nil }
 
-	rpt, err := host.NewReporter(hostID, hostname).Report()
+	rpt, err := host.NewReporter(hostID, hostname, probeID, nil).Report()
 	if err != nil {
 		t.Fatal(err)
 	}

prog/probe.go

@@ -140,9 +140,11 @@ func probeMain() {
 
 	p := probe.New(*spyInterval, *publishInterval, clients)
 	p.AddTicker(processCache)
+	hostReporter := host.NewReporter(hostID, hostName, probeID, clients)
+	defer hostReporter.Stop()
 	p.AddReporter(
 		endpointReporter,
-		host.NewReporter(hostID, hostName),
+		hostReporter,
 		process.NewReporter(processCache, hostID, process.GetDeltaTotalJiffies),
 	)
 	p.AddTagger(probe.NewTopologyTagger(), host.NewTagger(hostID))