Skip to content

Commit 42e8190

Browse files
mwainmwain
committed
Add Prometheus Metrics Exporter
1 parent d896a3f commit 42e8190

5 files changed

Lines changed: 346 additions & 0 deletions

File tree

‎core/internal/httpserver/coordinator.go‎

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

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

‎core/internal/httpserver/prometheus.go‎

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

‎core/internal/httpserver/prometheus_test.go‎

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

‎go.mod‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -18,6 +18,7 @@ require (
1818
github.com/pelletier/go-toml v1.6.0 // indirect
1919
github.com/pierrec/lz4 v2.4.1+incompatible // indirect
2020
github.com/pkg/errors v0.9.1
21+
github.com/prometheus/client_golang v0.9.3
2122
github.com/rcrowley/go-metrics v0.0.0-20190826022208-cac0b30c2563 // indirect
2223
github.com/samuel/go-zookeeper v0.0.0-20190923202752-2cc03de413da
2324
github.com/smartystreets/assertions v1.0.1 // 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=
@@ -54,6 +55,7 @@ github.com/golang/glog v0.0.0-20160126235308-23def4e6c14b/go.mod h1:SBH7ygxi8pfU
5455
github.com/golang/groupcache v0.0.0-20190129154638-5b532d6fd5ef/go.mod h1:cIg4eruTrX1D+g88fzRXU5OdNfaM+9IcxsU14FzY7Hc=
5556
github.com/golang/mock v1.1.1/go.mod h1:oTYuIxOrZwtPieC+H1uAHpcLFnEyAGVDL/k47Jfbm0A=
5657
github.com/golang/protobuf v1.2.0/go.mod h1:6lQm79b+lXiMfvg/cZm0SGofjICqVBUtrP5yJMmIC1U=
58+
github.com/golang/protobuf v1.3.1 h1:YF8+flBXS5eO826T4nzqPrxfhQThhXl0YzfuUPu4SBg=
5759
github.com/golang/protobuf v1.3.1/go.mod h1:6lQm79b+lXiMfvg/cZm0SGofjICqVBUtrP5yJMmIC1U=
5860
github.com/golang/snappy v0.0.1 h1:Qgr9rKW7uDUkrbSmQeiDsGa8SjGyCOGtuasMWwvp2P4=
5961
github.com/golang/snappy v0.0.1/go.mod h1:/XxbfmMg8lxefKM7IXC3fBNl/7bRcc72aCRzEWrmP2Q=
@@ -111,6 +113,7 @@ github.com/kr/text v0.1.0 h1:45sCR5RtlFHMR4UwH9sdQ5TC8v0qDQCHnXt+kaKSTVE=
111113
github.com/kr/text v0.1.0/go.mod h1:4Jbv+DJW3UT/LiOwJeYQe1efqtUx/iVham/4vfdArNI=
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=
@@ -133,12 +136,16 @@ github.com/pkg/errors v0.9.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINE
133136
github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
134137
github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
135138
github.com/prometheus/client_golang v0.9.1/go.mod h1:7SWBe2y4D6OKWSNQJUaRYU/AaXPKyh/dDVn+NZz0KFw=
139+
github.com/prometheus/client_golang v0.9.3 h1:9iH4JKXLzFbOAdtqv/a+j8aewx2Y8lAjAydhbaScPF8=
136140
github.com/prometheus/client_golang v0.9.3/go.mod h1:/TN21ttK/J9q6uSwhBd54HahCDft0ttaMvbicHlPoso=
137141
github.com/prometheus/client_model v0.0.0-20180712105110-5c3871d89910/go.mod h1:MbSGuTsp3dbXC40dX6PRTWyKYBIrTGTE9sqQNg2J8bo=
142+
github.com/prometheus/client_model v0.0.0-20190129233127-fd36f4220a90 h1:S/YWwWx/RA8rT8tKFRuGUZhuA90OyIBpPCXkcbwU8DE=
138143
github.com/prometheus/client_model v0.0.0-20190129233127-fd36f4220a90/go.mod h1:xMI15A0UPsDsEKsMN9yxemIoYk6Tm2C1GtYGdfGttqA=
139144
github.com/prometheus/common v0.0.0-20181113130724-41aa239b4cce/go.mod h1:daVV7qP5qjZbuso7PdcryaAu0sAZbrN9i7WWcTMWvro=
145+
github.com/prometheus/common v0.4.0 h1:7etb9YClo3a6HjLzfl6rIQaU+FDfi0VSX39io3aQ+DM=
140146
github.com/prometheus/common v0.4.0/go.mod h1:TNfzLD0ON7rHzMJeJkieUDPYmFC7Snx/y86RQel1bk4=
141147
github.com/prometheus/procfs v0.0.0-20181005140218-185b4288413d/go.mod h1:c3At6R/oaqEKCNdg8wHV1ftS6bRYblBhIjjI8uT2IGk=
148+
github.com/prometheus/procfs v0.0.0-20190507164030-5867b95ac084 h1:sofwID9zm4tzrgykg80hfFph1mryUeLRsUfoocVVmRY=
142149
github.com/prometheus/procfs v0.0.0-20190507164030-5867b95ac084/go.mod h1:TjEm7ze935MbeOT/UhFTIMYKhuLP4wbCsTZCD3I8kEA=
143150
github.com/prometheus/tsdb v0.7.1/go.mod h1:qhTCs0VvXwvX/y3TZrWD7rabWM+ijKTux40TwIPHuXU=
144151
github.com/rcrowley/go-metrics v0.0.0-20181016184325-3113b8401b8a h1:9ZKAASQSHhDYGoxY8uLVpewe1GDZ2vu2Tr/vTdVAkFQ=

0 commit comments

Comments
 (0)