Merge remote-tracking branch 'origin/main' into resource-store

pkg/middleware/request_metrics.go

@@ -51,12 +51,19 @@ func RequestMetrics(features featuremgmt.FeatureToggles, cfg *setting.Cfg, promR
 	if features.IsEnabledGlobally(featuremgmt.FlagEnableNativeHTTPHistogram) {
 		// the recommended default value from the prom_client
 		// https://github.com/prometheus/client_golang/blob/main/prometheus/histogram.go#L411
-		// Giving this variable an value means the client will expose the histograms as an
-		// native histogram instead of normal a normal histogram.
+		// Giving this variable a value means the client will expose a native
+		// histogram.
 		histogramOptions.NativeHistogramBucketFactor = 1.1
 		// The default value in OTel. It probably good enough for us as well.
 		histogramOptions.NativeHistogramMaxBucketNumber = 160
 		histogramOptions.NativeHistogramMinResetDuration = time.Hour
+
+		if features.IsEnabledGlobally(featuremgmt.FlagDisableClassicHTTPHistogram) {
+			// setting Buckets to nil with native options set means the classic
+			// histogram will no longer be exposed - this can be a good way to
+			// reduce cardinality in the exposed metrics
+			histogramOptions.Buckets = nil
+		}
 	}
 
 	httpRequestDurationHistogram := prometheus.NewHistogramVec(

pkg/services/authz/zanzana/logger.go

@@ -0,0 +1,86 @@
+package zanzana
+
+import (
+	"context"
+
+	"go.uber.org/zap"
+
+	"github.com/grafana/grafana/pkg/infra/log"
+)
+
+// zanzanaLogger is a grafana logger wrapper compatible with OpenFGA logger interface
+type zanzanaLogger struct {
+	logger *log.ConcreteLogger
+}
+
+func newZanzanaLogger() *zanzanaLogger {
+	logger := log.New("openfga-server")
+
+	return &zanzanaLogger{
+		logger: logger,
+	}
+}
+
+// Simple converter for zap logger fields
+func zapFieldsToArgs(fields []zap.Field) []any {
+	args := make([]any, 0)
+	for _, f := range fields {
+		args = append(args, f.Key)
+		if f.Interface != nil {
+			args = append(args, f.Interface)
+		} else if f.String != "" {
+			args = append(args, f.String)
+		} else {
+			args = append(args, f.Integer)
+		}
+	}
+	return args
+}
+
+func (l *zanzanaLogger) Debug(msg string, fields ...zap.Field) {
+	l.logger.Debug(msg, zapFieldsToArgs(fields)...)
+}
+
+func (l *zanzanaLogger) Info(msg string, fields ...zap.Field) {
+	l.logger.Info(msg, zapFieldsToArgs(fields)...)
+}
+
+func (l *zanzanaLogger) Warn(msg string, fields ...zap.Field) {
+	l.logger.Warn(msg, zapFieldsToArgs(fields)...)
+}
+
+func (l *zanzanaLogger) Error(msg string, fields ...zap.Field) {
+	l.logger.Error(msg, zapFieldsToArgs(fields)...)
+}
+
+func (l *zanzanaLogger) Panic(msg string, fields ...zap.Field) {
+	l.logger.Error(msg, zapFieldsToArgs(fields)...)
+}
+
+func (l *zanzanaLogger) Fatal(msg string, fields ...zap.Field) {
+	l.logger.Error(msg, zapFieldsToArgs(fields)...)
+}
+
+func (l *zanzanaLogger) DebugWithContext(ctx context.Context, msg string, fields ...zap.Field) {
+	l.logger.Debug(msg, zapFieldsToArgs(fields)...)
+}
+
+func (l *zanzanaLogger) InfoWithContext(ctx context.Context, msg string, fields ...zap.Field) {
+	l.logger.Info(msg, zapFieldsToArgs(fields)...)
+}
+
+func (l *zanzanaLogger) WarnWithContext(ctx context.Context, msg string, fields ...zap.Field) {
+	l.logger.Warn(msg, zapFieldsToArgs(fields)...)
+}
+
+func (l *zanzanaLogger) ErrorWithContext(ctx context.Context, msg string, fields ...zap.Field) {
+	l.logger.Error(msg, zapFieldsToArgs(fields)...)
+}
+
+func (l *zanzanaLogger) PanicWithContext(ctx context.Context, msg string, fields ...zap.Field) {
+	l.logger.Error(msg, zapFieldsToArgs(fields)...)
+}
+
+func (l *zanzanaLogger) FatalWithContext(ctx context.Context, msg string, fields ...zap.Field) {
+	l.logger.Error(msg, zapFieldsToArgs(fields)...)
+}

pkg/services/authz/zanzana/server.go

@@ -1,7 +1,6 @@
 package zanzana
 
 import (
-	"github.com/openfga/openfga/pkg/logger"
 	"github.com/openfga/openfga/pkg/server"
 	"github.com/openfga/openfga/pkg/storage"
 )
@@ -10,8 +9,7 @@ func NewServer(store storage.OpenFGADatastore) (*server.Server, error) {
 	// FIXME(kalleep): add support for more options, configure logging, tracing etc
 	opts := []server.OpenFGAServiceV1Option{
 		server.WithDatastore(store),
-		// FIXME(kalleep): Write and log adapter for open fga logging interface
-		server.WithLogger(logger.NewNoopLogger()),
+		server.WithLogger(newZanzanaLogger()),
 	}
 
 	// FIXME(kalleep): Interceptors

pkg/services/featuremgmt/registry.go

@@ -700,11 +700,24 @@ var (
 			FrontendOnly: true,
 		},
 		{
-			Name:         "enableNativeHTTPHistogram",
-			Description:  "Enables native HTTP Histograms",
-			Stage:        FeatureStageExperimental,
-			FrontendOnly: false,
-			Owner:        hostedGrafanaTeam,
+			Name:              "enableNativeHTTPHistogram",
+			Description:       "Enables native HTTP Histograms",
+			Stage:             FeatureStageExperimental,
+			FrontendOnly:      false,
+			Owner:             grafanaBackendServicesSquad,
+			HideFromAdminPage: true,
+			AllowSelfServe:    false,
+			RequiresRestart:   true,
+		},
+		FeatureFlag{
+			Name:              "disableClassicHTTPHistogram",
+			Description:       "Disables classic HTTP Histogram (use with enableNativeHTTPHistogram)",
+			Stage:             FeatureStageExperimental,
+			FrontendOnly:      false,
+			Owner:             grafanaBackendServicesSquad,
+			HideFromAdminPage: true,
+			AllowSelfServe:    false,
+			RequiresRestart:   true,
 		},
 		{
 			Name:         "formatString",

pkg/services/featuremgmt/toggles_gen.csv

@@ -92,7 +92,8 @@ pluginsAPIMetrics,experimental,@grafana/plugins-platform-backend,false,false,tru
 idForwarding,experimental,@grafana/identity-access-team,false,false,false
 externalServiceAccounts,preview,@grafana/identity-access-team,false,false,false
 panelMonitoring,GA,@grafana/dataviz-squad,false,false,true
-enableNativeHTTPHistogram,experimental,@grafana/hosted-grafana-team,false,false,false
+enableNativeHTTPHistogram,experimental,@grafana/grafana-backend-services-squad,false,true,false
+disableClassicHTTPHistogram,experimental,@grafana/grafana-backend-services-squad,false,true,false
 formatString,preview,@grafana/dataviz-squad,false,false,true
 transformationsVariableSupport,GA,@grafana/dataviz-squad,false,false,true
 kubernetesPlaylists,GA,@grafana/grafana-app-platform-squad,false,true,false

pkg/services/featuremgmt/toggles_gen.go

@@ -383,6 +383,10 @@ const (
 	// Enables native HTTP Histograms
 	FlagEnableNativeHTTPHistogram = "enableNativeHTTPHistogram"
 
+	// FlagDisableClassicHTTPHistogram
+	// Disables classic HTTP Histogram (use with enableNativeHTTPHistogram)
+	FlagDisableClassicHTTPHistogram = "disableClassicHTTPHistogram"
+
 	// FlagFormatString
 	// Enable format string transformer
 	FlagFormatString = "formatString"

pkg/services/ngalert/remote/alertmanager.go

@@ -15,7 +15,6 @@ import (
 	amalert "github.com/prometheus/alertmanager/api/v2/client/alert"
 	amalertgroup "github.com/prometheus/alertmanager/api/v2/client/alertgroup"
 	amgeneral "github.com/prometheus/alertmanager/api/v2/client/general"
-	amreceiver "github.com/prometheus/alertmanager/api/v2/client/receiver"
 	amsilence "github.com/prometheus/alertmanager/api/v2/client/silence"
 	"github.com/prometheus/client_golang/prometheus"
 
@@ -487,21 +486,7 @@ func (am *Alertmanager) GetStatus(ctx context.Context) (apimodels.GettableStatus
 }
 
 func (am *Alertmanager) GetReceivers(ctx context.Context) ([]apimodels.Receiver, error) {
-	params := amreceiver.NewGetReceiversParamsWithContext(ctx)
-	res, err := am.amClient.Receiver.GetReceivers(params)
-	if err != nil {
-		return []apimodels.Receiver{}, err
-	}
-
-	rcvs := make([]apimodels.Receiver, len(res.Payload))
-	for i, rcv := range res.Payload {
-		rcvs[i] = apimodels.Receiver{
-			Name:         *rcv.Name,
-			Integrations: []apimodels.Integration{},
-		}
-	}
-
-	return rcvs, nil
+	return am.mimirClient.GetReceivers(ctx)
 }
 
 func (am *Alertmanager) TestReceivers(ctx context.Context, c apimodels.TestReceiversConfigBodyParams) (*notifier.TestReceiversResult, error) {

pkg/services/ngalert/remote/alertmanager_test.go

@@ -717,7 +717,13 @@ func TestIntegrationRemoteAlertmanagerReceivers(t *testing.T) {
 	// We should start with the default config.
 	rcvs, err := am.GetReceivers(context.Background())
 	require.NoError(t, err)
-	require.Equal(t, "empty-receiver", rcvs[0].Name)
+	require.Equal(t, []apimodels.Receiver{
+		{
+			Active:       true,
+			Name:         "empty-receiver",
+			Integrations: []apimodels.Integration{},
+		},
+	}, rcvs)
 }
 
 func genAlert(active bool, labels map[string]string) amv2.PostableAlert {

pkg/services/ngalert/remote/client/alertmanager_configuration.go

@@ -11,7 +11,8 @@ import (
 )
 
 const (
-	grafanaAlertmanagerConfigPath = "/api/v1/grafana/config"
+	grafanaAlertmanagerConfigPath    = "/api/v1/grafana/config"
+	grafanaAlertmanagerReceiversPath = "/api/v1/grafana/receivers"
 )
 
 type UserGrafanaConfig struct {
@@ -63,3 +64,16 @@ func (mc *Mimir) CreateGrafanaAlertmanagerConfig(ctx context.Context, cfg *apimo
 func (mc *Mimir) DeleteGrafanaAlertmanagerConfig(ctx context.Context) error {
 	return mc.doOK(ctx, grafanaAlertmanagerConfigPath, http.MethodDelete, nil)
 }
+
+func (mc *Mimir) GetReceivers(ctx context.Context) ([]apimodels.Receiver, error) {
+	response := []apimodels.Receiver{}
+
+	// nolint:bodyclose
+	// closed within `do`
+	_, err := mc.do(ctx, grafanaAlertmanagerReceiversPath, http.MethodGet, nil, &response)
+	if err != nil {
+		return nil, err
+	}
+
+	return response, nil
+}

pkg/services/ngalert/remote/client/mimir.go

@@ -29,6 +29,9 @@ type MimirClient interface {
 	DeleteGrafanaAlertmanagerConfig(ctx context.Context) error
 
 	ShouldPromoteConfig() bool
+
+	// Mimir implements an extended version of the receivers API under a different path.
+	GetReceivers(ctx context.Context) ([]apimodels.Receiver, error)
 }
 
 type Mimir struct {

pkg/services/notifications/webhook.go

@@ -4,10 +4,12 @@ import (
 	"bytes"
 	"context"
 	"crypto/tls"
+	"errors"
 	"fmt"
 	"io"
 	"net"
 	"net/http"
+	"net/url"
 	"time"
 
 	"github.com/grafana/grafana/pkg/util"
@@ -62,6 +64,11 @@ func (ns *NotificationService) sendWebRequestSync(ctx context.Context, webhook *
 	if err != nil {
 		return err
 	}
+	url, err := url.Parse(webhook.Url)
+	if err != nil {
+		// Should not be possible - NewRequestWithContext should also err if the URL is bad.
+		return err
+	}
 
 	if webhook.ContentType == "" {
 		webhook.ContentType = "application/json"
@@ -80,7 +87,7 @@ func (ns *NotificationService) sendWebRequestSync(ctx context.Context, webhook *
 
 	resp, err := netClient.Do(request)
 	if err != nil {
-		return err
+		return redactURL(err)
 	}
 	defer func() {
 		if err := resp.Body.Close(); err != nil {
@@ -96,16 +103,25 @@ func (ns *NotificationService) sendWebRequestSync(ctx context.Context, webhook *
 	if webhook.Validation != nil {
 		err := webhook.Validation(body, resp.StatusCode)
 		if err != nil {
-			ns.log.Debug("Webhook failed validation", "url", webhook.Url, "statuscode", resp.Status, "body", string(body))
+			ns.log.Debug("Webhook failed validation", "url", url.Redacted(), "statuscode", resp.Status, "body", string(body), "error", err)
 			return fmt.Errorf("webhook failed validation: %w", err)
 		}
 	}
 
 	if resp.StatusCode/100 == 2 {
-		ns.log.Debug("Webhook succeeded", "url", webhook.Url, "statuscode", resp.Status)
+		ns.log.Debug("Webhook succeeded", "url", url.Redacted(), "statuscode", resp.Status)
 		return nil
 	}
 
-	ns.log.Debug("Webhook failed", "url", webhook.Url, "statuscode", resp.Status, "body", string(body))
+	ns.log.Debug("Webhook failed", "url", url.Redacted(), "statuscode", resp.Status, "body", string(body))
 	return fmt.Errorf("webhook response status %v", resp.Status)
 }
+
+func redactURL(err error) error {
+	var e *url.Error
+	if !errors.As(err, &e) {
+		return err
+	}
+	e.URL = "<redacted>"
+	return e
+}

pkg/tsdb/influxdb/flux/macros.go

@@ -25,27 +25,40 @@ func interpolateInterval(flux string, interval time.Duration) string {
 var fluxVariableFilterExp = regexp.MustCompile(`(?m)([a-zA-Z]+)\.([a-zA-Z]+)`)
 
 func interpolateFluxSpecificVariables(query queryModel) string {
+	rawQuery := query.RawQuery
 	flux := query.RawQuery
 
-	matches := fluxVariableFilterExp.FindAllStringSubmatch(flux, -1)
+	matches := fluxVariableFilterExp.FindAllStringSubmatchIndex(rawQuery, -1)
 	if matches != nil {
 		timeRange := query.TimeRange
 		from := timeRange.From.UTC().Format(time.RFC3339Nano)
 		to := timeRange.To.UTC().Format(time.RFC3339Nano)
 		for _, match := range matches {
-			switch match[2] {
+			// For query "range(start: v.timeRangeStart, stop: v.timeRangeStop)"
+			// rawQuery[match[0]:match[1]] will be v.timeRangeStart
+			// rawQuery[match[2]:match[3]] will be v
+			// rawQuery[match[4]:match[5]] will be timeRangeStart
+			fullMatch := rawQuery[match[0]:match[1]]
+			key := rawQuery[match[4]:match[5]]
+
+			switch key {
 			case "timeRangeStart":
-				flux = strings.ReplaceAll(flux, match[0], from)
+				flux = strings.ReplaceAll(flux, fullMatch, from)
 			case "timeRangeStop":
-				flux = strings.ReplaceAll(flux, match[0], to)
+				flux = strings.ReplaceAll(flux, fullMatch, to)
 			case "windowPeriod":
-				flux = strings.ReplaceAll(flux, match[0], query.Interval.String())
+				flux = strings.ReplaceAll(flux, fullMatch, query.Interval.String())
 			case "bucket":
-				flux = strings.ReplaceAll(flux, match[0], "\""+query.Options.Bucket+"\"")
+				// Check if 'bucket' is part of a join query
+				beforeMatch := rawQuery[:match[0]]
+				if strings.Contains(beforeMatch, "join.") {
+					continue
+				}
+				flux = strings.ReplaceAll(flux, fullMatch, "\""+query.Options.Bucket+"\"")
 			case "defaultBucket":
-				flux = strings.ReplaceAll(flux, match[0], "\""+query.Options.DefaultBucket+"\"")
+				flux = strings.ReplaceAll(flux, fullMatch, "\""+query.Options.DefaultBucket+"\"")
 			case "organization":
-				flux = strings.ReplaceAll(flux, match[0], "\""+query.Options.Organization+"\"")
+				flux = strings.ReplaceAll(flux, fullMatch, "\""+query.Options.Organization+"\"")
 			}
 		}
 	}

pkg/tsdb/influxdb/flux/macros_test.go

@@ -31,6 +31,11 @@ func TestInterpolate(t *testing.T) {
 			before: `v.timeRangeStart, something.timeRangeStop, XYZ.bucket, uuUUu.defaultBucket, aBcDefG.organization, window.windowPeriod, a91{}.bucket, $__interval, $__interval_ms`,
 			after:  `2021-09-22T10:12:51.310985041Z, 2021-09-22T11:12:51.310985042Z, "grafana2", "grafana3", "grafana1", 1m1.258s, a91{}.bucket, 1m, 61258`,
 		},
+		{
+			name:   "don't interpolate bucket variable in join query",
+			before: `range(start: v.timeRangeStart, stop: v.timeRangeStop) join.left(left: left |> group(), right: right,on:((l,r) => l.bucket == r.id), as: ((l, r) => ({l with name: r.name})))`,
+			after:  `range(start: 2021-09-22T10:12:51.310985041Z, stop: 2021-09-22T11:12:51.310985042Z) join.left(left: left |> group(), right: right,on:((l,r) => l.bucket == r.id), as: ((l, r) => ({l with name: r.name})))`,
+		},
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {