pipeline.go

//
// January 2016, cisco
//
// Copyright (c) 2016 by cisco Systems, Inc.
// All rights reserved.
//
//
// Code orchestrating the various bits of the pipeline
//
// Go generate directives. Think of this as limited makefile. Some
// autogeration of source is required. Currently, we have:
//
//  - autogeneration of .pb.go from xport_grpc_out.proto
//  - auto patching of vendor directory, until and if changes can be
//    back ported.
//
//go:generate protoc --go_out=plugins=grpc:. xport_grpc_out.proto
//
package main

import (
	"bytes"
	"flag"
	"fmt"
	telem "github.com/cisco/bigmuddy-network-telemetry-proto/proto_go"
	"github.com/dlintw/goconf"
	"github.com/evalphobia/logrus_fluent"
	"github.com/prometheus/client_golang/prometheus"
	log "github.com/sirupsen/logrus"
	"io/ioutil"
	"os"
	"os/signal"
	"runtime"
	"strconv"
	"strings"
	"syscall"
	"text/template"
	"time"
)

var theLogger = log.New()

//
// This is the entry with a common tag identifying pipeline. All
// logging in this application will hang off this context.
var logger *log.Entry

type nodeConfig struct {
	config *goconf.ConfigFile
}

type outputNodeModule interface {
	configure(nodename string, nc nodeConfig) (
		error, chan<- dataMsg, chan<- *ctrlMsg)
}

type node struct {
	name     string
	ctrlChan chan<- *ctrlMsg
}

type outputNode struct {
	node
	dataChan chan<- dataMsg
}

type inputNodeModule interface {
	configure(nodename string, nc nodeConfig,
		chans []chan<- dataMsg) (error, chan<- *ctrlMsg)
}

type inputNode struct {
	node
}

//
// Conductor state holds the global configuration in a way which is
// convenient to export if asked.
type conductorState struct {
	Started        time.Time `json:"started"`
	Debug          bool      `json:"debugging"`
	Version        string    `json:"version"`
	Configfile     string    `json:"configfile"`
	Logfile        string    `json:"logfile"`
	ID             string    `json:"id"`
	logging2Stdout bool
	// Meta monitoring object for global oper state
	monitor *ConductorMetaMonitor
	// Output node modules keyed by module type
	outputNodeModules map[string](func() outputNodeModule)
	// Output nodes keyed by instance name
	outputNodes map[string]*outputNode
	// Input node modules keyed by module type
	inputNodeModules map[string](func() inputNodeModule)
	// Input nodes keyed by instance name
	inputNodes map[string]*inputNode
	// Configured?
	configured bool
}

var appVersion string = "unspecified"

//
// Catch all for system constants
const (
	DATACHANNELDEPTH = 1000
	LOGDATALIMIT     = 32
	ID               = "pipeline"
)

// CLI options, keep to a minimum.  Configuration should be in .conf
// if startup, or, eventually, programmatic for run time config.
var (
	//
	// global options
	debugFlag = flag.Bool("debug", false,
		"Dump debugging information to logfile (for event dump see 'logdata' in config file)")
	logFileName = flag.String("log", "pipeline.log",
		"Specify the path and name of log file [-log= logs to stdout]")
	configFileName = flag.String("config", "pipeline.conf",
		"Specify the path and name of the configuration file")
	pemFileName = flag.String("pem", "",
		"Specify the RSA key pair used to protect passwords in config "+
			"(e.g. id_rsa as generated by 'ssh-keygen -t rsa -b 4096')")
	logModels = flag.String("logModels", "models.txt", "dump the GPB models supported")
)

// Conductor object
var (
	conductor conductorState
)

//
// Metamonitoring in the conductor part involves tracking global state
// of the pipeline. (Collector based on
// prometheus/example_memstats_test.go example).
//
// In the object we track the descriptors of metrics collected. We
// also implement the Collector interface against this object.
//
type ConductorMetaMonitor struct {
	NumGoroutine  *prometheus.Desc
	NumCPU        *prometheus.Desc
	MemAlloc      *prometheus.Desc
	MemTotalAlloc *prometheus.Desc
	MemMallocs    *prometheus.Desc
	MemFrees      *prometheus.Desc
	MemNumGC      *prometheus.Desc
	UpTime        *prometheus.Desc
	CfgInSection  *prometheus.Desc
	CfgOutSection *prometheus.Desc
	ChanOccupancy *prometheus.Desc
}

