HMS-5056: add layered RH repos with feature name (#941)

* HMS-5056: add layered RH repos with feature name

* store feature_name in db

* add index

Author: xbhouse

configs/config.yaml.example

@@ -62,6 +62,7 @@ options:
   enable_notifications: true
   template_event_topic: "platform.content-sources.template"
   snapshot_retain_days_limit: 365
+  feature_filter: ["RHEL-OS-x86_64"]
 metrics:
   path: "/metrics"
   port: 9000

db/migrations.latest

@@ -1 +1 @@
-20241203143614
+20250115112007

db/migrations/20250115112007_add_feature_name_to_repo_configs.down.sql

@@ -0,0 +1,6 @@
+BEGIN;
+
+DROP INDEX IF EXISTS repo_config_feature_name;
+ALTER TABLE repository_configurations DROP COLUMN feature_name;
+
+COMMIT;

db/migrations/20250115112007_add_feature_name_to_repo_configs.up.sql

@@ -0,0 +1,8 @@
+BEGIN;
+
+ALTER TABLE repository_configurations
+    ADD COLUMN IF NOT EXISTS feature_name VARCHAR (255) DEFAULT NULL;
+
+CREATE INDEX IF NOT EXISTS repo_config_feature_name ON repository_configurations(feature_name);
+
+COMMIT;

deployments/deployment.yaml

@@ -134,6 +134,8 @@ objects:
                 value: ${OPTIONS_ENABLE_NOTIFICATIONS}
               - name: OPTIONS_REPOSITORY_IMPORT_FILTER
                 value: ${OPTIONS_REPOSITORY_IMPORT_FILTER}
+              - name: OPTIONS_FEATURE_FILTER
+                value: ${OPTIONS_FEATURE_FILTER}
               - name: TASKING_WORKER_COUNT
                 value: ${TASKING_WORKER_COUNT}
               - name: CLIENTS_CANDLEPIN_SERVER
@@ -860,3 +862,5 @@ parameters:
     description: Number of task workers running within a single worker process
   - name: CLIENTS_CANDLEPIN_SERVER
     default: ''
+  - name: OPTIONS_FEATURE_FILTER
+    description: Comma separated list of features that determine which repos to import

pkg/config/config.go

@@ -175,8 +175,9 @@ type Options struct {
 	// url (https://servername) to access the api, used to reference gpg keys
 	// Supports partial hostnames (i.e. http://.server.example.com).
 	// If this is encountered (and clowder is used), it will prepend the envName from clowder
-	ExternalURL             string `mapstructure:"external_url"`
-	SnapshotRetainDaysLimit int    `mapstructure:"snapshot_retain_days_limit"`
+	ExternalURL             string   `mapstructure:"external_url"`
+	SnapshotRetainDaysLimit int      `mapstructure:"snapshot_retain_days_limit"`
+	FeatureFilter           []string `mapstructure:"feature_filter"` // Used to control which repos are imported based on feature name
 }
 
 type Metrics struct {
@@ -197,6 +198,8 @@ const (
 	DefaultIntrospectApiTimeLimitSec = 30
 )
 
+var featureFilter = [...]string{"RHEL-OS-x86_64"}
+
 var LoadedConfig Configuration
 
 func Get() *Configuration {
@@ -243,6 +246,7 @@ func setDefaults(v *viper.Viper) {
 	v.SetDefault("options.enable_notifications", false)
 	v.SetDefault("options.template_event_topic", "platform.content-sources.template")
 	v.SetDefault("options.repository_import_filter", "")
+	v.SetDefault("options.feature_filter", featureFilter)
 	v.SetDefault("options.external_url", "http://pulp.content:8000")
 	v.SetDefault("options.snapshot_retain_days_limit", 365)
 	v.SetDefault("logging.level", "info")

pkg/dao/interfaces.go

@@ -76,7 +76,7 @@ type RepositoryConfigDao interface {
 	InternalOnly_FetchRepoConfigsForRepoUUID(ctx context.Context, uuid string) []api.RepositoryResponse
 	UpdateLastSnapshotTask(ctx context.Context, taskUUID string, orgID string, repoUUID string) error
 	UpdateLastSnapshot(ctx context.Context, orgID string, repoConfigUUID string, snapUUID string) error
-	InternalOnly_RefreshRedHatRepo(ctx context.Context, request api.RepositoryRequest, label string) (*api.RepositoryResponse, error)
+	InternalOnly_RefreshRedHatRepo(ctx context.Context, request api.RepositoryRequest, label string, featureName string) (*api.RepositoryResponse, error)
 	FetchWithoutOrgID(ctx context.Context, uuid string) (api.RepositoryResponse, error)
 	BulkExport(ctx context.Context, orgID string, reposToExport api.RepositoryExportRequest) ([]api.RepositoryExportResponse, error)
 	BulkImport(ctx context.Context, reposToImport []api.RepositoryRequest) ([]api.RepositoryImportResponse, []error)

pkg/dao/repository_configs.go

@@ -1247,7 +1247,7 @@ func isTimeout(err error) bool {
 	return false
 }
 
-func (r repositoryConfigDaoImpl) InternalOnly_RefreshRedHatRepo(ctx context.Context, request api.RepositoryRequest, label string) (*api.RepositoryResponse, error) {
+func (r repositoryConfigDaoImpl) InternalOnly_RefreshRedHatRepo(ctx context.Context, request api.RepositoryRequest, label string, featureName string) (*api.RepositoryResponse, error) {
 	newRepoConfig := models.RepositoryConfiguration{}
 	newRepo := models.Repository{}
 
@@ -1256,6 +1256,7 @@ func (r repositoryConfigDaoImpl) InternalOnly_RefreshRedHatRepo(ctx context.Cont
 
 	newRepoConfig.OrgID = config.RedHatOrg
 	newRepoConfig.Label = label
+	newRepoConfig.FeatureName = featureName
 	newRepo.Origin = config.OriginRedHat
 	newRepo.Public = true // Ensure all RH repos can be searched
 
@@ -1278,7 +1279,7 @@ func (r repositoryConfigDaoImpl) InternalOnly_RefreshRedHatRepo(ctx context.Cont
 	result = r.db.WithContext(ctx).Clauses(clause.OnConflict{
 		Columns:     []clause.Column{{Name: "repository_uuid"}, {Name: "org_id"}},
 		TargetWhere: clause.Where{Exprs: []clause.Expression{clause.Eq{Column: "deleted_at", Value: nil}}},
-		DoUpdates:   clause.AssignmentColumns([]string{"name", "arch", "versions", "gpg_key", "label"})}).
+		DoUpdates:   clause.AssignmentColumns([]string{"name", "arch", "versions", "gpg_key", "label", "feature_name"})}).
 		Create(&newRepoConfig)
 	if result.Error != nil {
 		return nil, result.Error

pkg/dao/repository_configs_mock.go

@@ -2633,7 +2633,7 @@ func (suite *RepositoryConfigSuite) TestRefreshRedHatRepo() {
 		ContentType:          utils.Ptr(config.ContentTypeRpm),
 		Snapshot:             utils.Ptr(true),
 	}
-	response, err := dao.InternalOnly_RefreshRedHatRepo(context.Background(), rhRepo, "another-label")
+	response, err := dao.InternalOnly_RefreshRedHatRepo(context.Background(), rhRepo, "another-label", "test-feature")
 	assert.NoError(suite.T(), err)
 
 	assert.NotEmpty(suite.T(), response.UUID)
@@ -2644,7 +2644,7 @@ func (suite *RepositoryConfigSuite) TestRefreshRedHatRepo() {
 	// Change the name
 	rhRepo.Name = utils.Ptr("another name")
 
-	response, err = dao.InternalOnly_RefreshRedHatRepo(context.Background(), rhRepo, "some-label")
+	response, err = dao.InternalOnly_RefreshRedHatRepo(context.Background(), rhRepo, "some-label", "test-feature")
 	assert.NoError(suite.T(), err)
 
 	assert.Equal(suite.T(), *rhRepo.Name, response.Name)

pkg/dao/repository_configs_test.go

@@ -27,6 +27,7 @@ type RedHatRepo struct {
 	Selector            string `json:"selector"`
 	GpgKey              string `json:"gpg_key"`
 	Label               string `json:"content_label"`
+	FeatureName         string `json:"feature_name"`
 }
 
 func (rhr RedHatRepo) ToRepositoryRequest() api.RepositoryRequest {
@@ -63,7 +64,7 @@ func (rhr *RedHatRepoImporter) LoadAndSave(ctx context.Context) error {
 	}
 	for _, r := range repos {
 		r.GpgKey = gpgKey
-		_, err = rhr.daoReg.RepositoryConfig.InternalOnly_RefreshRedHatRepo(ctx, r.ToRepositoryRequest(), r.Label)
+		_, err = rhr.daoReg.RepositoryConfig.InternalOnly_RefreshRedHatRepo(ctx, r.ToRepositoryRequest(), r.Label, r.FeatureName)
 		if err != nil {
 			return err
 		}
@@ -96,9 +97,12 @@ func (rhr *RedHatRepoImporter) loadFromFile() ([]RedHatRepo, error) {
 	}
 	filteredRepos := []RedHatRepo{}
 	filter := config.Get().Options.RepositoryImportFilter
+	features := config.Get().Options.FeatureFilter
 	for _, repo := range repos {
 		if filter == "" || repo.Selector == filter {
-			filteredRepos = append(filteredRepos, repo)
+			if utils.Contains(features, repo.FeatureName) {
+				filteredRepos = append(filteredRepos, repo)
+			}
 		}
 	}
 	return filteredRepos, nil