Use watcher and get rid of RefreshState()

This change uses the database watcher to watch for changes to the
github entities, credentials and controller info.

Signed-off-by: Gabriel Adrian Samfira <gsamfira@cloudbasesolutions.com>

Author: gabriel-samfira

auth/instance_middleware.go

@@ -46,7 +46,20 @@ type InstanceJWTClaims struct {
 	jwt.RegisteredClaims
 }
 
-func NewInstanceJWTToken(instance params.Instance, secret, entity string, poolType params.GithubEntityType, ttlMinutes uint) (string, error) {
+func NewInstanceTokenGetter(jwtSecret string) (InstanceTokenGetter, error) {
+	if jwtSecret == "" {
+		return nil, fmt.Errorf("jwt secret is required")
+	}
+	return &instanceToken{
+		jwtSecret: jwtSecret,
+	}, nil
+}
+
+type instanceToken struct {
+	jwtSecret string
+}
+
+func (i *instanceToken) NewInstanceJWTToken(instance params.Instance, entity string, poolType params.GithubEntityType, ttlMinutes uint) (string, error) {
 	// Token expiration is equal to the bootstrap timeout set on the pool plus the polling
 	// interval garm uses to check for timed out runners. Runners that have not sent their info
 	// by the end of this interval are most likely failed and will be reaped by garm anyway.
@@ -67,7 +80,7 @@ func NewInstanceJWTToken(instance params.Instance, secret, entity string, poolTy
 		CreateAttempt: instance.CreateAttempt,
 	}
 	token := jwt.NewWithClaims(jwt.SigningMethodHS256, claims)
-	tokenString, err := token.SignedString([]byte(secret))
+	tokenString, err := token.SignedString([]byte(i.jwtSecret))
 	if err != nil {
 		return "", errors.Wrap(err, "signing token")
 	}

auth/interfaces.go

@@ -14,9 +14,17 @@
 
 package auth
 
-import "net/http"
+import (
+	"net/http"
+
+	"github.com/cloudbase/garm/params"
+)
 
 // Middleware defines an authentication middleware
 type Middleware interface {
 	Middleware(next http.Handler) http.Handler
 }
+
+type InstanceTokenGetter interface {
+	NewInstanceJWTToken(instance params.Instance, entity string, poolType params.GithubEntityType, ttlMinutes uint) (string, error)
+}

cmd/garm-cli/cmd/github_credentials.go

@@ -303,6 +303,10 @@ func parseCredentialsAddParams() (ret params.CreateGithubCredentialsParams, err
 func parseCredentialsUpdateParams() (params.UpdateGithubCredentialsParams, error) {
 	var updateParams params.UpdateGithubCredentialsParams
 
+	if credentialsAppInstallationID != 0 || credentialsAppID != 0 || credentialsPrivateKeyPath != "" {
+		updateParams.App = &params.GithubApp{}
+	}
+
 	if credentialsName != "" {
 		updateParams.Name = &credentialsName
 	}
@@ -312,6 +316,9 @@ func parseCredentialsUpdateParams() (params.UpdateGithubCredentialsParams, error
 	}
 
 	if credentialsOAuthToken != "" {
+		if updateParams.PAT == nil {
+			updateParams.PAT = &params.GithubPAT{}
+		}
 		updateParams.PAT.OAuth2Token = credentialsOAuthToken
 	}
 

database/sql/enterprise.go

@@ -132,7 +132,7 @@ func (s *sqlDatabase) ListEnterprises(_ context.Context) ([]params.Enterprise, e
 }
 
 func (s *sqlDatabase) DeleteEnterprise(ctx context.Context, enterpriseID string) error {
-	enterprise, err := s.getEnterpriseByID(ctx, s.conn, enterpriseID, "Endpoint", "Credentials")
+	enterprise, err := s.getEnterpriseByID(ctx, s.conn, enterpriseID, "Endpoint", "Credentials", "Credentials.Endpoint")
 	if err != nil {
 		return errors.Wrap(err, "fetching enterprise")
 	}
@@ -206,17 +206,13 @@ func (s *sqlDatabase) UpdateEnterprise(ctx context.Context, enterpriseID string,
 			return errors.Wrap(q.Error, "saving enterprise")
 		}
 
-		if creds.ID != 0 {
-			enterprise.Credentials = creds
-		}
-
 		return nil
 	})
 	if err != nil {
 		return params.Enterprise{}, errors.Wrap(err, "updating enterprise")
 	}
 
-	enterprise, err = s.getEnterpriseByID(ctx, s.conn, enterpriseID, "Endpoint", "Credentials")
+	enterprise, err = s.getEnterpriseByID(ctx, s.conn, enterpriseID, "Endpoint", "Credentials", "Credentials.Endpoint")
 	if err != nil {
 		return params.Enterprise{}, errors.Wrap(err, "updating enterprise")
 	}

database/sql/organizations.go

@@ -123,7 +123,7 @@ func (s *sqlDatabase) ListOrganizations(_ context.Context) ([]params.Organizatio
 }
 
 func (s *sqlDatabase) DeleteOrganization(ctx context.Context, orgID string) (err error) {
-	org, err := s.getOrgByID(ctx, s.conn, orgID, "Endpoint", "Credentials")
+	org, err := s.getOrgByID(ctx, s.conn, orgID, "Endpoint", "Credentials", "Credentials.Endpoint")
 	if err != nil {
 		return errors.Wrap(err, "fetching org")
 	}
@@ -198,17 +198,13 @@ func (s *sqlDatabase) UpdateOrganization(ctx context.Context, orgID string, para
 			return errors.Wrap(q.Error, "saving org")
 		}
 
-		if creds.ID != 0 {
-			org.Credentials = creds
-		}
-
 		return nil
 	})
 	if err != nil {
 		return params.Organization{}, errors.Wrap(err, "saving org")
 	}
 
-	org, err = s.getOrgByID(ctx, s.conn, orgID, "Endpoint", "Credentials")
+	org, err = s.getOrgByID(ctx, s.conn, orgID, "Endpoint", "Credentials", "Credentials.Endpoint")
 	if err != nil {
 		return params.Organization{}, errors.Wrap(err, "updating enterprise")
 	}

database/sql/repositories.go

@@ -122,7 +122,7 @@ func (s *sqlDatabase) ListRepositories(_ context.Context) ([]params.Repository,
 }
 
 func (s *sqlDatabase) DeleteRepository(ctx context.Context, repoID string) (err error) {
-	repo, err := s.getRepoByID(ctx, s.conn, repoID, "Endpoint", "Credentials")
+	repo, err := s.getRepoByID(ctx, s.conn, repoID, "Endpoint", "Credentials", "Credentials.Endpoint")
 	if err != nil {
 		return errors.Wrap(err, "fetching repo")
 	}
@@ -197,16 +197,13 @@ func (s *sqlDatabase) UpdateRepository(ctx context.Context, repoID string, param
 			return errors.Wrap(q.Error, "saving repo")
 		}
 
-		if creds.ID != 0 {
-			repo.Credentials = creds
-		}
 		return nil
 	})
 	if err != nil {
 		return params.Repository{}, errors.Wrap(err, "saving repo")
 	}
 
-	repo, err = s.getRepoByID(ctx, s.conn, repoID, "Endpoint", "Credentials")
+	repo, err = s.getRepoByID(ctx, s.conn, repoID, "Endpoint", "Credentials", "Credentials.Endpoint")
 	if err != nil {
 		return params.Repository{}, errors.Wrap(err, "updating enterprise")
 	}

database/watcher/filters.go

@@ -32,6 +32,18 @@ func WithAny(filters ...dbCommon.PayloadFilterFunc) dbCommon.PayloadFilterFunc {
 	}
 }
 
+// WithAll returns a filter function that returns true if all of the provided filters return true.
+func WithAll(filters ...dbCommon.PayloadFilterFunc) dbCommon.PayloadFilterFunc {
+	return func(payload dbCommon.ChangePayload) bool {
+		for _, filter := range filters {
+			if !filter(payload) {
+				return false
+			}
+		}
+		return true
+	}
+}
+
 // WithEntityTypeFilter returns a filter function that filters payloads by entity type.
 // The filter function returns true if the payload's entity type matches the provided entity type.
 func WithEntityTypeFilter(entityType dbCommon.DatabaseEntityType) dbCommon.PayloadFilterFunc {
@@ -139,3 +151,17 @@ func WithEntityJobFilter(ghEntity params.GithubEntity) dbCommon.PayloadFilterFun
 		}
 	}
 }
+
+// WithGithubCredentialsFilter returns a filter function that filters payloads by Github credentials.
+func WithGithubCredentialsFilter(creds params.GithubCredentials) dbCommon.PayloadFilterFunc {
+	return func(payload dbCommon.ChangePayload) bool {
+		if payload.EntityType != dbCommon.GithubCredentialsEntityType {
+			return false
+		}
+		credsPayload, ok := payload.Payload.(params.GithubCredentials)
+		if !ok {
+			return false
+		}
+		return credsPayload.ID == creds.ID
+	}
+}

