Skip to content

Commit 6915258

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

21 files changed

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

0 commit comments

Comments
 (0)