From a8dc50d75b54ccc0bb86934397bab9fa56984693 Mon Sep 17 00:00:00 2001 From: Stephan Renatus Date: Thu, 2 Oct 2025 09:13:30 +0200 Subject: [PATCH 1/3] service: add Trigger(name) to trigger bundle rebuild Adds the machinery to pool and service to trigger bundle rebuilds by name. If the bundle worker is queued, it'll be pulled into the front of the line, to be picked up next; if it's currently not in the queue, it must be executing: if so, we're overriding the next deadline to be _now_, causing an immediate re-run of the build process. For now, this is only wired up with signal handling: send a HUP signal to rebuild. This is because we haven't discussed proper permissions and HTTP API design for this yet. Signed-off-by: Stephan Renatus --- cmd/run/run.go | 19 ++++++++ config/schema.json | 12 +++++ internal/config/config.go | 10 +++++ internal/pool/pool.go | 72 ++++++++++++++++++++++-------- internal/pool/pool_test.go | 62 ++++++++++++++++++++++++-- internal/service/service.go | 87 +++++++++++++++++++++++++------------ internal/service/worker.go | 8 ++-- 7 files changed, 219 insertions(+), 51 deletions(-) diff --git a/cmd/run/run.go b/cmd/run/run.go index 9832fc37..60a0dce6 100644 --- a/cmd/run/run.go +++ b/cmd/run/run.go @@ -1,7 +1,10 @@ package cmd import ( + "context" "os" + "os/signal" + "syscall" "github.com/open-policy-agent/opa-control-plane/cmd" "github.com/open-policy-agent/opa-control-plane/cmd/internal/flags" @@ -70,6 +73,10 @@ func init() { log.Fatalf("initialize service: %v", err) } + // NB(sr): Preliminary, not necessarily something we'll want to keep: + // Rebuild all bundles on SIGHUP. + signalTrigger(svc, log) + go func() { if err := server.New().WithDatabase(svc.Database()).WithReadiness(svc.Ready).WithConfig(config.Service).Init().ListenAndServe(params.addr); err != nil { log.Fatalf("failed to start server: %v", err) @@ -94,3 +101,15 @@ func init() { run, ) } + +func signalTrigger(s *service.Service, l *logging.Logger) { + sigs := make(chan os.Signal, 1) + signal.Notify(sigs, syscall.SIGHUP) + go func() { + for range sigs { + if err := s.TriggerAll(context.Background()); err != nil { + l.Error(err.Error()) + } + } + }() +} diff --git a/config/schema.json b/config/schema.json index 1321c0c5..e0c5988b 100644 --- a/config/schema.json +++ b/config/schema.json @@ -335,6 +335,18 @@ "api_prefix": { "pattern": "^/([^/].*[^/])?$", "type": "string" + }, + "bundle_rebuild_interval": { + "type": [ + "null", + "integer" + ] + }, + "reconfiguration_interval": { + "type": [ + "null", + "integer" + ] } }, "type": "object" diff --git a/internal/config/config.go b/internal/config/config.go index 5106dd79..31383afd 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -1072,6 +1072,16 @@ type Service struct { // For example `/my/path` will make health endpoint be accessible under `/my/path/health` ApiPrefix string `json:"api_prefix,omitempty" pattern:"^/([^/].*[^/])?$"` + // ReconfigurationInterval is the duration between configuration checks, i.e. when a change + // to a bundle/stack/source will have an effect on the internal bundle workers. + // String of a duration, e.g. "1m". Defaults to "15s". + ReconfigurationInterval *time.Duration `json:"reconfiguration_interval,omitempty" yaml:"reconfiguration_interval,omitempty"` + + // BundleRebuildInterval is the time between bundle builds: After a bundle build as finished, + // OCP will wait _this long_ until it's build again (unless the bundle build is triggered by + // other means). String duration, e.g. "90s". Defaults to "30s". + BundleRebuildInterval *time.Duration `json:"bundle_rebuild_interval,omitempty" yaml:"bundle_rebuild_interval,omitempty"` + _ struct{} `additionalProperties:"false"` } diff --git a/internal/pool/pool.go b/internal/pool/pool.go index 961eeb34..512a6c99 100644 --- a/internal/pool/pool.go +++ b/internal/pool/pool.go @@ -2,6 +2,7 @@ package pool import ( "context" + "fmt" "slices" "sync" "time" @@ -15,17 +16,20 @@ import ( // the waiting goroutine to process the new task immediately. type Pool struct { mu sync.Mutex - tasks []*task + queue []*task + reg map[string]*task wait chan struct{} } type task struct { + name string fn func(context.Context) time.Time deadline time.Time + rerun bool } func New(workers int) *Pool { - var pool Pool + pool := Pool{reg: make(map[string]*task)} for range workers { go pool.work() @@ -34,8 +38,8 @@ func New(workers int) *Pool { return &pool } -func (p *Pool) Add(fn func(context.Context) time.Time) { - p.enqueue(&task{fn: fn, deadline: time.Now()}) +func (p *Pool) Add(name string, fn func(context.Context) time.Time) { + p.enqueue(&task{name: name, fn: fn, deadline: time.Now()}) } // work is the main loop for each worker goroutine. @@ -46,18 +50,34 @@ func (p *Pool) work() { } } -func (p *Pool) enqueue(t *task) { - if t.deadline.IsZero() { - // Task requested removal from the pool. - return - } - +// Trigger runs the named task NOW, if it is in the queue, regardless of the +// previous deadline, by pulling it into the front of the queue. If the named +// task is not queued, it's running. In that case, we'll have it override its +// next deadline to NOW, causing an immediate re-run after the current run. +// Subsequent runs will use the deadline returned by the task's `fn`. +func (p *Pool) Trigger(n string) error { p.mu.Lock() defer p.mu.Unlock() + if i := slices.IndexFunc(p.queue, func(t *task) bool { return t.name == n }); i != -1 { + p.queue[i].deadline = time.Now() + p.sortAndWake() + return nil + } + // if it's not in p.queue, it must be running at the moment + if t, ok := p.reg[n]; ok { + t.rerun = true + return nil + } + + return fmt.Errorf("no task with name %s", n) +} + +// sortAndWake is used in multiple places, but always needs to be run +// within a p.mu lock! +func (p *Pool) sortAndWake() { // Maintain the tasks in deadline order. - p.tasks = append(p.tasks, t) - slices.SortFunc(p.tasks, func(a, b *task) int { + slices.SortFunc(p.queue, func(a, b *task) int { return a.deadline.Compare(b.deadline) }) @@ -68,6 +88,20 @@ func (p *Pool) enqueue(t *task) { } } +func (p *Pool) enqueue(t *task) { + if t.deadline.IsZero() { + // Task requested removal from the pool. + delete(p.reg, t.name) + return + } + + p.mu.Lock() + p.reg[t.name] = t + p.queue = append(p.queue, t) + p.sortAndWake() + p.mu.Unlock() +} + func (p *Pool) dequeue() *task { p.mu.Lock() defer p.mu.Unlock() @@ -75,10 +109,10 @@ func (p *Pool) dequeue() *task { for { var t *task - if len(p.tasks) == 0 { - t = &task{deadline: time.Now().Add(time.Hour * 24 * 365)} // Default to a far future deadline + if len(p.queue) == 0 { + t = &task{name: "dummy", deadline: time.Now().Add(time.Hour * 24 * 365)} // Default to a far future deadline } else { - t = p.tasks[0] + t = p.queue[0] } if t.deadline.After(time.Now()) { @@ -105,12 +139,16 @@ func (p *Pool) dequeue() *task { break } - t := p.tasks[0] - p.tasks = slices.Delete(p.tasks, 0, 1) + var t *task + t, p.queue = p.queue[0], p.queue[1:] return t } func (t *task) Execute(ctx context.Context) *task { t.deadline = t.fn(ctx) + if t.rerun { + t.rerun = false + t.deadline = time.Now() + } return t } diff --git a/internal/pool/pool_test.go b/internal/pool/pool_test.go index 415df57b..e5a999d8 100644 --- a/internal/pool/pool_test.go +++ b/internal/pool/pool_test.go @@ -10,17 +10,17 @@ func TestPool(t *testing.T) { p := New(2) // Add a task that returns a deadline in the future. - p.Add(func(_ context.Context) time.Time { + p.Add("a", func(context.Context) time.Time { return time.Now().Add(100 * time.Millisecond) }) // Add a task that returns a deadline in the past. - p.Add(func(_ context.Context) time.Time { + p.Add("b", func(context.Context) time.Time { return time.Now().Add(-100 * time.Millisecond) }) // Add a task that returns a deadline in the future. - p.Add(func(_ context.Context) time.Time { + p.Add("c", func(context.Context) time.Time { return time.Now().Add(200 * time.Millisecond) }) @@ -28,5 +28,61 @@ func TestPool(t *testing.T) { time.Sleep(300 * time.Millisecond) // The pool should have processed all tasks without deadlock. + // If it had gotten stuck, we'd never reach this line. t.Log("All tasks processed successfully") } + +type run struct { + left int + ran int + sleep time.Duration + deadline time.Duration +} + +func (t *run) Execute(context.Context) time.Time { + if t.left > 0 { + time.Sleep(t.sleep) + t.left-- + t.ran++ + return time.Now().Add(t.deadline) + } + + var zero time.Time + return zero // dequeue task +} + +func TestTrigger(t *testing.T) { + t.Run("trigger pulls queued task up from", func(t *testing.T) { + p := New(2) + + rx := &run{left: 3, deadline: 200 * time.Millisecond} + + p.Add("t", rx.Execute) // will run once (run #1), and be queued for 200 ms + + _ = p.Trigger("t") // pulled in front, run #2 + time.Sleep(50 * time.Millisecond) + _ = p.Trigger("t") // pulled in front, run #3 + time.Sleep(300 * time.Millisecond) // no other runs, third run dequeued + + if exp, act := 3, rx.ran; exp != act { + t.Errorf("expected counter of %d, got %d", exp, act) + } + }) + + t.Run("trigger reruns executing task right away", func(t *testing.T) { + p := New(2) + + // if it wasn't triggered, we'd not see a second run: the next deadline is 1s + rx := &run{left: 3, sleep: 100 * time.Millisecond, deadline: time.Second} + + p.Add("t", rx.Execute) // will run once (run #1), and be queued for 200 ms + time.Sleep(50 * time.Millisecond) + _ = p.Trigger("t") // re-run after it's done, run #2 + + time.Sleep(300 * time.Millisecond) + + if exp, act := 2, rx.ran; exp != act { + t.Errorf("expected counter of %d, got %d", exp, act) + } + }) +} diff --git a/internal/service/service.go b/internal/service/service.go index 3af61e80..d65040ab 100644 --- a/internal/service/service.go +++ b/internal/service/service.go @@ -35,9 +35,11 @@ import ( ) const ( - internalPrincipal = "internal" - defaultTenant = "default" - reconfigurationInterval = 15 * time.Second + internalPrincipal = "internal" + defaultTenant = "default" + reconfigurationInterval = 15 * time.Second + defaultReconfigurationInterval = 15 * time.Second + defaultBuildInterval = 30 * time.Second ) var ( @@ -45,21 +47,23 @@ var ( ) type Service struct { - config *config.Root - persistenceDir string - pool *pool.Pool - workers map[string]*BundleWorker - readyMutex sync.Mutex - ready bool - failures map[string]Status - database database.Database - builtinFS fs.FS - singleShot bool - report *Report - log *logging.Logger - noninteractive bool - migrateDB bool - initialized bool + config *config.Root + persistenceDir string + pool *pool.Pool + workers map[string]*BundleWorker + readyMutex sync.Mutex + ready bool + failures map[string]Status + database database.Database + builtinFS fs.FS + singleShot bool + report *Report + log *logging.Logger + noninteractive bool + migrateDB bool + initialized bool + reconfigurationInterval time.Duration + buildInterval time.Duration } type Report struct { @@ -106,11 +110,13 @@ type Status struct { func New() *Service { return &Service{ - pool: pool.New(10), - workers: make(map[string]*BundleWorker), - failures: make(map[string]Status), - noninteractive: true, - migrateDB: false, + pool: pool.New(10), + workers: make(map[string]*BundleWorker), + failures: make(map[string]Status), + noninteractive: true, + migrateDB: false, + reconfigurationInterval: defaultReconfigurationInterval, + buildInterval: defaultBuildInterval, } } @@ -122,6 +128,14 @@ func (s *Service) WithPersistenceDir(d string) *Service { func (s *Service) WithConfig(config *config.Root) *Service { s.config = config s.database = *s.database.WithConfig(config.Database) + if s.config.Service != nil { + if s.config.Service.ReconfigurationInterval != nil { + s.reconfigurationInterval = *s.config.Service.ReconfigurationInterval + } + if s.config.Service.BundleRebuildInterval != nil { + s.buildInterval = *s.config.Service.BundleRebuildInterval + } + } return s } @@ -184,11 +198,12 @@ shutdown: break shutdown } + // NB(sr): if a worker fails to shut down, we'll be stuck here forever time.Sleep(100 * time.Millisecond) } select { - case <-time.After(reconfigurationInterval): + case <-time.After(s.reconfigurationInterval): case <-ctx.Done(): break shutdown } @@ -215,6 +230,23 @@ func (s *Service) Report() *Report { return s.report } +func (s *Service) TriggerAll(ctx context.Context) error { + for name := range s.workers { + return s.Trigger(ctx, name) + } + return nil +} + +func (s *Service) Trigger(_ context.Context, name string) error { + err := s.pool.Trigger(name) + if err != nil { + s.log.Errorf("trigger bundle build for %s: %v", name, err) + } else { + s.log.Debugf("triggered bundle build for %s", name) + } + return err +} + func (s *Service) Ready(context.Context) error { s.readyMutex.Lock() defer s.readyMutex.Unlock() @@ -302,7 +334,7 @@ func (s *Service) launchWorkers(ctx context.Context) { // Start any new workers for bundles that are in the current configuration but not yet running. Inform any existing // workers of the current configuration, which will cause them to shutdown if configuration has changed. // - // For each bundle, create the following directory structure under persistencyDir for the builder to use + // For each bundle, create the following directory structure under persistenceDir for the builder to use // when constructing bundles: // // persistenceDir/ @@ -380,13 +412,13 @@ func (s *Service) launchWorkers(ctx context.Context) { continue } - w := NewBundleWorker(bundleDir, b, sourceDefs, stacks, s.log, bar). + w := NewBundleWorker(bundleDir, b, sourceDefs, stacks, s.log, bar, s.buildInterval). WithSources(sources). WithSynchronizers(syncs). WithStorage(storage). WithInterval(b.Interval). WithSingleShot(s.singleShot) - s.pool.Add(w.Execute) + s.pool.Add(b.Name, w.Execute) // TODO(sr): name not unique s.workers[bName] = w } @@ -465,6 +497,7 @@ func (src *source) addFS(fsys fs.FS) { } func (src *source) SyncGit(syncs *[]Synchronizer, sourceName string, git config.Git, repoDir string, reqCommit string) *source { + // Q(sr): Why errorDelay if we use this function to report BuildStateSuccess, too? if git.Repo != "" { srcDir := repoDir if git.Path != nil { diff --git a/internal/service/worker.go b/internal/service/worker.go index faf05e7f..6c9271f3 100644 --- a/internal/service/worker.go +++ b/internal/service/worker.go @@ -46,7 +46,7 @@ type Synchronizer interface { Close(ctx context.Context) } -func NewBundleWorker(bundleDir string, b *config.Bundle, sources []*config.Source, stacks []*config.Stack, logger *logging.Logger, bar *progress.Bar) *BundleWorker { +func NewBundleWorker(bundleDir string, b *config.Bundle, sources []*config.Source, stacks []*config.Stack, logger *logging.Logger, bar *progress.Bar, buildInterval time.Duration) *BundleWorker { return &BundleWorker{ bundleDir: bundleDir, bundleConfig: b, @@ -54,8 +54,9 @@ func NewBundleWorker(bundleDir string, b *config.Bundle, sources []*config.Sourc stackConfigs: stacks, log: logger, bar: bar, - changed: make(chan struct{}), done: make(chan struct{}), - interval: defaultInterval, + interval: buildInterval, + changed: make(chan struct{}), + done: make(chan struct{}), } } @@ -107,7 +108,6 @@ func (w *BundleWorker) Execute(ctx context.Context) time.Time { defer w.bar.Add(1) // If a configuration change was requested, request the worker to be removed from the pool and signal this worker being done. - if w.configurationChanged() { return w.die(ctx) } From 3dd8f314942f22aefeaa87526a54581d676e6733 Mon Sep 17 00:00:00 2001 From: Stephan Renatus Date: Mon, 6 Oct 2025 10:23:55 +0200 Subject: [PATCH 2/3] server+service+database: add bundle trigger endpoint (with authz) Some arbitrary choices here that we'll need to discuss before merging! Signed-off-by: Stephan Renatus --- cmd/run/run.go | 26 ++++++--------------- e2e/cli/run_with_trigger.txtar | 42 ++++++++++++++++++++++++++++++++++ internal/authz/authz.rego | 6 +++++ internal/config/config.go | 2 +- internal/database/database.go | 11 ++++++++- internal/server/server.go | 26 +++++++++++++++++++++ internal/server/types/types.go | 2 ++ internal/service/service.go | 17 ++++++++------ 8 files changed, 104 insertions(+), 28 deletions(-) create mode 100644 e2e/cli/run_with_trigger.txtar diff --git a/cmd/run/run.go b/cmd/run/run.go index 60a0dce6..e3d5ef30 100644 --- a/cmd/run/run.go +++ b/cmd/run/run.go @@ -3,8 +3,6 @@ package cmd import ( "context" "os" - "os/signal" - "syscall" "github.com/open-policy-agent/opa-control-plane/cmd" "github.com/open-policy-agent/opa-control-plane/cmd/internal/flags" @@ -73,12 +71,14 @@ func init() { log.Fatalf("initialize service: %v", err) } - // NB(sr): Preliminary, not necessarily something we'll want to keep: - // Rebuild all bundles on SIGHUP. - signalTrigger(svc, log) - go func() { - if err := server.New().WithDatabase(svc.Database()).WithReadiness(svc.Ready).WithConfig(config.Service).Init().ListenAndServe(params.addr); err != nil { + if err := server.New(). + WithService(svc). + WithConfig(config.Service). + WithDatabase(svc.Database()). + WithReadiness(svc.Ready). + Init(). + ListenAndServe(params.addr); err != nil { log.Fatalf("failed to start server: %v", err) } }() @@ -101,15 +101,3 @@ func init() { run, ) } - -func signalTrigger(s *service.Service, l *logging.Logger) { - sigs := make(chan os.Signal, 1) - signal.Notify(sigs, syscall.SIGHUP) - go func() { - for range sigs { - if err := s.TriggerAll(context.Background()); err != nil { - l.Error(err.Error()) - } - } - }() -} diff --git a/e2e/cli/run_with_trigger.txtar b/e2e/cli/run_with_trigger.txtar new file mode 100644 index 00000000..867e249b --- /dev/null +++ b/e2e/cli/run_with_trigger.txtar @@ -0,0 +1,42 @@ +! exec $OPACTL run --config config.d/bundle.yml --data-dir tmp --addr 127.0.0.1:8285 --log-level debug &opactl& +! stderr . +! stdout . + +exec curl --retry 5 --retry-all-errors http://127.0.0.1:8285/health + +# automation token +exec curl -H 'Authorization: bearer sesame' http://127.0.0.1:8285/v1/bundles/hello-world -XPOST + +# admin token +exec curl -H 'Authorization: bearer admin' http://127.0.0.1:8285/v1/bundles/hello-world -XPOST + +kill opactl +wait opactl +stderr 'triggered bundle build for hello-world' + +-- data/common.json -- +{ "common": true } +-- config.d/bundle.yml -- +bundles: + hello-world: + object_storage: + filesystem: + path: bundles/hello-world/bundle.tar.gz + requirements: + - source: global-data +sources: + global-data: + paths: + - data/common.json +service: + bundle_rebuild_interval: 1h + reconfiguration_interval: 1h +tokens: + admin-user: + api_key: admin + scopes: + - role: administrator + trigger-token: + api_key: sesame + scopes: + - role: automation diff --git a/internal/authz/authz.rego b/internal/authz/authz.rego index 6bdd61a1..b949d098 100644 --- a/internal/authz/authz.rego +++ b/internal/authz/authz.rego @@ -60,3 +60,9 @@ in_tenant(tenant_id) if { data.tenants.name == input.tenant tenant_id == data.tenants.id } + +allow if { + data.principals.id == input.principal + data.principals.role == "automation" + input.permission == "bundles.trigger" +} diff --git a/internal/config/config.go b/internal/config/config.go index 31383afd..4387630c 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -801,7 +801,7 @@ func (t *Token) Equal(other *Token) bool { } type Scope struct { - Role string `json:"role" enum:"administrator,viewer,owner,stack_owner"` + Role string `json:"role" enum:"administrator,viewer,owner,stack_owner,automation"` } func scopesEqual(a, b []Scope) bool { diff --git a/internal/database/database.go b/internal/database/database.go index e0b5bcae..5e5f8d9a 100644 --- a/internal/database/database.go +++ b/internal/database/database.go @@ -491,7 +491,7 @@ func (d *Database) sourcesDataPut(ctx context.Context, sourceName, path string, Name: sourceName, }) if !allowed { - return errors.New("unauthorized") + return ErrNotAuthorized } sourceID, err := d.lookupID(ctx, tx, tenant, "sources", sourceName) @@ -1914,6 +1914,15 @@ func (d *Database) deleteNotIn(ctx context.Context, tx *sql.Tx, table, keyColumn return err } +func (d *Database) Check(ctx context.Context, a authz.Access) error { + return tx1(ctx, d, func(tx *sql.Tx) error { + if !authz.Check(ctx, tx, d.arg, a) { + return ErrNotAuthorized + } + return nil + }) +} + func (d *Database) arg(i int) string { switch d.kind { case postgres, cockroach: diff --git a/internal/server/server.go b/internal/server/server.go index 6d66af2b..f5384884 100644 --- a/internal/server/server.go +++ b/internal/server/server.go @@ -19,6 +19,7 @@ import ( "github.com/open-policy-agent/opa-control-plane/internal/metrics" "github.com/open-policy-agent/opa-control-plane/internal/server/chain" "github.com/open-policy-agent/opa-control-plane/internal/server/types" + "github.com/open-policy-agent/opa-control-plane/internal/service" ) const defaultTenant = "default" @@ -26,6 +27,7 @@ const defaultTenant = "default" type Server struct { router *http.ServeMux db *database.Database + svc *service.Service readyFn func(context.Context) error apiPrefix string } @@ -64,6 +66,7 @@ func (s *Server) Init() *Server { setup("GET", "/v1/bundles/{bundle}", s.v1BundlesGet) setup("PUT", "/v1/bundles/{bundle}", s.v1BundlesPut) setup("DELETE", "/v1/bundles/{bundle}", s.v1BundlesDelete) + setup("POST", "/v1/bundles/{bundle}", s.v1BundlesPost) setup("GET", "/v1/stacks", s.v1StacksList) setup("GET", "/v1/stacks/{stack}", s.v1StacksGet) @@ -83,6 +86,11 @@ func (s *Server) WithRouter(router *http.ServeMux) *Server { return s } +func (s *Server) WithService(svc *service.Service) *Server { + s.svc = svc + return s +} + func (s *Server) WithDatabase(db *database.Database) *Server { s.db = db return s @@ -184,6 +192,24 @@ func (s *Server) v1BundlesGet(w http.ResponseWriter, r *http.Request) { JSONOK(w, resp, pretty(r)) } +func (s *Server) v1BundlesPost(w http.ResponseWriter, r *http.Request) { + ctx := r.Context() + + name, err := url.PathUnescape(r.PathValue("bundle")) + if err != nil { + ErrorString(w, http.StatusBadRequest, types.CodeInvalidParameter, err) + return + } + + if err := s.svc.Trigger(ctx, s.auth(r), name); err != nil { + errorAuto(w, err) + return + } + + resp := types.BundlesPostResponseV1{} + JSONOK(w, resp, pretty(r)) +} + func (s *Server) v1BundlesDelete(w http.ResponseWriter, r *http.Request) { ctx := r.Context() diff --git a/internal/server/types/types.go b/internal/server/types/types.go index 37d7c8c5..0b991cf0 100644 --- a/internal/server/types/types.go +++ b/internal/server/types/types.go @@ -45,6 +45,8 @@ type BundlesGetResponseV1 struct { type BundlesPutResponseV1 struct{} +type BundlesPostResponseV1 struct{} + type BundlesDeleteResponseV1 struct{} type SourcesGetResponseV1 struct { diff --git a/internal/service/service.go b/internal/service/service.go index d65040ab..9b49fb8e 100644 --- a/internal/service/service.go +++ b/internal/service/service.go @@ -20,6 +20,7 @@ import ( "github.com/open-policy-agent/opa/v1/ast" _ "modernc.org/sqlite" + "github.com/open-policy-agent/opa-control-plane/internal/authz" "github.com/open-policy-agent/opa-control-plane/internal/builder" "github.com/open-policy-agent/opa-control-plane/internal/config" "github.com/open-policy-agent/opa-control-plane/internal/database" @@ -37,7 +38,6 @@ import ( const ( internalPrincipal = "internal" defaultTenant = "default" - reconfigurationInterval = 15 * time.Second defaultReconfigurationInterval = 15 * time.Second defaultBuildInterval = 30 * time.Second ) @@ -230,14 +230,17 @@ func (s *Service) Report() *Report { return s.report } -func (s *Service) TriggerAll(ctx context.Context) error { - for name := range s.workers { - return s.Trigger(ctx, name) +func (s *Service) Trigger(ctx context.Context, principal, name string) error { + a := authz.Access{ + Principal: principal, + Resource: "bundles", + Permission: "bundles.trigger", + Name: name, + } + if err := s.database.Check(ctx, a); err != nil { + return err } - return nil -} -func (s *Service) Trigger(_ context.Context, name string) error { err := s.pool.Trigger(name) if err != nil { s.log.Errorf("trigger bundle build for %s: %v", name, err) From c99aa83e316eb2343356fcbed44d417d7a79866f Mon Sep 17 00:00:00 2001 From: Stephan Renatus Date: Tue, 9 Dec 2025 14:38:29 +0100 Subject: [PATCH 3/3] tenants: post-rebase adaptations Signed-off-by: Stephan Renatus --- cmd/run/run.go | 1 - config/schema.json | 13 ++++--------- internal/config/config.go | 4 ++-- internal/pool/pool.go | 28 ++++++++++++++++++---------- internal/pool/pool_test.go | 18 ++++++++++-------- internal/server/server.go | 3 ++- internal/service/service.go | 37 ++++++++++++++++++++++++------------- 7 files changed, 60 insertions(+), 44 deletions(-) diff --git a/cmd/run/run.go b/cmd/run/run.go index e3d5ef30..452cae5e 100644 --- a/cmd/run/run.go +++ b/cmd/run/run.go @@ -1,7 +1,6 @@ package cmd import ( - "context" "os" "github.com/open-policy-agent/opa-control-plane/cmd" diff --git a/config/schema.json b/config/schema.json index e0c5988b..0ad0230b 100644 --- a/config/schema.json +++ b/config/schema.json @@ -310,7 +310,8 @@ "administrator", "viewer", "owner", - "stack_owner" + "stack_owner", + "automation" ], "type": "string" } @@ -337,16 +338,10 @@ "type": "string" }, "bundle_rebuild_interval": { - "type": [ - "null", - "integer" - ] + "type": "string" }, "reconfiguration_interval": { - "type": [ - "null", - "integer" - ] + "type": "string" } }, "type": "object" diff --git a/internal/config/config.go b/internal/config/config.go index 4387630c..572c53c1 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -1075,12 +1075,12 @@ type Service struct { // ReconfigurationInterval is the duration between configuration checks, i.e. when a change // to a bundle/stack/source will have an effect on the internal bundle workers. // String of a duration, e.g. "1m". Defaults to "15s". - ReconfigurationInterval *time.Duration `json:"reconfiguration_interval,omitempty" yaml:"reconfiguration_interval,omitempty"` + ReconfigurationInterval Duration `json:"reconfiguration_interval,omitempty"` // BundleRebuildInterval is the time between bundle builds: After a bundle build as finished, // OCP will wait _this long_ until it's build again (unless the bundle build is triggered by // other means). String duration, e.g. "90s". Defaults to "30s". - BundleRebuildInterval *time.Duration `json:"bundle_rebuild_interval,omitempty" yaml:"bundle_rebuild_interval,omitempty"` + BundleRebuildInterval Duration `json:"bundle_rebuild_interval,omitempty"` _ struct{} `additionalProperties:"false"` } diff --git a/internal/pool/pool.go b/internal/pool/pool.go index 512a6c99..1233e57b 100644 --- a/internal/pool/pool.go +++ b/internal/pool/pool.go @@ -17,19 +17,24 @@ import ( type Pool struct { mu sync.Mutex queue []*task - reg map[string]*task + reg map[tenantName]*task wait chan struct{} } +type tenantName struct { + tenant string + name string +} + type task struct { - name string + name tenantName fn func(context.Context) time.Time deadline time.Time rerun bool } func New(workers int) *Pool { - pool := Pool{reg: make(map[string]*task)} + pool := Pool{reg: make(map[tenantName]*task)} for range workers { go pool.work() @@ -38,8 +43,8 @@ func New(workers int) *Pool { return &pool } -func (p *Pool) Add(name string, fn func(context.Context) time.Time) { - p.enqueue(&task{name: name, fn: fn, deadline: time.Now()}) +func (p *Pool) Add(tenant, name string, fn func(context.Context) time.Time) { + p.enqueue(&task{name: tenantName{name: name, tenant: tenant}, fn: fn, deadline: time.Now()}) } // work is the main loop for each worker goroutine. @@ -55,22 +60,22 @@ func (p *Pool) work() { // task is not queued, it's running. In that case, we'll have it override its // next deadline to NOW, causing an immediate re-run after the current run. // Subsequent runs will use the deadline returned by the task's `fn`. -func (p *Pool) Trigger(n string) error { +func (p *Pool) Trigger(tenant, name string) error { p.mu.Lock() defer p.mu.Unlock() - if i := slices.IndexFunc(p.queue, func(t *task) bool { return t.name == n }); i != -1 { + if i := slices.IndexFunc(p.queue, func(t *task) bool { return t.name.name == name && t.name.tenant == tenant }); i != -1 { p.queue[i].deadline = time.Now() p.sortAndWake() return nil } // if it's not in p.queue, it must be running at the moment - if t, ok := p.reg[n]; ok { + if t, ok := p.reg[tenantName{tenant: tenant, name: name}]; ok { t.rerun = true return nil } - return fmt.Errorf("no task with name %s", n) + return fmt.Errorf("no task with name %s (tenant %s)", name, tenant) } // sortAndWake is used in multiple places, but always needs to be run @@ -110,7 +115,10 @@ func (p *Pool) dequeue() *task { var t *task if len(p.queue) == 0 { - t = &task{name: "dummy", deadline: time.Now().Add(time.Hour * 24 * 365)} // Default to a far future deadline + t = &task{ + name: tenantName{tenant: "dummy", name: "dummy"}, + deadline: time.Now().Add(time.Hour * 24 * 365), // Default to a far future deadline + } } else { t = p.queue[0] } diff --git a/internal/pool/pool_test.go b/internal/pool/pool_test.go index e5a999d8..19d508a5 100644 --- a/internal/pool/pool_test.go +++ b/internal/pool/pool_test.go @@ -6,21 +6,23 @@ import ( "time" ) +const tenant = "default" + func TestPool(t *testing.T) { p := New(2) // Add a task that returns a deadline in the future. - p.Add("a", func(context.Context) time.Time { + p.Add(tenant, "a", func(context.Context) time.Time { return time.Now().Add(100 * time.Millisecond) }) // Add a task that returns a deadline in the past. - p.Add("b", func(context.Context) time.Time { + p.Add(tenant, "b", func(context.Context) time.Time { return time.Now().Add(-100 * time.Millisecond) }) // Add a task that returns a deadline in the future. - p.Add("c", func(context.Context) time.Time { + p.Add(tenant, "c", func(context.Context) time.Time { return time.Now().Add(200 * time.Millisecond) }) @@ -57,11 +59,11 @@ func TestTrigger(t *testing.T) { rx := &run{left: 3, deadline: 200 * time.Millisecond} - p.Add("t", rx.Execute) // will run once (run #1), and be queued for 200 ms + p.Add(tenant, "t", rx.Execute) // will run once (run #1), and be queued for 200 ms - _ = p.Trigger("t") // pulled in front, run #2 + _ = p.Trigger(tenant, "t") // pulled in front, run #2 time.Sleep(50 * time.Millisecond) - _ = p.Trigger("t") // pulled in front, run #3 + _ = p.Trigger(tenant, "t") // pulled in front, run #3 time.Sleep(300 * time.Millisecond) // no other runs, third run dequeued if exp, act := 3, rx.ran; exp != act { @@ -75,9 +77,9 @@ func TestTrigger(t *testing.T) { // if it wasn't triggered, we'd not see a second run: the next deadline is 1s rx := &run{left: 3, sleep: 100 * time.Millisecond, deadline: time.Second} - p.Add("t", rx.Execute) // will run once (run #1), and be queued for 200 ms + p.Add(tenant, "t", rx.Execute) // will run once (run #1), and be queued for 200 ms time.Sleep(50 * time.Millisecond) - _ = p.Trigger("t") // re-run after it's done, run #2 + _ = p.Trigger(tenant, "t") // re-run after it's done, run #2 time.Sleep(300 * time.Millisecond) diff --git a/internal/server/server.go b/internal/server/server.go index f5384884..444e7c6d 100644 --- a/internal/server/server.go +++ b/internal/server/server.go @@ -201,7 +201,8 @@ func (s *Server) v1BundlesPost(w http.ResponseWriter, r *http.Request) { return } - if err := s.svc.Trigger(ctx, s.auth(r), name); err != nil { + principal, tenant := s.auth(r) + if err := s.svc.Trigger(ctx, principal, tenant, name); err != nil { errorAuto(w, err) return } diff --git a/internal/service/service.go b/internal/service/service.go index 9b49fb8e..abca9a89 100644 --- a/internal/service/service.go +++ b/internal/service/service.go @@ -46,11 +46,19 @@ var ( defaultStackMountPrefix = ast.DefaultRootRef.Append(ast.StringTerm("stacks")) ) +type TenantName struct { + Tenant, Name string +} + +func (tn *TenantName) String() string { + return tn.Tenant + ":" + tn.Name +} + type Service struct { config *config.Root persistenceDir string pool *pool.Pool - workers map[string]*BundleWorker + workers map[TenantName]*BundleWorker readyMutex sync.Mutex ready bool failures map[string]Status @@ -111,7 +119,7 @@ type Status struct { func New() *Service { return &Service{ pool: pool.New(10), - workers: make(map[string]*BundleWorker), + workers: make(map[TenantName]*BundleWorker), failures: make(map[string]Status), noninteractive: true, migrateDB: false, @@ -129,11 +137,11 @@ func (s *Service) WithConfig(config *config.Root) *Service { s.config = config s.database = *s.database.WithConfig(config.Database) if s.config.Service != nil { - if s.config.Service.ReconfigurationInterval != nil { - s.reconfigurationInterval = *s.config.Service.ReconfigurationInterval + if s.config.Service.ReconfigurationInterval != 0 { + s.reconfigurationInterval = time.Duration(s.config.Service.ReconfigurationInterval) } - if s.config.Service.BundleRebuildInterval != nil { - s.buildInterval = *s.config.Service.BundleRebuildInterval + if s.config.Service.BundleRebuildInterval != 0 { + s.buildInterval = time.Duration(s.config.Service.BundleRebuildInterval) } } return s @@ -220,6 +228,8 @@ shutdown: for _, w := range s.workers { s.report.Bundles[w.bundleConfig.Name] = w.status } + + // For singleshot, we don't need to worry about tenants: maps.Copy(s.report.Bundles, s.failures) } @@ -230,9 +240,10 @@ func (s *Service) Report() *Report { return s.report } -func (s *Service) Trigger(ctx context.Context, principal, name string) error { +func (s *Service) Trigger(ctx context.Context, principal, tenant, name string) error { a := authz.Access{ Principal: principal, + Tenant: tenant, Resource: "bundles", Permission: "bundles.trigger", Name: name, @@ -241,7 +252,7 @@ func (s *Service) Trigger(ctx context.Context, principal, name string) error { return err } - err := s.pool.Trigger(name) + err := s.pool.Trigger(tenant, name) if err != nil { s.log.Errorf("trigger bundle build for %s: %v", name, err) } else { @@ -316,9 +327,9 @@ func (s *Service) launchWorkers(ctx context.Context) { return } - activeBundles := make(map[string]struct{}) + activeBundles := make(map[TenantName]struct{}) for _, b := range bundles { - bName := tenant + "_" + b.Name + bName := TenantName{Tenant: tenant, Name: b.Name} activeBundles[bName] = struct{}{} } @@ -353,7 +364,7 @@ func (s *Service) launchWorkers(ctx context.Context) { failures := make(map[string]Status) for _, b := range bundles { - bName := tenant + "_" + b.Name + bName := TenantName{Tenant: tenant, Name: b.Name} if w, ok := s.workers[bName]; ok { w.UpdateConfig(b, sourceDefs, stacks) continue @@ -392,7 +403,7 @@ func (s *Service) launchWorkers(ctx context.Context) { syncs := []Synchronizer{} sources := []*builder.Source{&root.Source} - bundleDir := join(s.persistenceDir, md5sum(bName)) + bundleDir := join(s.persistenceDir, md5sum(bName.String())) for _, dep := range deps { // NB(sr): dep.Name could contain a `:` which cause build errors in OPA's bundle build machinery @@ -421,7 +432,7 @@ func (s *Service) launchWorkers(ctx context.Context) { WithStorage(storage). WithInterval(b.Interval). WithSingleShot(s.singleShot) - s.pool.Add(b.Name, w.Execute) // TODO(sr): name not unique + s.pool.Add(tenant, b.Name, w.Execute) s.workers[bName] = w }