params/params.go

@@ -419,10 +419,13 @@ func (r Repository) GetEntity() (GithubEntity, error) {
 		return GithubEntity{}, fmt.Errorf("repository has no ID")
 	}
 	return GithubEntity{
-		ID:         r.ID,
-		EntityType: GithubEntityTypeRepository,
-		Owner:      r.Owner,
-		Name:       r.Name,
+		ID:               r.ID,
+		EntityType:       GithubEntityTypeRepository,
+		Owner:            r.Owner,
+		Name:             r.Name,
+		PoolBalancerType: r.PoolBalancerType,
+		Credentials:      r.Credentials,
+		WebhookSecret:    r.WebhookSecret,
 	}, nil
 }
 
@@ -470,10 +473,12 @@ func (o Organization) GetEntity() (GithubEntity, error) {
 		return GithubEntity{}, fmt.Errorf("organization has no ID")
 	}
 	return GithubEntity{
-		ID:            o.ID,
-		EntityType:    GithubEntityTypeOrganization,
-		Owner:         o.Name,
-		WebhookSecret: o.WebhookSecret,
+		ID:               o.ID,
+		EntityType:       GithubEntityTypeOrganization,
+		Owner:            o.Name,
+		WebhookSecret:    o.WebhookSecret,
+		PoolBalancerType: o.PoolBalancerType,
+		Credentials:      o.Credentials,
 	}, nil
 }
 
@@ -517,10 +522,12 @@ func (e Enterprise) GetEntity() (GithubEntity, error) {
 		return GithubEntity{}, fmt.Errorf("enterprise has no ID")
 	}
 	return GithubEntity{
-		ID:            e.ID,
-		EntityType:    GithubEntityTypeEnterprise,
-		Owner:         e.Name,
-		WebhookSecret: e.WebhookSecret,
+		ID:               e.ID,
+		EntityType:       GithubEntityTypeEnterprise,
+		Owner:            e.Name,
+		WebhookSecret:    e.WebhookSecret,
+		PoolBalancerType: e.PoolBalancerType,
+		Credentials:      e.Credentials,
 	}, nil
 }
 
@@ -685,11 +692,6 @@ type Provider struct {
 // used by swagger client generated code
 type Providers []Provider
 
-type UpdatePoolStateParams struct {
-	WebhookSecret  string
-	InternalConfig *Internal
-}
-
 type PoolManagerStatus struct {
 	IsRunning     bool   `json:"running"`
 	FailureReason string `json:"failure_reason,omitempty"`
@@ -788,15 +790,23 @@ type UpdateSystemInfoParams struct {
 }
 
 type GithubEntity struct {
-	Owner       string            `json:"owner"`
-	Name        string            `json:"name"`
-	ID          string            `json:"id"`
-	EntityType  GithubEntityType  `json:"entity_type"`
-	Credentials GithubCredentials `json:"credentials"`
+	Owner            string            `json:"owner"`
+	Name             string            `json:"name"`
+	ID               string            `json:"id"`
+	EntityType       GithubEntityType  `json:"entity_type"`
+	Credentials      GithubCredentials `json:"credentials"`
+	PoolBalancerType PoolBalancerType  `json:"pool_balancing_type"`
 
 	WebhookSecret string `json:"-"`
 }
 
+func (g GithubEntity) GetPoolBalancerType() PoolBalancerType {
+	if g.PoolBalancerType == "" {
+		return PoolBalancerTypeRoundRobin
+	}
+	return g.PoolBalancerType
+}
+
 func (g GithubEntity) LabelScope() string {
 	switch g.EntityType {
 	case GithubEntityTypeRepository:

runner/common/mocks/PoolManager.go

@@ -53,8 +53,6 @@ type PoolManager interface {
 	// a repo, org or enterprise, we determine the destination of that webhook, retrieve the pool manager
 	// for it and call this function with the WorkflowJob as a parameter.
 	HandleWorkflowJob(job params.WorkflowJob) error
-	// RefreshState allows us to update webhook secrets and configuration for a pool manager.
-	RefreshState(param params.UpdatePoolStateParams) error
 
 	// DeleteRunner will attempt to remove a runner from the pool. If forceRemove is true, any error
 	// received from the provider will be ignored and we will proceed to remove the runner from the database.