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 <stephan.renatus@gmail.com>

Author: srenatus

cmd/run/run.go

@@ -4,6 +4,8 @@ import (
 	"bytes"
 	"context"
 	"os"
+	"os/signal"
+	"syscall"
 
 	"github.com/spf13/cobra"
 	"github.com/styrainc/opa-control-plane/cmd"
@@ -73,6 +75,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).Init().ListenAndServe(params.addr); err != nil {
 					log.Fatalf("failed to start server: %v", err)
@@ -97,3 +103,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())
+			}
+		}
+	}()
+}

internal/config/config.go

@@ -14,6 +14,7 @@ import (
 	"reflect"
 	"slices"
 	"sort"
+	"time"
 
 	"github.com/gobwas/glob"
 	"github.com/pkg/errors"
@@ -40,6 +41,7 @@ type Root struct {
 	Secrets  map[string]*Secret `json:"secrets,omitempty" yaml:"secrets,omitempty"` // Schema validation overrides Secret to object type.
 	Tokens   map[string]*Token  `json:"tokens,omitempty" yaml:"tokens,omitempty"`
 	Database *Database          `json:"database,omitempty" yaml:"database,omitempty"`
+	Service  *Service           `json:"service,omitempty" yaml:"service,omitempty"`
 }
 
 // SetSQLitePersistentByDefault sets the database configuration to use a SQLite
@@ -228,7 +230,7 @@ func (r *Root) Validate() error {
 		return err
 	}
 
-	var config interface{}
+	var config any
 	if err := json.Unmarshal(data, &config); err != nil {
 		return err
 	}
@@ -966,6 +968,18 @@ func (a Datasources) Equal(b Datasources) bool {
 	return setEqual(a, b, func(ds Datasource) string { return ds.Name }, func(a, b Datasource) bool { return a.Equal(&b) })
 }
 
+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"`
+
+	// 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"`
+}
+
 type Database struct {
 	SQL    *SQLDatabase `json:"sql,omitempty" yaml:"sql,omitempty"`
 	AWSRDS *AmazonRDS   `json:"aws_rds,omitempty" yaml:"aws_rds,omitempty"`

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,17 +88,31 @@ 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: "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
 }

internal/pool/pool_test.go

@@ -10,23 +10,79 @@ 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)
 	})
 
 	// 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("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)
+		}
+	})
+}