package autoradio import ( "bytes" "crypto/rand" "encoding/base64" "encoding/json" "errors" "strings" "sync" "time" "git.autistici.org/ale/autoradio/third_party/github.com/coreos/go-etcd/etcd" ) var ( MasterElectionPath = "/icecast/cluster/master" MountPrefix = "/icecast/mounts/" NodePrefix = "/icecast/nodes/" IcecastPort = 8000 ErrIsDirectory = errors.New("key is a directory") ErrIsFile = errors.New("key is a file") ) // A mountpoint for a stream. type Mount struct { // Name (path to the mountpoint). Name string // Username for source authentication. Username string // 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:] } // Status of a mount on an individual Icecast server. type IcecastMountStatus struct { Name string Listeners int BitRate int Quality float64 VideoQuality float64 FrameSize string FrameRate float64 } // Status of a node. This is used to report load and stream status. type NodeStatus struct { // Public IP of this server. IP string // Is the Icecast server up? IcecastUp bool // List of mount points. Mounts []IcecastMountStatus `xml:"mount"` // Bandwidth utilization. BandwidthUsage float64 } func (ns *NodeStatus) NumListeners() int { listeners := 0 for _, m := range ns.Mounts { listeners += m.Listeners } return listeners } // Cache the list of active nodes (the front-ends that need to // retrieve this information continuously, so we limit them to 2qps). type nodesCache struct { ttl time.Duration nodes []*NodeStatus deadline time.Time lock sync.Mutex } type getNodesFunc func() ([]*NodeStatus, error) func newNodesCache() *nodesCache { return &nodesCache{ ttl: 500 * time.Millisecond, } } // Get returns the cached value of 'fn', if valid. If the value is // expired and we get an error from 'fn', we will attempt to return // the previously cached value anyway, along with the error: the // caller can then pick the right failure behavior. func (nc *nodesCache) Get(fn getNodesFunc) ([]*NodeStatus, error) { nc.lock.Lock() defer nc.lock.Unlock() var err error now := time.Now() if now.After(nc.deadline) { var nodes []*NodeStatus if nodes, err = fn(); err == nil { nc.nodes = nodes nc.deadline = now.Add(nc.ttl) } } return nc.nodes, err } // RadioAPI is the actual API to the streaming cluster's database. type RadioAPI struct { client *etcd.Client activeNodesCache *nodesCache } func NewRadioAPI(client *etcd.Client) *RadioAPI { return &RadioAPI{client, newNodesCache()} } // 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) if err != nil || response.Node == nil { return nil, err } if response.Node.Dir { return nil, ErrIsDirectory } var m Mount if err := json.NewDecoder(strings.NewReader(response.Node.Value)).Decode(&m); err != nil { return nil, err } return &m, nil } // SetMount creates or updates a mountpoint. func (r *RadioAPI) SetMount(m *Mount) error { var buf bytes.Buffer if err := json.NewEncoder(&buf).Encode(m); err != nil { return err } _, err := r.client.Set(mountPath(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) return err } // ListMounts returns a list of all the configured mountpoints. func (r *RadioAPI) ListMounts() ([]*Mount, error) { response, err := r.client.Get(MountPrefix, true, false) if err != nil || response.Node == nil { return nil, err } if !response.Node.Dir { return nil, ErrIsFile } result := make([]*Mount, 0, len(response.Node.Nodes)) for _, n := range response.Node.Nodes { if n.Dir { continue } var m Mount if err := json.NewDecoder(strings.NewReader(n.Value)).Decode(&m); err != nil { continue } result = append(result, &m) } return result, nil } // GetMasterAddr returns the address of the current master server. func (r *RadioAPI) GetMasterAddr() (string, error) { response, err := r.client.Get(MasterElectionPath, false, false) if err != nil || response.Node == nil { return "", err } if response.Node.Dir { return "", ErrIsDirectory } return response.Node.Value, nil } // GetNodes returns the list of active cluster nodes. func (r *RadioAPI) doGetNodes() ([]*NodeStatus, error) { response, err := r.client.Get(NodePrefix, false, false) if err != nil || response.Node == nil { return nil, err } if !response.Node.Dir { return nil, ErrIsFile } result := make([]*NodeStatus, 0, len(response.Node.Nodes)) for _, entry := range response.Node.Nodes { var ns NodeStatus if err := json.NewDecoder(strings.NewReader(entry.Value)).Decode(&ns); err == nil { result = append(result, &ns) } } return result, nil } func (r *RadioAPI) GetNodes() ([]*NodeStatus, error) { return r.activeNodesCache.Get(r.doGetNodes) } func (r *RadioAPI) GetNodeIPs() ([]string, error) { nodes, err := r.GetNodes() if err != nil { return nil, err } ips := make([]string, 0, len(nodes)) for _, n := range nodes { ips = append(ips, n.IP) } return ips, nil } // GeneratePassword returns a new random password. func GeneratePassword() string { b := make([]byte, 6) rand.Read(b) return base64.StdEncoding.EncodeToString(b) }