diff --git a/pkg/apis/provisioning/v0alpha1/jobs.go b/pkg/apis/provisioning/v0alpha1/jobs.go index f68e282a6d1..4bfd476b377 100644 --- a/pkg/apis/provisioning/v0alpha1/jobs.go +++ b/pkg/apis/provisioning/v0alpha1/jobs.go @@ -68,6 +68,13 @@ type JobSpec struct { // URL to the originator (eg, PR URL) URL string `json:"url,omitempty"` + // When we know the commits, these will be passed along + Commits []CommitInfo `json:"commits,omitempty"` +} + +type CommitInfo struct { + SHA1 string `json:"sha1,omitempty"` + Added []FileRef `json:"added,omitempty"` Modified []FileRef `json:"modified,omitempty"` Removed []FileRef `json:"removed,omitempty"` @@ -80,10 +87,11 @@ type FileRef struct { // The job status type JobStatus struct { - Updated metav1.Time `json:"updated,omitempty"` - State JobState `json:"state,omitempty"` - Message string `json:"message,omitempty"` - Errors []string `json:"errors,omitempty"` + State JobState `json:"state,omitempty"` + Started int64 `json:"started,omitempty"` + Finished int64 `json:"finished,omitempty"` + Message string `json:"message,omitempty"` + Errors []string `json:"errors,omitempty"` } // Filter ignorable files @@ -102,14 +110,14 @@ type WebhookResponse struct { metav1.TypeMeta `json:",inline"` // HTTP Status code + // 200 implies that the payload was understood but nothing is required + // 202 implies that an async job has been scheduled to handle the request Code int `json:"code,omitempty"` // Optional message Message string `json:"added,omitempty"` // Jobs to be processed - Jobs []JobSpec `json:"jobs,omitempty"` - - // The queued jobs - IDs []string `json:"ids,omitempty"` + // When the response is 202 (Accepted) the queued jobs will be returned + Job *JobSpec `json:"job,omitempty"` } diff --git a/pkg/apis/provisioning/v0alpha1/register.go b/pkg/apis/provisioning/v0alpha1/register.go index fc59afb175e..b2d64be15d3 100644 --- a/pkg/apis/provisioning/v0alpha1/register.go +++ b/pkg/apis/provisioning/v0alpha1/register.go @@ -65,6 +65,8 @@ var JobResourceInfo = utils.NewResourceInfo(GROUP, VERSION, {Name: "Name", Type: "string", Format: "name"}, {Name: "Created At", Type: "date"}, {Name: "Action", Type: "string"}, + {Name: "State", Type: "string"}, + {Name: "Repository", Type: "string"}, }, Reader: func(obj any) ([]interface{}, error) { m, ok := obj.(*Job) @@ -76,6 +78,8 @@ var JobResourceInfo = utils.NewResourceInfo(GROUP, VERSION, m.Name, // may our may not be nice to read m.CreationTimestamp.UTC().Format(time.RFC3339), m.Spec.Action, + m.Status.State, + m.Labels["repository"], }, nil }, }) diff --git a/pkg/apis/provisioning/v0alpha1/zz_generated.deepcopy.go b/pkg/apis/provisioning/v0alpha1/zz_generated.deepcopy.go index 822a1daeaca..f26a4ab1307 100644 --- a/pkg/apis/provisioning/v0alpha1/zz_generated.deepcopy.go +++ b/pkg/apis/provisioning/v0alpha1/zz_generated.deepcopy.go @@ -27,6 +27,37 @@ func (in *Author) DeepCopy() *Author { return out } +// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil. +func (in *CommitInfo) DeepCopyInto(out *CommitInfo) { + *out = *in + if in.Added != nil { + in, out := &in.Added, &out.Added + *out = make([]FileRef, len(*in)) + copy(*out, *in) + } + if in.Modified != nil { + in, out := &in.Modified, &out.Modified + *out = make([]FileRef, len(*in)) + copy(*out, *in) + } + if in.Removed != nil { + in, out := &in.Removed, &out.Removed + *out = make([]FileRef, len(*in)) + copy(*out, *in) + } + return +} + +// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new CommitInfo. +func (in *CommitInfo) DeepCopy() *CommitInfo { + if in == nil { + return nil + } + out := new(CommitInfo) + in.DeepCopyInto(out) + return out +} + // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil. func (in *EditingOptions) DeepCopyInto(out *EditingOptions) { *out = *in @@ -265,20 +296,12 @@ func (in *JobList) DeepCopyObject() runtime.Object { // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil. func (in *JobSpec) DeepCopyInto(out *JobSpec) { *out = *in - if in.Added != nil { - in, out := &in.Added, &out.Added - *out = make([]FileRef, len(*in)) - copy(*out, *in) - } - if in.Modified != nil { - in, out := &in.Modified, &out.Modified - *out = make([]FileRef, len(*in)) - copy(*out, *in) - } - if in.Removed != nil { - in, out := &in.Removed, &out.Removed - *out = make([]FileRef, len(*in)) - copy(*out, *in) + if in.Commits != nil { + in, out := &in.Commits, &out.Commits + *out = make([]CommitInfo, len(*in)) + for i := range *in { + (*in)[i].DeepCopyInto(&(*out)[i]) + } } return } @@ -296,7 +319,6 @@ func (in *JobSpec) DeepCopy() *JobSpec { // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil. func (in *JobStatus) DeepCopyInto(out *JobStatus) { *out = *in - in.Updated.DeepCopyInto(&out.Updated) if in.Errors != nil { in, out := &in.Errors, &out.Errors *out = make([]string, len(*in)) @@ -552,17 +574,10 @@ func (in *S3RepositoryConfig) DeepCopy() *S3RepositoryConfig { func (in *WebhookResponse) DeepCopyInto(out *WebhookResponse) { *out = *in out.TypeMeta = in.TypeMeta - if in.Jobs != nil { - in, out := &in.Jobs, &out.Jobs - *out = make([]JobSpec, len(*in)) - for i := range *in { - (*in)[i].DeepCopyInto(&(*out)[i]) - } - } - if in.IDs != nil { - in, out := &in.IDs, &out.IDs - *out = make([]string, len(*in)) - copy(*out, *in) + if in.Job != nil { + in, out := &in.Job, &out.Job + *out = new(JobSpec) + (*in).DeepCopyInto(*out) } return } diff --git a/pkg/apis/provisioning/v0alpha1/zz_generated.openapi.go b/pkg/apis/provisioning/v0alpha1/zz_generated.openapi.go index fa35e295390..0cd6db94a90 100644 --- a/pkg/apis/provisioning/v0alpha1/zz_generated.openapi.go +++ b/pkg/apis/provisioning/v0alpha1/zz_generated.openapi.go @@ -15,6 +15,7 @@ import ( func GetOpenAPIDefinitions(ref common.ReferenceCallback) map[string]common.OpenAPIDefinition { return map[string]common.OpenAPIDefinition{ "github.com/grafana/grafana/pkg/apis/provisioning/v0alpha1.Author": schema_pkg_apis_provisioning_v0alpha1_Author(ref), + "github.com/grafana/grafana/pkg/apis/provisioning/v0alpha1.CommitInfo": schema_pkg_apis_provisioning_v0alpha1_CommitInfo(ref), "github.com/grafana/grafana/pkg/apis/provisioning/v0alpha1.EditingOptions": schema_pkg_apis_provisioning_v0alpha1_EditingOptions(ref), "github.com/grafana/grafana/pkg/apis/provisioning/v0alpha1.FileItem": schema_pkg_apis_provisioning_v0alpha1_FileItem(ref), "github.com/grafana/grafana/pkg/apis/provisioning/v0alpha1.FileList": schema_pkg_apis_provisioning_v0alpha1_FileList(ref), @@ -74,6 +75,65 @@ func schema_pkg_apis_provisioning_v0alpha1_Author(ref common.ReferenceCallback) } } +func schema_pkg_apis_provisioning_v0alpha1_CommitInfo(ref common.ReferenceCallback) common.OpenAPIDefinition { + return common.OpenAPIDefinition{ + Schema: spec.Schema{ + SchemaProps: spec.SchemaProps{ + Type: []string{"object"}, + Properties: map[string]spec.Schema{ + "sha1": { + SchemaProps: spec.SchemaProps{ + Type: []string{"string"}, + Format: "", + }, + }, + "added": { + 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.FileRef"), + }, + }, + }, + }, + }, + "modified": { + 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.FileRef"), + }, + }, + }, + }, + }, + "removed": { + 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.FileRef"), + }, + }, + }, + }, + }, + }, + }, + }, + Dependencies: []string{ + "github.com/grafana/grafana/pkg/apis/provisioning/v0alpha1.FileRef"}, + } +} + func schema_pkg_apis_provisioning_v0alpha1_EditingOptions(ref common.ReferenceCallback) common.OpenAPIDefinition { return common.OpenAPIDefinition{ Schema: spec.Schema{ @@ -572,40 +632,15 @@ func schema_pkg_apis_provisioning_v0alpha1_JobSpec(ref common.ReferenceCallback) Format: "", }, }, - "added": { + "commits": { SchemaProps: spec.SchemaProps{ - Type: []string{"array"}, + Description: "When we know the commits, these will be passed along", + 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.FileRef"), - }, - }, - }, - }, - }, - "modified": { - 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.FileRef"), - }, - }, - }, - }, - }, - "removed": { - 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.FileRef"), + Ref: ref("github.com/grafana/grafana/pkg/apis/provisioning/v0alpha1.CommitInfo"), }, }, }, @@ -616,7 +651,7 @@ func schema_pkg_apis_provisioning_v0alpha1_JobSpec(ref common.ReferenceCallback) }, }, Dependencies: []string{ - "github.com/grafana/grafana/pkg/apis/provisioning/v0alpha1.FileRef"}, + "github.com/grafana/grafana/pkg/apis/provisioning/v0alpha1.CommitInfo"}, } } @@ -627,11 +662,6 @@ func schema_pkg_apis_provisioning_v0alpha1_JobStatus(ref common.ReferenceCallbac Description: "The job status", Type: []string{"object"}, Properties: map[string]spec.Schema{ - "updated": { - SchemaProps: spec.SchemaProps{ - Ref: ref("k8s.io/apimachinery/pkg/apis/meta/v1.Time"), - }, - }, "state": { SchemaProps: spec.SchemaProps{ Description: "Possible enum values:\n - `\"error\"` Finished with errors\n - `\"pending\"` Job has been submitted, but not processed yet\n - `\"success\"` Finished with success\n - `\"working\"` The job is running", @@ -640,6 +670,18 @@ func schema_pkg_apis_provisioning_v0alpha1_JobStatus(ref common.ReferenceCallbac Enum: []interface{}{"error", "pending", "success", "working"}, }, }, + "started": { + SchemaProps: spec.SchemaProps{ + Type: []string{"integer"}, + Format: "int64", + }, + }, + "finished": { + SchemaProps: spec.SchemaProps{ + Type: []string{"integer"}, + Format: "int64", + }, + }, "message": { SchemaProps: spec.SchemaProps{ Type: []string{"string"}, @@ -663,8 +705,6 @@ func schema_pkg_apis_provisioning_v0alpha1_JobStatus(ref common.ReferenceCallbac }, }, }, - Dependencies: []string{ - "k8s.io/apimachinery/pkg/apis/meta/v1.Time"}, } } @@ -1151,7 +1191,7 @@ func schema_pkg_apis_provisioning_v0alpha1_WebhookResponse(ref common.ReferenceC }, "code": { SchemaProps: spec.SchemaProps{ - Description: "HTTP Status code", + Description: "HTTP Status code 200 implies that the payload was understood but nothing is required 202 implies that an async job has been scheduled to handle the request", Type: []string{"integer"}, Format: "int32", }, @@ -1163,33 +1203,10 @@ func schema_pkg_apis_provisioning_v0alpha1_WebhookResponse(ref common.ReferenceC Format: "", }, }, - "jobs": { + "job": { SchemaProps: spec.SchemaProps{ - Description: "Jobs to be processed", - 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.JobSpec"), - }, - }, - }, - }, - }, - "ids": { - SchemaProps: spec.SchemaProps{ - Description: "The queued jobs", - Type: []string{"array"}, - Items: &spec.SchemaOrArray{ - Schema: &spec.Schema{ - SchemaProps: spec.SchemaProps{ - Default: "", - Type: []string{"string"}, - Format: "", - }, - }, - }, + Description: "Jobs to be processed When the response is 202 (Accepted) the queued jobs will be returned", + Ref: ref("github.com/grafana/grafana/pkg/apis/provisioning/v0alpha1.JobSpec"), }, }, }, diff --git a/pkg/apis/provisioning/v0alpha1/zz_generated.openapi_violation_exceptions.list b/pkg/apis/provisioning/v0alpha1/zz_generated.openapi_violation_exceptions.list index 4e943ca013e..5aa8c603c40 100644 --- a/pkg/apis/provisioning/v0alpha1/zz_generated.openapi_violation_exceptions.list +++ b/pkg/apis/provisioning/v0alpha1/zz_generated.openapi_violation_exceptions.list @@ -1,14 +1,13 @@ +API rule violation: list_type_missing,github.com/grafana/grafana/pkg/apis/provisioning/v0alpha1,CommitInfo,Added +API rule violation: list_type_missing,github.com/grafana/grafana/pkg/apis/provisioning/v0alpha1,CommitInfo,Modified +API rule violation: list_type_missing,github.com/grafana/grafana/pkg/apis/provisioning/v0alpha1,CommitInfo,Removed API rule violation: list_type_missing,github.com/grafana/grafana/pkg/apis/provisioning/v0alpha1,HistoryItem,Authors -API rule violation: list_type_missing,github.com/grafana/grafana/pkg/apis/provisioning/v0alpha1,JobSpec,Added -API rule violation: list_type_missing,github.com/grafana/grafana/pkg/apis/provisioning/v0alpha1,JobSpec,Modified -API rule violation: list_type_missing,github.com/grafana/grafana/pkg/apis/provisioning/v0alpha1,JobSpec,Removed +API rule violation: list_type_missing,github.com/grafana/grafana/pkg/apis/provisioning/v0alpha1,JobSpec,Commits API rule violation: list_type_missing,github.com/grafana/grafana/pkg/apis/provisioning/v0alpha1,JobStatus,Errors API rule violation: list_type_missing,github.com/grafana/grafana/pkg/apis/provisioning/v0alpha1,ResourceWrapper,Errors API rule violation: list_type_missing,github.com/grafana/grafana/pkg/apis/provisioning/v0alpha1,ResourceWrapper,Lint -API rule violation: list_type_missing,github.com/grafana/grafana/pkg/apis/provisioning/v0alpha1,WebhookResponse,IDs -API rule violation: list_type_missing,github.com/grafana/grafana/pkg/apis/provisioning/v0alpha1,WebhookResponse,Jobs +API rule violation: names_match,github.com/grafana/grafana/pkg/apis/provisioning/v0alpha1,CommitInfo,SHA1 API rule violation: names_match,github.com/grafana/grafana/pkg/apis/provisioning/v0alpha1,FileList,Items API rule violation: names_match,github.com/grafana/grafana/pkg/apis/provisioning/v0alpha1,RepositorySpec,GitHub API rule violation: names_match,github.com/grafana/grafana/pkg/apis/provisioning/v0alpha1,RepositorySpec,PreferYAML -API rule violation: names_match,github.com/grafana/grafana/pkg/apis/provisioning/v0alpha1,WebhookResponse,IDs API rule violation: names_match,github.com/grafana/grafana/pkg/apis/provisioning/v0alpha1,WebhookResponse,Message diff --git a/pkg/registry/apis/provisioning/jobs/queue.go b/pkg/registry/apis/provisioning/jobs/queue.go index 61c1e96b4eb..e12980cd251 100644 --- a/pkg/registry/apis/provisioning/jobs/queue.go +++ b/pkg/registry/apis/provisioning/jobs/queue.go @@ -10,18 +10,21 @@ import ( metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/labels" - - "github.com/google/uuid" + "k8s.io/apimachinery/pkg/watch" + "k8s.io/apiserver/pkg/storage" provisioning "github.com/grafana/grafana/pkg/apis/provisioning/v0alpha1" + "github.com/grafana/grafana/pkg/util" ) func NewJobQueue(capacity int) JobQueue { return &jobStore{ - rv: 1, - capacity: capacity, - jobs: []provisioning.Job{}, - logger: slog.Default().With("logger", "job-queue"), + rv: 1, + capacity: capacity, + jobs: []provisioning.Job{}, + watchSet: NewWatchSet(), + versioner: &storage.APIObjectVersioner{}, + logger: slog.Default().With("logger", "job-queue"), } } @@ -31,8 +34,10 @@ type jobStore struct { workers []Worker // All jobs - jobs []provisioning.Job - rv int64 // updates whenever changed + jobs []provisioning.Job + rv int64 // updates whenever changed + watchSet *WatchSet + versioner storage.Versioner mutex sync.RWMutex } @@ -45,25 +50,25 @@ func (s *jobStore) Register(worker Worker) { } // Add implements JobQueue. -func (s *jobStore) Add(ctx context.Context, job provisioning.Job) (string, error) { +func (s *jobStore) Add(ctx context.Context, job *provisioning.Job) (*provisioning.Job, error) { if job.Namespace == "" { - return "", fmt.Errorf("expecting namespace") + return nil, fmt.Errorf("expecting namespace") } _, ok := job.Labels["repository"] if !ok { - return "", fmt.Errorf("expecting repository name label") + return nil, fmt.Errorf("expecting repository name label") } _, ok = job.Labels["repository.type"] // TODO: ideally just based on the action type if !ok { - return "", fmt.Errorf("expecting repository name label") + return nil, fmt.Errorf("expecting repository name label") } if job.Spec.Action == "" { - return "", fmt.Errorf("missing spec.action") + return nil, fmt.Errorf("missing spec.action") } // Only for add if job.Status.State != "" { - return "", fmt.Errorf("must add jobs with empty state") + return nil, fmt.Errorf("must add jobs with empty state") } s.mutex.Lock() @@ -74,33 +79,44 @@ func (s *jobStore) Add(ctx context.Context, job provisioning.Job) (string, error job.Status.State = provisioning.JobStatePending job.Labels["action"] = string(job.Spec.Action) // make it searchable job.Labels["state"] = string(job.Status.State) - job.Name = "j" + uuid.NewString() + job.Name = util.GenerateShortUID() job.CreationTimestamp = metav1.NewTime(time.Now()) jobs := make([]provisioning.Job, 0, len(s.jobs)+2) - jobs = append(jobs, job) + jobs = append(jobs, *job) for i, j := range s.jobs { if i >= s.capacity { - break + // Remove the old jobs + s.watchSet.notifyWatchers(watch.Event{ + Object: j.DeepCopyObject(), + Type: watch.Deleted, + }, nil) + continue } jobs = append(jobs, j) } + // Send add event + s.watchSet.notifyWatchers(watch.Event{ + Object: job.DeepCopyObject(), + Type: watch.Added, + }, nil) + // For now, start a thread processing each job go func() { - time.Sleep(time.Millisecond * 100) + time.Sleep(time.Millisecond * 200) s.drainPending() }() s.jobs = jobs // replace existing list - return job.Name, nil + return job, nil } // Reads the queue until no jobs remain func (s *jobStore) drainPending() { var err error for { - time.Sleep(time.Microsecond * 100) + time.Sleep(time.Microsecond * 200) ctx := context.Background() job := s.Checkout(ctx, nil) @@ -150,12 +166,19 @@ func (s *jobStore) Checkout(ctx context.Context, query labels.Selector) *provisi // The oldest jobs should be checked out first for i := len(s.jobs) - 1; i >= 0; i-- { if s.jobs[i].Status.State == provisioning.JobStatePending { + oldObj := s.jobs[i].DeepCopyObject() + s.rv++ s.jobs[i].ResourceVersion = strconv.FormatInt(s.rv, 10) s.jobs[i].Status.State = provisioning.JobStateWorking s.jobs[i].Labels["state"] = string(provisioning.JobStateWorking) - s.jobs[i].Status.Updated = metav1.NewTime(time.Now()) + s.jobs[i].Status.Started = time.Now().UnixMilli() job := s.jobs[i] + + s.watchSet.notifyWatchers(watch.Event{ + Object: job.DeepCopyObject(), + Type: watch.Modified, + }, oldObj) return &job } } @@ -171,11 +194,18 @@ func (s *jobStore) Complete(ctx context.Context, namespace string, name string, for idx, job := range s.jobs { if job.Name == name && job.Namespace == namespace { - status.Updated = metav1.NewTime(time.Now()) + oldObj := job.DeepCopyObject() + + status.Finished = time.Now().UnixMilli() job.ResourceVersion = strconv.FormatInt(s.rv, 10) job.Status = status job.Labels["state"] = string(status.State) s.jobs[idx] = job + + s.watchSet.notifyWatchers(watch.Event{ + Object: job.DeepCopyObject(), + Type: watch.Modified, + }, oldObj) return nil } } diff --git a/pkg/registry/apis/provisioning/jobs/store.go b/pkg/registry/apis/provisioning/jobs/store.go index 76364290978..9baa5f19699 100644 --- a/pkg/registry/apis/provisioning/jobs/store.go +++ b/pkg/registry/apis/provisioning/jobs/store.go @@ -8,10 +8,13 @@ import ( 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/fields" "k8s.io/apimachinery/pkg/labels" "k8s.io/apimachinery/pkg/runtime" + "k8s.io/apimachinery/pkg/watch" "k8s.io/apiserver/pkg/endpoints/request" "k8s.io/apiserver/pkg/registry/rest" + "k8s.io/apiserver/pkg/storage" provisioning "github.com/grafana/grafana/pkg/apis/provisioning/v0alpha1" ) @@ -22,6 +25,7 @@ var ( _ rest.Getter = (*jobStore)(nil) _ rest.Lister = (*jobStore)(nil) _ rest.Storage = (*jobStore)(nil) + _ rest.Watcher = (*jobStore)(nil) resourceInfo = provisioning.JobResourceInfo ) @@ -99,3 +103,20 @@ func (s *jobStore) Get(ctx context.Context, name string, options *metav1.GetOpti return nil, apierrors.NewNotFound(resourceInfo.GroupResource(), name) } + +func (s *jobStore) Watch(ctx context.Context, opts *internalversion.ListOptions) (watch.Interface, error) { + ns, ok := request.NamespaceFrom(ctx) + if !ok { + return nil, fmt.Errorf("missing namespace") + } + + p := storage.SelectionPredicate{ + Label: labels.Everything(), // TODO... limit + Field: fields.Everything(), + } + + // Can watch by label selection + jw := s.watchSet.newWatch(ctx, 0, p, s.versioner, &ns) + jw.Start() + return jw, nil +} diff --git a/pkg/registry/apis/provisioning/jobs/types.go b/pkg/registry/apis/provisioning/jobs/types.go index 59262034c53..55d76d1782e 100644 --- a/pkg/registry/apis/provisioning/jobs/types.go +++ b/pkg/registry/apis/provisioning/jobs/types.go @@ -17,7 +17,7 @@ var ( type JobQueue interface { rest.Storage // temporary.. simplifies registration - Add(ctx context.Context, job provisioning.Job) (string, error) + Add(ctx context.Context, job *provisioning.Job) (*provisioning.Job, error) Checkout(ctx context.Context, query labels.Selector) *provisioning.Job Complete(ctx context.Context, namespace string, name string, status provisioning.JobStatus) error diff --git a/pkg/registry/apis/provisioning/jobs/watchset.go b/pkg/registry/apis/provisioning/jobs/watchset.go new file mode 100644 index 00000000000..f9a2934482d --- /dev/null +++ b/pkg/registry/apis/provisioning/jobs/watchset.go @@ -0,0 +1,379 @@ +// SPDX-License-Identifier: AGPL-3.0-only +// Provenance-includes-location: https://github.com/tilt-dev/tilt-apiserver/blob/main/pkg/storage/filepath/watchset.go +// Provenance-includes-license: Apache-2.0 +// Provenance-includes-copyright: The Kubernetes Authors. + +// See also +// https://github.com/grafana/grafana/blob/v11.1.9/pkg/apiserver/storage/file/watchset.go + +package jobs + +import ( + "context" + "fmt" + "sync" + "sync/atomic" + + "k8s.io/apimachinery/pkg/api/meta" + "k8s.io/apimachinery/pkg/runtime" + "k8s.io/apimachinery/pkg/watch" + "k8s.io/apiserver/pkg/storage" + "k8s.io/klog/v2" +) + +const ( + UpdateChannelSize = 25 + InitialWatchNodesSize = 20 + InitialBufferedEventsSize = 25 +) + +type eventWrapper struct { + ev watch.Event + // optional: oldObject is only set for modifications for determining their type as necessary (when using predicate filtering) + oldObject runtime.Object +} + +type watchNode struct { + ctx context.Context + s *WatchSet + id uint64 + updateCh chan eventWrapper + outCh chan watch.Event + requestedRV uint64 + // the watch may or may not be namespaced for a namespaced resource. This is always nil for cluster-scoped kinds + watchNamespace *string + predicate storage.SelectionPredicate + versioner storage.Versioner +} + +// Keeps track of which watches need to be notified +type WatchSet struct { + mu sync.RWMutex + // mu protects both nodes and counter + nodes map[uint64]*watchNode + counter atomic.Uint64 + buffered []eventWrapper + bufferedMutex sync.RWMutex +} + +func NewWatchSet() *WatchSet { + return &WatchSet{ + buffered: make([]eventWrapper, 0, InitialBufferedEventsSize), + nodes: make(map[uint64]*watchNode, InitialWatchNodesSize), + } +} + +// Creates a new watch with a unique id, but +// does not start sending events to it until start() is called. +func (s *WatchSet) newWatch(ctx context.Context, requestedRV uint64, p storage.SelectionPredicate, versioner storage.Versioner, namespace *string) *watchNode { + s.counter.Add(1) + + node := &watchNode{ + ctx: ctx, + requestedRV: requestedRV, + id: s.counter.Load(), + s: s, + // updateCh size needs to be > 1 to allow slower clients to not block passing new events + updateCh: make(chan eventWrapper, UpdateChannelSize), + // outCh size needs to be > 1 for single process use-cases such as tests where watch and event seeding from CUD + // events is happening on the same thread + outCh: make(chan watch.Event, UpdateChannelSize), + predicate: p, + watchNamespace: namespace, + versioner: versioner, + } + + return node +} + +func (s *WatchSet) CleanupWatchers() { + s.mu.Lock() + defer s.mu.Unlock() + for _, w := range s.nodes { + w.stop() + } +} + +// oldObject is only passed in the event of a modification +// in case a predicate filtered watch is impacted as a result of modification +// NOTE: this function gives one the misperception that a newly added node will never +// get a double event, one from buffered and one from the update channel +// That perception is not true. Even though this function maintains the lock throughout the function body +// it is not true of the Start function. So basically, the Start function running after this function +// fully stands the chance of another future notifyWatchers double sending it the event through the two means mentioned +func (s *WatchSet) notifyWatchers(ev watch.Event, oldObject runtime.Object) { + s.mu.RLock() + defer s.mu.RUnlock() + + updateEv := eventWrapper{ + ev: ev, + } + if oldObject != nil { + updateEv.oldObject = oldObject + } + + // Events are always buffered. + // this is because of an inadvertent delay which is built into the watch process + // Watch() from storage returns Watch.Interface with a async start func. + // The only way to guarantee that we can interpret the passed RV correctly is to play it against missed events + // (notice the loop below over s.nodes isn't exactly going to work on a new node + // unless start is called on it) + s.bufferedMutex.Lock() + s.buffered = append(s.buffered, updateEv) + s.bufferedMutex.Unlock() + + for _, w := range s.nodes { + w.updateCh <- updateEv + } +} + +// isValid is not necessary to be called on oldObject in UpdateEvents - assuming the Watch pushes correctly setup eventWrapper our way +// first bool is whether the event is valid for current watcher +// second bool is whether checking the old value against the predicate may be valuable to the caller +// second bool may be a helpful aid to establish context around MODIFIED events +// (note that this second bool is only marked true if we pass other checks first, namely RV and namespace) +func (w *watchNode) isValid(e eventWrapper) (bool, bool, error) { + obj, err := meta.Accessor(e.ev.Object) + if err != nil { + klog.Error("Could not get accessor to object in event") + return false, false, nil + } + + eventRV, err := w.getResourceVersionAsInt(e.ev.Object) + if err != nil { + return false, false, err + } + + if eventRV < w.requestedRV { + return false, false, nil + } + + if w.watchNamespace != nil && *w.watchNamespace != obj.GetNamespace() { + return false, false, err + } + + valid, err := w.predicate.Matches(e.ev.Object) + if err != nil { + return false, false, err + } + + return valid, e.ev.Type == watch.Modified, nil +} + +// Only call this method if current object matches the predicate +func (w *watchNode) handleAddedForFilteredList(e eventWrapper) (*watch.Event, error) { + if e.oldObject == nil { + return nil, fmt.Errorf("oldObject should be set for modified events") + } + + ok, err := w.predicate.Matches(e.oldObject) + if err != nil { + return nil, err + } + + if !ok { + e.ev.Type = watch.Added + return &e.ev, nil + } + + return nil, nil +} + +func (w *watchNode) handleDeletedForFilteredList(e eventWrapper) (*watch.Event, error) { + if e.oldObject == nil { + return nil, fmt.Errorf("oldObject should be set for modified events") + } + + ok, err := w.predicate.Matches(e.oldObject) + if err != nil { + return nil, err + } + + if !ok { + return nil, nil + } + + // isn't a match but used to be + e.ev.Type = watch.Deleted + + oldObjectAccessor, err := meta.Accessor(e.oldObject) + if err != nil { + klog.Errorf("Could not get accessor to correct the old RV of filtered out object") + return nil, err + } + + currentRV, err := getResourceVersion(e.ev.Object) + if err != nil { + klog.Errorf("Could not get accessor to object in event") + return nil, err + } + + oldObjectAccessor.SetResourceVersion(currentRV) + e.ev.Object = e.oldObject + + return &e.ev, nil +} + +func (w *watchNode) processEvent(e eventWrapper, isInitEvent bool) error { + if isInitEvent { + // Init events have already been vetted against the predicate and other RV behavior + // Let them pass through + w.outCh <- e.ev + return nil + } + + valid, runDeleteFromFilteredListHandler, err := w.isValid(e) + if err != nil { + klog.Errorf("Could not determine validity of the event: %v", err) + return err + } + if valid { + if e.ev.Type == watch.Modified { + ev, err := w.handleAddedForFilteredList(e) + if err != nil { + return err + } + if ev != nil { + w.outCh <- *ev + } else { + // forward the original event if add handling didn't signal any impact + w.outCh <- e.ev + } + } else { + w.outCh <- e.ev + } + return nil + } + + if runDeleteFromFilteredListHandler { + if e.ev.Type == watch.Modified { + ev, err := w.handleDeletedForFilteredList(e) + if err != nil { + return err + } + if ev != nil { + w.outCh <- *ev + } + } // explicitly doesn't have an event forward for the else case here + return nil + } + + return nil +} + +// Start sending events to this watch. +func (w *watchNode) Start(initEvents ...watch.Event) { + w.s.mu.Lock() + w.s.nodes[w.id] = w + w.s.mu.Unlock() + + go func() { + maxRV := uint64(0) + for _, ev := range initEvents { + currentRV, err := w.getResourceVersionAsInt(ev.Object) + if err != nil { + klog.Errorf("Could not determine init event RV for deduplication of buffered events: %v", err) + continue + } + + if maxRV < currentRV { + maxRV = currentRV + } + + if err := w.processEvent(eventWrapper{ev: ev}, true); err != nil { + klog.Errorf("Could not process event: %v", err) + } + } + + // If we had no init events, simply rely on the passed RV + if maxRV == 0 { + maxRV = w.requestedRV + } + + w.s.bufferedMutex.RLock() + for _, e := range w.s.buffered { + eventRV, err := w.getResourceVersionAsInt(e.ev.Object) + if err != nil { + klog.Errorf("Could not determine RV for deduplication of buffered events: %v", err) + continue + } + + if maxRV >= eventRV { + continue + } else { + maxRV = eventRV + } + + if err := w.processEvent(e, false); err != nil { + klog.Errorf("Could not process event: %v", err) + } + } + w.s.bufferedMutex.RUnlock() + + for { + select { + case e, ok := <-w.updateCh: + if !ok { + close(w.outCh) + return + } + + eventRV, err := w.getResourceVersionAsInt(e.ev.Object) + if err != nil { + klog.Errorf("Could not determine RV for deduplication of channel events: %v", err) + continue + } + + if maxRV >= eventRV { + continue + } else { + maxRV = eventRV + } + + if err := w.processEvent(e, false); err != nil { + klog.Errorf("Could not process event: %v", err) + } + case <-w.ctx.Done(): + close(w.outCh) + return + } + } + }() +} + +func (w *watchNode) Stop() { + w.s.mu.Lock() + defer w.s.mu.Unlock() + w.stop() +} + +// Unprotected func: ensure mutex on the parent watch set is locked before calling +func (w *watchNode) stop() { + if _, ok := w.s.nodes[w.id]; ok { + delete(w.s.nodes, w.id) + close(w.updateCh) + } +} + +func (w *watchNode) ResultChan() <-chan watch.Event { + return w.outCh +} + +func getResourceVersion(obj runtime.Object) (string, error) { + accessor, err := meta.Accessor(obj) + if err != nil { + klog.Error("Could not get accessor to object in event") + return "", err + } + return accessor.GetResourceVersion(), nil +} + +func (w *watchNode) getResourceVersionAsInt(obj runtime.Object) (uint64, error) { + accessor, err := meta.Accessor(obj) + if err != nil { + klog.Error("Could not get accessor to object in event") + return 0, err + } + + return w.versioner.ParseResourceVersion(accessor.GetResourceVersion()) +} diff --git a/pkg/registry/apis/provisioning/repository/github.go b/pkg/registry/apis/provisioning/repository/github.go index 623288d0bb3..171e173ff0f 100644 --- a/pkg/registry/apis/provisioning/repository/github.go +++ b/pkg/registry/apis/provisioning/repository/github.go @@ -431,23 +431,23 @@ func (r *githubRepository) parsePushEvent(event *github.PushEvent) (*provisionin return &provisioning.WebhookResponse{Code: http.StatusOK}, nil } - count := 0 - - jobs := make([]provisioning.JobSpec, 0, len(event.Commits)) + job := &provisioning.JobSpec{ + Action: provisioning.JobActionMergeBranch, + } beforeRef := event.GetBefore() for _, commit := range event.Commits { - job := provisioning.JobSpec{ - Action: provisioning.JobActionMergeBranch, - Ref: commit.GetID(), + commitInfo := provisioning.CommitInfo{ + SHA1: commit.GetID(), } + count := 0 for _, file := range commit.Added { if r.ignore(file) { continue } - job.Added = append(job.Added, provisioning.FileRef{ + commitInfo.Added = append(commitInfo.Added, provisioning.FileRef{ Ref: commit.GetID(), Path: file, }) @@ -458,7 +458,7 @@ func (r *githubRepository) parsePushEvent(event *github.PushEvent) (*provisionin if r.ignore(file) { continue } - job.Modified = append(job.Modified, provisioning.FileRef{ + commitInfo.Modified = append(commitInfo.Modified, provisioning.FileRef{ Ref: commit.GetID(), Path: file, }) @@ -470,18 +470,20 @@ func (r *githubRepository) parsePushEvent(event *github.PushEvent) (*provisionin continue } - job.Removed = append(job.Removed, provisioning.FileRef{ + commitInfo.Removed = append(commitInfo.Removed, provisioning.FileRef{ Ref: beforeRef, Path: file, }) count++ } - jobs = append(jobs, job) + if count > 0 { + job.Commits = append(job.Commits, commitInfo) + } beforeRef = commit.GetID() } - if count == 0 { + if len(job.Commits) == 0 { return &provisioning.WebhookResponse{ Code: http.StatusOK, // Nothing needed Message: "no files require updates", @@ -489,9 +491,8 @@ func (r *githubRepository) parsePushEvent(event *github.PushEvent) (*provisionin } return &provisioning.WebhookResponse{ - Code: http.StatusAccepted, - Message: fmt.Sprintf("%d files", count), - Jobs: jobs, + Code: http.StatusAccepted, + Job: job, }, nil } @@ -538,14 +539,12 @@ func (r *githubRepository) parsePullRequestEvent(event *github.PullRequestEvent) return &provisioning.WebhookResponse{ Code: http.StatusAccepted, // Nothing needed Message: fmt.Sprintf("pull request: %s", action), - Jobs: []provisioning.JobSpec{ - { - Action: provisioning.JobActionPullRequest, - URL: pr.GetHTMLURL(), - PR: pr.GetNumber(), - Ref: pr.GetHead().GetRef(), - Hash: pr.GetHead().GetSHA(), - }, + Job: &provisioning.JobSpec{ + Action: provisioning.JobActionPullRequest, + URL: pr.GetHTMLURL(), + PR: pr.GetNumber(), + Ref: pr.GetHead().GetRef(), + Hash: pr.GetHead().GetSHA(), }, }, nil } @@ -564,47 +563,47 @@ func (r *githubRepository) Process(ctx context.Context, logger *slog.Logger, wra return r.processPR(ctx, logger, job, replicator) } - // ??????? - // Is everything below here generic? not github specific? + // NOTE: Everything below here is not git specific + for _, commit := range job.Commits { + // NOT PR, this is processing the actual files + for _, v := range commit.Added { + fileInfo, err := r.Read(ctx, logger, v.Path, v.Ref) + if err != nil { + logger.ErrorContext(ctx, "failed to read added resource", "file", v, "error", err) + continue + } - // NOT PR, this is processing the actual files - for _, v := range job.Added { - fileInfo, err := r.Read(ctx, logger, v.Path, v.Ref) - if err != nil { - logger.ErrorContext(ctx, "failed to read added resource", "file", v, "error", err) - continue + err = replicator.Replicate(ctx, fileInfo) + if err != nil { + logger.ErrorContext(ctx, "failed to replicate added resource", "file", v, "error", err) + continue + } + } + for _, v := range commit.Modified { + fileInfo, err := r.Read(ctx, logger, v.Path, v.Ref) + if err != nil { + logger.ErrorContext(ctx, "failed to read modified resource", "file", v, "error", err) + continue + } + + err = replicator.Replicate(ctx, fileInfo) + if err != nil { + logger.ErrorContext(ctx, "failed to replicate modified resource", "file", v, "error", err) + continue + } } - err = replicator.Replicate(ctx, fileInfo) - if err != nil { - logger.ErrorContext(ctx, "failed to replicate added resource", "file", v, "error", err) - continue - } - } - for _, v := range job.Modified { - fileInfo, err := r.Read(ctx, logger, v.Path, v.Ref) - if err != nil { - logger.ErrorContext(ctx, "failed to read modified resource", "file", v, "error", err) - continue - } + for _, v := range commit.Removed { + fileInfo, err := r.Read(ctx, logger, v.Path, v.Ref) + if err != nil { + logger.ErrorContext(ctx, "failed to read removed resource", "file", v, "error", err) + continue + } - err = replicator.Replicate(ctx, fileInfo) - if err != nil { - logger.ErrorContext(ctx, "failed to replicate modified resource", "file", v, "error", err) - continue - } - } - - for _, v := range job.Removed { - fileInfo, err := r.Read(ctx, logger, v.Path, v.Ref) - if err != nil { - logger.ErrorContext(ctx, "failed to read removed resource", "file", v, "error", err) - continue - } - - if err := replicator.Delete(ctx, fileInfo); err != nil { - logger.ErrorContext(ctx, "failed to delete resource", "file", v, "error", err) - continue + if err := replicator.Delete(ctx, fileInfo); err != nil { + logger.ErrorContext(ctx, "failed to delete resource", "file", v, "error", err) + continue + } } } diff --git a/pkg/registry/apis/provisioning/repository/github_test.go b/pkg/registry/apis/provisioning/repository/github_test.go index 6869b859c28..2c2833183a3 100644 --- a/pkg/registry/apis/provisioning/repository/github_test.go +++ b/pkg/registry/apis/provisioning/repository/github_test.go @@ -65,14 +65,12 @@ func TestParseWebhooks(t *testing.T) { }}, {"pull_request", "opened", provisioning.WebhookResponse{ Code: http.StatusAccepted, // 202 - Jobs: []provisioning.JobSpec{ - { - Action: provisioning.JobActionPullRequest, - Ref: "dashboard/1733653266690", - Hash: "ab5446a53df9e5f8bdeed52250f51fad08e822bc", - PR: 12, - URL: "https://github.com/grafana/git-ui-sync-demo/pull/12", - }, + Job: &provisioning.JobSpec{ + Action: provisioning.JobActionPullRequest, + Ref: "dashboard/1733653266690", + Hash: "ab5446a53df9e5f8bdeed52250f51fad08e822bc", + PR: 12, + URL: "https://github.com/grafana/git-ui-sync-demo/pull/12", }, }}, {"push", "ignored", provisioning.WebhookResponse{ @@ -80,10 +78,10 @@ func TestParseWebhooks(t *testing.T) { }}, {"push", "nested", provisioning.WebhookResponse{ Code: http.StatusAccepted, - Jobs: []provisioning.JobSpec{ - { - Action: provisioning.JobActionMergeBranch, - Ref: "5c816f9812e391c62b0c5555d0b473b296d9179c", + Job: &provisioning.JobSpec{ + Action: provisioning.JobActionMergeBranch, + Commits: []provisioning.CommitInfo{{ + SHA1: "5c816f9812e391c62b0c5555d0b473b296d9179c", Added: []provisioning.FileRef{ { Ref: "5c816f9812e391c62b0c5555d0b473b296d9179c", @@ -100,7 +98,7 @@ func TestParseWebhooks(t *testing.T) { Path: "first-dashboard.json", }, }, - }, + }}, }, }}, {"issue_comment", "created", provisioning.WebhookResponse{ @@ -136,7 +134,7 @@ func TestParseWebhooks(t *testing.T) { require.NoError(t, err) require.Equal(t, tt.expected.Code, rsp.Code) - require.Equal(t, tt.expected.Jobs, rsp.Jobs) + require.Equal(t, tt.expected.Job, rsp.Job) }) } } diff --git a/pkg/registry/apis/provisioning/repository/local.go b/pkg/registry/apis/provisioning/repository/local.go index 1442f663063..cfc06b750b1 100644 --- a/pkg/registry/apis/provisioning/repository/local.go +++ b/pkg/registry/apis/provisioning/repository/local.go @@ -279,9 +279,9 @@ func (r *localRepository) History(ctx context.Context, logger *slog.Logger, path func (r *localRepository) Webhook(ctx context.Context, logger *slog.Logger, req *http.Request) (*provisioning.WebhookResponse, error) { return &provisioning.WebhookResponse{ Code: http.StatusAccepted, - Jobs: []provisioning.JobSpec{{ + Job: &provisioning.JobSpec{ Action: provisioning.JobActionMergeBranch, // sync the latest changes - }}, + }, }, nil } diff --git a/pkg/registry/apis/provisioning/webhook.go b/pkg/registry/apis/provisioning/webhook.go index 9bbff67f288..c87f5757010 100644 --- a/pkg/registry/apis/provisioning/webhook.go +++ b/pkg/registry/apis/provisioning/webhook.go @@ -74,27 +74,25 @@ func (s *webhookConnector) Connect(ctx context.Context, name string, opts runtim responder.Error(fmt.Errorf("expecting a response")) return } - if rsp.Jobs != nil { + if rsp.Job != nil { // Add the job to the job queue - for _, spec := range rsp.Jobs { - job := provisioning.Job{ - ObjectMeta: v1.ObjectMeta{ - Namespace: namespace, - Labels: map[string]string{ - "repository": name, - "repository.type": string(repo.Config().Spec.Type), - }, + job := &provisioning.Job{ + ObjectMeta: v1.ObjectMeta{ + Namespace: namespace, + Labels: map[string]string{ + "repository": name, + "repository.type": string(repo.Config().Spec.Type), }, - Spec: spec, - } - id, err := s.jobs.Add(ctx, job) - if err != nil { - responder.Error(err) - return - } - job.Name = id - rsp.IDs = append(rsp.IDs, id) + }, + Spec: *rsp.Job, } + job, err := s.jobs.Add(ctx, job) + if err != nil { + responder.Error(err) + return + } + responder.Object(rsp.Code, job) + return } responder.Object(rsp.Code, rsp) }), nil diff --git a/pkg/tests/apis/provisioning/provisioning_test.go b/pkg/tests/apis/provisioning/provisioning_test.go index 3dd15a96a87..c304d212b49 100644 --- a/pkg/tests/apis/provisioning/provisioning_test.go +++ b/pkg/tests/apis/provisioning/provisioning_test.go @@ -81,7 +81,8 @@ func TestIntegrationProvisioning(t *testing.T) { "kind": "Job", "verbs": [ "get", - "list" + "list", + "watch" ] }, {