Fixes 4775: add weekly cron for pulp orphan cleanup

cmd/external-repos/main.go

@@ -4,8 +4,10 @@ import (
 	"context"
 	"fmt"
 	"os"
+	"slices"
 	"sort"
 	"strconv"
+	"sync"
 	"time"
 
 	"github.com/content-services/content-sources-backend/pkg/config"
@@ -39,7 +41,7 @@ func main() {
 	dao.SetupGormTableOrFail(db.DB)
 
 	if len(args) < 2 {
-		log.Fatal().Msg("Requires arguments: download, import, introspect, snapshot, nightly-jobs [INTERVAL]")
+		log.Fatal().Msg("Requires arguments: download, import, introspect, snapshot, nightly-jobs [INTERVAL], pulp-orphan-cleanup [BATCH_SIZE]")
 	}
 	if args[1] == "download" {
 		if len(args) < 3 {
@@ -110,6 +112,22 @@ func main() {
 				log.Error().Err(err).Msg("error queueing snapshot tasks")
 			}
 		}
+	} else if args[1] == "pulp-orphan-cleanup" {
+		batchSize := 5
+		if len(args) > 2 {
+			parsed, err := strconv.ParseInt(args[2], 10, 0)
+			if err != nil {
+				log.Logger.Fatal().Err(err).Msgf("could not parse integer interval %v", args[2])
+			}
+			batchSize = int(parsed)
+		}
+		if !config.PulpConfigured() {
+			log.Error().Msg("cannot run orphan cleanup if pulp is not configured")
+		}
+		err := pulpOrphanCleanup(ctx, db.DB, batchSize)
+		if err != nil {
+			log.Error().Err(err).Msg("error starting pulp orphan cleanup tasks")
+		}
 	}
 }
 
@@ -274,3 +292,44 @@ func enqueueSnapshotRepos(ctx context.Context, urls *[]string, interval *int) er
 	}
 	return nil
 }
+
+func pulpOrphanCleanup(ctx context.Context, db *gorm.DB, batchSize int) error {
+	var err error
+	daoReg := dao.GetDaoRegistry(db)
+
+	domains, err := daoReg.Domain.List(ctx)
+	if err != nil {
+		log.Panic().Err(err).Msg("orphan cleanup error: error listing orgs")
+	}
+
+	for batch := range slices.Chunk(domains, batchSize) {
+		wg := sync.WaitGroup{}
+		for _, domain := range batch {
+			org := domain.OrgId
+			domainName := domain.DomainName
+
+			logger := log.Logger.With().Str("org_id", org).Str("pulp_domain_name", domainName).Logger()
+
+			wg.Add(1)
+			go func() {
+				defer wg.Done()
+
+				pulpClient := pulp_client.GetPulpClientWithDomain(domainName)
+				cleanupTask, err := pulpClient.OrphanCleanup(ctx)
+				if err != nil {
+					logger.Error().Err(err).Msgf("error starting orphan cleanup")
+					return
+				}
+				logger.Info().Str("task_href", cleanupTask).Msgf("running orphan cleanup for org: %v", org)
+
+				_, err = pulp_client.GetGlobalPulpClient().PollTask(ctx, cleanupTask)
+				if err != nil {
+					logger.Error().Err(err).Msgf("error polling pulp task for orphan cleanup")
+					return
+				}
+			}()
+		}
+		wg.Wait()
+	}
+	return nil
+}

create-repos.sh

@@ -0,0 +1,22 @@
+total_orgs=1
+
+for ((i=0; i<$total_orgs; ++i))
+do
+
+ORG_ID=`tr -dc A-Za-z0-9 </dev/urandom | head -c 13`
+
+HEADER=`./scripts/header.sh $ORG_ID`
+
+UUID=$(curl -X POST --location "http://localhost:8000/api/content-sources/v1.0/repositories/" \
+    -H "${HEADER}" \
+    -H "Content-Type: application/json" \
+    -d '{
+          "name": "comps repo 2",
+          "url": "https://rverdile.fedorapeople.org/dummy-repos/comps/repo2/",
+          "snapshot": true
+  }' | jq -r .uuid)
+
+curl -X DELETE --location "http://localhost:8000/api/content-sources/v1.0/repositories/$UUID" \
+    -H "${HEADER}" \
+    -H "Content-Type: application/json"
+done

