diff --git a/AGENTS.md b/AGENTS.md index 76f9c9842..cb5e3959a 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -29,6 +29,7 @@ Prefer these sources over guessing when behavior or schema matters. | HTTP REST API | `manager/` | | External messages | `listener/`, topic names in code and `conf/` | | Evaluation | `evaluator/`, topic names in code and `conf/` | +| Advisory aggregation | `aggregator/` | | Advisory sync | `tasks/vmaas_sync/` | | Migrations | `database_admin/migrations/` (verify naming against existing migrations) | | Migration flow, session flags, ops runbook | `database_admin/update.go`, [docs/md/database.md#migrations](docs/md/database.md#migrations), [docs/md/major-migration-runbook.md](docs/md/major-migration-runbook.md) | @@ -55,11 +56,17 @@ Listener Component ↓ Evaluator-Upload Component ↓ (calls VMaaS /updates) - ↓ (updates system_advisories, advisory_account_data) + ↓ (updates system_advisories) + ↓ (updates advisory_account_data — legacy table, to be removed) ↓ -[platform.notifications.ingress] (optional) [platform.remediation-updates.patch] (optional) [platform.inventory.host-apps] (optional) +[patchman.advisory.update] Kafka Topic (changed advisory IDs) + ↓ +Aggregator Component + ↓ (recounts from system_advisories, writes aggregates to account_advisory — new workspace-scoped table) + ↓ +[platform.notifications.ingress] (optional) ``` ### Advisory Sync Flow diff --git a/Dockerfile b/Dockerfile index ddc5f0d75..652dab0b0 100644 --- a/Dockerfile +++ b/Dockerfile @@ -41,6 +41,7 @@ ADD --chown=insights:insights scripts /go/src/app/scripts ADD --chown=insights:insights database_admin /go/src/app/database_admin ADD --chown=insights:insights docs /go/src/app/docs ADD --chown=insights:insights evaluator /go/src/app/evaluator +ADD --chown=insights:insights aggregator /go/src/app/aggregator ADD --chown=insights:insights listener /go/src/app/listener ADD --chown=insights:insights tasks /go/src/app/tasks ADD --chown=insights:insights base /go/src/app/base diff --git a/aggregator/aggregator.go b/aggregator/aggregator.go new file mode 100644 index 000000000..4bcbf320d --- /dev/null +++ b/aggregator/aggregator.go @@ -0,0 +1,64 @@ +package aggregator + +import ( + "app/base" + "app/base/core" + "app/base/mqueue" + "app/base/utils" + "sync" + + "github.com/gin-gonic/gin" +) + +var ( + advisoryUpdateTopic string + enableNotifications bool + consumerCount int +) + +func readPodConfig() { + consumerCount = utils.PodConfig.GetInt("consumer_count", 1) + enableNotifications = utils.PodConfig.GetBool("instant_notifications", false) +} + +func configure() { + advisoryUpdateTopic = utils.FailIfEmpty(utils.CoreCfg.AdvisoryUpdateTopic, "ADVISORY_UPDATE_TOPIC") + configureNotifications() +} + +func runServer() { + app := gin.New() + core.InitProbes(app) + go base.TryExposeOnMetricsPort(app) + + err := utils.RunServer(base.Context, app, utils.CoreCfg.PublicPort) + if err != nil { + utils.LogError("err", err.Error()) + panic(err) + } +} + +func subscribeToAdvisoryUpdates(wg *sync.WaitGroup, readerBuilder mqueue.CreateReader) { + handler := mqueue.MakeRetryingHandler(advisoryUpdateHandler) + for i := 0; i < consumerCount; i++ { + mqueue.SpawnReader(base.Context, wg, advisoryUpdateTopic, readerBuilder, handler) + utils.LogDebug("spawned advisory update reader", i) + } + utils.LogInfo("connected to kafka topics") +} + +func RunAggregator() { + var wg sync.WaitGroup + + utils.LogInfo("aggregator starting") + core.ConfigureApp() + readPodConfig() + configure() + + go runServer() + go utils.RunProfiler() + subscribeToAdvisoryUpdates(&wg, mqueue.NewKafkaReaderFromEnv) + + wg.Wait() + utils.LogInfo("aggregator completed") +} diff --git a/aggregator/events.go b/aggregator/events.go new file mode 100644 index 000000000..472ee4bc1 --- /dev/null +++ b/aggregator/events.go @@ -0,0 +1,10 @@ +package aggregator + +import ( + "app/base/mqueue" +) + +// TODO: stub - will process advisory update events in batches and update account_advisory table +func advisoryUpdateHandler(m mqueue.KafkaMessage) error { + return nil +} diff --git a/aggregator/notifications.go b/aggregator/notifications.go new file mode 100644 index 000000000..bc48d0fae --- /dev/null +++ b/aggregator/notifications.go @@ -0,0 +1,14 @@ +package aggregator + +import ( + "app/base/mqueue" + "app/base/utils" +) + +var notificationsPublisher mqueue.Writer + +func configureNotifications() { + if topic := utils.CoreCfg.NotificationsTopic; topic != "" { + notificationsPublisher = mqueue.NewKafkaWriterFromEnv(topic) + } +} diff --git a/conf/aggregator.env b/conf/aggregator.env new file mode 100644 index 000000000..81ec9eb49 --- /dev/null +++ b/conf/aggregator.env @@ -0,0 +1 @@ +POD_CONFIG=label=aggregator diff --git a/conf/aggregator_common.env b/conf/aggregator_common.env new file mode 100644 index 000000000..d7dd977ec --- /dev/null +++ b/conf/aggregator_common.env @@ -0,0 +1,2 @@ +DB_USER=evaluator +DB_PASSWD=evaluator diff --git a/deploy/clowdapp.yaml b/deploy/clowdapp.yaml index 6bffb7b0b..d729d43f4 100644 --- a/deploy/clowdapp.yaml +++ b/deploy/clowdapp.yaml @@ -306,6 +306,51 @@ objects: limits: {cpu: '${CPU_LIMIT_EVALUATOR_USER_EVALUATION}', memory: '${MEM_LIMIT_EVALUATOR_USER_EVALUATION}'} requests: {cpu: '${CPU_REQUEST_EVALUATOR_USER_EVALUATION}', memory: '${MEM_REQUEST_EVALUATOR_USER_EVALUATION}'} + - name: aggregator + replicas: ${{REPLICAS_AGGREGATOR}} + webServices: + public: + enabled: true + private: + enabled: true + metrics: + enabled: true + podSpec: + image: ${IMAGE}:${IMAGE_TAG} + initContainers: + - name: check-for-db + image: ${IMAGE}:${IMAGE_TAG} + command: + - ./database_admin/check-upgraded.sh + env: + - {name: POD_CONFIG, value: '${DATABASE_ADMIN_CONFIG}'} + command: + - ./scripts/entrypoint.sh + - aggregator + env: + - {name: LOG_LEVEL, value: '${LOG_LEVEL_AGGREGATOR}'} + - {name: GIN_MODE, value: '${GIN_MODE}'} + - {name: SENTRY_DSN, valueFrom: {secretKeyRef: {name: patchman-sentry, key: sentry-dsn}}} + - {name: SHOW_CLOWDER_VARS, value: ''} + - {name: DB_DEBUG, value: '${DB_DEBUG_AGGREGATOR}'} + - {name: DB_USER, value: evaluator} + - {name: DB_PASSWD, valueFrom: {secretKeyRef: {name: patchman-engine-database-passwords, + key: evaluator-database-password}}} + - {name: KAFKA_GROUP, value: patchman} + - {name: KAFKA_READER_MAX_ATTEMPTS, value: '${KAFKA_READER_MAX_ATTEMPTS}'} + - {name: KAFKA_WRITER_MAX_ATTEMPTS, value: '${KAFKA_WRITER_MAX_ATTEMPTS}'} + - {name: NOTIFICATIONS_TOPIC, value: 'platform.notifications.ingress'} + - {name: ADVISORY_UPDATE_TOPIC, value: 'patchman.advisory.update'} + - {name: SSL_CERT_DIR, value: '${SSL_CERT_DIR}'} + - {name: GOGC, value: '${GOGC}'} + - {name: ENABLE_PROFILER, value: '${ENABLE_PROFILER_AGGREGATOR}'} + - {name: GOMEMLIMIT, value: '${GOMEMLIMIT_AGGREGATOR}'} + - {name: POD_CONFIG, value: 'label=aggregator;${AGGREGATOR_CONFIG}'} + - {name: CONSOLEDOT_HOSTNAME, value: '${CONSOLEDOT_HOSTNAME}'} + resources: + limits: {cpu: '${CPU_LIMIT_AGGREGATOR}', memory: '${MEM_LIMIT_AGGREGATOR}'} + requests: {cpu: '${CPU_REQUEST_AGGREGATOR}', memory: '${MEM_REQUEST_AGGREGATOR}'} + jobs: - name: db-migration completions: 1 @@ -785,6 +830,18 @@ parameters: - {name: ENABLE_PROFILER_EVALUATOR_USER_EVALUATION, value: 'false'} - {name: EVALUATOR_USER_EVALUATION_CONFIG, value: ''} +# Aggregator +- {name: REPLICAS_AGGREGATOR, value: '0'} +- {name: LOG_LEVEL_AGGREGATOR, value: debug} +- {name: DB_DEBUG_AGGREGATOR, value: 'false'} +- {name: CPU_LIMIT_AGGREGATOR, value: 500m} +- {name: MEM_LIMIT_AGGREGATOR, value: 512Mi} +- {name: CPU_REQUEST_AGGREGATOR, value: 250m} +- {name: MEM_REQUEST_AGGREGATOR, value: 256Mi} +- {name: ENABLE_PROFILER_AGGREGATOR, value: 'false'} +- {name: GOMEMLIMIT_AGGREGATOR, value: '460MiB'} +- {name: AGGREGATOR_CONFIG, value: ''} + # JOBS - {name: LOG_LEVEL_JOBS, value: debug} - {name: DB_DEBUG_JOBS, value: 'false'} diff --git a/docker-compose.prod.yml b/docker-compose.prod.yml index 7b6ff5d41..a29b321a8 100644 --- a/docker-compose.prod.yml +++ b/docker-compose.prod.yml @@ -194,6 +194,27 @@ services: security_opt: - label=disable + aggregator: + container_name: aggregator + image: patchman-engine-app + env_file: + - ./conf/common.env + - ./conf/aggregator_common.env + - ./conf/aggregator.env + command: ./dev/scripts/docker-compose-entrypoint.sh aggregator + ports: + - 8087:8080 + - 9007:9000 # private port - pprof + depends_on: + - db + - platform + volumes: + - ./dev:/go/src/app/dev + - ./dev/database/secrets:/opt/postgresql + - ./dev/kafka/secrets:/opt/kafka + security_opt: + - label=disable + vmaas_sync: container_name: vmaas_sync image: patchman-engine-app diff --git a/docker-compose.yml b/docker-compose.yml index 131109a08..3e686f7c2 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -188,6 +188,26 @@ services: security_opt: - label=disable + aggregator: + container_name: aggregator + image: patchman-engine-app + env_file: + - ./conf/common.env + - ./conf/aggregator_common.env + - ./conf/aggregator.env + - ./conf/gorun.env + command: ./dev/scripts/docker-compose-entrypoint.sh aggregator + ports: + - 8087:8080 + - 9007:9000 # private port - pprof + depends_on: + - db + - platform + volumes: + - ./:/go/src/app + security_opt: + - label=disable + vmaas_sync: container_name: vmaas_sync image: patchman-engine-app diff --git a/main.go b/main.go index 90f4828aa..6bc9d5965 100644 --- a/main.go +++ b/main.go @@ -1,6 +1,7 @@ package main import ( + "app/aggregator" "app/base" "app/base/utils" "app/database_admin" @@ -38,6 +39,9 @@ func main() { case "evaluator": evaluator.RunEvaluator() return + case "aggregator": + aggregator.RunAggregator() + return case "migrate": database_admin.UpdateDB(os.Args[2]) return diff --git a/scripts/check-dockercomposes.sh b/scripts/check-dockercomposes.sh index c805d31d6..9b96b15d4 100755 --- a/scripts/check-dockercomposes.sh +++ b/scripts/check-dockercomposes.sh @@ -9,7 +9,7 @@ sed \ -e "s|INSTALL_TOOLS=yes|INSTALL_TOOLS=no|" \ -e "s|target: buildimg|target: runtimeimg|" \ -e "/ - \.\/conf\/gorun.env/ d" \ - -e "/ \(db_admin\|db_feed\|manager\|listener\|evaluator_recalc\|evaluator_upload\|evaluator_user_evaluation\|vmaas_sync\|admin\|migrate_system_package2\):/,/^$/ { + -e "/ \(db_admin\|db_feed\|manager\|listener\|evaluator_recalc\|evaluator_upload\|evaluator_user_evaluation\|aggregator\|vmaas_sync\|admin\|migrate_system_package2\):/,/^$/ { s/- \.\/:\/go\/src\/app/- \.\/dev:\/go\/src\/app\/dev\n\ - .\/dev\/database\/secrets:\/opt\/postgresql\n\ - \.\/dev\/kafka\/secrets:\/opt\/kafka/