From 3810aea532f2a32e1dbcd7ec438f60ce32d0ab12 Mon Sep 17 00:00:00 2001 From: Mariell Hoversholm Date: Tue, 11 Feb 2025 10:29:11 +0100 Subject: [PATCH] Provisioning: Fork context on Kubernetes requests (#100403) This'll be changed with something on the API Server infra level eventually. For now, let's unblock our engineers. --- pkg/registry/apis/provisioning/files.go | 7 ++ .../apis/provisioning/jobs/export/folders.go | 7 ++ .../apis/provisioning/jobs/export/job.go | 7 ++ .../apis/provisioning/jobs/sync/worker.go | 8 +- pkg/registry/apis/provisioning/k8sctx/ctx.go | 85 +++++++++++++++++++ pkg/registry/apis/provisioning/register.go | 7 ++ .../apis/provisioning/resources/object.go | 13 +++ .../apis/provisioning/resources/parser.go | 7 ++ 8 files changed, 140 insertions(+), 1 deletion(-) create mode 100644 pkg/registry/apis/provisioning/k8sctx/ctx.go diff --git a/pkg/registry/apis/provisioning/files.go b/pkg/registry/apis/provisioning/files.go index 2ac7eb90746..4f8877165d5 100644 --- a/pkg/registry/apis/provisioning/files.go +++ b/pkg/registry/apis/provisioning/files.go @@ -17,6 +17,7 @@ import ( "github.com/grafana/grafana-app-sdk/logging" provisioning "github.com/grafana/grafana/pkg/apis/provisioning/v0alpha1" + "github.com/grafana/grafana/pkg/registry/apis/provisioning/k8sctx" "github.com/grafana/grafana/pkg/registry/apis/provisioning/repository" "github.com/grafana/grafana/pkg/registry/apis/provisioning/resources" ) @@ -254,6 +255,12 @@ func (s *filesConnector) doWrite(ctx context.Context, update bool, repo reposito return nil, err } + ctx, cancel, err := k8sctx.Fork(ctx) + if err != nil { + return nil, err + } + defer cancel() + // Which context? request of background??? // Behaves the same running sync after writing if parsed.Existing == nil { diff --git a/pkg/registry/apis/provisioning/jobs/export/folders.go b/pkg/registry/apis/provisioning/jobs/export/folders.go index 0d2b640178f..b7e16bc94ac 100644 --- a/pkg/registry/apis/provisioning/jobs/export/folders.go +++ b/pkg/registry/apis/provisioning/jobs/export/folders.go @@ -10,10 +10,17 @@ import ( apiutils "github.com/grafana/grafana/pkg/apimachinery/utils" "github.com/grafana/grafana/pkg/cmd/grafana-cli/logger" + "github.com/grafana/grafana/pkg/registry/apis/provisioning/k8sctx" "github.com/grafana/grafana/pkg/registry/apis/provisioning/resources" ) func readFolders(ctx context.Context, client dynamic.ResourceInterface, skip string) (*resources.FolderTree, error) { + ctx, cancel, err := k8sctx.Fork(ctx) + if err != nil { + return nil, err + } + defer cancel() + // TODO: handle pagination rawList, err := client.List(ctx, metav1.ListOptions{Limit: 10000}) if err != nil { diff --git a/pkg/registry/apis/provisioning/jobs/export/job.go b/pkg/registry/apis/provisioning/jobs/export/job.go index 23557c072a3..b63ad924dca 100644 --- a/pkg/registry/apis/provisioning/jobs/export/job.go +++ b/pkg/registry/apis/provisioning/jobs/export/job.go @@ -19,6 +19,7 @@ import ( "github.com/grafana/grafana/pkg/cmd/grafana-cli/logger" "github.com/grafana/grafana/pkg/infra/slugify" "github.com/grafana/grafana/pkg/registry/apis/provisioning/jobs" + "github.com/grafana/grafana/pkg/registry/apis/provisioning/k8sctx" "github.com/grafana/grafana/pkg/registry/apis/provisioning/repository" "github.com/grafana/grafana/pkg/registry/apis/provisioning/resources" "github.com/grafana/grafana/pkg/registry/apis/provisioning/safepath" @@ -149,6 +150,12 @@ func (r *exportJob) loadFolders(ctx context.Context) error { } func (r *exportJob) export(ctx context.Context, kind schema.GroupVersionResource) error { + ctx, cancel, err := k8sctx.Fork(ctx) + if err != nil { + return err + } + defer cancel() + r.jobStatus.Message = "Exporting " + kind.Resource + "..." r.maybeNotify(ctx) client := r.client.Resource(kind) diff --git a/pkg/registry/apis/provisioning/jobs/sync/worker.go b/pkg/registry/apis/provisioning/jobs/sync/worker.go index b59ff457290..c3b4cc13396 100644 --- a/pkg/registry/apis/provisioning/jobs/sync/worker.go +++ b/pkg/registry/apis/provisioning/jobs/sync/worker.go @@ -24,6 +24,7 @@ import ( provisioning "github.com/grafana/grafana/pkg/apis/provisioning/v0alpha1" client "github.com/grafana/grafana/pkg/generated/clientset/versioned/typed/provisioning/v0alpha1" "github.com/grafana/grafana/pkg/registry/apis/provisioning/jobs" + "github.com/grafana/grafana/pkg/registry/apis/provisioning/k8sctx" "github.com/grafana/grafana/pkg/registry/apis/provisioning/repository" "github.com/grafana/grafana/pkg/registry/apis/provisioning/resources" "github.com/grafana/grafana/pkg/registry/apis/provisioning/safepath" @@ -58,7 +59,12 @@ func (r *SyncWorker) Process(ctx context.Context, job provisioning.Job, progress jobs.ProgressFn, ) (*provisioning.JobStatus, error) { - var err error + ctx, cancel, err := k8sctx.Fork(ctx) + if err != nil { + return nil, err + } + defer cancel() + cfg := repo.Config() logger := logging.FromContext(ctx).With("job", job.GetName(), "namespace", job.GetNamespace()) diff --git a/pkg/registry/apis/provisioning/k8sctx/ctx.go b/pkg/registry/apis/provisioning/k8sctx/ctx.go new file mode 100644 index 00000000000..2c15a5612ca --- /dev/null +++ b/pkg/registry/apis/provisioning/k8sctx/ctx.go @@ -0,0 +1,85 @@ +package k8sctx + +import ( + "context" + "fmt" + + "github.com/grafana/grafana-app-sdk/logging" + "github.com/grafana/grafana/pkg/apimachinery/identity" + "github.com/grafana/grafana/pkg/infra/log" + "k8s.io/apiserver/pkg/authentication/user" + "k8s.io/apiserver/pkg/endpoints/request" + "k8s.io/component-base/tracing" + "k8s.io/klog/v2" +) + +// Fork creates a new context for Kubernetes APIs to use. +func Fork(refCtx context.Context) (context.Context, context.CancelFunc, error) { + newCtx := context.Background() + + if ns, ok := request.NamespaceFrom(refCtx); ok { + newCtx = request.WithNamespace(newCtx, ns) + } + if signal := request.ServerShutdownSignalFrom(refCtx); signal != nil { + newCtx = request.WithServerShutdownSignal(newCtx, signal) + } + + requester, _ := identity.GetRequester(refCtx) + if requester != nil { + newCtx = identity.WithRequester(newCtx, requester) + } + + usr, ok := request.UserFrom(refCtx) + if !ok && requester != nil { + // add in k8s user if not there yet + var ok bool + usr, ok = requester.(user.Info) + if !ok { + return nil, nil, fmt.Errorf("could not convert user to Kubernetes user") + } + } + if ok { + newCtx = request.WithUser(newCtx, usr) + } + + // App SDK logger + appLogger := logging.FromContext(refCtx) + newCtx = logging.Context(newCtx, appLogger) + // Klog logger + klogger := klog.FromContext(refCtx) + if klogger.Enabled() { + newCtx = klog.NewContext(newCtx, klogger) + } + // Grafana infra + infraLogger := log.FromContext(refCtx) + if len(infraLogger) > 0 { + newCtx = log.WithContextualAttributes(newCtx, infraLogger) + } + + // The tracing package deals with both k8s trace and otel. + if span := tracing.SpanFromContext(refCtx); span != nil && *span != (tracing.Span{}) { + newCtx = tracing.ContextWithSpan(newCtx, span) + } + + deadlineCancel := context.CancelFunc(func() {}) + if deadline, ok := refCtx.Deadline(); ok { + newCtx, deadlineCancel = context.WithDeadline(newCtx, deadline) + } + + newCtx, cancel := context.WithCancelCause(newCtx) + // We intentionally do not defer a cancel(nil) here. It wouldn't make sense to cancel until (*ResponseAdapter).Close() is called. + go func() { // Even context's own impls do goroutines for this type of pattern. + select { + case <-newCtx.Done(): + // We don't have to do anything! + case <-refCtx.Done(): + cancel(context.Cause(refCtx)) + } + deadlineCancel() + }() + + return newCtx, context.CancelFunc(func() { + cancel(nil) + deadlineCancel() + }), nil +} diff --git a/pkg/registry/apis/provisioning/register.go b/pkg/registry/apis/provisioning/register.go index 2939c5c7982..10f853bd178 100644 --- a/pkg/registry/apis/provisioning/register.go +++ b/pkg/registry/apis/provisioning/register.go @@ -34,6 +34,7 @@ import ( "github.com/grafana/grafana/pkg/registry/apis/provisioning/jobs/export" "github.com/grafana/grafana/pkg/registry/apis/provisioning/jobs/pullrequest" "github.com/grafana/grafana/pkg/registry/apis/provisioning/jobs/sync" + "github.com/grafana/grafana/pkg/registry/apis/provisioning/k8sctx" "github.com/grafana/grafana/pkg/registry/apis/provisioning/repository" "github.com/grafana/grafana/pkg/registry/apis/provisioning/repository/github" "github.com/grafana/grafana/pkg/registry/apis/provisioning/resources" @@ -225,6 +226,12 @@ func (b *APIBuilder) UpdateAPIGroupInfo(apiGroupInfo *genericapiserver.APIGroupI } func (b *APIBuilder) GetRepository(ctx context.Context, name string) (repository.Repository, error) { + ctx, cancel, err := k8sctx.Fork(ctx) + if err != nil { + return nil, err + } + defer cancel() + obj, err := b.getter.Get(ctx, name, &metav1.GetOptions{}) if err != nil { return nil, err diff --git a/pkg/registry/apis/provisioning/resources/object.go b/pkg/registry/apis/provisioning/resources/object.go index 2d511d7090b..33b4b007b97 100644 --- a/pkg/registry/apis/provisioning/resources/object.go +++ b/pkg/registry/apis/provisioning/resources/object.go @@ -4,6 +4,7 @@ import ( "context" provisioning "github.com/grafana/grafana/pkg/apis/provisioning/v0alpha1" + "github.com/grafana/grafana/pkg/registry/apis/provisioning/k8sctx" "github.com/grafana/grafana/pkg/storage/unified/resource" ) @@ -25,6 +26,12 @@ func NewResourceLister(index resource.RepositoryIndexClient) ResourceLister { // List implements ResourceLister. func (o *ResourceListerFromSearch) List(ctx context.Context, namespace, repository string) (*provisioning.ResourceList, error) { + ctx, cancel, err := k8sctx.Fork(ctx) + if err != nil { + return nil, err + } + defer cancel() + objects, err := o.index.ListRepositoryObjects(ctx, &resource.ListRepositoryObjectsRequest{ Namespace: namespace, Name: repository, @@ -54,6 +61,12 @@ func (o *ResourceListerFromSearch) List(ctx context.Context, namespace, reposito // Stats implements ResourceLister. func (o *ResourceListerFromSearch) Stats(ctx context.Context, namespace, repository string) (*provisioning.ResourceStats, error) { + ctx, cancel, err := k8sctx.Fork(ctx) + if err != nil { + return nil, err + } + defer cancel() + counts, err := o.index.CountRepositoryObjects(ctx, &resource.CountRepositoryObjectsRequest{ Namespace: namespace, Name: repository, diff --git a/pkg/registry/apis/provisioning/resources/parser.go b/pkg/registry/apis/provisioning/resources/parser.go index 52147acddec..bba50230fcd 100644 --- a/pkg/registry/apis/provisioning/resources/parser.go +++ b/pkg/registry/apis/provisioning/resources/parser.go @@ -20,6 +20,7 @@ import ( "github.com/grafana/grafana/pkg/apimachinery/apis/common/v0alpha1" "github.com/grafana/grafana/pkg/apimachinery/utils" provisioning "github.com/grafana/grafana/pkg/apis/provisioning/v0alpha1" + "github.com/grafana/grafana/pkg/registry/apis/provisioning/k8sctx" "github.com/grafana/grafana/pkg/registry/apis/provisioning/lint" "github.com/grafana/grafana/pkg/registry/apis/provisioning/repository" ) @@ -201,6 +202,12 @@ func (r *Parser) Parse(ctx context.Context, info *repository.FileInfo, validate return parsed, nil } + ctx, cancel, err := k8sctx.Fork(ctx) + if err != nil { + return nil, err + } + defer cancel() + // Dry run CREATE or UPDATE parsed.Existing, _ = parsed.Client.Get(ctx, obj.GetName(), metav1.GetOptions{}) if parsed.Existing == nil {