Merge pull request strukturag#47 from strukturag/proxy-list-etcd

Add support for fetching proxy URLs from etcd cluster.

server.conf.in

@@ -99,12 +99,12 @@ connectionsperhost = 8
 
 [mcu]
 # The type of the MCU to use. Currently only "janus" and "proxy" are supported.
-type = janus
+# Leave empty to disable MCU functionality.
+#type =
 
 # For type "janus": the URL to the websocket endpoint of the MCU server.
 # For type "proxy": a space-separated list of proxy URLs to connect to.
-# Leave empty to disable MCU functionality.
-url =
+#url =
 
 # For type "janus": the maximum bitrate per publishing stream (in bits per
 # second).
@@ -116,13 +116,46 @@ url =
 # Default is 2 mbit/sec.
 #maxscreenbitrate = 2097152
 
+# For type "proxy": type of URL configuration for proxy servers.
+# Defaults to "static".
+#
+# Possible values:
+# - static: A space-separated list of proxy URLs is given in the "url" option.
+# - etcd: Proxy URLs are retrieved from an etcd cluster (see below).
+#urltype = "static"
+
 # For type "proxy": the id of the token to use when connecting to proxy servers.
 #token_id = server1
 
 # For type "proxy": the private key for the configured token id to use when
 # connecting to proxy servers.
 #token_key = privkey.pem
 
+# For url type "etcd": Comma-separated list of static etcd endpoints to
+# connect to.
+#endpoints = 127.0.0.1:2379,127.0.0.1:22379,127.0.0.1:32379
+
+# For url type "etcd": Options to perform endpoint discovery through DNS SRV.
+# Only used if no endpoints are configured manually.
+#discoverysrv = example.com
+#discoveryservice = foo
+
+# For url type "etcd": Path to private key, client certificate and CA
+# certificate if TLS authentication should be used.
+#clientkey = /path/to/etcd-client.key
+#clientcert = /path/to/etcd-client.crt
+#cacert = /path/to/etcd-ca.crt
+
+# For url type "etcd": Key prefix of MCU proxy entries. All keys below will be
+# watched and assumed to contain a JSON document. The entry "address" from this
+# document will be used as proxy URL, other contents in the document will be
+# ignored.
+#
+# Example:
+# "/signaling/proxy/server/one" -> {"address": "https://proxy1.domain.invalid"}
+# "/signaling/proxy/server/two" -> {"address": "https://proxy2.domain.invalid"}
+#keyprefix = /signaling/proxy/server
+
 [turn]
 # API key that the MCU will need to send when requesting TURN credentials.
 #apikey = the-api-key-for-the-rest-service

src/server/main.go

@@ -91,7 +91,7 @@ func createTLSListener(addr string, certFile, keyFile string) (net.Listener, err
 }
 
 func main() {
-	log.SetFlags(log.Ldate | log.Ltime | log.Lmicroseconds | log.Lshortfile)
+	log.SetFlags(log.Lshortfile)
 	flag.Parse()
 
 	if *showVersion {
@@ -155,12 +155,13 @@ func main() {
 	}
 
 	mcuUrl, _ := config.GetString("mcu", "url")
-	if mcuUrl != "" {
-		mcuType, _ := config.GetString("mcu", "type")
-		if mcuType == "" {
-			mcuType = signaling.McuTypeDefault
-		}
+	mcuType, _ := config.GetString("mcu", "type")
+	if mcuType == "" && mcuUrl != "" {
+		log.Printf("WARNING: Old-style MCU configuration detected with url but no type, defaulting to type %s", signaling.McuTypeJanus)
+		mcuType = signaling.McuTypeJanus
+	}
 
+	if mcuType != "" {
 		var mcu signaling.Mcu
 		mcuRetry := initialMcuRetry
 		mcuRetryTimer := time.NewTimer(mcuRetry)
@@ -169,21 +170,21 @@ func main() {
 			case signaling.McuTypeJanus:
 				mcu, err = signaling.NewMcuJanus(mcuUrl, config, nats)
 			case signaling.McuTypeProxy:
-				mcu, err = signaling.NewMcuProxy(mcuUrl, config)
+				mcu, err = signaling.NewMcuProxy(config)
 			default:
 				log.Fatal("Unsupported MCU type: ", mcuType)
 			}
 			if err == nil {
 				err = mcu.Start()
 				if err != nil {
-					log.Printf("Could not create %s MCU at %s: %s", mcuType, mcuUrl, err)
+					log.Printf("Could not create %s MCU: %s", mcuType, err)
 				}
 			}
 			if err == nil {
 				break
 			}
 
-			log.Printf("Could not initialize %s MCU at %s (%s) will retry in %s", mcuType, mcuUrl, err, mcuRetry)
+			log.Printf("Could not initialize %s MCU (%s) will retry in %s", mcuType, err, mcuRetry)
 			mcuRetryTimer.Reset(mcuRetry)
 			select {
 			case sig := <-sigChan:
@@ -197,8 +198,9 @@ func main() {
 					} else {
 						mcuUrl, _ = config.GetString("mcu", "url")
 						mcuType, _ = config.GetString("mcu", "type")
-						if mcuType == "" {
-							mcuType = signaling.McuTypeDefault
+						if mcuType == "" && mcuUrl != "" {
+							log.Printf("WARNING: Old-style MCU configuration detected with url but no type, defaulting to type %s", signaling.McuTypeJanus)
+							mcuType = signaling.McuTypeJanus
 						}
 					}
 				}
@@ -212,7 +214,7 @@ func main() {
 		}
 		defer mcu.Stop()
 
-		log.Printf("Using MCU %s at %s\n", mcuType, mcuUrl)
+		log.Printf("Using %s MCU", mcuType)
 		hub.SetMcu(mcu)
 	}
 

src/signaling/api_proxy.go

@@ -252,3 +252,19 @@ type EventProxyServerMessage struct {
 	ClientId string `json:"clientId,omitempty"`
 	Load     int64  `json:"load,omitempty"`
 }
+
+// Information on a proxy in the etcd cluster.
+
+type ProxyInformationEtcd struct {
+	Address string `json:"address"`
+}
+
+func (p *ProxyInformationEtcd) CheckValid() error {
+	if p.Address == "" {
+		return fmt.Errorf("address missing")
+	}
+	if p.Address[len(p.Address)-1] != '/' {
+		p.Address += "/"
+	}
+	return nil
+}