unify the request namespace; support stream proxying

This allows running the service on a single port: with the
--enable-icecast-proxy option, all requests can go through
redirectord.

api.go

@@ -18,7 +18,8 @@ var (
 	MountPrefix        = "/icecast/mounts/"
 	NodePrefix         = "/icecast/nodes/"
 
-	IcecastPort = 8000
+	IcecastPort        = 8000
+	IcecastMountPrefix = "/_stream"
 
 	ErrIsDirectory = errors.New("key is a directory")
 	ErrIsFile      = errors.New("key is a file")
@@ -49,10 +50,21 @@ func (m *Mount) IsRelay() bool {
 	return m.RelayUrl != ""
 }
 
-func mountPath(mountName string) string {
+// Return the path in etcd used to store mountpoint configuration.
+func mountEtcdPath(mountName string) string {
 	return MountPrefix + mountName[1:]
 }
 
+// Return the Icecast mount path for the given public mount name.
+func MountNameToIcecastPath(mountName string) string {
+	return IcecastMountPrefix + mountName
+}
+
+// Return the public mount name from an Icecast mount path.
+func IcecastPathToMountName(path string) string {
+	return strings.TrimPrefix(path, IcecastMountPrefix)
+}
+
 // Status of a mount on an individual Icecast server.
 type IcecastMountStatus struct {
 	Name         string
@@ -137,7 +149,7 @@ func NewRadioAPI(client *etcd.Client) *RadioAPI {
 // GetMount returns data on a specific mountpoint (returns nil if not
 // found).
 func (r *RadioAPI) GetMount(mountName string) (*Mount, error) {
-	response, err := r.client.Get(mountPath(mountName), false, false)
+	response, err := r.client.Get(mountEtcdPath(mountName), false, false)
 	if err != nil || response.Node == nil {
 		return nil, err
 	}
@@ -159,13 +171,13 @@ func (r *RadioAPI) SetMount(m *Mount) error {
 		return err
 	}
 
-	_, err := r.client.Set(mountPath(m.Name), buf.String(), 0)
+	_, err := r.client.Set(mountEtcdPath(m.Name), buf.String(), 0)
 	return err
 }
 
 // DelMount removes a mountpoint.
 func (r *RadioAPI) DelMount(mountName string) error {
-	_, err := r.client.Delete(mountPath(mountName), false)
+	_, err := r.client.Delete(mountEtcdPath(mountName), false)
 	return err
 }
 

fe/http.go

@@ -2,18 +2,21 @@ package fe
 
 import (
 	"bytes"
+	"flag"
 	"fmt"
 	"html/template"
 	"io"
 	"log"
 	"net"
 	"net/http"
+	"net/url"
 	"path/filepath"
 	"strconv"
 	"strings"
 	"sync"
 	"time"
 
+	"net/http/httputil"
 	_ "net/http/pprof"
 
 	"git.autistici.org/ale/autoradio"
@@ -22,6 +25,8 @@ import (
 )
 
 var (
+	proxyStreams = flag.Bool("enable-icecast-proxy", false, "Proxy the local icecast")
+
 	httpStatusCodes   = instrumentation.NewCounter("http.status")
 	httpTargetStats   = instrumentation.NewCounter("http.target")
 	sourceConnections = instrumentation.NewCounter("http.source_connections")
@@ -89,12 +94,18 @@ func (h *HttpRedirector) pickActiveNode() string {
 	return result.IP
 }
 
-func makeIcecastAddr(server string) string {
+func icecastAddr(server string) string {
 	return net.JoinHostPort(server, strconv.Itoa(autoradio.IcecastPort))
 }
 
-func makeIcecastUrl(server, mountName string) string {
-	return fmt.Sprintf("http://%s%s", makeIcecastAddr(server), mountName)
+func streamUrl(server, mountName string) string {
+	var serverAddr string
+	if *proxyStreams {
+		serverAddr = server
+	} else {
+		serverAddr = icecastAddr(server)
+	}
+	return fmt.Sprintf("http://%s%s", serverAddr, autoradio.MountNameToIcecastPath(mountName))
 }
 
 // Request wrapper that passes a Mount along with the HTTP request.
@@ -110,6 +121,16 @@ func (h *HttpRedirector) withMount(f func(*autoradio.Mount, http.ResponseWriter,
 	})
 }
 
+// Serve a M3U response. This simply points back at the stream
+// redirect handler.
+func (h *HttpRedirector) serveM3U(mount *autoradio.Mount, w http.ResponseWriter, r *http.Request) {
+	m3u := strings.TrimSuffix(r.URL.String(), ".m3u") + "\n"
+	w.Header().Set("Content-Length", strconv.Itoa(len(m3u)))
+	w.Header().Set("Content-Type", "audio/x-mpegurl")
+	addDefaultHeaders(w)
+	io.WriteString(w, m3u)
+}
+
 // Serve a response for a client connection to a relay.
 func (h *HttpRedirector) serveRelay(mount *autoradio.Mount, w http.ResponseWriter, r *http.Request) {
 	// Find an active node.
@@ -120,16 +141,11 @@ func (h *HttpRedirector) serveRelay(mount *autoradio.Mount, w http.ResponseWrite
 	}
 	httpTargetStats.IncrVar(ipToMetric(relayAddr))
 
-	targetURL := makeIcecastUrl(relayAddr, mount.Name)
-
 	// See if we need to serve a M3U response or a redirect.
 	if strings.HasSuffix(r.URL.Path, ".m3u") {
-		m3u := targetURL + "\n"
-		w.Header().Set("Content-Length", strconv.Itoa(len(m3u)))
-		w.Header().Set("Content-Type", "audio/x-mpegurl")
-		addDefaultHeaders(w)
-		io.WriteString(w, m3u)
+		h.serveM3U(mount, w, r)
 	} else {
+		targetURL := streamUrl(relayAddr, mount.Name)
 		http.Redirect(w, r, targetURL, 302)
 	}
 }
@@ -161,16 +177,17 @@ func (h *HttpRedirector) serveSource(mount *autoradio.Mount, w http.ResponseWrit
 		return
 	}
 
-	// Create the upstream connection, and write the original
-	// request to it as-is (the URL path on the backend is the
-	// same, and the headers do not need to change).
-	upstream, err := net.Dial("tcp", makeIcecastAddr(masterAddr))
+	// Create the upstream connection, and write the HTTP request
+	// to it (with the right URL path).
+	upstream, err := net.Dial("tcp", icecastAddr(masterAddr))
 	if err != nil {
 		log.Printf("source: dial upstream: %v", err)
 		sendErr(err)
 		return
 	}
 	defer upstream.Close()
+
+	r.URL.Path = autoradio.MountNameToIcecastPath(mount.Name)
 	if err := r.Write(upstream); err != nil {
 		log.Printf("source: write upstream request: %v", err)
 		sendErr(err)
@@ -247,22 +264,28 @@ func (h *HttpRedirector) Run(addr, staticDir, templateDir string) {
 		template.ParseGlob(
 			filepath.Join(templateDir, "*.html")))
 
-	// Create our HTTP handler stack. Passes the /debug/ queries
-	// along to the global ServeMux (where moodules such as pprof
-	// install their handlers).
-	relayHandler := h.withMount(h.serveRelay)
+	// Create our HTTP handler stack.
 	mux := http.NewServeMux()
 	mux.Handle(
 		"/static/",
 		http.StripPrefix(
 			"/static/",
 			http.FileServer(http.Dir(staticDir))))
+
+	// Optionally enable a reverse proxy to the local Icecast.
+	if *proxyStreams {
+		iceurl, _ := url.Parse(fmt.Sprintf("http://localhost:%d", autoradio.IcecastPort))
+		mux.Handle(autoradio.IcecastMountPrefix+"/", httputil.NewSingleHostReverseProxy(iceurl))
+	}
+
+	// Pass /debug/ to the default ServeMux.
+	mux.Handle("/debug/", http.DefaultServeMux)
+
+	relayHandler := h.withMount(h.serveRelay)
 	mux.HandleFunc("/", func(w http.ResponseWriter, r *http.Request) {
 		switch {
 		case r.URL.Path == "" || r.URL.Path == "/":
 			h.serveStatusPage(w, r)
-		case strings.HasPrefix(r.URL.Path, "/debug/"):
-			http.DefaultServeMux.ServeHTTP(w, r)
 		default:
 			relayHandler.ServeHTTP(w, r)
 		}

node/icecast.go

@@ -153,7 +153,7 @@ func (ic *IcecastController) parseStatusPage(input io.Reader) (*IcecastStatus, e
 	}
 	for _, um := range ustatus.Mounts {
 		m := autoradio.IcecastMountStatus{
-			Name:         um.Name,
+			Name:         autoradio.IcecastPathToMountName(um.Name),
 			Listeners:    toi(um.Listeners),
 			BitRate:      toi(um.BitRate),
 			Quality:      tof(um.Quality),

node/icecast_config.go

@@ -19,15 +19,15 @@ var (
 )
 
 type iceLimitsConfig struct {
-	Clients        int `xml:"clients"`
-	Sources        int `xml:"sources"`
+	Clients int `xml:"clients"`
+	Sources int `xml:"sources"`
 	// Threadpool     int `xml:"threadpool"`
-	QueueSize      int `xml:"queue-size"`
-	ClientTimeout  int `xml:"client-timeout"`
-	HeaderTimeout  int `xml:"header-timeout"`
-	SourceTimeout  int `xml:"source-timeout"`
+	QueueSize     int `xml:"queue-size"`
+	ClientTimeout int `xml:"client-timeout"`
+	HeaderTimeout int `xml:"header-timeout"`
+	SourceTimeout int `xml:"source-timeout"`
 	// BurstOnConnect int `xml:"burst-on-connect"`
-	BurstSize      int `xml:"burst-size"`
+	BurstSize int `xml:"burst-size"`
 }
 
 type iceAuthenticationConfig struct {
@@ -80,8 +80,8 @@ type iceMountConfig struct {
 	FallbackOverride int    `xml:"fallback-override,omitempty"`
 	Hidden           int    `xml:"hidden"`
 	// NoYp             int    `xml:"no-yp"`
-	OnConnect        string `xml:"on-connect,omitempty"`
-	OnDisconnect     string `xml:"on-disconnect,omitempty"`
+	OnConnect    string `xml:"on-connect,omitempty"`
+	OnDisconnect string `xml:"on-disconnect,omitempty"`
 }
 
 // Configuration of the local Icecast daemon (meant for serialization
@@ -123,15 +123,15 @@ func defaultDebianConfig(publicIp string) *icecastConfig {
 	return &icecastConfig{
 		XMLName: xml.Name{"", "icecast"},
 		Limits: iceLimitsConfig{
-			Clients:        maxClients,
-			Sources:        maxClients / 2,
+			Clients: maxClients,
+			Sources: maxClients / 2,
 			// Threadpool:     16,
-			QueueSize:      1 << 20,
-			ClientTimeout:  30,
-			HeaderTimeout:  15,
-			SourceTimeout:  60,
+			QueueSize:     1 << 20,
+			ClientTimeout: 30,
+			HeaderTimeout: 15,
+			SourceTimeout: 60,
 			// BurstOnConnect: 1,
-			BurstSize:      65535,
+			BurstSize: 65535,
 		},
 		Auth: iceAuthenticationConfig{
 			SourcePassword: sourcePw,
@@ -197,7 +197,7 @@ func (c *icecastConfig) EncodeToFile(path string) error {
 
 func mountToConfig(m *autoradio.Mount) iceMountConfig {
 	mconfig := iceMountConfig{
-		Name:     m.Name,
+		Name:     autoradio.MountNameToIcecastPath(m.Name),
 		Username: m.Username,
 		Password: m.Password,
 		Hidden:   0,
@@ -229,7 +229,7 @@ func relayToConfig(m *autoradio.Mount) (iceRelayConfig, bool) {
 
 	rc := iceRelayConfig{
 		Mount:                  u.Path,
-		LocalMount:             m.Name,
+		LocalMount:             autoradio.MountNameToIcecastPath(m.Name),
 		Server:                 server,
 		Port:                   iport,
 		OnDemand:               1,
@@ -246,8 +246,8 @@ func relayToConfig(m *autoradio.Mount) (iceRelayConfig, bool) {
 
 func mountToRelayConfig(masterAddr string, m *autoradio.Mount) iceRelayConfig {
 	return iceRelayConfig{
-		Mount:                  m.Name,
-		LocalMount:             m.Name,
+		Mount:                  autoradio.MountNameToIcecastPath(m.Name),
+		LocalMount:             autoradio.MountNameToIcecastPath(m.Name),
 		Server:                 masterAddr,
 		Port:                   autoradio.IcecastPort,
 		Username:               m.Username,