path: root/src/hub/src/spreadspace.org/sfive/s5srv.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-2017 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 sfive

import (
	"context"
	"errors"
	"net"
	"net/http"
	"runtime"
	"sync"
	"time"
)

type ingestToken struct {
	updates  []*UpdateFull
	response chan error
}

type Server struct {
	cfg           SrvConfig
	store         *Store
	numWorker     int
	anonymization AnonymizationAlgo
	geoip         GeoIPLookup
	wgWorker      *sync.WaitGroup
	ingestChan    chan ingestToken
	interfaces    struct {
		pipe     net.Listener
		pipegram net.PacketConn
		web      *http.Server
	}
}

func (srv *Server) transform(update *UpdateFull) *UpdateFull {
	bytesSentTotal := uint(0)
	clients := []Client{}
	for _, client := range update.Data.Clients {
		bytesSentTotal += client.BytesSent

		if srv.geoip != nil {
			if info, err := srv.geoip.Lookup(client.IP); err != nil {
				s5l.Printf("srv|xfrm: Geo-IP lookup failed: %v", err)
			} else {
				client.GeoInfo = *info
			}
		}

		if srv.anonymization != nil {
			if aIP, err := srv.anonymization.Anonymize(client.IP); err != nil {
				s5l.Printf("srv|xfrm: anonymization failed: %v", err)
			} else {
				client.IP = aIP
			}
		}

		clients = append(clients, client)
	}
	update.Data.Clients = clients

	if uint(len(update.Data.Clients)) > update.Data.ClientCount {
		if update.Data.ClientCount > 0 {
			s5l.Printf("srv|xfrm: fixing client-count: %d -> %d", update.Data.ClientCount, len(update.Data.Clients))
		}
		update.Data.ClientCount = uint(len(update.Data.Clients))
	}
	if bytesSentTotal > update.Data.BytesSent {
		if update.Data.BytesSent > 0 {
			s5l.Printf("srv|xfrm: fixing bytes-sent: %d -> %d", update.Data.BytesSent, bytesSentTotal)
		}
		update.Data.BytesSent = bytesSentTotal
	}

	return update
}

func (srv *Server) transformMany(updates []*UpdateFull) {
	for _, update := range updates {
		srv.transform(update)
	}
}

func (srv *Server) ingestWorker(idx int) {
	for {
		select {
		case token, ok := <-srv.ingestChan:
			if !ok {
				return
			}
			srv.transformMany(token.updates)
			token.response <- srv.store.AppendMany(token.updates)
		}
	}
}

func (srv *Server) Ingest(update *UpdateFull) error {
	return srv.IngestMany([]*UpdateFull{update})
}

func (srv *Server) IngestMany(updates []*UpdateFull) error {
	token := ingestToken{updates: updates, response: make(chan error, 1)}
	defer close(token.response)
	srv.ingestChan <- token
	return <-token.response
}

func (srv *Server) shutdownInterfaces() (errors int) {
	ctx, cancel := context.WithTimeout(context.Background(), 90*time.Second)
	c := make(chan error)
	go func() { c <- srv.webStop(ctx) }()
	go func() { c <- srv.pipeStop(ctx) }()
	go func() { c <- srv.pipegramStop(ctx) }()

	errors = 0
	for i := 0; i < 3; i++ {
		if err := <-c; err != nil {
			s5l.Printf("srv: interface shutdown failed failed: %v", err)
			errors++
		}
	}
	close(c) // closing channel here in the hopes that this leads to a panic
	// in case the number of channel reads (for loop above) doesn't match the
	// number of interfaces to be stopped
	cancel()
	return
}

func (srv *Server) Start() (wg sync.WaitGroup, err error) {
	if srv.cfg.Interfaces.Pipe.ListenPath != "" {
		wg.Add(1)
		go func() {
			defer wg.Done()
			srv.ServePipe(srv.cfg.Interfaces.Pipe)
		}()
	}

	if srv.cfg.Interfaces.Pipegram.ListenPath != "" {
		wg.Add(1)
		go func() {
			defer wg.Done()
			srv.ServePipegram(srv.cfg.Interfaces.Pipegram)
		}()
	}

	if srv.cfg.Interfaces.Web.ListenAddr != "" {
		wg.Add(1)
		go func() {
			defer wg.Done()
			srv.ServeWeb(srv.cfg.Interfaces.Web)
		}()
	}

	if srv.cfg.Forwards.SFive.URL != "" {
		wg.Add(1)
		go func() {
			defer wg.Done()
			srv.RunForwarding(srv.cfg.Forwards.SFive)
		}()
	}

	if srv.cfg.Forwards.Elasticsearch.URL != "" {
		wg.Add(1)
		go func() {
			defer wg.Done()
			srv.RunForwardingEs(srv.cfg.Forwards.Elasticsearch)
		}()
	}

	if srv.cfg.Forwards.Graphite.Host != "" {
		wg.Add(1)
		go func() {
			defer wg.Done()
			srv.RunForwardingGraphite(srv.cfg.Forwards.Graphite)
		}()
	}

	if srv.cfg.Forwards.Piwik.URL != "" {
		wg.Add(1)
		go func() {
			defer wg.Done()
			srv.RunForwardingPiwik(srv.cfg.Forwards.Piwik)
		}()
	}
	return wg, nil
}

func (srv *Server) Shutdown() {
	s5l.Printf("srv: shutting down")

	if errors := srv.shutdownInterfaces(); errors != 0 {
		s5l.Printf("srv: shutdown of at least one interface failed, this is an unclean shutdown!!!")
	}

	close(srv.ingestChan) // close ingest channel to tell worker to stop
	srv.wgWorker.Wait()   // wait for worker to finish up
	s5l.Printf("srv: all worker stopped")

	srv.store.Close()
	s5l.Printf("srv: finished")
}

func NewServer(cfg SrvConfig) (srv *Server, err error) {
	srv = &Server{cfg: cfg}
	if srv.store, err = NewStore(cfg.Store); err != nil {
		return
	}

	if cfg.Transform.Anonymize {
		if srv.anonymization, err = NewCryptopanAnonymization(cfg.Transform.AnonKeyfile); err != nil {
			err = errors.New("failed to initialize IP address anonymization: " + err.Error())
			return
		}
		s5l.Printf("srv|xfrm: using IP address anonymization: %s", srv.anonymization)
	}

	if cfg.Transform.GeoipDB != "" {
		if srv.geoip, err = NewMaxMindGeoIP2(cfg.Transform.GeoipDB); err != nil {
			err = errors.New("failed to initialize Geo-IP Lookup: " + err.Error())
			return
		}
		s5l.Printf("srv|xfrm: using Geo-IP Lookup: %s", srv.geoip)
	}

	srv.numWorker = runtime.NumCPU()
	if cfg.Workers > 0 {
		srv.numWorker = cfg.Workers
	}

	srv.wgWorker = &sync.WaitGroup{}
	srv.ingestChan = make(chan ingestToken, srv.numWorker)
	for i := 0; i < srv.numWorker; i = i + 1 {
		srv.wgWorker.Add(1)
		go func(idx int) {
			defer srv.wgWorker.Done()
			srv.ingestWorker(idx)
		}(i)
	}
	s5l.Printf("srv: started with %d worker", srv.numWorker)
	return
}