Merge pull request #2547 from weaveworks/prometheus-metrics

Prometheus metrics

.gitmodules

@@ -82,3 +82,24 @@
 [submodule "vendor/github.com/weaveworks/docker"]
 	path = vendor/github.com/weaveworks/docker
 	url = https://github.com/weaveworks/docker
+[submodule "vendor/github.com/beorn7/perks"]
+	path = vendor/github.com/beorn7/perks
+	url = https://github.com/beorn7/perks
+[submodule "vendor/github.com/prometheus/procfs"]
+	path = vendor/github.com/prometheus/procfs
+	url = https://github.com/prometheus/procfs
+[submodule "vendor/github.com/prometheus/client_golang"]
+	path = vendor/github.com/prometheus/client_golang
+	url = https://github.com/prometheus/client_golang
+[submodule "vendor/github.com/prometheus/client_model"]
+	path = vendor/github.com/prometheus/client_model
+	url = https://github.com/prometheus/client_model
+[submodule "vendor/github.com/prometheus/common"]
+	path = vendor/github.com/prometheus/common
+	url = https://github.com/prometheus/common
+[submodule "vendor/github.com/golang/protobuf"]
+	path = vendor/github.com/golang/protobuf
+	url = https://github.com/golang/protobuf
+[submodule "vendor/github.com/matttproud/golang_protobuf_extensions"]
+	path = vendor/github.com/matttproud/golang_protobuf_extensions
+	url = https://github.com/matttproud/golang_protobuf_extensions

ipam/ring/expvar.go

@@ -1,31 +0,0 @@
-package ring
-
-/* Exported variables for monitoring and management */
-
-import (
-	"expvar"
-	"strconv"
-)
-
-var (
-	expRingSize    = expvar.NewMap("ipam.ringSize")
-	expRingEntries = expvar.NewMap("ipam.ringEntries")
-)
-
-type _int int
-
-func (i _int) String() string {
-	return strconv.Itoa(int(i))
-}
-
-type _uint32 uint32
-
-func (i _uint32) String() string {
-	return strconv.FormatUint(uint64(i), 10)
-}
-
-func (r *Ring) updateExportedVariables() {
-	ringName := r.Start.String()
-	expRingSize.Set(ringName, _uint32(r.End-r.Start))
-	expRingEntries.Set(ringName, _int(len(r.Entries)))
-}

ipam/ring/ring.go

@@ -114,7 +114,6 @@ func New(start, end address.Address, peer mesh.PeerName, f OnUpdate) *Ring {
 	common.Assert(start < end)
 
 	ring := &Ring{Start: start, End: end, Peer: peer, Entries: make([]*entry, 0), onUpdate: f}
-	ring.updateExportedVariables()
 	return ring
 }
 
@@ -151,7 +150,6 @@ func (r *Ring) GrantRangeToHost(start, end address.Address, peer mesh.PeerName)
 	r.assertInvariants()
 	defer r.trackUpdates()()
 	defer r.assertInvariants()
-	defer r.updateExportedVariables()
 
 	// ----------------- Start of Checks -----------------
 
