[WIP] Kafka

database/kafka.go

@@ -0,0 +1,46 @@
+package database
+
+import (
+	"log"
+
+	kafka "github.com/Shopify/sarama"
+	// "github.com/hyperpilotio/ingestor/config"
+	// "github.com/hyperpilotio/ingestor/log"
+)
+
+var producer kafka.SyncProducer
+
+func init() {
+	// FIXME pass configuration into this function
+	producer = newKafkaProducer()
+}
+
+// func newKafkaProducer(config config.Provider) {}
+func newKafkaProducer() kafka.SyncProducer {
+	// FIXME Change the fixed value
+	producer, err := kafka.NewSyncProducer([]string{"localhost:9092"}, nil)
+	if err != nil {
+		log.Fatalln(err)
+	}
+	return producer
+}
+
+func Producer() *kafka.SyncProducer {
+	return &producer
+}
+
+func CloseProducer() {
+	if err := producer.Close(); err != nil {
+		log.Fatalln(err)
+	}
+}
+
+func Send(topic, value string) {
+	msg := &kafka.ProducerMessage{Topic: topic, Value: kafka.StringEncoder(value)}
+	partition, offset, err := producer.SendMessage(msg)
+	if err != nil {
+		log.Printf("FAILED to send message: %s\n", err)
+	} else {
+		log.Printf("> message sent to partition %d at offset %d\n", partition, offset)
+	}
+}

glide.yaml

@@ -9,10 +9,14 @@ import:
   - service/ec2
   - service/ecs
   - service/iam
-- package: github.com/gin-gonic/gin
-  version: ~1.1.4
 - package: github.com/golang/glog
 - package: github.com/spf13/viper
-- package: k8s.io/apimachinery
 - package: k8s.io/client-go
   version: v2.0.0-alpha.1
+- package: k8s.io/apimachinery
+- package: github.com/gin-gonic/gin
+  version: ^1.2.0
+- package: github.com/rifflock/lfshook
+  version: ^1.7.0
+- package: github.com/Shopify/sarama
+  version: ^1.12.0