From 8d57fc8fa2f15a63d7f43f0f24107d4f0889ea18 Mon Sep 17 00:00:00 2001 From: Gabriel Adrian Samfira Date: Wed, 3 Apr 2024 14:46:32 +0000 Subject: [PATCH] Add rudimentary database watcher Adds a simple database watcher. At this point it's just one process, but the plan is to allow different implementations that inform the local running workers of changes that have occured on entities of interest in the database. Signed-off-by: Gabriel Adrian Samfira --- README.md | 2 +- cmd/garm/main.go | 3 + database/common/errors.go | 12 ++ database/common/{common.go => store.go} | 7 +- database/common/watcher.go | 50 ++++++++ database/sql/instances.go | 20 ---- database/sql/pools.go | 33 +++++- database/sql/repositories.go | 34 +++++- database/sql/repositories_test.go | 7 ++ database/sql/sql.go | 19 ++- database/sql/util.go | 29 +++++ database/watcher/consumer.go | 75 ++++++++++++ database/watcher/producer.go | 49 ++++++++ database/watcher/test_export.go | 12 ++ database/watcher/watcher.go | 151 ++++++++++++++++++++++++ internal/testing/mock_watcher.go | 45 +++++++ runner/pool/pool.go | 2 +- runner/repositories_test.go | 5 + 18 files changed, 514 insertions(+), 41 deletions(-) create mode 100644 database/common/errors.go rename database/common/{common.go => store.go} (98%) create mode 100644 database/common/watcher.go create mode 100644 database/watcher/consumer.go create mode 100644 database/watcher/producer.go create mode 100644 database/watcher/test_export.go create mode 100644 database/watcher/watcher.go create mode 100644 internal/testing/mock_watcher.go diff --git a/README.md b/README.md index 9536b8fb..f68da05b 100644 --- a/README.md +++ b/README.md @@ -12,7 +12,7 @@ GARM supports creating pools on either GitHub itself or on your own deployment o Through the use of providers, `GARM` can create runners in a variety of environments using the same `GARM` instance. Whether you want to create pools of runners in your OpenStack cloud, your Azure cloud and your Kubernetes cluster, that is easily achieved by just installing the appropriate providers, configuring them in `GARM` and creating pools that use them. You can create zero-runner pools for instances with high costs (large VMs, GPU enabled instances, etc) and have them spin up on demand, or you can create large pools of k8s backed runners that can be used for your CI/CD pipelines at a moment's notice. You can mix them up and create pools in any combination of providers or resource allocations you want. -:warning: **Important note**: The README and documentation in the `main` branch are relevant to the not yet released code that is present in `main`. Following the documentation from the `main` branch for a stable release of GARM, may lead to errors. To view the documentation for the latest stable release, please switch to the appropriate tag. For information about setting up `v0.1.4`, please refer to the [v0.1.4 tag](https://github.com/cloudbase/garm/tree/v0.1.4) +:warning: **Important note**: The README and documentation in the `main` branch are relevant to the not yet released code that is present in `main`. Following the documentation from the `main` branch for a stable release of GARM, may lead to errors. To view the documentation for the latest stable release, please switch to the appropriate tag. For information about setting up `v0.1.4`, please refer to the [v0.1.4 tag](https://github.com/cloudbase/garm/tree/v0.1.4). ## Join us on slack diff --git a/cmd/garm/main.go b/cmd/garm/main.go index 83d70326..d8eed80c 100644 --- a/cmd/garm/main.go +++ b/cmd/garm/main.go @@ -40,6 +40,7 @@ import ( "github.com/cloudbase/garm/config" "github.com/cloudbase/garm/database" "github.com/cloudbase/garm/database/common" + "github.com/cloudbase/garm/database/watcher" "github.com/cloudbase/garm/metrics" "github.com/cloudbase/garm/params" "github.com/cloudbase/garm/runner" //nolint:typecheck @@ -183,6 +184,7 @@ func main() { } ctx, stop := signal.NotifyContext(context.Background(), signals...) defer stop() + watcher.InitWatcher(ctx) ctx = auth.GetAdminContext(ctx) @@ -313,6 +315,7 @@ func main() { }() <-ctx.Done() + shutdownCtx, shutdownCancel := context.WithTimeout(context.Background(), 60*time.Second) defer shutdownCancel() if err := srv.Shutdown(shutdownCtx); err != nil { diff --git a/database/common/errors.go b/database/common/errors.go new file mode 100644 index 00000000..b68d8311 --- /dev/null +++ b/database/common/errors.go @@ -0,0 +1,12 @@ +package common + +import "fmt" + +var ( + ErrProducerClosed = fmt.Errorf("producer is closed") + ErrProducerTimeoutErr = fmt.Errorf("producer timeout error") + ErrProducerAlreadyRegistered = fmt.Errorf("producer already registered") + ErrConsumerAlreadyRegistered = fmt.Errorf("consumer already registered") + ErrWatcherAlreadyStarted = fmt.Errorf("watcher already started") + ErrWatcherNotInitialized = fmt.Errorf("watcher not initialized") +) diff --git a/database/common/common.go b/database/common/store.go similarity index 98% rename from database/common/common.go rename to database/common/store.go index 4f0df368..18075c1d 100644 --- a/database/common/common.go +++ b/database/common/store.go @@ -119,7 +119,7 @@ type JobsStore interface { DeleteCompletedJobs(ctx context.Context) error } -type EntityPools interface { +type EntityPoolStore interface { CreateEntityPool(ctx context.Context, entity params.GithubEntity, param params.CreatePoolParams) (params.Pool, error) GetEntityPool(ctx context.Context, entity params.GithubEntity, poolID string) (params.Pool, error) DeleteEntityPool(ctx context.Context, entity params.GithubEntity, poolID string) error @@ -144,8 +144,11 @@ type Store interface { UserStore InstanceStore JobsStore - EntityPools GithubEndpointStore GithubCredentialsStore ControllerStore + EntityPoolStore + + ControllerInfo() (params.ControllerInfo, error) + InitController() (params.ControllerInfo, error) } diff --git a/database/common/watcher.go b/database/common/watcher.go new file mode 100644 index 00000000..4903e4ab --- /dev/null +++ b/database/common/watcher.go @@ -0,0 +1,50 @@ +package common + +type ( + DatabaseEntityType string + OperationType string + PayloadFilterFunc func(ChangePayload) bool +) + +const ( + RepositoryEntityType DatabaseEntityType = "repository" + OrganizationEntityType DatabaseEntityType = "organization" + EnterpriseEntityType DatabaseEntityType = "enterprise" + PoolEntityType DatabaseEntityType = "pool" + UserEntityType DatabaseEntityType = "user" + InstanceEntityType DatabaseEntityType = "instance" + JobEntityType DatabaseEntityType = "job" + ControllerEntityType DatabaseEntityType = "controller" + GithubCredentialsEntityType DatabaseEntityType = "github_credentials" + GithubEndpointEntityType DatabaseEntityType = "github_endpoint" +) + +const ( + CreateOperation OperationType = "create" + UpdateOperation OperationType = "update" + DeleteOperation OperationType = "delete" +) + +type ChangePayload struct { + EntityType DatabaseEntityType + Operation OperationType + Payload interface{} +} + +type Consumer interface { + Watch() <-chan ChangePayload + IsClosed() bool + Close() + SetFilters(filters ...PayloadFilterFunc) +} + +type Producer interface { + Notify(ChangePayload) error + IsClosed() bool + Close() +} + +type Watcher interface { + RegisterProducer(ID string) (Producer, error) + RegisterConsumer(ID string, filters ...PayloadFilterFunc) (Consumer, error) +} diff --git a/database/sql/instances.go b/database/sql/instances.go index 1544090b..c09b60f3 100644 --- a/database/sql/instances.go +++ b/database/sql/instances.go @@ -25,29 +25,9 @@ import ( "gorm.io/gorm/clause" runnerErrors "github.com/cloudbase/garm-provider-common/errors" - "github.com/cloudbase/garm-provider-common/util" "github.com/cloudbase/garm/params" ) -func (s *sqlDatabase) marshalAndSeal(data interface{}) ([]byte, error) { - enc, err := json.Marshal(data) - if err != nil { - return nil, errors.Wrap(err, "marshalling data") - } - return util.Seal(enc, []byte(s.cfg.Passphrase)) -} - -func (s *sqlDatabase) unsealAndUnmarshal(data []byte, target interface{}) error { - decrypted, err := util.Unseal(data, []byte(s.cfg.Passphrase)) - if err != nil { - return errors.Wrap(err, "decrypting data") - } - if err := json.Unmarshal(decrypted, target); err != nil { - return errors.Wrap(err, "unmarshalling data") - } - return nil -} - func (s *sqlDatabase) CreateInstance(_ context.Context, poolID string, param params.CreateInstanceParams) (params.Instance, error) { pool, err := s.getPoolByID(s.conn, poolID) if err != nil { diff --git a/database/sql/pools.go b/database/sql/pools.go index 161b1d58..4d33343d 100644 --- a/database/sql/pools.go +++ b/database/sql/pools.go @@ -24,6 +24,7 @@ import ( "gorm.io/gorm" runnerErrors "github.com/cloudbase/garm-provider-common/errors" + "github.com/cloudbase/garm/database/common" "github.com/cloudbase/garm/params" ) @@ -66,12 +67,18 @@ func (s *sqlDatabase) GetPoolByID(_ context.Context, poolID string) (params.Pool return s.sqlToCommonPool(pool) } -func (s *sqlDatabase) DeletePoolByID(_ context.Context, poolID string) error { +func (s *sqlDatabase) DeletePoolByID(_ context.Context, poolID string) (err error) { pool, err := s.getPoolByID(s.conn, poolID) if err != nil { return errors.Wrap(err, "fetching pool by ID") } + defer func() { + if err == nil { + s.sendNotify(common.PoolEntityType, common.DeleteOperation, pool) + } + }() + if q := s.conn.Unscoped().Delete(&pool); q.Error != nil { return errors.Wrap(q.Error, "removing pool") } @@ -247,11 +254,17 @@ func (s *sqlDatabase) FindPoolsMatchingAllTags(_ context.Context, entityType par return pools, nil } -func (s *sqlDatabase) CreateEntityPool(_ context.Context, entity params.GithubEntity, param params.CreatePoolParams) (params.Pool, error) { +func (s *sqlDatabase) CreateEntityPool(_ context.Context, entity params.GithubEntity, param params.CreatePoolParams) (pool params.Pool, err error) { if len(param.Tags) == 0 { return params.Pool{}, runnerErrors.NewBadRequestError("no tags specified") } + defer func() { + if err == nil { + s.sendNotify(common.PoolEntityType, common.CreateOperation, pool) + } + }() + newPool := Pool{ ProviderName: param.ProviderName, MaxRunners: param.MaxRunners, @@ -313,12 +326,12 @@ func (s *sqlDatabase) CreateEntityPool(_ context.Context, entity params.GithubEn return params.Pool{}, err } - pool, err := s.getPoolByID(s.conn, newPool.ID.String(), "Tags", "Instances", "Enterprise", "Organization", "Repository") + dbPool, err := s.getPoolByID(s.conn, newPool.ID.String(), "Tags", "Instances", "Enterprise", "Organization", "Repository") if err != nil { return params.Pool{}, errors.Wrap(err, "fetching pool") } - return s.sqlToCommonPool(pool) + return s.sqlToCommonPool(dbPool) } func (s *sqlDatabase) GetEntityPool(_ context.Context, entity params.GithubEntity, poolID string) (params.Pool, error) { @@ -329,12 +342,21 @@ func (s *sqlDatabase) GetEntityPool(_ context.Context, entity params.GithubEntit return s.sqlToCommonPool(pool) } -func (s *sqlDatabase) DeleteEntityPool(_ context.Context, entity params.GithubEntity, poolID string) error { +func (s *sqlDatabase) DeleteEntityPool(_ context.Context, entity params.GithubEntity, poolID string) (err error) { entityID, err := uuid.Parse(entity.ID) if err != nil { return errors.Wrap(runnerErrors.ErrBadRequest, "parsing id") } + defer func() { + if err == nil { + pool := params.Pool{ + ID: poolID, + } + s.sendNotify(common.PoolEntityType, common.DeleteOperation, pool) + } + }() + poolUUID, err := uuid.Parse(poolID) if err != nil { return errors.Wrap(runnerErrors.ErrBadRequest, "parsing pool id") @@ -374,6 +396,7 @@ func (s *sqlDatabase) UpdateEntityPool(_ context.Context, entity params.GithubEn if err != nil { return params.Pool{}, err } + s.sendNotify(common.PoolEntityType, common.UpdateOperation, updatedPool) return updatedPool, nil } diff --git a/database/sql/repositories.go b/database/sql/repositories.go index 510a9959..7ab1c522 100644 --- a/database/sql/repositories.go +++ b/database/sql/repositories.go @@ -24,10 +24,17 @@ import ( runnerErrors "github.com/cloudbase/garm-provider-common/errors" "github.com/cloudbase/garm-provider-common/util" + "github.com/cloudbase/garm/database/common" "github.com/cloudbase/garm/params" ) -func (s *sqlDatabase) CreateRepository(ctx context.Context, owner, name, credentialsName, webhookSecret string, poolBalancerType params.PoolBalancerType) (params.Repository, error) { +func (s *sqlDatabase) CreateRepository(ctx context.Context, owner, name, credentialsName, webhookSecret string, poolBalancerType params.PoolBalancerType) (param params.Repository, err error) { + defer func() { + if err == nil { + s.sendNotify(common.RepositoryEntityType, common.CreateOperation, param) + } + }() + if webhookSecret == "" { return params.Repository{}, errors.New("creating repo: missing secret") } @@ -68,7 +75,7 @@ func (s *sqlDatabase) CreateRepository(ctx context.Context, owner, name, credent return params.Repository{}, errors.Wrap(err, "creating repository") } - param, err := s.sqlToCommonRepository(newRepo, true) + param, err = s.sqlToCommonRepository(newRepo, true) if err != nil { return params.Repository{}, errors.Wrap(err, "creating repository") } @@ -113,12 +120,21 @@ func (s *sqlDatabase) ListRepositories(_ context.Context) ([]params.Repository, return ret, nil } -func (s *sqlDatabase) DeleteRepository(ctx context.Context, repoID string) error { +func (s *sqlDatabase) DeleteRepository(ctx context.Context, repoID string) (err error) { repo, err := s.getRepoByID(ctx, s.conn, repoID) if err != nil { return errors.Wrap(err, "fetching repo") } + defer func(repo Repository) { + if err == nil { + asParam, innerErr := s.sqlToCommonRepository(repo, true) + if innerErr == nil { + s.sendNotify(common.RepositoryEntityType, common.DeleteOperation, asParam) + } + } + }(repo) + q := s.conn.Unscoped().Delete(&repo) if q.Error != nil && !errors.Is(q.Error, gorm.ErrRecordNotFound) { return errors.Wrap(q.Error, "deleting repo") @@ -127,10 +143,15 @@ func (s *sqlDatabase) DeleteRepository(ctx context.Context, repoID string) error return nil } -func (s *sqlDatabase) UpdateRepository(ctx context.Context, repoID string, param params.UpdateEntityParams) (params.Repository, error) { +func (s *sqlDatabase) UpdateRepository(ctx context.Context, repoID string, param params.UpdateEntityParams) (newParams params.Repository, err error) { + defer func() { + if err == nil { + s.sendNotify(common.RepositoryEntityType, common.UpdateOperation, newParams) + } + }() var repo Repository var creds GithubCredentials - err := s.conn.Transaction(func(tx *gorm.DB) error { + err = s.conn.Transaction(func(tx *gorm.DB) error { var err error repo, err = s.getRepoByID(ctx, tx, repoID) if err != nil { @@ -186,7 +207,8 @@ func (s *sqlDatabase) UpdateRepository(ctx context.Context, repoID string, param if err != nil { return params.Repository{}, errors.Wrap(err, "updating enterprise") } - newParams, err := s.sqlToCommonRepository(repo, true) + + newParams, err = s.sqlToCommonRepository(repo, true) if err != nil { return params.Repository{}, errors.Wrap(err, "saving repo") } diff --git a/database/sql/repositories_test.go b/database/sql/repositories_test.go index 6eb20a2c..34b07bd1 100644 --- a/database/sql/repositories_test.go +++ b/database/sql/repositories_test.go @@ -30,6 +30,7 @@ import ( "github.com/cloudbase/garm/auth" dbCommon "github.com/cloudbase/garm/database/common" + "github.com/cloudbase/garm/database/watcher" garmTesting "github.com/cloudbase/garm/internal/testing" "github.com/cloudbase/garm/params" ) @@ -827,5 +828,11 @@ func (s *RepoTestSuite) TestUpdateRepositoryPoolInvalidRepoID() { func TestRepoTestSuite(t *testing.T) { t.Parallel() + + watcher.SetWatcher(&garmTesting.MockWatcher{}) suite.Run(t, new(RepoTestSuite)) } + +func init() { + watcher.SetWatcher(&garmTesting.MockWatcher{}) +} diff --git a/database/sql/sql.go b/database/sql/sql.go index 5f970558..6ee8a2d9 100644 --- a/database/sql/sql.go +++ b/database/sql/sql.go @@ -31,6 +31,7 @@ import ( "github.com/cloudbase/garm/auth" "github.com/cloudbase/garm/config" "github.com/cloudbase/garm/database/common" + "github.com/cloudbase/garm/database/watcher" "github.com/cloudbase/garm/params" "github.com/cloudbase/garm/util/appdefaults" ) @@ -68,10 +69,15 @@ func NewSQLDatabase(ctx context.Context, cfg config.Database) (common.Store, err if err != nil { return nil, errors.Wrap(err, "creating DB connection") } + producer, err := watcher.RegisterProducer("sql") + if err != nil { + return nil, errors.Wrap(err, "registering producer") + } db := &sqlDatabase{ - conn: conn, - ctx: ctx, - cfg: cfg, + conn: conn, + ctx: ctx, + cfg: cfg, + producer: producer, } if err := db.migrateDB(); err != nil { @@ -81,9 +87,10 @@ func NewSQLDatabase(ctx context.Context, cfg config.Database) (common.Store, err } type sqlDatabase struct { - conn *gorm.DB - ctx context.Context - cfg config.Database + conn *gorm.DB + ctx context.Context + cfg config.Database + producer common.Producer } var renameTemplate = ` diff --git a/database/sql/util.go b/database/sql/util.go index b7f8a058..5814483d 100644 --- a/database/sql/util.go +++ b/database/sql/util.go @@ -26,6 +26,7 @@ import ( runnerErrors "github.com/cloudbase/garm-provider-common/errors" commonParams "github.com/cloudbase/garm-provider-common/params" "github.com/cloudbase/garm-provider-common/util" + dbCommon "github.com/cloudbase/garm/database/common" "github.com/cloudbase/garm/params" ) @@ -467,3 +468,31 @@ func (s *sqlDatabase) hasGithubEntity(tx *gorm.DB, entityType params.GithubEntit } return nil } + +func (s *sqlDatabase) marshalAndSeal(data interface{}) ([]byte, error) { + enc, err := json.Marshal(data) + if err != nil { + return nil, errors.Wrap(err, "marshalling data") + } + return util.Seal(enc, []byte(s.cfg.Passphrase)) +} + +func (s *sqlDatabase) unsealAndUnmarshal(data []byte, target interface{}) error { + decrypted, err := util.Unseal(data, []byte(s.cfg.Passphrase)) + if err != nil { + return errors.Wrap(err, "decrypting data") + } + if err := json.Unmarshal(decrypted, target); err != nil { + return errors.Wrap(err, "unmarshalling data") + } + return nil +} + +func (s *sqlDatabase) sendNotify(entityType dbCommon.DatabaseEntityType, op dbCommon.OperationType, payload interface{}) { + message := dbCommon.ChangePayload{ + Operation: op, + Payload: payload, + EntityType: entityType, + } + s.producer.Notify(message) +} diff --git a/database/watcher/consumer.go b/database/watcher/consumer.go new file mode 100644 index 00000000..369344ba --- /dev/null +++ b/database/watcher/consumer.go @@ -0,0 +1,75 @@ +package watcher + +import ( + "log/slog" + "sync" + "time" + + "github.com/cloudbase/garm/database/common" +) + +type consumer struct { + messages chan common.ChangePayload + filters []common.PayloadFilterFunc + id string + + mux sync.Mutex + closed bool + quit chan struct{} +} + +func (w *consumer) SetFilters(filters ...common.PayloadFilterFunc) { + w.mux.Lock() + defer w.mux.Unlock() + w.filters = filters +} + +func (w *consumer) Watch() <-chan common.ChangePayload { + return w.messages +} + +func (w *consumer) Close() { + w.mux.Lock() + defer w.mux.Unlock() + if w.closed { + return + } + close(w.messages) + close(w.quit) + w.closed = true +} + +func (w *consumer) IsClosed() bool { + w.mux.Lock() + defer w.mux.Unlock() + return w.closed +} + +func (w *consumer) Send(payload common.ChangePayload) { + w.mux.Lock() + defer w.mux.Unlock() + + if w.closed { + return + } + + if len(w.filters) > 0 { + shouldSend := false + for _, filter := range w.filters { + if filter(payload) { + shouldSend = true + break + } + } + + if !shouldSend { + return + } + } + + slog.Info("Sending payload to consumer", "consumer", w.id) + select { + case w.messages <- payload: + case <-time.After(1 * time.Second): + } +} diff --git a/database/watcher/producer.go b/database/watcher/producer.go new file mode 100644 index 00000000..70578004 --- /dev/null +++ b/database/watcher/producer.go @@ -0,0 +1,49 @@ +package watcher + +import ( + "sync" + + "github.com/cloudbase/garm/database/common" +) + +type producer struct { + closed bool + mux sync.Mutex + id string + + messages chan common.ChangePayload + quit chan struct{} +} + +func (w *producer) Notify(payload common.ChangePayload) error { + w.mux.Lock() + defer w.mux.Unlock() + + if w.closed { + return common.ErrProducerClosed + } + + select { + case w.messages <- payload: + default: + return common.ErrProducerTimeoutErr + } + return nil +} + +func (w *producer) Close() { + w.mux.Lock() + defer w.mux.Unlock() + if w.closed { + return + } + w.closed = true + close(w.messages) + close(w.quit) +} + +func (w *producer) IsClosed() bool { + w.mux.Lock() + defer w.mux.Unlock() + return w.closed +} diff --git a/database/watcher/test_export.go b/database/watcher/test_export.go new file mode 100644 index 00000000..4c75233e --- /dev/null +++ b/database/watcher/test_export.go @@ -0,0 +1,12 @@ +//go:build testing +// +build testing + +package watcher + +import "github.com/cloudbase/garm/database/common" + +// SetWatcher sets the watcher to be used by the database package. +// This function is intended for use in tests only. +func SetWatcher(w common.Watcher) { + databaseWatcher = w +} diff --git a/database/watcher/watcher.go b/database/watcher/watcher.go new file mode 100644 index 00000000..23400e21 --- /dev/null +++ b/database/watcher/watcher.go @@ -0,0 +1,151 @@ +package watcher + +import ( + "context" + "sync" + + "github.com/cloudbase/garm/database/common" +) + +var databaseWatcher common.Watcher + +func InitWatcher(ctx context.Context) { + if databaseWatcher != nil { + return + } + w := &watcher{ + producers: make(map[string]*producer), + consumers: make(map[string]*consumer), + quit: make(chan struct{}), + ctx: ctx, + } + + go w.loop() + databaseWatcher = w +} + +func RegisterProducer(id string) (common.Producer, error) { + if databaseWatcher == nil { + return nil, common.ErrWatcherNotInitialized + } + return databaseWatcher.RegisterProducer(id) +} + +func RegisterConsumer(id string, filters ...common.PayloadFilterFunc) (common.Consumer, error) { + if databaseWatcher == nil { + return nil, common.ErrWatcherNotInitialized + } + return databaseWatcher.RegisterConsumer(id, filters...) +} + +type watcher struct { + producers map[string]*producer + consumers map[string]*consumer + + mux sync.Mutex + closed bool + quit chan struct{} + ctx context.Context +} + +func (w *watcher) RegisterProducer(id string) (common.Producer, error) { + if _, ok := w.producers[id]; ok { + return nil, common.ErrProducerAlreadyRegistered + } + p := &producer{ + id: id, + messages: make(chan common.ChangePayload, 1), + quit: make(chan struct{}), + } + w.producers[id] = p + go w.serviceProducer(p) + return p, nil +} + +func (w *watcher) serviceProducer(prod *producer) { + defer func() { + w.mux.Lock() + defer w.mux.Unlock() + prod.Close() + delete(w.producers, prod.id) + }() + for { + select { + case <-w.quit: + return + case <-w.ctx.Done(): + return + case payload := <-prod.messages: + for _, c := range w.consumers { + go c.Send(payload) + } + } + } +} + +func (w *watcher) RegisterConsumer(id string, filters ...common.PayloadFilterFunc) (common.Consumer, error) { + if _, ok := w.consumers[id]; ok { + return nil, common.ErrConsumerAlreadyRegistered + } + c := &consumer{ + messages: make(chan common.ChangePayload, 1), + filters: filters, + quit: make(chan struct{}), + id: id, + } + w.consumers[id] = c + go w.serviceConsumer(c) + return c, nil +} + +func (w *watcher) serviceConsumer(consumer *consumer) { + defer func() { + w.mux.Lock() + defer w.mux.Unlock() + consumer.Close() + delete(w.consumers, consumer.id) + }() + for { + select { + case <-consumer.quit: + return + case <-w.quit: + return + case <-w.ctx.Done(): + return + } + } +} + +func (w *watcher) Close() { + w.mux.Lock() + defer w.mux.Unlock() + if w.closed { + return + } + + close(w.quit) + w.closed = true + + for _, p := range w.producers { + p.Close() + } + + for _, c := range w.consumers { + c.Close() + } +} + +func (w *watcher) loop() { + defer func() { + w.Close() + }() + for { + select { + case <-w.quit: + return + case <-w.ctx.Done(): + return + } + } +} diff --git a/internal/testing/mock_watcher.go b/internal/testing/mock_watcher.go new file mode 100644 index 00000000..394091bd --- /dev/null +++ b/internal/testing/mock_watcher.go @@ -0,0 +1,45 @@ +//go:build testing +// +build testing + +package testing + +import "github.com/cloudbase/garm/database/common" + +type MockWatcher struct{} + +func (w *MockWatcher) RegisterProducer(_ string) (common.Producer, error) { + return &MockProducer{}, nil +} + +func (w *MockWatcher) RegisterConsumer(_ string, _ ...common.PayloadFilterFunc) (common.Consumer, error) { + return &MockConsumer{}, nil +} + +type MockProducer struct{} + +func (p *MockProducer) Notify(_ common.ChangePayload) error { + return nil +} + +func (p *MockProducer) IsClosed() bool { + return false +} + +func (p *MockProducer) Close() { +} + +type MockConsumer struct{} + +func (c *MockConsumer) Watch() <-chan common.ChangePayload { + return nil +} + +func (c *MockConsumer) SetFilters(_ ...common.PayloadFilterFunc) { +} + +func (c *MockConsumer) Close() { +} + +func (c *MockConsumer) IsClosed() bool { + return false +} diff --git a/runner/pool/pool.go b/runner/pool/pool.go index 1261ad21..f7a8ea68 100644 --- a/runner/pool/pool.go +++ b/runner/pool/pool.go @@ -69,7 +69,7 @@ type urls struct { } func NewEntityPoolManager(ctx context.Context, entity params.GithubEntity, cfgInternal params.Internal, providers map[string]common.Provider, store dbCommon.Store) (common.PoolManager, error) { - ctx = garmUtil.WithContext(ctx, slog.Any("pool_mgr", entity.String()), slog.Any("pool_type", params.GithubEntityTypeRepository)) + ctx = garmUtil.WithContext(ctx, slog.Any("pool_mgr", entity.String()), slog.Any("pool_type", entity.EntityType)) ghc, err := garmUtil.GithubClient(ctx, entity, cfgInternal.GithubCredentialsDetails) if err != nil { return nil, errors.Wrap(err, "getting github client") diff --git a/runner/repositories_test.go b/runner/repositories_test.go index d0a6ab61..f17bd93a 100644 --- a/runner/repositories_test.go +++ b/runner/repositories_test.go @@ -26,6 +26,7 @@ import ( runnerErrors "github.com/cloudbase/garm-provider-common/errors" "github.com/cloudbase/garm/database" dbCommon "github.com/cloudbase/garm/database/common" + "github.com/cloudbase/garm/database/watcher" garmTesting "github.com/cloudbase/garm/internal/testing" "github.com/cloudbase/garm/params" "github.com/cloudbase/garm/runner/common" @@ -51,6 +52,10 @@ type RepoTestFixtures struct { PoolMgrCtrlMock *runnerMocks.PoolManagerController } +func init() { + watcher.SetWatcher(&garmTesting.MockWatcher{}) +} + type RepoTestSuite struct { suite.Suite Fixtures *RepoTestFixtures