Metricset/kafka consumer groups #3240
Conversation
urso
force-pushed
the
metricset/kafka-consumer-groups
branch
2 times, most recently
from
4baa87b to
10939b1
Compare
urso
force-pushed
the
metricset/kafka-consumer-groups
branch
from
378a540 to
797f695
Compare
urso
added
review
and removed
in progress
urso
force-pushed
the
metricset/kafka-consumer-groups
branch
from
797f695 to
35cdcc3
Compare
| queryGroups []string, | ||
| ) (map[string]GroupDescription, error) { | ||
| requ := &sarama.DescribeGroupsRequest{Groups: queryGroups} | ||
| resp, err := b.b.DescribeGroups(requ) |
There was a problem hiding this comment.
b.b. is not very readable ;-)
| }, | ||
| "kafka":{ | ||
| "consumergroup":{ | ||
| "example": "consumergroup" |
There was a problem hiding this comment.
We should populate this for the docs.
| description: > | ||
| Broker address | ||
|
|
||
| - name: group |
There was a problem hiding this comment.
This should be id, as it will be kafka.consumegroup.id. Alternative is kafka.conumser.group. Is the partition, topic, offset specific to a consumergroup or a consumer?
| Topics []string `config:"topics"` | ||
| } | ||
|
|
||
| var defaultConfig = metricsetConfig{ |
There was a problem hiding this comment.
We should move this to the kafka module later as I think it is the same for both metricsets.
| import ( | ||
| "crypto/tls" | ||
|
|
||
| "github.com/Shopify/sarama" |
| clientHost string | ||
| } | ||
|
|
||
| var debugf = logp.MakeDebug("kafka") |
There was a problem hiding this comment.
partitions is using kafka as well. we should clean this in a followup PR. Do we follow a debug selector naming standard in metricbeat?
| "fmt" | ||
| "math/rand" | ||
|
|
||
| "github.com/Shopify/sarama" |
| wg := sync.WaitGroup{} | ||
| results := make(chan result, len(groups)) | ||
| for _, group := range groups { | ||
| group := group |
There was a problem hiding this comment.
Yes, as group is used inside the go-routine. Copy is used to capture the group in the local scope.
Alternatively one can write:
for group := range groups {
go func(group string) {
...
}(group)
}
which is similar to writing:
for group := range groups {
group := group
go func() {
...
}()
}
|
|
||
| groupLoop: | ||
| for groupID, info := range resp { | ||
| G := groups[groupID] |
There was a problem hiding this comment.
I tend to use uppercase for local sets by convention (e.g. like N,M,L for length). Here G for group and T for topic. So not to shadow (overuse) topic and group used for correlating information. The 'sets' G and T hold results, so using a single upper-case makes them kinda stick out in this 'heavy' (potentially hard to follow) function.
| } | ||
|
|
||
| for topic, partitions := range memberDescr.Topics { | ||
| T := G[topic] |
urso
force-pushed
the
metricset/kafka-consumer-groups
branch
2 times, most recently
from
21ba3f9 to
cba815d
Compare
urso
force-pushed
the
metricset/kafka-consumer-groups
branch
from
cba815d to
032d0f5
Compare
- mark kafka module as experimental - add new metricset - includes unit tests only - fix kafka API version settings
urso
force-pushed
the
metricset/kafka-consumer-groups
branch
from
032d0f5 to
595fd0a
Compare
Requires #3239
Add kafka consumer group metricset to query kafka consumer groups via kafka 0.9 consumer group API (queries kafka, not zookeeper). Requires consumer groups to be managed by kafka.