path: root/src/daq/s5proxy/src/s5proxy/stats.go

//
// sfive
//
// sfive - spreadspace streaming statistics suite is a generic
// statistic collection tool for streaming server infrastuctures.
// The system collects and stores meta data like number of views
// and throughput from a number of streaming servers and stores
// it in a global data store.
// The data acquisition is designed to be generic and extensible in
// order to support different streaming software.
// sfive also contains tools and applications to filter and visualize
// live and recorded data.
//
//
// Copyright (C) 2014-2018 Christian Pointner <equinox@spreadspace.org>
//                         Markus Grüneis <gimpf@gimpf.org>
//
// This file is part of sfive.
//
// sfive is free software: you can redistribute it and/or modify
// it under the terms of the GNU General Public License version 3
// as published by the Free Software Foundation.
//
// sfive is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
// GNU General Public License for more details.
//
// You should have received a copy of the GNU General Public License
// along with sfive. If not, see <http://www.gnu.org/licenses/>.
//

package main

import (
	"encoding/json"
	"errors"
	"fmt"
	"net"
	"os"
	"strings"
	"time"
)

// TODO: this is basically a copy from src/hub....

const (
	ProtocolVersion = 2
)

type Stream struct {
	ContentID string `json:"content"`
	Format    string `json:"format"`
	Quality   string `json:"quality"`
}

type Source struct {
	Hostname string   `json:"hostname"`
	Stream   Stream   `json:"stream"`
	Tags     []string `json:"tags,omitempty"`
}

type GeoInfo struct {
	CountryName  string  `json:"country,omitempty"`
	CountryCode2 string  `json:"country-code2,omitempty"`
	RegionName   string  `json:"region,omitempty"`
	RegionCode   string  `json:"region-code,omitempty"`
	CityName     string  `json:"city,omitempty"`
	Latitude     float64 `json:"latitude,omitempty"`
	Longitude    float64 `json:"longitude,omitempty"`
}

type Client struct {
	IP        string `json:"ip"`
	Port      uint   `json:"port,omitempty"`
	UserAgent string `json:"user-agent,omitempty"`
	BytesSent uint   `json:"bytes-sent,omitempty"`
	GeoInfo
}

type UpdateData struct {
	ClientCount   uint     `json:"client-count"`
	BytesReceived uint     `json:"bytes-received,omitempty"`
	BytesSent     uint     `json:"bytes-sent"`
	Clients       []Client `json:"clients,omitempty"`
}

type Update struct {
	StartTime time.Time  `json:"start-time"`
	Duration  int64      `json:"duration-ms"`
	Data      UpdateData `json:"data"`
}

type Header struct {
	Version            uint   `json:"version,omitempty"` // omitempty is needed for data only messages and for REST API
	SourceHubUUID      string `json:"SourceHubUuid,omitempty"`
	SourceHubUpdateID  int    `json:"SourceHubUpdateId,omitempty"`
	ForwardHubUUID     string `json:"ForwardHubUuid,omitempty"`
	ForwardHubUpdateID int    `json:"ForwardHubUpdateId,omitempty"`
}

type UpdateFull struct {
	Header
	Source
	Update
}

type StatsWorker struct {
	stream  Stream
	current map[string]*Client
	trigger chan time.Time
	output  chan<- *UpdateFull
	input   chan *Client
}

func NewStatsWorker(stream Stream, updates chan<- *UpdateFull) (sw StatsWorker) {
	sw.stream = stream
	sw.current = make(map[string]*Client)
	sw.trigger = make(chan time.Time)
	sw.output = updates
	sw.input = make(chan *Client, 100)
	return
}

func (sw StatsWorker) Run() {
	for {
		select {
		case t := <-sw.trigger:
			if t.UnixNano() != 0 {
				upd := &UpdateFull{Source: Source{Stream: sw.stream}, Update: Update{StartTime: t}}
				upd.Data.ClientCount = uint(len(sw.current))
				for _, c := range sw.current {
					upd.Data.Clients = append(upd.Data.Clients, *c)
					upd.Data.BytesSent += c.BytesSent
				}
				select {
				case sw.output <- upd:
				default:
					s5l.Printf("STATS: worker(%v): writing to output channel would block, dropping data update...", sw.stream)
				}
			}
			sw.current = make(map[string]*Client)
		case upd := <-sw.input:
			if data, exists := sw.current[upd.IP]; exists {
				data.BytesSent += upd.BytesSent
			} else {
				sw.current[upd.IP] = upd
			}
		}
	}
}

