add tool for replicating metrics to second kafka cluster #435

woodsaj · 2016-12-23T16:20:45Z

No description provided.

Dieterbe · 2017-01-03T14:43:46Z

scripts/build_tools.sh

@@ -17,6 +17,6 @@ export CGO_ENABLED=0
 # Build binary
 cd $GOPATH/src/github.com/raintank/metrictank/cmd
 for tool in *; do
-  cd $tool
+  cd $GOPATH/src/github.com/raintank/metrictank/cmd/$tool


do we still want this change? b89b20e should fix the same problem. we can probably just remove this commit

Dieterbe · 2017-01-03T14:53:34Z

cmd/mt-replicator/main.go

+	showVersion = flag.Bool("version", false, "print version string")
+	logLevel    = flag.Int("log-level", 2, "log level. 0=TRACE|1=DEBUG|2=INFO|3=WARN|4=ERROR|5=CRITICAL|6=FATAL")
+
+	partitionBy  = flag.String("partition-by", "byOrg", "method used for paritioning metrics. (byOrg|bySeries)")


same thought here as https://github.com/raintank/tsdb-gw/pull/20#pullrequestreview-14312450

Dieterbe · 2017-01-03T15:02:23Z

Since we decided not to use sarama-cluster (rather use sarama directly) for input plugins so that we had full control over our offset (#236 ), why are we using it here?

also it's interesting that we decode all data in the consumer, pass the structs to the producer, which encodes them again. we could probably just pass the binary data directly? but i guess decoding might be useful if we want to print what's going on, or something.

woodsaj · 2017-01-03T18:26:27Z

we use sarama-cluster because unlike MT, this tool is a typical consumer.
- We want to only consume messages once.
- we want to support running multiple instances of the tool allowing for HA/scalability.
we have to decode data so we can read the metric Name or OrgId to be able to generate the new partition key

Dieterbe · 2017-01-03T19:23:40Z

cmd/mt-replicator/consume.go

+	config.Group.Return.Notifications = true
+	config.ChannelBufferSize = 1000
+	config.Consumer.Fetch.Min = int32(1)
+	config.Consumer.Fetch.Default = int32(32768)


I don't think you need to specify the type explicitly. the compiler should be able to deduce it from the type of the attribute you're assigning to.

https://play.golang.org/p/r1VsIXSsB6

Dieterbe · 2017-01-03T19:25:49Z

cmd/mt-replicator/consume.go

+				log.Debug("flushing metricData buffer to kafka.")
+				complete := false
+				for !complete {
+					if err = publisher.Send(buf); err != nil {


at our volume levels, it's probably not really a concern, so I'm not asking we make a change at this point, but if we decoupled publisher and consumer more they could work concurrently and faster.

This was a deliberate choice to reduce complexity. Performance can easily be scaled by just running multiple mt-replicators.

Dieterbe · 2017-01-03T19:31:22Z

cmd/mt-replicator/consume.go

-			buf = buf[:0]
-			c.consumer.MarkPartitionOffset(m.Topic, m.Partition, m.Offset, "")
+		case <-ticker.C:
+			log.Info("%d metrics procesed in last 10seconds.", counter)


I think it's very possible that the prior case will keep the select case rather busy (especially when it hits a failure and sleeps a bit), in which case this ticker can silently drop ticks. This is not a big issue in this case, except we shouldn't claim it's in last 10seconds. it might very well be in the last 20s or 30s. so i would just remove that part, we can just use the timestamps of the messages for guidance. (or actually measure how long it took since last time)

just tracking the time between ticker.C channel reads and using that instead of printing "10seconds" is the best option I think

Dieterbe · 2017-01-03T19:39:02Z

cmd/mt-replicator/publish.go

+	payload := make([]*sarama.ProducerMessage, len(metrics))
+
+	for i, metric := range metrics {
+		data, err := metric.MarshalMsg(data[:])


wouldn't this keep accumulating data, because every time it appends the msgp bytes for a metric, it does so after the msgp bytes of prior metrics?

yep, that should be data[:0]

Dieterbe · 2017-01-03T19:40:47Z

scripts/build_tools.sh

@@ -19,4 +19,5 @@ cd $GOPATH/src/github.com/raintank/metrictank/cmd
 for tool in *; do
  cd $tool
  go build -ldflags "-X main.GitHash=$GITVERSION" -o $BUILDDIR/$tool
+  cd ..


FWIW you can just rebase this branch on top of master and then you don't need this commit anymore.

Dieterbe · 2017-01-04T21:09:05Z

cmd/mt-replicator/publish.go

+			Key:   sarama.ByteEncoder(key),
+			Topic: p.topic,
+			Value: sarama.ByteEncoder(data),
+		}


I think if we do this we'll hit the same problem as https://github.com/raintank/tsdb-gw/issues/3 because a ProducerMessage's Value will point to an array that will be overwritten by subsequent loop iterations. (sarama.ByteEncoder doesn't copy or encode anything when called. it's just an alias for []byte which it will "encode" (return) later)

yeah, i should just stop trying to be clever with this and just accept the allocations. We can use a bufferPool later if we need the performance boost.

This tool will read from one kafka cluster and write to another. The source and destination kafka topics can have different names and different number of partitions.

woodsaj · 2017-01-10T18:00:47Z

@Dieterbe anything preventing this from being merged?

Dieterbe · 2017-01-10T18:05:54Z

yes #435 (comment)

woodsaj · 2017-01-10T18:20:30Z

that comment wasnt showing in github for me. must be some caching issue.

fixed.

Dieterbe · 2017-01-10T19:00:09Z

sweet, thanks @woodsaj !

Dieterbe self-requested a review December 23, 2016 18:27

Dieterbe self-assigned this Dec 23, 2016

Dieterbe added this to the hosted-metrics-alpha milestone Jan 1, 2017

Dieterbe reviewed Jan 3, 2017

View reviewed changes

woodsaj force-pushed the replicator branch 2 times, most recently from 9cad750 to f8fa86e Compare January 3, 2017 18:48

Dieterbe reviewed Jan 3, 2017

View reviewed changes

woodsaj force-pushed the replicator branch from f8fa86e to 52cf529 Compare January 4, 2017 06:54

Dieterbe reviewed Jan 4, 2017

View reviewed changes

Dieterbe removed their assignment Jan 10, 2017

woodsaj added 4 commits January 11, 2017 01:59

add mt-replicator tool

766e386

This tool will read from one kafka cluster and write to another. The source and destination kafka topics can have different names and different number of partitions.

periodically print number of metrics processed.

f6b682e

rename paritionBy to partitionScheme

3b4cb2c

dont try and reuse byte slice when publishing

a3eabf8

woodsaj force-pushed the replicator branch from b65b116 to a3eabf8 Compare January 10, 2017 17:59

use correct duration between printing processed count

bf9b6a4

Dieterbe merged commit b6525a6 into master Jan 10, 2017

woodsaj deleted the replicator branch January 10, 2017 19:02

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

add tool for replicating metrics to second kafka cluster #435

add tool for replicating metrics to second kafka cluster #435

woodsaj commented Dec 23, 2016

Dieterbe Jan 3, 2017

Dieterbe Jan 3, 2017

Dieterbe commented Jan 3, 2017

woodsaj commented Jan 3, 2017

Dieterbe Jan 3, 2017

Dieterbe Jan 3, 2017 •

edited

Loading

woodsaj Jan 4, 2017

Dieterbe Jan 3, 2017

Dieterbe Jan 10, 2017 •

edited

Loading

Dieterbe Jan 3, 2017

woodsaj Jan 4, 2017

Dieterbe Jan 3, 2017

Dieterbe Jan 4, 2017

woodsaj Jan 6, 2017

woodsaj commented Jan 10, 2017 •

edited

Loading

Dieterbe commented Jan 10, 2017

woodsaj commented Jan 10, 2017

Dieterbe commented Jan 10, 2017

add tool for replicating metrics to second kafka cluster #435

add tool for replicating metrics to second kafka cluster #435

Conversation

woodsaj commented Dec 23, 2016

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Dieterbe commented Jan 3, 2017

woodsaj commented Jan 3, 2017

Choose a reason for hiding this comment

Dieterbe Jan 3, 2017 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Dieterbe Jan 10, 2017 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

woodsaj commented Jan 10, 2017 • edited Loading

Dieterbe commented Jan 10, 2017

woodsaj commented Jan 10, 2017

Dieterbe commented Jan 10, 2017

Dieterbe Jan 3, 2017 •

edited

Loading

Dieterbe Jan 10, 2017 •

edited

Loading

woodsaj commented Jan 10, 2017 •

edited

Loading