func sendToLibhoney(ctx context.Context, toBeSent chan event.Event, toBeResent chan event.Event,
	delaySending chan int, doneSending chan bool) {
	for {
		select {
		case delay := <-delaySending:
			time.Sleep(time.Duration(delay) * time.Millisecond)
		default:
		}
		select {
		case ev := <-toBeResent:
			sendEvent(ev)
			continue
		default:
		}
		select {
		case ev, ok := <-toBeSent:
			if !ok {
				doneSending <- true
				return
			}
			sendEvent(ev)
			continue
		default:
		}
		time.Sleep(100 * time.Millisecond)
	}
}
sendToLibhoney reads from the toBeSent channel and shoves the events into
libhoney events, sending them on their way.
package main
import (
	"bufio"
	"context"
	"crypto/sha256"
	"encoding/json"
	"fmt"
	"io"
	"io/ioutil"
	"math/rand"
	"os"
	"os/signal"
	"path/filepath"
	"regexp"
	"sort"
	"strconv"
	"strings"
	"sync"
	"syscall"
	"time"
	dynsampler "github.com/honeycombio/dynsampler-go"
	"github.com/honeycombio/libhoney-go"
	"github.com/honeycombio/libhoney-go/transmission"
	"github.com/honeycombio/urlshaper"
	"github.com/sirupsen/logrus"
	"github.com/honeycombio/honeytail/event"
	"github.com/honeycombio/honeytail/parsers"
	"github.com/honeycombio/honeytail/parsers/arangodb"
	"github.com/honeycombio/honeytail/parsers/csv"
	"github.com/honeycombio/honeytail/parsers/htjson"
	"github.com/honeycombio/honeytail/parsers/keyval"
	"github.com/honeycombio/honeytail/parsers/mongodb"
	"github.com/honeycombio/honeytail/parsers/mysql"
	"github.com/honeycombio/honeytail/parsers/nginx"
	"github.com/honeycombio/honeytail/parsers/postgresql"
	"github.com/honeycombio/honeytail/parsers/regex"
	"github.com/honeycombio/honeytail/parsers/syslog"
	"github.com/honeycombio/honeytail/sample"
	"github.com/honeycombio/honeytail/tail"
)
const backfillMessage = `Running in backfill mode may result in rate-limited events for this dataset. This is expected behavior.
Be aware that if you are also sending data from other sources to this dataset, this may result in events
being dropped.`
const rateLimitMessageBackoff = `One or more of your events sent by honeytail has been rate-limited, and is being resent. This may result in a
notification to your team administrators that rate-limiting has been triggered on this dataset. While this is
expected behavior, it is possible that if there are other sources sending data to this dataset, that this may
result in events being dropped. 
`
var previouslyRateLimited = false
func run(ctx context.Context, options GlobalOptions) {
	logrus.Info("Starting honeytail")
	stats := newResponseStats()
	sigs := make(chan os.Signal, 1)
	ctx, cancel := context.WithCancel(ctx)
	signal.Notify(sigs, syscall.SIGINT, syscall.SIGTERM)
	libhConfig := libhoney.Config{
		WriteKey:             options.Reqs.WriteKey,
		Dataset:              options.Reqs.Dataset,
		APIHost:              options.APIHost,
		MaxConcurrentBatches: options.NumSenders,
		SendFrequency:        time.Duration(options.BatchFrequencyMs) * time.Millisecond,
		MaxBatchSize:         options.BatchSize,
		BlockOnSend: true,
		BlockOnResponse: true,
		PendingWorkCapacity: 20 * options.NumSenders,
	}
	if options.DebugOut {
		libhConfig.Output = &libhoney.WriterOutput{}
	}
	if err := libhoney.Init(libhConfig); err != nil {
		logrus.WithFields(logrus.Fields{"err": err}).Fatal(
			"Error occured while spinning up Transimission")
	}
	if options.Backfill {
		logrus.Info(backfillMessage)
	}
	var prefixRegex *parsers.ExtRegexp
	if options.PrefixRegex == "" {
		prefixRegex = nil
	} else {
		prefixRegex = &parsers.ExtRegexp{regexp.MustCompile(options.PrefixRegex)}
	}
	var linesChans []chan string
	var err error
	tc := tail.Config{
		Paths:       options.Reqs.LogFiles,
		FilterPaths: options.FilterFiles,
		Type:        tail.RotateStyleSyslog,
		Options:     options.Tail,
	}
	if options.TailSample {
		linesChans, err = tail.GetSampledEntries(ctx, tc, options.SampleRate)
	} else {
		linesChans, err = tail.GetEntries(ctx, tc)
	}
	if err != nil {
		logrus.WithFields(logrus.Fields{"err": err}).Fatal(
			"Error occurred while trying to tail logfile")
	}
	go func() {
		sig := <-sigs
		fmt.Fprintf(os.Stderr, "Aborting! Caught signal \"%s\"\n", sig)
		fmt.Fprintf(os.Stderr, "Cleaning up...\n")
		cancel()
		select {
		case <-sigs:
			fmt.Fprintf(os.Stderr, "Caught second signal... Aborting.\n")
			os.Exit(1)
		case <-time.After(10 * time.Second):
			fmt.Fprintf(os.Stderr, "Taking too long... Aborting.\n")
			os.Exit(1)
		}
	}()
	parsersWG := sync.WaitGroup{}
	responsesWG := sync.WaitGroup{}
	for _, lines := range linesChans {
		parser, opts := getParserAndOptions(options)
		if parser == nil {
			logrus.WithFields(logrus.Fields{"parser": options.Reqs.ParserName}).Fatal(
				"Parser not found. Use --list to show valid parsers")
		}
		if err := parser.Init(opts); err != nil {
			logrus.Fatalf(
				"Error initializing %s parser module: %v", options.Reqs.ParserName, err)
		}
		toBeSent := make(chan event.Event, options.NumSenders)
		doneSending := make(chan bool)
		toBeResent := make(chan event.Event, 2*options.NumSenders)
		delaySending := make(chan int, 2*options.NumSenders)
		modifiedToBeSent := modifyEventContents(toBeSent, options)
		realToBeSent := make(chan event.Event, 10*options.NumSenders)
		go func() {
			wg := sync.WaitGroup{}
			for i := uint(0); i < options.NumSenders; i++ {
				wg.Add(1)
				go func() {
					for ev := range modifiedToBeSent {
						realToBeSent <- ev
					}
					wg.Done()
				}()
			}
			wg.Wait()
			close(realToBeSent)
		}()
		go sendToLibhoney(ctx, realToBeSent, toBeResent, delaySending, doneSending)
		responses := libhoney.TxResponses()
		responsesWG.Add(1)
		go func() {
			handleResponses(responses, stats, toBeResent, delaySending, options)
			responsesWG.Done()
		}()
		parsersWG.Add(1)
		go func(plines chan string) {
			parser.ProcessLines(plines, toBeSent, prefixRegex)
			close(toBeSent)
			<-doneSending
			parsersWG.Done()
		}(lines)
	}
	parsersWG.Wait()
	libhoney.Close()
	responsesWG.Wait()
	stats.log()
	stats.logFinal()
	logrus.Info("Honeytail is all done, goodbye!")
}
func getParserAndOptions(options GlobalOptions) (parsers.Parser, interface{}) {
	var parser parsers.Parser
	var opts interface{}
	switch options.Reqs.ParserName {
	case "regex":
		parser = &regex.Parser{}
		opts = &options.Regex
		opts.(*regex.Options).NumParsers = int(options.NumSenders)
	case "nginx":
		parser = &nginx.Parser{}
		opts = &options.Nginx
		opts.(*nginx.Options).NumParsers = int(options.NumSenders)
	case "json":
		parser = &htjson.Parser{}
		opts = &options.JSON
		opts.(*htjson.Options).NumParsers = int(options.NumSenders)
	case "keyval":
		parser = &keyval.Parser{}
		opts = &options.KeyVal
		opts.(*keyval.Options).NumParsers = int(options.NumSenders)
	case "mongo", "mongodb":
		parser = &mongodb.Parser{}
		opts = &options.Mongo
		opts.(*mongodb.Options).NumParsers = int(options.NumSenders)
	case "mysql":
		parser = &mysql.Parser{
			SampleRate: int(options.SampleRate),
		}
		opts = &options.MySQL
		opts.(*mysql.Options).NumParsers = int(options.NumSenders)
	case "postgresql":
		opts = &options.PostgreSQL
		parser = &postgresql.Parser{}
	case "arangodb":
		parser = &arangodb.Parser{}
		opts = &options.ArangoDB
	case "csv":
		parser = &csv.Parser{}
		opts = &options.CSV
		opts.(*csv.Options).NumParsers = int(options.NumSenders)
	case "syslog":
		parser = &syslog.Parser{}
		opts = &options.Syslog
		opts.(*syslog.Options).NumParsers = int(options.NumSenders)
	}
	parser, _ = parser.(parsers.Parser)
	return parser, opts
}
func modifyEventContents(toBeSent chan event.Event, options GlobalOptions) chan event.Event {
	parsedAddFields := map[string]string{}
	for _, addField := range options.AddFields {
		splitField := strings.SplitN(addField, "=", 2)
		if len(splitField) != 2 {
			logrus.WithFields(logrus.Fields{
				"add_field": addField,
			}).Fatal("unable to separate provided field into a key=val pair")
		}
		parsedAddFields[splitField[0]] = splitField[1]
	}
	shaper := &requestShaper{}
	if len(options.RequestShape) != 0 {
		shaper.pr = &urlshaper.Parser{}
		if options.ShapePrefix != "" {
			shaper.prefix = options.ShapePrefix + "_"
		}
		for _, rpat := range options.RequestPattern {
			pat := urlshaper.Pattern{Pat: rpat}
			if err := pat.Compile(); err != nil {
				logrus.WithField("request_pattern", rpat).WithError(err).Fatal(
					"Failed to compile provided pattern.")
			}
			shaper.pr.Patterns = append(shaper.pr.Patterns, &pat)
		}
	}
	var dynamicSampler dynsampler.Sampler
	if len(options.DynSample) != 0 {
		dynamicSampler = &dynsampler.AvgSampleWithMin{
			GoalSampleRate:    options.GoalSampleRate,
			ClearFrequencySec: options.DynWindowSec,
			MinEventsPerSec:   options.MinSampleRate,
		}
		if err := dynamicSampler.Start(); err != nil {
			logrus.WithField("error", err).Fatal("dynsampler failed to start")
		}
	}
	var deterministicSampler *sample.DeterministicSampler
	if options.DeterministicSample != "" {
		var err error
		deterministicSampler, err = sample.NewDeterministicSampler(options.SampleRate)
		if err != nil {
			logrus.WithField("error", err).Fatal("error creating deterministic sampler")
		}
	}
	type DataAugmentationMap map[string]map[string]map[string]interface{}
	var daMap DataAugmentationMap
	if options.DAMapFile != "" {
		raw, err := ioutil.ReadFile(options.DAMapFile)
		if err != nil {
			logrus.WithField("error", err).Fatal("failed to read Data Augmentation Map file")
		}
		err = json.Unmarshal(raw, &daMap)
		if err != nil {
			logrus.WithField("error", err).Fatal("failed to unmarshal Data Augmentation Map from JSON")
		}
	}
	var baseTime, startTime time.Time
	if options.RebaseTime {
		var err error
		baseTime, err = getBaseTime(options)
		if err != nil {
			logrus.WithError(err).Fatal("--rebase_time specified but cannot rebase")
		}
		startTime = time.Now()
	}
	newSent := make(chan event.Event, options.NumSenders)
	go func() {
		wg := sync.WaitGroup{}
		for i := uint(0); i < options.NumSenders; i++ {
			wg.Add(1)
			go func() {
				for ev := range toBeSent {
					for _, field := range options.RequestShape {
						shaper.requestShape(field, &ev, options)
					}
					for sourceField, augmentableVals := range daMap {
						if val, ok := ev.Data[sourceField]; ok {
							if val, ok := val.(string); ok {
								if newFields, ok := augmentableVals[val]; ok {
									for k, v := range newFields {
										ev.Data[k] = v
									}
								}
							}
						}
					}
					for _, field := range options.DropFields {
						delete(ev.Data, field)
					}
					for _, field := range options.ScrubFields {
						if val, ok := ev.Data[field]; ok {
							newVal := sha256.Sum256([]byte(fmt.Sprintf("%v", val)))
							ev.Data[field] = fmt.Sprintf("%x", newVal)
						}
					}
					for k, v := range parsedAddFields {
						ev.Data[k] = v
					}
					if options.PreSampledField != "" {
						var presampledRate int
						if psr, ok := ev.Data[options.PreSampledField]; ok {
							switch psr := psr.(type) {
							case float64:
								presampledRate = int(psr)
							case string:
								if val, err := strconv.Atoi(psr); err == nil {
									presampledRate = val
								}
							}
						}
						ev.SampleRate = presampledRate
					} else {
						ev.SampleRate = int(options.SampleRate)
						if dynamicSampler != nil {
							key := makeDynsampleKey(&ev, options)
							sr := dynamicSampler.GetSampleRate(key)
							if rand.Intn(sr) != 0 {
								ev.SampleRate = -1
							} else {
								ev.SampleRate = sr
							}
						}
						if deterministicSampler != nil {
							sampleKey, ok := ev.Data[options.DeterministicSample].(string)
							if !ok {
								logrus.WithField("event_data", ev.Data).
									WithField("field", options.DeterministicSample).
									Error("Field to deterministically sample on does not exist in event, leaving it to random chance")
								if rand.Intn(int(options.SampleRate)) != 0 {
									ev.SampleRate = -1
								}
							} else {
								if !deterministicSampler.Sample(sampleKey) {
									ev.SampleRate = -1
								}
							}
						}
					}
					if options.RebaseTime {
						ev.Timestamp = rebaseTime(baseTime, startTime, ev.Timestamp)
					}
					if len(options.JSONFields) > 0 {
						for _, field := range options.JSONFields {
							jsonVal, ok := ev.Data[field].(string)
							if !ok {
								logrus.WithField("field", field).
									Warn("Error asserting given field as string")
								continue
							}
							var jsonMap map[string]interface{}
							if err := json.Unmarshal([]byte(jsonVal), &jsonMap); err != nil {
								logrus.WithField("field", field).
									Warn("Error unmarshalling field as JSON")
								continue
							}
							ev.Data[field] = jsonMap
						}
					}
					if len(options.RenameFields) > 0 {
						for _, kv := range options.RenameFields {
							kvPair := strings.Split(kv, "=")
							if len(kvPair) != 2 {
								logrus.WithField("arg", kv).
									Error("Invalid --rename_field arg. Should be format 'before=after' ")
								continue
							}
							val, ok := ev.Data[kvPair[0]]
							if !ok {
								logrus.WithField("before_field", kvPair[0]).
									WithField("after_field", kvPair[1]).
									Error("Did not find before_field in event.")
								continue
							}
							delete(ev.Data, kvPair[0])
							ev.Data[kvPair[1]] = val
						}
					}
					newSent <- ev
				}
				wg.Done()
			}()
		}
		wg.Wait()
		close(newSent)
	}()
	return newSent
}
func makeDynsampleKey(ev *event.Event, options GlobalOptions) string {
	key := make([]string, len(options.DynSample))
	for i, field := range options.DynSample {
		if val, ok := ev.Data[field]; ok {
			switch val := val.(type) {
			case bool:
				key[i] = strconv.FormatBool(val)
			case int64:
				key[i] = strconv.FormatInt(val, 10)
			case float64:
				key[i] = strconv.FormatFloat(val, 'E', -1, 64)
			case string:
				key[i] = val
			default:
				key[i] = ""  
			}
		}
	}
	return strings.Join(key, "_")
}
type requestShaper struct {
	prefix string
	pr     *urlshaper.Parser
}
func (r *requestShaper) requestShape(field string, ev *event.Event,
	options GlobalOptions) {
	if val, ok := ev.Data[field]; ok {
		strval, ok := val.(string)
		if !ok {
			logrus.WithFields(logrus.Fields{
				"value": val,
				"field": field,
				"event": *ev,
			}).Error("Error! Value did not correctly assert to be type string in request shaping. Skipping shaping.")
			return
		}
		parts := strings.Split(strval, " ")
		var path string
		if len(parts) == 3 {
			ev.Data[r.prefix+field+"_method"] = parts[0]
			ev.Data[r.prefix+field+"_protocol_version"] = parts[2]
			path = parts[1]
		} else {
			path = parts[0]
		}
		res, err := r.pr.Parse(path)
		if err != nil {
			return
		}
		ev.Data[r.prefix+field+"_uri"] = res.URI
		ev.Data[r.prefix+field+"_path"] = res.Path
		if res.Query != "" {
			ev.Data[r.prefix+field+"_query"] = res.Query
		}
		for k, v := range res.QueryFields {
			if options.RequestParseQuery == "all" ||
				whitelistKey(options.RequestQueryKeys, k) {
				if len(v) > 1 {
					sort.Strings(v)
				}
				ev.Data[r.prefix+field+"_query_"+k] = strings.Join(v, ", ")
			}
		}
		for k, v := range res.PathFields {
			ev.Data[r.prefix+field+"_path_"+k] = v[0]
		}
		ev.Data[r.prefix+field+"_shape"] = res.Shape
		ev.Data[r.prefix+field+"_pathshape"] = res.PathShape
		if res.QueryShape != "" {
			ev.Data[r.prefix+field+"_queryshape"] = res.QueryShape
		}
	}
}
func whitelistKey(whiteKeys []string, key string) bool {
	for _, whiteKey := range whiteKeys {
		if key == whiteKey {
			return true
		}
	}
	return false
}