<commit_before>package main

import (
	"flag"
	"hash/fnv"
	"net"
	"sync"
	"time"

	log "github.com/Sirupsen/logrus"
	"github.com/stripe/veneur"
)

var (
	configFile = flag.String("f", "", "The config file to read for settings.")
)

type packet struct {
	buf []byte
	ip  string
}

func main() {
	flag.Parse()

	if configFile == nil || *configFile == "" {
		log.Fatal("You must specify a config file")
	}

	err := veneur.ReadConfig(*configFile)
	if err != nil {
		log.WithError(err).Fatal("Error reading config file")
	}

	if veneur.Config.Debug {
		log.SetLevel(log.DebugLevel)
		log.WithField("config", veneur.Config).Debug("Starting with config")
	}

	veneur.InitSentry()
	defer func() {
		veneur.ConsumePanic(recover())
	}()
	veneur.InitStats()

	// Start the dispatcher.
	log.WithField("number", veneur.Config.NumWorkers).Info("Starting workers")
	workers := make([]*veneur.Worker, veneur.Config.NumWorkers)
	for i := 0; i < veneur.Config.NumWorkers; i++ {
		worker := veneur.NewWorker(i+1, veneur.Config.Percentiles, veneur.Config.HistCounters, veneur.Config.SetSize, veneur.Config.SetAccuracy)
		worker.Start()
		workers[i] = worker
	}

	// Start the UDP server!
	log.WithFields(log.Fields{
		"address":        veneur.Config.UDPAddr,
		"flush_interval": veneur.Config.Interval,
	}).Info("UDP server listening")
	serverAddr, err := net.ResolveUDPAddr("udp", veneur.Config.UDPAddr)
	if err != nil {
		log.WithError(err).Fatal("Error resolving address")
	}

	serverConn, err := net.ListenUDP("udp", serverAddr)

	if err != nil {
		log.WithError(err).Fatal("Error listening for UDP")
	}
	if err := serverConn.SetReadBuffer(veneur.Config.ReadBufferSizeBytes); err != nil {
		log.WithError(err).Fatal("Could not set recvbuf size")
	}

	defer serverConn.Close()

	ticker := time.NewTicker(veneur.Config.Interval)
	go func() {
		defer func() {
			veneur.ConsumePanic(recover())
		}()
		for t := range ticker.C {
			metrics := make([][]veneur.DDMetric, veneur.Config.NumWorkers)
			for i, w := range workers {
				log.WithFields(log.Fields{
					"worker": i,
					"tick":   t,
				}).Debug("Flushing")
				metrics = append(metrics, w.Flush(veneur.Config.Interval))
			}
			veneur.Flush(metrics, veneur.Config.FlushLimit)
		}
	}()

	packetPool := &sync.Pool{
		New: func() interface{} {
			return packet{
				buf: make([]byte, veneur.Config.MetricMaxLength),
				ip:  "",
			}
		},
	}

	// precompute digest for the "veneur.unique_sender_ips" set, since it's the
	// same every time
	veneurIPdigester := fnv.New32()
	veneurIPdigester.Write([]byte("veneur.unique_sender_ips"))
	veneurIPdigester.Write([]byte("set"))
	uniqueSenderIPDigest := veneurIPdigester.Sum32()

	// Creates N workers to handle incoming packets, parsing them,
	// hashing them and dispatching them on to workers that do the storage.
	parserChan := make(chan packet)
	for i := 0; i < veneur.Config.NumWorkers; i++ {
		go func() {
			defer func() {
				veneur.ConsumePanic(recover())
			}()
			for packetbuf := range parserChan {
				handlePacket(workers, packetbuf.buf)
				// we don't want to talk to statsd in the packet-processing code
				// path - it causes packet drops (even in the parsing loop).
				// instead, we can push the metric to a worker and let it handle
				// the IO at flush time, which has much less performance impact
				// note that all of these metrics hit a single worker, we might
				// want to have a dedicated worker for this in the future
				workers[uniqueSenderIPDigest%uint32(veneur.Config.NumWorkers)].WorkChan <- veneur.Metric{
					Name:       "veneur.unique_sender_ips",
					Digest:     uniqueSenderIPDigest,
					Value:      packetbuf.ip,
					SampleRate: 1.0,
					Type:       "set",
				}
				packetbuf.buf = packetbuf.buf[:cap(packetbuf.buf)]
				// handlePacket generates a Metric struct which contains only
				// strings, so there are no outstanding references to the byte
				// slice containing the packet
				// therefore, at this point we can return it to the pool
				packetPool.Put(packetbuf)
			}
		}()
	}

	// Read forever!
	for {
		packetbuf := packetPool.Get().(packet)
		n, addr, err := serverConn.ReadFromUDP(packetbuf.buf)
		if err != nil {
			log.WithError(err).Error("Error reading packetbuf from UDP")
			continue
		}
		packetbuf.buf = packetbuf.buf[:n]
		packetbuf.ip = addr.IP.String()
		parserChan <- packetbuf<commit_msg>Stringify IP in parser loop<commit_after>package main

import (
	"flag"
	"hash/fnv"
	"net"
	"sync"
	"time"

	log "github.com/Sirupsen/logrus"
	"github.com/stripe/veneur"
)

var (
	configFile = flag.String("f", "", "The config file to read for settings.")
)

type packet struct {
	buf []byte
	ip  net.IP
}

func main() {
	flag.Parse()

	if configFile == nil || *configFile == "" {
		log.Fatal("You must specify a config file")
	}

	err := veneur.ReadConfig(*configFile)
	if err != nil {
		log.WithError(err).Fatal("Error reading config file")
	}

	if veneur.Config.Debug {
		log.SetLevel(log.DebugLevel)
		log.WithField("config", veneur.Config).Debug("Starting with config")
	}

	veneur.InitSentry()
	defer func() {
		veneur.ConsumePanic(recover())
	}()
	veneur.InitStats()

	// Start the dispatcher.
	log.WithField("number", veneur.Config.NumWorkers).Info("Starting workers")
	workers := make([]*veneur.Worker, veneur.Config.NumWorkers)
	for i := 0; i < veneur.Config.NumWorkers; i++ {
		worker := veneur.NewWorker(i+1, veneur.Config.Percentiles, veneur.Config.HistCounters, veneur.Config.SetSize, veneur.Config.SetAccuracy)
		worker.Start()
		workers[i] = worker
	}

	// Start the UDP server!
	log.WithFields(log.Fields{
		"address":        veneur.Config.UDPAddr,
		"flush_interval": veneur.Config.Interval,
	}).Info("UDP server listening")
	serverAddr, err := net.ResolveUDPAddr("udp", veneur.Config.UDPAddr)
	if err != nil {
		log.WithError(err).Fatal("Error resolving address")
	}

	serverConn, err := net.ListenUDP("udp", serverAddr)

	if err != nil {
		log.WithError(err).Fatal("Error listening for UDP")
	}
	if err := serverConn.SetReadBuffer(veneur.Config.ReadBufferSizeBytes); err != nil {
		log.WithError(err).Fatal("Could not set recvbuf size")
	}

	defer serverConn.Close()

	ticker := time.NewTicker(veneur.Config.Interval)
	go func() {
		defer func() {
			veneur.ConsumePanic(recover())
		}()
		for t := range ticker.C {
			metrics := make([][]veneur.DDMetric, veneur.Config.NumWorkers)
			for i, w := range workers {
				log.WithFields(log.Fields{
					"worker": i,
					"tick":   t,
				}).Debug("Flushing")
				metrics = append(metrics, w.Flush(veneur.Config.Interval))
			}
			veneur.Flush(metrics, veneur.Config.FlushLimit)
		}
	}()

	packetPool := &sync.Pool{
		New: func() interface{} {
			return packet{
				buf: make([]byte, veneur.Config.MetricMaxLength),
				ip:  nil,
			}
		},
	}

	// precompute digest for the "veneur.unique_sender_ips" set, since it's the
	// same every time
	veneurIPdigester := fnv.New32()
	veneurIPdigester.Write([]byte("veneur.unique_sender_ips"))
	veneurIPdigester.Write([]byte("set"))
	uniqueSenderIPDigest := veneurIPdigester.Sum32()

	// Creates N workers to handle incoming packets, parsing them,
	// hashing them and dispatching them on to workers that do the storage.
	parserChan := make(chan packet)
	for i := 0; i < veneur.Config.NumWorkers; i++ {
		go func() {
			defer func() {
				veneur.ConsumePanic(recover())
			}()
			for packetbuf := range parserChan {
				handlePacket(workers, packetbuf.buf)
				// we don't want to talk to statsd in the packet-processing code
				// path - it causes packet drops (even in the parsing loop).
				// instead, we can push the metric to a worker and let it handle
				// the IO at flush time, which has much less performance impact
				// note that all of these metrics hit a single worker, we might
				// want to have a dedicated worker for this in the future
				workers[uniqueSenderIPDigest%uint32(veneur.Config.NumWorkers)].WorkChan <- veneur.Metric{
					Name:       "veneur.unique_sender_ips",
					Digest:     uniqueSenderIPDigest,
					Value:      packetbuf.ip.String(),
					SampleRate: 1.0,
					Type:       "set",
				}
				packetbuf.buf = packetbuf.buf[:cap(packetbuf.buf)]
				// handlePacket generates a Metric struct which contains only
				// strings, so there are no outstanding references to the byte
				// slice containing the packet
				// therefore, at this point we can return it to the pool
				packetPool.Put(packetbuf)
			}
		}()
	}

	// Read forever!
	for {
		packetbuf := packetPool.Get().(packet)
		n, addr, err := serverConn.ReadFromUDP(packetbuf.buf)
		if err != nil {
			log.WithError(err).Error("Error reading packetbuf from UDP")
			continue
		}
		packetbuf.buf = packetbuf.buf[:n]
		packetbuf.ip = addr.IP
		parserChan <- packetbuf
{'fraction_non_alphanumeric': 0.13522192912031197, 'fraction_numerical': 0.0025232251404977636, 'mean_word_length': 3.555903866248694, 'pattern_counts': {'":': 8, '<': 11, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5479084', 'n_tokens_mistral': 3326, 'n_tokens_neox': 2950, 'n_words': 746}