<commit_before>package main

import (
	"bufio"
	"flag"
	"fmt"
	"github.com/Shopify/sarama"
	"github.com/cenkalti/backoff"
	rfc3164 "github.com/jeromer/syslogparser/rfc3164"
	"github.com/rcrowley/go-metrics"
	"log"
	"net"
	"os"
	"os/signal"
	"syscall"
	"time"
)

var (
	cfg      *config
	logger   = log.New(os.Stderr, "", log.Ldate|log.Ltime|log.Lshortfile)
	port     = flag.Int("port", 514, "port on which to listen")
	verbose  = flag.Bool("verbose", false, "print all messages to stdout")
	publish  = flag.Bool("publish", false, "publish messages to kafka")
	topic    = flag.String("topic", "syslog", "kafka topic to publish on")
	zkstring = flag.String("zkstring", "localhost:2181", "ZooKeeper broker connection string")
	riemann  = flag.String("riemann", "localhost:5555", "Riemann TCP host:port")

	connectionsCounter = metrics.NewCounter()
	sendMeter          = metrics.NewMeter()
	sendErrorsMeter    = metrics.NewMeter()
	sendHistogram      = metrics.NewHistogram(metrics.NewExpDecaySample(1028, 0.015))
)

type config struct {
	port     int
	verbose  bool
	publish  bool
	topic    string
	zkstring string
	riemann  string
}

func (c *config) String() string {
	return fmt.Sprintf("port: %d, verbose: %t, publish: %t, topic: %s, zkstring: %s, riemann: %s",
		c.port, c.verbose, c.publish, c.topic, c.zkstring, c.riemann)
}

type openConnection struct {
	connection net.Conn
	done       chan bool
}

type server struct {
	listener    net.Listener
	connections []*openConnection
	client      *sarama.Client
	producer    *sarama.Producer
	shutdown    chan bool
}

func (s *server) process(line []byte) {
	p := rfc3164.NewParser(line)
	if err := p.Parse(); err != nil {
		logger.Println("failed to parse:", err)
		return
	}

	parts := p.Dump()
	content := parts["content"].(string)

	if cfg.publish {
		if cfg.verbose {
			logger.Println("enqueuing", content)
		}

		put := func() error {
			started := time.Now()
			err := s.producer.SendMessage(cfg.topic, nil, sarama.StringEncoder(content))
			duration := time.Since(started)
			sendHistogram.Update(duration.Nanoseconds() / 1000000)

			if err != nil {
				logger.Println("error sending message, will retry.", err)
				sendErrorsMeter.Mark(1)
			} else {
				sendMeter.Mark(1)
			}
			return err
		}

		policy := backoff.NewExponentialBackOff()
		policy.MaxElapsedTime = time.Minute * 5
		err := backoff.Retry(put, policy)
		if err != nil {
			// retrying a bunch of times failed...
			logger.Println("failed sending message.", err)
		}

	}

	if cfg.verbose {
		logger.Println(content)
	}
}

func (s *server) handleConnection(conn *openConnection) {
	defer conn.connection.Close()

	logger.Println("got connection from:", conn.connection.RemoteAddr())

	incomingMessageMeter := metrics.NewMeter()
	metrics.Register(fmt.Sprintf("%s incomingMessages", conn.connection.RemoteAddr()), incomingMessageMeter)

	scanner := bufio.NewScanner(conn.connection)
	for scanner.Scan() {
		b := []byte(scanner.Text())
		incomingMessageMeter.Mark(1)
		s.process(b)<commit_msg>publish json bytes of message<commit_after>package main

import (
	"bufio"
	"encoding/json"
	"flag"
	"fmt"
	"github.com/Shopify/sarama"
	"github.com/cenkalti/backoff"
	rfc3164 "github.com/jeromer/syslogparser/rfc3164"
	"github.com/rcrowley/go-metrics"
	"log"
	"net"
	"os"
	"os/signal"
	"syscall"
	"time"
)

var (
	cfg      *config
	logger   = log.New(os.Stderr, "", log.Ldate|log.Ltime|log.Lshortfile)
	port     = flag.Int("port", 514, "port on which to listen")
	verbose  = flag.Bool("verbose", false, "print all messages to stdout")
	publish  = flag.Bool("publish", false, "publish messages to kafka")
	topic    = flag.String("topic", "syslog", "kafka topic to publish on")
	zkstring = flag.String("zkstring", "localhost:2181", "ZooKeeper broker connection string")
	riemann  = flag.String("riemann", "localhost:5555", "Riemann TCP host:port")

	connectionsCounter = metrics.NewCounter()
	sendMeter          = metrics.NewMeter()
	sendErrorsMeter    = metrics.NewMeter()
	sendHistogram      = metrics.NewHistogram(metrics.NewExpDecaySample(1028, 0.015))
)

type config struct {
	port     int
	verbose  bool
	publish  bool
	topic    string
	zkstring string
	riemann  string
}

func (c *config) String() string {
	return fmt.Sprintf("port: %d, verbose: %t, publish: %t, topic: %s, zkstring: %s, riemann: %s",
		c.port, c.verbose, c.publish, c.topic, c.zkstring, c.riemann)
}

type openConnection struct {
	connection net.Conn
	done       chan bool
}

type server struct {
	listener    net.Listener
	connections []*openConnection
	client      *sarama.Client
	producer    *sarama.Producer
	shutdown    chan bool
}

func (s *server) process(line []byte) {
	p := rfc3164.NewParser(line)
	if err := p.Parse(); err != nil {
		logger.Println("failed to parse:", err)
		return
	}

	parts := p.Dump()

	jsonBytes, err := json.Marshal(parts)
	if err != nil {
		logger.Println("error marshaling message, skipping.", err)
		return
	}

	if cfg.publish {
		if cfg.verbose {
			logger.Println("enqueuing", string(jsonBytes))
		}

		put := func() error {
			started := time.Now()
			err := s.producer.SendMessage(cfg.topic, nil, sarama.ByteEncoder(jsonBytes))
			duration := time.Since(started)
			sendHistogram.Update(duration.Nanoseconds() / 1000000)

			if err != nil {
				logger.Println("error sending message, will retry.", err)
				sendErrorsMeter.Mark(1)
			} else {
				sendMeter.Mark(1)
			}
			return err
		}

		policy := backoff.NewExponentialBackOff()
		policy.MaxElapsedTime = time.Minute * 5
		err := backoff.Retry(put, policy)
		if err != nil {
			// retrying a bunch of times failed...
			logger.Println("failed sending message.", err)
		}

	}
}

func (s *server) handleConnection(conn *openConnection) {
	defer conn.connection.Close()

	logger.Println("got connection from:", conn.connection.RemoteAddr())

	incomingMessageMeter := metrics.NewMeter()
	metrics.Register(fmt.Sprintf("%s incomingMessages", conn.connection.RemoteAddr()), incomingMessageMeter)

	scanner := bufio.NewScanner(conn.connection)
	for scanner.Scan() {
		b := []byte(scanner.Text())
		incomingMessageMeter.Mark(1)
		s.process(b)
{'fraction_non_alphanumeric': 0.14026483570377635, 'fraction_numerical': 0.01373222167729279, 'mean_word_length': 3.9418416801292406, 'pattern_counts': {'":': 0, '<': 3, '<?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': '160838', 'n_tokens_mistral': 2335, 'n_tokens_neox': 2116, 'n_words': 491}