From fd14d4a5ed3ad8dfd5948a9f2d7b9074e15ea655 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jean-Philippe=20Qu=C3=A9m=C3=A9ner?= Date: Thu, 6 Nov 2025 12:42:46 +0100 Subject: [PATCH] feat(unified-storage): add tracing to dual writer and legacy storage (#113504) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-authored-by: Mustafa Sencer Özcan <32759850+mustafasencer@users.noreply.github.com> --- .../apis/dashboard/legacy/sql_dashboards.go | 19 +++- pkg/registry/apis/dashboard/legacy/storage.go | 30 +++++ pkg/storage/legacysql/dualwrite/dualwriter.go | 107 ++++++++++++------ 3 files changed, 121 insertions(+), 35 deletions(-) diff --git a/pkg/registry/apis/dashboard/legacy/sql_dashboards.go b/pkg/registry/apis/dashboard/legacy/sql_dashboards.go index 5676566fd1a..0fca282fa82 100644 --- a/pkg/registry/apis/dashboard/legacy/sql_dashboards.go +++ b/pkg/registry/apis/dashboard/legacy/sql_dashboards.go @@ -10,6 +10,7 @@ import ( "sync" "time" + "go.opentelemetry.io/otel" apierrors "k8s.io/apimachinery/pkg/api/errors" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/utils/ptr" @@ -40,7 +41,8 @@ import ( ) var ( - _ DashboardAccess = (*dashboardSqlAccess)(nil) + _ DashboardAccess = (*dashboardSqlAccess)(nil) + tracer = otel.Tracer("github.com/grafana/grafana/pkg/registry/apis/dashboard/legacy") ) type dashboardRow struct { @@ -105,6 +107,9 @@ func NewDashboardAccess(sql legacysql.LegacyDatabaseProvider, } func (a *dashboardSqlAccess) getRows(ctx context.Context, sql *legacysql.LegacyDatabaseHelper, query *DashboardQuery) (*rowsWrapper, error) { + ctx, span := tracer.Start(ctx, "legacy.dashboardSqlAccess.getRows") + defer span.End() + if len(query.Labels) > 0 { return nil, fmt.Errorf("labels not yet supported") // if query.Requirements.Folder != nil { @@ -416,6 +421,9 @@ func getUserID(v sql.NullString, id sql.NullInt64) string { // DeleteDashboard implements DashboardAccess. func (a *dashboardSqlAccess) DeleteDashboard(ctx context.Context, orgId int64, uid string) (*dashboardV1.Dashboard, bool, error) { + ctx, span := tracer.Start(ctx, "legacy.dashboardSqlAccess.DeleteDashboard") + defer span.End() + dash, _, err := a.GetDashboard(ctx, orgId, uid, 0) if err != nil { return nil, false, err @@ -432,6 +440,9 @@ func (a *dashboardSqlAccess) DeleteDashboard(ctx context.Context, orgId int64, u } func (a *dashboardSqlAccess) buildSaveDashboardCommand(ctx context.Context, orgId int64, dash *dashboardV1.Dashboard) (*dashboards.SaveDashboardCommand, bool, error) { + ctx, span := tracer.Start(ctx, "legacy.dashboardSqlAccess.buildSaveDashboardCommand") + defer span.End() + created := false user, ok := claims.AuthInfoFrom(ctx) if !ok || user == nil { @@ -495,6 +506,9 @@ func (a *dashboardSqlAccess) buildSaveDashboardCommand(ctx context.Context, orgI } func (a *dashboardSqlAccess) SaveDashboard(ctx context.Context, orgId int64, dash *dashboardV1.Dashboard, failOnExisting bool) (*dashboardV1.Dashboard, bool, error) { + ctx, span := tracer.Start(ctx, "legacy.dashboardSqlAccess.SaveDashboard") + defer span.End() + user, ok := claims.AuthInfoFrom(ctx) if !ok || user == nil { return nil, false, fmt.Errorf("no user found in context") @@ -565,6 +579,9 @@ type panel struct { } func (a *dashboardSqlAccess) GetLibraryPanels(ctx context.Context, query LibraryPanelQuery) (*dashboardV0.LibraryPanelList, error) { + ctx, span := tracer.Start(ctx, "legacy.dashboardSqlAccess.GetLibraryPanels") + defer span.End() + limit := int(query.Limit) query.Limit += 1 // for continue if query.OrgID == 0 { diff --git a/pkg/registry/apis/dashboard/legacy/storage.go b/pkg/registry/apis/dashboard/legacy/storage.go index d5947ab24f5..d8c948c9664 100644 --- a/pkg/registry/apis/dashboard/legacy/storage.go +++ b/pkg/registry/apis/dashboard/legacy/storage.go @@ -78,6 +78,9 @@ func isDashboardKey(key *resourcepb.ResourceKey, requireName bool) error { } func (a *dashboardSqlAccess) WriteEvent(ctx context.Context, event resource.WriteEvent) (rv int64, err error) { + ctx, span := tracer.Start(ctx, "legacy.dashboardSqlAccess.WriteEvent") + defer span.End() + info, err := claims.ParseNamespace(event.Key.Namespace) if err == nil { err = isDashboardKey(event.Key, true) @@ -170,6 +173,9 @@ func (a *dashboardSqlAccess) WriteEvent(ctx context.Context, event resource.Writ } func (a *dashboardSqlAccess) GetDashboard(ctx context.Context, orgId int64, uid string, v int64) (*dashboard.Dashboard, int64, error) { + ctx, span := tracer.Start(ctx, "legacy.dashboardSqlAccess.GetDashboard") + defer span.End() + sql, err := a.sql(ctx) if err != nil { return nil, 0, err @@ -197,6 +203,9 @@ func (a *dashboardSqlAccess) GetDashboard(ctx context.Context, orgId int64, uid // Read implements ResourceStoreServer. func (a *dashboardSqlAccess) ReadResource(ctx context.Context, req *resourcepb.ReadRequest) *resource.BackendReadResponse { + ctx, span := tracer.Start(ctx, "legacy.dashboardSqlAccess.ReadResource") + defer span.End() + rsp := &resource.BackendReadResponse{} info, err := claims.ParseNamespace(req.Key.Namespace) if err == nil { @@ -238,16 +247,25 @@ func (a *dashboardSqlAccess) ReadResource(ctx context.Context, req *resourcepb.R // ListHistory implements StorageBackend. func (a *dashboardSqlAccess) ListHistory(ctx context.Context, req *resourcepb.ListRequest, cb func(resource.ListIterator) error) (int64, error) { + ctx, span := tracer.Start(ctx, "legacy.dashboardSqlAccess.ListHistory") + defer span.End() + return a.ListIterator(ctx, req, cb) } func (a *dashboardSqlAccess) ListModifiedSince(ctx context.Context, key resource.NamespacedResource, sinceRv int64) (int64, iter.Seq2[*resource.ModifiedResource, error]) { + _, span := tracer.Start(ctx, "legacy.dashboardSqlAccess.ListModifiedSince") + defer span.End() + return 0, func(yield func(*resource.ModifiedResource, error) bool) { yield(nil, errors.New("not implemented")) } } func (a *dashboardSqlAccess) GetResourceLastImportTimes(ctx context.Context) iter.Seq2[resource.ResourceLastImportTime, error] { + _, span := tracer.Start(ctx, "legacy.dashboardSqlAccess.GetResourceLastImportTimes") + defer span.End() + return func(yield func(resource.ResourceLastImportTime, error) bool) { yield(resource.ResourceLastImportTime{}, errors.New("not implemented")) } @@ -255,6 +273,9 @@ func (a *dashboardSqlAccess) GetResourceLastImportTimes(ctx context.Context) ite // List implements StorageBackend. func (a *dashboardSqlAccess) ListIterator(ctx context.Context, req *resourcepb.ListRequest, cb func(resource.ListIterator) error) (int64, error) { + ctx, span := tracer.Start(ctx, "legacy.dashboardSqlAccess.ListIterator") + defer span.End() + if req.ResourceVersion != 0 { return 0, apierrors.NewBadRequest("List with explicit resourceVersion is not supported with this storage backend") } @@ -348,10 +369,16 @@ func (a *dashboardSqlAccess) WatchWriteEvents(ctx context.Context) (<-chan *reso // Simple wrapper for index implementation func (a *dashboardSqlAccess) Read(ctx context.Context, req *resourcepb.ReadRequest) (*resource.BackendReadResponse, error) { + ctx, span := tracer.Start(ctx, "legacy.dashboardSqlAccess.Read") + defer span.End() + return a.ReadResource(ctx, req), nil } func (a *dashboardSqlAccess) Search(ctx context.Context, req *resourcepb.ResourceSearchRequest) (*resourcepb.ResourceSearchResponse, error) { + ctx, span := tracer.Start(ctx, "legacy.dashboardSqlAccess.Search") + defer span.End() + return a.dashboardSearchClient.Search(ctx, req) } @@ -365,5 +392,8 @@ func (a *dashboardSqlAccess) CountManagedObjects(context.Context, *resourcepb.Co // GetStats implements ResourceServer. func (a *dashboardSqlAccess) GetStats(ctx context.Context, req *resourcepb.ResourceStatsRequest) (*resourcepb.ResourceStatsResponse, error) { + ctx, span := tracer.Start(ctx, "legacy.dashboardSqlAccess.GetStats") + defer span.End() + return a.dashboardSearchClient.GetStats(ctx, req) } diff --git a/pkg/storage/legacysql/dualwrite/dualwriter.go b/pkg/storage/legacysql/dualwrite/dualwriter.go index 966340db779..8aed81d7200 100644 --- a/pkg/storage/legacysql/dualwrite/dualwriter.go +++ b/pkg/storage/legacysql/dualwrite/dualwriter.go @@ -5,6 +5,9 @@ import ( "fmt" "time" + "go.opentelemetry.io/otel" + "go.opentelemetry.io/otel/attribute" + "go.opentelemetry.io/otel/trace" apierrors "k8s.io/apimachinery/pkg/api/errors" "k8s.io/apimachinery/pkg/api/meta" metainternalversion "k8s.io/apimachinery/pkg/apis/meta/internalversion" @@ -19,41 +22,17 @@ import ( ) var ( - _ grafanarest.Storage = (*dualWriter)(nil) + _ grafanarest.Storage = (*dualWriter)(nil) + tracer = otel.Tracer("github.com/grafana/grafana/pkg/storage/legacysql/dualwrite") ) -func objectInfo(obj runtime.Object) map[string]interface{} { - if obj == nil { - return map[string]interface{}{"object": "nil"} - } - - acc, err := meta.Accessor(obj) - if err != nil { - return map[string]interface{}{"object": fmt.Sprintf("%T", obj), "error": err.Error()} - } - - info := map[string]interface{}{ - "name": acc.GetName(), - } - - if ns := acc.GetNamespace(); ns != "" { - info["namespace"] = ns - } - if uid := acc.GetUID(); uid != "" { - info["uid"] = string(uid) - } - if rv := acc.GetResourceVersion(); rv != "" { - info["resourceVersion"] = rv - } - - return info -} - -// Let's give the background queries a bit more time to complete -// as we also run them as part of load tests that might need longer -// to complete. Those run in the background and won't impact the -// user experience in any way. -const backgroundReqTimeout = time.Minute +const ( + // Let's give the background queries a bit more time to complete + // as we also run them as part of load tests that might need longer + // to complete. Those run in the background and won't impact the + // user experience in any way. + backgroundReqTimeout = time.Minute +) // dualWriter will write first to legacy, then to unified keeping the same internal ID type dualWriter struct { @@ -64,6 +43,12 @@ type dualWriter struct { } func (d *dualWriter) Get(ctx context.Context, name string, options *metav1.GetOptions) (runtime.Object, error) { + ctx, span := tracer.Start(ctx, "dualwrite.dualWriter.Get", + trace.WithAttributes( + attribute.Bool("errorIsOK", d.errorIsOK), + attribute.Bool("readUnified", d.readUnified))) + defer span.End() + log := logging.FromContext(ctx).With("method", "Get", "name", name) // If we read from unified, we can just do that and return. if d.readUnified { @@ -96,6 +81,12 @@ func (d *dualWriter) Get(ctx context.Context, name string, options *metav1.GetOp } func (d *dualWriter) List(ctx context.Context, options *metainternalversion.ListOptions) (runtime.Object, error) { + ctx, span := tracer.Start(ctx, "dualwrite.dualWriter.List", + trace.WithAttributes( + attribute.Bool("errorIsOK", d.errorIsOK), + attribute.Bool("readUnified", d.readUnified))) + defer span.End() + // Always work on *copies* so we never mutate the caller's ListOptions. var ( legacyOptions = options.DeepCopy() @@ -204,6 +195,12 @@ func (d *dualWriter) List(ctx context.Context, options *metainternalversion.List // Create overrides the behavior of the generic DualWriter and writes to LegacyStorage and Storage. func (d *dualWriter) Create(ctx context.Context, in runtime.Object, createValidation rest.ValidateObjectFunc, options *metav1.CreateOptions) (runtime.Object, error) { + ctx, span := tracer.Start(ctx, "dualwrite.dualWriter.Create", + trace.WithAttributes( + attribute.Bool("errorIsOK", d.errorIsOK), + attribute.Bool("readUnified", d.readUnified))) + defer span.End() + log := logging.FromContext(ctx).With("method", "Create") accIn, err := meta.Accessor(in) @@ -316,6 +313,11 @@ func (d *dualWriter) Delete(ctx context.Context, name string, deleteValidation r // By setting RemovePermissions to false in the context, we will skip the deletion of permissions // in the legacy store. This is needed as otherwise the permissions would be missing when executing // the delete operation in the unified storage store. + ctx, span := tracer.Start(ctx, "dualwrite.dualWriter.Delete", + trace.WithAttributes( + attribute.Bool("errorIsOK", d.errorIsOK), + attribute.Bool("readUnified", d.readUnified))) + defer span.End() log := logging.FromContext(ctx).With("method", "Delete", "name", name) ctx = utils.SetFolderRemovePermissions(ctx, false) @@ -357,8 +359,12 @@ func (d *dualWriter) Delete(ctx context.Context, name string, deleteValidation r // Update overrides the behavior of the generic DualWriter and writes first to Storage and then to LegacyStorage. func (d *dualWriter) Update(ctx context.Context, name string, objInfo rest.UpdatedObjectInfo, createValidation rest.ValidateObjectFunc, updateValidation rest.ValidateObjectUpdateFunc, forceAllowCreate bool, options *metav1.UpdateOptions) (runtime.Object, bool, error) { + ctx, span := tracer.Start(ctx, "dualwrite.dualWriter.Update", + trace.WithAttributes( + attribute.Bool("errorIsOK", d.errorIsOK), + attribute.Bool("readUnified", d.readUnified))) + defer span.End() log := logging.FromContext(ctx).With("method", "Update", "name", name) - // update in legacy first, and then unistore. Will return a failure if either fails. // // we want to update in legacy first, otherwise if the update from unistore was successful, @@ -420,6 +426,12 @@ func (d *dualWriter) Update(ctx context.Context, name string, objInfo rest.Updat // DeleteCollection overrides the behavior of the generic DualWriter and deletes from both LegacyStorage and Storage. func (d *dualWriter) DeleteCollection(ctx context.Context, deleteValidation rest.ValidateObjectFunc, options *metav1.DeleteOptions, listOptions *metainternalversion.ListOptions) (runtime.Object, error) { + ctx, span := tracer.Start(ctx, "dualwrite.dualWriter.DeleteCollection", + trace.WithAttributes( + attribute.Bool("errorIsOK", d.errorIsOK), + attribute.Bool("readUnified", d.readUnified))) + defer span.End() + log := logging.FromContext(ctx).With("method", "DeleteCollection", "resourceVersion", listOptions.ResourceVersion) // delete from legacy first, and anything that is successful can be deleted in unistore too. @@ -528,3 +540,30 @@ func (w *wrappedUpdateInfo) UpdatedObject(ctx context.Context, oldObj runtime.Ob meta.SetUID("") return obj, err } + +func objectInfo(obj runtime.Object) map[string]interface{} { + if obj == nil { + return map[string]interface{}{"object": "nil"} + } + + acc, err := meta.Accessor(obj) + if err != nil { + return map[string]interface{}{"object": fmt.Sprintf("%T", obj), "error": err.Error()} + } + + info := map[string]interface{}{ + "name": acc.GetName(), + } + + if ns := acc.GetNamespace(); ns != "" { + info["namespace"] = ns + } + if uid := acc.GetUID(); uid != "" { + info["uid"] = string(uid) + } + if rv := acc.GetResourceVersion(); rv != "" { + info["resourceVersion"] = rv + } + + return info +}