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

Provisioning: Hold recent jobs in memory (#103678)

Browse Source
This commit is contained in:
Ryan McKinley
2025-04-09 12:46:01 +03:00
committed by GitHub
parent 0fd6144969
commit 6df04f3949
9 changed files with 44 additions and 349 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/apis/provisioning/v0alpha1/jobs.go
-20
View File
@@ -22,26 +22,6 @@ type JobList struct {
Items []Job `json:"items,omitempty"`
}
// HistoricJob is a history entry of Job. It is used to store Jobs that have been processed.
//
// The repository name and type are stored as labels.
// +k8s:deepcopy-gen:interfaces=k8s.io/apimachinery/pkg/runtime.Object
type HistoricJob struct {
metav1.TypeMeta `json:",inline"`
metav1.ObjectMeta `json:"metadata,omitempty"`
Spec JobSpec `json:"spec,omitempty"`
Status JobStatus `json:"status,omitempty"`
}
// +k8s:deepcopy-gen:interfaces=k8s.io/apimachinery/pkg/runtime.Object
type HistoricJobList struct {
metav1.TypeMeta `json:",inline"`
metav1.ListMeta `json:"metadata,omitempty"`
Items []HistoricJob `json:"items,omitempty"`
}
// +enum
type JobAction string
pkg/apis/provisioning/v0alpha1/register.go
-30
View File
@@ -81,34 +81,6 @@ var JobResourceInfo = utils.NewResourceInfo(GROUP, VERSION,
},
})
var HistoricJobResourceInfo = utils.NewResourceInfo(GROUP, VERSION,
"historicjobs", "historicjob", "HistoricJob",
func() runtime.Object { return &HistoricJob{} }, // newObj
func() runtime.Object { return &HistoricJobList{} }, // newList
utils.TableColumns{ // Returned by `kubectl get`. Doesn't affect disk storage.
Definition: []metav1.TableColumnDefinition{
{Name: "Name", Type: "string", Format: "name"},
{Name: "Created At", Type: "date"},
{Name: "Action", Type: "string"},
{Name: "State", Type: "string"},
{Name: "Message", Type: "string"},
},
Reader: func(obj any) ([]interface{}, error) {
m, ok := obj.(*HistoricJob)
if !ok {
return nil, errors.New("expected HistoricJob")
}
return []interface{}{
m.Name, // may our may not be nice to read
m.CreationTimestamp.UTC().Format(time.RFC3339),
m.Spec.Action,
m.Status.State,
m.Status.Message,
}, nil
},
})
var (
// SchemeGroupVersion is group version used to register these objects
SchemeGroupVersion = schema.GroupVersion{Group: GROUP, Version: VERSION}
@@ -145,8 +117,6 @@ func AddKnownTypes(gv schema.GroupVersion, scheme *runtime.Scheme) error {
&ResourceStats{},
&Job{},
&JobList{},
&HistoricJob{},
&HistoricJobList{},
)
return nil
}
pkg/apis/provisioning/v0alpha1/zz_generated.deepcopy.go
-61
View File
@@ -132,67 +132,6 @@ func (in *HealthStatus) DeepCopy() *HealthStatus {
return out
}
// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
func (in *HistoricJob) DeepCopyInto(out *HistoricJob) {
*out = *in
out.TypeMeta = in.TypeMeta
in.ObjectMeta.DeepCopyInto(&out.ObjectMeta)
in.Spec.DeepCopyInto(&out.Spec)
in.Status.DeepCopyInto(&out.Status)
return
}
// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new HistoricJob.
func (in *HistoricJob) DeepCopy() *HistoricJob {
if in == nil {
return nil
}
out := new(HistoricJob)
in.DeepCopyInto(out)
return out
}
// DeepCopyObject is an autogenerated deepcopy function, copying the receiver, creating a new runtime.Object.
func (in *HistoricJob) DeepCopyObject() runtime.Object {
if c := in.DeepCopy(); c != nil {
return c
}
return nil
}
// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
func (in *HistoricJobList) DeepCopyInto(out *HistoricJobList) {
*out = *in
out.TypeMeta = in.TypeMeta
in.ListMeta.DeepCopyInto(&out.ListMeta)
if in.Items != nil {
in, out := &in.Items, &out.Items
*out = make([]HistoricJob, len(*in))
for i := range *in {
(*in)[i].DeepCopyInto(&(*out)[i])
}
}
return
}
// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new HistoricJobList.
func (in *HistoricJobList) DeepCopy() *HistoricJobList {
if in == nil {
return nil
}
out := new(HistoricJobList)
in.DeepCopyInto(out)
return out
}
// DeepCopyObject is an autogenerated deepcopy function, copying the receiver, creating a new runtime.Object.
func (in *HistoricJobList) DeepCopyObject() runtime.Object {
if c := in.DeepCopy(); c != nil {
return c
}
return nil
}
// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
func (in *HistoryItem) DeepCopyInto(out *HistoryItem) {
*out = *in
pkg/apis/provisioning/v0alpha1/zz_generated.openapi.go
-96
View File
@@ -20,8 +20,6 @@ func GetOpenAPIDefinitions(ref common.ReferenceCallback) map[string]common.OpenA
"github.com/grafana/grafana/pkg/apis/provisioning/v0alpha1.FileList": schema_pkg_apis_provisioning_v0alpha1_FileList(ref),
"github.com/grafana/grafana/pkg/apis/provisioning/v0alpha1.GitHubRepositoryConfig": schema_pkg_apis_provisioning_v0alpha1_GitHubRepositoryConfig(ref),
"github.com/grafana/grafana/pkg/apis/provisioning/v0alpha1.HealthStatus": schema_pkg_apis_provisioning_v0alpha1_HealthStatus(ref),
"github.com/grafana/grafana/pkg/apis/provisioning/v0alpha1.HistoricJob": schema_pkg_apis_provisioning_v0alpha1_HistoricJob(ref),
"github.com/grafana/grafana/pkg/apis/provisioning/v0alpha1.HistoricJobList": schema_pkg_apis_provisioning_v0alpha1_HistoricJobList(ref),
"github.com/grafana/grafana/pkg/apis/provisioning/v0alpha1.HistoryItem": schema_pkg_apis_provisioning_v0alpha1_HistoryItem(ref),
"github.com/grafana/grafana/pkg/apis/provisioning/v0alpha1.HistoryList": schema_pkg_apis_provisioning_v0alpha1_HistoryList(ref),
"github.com/grafana/grafana/pkg/apis/provisioning/v0alpha1.Job": schema_pkg_apis_provisioning_v0alpha1_Job(ref),
@@ -329,100 +327,6 @@ func schema_pkg_apis_provisioning_v0alpha1_HealthStatus(ref common.ReferenceCall
}
}
func schema_pkg_apis_provisioning_v0alpha1_HistoricJob(ref common.ReferenceCallback) common.OpenAPIDefinition {
return common.OpenAPIDefinition{
Schema: spec.Schema{
SchemaProps: spec.SchemaProps{
Description: "HistoricJob is a history entry of Job. It is used to store Jobs that have been processed.\n\nThe repository name and type are stored as labels.",
Type: []string{"object"},
Properties: map[string]spec.Schema{
"kind": {
SchemaProps: spec.SchemaProps{
Description: "Kind is a string value representing the REST resource this object represents. Servers may infer this from the endpoint the client submits requests to. Cannot be updated. In CamelCase. More info: https://git.k8s.io/community/contributors/devel/sig-architecture/api-conventions.md#types-kinds",
Type: []string{"string"},
Format: "",
},
},
"apiVersion": {
SchemaProps: spec.SchemaProps{
Description: "APIVersion defines the versioned schema of this representation of an object. Servers should convert recognized schemas to the latest internal value, and may reject unrecognized values. More info: https://git.k8s.io/community/contributors/devel/sig-architecture/api-conventions.md#resources",
Type: []string{"string"},
Format: "",
},
},
"metadata": {
SchemaProps: spec.SchemaProps{
Default: map[string]interface{}{},
Ref: ref("k8s.io/apimachinery/pkg/apis/meta/v1.ObjectMeta"),
},
},
"spec": {
SchemaProps: spec.SchemaProps{
Default: map[string]interface{}{},
Ref: ref("github.com/grafana/grafana/pkg/apis/provisioning/v0alpha1.JobSpec"),
},
},
"status": {
SchemaProps: spec.SchemaProps{
Default: map[string]interface{}{},
Ref: ref("github.com/grafana/grafana/pkg/apis/provisioning/v0alpha1.JobStatus"),
},
},
},
},
},
Dependencies: []string{
"github.com/grafana/grafana/pkg/apis/provisioning/v0alpha1.JobSpec", "github.com/grafana/grafana/pkg/apis/provisioning/v0alpha1.JobStatus", "k8s.io/apimachinery/pkg/apis/meta/v1.ObjectMeta"},
}
}
func schema_pkg_apis_provisioning_v0alpha1_HistoricJobList(ref common.ReferenceCallback) common.OpenAPIDefinition {
return common.OpenAPIDefinition{
Schema: spec.Schema{
SchemaProps: spec.SchemaProps{
Type: []string{"object"},
Properties: map[string]spec.Schema{
"kind": {
SchemaProps: spec.SchemaProps{
Description: "Kind is a string value representing the REST resource this object represents. Servers may infer this from the endpoint the client submits requests to. Cannot be updated. In CamelCase. More info: https://git.k8s.io/community/contributors/devel/sig-architecture/api-conventions.md#types-kinds",
Type: []string{"string"},
Format: "",
},
},
"apiVersion": {
SchemaProps: spec.SchemaProps{
Description: "APIVersion defines the versioned schema of this representation of an object. Servers should convert recognized schemas to the latest internal value, and may reject unrecognized values. More info: https://git.k8s.io/community/contributors/devel/sig-architecture/api-conventions.md#resources",
Type: []string{"string"},
Format: "",
},
},
"metadata": {
SchemaProps: spec.SchemaProps{
Default: map[string]interface{}{},
Ref: ref("k8s.io/apimachinery/pkg/apis/meta/v1.ListMeta"),
},
},
"items": {
SchemaProps: spec.SchemaProps{
Type: []string{"array"},
Items: &spec.SchemaOrArray{
Schema: &spec.Schema{
SchemaProps: spec.SchemaProps{
Default: map[string]interface{}{},
Ref: ref("github.com/grafana/grafana/pkg/apis/provisioning/v0alpha1.HistoricJob"),
},
},
},
},
},
},
},
},
Dependencies: []string{
"github.com/grafana/grafana/pkg/apis/provisioning/v0alpha1.HistoricJob", "k8s.io/apimachinery/pkg/apis/meta/v1.ListMeta"},
}
}
func schema_pkg_apis_provisioning_v0alpha1_HistoryItem(ref common.ReferenceCallback) common.OpenAPIDefinition {
return common.OpenAPIDefinition{
Schema: spec.Schema{
pkg/registry/apis/provisioning/jobs/history.go
+41 -82
View File
@@ -3,13 +3,9 @@ package jobs
import (
"context"
"fmt"
"sync"
apierrors "k8s.io/apimachinery/pkg/api/errors"
"k8s.io/apimachinery/pkg/apis/meta/internalversion"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/apimachinery/pkg/labels"
"k8s.io/apiserver/pkg/endpoints/request"
"k8s.io/apiserver/pkg/registry/rest"
provisioning "github.com/grafana/grafana/pkg/apis/provisioning/v0alpha1"
)
@@ -28,100 +24,63 @@ type History interface {
GetJob(ctx context.Context, namespace, repo, uid string) (*provisioning.Job, error)
}
// NewStorageBackedHistory creates a History client backed by unified storage
// This should be replaced by loki when running in cloud
func NewStorageBackedHistory(store rest.Storage) (History, error) {
var ok bool
history := &storageBackedHistory{}
history.creator, ok = store.(rest.Creater)
if !ok {
return nil, fmt.Errorf("storage does not implement rest.Creater")
// NewJobHistoryCache creates a History client
func NewJobHistoryCache() History {
history := &recentHistory{
maxJobs: 9, // 10-1
repoHistory: make(map[string]provisioning.JobList),
}
history.lister, ok = store.(rest.Lister)
if !ok {
return nil, fmt.Errorf("storage does not implement rest.Lister")
}
return history, nil
return history
}
type storageBackedHistory struct {
creator rest.Creater
lister rest.Lister
type recentHistory struct {
maxJobs int
repoMu sync.Mutex
repoHistory map[string]provisioning.JobList
}
// Write implements History.
func (s *storageBackedHistory) WriteJob(ctx context.Context, job *provisioning.Job) error {
if job.UID == "" {
return fmt.Errorf("missing UID in job '%s'", job.GetName())
}
if job.Labels == nil {
job.Labels = make(map[string]string)
}
job.Labels[LabelRepository] = job.Spec.Repository
job.Labels[LabelJobOriginalUID] = string(job.UID)
func (h *recentHistory) WriteJob(ctx context.Context, job *provisioning.Job) error {
h.repoMu.Lock()
defer h.repoMu.Unlock()
// Generate a new name based on the input job
job.GenerateName = job.Name + "-"
job.Name = ""
// We also reset the UID as this is not the same object.
job.ObjectMeta.UID = ""
// We aren't allowed to write with ResourceVersion set.
job.ResourceVersion = ""
copy := job.DeepCopy()
delete(copy.Labels, LabelJobClaim)
_, err := s.creator.Create(ctx, &provisioning.HistoricJob{
ObjectMeta: job.ObjectMeta,
Spec: job.Spec,
Status: job.Status,
}, nil, &metav1.CreateOptions{})
return err
}
func (s *storageBackedHistory) getJobs(ctx context.Context, namespace string, labels labels.Set) (*provisioning.JobList, error) {
ctx = request.WithNamespace(ctx, namespace)
obj, err := s.lister.List(ctx, &internalversion.ListOptions{
LabelSelector: labels.AsSelector(),
})
if err != nil {
return nil, err
items := []provisioning.Job{*copy}
key := fmt.Sprintf("%s/%s", job.Namespace, job.Spec.Repository)
v, ok := h.repoHistory[key]
if ok {
max := min(len(v.Items), h.maxJobs)
items = append(items, v.Items[0:max]...)
}
historic, ok := obj.(*provisioning.HistoricJobList)
if !ok {
return nil, fmt.Errorf("expected HistoricJobList, found %T", historic)
}
jobs := &provisioning.JobList{
ListMeta: historic.ListMeta,
}
for _, job := range historic.Items {
jobs.Items = append(jobs.Items, provisioning.Job{
ObjectMeta: job.ObjectMeta,
Spec: job.Spec,
Status: job.Status,
})
}
return jobs, nil
h.repoHistory[key] = provisioning.JobList{Items: items}
return nil
}
// Recent implements History.
func (s *storageBackedHistory) RecentJobs(ctx context.Context, namespace, repo string) (*provisioning.JobList, error) {
return s.getJobs(ctx, namespace, labels.Set{
LabelRepository: repo,
})
func (h *recentHistory) RecentJobs(ctx context.Context, namespace, repo string) (*provisioning.JobList, error) {
h.repoMu.Lock()
defer h.repoMu.Unlock()
rsp := &provisioning.JobList{}
key := fmt.Sprintf("%s/%s", namespace, repo)
v, ok := h.repoHistory[key]
if ok {
rsp.Items = v.Items
}
return rsp, nil
}
// GetJob implements History.
func (s *storageBackedHistory) GetJob(ctx context.Context, namespace, repo, job string) (*provisioning.Job, error) {
jobs, err := s.getJobs(ctx, namespace, labels.Set{
LabelJobOriginalUID: job,
})
func (h *recentHistory) GetJob(ctx context.Context, namespace, repo, job string) (*provisioning.Job, error) {
jobs, err := h.RecentJobs(ctx, namespace, repo)
if err != nil {
return nil, err
}
if len(jobs.Items) == 1 {
return &jobs.Items[0], nil
for _, item := range jobs.Items {
if string(item.UID) == job {
return &item, nil
}
}
return nil, apierrors.NewNotFound(provisioning.JobResourceInfo.GroupResource(), job)
}
pkg/registry/apis/provisioning/jobs/persistentstore.go
-2
View File
@@ -28,8 +28,6 @@ const (
// The label must be formatted as milliseconds from Epoch. This grants a natural ordering, allowing for less-than operators in label selectors.
// The natural ordering would be broken if the number rolls over into 1 more digit. This won't happen before Nov, 2286.
LabelJobClaim = "provisioning.grafana.app/claim"
// LabelJobOriginalUID contains the Job's original uid as a label. This allows for label selectors to find the archived version of a job.
LabelJobOriginalUID = "provisioning.grafana.app/original-uid"
// LabelRepository contains the repository name as a label. This allows for label selectors to find the archived version of a job.
LabelRepository = "provisioning.grafana.app/repository"
)
pkg/registry/apis/provisioning/register.go
+2 -12
View File
@@ -140,6 +140,7 @@ func NewAPIBuilder(
storageStatus: storageStatus,
unified: unified,
secrets: secrets,
jobHistory: jobs.NewJobHistoryCache(),
}
}
@@ -287,8 +288,7 @@ func (b *APIBuilder) GetAuthorizer() authorizer.Authorizer {
}
return authorizer.DecisionDeny, "viewer role is required", nil
case provisioning.JobResourceInfo.GetName(),
provisioning.HistoricJobResourceInfo.GetName():
case provisioning.JobResourceInfo.GetName():
// Jobs are shown on the configuration page.
if id.GetOrgRole().Includes(identity.RoleAdmin) {
return authorizer.DecisionAllow, "", nil
@@ -343,16 +343,6 @@ func (b *APIBuilder) UpdateAPIGroupInfo(apiGroupInfo *genericapiserver.APIGroupI
return fmt.Errorf("failed to create job storage: %w", err)
}
historicJobStore, err := grafanaregistry.NewCompleteRegistryStore(opts.Scheme, provisioning.HistoricJobResourceInfo, opts.OptsGetter)
if err != nil {
return fmt.Errorf("failed to create historic job storage: %w", err)
}
b.jobHistory, err = jobs.NewStorageBackedHistory(historicJobStore)
if err != nil {
return fmt.Errorf("failed to create historic job wrapper: %w", err)
}
b.jobs, err = jobs.NewStore(realJobStore, time.Second*30)
if err != nil {
return fmt.Errorf("failed to create job store: %w", err)
pkg/tests/apis/openapi_snapshots/provisioning.grafana.app-v0alpha1.json
-45
View File
@@ -2601,51 +2601,6 @@
}
}
},
"com.github.grafana.grafana.pkg.apis.provisioning.v0alpha1.HistoricJob": {
"description": "HistoricJob is a history entry of Job. It is used to store Jobs that have been processed.\n\nThe repository name and type are stored as labels.",
"type": "object",
"properties": {
"apiVersion": {
"description": "APIVersion defines the versioned schema of this representation of an object. Servers should convert recognized schemas to the latest internal value, and may reject unrecognized values. More info: https://git.k8s.io/community/contributors/devel/sig-architecture/api-conventions.md#resources",
"type": "string"
},
"kind": {
"description": "Kind is a string value representing the REST resource this object represents. Servers may infer this from the endpoint the client submits requests to. Cannot be updated. In CamelCase. More info: https://git.k8s.io/community/contributors/devel/sig-architecture/api-conventions.md#types-kinds",
"type": "string"
},
"metadata": {
"default": {}
},
"spec": {
"default": {}
},
"status": {
"default": {}
}
}
},
"com.github.grafana.grafana.pkg.apis.provisioning.v0alpha1.HistoricJobList": {
"type": "object",
"properties": {
"apiVersion": {
"description": "APIVersion defines the versioned schema of this representation of an object. Servers should convert recognized schemas to the latest internal value, and may reject unrecognized values. More info: https://git.k8s.io/community/contributors/devel/sig-architecture/api-conventions.md#resources",
"type": "string"
},
"items": {
"type": "array",
"items": {
"default": {}
}
},
"kind": {
"description": "Kind is a string value representing the REST resource this object represents. Servers may infer this from the endpoint the client submits requests to. Cannot be updated. In CamelCase. More info: https://git.k8s.io/community/contributors/devel/sig-architecture/api-conventions.md#types-kinds",
"type": "string"
},
"metadata": {
"default": {}
}
}
},
"com.github.grafana.grafana.pkg.apis.provisioning.v0alpha1.HistoryItem": {
"type": "object",
"required": [
public/app/features/provisioning/Job/RecentJobs.tsx
+1 -1
View File
@@ -201,7 +201,7 @@ export function RecentJobs({ repo }: Props) {
<InteractiveTable
data={jobs}
columns={jobColumns}
getRowId={(item) => `${item.metadata?.name}`}
getRowId={(item) => `${item.metadata?.uid}`}
renderExpandedRow={(row) => <ExpandedRow row={row} />}
pageSize={10}
/>