add support for global relays of external streams

api.go

@@ -35,10 +35,20 @@ type Mount struct {
 	// Password for source authentication.
 	Password string
 
+	// Is this field is set, the mount point will be relaying an
+	// external stream and no source connections will be accepted.
+	// Each node will pull from the external source directly,
+	// ignoring our master election protocol.
+	RelayUrl string
+
 	// Fallback stream name (optional).
 	Fallback string
 }
 
+func (m *Mount) IsRelay() bool {
+	return m.RelayUrl != ""
+}
+
 func mountPath(mountName string) string {
 	return MountPrefix + mountName[1:]
 }
@@ -62,8 +72,8 @@ type NodeStatus struct {
 	// Is the Icecast server up?
 	IcecastUp bool
 
-	// List of 
-	Mounts  []IcecastMountStatus `xml:"mount"`
+	// List of
+	Mounts []IcecastMountStatus `xml:"mount"`
 }
 
 func (ns *NodeStatus) NumListeners() int {
@@ -221,7 +231,7 @@ func (r *RadioAPI) GetNodeIPs() ([]string, error) {
 		return nil, err
 	}
 	ips := make([]string, 0, len(nodes))
-	for _, n := range(nodes) {
+	for _, n := range nodes {
 		ips = append(ips, n.IP)
 	}
 	return ips, nil

cmd/radioctl/radioctl.go

@@ -7,6 +7,7 @@ import (
 	"log"
 	"os"
 	"path/filepath"
+	"net/url"
 	"strings"
 
 	"git.autistici.org/ale/radioai"
@@ -102,8 +103,7 @@ func generateUsername(path string) string {
 	return fmt.Sprintf("source%d", crc32.ChecksumIEEE([]byte(path)))
 }
 
-func createMount(args []string) {
-	path := args[0]
+func doCreateMount(path, relayUrl string) {
 	if strings.Contains(path, "/") {
 		log.Fatal("Mount points should not contain a slash ('/').")
 	}
@@ -111,18 +111,24 @@ func createMount(args []string) {
 
 	// Check if the mount already exists.
 	client := getClient()
-	oldm, _ := client.GetMount(path)
-	if oldm != nil {
+	if oldm, _ := client.GetMount(path); oldm != nil {
 		log.Fatal("A mount with that name already exists!")
 	}
 
-	// Create the new mount, randomly generate source authentication.
-	username := generateUsername(path)
-	password := radioai.GeneratePassword()
-	m := &radioai.Mount{
-		Name:     path,
-		Username: username,
-		Password: password,
+	// Create the new mount and set the relevant fields (depending
+	// on the relayUrl value).
+	m := &radioai.Mount{Name: path}
+	if relayUrl == "" {
+		// Randomly generate source credentials.
+		m.Username = generateUsername(path)
+		m.Password = radioai.GeneratePassword()
+	} else {
+		// Validate the given relay URL.
+		u, err := url.Parse(relayUrl)
+		if err != nil {
+			log.Fatal(err)
+		}
+		m.RelayUrl = u.String()
 	}
 
 	if err := client.SetMount(m); err != nil {
@@ -132,8 +138,17 @@ func createMount(args []string) {
 	fmt.Printf("%+v\n", m)
 }
 
+func createMount(args []string) {
+	doCreateMount(args[0], "")
+}
+
+func createRelay(args []string) {
+	doCreateMount(args[0], args[1])
+}
+
 var commands = CommandList{
 	{"create-mount", "<path>", "Create a new mountpoint", createMount, 1, 1},
+	{"create-relay", "<path> <source_url>", "Create a new relay", createRelay, 2, 2},
 	{"delete-mount", "<path>", "Delete a mountpoint", deleteMount, 1, 1},
 	{"list-mounts", "", "List all known mountpoints", listMounts, 0, 0},
 	{"show-mount", "<path>", "Show configuration of a mount", showMount, 1, 1},

fe/http.go

@@ -130,9 +130,13 @@ func (h *HttpRedirector) serveRelay(w http.ResponseWriter, r *http.Request) {
 }
 
 func (h *HttpRedirector) serveSource(w http.ResponseWriter, r *http.Request) {
-	_, err := h.getMount(r)
-	if err != nil {
-		log.Printf("source: error retrieving mount for %+v: %s", r, err)
+	m, err := h.getMount(r)
+	if err != nil || m.IsRelay() {
+		if err != nil {
+			log.Printf("source: error retrieving mount for %+v: %s", r, err)
+		} else {
+			log.Printf("source: connection to relay stream %s", m.Name)
+		}			
 		http.Error(w, "Not Found", http.StatusNotFound)
 		h.stats.Add("source_404", 1)
 		return

node/icecast_config.go

@@ -4,14 +4,18 @@ import (
 	"bytes"
 	"encoding/xml"
 	"io"
+	"log"
+	"net"
+	"net/url"
 	"os"
+	"strconv"
 
 	"git.autistici.org/ale/radioai"
 )
 
 var (
 	//shoutHttpPort = 8001
-	maxClients    = 10000
+	maxClients = 10000
 )
 
 type iceLimitsConfig struct {
@@ -134,7 +138,7 @@ func defaultDebianConfig(publicIp string) *icecastConfig {
 			AdminUser:      "admin",
 			AdminPassword:  adminPw,
 		},
-		Hostname: publicIp,
+		Hostname:  publicIp,
 		Fileserve: 1,
 		Paths: icePathsConfig{
 			Basedir:   "/usr/share/icecast2",
@@ -207,7 +211,40 @@ func mountToConfig(m *radioai.Mount) iceMountConfig {
 	return mconfig
 }
 
-func mountToRelay(masterAddr string, m *radioai.Mount) iceRelayConfig {
+func relayToConfig(m *radioai.Mount) (iceRelayConfig, bool) {
+	u, err := url.Parse(m.RelayUrl)
+	if err != nil {
+		// A failure here is almost invisible and not very
+		// useful, but at least we can prevent garbling the
+		// resulting icecast config.
+		log.Printf("can't parse relay url for %s: %s", m.Name, err)
+		return iceRelayConfig{}, false
+	}
+	server, port, err := net.SplitHostPort(u.Host)
+	if err != nil {
+		server = u.Host
+		port = "80"
+	}
+	iport, _ := strconv.Atoi(port)
+
+	rc := iceRelayConfig{
+		Mount:                  u.Path,
+		LocalMount:             m.Name,
+		Server:                 server,
+		Port:                   iport,
+		OnDemand:               1,
+		RelayShoutcastMetadata: 1,
+	}
+	if u.User != nil {
+		rc.Username = u.User.Username()
+		if p, ok := u.User.Password(); ok {
+			rc.Password = p
+		}
+	}
+	return rc, true
+}
+
+func mountToRelayConfig(masterAddr string, m *radioai.Mount) iceRelayConfig {
 	return iceRelayConfig{
 		Mount:                  m.Name,
 		LocalMount:             m.Name,
@@ -226,17 +263,26 @@ func mountToRelay(masterAddr string, m *radioai.Mount) iceRelayConfig {
 func (ic *icecastConfig) Update(config *ClusterConfig, isMaster bool, masterAddr string) {
 	ic.Mounts = nil
 	ic.Relays = nil
-	if isMaster {
-		mounts := make([]iceMountConfig, 0)
-		for _, m := range config.ListMounts() {
+	mounts := make([]iceMountConfig, 0)
+	relays := make([]iceRelayConfig, 0)
+
+	for _, m := range config.ListMounts() {
+		switch {
+		case m.IsRelay():
+			if rc, ok := relayToConfig(m); ok {
+				relays = append(relays, rc)
+			}
+		case isMaster:
 			mounts = append(mounts, mountToConfig(m))
+		default:
+			relays = append(relays, mountToRelayConfig(masterAddr, m))
 		}
+	}
+
+	if len(mounts) > 0 {
 		ic.Mounts = mounts
-	} else {
-		relays := make([]iceRelayConfig, 0)
-		for _, m := range config.ListMounts() {
-			relays = append(relays, mountToRelay(masterAddr, m))
-		}
+	}
+	if len(relays) > 0 {
 		ic.Relays = relays
 	}
 }