// Describe sends the Descriptor objects for the Metrics we intend to
// collect.
func (c ConductorMetaMonitor) Describe(ch chan<- *prometheus.Desc) {
	ch <- c.NumGoroutine
	ch <- c.NumCPU
	ch <- c.MemAlloc
	ch <- c.MemTotalAlloc
	ch <- c.MemMallocs
	ch <- c.MemFrees
	ch <- c.MemNumGC
	ch <- c.UpTime
	ch <- c.CfgInSection
	ch <- c.CfgOutSection
	ch <- c.ChanOccupancy
}

// If metamonitoring is setup, the function will be called
// periodically causing us to collect and report on the conductor
// state
func (c ConductorMetaMonitor) Collect(ch chan<- prometheus.Metric) {

	//
	// Collect memory statistics
	var ms runtime.MemStats
	runtime.ReadMemStats(&ms)

	ch <- prometheus.MustNewConstMetric(
		c.NumGoroutine,
		prometheus.GaugeValue,
		float64(runtime.NumGoroutine()),
	)

	ch <- prometheus.MustNewConstMetric(
		c.NumCPU,
		prometheus.GaugeValue,
		float64(runtime.NumCPU()),
	)

	ch <- prometheus.MustNewConstMetric(
		c.MemAlloc,
		prometheus.GaugeValue,
		float64(ms.Alloc),
	)

	ch <- prometheus.MustNewConstMetric(
		c.MemTotalAlloc,
		prometheus.GaugeValue,
		float64(ms.TotalAlloc),
	)

	ch <- prometheus.MustNewConstMetric(
		c.MemMallocs,
		prometheus.GaugeValue,
		float64(ms.Mallocs),
	)

	ch <- prometheus.MustNewConstMetric(
		c.MemFrees,
		prometheus.GaugeValue,
		float64(ms.Frees),
	)

	ch <- prometheus.MustNewConstMetric(
		c.MemNumGC,
		prometheus.CounterValue,
		float64(ms.NumGC),
	)

	ch <- prometheus.MustNewConstMetric(
		c.UpTime,
		prometheus.CounterValue,
		float64(time.Since(conductor.Started)))

	ch <- prometheus.MustNewConstMetric(
		c.CfgInSection,
		prometheus.GaugeValue,
		float64(len(conductor.inputNodes)),
	)

	ch <- prometheus.MustNewConstMetric(
		c.CfgOutSection,
		prometheus.GaugeValue,
		float64(len(conductor.outputNodes)),
	)

	for nodename, node := range conductor.outputNodes {
		ch <- prometheus.MustNewConstMetric(
			c.ChanOccupancy,
			prometheus.GaugeValue,
			float64(100*len(node.dataChan)/cap(node.dataChan)),
			nodename)
	}

	//
	// We could also allocate the metrics and stash them once
	// (like we do with descriptors), and then update with every
	// invocation.
	//
}

func startupConductor(
	configFIlename *string,
	logFilename *string,
	debugFlag *bool) {

	// Initialise input and output module factories. Eventually a
	// plugin pattern might be adopted.
	activeOutputNodeModule := map[string](func() outputNodeModule){
		"kafka":   kafkaOutputModuleNew,
		"tap":     tapOutputModuleNew,
		"metrics": metricsOutputModuleNew,
		"grpc":    grpcOutputModuleNew,
	}
	activeInputNodeModule := map[string](func() inputNodeModule){
		"replay": replayInputModuleNew,
		"tcp":    tcpInputModuleNew,
		"grpc":   grpcInputModuleNew,
		"kafka":  kafkaInputModuleNew,
		"udp":    udpInputModuleNew,
		"gnmi":   gnmiInputModuleNew,
	}

	conductor = conductorState{
		Started:           time.Now(),
		Debug:             *debugFlag,
		Version:           appVersion,
		Configfile:        *configFileName,
		Logfile:           *logFileName,
		outputNodeModules: activeOutputNodeModule,
		outputNodes:       make(map[string]*outputNode),
		inputNodeModules:  activeInputNodeModule,
		inputNodes:        make(map[string]*inputNode),
		configured:        true,
	}
}

