adalal/gotenberg
1
0
Fork 0
mirror of https://github.com/gotenberg/gotenberg.git synced 2026-08-17 20:52:14 +01:00
Code Issues Packages Projects Releases Wiki Activity

refactor(gotenberg): thread engine label into NewProcessSupervisor

Browse Source
This commit is contained in:
Julien Neuhart
2026-06-02 19:25:47 +02:00
parent 2b20399a80
commit de572fe6b5
4 changed files with 31 additions and 23 deletions
Download Patch File Download Diff File Expand all files Collapse all files

12
pkg/gotenberg/supervisor.go
View File

@@ -101,6 +101,7 @@ const healthFailureThreshold = 2
type processSupervisor struct { type processSupervisor struct {
logger *slog.Logger logger *slog.Logger
engine string
process Process process Process
maxReqLimit int64 maxReqLimit int64
maxQueueSize int64 maxQueueSize int64
@@ -132,14 +133,21 @@ type processSupervisor struct {
idleStopChan chan struct{} // signal to stop the idle ticker goroutine idleStopChan chan struct{} // signal to stop the idle ticker goroutine
} }
// NewProcessSupervisor initializes a new [ProcessSupervisor]. // NewProcessSupervisor initializes a new [ProcessSupervisor]. engine names the
func NewProcessSupervisor(logger *slog.Logger, process Process, maxReqLimit, maxQueueSize, maxConcurrency int64, idleShutdownTimeout time.Duration) ProcessSupervisor { // managed process (for example "chromium" or "libreoffice") and prefixes the
// telemetry sub-spans; an empty engine falls back to "process".
func NewProcessSupervisor(logger *slog.Logger, engine string, process Process, maxReqLimit, maxQueueSize, maxConcurrency int64, idleShutdownTimeout time.Duration) ProcessSupervisor {
if maxConcurrency < 1 { if maxConcurrency < 1 {
maxConcurrency = 1 maxConcurrency = 1
} }
if engine == "" {
engine = "process"
}
b := &processSupervisor{ b := &processSupervisor{
logger: logger, logger: logger,
engine: engine,
process: process, process: process,
semaphore: make(chan struct{}, maxConcurrency), semaphore: make(chan struct{}, maxConcurrency),
maxReqLimit: maxReqLimit, maxReqLimit: maxReqLimit,

38
pkg/gotenberg/supervisor_test.go
View File

@@ -45,7 +45,7 @@ func TestProcessSupervisor_Launch(t *testing.T) {
}, },
} }
ps := NewProcessSupervisor(logger, process, 5, 0, 1, 0).(*processSupervisor) ps := NewProcessSupervisor(logger, "test", process, 5, 0, 1, 0).(*processSupervisor)
if tc.firstStartSet { if tc.firstStartSet {
ps.firstStart.Store(true) ps.firstStart.Store(true)
} }
@@ -93,7 +93,7 @@ func TestProcessSupervisor_Shutdown(t *testing.T) {
}, },
} }
ps := NewProcessSupervisor(logger, process, 5, 0, 1, 0) ps := NewProcessSupervisor(logger, "test", process, 5, 0, 1, 0)
err := ps.Shutdown() err := ps.Shutdown()
if !tc.expectError && err != nil { if !tc.expectError && err != nil {
@@ -145,7 +145,7 @@ func TestProcessSupervisor_restart(t *testing.T) {
}, },
} }
ps := NewProcessSupervisor(logger, process, 5, 0, 1, 0).(*processSupervisor) ps := NewProcessSupervisor(logger, "test", process, 5, 0, 1, 0).(*processSupervisor)
err := ps.restart() err := ps.restart()
@@ -201,7 +201,7 @@ func TestProcessSupervisor_Healthy(t *testing.T) {
}, },
} }
ps := NewProcessSupervisor(logger, process, 5, 0, 1, 0).(*processSupervisor) ps := NewProcessSupervisor(logger, "test", process, 5, 0, 1, 0).(*processSupervisor)
if tc.initiallyStarted { if tc.initiallyStarted {
ps.firstStart.Store(true) ps.firstStart.Store(true)
} }
@@ -229,7 +229,7 @@ func TestProcessSupervisor_Healthy_ConsecutiveFailures(t *testing.T) {
HealthyMock: func(_ *slog.Logger) bool { return processHealthy.Load() }, HealthyMock: func(_ *slog.Logger) bool { return processHealthy.Load() },
} }
ps := NewProcessSupervisor(logger, process, 5, 0, 1, 0).(*processSupervisor) ps := NewProcessSupervisor(logger, "test", process, 5, 0, 1, 0).(*processSupervisor)
ps.firstStart.Store(true) ps.firstStart.Store(true)
processHealthy.Store(false) processHealthy.Store(false)
@@ -272,7 +272,7 @@ func TestProcessSupervisor_Healthy_CachesPositiveResult(t *testing.T) {
}, },
} }
ps := NewProcessSupervisor(logger, process, 5, 0, 1, 0).(*processSupervisor) ps := NewProcessSupervisor(logger, "test", process, 5, 0, 1, 0).(*processSupervisor)
ps.firstStart.Store(true) ps.firstStart.Store(true)
for range 5 { for range 5 {
@@ -302,7 +302,7 @@ func TestProcessSupervisor_Healthy_DoesNotCacheNegativeResult(t *testing.T) {
}, },
} }
ps := NewProcessSupervisor(logger, process, 5, 0, 1, 0).(*processSupervisor) ps := NewProcessSupervisor(logger, "test", process, 5, 0, 1, 0).(*processSupervisor)
ps.firstStart.Store(true) ps.firstStart.Store(true)
processHealthy.Store(false) processHealthy.Store(false)
@@ -489,7 +489,7 @@ func TestProcessSupervisor_Run(t *testing.T) {
}, },
} }
ps := NewProcessSupervisor(logger, process, tc.maxReqLimit, tc.maxQueueSize, 1, 0).(*processSupervisor) ps := NewProcessSupervisor(logger, "test", process, tc.maxReqLimit, tc.maxQueueSize, 1, 0).(*processSupervisor)
if tc.initiallyStarted { if tc.initiallyStarted {
ps.firstStart.Store(true) ps.firstStart.Store(true)
} }
@@ -573,7 +573,7 @@ func TestProcessSupervisor_runWithDeadline(t *testing.T) {
}, },
} { } {
t.Run(tc.scenario, func(t *testing.T) { t.Run(tc.scenario, func(t *testing.T) {
ps := NewProcessSupervisor(slog.New(slog.DiscardHandler), new(ProcessMock), 0, 0, 1, 0).(*processSupervisor) ps := NewProcessSupervisor(slog.New(slog.DiscardHandler), "test", new(ProcessMock), 0, 0, 1, 0).(*processSupervisor)
ctx, cancel := context.WithTimeout(context.Background(), 1*time.Second) ctx, cancel := context.WithTimeout(context.Background(), 1*time.Second)
if tc.ctxDone { if tc.ctxDone {
@@ -607,7 +607,7 @@ func TestProcessSupervisor_ReqQueueSize(t *testing.T) {
return true return true
}, },
} }
ps := NewProcessSupervisor(logger, process, 0, 0, 1, 0).(*processSupervisor) ps := NewProcessSupervisor(logger, "test", process, 0, 0, 1, 0).(*processSupervisor)
// Simulating a lock. // Simulating a lock.
ps.semaphore <- struct{}{} ps.semaphore <- struct{}{}
@@ -668,7 +668,7 @@ func TestProcessSupervisor_QueueSizeCAS(t *testing.T) {
maxQueueSize := int64(50) maxQueueSize := int64(50)
// maxConcurrency=1 so all goroutines block on the semaphore, exercising queue logic. // maxConcurrency=1 so all goroutines block on the semaphore, exercising queue logic.
ps := NewProcessSupervisor(logger, process, 0, maxQueueSize, 1, 0).(*processSupervisor) ps := NewProcessSupervisor(logger, "test", process, 0, maxQueueSize, 1, 0).(*processSupervisor)
// Simulating a lock so that all goroutines queue up. // Simulating a lock so that all goroutines queue up.
ps.semaphore <- struct{}{} ps.semaphore <- struct{}{}
@@ -722,7 +722,7 @@ func TestProcessSupervisor_QueueSizeIncludesActiveTasks(t *testing.T) {
} }
// maxQueueSize=1, maxConcurrency=1: only one request at a time. // maxQueueSize=1, maxConcurrency=1: only one request at a time.
ps := NewProcessSupervisor(logger, process, 0, 1, 1, 0) ps := NewProcessSupervisor(logger, "test", process, 0, 1, 1, 0)
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel() defer cancel()
@@ -811,7 +811,7 @@ func TestProcessSupervisor_RestartsCount(t *testing.T) {
}, },
} }
ps := NewProcessSupervisor(logger, process, 0, 0, 1, 0).(*processSupervisor) ps := NewProcessSupervisor(logger, "test", process, 0, 0, 1, 0).(*processSupervisor)
ps.restartsCounter.Store(tc.initialRestartsCount) ps.restartsCounter.Store(tc.initialRestartsCount)
for i := 0; i < tc.restartAttempts; i++ { for i := 0; i < tc.restartAttempts; i++ {
@@ -844,7 +844,7 @@ func TestProcessSupervisor_ConcurrentRun(t *testing.T) {
} }
maxConcurrency := int64(3) maxConcurrency := int64(3)
ps := NewProcessSupervisor(logger, process, 0, 0, maxConcurrency, 0).(*processSupervisor) ps := NewProcessSupervisor(logger, "test", process, 0, 0, maxConcurrency, 0).(*processSupervisor)
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
defer cancel() defer cancel()
@@ -906,7 +906,7 @@ func TestProcessSupervisor_RestartDrainsAllSlots(t *testing.T) {
} }
maxConcurrency := int64(3) maxConcurrency := int64(3)
ps := NewProcessSupervisor(logger, process, 3, 0, maxConcurrency, 0).(*processSupervisor) ps := NewProcessSupervisor(logger, "test", process, 3, 0, maxConcurrency, 0).(*processSupervisor)
ps.firstStart.Store(true) ps.firstStart.Store(true)
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
@@ -963,7 +963,7 @@ func TestProcessSupervisor_IdleShutdown(t *testing.T) {
} }
idleTimeout := 50 * time.Millisecond idleTimeout := 50 * time.Millisecond
ps := NewProcessSupervisor(logger, process, 0, 0, 1, idleTimeout).(*processSupervisor) ps := NewProcessSupervisor(logger, "test", process, 0, 0, 1, idleTimeout).(*processSupervisor)
ctx := context.Background() ctx := context.Background()
err := ps.Run(ctx, logger, func() error { err := ps.Run(ctx, logger, func() error {
@@ -1023,7 +1023,7 @@ func TestProcessSupervisor_RetryAfterFailedFirstStart(t *testing.T) {
}, },
} }
ps := NewProcessSupervisor(logger, process, 0, 0, 1, 0).(*processSupervisor) ps := NewProcessSupervisor(logger, "test", process, 0, 0, 1, 0).(*processSupervisor)
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel() defer cancel()
@@ -1069,7 +1069,7 @@ func TestProcessSupervisor_IdleShutdownSkippedWhenActive(t *testing.T) {
} }
idleTimeout := 50 * time.Millisecond idleTimeout := 50 * time.Millisecond
ps := NewProcessSupervisor(logger, process, 0, 0, 1, idleTimeout) ps := NewProcessSupervisor(logger, "test", process, 0, 0, 1, idleTimeout)
ctx := context.Background() ctx := context.Background()
go func() { go func() {
@@ -1098,7 +1098,7 @@ func TestProcessSupervisor_ConversionsSinceRestart(t *testing.T) {
StopMock: func(*slog.Logger) error { return nil }, StopMock: func(*slog.Logger) error { return nil },
HealthyMock: func(*slog.Logger) bool { return true }, HealthyMock: func(*slog.Logger) bool { return true },
} }
s := NewProcessSupervisor(slog.New(slog.DiscardHandler), process, 0, 0, 1, 0).(*processSupervisor) s := NewProcessSupervisor(slog.New(slog.DiscardHandler), "test", process, 0, 0, 1, 0).(*processSupervisor)
if got := s.ConversionsSinceRestart(); got != 0 { if got := s.ConversionsSinceRestart(); got != 0 {
t.Errorf("expected 0 conversions initially, got %d", got) t.Errorf("expected 0 conversions initially, got %d", got)

2
pkg/modules/chromium/chromium.go
View File

@@ -517,7 +517,7 @@ func (mod *Chromium) Provision(ctx *gotenberg.Context) error {
// Process. // Process.
mod.browser = newChromiumBrowser(mod.args) mod.browser = newChromiumBrowser(mod.args)
mod.supervisor = gotenberg.NewProcessSupervisor(mod.logger, mod.browser, flags.MustInt64("chromium-restart-after"), flags.MustInt64("chromium-max-queue-size"), mod.maxConcurrency, flags.MustDuration("chromium-idle-shutdown-timeout")) mod.supervisor = gotenberg.NewProcessSupervisor(mod.logger, "chromium", mod.browser, flags.MustInt64("chromium-restart-after"), flags.MustInt64("chromium-max-queue-size"), mod.maxConcurrency, flags.MustDuration("chromium-idle-shutdown-timeout"))
// PDF Engine. // PDF Engine.
provider, err := ctx.Module(new(gotenberg.PdfEngineProvider)) provider, err := ctx.Module(new(gotenberg.PdfEngineProvider))

2
pkg/modules/libreoffice/api/api.go
View File

@@ -370,7 +370,7 @@ func (a *Api) Provision(ctx *gotenberg.Context) error {
// Process. // Process.
a.libreOffice = newLibreOfficeProcess(a.args) a.libreOffice = newLibreOfficeProcess(a.args)
a.supervisor = gotenberg.NewProcessSupervisor(a.logger, a.libreOffice, flags.MustInt64("libreoffice-restart-after"), flags.MustInt64("libreoffice-max-queue-size"), 1, flags.MustDuration("libreoffice-idle-shutdown-timeout")) a.supervisor = gotenberg.NewProcessSupervisor(a.logger, "libreoffice", a.libreOffice, flags.MustInt64("libreoffice-restart-after"), flags.MustInt64("libreoffice-max-queue-size"), 1, flags.MustDuration("libreoffice-idle-shutdown-timeout"))
// Metrics. // Metrics.
meter := gotenberg.Meter() meter := gotenberg.Meter()