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

Provisioning: Fork context on Kubernetes requests (#100403)

Browse Source 
This'll be changed with something on the API Server infra level eventually. For now, let's unblock our engineers.
This commit is contained in:
Mariell Hoversholm authored and GitHub committed 2025-02-11 10:29:11 +01:00
1 parent 645395bee1
commit 3810aea532
8 files changed
+140 -1
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

pkg/registry/apis/provisioning/files.go
+7
View File
@@ -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 {
pkg/registry/apis/provisioning/jobs/export/folders.go
+7
View File
@@ -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 {
pkg/registry/apis/provisioning/jobs/export/job.go
+7
View File
@@ -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)
pkg/registry/apis/provisioning/jobs/sync/worker.go
+7 -1
View File
@@ -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())
pkg/registry/apis/provisioning/k8sctx/ctx.go
+85
View File
@@ -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
}
pkg/registry/apis/provisioning/register.go
+7
View File
@@ -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
pkg/registry/apis/provisioning/resources/object.go
+13
View File
@@ -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,
pkg/registry/apis/provisioning/resources/parser.go
+7
View File
@@ -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 {