Skip to content

Commit fb61f5a

Browse files
SungJin1212SungJin1212
committed
Allow ruler to retrieve proto format query response
Signed-off-by: SungJin1212 <[email protected]>
1 parent 7fb98ab commit fb61f5a

21 files changed

+649
-75
lines changed

‎CHANGELOG.md‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -9,6 +9,7 @@
99
* [CHANGE] Change all max async concurrency default values `50` to `3` #6268
1010
* [CHANGE] Change default value of `-blocks-storage.bucket-store.index-cache.multilevel.max-async-concurrency` from `50` to `3` #6265
1111
* [CHANGE] Enable Compactor and Alertmanager in target all. #6204
12+
* [FEATURE] Ruler: Add an experimental flag `-ruler.query-response-format` to retrieve query response as a proto format. #6345
1213
* [FEATURE] Ruler: Pagination support for List Rules API. #6299
1314
* [FEATURE] Query Frontend/Querier: Add protobuf codec `-api.querier-default-codec` and the option to choose response compression type `-querier.response-compression`. #5527
1415
* [FEATURE] Ruler: Experimental: Add `ruler.frontend-address` to allow query to query frontends instead of ingesters. #6151

‎docs/configuration/config-file-reference.md‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -4252,6 +4252,11 @@ The `ruler_config` configures the Cortex ruler.
42524252
# CLI flag: -ruler.frontend-address
42534253
[frontend_address: <string> | default = ""]
42544254
4255+
# [Experimental] Query response format to get query results from Query Frontend
4256+
# when the rule evaluation. Supported values: json,protobuf
4257+
# CLI flag: -ruler.query-response-format
4258+
[query_response_format: <string> | default = "protobuf"]
4259+
42554260
frontend_client:
42564261
# gRPC client max receive message size (bytes).
42574262
# CLI flag: -ruler.frontendClient.grpc-max-recv-msg-size