@@ -218,7 +216,6 @@ func (r *Ring) GrantRangeToHost(start, end address.Address, peer mesh.PeerName)
 func (r *Ring) Merge(gossip Ring) (bool, error) {
 	r.assertInvariants()
 	defer r.trackUpdates()()
-	defer r.updateExportedVariables()
 
 	// Don't panic when checking the gossiped in ring.
 	// In this case just return any error found.
@@ -404,7 +401,6 @@ func (r *Ring) ClaimForPeers(peers []mesh.PeerName) {
 
 	defer r.trackUpdates()()
 	defer r.assertInvariants()
-	defer r.updateExportedVariables()
 	defer func() {
 		e := r.Entries[len(r.Entries)-1]
 		common.Assert(address.Add(e.Token, address.Offset(e.Free)) == r.End)
@@ -455,7 +451,6 @@ func (r *Ring) String() string {
 func (r *Ring) ReportFree(freespace map[address.Address]address.Count) (updated bool) {
 	r.assertInvariants()
 	defer r.assertInvariants()
-	defer r.updateExportedVariables()
 
 	common.Assert(!r.Empty())
 	entries := r.Entries
@@ -564,7 +559,6 @@ func (r *Ring) Transfer(from, to mesh.PeerName) []address.Range {
 	defer r.trackUpdates()()
 	defer r.trackUpdatesOfPeer(from)()
 	defer r.assertInvariants()
-	defer r.updateExportedVariables()
 
 	var newRanges []address.Range
 

ipam/space/space.go

@@ -88,6 +88,14 @@ func (s *Space) Claim(addr address.Address) error {
 	return nil
 }
 
+func (s *Space) NumOwnedAddresses() address.Count {
+	res := address.Count(0)
+	for i := 0; i < len(s.ours); i += 2 {
+		res += address.Length(s.ours[i+1], s.ours[i])
+	}
+	return res
+}
+
 func (s *Space) NumFreeAddresses() address.Count {
 	res := address.Count(0)
 	for i := 0; i < len(s.free); i += 2 {

ipam/space/space_test.go

@@ -98,16 +98,19 @@ func TestSpaceAllocate(t *testing.T) {
 
 	space1 := makeSpace(start, size)
 	require.Equal(t, address.Count(20), space1.NumFreeAddresses())
+	require.Equal(t, address.Count(0), space1.NumOwnedAddresses())
 	space1.assertInvariants()
 
 	_, addr1 := space1.Allocate(address.NewRange(start, size))
 	require.Equal(t, testAddr1, addr1.String(), "address")
 	require.Equal(t, address.Count(19), space1.NumFreeAddresses())
+	require.Equal(t, address.Count(1), space1.NumOwnedAddresses())
 	space1.assertInvariants()
 
 	_, addr2 := space1.Allocate(address.NewRange(start, size))
 	require.False(t, addr2.String() == testAddr1, "address")
 	require.Equal(t, address.Count(18), space1.NumFreeAddresses())
+	require.Equal(t, address.Count(2), space1.NumOwnedAddresses())
 	require.Equal(t, address.Count(13), space1.NumFreeAddressesInRange(address.Range{Start: ip(testAddr1), End: ip(testAddrx)}))
 	require.Equal(t, address.Count(18), space1.NumFreeAddressesInRange(address.Range{Start: ip(testAddr1), End: ip(testAddry)}))
 	space1.assertInvariants()

ipam/status.go

@@ -10,6 +10,7 @@ type Status struct {
 	Paxos            *paxos.Status
 	Range            string
 	RangeNumIPs      int
+	ActiveIPs        int
 	DefaultSubnet    string
 	Entries          []EntryStatus
 	PendingClaims    []ClaimStatus
@@ -52,6 +53,7 @@ func NewStatus(allocator *Allocator, defaultSubnet address.CIDR) *Status {
 			paxosStatus,
 			allocator.universe.String(),
 			int(allocator.universe.Size()),
+			int(allocator.space.NumOwnedAddresses()),
 			defaultSubnet.String(),
 			newEntryStatusSlice(allocator),
 			newClaimStatusSlice(allocator),

prog/weaver/http.go

@@ -20,15 +20,7 @@ import (
 )
 
 var rootTemplate = template.New("root").Funcs(map[string]interface{}{
-	"countDNSEntries": func(entries []nameserver.EntryStatus) int {
-		count := 0
-		for _, entry := range entries {
-			if entry.Tombstone == 0 {
-				count++
-			}
-		}
-		return count
-	},
+	"countDNSEntries": countDNSEntries,
 	"printList": func(list []string) string {
 		if len(list) == 0 {
 			return "none"
@@ -55,27 +47,28 @@ var rootTemplate = template.New("root").Funcs(map[string]interface{}{
 			s.ips += entry.Size
 		}
 
-		printOwned := func(name string, nickName string, reachable bool, ips uint32) {
-			reachableStr := ""
-			if !reachable {
-				reachableStr = "- unreachable!"
-			}
+		printOwned := func(name string, nickName string, info string, ips uint32) {
 			percentageRanges := float32(ips) * 100.0 / float32(status.RangeNumIPs)
 
 			displayName := name + "(" + nickName + ")"
 			fmt.Fprintf(&buffer, "%-37v %8d IPs (%04.1f%% of total) %s\n",
-				displayName, ips, percentageRanges, reachableStr)
+				displayName, ips, percentageRanges, info)
 		}
 
 		// print the local info first
 		if ourStats := peerStats[router.Name]; ourStats != nil {
-			printOwned(router.Name, ourStats.nickname, true, ourStats.ips)
+			activeStr := fmt.Sprintf("(%d active)", status.ActiveIPs)
+			printOwned(router.Name, ourStats.nickname, activeStr, ourStats.ips)
 		}
 
 		// and then the rest
 		for peer, stats := range peerStats {
 			if peer != router.Name {
-				printOwned(peer, stats.nickname, stats.reachable, stats.ips)
+				reachableStr := ""
+				if !stats.reachable {
+					reachableStr = "- unreachable!"
+				}
+				printOwned(peer, stats.nickname, reachableStr, stats.ips)
 			}
 		}
 
@@ -118,6 +111,26 @@ var rootTemplate = template.New("root").Funcs(map[string]interface{}{
 	"trimSuffix": strings.TrimSuffix,
 })
 
+func countDNSEntries(entries []nameserver.EntryStatus) int {
+	count := 0
+	for _, entry := range entries {
+		if entry.Tombstone == 0 {
+			count++
+		}
+	}
+	return count
+}
+
+func countDNSEntriesForPeer(peername string, entries []nameserver.EntryStatus) int {
+	count := 0
+	for _, entry := range entries {
+		if entry.Tombstone == 0 && entry.Origin == peername {
+			count++
+		}
+	}
+	return count
+}
+
 // Print counts in a specified order
 func printCounts(counts map[string]int, keys []string) string {
 	var stringCounts []string
@@ -263,6 +276,7 @@ type WeaveStatus struct {
 	DNS          *nameserver.Status         `json:"DNS,omitempty"`
 }
 
+// Read-only functions, suitable for exposing on an unprotected socket
 func HandleHTTP(muxRouter *mux.Router, version string, router *weave.NetworkRouter, allocator *ipam.Allocator, defaultSubnet address.CIDR, ns *nameserver.Nameserver, dnsserver *nameserver.DNSServer) {
 	status := func() WeaveStatus {
 		return WeaveStatus{

prog/weaver/main.go

@@ -132,6 +132,7 @@ func main() {
 		bufSzMB            int
 		noDiscovery        bool
 		httpAddr           string
+		statusAddr         string
 		ipamConfig         ipamConfig
 		dockerAPI          string
 		peers              []string
@@ -165,6 +166,7 @@ func main() {
 	mflag.BoolVar(&noDiscovery, []string{"#nodiscovery", "#-nodiscovery", "-no-discovery"}, false, "disable peer discovery")
 	mflag.IntVar(&bufSzMB, []string{"#bufsz", "-bufsz"}, 8, "capture buffer size in MB")
 	mflag.StringVar(&httpAddr, []string{"#httpaddr", "#-httpaddr", "-http-addr"}, "", "address to bind HTTP interface to (disabled if blank, absolute path indicates unix domain socket)")
+	mflag.StringVar(&statusAddr, []string{"-status-addr"}, "", "address to bind status+metrics interface to (disabled if blank, absolute path indicates unix domain socket)")
 	mflag.StringVar(&ipamConfig.Mode, []string{"-ipalloc-init"}, "", "allocator initialisation strategy (consensus, seed or observer)")
 	mflag.StringVar(&ipamConfig.IPRangeCIDR, []string{"#iprange", "#-iprange", "-ipalloc-range"}, "", "IP address range reserved for automatic allocation, in CIDR notation")
 	mflag.StringVar(&ipamConfig.IPSubnetCIDR, []string{"#ipsubnet", "#-ipsubnet", "-ipalloc-default-subnet"}, "", "subnet to allocate within by default, in CIDR notation")
@@ -337,9 +339,20 @@ func main() {
 		}
 		router.HandleHTTP(muxRouter)
 		HandleHTTP(muxRouter, version, router, allocator, defaultSubnet, ns, dnsserver)
+		muxRouter.Methods("GET").Path("/metrics").Handler(metricsHandler(router, allocator, ns, dnsserver))
 		http.Handle("/", common.LoggingHTTPHandler(muxRouter))
 		Log.Println("Listening for HTTP control messages on", httpAddr)
-		go listenAndServeHTTP(httpAddr)
+		go listenAndServeHTTP(httpAddr, nil)
+	}
+
+	if statusAddr != "" {
+		muxRouter := mux.NewRouter()
+		HandleHTTP(muxRouter, version, router, allocator, defaultSubnet, ns, dnsserver)
+		muxRouter.Methods("GET").Path("/metrics").Handler(metricsHandler(router, allocator, ns, dnsserver))
+		statusMux := http.NewServeMux()
+		statusMux.Handle("/", muxRouter)
+		Log.Println("Listening for metrics requests on", statusAddr)
+		go listenAndServeHTTP(statusAddr, statusMux)
 	}
 
 	common.SignalHandlerLoop(router)
@@ -555,7 +568,7 @@ func parsePeerNames(s string) ([]mesh.PeerName, error) {
 	return peerNames, nil
 }
 
-func listenAndServeHTTP(httpAddr string) {
+func listenAndServeHTTP(httpAddr string, handler http.Handler) {
 	protocol := "tcp"
 	if strings.HasPrefix(httpAddr, "/") {
 		os.Remove(httpAddr) // in case it's there from last time
@@ -565,7 +578,7 @@ func listenAndServeHTTP(httpAddr string) {
 	if err != nil {
 		Log.Fatal("Unable to create http listener socket: ", err)
 	}
-	err = http.Serve(l, nil)
+	err = http.Serve(l, handler)
 	if err != nil {
 		Log.Fatal("Unable to create http server", err)
 	}

prog/weaver/metrics.go

@@ -0,0 +1,129 @@
+package main
+
+import (
+	"net/http"
+	"os"
+
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/prometheus/client_golang/prometheus/promhttp"
+
+	"github.com/weaveworks/weave/ipam"
+	"github.com/weaveworks/weave/nameserver"
+	"github.com/weaveworks/weave/net/address"
+	weave "github.com/weaveworks/weave/router"
+)
+
+func metricsHandler(router *weave.NetworkRouter, allocator *ipam.Allocator, ns *nameserver.Nameserver, dnsserver *nameserver.DNSServer) http.Handler {
+	reg := prometheus.NewRegistry()
+	reg.MustRegister(prometheus.NewProcessCollector(os.Getpid(), ""))
+	reg.MustRegister(newMetrics(router, allocator, ns, dnsserver))
+	return promhttp.HandlerFor(reg, promhttp.HandlerOpts{})
+}
+
+type collector struct {
+	router    *weave.NetworkRouter
+	allocator *ipam.Allocator
+	ns        *nameserver.Nameserver
+	dnsserver *nameserver.DNSServer
+}
+
+type metric struct {
+	*prometheus.Desc
+	Collect func(WeaveStatus, *prometheus.Desc, chan<- prometheus.Metric)
+}
+
+func desc(fqName, help string, variableLabels ...string) *prometheus.Desc {
+	return prometheus.NewDesc(fqName, help, variableLabels, prometheus.Labels{})
+}
+
+func intGauge(desc *prometheus.Desc, val int, labels ...string) prometheus.Metric {
+	return prometheus.MustNewConstMetric(desc, prometheus.GaugeValue, float64(val), labels...)
+}
+func uint64Counter(desc *prometheus.Desc, val uint64, labels ...string) prometheus.Metric {
+	return prometheus.MustNewConstMetric(desc, prometheus.CounterValue, float64(val), labels...)
+}
+
+var metrics []metric = []metric{
+	{desc("weave_connections", "Number of peer-to-peer connections.", "state"),
+		func(s WeaveStatus, desc *prometheus.Desc, ch chan<- prometheus.Metric) {
+			established := 0
+			for _, conn := range s.Router.Connections {
+				if conn.State == "established" {
+					established++
+				}
+			}
+
+			ch <- intGauge(desc, len(s.Router.Connections)-established, "non-established")
+			ch <- intGauge(desc, established, "established")
+		}},
+	{desc("weave_connection_termination_count", "Number of peer-to-peer connections terminated."),
+		func(s WeaveStatus, desc *prometheus.Desc, ch chan<- prometheus.Metric) {
+			ch <- uint64Counter(desc, uint64(s.Router.TerminationCount))
+		}},
+	{desc("weave_ips", "Number of IP addresses.", "state"),
+		func(s WeaveStatus, desc *prometheus.Desc, ch chan<- prometheus.Metric) {
+			ch <- intGauge(desc, s.IPAM.RangeNumIPs, "total")
+			ch <- intGauge(desc, s.IPAM.ActiveIPs, "local-used")
+		}},
+	{desc("weave_dns_entries", "Number of DNS entries.", "state"),
+		func(s WeaveStatus, desc *prometheus.Desc, ch chan<- prometheus.Metric) {
+			ch <- intGauge(desc, countDNSEntries(s.DNS.Entries), "total")
+			ch <- intGauge(desc, countDNSEntriesForPeer(s.Router.Name, s.DNS.Entries), "local")
+		}},
+	{desc("weave_flows", "Number of FastDP flows."),
+		func(s WeaveStatus, desc *prometheus.Desc, ch chan<- prometheus.Metric) {
+			if metrics := fastDPMetrics(s); metrics != nil {
+				ch <- intGauge(desc, metrics.Flows)
+			}
+		}},
+	{desc("weave_packets_total", "Number of packets transferred.", "flow"),
+		func(s WeaveStatus, desc *prometheus.Desc, ch chan<- prometheus.Metric) {
+			if metrics := fastDPMetrics(s); metrics != nil {
+				ch <- uint64Counter(desc, metrics.TotalPackets, "total")
+			}
+		}},
+	{desc("weave_bytes_total", "Number of bytes transferred.", "flow"),
+		func(s WeaveStatus, desc *prometheus.Desc, ch chan<- prometheus.Metric) {
+			if metrics := fastDPMetrics(s); metrics != nil {
+				ch <- uint64Counter(desc, metrics.TotalBytes, "total")
+			}
+		}},
+}
+
+func fastDPMetrics(s WeaveStatus) *weave.FastDPMetrics {
+	if diagMap, ok := s.Router.OverlayDiagnostics.(map[string]interface{}); ok {
+		if diag, ok := diagMap["fastdp"]; ok {
+			if fastDPStats, ok := diag.(weave.FastDPStatus); ok {
+				return fastDPStats.Metrics().(*weave.FastDPMetrics)
+			}
+		}
+	}
+	return nil
+}
+
+func newMetrics(router *weave.NetworkRouter, allocator *ipam.Allocator, ns *nameserver.Nameserver, dnsserver *nameserver.DNSServer) *collector {
+	return &collector{
+		router:    router,
+		allocator: allocator,
+		ns:        ns,
+		dnsserver: dnsserver,
+	}
+}
+
+func (m *collector) Collect(ch chan<- prometheus.Metric) {
+
+	status := WeaveStatus{"", nil,
+		weave.NewNetworkRouterStatus(m.router),
+		ipam.NewStatus(m.allocator, address.CIDR{}),
+		nameserver.NewStatus(m.ns, m.dnsserver)}
+
+	for _, metric := range metrics {
+		metric.Collect(status, metric.Desc, ch)
+	}
+}
+
+func (m *collector) Describe(ch chan<- *prometheus.Desc) {
+	for _, metric := range metrics {
+		ch <- metric.Desc
+	}
+}

router/fastdp.go

@@ -318,6 +318,11 @@ func (fastDatapathOverlay) AddFeaturesTo(features map[string]string) {
 	// OverlaySwitch.
 }
 
+type FastDPStatus struct {
+	Vports []VportStatus
+	Flows  []FlowStatus
+}
+
 type FlowStatus odp.FlowInfo
 
 func (flowStatus *FlowStatus) MarshalJSON() ([]byte, error) {
@@ -374,15 +379,28 @@ func (fastdp fastDatapathOverlay) Diagnostics() interface{} {
 		flowStatuses = append(flowStatuses, FlowStatus(flow))
 	}
 
-	return struct {
-		Vports []VportStatus
-		Flows  []FlowStatus
-	}{
+	return FastDPStatus{
 		vportStatuses,
 		flowStatuses,
 	}
 }
 
+type FastDPMetrics struct {
+	Flows        int
+	TotalPackets uint64
+	TotalBytes   uint64
+}
+
+func (s FastDPStatus) Metrics() interface{} {
+	var m FastDPMetrics
+	m.Flows = len(s.Flows)
+	for _, flow := range s.Flows {
+		m.TotalPackets += flow.Packets
+		m.TotalBytes += flow.Bytes
+	}
+	return &m
+}
+
 func (fastdp fastDatapathOverlay) StartConsumingPackets(localPeer *mesh.Peer, peers *mesh.Peers, consumer OverlayConsumer) error {
 	fastdp.lock.Lock()
 	defer fastdp.lock.Unlock()