Merge pull request #1 from inloco/chore/run_utils

Add local run utils

Author: racevedoo

.circleci/config.yml

@@ -77,7 +77,7 @@ jobs:
           command: |
             count=0; \
               until nc -z localhost 9200 || ((count ++ >= 10)); \
-              do echo "Retrying: Verify if Cassandra is ready"; sleep 5; done
+              do echo "Retrying: Verify if Elasticsearch is ready"; sleep 5; done
       - run:
           name: "Run project tests (excluding vendor)"
           command: go test $(go list ./... | grep -v /vendor/)

Makefile

@@ -3,3 +3,17 @@ go/deps:
 
 test:
 	go test $$(go list ./... | grep -v /vendor/)
+
+docker/build:
+	GOOS=linux GOARCH=386 go build -o bin/injector cmd/injector.go
+	GOOS=linux GOARCH=386 go build -o bin/producer util/producer/producer.go
+	docker build --rm=false -t "inlocomedia/kafka-elasticsearch-injector:local" -f cmd/Dockerfile .
+	docker build --rm=false -t "inlocomedia/kafka-elasticsearch-injector:producer-local" -f util/producer/Dockerfile .
+
+docker/run:
+	docker-compose up -d zookeeper kafka schema-registry elasticsearch kibana
+	count=0; \
+          until curl localhost:9200 || ((count ++ >= 10)); \
+          do echo "Retrying: Verify if Elasticsearch is ready"; sleep 5; done
+	curl -XPOST "localhost:9200/_template/my-topic" --data '{"template":"my-topic-*","settings":{"refresh_interval":"30s","number_of_replicas":0},"mappings":{"_default_":{"_all":{"enabled":"false"},"_source":{"enabled":"true"},"properties":{"@timestamp":{"format":"epoch_millis","ignore_malformed":true,"type":"date"}},"dynamic_templates":[{"strings":{"match_mapping_type":"string","mapping":{"type":"text","index":false}}}]}}}'
+	docker-compose up -d producer app

README.md

