diff --git a/registry/handlers/app.go b/registry/handlers/app.go index 301fc970680..fc3261271fd 100644 --- a/registry/handlers/app.go +++ b/registry/handlers/app.go @@ -5,6 +5,7 @@ import ( "crypto/rand" "crypto/tls" "crypto/x509" + "errors" "expvar" "fmt" "math" @@ -61,6 +62,9 @@ const defaultCheckInterval = 10 * time.Second type App struct { context.Context + // cancel stops background goroutines started by NewApp (e.g. the upload purger). + cancel context.CancelFunc + Config *configuration.Configuration router *mux.Router // main application router, configured with dispatchers @@ -92,9 +96,13 @@ type App struct { // requests. The app only implements ServeHTTP and can be wrapped in other // handlers accordingly. func NewApp(ctx context.Context, config *configuration.Configuration) *App { + // Make the app's context cancelable so Shutdown can stop background + // goroutines (e.g. the upload purger) that would otherwise leak the app. + ctx, cancel := context.WithCancel(ctx) app := &App{ Config: config, Context: ctx, + cancel: cancel, router: v2.RouterWithPrefix(config.HTTP.Prefix), isCache: config.Proxy.RemoteURL != "", } @@ -449,10 +457,21 @@ func (app *App) RegisterHealthChecks(healthRegistries ...*health.Registry) { // Shutdown close the underlying registry func (app *App) Shutdown() error { + if app.cancel != nil { + app.cancel() + } + var errs []error + if app.events.sink != nil { + if err := app.events.sink.Close(); err != nil { + errs = append(errs, err) + } + } if r, ok := app.registry.(proxy.Closer); ok { - return r.Close() + if err := r.Close(); err != nil { + errs = append(errs, err) + } } - return nil + return errors.Join(errs...) } // register a handler with the application, by route name. The handler will be @@ -1086,12 +1105,19 @@ func startUploadPurger(ctx context.Context, storageDriver storagedriver.StorageD } jitter := time.Duration(randInt.Int64()%60) * time.Minute log.Infof("Starting upload purge in %s", jitter) - time.Sleep(jitter) + timer := time.NewTimer(jitter) + defer timer.Stop() for { + select { + case <-ctx.Done(): + log.Info("Upload purger stopped") + return + case <-timer.C: + } storage.PurgeUploads(ctx, storageDriver, time.Now().Add(-purgeAgeDuration), !dryRunBool) log.Infof("Starting upload purge in %s", intervalDuration) - time.Sleep(intervalDuration) + timer.Reset(intervalDuration) } }() } diff --git a/registry/handlers/app_test.go b/registry/handlers/app_test.go index 7725df83a1e..df5edac5b4d 100644 --- a/registry/handlers/app_test.go +++ b/registry/handlers/app_test.go @@ -2,11 +2,13 @@ package handlers import ( "encoding/json" + "errors" "net/http" "net/http/httptest" "net/url" "reflect" "testing" + "time" "github.com/distribution/distribution/v3/configuration" "github.com/distribution/distribution/v3/internal/dcontext" @@ -17,6 +19,7 @@ import ( "github.com/distribution/distribution/v3/registry/storage" memorycache "github.com/distribution/distribution/v3/registry/storage/cache/memory" "github.com/distribution/distribution/v3/registry/storage/driver/inmemory" + events "github.com/docker/go-events" ) // TestAppDispatcher builds an application with a test dispatcher and ensures @@ -275,3 +278,28 @@ func TestAppendAccessRecords(t *testing.T) { t.Fatal("Actual access record differs from expected") } } + +// TestAppShutdownCancelsContext ensures Shutdown cancels the app's context so +// background goroutines started by NewApp (e.g. the upload purger) terminate +// instead of pinning the app in memory. +func TestAppShutdownCancelsContext(t *testing.T) { + app := NewApp(dcontext.Background(), &configuration.Configuration{ + Storage: configuration.Storage{ + "inmemory": nil, + }, + }) + + if err := app.Shutdown(); err != nil { + t.Fatalf("unexpected error shutting down app: %v", err) + } + + select { + case <-app.Done(): + case <-time.After(5 * time.Second): + t.Fatal("app context not canceled after Shutdown") + } + + if err := app.events.sink.Write(nil); !errors.Is(err, events.ErrSinkClosed) { + t.Fatalf("expected event sink to be closed after Shutdown, got %v", err) + } +} diff --git a/registry/leak_test.go b/registry/leak_test.go new file mode 100644 index 00000000000..e3953e94c75 --- /dev/null +++ b/registry/leak_test.go @@ -0,0 +1,58 @@ +package registry + +import ( + "context" + "runtime" + "testing" + "time" + + "github.com/distribution/distribution/v3/configuration" + "github.com/distribution/distribution/v3/internal/dcontext" + _ "github.com/distribution/distribution/v3/registry/storage/driver/inmemory" +) + +// TestRebuildDoesNotLeakGoroutines simulates a process that repeatedly tears +// down and recreates a registry (e.g. on credential rotation) and asserts the +// old registries' background goroutines (upload purger, event broadcaster) do +// not accumulate. +func TestRebuildDoesNotLeakGoroutines(t *testing.T) { + config := &configuration.Configuration{ + Storage: configuration.Storage{"inmemory": configuration.Parameters{}}, + } + config.HTTP.Addr = "127.0.0.1:15999" + config.Log.Level = "error" + + rebuild := func() { + reg, err := NewRegistry(dcontext.Background(), config) + if err != nil { + t.Fatalf("NewRegistry: %v", err) + } + done := make(chan struct{}) + go func() { _ = reg.ListenAndServe(); close(done) }() + time.Sleep(100 * time.Millisecond) // let it bind + if err := reg.Shutdown(context.Background()); err != nil { + t.Fatalf("Shutdown: %v", err) + } + <-done + } + + rebuild() // warm up process-global state (otel, health) + runtime.GC() + before := runtime.NumGoroutine() + + const iterations = 10 + for i := 0; i < iterations; i++ { + rebuild() + } + + var after int + for i := 0; i < 50; i++ { // allow stragglers to exit + runtime.GC() + after = runtime.NumGoroutine() + if after <= before+2 { + return + } + time.Sleep(100 * time.Millisecond) + } + t.Fatalf("goroutines grew from %d to %d over %d rebuilds", before, after, iterations) +} diff --git a/registry/registry.go b/registry/registry.go index 29fc1c401e6..e827d8583c2 100644 --- a/registry/registry.go +++ b/registry/registry.go @@ -165,6 +165,10 @@ func NewRegistry(ctx context.Context, config *configuration.Configuration) (*Reg err = tracing.InitOpenTelemetry(app.Context) if err != nil { + // The app is being discarded; stop its background goroutines. + if shutdownErr := app.Shutdown(); shutdownErr != nil { + err = errors.Join(err, shutdownErr) + } return nil, fmt.Errorf("error during open telemetry initialization: %v", err) } if config.HTTP.H2C.Enabled { diff --git a/tracing/tracing.go b/tracing/tracing.go index dca9d584fe1..907db5ab7ac 100644 --- a/tracing/tracing.go +++ b/tracing/tracing.go @@ -2,6 +2,7 @@ package tracing import ( "context" + "sync" "github.com/distribution/distribution/v3/internal/dcontext" "github.com/distribution/distribution/v3/version" @@ -26,9 +27,33 @@ const ( AttributePrefix = "io.cncf.distribution." ) +var ( + initMu sync.Mutex + initDone bool +) + // InitOpenTelemetry initializes OpenTelemetry for the application. This function sets up the // necessary components for collecting telemetry data, such as traces. +// +// It configures process-global state (the global TracerProvider, error handler, and +// propagator), so it initializes at most once per process; once it has succeeded, +// subsequent calls are no-ops (a failed attempt may be retried). This also keeps +// repeated registry construction (e.g. on config reload) from leaking a +// BatchSpanProcessor per call. func InitOpenTelemetry(ctx context.Context) error { + initMu.Lock() + defer initMu.Unlock() + if initDone { + return nil + } + if err := initOpenTelemetry(ctx); err != nil { + return err + } + initDone = true + return nil +} + +func initOpenTelemetry(ctx context.Context) error { res := resource.NewWithAttributes( semconv.SchemaURL, semconv.ServiceNameKey.String(serviceName),