github/weave-scope
1
0
Fork 0
mirror of https://github.com/weaveworks/scope.git synced 2026-07-18 12:59:31 +00:00
Code Issues Packages Projects Releases Wiki Activity

Probes POST to apps

Browse Source 
- App takes POST report on /api/report
- Probe publishes to configured target(s)
- Name resolution happens on probe-side
- There's no longer an xfer.ProbePort
- xfer.Collector responsibility is reduced
- Fixes to remaining experimental components.
- rm experimental/bridge: it's not being used, and by changing the
  app/probe comm model, it would require a complete refactor anyway. We
  can easily rebuild it when we need to. It will even be much simpler.
- rm experimental/graphviz: it's broken for some time anyway, and we
  don't really need to play around with it as a rendering option
  anymore.
- rm experimental/oneshot: we never use this anymore.
This commit is contained in:
Peter Bourgon
2015-08-06 15:20:47 +02:00
parent 5a29b1b9fe
commit 7d51f173ea
33 changed files with 446 additions and 1609 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
app/api_report.go
View File

@@ -2,10 +2,12 @@ package main
import (
"net/http"
"github.com/weaveworks/scope/xfer"
)
// Raw report handler
func makeRawReportHandler(rep Reporter) func(http.ResponseWriter, *http.Request) {
func makeRawReportHandler(rep xfer.Reporter) func(http.ResponseWriter, *http.Request) {
return func(w http.ResponseWriter, r *http.Request) {
// r.ParseForm()
respondWith(w, http.StatusOK, rep.Report())

3
app/api_topologies.go
View File

@@ -4,6 +4,7 @@ import (
"net/http"
"github.com/weaveworks/scope/render"
"github.com/weaveworks/scope/xfer"
)
// APITopologyDesc is returned in a list by the /api/topology handler.
@@ -21,7 +22,7 @@ type topologyStats struct {
}
// makeTopologyList returns a handler that yields an APITopologyList.
func makeTopologyList(rep Reporter) func(w http.ResponseWriter, r *http.Request) {
func makeTopologyList(rep xfer.Reporter) func(w http.ResponseWriter, r *http.Request) {
return func(w http.ResponseWriter, r *http.Request) {
var (
rpt = rep.Report()

11
app/api_topology.go
View File

@@ -9,6 +9,7 @@ import (
"github.com/weaveworks/scope/render"
"github.com/weaveworks/scope/report"
"github.com/weaveworks/scope/xfer"
)
const (
@@ -32,14 +33,14 @@ type APIEdge struct {
}
// Full topology.
func handleTopology(rep Reporter, t topologyView, w http.ResponseWriter, r *http.Request) {
func handleTopology(rep xfer.Reporter, t topologyView, w http.ResponseWriter, r *http.Request) {
respondWith(w, http.StatusOK, APITopology{
Nodes: t.renderer.Render(rep.Report()),
})
}
// Websocket for the full topology. This route overlaps with the next.
func handleWs(rep Reporter, t topologyView, w http.ResponseWriter, r *http.Request) {
func handleWs(rep xfer.Reporter, t topologyView, w http.ResponseWriter, r *http.Request) {
if err := r.ParseForm(); err != nil {
respondWith(w, http.StatusInternalServerError, err.Error())
return
@@ -56,7 +57,7 @@ func handleWs(rep Reporter, t topologyView, w http.ResponseWriter, r *http.Reque
}
// Individual nodes.
func handleNode(rep Reporter, t topologyView, w http.ResponseWriter, r *http.Request) {
func handleNode(rep xfer.Reporter, t topologyView, w http.ResponseWriter, r *http.Request) {
var (
vars = mux.Vars(r)
nodeID = vars["id"]
@@ -71,7 +72,7 @@ func handleNode(rep Reporter, t topologyView, w http.ResponseWriter, r *http.Req
}
// Individual edges.
func handleEdge(rep Reporter, t topologyView, w http.ResponseWriter, r *http.Request) {
func handleEdge(rep xfer.Reporter, t topologyView, w http.ResponseWriter, r *http.Request) {
var (
vars = mux.Vars(r)
localID = vars["local"]
@@ -90,7 +91,7 @@ var upgrader = websocket.Upgrader{
func handleWebsocket(
w http.ResponseWriter,
r *http.Request,
rep Reporter,
rep xfer.Reporter,
t topologyView,
loop time.Duration,
) {

29
app/main.go
View File

@@ -2,7 +2,6 @@ package main
import (
"flag"
"fmt"
"log"
"math/rand"
"net/http"
@@ -21,35 +20,17 @@ var version = "dev"
func main() {
var (
defaultProbes = []string{fmt.Sprintf("localhost:%d", xfer.ProbePort), fmt.Sprintf("scope.weave.local:%d", xfer.ProbePort)}
batch = flag.Duration("batch", 1*time.Second, "batch interval")
window = flag.Duration("window", 15*time.Second, "window")
listen = flag.String("http.address", ":"+strconv.Itoa(xfer.AppPort), "webserver listen address")
printVersion = flag.Bool("version", false, "print version number and exit")
window = flag.Duration("window", 15*time.Second, "window")
listen = flag.String("http.address", ":"+strconv.Itoa(xfer.AppPort), "webserver listen address")
)
flag.Parse()
probes := append(defaultProbes, flag.Args()...)
if *printVersion {
fmt.Println(version)
return
}
rand.Seed(time.Now().UnixNano())
id := strconv.FormatInt(rand.Int63(), 16)
log.Printf("app starting, version %s, id %s", version, id)
log.Printf("app starting, version %s, ID %s", version, id)
// Collector deals with the probes, and generates merged reports.
c := xfer.NewCollector(*batch, id)
defer c.Stop()
r := newStaticResolver(probes, c.Add)
defer r.Stop()
lifo := NewReportLIFO(c, *window)
defer lifo.Stop()
http.Handle("/", Router(lifo))
c := xfer.NewCollector(*window)
http.Handle("/", Router(c))
irq := interrupt()
go func() {
log.Printf("listening on %s", *listen)

8
app/mock_reporter_test.go
View File

@@ -5,9 +5,9 @@ import (
"github.com/weaveworks/scope/test"
)
// StaticReport is used as know test data in api tests.
// StaticReport is used as a fixture in tests. It emulates an xfer.Collector.
type StaticReport struct{}
func (s StaticReport) Report() report.Report {
return test.Report
}
func (s StaticReport) Report() report.Report { return test.Report }
func (s StaticReport) Add(report.Report) {}

3
app/origin_host.go
View File

@@ -8,6 +8,7 @@ import (
"github.com/weaveworks/scope/probe/host"
"github.com/weaveworks/scope/report"
"github.com/weaveworks/scope/xfer"
)
// OriginHost represents a host that runs a probe, i.e. the origin host of
@@ -35,7 +36,7 @@ func getOriginHost(t report.Topology, nodeID string) (OriginHost, bool) {
}
// makeOriginHostHandler makes the /api/origin/* handler.
func makeOriginHostHandler(rep Reporter) http.HandlerFunc {
func makeOriginHostHandler(rep xfer.Reporter) http.HandlerFunc {
return func(w http.ResponseWriter, r *http.Request) {
var (
vars = mux.Vars(r)

96
app/report_lifo.go
View File

@@ -1,96 +0,0 @@
package main
import (
"time"
"github.com/weaveworks/scope/report"
)
// Reporter is something which generates a single 'current' report over a
// stream of incoming reports.
type Reporter interface {
Report() report.Report
}
type timedReport struct {
report.Report
Timestamp time.Time
}
// ReportLIFO keeps a short-term history of reports.
type ReportLIFO struct {
reports []timedReport
requests chan chan report.Report
quit chan chan struct{}
}
type reporter interface {
Reports() <-chan report.Report
}
// NewReportLIFO collects reports up to a certain age.
func NewReportLIFO(r reporter, maxAge time.Duration) *ReportLIFO {
l := ReportLIFO{
reports: []timedReport{},
requests: make(chan chan report.Report),
quit: make(chan chan struct{}),
}
go func() {
for {
select {
case report := <-r.Reports():
// Incoming report from the collecter.
tr := timedReport{
Timestamp: time.Now(),
Report: report,
}
l.reports = append(l.reports, tr)
l.reports = cleanOld(l.reports, time.Now().Add(-maxAge))
case req := <-l.requests:
// Request for the current report.
report := report.MakeReport()
oldest := time.Now()
for _, r := range l.reports {
if r.Timestamp.Before(oldest) {
oldest = r.Timestamp
}
report.Merge(r.Report)
}
report.Window = time.Now().Sub(oldest)
req <- report
case q := <-l.quit:
close(q)
return
}
}
}()
return &l
}
// Stop shuts down the monitor.
func (r *ReportLIFO) Stop() {
q := make(chan struct{})
r.quit <- q
<-q
}
// Report returns the latest report.
func (r *ReportLIFO) Report() report.Report {
req := make(chan report.Report)
r.requests <- req
return <-req
}
func cleanOld(reports []timedReport, threshold time.Time) []timedReport {
res := make([]timedReport, 0, len(reports))
for _, tr := range reports {
if tr.Timestamp.Before(threshold) {
continue
}
res = append(res, tr)
}
return res
}

30
app/router.go
View File

@@ -1,6 +1,7 @@
package main
import (
"encoding/gob"
"net/http"
"net/url"
"strings"
@@ -8,6 +9,8 @@ import (
"github.com/gorilla/mux"
"github.com/weaveworks/scope/render"
"github.com/weaveworks/scope/report"
"github.com/weaveworks/scope/xfer"
)
// URLMatcher uses request.RequestURI (the raw, unparsed request) to attempt
@@ -41,10 +44,17 @@ func URLMatcher(pattern string) mux.MatcherFunc {
}
}
// Router gives of the HTTP dispatcher. It will always use the embedded HTML
// resources.
func Router(c Reporter) *mux.Router {
type collector interface {
xfer.Reporter
xfer.Adder
}
// Router returns the HTTP dispatcher, managing API and UI requests, and
// accepting reports from probes.. It will always use the embedded HTML
// resources for the UI.
func Router(c collector) *mux.Router {
router := mux.NewRouter()
router.HandleFunc("/api/report", makeReportPostHandler(c)).Methods("POST")
get := router.Methods("GET").Subrouter()
get.HandleFunc("/api", apiHandler)
get.HandleFunc("/api/topology", makeTopologyList(c))
@@ -58,7 +68,19 @@ func Router(c Reporter) *mux.Router {
return router
}
func captureTopology(rep Reporter, f func(Reporter, topologyView, http.ResponseWriter, *http.Request)) func(http.ResponseWriter, *http.Request) {
func makeReportPostHandler(a xfer.Adder) http.HandlerFunc {
return func(w http.ResponseWriter, r *http.Request) {
var rpt report.Report
if err := gob.NewDecoder(r.Body).Decode(&rpt); err != nil {
http.Error(w, err.Error(), http.StatusBadRequest)
return
}
a.Add(rpt)
w.WriteHeader(http.StatusOK)
}
}
func captureTopology(rep xfer.Reporter, f func(xfer.Reporter, topologyView, http.ResponseWriter, *http.Request)) http.HandlerFunc {
return func(w http.ResponseWriter, r *http.Request) {
topology, ok := topologyRegistry[mux.Vars(r)["topology"]]
if !ok {

10
docker/entrypoint.sh
View File

@@ -54,10 +54,10 @@ if [ -n "$DNS_SERVER" -a -n "$SEARCHPATH" ]; then
fi
# End of the command line can optionally be some
# addresses of probes to connect to, for people not
# using Weave DNS. We stick these in /etc/weave/probes
# for the run-app script to pick up.
MANUAL_PROBES=$@
echo "$MANUAL_PROBES" >/etc/weave/probes
# addresses of apps to connect to, for people not
# using Weave DNS. We stick these in /etc/weave/apps
# for the run-probe script to pick up.
MANUAL_APPS=$@
echo "$MANUAL_APPS" >/etc/weave/apps
exec /sbin/runsvdir /etc/service

2
docker/run-app
View File

@@ -1,3 +1,3 @@
#!/bin/sh
exec /home/weave/scope-app $(cat /etc/weave/scope-app.args) $(cat /etc/weave/probes)
exec /home/weave/scope-app $(cat /etc/weave/scope-app.args)

2
docker/run-probe
View File

@@ -1,3 +1,3 @@
#!/bin/sh
exec /home/weave/scope-probe $(cat /etc/weave/scope-probe.args)
exec /home/weave/scope-probe $(cat /etc/weave/scope-probe.args) $(cat /etc/weave/apps)

19
experimental/bridge/Makefile
View File

@@ -1,19 +0,0 @@
.PHONY: all vet lint build test clean
all: build test vet lint
vet:
go vet ./...
lint:
golint .
build:
go build
test:
go test
clean:
go clean

207
experimental/bridge/main.go
View File

@@ -1,207 +0,0 @@
package main
import (
"flag"
"log"
"net"
_ "net/http/pprof"
"os"
"os/signal"
"strconv"
"strings"
"syscall"
"time"
"github.com/weaveworks/scope/render"
"github.com/weaveworks/scope/report"
"github.com/weaveworks/scope/xfer"
)
const (
trafficTimeout = 2 * time.Minute
)
func main() {
var (
listen = flag.String("listen", ":"+strconv.Itoa(xfer.ProbePort), "listen address")
probes = flag.String("probes", "", "list of all initial probes, comma separated")
batch = flag.Duration("batch", 1*time.Second, "batch interval")
version = flag.Bool("version", false, "print version number and exit")
)
flag.Parse()
if len(flag.Args()) != 0 {
flag.Usage()
os.Exit(1)
}
if *version {
//fmt.Printf("%s\n", probe.Version)
return
}
if *probes == "" {
log.Fatal("no probes given via -probes")
}
log.Printf("starting")
fixedAddresses := strings.Split(*probes, ",")
// Collector deals with the probes, and generates a single merged report
// every second.
c := xfer.NewCollector(*batch, "id")
for _, addr := range fixedAddresses {
c.Add(addr)
}
defer c.Stop()
publisher, err := xfer.NewTCPPublisher(*listen)
if err != nil {
log.Fatal(err)
}
defer publisher.Close()
log.Printf("listening on %s\n", *listen)
var fixedIPs []string
for _, a := range fixedAddresses {
if addr, _, err := net.SplitHostPort(a); err == nil {
fixedIPs = append(fixedIPs, addr)
}
}
go discover(c, publisher, fixedIPs)
<-interrupt()
log.Printf("shutting down")
}
func interrupt() chan os.Signal {
c := make(chan os.Signal)
signal.Notify(c, syscall.SIGINT, syscall.SIGTERM)
return c
}
type collector interface {
Reports() <-chan report.Report
Remove(string)
Add(string)
}
type publisher xfer.Publisher
// makeAvoid makes a map with IPs we don't want to consider in discover(). It
// is the set of IPs which the bridge is configured to connect to, and the all
// the IPs from the local interfaces.
func makeAvoid(fixed []string) map[string]struct{} {
avoid := map[string]struct{}{}
// Don't discover fixed probes. This way we'll never remove them.
for _, a := range fixed {
avoid[a] = struct{}{}
}
// Don't go Ouroboros.
if localNets, err := net.InterfaceAddrs(); err == nil {
for _, n := range localNets {
if net, ok := n.(*net.IPNet); ok {
avoid[net.IP.String()] = struct{}{}
}
}
}
return avoid
}
// discover reads reports from a collector and republishes them on the
// publisher, while scanning the reports for IPs to connect to. Only addresses
// in the network topology of the report are considered. IPs listed in fixed
// are always skipped.
func discover(c collector, p publisher, fixed []string) {
lastSeen := map[string]time.Time{}
avoid := makeAvoid(fixed)
for r := range c.Reports() {
// log.Printf("got a report")
p.Publish(r)
var (
now = time.Now()
localNets = render.LocalNetworks(r)
)
for _, adjacent := range r.Address.Adjacency {
for _, a := range adjacent {
ip := report.AddressIDAddresser(a) // address id -> IP
if ip == nil {
continue
}
addr := ip.String()
if _, ok := avoid[addr]; ok {
continue
}
// log.Printf("potential address: %v (via %s)", addr, src)
if _, ok := lastSeen[addr]; !ok {
if interestingAddress(localNets, addr) {
log.Printf("discovery %v: potential probe address", addr)
c.Add(addressToDial(addr))
} else {
log.Printf("discovery %v: non-probe address", addr)
}
}
// We always add addr to lastSeen[], even if it's a non-local
// address. This way they are part of the normal timeout logic,
// and we won't analyze the address again until it re-appears
// after a timeout.
lastSeen[addr] = now
}
}
for addr, last := range lastSeen {
if now.Sub(last) > trafficTimeout {
// Timeout can be for a non-local address, which we didn't
// connect to. In that case the RemoveAddress() call won't do
// anything.
log.Printf("discovery %v: traffic timeout", addr)
delete(lastSeen, addr)
c.Remove(addressToDial(addr))
}
}
}
}
// interestingAddress tells whether the address is a local and normal address,
// which we want to try to connect to.
func interestingAddress(localNets []*net.IPNet, addr string) bool {
if addr == "" {
return false
}
// The address is expected to be an IPv{4,6} address.
ip := net.ParseIP(addr)
if ip == nil {
return false
}
// Filter out localhost, broadcast, and other non-connectable addresses.
if !validateRemoteAddr(ip) {
return false
}
// Only connect to addresses we know are localnet.
for _, n := range localNets {
if n.Contains(ip) {
return true
}
}
return false
}
// addressToDial formats an IP address so we can pass it on to Dial().
func addressToDial(address string) string {
// return fmt.Sprintf("[%s]:%d", addr, probePort)
return net.JoinHostPort(address, strconv.Itoa(xfer.ProbePort))
}

71
experimental/bridge/net.go
View File

@@ -1,71 +0,0 @@
package main
import (
"net"
)
func validateRemoteAddr(ip net.IP) bool {
if ip == nil {
return false
}
if ip.IsInterfaceLocalMulticast() {
return false
}
if ip.IsLinkLocalMulticast() {
return false
}
if ip.IsLinkLocalUnicast() {
return false
}
if ip.IsLoopback() {
return false
}
if ip.IsMulticast() {
return false
}
if ip.IsUnspecified() {
return false
}
if isBroadcasty(ip) {
return false
}
return true
}
func isBroadcasty(ip net.IP) bool {
if ip4 := ip.To4(); ip4 != nil {
if ip4.Equal(net.IPv4bcast) {
return true
}
if ip4.Equal(net.IPv4allsys) {
return true
}
if ip4.Equal(net.IPv4allrouter) {
return true
}
if ip4.Equal(net.IPv4zero) {
return true
}
return false
}
if ip.Equal(net.IPv6zero) {
return true
}
if ip.Equal(net.IPv6unspecified) {
return true
}
if ip.Equal(net.IPv6loopback) {
return true
}
if ip.Equal(net.IPv6interfacelocalallnodes) {
return true
}
if ip.Equal(net.IPv6linklocalallnodes) {
return true
}
if ip.Equal(net.IPv6linklocalallrouters) {
return true
}
return false
}

41
experimental/bridge/net_test.go
View File

@@ -1,41 +0,0 @@
package main
import (
"net"
"testing"
)
func TestValidRemoteAddr(t *testing.T) {
for input, expected := range map[string]bool{
"localhost": false,
"127.0.0.1": false, // should be same as loopback
"127.1.2.3": false, // 127.0.0.0/8 is all loopback
"0.0.0.0": false,
"224.0.0.1": false, // all systems
"224.0.0.2": false, // all routers
"224.0.0.22": false,
"255.255.255.255": false, // broadcast
"1.2.3.4": true,
"::": false, // unspecified
"0:0:0:0:0:0:0:0": false, // unspecified (alternate form)
"b8:27:eb:a4:bf:6e": false,
"fe80::1240:f3ff:fe80:5474": false, // loopback
"fe80::1": false, // loopback
"::1": false, // loopback
"0:0:0:0:0:0:0:1": false, // loopback (alternate form)
"2001:db8::1:0:0:1": true, // valid
// http://www.iana.org/assignments/ipv6-multicast-addresses/ipv6-multicast-addresses.xhtml
"FF01:0:0:0:0:0:0:1": false, // Node-local all-nodes
// "FF01:0:0:0:0:0:0:2": false, // Node-local all-routers, isn't spec'd
// in package net/ip
"FF02:0:0:0:0:0:0:1": false, // Link-local all-nodes
"FF02:0:0:0:0:0:0:2": false, // Link-local all-routers
} {
if got := validateRemoteAddr(net.ParseIP(input)); expected != got {
t.Errorf("%s: expected %v, got %v", input, expected, got)
}
}
}

126
experimental/demoprobe/generate.go
View File

@@ -1,126 +0,0 @@
package main
import (
"fmt"
"math/rand"
"net"
"strconv"
"time"
"github.com/weaveworks/scope/probe/docker"
"github.com/weaveworks/scope/probe/process"
"github.com/weaveworks/scope/report"
)
func init() {
rand.Seed(time.Now().UnixNano())
}
// DemoReport makes up a report.
func DemoReport(nodeCount int) report.Report {
r := report.MakeReport()
// Make up some plausible IPv4 numbers
hosts := []string{}
ip := [4]int{192, 168, 1, 1}
for range make([]struct{}, nodeCount) {
hosts = append(hosts, fmt.Sprintf("%d.%d.%d.%d", ip[0], ip[1], ip[2], ip[3]))
ip[3]++
if ip[3] > 200 {
ip[2]++
ip[3] = 1
}
}
// Some non-local ones.
hosts = append(hosts, []string{"1.2.3.4", "2.3.4.5"}...)
_, localNet, err := net.ParseCIDR("192.168.0.0/16")
if err != nil {
panic(err)
}
type conn struct {
srcProc, dstProc string
dstPort int
}
procPool := []conn{
{srcProc: "curl", dstPort: 80, dstProc: "apache"},
{srcProc: "wget", dstPort: 80, dstProc: "apache"},
{srcProc: "curl", dstPort: 80, dstProc: "nginx"},
{srcProc: "curl", dstPort: 8080, dstProc: "app1"},
{srcProc: "nginx", dstPort: 8080, dstProc: "app1"},
{srcProc: "nginx", dstPort: 8080, dstProc: "app2"},
{srcProc: "nginx", dstPort: 8080, dstProc: "app3"},
}
connectionCount := nodeCount * 2
for i := 0; i < connectionCount; i++ {
var (
c = procPool[rand.Intn(len(procPool))]
src = hosts[rand.Intn(len(hosts))]
dst = hosts[rand.Intn(len(hosts))]
srcPort = rand.Intn(50000) + 10000
srcPortID = report.MakeEndpointNodeID("", src, strconv.Itoa(srcPort))
dstPortID = report.MakeEndpointNodeID("", dst, strconv.Itoa(c.dstPort))
srcID = report.MakeAdjacencyID(srcPortID)
dstID = report.MakeAdjacencyID(dstPortID)
srcAddressID = report.MakeAddressNodeID("", src)
dstAddressID = report.MakeAddressNodeID("", dst)
nodeSrcAddressID = report.MakeAdjacencyID(srcAddressID)
nodeDstAddressID = report.MakeAdjacencyID(dstAddressID)
)
// Endpoint topology
if _, ok := r.Endpoint.NodeMetadatas[srcPortID]; !ok {
r.Endpoint.NodeMetadatas[srcPortID] = report.MakeNodeMetadataWith(map[string]string{
process.PID: "4000",
"name": c.srcProc,
"domain": "node-" + src,
})
}
r.Endpoint.Adjacency[srcID] = r.Endpoint.Adjacency[srcID].Add(dstPortID)
if _, ok := r.Endpoint.NodeMetadatas[dstPortID]; !ok {
r.Endpoint.NodeMetadatas[dstPortID] = report.MakeNodeMetadataWith(map[string]string{
process.PID: "4000",
"name": c.dstProc,
"domain": "node-" + dst,
})
}
r.Endpoint.Adjacency[dstID] = r.Endpoint.Adjacency[dstID].Add(srcPortID)
var (
edgeKeyEgress = report.MakeEdgeID(srcPortID, dstPortID)
edgeKeyIngress = report.MakeEdgeID(dstPortID, srcPortID)
)
r.Endpoint.EdgeMetadatas[edgeKeyEgress] = report.EdgeMetadata{
MaxConnCountTCP: newu64(uint64(rand.Intn(100) + 10)),
}
r.Endpoint.EdgeMetadatas[edgeKeyIngress] = report.EdgeMetadata{
MaxConnCountTCP: newu64(uint64(rand.Intn(100) + 10)),
}
// Address topology
if _, ok := r.Address.NodeMetadatas[srcAddressID]; !ok {
r.Address.NodeMetadatas[srcAddressID] = report.MakeNodeMetadataWith(map[string]string{
docker.Name: src,
})
}
r.Address.Adjacency[nodeSrcAddressID] = r.Address.Adjacency[nodeSrcAddressID].Add(dstAddressID)
if _, ok := r.Address.NodeMetadatas[dstAddressID]; !ok {
r.Address.NodeMetadatas[dstAddressID] = report.MakeNodeMetadataWith(map[string]string{
docker.Name: dst,
})
}
r.Address.Adjacency[nodeDstAddressID] = r.Address.Adjacency[nodeDstAddressID].Add(srcAddressID)
// Host data
r.Host.NodeMetadatas["hostX"] = report.MakeNodeMetadataWith(map[string]string{
"ts": time.Now().UTC().Format(time.RFC3339Nano),
"host_name": "host-x",
"local_networks": localNet.String(),
"os": "linux",
})
}
return r
}
func newu64(value uint64) *uint64 { return &value }

148
experimental/demoprobe/main.go
View File

@@ -4,53 +4,141 @@ import (
"flag"
"fmt"
"log"
"os"
"os/signal"
"math/rand"
"net"
"strconv"
"syscall"
"time"
"github.com/weaveworks/scope/probe/docker"
"github.com/weaveworks/scope/probe/process"
"github.com/weaveworks/scope/report"
"github.com/weaveworks/scope/xfer"
)
func main() {
var (
version = flag.Bool("version", false, "print version number and exit")
publish = flag.String("publish", fmt.Sprintf("localhost:%d", xfer.AppPort), "publish target")
publishInterval = flag.Duration("publish.interval", 1*time.Second, "publish (output) interval")
listen = flag.String("listen", ":"+strconv.Itoa(xfer.ProbePort), "listen address")
hostCount = flag.Int("hostcount", 10, "Number of demo hosts to generate")
)
flag.Parse()
if len(flag.Args()) != 0 {
flag.Usage()
os.Exit(1)
}
// -version flag:
if *version {
fmt.Printf("unstable\n")
return
}
publisher, err := xfer.NewTCPPublisher(*listen)
publisher, err := xfer.NewHTTPPublisher(*publish, "demoprobe")
if err != nil {
log.Fatal(err)
}
defer publisher.Close()
go func() {
for {
publisher.Publish(DemoReport(*hostCount))
time.Sleep(*publishInterval)
rand.Seed(time.Now().UnixNano())
for range time.Tick(*publishInterval) {
if err := publisher.Publish(demoReport(*hostCount)); err != nil {
log.Print(err)
}
}()
log.Printf("%s", <-interrupt())
log.Printf("Shutting down...")
}
}
func interrupt() chan os.Signal {
c := make(chan os.Signal)
signal.Notify(c, syscall.SIGINT, syscall.SIGTERM)
return c
func demoReport(nodeCount int) report.Report {
r := report.MakeReport()
// Make up some plausible IPv4 numbers.
hosts := []string{}
ip := [4]int{192, 168, 1, 1}
for range make([]struct{}, nodeCount) {
hosts = append(hosts, fmt.Sprintf("%d.%d.%d.%d", ip[0], ip[1], ip[2], ip[3]))
ip[3]++
if ip[3] > 200 {
ip[2]++
ip[3] = 1
}
}
hosts = append(hosts, []string{"1.2.3.4", "2.3.4.5"}...) // Some non-local ones, too.
_, localNet, err := net.ParseCIDR("192.168.0.0/16")
if err != nil {
panic(err)
}
type conn struct {
srcProc, dstProc string
dstPort int
}
procPool := []conn{
{srcProc: "curl", dstPort: 80, dstProc: "apache"},
{srcProc: "wget", dstPort: 80, dstProc: "apache"},
{srcProc: "curl", dstPort: 80, dstProc: "nginx"},
{srcProc: "curl", dstPort: 8080, dstProc: "app1"},
{srcProc: "nginx", dstPort: 8080, dstProc: "app1"},
{srcProc: "nginx", dstPort: 8080, dstProc: "app2"},
{srcProc: "nginx", dstPort: 8080, dstProc: "app3"},
}
connectionCount := nodeCount * 2
for i := 0; i < connectionCount; i++ {
var (
c = procPool[rand.Intn(len(procPool))]
src = hosts[rand.Intn(len(hosts))]
dst = hosts[rand.Intn(len(hosts))]
srcPort = rand.Intn(50000) + 10000
srcPortID = report.MakeEndpointNodeID("", src, strconv.Itoa(srcPort))
dstPortID = report.MakeEndpointNodeID("", dst, strconv.Itoa(c.dstPort))
srcID = report.MakeAdjacencyID(srcPortID)
dstID = report.MakeAdjacencyID(dstPortID)
srcAddressID = report.MakeAddressNodeID("", src)
dstAddressID = report.MakeAddressNodeID("", dst)
nodeSrcAddressID = report.MakeAdjacencyID(srcAddressID)
nodeDstAddressID = report.MakeAdjacencyID(dstAddressID)
)
// Endpoint topology
if _, ok := r.Endpoint.NodeMetadatas[srcPortID]; !ok {
r.Endpoint.NodeMetadatas[srcPortID] = report.MakeNodeMetadataWith(map[string]string{
process.PID: "4000",
"name": c.srcProc,
"domain": "node-" + src,
})
}
r.Endpoint.Adjacency[srcID] = r.Endpoint.Adjacency[srcID].Add(dstPortID)
if _, ok := r.Endpoint.NodeMetadatas[dstPortID]; !ok {
r.Endpoint.NodeMetadatas[dstPortID] = report.MakeNodeMetadataWith(map[string]string{
process.PID: "4000",
"name": c.dstProc,
"domain": "node-" + dst,
})
}
r.Endpoint.Adjacency[dstID] = r.Endpoint.Adjacency[dstID].Add(srcPortID)
var (
edgeKeyEgress = report.MakeEdgeID(srcPortID, dstPortID)
edgeKeyIngress = report.MakeEdgeID(dstPortID, srcPortID)
)
r.Endpoint.EdgeMetadatas[edgeKeyEgress] = report.EdgeMetadata{
MaxConnCountTCP: newu64(uint64(rand.Intn(100) + 10)),
}
r.Endpoint.EdgeMetadatas[edgeKeyIngress] = report.EdgeMetadata{
MaxConnCountTCP: newu64(uint64(rand.Intn(100) + 10)),
}
// Address topology
if _, ok := r.Address.NodeMetadatas[srcAddressID]; !ok {
r.Address.NodeMetadatas[srcAddressID] = report.MakeNodeMetadataWith(map[string]string{
docker.Name: src,
})
}
r.Address.Adjacency[nodeSrcAddressID] = r.Address.Adjacency[nodeSrcAddressID].Add(dstAddressID)
if _, ok := r.Address.NodeMetadatas[dstAddressID]; !ok {
r.Address.NodeMetadatas[dstAddressID] = report.MakeNodeMetadataWith(map[string]string{
docker.Name: dst,
})
}
r.Address.Adjacency[nodeDstAddressID] = r.Address.Adjacency[nodeDstAddressID].Add(srcAddressID)
// Host data
r.Host.NodeMetadatas["hostX"] = report.MakeNodeMetadataWith(map[string]string{
"ts": time.Now().UTC().Format(time.RFC3339Nano),
"host_name": "host-x",
"local_networks": localNet.String(),
"os": "linux",
})
}
return r
}
func newu64(value uint64) *uint64 { return &value }

31
experimental/fixprobe/main.go
View File

@@ -7,7 +7,6 @@ import (
"fmt"
"log"
"os"
"strconv"
"time"
"github.com/weaveworks/scope/report"
@@ -16,35 +15,29 @@ import (
func main() {
var (
publish = flag.String("publish", fmt.Sprintf("localhost:%d", xfer.AppPort), "publish target")
publishInterval = flag.Duration("publish.interval", 1*time.Second, "publish (output) interval")
listenAddress = flag.String("listen", ":"+strconv.Itoa(xfer.ProbePort), "listen address")
)
flag.Parse()
if len(flag.Args()) != 1 {
fmt.Printf("usage: fixprobe [--args] report.json\n")
return
}
fixture := flag.Arg(0)
f, err := os.Open(fixture)
if err != nil {
fmt.Printf("json error: %v\n", err)
return
}
var fixedReport report.Report
if err := json.NewDecoder(f).Decode(&fixedReport); err != nil {
fmt.Printf("json error: %v\n", err)
return
log.Fatal("usage: fixprobe [--args] report.json")
}
publisher, err := xfer.NewTCPPublisher(*listenAddress)
f, err := os.Open(flag.Arg(0))
if err != nil {
log.Fatal(err)
}
defer publisher.Close()
var fixedReport report.Report
if err := json.NewDecoder(f).Decode(&fixedReport); err != nil {
log.Fatal(err)
}
f.Close()
log.Printf("listening on %s", *listenAddress)
publisher, err := xfer.NewHTTPPublisher(*publish, "fixprobe")
if err != nil {
log.Fatal(err)
}
for range time.Tick(*publishInterval) {
publisher.Publish(fixedReport)

115
experimental/graphviz/handle.go
View File

@@ -1,115 +0,0 @@
package main
import (
"fmt"
"io"
"net/http"
"os/exec"
"sort"
"strings"
"github.com/weaveworks/scope/render"
"github.com/weaveworks/scope/report"
)
func handleTXT(r Reporter) http.HandlerFunc {
return func(w http.ResponseWriter, req *http.Request) {
w.Header().Set("Content-Type", "text/plain")
renderer := render.LeafMap{Selector: report.SelectEndpoint, Mapper: mapFunc(req), Pseudo: nil}
dot(w, renderer.Render(r.Report()))
//report.Render(r.Report(), report.SelectEndpoint, mapFunc(req), report.NoPseudoNode))
}
}
func handleSVG(r Reporter) http.HandlerFunc {
return func(w http.ResponseWriter, req *http.Request) {
cmd := exec.Command(engine(req), "-Tsvg")
wc, err := cmd.StdinPipe()
if err != nil {
http.Error(w, err.Error(), http.StatusInternalServerError)
return
}
cmd.Stdout = w
renderer := render.LeafMap{Selector: report.SelectEndpoint, Mapper: mapFunc(req), Pseudo: nil}
dot(wc, renderer.Render(r.Report()))
wc.Close()
w.Header().Set("Content-Type", "image/svg+xml")
if err := cmd.Run(); err != nil {
http.Error(w, err.Error(), http.StatusInternalServerError)
return
}
}
}
func handleHTML(w http.ResponseWriter, r *http.Request) {
w.Header().Set("Content-Type", "text/html")
fmt.Fprintf(w, "<html><head>\n")
fmt.Fprintf(w, `<meta http-equiv="refresh" content="3">`+"\n")
fmt.Fprintf(w, "</head><body>\n")
fmt.Fprintf(w, `<center><img src="/svg?%s" width="100%%" height="95%%"></center>`+"\n", r.URL.Query().Encode())
fmt.Fprintf(w, "</body></html>\n")
}
func dot(w io.Writer, m map[string]render.RenderableNode) {
fmt.Fprintf(w, "digraph G {\n")
fmt.Fprintf(w, "\tgraph [ overlap=false ];\n")
fmt.Fprintf(w, "\tnode [ shape=circle, style=filled ];\n")
fmt.Fprintf(w, "\toutputorder=edgesfirst;\n")
fmt.Fprintf(w, "\n")
// Sorting the nodes seems to stop jumpiness.
nodes := make(sort.StringSlice, 0, len(m))
for _, node := range m {
nodes = append(nodes, fmt.Sprintf("\t\"%s\" [label=\"%s\n%s\"];\n", node.ID, node.LabelMajor, node.LabelMinor))
}
sort.Sort(nodes)
for _, s := range nodes {
fmt.Fprint(w, s)
}
fmt.Fprintf(w, "\n")
// Add ranking information by default.
// Non-dot engines don't seem to be harmed by it.
same := map[string][]string{}
for _, node := range m {
k, v := node.LabelMajor, fmt.Sprintf(`"%s"`, node.ID)
same[k] = append(same[k], v)
}
for _, ids := range same {
fmt.Fprintf(w, "\t{ rank=same; %s }\n", strings.Join(ids, " "))
}
fmt.Fprintf(w, "\n")
for _, src := range m {
for _, dstID := range src.Adjacency {
fmt.Fprintf(w, "\t\"%s\" -> \"%s\";\n", src.ID, dstID)
}
}
fmt.Fprintf(w, "}\n")
}
func engine(r *http.Request) string {
engine := r.FormValue("engine")
if engine == "" {
engine = "dot"
}
return engine
}
func mapFunc(r *http.Request) render.LeafMapFunc {
switch strings.ToLower(r.FormValue("map_func")) {
case "hosts", "networkhost", "networkhostname":
return render.MapAddressIdentity
}
return render.MapProcessIdentity
}
func classView(r *http.Request) bool {
return r.FormValue("class_view") == "true"
}

55
experimental/graphviz/main.go
View File

@@ -1,55 +0,0 @@
package main
import (
"flag"
"fmt"
"log"
"net/http"
"os"
"os/signal"
"strconv"
"strings"
"syscall"
"time"
"github.com/weaveworks/scope/xfer"
)
func main() {
var (
defaultProbes = fmt.Sprintf("localhost:%d", xfer.ProbePort)
probes = flag.String("probes", defaultProbes, "list of probe endpoints, comma separated")
batch = flag.Duration("batch", 1*time.Second, "batch interval")
window = flag.Duration("window", 15*time.Second, "window")
listen = flag.String("http.address", ":"+strconv.Itoa(xfer.AppPort), "webserver listen address")
)
flag.Parse()
xfer.MaxBackoff = 10 * time.Second
c := xfer.NewCollector(*batch, "id")
for _, addr := range strings.Split(*probes, ",") {
c.Add(addr)
}
defer c.Stop()
lifo := NewReportLIFO(c, *window)
defer lifo.Stop()
http.Handle("/svg", handleSVG(lifo))
http.Handle("/txt", handleTXT(lifo))
http.Handle("/", http.HandlerFunc(handleHTML))
irq := interrupt()
go func() {
log.Printf("listening on %s", *listen)
log.Print(http.ListenAndServe(*listen, nil))
irq <- syscall.SIGINT
}()
<-irq
log.Printf("shutting down")
}
func interrupt() chan os.Signal {
c := make(chan os.Signal)
signal.Notify(c, syscall.SIGINT, syscall.SIGTERM)
return c
}

90
experimental/graphviz/report_lifo.go
View File

@@ -1,90 +0,0 @@
package main
import (
"time"
"github.com/weaveworks/scope/report"
)
// Copy/paste from app/report_lifo.go
// Reporter XXX
type Reporter interface {
Report() report.Report
}
type timedReport struct {
report.Report
Timestamp time.Time
}
// ReportLIFO XXX
type ReportLIFO struct {
reports []timedReport
requests chan chan report.Report
quit chan chan struct{}
}
type reporter interface {
Reports() <-chan report.Report
}
// NewReportLIFO XXX
func NewReportLIFO(r reporter, maxAge time.Duration) *ReportLIFO {
l := ReportLIFO{
reports: []timedReport{},
requests: make(chan chan report.Report),
quit: make(chan chan struct{}),
}
go func() {
for {
select {
case report := <-r.Reports():
tr := timedReport{
Timestamp: time.Now(),
Report: report,
}
l.reports = append(l.reports, tr)
l.reports = cleanOld(l.reports, time.Now().Add(-maxAge))
case req := <-l.requests:
report := report.MakeReport()
for _, r := range l.reports {
report.Merge(r.Report)
}
req <- report
case q := <-l.quit:
close(q)
return
}
}
}()
return &l
}
// Stop XXX
func (r *ReportLIFO) Stop() {
q := make(chan struct{})
r.quit <- q
<-q
}
// Report XXX
func (r *ReportLIFO) Report() report.Report {
req := make(chan report.Report)
r.requests <- req
return <-req
}
func cleanOld(reports []timedReport, threshold time.Time) []timedReport {
res := make([]timedReport, 0, len(reports))
for _, tr := range reports {
if tr.Timestamp.Before(threshold) {
continue
}
res = append(res, tr)
}
return res
}

19
experimental/oneshot/Makefile
View File

@@ -1,19 +0,0 @@
.PHONY: all vet lint build test clean
all: build test vet lint
vet:
go vet ./...
lint:
golint .
build:
go build
test:
go test
clean:
go clean

60
experimental/oneshot/main.go
View File

@@ -1,60 +0,0 @@
package main
import (
"encoding/json"
"flag"
"fmt"
_ "net/http/pprof"
"os"
"os/signal"
"strings"
"syscall"
"time"
"github.com/weaveworks/scope/report"
"github.com/weaveworks/scope/xfer"
)
func main() {
var (
defaultProbes = fmt.Sprintf("localhost:%d", xfer.ProbePort)
probes = flag.String("probes", defaultProbes, "list of probe endpoints, comma separated")
)
flag.Parse()
if len(flag.Args()) != 0 {
flag.Usage()
os.Exit(1)
}
// Collector deals with the probes, and generates merged reports.
xfer.MaxBackoff = 1 * time.Second
c := xfer.NewCollector(1*time.Second, "id")
for _, addr := range strings.Split(*probes, ",") {
c.Add(addr)
}
defer c.Stop()
report := report.MakeReport()
irq := interrupt()
OUTER:
for {
select {
case r := <-c.Reports():
report.Merge(r)
case <-irq:
break OUTER
}
}
b, err := json.Marshal(report)
if err != nil {
panic(err)
}
fmt.Print(string(b))
}
func interrupt() chan os.Signal {
c := make(chan os.Signal)
signal.Notify(c, syscall.SIGINT, syscall.SIGTERM)
return c
}

32
probe/main.go
View File

@@ -2,6 +2,7 @@ package main
import (
"flag"
"fmt"
"log"
"net"
"net/http"
@@ -9,7 +10,6 @@ import (
"os"
"os/signal"
"runtime"
"strconv"
"strings"
"syscall"
"time"
@@ -29,10 +29,11 @@ var version = "dev" // set at build time
func main() {
var (
targets = []string{fmt.Sprintf("localhost:%d", xfer.AppPort), fmt.Sprintf("scope.weave.local:%d", xfer.AppPort)}
token = flag.String("token", "default-token", "probe token")
httpListen = flag.String("http.listen", "", "listen address for HTTP profiling and instrumentation server")
publishInterval = flag.Duration("publish.interval", 3*time.Second, "publish (output) interval")
spyInterval = flag.Duration("spy.interval", time.Second, "spy (scan) interval")
listen = flag.String("listen", ":"+strconv.Itoa(xfer.ProbePort), "listen address")
prometheusEndpoint = flag.String("prometheus.endpoint", "/metrics", "Prometheus metrics exposition endpoint (requires -http.listen)")
spyProcs = flag.Bool("processes", true, "report processes (needs root)")
dockerEnabled = flag.Bool("docker", false, "collect Docker-related attributes for processes")
@@ -46,14 +47,13 @@ func main() {
captureOff = flag.Duration("capture.off", 5*time.Second, "packet capture duty cycle 'off'")
)
flag.Parse()
log.SetFlags(log.Lmicroseconds)
if len(flag.Args()) != 0 {
flag.Usage()
os.Exit(1)
log.Printf("probe starting, version %s", version)
if len(flag.Args()) > 0 {
targets = flag.Args()
}
log.Printf("probe version %s", version)
log.Printf("publishing to: %s", strings.Join(targets, ", "))
procspy.SetProcRoot(*procRoot)
@@ -74,11 +74,10 @@ func main() {
log.Printf("warning: process reporting enabled, but that requires root to find everything")
}
publisher, err := xfer.NewTCPPublisher(*listen)
if err != nil {
log.Fatal(err)
}
defer publisher.Close()
publisherFactory := func(target string) (xfer.Publisher, error) { return xfer.NewHTTPPublisher(target, *token) }
publishers := xfer.NewMultiPublisher(publisherFactory)
resolver := newStaticResolver(targets, publishers.Add)
defer resolver.Stop()
addrs, err := net.InterfaceAddrs()
if err != nil {
@@ -147,8 +146,6 @@ func main() {
}
}
log.Printf("listening on %s", *listen)
quit := make(chan struct{})
defer close(quit)
go func() {
@@ -163,7 +160,9 @@ func main() {
case <-pubTick:
publishTicks.WithLabelValues().Add(1)
r.Window = *publishInterval
publisher.Publish(r)
if err := publishers.Publish(r); err != nil {
log.Printf("publish: %v", err)
}
r = report.MakeReport()
case <-spyTick:
@@ -184,7 +183,6 @@ func main() {
}
}
}()
log.Printf("%s", <-interrupt())
}

2
app/resolver.go → probe/resolver.go
View File

@@ -57,7 +57,7 @@ func prepareNames(strs []string) []peer {
continue
}
} else {
hostname, port = s, strconv.Itoa(xfer.ProbePort)
hostname, port = s, strconv.Itoa(xfer.AppPort)
}
results = append(results, peer{hostname, port})

10
app/resolver_test.go → probe/resolver_test.go
View File

@@ -43,26 +43,26 @@ func TestResolver(t *testing.T) {
t.Errorf("line %d: want %q, have %q", line, want, have)
}
case <-time.After(time.Millisecond):
t.Fatalf("line %d: didn't get add in time", line)
t.Errorf("line %d: didn't get add in time", line)
}
}
// Initial resolve should just give us IPs
assertAdd(ip1 + port)
assertAdd(fmt.Sprintf("%s:%d", ip2, xfer.ProbePort))
assertAdd(fmt.Sprintf("%s:%d", ip2, xfer.AppPort))
// Trigger another resolve with a tick; again,
// just want ips.
c <- time.Now()
assertAdd(ip1 + port)
assertAdd(fmt.Sprintf("%s:%d", ip2, xfer.ProbePort))
assertAdd(fmt.Sprintf("%s:%d", ip2, xfer.AppPort))
ip3 := "1.2.3.4"
ips = map[string][]net.IP{"symbolic.name": makeIPs(ip3)}
c <- time.Now() // trigger a resolve
assertAdd(ip3 + port) // we want 1 add
assertAdd(ip1 + port)
assertAdd(fmt.Sprintf("%s:%d", ip2, xfer.ProbePort))
assertAdd(fmt.Sprintf("%s:%d", ip2, xfer.AppPort))
ip4 := "10.10.10.10"
ips = map[string][]net.IP{"symbolic.name": makeIPs(ip3, ip4)}
@@ -70,7 +70,7 @@ func TestResolver(t *testing.T) {
assertAdd(ip3 + port) // first add
assertAdd(ip4 + port) // second add
assertAdd(ip1 + port)
assertAdd(fmt.Sprintf("%s:%d", ip2, xfer.ProbePort))
assertAdd(fmt.Sprintf("%s:%d", ip2, xfer.AppPort))
done := make(chan struct{})
go func() { r.Stop(); close(done) }()

270
xfer/collector.go
View File

@@ -1,233 +1,79 @@
package xfer
import (
"encoding/gob"
"log"
"net"
"sync"
"time"
"github.com/weaveworks/scope/report"
)
const (
connectTimeout = 10 * time.Second
initialBackoff = 2 * time.Second
)
var (
// MaxBackoff is the maximum time between connect retries.
// It's exported so it's externally configurable.
MaxBackoff = 1 * time.Minute
// This is extracted out for mocking.
tick = time.Tick
)
// Collector describes anything that can have addresses added and removed, and
// which produces reports that represent aggregate reports from all collected
// addresses.
type Collector interface {
Add(string)
Remove(string)
Reports() <-chan report.Report
Stop()
// Reporter is something that can produce reports on demand. It's a convenient
// interface for parts of the app, and several experimental components.
type Reporter interface {
Report() report.Report
}
// realCollector connects to probes over TCP and merges reports published by those
// probes into a single one.
type realCollector struct {
in chan report.Report
out chan report.Report
peekc chan chan report.Report
add chan string
remove chan string
quit chan struct{}
id string
// Adder is something that can accept reports. It's a convenient interface for
// parts of the app, and several experimental components.
type Adder interface {
Add(report.Report)
}
// NewCollector produces and returns a report collector.
func NewCollector(batchTime time.Duration, id string) Collector {
c := &realCollector{
in: make(chan report.Report),
out: make(chan report.Report),
peekc: make(chan chan report.Report),
add: make(chan string),
remove: make(chan string),
quit: make(chan struct{}),
id: id,
// Collector receives published reports from multiple producers. It yields a
// single merged report, representing all collected reports.
type Collector struct {
mtx sync.Mutex
reports []timestampReport
window time.Duration
}
// NewCollector returns a collector ready for use.
func NewCollector(window time.Duration) *Collector {
return &Collector{
window: window,
}
go c.loop(batchTime)
return c
}
func (c *realCollector) loop(batchTime time.Duration) {
var now = time.Now
// Add adds a report to the collector's internal state. It implements Adder.
func (c *Collector) Add(rpt report.Report) {
c.mtx.Lock()
defer c.mtx.Unlock()
c.reports = append(c.reports, timestampReport{now(), rpt})
c.reports = clean(c.reports, c.window)
}
// Report returns a merged report over all added reports. It implements
// Reporter.
func (c *Collector) Report() report.Report {
c.mtx.Lock()
defer c.mtx.Unlock()
c.reports = clean(c.reports, c.window)
report := report.MakeReport()
for _, tr := range c.reports {
report.Merge(tr.report)
}
return report
}
type timestampReport struct {
timestamp time.Time
report report.Report
}
func clean(reports []timestampReport, window time.Duration) []timestampReport {
var (
tick = tick(batchTime)
current = report.MakeReport()
addrs = map[string]chan struct{}{}
wg = &sync.WaitGroup{} // per-address goroutines
cleaned = make([]timestampReport, 0, len(reports))
oldest = now().Add(-window)
)
add := func(ip string) {
if _, ok := addrs[ip]; ok {
return
}
addrs[ip] = make(chan struct{})
wg.Add(1)
go func(quit chan struct{}) {
defer wg.Done()
c.reportCollector(ip, quit)
}(addrs[ip])
}
remove := func(ip string) {
q, ok := addrs[ip]
if !ok {
return // hmm
}
close(q)
delete(addrs, ip)
}
for {
select {
case <-tick:
c.out <- current
current = report.MakeReport()
case pc := <-c.peekc:
copy := report.MakeReport()
copy.Merge(current)
pc <- copy
case r := <-c.in:
if err := r.Validate(); err != nil {
log.Printf("Received invalid report: %v", err)
continue
}
current.Merge(r)
case ip := <-c.add:
add(ip)
case ip := <-c.remove:
remove(ip)
case <-c.quit:
for _, q := range addrs {
close(q)
}
wg.Wait()
return
}
}
}
// Add adds an address to be collected from.
func (c *realCollector) Add(addr string) {
c.add <- addr
}
// Remove removes a previously-added address.
func (c *realCollector) Remove(addr string) {
c.remove <- addr
}
// Reports returns the report chan. It must be consumed by the client, or the
// collector will break.
func (c *realCollector) Reports() <-chan report.Report {
return c.out
}
func (c *realCollector) peek() report.Report {
pc := make(chan report.Report)
c.peekc <- pc
return <-pc
}
// Stop terminates the collector.
func (c *realCollector) Stop() {
close(c.quit)
}
// reportCollector is the loop to connect to a single Probe. It'll keep
// running until the quit channel is closed.
func (c *realCollector) reportCollector(ip string, quit <-chan struct{}) {
backoff := initialBackoff / 2
for {
backoff *= 2
if backoff > MaxBackoff {
backoff = MaxBackoff
}
select {
default:
case <-quit:
return
}
log.Printf("dialing %v (backoff %v)", ip, backoff)
conn, err := net.DialTimeout("tcp", ip, connectTimeout)
if err != nil {
log.Print(err)
select {
case <-time.After(backoff):
continue
case <-quit:
return
}
}
log.Printf("connected to %v", ip)
go func() {
<-quit
log.Printf("closing %v collector", ip)
conn.Close()
}()
// Connection accepted.
if err := gob.NewEncoder(conn).Encode(HandshakeRequest{ID: c.id}); err != nil {
log.Printf("handshake error: %v", err)
break
}
dec := gob.NewDecoder(conn)
for {
var report report.Report
err := dec.Decode(&report)
// Don't complain of errors when shutting down.
select {
default:
case <-quit:
return
}
if err != nil {
log.Printf("decode error: %v", err)
break
}
select {
case c.in <- report:
case <-quit:
return
}
// Reset the backoff iff we have a connection which works. This
// prevents us from spamming probes with multiple addresses (since
// the probe closes everything but a single connection).
backoff = initialBackoff
}
// Prevent a 100% CPU loop when the probe is closing the
// connection right away (which happens on a probe which already
// has a client)
select {
case <-time.After(backoff):
case <-quit:
return
for _, tr := range reports {
if tr.timestamp.Before(oldest) {
continue
}
cleaned = append(cleaned, tr)
}
return cleaned
}

103
xfer/collector_internal_test.go
View File

@@ -1,103 +0,0 @@
package xfer
import (
"encoding/gob"
"io/ioutil"
"log"
"net"
"reflect"
"runtime"
"testing"
"time"
"github.com/weaveworks/scope/report"
"github.com/weaveworks/scope/test"
)
func TestCollector(t *testing.T) {
log.SetOutput(ioutil.Discard)
// Swap out ticker
publish := make(chan time.Time)
oldTick := tick
tick = func(time.Duration) <-chan time.Time { return publish }
defer func() { tick = oldTick }()
// Build a collector
collector := NewCollector(time.Second, "id")
defer collector.Stop()
concreteCollector, ok := collector.(*realCollector)
if !ok {
t.Fatal("type assertion failure")
}
// Build a test publisher
reports := make(chan interface{})
ln := testPublisher(t, reports)
defer ln.Close()
// Connect the collector to the test publisher
addr := ln.Addr().String()
collector.Add(addr)
collector.Add(addr) // test duplicate case
runtime.Gosched() // make sure it connects
// Push a report through everything
r := report.Report{
Address: report.Topology{
NodeMetadatas: report.NodeMetadatas{
report.MakeAddressNodeID("a", "b"): report.MakeNodeMetadata(),
},
},
}
reports <- r
test.Poll(t, 100*time.Millisecond, 1, func() interface{} {
return len(concreteCollector.peek().Address.NodeMetadatas)
})
go func() { publish <- time.Now() }()
collected := <-collector.Reports()
if reflect.DeepEqual(r, collected) {
t.Errorf(test.Diff(r, collected))
}
collector.Remove(addr)
collector.Remove(addr) // test duplicate case
}
func TestCollectorQuitWithActiveConnections(t *testing.T) {
c := NewCollector(time.Second, "id")
c.Add("1.2.3.4:56789")
c.Stop()
}
func testPublisher(t *testing.T, input <-chan interface{}) net.Listener {
addr, err := net.ResolveTCPAddr("tcp4", "127.0.0.1:0")
if err != nil {
t.Fatal(err)
}
ln, err := net.ListenTCP("tcp4", addr)
if err != nil {
t.Fatal(err)
}
go func() {
conn, err := ln.Accept()
if err != nil {
t.Log(err)
return
}
defer conn.Close()
for {
enc := gob.NewEncoder(conn)
for v := range input {
if err := enc.Encode(v); err != nil {
t.Error(err)
return
}
}
}
}()
return ln
}

41
xfer/collector_test.go Normal file
View File

@@ -0,0 +1,41 @@
package xfer_test
import (
"reflect"
"time"
"github.com/weaveworks/scope/test"
"github.com/weaveworks/scope/report"
"github.com/weaveworks/scope/xfer"
"testing"
)
func TestCollector(t *testing.T) {
window := time.Millisecond
c := xfer.NewCollector(window)
r1 := report.MakeReport()
r1.Endpoint.NodeMetadatas["foo"] = report.MakeNodeMetadata()
r2 := report.MakeReport()
r2.Endpoint.NodeMetadatas["bar"] = report.MakeNodeMetadata()
if want, have := report.MakeReport(), c.Report(); !reflect.DeepEqual(want, have) {
t.Error(test.Diff(want, have))
}
c.Add(r1)
if want, have := r1, c.Report(); !reflect.DeepEqual(want, have) {
t.Error(test.Diff(want, have))
}
c.Add(r2)
merged := report.MakeReport()
merged.Merge(r1)
merged.Merge(r2)
if want, have := merged, c.Report(); !reflect.DeepEqual(want, have) {
t.Error(test.Diff(want, have))
}
}

72
xfer/merge_test.go
View File

@@ -1,72 +0,0 @@
package xfer_test
import (
"io/ioutil"
"log"
"testing"
"time"
"github.com/weaveworks/scope/report"
"github.com/weaveworks/scope/xfer"
)
func TestMerge(t *testing.T) {
log.SetOutput(ioutil.Discard)
var (
p1Addr = "localhost:7888"
p2Addr = "localhost:7889"
)
p1, err := xfer.NewTCPPublisher(p1Addr)
if err != nil {
t.Fatal(err)
}
defer p1.Close()
p2, err := xfer.NewTCPPublisher(p2Addr)
if err != nil {
t.Fatal(err)
}
defer p2.Close()
batchTime := 100 * time.Millisecond
c := xfer.NewCollector(batchTime, "id")
c.Add(p1Addr)
c.Add(p2Addr)
defer c.Stop()
time.Sleep(batchTime / 10) // connect
k1, k2 := report.MakeHostNodeID("p1"), report.MakeHostNodeID("p2")
{
r := report.MakeReport()
r.Host.NodeMetadatas[k1] = report.MakeNodeMetadataWith(map[string]string{"host_name": "test1"})
p1.Publish(r)
}
{
r := report.MakeReport()
r.Host.NodeMetadatas[k2] = report.MakeNodeMetadataWith(map[string]string{"host_name": "test2"})
p2.Publish(r)
}
success := make(chan struct{})
go func() {
defer close(success)
for r := range c.Reports() {
if r.Host.NodeMetadatas[k1].Metadata["host_name"] != "test1" {
continue
}
if r.Host.NodeMetadatas[k2].Metadata["host_name"] != "test2" {
continue
}
return
}
}()
select {
case <-success:
case <-time.After(2 * batchTime):
t.Errorf("collector didn't capture both reports")
}
}

6
xfer/ports.go
View File

@@ -1,10 +1,8 @@
package xfer
var (
// ProbePort is the default port that the probe(s) will listen on to
// publish reports.
ProbePort = 4030
// 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
)

193
xfer/publisher.go
View File

@@ -1,132 +1,119 @@
package xfer
import (
"bytes"
"encoding/gob"
"io"
"fmt"
"log"
"net"
"net/http"
"net/url"
"strings"
"sync"
"github.com/weaveworks/scope/report"
)
// Publisher provides a way to send reports upstream.
// Publisher is something which can send a report to a remote collector.
type Publisher interface {
Publish(report.Report)
Close()
Publish(report.Report) error
}
// TCPPublisher is a Publisher implementation which uses TCP and gob encoding.
type TCPPublisher struct {
msg chan report.Report
closer io.Closer
// HTTPPublisher publishes reports by POST to a fixed endpoint.
type HTTPPublisher struct {
url string
token string
}
// HandshakeRequest contains the unique ID of the connecting app.
type HandshakeRequest struct {
ID string
}
// NewTCPPublisher listens for connections on listenAddress. Only one client
// is accepted at a time; other clients are accepted, but disconnected right
// away. Reports published via publish() will be written to the connected
// client, if any. Gentle shutdown of the returned publisher via close().
func NewTCPPublisher(listenAddress string) (*TCPPublisher, error) {
listener, err := net.Listen("tcp", listenAddress)
// NewHTTPPublisher returns an HTTPPublisher ready for use.
func NewHTTPPublisher(target, token string) (*HTTPPublisher, error) {
if !strings.HasPrefix(target, "http") {
target = "http://" + target
}
u, err := url.Parse(target)
if err != nil {
return nil, err
}
if u.Path == "" {
u.Path = "/api/report"
}
return &HTTPPublisher{
url: u.String(),
token: token,
}, nil
}
p := &TCPPublisher{
msg: make(chan report.Report),
closer: listener,
// Publish publishes the report to the URL.
func (p HTTPPublisher) Publish(rpt report.Report) error {
var buf bytes.Buffer
if err := gob.NewEncoder(&buf).Encode(rpt); err != nil {
return err
}
req, err := http.NewRequest("POST", p.url, &buf)
if err != nil {
return err
}
req.Header.Set("Authorization", p.token)
resp, err := http.DefaultClient.Do(req)
if err != nil {
return err
}
defer resp.Body.Close()
if resp.StatusCode != http.StatusOK {
return fmt.Errorf(resp.Status)
}
return nil
}
// MultiPublisher implements Publisher over a set of publishers.
type MultiPublisher struct {
mtx sync.RWMutex
factory func(string) (Publisher, error)
m map[string]Publisher
}
// NewMultiPublisher returns a new MultiPublisher ready for use. The factory
// should be e.g. NewHTTPPublisher, except you need to curry it over the
// probe token.
func NewMultiPublisher(factory func(string) (Publisher, error)) *MultiPublisher {
return &MultiPublisher{
factory: factory,
m: map[string]Publisher{},
}
}
// Add allows additional targets to be added dynamically. It will dedupe
// identical targets. TODO we have no good mechanism to remove.
func (p *MultiPublisher) Add(target string) {
p.mtx.Lock()
defer p.mtx.Unlock()
if _, ok := p.m[target]; ok {
return
}
go p.loop(fwd(listener))
return p, nil
}
// Close stops a TCPPublisher and closes the socket.
func (p *TCPPublisher) Close() {
close(p.msg)
p.closer.Close()
}
// Publish sens a Report to the client, if any.
func (p *TCPPublisher) Publish(msg report.Report) {
p.msg <- msg
}
func (p *TCPPublisher) loop(incoming <-chan net.Conn) {
type connEncoder struct {
net.Conn
*gob.Encoder
publisher, err := p.factory(target)
if err != nil {
log.Printf("multi-publisher: %v", err)
return
}
activeConns := map[string]connEncoder{} // host: connEncoder
p.m[target] = publisher
}
for {
select {
case conn, ok := <-incoming:
if !ok {
return // someone closed our connection chan -- weird?
}
// Publish implements Publisher by emitting the report to all publishers.
func (p *MultiPublisher) Publish(rpt report.Report) error {
p.mtx.RLock()
defer p.mtx.RUnlock()
// Don't allow multiple connections from the same remote host.
listenerID, err := getListenerID(conn)
if err != nil {
log.Printf("incoming connection: %s: %v (dropped)", conn.RemoteAddr(), err)
conn.Close()
continue
}
if _, ok := activeConns[listenerID]; ok {
log.Printf("duplicate connection: %s (dropped)", conn.RemoteAddr())
conn.Close()
continue
}
log.Printf("connection initiated: %s (%s)", conn.RemoteAddr(), listenerID)
activeConns[listenerID] = connEncoder{conn, gob.NewEncoder(conn)}
case msg, ok := <-p.msg:
if !ok {
return // someone closed our msg chan, so we're done
}
for host, connEncoder := range activeConns {
if err := connEncoder.Encoder.Encode(msg); err != nil {
log.Printf("connection terminated: %s: %v", connEncoder.Conn.RemoteAddr(), err)
connEncoder.Conn.Close()
delete(activeConns, host)
}
}
var errs []string
for _, publisher := range p.m {
if err := publisher.Publish(rpt); err != nil {
errs = append(errs, err.Error())
}
}
}
func getListenerID(c net.Conn) (string, error) {
var req HandshakeRequest
if err := gob.NewDecoder(c).Decode(&req); err != nil {
return "", err
if len(errs) > 0 {
return fmt.Errorf(strings.Join(errs, "; "))
}
return req.ID, nil
}
func fwd(ln net.Listener) chan net.Conn {
c := make(chan net.Conn)
go func() {
defer close(c)
for {
conn, err := ln.Accept()
if err != nil {
log.Printf("%s: %s", ln.Addr(), err)
return
}
c <- conn
}
}()
return c
return nil
}

146
xfer/publisher_test.go
View File

@@ -2,120 +2,72 @@ package xfer_test
import (
"encoding/gob"
"fmt"
"io/ioutil"
"log"
"net"
"net/http"
"net/http/httptest"
"reflect"
"testing"
"time"
"github.com/weaveworks/scope/report"
"github.com/weaveworks/scope/test"
"github.com/weaveworks/scope/xfer"
)
func TestTCPPublisher(t *testing.T) {
log.SetOutput(ioutil.Discard)
func TestHTTPPublisher(t *testing.T) {
var (
token = "abcdefg"
rpt = report.MakeReport()
)
// Choose a port
port, err := getFreePort()
s := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
if want, have := token, r.Header.Get("Authorization"); want != have {
t.Errorf("want %q, have %q", want, have)
}
var have report.Report
if err := gob.NewDecoder(r.Body).Decode(&have); err != nil {
t.Error(err)
return
}
if want := rpt; !reflect.DeepEqual(want, have) {
t.Error(test.Diff(want, have))
return
}
w.WriteHeader(http.StatusOK)
}))
defer s.Close()
p, err := xfer.NewHTTPPublisher(s.URL, token)
if err != nil {
t.Fatal(err)
}
// Start a publisher
p, err := xfer.NewTCPPublisher(port)
if err != nil {
t.Fatal(err)
}
defer p.Close()
// Start a raw listener
conn, err := net.Dial("tcp4", "127.0.0.1"+port)
if err != nil {
t.Fatal(err)
}
defer conn.Close()
time.Sleep(time.Millisecond)
// Send handshake
if err := gob.NewEncoder(conn).Encode(xfer.HandshakeRequest{ID: "foo"}); err != nil {
t.Fatal(err)
}
// Publish a message
p.Publish(report.Report{})
// Check it was received
var r report.Report
if err := gob.NewDecoder(conn).Decode(&r); err != nil {
t.Fatal(err)
if err := p.Publish(rpt); err != nil {
t.Error(err)
}
}
func TestPublisherClosesDuplicateConnections(t *testing.T) {
log.SetOutput(ioutil.Discard)
func TestMultiPublisher(t *testing.T) {
var (
p = &mockPublisher{}
factory = func(string) (xfer.Publisher, error) { return p, nil }
multiPublisher = xfer.NewMultiPublisher(factory)
)
// Choose a port
port, err := getFreePort()
if err != nil {
t.Fatal(err)
multiPublisher.Add("first")
if err := multiPublisher.Publish(report.MakeReport()); err != nil {
t.Error(err)
}
if want, have := 1, p.count; want != have {
t.Errorf("want %d, have %d", want, have)
}
// Start a publisher
p, err := xfer.NewTCPPublisher(port)
if err != nil {
t.Fatal(err)
multiPublisher.Add("second") // but factory returns same mockPublisher
if err := multiPublisher.Publish(report.MakeReport()); err != nil {
t.Error(err)
}
defer p.Close()
// Connect a listener
conn, err := net.Dial("tcp4", "127.0.0.1"+port)
if err != nil {
t.Fatal(err)
}
defer conn.Close()
if err := gob.NewEncoder(conn).Encode(xfer.HandshakeRequest{ID: "foo"}); err != nil {
t.Fatal(err)
}
time.Sleep(time.Millisecond)
// Try to connect the same listener
dupconn, err := net.Dial("tcp4", "127.0.0.1"+port)
if err != nil {
t.Fatal(err)
}
// Send handshake
if err := gob.NewEncoder(dupconn).Encode(xfer.HandshakeRequest{ID: "foo"}); err != nil {
t.Fatal(err)
}
defer dupconn.Close()
// Publish a message
p.Publish(report.Report{})
// The first listener should receive it
var r report.Report
if err := gob.NewDecoder(conn).Decode(&r); err != nil {
t.Fatal(err)
}
// The duplicate listener should have an error
if err := gob.NewDecoder(dupconn).Decode(&r); err == nil {
t.Errorf("expected error, got none")
} else {
t.Logf("dupconn got expected error: %v", err)
if want, have := 3, p.count; want != have {
t.Errorf("want %d, have %d", want, have)
}
}
func getFreePort() (string, error) {
ln, err := net.Listen("tcp4", ":0")
if err != nil {
return "", fmt.Errorf("Listen: %v", err)
}
defer ln.Close()
_, port, err := net.SplitHostPort(ln.Addr().String())
if err != nil {
return "", fmt.Errorf("SplitHostPort(%s): %v", ln.Addr().String(), err)
}
return ":" + port, nil
}
type mockPublisher struct{ count int }
func (p *mockPublisher) Publish(report.Report) error { p.count++; return nil }