deployments/deployment.yaml

@@ -578,6 +578,86 @@ objects:
                     name: content-sources-candlepin
                     key: key
                     optional: true
+        - name: pulp-orphan-cleanup
+          # https://crontab.guru/
+          schedule: ${WEEKLY_CRON_JOB}
+          suspend: ${{SUSPEND_CRON_JOB}}
+          concurrencyPolicy: "Forbid"
+          podSpec:
+            securityContext:
+              runAsNonRoot: true
+              runAsUser: 1001
+            image: ${IMAGE}:${IMAGE_TAG}
+            inheritEnv: true
+            command:
+              - /external-repos
+              - pulp-orphan-cleanup 5
+            env:
+              - name: CLOWDER_ENABLED
+                value: ${CLOWDER_ENABLED}
+              - name: RH_CDN_CERT_PAIR
+                valueFrom:
+                  secretKeyRef:
+                    name: content-sources-certs
+                    key: cdn.redhat.com
+              - name: SENTRY_DSN
+                valueFrom:
+                  secretKeyRef:
+                    name: content-sources-sentry
+                    key: dsn
+                    optional: true
+              - name: CLIENTS_PULP_SERVER
+                value: ${{CLIENTS_PULP_SERVER}}
+              - name: CLIENTS_PULP_CUSTOM_REPO_CONTENT_GUARDS
+                value: ${CLIENTS_PULP_CUSTOM_REPO_CONTENT_GUARDS}
+              - name: CLIENTS_PULP_GUARD_SUBJECT_DN
+                value: ${{CLIENTS_PULP_GUARD_SUBJECT_DN}}
+              - name: CLIENTS_PULP_DOWNLOAD_POLICY
+                value: ${{CLIENTS_PULP_DOWNLOAD_POLICY}}
+              - name: CLIENTS_PULP_USERNAME
+                value: ${{CLIENTS_PULP_USERNAME}}
+              - name: CLIENTS_PULP_PASSWORD
+                valueFrom:
+                  secretKeyRef:
+                    name: pulp-content-sources-password
+                    key: password
+                    optional: true
+              - name: LOGGING_LEVEL
+                value: ${{LOGGING_LEVEL}}
+              - name: OPTIONS_EXTERNAL_URL
+                value: ${OPTIONS_EXTERNAL_URL}
+              - name: FEATURES_SNAPSHOTS_ENABLED
+                value: ${FEATURES_SNAPSHOTS_ENABLED}
+              - name: FEATURES_SNAPSHOTS_ACCOUNTS
+                value: ${FEATURES_SNAPSHOTS_ACCOUNTS}
+              - name: FEATURES_SNAPSHOTS_ORGANIZATIONS
+                value: ${FEATURES_SNAPSHOTS_ORGANIZATIONS}
+              - name: FEATURES_ADMIN_TASKS_ENABLED
+                value: ${FEATURES_ADMIN_TASKS_ENABLED}
+              - name: FEATURES_ADMIN_TASKS_ACCOUNTS
+                value: ${FEATURES_ADMIN_TASKS_ACCOUNTS}
+              - name: FEATURES_ADMIN_TASKS_ORGANIZATIONS
+                value: ${FEATURES_ADMIN_TASKS_ORGANIZATIONS}
+              - name: CLIENTS_RBAC_BASE_URL
+                value: ${{CLIENTS_RBAC_BASE_URL}}
+              - name: OPTIONS_ALWAYS_RUN_CRON_TASKS
+                value: ${OPTIONS_ALWAYS_RUN_CRON_TASKS}
+              - name: OPTIONS_ENABLE_NOTIFICATIONS
+                value: ${OPTIONS_ENABLE_NOTIFICATIONS}
+              - name: CLIENTS_CANDLEPIN_SERVER
+                value: ${CLIENTS_CANDLEPIN_SERVER}
+              - name: CLIENTS_CANDLEPIN_CLIENT_CERT
+                valueFrom:
+                  secretKeyRef:
+                    name: content-sources-candlepin
+                    key: cert
+                    optional: true
+              - name: CLIENTS_CANDLEPIN_CLIENT_KEY
+                valueFrom:
+                  secretKeyRef:
+                    name: content-sources-candlepin
+                    key: key
+                    optional: true
       database:
         name: content-sources
         version: 15
