feat: support background runner prunning (#50)

Enables automatic pruning of Docker images.

Author: w-h-a

cmd/engine.go

@@ -155,6 +155,7 @@ func initRunner() runner.Runner {
 		runner.WithHost(config.RunnerHost()),
 		runner.WithRegistryUser(config.RunnerRegistryUser()),
 		runner.WithRegistryPass(config.RunnerRegistryPass()),
+		runner.WithPruneInterval(config.RunnerPruneInterval()),
 	)
 }
 

internal/engine/clients/runner/docker/runner.go

@@ -10,6 +10,7 @@ import (
 	"os"
 	"strings"
 	"sync"
+	"time"
 
 	"github.com/docker/docker/api/types/container"
 	"github.com/docker/docker/api/types/filters"
@@ -310,6 +311,38 @@ func (r *dockerRunner) remove(ctx context.Context, id string) error {
 	return nil
 }
 
+func (r *dockerRunner) pruneImages() {
+	r.wg.Add(1)
+	defer r.wg.Done()
+
+	ticker := time.NewTicker(r.options.PruneInterval)
+	defer ticker.Stop()
+
+	for {
+		select {
+		case <-ticker.C:
+			ctx := context.Background()
+			r.prune(ctx)
+		case <-r.exit:
+			return
+		}
+	}
+}
+
+func (r *dockerRunner) prune(ctx context.Context) {
+	filters := filters.NewArgs()
+
+	filters.Add("until", "24h")
+
+	pruneReport, err := r.client.ImagesPrune(ctx, filters)
+	if err != nil {
+		slog.ErrorContext(ctx, "failed to prune images", "error", err)
+		return
+	}
+
+	slog.InfoContext(ctx, "pruning complete", "images-deleted", len(pruneReport.ImagesDeleted), "space-reclaimed", pruneReport.SpaceReclaimed)
+}
+
 func NewRunner(opts ...runner.Option) runner.Runner {
 	options := runner.NewOptions(opts...)
 
@@ -333,5 +366,7 @@ func NewRunner(opts ...runner.Option) runner.Runner {
 		wg:      sync.WaitGroup{},
 	}
 
+	go dr.pruneImages()
+
 	return dr
 }

internal/engine/clients/runner/options.go

@@ -1,14 +1,18 @@
 package runner
 
-import "context"
+import (
+	"context"
+	"time"
+)
 
 type Option func(o *Options)
 
 type Options struct {
-	Host         string
-	RegistryUser string
-	RegistryPass string
-	Context      context.Context
+	Host          string
+	RegistryUser  string
+	RegistryPass  string
+	PruneInterval time.Duration
+	Context       context.Context
 }
 
 func WithHost(host string) Option {
@@ -29,9 +33,16 @@ func WithRegistryPass(pass string) Option {
 	}
 }
 
+func WithPruneInterval(interval time.Duration) Option {
+	return func(o *Options) {
+		o.PruneInterval = interval
+	}
+}
+
 func NewOptions(opts ...Option) Options {
 	options := Options{
-		Context: context.Background(),
+		Context:       context.Background(),
+		PruneInterval: 24 * time.Hour,
 	}
 
 	for _, fn := range opts {

internal/engine/config/config.go

@@ -6,6 +6,7 @@ import (
 	"strconv"
 	"strings"
 	"sync"
+	"time"
 
 	"github.com/w-h-a/workflow/internal/engine/clients/broker"
 	"github.com/w-h-a/workflow/internal/engine/clients/readwriter"
@@ -19,43 +20,45 @@ var (
 )
 
 type config struct {
-	env                string
-	name               string
-	version            string
-	mode               string
-	coordinatorHttp    string
-	workerHttp         string
-	workerQueues       map[string]int
-	broker             string
-	brokerLocation     string
-	brokerDurable      bool
-	runner             string
-	runnerHost         string
-	runnerRegistryUser string
-	runnerRegistryPass string
-	readwriter         string
-	readwriterLocation string
+	env                 string
+	name                string
+	version             string
+	mode                string
+	coordinatorHttp     string
+	workerHttp          string
+	workerQueues        map[string]int
+	broker              string
+	brokerLocation      string
+	brokerDurable       bool
+	runner              string
+	runnerHost          string
+	runnerRegistryUser  string
+	runnerRegistryPass  string
+	runnerPruneInterval time.Duration
+	readwriter          string
+	readwriterLocation  string
 }
 
 func New() {
 	once.Do(func() {
 		instance = &config{
-			env:                "dev",
-			name:               "workflow",
-			version:            "0.1.0-alpha.0",
-			mode:               "standalone",
-			coordinatorHttp:    ":4000",
-			workerHttp:         ":4001",
-			workerQueues:       map[string]int{string(task.Scheduled): 1, string(task.Cancelled): 1},
-			broker:             "memory",
-			brokerLocation:     "",
-			brokerDurable:      false,
-			runner:             "docker",
-			runnerHost:         "unix:///var/run/docker.sock",
-			runnerRegistryUser: "",
-			runnerRegistryPass: "",
-			readwriter:         "memory",
-			readwriterLocation: "",
+			env:                 "dev",
+			name:                "workflow",
+			version:             "0.1.0-alpha.0",
+			mode:                "standalone",
+			coordinatorHttp:     ":4000",
+			workerHttp:          ":4001",
+			workerQueues:        map[string]int{string(task.Scheduled): 1, string(task.Cancelled): 1},
+			broker:              "memory",
+			brokerLocation:      "",
+			brokerDurable:       false,
+			runner:              "docker",
+			runnerHost:          "unix:///var/run/docker.sock",
+			runnerRegistryUser:  "",
+			runnerRegistryPass:  "",
+			runnerPruneInterval: 24 * time.Hour,
+			readwriter:          "memory",
+			readwriterLocation:  "",
 		}
 
 		env := os.Getenv("ENV")
@@ -154,6 +157,18 @@ func New() {
 			instance.runnerRegistryPass = runnerRegistryPass
 		}
 
+		runnerPruneInterval := os.Getenv("RUNNER_PRUNE_INTERVAL")
+		if len(runnerPruneInterval) > 0 {
+			dur, err := time.ParseDuration(runnerPruneInterval)
+			if err != nil {
+				panic("invalid runner prune interval")
+			}
+			if dur <= 0 {
+				panic("runner prune interval must be a positive duration")
+			}
+			instance.runnerPruneInterval = dur
+		}
+
 		rw := os.Getenv("READ_WRITER")
 		if len(rw) > 0 {
 			if _, ok := readwriter.ReadWriterTypes[rw]; ok {
@@ -286,6 +301,14 @@ func RunnerRegistryPass() string {
 	return instance.runnerRegistryPass
 }
 
+func RunnerPruneInterval() time.Duration {
+	if instance == nil {
+		panic("cfg is nil")
+	}
+
+	return instance.runnerPruneInterval
+}
+
 func ReadWriter() string {
 	if instance == nil {
 		panic("cfg is nil")