RHINENG-10032: implement pg_repack job and admin API

Dockerfile

@@ -8,7 +8,7 @@ ARG INSTALL_TOOLS=no
 RUN dnf module enable -y postgresql:16 || curl -o /etc/yum.repos.d/postgresql.repo \
         https://copr.fedorainfracloud.org/coprs/mmraka/postgresql-16/repo/epel-8/mmraka-postgresql-16-epel-8.repo
 
-RUN dnf install -y go-toolset postgresql diffutils rpm-devel && \
+RUN dnf install -y go-toolset postgresql diffutils rpm-devel pg_repack && \
     ln -s /usr/libexec/platform-python /usr/bin/python3
 
 ENV GOPATH=/go \
@@ -53,7 +53,7 @@ RUN go build -v main.go
 
 # libs to be copied into runtime
 RUN mkdir -p /go/lib64 && \
-    ldd /go/src/app/main \
+    ldd /go/src/app/main /usr/bin/pg_repack \
     | awk '/=>/ {print $3}' \
     | sort -u \
     | while read lib ; do \
@@ -79,6 +79,7 @@ COPY --from=buildimg /etc/pki/tls/certs/ca-bundle.crt /etc/pki/tls/certs/
 COPY --from=buildimg /etc/crypto-policies/ /etc/crypto-policies/
 COPY --from=buildimg /usr/lib64/.lib* /usr/lib64/
 COPY --from=buildimg /usr/lib64/libssl* /usr/lib64/
+COPY --from=buildimg /usr/bin/pg_repack /usr/bin/
 
 # copy libs needed by main
 COPY --from=buildimg /go/lib64/* /lib64/

database_admin/migrations/129_create_pg_repack.up.sql

@@ -0,0 +1,7 @@
+DO $$
+BEGIN
+    IF EXISTS (SELECT 1 FROM pg_available_extensions WHERE name = 'pg_repack') THEN
+        CREATE EXTENSION IF NOT EXISTS pg_repack;
+    END IF;
+END
+$$;

database_admin/schema/create_schema.sql

@@ -7,7 +7,7 @@ CREATE TABLE IF NOT EXISTS schema_migrations
 
 
 INSERT INTO schema_migrations
-VALUES (128, false);
+VALUES (129, false);
 
 -- ---------------------------------------------------------------------------
 -- Functions

deploy/clowdapp.yaml

@@ -399,6 +399,34 @@ objects:
                                                       key: vmaas-sync-database-password}}}
         - {name: POD_CONFIG, value: '${JOBS_CONFIG}'}
 
+    - name: repack
+      activeDeadlineSeconds: ${{JOBS_TIMEOUT}}
+      schedule: ${REPACK_SCHEDULE}
+      suspend: ${{REPACK_SUSPEND}}
+      concurrencyPolicy: Forbid
+      podSpec:
+        image: ${IMAGE}:${IMAGE_TAG_JOBS}
+        initContainers:
+          - name: check-for-db
+            image: ${IMAGE}:${IMAGE_TAG_DATABASE_ADMIN}
+            command:
+              - ./database_admin/check-upgraded.sh
+            env:
+            - {name: POD_CONFIG, value: '${DATABASE_ADMIN_CONFIG}'}
+        command:
+          - ./scripts/entrypoint.sh
+          - job
+          - repack
+        env:
+        - {name: LOG_LEVEL, value: '${LOG_LEVEL_JOBS}'}
+        - {name: GIN_MODE, value: '${GIN_MODE}'}
+        - {name: DB_DEBUG, value: '${DB_DEBUG_JOBS}'}
+        - {name: DB_USER, value: vmaas_sync}
+        - {name: DB_PASSWD, valueFrom: {secretKeyRef: {name: patchman-engine-database-passwords,
+                                                       key: vmaas-sync-database-password}}}
+        - {name: PROMETHEUS_PUSHGATEWAY,value: '${PROMETHEUS_PUSHGATEWAY}'}
+        - {name: POD_CONFIG, value: '${JOBS_CONFIG}'}
+
     database:
       name: patchman
       version: 16
@@ -595,6 +623,10 @@ parameters:
 - {name: PKG_REFRESH_SUSPEND, value: 'false'} # Disable cronjob execution
 - {name: ADVISORY_REFRESH_SCHEDULE, value: '*/15 * * * *'} # Cronjob schedule definition
 - {name: ADVISORY_REFRESH_SUSPEND, value: 'true'} # Disable cronjob execution
