kafka influx message generator

influx metrics generator and kafka producer

package main

import (
	"fmt"
	"time"
	"math/rand"
	"gopkg.in/confluentinc/confluent-kafka-go.v1/kafka"
)

func makeTimestamp() int64 {
	return time.Now().UnixNano() 
}

func main() {

	p, err := kafka.NewProducer(&kafka.ConfigMap{"bootstrap.servers": "my-cluster-kafka-brokers.kafka.svc.cluster.local:9092"})
	if err != nil {
		panic(err)
	}

	defer p.Close()

	// Delivery report handler for produced messages
	go func() {
		for e := range p.Events() {
			switch ev := e.(type) {
			case *kafka.Message:
				if ev.TopicPartition.Error != nil {
					// fmt.Printf("Delivery failed: %v\n", ev.TopicPartition)
				}
			}
		}
	}()

	// Produce messages to topic (asynchronously)
	topic := "topic"
	for {
		word := fmt.Sprintf("fffff,go_version=1.15.6,version=1.17.0 metrics_gathered=%vi,gather_errors=0i,metrics_written=39368i,metrics_dropped=0i %v", rand.Intn(10000000), makeTimestamp())
		p.Produce(&kafka.Message{
		TopicPartition: kafka.TopicPartition{Topic: &topic, Partition: kafka.PartitionAny},
		Value:          []byte(word),
	}, nil)
		time.Sleep(500 * time.Millisecond)
		}
	p.Flush(15 * 1000)
}