Rationalize utilization in Status proto, and add transcoderd

cmd/radiod/radiod.go

@@ -8,6 +8,7 @@ import (
 	"os"
 	"os/signal"
 	"strings"
+	"sync"
 	"syscall"
 	"time"
 
@@ -28,7 +29,7 @@ var (
 	httpPort      = flag.Int("http-port", 80, "HTTP port")
 	dnsPort       = flag.Int("dns-port", 53, "DNS port")
 	gossipPort    = flag.Int("gossip-port", 2323, "Gossip GRPC port")
-	bwLimit       = flag.Int("bwlimit", 100, "Bandwidth usage limit (Mbps), for load-balancing")
+	bwlimit       = flag.Int("bwlimit", 100, "Bandwidth usage limit (Mbps), for load-balancing")
 	maxClients    = flag.Int("max-clients", 1000, "Maximum number of connected clients, for load-balancing")
 	etcdEndpoints = flag.String("etcd", "http://localhost:2379", "Etcd endpoints (comma-separated list of URLs)")
 
@@ -76,16 +77,23 @@ func main() {
 	}
 
 	// Create the etcd client and establish the Session that will
-	// control the lifecycle of the node.
+	// control the lifecycle of the node. Enable auto-sync because
+	// this is a long-lived connection.
 	etcd, err := clientv3.New(clientv3.Config{
-		Endpoints:   strings.Split(*etcdEndpoints, ","),
-		DialTimeout: 5 * time.Second,
+		Endpoints:        strings.Split(*etcdEndpoints, ","),
+		DialTimeout:      5 * time.Second,
+		AutoSyncInterval: 5 * time.Minute,
 	})
 	if err != nil {
 		log.Fatalf("failed to connect to etcd: %v", err)
 	}
 	defer etcd.Close()
 
+	// Log a message here because the next step will hang forever
+	// if etcd is not reachable. An alternative would be to set a
+	// timeout (with custom Context and concurrency.WithContext),
+	// and rely on the service restart.
+	log.Printf("establishing etcd session...")
 	session, err := concurrency.NewSession(etcd, concurrency.WithTTL(sessionTTL))
 	if err != nil {
 		log.Fatalf("could not establish etcd session: %v", err)
@@ -98,8 +106,9 @@ func main() {
 	// outbound requests.
 	ctx, cancel := context.WithCancel(context.Background())
 	go func() {
-		// Oops, the session is gone, stop everything.
+		// Stop everything if the etcd session goes away.
 		<-session.Done()
+		log.Printf("etcd session gone, terminating...")
 		cancel()
 	}()
 
@@ -120,15 +129,35 @@ func main() {
 	}
 
 	// Create and start the local Node.
-	n, err := node.New(ctx, session, ice, *name, *publicIPs, *peerIP, *gossipPort, *lbSpec)
+	bwlimitBytes := *bwlimit / 8 * 1000000
+	n, err := node.New(ctx, session, ice, *name, *publicIPs, *peerIP, *gossipPort, *lbSpec, bwlimitBytes, *maxClients)
 	if err != nil {
 		log.Fatalf("could not initialize node: %v", err)
 	}
 
 	// Start all the network services.
 	srv := node.NewServer(n, *domain, strings.Split(*nameservers, ","), *publicIPs, *peerIP, *httpPort, *dnsPort, *gossipPort, autoradio.IcecastPort)
-	defer srv.Stop()
 
-	// Just wait for it to complete.
-	n.Wait()
+	// Wait until the Node and the Server terminate. A failure in
+	// either the network services or the Node itself should cause
+	// the other to terminate. Since the Server does not have a
+	// controlling Context, we just call its Stop method.
+	var wg sync.WaitGroup
+	wg.Add(2)
+	go func() {
+		err := srv.Wait()
+		if err != nil {
+			log.Printf("server error: %v", err)
+			cancel()
+		}
+		wg.Done()
+	}()
+
+	go func() {
+		n.Wait()
+		srv.Stop()
+		wg.Done()
+	}()
+
+	wg.Wait()
 }

cmd/transcoderd/transcoderd.go

+package main
+
+import (
+	"context"
+	"flag"
+	"log"
+	"os"
+	"os/signal"
+	"strings"
+	"syscall"
+	"time"
+
+	"git.autistici.org/ale/autoradio/transcoder"
+	"go.etcd.io/etcd/clientv3"
+	"go.etcd.io/etcd/clientv3/concurrency"
+)
+
+var (
+	name          = flag.String("name", shortHostname(), "Name for this node")
+	etcdEndpoints = flag.String("etcd", "http://localhost:2379", "Etcd endpoints (comma-separated list of URLs)")
+
+	sessionTTL = 5
+)
+
+func shortHostname() string {
+	hostname, _ := os.Hostname()
+	if r := strings.Index(hostname, "."); r >= 0 {
+		return hostname[:r]
+	}
+	return hostname
+}
+
+func main() {
+	log.SetFlags(0)
+	flag.Parse()
+
+	if *name == "" {
+		log.Fatal("--name must be set")
+	}
+
+	// Create the etcd client and establish the Session that will
+	// control the lifecycle of the node. Enable auto-sync because
+	// this is a long-lived connection.
+	etcd, err := clientv3.New(clientv3.Config{
+		Endpoints:        strings.Split(*etcdEndpoints, ","),
+		DialTimeout:      5 * time.Second,
+		AutoSyncInterval: 5 * time.Minute,
+	})
+	if err != nil {
+		log.Fatalf("failed to connect to etcd: %v", err)
+	}
+	defer etcd.Close()
+
+	// Log a message here because the next step will hang forever
+	// if etcd is not reachable. An alternative would be to set a
+	// timeout (with custom Context and concurrency.WithContext),
+	// and rely on the service restart.
+	log.Printf("establishing etcd session...")
+	session, err := concurrency.NewSession(etcd, concurrency.WithTTL(sessionTTL))
+	if err != nil {
+		log.Fatalf("could not establish etcd session: %v", err)
+	}
+
+	// Create a top-level Context that can be canceled when the
+	// program must terminate. This Context controls the lifetime
+	// of the Node itself, and all the associated background
+	// goroutines: canceling it immediately terminates all
+	// outbound requests.
+	ctx, cancel := context.WithCancel(context.Background())
+	go func() {
+		// Stop everything if the etcd session goes away.
+		<-session.Done()
+		log.Printf("etcd session gone, terminating...")
+		cancel()
+	}()
+
+	// Set up a clean shutdown function on SIGTERM that will
+	// cancel the controlling Context.
+	sigCh := make(chan os.Signal)
+	go func() {
+		<-sigCh
+		log.Printf("terminating due to signal...")
+		cancel()
+	}()
+	signal.Notify(sigCh, syscall.SIGTERM, syscall.SIGINT)
+
+	m := transcoder.New(ctx, session, *name)
+	defer m.Close()
+
+	// Wait until termination is requested via signal.
+	<-ctx.Done()
+}

debian/autoradio-server.radiod.service

@@ -6,10 +6,9 @@ Wants=etcd.service
 [Service]
 User=icecast2
 EnvironmentFile=-/etc/default/autoradio
-ExecStart=/usr/sbin/radiod $ETCD_SERVER $PUBLIC_IP $INTERFACE $RADIOD_OPTIONS
+ExecStart=/usr/bin/radiod $ETCD_SERVER $PUBLIC_IP $INTERFACE $RADIOD_OPTIONS
 Restart=always
 RestartSec=1
 
 [Install]
 WantedBy=multi-user.target
-

debian/changelog

+autoradio (1.99) unstable; urgency=medium
+
+  * First 2.0 pre-release.
+
+ -- ale <ale@incal.net>  Thu, 11 Apr 2019 22:01:38 +0100
+
 autoradio (0.7.2) unstable; urgency=medium
 
   * Add support for static redirects.

debian/rules

 #!/usr/bin/make -f
-# -*- makefile -*-
-# Sample debian/rules that uses debhelper.
-#
-# This file was originally written by Joey Hess and Craig Small.
-# As a special exception, when this file is copied by dh-make into a
-# dh-make output file, you may use that output file without restriction.
-# This special exception was added by Craig Small in version 0.37 of dh-make.
-#
-# Modified to make a template file for a multi-binary package with separated
-# build-arch and build-indep targets  by Bill Allombert 2001
 
-# Uncomment this to turn on verbose mode.
-#export DH_VERBOSE=1
-
-# This has to be exported to make some magic below work.
-export DH_OPTIONS
-
-GOPKG = git.autistici.org/ale/autoradio
-
-DESTDIR = $(CURDIR)/debian/autoradio-server
+export DH_GOPKG = git.autistici.org/ale/autoradio
+export DH_GOLANG_EXCLUDES = vendor
 
 %:
-	dh $@ --with systemd
+	dh $@ --with systemd --with golang --buildsystem golang
 
-override_dh_install:
-	# Build the binary
-	(mkdir -p $(CURDIR)/build/src/$(dir $(GOPKG)) && \
-	 ln -s $(CURDIR) $(CURDIR)/build/src/$(GOPKG) && \
-	 cd $(CURDIR)/build/src/$(GOPKG) && \
-	 env GOPATH=$(CURDIR)/build PATH=$$PATH:/usr/local/bin go install -v ./...)
-	install -m 755 -o root -g root -d $(DESTDIR)/usr/bin
-	(for bin in radioctl radiobench ; do \
-	  install -m 755 -o root -g root $(CURDIR)/build/bin/$$bin $(DESTDIR)/usr/bin/$$bin ; \
-	 done)
-	install -m 755 -o root -g root -d $(DESTDIR)/usr/sbin
-	(for bin in redirectord radiod ; do \
-	  install -m 755 -o root -g root $(CURDIR)/build/bin/$$bin $(DESTDIR)/usr/sbin/$$bin ; \
-	 done)
-
-	# Install the static files and templates for the HTTP dashboard.
-	install -d -m 755 -o root -g root $(DESTDIR)/usr/share/autoradio
-	install -d -m 755 -o root -g root $(DESTDIR)/usr/share/autoradio/htdocs/static
-	(for f in $(CURDIR)/fe/static/* ; do \
-	  install -o root -g root -m 644 $$f $(DESTDIR)/usr/share/autoradio/htdocs/static/ ; \
-	 done)
-	install -d -m 755 -o root -g root $(DESTDIR)/usr/share/autoradio/htdocs/templates
-	(for f in $(CURDIR)/fe/templates/*.html ; do \
-	  install -o root -g root -m 644 $$f $(DESTDIR)/usr/share/autoradio/htdocs/templates/ ; \
-	 done)
+override_dh_auto_install:
+	dh_auto_install
+	$(RM) -rv debian/tmp/usr/share/gocode
 
 	# Icecast2 status XSL template.
 	install -d -m 755 -o root -g root $(DESTDIR)/usr/share/icecast2/web
@@ -60,13 +21,8 @@ override_dh_install:
 		> $(DESTDIR)/etc/sudoers.d/autoradio
 	chmod 0440 $(DESTDIR)/etc/sudoers.d/autoradio
 
-override_dh_clean:
-	-rm -fr build
-	dh_clean
-
 override_dh_installinit:
 	dh_installinit --name=autoradio
 
 override_dh_systemd_enable:
 	dh_systemd_enable --name=radiod
-	dh_systemd_enable --name=redirectord

node/loadbalancing.go

@@ -12,10 +12,10 @@ import (
 	"sync"
 	"time"
 
-	"git.autistici.org/ale/autoradio/util"
 	"git.autistici.org/ale/autoradio/coordination/presence"
 	"git.autistici.org/ale/autoradio/node/lbv2"
 	pb "git.autistici.org/ale/autoradio/proto"
+	"git.autistici.org/ale/autoradio/util"
 )
 
 type loadBalancer struct {
@@ -103,13 +103,19 @@ func (n *nodeInfo) Utilization(dimension int) lbv2.NodeUtilization {
 	if n.status == nil {
 		return lbv2.NodeUtilization{Utilization: 1}
 	}
-	nl := n.status.NumListeners()
+	nl := int(n.status.NumListeners)
 	var u float64
 	switch dimension {
 	case utilBandwidth:
-		u = float64(n.status.BwUsage)
+		u = float64(n.status.CurBandwidth) / float64(n.status.MaxBandwidth)
 	case utilListeners:
-		u = float64(nl) / float64(n.status.MaxListeners)
+		u = float64(n.status.NumListeners) / float64(n.status.MaxListeners)
+	}
+	if u < 0 {
+		u = 0
+	}
+	if u > 1 {
+		u = 1
 	}
 	return lbv2.NodeUtilization{
 		Utilization: u,

node/node.go

@@ -29,7 +29,9 @@ type Icecast interface {
 // frontends. A Node comes with its own etcd Session, that it
 // maintains until the controlling Context is canceled.
 type Node struct {
-	name string
+	name         string
+	maxBandwidth int
+	maxListeners int
 
 	mounts      client.MountConfig
 	watcher     *election.ElectionWatcher
@@ -44,7 +46,7 @@ type Node struct {
 
 // New returns a new Node with a controlling Context, scoped to an etcd
 // Session.
-func New(parentCtx context.Context, session *concurrency.Session, ice Icecast, nodeID string, publicAddrs []net.IP, peerAddr net.IP, gossipPort int, lbSpec string) (*Node, error) {
+func New(parentCtx context.Context, session *concurrency.Session, ice Icecast, nodeID string, publicAddrs []net.IP, peerAddr net.IP, gossipPort int, lbSpec string, maxBandwidth, maxListeners int) (*Node, error) {
 	// Create a sub-Context that can be canceled when Stop is called. This
 	// Context controls the lifetime of the Node itself, and all the
 	// associated background goroutines: canceling it immediately
@@ -57,10 +59,12 @@ func New(parentCtx context.Context, session *concurrency.Session, ice Icecast, n
 	}()
 
 	n := &Node{
-		ice:      ice,
-		ctx:      ctx,
-		name:     nodeID,
-		updateCh: make(chan struct{}, 1),
+		ice:          ice,
+		ctx:          ctx,
+		name:         nodeID,
+		maxBandwidth: maxBandwidth,
+		maxListeners: maxListeners,
+		updateCh:     make(chan struct{}, 1),
 	}
 
 	// The runtime configuration is just a list of mounts. Synchronize it
@@ -178,6 +182,16 @@ func (n *Node) updateIcecast() {
 	}
 }
 
+// NumListeners returns the total number of listeners for all streams
+// on this node.
+func numListeners(icecastMounts []*pb.IcecastMount) int {
+	var l int32
+	for _, m := range icecastMounts {
+		l += m.Listeners
+	}
+	return int(l)
+}
+
 // Return the Status protobuf for this node.
 func (n *Node) getStatus() *pb.Status {
 	iceMounts, iceOk := n.ice.GetStatus()
@@ -186,8 +200,10 @@ func (n *Node) getStatus() *pb.Status {
 		Timestamp:     uint64(time.Now().UTC().UnixNano()),
 		IcecastOk:     iceOk,
 		IcecastMounts: iceMounts,
-		BwUsage:       0,
-		MaxListeners:  1000,
+		CurBandwidth:  int32(getCurrentBandwidthUsage()),
+		MaxBandwidth:  int32(n.maxBandwidth),
+		NumListeners:  int32(numListeners(iceMounts)),
+		MaxListeners:  int32(n.maxListeners),
 	}
 	return &ns
 }

node/node_test.go

@@ -51,6 +51,7 @@ func TestNode(t *testing.T) {
 			net.ParseIP("127.0.0.1"),
 			4014,
 			"random",
+			0, 0,
 		)
 		if err != nil {
 			t.Fatalf("NewNode: %v", err)

node/proxy.go

@@ -17,6 +17,7 @@ import (
 	"net/url"
 	"strings"
 	"sync"
+	"sync/atomic"
 	"time"
 
 	"github.com/prometheus/client_golang/prometheus"
@@ -143,7 +144,11 @@ func doIcecastProxy(rw http.ResponseWriter, req *http.Request, target *url.URL,
 	handleProxy(conn.(*net.TCPConn), upstream.(*net.TCPConn), streamName)
 }
 
-func copyStream(out, in *net.TCPConn, promCounter prometheus.Counter) {
+// Copy data between two network connections. On recent Go versions
+// (>1.11), this is quite fast as io.CopyBuffer uses the splice()
+// system call internally (in exchange we lose the ability to figure
+// out which connection is the source of the error).
+func copyStream(out, in *net.TCPConn, promCounter prometheus.Counter, cntr *uint64) {
 	buf := getBuf()
 	defer releaseBuf(buf)
 	defer in.CloseRead()   //nolint
@@ -152,6 +157,9 @@ func copyStream(out, in *net.TCPConn, promCounter prometheus.Counter) {
 	for {
 		n, err := io.CopyBuffer(out, in, buf)
 		promCounter.Add(float64(n))
+		if cntr != nil {
+			atomic.AddUint64(cntr, uint64(n))
+		}
 		if err != nil {
 			if err != io.EOF {
 				log.Printf("http: proxy error: %v", err)
@@ -168,12 +176,14 @@ func handleProxy(conn *net.TCPConn, upstream *net.TCPConn, streamName string) {
 	wg.Add(2)
 	streamListeners.WithLabelValues(streamName).Inc()
 
+	// Instrument both directions of the stream, but let the
+	// bandwidth estimator count only the bytes sent to the user.
 	go func() {
-		copyStream(conn, upstream, streamSentBytes.WithLabelValues(streamName))
+		copyStream(conn, upstream, streamSentBytes.WithLabelValues(streamName), &bwBytesSent)
 		wg.Done()
 	}()
 	go func() {
-		copyStream(upstream, conn, streamRcvdBytes.WithLabelValues(streamName))
+		copyStream(upstream, conn, streamRcvdBytes.WithLabelValues(streamName), nil)
 		wg.Done()
 	}()
 
@@ -212,3 +222,37 @@ func releaseBuf(b []byte) {
 	default:
 	}
 }
+
+// Simple bandwidth meter that keeps track of the current
+// (approximate) rate of bytes sent through the proxy.
+var (
+	bwBytesSent     uint64
+	bwLastBytesSent uint64
+	bwLastTS        time.Time
+
+	bwMx      sync.Mutex
+	bwCurrent float64
+)
+
+func init() {
+	tick := time.NewTicker(10 * time.Second)
+	go func() {
+		for t := range tick.C {
+			bytesSent := atomic.LoadUint64(&bwBytesSent)
+			bw := float64(bytesSent-bwLastBytesSent) / t.Sub(bwLastTS).Seconds()
+			bwLastBytesSent = bytesSent
+			bwLastTS = t
+
+			bwMx.Lock()
+			bwCurrent = bw
+			bwMx.Unlock()
+		}
+	}()
+}
+
+// Returns current usage (through the proxy) in bytes per second.
+func getCurrentBandwidthUsage() float64 {
+	bwMx.Lock()
+	defer bwMx.Unlock()
+	return bwCurrent
+}

node/server.go

@@ -49,8 +49,13 @@ func buildServer(servers ...genericServer) *Server {
 	return &ms
 }
 
-func (s *Server) Stop() error {
+// Stop all network services.
+func (s *Server) Stop() {
 	close(s.stopCh)
+}
+
+// Wait for the services to terminate.
+func (s *Server) Wait() error {
 	s.wg.Wait()
 
 	// There may or may not be an error stored in errCh.

proto/status.go

-package autoradio
-
-// NumListeners returns the total number of listeners for all streams
-// on this node.
-func (s *Status) NumListeners() int {
-	var l int32
-	for _, m := range s.IcecastMounts {
-		l += m.Listeners
-	}
-	return int(l)
-}

proto/status.pb.go

@@ -114,8 +114,10 @@ type Status struct {
 	Timestamp            uint64          `protobuf:"varint,2,opt,name=timestamp,proto3" json:"timestamp,omitempty"`
 	IcecastOk            bool            `protobuf:"varint,3,opt,name=icecast_ok,json=icecastOk,proto3" json:"icecast_ok,omitempty"`
 	IcecastMounts        []*IcecastMount `protobuf:"bytes,4,rep,name=icecast_mounts,json=icecastMounts,proto3" json:"icecast_mounts,omitempty"`
-	MaxListeners         int32           `protobuf:"varint,6,opt,name=max_listeners,json=maxListeners,proto3" json:"max_listeners,omitempty"`
-	BwUsage              float32         `protobuf:"fixed32,5,opt,name=bw_usage,json=bwUsage,proto3" json:"bw_usage,omitempty"`
+	CurBandwidth         int32           `protobuf:"varint,5,opt,name=cur_bandwidth,json=curBandwidth,proto3" json:"cur_bandwidth,omitempty"`
+	MaxBandwidth         int32           `protobuf:"varint,6,opt,name=max_bandwidth,json=maxBandwidth,proto3" json:"max_bandwidth,omitempty"`
+	MaxListeners         int32           `protobuf:"varint,7,opt,name=max_listeners,json=maxListeners,proto3" json:"max_listeners,omitempty"`
+	NumListeners         int32           `protobuf:"varint,8,opt,name=num_listeners,json=numListeners,proto3" json:"num_listeners,omitempty"`
 	XXX_NoUnkeyedLiteral struct{}        `json:"-"`
 	XXX_unrecognized     []byte          `json:"-"`
 	XXX_sizecache        int32           `json:"-"`
@@ -174,6 +176,20 @@ func (m *Status) GetIcecastMounts() []*IcecastMount {
 	return nil
 }
 
+func (m *Status) GetCurBandwidth() int32 {
+	if m != nil {
+		return m.CurBandwidth
+	}
+	return 0
+}
+
+func (m *Status) GetMaxBandwidth() int32 {
+	if m != nil {
+		return m.MaxBandwidth
+	}
+	return 0
+}
+
 func (m *Status) GetMaxListeners() int32 {
 	if m != nil {
 		return m.MaxListeners
@@ -181,9 +197,9 @@ func (m *Status) GetMaxListeners() int32 {
 	return 0
 }
 
-func (m *Status) GetBwUsage() float32 {
+func (m *Status) GetNumListeners() int32 {
 	if m != nil {
-		return m.BwUsage
+		return m.NumListeners
 	}
 	return 0
 }
@@ -276,32 +292,34 @@ func init() {
 func init() { proto.RegisterFile("status.proto", fileDescriptor_dfe4fce6682daf5b) }
 
 var fileDescriptor_dfe4fce6682daf5b = []byte{
-	// 390 bytes of a gzipped FileDescriptorProto
-	0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0x8c, 0x92, 0x4f, 0x6f, 0xd3, 0x40,
-	0x10, 0xc5, 0x71, 0x9b, 0xc4, 0xf1, 0xe0, 0xf0, 0x67, 0x2f, 0x2c, 0x05, 0x24, 0xcb, 0x1c, 0xf0,
-	0x29, 0x87, 0x72, 0x03, 0x89, 0x5b, 0x85, 0x90, 0x40, 0x88, 0x8d, 0xe0, 0x6a, 0xad, 0x9d, 0xa1,
-	0x5d, 0xb5, 0xf6, 0xba, 0x9e, 0x71, 0x1b, 0xf2, 0x49, 0xe1, 0xdb, 0xa0, 0x5d, 0xc7, 0xd8, 0x20,
-	0x0e, 0xbd, 0xed, 0xbc, 0xe7, 0xf9, 0xf3, 0x7e, 0x32, 0xc4, 0xc4, 0x9a, 0x3b, 0x5a, 0x37, 0xad,
-	0x65, 0x2b, 0x22, 0xdd, 0xb1, 0x6d, 0xf5, 0xd6, 0xd8, 0xf4, 0x67, 0x00, 0xf1, 0x87, 0x12, 0x4b,
-	0x4d, 0xfc, 0xc9, 0x76, 0x35, 0x0b, 0x01, 0xb3, 0x46, 0xf3, 0x85, 0x0c, 0x92, 0x20, 0x8b, 0x94,
-	0x7f, 0x8b, 0xe7, 0x10, 0x5d, 0x19, 0x62, 0xac, 0xb1, 0x25, 0x79, 0x94, 0x04, 0xd9, 0x5c, 0x8d,
-	0x82, 0x78, 0x0a, 0xcb, 0xc2, 0x70, 0xde, 0x6a, 0x46, 0x79, 0xec, 0xcd, 0xb0, 0x30, 0xac, 0x34,
-	0xa3, 0x90, 0x10, 0x5e, 0x77, 0xfa, 0xca, 0xf0, 0x0f, 0x39, 0x4b, 0x82, 0xec, 0x48, 0x0d, 0xa5,
-	0x78, 0x09, 0xab, 0x1b, 0xb3, 0x45, 0x9b, 0x0f, 0xfe, 0xdc, 0xfb, 0xb1, 0x17, 0xbf, 0x1c, 0x3e,
-	0x7a, 0x01, 0xf0, 0xbd, 0xd5, 0x15, 0xe6, 0x64, 0xf6, 0x28, 0x17, 0xfe, 0xa2, 0xc8, 0x2b, 0x1b,
-	0xb3, 0xc7, 0xd1, 0xf6, 0xab, 0x43, 0x3f, 0xa0, 0xb7, 0xdd, 0xf2, 0xf4, 0x57, 0x00, 0x8b, 0x8d,
-	0x8f, 0xed, 0x42, 0xd5, 0xba, 0xc2, 0x21, 0x94, 0x7b, 0xbb, 0x50, 0x6c, 0x2a, 0x24, 0xd6, 0x55,
-	0xe3, 0x43, 0xcd, 0xd4, 0x28, 0xb8, 0xd9, 0xa6, 0xc7, 0x92, 0xdb, 0x4b, 0x1f, 0x6b, 0xa9, 0xa2,
-	0x83, 0xf2, 0xf9, 0x52, 0xbc, 0x83, 0x07, 0x83, 0x5d, 0x39, 0x6c, 0x24, 0x67, 0xc9, 0x71, 0x76,
-	0xff, 0xf4, 0xc9, 0xfa, 0x0f, 0xda, 0xf5, 0x14, 0xab, 0x5a, 0x99, 0x49, 0x45, 0x2e, 0x7e, 0xa5,
-	0x77, 0xf9, 0x48, 0x75, 0xe1, 0xc1, 0xc5, 0x95, 0xde, 0x7d, 0xfc, 0x0b, 0xec, 0x6d, 0xde, 0x91,
-	0x3e, 0xc7, 0x03, 0x9e, 0xb0, 0xb8, 0xfd, 0xea, 0xca, 0xf4, 0x0d, 0x3c, 0x3c, 0xdb, 0x95, 0x17,
-	0xba, 0x3e, 0x47, 0x85, 0xd7, 0x1d, 0x12, 0x8b, 0x57, 0x30, 0xaf, 0xed, 0x16, 0x49, 0x06, 0xfe,
-	0x92, 0xc7, 0x93, 0x4b, 0x7a, 0x0a, 0xaa, 0xf7, 0xd3, 0xb7, 0xf0, 0x68, 0xec, 0xa5, 0xc6, 0xd6,
-	0x84, 0x77, 0x6e, 0x3e, 0xfd, 0x06, 0xab, 0xf7, 0x96, 0xc8, 0x34, 0x1b, 0x6c, 0x6f, 0x4c, 0x89,
-	0xe2, 0x0c, 0x96, 0xc3, 0x34, 0x71, 0x32, 0x69, 0xfb, 0xe7, 0xbc, 0x93, 0x67, 0xff, 0xf5, 0xfa,
-	0xf5, 0xe9, 0xbd, 0x62, 0xe1, 0xff, 0xcc, 0xd7, 0xbf, 0x03, 0x00, 0x00, 0xff, 0xff, 0x2c, 0x5f,
-	0x4c, 0x43, 0xa9, 0x02, 0x00, 0x00,
+	// 418 bytes of a gzipped FileDescriptorProto
+	0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0x8c, 0x92, 0x4d, 0x8f, 0xd3, 0x30,
+	0x10, 0x86, 0x49, 0xb7, 0x5f, 0x19, 0x5a, 0x3e, 0x7c, 0x21, 0x2c, 0x20, 0x55, 0xd9, 0x03, 0x3d,
+	0xf5, 0xb0, 0xdc, 0x40, 0xe2, 0x80, 0xb4, 0x42, 0x48, 0x20, 0x84, 0x2b, 0x71, 0x8d, 0xdc, 0x64,
+	0xa0, 0xd6, 0xae, 0xed, 0xac, 0x3d, 0x5e, 0xca, 0xfe, 0x27, 0xfe, 0x0f, 0x3f, 0x07, 0xd9, 0x69,
+	0x3e, 0x04, 0x1c, 0xf6, 0x16, 0xbf, 0xef, 0x3b, 0x33, 0x7e, 0xc6, 0x81, 0x85, 0x23, 0x41, 0xde,
+	0x6d, 0x6a, 0x6b, 0xc8, 0xb0, 0x54, 0x78, 0x32, 0x56, 0x54, 0xd2, 0xe4, 0xbf, 0x13, 0x58, 0x7c,
+	0x28, 0xb1, 0x14, 0x8e, 0x3e, 0x19, 0xaf, 0x89, 0x31, 0x18, 0xd7, 0x82, 0xf6, 0x59, 0xb2, 0x4a,
+	0xd6, 0x29, 0x8f, 0xdf, 0xec, 0x39, 0xa4, 0x57, 0xd2, 0x11, 0x6a, 0xb4, 0x2e, 0x1b, 0xad, 0x92,
+	0xf5, 0x84, 0xf7, 0x02, 0x7b, 0x0a, 0xf3, 0x9d, 0xa4, 0xc2, 0x0a, 0xc2, 0xec, 0x24, 0x9a, 0xb3,
+	0x9d, 0x24, 0x2e, 0x08, 0x59, 0x06, 0xb3, 0x6b, 0x2f, 0xae, 0x24, 0xfd, 0xcc, 0xc6, 0xab, 0x64,
+	0x3d, 0xe2, 0xed, 0x91, 0x9d, 0xc1, 0xf2, 0x46, 0x56, 0x68, 0x8a, 0xd6, 0x9f, 0x44, 0x7f, 0x11,
+	0xc5, 0x2f, 0xc7, 0xd0, 0x0b, 0x80, 0x6f, 0x56, 0x28, 0x2c, 0x9c, 0xbc, 0xc5, 0x6c, 0x1a, 0x6f,
+	0x94, 0x46, 0x65, 0x2b, 0x6f, 0xb1, 0xb7, 0xe3, 0xe8, 0x59, 0x6c, 0xd0, 0xd8, 0x61, 0x78, 0xfe,
+	0x6b, 0x04, 0xd3, 0x6d, 0xc4, 0x0e, 0x50, 0x5a, 0x28, 0x6c, 0xa1, 0xc2, 0x77, 0x80, 0x22, 0xa9,
+	0xd0, 0x91, 0x50, 0x75, 0x84, 0x1a, 0xf3, 0x5e, 0x08, 0xbd, 0x65, 0xb3, 0x96, 0xc2, 0x5c, 0x46,
+	0xac, 0x39, 0x4f, 0x8f, 0xca, 0xe7, 0x4b, 0xf6, 0x16, 0x1e, 0xb4, 0xb6, 0x0a, 0x6b, 0x73, 0xd9,
+	0x78, 0x75, 0xb2, 0xbe, 0x7f, 0xfe, 0x64, 0xd3, 0xad, 0x76, 0x33, 0x5c, 0x2b, 0x5f, 0xca, 0xc1,
+	0xc9, 0x05, 0xfc, 0xd2, 0xdb, 0x62, 0x27, 0x74, 0xf5, 0x43, 0x56, 0xb4, 0x8f, 0xf8, 0x13, 0xbe,
+	0x28, 0xbd, 0x7d, 0xd7, 0x6a, 0x21, 0xa4, 0xc4, 0x61, 0x10, 0x9a, 0x36, 0x21, 0x25, 0x0e, 0xff,
+	0x84, 0xfa, 0xf7, 0x99, 0x75, 0xa1, 0x8f, 0xdd, 0x13, 0x9d, 0xc1, 0x52, 0x7b, 0x35, 0x08, 0xcd,
+	0x9b, 0x90, 0xf6, 0xaa, 0x0b, 0xe5, 0xaf, 0xe1, 0xe1, 0xc5, 0xa1, 0xdc, 0x0b, 0xfd, 0x1d, 0x39,
+	0x5e, 0x7b, 0x74, 0xc4, 0x5e, 0xc2, 0x44, 0x9b, 0x0a, 0x5d, 0x96, 0x44, 0xba, 0xc7, 0x03, 0xba,
+	0x66, 0xb3, 0xbc, 0xf1, 0xf3, 0x37, 0xf0, 0xa8, 0xaf, 0x75, 0xb5, 0xd1, 0x0e, 0xef, 0x5c, 0x7c,
+	0xfe, 0x15, 0x96, 0xef, 0x8d, 0x73, 0xb2, 0xde, 0xa2, 0xbd, 0x91, 0x25, 0xb2, 0x0b, 0x98, 0xb7,
+	0xdd, 0xd8, 0xe9, 0xa0, 0xec, 0xaf, 0xeb, 0x9d, 0x3e, 0xfb, 0xaf, 0xd7, 0x8c, 0xcf, 0xef, 0xed,
+	0xa6, 0xf1, 0x6f, 0x7f, 0xf5, 0x27, 0x00, 0x00, 0xff, 0xff, 0x9d, 0xc8, 0xe4, 0xa7, 0xfd, 0x02,
+	0x00, 0x00,
 }
 
 // Reference imports to suppress errors if they are not otherwise used.

proto/status.proto

@@ -23,9 +23,11 @@ message Status {
 
   bool icecast_ok = 3;
   repeated IcecastMount icecast_mounts = 4;
-  int32 max_listeners = 6;
 
-  float bw_usage = 5;
+  int32 cur_bandwidth = 5;
+  int32 max_bandwidth = 6;
+  int32 max_listeners = 7;
+  int32 num_listeners = 8;
 }
 
 // The GossipService is used by frontends to exchange NodeStatus

transcoder/manager.go

@@ -3,20 +3,32 @@ package transcoder
 import (
 	"context"
 
+	"git.autistici.org/ale/autoradio/client"
 	pb "git.autistici.org/ale/autoradio/proto"
 	"github.com/golang/protobuf/proto"
 	"go.etcd.io/etcd/clientv3/concurrency"
 )
 
-var electionPath = "/icecast/5/election/transcoder/"
-
 type Manager struct {
-	nodeID      string
-	ctx         context.Context
-	session     *concurrency.Session
+	ctx     context.Context
+	session *concurrency.Session
+	nodeID  string
+
 	transcoders map[string]*transcoder
 }
 
+func New(ctx context.Context, session *concurrency.Session, nodeID string) *Manager {
+	m := &Manager{
+		ctx:         ctx,
+		session:     session,
+		nodeID:      nodeID,
+		transcoders: make(map[string]*transcoder),
+	}
+	configReady, _ := client.WatchConfig(ctx, session.Client(), m)
+	<-configReady
+	return m
+}
+
 func (t *Manager) Reset(mounts map[string]*pb.Mount) {
 	// Figure out what the differences are.
 	newTC := make(map[string]*transcoder)
@@ -57,3 +69,9 @@ func (t *Manager) Delete(path string) {
 		delete(t.transcoders, path)
 	}
 }
+
+func (t *Manager) Close() {
+	for _, tc := range t.transcoders {
+		tc.stop()
+	}
+}

transcoder/transcoder.go

@@ -7,8 +7,10 @@ import (
 	"math/rand"
 	"os"
 	"os/exec"
+	"strings"
 	"time"
 
+	"git.autistici.org/ale/autoradio"
 	"git.autistici.org/ale/autoradio/coordination/election"
 	pb "git.autistici.org/ale/autoradio/proto"
 	"go.etcd.io/etcd/clientv3/concurrency"
@@ -35,11 +37,20 @@ func newTranscoder(ctx context.Context, session *concurrency.Session, m *pb.Moun
 	// Start a leader election to run a single transcoder
 	// globally.  Sleep a small random amount of time before
 	// participating in the election, so that leadership is more
-	// easily spread around nodes.
-	el := election.New(session, electionPath, nodeID, t.run)
+	// easily spread around nodes. The election prefix should not
+	// contain nested directories (these are key prefixes, not
+	// paths), so we perform a simple character substitution to
+	// obtain the mount-specific election prefix. Furthermore,
+	// the election API requires an Endpoint proto, but we aren't
+	// providing any network service, so we just create a weird
+	// Endpoint without Addrs.x
+	electionPath := autoradio.TranscoderElectionPrefix + strings.Replace(m.Path, "/", "_", -1) + "/"
+	el := election.New(session, electionPath, &pb.Endpoint{
+		Name: nodeID,
+	})
 	go func() {
 		sleepTinyBit()
-		err := el.Run(tctx)
+		err := el.Run(tctx, t.run)
 		if err != nil && err != context.Canceled {
 			log.Printf("transcoder failed: %s: %v", m.Path,