Use Confluent and move the kafka-consumers into one consumer struct

.circleci/config.yml

@@ -16,23 +16,32 @@ jobs:
       - persist_to_workspace:
           root: .
           paths:
-            - build
             - build_docker
 
   test:
     working_directory: /go/src/github.com/grafana/metrictank
     docker:
       - image: circleci/golang:1.10
+    environment:
+      - TMP_DIR: /tmp/tmpdir
+      - PKG_CONFIG_PATH: /tmp/tmpdir/lib/pkgconfig
+      - LD_LIBRARY_PATH: /tmp/tmpdir/lib
     steps:
       - checkout
-      - run: go test -v -race $(go list ./... | grep -v github.com/grafana/metrictank/stacktest)
+      - run: scripts/build_deps.sh
+      - run: go test -v -race -tags static $(go list ./... | grep -v github.com/grafana/metrictank/stacktest)
 
   qa:
     working_directory: /go/src/github.com/grafana/metrictank
     docker:
       - image: circleci/golang:1.10
+    environment:
+      - TMP_DIR: /tmp/tmpdir
+      - PKG_CONFIG_PATH: /tmp/tmpdir/lib/pkgconfig
+      - LD_LIBRARY_PATH: /tmp/tmpdir/lib
     steps:
       - checkout
+      - run: scripts/build_deps.sh
       - run: scripts/qa/gofmt.sh
       - run: scripts/qa/go-generate.sh
       - run: scripts/qa/ineffassign.sh
@@ -43,12 +52,20 @@ jobs:
 
   qa-post-build:
     working_directory: /home/circleci/.go_workspace/src/github.com/grafana/metrictank
-    machine: true
+    machine:
+      image: true
+    environment:
+      - TMP_DIR: /tmp/tmpdir
+      - PKG_CONFIG_PATH: /tmp/tmpdir/lib/pkgconfig
+      - LD_LIBRARY_PATH: /tmp/tmpdir/lib
     steps:
       - checkout
+      - run: cd /usr/local; curl https://dl.google.com/go/go1.10.1.linux-amd64.tar.gz | sudo tar -xz 
+      - run: scripts/build_deps.sh
+      - run: scripts/build_tools.sh
+      - run: scripts/qa/docs.sh
       - attach_workspace:
           at: .
-      - run: scripts/qa/docs.sh
       - run: docker load -i build_docker/metrictank.tar
       - run: go test -v ./stacktest/tests/end2end_carbon
 
@@ -59,13 +76,13 @@ jobs:
       - checkout
       - attach_workspace:
           at: .
+      - setup_remote_docker
       - run: scripts/depends.sh
       - run: scripts/build_packages.sh
       - store_artifacts:
           path: build
       - store_artifacts:
           path: build_pkg
-      - setup_remote_docker
       - deploy:
           command: |
             if [ "${CIRCLE_BRANCH}" == "master" ]; then

cmd/mt-kafka-mdm-sniff-out-of-order/main.go