@@ -113,7 +113,7 @@ go get -u github.com/golang/dep/...
 dep ensure -v
 ```
 
-To run tests, run `docker-compose up -d` and run `make test`. 
+To run tests, run `docker-compose up -d zookeeper kafka schema-registry elasticsearch` and run `make test`. 
 
 ### Versioning
 

VERSION

@@ -1 +1 @@
-0.4.0
+0.4.1

docker-compose.yml

@@ -2,10 +2,12 @@ version: '2'
 services:
   zookeeper:
     image: 'confluentinc/cp-zookeeper:3.1.1'
+    container_name: zookeeper
     environment:
       - ZOOKEEPER_CLIENT_PORT=2181
   kafka:
     image: 'confluentinc/cp-kafka:3.1.1'
+    container_name: kafka
     environment:
       - KAFKA_ZOOKEEPER_CONNECT=zookeeper:2181
       - KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://kafka:9092
@@ -15,6 +17,7 @@ services:
       - "9092:9092"
   schema-registry:
     image: 'confluentinc/cp-schema-registry:3.1.1'
+    container_name: schema-registry
     environment:
       - SCHEMA_REGISTRY_KAFKASTORE_CONNECTION_URL=zookeeper:2181
       - SCHEMA_REGISTRY_HOST_NAME=schema-registry
@@ -23,5 +26,46 @@ services:
       - "8081:8081"
   elasticsearch:
     image: 'elasticsearch:5-alpine'
+    container_name: elasticsearch
     ports:
-      - '9200:9200'
+      - '9200:9200'
+  kibana:
+    image: 'kibana:5'
+    container_name: kibana
+    depends_on:
+      - elasticsearch
+    environment:
+      - ELASTICSEARCH_URL=http://elasticsearch:9200
+      - SERVER_HOST=0.0.0.0
+    ports:
+      - "5601:5601"
+  app:
+    image: 'inlocomedia/kafka-elasticsearch-injector:local'
+    container_name: app
+    depends_on:
+      - kafka
+      - elasticsearch
+      - schema-registry
+    environment:
+      - KAFKA_ADDRESS=kafka:9092
+      - SCHEMA_REGISTRY_URL=http://schema-registry:8081
+      - KAFKA_TOPICS=my-topic
+      - KAFKA_CONSUMER_GROUP=my-topic-es-injector
+      - ELASTICSEARCH_HOST=http://elasticsearch:9200
+      - PROBES_PORT=5000
+      - K8S_LIVENESS_ROUTE=/liveness
+      - K8S_READINESS_ROUTE=/readiness
+      - KAFKA_CONSUMER_CONCURRENCY=10
+      - KAFKA_CONSUMER_BATCH_SIZE=10
+      - METRICS_PORT=9102
+      - LOG_LEVEL=DEBUG
+  producer:
+    image: 'inlocomedia/kafka-elasticsearch-injector:producer-local'
+    container_name: producer
+    depends_on:
+      - kafka
+      - schema-registry
+    environment:
+      - KAFKA_ADDRESS=kafka:9092
+      - SCHEMA_REGISTRY_URL=http://schema-registry:8081
+      - LOG_LEVEL=DEBUG

src/kafka/consumer.go

@@ -8,12 +8,12 @@ import (
 
 	"sync"
 
-	"github.com/inloco/kafka-elasticsearch-injector/src/models"
 	"github.com/Shopify/sarama"
 	"github.com/bsm/sarama-cluster"
 	"github.com/go-kit/kit/endpoint"
 	"github.com/go-kit/kit/log"
 	"github.com/go-kit/kit/log/level"
+	"github.com/inloco/kafka-elasticsearch-injector/src/models"
 )
 
 type Notification int32

src/kafka/consumer_test.go

@@ -12,13 +12,13 @@ import (
 
 	"encoding/json"
 
+	"github.com/Shopify/sarama"
+	"github.com/go-kit/kit/endpoint"
 	"github.com/inloco/kafka-elasticsearch-injector/src/elasticsearch"
 	"github.com/inloco/kafka-elasticsearch-injector/src/kafka/fixtures"
 	"github.com/inloco/kafka-elasticsearch-injector/src/logger_builder"
 	"github.com/inloco/kafka-elasticsearch-injector/src/models"
 	"github.com/inloco/kafka-elasticsearch-injector/src/schema_registry"
-	"github.com/Shopify/sarama"
-	"github.com/go-kit/kit/endpoint"
 	"github.com/olivere/elastic"
 	"github.com/stretchr/testify/assert"
 )
@@ -102,7 +102,7 @@ func TestKafka_Start(t *testing.T) {
 	var msg *sarama.ProducerMessage
 	if assert.NoError(t, err) {
 
-		err = producer.Publish(&rec)
+		err = producer.Publish(rec)
 		if assert.NoError(t, err) {
 			msg = <-producer.GetSuccesses()
 		} else {

src/kafka/fixtures/fixtures.go

@@ -5,8 +5,8 @@ import (
 
 	"math/rand"
 
-	"github.com/inloco/kafka-elasticsearch-injector/src/models"
 	"github.com/inloco/goavro"
+	"github.com/inloco/kafka-elasticsearch-injector/src/models"
 )
 
 const DefaultTopic = "my-topic"
@@ -33,8 +33,8 @@ func (r *FixtureRecord) ToAvroSerialization() ([]byte, error) {
 	return codec.BinaryFromNative(nil, map[string]interface{}{"id": r.Id})
 }
 
-func NewFixtureRecord() FixtureRecord {
-	return FixtureRecord{Id: rand.Int31()}
+func NewFixtureRecord() *FixtureRecord {
+	return &FixtureRecord{Id: rand.Int31()}
 }
 
 func NewRecord(ts time.Time) (*models.Record, int32, int32) {

util/producer/Dockerfile

@@ -0,0 +1,5 @@
+FROM alpine:latest
+RUN apk add --update ca-certificates
+COPY bin/producer /
+RUN chmod +x producer
+ENTRYPOINT ["/producer"]

util/producer/producer.go

@@ -0,0 +1,53 @@
+package main
+
+import (
+	"github.com/Shopify/sarama"
+	"github.com/go-kit/kit/log/level"
+	"github.com/inloco/kafka-elasticsearch-injector/src/kafka/fixtures"
+	"github.com/inloco/kafka-elasticsearch-injector/src/logger_builder"
+	"github.com/inloco/kafka-elasticsearch-injector/src/schema_registry"
+	"os"
+	"os/signal"
+	"syscall"
+	"time"
+)
+
+func main() {
+	logger := logger_builder.NewLogger("test-producer")
+	registry, err := schema_registry.NewSchemaRegistry(os.Getenv("SCHEMA_REGISTRY_URL"))
+	if err != nil {
+		panic(err)
+	}
+	config := sarama.NewConfig()
+	config.Producer.Return.Successes = true
+	config.Producer.MaxMessageBytes = 20 * 1024 * 1024 // 20mb
+	config.Producer.Flush.Frequency = 1 * time.Millisecond
+	config.Version = sarama.V0_10_0_0 // This version is the same as in production
+	producer, err := fixtures.NewProducer(os.Getenv("KAFKA_ADDRESS"), config, registry)
+	if err != nil {
+		panic(err)
+	}
+	numProducers := 1
+	for i := 0; i < numProducers; i++ {
+		go func() {
+			for {
+				rec := fixtures.NewFixtureRecord()
+				err := producer.Publish(rec)
+				if err != nil {
+					level.Error(logger).Log("msg", "error publishing to kafka", "err", err)
+					continue
+				}
+				msg := <-producer.GetSuccesses()
+				level.Debug(logger).Log("msg", "produced!", "value", rec, "msg", *msg)
+			}
+		}()
+	}
+	signals := make(chan os.Signal, 1)
+	signal.Notify(signals, syscall.SIGINT, syscall.SIGTERM)
+	for {
+		select {
+		case <-signals:
+			return
+		}
+	}
+}