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

fix(unified-storage): resource server tracing (#113582)

Browse Source
This commit is contained in:
Jean-Philippe Quéméner
2025-11-07 11:51:32 +01:00
committed by GitHub
parent b4d2d1eaf5
commit 589435b7c2
5 changed files with 22 additions and 39 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/storage/unified/resource/bulk.go
+1 -1
View File
@@ -110,7 +110,7 @@ func NewBulkSettings(md metadata.MD) (BulkSettings, error) {
// All requests must be to the same NAMESPACE/GROUP/RESOURCE
func (s *server) BulkProcess(stream resourcepb.BulkStore_BulkProcessServer) error {
ctx := stream.Context()
ctx, span := s.tracer.Start(ctx, "resource.server.BulkProcess")
ctx, span := tracer.Start(ctx, "resource.server.BulkProcess")
defer span.End()
sendAndClose := func(rsp *resourcepb.BulkResponse) error {
pkg/storage/unified/resource/search.go
+6 -13
View File
@@ -127,11 +127,8 @@ type SearchBackend interface {
GetOpenIndexes() []NamespacedResource
}
const tracingPrexfixSearch = "unified_search."
// This supports indexing+search regardless of implementation
type searchSupport struct {
tracer trace.Tracer
log *slog.Logger
storage StorageBackend
search SearchBackend
@@ -163,14 +160,11 @@ var (
_ resourcepb.ManagedObjectIndexServer = (*searchSupport)(nil)
)
func newSearchSupport(opts SearchOptions, storage StorageBackend, access types.AccessClient, blob BlobSupport, tracer trace.Tracer, indexMetrics *BleveIndexMetrics, ownsIndexFn func(key NamespacedResource) (bool, error)) (support *searchSupport, err error) {
func newSearchSupport(opts SearchOptions, storage StorageBackend, access types.AccessClient, blob BlobSupport, indexMetrics *BleveIndexMetrics, ownsIndexFn func(key NamespacedResource) (bool, error)) (support *searchSupport, err error) {
// No backend search support
if opts.Backend == nil {
return nil, nil
}
if tracer == nil {
return nil, fmt.Errorf("missing tracer")
}
if opts.InitWorkerThreads < 1 {
opts.InitWorkerThreads = 1
@@ -188,7 +182,6 @@ func newSearchSupport(opts SearchOptions, storage StorageBackend, access types.A
support = &searchSupport{
access: access,
tracer: tracer,
storage: storage,
search: opts.Backend,
log: slog.Default().With("logger", "resource-search"),
@@ -341,7 +334,7 @@ func (s *searchSupport) CountManagedObjects(ctx context.Context, req *resourcepb
// Search implements ResourceIndexServer.
func (s *searchSupport) Search(ctx context.Context, req *resourcepb.ResourceSearchRequest) (*resourcepb.ResourceSearchResponse, error) {
ctx, span := s.tracer.Start(ctx, tracingPrexfixSearch+"Search")
ctx, span := tracer.Start(ctx, "resource.searchSupport.Search")
defer span.End()
if req.Options.Key.Namespace == "" || req.Options.Key.Group == "" || req.Options.Key.Resource == "" {
@@ -499,7 +492,7 @@ func (s *searchSupport) buildIndexes(ctx context.Context) (int, error) {
func (s *searchSupport) init(ctx context.Context) error {
origCtx := ctx
ctx, span := s.tracer.Start(ctx, tracingPrexfixSearch+"Init")
ctx, span := tracer.Start(ctx, "resource.searchSupport.init")
defer span.End()
start := time.Now().Unix()
@@ -632,7 +625,7 @@ func (s *searchSupport) runIndexRebuilder(ctx context.Context) {
}
func (s *searchSupport) rebuildIndex(ctx context.Context, req rebuildRequest) {
ctx, span := s.tracer.Start(ctx, tracingPrexfixSearch+"RebuildIndex")
ctx, span := tracer.Start(ctx, "resource.searchSupport.rebuildIndex")
defer span.End()
l := s.log.With("namespace", req.Namespace, "group", req.Group, "resource", req.Resource)
@@ -731,7 +724,7 @@ func (s *searchSupport) getOrCreateIndex(ctx context.Context, key NamespacedReso
return nil, fmt.Errorf("search is not configured properly (missing unifiedStorageSearch feature toggle?)")
}
ctx, span := s.tracer.Start(ctx, tracingPrexfixSearch+"GetOrCreateIndex")
ctx, span := tracer.Start(ctx, "resource.searchSupport.getOrCreateIndex")
defer span.End()
span.SetAttributes(
attribute.String("namespace", key.Namespace),
@@ -808,7 +801,7 @@ func (s *searchSupport) getOrCreateIndex(ctx context.Context, key NamespacedReso
}
func (s *searchSupport) build(ctx context.Context, nsr NamespacedResource, size int64, indexBuildReason string, rebuild bool) (ResourceIndex, error) {
ctx, span := s.tracer.Start(ctx, tracingPrexfixSearch+"Build")
ctx, span := tracer.Start(ctx, "resource.searchSupport.build")
defer span.End()
span.SetAttributes(
pkg/storage/unified/resource/search_test.go
+5 -6
View File
@@ -13,7 +13,6 @@ import (
"github.com/grafana/authlib/types"
"github.com/stretchr/testify/mock"
"github.com/stretchr/testify/require"
"go.opentelemetry.io/otel/trace/noop"
dashboardv1 "github.com/grafana/grafana/apps/dashboard/pkg/apis/dashboard/v1beta1"
"github.com/grafana/grafana/pkg/storage/unified/resourcepb"
@@ -211,7 +210,7 @@ func TestSearchGetOrCreateIndex(t *testing.T) {
InitMinCount: 1, // set min count to default for this test
}
support, err := newSearchSupport(opts, storage, nil, nil, noop.NewTracerProvider().Tracer("test"), nil, nil)
support, err := newSearchSupport(opts, storage, nil, nil, nil, nil)
require.NoError(t, err)
require.NotNil(t, support)
@@ -267,7 +266,7 @@ func TestSearchGetOrCreateIndexWithIndexUpdate(t *testing.T) {
}
// Enable searchAfterWrite
support, err := newSearchSupport(opts, storage, nil, nil, noop.NewTracerProvider().Tracer("test"), nil, nil)
support, err := newSearchSupport(opts, storage, nil, nil, nil, nil)
require.NoError(t, err)
require.NotNil(t, support)
@@ -316,7 +315,7 @@ func TestSearchGetOrCreateIndexWithCancellation(t *testing.T) {
InitMinCount: 1, // set min count to default for this test
}
support, err := newSearchSupport(opts, storage, nil, nil, noop.NewTracerProvider().Tracer("test"), nil, nil)
support, err := newSearchSupport(opts, storage, nil, nil, nil, nil)
require.NoError(t, err)
require.NotNil(t, support)
@@ -594,7 +593,7 @@ func TestFindIndexesForRebuild(t *testing.T) {
MinBuildVersion: semver.MustParse("5.5.5"),
}
support, err := newSearchSupport(opts, storage, nil, nil, noop.NewTracerProvider().Tracer("test"), nil, nil)
support, err := newSearchSupport(opts, storage, nil, nil, nil, nil)
require.NoError(t, err)
require.NotNil(t, support)
@@ -665,7 +664,7 @@ func TestRebuildIndexes(t *testing.T) {
Resources: supplier,
}
support, err := newSearchSupport(opts, storage, nil, nil, noop.NewTracerProvider().Tracer("test"), nil, nil)
support, err := newSearchSupport(opts, storage, nil, nil, nil, nil)
require.NoError(t, err)
require.NotNil(t, support)
pkg/storage/unified/resource/server.go
+10 -18
View File
@@ -14,8 +14,7 @@ import (
"github.com/Masterminds/semver"
"github.com/google/uuid"
"github.com/prometheus/client_golang/prometheus"
"go.opentelemetry.io/otel/trace"
"go.opentelemetry.io/otel/trace/noop"
"go.opentelemetry.io/otel"
apierrors "k8s.io/apimachinery/pkg/api/errors"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
@@ -30,6 +29,8 @@ import (
"github.com/grafana/grafana/pkg/util/scheduler"
)
var tracer = otel.Tracer("github.com/grafana/grafana/pkg/storage/unified/resource")
// ResourceServer implements all gRPC services
type ResourceServer interface {
resourcepb.ResourceStoreServer
@@ -210,9 +211,6 @@ type SearchOptions struct {
}
type ResourceServerOptions struct {
// OTel tracer
Tracer trace.Tracer
// Real storage backend
Backend StorageBackend
@@ -259,10 +257,6 @@ type ResourceServerOptions struct {
}
func NewResourceServer(opts ResourceServerOptions) (*server, error) {
if opts.Tracer == nil {
opts.Tracer = noop.NewTracerProvider().Tracer("resource-server")
}
if opts.Backend == nil {
return nil, fmt.Errorf("missing Backend implementation")
}
@@ -314,8 +308,8 @@ func NewResourceServer(opts ResourceServerOptions) (*server, error) {
}
blobstore, err = NewCDKBlobSupport(ctx, CDKBlobSupportOptions{
Tracer: opts.Tracer,
Bucket: NewInstrumentedBucket(bucket, opts.Reg, opts.Tracer),
Tracer: tracer,
Bucket: NewInstrumentedBucket(bucket, opts.Reg, tracer),
})
if err != nil {
return nil, err
@@ -331,7 +325,6 @@ func NewResourceServer(opts ResourceServerOptions) (*server, error) {
// Make this cancelable
ctx, cancel := context.WithCancel(context.Background())
s := &server{
tracer: opts.Tracer,
log: logger,
backend: opts.Backend,
blob: blobstore,
@@ -355,7 +348,7 @@ func NewResourceServer(opts ResourceServerOptions) (*server, error) {
if opts.Search.Resources != nil {
var err error
s.search, err = newSearchSupport(opts.Search, s.backend, s.access, s.blob, opts.Tracer, opts.IndexMetrics, opts.OwnsIndexFn)
s.search, err = newSearchSupport(opts.Search, s.backend, s.access, s.blob, opts.IndexMetrics, opts.OwnsIndexFn)
if err != nil {
return nil, err
}
@@ -373,7 +366,6 @@ func NewResourceServer(opts ResourceServerOptions) (*server, error) {
var _ ResourceServer = &server{}
type server struct {
tracer trace.Tracer
log *slog.Logger
backend StorageBackend
blob BlobSupport
@@ -651,7 +643,7 @@ func (s *server) checkFolderMovePermissions(ctx context.Context, user claims.Aut
}
func (s *server) Create(ctx context.Context, req *resourcepb.CreateRequest) (*resourcepb.CreateResponse, error) {
ctx, span := s.tracer.Start(ctx, "storage_server.Create")
ctx, span := tracer.Start(ctx, "resource.server.Create")
defer span.End()
if r := verifyRequestKey(req.Key); r != nil {
@@ -738,7 +730,7 @@ func (s *server) sleepAfterSuccessfulWriteOperation(res responseWithErrorResult,
}
func (s *server) Update(ctx context.Context, req *resourcepb.UpdateRequest) (*resourcepb.UpdateResponse, error) {
ctx, span := s.tracer.Start(ctx, "storage_server.Update")
ctx, span := tracer.Start(ctx, "resource.server.Update")
defer span.End()
rsp := &resourcepb.UpdateResponse{}
@@ -812,7 +804,7 @@ func (s *server) update(ctx context.Context, user claims.AuthInfo, req *resource
}
func (s *server) Delete(ctx context.Context, req *resourcepb.DeleteRequest) (*resourcepb.DeleteResponse, error) {
ctx, span := s.tracer.Start(ctx, "storage_server.Delete")
ctx, span := tracer.Start(ctx, "resource.server.Delete")
defer span.End()
rsp := &resourcepb.DeleteResponse{}
@@ -983,7 +975,7 @@ func (s *server) read(ctx context.Context, user claims.AuthInfo, req *resourcepb
}
func (s *server) List(ctx context.Context, req *resourcepb.ListRequest) (*resourcepb.ListResponse, error) {
ctx, span := s.tracer.Start(ctx, "storage_server.List")
ctx, span := tracer.Start(ctx, "resource.server.List")
defer span.End()
// The history + trash queries do not yet support additional filters
pkg/storage/unified/sql/server.go
-1
View File
@@ -62,7 +62,6 @@ func NewResourceServer(opts ServerOptions) (resource.ResourceServer, error) {
}
serverOptions := resource.ResourceServerOptions{
Tracer: opts.Tracer,
Blob: resource.BlobConfig{
URL: apiserverCfg.Key("blob_url").MustString(""),
},