service: add Trigger(name) to trigger bundle rebuild

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)
 				}
 			}()

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"

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

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

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

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:

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,17 +93,34 @@ 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()
 
 	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
 }

internal/pool/pool_test.go

@@ -6,27 +6,85 @@ 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)
 	})
 
 	// Wait for a short period to allow tasks to be processed.
 	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)
+		}
+	})
+}