diff --git a/go.mod b/go.mod index 1a7d527a7..f72bfe535 100644 --- a/go.mod +++ b/go.mod @@ -10,6 +10,7 @@ require ( github.com/coreos/go-oidc v2.5.0+incompatible github.com/gin-contrib/cors v1.7.7 github.com/gin-gonic/gin v1.12.0 + github.com/go-co-op/gocron v1.37.0 github.com/go-sql-driver/mysql v1.10.0 github.com/goark/go-cvss v1.6.7 github.com/golang-jwt/jwt/v5 v5.3.1 @@ -52,9 +53,11 @@ require ( github.com/lann/ps v0.0.0-20150810152359-62de8c46ede0 // indirect github.com/olekukonko/cat v0.0.0-20250911104152-50322a0618f6 // indirect github.com/openfga/api/proto v0.0.0-20260319214821-f153694bfc20 // indirect + github.com/robfig/cron/v3 v3.0.1 // indirect github.com/sergi/go-diff v1.4.0 // indirect github.com/urfave/cli/v3 v3.8.0 // indirect go.mongodb.org/mongo-driver/v2 v2.5.0 // indirect + go.uber.org/atomic v1.9.0 // indirect go.uber.org/multierr v1.11.0 // indirect golang.org/x/exp v0.0.0-20260312153236-7ab1446f8b90 // indirect google.golang.org/genproto/googleapis/api v0.0.0-20260319201613-d00831a3d3e7 // indirect diff --git a/go.sum b/go.sum index 800316dfe..2a3b50775 100644 --- a/go.sum +++ b/go.sum @@ -59,6 +59,7 @@ github.com/containerd/errdefs/pkg v0.3.0 h1:9IKJ06FvyNlexW690DXuQNx2KA2cUJXx151X github.com/containerd/errdefs/pkg v0.3.0/go.mod h1:NJw6s9HwNuRhnjJhM7pylWwMyAkmCQvQ4GpJHEqRLVk= github.com/coreos/go-oidc v2.5.0+incompatible h1:6W0vGJR3Tu0r0PwfmjOrRZSlfxeEln8dsejt3ZWIvwo= github.com/coreos/go-oidc v2.5.0+incompatible/go.mod h1:CgnwVTmzoESiwO9qyAFEMiHoZ1nMCKZlZ9V6mm3/LKc= +github.com/creack/pty v1.1.9/go.mod h1:oKZEueFk5CKHvIhNR5MUki03XCEU+Q6VDXinZuGJ33E= github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc h1:U9qPSI2PIWSS1VwoXQT9A3Wy9MM3WgvqSxFWenqJduM= @@ -95,6 +96,8 @@ github.com/gkampitakis/go-diff v1.3.2 h1:Qyn0J9XJSDTgnsgHRdz9Zp24RaJeKMUHg2+PDZZ github.com/gkampitakis/go-diff v1.3.2/go.mod h1:LLgOrpqleQe26cte8s36HTWcTmMEur6OPYerdAAS9tk= github.com/gkampitakis/go-snaps v0.5.15 h1:amyJrvM1D33cPHwVrjo9jQxX8g/7E2wYdZ+01KS3zGE= github.com/gkampitakis/go-snaps v0.5.15/go.mod h1:HNpx/9GoKisdhw9AFOBT1N7DBs9DiHo/hGheFGBZ+mc= +github.com/go-co-op/gocron v1.37.0 h1:ZYDJGtQ4OMhTLKOKMIch+/CY70Brbb1dGdooLEhh7b0= +github.com/go-co-op/gocron v1.37.0/go.mod h1:3L/n6BkO7ABj+TrfSVXLRzsP26zmikL4ISkLQ0O8iNY= github.com/go-logr/logr v1.2.2/go.mod h1:jdQByPbusPIv2/zmleS9BjJVeZ6kBagPoEUsqbVz/1A= github.com/go-logr/logr v1.4.3 h1:CjnDlHq8ikf6E492q6eKboGOC0T8CDaOvkHCIg8idEI= github.com/go-logr/logr v1.4.3/go.mod h1:9T104GzyrTigFIr8wt5mBrctHMim0Nb2HLGrmQ40KvY= @@ -136,6 +139,7 @@ github.com/google/go-cmp v0.7.0/go.mod h1:pXiqmnSA92OHEEa9HXL2W4E7lf9JzCmGVUdgjX github.com/google/gofuzz v1.0.0/go.mod h1:dBl0BpW6vV/+mYPU4Po3pmUjxk6FQPldtuIdl/M65Eg= github.com/google/pprof v0.0.0-20260402051712-545e8a4df936 h1:EwtI+Al+DeppwYX2oXJCETMO23COyaKGP6fHVpkpWpg= github.com/google/pprof v0.0.0-20260402051712-545e8a4df936/go.mod h1:MxpfABSjhmINe3F1It9d+8exIHFvUqtLIRCdOGNXqiI= +github.com/google/uuid v1.4.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo= github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0= github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo= github.com/gorilla/websocket v1.5.0 h1:PPwGk2jz7EePpoHN/+ClbZu8SPxiqlu12wZP/3sWmnc= @@ -164,6 +168,8 @@ github.com/klauspost/compress v1.18.0/go.mod h1:2Pp+KzxcywXVXMr50+X0Q/Lsb43OQHYW github.com/klauspost/cpuid/v2 v2.3.0 h1:S4CRMLnYUhGeDFDqkGriYKdfoFlDnMtqTiI/sFzhA9Y= github.com/klauspost/cpuid/v2 v2.3.0/go.mod h1:hqwkgyIinND0mEev00jJYCxPNVRVXFQeu1XKlok6oO0= github.com/kr/pretty v0.1.0/go.mod h1:dAy3ld7l9f0ibDNOQOHHMYYIIbhfbHSm3C4ZsoJORNo= +github.com/kr/pretty v0.2.1/go.mod h1:ipq/a2n7PKx3OHsz4KJII5eveXtPO4qwEXGdVfWzfnI= +github.com/kr/pretty v0.3.0/go.mod h1:640gp4NfQd8pI5XOwp5fnNeVWj67G7CFk/SaSQn7NBk= github.com/kr/pretty v0.3.1 h1:flRD4NNwYAUpkphVc1HcthR4KEIFJ65n8Mw5qdRn3LE= github.com/kr/pretty v0.3.1/go.mod h1:hoEshYVHaxMs3cyo3Yncou5ZscifuDolrwPKZanG3xk= github.com/kr/pty v1.1.1/go.mod h1:pFQYn66WHrOpPYNljwOMqo10TkYh1fy3cYio2l3bCsQ= @@ -239,6 +245,7 @@ github.com/patrickmn/go-cache v2.1.0+incompatible h1:HRMgzkcYKYpi3C8ajMPV8OFXaaR github.com/patrickmn/go-cache v2.1.0+incompatible/go.mod h1:3Qf8kWWT7OJRJbdiICTKqZju1ZixQ/KpMGzzAfe6+WQ= github.com/pelletier/go-toml/v2 v2.2.4 h1:mye9XuhQ6gvn5h28+VilKrrPoQVanw5PMw/TB0t5Ec4= github.com/pelletier/go-toml/v2 v2.2.4/go.mod h1:2gIqNv+qfxSVS7cM2xJQKtLSTLUE9V8t9Stt+h56mCY= +github.com/pkg/diff v0.0.0-20210226163009-20ebb0f2a09e/go.mod h1:pJLUxLENpZxwdsKMEsNbx1VGcRFpLqf3715MtcvvzbA= github.com/pkg/errors v0.9.1 h1:FEBLx1zS214owpjy7qsBeixbURkuhQAwrK5UwLGTwt4= github.com/pkg/errors v0.9.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0= github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= @@ -258,6 +265,10 @@ github.com/quic-go/qpack v0.6.0 h1:g7W+BMYynC1LbYLSqRt8PBg5Tgwxn214ZZR34VIOjz8= github.com/quic-go/qpack v0.6.0/go.mod h1:lUpLKChi8njB4ty2bFLX2x4gzDqXwUpaO1DP9qMDZII= github.com/quic-go/quic-go v0.59.1 h1:0Gmua0HW1Tv7ANR7hUYwRyD0MG5OJfgvYSZasGZzBic= github.com/quic-go/quic-go v0.59.1/go.mod h1:upnsH4Ju1YkqpLXC305eW3yDZ4NfnNbmQRCMWS58IKU= +github.com/robfig/cron/v3 v3.0.1 h1:WdRxkvbJztn8LMz/QEvLN5sBU+xKpSqwwUO1Pjr4qDs= +github.com/robfig/cron/v3 v3.0.1/go.mod h1:eQICP3HwyT7UooqI/z+Ov+PtYAWygg1TEWWzGIFLtro= +github.com/rogpeppe/go-internal v1.6.1/go.mod h1:xXDCJY+GAPziupqXw64V24skbSoqbTEfhy4qGm1nDQc= +github.com/rogpeppe/go-internal v1.8.1/go.mod h1:JeRgkft04UBgHMgCIwADu4Pn6Mtm5d4nPKWu0nJ5d+o= github.com/rogpeppe/go-internal v1.14.1 h1:UQB4HGPB6osV0SQTLymcB4TgvyWu6ZyliaW0tI/otEQ= github.com/rogpeppe/go-internal v1.14.1/go.mod h1:MaRKkUm5W0goXpeCfT7UZI6fk/L7L7so1lCWt35ZSgc= github.com/samber/lo v1.53.0 h1:t975lj2py4kJPQ6haz1QMgtId2gtmfktACxIXArw3HM= @@ -285,6 +296,7 @@ github.com/stretchr/testify v1.4.0/go.mod h1:j7eGeouHqKxXV5pUuKE4zz7dFj8WfuZ+81P github.com/stretchr/testify v1.6.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg= github.com/stretchr/testify v1.7.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg= github.com/stretchr/testify v1.8.0/go.mod h1:yNjHg4UonilssWZ8iaSj1OCr/vHnekPRkoO+kdMU+MU= +github.com/stretchr/testify v1.8.2/go.mod h1:w2LPCIKwWwSfY2zedu0+kehJoqGctiVI29o6fzry7u4= github.com/stretchr/testify v1.8.4/go.mod h1:sz/lmYIOXD/1dqDmKjjqLyZ2RngseejIcXlSw2iwfAo= github.com/stretchr/testify v1.10.0/go.mod h1:r2ic/lqez/lEtzL7wO/rwa5dbSLXVDPFyf8C91i36aY= github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu7U= @@ -323,6 +335,8 @@ go.opentelemetry.io/otel/sdk/metric v1.39.0 h1:cXMVVFVgsIf2YL6QkRF4Urbr/aMInf+2W go.opentelemetry.io/otel/sdk/metric v1.39.0/go.mod h1:xq9HEVH7qeX69/JnwEfp6fVq5wosJsY1mt4lLfYdVew= go.opentelemetry.io/otel/trace v1.43.0 h1:BkNrHpup+4k4w+ZZ86CZoHHEkohws8AY+WTX09nk+3A= go.opentelemetry.io/otel/trace v1.43.0/go.mod h1:/QJhyVBUUswCphDVxq+8mld+AvhXZLhe+8WVFxiFff0= +go.uber.org/atomic v1.9.0 h1:ECmE8Bn/WFTYwEW/bpKD3M8VtR/zQVbavAoalC1PYyE= +go.uber.org/atomic v1.9.0/go.mod h1:fEN4uk6kAWBTFdckzkM89CLk9XfWZrxpCo0nPH17wJc= go.uber.org/goleak v1.3.0 h1:2K3zAYmnTNqV73imy9J1T3WC+gmCePx2hEGkimedGto= go.uber.org/goleak v1.3.0/go.mod h1:CoHD4mav9JJNrW/WLlf7HGZPjdw8EucARQHekz1X6bE= go.uber.org/mock v0.6.0 h1:hyF9dfmbgIX5EfOdasqLsWD6xqpNZlXblLB/Dbnwv3Y= @@ -373,9 +387,11 @@ google.golang.org/grpc v1.79.3/go.mod h1:KmT0Kjez+0dde/v2j9vzwoAScgEPx/Bw1CYChhH google.golang.org/protobuf v1.36.11 h1:fV6ZwhNocDyBLK0dj+fg8ektcVegBBuEolpbTQyBNVE= google.golang.org/protobuf v1.36.11/go.mod h1:HTf+CrKn2C3g5S8VImy6tdcUvCska2kB7j23XfzDpco= gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= +gopkg.in/check.v1 v1.0.0-20180628173108-788fd7840127/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= gopkg.in/check.v1 v1.0.0-20190902080502-41f04d3bba15/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c h1:Hei/4ADfdWqJk1ZMxUNpqntNwaWcugrBjAiHlqqRiVk= gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c/go.mod h1:JHkPIbrfpd72SG/EVd6muEfDQjcINNoR0C8j2r3qZ4Q= +gopkg.in/errgo.v2 v2.1.0/go.mod h1:hNsd1EY+bozCKY1Ytp96fpM3vjJbqLJn88ws8XvfDNI= gopkg.in/go-jose/go-jose.v2 v2.6.3 h1:nt80fvSDlhKWQgSWyHyy5CfmlQr+asih51R8PTWNKKs= gopkg.in/go-jose/go-jose.v2 v2.6.3/go.mod h1:zzZDPkNNw/c9IE7Z9jr11mBZQhKQTMzoEEIoEdZlFBI= gopkg.in/yaml.v2 v2.2.2/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI= diff --git a/internal/app/heureka.go b/internal/app/heureka.go index 27dab8cdd..a69a3ff60 100644 --- a/internal/app/heureka.go +++ b/internal/app/heureka.go @@ -289,6 +289,6 @@ func (h HeurekaApp) GetCache() cache.Cache { return h.cache } -func (h HeurekaApp) WaitPostMigrations() error { - return h.database.WaitPostMigrations() +func (h HeurekaApp) WaitPostMigrations() { + h.database.WaitPostMigrations() } diff --git a/internal/database/interface.go b/internal/database/interface.go index 0335a4171..56e9531a4 100644 --- a/internal/database/interface.go +++ b/internal/database/interface.go @@ -175,7 +175,7 @@ type Database interface { Autopatch(context.Context) (bool, error) - WaitPostMigrations() error + WaitPostMigrations() // Batch pre-load methods for GetVulnerabilities query optimization GetMaxSeverityByIssueIDs(context.Context, []int64) (map[int64]string, error) diff --git a/internal/database/mariadb/database.go b/internal/database/mariadb/database.go index 04bbaebec..5f3e8e525 100644 --- a/internal/database/mariadb/database.go +++ b/internal/database/mariadb/database.go @@ -28,7 +28,6 @@ type SqlDatabase struct { defaultIssuePriority int64 defaultRepositoryName string dbName string - postMigrationCtx postMigrationContext } func (s *SqlDatabase) CloseConnection() error { diff --git a/internal/database/mariadb/db.go b/internal/database/mariadb/db.go index 03603b1b9..65e8f0710 100644 --- a/internal/database/mariadb/db.go +++ b/internal/database/mariadb/db.go @@ -41,6 +41,7 @@ type Db interface { Query(query string, args ...any) (SqlRows, error) QueryContext(ctx context.Context, query string, args ...any) (SqlRows, error) QueryRow(query string, args ...any) *sql.Row + BeginTx(ctx context.Context, opts *sql.TxOptions) (*sql.Tx, error) } func NewDb(cfg util.Config) (Db, error) { diff --git a/internal/database/mariadb/migration.go b/internal/database/mariadb/migration.go index 8ea841a41..97075f63c 100644 --- a/internal/database/mariadb/migration.go +++ b/internal/database/mariadb/migration.go @@ -8,15 +8,12 @@ import ( "fmt" "io" "io/fs" - "strings" - "sync" "github.com/cloudoperators/heureka/internal/util" _ "github.com/go-sql-driver/mysql" "github.com/golang-migrate/migrate/v4" "github.com/golang-migrate/migrate/v4/database/mysql" "github.com/golang-migrate/migrate/v4/source/iofs" - "github.com/sirupsen/logrus" ) //go:embed migrations/*.sql @@ -100,145 +97,6 @@ func versionToString(v uint, dirty bool) string { return fmt.Sprintf("%d%s", v, dirtyStr) } -func RunMigrations(cfg util.Config) error { - err := runNewUpMigrations(cfg) - if err != nil { - return err - } - - err = runNewPostMigrationsAsync(cfg) - if err != nil { - return err - } - - err = enableScheduler(cfg) - if err != nil { - return fmt.Errorf("error while Enabling Scheduler Db: %w", err) - } - - err = registerEvents(cfg) - if err != nil { - return fmt.Errorf("error when registering events in scheduler: %w", err) - } - - return nil -} - -func enableScheduler(cfg util.Config) error { - db, err := GetSqlxRootConnection(cfg) - if err != nil { - logrus.WithError(err).Error(err) - return err - } - - defer func() { - if err := db.Close(); err != nil { - logrus.Warnf("failed to close DB connection: %s", err) - } - }() - - _, err = db.Exec("SET GLOBAL event_scheduler = ON;") - if err != nil { - logrus.WithError(err).Error(err) - return err - } - - return nil -} - -func registerEvents(cfg util.Config) error { - events := []string{ - "refresh_mvServiceIssueCounts", - "refresh_mvSingleComponentByServiceVulnerabilityCounts", - "refresh_mvAllComponentsByServiceVulnerabilityCounts", - "refresh_mvCountIssueRatingsUniqueService", - "refresh_mvCountIssueRatingsService", - "refresh_mvCountIssueRatingsServiceWithoutSupportGroup", - "refresh_mvCountIssueRatingsSupportGroup", - "refresh_mvCountIssueRatingsComponentVersion", - "refresh_mvCountIssueRatingsServiceId", - "refresh_mvCountIssueRatingsOther", - "refresh_mvVulnerabilityList", - "refresh_mvVulnerabilityService", - "refresh_mvComponentService", - } - - db, err := GetSqlxConnection(cfg) - if err != nil { - return fmt.Errorf("error while Creating Db: %w", err) - } - - defer func() { - if err := db.Close(); err != nil { - logrus.Warnf("failed to close DB connection: %s", err) - } - }() - - periodMinutes := cfg.DBMvCalcPeriodMinutes - if periodMinutes <= 0 { - periodMinutes = 200 - } - - logrus.Debugf("mv scheduling period set to %d minutes", periodMinutes) - - var query string - for _, e := range events { - query = fmt.Sprintf( - "%s DROP EVENT IF EXISTS %s; CREATE EVENT %s ON SCHEDULE EVERY %d MINUTE DO CALL %s_proc();", - query, - e, - e, - periodMinutes, - e, - ) - } - - _, err = db.Exec(query) - if err != nil { - logrus.WithError(err).Error(err) - return err - } - - return nil -} - -func runNewPostMigrationsAsync(cfg util.Config) error { - _, err := runNewPostMigrations(cfg) - if err != nil { - return err - } - - return nil -} - -func RunMigrationsSync(cfg util.Config) error { - err := runNewUpMigrations(cfg) - if err != nil { - return err - } - - err = runNewPostMigrationsSync(cfg) - if err != nil { - return err - } - - return nil -} - -func runNewPostMigrationsSync(cfg util.Config) error { - db, err := runNewPostMigrations(cfg) - if err != nil { - return err - } - - err = db.WaitPostMigrations() - if err != nil { - return fmt.Errorf("error while waiting for post migration procedures: %w", err) - } - - return nil -} - func (s *SqlDatabase) runUpMigrations() error { m, err := s.openMigration() if err != nil { @@ -257,21 +115,7 @@ func (s *SqlDatabase) runUpMigrations() error { return nil } -func runNewPostMigrations(cfg util.Config) (*SqlDatabase, error) { - db, err := NewSqlDatabase(cfg) - if err != nil { - return nil, fmt.Errorf("error while Creating Db: %w", err) - } - - err = db.runPostMigrations() - if err != nil { - return nil, fmt.Errorf("error while starting Post Migration procedures: %w", err) - } - - return db, nil -} - -func runNewUpMigrations(cfg util.Config) error { +func RunMigrations(cfg util.Config) error { db, err := NewSqlDatabase(cfg) if err != nil { return fmt.Errorf("error while Creating Db: %w", err) @@ -285,157 +129,6 @@ func runNewUpMigrations(cfg util.Config) error { return nil } -func (s *SqlDatabase) tableExists(table string) (bool, error) { - var count int - - err := s.db.Get(&count, ` - SELECT COUNT(*) - FROM information_schema.tables - WHERE table_schema = DATABASE() - AND table_name = ? - `, table) - if err != nil { - return false, err - } - - return count > 0, nil -} - -const PostMigrationProcedureRegistryTable = "post_migration_procedure_registry" - -func (s *SqlDatabase) getPostMigrationProcedures() ([]string, error) { - var procs []string - - exists, err := s.tableExists(PostMigrationProcedureRegistryTable) - if err != nil { - return procs, fmt.Errorf("could not check if table exists: %w", err) - } else if !exists { - return procs, nil - } - - err = s.db.Select( - &procs, - fmt.Sprintf("SELECT name FROM %s", PostMigrationProcedureRegistryTable), - ) - if err != nil { - return nil, err - } - - return procs, nil -} - -type postMigrationContext struct { - wg sync.WaitGroup - mu sync.Mutex - errs []string -} - -func (pmc *postMigrationContext) appendErrorMessage(msg string) { - pmc.mu.Lock() - pmc.errs = append(pmc.errs, msg) - pmc.mu.Unlock() -} - -func (pmc *postMigrationContext) hasError() bool { - return len(pmc.errs) > 0 -} - -func (pmc *postMigrationContext) getError() error { - return fmt.Errorf("error when exeute joined callers: [%s]", strings.Join(pmc.errs, "; ")) -} - -func (s *SqlDatabase) runPostMigrations() error { - procs, err := s.getPostMigrationProcedures() - if err != nil { - return fmt.Errorf("failed to get post migration procedures: %w", err) - } - - if err := s.checkProceduresExist(procs); err != nil { - return err - } - - s.runPostMigrationProcessInBackground(procs) - - return nil -} - -func (s *SqlDatabase) checkProceduresExist(procs []string) error { - exists, err := s.proceduresExist(procs) - if err != nil { - return fmt.Errorf("could not check procedures exist: %w", err) - } else if !exists { - return fmt.Errorf("some procedures [%s] do not exist", strings.Join(procs, ", ")) - } - - return nil -} - -func (s *SqlDatabase) proceduresExist(procedures []string) (bool, error) { - if len(procedures) == 0 { - return true, nil - } - - placeholders := make([]string, len(procedures)) - args := make([]any, len(procedures)) - - for i, p := range procedures { - placeholders[i] = "?" - args[i] = p - } - - query := fmt.Sprintf(` - SELECT COUNT(*) - FROM information_schema.routines - WHERE routine_schema = DATABASE() - AND routine_type = 'PROCEDURE' - AND routine_name IN (%s); - `, strings.Join(placeholders, ",")) - - var count int - - err := s.db.QueryRow(query, args...).Scan(&count) - if err != nil { - return false, fmt.Errorf("could not check if procedures exist: %w", err) - } - - if count == len(procedures) { - return true, nil - } - - return false, nil -} - -func (s *SqlDatabase) runPostMigrationProcessInBackground(procs []string) { - s.runPostMigrationProceduresInBackground(procs) - s.runPostMigrationCleanupRoutineInBackground() -} - -func (s *SqlDatabase) runPostMigrationProceduresInBackground(procs []string) { - for _, p := range procs { - s.postMigrationCtx.wg.Go(func() { - if _, err := s.db.Exec(fmt.Sprintf("CALL %s();", p)); err != nil { - s.postMigrationCtx.appendErrorMessage(fmt.Sprintf("%s: %v", p, err)) - } - }) - } -} - -func (s *SqlDatabase) runPostMigrationCleanupRoutineInBackground() { - go func() { - if err := s.WaitPostMigrations(); err != nil { - logrus.WithError(err).Error(err) - } - - _ = s.CloseConnection() - }() -} - -func (s *SqlDatabase) WaitPostMigrations() error { - s.postMigrationCtx.wg.Wait() - - if s.postMigrationCtx.hasError() { - return s.postMigrationCtx.getError() - } - - return nil +func (s *SqlDatabase) WaitPostMigrations() { + WaitMVEForFirstRun() } diff --git a/internal/database/mariadb/migrations/20260701112108_app_scheduler.down.sql b/internal/database/mariadb/migrations/20260701112108_app_scheduler.down.sql new file mode 100644 index 000000000..5acea171c --- /dev/null +++ b/internal/database/mariadb/migrations/20260701112108_app_scheduler.down.sql @@ -0,0 +1,48 @@ +-- SPDX-FileCopyrightText: 2026 SAP SE or an SAP affiliate company and Greenhouse contributors +-- SPDX-License-Identifier: Apache-2.0 + +CREATE TABLE IF NOT EXISTS post_migration_procedure_registry (name VARCHAR(128) PRIMARY KEY); + +DROP PROCEDURE IF EXISTS call_registered_post_migration_procedures; + +CREATE PROCEDURE call_registered_post_migration_procedures() +BEGIN + DECLARE done INT DEFAULT 0; + DECLARE pname VARCHAR(128); + DECLARE cur CURSOR FOR SELECT name FROM post_migration_procedure_registry; + DECLARE CONTINUE HANDLER FOR NOT FOUND SET done = 1; + + OPEN cur; + + read_loop: LOOP + FETCH cur INTO pname; + IF done THEN + LEAVE read_loop; + END IF; + + SET @sql = CONCAT('CALL ', pname, '();'); + PREPARE stmt FROM @sql; + EXECUTE stmt; + DEALLOCATE PREPARE stmt; + END LOOP; + + CLOSE cur; +END; + +DROP PROCEDURE IF EXISTS add_post_migration_procedure; + +CREATE PROCEDURE add_post_migration_procedure(IN p_name VARCHAR(128)) +BEGIN + INSERT IGNORE INTO post_migration_procedure_registry (name) + VALUES (p_name); +END +; + +DROP PROCEDURE IF EXISTS remove_post_migration_procedure; + +CREATE PROCEDURE remove_post_migration_procedure(IN p_name VARCHAR(128)) +BEGIN + DELETE FROM post_migration_procedure_registry + WHERE name = p_name; +END; + diff --git a/internal/database/mariadb/migrations/20260701112108_app_scheduler.up.sql b/internal/database/mariadb/migrations/20260701112108_app_scheduler.up.sql new file mode 100644 index 000000000..9b1a17165 --- /dev/null +++ b/internal/database/mariadb/migrations/20260701112108_app_scheduler.up.sql @@ -0,0 +1,7 @@ +-- SPDX-FileCopyrightText: 2026 SAP SE or an SAP affiliate company and Greenhouse contributors +-- SPDX-License-Identifier: Apache-2.0 + +DROP TABLE IF EXISTS post_migration_procedure_registry; +DROP PROCEDURE IF EXISTS call_registered_post_migration_procedures; +DROP PROCEDURE IF EXISTS add_post_migration_procedure; +DROP PROCEDURE IF EXISTS remove_post_migration_procedure; diff --git a/internal/database/mariadb/mve.go b/internal/database/mariadb/mve.go new file mode 100644 index 000000000..0ebb9330e --- /dev/null +++ b/internal/database/mariadb/mve.go @@ -0,0 +1,218 @@ +// SPDX-FileCopyrightText: 2026 SAP SE or an SAP affiliate company and Greenhouse contributors +// SPDX-License-Identifier: Apache-2.0 + +package mariadb + +import ( + "fmt" + "strings" + "sync" + "time" + + "github.com/cloudoperators/heureka/internal/util" + "github.com/go-co-op/gocron" + "github.com/jmoiron/sqlx" + "github.com/sirupsen/logrus" +) + +var MVSqlProcedures = []string{ + //"refresh_mvServiceIssueCounts_proc", // 1 + "refresh_mvSingleComponentByServiceVulnerabilityCounts_proc", + "refresh_mvAllComponentsByServiceVulnerabilityCounts_proc", + //"refresh_mvCountIssueRatingsUniqueService_proc", // 3 + "refresh_mvCountIssueRatingsService_proc", + "refresh_mvCountIssueRatingsServiceWithoutSupportGroup_proc", + "refresh_mvCountIssueRatingsSupportGroup_proc", + "refresh_mvCountIssueRatingsComponentVersion_proc", + //"refresh_mvCountIssueRatingsServiceId_proc", // 2 + "refresh_mvCountIssueRatingsOther_proc", + "refresh_mvVulnerabilityList_proc", + "refresh_mvVulnerabilityService_proc", + "refresh_mvComponentService_proc", +} + +func TriggerMVE(cfg util.Config) error { + db, err := GetSqlxConnection(cfg) + if err != nil { + return fmt.Errorf("error while Creating Db: %w", err) + } + + if err := checkProceduresExist(db, MVSqlProcedures); err != nil { + return err + } + + return runInBackground(db, MVSqlProcedures).Wait() +} + +func StartMVEScheduler(cfg util.Config) { + mve := getMVE() + + periodMinutes := cfg.DBMvCalcPeriodMinutes + if periodMinutes <= 0 { + periodMinutes = 200 + } + + logrus.Debugf("MVE scheduling period set to %d minutes", periodMinutes) + + _, err := mve.scheduler.Every(periodMinutes).Minutes().SingletonMode().Do(func() { + err := TriggerMVE(cfg) + if err != nil { + logrus.WithError(err).Error("MVE Trigger error") + } + + mve.once.Do(func() { + close(mve.firstRunDone) + }) + }) + if err != nil { + logrus.WithError(err).Error("MVE Do() error") + mve.once.Do(func() { + close(mve.firstRunDone) + }) + } + + mve.scheduler.StartAsync() +} + +func StopMVEScheduler() { + mve := getMVE() + mve.Shutdown() +} + +func WaitMVEForFirstRun() { + mve := getMVE() + <-mve.firstRunDone +} + +////////// Internals + +var mvEngine *MvEngine + +type MvEngine struct { + scheduler *gocron.Scheduler + firstRunDone chan struct{} + once sync.Once +} + +func NewMvEngine() *MvEngine { + return &MvEngine{ + scheduler: gocron.NewScheduler(time.UTC), + firstRunDone: make(chan struct{}), + } +} + +func getMVE() *MvEngine { + if mvEngine == nil { + mvEngine = NewMvEngine() + } + + return mvEngine +} + +func (mve *MvEngine) Shutdown() { + mve.scheduler.Clear() + // This is not advisory as it may hang for a long time: + // mve.scheduler.Stop() +} + +type mveCtx struct { + wg sync.WaitGroup + mu sync.Mutex + errs []string +} + +func (mc *mveCtx) appendErrorMessage(msg string) { + mc.mu.Lock() + mc.errs = append(mc.errs, msg) + mc.mu.Unlock() +} + +func (mc *mveCtx) hasError() bool { + return len(mc.errs) > 0 +} + +func (mc *mveCtx) getError() error { + return fmt.Errorf("error when execute joined callers: [%s]", strings.Join(mc.errs, "; ")) +} + +func (mc *mveCtx) Wait() error { + mc.wg.Wait() + + if mc.hasError() { + return mc.getError() + } + + return nil +} + +func checkProceduresExist(db *sqlx.DB, procs []string) error { + exists, err := proceduresExist(db, procs) + if err != nil { + return fmt.Errorf("could not check procedures exist: %w", err) + } else if !exists { + return fmt.Errorf("some procedures [%s] do not exist", strings.Join(procs, ", ")) + } + + return nil +} + +func proceduresExist(db *sqlx.DB, procedures []string) (bool, error) { + if len(procedures) == 0 { + return true, nil + } + + placeholders := make([]string, len(procedures)) + args := make([]any, len(procedures)) + + for i, p := range procedures { + placeholders[i] = "?" + args[i] = p + } + + query := fmt.Sprintf(` + SELECT COUNT(*) + FROM information_schema.routines + WHERE routine_schema = DATABASE() + AND routine_type = 'PROCEDURE' + AND routine_name IN (%s); + `, strings.Join(placeholders, ",")) + + var count int + + err := db.QueryRow(query, args...).Scan(&count) + if err != nil { + return false, fmt.Errorf("could not check if procedures exist: %w", err) + } + + if count == len(procedures) { + return true, nil + } + + return false, nil +} + +func runInBackground(db *sqlx.DB, procs []string) *mveCtx { + mc := &mveCtx{} + + for _, p := range procs { + mc.wg.Go(func() { + if _, err := db.Exec(fmt.Sprintf("CALL %s();", p)); err != nil { + mc.appendErrorMessage(fmt.Sprintf("%s: %v", p, err)) + } + }) + } + + runCleanupRoutineInBackground(db, mc) + + return mc +} + +func runCleanupRoutineInBackground(db *sqlx.DB, mc *mveCtx) { + go func() { + if err := mc.Wait(); err != nil { + logrus.WithError(err).Error("MVE background execution failed") + } + + _ = db.Close() + }() +} diff --git a/internal/database/mariadb/mvproc.go b/internal/database/mariadb/mvproc.go new file mode 100644 index 000000000..287b8687d --- /dev/null +++ b/internal/database/mariadb/mvproc.go @@ -0,0 +1,358 @@ +// SPDX-FileCopyrightText: 2026 SAP SE or an SAP affiliate company and Greenhouse contributors +// SPDX-License-Identifier: Apache-2.0 + +package mariadb + +import ( + "context" + + sq "github.com/Masterminds/squirrel" +) + +type MVProcedure func(ctx context.Context, db Db) error + +var MVProcedures []MVProcedure = []MVProcedure{ + RefreshMVServiceIssueCounts, +} + +func RefreshMVServiceIssueCounts(ctx context.Context, db Db) error { + tx, err := db.BeginTx(ctx, nil) + if err != nil { + return err + } + + defer func() { + if err != nil { + _ = tx.Rollback() + } + }() + + // Match the stored procedure's session isolation level. + if _, err = tx.ExecContext(ctx, ` + SET SESSION TRANSACTION ISOLATION LEVEL READ COMMITTED`); err != nil { + return err + } + + // Clean up any leftover tables. + if _, err = tx.ExecContext(ctx, + `DROP TABLE IF EXISTS mvServiceIssueCounts_tmp`); err != nil { + return err + } + + if _, err = tx.ExecContext(ctx, + `DROP TABLE IF EXISTS mvServiceIssueCounts_old`); err != nil { + return err + } + + // Create the temporary table. + if _, err = tx.ExecContext(ctx, ` + CREATE TABLE mvServiceIssueCounts_tmp + LIKE mvServiceIssueCounts`); err != nil { + return err + } + + selectBuilder := sq. + Select( + "S.service_id", + `COUNT(DISTINCT CASE + WHEN IV.issuevariant_rating = 'Critical' + THEN CONCAT(CI.componentinstance_component_version_id, ',', I.issue_id) + END)`, + `COUNT(DISTINCT CASE + WHEN IV.issuevariant_rating = 'High' + THEN CONCAT(CI.componentinstance_component_version_id, ',', I.issue_id) + END)`, + `COUNT(DISTINCT CASE + WHEN IV.issuevariant_rating = 'Medium' + THEN CONCAT(CI.componentinstance_component_version_id, ',', I.issue_id) + END)`, + `COUNT(DISTINCT CASE + WHEN IV.issuevariant_rating = 'Low' + THEN CONCAT(CI.componentinstance_component_version_id, ',', I.issue_id) + END)`, + `COUNT(DISTINCT CASE + WHEN IV.issuevariant_rating = 'None' + THEN CONCAT(CI.componentinstance_component_version_id, ',', I.issue_id) + END)`, + ). + From("Service S"). + LeftJoin(`ComponentInstance CI + ON S.service_id = CI.componentinstance_service_id + AND CI.componentinstance_deleted_at IS NULL`). + LeftJoin(`IssueMatch IM + ON CI.componentinstance_id = IM.issuematch_component_instance_id + AND IM.issuematch_deleted_at IS NULL`). + LeftJoin(`Issue I + ON IM.issuematch_issue_id = I.issue_id + AND I.issue_deleted_at IS NULL`). + LeftJoin("IssueVariant IV ON IV.issuevariant_issue_id = I.issue_id"). + Where("S.service_deleted_at IS NULL"). + Where(` + NOT EXISTS ( + SELECT 1 + FROM Remediation R + WHERE R.remediation_service_id = S.service_id + AND R.remediation_issue_id = I.issue_id + AND R.remediation_deleted_at IS NULL + AND ( + R.remediation_expiration_date IS NULL + OR R.remediation_expiration_date >= CURDATE() + ) + )`). + GroupBy("S.service_id") + + insertBuilder := sq. + Insert("mvServiceIssueCounts_tmp"). + Columns( + "service_id", + "critical_count", + "high_count", + "medium_count", + "low_count", + "none_count", + ). + Select(selectBuilder) + + insertSQL, args, err := insertBuilder.ToSql() + if err != nil { + return err + } + + if _, err = tx.ExecContext(ctx, insertSQL, args...); err != nil { + return err + } + + // Atomically swap the tables. + if _, err = tx.ExecContext(ctx, ` + RENAME TABLE + mvServiceIssueCounts TO mvServiceIssueCounts_old, + mvServiceIssueCounts_tmp TO mvServiceIssueCounts`); err != nil { + return err + } + + // Remove the previous table. + if _, err = tx.ExecContext(ctx, + `DROP TABLE IF EXISTS mvServiceIssueCounts_old`); err != nil { + return err + } + + return tx.Commit() +} + +func RefreshMVCountIssueRatingsServiceId(ctx context.Context, db Db) error { + tx, err := db.BeginTx(ctx, nil) + if err != nil { + return err + } + + defer func() { + if err != nil { + _ = tx.Rollback() + } + }() + + // Match the stored procedure's session isolation level. + if _, err = tx.ExecContext(ctx, ` + SET SESSION TRANSACTION ISOLATION LEVEL READ COMMITTED`); err != nil { + return err + } + + // Mark all current rows as inactive. + if _, err = tx.ExecContext(ctx, ` + UPDATE mvCountIssueRatingsServiceId + SET is_active = 0 + WHERE is_active = 1`); err != nil { + return err + } + + selectBuilder := sq. + Select( + "S.service_id", + "S.service_ccrn", + `COUNT(DISTINCT CASE + WHEN IV.issuevariant_rating = 'Critical' + THEN CONCAT(CI.componentinstance_component_version_id, ',', I.issue_id) + END)`, + `COUNT(DISTINCT CASE + WHEN IV.issuevariant_rating = 'High' + THEN CONCAT(CI.componentinstance_component_version_id, ',', I.issue_id) + END)`, + `COUNT(DISTINCT CASE + WHEN IV.issuevariant_rating = 'Medium' + THEN CONCAT(CI.componentinstance_component_version_id, ',', I.issue_id) + END)`, + `COUNT(DISTINCT CASE + WHEN IV.issuevariant_rating = 'Low' + THEN CONCAT(CI.componentinstance_component_version_id, ',', I.issue_id) + END)`, + `COUNT(DISTINCT CASE + WHEN IV.issuevariant_rating = 'None' + THEN CONCAT(CI.componentinstance_component_version_id, ',', I.issue_id) + END)`, + "1", + ). + From("Service S"). + LeftJoin(`ComponentInstance CI + ON S.service_id = CI.componentinstance_service_id + AND CI.componentinstance_deleted_at IS NULL`). + LeftJoin(`IssueMatch IM + ON CI.componentinstance_id = IM.issuematch_component_instance_id + AND IM.issuematch_deleted_at IS NULL`). + LeftJoin(`Issue I + ON IM.issuematch_issue_id = I.issue_id + AND I.issue_deleted_at IS NULL`). + LeftJoin("IssueVariant IV ON IV.issuevariant_issue_id = I.issue_id"). + Where("S.service_deleted_at IS NULL"). + Where("IM.issuematch_id IS NOT NULL"). + Where(sq.Expr(` + NOT EXISTS ( + SELECT 1 + FROM Remediation R + WHERE R.remediation_service_id = S.service_id + AND R.remediation_issue_id = I.issue_id + AND R.remediation_deleted_at IS NULL + AND ( + R.remediation_expiration_date IS NULL + OR R.remediation_expiration_date >= CURDATE() + ) + )`)). + GroupBy("S.service_id") + + insertBuilder := sq. + Insert("mvCountIssueRatingsServiceId"). + Columns( + "service_id", + "service_ccrn", + "critical_count", + "high_count", + "medium_count", + "low_count", + "none_count", + "is_active", + ). + Select(selectBuilder) + + insertSQL, args, err := insertBuilder.ToSql() + if err != nil { + return err + } + + insertSQL += ` + ON DUPLICATE KEY UPDATE + service_ccrn = VALUES(service_ccrn), + critical_count = VALUES(critical_count), + high_count = VALUES(high_count), + medium_count = VALUES(medium_count), + low_count = VALUES(low_count), + none_count = VALUES(none_count), + is_active = 1` + + if _, err = tx.ExecContext(ctx, insertSQL, args...); err != nil { + return err + } + + // Remove rows that were not refreshed. + if _, err = tx.ExecContext(ctx, ` + DELETE FROM mvCountIssueRatingsServiceId + WHERE is_active = 0`); err != nil { + return err + } + + return tx.Commit() +} + +func RefreshMVCountIssueRatingsUniqueService(ctx context.Context, db Db) error { + tx, err := db.BeginTx(ctx, nil) + if err != nil { + return err + } + + defer func() { + if err != nil { + _ = tx.Rollback() + } + }() + + // Match the stored procedure's session isolation level. + if _, err = tx.ExecContext(ctx, ` + SET SESSION TRANSACTION ISOLATION LEVEL READ COMMITTED`); err != nil { + return err + } + + // Mark all current rows as inactive. + if _, err = tx.ExecContext(ctx, ` + UPDATE mvCountIssueRatingsUniqueService + SET is_active = 0 + WHERE is_active = 1`); err != nil { + return err + } + + selectBuilder := sq. + Select( + "1", + `COUNT(DISTINCT CASE + WHEN IV.issuevariant_rating = 'Critical' + THEN IV.issuevariant_issue_id + END)`, + `COUNT(DISTINCT CASE + WHEN IV.issuevariant_rating = 'High' + THEN IV.issuevariant_issue_id + END)`, + `COUNT(DISTINCT CASE + WHEN IV.issuevariant_rating = 'Medium' + THEN IV.issuevariant_issue_id + END)`, + `COUNT(DISTINCT CASE + WHEN IV.issuevariant_rating = 'Low' + THEN IV.issuevariant_issue_id + END)`, + `COUNT(DISTINCT CASE + WHEN IV.issuevariant_rating = 'None' + THEN IV.issuevariant_issue_id + END)`, + "1", + ). + From("Issue I"). + LeftJoin("IssueVariant IV ON IV.issuevariant_issue_id = I.issue_id"). + Where("I.issue_deleted_at IS NULL") + + insertBuilder := sq. + Insert("mvCountIssueRatingsUniqueService"). + Columns( + "id", + "critical_count", + "high_count", + "medium_count", + "low_count", + "none_count", + "is_active", + ). + Select(selectBuilder) + + insertSQL, args, err := insertBuilder.ToSql() + if err != nil { + return err + } + + insertSQL += ` + ON DUPLICATE KEY UPDATE + critical_count = VALUES(critical_count), + high_count = VALUES(high_count), + medium_count = VALUES(medium_count), + low_count = VALUES(low_count), + none_count = VALUES(none_count), + is_active = 1` + + if _, err = tx.ExecContext(ctx, insertSQL, args...); err != nil { + return err + } + + // Remove rows that were not refreshed. + if _, err = tx.ExecContext(ctx, ` + DELETE FROM mvCountIssueRatingsUniqueService + WHERE is_active = 0`); err != nil { + return err + } + + return tx.Commit() +} diff --git a/internal/database/mariadb/quiet_db.go b/internal/database/mariadb/quiet_db.go index a26626b9e..9a73d6ff1 100644 --- a/internal/database/mariadb/quiet_db.go +++ b/internal/database/mariadb/quiet_db.go @@ -57,3 +57,7 @@ func (qdb *QuietDb) QueryContext(ctx context.Context, query string, args ...any) func (qdb *QuietDb) QueryRow(query string, args ...any) *sql.Row { return qdb.db.QueryRow(query, args...) } + +func (qdb *QuietDb) BeginTx(ctx context.Context, opts *sql.TxOptions) (*sql.Tx, error) { + return qdb.db.BeginTx(ctx, opts) +} diff --git a/internal/database/mariadb/test/database_manager.go b/internal/database/mariadb/test/database_manager.go index 389c62702..8bc29d491 100644 --- a/internal/database/mariadb/test/database_manager.go +++ b/internal/database/mariadb/test/database_manager.go @@ -61,7 +61,7 @@ func (dbm *DatabaseManager) loadDBClient() { } func (dbm *DatabaseManager) Setup() error { - err := mariadb.RunMigrationsSync(dbm.rootUserConfig()) + err := dbm.runMigrationsAndTriggerMvUpdates(dbm.rootUserConfig()) if err != nil { ginkgo.GinkgoLogr.WithCallDepth(5).Error(err, "Failure while setting migrations schema") return err @@ -70,6 +70,19 @@ func (dbm *DatabaseManager) Setup() error { return nil } +func (dbm *DatabaseManager) runMigrationsAndTriggerMvUpdates(cfg util.Config) error { + if err := mariadb.RunMigrations(cfg); err != nil { + ginkgo.GinkgoLogr.WithCallDepth(5).Error(err, "Failure while setting migrations schema") + return err + } + + if err := mariadb.TriggerMVE(cfg); err != nil { + return err + } + + return nil +} + func (dbm *DatabaseManager) ResetSchema(dbName string) error { // load the client dbm.loadDBClientIfNeeded() @@ -87,7 +100,7 @@ func (dbm *DatabaseManager) ResetSchema(dbName string) error { return err } - err = mariadb.RunMigrationsSync(dbm.DbConfig()) + err = dbm.runMigrationsAndTriggerMvUpdates(dbm.DbConfig()) if err != nil { ginkgo.GinkgoLogr.WithCallDepth(5).Error(err, "Failure while resetting migrations schema") return err @@ -179,7 +192,7 @@ func (dbm *DatabaseManager) NewTestSchema() *mariadb.SqlDatabase { ginkgo.GinkgoLogr.WithCallDepth(5).Error(err, "failed to prepare DB schema") } - err = mariadb.RunMigrationsSync(dbm.DbConfig()) + err = dbm.runMigrationsAndTriggerMvUpdates(dbm.DbConfig()) if err != nil { ginkgo.GinkgoLogr.WithCallDepth(5).Error(err, "failure while resetting migrations schema") } diff --git a/internal/database/mariadb/test/fixture.go b/internal/database/mariadb/test/fixture.go index 4cf7f33fe..9fc17805a 100644 --- a/internal/database/mariadb/test/fixture.go +++ b/internal/database/mariadb/test/fixture.go @@ -4,6 +4,7 @@ package test import ( + "context" "database/sql" "fmt" "math/rand" @@ -17,7 +18,6 @@ import ( "github.com/brianvoe/gofakeit/v7" "github.com/cloudoperators/heureka/internal/database/mariadb" "github.com/cloudoperators/heureka/internal/util" - "github.com/jmoiron/sqlx" "github.com/samber/lo" "github.com/sirupsen/logrus" ) @@ -365,11 +365,11 @@ func (s *SeedCollection) FindMatchingSupportGroupAndIssueMatch() (mariadb.Suppor } type DatabaseSeeder struct { - db *sqlx.DB + db mariadb.Db } func NewDatabaseSeeder(cfg util.Config) (*DatabaseSeeder, error) { - db, err := mariadb.Connect(cfg) + db, err := mariadb.NewDb(cfg) if err != nil { return nil, err } @@ -2097,25 +2097,37 @@ func (s *DatabaseSeeder) Clear() error { } func (s *DatabaseSeeder) RefreshServiceIssueCounters() error { - _, err := s.db.Exec(` - CALL refresh_mvServiceIssueCounts_proc(); - `) + // 1 + return mariadb.RefreshMVServiceIssueCounts(context.Background(), s.db) +} - return err +func (s *DatabaseSeeder) RefreshCountIssueRatingsServiceId() error { + // 2 + return mariadb.RefreshMVCountIssueRatingsServiceId(context.Background(), s.db) +} + +func (s *DatabaseSeeder) RefreshCountIssueRatingsUniqueService() error { + // 3 + return mariadb.RefreshMVCountIssueRatingsUniqueService(context.Background(), s.db) } func (s *DatabaseSeeder) RefreshCountIssueRatings() error { _, err := s.db.Exec(` - CALL refresh_mvCountIssueRatingsUniqueService_proc(); CALL refresh_mvCountIssueRatingsService_proc(); CALL refresh_mvCountIssueRatingsServiceWithoutSupportGroup_proc(); CALL refresh_mvCountIssueRatingsSupportGroup_proc(); CALL refresh_mvCountIssueRatingsComponentVersion_proc(); - CALL refresh_mvCountIssueRatingsServiceId_proc(); CALL refresh_mvCountIssueRatingsOther_proc(); `) + if err != nil { + return err + } - return err + if err = s.RefreshCountIssueRatingsUniqueService(); err != nil { + return err + } + + return s.RefreshCountIssueRatingsServiceId() } func (s *DatabaseSeeder) RefreshComponentVulnerabilityCounts() error { diff --git a/internal/database/mariadb/trace_db.go b/internal/database/mariadb/trace_db.go index 3fb6ef4c0..ec697d24b 100644 --- a/internal/database/mariadb/trace_db.go +++ b/internal/database/mariadb/trace_db.go @@ -177,3 +177,8 @@ func (tdb *TraceDb) QueryContext(ctx context.Context, query string, args ...any) defer NewTrace("QueryContext", query).exitTrace() return tdb.db.QueryContext(ctx, query, args...) } + +func (tdb *TraceDb) BeginTx(ctx context.Context, opts *sql.TxOptions) (*sql.Tx, error) { + defer NewTrace("BeginTx", "").exitTrace() + return tdb.db.BeginTx(ctx, opts) +} diff --git a/internal/e2e/common/server.go b/internal/e2e/common/server.go index 3b8d456c8..7d417b9bb 100644 --- a/internal/e2e/common/server.go +++ b/internal/e2e/common/server.go @@ -6,14 +6,12 @@ package e2e_common import ( "github.com/cloudoperators/heureka/internal/server" "github.com/cloudoperators/heureka/internal/util" - . "github.com/onsi/gomega" ) func NewRunningServer(cfg util.Config) *server.Server { s := server.NewServer(cfg) s.NonBlockingStart() - err := s.GetApp().WaitPostMigrations() - Expect(err).Should(BeNil()) + s.GetApp().WaitPostMigrations() return s } diff --git a/internal/e2e/db_migration_test.go b/internal/e2e/db_migration_test.go index a731d8c4f..11678864d 100644 --- a/internal/e2e/db_migration_test.go +++ b/internal/e2e/db_migration_test.go @@ -34,9 +34,6 @@ var migrationABFiles embed.FS //go:embed migrations/*_mvTestTable.*.sql var migrationMvTestTableMigrationFiles embed.FS -//go:embed migrations/*_add_post_migration.*.sql -var migrationAddPostMigrationMigrationFiles embed.FS - // Merge multiple fs.FS into one MapFS with "migrations/" prefix func MergeToMapFS(sources ...fs.FS) (fstest.MapFS, error) { merged := fstest.MapFS{} @@ -71,6 +68,8 @@ func MergeToMapFS(sources ...fs.FS) (fstest.MapFS, error) { } func setDbTestMigration(migrationFiles fs.FS) { + setMvSqlProceduresInMVE([]string{}) + mariadb.MigrationFs = migrationFiles } @@ -82,20 +81,16 @@ func setDbABMigration() { setDbTestMigration(&migrationABFiles) } -func addDbMvTestTableMigration() { - mapFS, err := MergeToMapFS(mariadb.MigrationFs, migrationMvTestTableMigrationFiles) - Expect(err).To(BeNil()) - setDbTestMigration(&mapFS) +func setDbMvTestTableMigration() { + setDbTestMigration(&migrationMvTestTableMigrationFiles) } -func addDbMvTestTableAndAddPostMigrationMigration() { - mapFS, err := MergeToMapFS( - mariadb.MigrationFs, - migrationMvTestTableMigrationFiles, - migrationAddPostMigrationMigrationFiles, - ) - Expect(err).To(BeNil()) - setDbTestMigration(&mapFS) +func setMvSqlProceduresInMVE(procs []string) { + mariadb.MVSqlProcedures = procs +} + +func setMvTestTableInMVE() { + setMvSqlProceduresInMVE([]string{"refresh_mvTestData_proc"}) } func extractVersion(filename string) string { @@ -224,11 +219,10 @@ func (dbmt *DbMigrationTest) dbVersionIsB() { func (dbmt *DbMigrationTest) dbVersionIsMvTestTable() { dbmt.dbVersionShouldBeZero() dbmt.dbShouldNotContainPostMigrationProcedureRefreshTable() - addDbMvTestTableMigration() + setDbMvTestTableMigration() dbmt.createHeurekaServer() dbmt.dbShouldNotContainPostMigrationProcedureData() dbmt.dbVersionShouldBeMvTestTable() - addDbMvTestTableAndAddPostMigrationMigration() } func (dbmt *DbMigrationTest) createHeurekaServer() { @@ -367,6 +361,7 @@ var _ = Describe( func() { It("executes post migration procedure after successful migration", func() { migrationTest.dbVersionIsMvTestTable() + setMvTestTableInMVE() migrationTest.createHeurekaServer() waitForPostMigration() migrationTest.dbShouldContainPostMigrationProcedureData() diff --git a/internal/e2e/migrations/70250915164215_add_post_migration.down.sql b/internal/e2e/migrations/70250915164215_add_post_migration.down.sql deleted file mode 100644 index 7145581c4..000000000 --- a/internal/e2e/migrations/70250915164215_add_post_migration.down.sql +++ /dev/null @@ -1,3 +0,0 @@ --- SPDX-FileCopyrightText: 2025 SAP SE or an SAP affiliate company and Greenhouse contributors --- SPDX-License-Identifier: Apache-2.0 - diff --git a/internal/e2e/migrations/70250915164215_add_post_migration.up.sql b/internal/e2e/migrations/70250915164215_add_post_migration.up.sql deleted file mode 100644 index b33543fb9..000000000 --- a/internal/e2e/migrations/70250915164215_add_post_migration.up.sql +++ /dev/null @@ -1,5 +0,0 @@ --- SPDX-FileCopyrightText: 2025 SAP SE or an SAP affiliate company and Greenhouse contributors --- SPDX-License-Identifier: Apache-2.0 - -CALL add_post_migration_procedure('refresh_mvTestData_proc'); - diff --git a/internal/server/server.go b/internal/server/server.go index 8f3e2fae0..007219638 100644 --- a/internal/server/server.go +++ b/internal/server/server.go @@ -60,6 +60,8 @@ func NewServer(cfg util.Config) *Server { logrus.WithError(err).Fatalln("Error while Migrating Db") } + mariadb.StartMVEScheduler(cfg) + db, err := mariadb.NewSqlDatabase(cfg) if err != nil { logrus.WithError(err).Fatalln("Error while Creating Db") @@ -195,6 +197,8 @@ func (s *Server) BlockingStop() { ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second) defer cancel() + mariadb.StopMVEScheduler() + if err := s.nonBlockingSrv.Shutdown(ctx); err != nil { log.Fatal("Server forced to shutdown: ", err) }