diff --git a/app/pipes_internal_test.go b/app/pipes_internal_test.go index 5b3643566..312e02bbe 100644 --- a/app/pipes_internal_test.go +++ b/app/pipes_internal_test.go @@ -84,12 +84,8 @@ func TestPipeClose(t *testing.T) { } defer client.Stop() - oldClient := controls.Client - defer func() { controls.Client = oldClient }() - controls.Client = adapter{client} - // this is the probe end of the pipe - pipeID, pipe, err := controls.NewPipe("appid") + pipeID, pipe, err := controls.NewPipe(adapter{client}, "appid") if err != nil { t.Fatal(err) } diff --git a/probe/controls/pipes.go b/probe/controls/pipes.go index 9b7016dc0..eec4da28f 100644 --- a/probe/controls/pipes.go +++ b/probe/controls/pipes.go @@ -7,8 +7,8 @@ import ( "github.com/weaveworks/scope/xfer" ) -// Client is the thing the probe uses to make pipe connections. -var Client interface { +// PipeClient is the type of the thing the probe uses to make pipe connections. +type PipeClient interface { PipeConnection(string, string, xfer.Pipe) error PipeClose(string, string) error } @@ -18,17 +18,19 @@ var Client interface { type pipe struct { xfer.Pipe id, appID string + client PipeClient } // NewPipe creats a new pipe and connects it to the app. -var NewPipe = func(appID string) (string, xfer.Pipe, error) { +var NewPipe = func(c PipeClient, appID string) (string, xfer.Pipe, error) { pipeID := fmt.Sprintf("pipe-%d", rand.Int63()) pipe := &pipe{ - Pipe: xfer.NewPipe(), - appID: appID, - id: pipeID, + Pipe: xfer.NewPipe(), + appID: appID, + id: pipeID, + client: c, } - if err := Client.PipeConnection(appID, pipeID, pipe.Pipe); err != nil { + if err := c.PipeConnection(appID, pipeID, pipe.Pipe); err != nil { return "", nil, err } return pipeID, pipe, nil @@ -36,7 +38,7 @@ var NewPipe = func(appID string) (string, xfer.Pipe, error) { func (p *pipe) Close() error { err1 := p.Pipe.Close() - err2 := Client.PipeClose(p.appID, p.id) + err2 := p.client.PipeClose(p.appID, p.id) if err1 != nil { return err1 } diff --git a/probe/docker/controls.go b/probe/docker/controls.go index 7d1270014..fdb63a19a 100644 --- a/probe/docker/controls.go +++ b/probe/docker/controls.go @@ -55,7 +55,7 @@ func (r *registry) attachContainer(containerID string, req xfer.Request) xfer.Re } hasTTY := c.HasTTY() - id, pipe, err := controls.NewPipe(req.AppID) + id, pipe, err := controls.NewPipe(r.pipes, req.AppID) if err != nil { xfer.ResponseError(err) } @@ -106,7 +106,7 @@ func (r *registry) execContainer(containerID string, req xfer.Request) xfer.Resp xfer.ResponseError(err) } - id, pipe, err := controls.NewPipe(req.AppID) + id, pipe, err := controls.NewPipe(r.pipes, req.AppID) if err != nil { xfer.ResponseError(err) } diff --git a/probe/docker/controls_test.go b/probe/docker/controls_test.go index cca2d5586..67a658958 100644 --- a/probe/docker/controls_test.go +++ b/probe/docker/controls_test.go @@ -18,7 +18,7 @@ import ( func TestControls(t *testing.T) { mdc := newMockClient() setupStubs(mdc, func() { - registry, _ := docker.NewRegistry(10 * time.Second) + registry, _ := docker.NewRegistry(10*time.Second, nil) defer registry.Stop() for _, tc := range []struct{ command, result string }{ @@ -52,13 +52,13 @@ func (mockPipe) OnClose(func()) {} func TestPipes(t *testing.T) { oldNewPipe := controls.NewPipe defer func() { controls.NewPipe = oldNewPipe }() - controls.NewPipe = func(_ string) (string, xfer.Pipe, error) { + controls.NewPipe = func(_ controls.PipeClient, _ string) (string, xfer.Pipe, error) { return "pipeid", mockPipe{}, nil } mdc := newMockClient() setupStubs(mdc, func() { - registry, _ := docker.NewRegistry(10 * time.Second) + registry, _ := docker.NewRegistry(10*time.Second, nil) defer registry.Stop() test.Poll(t, 100*time.Millisecond, true, func() interface{} { diff --git a/probe/docker/registry.go b/probe/docker/registry.go index 9a040f1c1..bf805eb48 100644 --- a/probe/docker/registry.go +++ b/probe/docker/registry.go @@ -6,6 +6,8 @@ import ( "time" docker_client "github.com/fsouza/go-dockerclient" + + "github.com/weaveworks/scope/probe/controls" ) // Consts exported for testing. @@ -43,6 +45,7 @@ type registry struct { quit chan chan struct{} interval time.Duration client Client + pipes controls.PipeClient watchers []ContainerUpdateWatcher containers map[string]Container @@ -73,7 +76,7 @@ func newDockerClient(endpoint string) (Client, error) { } // NewRegistry returns a usable Registry. Don't forget to Stop it. -func NewRegistry(interval time.Duration) (Registry, error) { +func NewRegistry(interval time.Duration, pipes controls.PipeClient) (Registry, error) { client, err := NewDockerClientStub(endpoint) if err != nil { return nil, err @@ -85,6 +88,7 @@ func NewRegistry(interval time.Duration) (Registry, error) { images: map[string]*docker_client.APIImages{}, client: client, + pipes: pipes, interval: interval, quit: make(chan chan struct{}), } diff --git a/probe/docker/registry_test.go b/probe/docker/registry_test.go index 1dda7cc6e..caba06996 100644 --- a/probe/docker/registry_test.go +++ b/probe/docker/registry_test.go @@ -242,7 +242,7 @@ func allImages(r docker.Registry) []*client.APIImages { func TestRegistry(t *testing.T) { mdc := newMockClient() setupStubs(mdc, func() { - registry, _ := docker.NewRegistry(10 * time.Second) + registry, _ := docker.NewRegistry(10*time.Second, nil) defer registry.Stop() runtime.Gosched() @@ -265,7 +265,7 @@ func TestRegistry(t *testing.T) { func TestLookupByPID(t *testing.T) { mdc := newMockClient() setupStubs(mdc, func() { - registry, _ := docker.NewRegistry(10 * time.Second) + registry, _ := docker.NewRegistry(10*time.Second, nil) defer registry.Stop() want := docker.Container(&mockContainer{container1}) @@ -282,7 +282,7 @@ func TestLookupByPID(t *testing.T) { func TestRegistryEvents(t *testing.T) { mdc := newMockClient() setupStubs(mdc, func() { - registry, _ := docker.NewRegistry(10 * time.Second) + registry, _ := docker.NewRegistry(10*time.Second, nil) defer registry.Stop() runtime.Gosched() diff --git a/prog/probe.go b/prog/probe.go index da19ea684..a76ea0a84 100644 --- a/prog/probe.go +++ b/prog/probe.go @@ -114,7 +114,6 @@ func probeMain() { ) }) defer clients.Stop() - controls.Client = clients resolver := xfer.NewStaticResolver(targets, clients.Set) defer resolver.Stop() @@ -136,7 +135,7 @@ func probeMain() { if err := report.AddLocalBridge(*dockerBridge); err != nil { log.Printf("Docker: problem with bridge %s: %v", *dockerBridge, err) } - if registry, err := docker.NewRegistry(*dockerInterval); err == nil { + 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))