github/weave-scope
1
0
Fork 0
mirror of https://github.com/weaveworks/scope.git synced 2026-07-28 17:51:21 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #794 from weaveworks/641-xfer-mv

Browse Source 
Split and move xfer package.
This commit is contained in:
Paul Bellamy
2016-01-06 15:56:55 +00:00
parent a4f7042f3e ffa955a21b
commit 80c7436036
29 changed files with 104 additions and 93 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
app/controls.go
View File

@@ -9,7 +9,7 @@ import (
"github.com/gorilla/mux"
"github.com/weaveworks/scope/xfer"
"github.com/weaveworks/scope/common/xfer"
)
// RegisterControlRoutes registers the various control routes with a http mux.

7
app/controls_test.go
View File

@@ -12,7 +12,8 @@ import (
"github.com/gorilla/mux"
"github.com/weaveworks/scope/app"
"github.com/weaveworks/scope/xfer"
"github.com/weaveworks/scope/common/xfer"
"github.com/weaveworks/scope/probe/appclient"
)
func TestControl(t *testing.T) {
@@ -26,7 +27,7 @@ func TestControl(t *testing.T) {
t.Fatal(err)
}
probeConfig := xfer.ProbeConfig{
probeConfig := appclient.ProbeConfig{
ProbeID: "foo",
}
controlHandler := xfer.ControlHandlerFunc(func(req xfer.Request) xfer.Response {
@@ -42,7 +43,7 @@ func TestControl(t *testing.T) {
Value: "foo",
}
})
client, err := xfer.NewAppClient(probeConfig, ip+":"+port, ip+":"+port, controlHandler)
client, err := appclient.NewAppClient(probeConfig, ip+":"+port, ip+":"+port, controlHandler)
if err != nil {
t.Fatal(err)
}

2
app/pipes.go
View File

@@ -10,7 +10,7 @@ import (
"github.com/gorilla/mux"
"github.com/weaveworks/scope/common/mtime"
"github.com/weaveworks/scope/xfer"
"github.com/weaveworks/scope/common/xfer"
)
const (

9
app/pipes_internal_test.go
View File

@@ -14,9 +14,10 @@ import (
"github.com/gorilla/websocket"
"github.com/weaveworks/scope/common/mtime"
"github.com/weaveworks/scope/common/xfer"
"github.com/weaveworks/scope/probe/appclient"
"github.com/weaveworks/scope/probe/controls"
"github.com/weaveworks/scope/test"
"github.com/weaveworks/scope/xfer"
)
func TestPipeTimeout(t *testing.T) {
@@ -50,7 +51,7 @@ func TestPipeTimeout(t *testing.T) {
}
type adapter struct {
c xfer.AppClient
c appclient.AppClient
}
func (a adapter) PipeConnection(_, pipeID string, pipe xfer.Pipe) error {
@@ -75,10 +76,10 @@ func TestPipeClose(t *testing.T) {
t.Fatal(err)
}
probeConfig := xfer.ProbeConfig{
probeConfig := appclient.ProbeConfig{
ProbeID: "foo",
}
client, err := xfer.NewAppClient(probeConfig, ip+":"+port, ip+":"+port, nil)
client, err := appclient.NewAppClient(probeConfig, ip+":"+port, ip+":"+port, nil)
if err != nil {
t.Fatal(err)
}

2
app/router.go
View File

@@ -12,8 +12,8 @@ import (
"github.com/gorilla/mux"
"github.com/weaveworks/scope/common/hostname"
"github.com/weaveworks/scope/common/xfer"
"github.com/weaveworks/scope/report"
"github.com/weaveworks/scope/xfer"
)
var (

19
common/xfer/constants.go Normal file
View File

@@ -0,0 +1,19 @@
package xfer
const (
// AppPort is the default port that the app will use for its HTTP server.
// The app publishes the API and user interface, and receives reports from
// probes, on this port.
AppPort = 4040
// ScopeProbeIDHeader is the header we use to carry the probe's unique ID. The
// ID is currently set to the a random string on probe startup.
ScopeProbeIDHeader = "X-Scope-Probe-ID"
)
// Details are some generic details that can be fetched from /api
type Details struct {
ID string `json:"id"`
Version string `json:"version"`
Hostname string `json:"hostname"`
}

0
xfer/controls.go → common/xfer/controls.go
View File

0
xfer/pipes.go → common/xfer/pipes.go
View File

7
experimental/demoprobe/main.go
View File

@@ -9,10 +9,11 @@ import (
"strconv"
"time"
"github.com/weaveworks/scope/common/xfer"
"github.com/weaveworks/scope/probe/appclient"
"github.com/weaveworks/scope/probe/docker"
"github.com/weaveworks/scope/probe/process"
"github.com/weaveworks/scope/report"
"github.com/weaveworks/scope/xfer"
)
func main() {
@@ -23,7 +24,7 @@ func main() {
)
flag.Parse()
client, err := xfer.NewAppClient(xfer.ProbeConfig{
client, err := appclient.NewAppClient(appclient.ProbeConfig{
Token: "demoprobe",
ProbeID: "demoprobe",
Insecure: false,
@@ -31,7 +32,7 @@ func main() {
if err != nil {
log.Fatal(err)
}
rp := xfer.NewReportPublisher(client)
rp := appclient.NewReportPublisher(client)
rand.Seed(time.Now().UnixNano())
for range time.Tick(*publishInterval) {

7
experimental/fixprobe/main.go
View File

@@ -9,8 +9,9 @@ import (
"os"
"time"
"github.com/weaveworks/scope/common/xfer"
"github.com/weaveworks/scope/probe/appclient"
"github.com/weaveworks/scope/report"
"github.com/weaveworks/scope/xfer"
)
func main() {
@@ -34,7 +35,7 @@ func main() {
}
f.Close()
client, err := xfer.NewAppClient(xfer.ProbeConfig{
client, err := appclient.NewAppClient(appclient.ProbeConfig{
Token: "fixprobe",
ProbeID: "fixprobe",
Insecure: false,
@@ -43,7 +44,7 @@ func main() {
log.Fatal(err)
}
rp := xfer.NewReportPublisher(client)
rp := appclient.NewReportPublisher(client)
for range time.Tick(*publishInterval) {
rp.Publish(fixedReport)
}

28
xfer/app_client.go → probe/appclient/app_client.go
View File

@@ -1,4 +1,4 @@
package xfer
package appclient
import (
"encoding/json"
@@ -13,6 +13,7 @@ import (
"github.com/gorilla/websocket"
"github.com/weaveworks/scope/common/sanitize"
"github.com/weaveworks/scope/common/xfer"
)
const (
@@ -20,18 +21,11 @@ const (
maxBackoff = 60 * time.Second
)
// Details are some generic details that can be fetched from /api
type Details struct {
ID string `json:"id"`
Version string `json:"version"`
Hostname string `json:"hostname"`
}
// AppClient is a client to an app for dealing with controls.
type AppClient interface {
Details() (Details, error)
Details() (xfer.Details, error)
ControlConnection()
PipeConnection(string, Pipe)
PipeConnection(string, xfer.Pipe)
PipeClose(string) error
Publish(r io.Reader) error
Stop()
@@ -58,11 +52,11 @@ type appClient struct {
readers chan io.Reader
// For controls
control ControlHandler
control xfer.ControlHandler
}
// NewAppClient makes a new appClient.
func NewAppClient(pc ProbeConfig, hostname, target string, control ControlHandler) (AppClient, error) {
func NewAppClient(pc ProbeConfig, hostname, target string, control xfer.ControlHandler) (AppClient, error) {
httpTransport, err := pc.getHTTPTransport(hostname)
if err != nil {
return nil, err
@@ -144,8 +138,8 @@ func (c *appClient) Stop() {
}
// Details fetches the details (version, id) of the app.
func (c *appClient) Details() (Details, error) {
result := Details{}
func (c *appClient) Details() (xfer.Details, error) {
result := xfer.Details{}
req, err := c.ProbeConfig.authorizedRequest("GET", sanitize.URL("", 0, "/api")(c.target), nil)
if err != nil {
return result, err
@@ -202,7 +196,7 @@ func (c *appClient) controlConnection() (bool, error) {
conn.Close()
}()
codec := NewJSONWebsocketCodec(conn)
codec := xfer.NewJSONWebsocketCodec(conn)
server := rpc.NewServer()
if err := server.RegisterName("control", c.control); err != nil {
return false, err
@@ -271,7 +265,7 @@ func (c *appClient) Publish(r io.Reader) error {
return nil
}
func (c *appClient) pipeConnection(id string, pipe Pipe) (bool, error) {
func (c *appClient) pipeConnection(id string, pipe xfer.Pipe) (bool, error) {
headers := http.Header{}
c.ProbeConfig.authorizeHeaders(headers)
url := sanitize.URL("ws://", 0, fmt.Sprintf("/api/pipe/%s/probe", id))(c.target)
@@ -295,7 +289,7 @@ func (c *appClient) pipeConnection(id string, pipe Pipe) (bool, error) {
return false, pipe.CopyToWebsocket(remote, conn)
}
func (c *appClient) PipeConnection(id string, pipe Pipe) {
func (c *appClient) PipeConnection(id string, pipe xfer.Pipe) {
go func() {
log.Printf("Pipe %s connection to %s starting", id, c.target)
defer log.Printf("Pipe %s connection to %s exiting", id, c.target)

7
xfer/app_client_internal_test.go → probe/appclient/app_client_internal_test.go
View File

@@ -1,4 +1,4 @@
package xfer
package appclient
import (
"compress/gzip"
@@ -15,6 +15,7 @@ import (
"time"
"github.com/gorilla/handlers"
"github.com/weaveworks/scope/common/xfer"
"github.com/weaveworks/scope/report"
"github.com/weaveworks/scope/test"
)
@@ -33,7 +34,7 @@ func dummyServer(t *testing.T, expectedToken, expectedID string, expectedReport
t.Errorf("want %q, have %q", expectedToken, have)
}
if have := r.Header.Get(ScopeProbeIDHeader); expectedID != have {
if have := r.Header.Get(xfer.ScopeProbeIDHeader); expectedID != have {
t.Errorf("want %q, have %q", expectedID, have)
}
@@ -151,7 +152,7 @@ func TestAppClientPublish(t *testing.T) {
var (
id = "foobarbaz"
version = "imalittleteapot"
want = Details{ID: id, Version: version}
want = xfer.Details{ID: id, Version: version}
)
handler := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {

16
xfer/multi_client.go → probe/appclient/multi_client.go
View File

@@ -1,4 +1,4 @@
package xfer
package appclient
import (
"bytes"
@@ -10,6 +10,7 @@ import (
"strings"
"sync"
"github.com/weaveworks/scope/common/xfer"
"github.com/weaveworks/scope/report"
)
@@ -29,15 +30,22 @@ type multiClient struct {
}
type clientTuple struct {
Details
xfer.Details
AppClient
}
// Publisher is something which can send a stream of data somewhere, probably
// to a remote collector.
type Publisher interface {
Publish(io.Reader) error
Stop()
}
// MultiAppClient maintains a set of upstream apps, and ensures we have an
// AppClient for each one.
type MultiAppClient interface {
Set(hostname string, endpoints []string)
PipeConnection(appID, pipeID string, pipe Pipe) error
PipeConnection(appID, pipeID string, pipe xfer.Pipe) error
PipeClose(appID, pipeID string) error
Stop()
Publish(io.Reader) error
@@ -122,7 +130,7 @@ func (c *multiClient) withClient(appID string, f func(AppClient) error) error {
return f(client)
}
func (c *multiClient) PipeConnection(appID, pipeID string, pipe Pipe) error {
func (c *multiClient) PipeConnection(appID, pipeID string, pipe xfer.Pipe) error {
return c.withClient(appID, func(client AppClient) error {
client.PipeConnection(pipeID, pipe)
return nil

2
xfer/multi_client_internal_test.go → probe/appclient/multi_client_internal_test.go
View File

@@ -1,4 +1,4 @@
package xfer
package appclient
import (
"testing"

11
xfer/multi_client_test.go → probe/appclient/multi_client_test.go
View File

@@ -1,4 +1,4 @@
package xfer_test
package appclient_test
import (
"bytes"
@@ -6,7 +6,8 @@ import (
"runtime"
"testing"
"github.com/weaveworks/scope/xfer"
"github.com/weaveworks/scope/common/xfer"
"github.com/weaveworks/scope/probe/appclient"
)
type mockClient struct {
@@ -41,7 +42,7 @@ var (
a2 = &mockClient{id: "2"} // hostname a, app id 2
b2 = &mockClient{id: "2"} // hostname b, app id 2 (duplicate)
b3 = &mockClient{id: "3"} // hostname b, app id 3
factory = func(hostname, target string) (xfer.AppClient, error) {
factory = func(hostname, target string) (appclient.AppClient, error) {
switch target {
case "a1":
return a1, nil
@@ -66,7 +67,7 @@ func TestMultiClient(t *testing.T) {
}
)
mp := xfer.NewMultiAppClient(factory)
mp := appclient.NewMultiAppClient(factory)
defer mp.Stop()
// Add two hostnames with overlapping apps, check we don't add the same app twice
@@ -88,7 +89,7 @@ func TestMultiClient(t *testing.T) {
}
func TestMultiClientPublish(t *testing.T) {
mp := xfer.NewMultiAppClient(factory)
mp := appclient.NewMultiAppClient(factory)
defer mp.Stop()
sum := func() int { return a1.publish + a2.publish + b2.publish + b3.publish }

9
xfer/probe_config.go → probe/appclient/probe_config.go
View File

@@ -1,4 +1,4 @@
package xfer
package appclient
import (
"crypto/tls"
@@ -9,12 +9,9 @@ import (
"net/http"
"github.com/certifi/gocertifi"
"github.com/weaveworks/scope/common/xfer"
)
// ScopeProbeIDHeader is the header we use to carry the probe's unique ID. The
// ID is currently set to the a random string on probe startup.
const ScopeProbeIDHeader = "X-Scope-Probe-ID"
var certPool *x509.CertPool
func init() {
@@ -34,7 +31,7 @@ type ProbeConfig struct {
func (pc ProbeConfig) authorizeHeaders(headers http.Header) {
headers.Set("Authorization", fmt.Sprintf("Scope-Probe token=%s", pc.Token))
headers.Set(ScopeProbeIDHeader, pc.ProbeID)
headers.Set(xfer.ScopeProbeIDHeader, pc.ProbeID)
}
func (pc ProbeConfig) authorizedRequest(method string, urlStr string, body io.Reader) (*http.Request, error) {

2
xfer/report_publisher.go → probe/appclient/report_publisher.go
View File

@@ -1,4 +1,4 @@
package xfer
package appclient
import (
"bytes"

6
xfer/resolver.go → probe/appclient/resolver.go
View File

@@ -1,4 +1,4 @@
package xfer
package appclient
import (
"log"
@@ -6,6 +6,8 @@ import (
"strconv"
"strings"
"time"
"github.com/weaveworks/scope/common/xfer"
)
const (
@@ -76,7 +78,7 @@ func prepare(strs []string) []target {
continue
}
} else {
host, port = s, strconv.Itoa(AppPort)
host, port = s, strconv.Itoa(xfer.AppPort)
}
targets = append(targets, target{host, port})
}

12
xfer/resolver_internal_test.go → probe/appclient/resolver_internal_test.go
View File

@@ -1,4 +1,4 @@
package xfer
package appclient
import (
"fmt"
@@ -7,6 +7,8 @@ import (
"sync"
"testing"
"time"
"github.com/weaveworks/scope/common/xfer"
)
func TestResolver(t *testing.T) {
@@ -68,22 +70,22 @@ func TestResolver(t *testing.T) {
}
// Initial resolve should just give us IPs
assertAdd(ip1+port, fmt.Sprintf("%s:%d", ip2, AppPort))
assertAdd(ip1+port, fmt.Sprintf("%s:%d", ip2, xfer.AppPort))
// Trigger another resolve with a tick; again,
// just want ips.
c <- time.Now()
assertAdd(ip1+port, fmt.Sprintf("%s:%d", ip2, AppPort))
assertAdd(ip1+port, fmt.Sprintf("%s:%d", ip2, xfer.AppPort))
ip3 := "1.2.3.4"
updateIPs("symbolic.name", makeIPs(ip3))
c <- time.Now() // trigger a resolve
assertAdd(ip3+port, ip1+port, fmt.Sprintf("%s:%d", ip2, AppPort))
assertAdd(ip3+port, ip1+port, fmt.Sprintf("%s:%d", ip2, xfer.AppPort))
ip4 := "10.10.10.10"
updateIPs("symbolic.name", makeIPs(ip3, ip4))
c <- time.Now() // trigger another resolve, this time with 2 adds
assertAdd(ip3+port, ip4+port, ip1+port, fmt.Sprintf("%s:%d", ip2, AppPort))
assertAdd(ip3+port, ip4+port, ip1+port, fmt.Sprintf("%s:%d", ip2, xfer.AppPort))
done := make(chan struct{})
go func() { r.Stop(); close(done) }()

2
probe/controls/controls.go
View File

@@ -3,7 +3,7 @@ package controls
import (
"sync"
"github.com/weaveworks/scope/xfer"
"github.com/weaveworks/scope/common/xfer"
)
var (

2
probe/controls/controls_test.go
View File

@@ -4,9 +4,9 @@ import (
"reflect"
"testing"
"github.com/weaveworks/scope/common/xfer"
"github.com/weaveworks/scope/probe/controls"
"github.com/weaveworks/scope/test"
"github.com/weaveworks/scope/xfer"
)
func TestControls(t *testing.T) {

2
probe/controls/pipes.go
View File

@@ -4,7 +4,7 @@ import (
"fmt"
"math/rand"
"github.com/weaveworks/scope/xfer"
"github.com/weaveworks/scope/common/xfer"
)
// PipeClient is the type of the thing the probe uses to make pipe connections.

2
probe/docker/controls.go
View File

@@ -5,9 +5,9 @@ import (
docker_client "github.com/fsouza/go-dockerclient"
"github.com/weaveworks/scope/common/xfer"
"github.com/weaveworks/scope/probe/controls"
"github.com/weaveworks/scope/report"
"github.com/weaveworks/scope/xfer"
)
// Control IDs used by the docker intergation.

2
probe/docker/controls_test.go
View File

@@ -8,11 +8,11 @@ import (
"github.com/gorilla/websocket"
"github.com/weaveworks/scope/common/xfer"
"github.com/weaveworks/scope/probe/controls"
"github.com/weaveworks/scope/probe/docker"
"github.com/weaveworks/scope/report"
"github.com/weaveworks/scope/test"
"github.com/weaveworks/scope/xfer"
)
func TestControls(t *testing.T) {

8
probe/probe.go
View File

@@ -7,8 +7,8 @@ import (
"github.com/armon/go-metrics"
"github.com/weaveworks/scope/probe/appclient"
"github.com/weaveworks/scope/report"
"github.com/weaveworks/scope/xfer"
)
const (
@@ -18,7 +18,7 @@ const (
// Probe sits there, generating and publishing reports.
type Probe struct {
spyInterval, publishInterval time.Duration
publisher *xfer.ReportPublisher
publisher *appclient.ReportPublisher
tickers []Ticker
reporters []Reporter
@@ -52,11 +52,11 @@ type Ticker interface {
}
// New makes a new Probe.
func New(spyInterval, publishInterval time.Duration, publisher xfer.Publisher) *Probe {
func New(spyInterval, publishInterval time.Duration, publisher appclient.Publisher) *Probe {
result := &Probe{
spyInterval: spyInterval,
publishInterval: publishInterval,
publisher: xfer.NewReportPublisher(publisher),
publisher: appclient.NewReportPublisher(publisher),
quit: make(chan struct{}),
spiedReports: make(chan report.Report, reportBufferSize),
shortcutReports: make(chan report.Report, reportBufferSize),

2
prog/app.go
View File

@@ -14,7 +14,7 @@ import (
"github.com/weaveworks/weave/common"
"github.com/weaveworks/scope/app"
"github.com/weaveworks/scope/xfer"
"github.com/weaveworks/scope/common/xfer"
)
// Router creates the mux for all the various app components.

11
prog/probe.go
View File

@@ -17,7 +17,9 @@ import (
"github.com/weaveworks/weave/common"
"github.com/weaveworks/scope/common/hostname"
"github.com/weaveworks/scope/common/xfer"
"github.com/weaveworks/scope/probe"
"github.com/weaveworks/scope/probe/appclient"
"github.com/weaveworks/scope/probe/controls"
"github.com/weaveworks/scope/probe/docker"
"github.com/weaveworks/scope/probe/endpoint"
@@ -26,7 +28,6 @@ import (
"github.com/weaveworks/scope/probe/overlay"
"github.com/weaveworks/scope/probe/process"
"github.com/weaveworks/scope/report"
"github.com/weaveworks/scope/xfer"
)
// Main runs the probe
@@ -94,20 +95,20 @@ func probeMain() {
}
log.Printf("publishing to: %s", strings.Join(targets, ", "))
probeConfig := xfer.ProbeConfig{
probeConfig := appclient.ProbeConfig{
Token: *token,
ProbeID: probeID,
Insecure: *insecure,
}
clients := xfer.NewMultiAppClient(func(hostname, endpoint string) (xfer.AppClient, error) {
return xfer.NewAppClient(
clients := appclient.NewMultiAppClient(func(hostname, endpoint string) (appclient.AppClient, error) {
return appclient.NewAppClient(
probeConfig, hostname, endpoint,
xfer.ControlHandlerFunc(controls.HandleControlRequest),
)
})
defer clients.Stop()
resolver := xfer.NewStaticResolver(targets, clients.Set)
resolver := appclient.NewStaticResolver(targets, clients.Set)
defer resolver.Stop()
processCache := process.NewCachingWalker(process.NewWalker(*procRoot))

8
xfer/ports.go
View File

@@ -1,8 +0,0 @@
package xfer
var (
// AppPort is the default port that the app will use for its HTTP server.
// The app publishes the API and user interface, and receives reports from
// probes, on this port.
AppPort = 4040
)

10
xfer/publisher.go
View File

@@ -1,10 +0,0 @@
package xfer
import "io"
// Publisher is something which can send a stream of data somewhere, probably
// to a remote collector.
type Publisher interface {
Publish(io.Reader) error
Stop()
}