func NewApp()

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
}