@@ -631,6 +711,8 @@ parameters:
     required: true
   - name: IMAGE
     value: quay.io/cloudservices/content-sources-backend
+  - name: WEEKLY_CRON_JOB
+    value: "0 5 * * 1"
   - name: NIGHTLY_CRON_JOB
     value: "0 0/1 * * *"
   - name: SUSPEND_CRON_JOB

pkg/dao/domains.go

@@ -59,3 +59,12 @@ func (dDao domainDaoImpl) Fetch(ctx context.Context, orgId string) (string, erro
 	}
 	return found[0].DomainName, nil
 }
+
+func (dDao domainDaoImpl) List(ctx context.Context) ([]models.Domain, error) {
+	var domains []models.Domain
+	result := dDao.db.WithContext(ctx).Table("domains").Find(&domains)
+	if result.Error != nil {
+		return nil, result.Error
+	}
+	return domains, nil
+}

pkg/dao/domains_test.go

@@ -6,6 +6,7 @@ import (
 	"testing"
 
 	"github.com/content-services/content-sources-backend/pkg/db"
+	"github.com/content-services/content-sources-backend/pkg/models"
 	uuid2 "github.com/google/uuid"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/suite"
@@ -38,6 +39,35 @@ func (ds *DomainSuite) TestCreate() {
 	assert.NotEmpty(ds.T(), name)
 }
 
+func (ds *DomainSuite) TestList() {
+	dd := domainDaoImpl{db: ds.tx}
+	numOrgs := 5
+
+	var existingOrgs []models.Domain
+	res := ds.tx.Model(&models.Domain{}).Find(&existingOrgs)
+	assert.NoError(ds.T(), res.Error)
+
+	newOrgs := make([]models.Domain, numOrgs)
+	for i := 0; i < numOrgs; i++ {
+		orgID := randomHexadecimal(10)
+
+		name, err := dd.Create(context.Background(), orgID)
+		assert.NoError(ds.T(), err)
+		assert.NotEmpty(ds.T(), name)
+
+		newOrgs[i].OrgId = orgID
+		newOrgs[i].DomainName = name
+	}
+
+	expectedOrgs := append(newOrgs, existingOrgs...)
+	expectedCount := len(newOrgs) + len(existingOrgs)
+
+	orgs, err := dd.List(context.Background())
+	assert.NoError(ds.T(), err)
+	assert.Equal(ds.T(), expectedCount, len(orgs))
+	assert.ElementsMatch(ds.T(), expectedOrgs, orgs)
+}
+
 func TestConcurrentGetDomainName(t *testing.T) {
 	// Note, this test does not use a transaction, as it fails when multiple go routines are trying to do that
 	orgId := uuid2.NewString()

pkg/dao/interfaces.go

@@ -147,6 +147,7 @@ type AdminTaskDao interface {
 type DomainDao interface {
 	FetchOrCreateDomain(ctx context.Context, orgId string) (string, error)
 	Fetch(ctx context.Context, orgId string) (string, error)
+	List(ctx context.Context) ([]models.Domain, error)
 }
 
 type PackageGroupDao interface {

pkg/utils/utils.go

@@ -34,7 +34,7 @@ func Contains[T comparable](elems []T, v T) bool {
 	return false
 }
 
-// Converts any struct to a pointer to that struct
+// Ptr converts any value to a pointer to that value
 func Ptr[T any](item T) *T {
 	return &item
 }