Prometheus: Make Prometheus streaming parser as default client (#58365)

* Introduce a new feature flag for prometheus buffered client

* Use querydata client as default and put buffered client behind the feature flag

* Remove prometheusStreamingJSONParser feature flag as it is not needed anymore

* Update tests

* Fix unit tests

* Update feature flag description

packages/grafana-data/src/types/featureToggles.gen.ts

@@ -52,7 +52,7 @@ export interface FeatureToggles {
   cloudWatchDynamicLabels?: boolean;
   datasourceQueryMultiStatus?: boolean;
   traceToMetrics?: boolean;
-  prometheusStreamingJSONParser?: boolean;
+  prometheusBufferedClient?: boolean;
   newDBLibrary?: boolean;
   validateDashboardsOnSave?: boolean;
   autoMigrateGraphPanels?: boolean;

pkg/services/featuremgmt/registry.go

@@ -210,9 +210,9 @@ var (
 			FrontendOnly: true,
 		},
 		{
-			Name:        "prometheusStreamingJSONParser",
+			Name:        "prometheusBufferedClient",
-			Description: "Enable streaming JSON parser for Prometheus datasource",
+			Description: "Enable buffered (old) client for Prometheus datasource as default instead of streaming JSON parser client (new)",
-			State:       FeatureStateBeta,
+			State:       FeatureStateStable,
 		},
 		{
 			Name:        "newDBLibrary",

pkg/services/featuremgmt/toggles_gen.go

@@ -151,9 +151,9 @@ const (
 	// Enable trace to metrics links
 	FlagTraceToMetrics = "traceToMetrics"
 
-	// FlagPrometheusStreamingJSONParser
+	// FlagPrometheusBufferedClient
-	// Enable streaming JSON parser for Prometheus datasource
+	// Enable buffered (old) client for Prometheus datasource as default instead of streaming JSON parser client (new)
-	FlagPrometheusStreamingJSONParser = "prometheusStreamingJSONParser"
+	FlagPrometheusBufferedClient = "prometheusBufferedClient"
 
 	// FlagNewDBLibrary
 	// Use jmoiron/sqlx rather than xorm for a few backend services

pkg/tests/api/prometheus/prometheus_test.go

@@ -79,8 +79,8 @@ func TestIntegrationPrometheusBuffered(t *testing.T) {
 		})
 		buf1 := &bytes.Buffer{}
 		err = json.NewEncoder(buf1).Encode(dtos.MetricRequest{
-			From:    "now-1h",
+			From:    "1668078080000",
-			To:      "now",
+			To:      "1668081680000",
 			Queries: []*simplejson.Json{query},
 		})
 		require.NoError(t, err)
@@ -88,7 +88,7 @@ func TestIntegrationPrometheusBuffered(t *testing.T) {
 		// nolint:gosec
 		resp, err := http.Post(u, "application/json", buf1)
 		require.NoError(t, err)
-		require.Equal(t, http.StatusBadRequest, resp.StatusCode)
+		require.Equal(t, http.StatusInternalServerError, resp.StatusCode)
 		t.Cleanup(func() {
 			err := resp.Body.Close()
 			require.NoError(t, err)
@@ -97,7 +97,8 @@ func TestIntegrationPrometheusBuffered(t *testing.T) {
 		require.NoError(t, err)
 
 		require.NotNil(t, outgoingRequest)
-		require.Equal(t, "/api/v1/query_range?q1=1&q2=2", outgoingRequest.URL.String())
+		require.Equal(t, "/api/v1/query_range?end=1668081660&q1=1&q2=2&query=up&start=1668078060&step=30",
+			outgoingRequest.URL.String())
 		require.Equal(t, "custom-header-value", outgoingRequest.Header.Get("X-CUSTOM-HEADER"))
 		username, pwd, ok := outgoingRequest.BasicAuth()
 		require.True(t, ok)
@@ -110,9 +111,7 @@ func TestIntegrationPrometheusClient(t *testing.T) {
 	if testing.Short() {
 		t.Skip("skipping integration test")
 	}
-	dir, path := testinfra.CreateGrafDir(t, testinfra.GrafanaOpts{
+	dir, path := testinfra.CreateGrafDir(t, testinfra.GrafanaOpts{})
-		EnableFeatureToggles: []string{"prometheusStreamingJSONParser"},
-	})
 
 	grafanaListeningAddr, testEnv := testinfra.StartGrafanaEnv(t, dir, path)
 	ctx := context.Background()

pkg/tsdb/prometheus/prometheus.go

@@ -93,11 +93,11 @@ func (s *Service) QueryData(ctx context.Context, req *backend.QueryDataRequest)
 		return nil, err
 	}
 
-	if s.features.IsEnabled(featuremgmt.FlagPrometheusStreamingJSONParser) || s.features.IsEnabled(featuremgmt.FlagPrometheusWideSeries) {
+	if s.features.IsEnabled(featuremgmt.FlagPrometheusBufferedClient) {
-		return i.queryData.Execute(ctx, req)
+		return i.buffered.ExecuteTimeSeriesQuery(ctx, req)
 	}
 
-	return i.buffered.ExecuteTimeSeriesQuery(ctx, req)
+	return i.queryData.Execute(ctx, req)
 }
 
 func (s *Service) CallResource(ctx context.Context, req *backend.CallResourceRequest, sender backend.CallResourceResponseSender) error {

pkg/tsdb/prometheus/querydata/request_test.go

@@ -415,7 +415,8 @@ func setup(wideFrames bool) (*testContext, error) {
 		JSONData: json.RawMessage(`{"timeInterval": "15s"}`),
 	}
 
-	features := &fakeFeatureToggles{flags: map[string]bool{"prometheusStreamingJSONParser": true, "prometheusWideSeries": wideFrames}}
+	features := &fakeFeatureToggles{flags: map[string]bool{"prometheusBufferedClient": false,
+		"prometheusWideSeries": wideFrames}}
 
 	opts, err := client.CreateTransportOptions(settings, &setting.Cfg{}, &logtest.Fake{})
 	if err != nil {