//
// Parse command line options, and get setup.
func startup() *os.File {

	if conductor.configured {
		return nil
	}

	//
	// Overwrite the usage message
	flag.Usage = func() {
		fmt.Printf("pipeline, version %s\n\n", appVersion)
		flag.PrintDefaults()
	}

	//
	// Parse command line options
	flag.Parse()
	startupConductor(configFileName, logFileName, debugFlag)

	var outfile *os.File
	var err error
	if conductor.Logfile != "" {
		outfile, err = os.OpenFile(conductor.Logfile,
			os.O_APPEND|os.O_CREATE|os.O_RDWR, 0666)
		if err != nil {
			fmt.Printf("Failed opening logfile %s: %v\n",
				conductor.Logfile, err)
			os.Exit(-1)
		}
	} else {
		conductor.logging2Stdout = true
		outfile = os.Stdout
	}

	theLogger.Formatter = &log.TextFormatter{
		FullTimestamp:   true,
		TimestampFormat: "2006-01-02 15:04:05.000000"}
	//logger.Formatter = new(log.JSONFormatter)

	//
	// Use file handle for output logging. From here on in logging
	// is up and running.
	theLogger.Out = outfile

	//
	// Setup logging level.
	if conductor.Debug {
		theLogger.Level = log.DebugLevel
	} else {
		theLogger.Level = log.InfoLevel
	}

	return outfile
}

type ConfigEnvironment struct {
}

func (c *ConfigEnvironment) Env(k string) (string, error) {
	//
	// Read environment into a map to apply to text template.
	env := make(map[string]string)
	for _, pair := range os.Environ() {
		sep := strings.Index(pair, "=")
		env[pair[0:sep]] = pair[sep+1:]
	}

	v, ok := env[k]
	if ok {
		return v, nil
	} else {
		return "", fmt.Errorf(
			"config includes '%s', but not set in environment variable.", k)
	}
}

