public/grafana
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

chore(unified-storage): align how we do tracing (#114998)

Browse Source
This commit is contained in:
Jean-Philippe Quéméner
2025-12-09 14:53:53 +01:00
committed by GitHub
parent 3e66c7ed21
commit 1f5fd1c0da
22 changed files with 62 additions and 148 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/server/search_server_distributor_test.go
+1 -1
View File
@@ -389,7 +389,7 @@ func createBaselineServer(t *testing.T, dbType, dbConnStr string, testNamespaces
require.NoError(t, err)
tracer := noop.NewTracerProvider().Tracer("test-tracer")
require.NoError(t, err)
searchOpts, err := search.NewSearchOptions(features, cfg, tracer, docBuilders, nil, nil)
searchOpts, err := search.NewSearchOptions(features, cfg, docBuilders, nil, nil)
require.NoError(t, err)
server, err := sql.NewResourceServer(sql.ServerOptions{
DB: nil,
pkg/storage/unified/client.go
+1 -1
View File
@@ -168,7 +168,7 @@ func newClient(opts options.StorageOptions,
return resource.NewResourceClient(conn, indexConn, cfg, features, tracer)
default:
searchOptions, err := search.NewSearchOptions(features, cfg, tracer, docs, indexMetrics, nil)
searchOptions, err := search.NewSearchOptions(features, cfg, docs, indexMetrics, nil)
if err != nil {
return nil, err
}
pkg/storage/unified/resource/access.go
+2 -9
View File
@@ -10,7 +10,6 @@ import (
"go.opentelemetry.io/otel/attribute"
"go.opentelemetry.io/otel/codes"
"go.opentelemetry.io/otel/trace"
"go.opentelemetry.io/otel/trace/noop"
claims "github.com/grafana/authlib/types"
@@ -77,21 +76,16 @@ type authzLimitedClient struct {
// allowlist is a map of group to resources that are compatible with RBAC.
allowlist groupResource
logger log.Logger
tracer trace.Tracer
metrics *accessMetrics
}
type AuthzOptions struct {
Tracer trace.Tracer
Registry prometheus.Registerer
}
// NewAuthzLimitedClient creates a new authzLimitedClient.
func NewAuthzLimitedClient(client claims.AccessClient, opts AuthzOptions) claims.AccessClient {
logger := log.New("limited-authz-client")
if opts.Tracer == nil {
opts.Tracer = noop.NewTracerProvider().Tracer("limited-authz-client")
}
if opts.Registry == nil {
opts.Registry = prometheus.DefaultRegisterer
}
@@ -102,7 +96,6 @@ func NewAuthzLimitedClient(client claims.AccessClient, opts AuthzOptions) claims
"folder.grafana.app": map[string]interface{}{"folders": nil},
},
logger: logger,
tracer: opts.Tracer,
metrics: newMetrics(opts.Registry),
}
}
@@ -110,7 +103,7 @@ func NewAuthzLimitedClient(client claims.AccessClient, opts AuthzOptions) claims
// Check implements claims.AccessClient.
func (c authzLimitedClient) Check(ctx context.Context, id claims.AuthInfo, req claims.CheckRequest, folder string) (claims.CheckResponse, error) {
t := time.Now()
ctx, span := c.tracer.Start(ctx, "authzLimitedClient.Check", trace.WithAttributes(
ctx, span := tracer.Start(ctx, "resource.authzLimitedClient.Check", trace.WithAttributes(
attribute.String("group", req.Group),
attribute.String("resource", req.Resource),
attribute.String("namespace", req.Namespace),
@@ -163,7 +156,7 @@ func (c authzLimitedClient) Check(ctx context.Context, id claims.AuthInfo, req c
func (c authzLimitedClient) Compile(ctx context.Context, id claims.AuthInfo, req claims.ListRequest) (claims.ItemChecker, claims.Zookie, error) {
t := time.Now()
fallbackUsed := FallbackUsed(ctx)
ctx, span := c.tracer.Start(ctx, "authzLimitedClient.Compile", trace.WithAttributes(
ctx, span := tracer.Start(ctx, "resource.authzLimitedClient.Compile", trace.WithAttributes(
attribute.String("group", req.Group),
attribute.String("resource", req.Resource),
attribute.String("namespace", req.Namespace),
pkg/storage/unified/resource/cdk_blob.go
-9
View File
@@ -11,8 +11,6 @@ import (
"time"
"github.com/google/uuid"
"go.opentelemetry.io/otel/trace"
"go.opentelemetry.io/otel/trace/noop"
"gocloud.dev/blob"
"github.com/grafana/grafana/pkg/apimachinery/utils"
@@ -27,7 +25,6 @@ import (
)
type CDKBlobSupportOptions struct {
Tracer trace.Tracer
Bucket CDKBucket
RootFolder string
URLExpiration time.Duration
@@ -47,10 +44,6 @@ func OpenBlobBucket(ctx context.Context, url string) (*blob.Bucket, error) {
}
func NewCDKBlobSupport(ctx context.Context, opts CDKBlobSupportOptions) (BlobSupport, error) {
if opts.Tracer == nil {
opts.Tracer = noop.NewTracerProvider().Tracer("cdk-blob-store")
}
if opts.Bucket == nil {
return nil, fmt.Errorf("missing bucket")
}
@@ -70,7 +63,6 @@ func NewCDKBlobSupport(ctx context.Context, opts CDKBlobSupportOptions) (BlobSup
}
return &cdkBlobSupport{
tracer: opts.Tracer,
bucket: opts.Bucket,
root: opts.RootFolder,
cansignurls: false, // TODO depends on the implementation
@@ -79,7 +71,6 @@ func NewCDKBlobSupport(ctx context.Context, opts CDKBlobSupportOptions) (BlobSup
}
type cdkBlobSupport struct {
tracer trace.Tracer
bucket CDKBucket
root string
cansignurls bool
pkg/storage/unified/resource/cdk_bucket.go
+7 -10
View File
@@ -7,7 +7,6 @@ import (
"github.com/prometheus/client_golang/prometheus"
"github.com/prometheus/client_golang/prometheus/promauto"
"go.opentelemetry.io/otel/codes"
"go.opentelemetry.io/otel/trace"
"gocloud.dev/blob"
)
@@ -36,14 +35,12 @@ const (
type InstrumentedBucket struct {
requests *prometheus.CounterVec
latency *prometheus.HistogramVec
tracer trace.Tracer
bucket CDKBucket
}
func NewInstrumentedBucket(bucket CDKBucket, reg prometheus.Registerer, tracer trace.Tracer) *InstrumentedBucket {
func NewInstrumentedBucket(bucket CDKBucket, reg prometheus.Registerer) *InstrumentedBucket {
b := &InstrumentedBucket{
bucket: bucket,
tracer: tracer,
}
b.initMetrics(reg)
return b
@@ -66,7 +63,7 @@ func (b *InstrumentedBucket) initMetrics(reg prometheus.Registerer) {
}
func (b *InstrumentedBucket) Attributes(ctx context.Context, key string) (*blob.Attributes, error) {
ctx, span := b.tracer.Start(ctx, "InstrumentedBucket/Attributes")
ctx, span := tracer.Start(ctx, "resource.InstrumentedBucket.Attributes")
defer span.End()
start := time.Now()
retVal, err := b.bucket.Attributes(ctx, key)
@@ -94,7 +91,7 @@ func (b *InstrumentedBucket) List(opts *blob.ListOptions) *blob.ListIterator {
}
func (b *InstrumentedBucket) ListPage(ctx context.Context, pageToken []byte, pageSize int, opts *blob.ListOptions) ([]*blob.ListObject, []byte, error) {
ctx, span := b.tracer.Start(ctx, "InstrumentedBucket/ListPage")
ctx, span := tracer.Start(ctx, "resource.InstrumentedBucket.ListPage")
defer span.End()
start := time.Now()
retVal, nextPageToken, err := b.bucket.ListPage(ctx, pageToken, pageSize, opts)
@@ -116,7 +113,7 @@ func (b *InstrumentedBucket) ListPage(ctx context.Context, pageToken []byte, pag
}
func (b *InstrumentedBucket) ReadAll(ctx context.Context, key string) ([]byte, error) {
ctx, span := b.tracer.Start(ctx, "InstrumentedBucket/ReadAll")
ctx, span := tracer.Start(ctx, "resource.InstrumentedBucket.ReadAll")
defer span.End()
start := time.Now()
retVal, err := b.bucket.ReadAll(ctx, key)
@@ -139,7 +136,7 @@ func (b *InstrumentedBucket) ReadAll(ctx context.Context, key string) ([]byte, e
}
func (b *InstrumentedBucket) WriteAll(ctx context.Context, key string, p []byte, opts *blob.WriterOptions) error {
ctx, span := b.tracer.Start(ctx, "InstrumentedBucket/WriteAll")
ctx, span := tracer.Start(ctx, "resource.InstrumentedBucket.WriteAll")
defer span.End()
start := time.Now()
err := b.bucket.WriteAll(ctx, key, p, opts)
@@ -162,7 +159,7 @@ func (b *InstrumentedBucket) WriteAll(ctx context.Context, key string, p []byte,
}
func (b *InstrumentedBucket) Delete(ctx context.Context, key string) error {
ctx, span := b.tracer.Start(ctx, "InstrumentedBucket/Delete")
ctx, span := tracer.Start(ctx, "resource.InstrumentedBucket.Delete")
defer span.End()
start := time.Now()
err := b.bucket.Delete(ctx, key)
@@ -185,7 +182,7 @@ func (b *InstrumentedBucket) Delete(ctx context.Context, key string) error {
}
func (b *InstrumentedBucket) SignedURL(ctx context.Context, key string, opts *blob.SignedURLOptions) (string, error) {
ctx, span := b.tracer.Start(ctx, "InstrumentedBucket/SignedURL")
ctx, span := tracer.Start(ctx, "resource.InstrumentedBucket.SignedURL")
defer span.End()
start := time.Now()
retVal, err := b.bucket.SignedURL(ctx, key, opts)
pkg/storage/unified/resource/cdk_bucket_test.go
+1 -3
View File
@@ -8,7 +8,6 @@ import (
"github.com/prometheus/client_golang/prometheus"
"github.com/prometheus/client_golang/prometheus/testutil"
"github.com/stretchr/testify/require"
"go.opentelemetry.io/otel"
"gocloud.dev/blob"
)
@@ -196,8 +195,7 @@ func TestInstrumentedBucket(t *testing.T) {
t.Run(op.name+" "+tc.name, func(t *testing.T) {
fakeBucket := &fakeCDKBucket{}
reg := prometheus.NewPedanticRegistry()
tracer := otel.Tracer("test")
instrumentedBucket := NewInstrumentedBucket(fakeBucket, reg, tracer)
instrumentedBucket := NewInstrumentedBucket(fakeBucket, reg)
op.setup(fakeBucket, tc.success)
err := op.call(instrumentedBucket)
pkg/storage/unified/resource/server.go
+1 -2
View File
@@ -313,8 +313,7 @@ func NewResourceServer(opts ResourceServerOptions) (*server, error) {
}
blobstore, err = NewCDKBlobSupport(ctx, CDKBlobSupportOptions{
Tracer: tracer,
Bucket: NewInstrumentedBucket(bucket, opts.Reg, tracer),
Bucket: NewInstrumentedBucket(bucket, opts.Reg),
})
if err != nil {
return nil, err
pkg/storage/unified/search/bleve.go
+13 -18
View File
@@ -25,8 +25,8 @@ import (
bleveSearch "github.com/blevesearch/bleve/v2/search/searcher"
index "github.com/blevesearch/bleve_index_api"
"github.com/prometheus/client_golang/prometheus"
"go.opentelemetry.io/otel"
"go.opentelemetry.io/otel/attribute"
"go.opentelemetry.io/otel/trace"
"go.uber.org/atomic"
"k8s.io/apimachinery/pkg/selection"
@@ -42,9 +42,6 @@ import (
)
const (
// tracingPrexfixBleve is the prefix used for tracing spans in the Bleve backend
tracingPrexfixBleve = "unified_search.bleve."
indexStorageMemory = "memory"
indexStorageFile = "file"
)
@@ -55,6 +52,8 @@ const (
internalBuildInfoKey = "build_info" // Encoded as JSON of buildInfo struct
)
var tracer = otel.Tracer("github.com/grafana/grafana/pkg/storage/unified/search")
var _ resource.SearchBackend = &bleveBackend{}
var _ resource.ResourceIndex = &bleveIndex{}
@@ -83,9 +82,8 @@ type BleveOptions struct {
}
type bleveBackend struct {
tracer trace.Tracer
log log.Logger
opts BleveOptions
log log.Logger
opts BleveOptions
// set from opts.OwnsIndex, always non-nil
ownsIndexFn func(key resource.NamespacedResource) (bool, error)
@@ -99,7 +97,7 @@ type bleveBackend struct {
bgTasksWg sync.WaitGroup
}
func NewBleveBackend(opts BleveOptions, tracer trace.Tracer, indexMetrics *resource.BleveIndexMetrics) (*bleveBackend, error) {
func NewBleveBackend(opts BleveOptions, indexMetrics *resource.BleveIndexMetrics) (*bleveBackend, error) {
if opts.Root == "" {
return nil, fmt.Errorf("bleve backend missing root folder configuration")
}
@@ -138,7 +136,6 @@ func NewBleveBackend(opts BleveOptions, tracer trace.Tracer, indexMetrics *resou
be := &bleveBackend{
log: l,
tracer: tracer,
cache: map[resource.NamespacedResource]*bleveIndex{},
opts: opts,
ownsIndexFn: ownFn,
@@ -358,7 +355,7 @@ func (b *bleveBackend) BuildIndex(
updater resource.UpdateFn,
rebuild bool,
) (resource.ResourceIndex, error) {
_, span := b.tracer.Start(ctx, tracingPrexfixBleve+"BuildIndex")
_, span := tracer.Start(ctx, "search.bleveBackend.BuildIndex")
defer span.End()
span.SetAttributes(
@@ -713,7 +710,6 @@ type bleveIndex struct {
// The values returned with all
allFields []*resourcepb.ResourceTableColumnDefinition
tracing trace.Tracer
logger log.Logger
updaterFn resource.UpdateFn
@@ -750,7 +746,6 @@ func (b *bleveBackend) newBleveIndex(
fields: fields,
allFields: allFields,
standard: standardSearchFields,
tracing: b.tracer,
logger: logger,
updaterFn: updaterFn,
minUpdateInterval: b.opts.IndexMinUpdateInterval,
@@ -1018,7 +1013,7 @@ func (b *bleveIndex) Search(
federate []resource.ResourceIndex, // For federated queries, these will match the values in req.federate
stats *resource.SearchStats,
) (*resourcepb.ResourceSearchResponse, error) {
ctx, span := b.tracing.Start(ctx, tracingPrexfixBleve+"Search")
ctx, span := tracer.Start(ctx, "search.bleveIndex.Search")
defer span.End()
if req.Options == nil || req.Options.Key == nil {
@@ -1098,7 +1093,7 @@ func (b *bleveIndex) Search(
}
func (b *bleveIndex) DocCount(ctx context.Context, folder string, stats *resource.SearchStats) (int64, error) {
ctx, span := b.tracing.Start(ctx, tracingPrexfixBleve+"DocCount")
ctx, span := tracer.Start(ctx, "search.bleveIndex.DocCount")
defer span.End()
if folder == "" {
@@ -1144,7 +1139,7 @@ func (b *bleveIndex) getIndex(
req *resourcepb.ResourceSearchRequest,
federate []resource.ResourceIndex,
) (bleve.Index, error) {
_, span := b.tracing.Start(ctx, tracingPrexfixBleve+"getIndex")
_, span := tracer.Start(ctx, "search.bleveIndex.getIndex")
defer span.End()
if len(req.Federated) != len(federate) {
@@ -1171,7 +1166,7 @@ func (b *bleveIndex) getIndex(
}
func (b *bleveIndex) toBleveSearchRequest(ctx context.Context, req *resourcepb.ResourceSearchRequest, access authlib.AccessClient) (*bleve.SearchRequest, *resourcepb.ErrorResult) {
ctx, span := b.tracing.Start(ctx, tracingPrexfixBleve+"toBleveSearchRequest")
ctx, span := tracer.Start(ctx, "search.bleveIndex.toBleveSearchRequest")
defer span.End()
facets := bleve.FacetsRequest{}
@@ -1493,7 +1488,7 @@ func (b *bleveIndex) runUpdater(ctx context.Context) {
}
func (b *bleveIndex) updateIndexWithLatestModifications(ctx context.Context, requests int) (int64, error) {
ctx, span := b.tracing.Start(ctx, tracingPrexfixBleve+"updateIndexWithLatestModifications")
ctx, span := tracer.Start(ctx, "search.bleveIndex.updateIndexWithLatestModifications")
defer span.End()
sinceRV := b.resourceVersion
@@ -1691,7 +1686,7 @@ func filterValue(field string, v string) string {
}
func (b *bleveIndex) hitsToTable(ctx context.Context, selectFields []string, hits search.DocumentMatchCollection, explain bool) (*resourcepb.ResourceTable, error) {
_, span := b.tracing.Start(ctx, tracingPrexfixBleve+"hitsToTable")
_, span := tracer.Start(ctx, "search.bleveIndex.hitsToTable")
defer span.End()
fields := []*resourcepb.ResourceTableColumnDefinition{}
pkg/storage/unified/search/bleve_integration_test.go
+2 -3
View File
@@ -6,7 +6,6 @@ import (
"github.com/stretchr/testify/require"
"github.com/grafana/grafana/pkg/infra/tracing"
"github.com/grafana/grafana/pkg/storage/unified/resource"
unitest "github.com/grafana/grafana/pkg/storage/unified/testing"
)
@@ -20,7 +19,7 @@ func TestBleveSearchBackend(t *testing.T) {
backend, err := NewBleveBackend(BleveOptions{
Root: tempDir,
FileThreshold: 5,
}, tracing.NewNoopTracerService(), nil)
}, nil)
require.NoError(t, err)
require.NotNil(t, backend)
@@ -45,7 +44,7 @@ func TestSearchBackendBenchmark(t *testing.T) {
// Create a new bleve backend
backend, err := NewBleveBackend(BleveOptions{
Root: tempDir,
}, tracing.NewNoopTracerService(), nil)
}, nil)
require.NoError(t, err)
require.NotNil(t, backend)
pkg/storage/unified/search/bleve_search_test.go
+1 -2
View File
@@ -10,7 +10,6 @@ import (
"github.com/stretchr/testify/require"
"github.com/grafana/grafana/pkg/apimachinery/identity"
"github.com/grafana/grafana/pkg/infra/tracing"
"github.com/grafana/grafana/pkg/services/store/kind/dashboard"
"github.com/grafana/grafana/pkg/services/user"
"github.com/grafana/grafana/pkg/storage/unified/resource"
@@ -259,7 +258,7 @@ func newTestDashboardsIndex(t testing.TB, threshold int64, size int64, writer re
backend, err := search.NewBleveBackend(search.BleveOptions{
Root: t.TempDir(),
FileThreshold: threshold, // use in-memory for tests
}, tracing.NewNoopTracerService(), nil)
}, nil)
require.NoError(t, err)
t.Cleanup(backend.Stop)
pkg/storage/unified/search/bleve_test.go
+3 -4
View File
@@ -26,7 +26,6 @@ import (
"github.com/grafana/grafana/pkg/apimachinery/identity"
"github.com/grafana/grafana/pkg/apimachinery/utils"
"github.com/grafana/grafana/pkg/infra/log"
"github.com/grafana/grafana/pkg/infra/tracing"
authzextv1 "github.com/grafana/grafana/pkg/services/authz/proto/v1"
"github.com/grafana/grafana/pkg/services/store/kind/dashboard"
"github.com/grafana/grafana/pkg/services/user"
@@ -51,7 +50,7 @@ func TestBleveBackend(t *testing.T) {
backend, err := NewBleveBackend(BleveOptions{
Root: tmpdir,
FileThreshold: 5, // with more than 5 items we create a file on disk
}, tracing.NewNoopTracerService(), nil)
}, nil)
require.NoError(t, err)
t.Cleanup(backend.Stop)
@@ -782,7 +781,7 @@ func setupBleveBackend(t *testing.T, options ...setupOption) (*bleveBackend, pro
opts.Root = t.TempDir()
}
backend, err := NewBleveBackend(opts, tracing.NewNoopTracerService(), metrics)
backend, err := NewBleveBackend(opts, metrics)
require.NoError(t, err)
require.NotNil(t, backend)
t.Cleanup(backend.Stop)
@@ -1558,7 +1557,7 @@ func TestInvalidBuildVersion(t *testing.T) {
Root: t.TempDir(),
BuildVersion: "invalid",
}
_, err := NewBleveBackend(opts, tracing.NewNoopTracerService(), nil)
_, err := NewBleveBackend(opts, nil)
require.ErrorContains(t, err, "cannot parse build version")
}
pkg/storage/unified/search/options.go
+1 -3
View File
@@ -5,7 +5,6 @@ import (
"path/filepath"
"github.com/Masterminds/semver"
"go.opentelemetry.io/otel/trace"
"github.com/grafana/grafana/pkg/services/featuremgmt"
"github.com/grafana/grafana/pkg/setting"
@@ -15,7 +14,6 @@ import (
func NewSearchOptions(
features featuremgmt.FeatureToggles,
cfg *setting.Cfg,
tracer trace.Tracer,
docs resource.DocumentBuilderSupplier,
indexMetrics *resource.BleveIndexMetrics,
ownsIndexFn func(key resource.NamespacedResource) (bool, error),
@@ -48,7 +46,7 @@ func NewSearchOptions(
BuildVersion: cfg.BuildVersion,
OwnsIndex: ownsIndexFn,
IndexMinUpdateInterval: cfg.IndexMinUpdateInterval,
}, tracer, indexMetrics)
}, indexMetrics)
if err != nil {
return resource.SearchOptions{}, err
pkg/storage/unified/sql/backend.go
+19 -25
View File
@@ -14,9 +14,9 @@ import (
"github.com/jackc/pgx/v5/pgconn"
"github.com/lib/pq"
"github.com/prometheus/client_golang/prometheus"
"go.opentelemetry.io/otel"
"go.opentelemetry.io/otel/attribute"
"go.opentelemetry.io/otel/trace"
"go.opentelemetry.io/otel/trace/noop"
"go.uber.org/atomic"
"google.golang.org/protobuf/proto"
apierrors "k8s.io/apimachinery/pkg/api/errors"
@@ -35,7 +35,8 @@ import (
"github.com/grafana/grafana/pkg/util/debouncer"
)
const tracePrefix = "sql.resource."
var tracer = otel.Tracer("github.com/grafana/grafana/pkg/storage/unified/sql")
const defaultPollingInterval = 100 * time.Millisecond
const defaultWatchBufferSize = 100 // number of events to buffer in the watch stream
const defaultPrunerHistoryLimit = 20
@@ -56,7 +57,6 @@ type Backend interface {
type BackendOptions struct {
DBProvider db.DBProvider
Tracer trace.Tracer
Reg prometheus.Registerer
PollingInterval time.Duration
WatchBufferSize int
@@ -74,9 +74,6 @@ func NewBackend(opts BackendOptions) (Backend, error) {
if opts.DBProvider == nil {
return nil, errors.New("no db provider")
}
if opts.Tracer == nil {
opts.Tracer = noop.NewTracerProvider().Tracer("sql-backend")
}
ctx, cancel := context.WithCancel(context.Background())
if opts.PollingInterval == 0 {
@@ -90,7 +87,6 @@ func NewBackend(opts BackendOptions) (Backend, error) {
done: ctx.Done(),
cancel: cancel,
log: logging.DefaultLogger.With("logger", "sql-resource-server"),
tracer: opts.Tracer,
reg: opts.Reg,
dbProvider: opts.DBProvider,
pollingInterval: opts.PollingInterval,
@@ -114,7 +110,6 @@ type backend struct {
// o11y
log logging.Logger
tracer trace.Tracer
reg prometheus.Registerer
storageMetrics *resource.StorageMetrics
@@ -171,7 +166,6 @@ func (b *backend) initLocked(ctx context.Context) error {
rvManager, err := NewResourceVersionManager(ResourceManagerOptions{
Dialect: b.dialect,
DB: b.db,
Tracer: b.tracer,
})
if err != nil {
return fmt.Errorf("failed to create resource version manager: %w", err)
@@ -264,7 +258,7 @@ func (b *backend) Stop(_ context.Context) error {
// GetResourceStats implements Backend.
func (b *backend) GetResourceStats(ctx context.Context, nsr resource.NamespacedResource, minCount int) ([]resource.ResourceStats, error) {
ctx, span := b.tracer.Start(ctx, tracePrefix+"GetResourceStats", trace.WithAttributes(
ctx, span := tracer.Start(ctx, "sql.backend.GetResourceStats", trace.WithAttributes(
attribute.String("namespace", nsr.Namespace),
attribute.String("group", nsr.Group),
attribute.String("resource", nsr.Resource),
@@ -304,7 +298,7 @@ func (b *backend) GetResourceStats(ctx context.Context, nsr resource.NamespacedR
}
func (b *backend) WriteEvent(ctx context.Context, event resource.WriteEvent) (int64, error) {
_, span := b.tracer.Start(ctx, tracePrefix+"WriteEvent")
_, span := tracer.Start(ctx, "sql.backend.WriteEvent")
defer span.End()
// TODO: validate key ?
switch event.Type {
@@ -320,7 +314,7 @@ func (b *backend) WriteEvent(ctx context.Context, event resource.WriteEvent) (in
}
func (b *backend) create(ctx context.Context, event resource.WriteEvent) (int64, error) {
ctx, span := b.tracer.Start(ctx, tracePrefix+"Create")
ctx, span := tracer.Start(ctx, "sql.backend.create")
defer span.End()
folder := ""
@@ -408,7 +402,7 @@ func IsRowAlreadyExistsError(err error) bool {
}
func (b *backend) update(ctx context.Context, event resource.WriteEvent) (int64, error) {
ctx, span := b.tracer.Start(ctx, tracePrefix+"Update")
ctx, span := tracer.Start(ctx, "sql.backend.update")
defer span.End()
folder := ""
@@ -468,7 +462,7 @@ func (b *backend) update(ctx context.Context, event resource.WriteEvent) (int64,
}
func (b *backend) delete(ctx context.Context, event resource.WriteEvent) (int64, error) {
ctx, span := b.tracer.Start(ctx, tracePrefix+"Delete")
ctx, span := tracer.Start(ctx, "sql.backend.delete")
defer span.End()
folder := ""
@@ -547,7 +541,7 @@ func (b *backend) checkConflict(res db.Result, key *resourcepb.ResourceKey, rv i
}
func (b *backend) ReadResource(ctx context.Context, req *resourcepb.ReadRequest) *resource.BackendReadResponse {
_, span := b.tracer.Start(ctx, tracePrefix+".Read")
_, span := tracer.Start(ctx, "sql.backend.ReadResource")
defer span.End()
// TODO: validate key ?
@@ -580,7 +574,7 @@ func (b *backend) ReadResource(ctx context.Context, req *resourcepb.ReadRequest)
}
func (b *backend) ListIterator(ctx context.Context, req *resourcepb.ListRequest, cb func(resource.ListIterator) error) (int64, error) {
ctx, span := b.tracer.Start(ctx, tracePrefix+"List")
ctx, span := tracer.Start(ctx, "sql.backend.ListIterator")
defer span.End()
if err := resource.MigrateListRequestVersionMatch(req, b.log); err != nil {
@@ -602,7 +596,7 @@ func (b *backend) ListIterator(ctx context.Context, req *resourcepb.ListRequest,
}
func (b *backend) ListHistory(ctx context.Context, req *resourcepb.ListRequest, cb func(resource.ListIterator) error) (int64, error) {
ctx, span := b.tracer.Start(ctx, tracePrefix+"ListHistory")
ctx, span := tracer.Start(ctx, "sql.backend.ListHistory")
defer span.End()
return b.getHistory(ctx, req, cb)
@@ -610,7 +604,7 @@ func (b *backend) ListHistory(ctx context.Context, req *resourcepb.ListRequest,
// listLatest fetches the resources from the resource table.
func (b *backend) listLatest(ctx context.Context, req *resourcepb.ListRequest, cb func(resource.ListIterator) error) (int64, error) {
ctx, span := b.tracer.Start(ctx, tracePrefix+"listLatest")
ctx, span := tracer.Start(ctx, "sql.backend.listLatest")
defer span.End()
if req.NextPageToken != "" {
@@ -724,7 +718,7 @@ func (b *backend) ListModifiedSince(ctx context.Context, key resource.Namespaced
// listAtRevision fetches the resources from the resource_history table at a specific revision.
func (b *backend) listAtRevision(ctx context.Context, req *resourcepb.ListRequest, cb func(resource.ListIterator) error) (int64, error) {
ctx, span := b.tracer.Start(ctx, tracePrefix+"listAtRevision")
ctx, span := tracer.Start(ctx, "sql.backend.listAtRevision")
defer span.End()
// Get the RV
@@ -784,7 +778,7 @@ func (b *backend) listAtRevision(ctx context.Context, req *resourcepb.ListReques
// readHistory fetches the resource history from the resource_history table.
func (b *backend) readHistory(ctx context.Context, key *resourcepb.ResourceKey, rv int64) *resource.BackendReadResponse {
_, span := b.tracer.Start(ctx, tracePrefix+".ReadHistory")
_, span := tracer.Start(ctx, "sql.backend.readHistory")
defer span.End()
readReq := &sqlResourceHistoryReadRequest{
@@ -815,7 +809,7 @@ func (b *backend) readHistory(ctx context.Context, key *resourcepb.ResourceKey,
// getHistory fetches the resource history from the resource_history table.
func (b *backend) getHistory(ctx context.Context, req *resourcepb.ListRequest, cb func(resource.ListIterator) error) (int64, error) {
ctx, span := b.tracer.Start(ctx, tracePrefix+"getHistory")
ctx, span := tracer.Start(ctx, "sql.backend.getHistory")
defer span.End()
listReq := sqlGetHistoryRequest{
SQLTemplate: sqltemplate.New(b.dialect),
@@ -903,7 +897,7 @@ func (b *backend) WatchWriteEvents(ctx context.Context) (<-chan *resource.Writte
// listLatestRVs returns the latest resource version for each (Group, Resource) pair.
func (b *backend) listLatestRVs(ctx context.Context) (groupResourceRV, error) {
ctx, span := b.tracer.Start(ctx, tracePrefix+"listLatestRVs")
ctx, span := tracer.Start(ctx, "sql.backend.listLatestRVs")
defer span.End()
var grvs []*groupResourceVersion
err := b.db.WithTx(ctx, ReadCommittedRO, func(ctx context.Context, tx db.Tx) error {
@@ -932,7 +926,7 @@ func (b *backend) listLatestRVs(ctx context.Context) (groupResourceRV, error) {
// fetchLatestRV returns the current maximum RV in the resource table
func (b *backend) fetchLatestRV(ctx context.Context, x db.ContextExecer, d sqltemplate.Dialect, group, resource string) (int64, error) {
ctx, span := b.tracer.Start(ctx, tracePrefix+"fetchLatestRV")
ctx, span := tracer.Start(ctx, "sql.backend.fetchLatestRV")
defer span.End()
res, err := dbutil.QueryRow(ctx, x, sqlResourceVersionGet, sqlResourceVersionGetRequest{
SQLTemplate: sqltemplate.New(d),
@@ -951,7 +945,7 @@ func (b *backend) fetchLatestRV(ctx context.Context, x db.ContextExecer, d sqlte
// fetchLatestHistoryRV returns the current maximum RV in the resource_history table
func (b *backend) fetchLatestHistoryRV(ctx context.Context, x db.ContextExecer, d sqltemplate.Dialect, key *resourcepb.ResourceKey, eventType resourcepb.WatchEvent_Type) (int64, error) {
ctx, span := b.tracer.Start(ctx, tracePrefix+"fetchLatestHistoryRV")
ctx, span := tracer.Start(ctx, "sql.backend.fetchLatestHistoryRV")
defer span.End()
res, err := dbutil.QueryRow(ctx, x, sqlResourceHistoryReadLatestRV, sqlResourceHistoryReadLatestRVRequest{
SQLTemplate: sqltemplate.New(d),
@@ -973,7 +967,7 @@ func (b *backend) fetchLatestHistoryRV(ctx context.Context, x db.ContextExecer,
const limitLastImportTimesDeletion = 1 * time.Hour
func (b *backend) GetResourceLastImportTimes(ctx context.Context) iter.Seq2[resource.ResourceLastImportTime, error] {
ctx, span := b.tracer.Start(ctx, tracePrefix+"GetLastImportTimes")
ctx, span := tracer.Start(ctx, "sql.backend.GetResourceLastImportTimes")
defer span.End()
// Delete old entries, if configured, and if enough time has passed since last deletion.
pkg/storage/unified/sql/blob.go
+2 -2
View File
@@ -27,7 +27,7 @@ func (b *backend) SupportsSignedURLs() bool {
}
func (b *backend) PutResourceBlob(ctx context.Context, req *resourcepb.PutBlobRequest) (*resourcepb.PutBlobResponse, error) {
ctx, span := b.tracer.Start(ctx, tracePrefix+"PutResourceBlob")
ctx, span := tracer.Start(ctx, "sql.backend.PutResourceBlob")
defer span.End()
if req.Method == resourcepb.PutBlobRequest_HTTP {
@@ -83,7 +83,7 @@ func (b *backend) PutResourceBlob(ctx context.Context, req *resourcepb.PutBlobRe
}
func (b *backend) GetResourceBlob(ctx context.Context, key *resourcepb.ResourceKey, info *utils.BlobInfo, mustProxy bool) (*resourcepb.GetBlobResponse, error) {
ctx, span := b.tracer.Start(ctx, tracePrefix+"GetResourceBlob")
ctx, span := tracer.Start(ctx, "sql.backend.GetResourceBlob")
defer span.End()
if info == nil {
pkg/storage/unified/sql/notifier.go
-1
View File
@@ -28,7 +28,6 @@ func newNotifier(b *backend) (eventNotifier, error) {
pollingInterval: b.pollingInterval,
watchBufferSize: b.watchBufferSize,
log: b.log,
tracer: b.tracer,
bulkLock: b.bulkLock,
listLatestRVs: b.listLatestRVs,
storageMetrics: b.storageMetrics,
pkg/storage/unified/sql/notifier_sql.go
+2 -11
View File
@@ -5,8 +5,6 @@ import (
"fmt"
"time"
"go.opentelemetry.io/otel/trace"
"github.com/grafana/grafana-app-sdk/logging"
"github.com/grafana/grafana/pkg/storage/unified/resource"
@@ -19,7 +17,6 @@ var (
errHistoryPollRequired = fmt.Errorf("historyPoll is required")
errListLatestRVsRequired = fmt.Errorf("listLatestRVs is required")
errBulkLockRequired = fmt.Errorf("bulkLock is required")
errTracerRequired = fmt.Errorf("tracer is required")
errLogRequired = fmt.Errorf("log is required")
errInvalidWatchBufferSize = fmt.Errorf("watchBufferSize must be greater than 0")
errInvalidPollingInterval = fmt.Errorf("pollingInterval must be greater than 0")
@@ -34,7 +31,6 @@ type pollingNotifier struct {
watchBufferSize int
log logging.Logger
tracer trace.Tracer
storageMetrics *resource.StorageMetrics
bulkLock *bulkLock
@@ -50,7 +46,6 @@ type pollingNotifierConfig struct {
watchBufferSize int
log logging.Logger
tracer trace.Tracer
storageMetrics *resource.StorageMetrics
bulkLock *bulkLock
@@ -70,9 +65,6 @@ func (cfg *pollingNotifierConfig) validate() error {
if cfg.bulkLock == nil {
return errBulkLockRequired
}
if cfg.tracer == nil {
return errTracerRequired
}
if cfg.log == nil {
return errLogRequired
}
@@ -100,7 +92,6 @@ func newPollingNotifier(cfg *pollingNotifierConfig) (*pollingNotifier, error) {
pollingInterval: cfg.pollingInterval,
watchBufferSize: cfg.watchBufferSize,
log: cfg.log,
tracer: cfg.tracer,
bulkLock: cfg.bulkLock,
listLatestRVs: cfg.listLatestRVs,
historyPoll: cfg.historyPoll,
@@ -131,7 +122,7 @@ func (p *pollingNotifier) poller(ctx context.Context, since groupResourceRV, str
case <-p.done:
return
case <-t.C:
ctx, span := p.tracer.Start(ctx, tracePrefix+"poller")
ctx, span := tracer.Start(ctx, "sql.pollingNotifier.poller")
// List the latest RVs to see if any of those are not have been seen before.
grv, err := p.listLatestRVs(ctx)
if err != nil {
@@ -174,7 +165,7 @@ func (p *pollingNotifier) poller(ctx context.Context, since groupResourceRV, str
}
func (p *pollingNotifier) poll(ctx context.Context, grp string, res string, since int64, stream chan<- *resource.WrittenEvent) (int64, error) {
ctx, span := p.tracer.Start(ctx, tracePrefix+"poll")
ctx, span := tracer.Start(ctx, "sql.pollingNotifier.poll")
defer span.End()
start := time.Now()
pkg/storage/unified/sql/notifier_sql_test.go
-30
View File
@@ -8,7 +8,6 @@ import (
"github.com/grafana/grafana-app-sdk/logging"
"github.com/stretchr/testify/require"
"go.opentelemetry.io/otel/trace/noop"
"github.com/grafana/grafana/pkg/storage/unified/resourcepb"
"github.com/grafana/grafana/pkg/storage/unified/sql/sqltemplate"
@@ -30,7 +29,6 @@ func TestPollingNotifierConfig(t *testing.T) {
},
listLatestRVs: func(ctx context.Context) (groupResourceRV, error) { return nil, nil },
bulkLock: &bulkLock{},
tracer: noop.NewTracerProvider().Tracer("test"),
log: &logging.NoOpLogger{},
watchBufferSize: 10,
pollingInterval: time.Second,
@@ -44,7 +42,6 @@ func TestPollingNotifierConfig(t *testing.T) {
config: &pollingNotifierConfig{
listLatestRVs: func(ctx context.Context) (groupResourceRV, error) { return nil, nil },
bulkLock: &bulkLock{},
tracer: noop.NewTracerProvider().Tracer("test"),
log: &logging.NoOpLogger{},
watchBufferSize: 10,
pollingInterval: time.Second,
@@ -60,7 +57,6 @@ func TestPollingNotifierConfig(t *testing.T) {
return nil, nil
},
bulkLock: &bulkLock{},
tracer: noop.NewTracerProvider().Tracer("test"),
log: &logging.NoOpLogger{},
watchBufferSize: 10,
pollingInterval: time.Second,
@@ -76,7 +72,6 @@ func TestPollingNotifierConfig(t *testing.T) {
return nil, nil
},
listLatestRVs: func(ctx context.Context) (groupResourceRV, error) { return nil, nil },
tracer: noop.NewTracerProvider().Tracer("test"),
log: &logging.NoOpLogger{},
watchBufferSize: 10,
pollingInterval: time.Second,
@@ -85,22 +80,6 @@ func TestPollingNotifierConfig(t *testing.T) {
},
expectedErr: errBulkLockRequired,
},
{
name: "missing tracer",
config: &pollingNotifierConfig{
historyPoll: func(ctx context.Context, grp string, res string, since int64) ([]*historyPollResponse, error) {
return nil, nil
},
listLatestRVs: func(ctx context.Context) (groupResourceRV, error) { return nil, nil },
bulkLock: &bulkLock{},
log: &logging.NoOpLogger{},
watchBufferSize: 10,
pollingInterval: time.Second,
done: make(chan struct{}),
dialect: sqltemplate.SQLite,
},
expectedErr: errTracerRequired,
},
{
name: "missing logger",
config: &pollingNotifierConfig{
@@ -109,7 +88,6 @@ func TestPollingNotifierConfig(t *testing.T) {
},
listLatestRVs: func(ctx context.Context) (groupResourceRV, error) { return nil, nil },
bulkLock: &bulkLock{},
tracer: noop.NewTracerProvider().Tracer("test"),
watchBufferSize: 10,
pollingInterval: time.Second,
done: make(chan struct{}),
@@ -125,7 +103,6 @@ func TestPollingNotifierConfig(t *testing.T) {
},
listLatestRVs: func(ctx context.Context) (groupResourceRV, error) { return nil, nil },
bulkLock: &bulkLock{},
tracer: noop.NewTracerProvider().Tracer("test"),
log: &logging.NoOpLogger{},
watchBufferSize: 0,
pollingInterval: time.Second,
@@ -142,7 +119,6 @@ func TestPollingNotifierConfig(t *testing.T) {
},
listLatestRVs: func(ctx context.Context) (groupResourceRV, error) { return nil, nil },
bulkLock: &bulkLock{},
tracer: noop.NewTracerProvider().Tracer("test"),
log: &logging.NoOpLogger{},
watchBufferSize: 10,
pollingInterval: 0,
@@ -159,7 +135,6 @@ func TestPollingNotifierConfig(t *testing.T) {
},
listLatestRVs: func(ctx context.Context) (groupResourceRV, error) { return nil, nil },
bulkLock: &bulkLock{},
tracer: noop.NewTracerProvider().Tracer("test"),
log: &logging.NoOpLogger{},
watchBufferSize: 10,
pollingInterval: time.Second,
@@ -175,7 +150,6 @@ func TestPollingNotifierConfig(t *testing.T) {
},
listLatestRVs: func(ctx context.Context) (groupResourceRV, error) { return nil, nil },
bulkLock: &bulkLock{},
tracer: noop.NewTracerProvider().Tracer("test"),
log: &logging.NoOpLogger{},
watchBufferSize: 10,
pollingInterval: time.Second,
@@ -255,7 +229,6 @@ func TestPollingNotifier(t *testing.T) {
pollingInterval: 10 * time.Millisecond,
watchBufferSize: 10,
log: &logging.NoOpLogger{},
tracer: noop.NewTracerProvider().Tracer("test"),
bulkLock: &bulkLock{},
listLatestRVs: listLatestRVs,
historyPoll: historyPoll,
@@ -309,7 +282,6 @@ func TestPollingNotifier(t *testing.T) {
pollingInterval: 10 * time.Millisecond,
watchBufferSize: 10,
log: &logging.NoOpLogger{},
tracer: noop.NewTracerProvider().Tracer("test"),
bulkLock: &bulkLock{},
listLatestRVs: listLatestRVs,
historyPoll: historyPoll,
@@ -343,7 +315,6 @@ func TestPollingNotifier(t *testing.T) {
pollingInterval: 10 * time.Millisecond,
watchBufferSize: 10,
log: &logging.NoOpLogger{},
tracer: noop.NewTracerProvider().Tracer("test"),
bulkLock: &bulkLock{},
listLatestRVs: func(ctx context.Context) (groupResourceRV, error) { return nil, nil },
historyPoll: func(ctx context.Context, grp string, res string, since int64) ([]*historyPollResponse, error) {
@@ -380,7 +351,6 @@ func TestPollingNotifier(t *testing.T) {
pollingInterval: 10 * time.Millisecond,
watchBufferSize: 10,
log: &logging.NoOpLogger{},
tracer: noop.NewTracerProvider().Tracer("test"),
bulkLock: &bulkLock{},
listLatestRVs: func(ctx context.Context) (groupResourceRV, error) { return nil, nil },
historyPoll: func(ctx context.Context, grp string, res string, since int64) ([]*historyPollResponse, error) {
pkg/storage/unified/sql/rv_manager.go
+2 -9
View File
@@ -12,7 +12,6 @@ import (
"github.com/prometheus/client_golang/prometheus/promauto"
"go.opentelemetry.io/otel/attribute"
"go.opentelemetry.io/otel/trace"
"go.opentelemetry.io/otel/trace/noop"
"github.com/grafana/grafana/pkg/storage/unified/resourcepb"
"github.com/grafana/grafana/pkg/storage/unified/sql/db"
@@ -66,7 +65,6 @@ const (
type resourceVersionManager struct {
dialect sqltemplate.Dialect
db db.DB
tracer trace.Tracer
batchMu sync.RWMutex
batchChMap map[string]chan *writeOp
@@ -98,7 +96,6 @@ type ResourceManagerOptions struct {
DB db.DB // The database to use
MaxBatchSize int // The maximum number of operations to batch together
MaxBatchWaitTime time.Duration // The maximum time to wait for a batch to be ready
Tracer trace.Tracer // The tracer to use for tracing
}
// NewResourceVersionManager creates a new ResourceVersionManager
@@ -109,9 +106,6 @@ func NewResourceVersionManager(opts ResourceManagerOptions) (*resourceVersionMan
if opts.MaxBatchWaitTime == 0 {
opts.MaxBatchWaitTime = defaultMaxBatchWaitTime
}
if opts.Tracer == nil {
opts.Tracer = noop.NewTracerProvider().Tracer("resource-version-manager")
}
if opts.Dialect == nil {
return nil, errors.New("dialect is required")
}
@@ -121,7 +115,6 @@ func NewResourceVersionManager(opts ResourceManagerOptions) (*resourceVersionMan
return &resourceVersionManager{
dialect: opts.Dialect,
db: opts.DB,
tracer: opts.Tracer,
batchChMap: make(map[string]chan *writeOp),
maxBatchSize: opts.MaxBatchSize,
maxBatchWaitTime: opts.MaxBatchWaitTime,
@@ -143,7 +136,7 @@ func (m *resourceVersionManager) ExecWithRV(ctx context.Context, key *resourcepb
}))
defer timer.ObserveDuration()
ctx, span := m.tracer.Start(ctx, "sql.rvmanager.ExecWithRV")
ctx, span := tracer.Start(ctx, "sql.resourceVersionManager.ExecWithRV")
defer span.End()
span.SetAttributes(
@@ -223,7 +216,7 @@ func (m *resourceVersionManager) startBatchProcessor(group, resource string) {
}
func (m *resourceVersionManager) execBatch(ctx context.Context, group, resource string, batch []writeOp) {
ctx, span := m.tracer.Start(ctx, "sql.rvmanager.execBatch")
ctx, span := tracer.Start(ctx, "sql.resourceVersionManager.execBatch")
defer span.End()
// Add batch size attribute
pkg/storage/unified/sql/search.go
+1 -1
View File
@@ -18,7 +18,7 @@ var _ resourcepb.ResourceIndexServer = &backend{}
// GetStats implements resource.ResourceIndexServer.
// This will use the SQL index to count values
func (b *backend) GetStats(ctx context.Context, req *resourcepb.ResourceStatsRequest) (*resourcepb.ResourceStatsResponse, error) {
ctx, span := b.tracer.Start(ctx, tracePrefix+"GetStats")
ctx, span := tracer.Start(ctx, "sql.backend.GetStats")
defer span.End()
sreq := &sqlStatsRequest{
pkg/storage/unified/sql/server.go
+1 -2
View File
@@ -70,7 +70,7 @@ func NewResourceServer(opts ServerOptions) (resource.ResourceServer, error) {
SecureValues: opts.SecureValues,
}
if opts.AccessClient != nil {
serverOptions.AccessClient = resource.NewAuthzLimitedClient(opts.AccessClient, resource.AuthzOptions{Tracer: opts.Tracer, Registry: opts.Reg})
serverOptions.AccessClient = resource.NewAuthzLimitedClient(opts.AccessClient, resource.AuthzOptions{Registry: opts.Reg})
}
// Support local file blob
if strings.HasPrefix(serverOptions.Blob.URL, "./data/") {
@@ -102,7 +102,6 @@ func NewResourceServer(opts ServerOptions) (resource.ResourceServer, error) {
backend, err := NewBackend(BackendOptions{
DBProvider: eDB,
Tracer: opts.Tracer,
Reg: opts.Reg,
IsHA: isHA,
storageMetrics: opts.StorageMetrics,
pkg/storage/unified/sql/service.go
+1 -1
View File
@@ -260,7 +260,7 @@ func (s *service) starting(ctx context.Context) error {
return err
}
searchOptions, err := search.NewSearchOptions(s.features, s.cfg, s.tracing, s.docBuilders, s.indexMetrics, s.OwnsIndex)
searchOptions, err := search.NewSearchOptions(s.features, s.cfg, s.docBuilders, s.indexMetrics, s.OwnsIndex)
if err != nil {
return err
}
pkg/storage/unified/sql/test/integration_test.go
+1 -1
View File
@@ -104,7 +104,7 @@ func TestIntegrationSearchAndStorage(t *testing.T) {
search, err := search.NewBleveBackend(search.BleveOptions{
FileThreshold: 0,
Root: t.TempDir(),
}, tracing.NewNoopTracerService(), nil)
}, nil)
require.NoError(t, err)
require.NotNil(t, search)
t.Cleanup(search.Stop)