Skip to content

Commit 6948d28

Browse files
cqin-confluentcqin-confluent
authored
Revert "DGS-24388 - Update schema counts to match UI and billing (#33… (#3378)
1 parent 9e53884 commit 6948d28

4 files changed

Lines changed: 18 additions & 114 deletions

File tree

‎internal/schema-registry/command_cluster_describe.go‎

Lines changed: 14 additions & 29 deletions
Original file line numberDiff line numberDiff line change
@@ -1,14 +1,15 @@
11
package schemaregistry
22

33
import (
4-
"encoding/json"
54
"fmt"
65
"math"
76
"strconv"
87
"strings"
98

109
"github.com/spf13/cobra"
1110

11+
metricsv2 "github.com/confluentinc/ccloud-sdk-go-v2/metrics/v2"
12+
1213
"github.com/confluentinc/cli/v4/pkg/ccloudv2"
1314
pcmd "github.com/confluentinc/cli/v4/pkg/cmd"
1415
"github.com/confluentinc/cli/v4/pkg/log"
@@ -95,11 +96,7 @@ func (c *command) clusterDescribe(cmd *cobra.Command, _ []string) error {
9596
return err
9697
}
9798

98-
queryBody, err := schemaCountQueryBodyFor(cluster.GetId())
99-
if err != nil {
100-
return err
101-
}
102-
metricsResponse, httpResp, err := metricsClient.MetricsDatasetQueryRaw("cloud", queryBody)
99+
metricsResponse, httpResp, err := metricsClient.MetricsDatasetQuery("cloud", schemaCountQueryFor(cluster.GetId()))
103100
if err := ccloudv2.UnmarshalFlatQueryResponseIfDataSchemaMatchError(err, metricsResponse, httpResp); err != nil {
104101
return err
105102
}
@@ -161,30 +158,18 @@ func (c *command) clusterDescribe(cmd *cobra.Command, _ []string) error {
161158
return table.Print()
162159
}
163160

164-
// schemaCountQueryBodyFor builds the Metrics-API request body for schema_count.
165-
// schema_count is a GAUGE. For legacy LSRCs whose schemas span two PSRCs, each
166-
// PSRC emits its own data point against the same LSRC ID, and the API's default
167-
// MEAN time aggregation under-counts. We mirror cc-billing-worker's query
168-
// (metrics/configurable/cloud_metrics_plugin.go) which uses the undocumented
169-
// "time_agg" field to force MAX per series before "agg":"SUM" combines them.
170-
// The v2 SDK doesn't expose "time_agg", so the body is built and sent raw.
171-
func schemaCountQueryBodyFor(schemaRegistryId string) ([]byte, error) {
172-
return json.Marshal(map[string]any{
173-
"aggregations": []map[string]any{{
174-
"time_agg": "MAX",
175-
"agg": "SUM",
176-
"metric": "io.confluent.kafka.schema_registry/schema_count",
177-
}},
178-
"filter": map[string]any{
179-
"field": "resource.schema_registry.id",
180-
"op": "EQ",
181-
"value": schemaRegistryId,
161+
func schemaCountQueryFor(schemaRegistryId string) metricsv2.QueryRequest {
162+
aggregations := []metricsv2.Aggregation{{Metric: "io.confluent.kafka.schema_registry/schema_count"}}
163+
filter := metricsv2.Filter{
164+
FieldFilter: &metricsv2.FieldFilter{
165+
Field: metricsv2.PtrString("resource.schema_registry.id"),
166+
Op: "EQ",
167+
Value: metricsv2.StringAsFieldFilterValue(metricsv2.PtrString(schemaRegistryId)),
182168
},
183-
"format": "FLAT",
184-
"granularity": "PT1H",
185-
"intervals": []string{"PT1H/now-2m|m"},
186-
"limit": 1000,
187-
})
169+
}
170+
req := metricsv2.NewQueryRequest(aggregations, "ALL", []string{"PT1M/now-2m|m"})
171+
req.SetFilter(filter)
172+
return *req
188173
}
189174

190175
func getMaxSchemaLimitPriceKey(serviceProvider, serviceProviderRegion, streamGovernancePackage string) string {

‎pkg/ccloudv2/metrics.go‎

Lines changed: 0 additions & 48 deletions
Original file line numberDiff line numberDiff line change
@@ -1,10 +1,8 @@
11
package ccloudv2
22

33
import (
4-
"bytes"
54
"context"
65
"encoding/json"
7-
"fmt"
86
"io"
97
"net/http"
108
"time"
@@ -51,52 +49,6 @@ func (c *MetricsClient) MetricsDatasetQuery(dataset string, query metricsv2.Quer
5149
return c.Version2Api.V2MetricsDatasetQueryPost(c.context(), dataset).QueryRequest(query).Execute()
5250
}
5351

54-
// MetricsDatasetQueryRaw posts a hand-built JSON body to /v2/metrics/{dataset}/query.
55-
// Use this when the request needs a field the typed SDK doesn't expose (e.g. the
56-
// undocumented "time_agg" knob that the Metrics API requires to override the
57-
// gauge MEAN time aggregation; see schema-registry cluster describe).
58-
func (c *MetricsClient) MetricsDatasetQueryRaw(dataset string, body []byte) (*metricsv2.QueryResponse, *http.Response, error) {
59-
cfg := c.GetConfig()
60-
if len(cfg.Servers) == 0 {
61-
return nil, nil, fmt.Errorf("metrics client has no configured server")
62-
}
63-
url := fmt.Sprintf("%s/v2/metrics/%s/query", cfg.Servers[0].URL, dataset)
64-
65-
req, err := http.NewRequestWithContext(context.Background(), http.MethodPost, url, bytes.NewReader(body))
66-
if err != nil {
67-
return nil, nil, err
68-
}
69-
req.Header.Set("Content-Type", "application/json")
70-
req.Header.Set("Accept", "application/json")
71-
if token, err := auth.GetDataplaneToken(c.cfg.Context()); err == nil {
72-
req.Header.Set("Authorization", "Bearer "+token)
73-
}
74-
75-
httpResp, err := cfg.HTTPClient.Do(req)
76-
if err != nil {
77-
return nil, httpResp, err
78-
}
79-
defer httpResp.Body.Close()
80-
81-
respBody, err := io.ReadAll(httpResp.Body)
82-
if err != nil {
83-
return nil, httpResp, err
84-
}
85-
if httpResp.StatusCode >= 400 {
86-
return nil, httpResp, fmt.Errorf("metrics API returned %d: %s", httpResp.StatusCode, string(respBody))
87-
}
88-
89-
var flat flatQueryResponse
90-
if err := json.Unmarshal(respBody, &flat); err != nil {
91-
return nil, httpResp, err
92-
}
93-
points := make([]metricsv2.Point, len(flat.Data))
94-
for i, p := range flat.Data {
95-
points[i] = metricsv2.Point{Value: p.Value, Timestamp: p.Timestamp}
96-
}
97-
return &metricsv2.QueryResponse{FlatQueryResponse: metricsv2.NewFlatQueryResponse(points)}, httpResp, nil
98-
}
99-
10052
func UnmarshalFlatQueryResponseIfDataSchemaMatchError(err error, metricsResponse *metricsv2.QueryResponse, httpResp *http.Response) error {
10153
if !IsDataMatchesMoreThanOneSchemaError(err) {
10254
return nil

‎test/fixtures/output/schema-registry/cluster/describe.golden‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -6,8 +6,8 @@
66
| Private Regional Endpoint URLs | us-east-1=https://lsrc-stk1d.us-east-1.aws.private.stag.cpdev.cloud |
77
| | us-west-2=https://lsrc-stgvk1d.us-west-2.aws.private.stag.cpdev.cloud |
88
| Catalog Endpoint URL | http://127.0.0.1:1030 |
9-
| Used Schemas | 7 |
10-
| Available Schemas | 993 |
9+
| Used Schemas | 0 |
10+
| Available Schemas | 1000 |
1111
| Free Schemas Limit | 1000 |
1212
| Global Compatibility | FULL |
1313
| Mode | READWRITE |

‎test/test-server/metrics_handlers.go‎

Lines changed: 2 additions & 35 deletions
Original file line numberDiff line numberDiff line change
@@ -2,9 +2,7 @@ package testserver
22

33
import (
44
"encoding/json"
5-
"io"
65
"net/http"
7-
"strings"
86
"testing"
97
"time"
108

@@ -15,47 +13,16 @@ import (
1513

1614
var queryTime = time.Date(2019, 12, 19, 16, 1, 0, 0, time.UTC)
1715

18-
// schemaCountTwoPsrcValue simulates an LSRC whose schemas span two PSRCs (e.g.
19-
// PSRC1=6, PSRC2=1 schemas). With the gauge's default MEAN time aggregation,
20-
// the API would return a fractional value (~3.5). With the billing-shaped query
21-
// (time_agg=MAX, agg=SUM), the per-PSRC counts collapse correctly.
22-
const schemaCountTwoPsrcValue float64 = 7
23-
2416
func handleMetricsQuery(t *testing.T) http.HandlerFunc {
2517
return func(w http.ResponseWriter, r *http.Request) {
26-
body, err := io.ReadAll(r.Body)
27-
require.NoError(t, err)
28-
29-
// schema_count is a GAUGE; for legacy LSRCs spanning two PSRCs, the
30-
// API's default MEAN under-counts. The CLI must send the undocumented
31-
// "time_agg":"MAX" alongside "agg":"SUM". See
32-
// internal/schema-registry/command_cluster_describe.go:schemaCountQueryBodyFor.
33-
// We parse as raw JSON because the v2 SDK type can't represent time_agg.
34-
var raw map[string]any
35-
require.NoError(t, json.Unmarshal(body, &raw))
36-
37-
value := 0.0
38-
if aggs, ok := raw["aggregations"].([]any); ok {
39-
for _, a := range aggs {
40-
agg, _ := a.(map[string]any)
41-
metric, _ := agg["metric"].(string)
42-
if !strings.HasSuffix(metric, "/schema_count") {
43-
continue
44-
}
45-
require.Equal(t, "MAX", agg["time_agg"], "schema_count query must set time_agg=MAX to override the gauge MEAN default")
46-
require.Equal(t, "SUM", agg["agg"], "schema_count query must set agg=SUM")
47-
value = schemaCountTwoPsrcValue
48-
}
49-
}
50-
5118
resp := &metricsv2.QueryResponse{
5219
FlatQueryResponse: &metricsv2.FlatQueryResponse{
5320
Data: []metricsv2.Point{
54-
{Value: float32(value), Timestamp: queryTime},
21+
{Value: 0.0, Timestamp: queryTime},
5522
},
5623
},
5724
}
58-
err = json.NewEncoder(w).Encode(resp)
25+
err := json.NewEncoder(w).Encode(resp)
5926
require.NoError(t, err)
6027
}
6128
}

0 commit comments

Comments
 (0)