+# Repack
+- {name: REPACK_SCHEDULE, value: '0 11 * * 1'} # Cronjob schedule definition
+- {name: REPACK_SUSPEND, value: 'false'} # Disable cronjob execution
+
 
 # Database admin
 - {name: IMAGE_TAG_DATABASE_ADMIN, value: v3.6.110}

dev/database/Dockerfile

@@ -1,5 +1,11 @@
 FROM quay.io/cloudservices/postgresql-rds:16-4649c84
 
+# install pg_repack
+USER root
+RUN curl -o /etc/yum.repos.d/postgresql.repo \
+        https://copr.fedorainfracloud.org/coprs/mmraka/postgresql-16/repo/epel-8/mmraka-postgresql-16-epel-8.repo
+RUN dnf install -y pg_repack
+
 ADD /dev/database/init.sh /docker-entrypoint-initdb.d/
 
 USER postgres

docs/admin/openapi.json

@@ -357,6 +357,70 @@
                 ]
             }
         },
+        "/repack/{table_name}": {
+            "get": {
+                "summary": "Reindex and cluster DB with pg_repack",
+                "description": "Reindex the table from `table_name`. If `columns` are provided, clustering is performed as well.",
+                "operationId": "repack",
+                "parameters": [
+                    {
+                        "name": "table_name",
+                        "in": "path",
+                        "description": "Table to reindex",
+                        "required": true,
+                        "schema": {
+                            "type": "string"
+                        }
+                    },
+                    {
+                        "name": "columns",
+                        "in": "query",
+                        "description": "Comma-separated columns to cluster by (optional)",
+                        "schema": {
+                            "type": "string"
+                        }
+                    }
+                ],
+                "responses": {
+                    "200": {
+                        "description": "OK",
+                        "content": {
+                            "application/json": {
+                                "schema": {
+                                    "type": "string"
+                                }
+                            }
+                        }
+                    },
+                    "400": {
+                        "description": "Bad Request",
+                        "content": {
+                            "application/json": {
+                                "schema": {
+                                    "type": "string"
+                                }
+                            }
+                        }
+                    },
+                    "500": {
+                        "description": "Internal Server Error",
+                        "content": {
+                            "application/json": {
+                                "schema": {
+                                    "type": "object",
+                                    "additionalProperties": true
+                                }
+                            }
+                        }
+                    }
+                },
+                "security": [
+                    {
+                        "RhIdentity": []
+                    }
+                ]
+            }
+        },
         "/sessions/{pid}": {
             "delete": {
                 "summary": "Terminate db session",

main.go

@@ -10,6 +10,7 @@ import (
 	"app/platform"
 	"app/tasks/caches"
 	"app/tasks/cleaning"
+	"app/tasks/repack"
 	"app/tasks/system_culling"
 	"app/tasks/vmaas_sync"
 	"app/turnpike"
@@ -69,5 +70,7 @@ func runJob(name string) {
 		cleaning.RunDeleteUnusedData()
 	case "packages_cache_refresh":
 		caches.RunPackageRefresh()
+	case "repack":
+		repack.RunRepack()
 	}
 }

manager/routes/routes.go

@@ -113,6 +113,7 @@ func InitAdmin(app *gin.Engine, enableTurnpikeAuth bool) {
 	api.GET("/sessions", admin.GetActiveSessionsHandler)
 	api.GET("/sessions/:search", admin.GetActiveSessionsHandler)
 	api.DELETE("/sessions/:pid", admin.TerminateSessionHandler)
+	api.GET("/repack/:table_name", admin.RepackHandler)
 
 	pprof := api.Group("/pprof")
 	pprof.GET("/evaluator_upload/:param", admin.GetEvaluatorUploadPprof)

tasks/repack/repack.go

@@ -0,0 +1,80 @@
+package repack
+
+import (
+	"app/base/core"
+	"app/base/utils"
+	"app/tasks"
+	"fmt"
+	"os"
+	"os/exec"
+)
+
+var pgRepackArgs = []string{
+	"--no-superuser-check",
+	"--no-password",
+	"-d", utils.CoreCfg.DBName,
+	"-h", utils.CoreCfg.DBHost,
+	"-p", fmt.Sprintf("%d", utils.CoreCfg.DBPort),
+	"-U", utils.CoreCfg.DBUser,
+}
+
+func configure() {
+	core.ConfigureApp()
+}
+
+// GetCmd returns command that calls gp_repack with table. Table must be a partitioned table.
+// Args are appended to the necessary pgRepackArgs. Stdout and stderr of the subprocess are redirected to host.
+func getCmd(table string, args ...string) *exec.Cmd {
+	fullArgs := pgRepackArgs
+	fullArgs = append(fullArgs, "-I", table)
+	fullArgs = append(fullArgs, args...)
+	cmd := exec.Command("pg_repack", fullArgs...)
+	cmd.Env = append(os.Environ(), fmt.Sprintf("PGPASSWORD=%s", utils.CoreCfg.DBPassword))
+	cmd.Stdout = os.Stdout
+	cmd.Stderr = os.Stderr
+	return cmd
+}
+
+// Repack runs pg_repack reindex with table. If columns are provided, cluster by these columns is executed as well.
+// Table must be a partitioned table.
+func Repack(table string, columns string) error {
+	reindexCmd := getCmd(table, "-x")
+	err := reindexCmd.Run()
+	if err != nil {
+		return err
+	}
+
+	if len(columns) == 0 {
+		utils.LogWarn("no columns provided, skipping repack clustering")
+		return nil
+	}
+	clusterCmd := getCmd(table, "-o", columns)
+	err = clusterCmd.Run()
+	if err != nil {
+		return err
+	}
+
+	return nil
+}
+
+// RunRepack wraps Repack call for a job.
+func RunRepack() {
+	tasks.HandleContextCancel(tasks.WaitAndExit)
+	utils.LogInfo("Starting repack job")
+	configure()
+
+	TABLES := map[string]string{
+		"system_package2":   "rh_account_id,system_id",
+		"system_platform":   "rh_account_id,id,inventory_id",
+		"system_advisories": "rh_account_id,system_id",
+	}
+
+	for table, columns := range TABLES {
+		err := Repack(table, columns)
+		if err != nil {
+			utils.LogError("err", err.Error(), fmt.Sprintf("Failed to repack table %s", table))
+			continue
+		}
+		utils.LogInfo(fmt.Sprintf("Successfully repacked table %s", table))
+	}
+}

turnpike/controllers/admin.go

@@ -4,10 +4,12 @@ import (
 	"app/base/database"
 	"app/base/utils"
 	"app/tasks/caches"
+	"app/tasks/repack"
 	sync "app/tasks/vmaas_sync"
 	"fmt"
 	"io"
 	"net/http"
+	"regexp"
 	"strconv"
 	"time"
 
@@ -198,6 +200,44 @@ func TerminateSessionHandler(c *gin.Context) {
 	c.JSON(http.StatusOK, fmt.Sprintf("pid: %s terminated", param))
 }
 
+// @Summary Reindex and cluster DB with pg_repack
+// @Description Reindex the table from `table_name`. If `columns` are provided, clustering is performed as well.
+// @ID repack
+// @Security RhIdentity
+// @Accept   json
+// @Produce  json
+// @Param    table_name path string true "Table to reindex"
+// @Param    columns query string false "Comma-separated columns to cluster by (optional)"
+// @Success 200 {object} string
+// @Failure 400 {object} string
+// @Failure 500 {object} map[string]interface{}
+// @Router /repack/{table_name} [get]
+func RepackHandler(c *gin.Context) {
+	utils.LogInfo("manual repack called...")
+
+	tableName := c.Param("table_name")
+	if ok, _ := regexp.MatchString(`^\w+$`, tableName); !ok {
+		c.JSON(http.StatusBadRequest, utils.ErrorResponse{Error: "invalid table_name"})
+		return
+	}
+
+	columns := c.Query("columns")
+	if ok, _ := regexp.MatchString(`^[\w,]*$`, columns); !ok {
+		c.JSON(http.StatusBadRequest, utils.ErrorResponse{Error: "invalid columns"})
+		return
+	}
+
+	err := repack.Repack(tableName, columns)
+	if err != nil {
+		utils.LogError("err", err.Error(), "manual repack call failed")
+		c.JSON(http.StatusInternalServerError, gin.H{"err": err.Error()})
+		return
+	}
+
+	utils.LogInfo("manual repack finished successfully")
+	c.JSON(http.StatusOK, "OK")
+}
+
 // @Summary Get profile info
 // @Description Get profile info
 // @ID getEvaluatorUploadPprof