remove consumer prefix from parameters and update example configs

Author: replay

docker/docker-chaos/metrictank.ini

@@ -178,6 +178,28 @@ 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
+# Maximum number of messages batched in one MessageSet
+batch-num-messages = 10000
+# 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
+# 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 = 100
+# 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)
@@ -188,21 +210,6 @@ topics = mdm
 offset = oldest
 # kafka partitions to consume. use '*' or a comma separated list of id's
 partitions = *
-# save interval for consumption lag
-lag-collection-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 = 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]
@@ -285,11 +292,28 @@ partition-scheme = bySeries
 offset = oldest
 # save interval for consumption lag
 lag-collection-interval = 5s
+# Maximum number of messages batched in one MessageSet
+batch-num-messages = 10000
+# 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
 # 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
+# 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 = 100
+# Client group session and failure detection timeout
+session-timeout = 10s
 
 ### nsq as transport for clustering messages
 [nsq-cluster]

docker/docker-cluster/metrictank.ini

@@ -178,30 +178,38 @@ 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
+# Maximum number of messages batched in one MessageSet
+batch-num-messages = 10000
+# 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
+# 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 = 100
+# 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 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 = oldest
 # kafka partitions to consume. use '*' or a comma separated list of id's
 partitions = *
-# save interval for consumption lag
-lag-collection-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 = 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]
@@ -279,14 +287,33 @@ 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 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 = oldest
 # save interval for consumption lag
 lag-collection-interval = 5s
+# Maximum number of messages batched in one MessageSet
+batch-num-messages = 10000
+# 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
 # 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
+# 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 = 100
+# Client group session and failure detection timeout
+session-timeout = 10s
 
 ### nsq as transport for clustering messages
 [nsq-cluster]

docker/docker-dev-custom-cfg-kafka/metrictank.ini

@@ -178,6 +178,28 @@ partition = 0
 enabled = true
 # For incoming MetricPoint messages without org-id, assume this org id
 org-id = 1
+# save interval for consumption lag
+lag-collection-interval = 5s
+# Maximum number of messages batched in one MessageSet
+batch-num-messages = 10000
+# 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
+# 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 = 100
+# 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)
@@ -188,21 +210,6 @@ topics = mdm
 offset = oldest
 # kafka partitions to consume. use '*' or a comma separated list of id's
 partitions = *
-# save interval for consumption lag
-lag-collection-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 = 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]
@@ -285,11 +292,28 @@ partition-scheme = bySeries
 offset = oldest
 # save interval for consumption lag
 lag-collection-interval = 5s
+# Maximum number of messages batched in one MessageSet
+batch-num-messages = 10000
+# 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
 # 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
+# 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 = 100
+# Client group session and failure detection timeout
+session-timeout = 10s
 
 ### nsq as transport for clustering messages
 [nsq-cluster]

docs/config.md

@@ -226,6 +226,28 @@ partition = 0
 enabled = false
 # For incoming MetricPoint messages without org-id, assume this org id
 org-id = 0
+# save interval for consumption lag
+lag-collection-interval = 5s
+# Maximum number of messages batched in one MessageSet
+batch-num-messages = 10000
+# 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
+# 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 = 100
+# 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)
@@ -236,18 +258,6 @@ topics = mdm
 offset = oldest
 # kafka partitions to consume. use '*' or a comma separated list of id's
 partitions = *
-# save interval for consumption lag
-lag-collection-interval = 5s
-# 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 = 1s
-# How many outstanding requests a connection is allowed to have before sending on it blocks
-net-max-open-requests = 100
 ```
 
 ## basic clustering settings ##
@@ -339,8 +349,28 @@ partition-scheme = bySeries
 offset = oldest
 # save interval for consumption lag
 lag-collection-interval = 5s
+# Maximum number of messages batched in one MessageSet
+batch-num-messages = 10000
+# 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
 # Maximum time backlog processing can block during metrictank startup.
 backlog-process-timeout = 60s
+# The minimum number of message bytes to fetch in a request
+fetch-min = 1
+# 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 = 100
+# Client group session and failure detection timeout
+session-timeout = 10s
 ```
 
 ### nsq as transport for clustering messages

