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

POC/Provisioning: Add multiple commits to one job w/ Watch (#97716)

Browse Source
This commit is contained in:
Ryan McKinley authored and GitHub committed 2024-12-10 13:10:34 +03:00
1 parent 9e734d8c94
commit ef5dacd289
14 files changed
+690 -221
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

pkg/apis/provisioning/v0alpha1/jobs.go
+16 -8
View File
@@ -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"`
}
pkg/apis/provisioning/v0alpha1/register.go
+4
View File
@@ -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
},
})
pkg/apis/provisioning/v0alpha1/zz_generated.deepcopy.go
+41 -26
View File
@@ -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
}
pkg/apis/provisioning/v0alpha1/zz_generated.openapi.go
+81 -64
View File
@@ -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"),
},
},
},
pkg/apis/provisioning/v0alpha1/zz_generated.openapi_violation_exceptions.list
+5 -6
View File
@@ -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
pkg/registry/apis/provisioning/jobs/queue.go
+52 -22
View File
@@ -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
}
}
pkg/registry/apis/provisioning/jobs/store.go
+21
View File
@@ -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
}
pkg/registry/apis/provisioning/jobs/types.go
+1 -1
View File
@@ -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
pkg/registry/apis/provisioning/jobs/watchset.go
+379
View File
@@ -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())
}
pkg/registry/apis/provisioning/repository/github.go
+58 -59
View File
@@ -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
}
}
}
pkg/registry/apis/provisioning/repository/github_test.go
+12 -14
View File
@@ -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)
})
}
}
pkg/registry/apis/provisioning/repository/local.go
+2 -2
View File
@@ -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
}
pkg/registry/apis/provisioning/webhook.go
+16 -18
View File
@@ -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
pkg/tests/apis/provisioning/provisioning_test.go
+2 -1
View File
@@ -81,7 +81,8 @@ func TestIntegrationProvisioning(t *testing.T) {
"kind": "Job",
"verbs": [
"get",
"list"
"list",
"watch"
]
},
{