diff --git a/cmd/run/run.go b/cmd/run/run.go index 9832fc37..452cae5e 100644 --- a/cmd/run/run.go +++ b/cmd/run/run.go @@ -71,7 +71,13 @@ func init() { } 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) } }() diff --git a/config/schema.json b/config/schema.json index 1321c0c5..0ad0230b 100644 --- a/config/schema.json +++ b/config/schema.json @@ -310,7 +310,8 @@ "administrator", "viewer", "owner", - "stack_owner" + "stack_owner", + "automation" ], "type": "string" } @@ -335,6 +336,12 @@ "api_prefix": { "pattern": "^/([^/].*[^/])?$", "type": "string" + }, + "bundle_rebuild_interval": { + "type": "string" + }, + "reconfiguration_interval": { + "type": "string" } }, "type": "object" 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 5106dd79..572c53c1 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 { @@ -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 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 Duration `json:"bundle_rebuild_interval,omitempty"` + _ struct{} `additionalProperties:"false"` } 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/pool/pool.go b/internal/pool/pool.go index 961eeb34..1233e57b 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,25 @@ import ( // the waiting goroutine to process the new task immediately. type Pool struct { mu sync.Mutex - tasks []*task + queue []*task + reg map[tenantName]*task wait chan struct{} } +type tenantName struct { + tenant string + name string +} + type task struct { + name tenantName fn func(context.Context) time.Time deadline time.Time + rerun bool } func New(workers int) *Pool { - var pool Pool + pool := Pool{reg: make(map[tenantName]*task)} for range workers { go pool.work() @@ -34,8 +43,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(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. @@ -46,18 +55,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(tenant, name string) error { p.mu.Lock() defer p.mu.Unlock() + 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[tenantName{tenant: tenant, name: name}]; ok { + t.rerun = true + return nil + } + + return fmt.Errorf("no task with name %s (tenant %s)", name, tenant) +} + +// 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 +93,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 +114,13 @@ 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: tenantName{tenant: "dummy", 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 +147,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..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(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(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(func(_ context.Context) time.Time { + p.Add(tenant, "c", func(context.Context) time.Time { return time.Now().Add(200 * time.Millisecond) }) @@ -28,5 +30,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(tenant, "t", rx.Execute) // will run once (run #1), and be queued for 200 ms + + _ = p.Trigger(tenant, "t") // pulled in front, run #2 + time.Sleep(50 * time.Millisecond) + _ = 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 { + 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(tenant, "t", rx.Execute) // will run once (run #1), and be queued for 200 ms + time.Sleep(50 * time.Millisecond) + _ = p.Trigger(tenant, "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/server/server.go b/internal/server/server.go index 6d66af2b..444e7c6d 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,25 @@ 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 + } + + principal, tenant := s.auth(r) + if err := s.svc.Trigger(ctx, principal, tenant, 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 3af61e80..abca9a89 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" @@ -35,31 +36,42 @@ import ( ) const ( - internalPrincipal = "internal" - defaultTenant = "default" - reconfigurationInterval = 15 * time.Second + internalPrincipal = "internal" + defaultTenant = "default" + defaultReconfigurationInterval = 15 * time.Second + defaultBuildInterval = 30 * time.Second ) 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 - 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[TenantName]*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 +118,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[TenantName]*BundleWorker), + failures: make(map[string]Status), + noninteractive: true, + migrateDB: false, + reconfigurationInterval: defaultReconfigurationInterval, + buildInterval: defaultBuildInterval, } } @@ -122,6 +136,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 != 0 { + s.reconfigurationInterval = time.Duration(s.config.Service.ReconfigurationInterval) + } + if s.config.Service.BundleRebuildInterval != 0 { + s.buildInterval = time.Duration(s.config.Service.BundleRebuildInterval) + } + } return s } @@ -184,11 +206,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 } @@ -205,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) } @@ -215,6 +240,27 @@ func (s *Service) Report() *Report { return s.report } +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, + } + if err := s.database.Check(ctx, a); err != nil { + return err + } + + err := s.pool.Trigger(tenant, 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() @@ -281,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{}{} } @@ -302,7 +348,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/ @@ -318,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 @@ -357,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 @@ -380,13 +426,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(tenant, b.Name, w.Execute) s.workers[bName] = w } @@ -465,6 +511,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) }