input/kafkamdm/kafkamdm.go

@@ -52,13 +52,13 @@ func ConfigSetup() {
 	inKafkaMdm.IntVar(&clientConf.BatchNumMessages, "batch-num-messages", 10000, "Maximum number of messages batched in one MessageSet")
 	inKafkaMdm.DurationVar(&clientConf.BufferMax, "metrics-buffer-max", time.Millisecond*100, "Delay in milliseconds to wait for messages in the producer queue to accumulate before constructing message batches (MessageSets) to transmit to brokers")
 	inKafkaMdm.IntVar(&clientConf.ChannelBufferSize, "channel-buffer-size", 1000, "Maximum number of messages allowed on the producer queue")
-	inKafkaMdm.IntVar(&clientConf.FetchMin, "consumer-fetch-min", 1, "Minimum number of bytes the broker responds with. If fetch.wait.max.ms expires the accumulated data will be sent to the client regardless of this setting")
-	inKafkaMdm.DurationVar(&clientConf.MaxWait, "consumer-max-wait", time.Millisecond*100, "Maximum time the broker may wait to fill the response with fetch.min.bytes")
+	inKafkaMdm.IntVar(&clientConf.FetchMin, "fetch-min", 1, "Minimum number of bytes the broker responds with. If fetch.wait.max.ms expires the accumulated data will be sent to the client regardless of this setting")
+	inKafkaMdm.DurationVar(&clientConf.MaxWait, "max-wait", time.Millisecond*100, "Maximum time the broker may wait to fill the response with fetch.min.bytes")
 	inKafkaMdm.DurationVar(&clientConf.MetadataBackoffTime, "metadata-backoff-time", time.Millisecond*500, "Time to wait between attempts to fetch metadata")
 	inKafkaMdm.IntVar(&clientConf.MetadataRetries, "metadata-retries", 5, "Number of retries to fetch metadata in case of failure")
-	inKafkaMdm.DurationVar(&clientConf.MetadataTimeout, "consumer-metadata-timeout", time.Second*10, "Maximum time to wait for the broker to reply to metadata queries")
+	inKafkaMdm.DurationVar(&clientConf.MetadataTimeout, "metadata-timeout", time.Second*10, "Maximum time to wait for the broker to reply to metadata queries")
 	inKafkaMdm.IntVar(&clientConf.NetMaxOpenRequests, "net-max-open-requests", 100, "Maximum number of in-flight requests per broker connection. This is a generic property applied to all broker communication, however it is primarily relevant to produce requests.")
-	inKafkaMdm.DurationVar(&clientConf.SessionTimeout, "consumer-session-timeout", time.Second*30, "Client group session and failure detection timeout")
+	inKafkaMdm.DurationVar(&clientConf.SessionTimeout, "session-timeout", time.Second*30, "Client group session and failure detection timeout")
 	inKafkaMdm.StringVar(&clientConf.Broker, "brokers", "kafka:9092", "tcp address for kafka (may be be given multiple times as a comma-separated list)")
 	inKafkaMdm.StringVar(&clientConf.StartAtOffset, "offset", "oldest", "Set the offset to start consuming from. Can be one of newest, oldest or a time duration")
 	inKafkaMdm.StringVar(&clientConf.Partitions, "partitions", "*", "kafka partitions to consume. use '*' or a comma separated list of id's")
@@ -76,7 +76,7 @@ func ConfigProcess(instance string) {
 	}
 
 	if clientConf.MaxWait == 0 {
-		log.Fatal(4, "kafkamdm: consumer-max-wait-time must be greater then 0")
+		log.Fatal(4, "kafkamdm: max-wait-time must be greater then 0")
 	}
 
 	if !clientConf.OffsetIsValid() {