//
// loadConfig loads configuration of nodes from .conf, kicks them off,
// and plugs them together.
func loadConfig() error {

	var configBuffer bytes.Buffer
	var nc nodeConfig
	var dataChans = make([]chan<- dataMsg, 0)
	var err error

	configAsTemplate, err := template.ParseFiles(conductor.Configfile)
	if err != nil {
		return fmt.Errorf("failed to open configuration file %s, %v\n",
			conductor.Configfile, err)
	}

	err = configAsTemplate.Execute(&configBuffer, &ConfigEnvironment{})
	if err != nil {
		return err
	}

	//
	// Read configuration to setup input and output nodes
	cfg, err := goconf.ReadConfigBytes(configBuffer.Bytes())
	if err != nil {
		return fmt.Errorf("read configuration file %s, %v\n",
			conductor.Configfile, err)
	}

	nc.config = cfg

	conductor.ID, err = nc.config.GetString("default", "id")
	if err != nil {
		hostname, err := os.Hostname()
		if err != nil {
			hostname = "unknown"
		}
		conductor.ID = fmt.Sprintf("%s.%s", ID, hostname)
	}

	//
	// Tag is what will be used by fluentd to route messages.
	logger = theLogger.WithFields(log.Fields{"tag": conductor.ID})

	logsocket, err := nc.config.GetString("default", "fluentd")
	if err == nil {
		logsocketsplit := strings.Split(logsocket, ":")
		if len(logsocketsplit) == 2 {
			port, err := strconv.Atoi(logsocketsplit[1])
			if err == nil {
				//
				// Api provides no indication of success/error?
				hook := logrus_fluent.NewHook(
					logsocketsplit[0], port)

				levels := []log.Level{
					log.PanicLevel,
					log.FatalLevel,
					log.ErrorLevel,
					log.WarnLevel,
					log.InfoLevel,
				}
				if conductor.Debug {
					levels = append(levels, log.DebugLevel)
				}
				hook.SetLevels(levels)

				theLogger.Hooks.Add(hook)
				theLogger.Formatter = new(log.JSONFormatter)

				logger.WithFields(log.Fields{
					"debug":   conductor.Debug,
					"fluentd": logsocket,
					"logfile": conductor.Logfile}).Info(
					"Logging via fluentd configured")

			} else {
				logger.WithError(err).WithFields(log.Fields{
					"debug":   conductor.Debug,
					"fluentd": logsocket,
					"logfile": conductor.Logfile}).Error(
					"Logging via fluentd, condfig specifies bad port")
			}
		} else {
			logger.WithFields(log.Fields{
				"debug":   conductor.Debug,
				"fluentd": logsocket,
				"logfile": conductor.Logfile}).Error(
				"'Logging via fluentd, config must be of the form <host>:<port>")
		}
	} else {
		err = nil
	}

	logger.WithFields(log.Fields{
		"debug":      conductor.Debug,
		"version":    appVersion,
		"config":     conductor.Configfile,
		"fluentd":    logsocket,
		"maxthreads": runtime.GOMAXPROCS(-1),
		"logfile":    conductor.Logfile}).Info(
		"Conductor says hello, loading config")

	models := telem.EncodingPathsSupported()
	modelDesc := make([]string, len(models))
	for i, p := range models {
		modelDesc[i] = fmt.Sprintf("\t%s %s", p.EncodingPath, p.Version)
	}
	ioutil.WriteFile(*logModels, []byte(strings.Join(modelDesc, "\n")), 0644)

	//
	// Initialise meta monitoring and setup ConductorMetaMonitor
	conductor.monitor = &ConductorMetaMonitor{
		NumGoroutine: prometheus.NewDesc(
			"goroutines",
			"currently active goroutines",
			nil, nil),
		NumCPU: prometheus.NewDesc(
			"cpus",
			"available CPUs for pipeline",
			nil, nil),
		MemAlloc: prometheus.NewDesc(
			"alloc",
			"currently allocated memory",
			nil, nil),
		MemTotalAlloc: prometheus.NewDesc(
			"totalalloc",
			"allocated memory, including historic allocations",
			nil, nil),
		MemMallocs: prometheus.NewDesc(
			"mallocs",
			"Calls to malloc memory",
			nil, nil),
		MemFrees: prometheus.NewDesc(
			"frees",
			"Calls to free memory",
			nil, nil),
		MemNumGC: prometheus.NewDesc(
			"gcruns",
			"Garbage collection invocations",
			nil, nil),
		UpTime: prometheus.NewDesc(
			"uptime",
			"Time since pipeline started",
			nil, nil),
		CfgInSection: prometheus.NewDesc(
			"cfginputs",
			"Configured input sections",
			nil, nil),
		CfgOutSection: prometheus.NewDesc(
			"cfgoutputs",
			"Configured output sections",
			nil, nil),
		ChanOccupancy: prometheus.NewDesc(
			"chanoccupancy",
			"Channel occupancy",
			[]string{"section"}, nil),
	}
	prometheus.MustRegister(conductor.monitor)
	//
	// Initialise meta monitoring server
	go metamonitoring_init(nc)

	codec_init(nc)

	// Setup output stages first in no particular order
	for _, section := range cfg.GetSections() {

		//
		// No default inheritance yet
		if section == "default" {
			continue
		}

		modstage, err := cfg.GetString(section, "stage")
		if err != nil {
			return fmt.Errorf("Sections require a 'stage' attribute: %v\n",
				err)
		}

		if modstage != "xport_output" {
			continue
		}

		logger.WithFields(log.Fields{
			"name":    "conductor",
			"section": section}).Debug(
			"Conductor processing section...")

		modtype, err := cfg.GetString(section, "type")
		//
		// Strict with any configuration errors.
		if err != nil {
			return fmt.Errorf("Sections require a 'type' attribute: %v\n",
				err)
		}

		logger.WithFields(log.Fields{
			"name":    "conductor",
			"section": section,
			"type":    modtype}).Debug(
			"Conductor processing section, type...")

		outfn, ok := conductor.outputNodeModules[modtype]
		if !ok {
			return fmt.Errorf("Unsupported 'type' attribute for output module [%s]\n",
				modtype)
		}
		module := outfn()

		logger.WithFields(
			log.Fields{
				"name":    "conductor",
				"stage":   modstage,
				"section": section,
			}).Info("Conductor starting up section")
		err, o_data, o_ctrl := module.configure(section, nc)
		if err != nil {
			logger.WithError(err).WithFields(
				log.Fields{
					"name":    "conductor",
					"stage":   modstage,
					"section": section,
				}).Error("Conductor failed to start up section")
			continue
		}

		onode := new(outputNode)
		onode.name = section
		onode.ctrlChan = o_ctrl
		onode.dataChan = o_data
		dataChans = append(dataChans, o_data)
		conductor.outputNodes[section] = onode
	}

	// Setup and connect input stages next in no particular order
	for _, section := range cfg.GetSections() {

		//
		// No default inheritance yet
		if section == "default" {
			continue
		}

		modstage, err := cfg.GetString(section, "stage")
		if err != nil {
			return fmt.Errorf("Sections require a 'stage' attribute: %v\n",
				err)
		}

		if modstage != "xport_input" {
			continue
		}

		logger.WithFields(log.Fields{
			"name":    "conductor",
			"section": section}).Debug(
			"Conductor processing section...")

		modtype, err := cfg.GetString(section, "type")
		//
		// Strict with any configuration errors.
		if err != nil {
			return fmt.Errorf("Sections require a 'type' attribute: %v\n",
				err)
		}

		logger.WithFields(log.Fields{
			"name":    "conductor",
			"section": section,
			"type":    modtype}).Debug(
			"Conductor processing section, type...")

		infn, ok := conductor.inputNodeModules[modtype]
		if !ok {
			return fmt.Errorf("Unsupported 'type' attribute for input module [%s]\n",
				modtype)
		}
		module := infn()

		logger.WithFields(
			log.Fields{
				"name":    "conductor",
				"stage":   modstage,
				"section": section,
			}).Info("Conductor starting up section")
		err, i_ctrl := module.configure(section, nc, dataChans)
		if err != nil {
			logger.WithError(err).WithFields(
				log.Fields{
					"name":    "conductor",
					"stage":   modstage,
					"section": section,
				}).Error("Conductor failed to start up section")
			continue
		}

		inode := new(inputNode)
		inode.name = section
		inode.ctrlChan = i_ctrl

		conductor.inputNodes[section] = inode
	}

	return nil
}

