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

feat(unified-storage): add tracing to dual writer and legacy storage (#113504)

Browse Source 
Co-authored-by: Mustafa Sencer Özcan <32759850+mustafasencer@users.noreply.github.com>
This commit is contained in:
Jean-Philippe Quéméner
2025-11-06 11:42:46 +00:00
committed by GitHub
co-authored by Mustafa Sencer Özcan
parent efd6b250d9
commit fd14d4a5ed
3 changed files with 121 additions and 35 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/storage/legacysql/dualwrite/dualwriter.go
+73 -34
View File
@@ -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
}