in registry/handlers/app.go [159:612]
func NewApp(ctx context.Context, config *configuration.Configuration) (*App, error) {
app := &App{
Config: config,
Context: ctx,
shutdownFuncs: make([]shutdownFunc, 0),
}
if err := app.initMetaRouter(); err != nil {
return nil, fmt.Errorf("initializing metaRouter: %w", err)
}
storageParams := config.Storage.Parameters()
if storageParams == nil {
storageParams = make(configuration.Parameters)
}
log := dcontext.GetLogger(app)
storageParams[storagedriver.ParamLogger] = log
var err error
app.driver, err = factory.Create(config.Storage.Type(), storageParams)
if err != nil {
return nil, err
}
purgeConfig := uploadPurgeDefaultConfig()
if mc, ok := config.Storage["maintenance"]; ok {
if v, ok := mc["uploadpurging"]; ok {
purgeConfig, ok = v.(map[any]any)
if !ok {
return nil, fmt.Errorf("uploadpurging config key must contain additional keys")
}
}
if v, ok := mc["readonly"]; ok {
readOnly, ok := v.(map[any]any)
if !ok {
return nil, fmt.Errorf("readonly config key must contain additional keys")
}
// nolint: revive // max-control-nesting
if readOnlyEnabled, ok := readOnly["enabled"]; ok {
app.readOnly, ok = readOnlyEnabled.(bool)
if !ok {
return nil, fmt.Errorf("readonly's enabled config key must have a boolean value")
}
if app.readOnly {
if enabled, ok := purgeConfig["enabled"].(bool); ok && enabled {
log.Info("disabled upload purging in readonly mode")
purgeConfig["enabled"] = false
}
}
}
}
}
if err := startUploadPurger(app, app.driver, log, purgeConfig); err != nil {
return nil, err
}
app.driver, err = applyStorageMiddleware(app.driver, config.Middleware["storage"])
if err != nil {
return nil, err
}
if err := app.configureSecret(config); err != nil {
return nil, err
}
app.configureEvents(config)
if err := app.configureRedisBlobDesc(ctx, config); err != nil {
return nil, err
}
if err := app.configureRedisCache(ctx, config); err != nil {
// Because the Redis cache is not a strictly required dependency (data will be served from the metadata DB if
// we're unable to serve or find it in cache) we simply log and report a failure here and proceed to not prevent
// the app from starting.
log.WithError(err).Error("failed configuring Redis cache")
errortracking.Capture(err, errortracking.WithContext(ctx), errortracking.WithStackTrace())
}
if err := app.configureRedisRateLimiter(ctx, config); err != nil {
// Redis rate-limiter is not a strictly required dependency, we simply log and report a failure here
// and proceed to not prevent the app from starting.
log.WithError(err).Error("failed configuring Redis rate-limiter")
errortracking.Capture(err, errortracking.WithContext(ctx), errortracking.WithStackTrace())
return nil, err
}
options := registrymiddleware.GetRegistryOptions()
// TODO: Once schema1 code is removed throughout the registry, we will not
// need to explicitly configure this.
options = append(options, storage.DisableSchema1Pulls)
if config.HTTP.Host != "" {
u, err := url.Parse(config.HTTP.Host)
if err != nil {
return nil, fmt.Errorf(`could not parse http "host" parameter: %w`, err)
}
app.httpHost = *u
}
// configure deletion
if d, ok := config.Storage["delete"]; ok {
e, ok := d["enabled"]
if ok {
if deleteEnabled, ok := e.(bool); ok && deleteEnabled {
options = append(options, storage.EnableDelete)
}
}
}
// configure redirects
var redirectDisabled bool
if redirectConfig, ok := config.Storage["redirect"]; ok {
v := redirectConfig["disable"]
switch v := v.(type) {
case bool:
redirectDisabled = v
case nil:
// disable is not mandatory as we default to false, so do nothing if it doesn't exist
default:
return nil, fmt.Errorf("invalid type %T for 'storage.redirect.disable' (boolean)", v)
}
}
if redirectDisabled {
log.Info("backend redirection disabled")
} else {
l := log
exceptions := config.Storage["redirect"]["exceptions"]
if exceptions, ok := exceptions.([]any); ok && len(exceptions) > 0 {
s := make([]string, len(exceptions))
for i, v := range exceptions {
s[i] = fmt.Sprint(v)
}
l.WithField("exceptions", s)
options = append(options, storage.EnableRedirectWithExceptions(s))
} else {
options = append(options, storage.EnableRedirect)
}
// expiry delay
delay := config.Storage["redirect"]["expirydelay"]
var d time.Duration
switch v := delay.(type) {
case time.Duration:
d = v
case string:
if d, err = time.ParseDuration(v); err != nil {
return nil, fmt.Errorf("%q value for 'storage.redirect.expirydelay' is not a valid duration", v)
}
case nil:
default:
return nil, fmt.Errorf("invalid type %[1]T for 'storage.redirect.expirydelay' (duration)", delay)
}
if d > 0 {
l = l.WithField("expiry_delay_s", d.Seconds())
options = append(options, storage.WithRedirectExpiryDelay(d))
}
l.Info("storage backend redirection enabled")
}
if !config.Validation.Enabled {
config.Validation.Enabled = !config.Validation.Disabled
}
// configure validation
if config.Validation.Enabled {
if len(config.Validation.Manifests.URLs.Allow) == 0 && len(config.Validation.Manifests.URLs.Deny) == 0 {
// If Allow and Deny are empty, allow nothing.
app.manifestURLs.Allow = regexp.MustCompile("^$")
options = append(options, storage.ManifestURLsAllowRegexp(app.manifestURLs.Allow))
} else {
// nolint: revive // max-control-nesting
if len(config.Validation.Manifests.URLs.Allow) > 0 {
for i, s := range config.Validation.Manifests.URLs.Allow {
// Validate via compilation.
if _, err := regexp.Compile(s); err != nil {
return nil, fmt.Errorf("validation.manifests.urls.allow: %w", err)
}
// Wrap with non-capturing group.
config.Validation.Manifests.URLs.Allow[i] = fmt.Sprintf("(?:%s)", s)
}
app.manifestURLs.Allow = regexp.MustCompile(strings.Join(config.Validation.Manifests.URLs.Allow, "|"))
options = append(options, storage.ManifestURLsAllowRegexp(app.manifestURLs.Allow))
}
// nolint: revive // max-control-nesting
if len(config.Validation.Manifests.URLs.Deny) > 0 {
for i, s := range config.Validation.Manifests.URLs.Deny {
// Validate via compilation.
if _, err := regexp.Compile(s); err != nil {
return nil, fmt.Errorf("validation.manifests.urls.deny: %w", err)
}
// Wrap with non-capturing group.
config.Validation.Manifests.URLs.Deny[i] = fmt.Sprintf("(?:%s)", s)
}
app.manifestURLs.Deny = regexp.MustCompile(strings.Join(config.Validation.Manifests.URLs.Deny, "|"))
options = append(options, storage.ManifestURLsDenyRegexp(app.manifestURLs.Deny))
}
}
app.manifestRefLimit = config.Validation.Manifests.ReferenceLimit
options = append(options, storage.ManifestReferenceLimit(app.manifestRefLimit))
app.manifestPayloadSizeLimit = config.Validation.Manifests.PayloadSizeLimit
options = append(options, storage.ManifestPayloadSizeLimit(app.manifestPayloadSizeLimit))
}
fsLocker := &storage.FilesystemInUseLocker{Driver: app.driver}
dbLocker := &storage.DatabaseInUseLocker{Driver: app.driver}
// Connect to the metadata database, if enabled.
if config.Database.Enabled {
// Temporary measure to enforce lock files while all the implementation is done
// Seehttps://gitlab.com/gitlab-org/container-registry/-/issues/1335
if feature.EnforceLockfiles.Enabled() {
fsLocked, err := fsLocker.IsLocked(ctx)
if err != nil {
log.WithError(err).Error("could not check if filesystem metadata is locked, see https://docs.gitlab.com/ee/administration/packages/container_registry_metadata_database.html")
return nil, err
}
if fsLocked {
return nil, ErrFilesystemInUse
}
}
log.Info("using the metadata database")
if config.GC.Disabled {
log.Warn("garbage collection is disabled")
}
// Do not write or check for repository layer link metadata on the filesystem when the database is enabled.
options = append(options, storage.UseDatabase)
dsn := &datastore.DSN{
Host: config.Database.Host,
Port: config.Database.Port,
User: config.Database.User,
Password: config.Database.Password,
DBName: config.Database.DBName,
SSLMode: config.Database.SSLMode,
SSLCert: config.Database.SSLCert,
SSLKey: config.Database.SSLKey,
SSLRootCert: config.Database.SSLRootCert,
ConnectTimeout: config.Database.ConnectTimeout,
}
dbOpts := []datastore.Option{
datastore.WithLogger(log.WithFields(logrus.Fields{"database": config.Database.DBName})),
datastore.WithLogLevel(config.Log.Level),
datastore.WithPreparedStatements(config.Database.PreparedStatements),
datastore.WithPoolConfig(&datastore.PoolConfig{
MaxIdle: config.Database.Pool.MaxIdle,
MaxOpen: config.Database.Pool.MaxOpen,
MaxLifetime: config.Database.Pool.MaxLifetime,
MaxIdleTime: config.Database.Pool.MaxIdleTime,
}),
}
// nolint: revive // max-control-nesting
if config.Database.LoadBalancing.Enabled {
if err := app.configureRedisLoadBalancingCache(ctx, config); err != nil {
return nil, err
}
dbOpts = append(dbOpts, datastore.WithLSNCache(datastore.NewCentralRepositoryCache(app.redisLBCache)))
// service discovery takes precedence over fixed hosts
if config.Database.LoadBalancing.Record != "" {
nameserver := config.Database.LoadBalancing.Nameserver
port := config.Database.LoadBalancing.Port
record := config.Database.LoadBalancing.Record
replicaCheckInterval := config.Database.LoadBalancing.ReplicaCheckInterval
log.WithFields(dlog.Fields{
"nameserver": nameserver,
"port": port,
"record": record,
"replica_check_interval_s": replicaCheckInterval.Seconds(),
}).Info("enabling database load balancing with service discovery")
resolver := datastore.NewDNSResolver(nameserver, port, record)
dbOpts = append(dbOpts,
datastore.WithServiceDiscovery(resolver),
datastore.WithReplicaCheckInterval(replicaCheckInterval),
)
} else if len(config.Database.LoadBalancing.Hosts) > 0 {
hosts := config.Database.LoadBalancing.Hosts
log.WithField("hosts", hosts).Info("enabling database load balancing with static hosts list")
dbOpts = append(dbOpts, datastore.WithFixedHosts(hosts))
}
}
if config.HTTP.Debug.Prometheus.Enabled {
dbOpts = append(dbOpts, datastore.WithMetricsCollection())
}
db, err := datastore.NewDBLoadBalancer(ctx, dsn, dbOpts...)
if err != nil {
return nil, fmt.Errorf("failed to initialize database connections: %w", err)
}
if config.Database.LoadBalancing.Enabled && config.Database.LoadBalancing.ReplicaCheckInterval != 0 {
startDBPoolRefresh(ctx, db)
startDBLagCheck(ctx, db)
}
inRecovery, err := datastore.IsInRecovery(ctx, db.Primary())
if err != nil {
log.WithError(err).Error("could not check database recovery status")
return nil, err
}
if inRecovery {
err = errors.New("the database is in read-only mode (in recovery)")
log.WithError(err).Error("could not connect to database")
log.Warn("if this is a Geo secondary instance, the registry must not point to the default replicated database. Please configure the registry to connect to a separate, writable database on the Geo secondary site.")
log.Warn("if this is not a Geo secondary instance, the database must not be in read-only mode. Please ensure the database is correctly configured and set to read-write mode.")
return nil, err
}
// Skip postdeployment migrations to prevent pending post deployment
// migrations from preventing the registry from starting.
m := premigrations.NewMigrator(db.Primary(), premigrations.SkipPostDeployment())
pending, err := m.HasPending()
if err != nil {
return nil, fmt.Errorf("failed to check database migrations status: %w", err)
}
if pending {
return nil, fmt.Errorf("there are pending database migrations, use the 'registry database migrate' CLI " +
"command to check and apply them")
}
app.db = db
app.registerShutdownFunc(
func(app *App, errCh chan error, l dlog.Logger) {
l.Info("closing database connections")
err := app.db.Close()
if err != nil {
err = fmt.Errorf("database shutdown: %w", err)
} else {
l.Info("database component has been shut down")
}
errCh <- err
},
)
options = append(options, storage.Database(app.db))
// update online GC settings (if needed) in the background to avoid delaying the app start
go func() {
if err := updateOnlineGCSettings(app.Context, app.db.Primary(), config); err != nil {
errortracking.Capture(err, errortracking.WithContext(app.Context), errortracking.WithStackTrace())
log.WithError(err).Error("failed to update online GC settings")
}
}()
startOnlineGC(app.Context, app.db.Primary(), app.driver, config)
// Now that we've started the database successfully, lock the filesystem
// to signal that this object storage needs to be managed by the database.
if feature.EnforceLockfiles.Enabled() {
if err := dbLocker.Lock(app.Context); err != nil {
log.WithError(err).Error("failed to mark filesystem for database only usage, see https://docs.gitlab.com/ee/administration/packages/container_registry_metadata_database.html")
return nil, err
}
}
if config.Database.BackgroundMigrations.Enabled && feature.BBMProcess.Enabled() {
startBackgroundMigrations(app.Context, app.db.Primary(), config)
}
}
// configure storage caches
// It's possible that the metadata database will fill the same original need
// as the blob descriptor cache (avoiding slow and/or expensive calls to
// external storage) and enable the cache to be removed long term.
//
// For now, disabling concurrent use of the metadata database and cache
// decreases the surface area which we are testing during database development.
if cc, ok := config.Storage["cache"]; ok && !config.Database.Enabled {
v, ok := cc["blobdescriptor"]
if !ok {
// Backwards compatible: "layerinfo" == "blobdescriptor"
v = cc["layerinfo"]
}
switch v {
case "redis":
if app.redisBlobDesc == nil {
return nil, fmt.Errorf("redis configuration required to use for layerinfo cache")
}
cacheProvider := rediscache.NewRedisBlobDescriptorCacheProvider(app.redisBlobDesc)
options = append(options, storage.BlobDescriptorCacheProvider(cacheProvider))
log.Info("using redis blob descriptor cache")
case "inmemory":
cacheProvider := memorycache.NewInMemoryBlobDescriptorCacheProvider()
options = append(options, storage.BlobDescriptorCacheProvider(cacheProvider))
log.Info("using inmemory blob descriptor cache")
default:
if v != "" {
log.WithField("type", config.Storage["cache"]).Warn("unknown cache type, caching disabled")
}
}
} else if ok && config.Database.Enabled {
log.Warn("blob descriptor cache is not compatible with metadata database, caching disabled")
}
if app.registry == nil {
// Initialize registry once with all options
app.registry, err = storage.NewRegistry(app.Context, app.driver, options...)
if err != nil {
return nil, fmt.Errorf("could not create registry: %w", err)
}
}
// Check filesystem lock file after registry is initialized
if !config.Database.Enabled {
if err := app.handleFilesystemLockFile(ctx, dbLocker, fsLocker); err != nil {
return nil, err
}
log.Info("registry filesystem metadata in use")
}
app.registry, err = applyRegistryMiddleware(app, app.registry, config.Middleware["registry"])
if err != nil {
return nil, err
}
authType := config.Auth.Type()
if authType != "" && !strings.EqualFold(authType, "none") {
accessController, err := auth.GetAccessController(config.Auth.Type(), config.Auth.Parameters())
if err != nil {
return nil, fmt.Errorf("unable to configure authorization (%s): %w", authType, err)
}
app.accessController = accessController
log.WithField("auth_type", authType).Debug("configured access controller")
}
var ok bool
app.repoRemover, ok = app.registry.(distribution.RepositoryRemover)
if !ok {
log.Warn("registry does not implement RepositoryRemover. Will not be able to delete repos and tags")
}
return app, nil
}