//
// run the loop waiting for messages, polling state periodically to
// report what is going on, and waiting for interrupt to exit clean.
func run() {

	// Wait for a SIGINT, (typically triggered from CTRL-C), TERM,
	// QUIT. Run cleanup when signal is received. Ideally use os
	// independent os.Interrupt, Kill (but need an exhaustive
	// list.
	signalChan := make(chan os.Signal, 1)
	signal.Notify(signalChan,
		syscall.SIGTERM, // ^C
		syscall.SIGINT,  // kill
		syscall.SIGQUIT, // QUIT
		syscall.SIGABRT)
	cleanupDone := make(chan bool)

	logger.WithFields(log.Fields{
		"debug":   conductor.Debug,
		"config":  conductor.Configfile,
		"logfile": conductor.Logfile}).Debug(
		"Conductor watching for shutdown...")

	go func() {
		for range signalChan {

			if !conductor.logging2Stdout {
				fmt.Printf("\nInterrupt, stopping gracefully\n\n")
			}

			for nodename, node := range conductor.inputNodes {
				if !conductor.logging2Stdout {
					fmt.Printf(" Stopping input '%s'\n", nodename)
				}
				respChan := make(chan *ctrlMsg)
				request := &ctrlMsg{
					id:       SHUTDOWN,
					respChan: respChan,
				}

				//
				// Send shutdown message
				node.ctrlChan <- request
				// Wait for ACK
				<-respChan
				close(node.ctrlChan)
			}

			for nodename, node := range conductor.outputNodes {
				if !conductor.logging2Stdout {
					fmt.Printf(" Stopping output '%s'\n", nodename)
				}
				respChan := make(chan *ctrlMsg)
				request := &ctrlMsg{
					id:       SHUTDOWN,
					respChan: respChan,
				}

				//
				// Send shutdown message
				node.ctrlChan <- request
				// Wait for ACK
				<-respChan
				close(node.ctrlChan)
			}
			//
			// Now that they are all done. Unblock
			cleanupDone <- true
		}
	}()

	//
	// Block here waiting for cleanup. This is likely to be in a
	// main select along other possible conditions (like a timeout
	// to update stats?)
	<-cleanupDone

	logger.Info("Goodbye")
	if !conductor.logging2Stdout {
		fmt.Printf("Done\n")
	}
}

func main() {

	logfile := startup()

	if !conductor.logging2Stdout {
		fmt.Printf("Startup pipeline\n")
		defer logfile.Close()
		fmt.Printf("Load config from [%s], logging in [%s]\n",
			conductor.Configfile,
			conductor.Logfile)
	}

	err := loadConfig()
	if err == nil {
		if !conductor.logging2Stdout {
			fmt.Printf("Wait for ^C to shutdown\n")
		}

		run()
		os.Exit(0)
	} else {
		fmt.Printf("Loading configuration stage failed [%v]\n", err)
		os.Exit(-1)
	}

}