‎go.mod‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -80,6 +80,7 @@ require (
8080
github.com/bboreham/go-loser v0.0.0-20230920113527-fcc2c21820a3
8181
github.com/cespare/xxhash/v2 v2.3.0
8282
github.com/google/go-cmp v0.6.0
83+
github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822
8384
github.com/sercand/kuberesolver/v4 v4.0.0
8485
go.opentelemetry.io/collector/pdata v1.19.0
8586
golang.org/x/exp v0.0.0-20240613232115-7f521ea00fb8
@@ -186,7 +187,6 @@ require (
186187
github.com/mitchellh/mapstructure v1.5.0 // indirect
187188
github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect
188189
github.com/modern-go/reflect2 v1.0.2 // indirect
189-
github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect
190190
github.com/mwitkow/go-conntrack v0.0.0-20190716064945-2f068394615f // indirect
191191
github.com/ncw/swift v1.0.53 // indirect
192192
github.com/oklog/run v1.1.0 // indirect

‎integration/ruler_test.go‎

Lines changed: 52 additions & 33 deletions
Original file line numberDiff line numberDiff line change
@@ -1670,42 +1670,61 @@ func TestRulerEvalWithQueryFrontend(t *testing.T) {
16701670
distributor := e2ecortex.NewDistributor("distributor", e2ecortex.RingStoreConsul, consul.NetworkHTTPEndpoint(), flags, "")
16711671
ingester := e2ecortex.NewIngester("ingester", e2ecortex.RingStoreConsul, consul.NetworkHTTPEndpoint(), flags, "")
16721672
require.NoError(t, s.StartAndWaitReady(distributor, ingester))
1673-
queryFrontend := e2ecortex.NewQueryFrontend("query-frontend", flags, "")
1674-
require.NoError(t, s.Start(queryFrontend))
1675-
1676-
ruler := e2ecortex.NewRuler("ruler", consul.NetworkHTTPEndpoint(), mergeFlags(flags, map[string]string{
1677-
"-ruler.frontend-address": queryFrontend.NetworkGRPCEndpoint(),
1678-
}), "")
1679-
querier := e2ecortex.NewQuerier("querier", e2ecortex.RingStoreConsul, consul.NetworkHTTPEndpoint(), mergeFlags(flags, map[string]string{
1680-
"-querier.frontend-address": queryFrontend.NetworkGRPCEndpoint(),
1681-
}), "")
1682-
require.NoError(t, s.StartAndWaitReady(ruler, querier))
1683-
1684-
c, err := e2ecortex.NewClient("", "", "", ruler.HTTPEndpoint(), user)
1685-
require.NoError(t, err)
1673+
for _, format := range []string{"protobuf", "json"} {
1674+
t.Run(fmt.Sprintf("format:%s", format), func(t *testing.T) {
1675+
queryFrontendFlag := mergeFlags(flags, map[string]string{
1676+
"-ruler.query-response-format": format,
1677+
})
1678+
queryFrontend := e2ecortex.NewQueryFrontend("query-frontend", queryFrontendFlag, "")
1679+
require.NoError(t, s.Start(queryFrontend))
16861680

1687-
expression := "metric"
1688-
groupName := "rule_group"
1689-
ruleName := "rule_name"
1690-
require.NoError(t, c.SetRuleGroup(ruleGroupWithRule(groupName, ruleName, expression), namespace))
1681+
querier := e2ecortex.NewQuerier("querier", e2ecortex.RingStoreConsul, consul.NetworkHTTPEndpoint(), mergeFlags(queryFrontendFlag, map[string]string{
1682+
"-querier.frontend-address": queryFrontend.NetworkGRPCEndpoint(),
1683+
}), "")
1684+
require.NoError(t, s.StartAndWaitReady(querier))
16911685

1692-
rgMatcher := ruleGroupMatcher(user, namespace, groupName)
1693-
// Wait until ruler has loaded the group.
1694-
require.NoError(t, ruler.WaitSumMetricsWithOptions(e2e.Equals(1), []string{"cortex_prometheus_rule_group_rules"}, e2e.WithLabelMatchers(rgMatcher), e2e.WaitMissingMetrics))
1695-
// Wait until rule group has tried to evaluate the rule.
1696-
require.NoError(t, ruler.WaitSumMetricsWithOptions(e2e.GreaterOrEqual(1), []string{"cortex_prometheus_rule_evaluations_total"}, e2e.WithLabelMatchers(rgMatcher), e2e.WaitMissingMetrics))
1686+
rulerFlag := mergeFlags(queryFrontendFlag, map[string]string{
1687+
"-ruler.frontend-address": queryFrontend.NetworkGRPCEndpoint(),
1688+
})
1689+
ruler := e2ecortex.NewRuler("ruler", consul.NetworkHTTPEndpoint(), rulerFlag, "")
1690+
require.NoError(t, s.StartAndWaitReady(ruler))
16971691

1698-
matcher := labels.MustNewMatcher(labels.MatchEqual, "user", user)
1699-
// Check that cortex_ruler_query_frontend_clients went up
1700-
require.NoError(t, ruler.WaitSumMetricsWithOptions(e2e.Equals(1), []string{"cortex_ruler_query_frontend_clients"}, e2e.WaitMissingMetrics))
1701-
// Check that cortex_ruler_queries_total went up
1702-
require.NoError(t, ruler.WaitSumMetricsWithOptions(e2e.GreaterOrEqual(1), []string{"cortex_ruler_queries_total"}, e2e.WithLabelMatchers(matcher), e2e.WaitMissingMetrics))
1703-
// Check that cortex_ruler_queries_failed_total is zero
1704-
require.NoError(t, ruler.WaitSumMetricsWithOptions(e2e.Equals(0), []string{"cortex_ruler_queries_failed_total"}, e2e.WithLabelMatchers(matcher), e2e.WaitMissingMetrics))
1705-
// Check that cortex_ruler_write_requests_total went up
1706-
require.NoError(t, ruler.WaitSumMetricsWithOptions(e2e.GreaterOrEqual(1), []string{"cortex_ruler_write_requests_total"}, e2e.WithLabelMatchers(matcher), e2e.WaitMissingMetrics))
1707-
// Check that cortex_ruler_write_requests_failed_total is zero
1708-
require.NoError(t, ruler.WaitSumMetricsWithOptions(e2e.Equals(0), []string{"cortex_ruler_write_requests_failed_total"}, e2e.WithLabelMatchers(matcher), e2e.WaitMissingMetrics))
1692+
t.Cleanup(func() {
1693+
_ = s.Stop(ruler)
1694+
_ = s.Stop(queryFrontend)
1695+
_ = s.Stop(querier)
1696+
})
1697+
1698+
c, err := e2ecortex.NewClient("", "", "", ruler.HTTPEndpoint(), user)
1699+
require.NoError(t, err)
1700+
1701+
expression := "metric" // vector
1702+
//expression := "scalar(count(up == 1)) > bool 1" // scalar
1703+
groupName := "rule_group"
1704+
ruleName := "rule_name"
1705+
require.NoError(t, c.SetRuleGroup(ruleGroupWithRule(groupName, ruleName, expression), namespace))
1706+
1707+
rgMatcher := ruleGroupMatcher(user, namespace, groupName)
1708+
// Wait until ruler has loaded the group.
1709+
require.NoError(t, ruler.WaitSumMetricsWithOptions(e2e.Equals(1), []string{"cortex_prometheus_rule_group_rules"}, e2e.WithLabelMatchers(rgMatcher), e2e.WaitMissingMetrics))
1710+
// Wait until rule group has tried to evaluate the rule.
1711+
require.NoError(t, ruler.WaitSumMetricsWithOptions(e2e.GreaterOrEqual(1), []string{"cortex_prometheus_rule_evaluations_total"}, e2e.WithLabelMatchers(rgMatcher), e2e.WaitMissingMetrics))
1712+
// Make sure not to fail
1713+
require.NoError(t, ruler.WaitSumMetricsWithOptions(e2e.Equals(0), []string{"cortex_prometheus_rule_evaluation_failures_total"}, e2e.WithLabelMatchers(rgMatcher), e2e.WaitMissingMetrics))
1714+
1715+
matcher := labels.MustNewMatcher(labels.MatchEqual, "user", user)
1716+
// Check that cortex_ruler_query_frontend_clients went up
1717+
require.NoError(t, ruler.WaitSumMetricsWithOptions(e2e.Equals(1), []string{"cortex_ruler_query_frontend_clients"}, e2e.WaitMissingMetrics))
1718+
// Check that cortex_ruler_queries_total went up
1719+
require.NoError(t, ruler.WaitSumMetricsWithOptions(e2e.GreaterOrEqual(1), []string{"cortex_ruler_queries_total"}, e2e.WithLabelMatchers(matcher), e2e.WaitMissingMetrics))
1720+
// Check that cortex_ruler_queries_failed_total is zero
1721+
require.NoError(t, ruler.WaitSumMetricsWithOptions(e2e.Equals(0), []string{"cortex_ruler_queries_failed_total"}, e2e.WithLabelMatchers(matcher), e2e.WaitMissingMetrics))
1722+
// Check that cortex_ruler_write_requests_total went up
1723+
require.NoError(t, ruler.WaitSumMetricsWithOptions(e2e.GreaterOrEqual(1), []string{"cortex_ruler_write_requests_total"}, e2e.WithLabelMatchers(matcher), e2e.WaitMissingMetrics))
1724+
// Check that cortex_ruler_write_requests_failed_total is zero
1725+
require.NoError(t, ruler.WaitSumMetricsWithOptions(e2e.Equals(0), []string{"cortex_ruler_write_requests_failed_total"}, e2e.WithLabelMatchers(matcher), e2e.WaitMissingMetrics))
1726+
})
1727+
}
17091728
}
17101729

17111730
func parseAlertFromRule(t *testing.T, rules interface{}) *alertingRule {

‎pkg/cortex/modules.go‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -474,7 +474,7 @@ func (t *Cortex) initQueryFrontendTripperware() (serv services.Service, err erro
474474
prometheusCodec := queryrange.NewPrometheusCodec(false, t.Cfg.Querier.ResponseCompression, t.Cfg.API.QuerierDefaultCodec)
475475
// ShardedPrometheusCodec is same as PrometheusCodec but to be used on the sharded queries (it sum up the stats)
476476
shardedPrometheusCodec := queryrange.NewPrometheusCodec(true, t.Cfg.Querier.ResponseCompression, t.Cfg.API.QuerierDefaultCodec)
477-
instantQueryCodec := instantquery.NewInstantQueryCodec(t.Cfg.Querier.ResponseCompression, t.Cfg.API.QuerierDefaultCodec)
477+
instantQueryCodec := instantquery.NewInstantQueryCodec(t.Cfg.Querier.ResponseCompression, t.Cfg.API.QuerierDefaultCodec, t.Cfg.Ruler.QueryResponseFormat)
478478

479479
queryRangeMiddlewares, cache, err := queryrange.Middlewares(
480480
t.Cfg.QueryRange,

‎pkg/querier/codec/protobuf_codec_test.go‎

Lines changed: 182 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,182 @@
1+
package codec
2+
3+
import (
4+
"testing"
5+
6+
"github.com/gogo/protobuf/proto"
7+
"github.com/prometheus/prometheus/model/labels"
8+
"github.com/prometheus/prometheus/promql"
9+
v1 "github.com/prometheus/prometheus/web/api/v1"
10+
"github.com/stretchr/testify/require"
11+
12+
"github.com/cortexproject/cortex/pkg/cortexpb"
13+
"github.com/cortexproject/cortex/pkg/querier/tripperware"
14+
)
15+
16+
func TestPrometheusResponse_ProtoMarshalUnMarshal(t *testing.T) {
17+
var tests = []struct {
18+
name string
19+
resp *v1.Response
20+
expectedUnmarshalResp tripperware.PrometheusResponse
21+
}{
22+
{
23+
name: "vector",
24+
resp: &v1.Response{
25+
Status: "success",
26+
Data: &v1.QueryData{
27+
ResultType: "vector",
28+
Result: promql.Vector{
29+
{
30+
Metric: labels.FromStrings("name", "value"),
31+
T: 1234,
32+
F: 5.67,
33+
},
34+
},
35+
},
36+
},
37+
expectedUnmarshalResp: tripperware.PrometheusResponse{
38+
Status: "success",
39+
Data: tripperware.PrometheusData{
40+
ResultType: "vector",
41+
Result: tripperware.PrometheusQueryResult{
42+
Result: &tripperware.PrometheusQueryResult_Vector{
43+
Vector: &tripperware.Vector{
44+
Samples: []tripperware.Sample{
45+
{
46+
Labels: cortexpb.FromLabelsToLabelAdapters(labels.FromStrings("name", "value")),
47+
Sample: &cortexpb.Sample{
48+
TimestampMs: 1234,
49+
Value: 5.67,
50+
},
51+
},
52+
},
53+
},
54+
},
55+
},
56+
},
57+
},
58+
},
59+
{
60+
name: "matrix",
61+
resp: &v1.Response{
62+
Status: "success",
63+
Data: &v1.QueryData{
64+
ResultType: "matrix",
65+
Result: promql.Matrix{
66+
{
67+
Metric: labels.FromStrings("name1", "value1"),
68+
Floats: []promql.FPoint{
69+
{T: 12, F: 3.4},
70+
{T: 56, F: 7.8},
71+
},
72+
},
73+
{
74+
Metric: labels.FromStrings("name2", "value2"),
75+
Floats: []promql.FPoint{
76+
{T: 12, F: 3.4},
77+
{T: 56, F: 7.8},
78+
},
79+
},
80+
},
81+
},
82+
},
83+
expectedUnmarshalResp: tripperware.PrometheusResponse{
84+
Status: "success",
85+
Data: tripperware.PrometheusData{
86+
ResultType: "matrix",
87+
Result: tripperware.PrometheusQueryResult{
88+
Result: &tripperware.PrometheusQueryResult_Matrix{
89+
Matrix: &tripperware.Matrix{
90+
SampleStreams: []tripperware.SampleStream{
91+
{
92+
Labels: cortexpb.FromLabelsToLabelAdapters(labels.FromStrings("name1", "value1")),
93+
Samples: []cortexpb.Sample{
94+
{
95+
TimestampMs: 12,
96+
Value: 3.4,
97+
},
98+
{
99+
TimestampMs: 56,
100+
Value: 7.8,
101+
},
102+
},
103+
},
104+
{
105+
Labels: cortexpb.FromLabelsToLabelAdapters(labels.FromStrings("name2", "value2")),
106+
Samples: []cortexpb.Sample{
107+
{
108+
TimestampMs: 12,
109+
Value: 3.4,
110+
},
111+
{
112+
TimestampMs: 56,
113+
Value: 7.8,
114+
},
115+
},
116+
},
117+
},
118+
},
119+
},
120+
},
121+
},
122+
},
123+
},
124+
{
125+
name: "scalar",
126+
resp: &v1.Response{
127+
Status: "success",
128+
Data: &v1.QueryData{
129+
ResultType: "scalar",
130+
Result: promql.Scalar{T: 1000, V: 2},
131+
},
132+
},
133+
expectedUnmarshalResp: tripperware.PrometheusResponse{
134+
Status: "success",
135+
Data: tripperware.PrometheusData{
136+
ResultType: "scalar",
137+
Result: tripperware.PrometheusQueryResult{
138+
Result: &tripperware.PrometheusQueryResult_RawBytes{
139+
RawBytes: []byte(`{"resultType":"scalar","result":[1,"2"]}`),
140+
},
141+
},
142+
},
143+
},
144+
},
145+
{
146+
name: "string",
147+
resp: &v1.Response{
148+
Status: "success",
149+
Data: &v1.QueryData{
150+
ResultType: "string",
151+
Result: promql.String{T: 1000, V: "2"},
152+
},
153+
},
154+
expectedUnmarshalResp: tripperware.PrometheusResponse{
155+
Status: "success",
156+
Data: tripperware.PrometheusData{
157+
ResultType: "string",
158+
Result: tripperware.PrometheusQueryResult{
159+
Result: &tripperware.PrometheusQueryResult_RawBytes{
160+
RawBytes: []byte(`{"resultType":"string","result":[1,"2"]}`),
161+
},
162+
},
163+
},
164+
},
165+
},
166+
}
167+
for _, test := range tests {
168+
t.Run(test.name, func(t *testing.T) {
169+
prometheusResponse, err := createPrometheusQueryResponse(test.resp)
170+
require.NoError(t, err)
171+
172+
b, err := proto.Marshal(prometheusResponse)
173+
require.NoError(t, err)
174+
175+
var resp tripperware.PrometheusResponse
176+
err = proto.Unmarshal(b, &resp)
177+
require.NoError(t, err)
178+
179+
require.Equal(t, test.expectedUnmarshalResp, resp)
180+
})
181+
}
182+
}

0 commit comments

Comments
 (0)