Skip to content

Commit ece489e

Browse files
baibai
authored
Merge pull request #628 from mwain/prometheus
Add Prometheus Metrics Exporter
2 parents bc09a91 + 3c918db commit ece489e

5 files changed

Lines changed: 348 additions & 0 deletions

File tree

‎core/internal/httpserver/coordinator.go‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -129,6 +129,8 @@ func (hc *Coordinator) Configure() {
129129
// This is a healthcheck URL. Please don't change it
130130
hc.router.GET("/burrow/admin", hc.handleAdmin)
131131

132+
hc.router.Handler(http.MethodGet, "/metrics", hc.handlePrometheusMetrics())
133+
132134
// All valid paths go here
133135
hc.router.GET("/v3/kafka", hc.handleClusterList)
134136
hc.router.GET("/v3/kafka/:cluster", hc.handleClusterDetail)

‎core/internal/httpserver/prometheus.go‎

Lines changed: 182 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,182 @@
1+
package httpserver
2+
3+
import (
4+
"net/http"
5+
"strconv"
6+
7+
"github.com/prometheus/client_golang/prometheus"
8+
9+
"github.com/linkedin/Burrow/core/protocol"
10+
11+
"github.com/prometheus/client_golang/prometheus/promauto"
12+
"github.com/prometheus/client_golang/prometheus/promhttp"
13+
)
14+
15+
var (
16+
consumerTotalLagGauge = promauto.NewGaugeVec(
17+
prometheus.GaugeOpts{
18+
Name: "burrow_kafka_consumer_lag_total",
19+
Help: "The sum of all partition current lag values for the group",
20+
},
21+
[]string{"cluster", "consumer_group"},
22+
)
23+
24+
consumerStatusGauge = promauto.NewGaugeVec(
25+
prometheus.GaugeOpts{
26+
Name: "burrow_kafka_consumer_status",
27+
Help: "The status of the consumer group. It is calculated from the highest status for the individual partitions. Statuses are an index list from NOTFOUND, OK, WARN, ERR, STOP, STALL, REWIND",
28+
},
29+
[]string{"cluster", "consumer_group"},
30+
)
31+
32+
consumerPartitionCurrentOffset = promauto.NewGaugeVec(
33+
prometheus.GaugeOpts{
34+
Name: "burrow_kafka_consumer_current_offset",
35+
Help: "Latest offset that Burrow is storing for this partition",
36+
},
37+
[]string{"cluster", "consumer_group", "topic", "partition"},
38+
)
39+
40+
consumerPartitionLagGauge = promauto.NewGaugeVec(
41+
prometheus.GaugeOpts{
42+
Name: "burrow_kafka_consumer_partition_lag",
43+
Help: "Number of messages the consumer group is behind by for a partition as reported by Burrow",
44+
},
45+
[]string{"cluster", "consumer_group", "topic", "partition"},
46+
)
47+
48+
topicPartitionOffsetGauge = promauto.NewGaugeVec(
49+
prometheus.GaugeOpts{
50+
Name: "burrow_kafka_topic_partition_offset",
51+
Help: "Latest offset the topic that Burrow is storing for this partition",
52+
},
53+
[]string{"cluster", "topic", "partition"},
54+
)
55+
)
56+
57+
func (hc *Coordinator) handlePrometheusMetrics() http.HandlerFunc {
58+
promHandler := promhttp.Handler()
59+
60+
return http.HandlerFunc(func(resp http.ResponseWriter, req *http.Request) {
61+
for _, cluster := range listClusters(hc.App) {
62+
for _, consumer := range listConsumers(hc.App, cluster) {
63+
consumerStatus := getFullConsumerStatus(hc.App, cluster, consumer)
64+
65+
if consumerStatus == nil ||
66+
consumerStatus.Status == protocol.StatusNotFound ||
67+
consumerStatus.Complete < 1.0 {
68+
continue
69+
}
70+
71+
labels := map[string]string{
72+
"cluster": cluster,
73+
"consumer_group": consumer,
74+
}
75+
76+
consumerTotalLagGauge.With(labels).Set(float64(consumerStatus.TotalLag))
77+
consumerStatusGauge.With(labels).Set(float64(consumerStatus.Status))
78+
79+
for _, partition := range consumerStatus.Partitions {
80+
if partition.Complete < 1.0 {
81+
continue
82+
}
83+
84+
labels := map[string]string{
85+
"cluster": cluster,
86+
"consumer_group": consumer,
87+
"topic": partition.Topic,
88+
"partition": strconv.FormatInt(int64(partition.Partition), 10),
89+
}
90+
91+
consumerPartitionCurrentOffset.With(labels).Set(float64(partition.End.Offset))
92+
consumerPartitionLagGauge.With(labels).Set(float64(partition.CurrentLag))
93+
}
94+
}
95+
96+
// Topics
97+
for _, topic := range listTopics(hc.App, cluster) {
98+
for partitionNumber, offset := range getTopicDetail(hc.App, cluster, topic) {
99+
topicPartitionOffsetGauge.With(map[string]string{
100+
"cluster": cluster,
101+
"topic": topic,
102+
"partition": strconv.FormatInt(int64(partitionNumber), 10),
103+
}).Set(float64(offset))
104+
}
105+
}
106+
}
107+
108+
promHandler.ServeHTTP(resp, req)
109+
})
110+
}
111+
112+
func listClusters(app *protocol.ApplicationContext) []string {
113+
request := &protocol.StorageRequest{
114+
RequestType: protocol.StorageFetchClusters,
115+
Reply: make(chan interface{}),
116+
}
117+
app.StorageChannel <- request
118+
response := <-request.Reply
119+
if response == nil {
120+
return []string{}
121+
}
122+
123+
return response.([]string)
124+
}
125+
126+
func listConsumers(app *protocol.ApplicationContext, cluster string) []string {
127+
request := &protocol.StorageRequest{
128+
RequestType: protocol.StorageFetchConsumers,
129+
Cluster: cluster,
130+
Reply: make(chan interface{}),
131+
}
132+
app.StorageChannel <- request
133+
response := <-request.Reply
134+
if response == nil {
135+
return []string{}
136+
}
137+
138+
return response.([]string)
139+
}
140+
141+
func getFullConsumerStatus(app *protocol.ApplicationContext, cluster, consumer string) *protocol.ConsumerGroupStatus {
142+
request := &protocol.EvaluatorRequest{
143+
Cluster: cluster,
144+
Group: consumer,
145+
ShowAll: true,
146+
Reply: make(chan *protocol.ConsumerGroupStatus),
147+
}
148+
app.EvaluatorChannel <- request
149+
response := <-request.Reply
150+
return response
151+
}
152+
153+
func listTopics(app *protocol.ApplicationContext, cluster string) []string {
154+
request := &protocol.StorageRequest{
155+
RequestType: protocol.StorageFetchTopics,
156+
Cluster: cluster,
157+
Reply: make(chan interface{}),
158+
}
159+
app.StorageChannel <- request
160+
response := <-request.Reply
161+
if response == nil {
162+
return []string{}
163+
}
164+
165+
return response.([]string)
166+
}
167+
168+
func getTopicDetail(app *protocol.ApplicationContext, cluster, topic string) []int64 {
169+
request := &protocol.StorageRequest{
170+
RequestType: protocol.StorageFetchTopic,
171+
Cluster: cluster,
172+
Topic: topic,
173+
Reply: make(chan interface{}),
174+
}
175+
app.StorageChannel <- request
176+
response := <-request.Reply
177+
if response == nil {
178+
return []int64{}
179+
}
180+
181+
return response.([]int64)
182+
}

‎core/internal/httpserver/prometheus_test.go‎

Lines changed: 156 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,156 @@
1+
package httpserver
2+
3+
import (
4+
"net/http"
5+
"net/http/httptest"
6+
"testing"
7+
8+
"github.com/stretchr/testify/assert"
9+
10+
"github.com/linkedin/Burrow/core/protocol"
11+
)
12+
13+
func TestHttpServer_handlePrometheusMetrics(t *testing.T) {
14+
coordinator := fixtureConfiguredCoordinator()
15+
16+
// Respond to the expected storage requests
17+
go func() {
18+
request := <-coordinator.App.StorageChannel
19+
assert.Equalf(t, protocol.StorageFetchClusters, request.RequestType, "Expected request of type StorageFetchClusters, not %v", request.RequestType)
20+
request.Reply <- []string{"testcluster"}
21+
close(request.Reply)
22+
23+
// List of consumers
24+
request = <-coordinator.App.StorageChannel
25+
assert.Equalf(t, protocol.StorageFetchConsumers, request.RequestType, "Expected request of type StorageFetchConsumers, not %v", request.RequestType)
26+
assert.Equalf(t, "testcluster", request.Cluster, "Expected request Cluster to be testcluster, not %v", request.Cluster)
27+
request.Reply <- []string{"testgroup", "testgroup2"}
28+
close(request.Reply)
29+
30+
// List of topics
31+
request = <-coordinator.App.StorageChannel
32+
assert.Equalf(t, protocol.StorageFetchTopics, request.RequestType, "Expected request of type StorageFetchTopics, not %v", request.RequestType)
33+
assert.Equalf(t, "testcluster", request.Cluster, "Expected request Cluster to be testcluster, not %v", request.Cluster)
34+
request.Reply <- []string{"testtopic", "testtopic1"}
35+
close(request.Reply)
36+
37+
// Topic details
38+
request = <-coordinator.App.StorageChannel
39+
assert.Equalf(t, protocol.StorageFetchTopic, request.RequestType, "Expected request of type StorageFetchTopic, not %v", request.RequestType)
40+
assert.Equalf(t, "testcluster", request.Cluster, "Expected request Cluster to be testcluster, not %v", request.Cluster)
41+
assert.Equalf(t, "testtopic", request.Topic, "Expected request Topic to be testtopic, not %v", request.Topic)
42+
request.Reply <- []int64{6556, 5566}
43+
close(request.Reply)
44+
45+
request = <-coordinator.App.StorageChannel
46+
assert.Equalf(t, protocol.StorageFetchTopic, request.RequestType, "Expected request of type StorageFetchTopic, not %v", request.RequestType)
47+
assert.Equalf(t, "testcluster", request.Cluster, "Expected request Cluster to be testcluster, not %v", request.Cluster)
48+
assert.Equalf(t, "testtopic1", request.Topic, "Expected request Topic to be testtopic, not %v", request.Topic)
49+
request.Reply <- []int64{54}
50+
close(request.Reply)
51+
}()
52+
53+
// Respond to the expected evaluator requests
54+
go func() {
55+
// testgroup happy paths
56+
request := <-coordinator.App.EvaluatorChannel
57+
assert.Equalf(t, "testcluster", request.Cluster, "Expected request Cluster to be testcluster, not %v", request.Cluster)
58+
assert.Equalf(t, "testgroup", request.Group, "Expected request Group to be testgroup, not %v", request.Group)
59+
assert.True(t, request.ShowAll, "Expected request ShowAll to be True")
60+
response := &protocol.ConsumerGroupStatus{
61+
Cluster: request.Cluster,
62+
Group: request.Group,
63+
Status: protocol.StatusOK,
64+
Complete: 1.0,
65+
Partitions: []*protocol.PartitionStatus{
66+
{
67+
Topic: "testtopic",
68+
Partition: 0,
69+
Status: protocol.StatusOK,
70+
CurrentLag: 100,
71+
Complete: 1.0,
72+
End: &protocol.ConsumerOffset{
73+
Offset: 22663,
74+
},
75+
},
76+
{
77+
Topic: "testtopic",
78+
Partition: 1,
79+
Status: protocol.StatusOK,
80+
CurrentLag: 10,
81+
Complete: 1.0,
82+
End: &protocol.ConsumerOffset{
83+
Offset: 2488,
84+
},
85+
},
86+
{
87+
Topic: "testtopic1",
88+
Partition: 0,
89+
Status: protocol.StatusOK,
90+
CurrentLag: 50,
91+
Complete: 1.0,
92+
End: &protocol.ConsumerOffset{
93+
Offset: 99888,
94+
},
95+
},
96+
{
97+
Topic: "incomplete",
98+
Partition: 0,
99+
Status: protocol.StatusOK,
100+
CurrentLag: 0,
101+
Complete: 0.2,
102+
End: &protocol.ConsumerOffset{
103+
Offset: 5335,
104+
},
105+
},
106+
},
107+
TotalPartitions: 2134,
108+
Maxlag: &protocol.PartitionStatus{},
109+
TotalLag: 2345,
110+
}
111+
request.Reply <- response
112+
close(request.Reply)
113+
114+
// testgroup2 not found
115+
request = <-coordinator.App.EvaluatorChannel
116+
assert.Equalf(t, "testcluster", request.Cluster, "Expected request Cluster to be testcluster, not %v", request.Cluster)
117+
assert.Equalf(t, "testgroup2", request.Group, "Expected request Group to be testgroup, not %v", request.Group)
118+
assert.True(t, request.ShowAll, "Expected request ShowAll to be True")
119+
response = &protocol.ConsumerGroupStatus{
120+
Cluster: request.Cluster,
121+
Group: request.Group,
122+
Status: protocol.StatusNotFound,
123+
}
124+
request.Reply <- response
125+
close(request.Reply)
126+
}()
127+
128+
// Set up a request
129+
req, err := http.NewRequest("GET", "/metrics", nil)
130+
assert.NoError(t, err, "Expected request setup to return no error")
131+
132+
// Call the handler via httprouter
133+
rr := httptest.NewRecorder()
134+
coordinator.router.ServeHTTP(rr, req)
135+
136+
assert.Equalf(t, http.StatusOK, rr.Code, "Expected response code to be 200, not %v", rr.Code)
137+
138+
promExp := rr.Body.String()
139+
assert.Contains(t, promExp, `burrow_kafka_consumer_status{cluster="testcluster",consumer_group="testgroup"} 1`)
140+
assert.Contains(t, promExp, `burrow_kafka_consumer_lag_total{cluster="testcluster",consumer_group="testgroup"} 2345`)
141+
142+
assert.Contains(t, promExp, `burrow_kafka_consumer_partition_lag{cluster="testcluster",consumer_group="testgroup",partition="0",topic="testtopic"} 100`)
143+
assert.Contains(t, promExp, `burrow_kafka_consumer_partition_lag{cluster="testcluster",consumer_group="testgroup",partition="1",topic="testtopic"} 10`)
144+
assert.Contains(t, promExp, `burrow_kafka_consumer_partition_lag{cluster="testcluster",consumer_group="testgroup",partition="0",topic="testtopic1"} 50`)
145+
146+
assert.Contains(t, promExp, `burrow_kafka_consumer_current_offset{cluster="testcluster",consumer_group="testgroup",partition="0",topic="testtopic"} 22663`)
147+
assert.Contains(t, promExp, `burrow_kafka_consumer_current_offset{cluster="testcluster",consumer_group="testgroup",partition="1",topic="testtopic"} 2488`)
148+
assert.Contains(t, promExp, `burrow_kafka_consumer_current_offset{cluster="testcluster",consumer_group="testgroup",partition="0",topic="testtopic1"} 99888`)
149+
150+
assert.Contains(t, promExp, `burrow_kafka_topic_partition_offset{cluster="testcluster",partition="0",topic="testtopic"} 6556`)
151+
assert.Contains(t, promExp, `burrow_kafka_topic_partition_offset{cluster="testcluster",partition="1",topic="testtopic"} 5566`)
152+
assert.Contains(t, promExp, `burrow_kafka_topic_partition_offset{cluster="testcluster",partition="0",topic="testtopic1"} 54`)
153+
154+
assert.NotContains(t, promExp, `burrow_kafka_consumer_partition_lag{cluster="testcluster",consumer_group="testgroup",partition="0",topic="incomplete"} 0`)
155+
assert.NotContains(t, promExp, "testgroup2")
156+
}

‎go.mod‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -17,6 +17,7 @@ require (
1717
github.com/pelletier/go-toml v1.7.0 // indirect
1818
github.com/pierrec/lz4 v2.5.2+incompatible // indirect
1919
github.com/pkg/errors v0.9.1
20+
github.com/prometheus/client_golang v0.9.3
2021
github.com/rcrowley/go-metrics v0.0.0-20200313005456-10cdbea86bc0 // indirect
2122
github.com/samuel/go-zookeeper v0.0.0-20190923202752-2cc03de413da
2223
github.com/smartystreets/assertions v1.1.0 // indirect

‎go.sum‎

Lines changed: 7 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -13,6 +13,7 @@ github.com/alecthomas/template v0.0.0-20160405071501-a0175ee3bccc/go.mod h1:LOuy
1313
github.com/alecthomas/units v0.0.0-20151022065526-2efee857e7cf/go.mod h1:ybxpYRFXyAe+OPACYpWeL0wqObRcbAqCMya13uyzqw0=
1414
github.com/armon/consul-api v0.0.0-20180202201655-eb2c6b5be1b6/go.mod h1:grANhF5doyWs3UAsr3K4I6qtAmlQcZDesFNEHPZAzj8=
1515
github.com/beorn7/perks v0.0.0-20180321164747-3a771d992973/go.mod h1:Dwedo/Wpr24TaqPxmxbtue+5NUziq4I4S80YR8gNf3Q=
16+
github.com/beorn7/perks v1.0.0 h1:HWo1m869IqiPhD389kmkxeTalrjNbbJTC8LXupb+sl0=
1617
github.com/beorn7/perks v1.0.0/go.mod h1:KWe93zE9D1o94FZ5RNwFwVgaQK1VOXiVxmqh+CedLV8=
1718
github.com/cespare/xxhash v1.1.0 h1:a6HrQnmkObjyL+Gs60czilIUGqrzKutQD6XZog3p+ko=
1819
github.com/cespare/xxhash v1.1.0/go.mod h1:XrSqR1VqqWfGrhpAt58auRo0WTKS1nRRg3ghfAqPWnc=
@@ -55,6 +56,7 @@ github.com/golang/glog v0.0.0-20160126235308-23def4e6c14b/go.mod h1:SBH7ygxi8pfU
5556
github.com/golang/groupcache v0.0.0-20190129154638-5b532d6fd5ef/go.mod h1:cIg4eruTrX1D+g88fzRXU5OdNfaM+9IcxsU14FzY7Hc=
5657
github.com/golang/mock v1.1.1/go.mod h1:oTYuIxOrZwtPieC+H1uAHpcLFnEyAGVDL/k47Jfbm0A=
5758
github.com/golang/protobuf v1.2.0/go.mod h1:6lQm79b+lXiMfvg/cZm0SGofjICqVBUtrP5yJMmIC1U=
59+
github.com/golang/protobuf v1.3.1 h1:YF8+flBXS5eO826T4nzqPrxfhQThhXl0YzfuUPu4SBg=
5860
github.com/golang/protobuf v1.3.1/go.mod h1:6lQm79b+lXiMfvg/cZm0SGofjICqVBUtrP5yJMmIC1U=
5961
github.com/golang/snappy v0.0.1 h1:Qgr9rKW7uDUkrbSmQeiDsGa8SjGyCOGtuasMWwvp2P4=
6062
github.com/golang/snappy v0.0.1/go.mod h1:/XxbfmMg8lxefKM7IXC3fBNl/7bRcc72aCRzEWrmP2Q=
@@ -111,6 +113,7 @@ github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY=
111113
github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE=
112114
github.com/magiconair/properties v1.8.1 h1:ZC2Vc7/ZFkGmsVC9KvOjumD+G5lXy2RtTKyzRKO2BQ4=
113115
github.com/magiconair/properties v1.8.1/go.mod h1:PppfXfuXeibc/6YijjN8zIbojt8czPbwD3XqdrwzmxQ=
116+
github.com/matttproud/golang_protobuf_extensions v1.0.1 h1:4hp9jkHxhMHkqkrB3Ix0jegS5sx/RkqARlsWZ6pIwiU=
114117
github.com/matttproud/golang_protobuf_extensions v1.0.1/go.mod h1:D8He9yQNgCq6Z5Ld7szi9bcBfOoFv/3dc6xSMkL2PC0=
115118
github.com/mitchellh/mapstructure v1.1.2 h1:fmNYVwqnSfB9mZU6OS2O6GsXM+wcskZDuKQzvN1EDeE=
116119
github.com/mitchellh/mapstructure v1.1.2/go.mod h1:FVVH3fgwuzCH5S8UJGiWEs2h04kUh9fWfEaFds41c1Y=
@@ -141,12 +144,16 @@ github.com/pkg/errors v0.9.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINE
141144
github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
142145
github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
143146
github.com/prometheus/client_golang v0.9.1/go.mod h1:7SWBe2y4D6OKWSNQJUaRYU/AaXPKyh/dDVn+NZz0KFw=
147+
github.com/prometheus/client_golang v0.9.3 h1:9iH4JKXLzFbOAdtqv/a+j8aewx2Y8lAjAydhbaScPF8=
144148
github.com/prometheus/client_golang v0.9.3/go.mod h1:/TN21ttK/J9q6uSwhBd54HahCDft0ttaMvbicHlPoso=
145149
github.com/prometheus/client_model v0.0.0-20180712105110-5c3871d89910/go.mod h1:MbSGuTsp3dbXC40dX6PRTWyKYBIrTGTE9sqQNg2J8bo=
150+
github.com/prometheus/client_model v0.0.0-20190129233127-fd36f4220a90 h1:S/YWwWx/RA8rT8tKFRuGUZhuA90OyIBpPCXkcbwU8DE=
146151
github.com/prometheus/client_model v0.0.0-20190129233127-fd36f4220a90/go.mod h1:xMI15A0UPsDsEKsMN9yxemIoYk6Tm2C1GtYGdfGttqA=
147152
github.com/prometheus/common v0.0.0-20181113130724-41aa239b4cce/go.mod h1:daVV7qP5qjZbuso7PdcryaAu0sAZbrN9i7WWcTMWvro=
153+
github.com/prometheus/common v0.4.0 h1:7etb9YClo3a6HjLzfl6rIQaU+FDfi0VSX39io3aQ+DM=
148154
github.com/prometheus/common v0.4.0/go.mod h1:TNfzLD0ON7rHzMJeJkieUDPYmFC7Snx/y86RQel1bk4=
149155
github.com/prometheus/procfs v0.0.0-20181005140218-185b4288413d/go.mod h1:c3At6R/oaqEKCNdg8wHV1ftS6bRYblBhIjjI8uT2IGk=
156+
github.com/prometheus/procfs v0.0.0-20190507164030-5867b95ac084 h1:sofwID9zm4tzrgykg80hfFph1mryUeLRsUfoocVVmRY=
150157
github.com/prometheus/procfs v0.0.0-20190507164030-5867b95ac084/go.mod h1:TjEm7ze935MbeOT/UhFTIMYKhuLP4wbCsTZCD3I8kEA=
151158
github.com/prometheus/tsdb v0.7.1/go.mod h1:qhTCs0VvXwvX/y3TZrWD7rabWM+ijKTux40TwIPHuXU=
152159
github.com/rcrowley/go-metrics v0.0.0-20190826022208-cac0b30c2563 h1:dY6ETXrvDG7Sa4vE8ZQG4yqWg6UnOcbqTAahkV813vQ=

0 commit comments

Comments
 (0)