tenants: post-rebase adaptations

Signed-off-by: Stephan Renatus <stephan.renatus@gmail.com>

Author: srenatus

cmd/run/run.go

@@ -1,7 +1,6 @@
 package cmd
 
 import (
-	"context"
 	"os"
 
 	"github.com/open-policy-agent/opa-control-plane/cmd"

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"

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"`
 }

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]
 		}

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
 	}

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
 		}