type Stats struct {
	conf        *Config
	sock        net.Conn
	dataEncoder *json.Encoder
	updates     chan *UpdateFull
	workers     map[string]StatsWorker
}

func (s *Stats) GetUpdateChannel(url string) chan<- *Client {
	s5l.Printf("STATS: got new client for url: %s", url)
	for name, worker := range s.workers {
		if strings.Contains(url, name) {
			return worker.input
		}
	}
	return nil
}

func parseStreamDescriptionElement(desc, desctype string) (e []string, err error) {
	parts := strings.Split(desc, ",")
	for _, p := range parts {
		if p = strings.TrimSpace(p); p == "" {
			err = fmt.Errorf("invalid streamer description %s: '%s'", desctype, desc)
			return
		}
		e = append(e, p)
	}
	return
}

func parseStreamDescription(desc string) (c, f, q []string, err error) {
	parts := strings.Split(desc, "/")
	if len(parts) != 3 {
		err = fmt.Errorf("invalid streamer description: '%s'", desc)
		return
	}
	if c, err = parseStreamDescriptionElement(parts[0], "content"); err != nil {
		return
	}
	if f, err = parseStreamDescriptionElement(parts[1], "format"); err != nil {
		return
	}
	if q, err = parseStreamDescriptionElement(parts[2], "quality"); err != nil {
		return
	}
	return
}

func generateStreamName(format, c, f, q string) string {
	return os.Expand(format, func(k string) string {
		switch k {
		case "content":
			return c
		case "format":
			return f
		case "quality":
			return q
		default:
			return ""
		}
	})
}

func NewStats(conf *Config) (s *Stats, err error) {
	s = &Stats{conf: conf}
	if err = s.connectToHub(); err != nil {
		return
	}

	s.updates = make(chan *UpdateFull, 1000)
	s.workers = make(map[string]StatsWorker)
	var content, format, quality []string
	for _, match := range conf.SFive.Matches {
		if content, format, quality, err = parseStreamDescription(match.Streams); err != nil {
			return
		}
		for _, c := range content {
			for _, f := range format {
				for _, q := range quality {
					name := generateStreamName(match.Format, c, f, q)
					s.workers[name] = NewStatsWorker(Stream{c, f, q}, s.updates)
					s5l.Printf("STATS: adding streamer '%s'", name)
				}
			}
		}
	}
	if len(s.workers) < 1 {
		err = errors.New("no streamers defined")
		return
	}
	return
}

func (s *Stats) connectToHub() (err error) {
	if s.sock, err = net.DialTimeout("unixgram", s.conf.SFive.Sock, time.Second); err != nil {
		return
	}
	s5l.Printf("STATS: connected to sfive-hub on '%s'", s.conf.SFive.Sock)
	s.dataEncoder = json.NewEncoder(s.sock)
	return
}

func (s *Stats) sendUpdates() {
	for {
		upd := <-s.updates
		upd.Version = ProtocolVersion
		upd.Hostname = s.conf.SFive.Hostname
		upd.Tags = s.conf.SFive.Tags
		upd.Duration = int64(s.conf.SFive.Duration)
		if err := s.dataEncoder.Encode(upd); err != nil {
			nErr, ok := err.(*net.OpError)
			if !ok {
				s5l.Printf("STATS: json encode(): %s", err)
			} else {
				if !nErr.Temporary() {
					for {
						time.Sleep(5 * time.Second)
						s5l.Printf("STATS: attempting reconnect to hub")
						if err := s.connectToHub(); err == nil {
							break
						}
					}
				}
			}
		}
	}
}

func (s *Stats) Run() (err error) {
	for _, worker := range s.workers {
		go worker.Run()
	}

	d := time.Duration(s.conf.SFive.Duration) * time.Millisecond
	offset := d - (time.Duration(time.Now().UnixNano()) % d)
	s5l.Printf("STATS: waiting %v for the next duration boundary", offset)
	time.Sleep(offset)
	s5l.Printf("STATS: starting ticker")
	ticker := time.NewTicker(d)
	for _, worker := range s.workers {
		worker.trigger <- time.Unix(0, 0) // stats-worker will not send data but just reset
	}
	go s.sendUpdates()
	for {
		t := <-ticker.C
		t = t.Add(-1 * d)
		for _, worker := range s.workers {
			worker.trigger <- t
		}
	}
}