func randMsg(m []byte, generator rand.Rand) {
	for i := range m {
		m[i] = chars[generator.Intn(len(chars))]
	}
}
randMsg returns a random message generated from the chars byte slice.
Message length of m bytes as defined by Config.msgSize.
package main
import (
	"crypto/tls"
	"crypto/x509"
	"flag"
	"fmt"
	"io/ioutil"
	"log"
	"math"
	"math/rand"
	"os"
	"strconv"
	"strings"
	"sync/atomic"
	"time"
	"github.com/Shopify/sarama"
	"github.com/jamiealquiza/tachymeter"
)
type config struct {
	brokers               []string
	topic                 string
	msgSize               int
	msgRate               uint64
	batchSize             int
	compression           sarama.CompressionCodec
	compressionName       string
	requiredAcks          sarama.RequiredAcks
	requiredAcksName      string
	workers               int
	writersPerWorker      int
	noop                  bool
	interval              int
	kafkaVersion          sarama.KafkaVersion
	kafkaVersionString    string
	tls                   bool
	tlsCaCertificate      string
	tlsCertificate        string
	tlsPrivateKey         string
	tlsInsecureSkipVerify bool
}
var (
	Config = &config{}
	chars = []byte("abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ1234567890!@#$^&*(){}][:<>.")
	sentCnt uint64
	errCnt  uint64
	validVersions = []string{
		"0.8.2.0",
		"0.8.2.1",
		"0.8.2.2",
		"0.9.0.0",
		"0.9.0.1",
		"0.10.0.0",
		"0.10.0.1",
		"0.10.1.0",
		"0.10.2.0",
		"0.11.0.0",
		"1.0.0.0",
		"1.1.0.0",
		"1.1.1.0",
		"2.0.0.0",
		"2.0.1.0",
		"2.1.0.0",
		"2.2.0.0",
		"2.2.1.0",
		"2.3.0.0",
		"2.4.0.0",
		"2.5.0.0",
		"2.6.0.0",
	}
)
func init() {
	flag.StringVar(&Config.topic, "topic", "sangrenel", "Kafka topic to produce to")
	flag.IntVar(&Config.msgSize, "message-size", 300, "Message size (bytes)")
	flag.Uint64Var(&Config.msgRate, "produce-rate", 100000000, "Global write rate limit (messages/sec)")
	flag.IntVar(&Config.batchSize, "message-batch-size", 500, "Messages per batch")
	flag.StringVar(&Config.compressionName, "compression", "none", "Message compression: none, gzip, snappy")
	flag.StringVar(&Config.requiredAcksName, "required-acks", "local", "RequiredAcks config: none, local, all")
	flag.BoolVar(&Config.noop, "noop", false, "Test message generation performance (does not connect to Kafka)")
	flag.IntVar(&Config.workers, "workers", 1, "Number of workers")
	flag.IntVar(&Config.writersPerWorker, "writers-per-worker", 5, "Number of writer (Kafka producer) goroutines per worker")
	brokerString := flag.String("brokers", "localhost:9092", "Comma delimited list of Kafka brokers")
	flag.IntVar(&Config.interval, "interval", 5, "Statistics output interval (seconds)")
	flag.StringVar(&Config.kafkaVersionString, "api-version", "0.10.2.0", "Explicit sarama.Version string")
	flag.BoolVar(&Config.tls, "tls", false, "Whether to enable TLS communcation")
	flag.StringVar(&Config.tlsCaCertificate, "tls-ca-cert", "", "Path to the CA SSL certificate")
	flag.StringVar(&Config.tlsCertificate, "tls-cert-file", "", "Path to the certificate file")
	flag.StringVar(&Config.tlsPrivateKey, "tls-key-file", "", "Path to the private key file")
	flag.BoolVar(&Config.tlsInsecureSkipVerify, "tls-insecure-skip-verify", false, "TLS insecure skip verify")
	flag.Parse()
	Config.brokers = strings.Split(*brokerString, ",")
	switch Config.compressionName {
	case "gzip":
		Config.compression = sarama.CompressionGZIP
	case "snappy":
		Config.compression = sarama.CompressionSnappy
	case "none":
		Config.compression = sarama.CompressionNone
	default:
		fmt.Printf("Invalid compression option: %s\n", Config.compressionName)
		os.Exit(1)
	}
	switch Config.requiredAcksName {
	case "none":
		Config.requiredAcks = sarama.NoResponse
	case "local":
		Config.requiredAcks = sarama.WaitForLocal
	case "all":
		Config.requiredAcks = sarama.WaitForAll
	default:
		fmt.Printf("Invalid required-acks option: %s\n", Config.requiredAcksName)
		os.Exit(1)
	}
}
func parseKafkaVersion(kafkaVersion string) sarama.KafkaVersion {
	version, err := sarama.ParseKafkaVersion(kafkaVersion)
	if err != nil {
		fmt.Printf("Invalid API version option: %s\n", Config.kafkaVersionString)
		fmt.Printf("Options: %+q\n", validVersions)
		os.Exit(1)
	}
	return version
}
func main() {
	if graphiteIp != "" {
		go graphiteWriter()
	}
	version := Config.kafkaVersionString
	if version == "" {
		version = "automatic"
	}
	fmt.Printf("\nStarting %d client workers, %d writers per worker\n", Config.workers, Config.writersPerWorker)
	fmt.Printf("Message size %d bytes, %d message limit per batch\n", Config.msgSize, Config.batchSize)
	fmt.Printf("API Version: %s, Compression: %s, RequiredAcks: %s\n",
		version, Config.compressionName, Config.requiredAcksName)
	t := tachymeter.New(&tachymeter.Config{Size: 300000, Safe: true})
	for i := 0; i < Config.workers; i++ {
		go worker(i+1, t)
	}
	var currSentCnt, lastSentCnt uint64
	var currErrCnt, lastErrCnt uint64
	interval := time.Duration(Config.interval) * time.Second
	ticker := time.Tick(interval)
	start := time.Now()
	for {
		<-ticker
		intervalTime := time.Since(start).Seconds()
		t.SetWallTime(time.Since(start))
		lastSentCnt = currSentCnt
		currSentCnt = atomic.LoadUint64(&sentCnt)
		sentSinceLastInterval := currSentCnt - lastSentCnt
		outputBytes, outputString := calcOutput(intervalTime, sentSinceLastInterval)
		lastErrCnt = currErrCnt
		currErrCnt = atomic.LoadUint64(&errCnt)
		errSinceLastInterval := currErrCnt - lastErrCnt
		errRate := (float64(errSinceLastInterval) / float64(sentSinceLastInterval)) * 100
		stats := t.Calc()
		metrics["rate"] = float64(sentSinceLastInterval) / intervalTime
		metrics["error_rate"] = errRate
		metrics["output"] = outputBytes
		metrics["p99"] = (float64(stats.Time.P99.Nanoseconds()) / 1000) / 1000
		metrics["timestamp"] = float64(time.Now().Unix())
		if graphiteIp != "" {
			metricsOutgoing <- metrics
		}
		fmt.Println()
		log.Printf("[ topic: %s ]\n", Config.topic)
		fmt.Printf("> Messages: %s @ %.0f msgs/sec. | error rate %.2f%%\n",
			outputString,
			metrics["rate"],
			metrics["error_rate"])
		if !Config.noop {
			fmt.Printf("> Batches: %.2f batches/sec. | %s p99 | %s HMean | %s Min | %s Max\n",
				stats.Rate.Second, round(stats.Time.P99), round(stats.Time.HMean), round(stats.Time.Min), round(stats.Time.Max))
			fmt.Println(stats.Histogram.String(50))
			if int(sentSinceLastInterval) > len(t.Times) {
				newTachy := tachymeter.New(&tachymeter.Config{Size: int(2 * sentSinceLastInterval)})
				*t = *newTachy
			} else {
				t.Reset()
			}
		}
		start = time.Now()
	}
}
func worker(n int, t *tachymeter.Tachymeter) {
	switch Config.noop {
	case false:
		cId := "worker_" + strconv.Itoa(n)
		conf := sarama.NewConfig()
		conf.Producer.Compression = Config.compression
		conf.Producer.Return.Successes = true
		conf.Producer.RequiredAcks = Config.requiredAcks
		conf.Producer.Flush.MaxMessages = Config.batchSize
		conf.Producer.MaxMessageBytes = Config.msgSize + 50
		conf.Version = parseKafkaVersion(Config.kafkaVersionString)
		if Config.tls {
			tlsConfig := &tls.Config{
				InsecureSkipVerify: Config.tlsInsecureSkipVerify,
			}
			if len(Config.tlsCaCertificate) > 0 {
				caCert, err := ioutil.ReadFile(Config.tlsCaCertificate)
				if err != nil {
					log.Println(err)
					os.Exit(1)
				}
				caCertPool := x509.NewCertPool()
				caCertPool.AppendCertsFromPEM(caCert)
				tlsConfig.RootCAs = caCertPool
			}
			if len(Config.tlsCertificate) > 0 && len(Config.tlsPrivateKey) > 0 {
				cert, err := tls.LoadX509KeyPair(Config.tlsCertificate, Config.tlsPrivateKey)
				if err != nil {
					log.Fatal(err)
				}
				tlsConfig.Certificates = []tls.Certificate{cert}
			}
			conf.Net.TLS.Enable = true
			conf.Net.TLS.Config = tlsConfig
		}
		client, err := sarama.NewClient(Config.brokers, conf)
		if err != nil {
			log.Println(err)
			os.Exit(1)
		} else {
			log.Printf("%s connected\n", cId)
		}
		for i := 0; i < Config.writersPerWorker; i++ {
			go writer(client, t)
		}
	default:
		for i := 0; i < Config.writersPerWorker; i++ {
			go dummyWriter(t)
		}
	}
	wait := make(chan bool)
	<-wait
}
func writer(c sarama.Client, t *tachymeter.Tachymeter) {
	producer, err := sarama.NewSyncProducerFromClient(c)
	if err != nil {
		log.Println(err.Error())
	}
	defer producer.Close()
	source := rand.NewSource(time.Now().UnixNano())
	generator := rand.New(source)
	msgBatch := make([]*sarama.ProducerMessage, 0, Config.batchSize)
	for {
		intervalEnd := time.Now().Add(time.Second)
		countStart := atomic.LoadUint64(&sentCnt)
		var sendTime time.Time
		var intervalSent uint64
		for {
			toSend := (Config.msgRate - intervalSent) / uint64((Config.workers * Config.writersPerWorker))
			n := int(math.Min(float64(toSend), float64(Config.batchSize)))
			for i := 0; i < n; i++ {
				msgData := make([]byte, Config.msgSize)
				randMsg(msgData, *generator)
				msg := &sarama.ProducerMessage{Topic: Config.topic, Value: sarama.ByteEncoder(msgData)}
				msgBatch = append(msgBatch, msg)
			}
			sendTime = time.Now()
			err = producer.SendMessages(msgBatch)
			if err != nil {
				atomic.AddUint64(&errCnt, uint64(len(err.(sarama.ProducerErrors))))
			}
			t.AddTime(time.Since(sendTime))
			atomic.AddUint64(&sentCnt, uint64(len(msgBatch)))
			msgBatch = msgBatch[:0]
			intervalSent = atomic.LoadUint64(&sentCnt) - countStart
			sendEstimate := intervalSent + uint64((Config.batchSize*Config.workers*Config.writersPerWorker)-Config.writersPerWorker)
			if sendEstimate >= Config.msgRate {
				break
			}
		}
		time.Sleep(intervalEnd.Sub(time.Now()))
	}
}
func dummyWriter(t *tachymeter.Tachymeter) {
	source := rand.NewSource(time.Now().UnixNano())
	generator := rand.New(source)
	msgBatch := make([]*sarama.ProducerMessage, 0, Config.batchSize)
	for {
		for i := 0; i < Config.batchSize; i++ {
			msgData := make([]byte, Config.msgSize)
			randMsg(msgData, *generator)
			msg := &sarama.ProducerMessage{Topic: Config.topic, Value: sarama.ByteEncoder(msgData)}
			msgBatch = append(msgBatch, msg)
		}
		atomic.AddUint64(&sentCnt, uint64(len(msgBatch)))
		t.AddTime(time.Duration(0))
		msgBatch = msgBatch[:0]
	}
}