diff --git a/cmd/main.go b/cmd/main.go index 1293dfdd309a..ce13f5426ad2 100644 --- a/cmd/main.go +++ b/cmd/main.go @@ -76,7 +76,6 @@ import ( "github.com/pkg/errors" "github.com/prometheus/client_golang/prometheus" "github.com/sirupsen/logrus" - "github.com/thoas/go-funk" "gorm.io/driver/postgres" "gorm.io/gorm" "gorm.io/gorm/logger" @@ -261,7 +260,7 @@ func main() { failOnError(err, "Failed to initialize OSImages") var releaseImagesArray = models.ReleaseImages{} - parseReleaseImages(&releaseImagesArray, failOnError) + versions.ParseReleaseImages(&releaseImagesArray, Options.ReleaseImages, failOnError) var releaseSourcesArray = models.ReleaseSources{} if Options.ReleaseSourcesConfig.ReleaseSources != "" { @@ -270,7 +269,7 @@ func main() { } var ignoredOpenshiftVersions = []string{} - parseIgnoredOpenshiftVersions(&ignoredOpenshiftVersions, failOnError) + versions.ParseIgnoredOpenshiftVersions(&ignoredOpenshiftVersions, Options.IgnoredOpenshiftVersions, failOnError) log.Println(fmt.Sprintf("Started service with OS Images %v, Release Images %v, Release Sources %v, Ignored OpenShift Versions %v", Options.OsImages, Options.ReleaseImages, Options.ReleaseSourcesConfig.ReleaseSources, Options.IgnoredOpenshiftVersions)) @@ -443,9 +442,16 @@ func main() { hostStateMonitor.Start() defer hostStateMonitor.Stop() - failOnError(addReleaseImagesToDBIfNeeded(db, releaseImagesArray, startupLeader, log), "error occured while adding configuration release images to the DB if needed") - openshiftReleaseSyncer := runOpenshiftReleaseSyncerIfNeeded(releaseSourcesArray, releaseImagesArray, db, log, releaseHandler, lead, startupLeader, failOnError) - defer stopOpenshiftReleaseSyncerIfNeeded(openshiftReleaseSyncer) + failOnError( + versions.AddReleaseImagesToDBIfNeeded(db, releaseImagesArray, startupLeader, log, Options.EnableKubeAPI, Options.ReleaseSourcesConfig.ReleaseSources), + "error occured while adding configuration release images to the DB if needed", + ) + openshiftReleaseSyncer, err := releasesources.RunOpenshiftReleaseSyncerIfNeeded( + releaseSourcesArray, releaseImagesArray, db, log, releaseHandler, lead, startupLeader, Options.EnableKubeAPI, Options.ReleaseSourcesConfig, + ) + + failOnError(err, "error occured while running OpenShift Release Syncer if needed") + defer releasesources.StopOpenshiftReleaseSyncerIfNeeded(openshiftReleaseSyncer) newUrl, err := s3wrapper.FixEndpointURL(Options.BMConfig.S3EndpointURL) failOnError(err, "failed to create valid bm config S3 endpoint URL from %s", Options.BMConfig.S3EndpointURL) @@ -945,108 +951,3 @@ func startPPROF(log *logrus.Logger) { } } } - -func runOpenshiftReleaseSyncerIfNeeded( - releaseSources models.ReleaseSources, - releaseImages models.ReleaseImages, - db *gorm.DB, - log logrus.FieldLogger, - releaseHandler oc.Release, - leader, - startupLeader leader.ElectorInterface, - failOnError func(err error, msg string, args ...interface{}), -) *thread.Thread { - if !Options.EnableKubeAPI && Options.ReleaseSourcesConfig.ReleaseSources != "" { - releaseSourcesHandler, err := releasesources.NewReleaseSourcesHandler( - releaseSources, releaseImages, log, db, Options.ReleaseSourcesConfig, leader, - ) - if err != nil { - failOnError(err, "error occurred while creating release sources handler") - } - - failOnError(startupLeader.RunWithLeader(context.Background(), func() error { - log.WithField("pkg", "releasesources").Info("Running Openshift Releases Syncer at startup") - return releaseSourcesHandler.SyncReleaseImages() - }), "error occurred while running Openshift Releases Syncer at startup") - - openShiftReleaseSyncer := thread.New( - log.WithField("pkg", "releasesources"), "Openshift Releases Syncer", Options.ReleaseSourcesConfig.OpenShiftReleaseSyncerInterval, releaseSourcesHandler.SyncReleaseImagesThreadFunc) - openShiftReleaseSyncer.Start() - return openShiftReleaseSyncer - } - - return nil -} - -func stopOpenshiftReleaseSyncerIfNeeded(openShiftReleaseSyncer *thread.Thread) { - if openShiftReleaseSyncer != nil { - openShiftReleaseSyncer.Stop() - } -} - -func addReleaseImagesToDBIfNeeded( - db *gorm.DB, - releaseImages models.ReleaseImages, - lead leader.ElectorInterface, - log logrus.FieldLogger, -) error { - // If it is kubeAPI mode or release sources was specified (therefore OpenShift Release Syncer will run), - // we should not add the release images to the DB. - if Options.EnableKubeAPI || Options.ReleaseSourcesConfig.ReleaseSources != "" { - return nil - } - - return lead.RunWithLeader(context.Background(), func() error { - tx := db.Begin() - - var count int64 - if err := tx.Model(&models.ReleaseImage{}).Count(&count).Error; err != nil { - return errors.Wrapf(err, "error occurred while trying to count the initial amount of release images in the DB") - } - - log.Debugf("Truncating all release_images table. '%d' records", count) - if err := tx.Exec("TRUNCATE TABLE release_images").Error; err != nil { - tx.Rollback() - return errors.Wrapf(err, "error occurred while trying to delete the release images in the DB") - } - - if len(releaseImages) > 0 { - log.Debugf("Inserting configuration release images to the DB. '%d' records", len(releaseImages)) - if err := tx.Create(releaseImages).Error; err != nil { - tx.Rollback() - return errors.Wrapf(err, "error occurred while trying to insert the release images to the DB") - } - } - - if err := tx.Commit().Error; err != nil { - tx.Rollback() - return errors.Wrapf(err, "error occurred while commiting changes to the DB") - } - - return nil - }) -} - -func parseIgnoredOpenshiftVersions(ignoredOpenshiftVersions *[]string, failOnError func(err error, msg string, args ...interface{})) { - if Options.IgnoredOpenshiftVersions != "" { - failOnError(json.Unmarshal([]byte(Options.IgnoredOpenshiftVersions), ignoredOpenshiftVersions), - "Failed to parse IGNORED_OPENSHIFT_VERSIONS json %s", Options.IgnoredOpenshiftVersions) - } -} - -func parseReleaseImages(releaseImages *models.ReleaseImages, failOnError func(err error, msg string, args ...interface{})) { - if Options.ReleaseImages == "" { - // ReleaseImages is optional (not used by the operator) - return - } else { - failOnError(json.Unmarshal([]byte(Options.ReleaseImages), releaseImages), - "Failed to parse RELEASE_IMAGES json %s", Options.ReleaseImages) - } - - // For backward compatibility with release images that lack the CPUArchitectures field. - funk.ForEach(*releaseImages, func(releaseImage *models.ReleaseImage) { - if releaseImage.CPUArchitectures == nil { - releaseImage.CPUArchitectures = []string{*releaseImage.CPUArchitecture} - } - }) -} diff --git a/internal/releasesources/release_sources.go b/internal/releasesources/release_sources.go index 2ca7ed5bb930..b2b45b1ac9bb 100644 --- a/internal/releasesources/release_sources.go +++ b/internal/releasesources/release_sources.go @@ -1,6 +1,8 @@ package releasesources import ( + "context" + errorspackage "errors" "fmt" "net/url" @@ -8,8 +10,10 @@ import ( "github.com/go-openapi/swag" goversion "github.com/hashicorp/go-version" "github.com/openshift/assisted-service/internal/common" + "github.com/openshift/assisted-service/internal/oc" "github.com/openshift/assisted-service/models" "github.com/openshift/assisted-service/pkg/leader" + "github.com/openshift/assisted-service/pkg/thread" "github.com/pkg/errors" "github.com/sirupsen/logrus" "github.com/thoas/go-funk" @@ -35,7 +39,7 @@ type enrichedReleaseImage struct { channel models.ReleaseChannel } -func NewReleaseSourcesHandler( +func newReleaseSourcesHandler( releaseSources models.ReleaseSources, releaseImages models.ReleaseImages, logger logrus.FieldLogger, @@ -199,8 +203,7 @@ func (h *releaseSourcesHandler) getDynamicReleaseImages() ([]*enrichedReleaseIma graph, err := h.releasesClient.getReleases(channel, openshiftVersion, cpuArchitecture) if err != nil { - h.log.WithError(err).Warnf("failed to get releases from api with: %s-%s, %s", channel, openshiftVersion, cpuArchitecture) - continue + return nil, errors.Wrapf(err, "failed to get releases from api with: %s-%s, %s", channel, openshiftVersion, cpuArchitecture) } if shouldSwitch { @@ -503,3 +506,61 @@ func getReleaseImageReference(version, cpuArchitecture string) string { } return fmt.Sprintf(releaseImageReferenceTemplateSaaS, version, cpuArchitecture) } + +func RunOpenshiftReleaseSyncerIfNeeded( + releaseSources models.ReleaseSources, + releaseImages models.ReleaseImages, + db *gorm.DB, + log logrus.FieldLogger, + releaseHandler oc.Release, + leader, + startupLeader leader.ElectorInterface, + enableKubeAPI bool, + releaseSourcesConfig Config, +) (*thread.Thread, error) { + if !enableKubeAPI && releaseSourcesConfig.ReleaseSources != "" { + releaseSourcesHandler, err := newReleaseSourcesHandler( + releaseSources, releaseImages, log, db, releaseSourcesConfig, leader, + ) + if err != nil { + return nil, errors.Wrap(err, "error occurred while creating release sources handler") + } + + err = startupLeader.RunWithLeader(context.Background(), func() error { + log.WithField("pkg", "releasesources").Info("Running Openshift Releases Syncer at startup") + if syncErr := releaseSourcesHandler.SyncReleaseImages(); syncErr != nil { + // if it failed at startup but the table contains data, we want to contine with stale data + var count int64 + if dbErr := db.Model(&models.ReleaseImage{}).Count(&count).Error; err != nil { + return errorspackage.Join(err, dbErr) + } + + // empty table + if count == 0 { + return syncErr + } + + log.WithError(err).Warn("Error occurred while running OpenShift Release Syncer at startup, proceeding with stale data") + } + + return nil + }) + + if err != nil { + return nil, errors.Wrap(err, "error occurred while running Openshift Releases Syncer at startup") + } + + openShiftReleaseSyncer := thread.New( + log.WithField("pkg", "releasesources"), "Openshift Releases Syncer", releaseSourcesConfig.OpenShiftReleaseSyncerInterval, releaseSourcesHandler.SyncReleaseImagesThreadFunc) + openShiftReleaseSyncer.Start() + return openShiftReleaseSyncer, nil + } + + return nil, nil +} + +func StopOpenshiftReleaseSyncerIfNeeded(openShiftReleaseSyncer *thread.Thread) { + if openShiftReleaseSyncer != nil { + openShiftReleaseSyncer.Stop() + } +} diff --git a/internal/releasesources/release_sources_test.go b/internal/releasesources/release_sources_test.go index 8adcfdeb4a32..ba3e8ead7648 100644 --- a/internal/releasesources/release_sources_test.go +++ b/internal/releasesources/release_sources_test.go @@ -183,7 +183,7 @@ var _ = Describe("SyncReleaseImages", func() { // Testing ReleaseSources with standard ReleaseImages It("Should not cause an error with empty release sources", func() { - handler, err = NewReleaseSourcesHandler( + handler, err = newReleaseSourcesHandler( models.ReleaseSources{}, staticReleaseImages, common.GetTestLog(), @@ -217,7 +217,7 @@ var _ = Describe("SyncReleaseImages", func() { }, } - handler, err = NewReleaseSourcesHandler( + handler, err = newReleaseSourcesHandler( releaseSources, staticReleaseImages, common.GetTestLog(), @@ -242,7 +242,7 @@ var _ = Describe("SyncReleaseImages", func() { }, } - handler, err = NewReleaseSourcesHandler( + handler, err = newReleaseSourcesHandler( releaseSources, staticReleaseImages, common.GetTestLog(), @@ -267,7 +267,7 @@ var _ = Describe("SyncReleaseImages", func() { }, } - handler, err = NewReleaseSourcesHandler( + handler, err = newReleaseSourcesHandler( releaseSources, staticReleaseImages, common.GetTestLog(), @@ -292,7 +292,7 @@ var _ = Describe("SyncReleaseImages", func() { }, } - handler, err = NewReleaseSourcesHandler( + handler, err = newReleaseSourcesHandler( releaseSources, staticReleaseImages, common.GetTestLog(), @@ -317,7 +317,7 @@ var _ = Describe("SyncReleaseImages", func() { }, } - handler, err = NewReleaseSourcesHandler( + handler, err = newReleaseSourcesHandler( releaseSources, staticReleaseImages, common.GetTestLog(), @@ -346,7 +346,7 @@ var _ = Describe("SyncReleaseImages", func() { }, } - handler, err = NewReleaseSourcesHandler( + handler, err = newReleaseSourcesHandler( releaseSources, staticReleaseImages, common.GetTestLog(), @@ -378,7 +378,7 @@ var _ = Describe("SyncReleaseImages", func() { }, }, } - handler, err = NewReleaseSourcesHandler( + handler, err = newReleaseSourcesHandler( releaseSources, staticReleaseImages, common.GetTestLog(), @@ -402,7 +402,7 @@ var _ = Describe("SyncReleaseImages", func() { }, } - handler, err = NewReleaseSourcesHandler( + handler, err = newReleaseSourcesHandler( releaseSources, staticReleaseImages, common.GetTestLog(), @@ -421,7 +421,7 @@ var _ = Describe("SyncReleaseImages", func() { }, } - handler, err = NewReleaseSourcesHandler( + handler, err = newReleaseSourcesHandler( releaseSources, staticReleaseImages, common.GetTestLog(), @@ -444,7 +444,7 @@ var _ = Describe("SyncReleaseImages", func() { }, }, } - handler, err = NewReleaseSourcesHandler( + handler, err = newReleaseSourcesHandler( releaseSources, staticReleaseImages, common.GetTestLog(), @@ -468,7 +468,7 @@ var _ = Describe("SyncReleaseImages", func() { }, } - handler, err = NewReleaseSourcesHandler( + handler, err = newReleaseSourcesHandler( releaseSources, staticReleaseImages, common.GetTestLog(), @@ -482,7 +482,7 @@ var _ = Describe("SyncReleaseImages", func() { // Testing ReleaseImages with standard ReleaseSources It("Should not cause an error with empty ReleaseImages", func() { - handler, err = NewReleaseSourcesHandler( + handler, err = newReleaseSourcesHandler( defaultReleaseSources, models.ReleaseImages{}, common.GetTestLog(), @@ -503,7 +503,7 @@ var _ = Describe("SyncReleaseImages", func() { }) It("Should not cause an error with nil ReleaseImages", func() { - handler, err = NewReleaseSourcesHandler( + handler, err = newReleaseSourcesHandler( defaultReleaseSources, nil, common.GetTestLog(), @@ -533,7 +533,7 @@ var _ = Describe("SyncReleaseImages", func() { }, } - handler, err = NewReleaseSourcesHandler( + handler, err = newReleaseSourcesHandler( defaultReleaseSources, releaseImages, common.GetTestLog(), @@ -561,7 +561,7 @@ var _ = Describe("SyncReleaseImages", func() { URL: swag.String("quay.io/openshift-release-dev/ocp-release:4.11.1-x86_64"), }, } - handler, err = NewReleaseSourcesHandler( + handler, err = newReleaseSourcesHandler( defaultReleaseSources, releaseImages, common.GetTestLog(), @@ -579,7 +579,7 @@ var _ = Describe("SyncReleaseImages", func() { }, } - handler, err = NewReleaseSourcesHandler( + handler, err = newReleaseSourcesHandler( defaultReleaseSources, releaseImages, common.GetTestLog(), @@ -597,7 +597,7 @@ var _ = Describe("SyncReleaseImages", func() { }, } - handler, err = NewReleaseSourcesHandler( + handler, err = newReleaseSourcesHandler( defaultReleaseSources, releaseImages, common.GetTestLog(), @@ -615,7 +615,7 @@ var _ = Describe("SyncReleaseImages", func() { }, } - handler, err = NewReleaseSourcesHandler( + handler, err = newReleaseSourcesHandler( defaultReleaseSources, releaseImages, common.GetTestLog(), @@ -636,7 +636,7 @@ var _ = Describe("SyncReleaseImages", func() { }, } - handler, err = NewReleaseSourcesHandler( + handler, err = newReleaseSourcesHandler( defaultReleaseSources, releaseImages, common.GetTestLog(), @@ -666,7 +666,7 @@ var _ = Describe("SyncReleaseImages", func() { }, } - handler, err = NewReleaseSourcesHandler( + handler, err = newReleaseSourcesHandler( defaultReleaseSources, releaseImages, common.GetTestLog(), @@ -692,7 +692,7 @@ var _ = Describe("SyncReleaseImages", func() { }, } - handler, err = NewReleaseSourcesHandler( + handler, err = newReleaseSourcesHandler( releaseSources, releaseImages, common.GetTestLog(), @@ -758,7 +758,7 @@ var _ = Describe("SyncReleaseImages", func() { }, } - handler, err = NewReleaseSourcesHandler( + handler, err = newReleaseSourcesHandler( releaseSources, releaseImages, common.GetTestLog(), @@ -856,7 +856,7 @@ var _ = Describe("SyncReleaseImages", func() { }, } - handler, err = NewReleaseSourcesHandler( + handler, err = newReleaseSourcesHandler( releaseSources, releaseImages, common.GetTestLog(), @@ -954,7 +954,7 @@ var _ = Describe("SyncReleaseImages", func() { }, } - handler, err = NewReleaseSourcesHandler( + handler, err = newReleaseSourcesHandler( releaseSources, releaseImages, common.GetTestLog(), @@ -1064,7 +1064,7 @@ var _ = Describe("SyncReleaseImages", func() { }, } - handler, err = NewReleaseSourcesHandler( + handler, err = newReleaseSourcesHandler( models.ReleaseSources{}, releaseImages, common.GetTestLog(), @@ -1181,7 +1181,7 @@ var _ = Describe("SyncReleaseImages", func() { }, } - handler, err = NewReleaseSourcesHandler( + handler, err = newReleaseSourcesHandler( releaseSources, models.ReleaseImages{}, common.GetTestLog(), @@ -1311,7 +1311,7 @@ var _ = Describe("SyncReleaseImages", func() { }, } - handler, err = NewReleaseSourcesHandler( + handler, err = newReleaseSourcesHandler( releaseSources, models.ReleaseImages{}, common.GetTestLog(), @@ -1463,7 +1463,7 @@ var _ = Describe("SyncReleaseImages", func() { }, } - handler, err = NewReleaseSourcesHandler( + handler, err = newReleaseSourcesHandler( releaseSources, releaseImages, common.GetTestLog(), @@ -1542,7 +1542,7 @@ var _ = Describe("SyncReleaseImages", func() { }, } - handler, err = NewReleaseSourcesHandler( + handler, err = newReleaseSourcesHandler( releaseSources, releaseImages, common.GetTestLog(), @@ -1577,7 +1577,7 @@ var _ = Describe("SyncReleaseImages", func() { }) It("Should be successfull with valid static and dynamic release images extended scenario", func() { - handler, err = NewReleaseSourcesHandler( + handler, err = newReleaseSourcesHandler( defaultReleaseSources, staticReleaseImages, common.GetTestLog(), diff --git a/internal/versions/common.go b/internal/versions/common.go index db8f095e4fca..49ee013c1007 100644 --- a/internal/versions/common.go +++ b/internal/versions/common.go @@ -2,14 +2,17 @@ package versions import ( context "context" + "encoding/json" "fmt" "sync" "github.com/openshift/assisted-service/internal/common" "github.com/openshift/assisted-service/internal/oc" models "github.com/openshift/assisted-service/models" + "github.com/openshift/assisted-service/pkg/leader" "github.com/pkg/errors" "github.com/sirupsen/logrus" + "github.com/thoas/go-funk" "golang.org/x/sync/semaphore" "gorm.io/gorm" "sigs.k8s.io/controller-runtime/pkg/client" @@ -155,3 +158,80 @@ func validateReleaseImageForRHCOS( return errors.Errorf("The requested RHCOS version (%s, arch: %s) does not have a matching OpenShift release image", *rhcosVersionPtr, cpuArchitecture) } + +func AddReleaseImagesToDBIfNeeded( + db *gorm.DB, + releaseImages models.ReleaseImages, + lead leader.ElectorInterface, + log logrus.FieldLogger, + enableKubeAPI bool, + releaseSourcesLiteral string, +) error { + // If it is kubeAPI mode or release sources was specified (therefore OpenShift Release Syncer will run), + // we should not add the release images to the DB. + if enableKubeAPI || releaseSourcesLiteral != "" { + return nil + } + + return lead.RunWithLeader(context.Background(), func() error { + tx := db.Begin() + + var count int64 + if err := tx.Model(&models.ReleaseImage{}).Count(&count).Error; err != nil { + return errors.Wrapf(err, "error occurred while trying to count the initial amount of release images in the DB") + } + + log.Debugf("Truncating all release_images table. '%d' records", count) + if err := tx.Exec("TRUNCATE TABLE release_images").Error; err != nil { + tx.Rollback() + return errors.Wrapf(err, "error occurred while trying to delete the release images in the DB") + } + + if len(releaseImages) > 0 { + log.Debugf("Inserting configuration release images to the DB. '%d' records", len(releaseImages)) + if err := tx.Create(releaseImages).Error; err != nil { + tx.Rollback() + return errors.Wrapf(err, "error occurred while trying to insert the release images to the DB") + } + } + + if err := tx.Commit().Error; err != nil { + tx.Rollback() + return errors.Wrapf(err, "error occurred while commiting changes to the DB") + } + + return nil + }) +} + +func ParseIgnoredOpenshiftVersions( + ignoredOpenshiftVersions *[]string, + ignoredOpenshiftVersionsLiteral string, + failOnError func(err error, msg string, args ...interface{}), +) { + if ignoredOpenshiftVersionsLiteral != "" { + failOnError(json.Unmarshal([]byte(ignoredOpenshiftVersionsLiteral), ignoredOpenshiftVersions), + "Failed to parse IGNORED_OPENSHIFT_VERSIONS json %s", ignoredOpenshiftVersionsLiteral) + } +} + +func ParseReleaseImages( + releaseImages *models.ReleaseImages, + releaseImagesLiteral string, + failOnError func(err error, msg string, args ...interface{}), +) { + if releaseImagesLiteral == "" { + // ReleaseImages is optional (not used by the operator) + return + } else { + failOnError(json.Unmarshal([]byte(releaseImagesLiteral), releaseImages), + "Failed to parse RELEASE_IMAGES json %s", releaseImagesLiteral) + } + + // For backward compatibility with release images that lack the CPUArchitectures field. + funk.ForEach(*releaseImages, func(releaseImage *models.ReleaseImage) { + if releaseImage.CPUArchitectures == nil { + releaseImage.CPUArchitectures = []string{*releaseImage.CPUArchitecture} + } + }) +}