@@ -199,8 +199,6 @@ func main() {
 
 	// config may have had it disabled
 	inKafkaMdm.Enabled = true
-	// important: we don't want to share the same offset tracker as the mdm input of MT itself
-	inKafkaMdm.DataDir = "/tmp/" + instance
 
 	inKafkaMdm.ConfigProcess(instance)
 

cmd/mt-kafka-mdm-sniff/main.go

@@ -115,8 +115,6 @@ func main() {
 
 	// config may have had it disabled
 	inKafkaMdm.Enabled = true
-	// important: we don't want to share the same offset tracker as the mdm input of MT itself
-	inKafkaMdm.DataDir = "/tmp/" + instance
 
 	inKafkaMdm.ConfigProcess(instance)
 

docker/docker-chaos/docker-compose.yml

@@ -7,7 +7,6 @@ services:
     ports:
      - "6060:6060"
     volumes:
-      - ../../build/metrictank:/usr/bin/metrictank
       - ./metrictank.ini:/etc/metrictank/metrictank.ini
     environment:
       WAIT_HOSTS: kafka:9092,cassandra:9042
@@ -32,7 +31,6 @@ services:
     ports:
       - "6061:6060"
     volumes:
-      - ../../build/metrictank:/usr/bin/metrictank
       - ./metrictank.ini:/etc/metrictank/metrictank.ini
     environment:
       WAIT_HOSTS: kafka:9092,cassandra:9042,metrictank0:6060
@@ -57,7 +55,6 @@ services:
     ports:
       - "6062:6060"
     volumes:
-      - ../../build/metrictank:/usr/bin/metrictank
       - ./metrictank.ini:/etc/metrictank/metrictank.ini
     environment:
       WAIT_HOSTS: kafka:9092,cassandra:9042,metrictank0:6060
@@ -82,7 +79,6 @@ services:
     ports:
       - "6063:6060"
     volumes:
-      - ../../build/metrictank:/usr/bin/metrictank
       - ./metrictank.ini:/etc/metrictank/metrictank.ini
     environment:
       WAIT_HOSTS: kafka:9092,cassandra:9042,metrictank0:6060
@@ -107,7 +103,6 @@ services:
     ports:
       - "6064:6060"
     volumes:
-      - ../../build/metrictank:/usr/bin/metrictank
       - ./metrictank.ini:/etc/metrictank/metrictank.ini
     environment:
       WAIT_HOSTS: kafka:9092,cassandra:9042,metrictank0:6060
@@ -132,7 +127,6 @@ services:
     ports:
       - "6065:6060"
     volumes:
-      - ../../build/metrictank:/usr/bin/metrictank
       - ./metrictank.ini:/etc/metrictank/metrictank.ini
     environment:
       WAIT_HOSTS: kafka:9092,cassandra:9042,metrictank0:6060

docker/docker-chaos/metrictank.ini

@@ -178,33 +178,42 @@ partition = 0
 enabled = true
 # For incoming MetricPoint messages without org-id, assume this org id
 org-id = 0
+# save interval for consumption lag
+lag-collection-interval = 5s
+# Number of kafka client threads
+consumer-threads = 8
+# Size of the channel that enqueues incoming events
+events-channel-size = 50
+# Delay in milliseconds to wait for messages in the producer queue to accumulate before constructing message batches (MessageSets) to transmit to brokers
+metrics-buffer-max = 100ms
+# The number of metrics to buffer in internal and external channels
+channel-buffer-size = 1000
+# The minimum number of message bytes to fetch in a request
+fetch-min = 1
+# Initial maximum number of bytes per topic+partition to request when fetching messages from the broker
+fetch-message-max = 32768
+# The maximum amount of time the broker will wait for Consumer.Fetch.Min bytes to become available before it
+max-wait = 1s
+# Time to wait between attempts to fetch metadata
+metadata-backoff-time = 500ms
+# Number of retries to fetch metadata in case of failure
+metadata-retries = 5
+# Maximum time to wait for the broker to reply to metadata queries
+metadata-timeout = 10s
+# How many outstanding requests a connection is allowed to have before sending on it blocks
+net-max-open-requests = 1000000
+# Client group session and failure detection timeout
+session-timeout = 10s
 # tcp address (may be given multiple times as a comma-separated list)
 brokers = kafka:9092
 # kafka topic (may be given multiple times as a comma-separated list)
 topics = mdm
-# offset to start consuming from. Can be one of newest, oldest,last or a time duration
+# offset to start consuming from. Can be one of newest, oldest or a time duration
 # When using a duration but the offset request fails (e.g. Kafka doesn't have data so far back), metrictank falls back to `oldest`.
 # the further back in time you go, the more old data you can load into metrictank, but the longer it takes to catch up to realtime data
-offset = last
+offset = oldest
 # kafka partitions to consume. use '*' or a comma separated list of id's
 partitions = *
-# save interval for offsets
-offset-commit-interval = 5s
-# directory to store partition offsets index. supports relative or absolute paths. empty means working dir.
-# it will be created (incl parent dirs) if not existing.
-data-dir = /var/lib/metrictank
-# The number of metrics to buffer in internal and external channels
-channel-buffer-size = 1000
-# The minimum number of message bytes to fetch in a request
-consumer-fetch-min = 1
-# The default number of message bytes to fetch in a request
-consumer-fetch-default = 32768
-# The maximum amount of time the broker will wait for Consumer.Fetch.Min bytes to become available before it
-consumer-max-wait-time = 1s
-#The maximum amount of time the consumer expects a message takes to process
-consumer-max-processing-time = 1s
-# How many outstanding requests a connection is allowed to have before sending on it blocks
-net-max-open-requests = 100
 
 ## basic clustering settings ##
 [cluster]
@@ -281,17 +290,36 @@ topic = metricpersist
 partitions = *
 # method used for partitioning metrics. This should match the settings of tsdb-gw. (byOrg|bySeries)
 partition-scheme = bySeries
-# offset to start consuming from. Can be one of newest, oldest,last or a time duration
+# offset to start consuming from. Can be one of newest, oldest or a time duration
 # When using a duration but the offset request fails (e.g. Kafka doesn't have data so far back), metrictank falls back to `oldest`.
 # Should match your kafka-mdm-in setting
-offset = last
-# save interval for offsets
-offset-commit-interval = 5s
+offset = oldest
+# save interval for consumption lag
+lag-collection-interval = 5s
+# Maximum number of messages batched in one MessageSet
+batch-num-messages = 100
+# Number of kafka client threads
+consumer-threads = 1
+# Size of the channel that enqueues incoming events
+events-channel-size = 5
 # Maximum time backlog processing can block during metrictank startup.
 backlog-process-timeout = 60s
-# directory to store partition offsets index. supports relative or absolute paths. empty means working dir.
-# it will be created (incl parent dirs) if not existing.
-data-dir = /var/lib/metrictank
+# The minimum number of message bytes to fetch in a request
+fetch-min = 1
+# Initial maximum number of bytes per topic+partition to request when fetching messages from the broker
+fetch-message-max = 32768
+# The maximum amount of time the broker will wait for Consumer.Fetch.Min bytes to become available before it
+max-wait = 1s
+# Time to wait between attempts to fetch metadata
+metadata-backoff-time = 500ms
+# Number of retries to fetch metadata in case of failure
+metadata-retries = 5
+# Maximum time to wait for the broker to reply to metadata queries
+metadata-timeout = 10s
+# How many outstanding requests a connection is allowed to have before sending on it blocks
+net-max-open-requests = 1000000
+# Client group session and failure detection timeout
+session-timeout = 10s
 
 ### nsq as transport for clustering messages
 [nsq-cluster]

docker/docker-cluster/docker-compose.yml

@@ -7,7 +7,6 @@ services:
     expose:
      - 6060
     volumes:
-      - ../../build/metrictank:/usr/bin/metrictank
       - ./metrictank.ini:/etc/metrictank/metrictank.ini
       - ./storage-schemas.conf:/etc/metrictank/storage-schemas.conf
       - ./storage-aggregation.conf:/etc/metrictank/storage-aggregation.conf
@@ -32,7 +31,6 @@ services:
     expose:
      - 6060
     volumes:
-      - ../../build/metrictank:/usr/bin/metrictank
       - ./metrictank.ini:/etc/metrictank/metrictank.ini
       - ./storage-schemas.conf:/etc/metrictank/storage-schemas.conf
       - ./storage-aggregation.conf:/etc/metrictank/storage-aggregation.conf
@@ -57,7 +55,6 @@ services:
     expose:
      - 6060
     volumes:
-      - ../../build/metrictank:/usr/bin/metrictank
       - ./metrictank.ini:/etc/metrictank/metrictank.ini
       - ./storage-schemas.conf:/etc/metrictank/storage-schemas.conf
       - ./storage-aggregation.conf:/etc/metrictank/storage-aggregation.conf
@@ -82,7 +79,6 @@ services:
     expose:
      - 6060
     volumes:
-      - ../../build/metrictank:/usr/bin/metrictank
       - ./metrictank.ini:/etc/metrictank/metrictank.ini
       - ./storage-schemas.conf:/etc/metrictank/storage-schemas.conf
       - ./storage-aggregation.conf:/etc/metrictank/storage-aggregation.conf