Add runner periodic cleanup check

Adds a periodic cleanup function that cross checks runners between github,
the provider and the GARM database. If an inconsistency is found, GARM will
attempt to fix it.

Signed-off-by: Gabriel Adrian Samfira <[email protected]>

Author: gabriel-samfira

database/watcher/filters.go

@@ -255,9 +255,10 @@ func WithScaleSetInstanceFilter(scaleset params.ScaleSet) dbCommon.PayloadFilter
 		}
 
 		instance, ok := payload.Payload.(params.Instance)
-		if !ok {
+		if !ok || instance.ScaleSetID == 0 {
 			return false
 		}
+
 		return instance.ScaleSetID == scaleset.ID
 	}
 }

locking/local_locker.go

@@ -33,19 +33,19 @@ var _ Locker = &keyMutex{}
 
 func (k *keyMutex) TryLock(key, identifier string) bool {
 	mux, _ := k.muxes.LoadOrStore(key, &lockWithIdent{
-		mux:   sync.Mutex{},
-		ident: identifier,
+		mux: sync.Mutex{},
 	})
 	keyMux := mux.(*lockWithIdent)
+	keyMux.ident = identifier
 	return keyMux.mux.TryLock()
 }
 
 func (k *keyMutex) Lock(key, identifier string) {
 	mux, _ := k.muxes.LoadOrStore(key, &lockWithIdent{
-		mux:   sync.Mutex{},
-		ident: identifier,
+		mux: sync.Mutex{},
 	})
 	keyMux := mux.(*lockWithIdent)
+	keyMux.ident = identifier
 	keyMux.mux.Lock()
 }
 
@@ -60,6 +60,7 @@ func (k *keyMutex) Unlock(key string, remove bool) {
 	}
 	_, filename, line, _ := runtime.Caller(1)
 	slog.Debug("unlocking", "key", key, "identifier", keyMux.ident, "caller", fmt.Sprintf("%s:%d", filename, line))
+	keyMux.ident = ""
 	keyMux.mux.Unlock()
 }
 

params/github.go

@@ -421,6 +421,7 @@ func (r RunnerScaleSetMessage) GetJobsFromBody() ([]ScaleSetJobMessage, error) {
 type RunnerReference struct {
 	ID                int64       `json:"id"`
 	Name              string      `json:"name"`
+	OS                string      `json:"os"`
 	RunnerScaleSetID  int         `json:"runnerScaleSetId"`
 	CreatedOn         interface{} `json:"createdOn"`
 	RunnerGroupID     uint64      `json:"runnerGroupId"`
@@ -431,9 +432,29 @@ type RunnerReference struct {
 	Status            interface{} `json:"status"`
 	DisableUpdate     bool        `json:"disableUpdate"`
 	ProvisioningState string      `json:"provisioningState"`
+	Busy              bool        `json:"busy"`
 	Labels            []Label     `json:"labels,omitempty"`
 }
 
+func (r RunnerReference) GetStatus() RunnerStatus {
+	status, ok := r.Status.(string)
+	if !ok {
+		return RunnerUnknown
+	}
+	runnerStatus := RunnerStatus(status)
+	if !runnerStatus.IsValid() {
+		return RunnerUnknown
+	}
+
+	if runnerStatus == RunnerOnline {
+		if r.Busy {
+			return RunnerActive
+		}
+		return RunnerIdle
+	}
+	return runnerStatus
+}
+
 type RunnerScaleSetJitRunnerConfig struct {
 	Runner           *RunnerReference `json:"runner"`
 	EncodedJITConfig string           `json:"encodedJITConfig"`

params/params.go

@@ -49,6 +49,18 @@ type (
 	ScaleSetMessageType string
 )
 
+func (s RunnerStatus) IsValid() bool {
+	switch s {
+	case RunnerIdle, RunnerPending, RunnerTerminated,
+		RunnerInstalling, RunnerFailed,
+		RunnerActive, RunnerOffline,
+		RunnerUnknown, RunnerOnline:
+
+		return true
+	}
+	return false
+}
+
 const (
 	// PoolBalancerTypeRoundRobin will try to cycle through the pools of an entity
 	// in a round robin fashion. For example, if a repository has multiple pools that
@@ -117,6 +129,9 @@ const (
 	RunnerInstalling RunnerStatus = "installing"
 	RunnerFailed     RunnerStatus = "failed"
 	RunnerActive     RunnerStatus = "active"
+	RunnerOffline    RunnerStatus = "offline"
+	RunnerOnline     RunnerStatus = "online"
+	RunnerUnknown    RunnerStatus = "unknown"
 )
 
 const (

runner/common/util.go

@@ -28,6 +28,10 @@ type GithubEntityOperations interface {
 	GithubBaseURL() *url.URL
 }
 
+type RateLimitClient interface {
+	RateLimit(ctx context.Context) (*github.RateLimits, error)
+}
+
 // GithubClient that describes the minimum list of functions we need to interact with github.
 // Allows for easier testing.
 //

runner/pool/pool.go

@@ -439,10 +439,14 @@ func (r *basePoolManager) cleanupOrphanedProviderRunners(runners []*github.Runne
 			// github so we let them be for now.
 			continue
 		}
+		pool, err := r.store.GetEntityPool(r.ctx, r.entity, instance.PoolID)
+		if err != nil {
+			return errors.Wrap(err, "fetching instance pool info")
+		}
 
 		switch instance.RunnerStatus {
 		case params.RunnerPending, params.RunnerInstalling:
-			if time.Since(instance.UpdatedAt).Minutes() < float64(instance.RunnerTimeout()) {
+			if time.Since(instance.UpdatedAt).Minutes() < float64(pool.RunnerTimeout()) {
 				// runner is still installing. We give it a chance to finish.
 				slog.DebugContext(
 					r.ctx, "runner is still installing, give it a chance to finish",
@@ -510,7 +514,11 @@ func (r *basePoolManager) reapTimedOutRunners(runners []*github.Runner) error {
 		}
 		defer locking.Unlock(instance.Name, false)
 
-		if time.Since(instance.UpdatedAt).Minutes() < float64(instance.RunnerTimeout()) {
+		pool, err := r.store.GetEntityPool(r.ctx, r.entity, instance.PoolID)
+		if err != nil {
+			return errors.Wrap(err, "fetching instance pool info")
+		}
+		if time.Since(instance.UpdatedAt).Minutes() < float64(pool.RunnerTimeout()) {
 			continue
 		}
 

util/github/client.go

@@ -490,6 +490,30 @@ func (g *githubClient) GithubBaseURL() *url.URL {
 	return g.cli.BaseURL
 }
 
+func NewRateLimitClient(ctx context.Context, credentials params.GithubCredentials) (common.RateLimitClient, error) {
+	httpClient, err := credentials.GetHTTPClient(ctx)
+	if err != nil {
+		return nil, errors.Wrap(err, "fetching http client")
+	}
+
+	slog.DebugContext(
+		ctx, "creating rate limit client",
+		"base_url", credentials.APIBaseURL,
+		"upload_url", credentials.UploadBaseURL)
+
+	ghClient, err := github.NewClient(httpClient).WithEnterpriseURLs(
+		credentials.APIBaseURL, credentials.UploadBaseURL)
+	if err != nil {
+		return nil, errors.Wrap(err, "fetching github client")
+	}
+	cli := &githubClient{
+		rateLimit: ghClient.RateLimit,
+		cli:       ghClient,
+	}
+
+	return cli, nil
+}
+
 func Client(ctx context.Context, entity params.GithubEntity) (common.GithubClient, error) {
 	// func GithubClient(ctx context.Context, entity params.GithubEntity) (common.GithubClient, error) {
 	httpClient, err := entity.Credentials.GetHTTPClient(ctx)

util/github/scalesets/message_sessions.go

@@ -141,16 +141,17 @@ func (m *MessageSession) maybeRefreshToken(ctx context.Context) error {
 	if m.session == nil {
 		return fmt.Errorf("session is nil")
 	}
-	// add some jitter
-	randInt, err := rand.Int(rand.Reader, big.NewInt(5000))
-	if err != nil {
-		return fmt.Errorf("failed to get a random number")
-	}
+
 	expiresAt, err := m.session.ExiresAt()
 	if err != nil {
 		return fmt.Errorf("failed to get expires at: %w", err)
 	}
-	expiresIn := time.Duration(randInt.Int64())*time.Millisecond + 10*time.Minute
+	// add some jitter (30 second interval)
+	randInt, err := rand.Int(rand.Reader, big.NewInt(30))
+	if err != nil {
+		return fmt.Errorf("failed to get a random number")
+	}
+	expiresIn := time.Duration(randInt.Int64())*time.Second + 10*time.Minute
 	slog.DebugContext(ctx, "checking if message session token needs refresh", "expires_at", expiresAt)
 	if m.session.ExpiresIn(expiresIn) {
 		if err := m.Refresh(ctx); err != nil {

workers/entity/worker_watcher.go

@@ -13,32 +13,28 @@ func (w *Worker) handleWorkerWatcherEvent(event dbCommon.ChangePayload) {
 	entityType := dbCommon.DatabaseEntityType(w.Entity.EntityType)
 	switch event.EntityType {
 	case entityType:
-		entityGetter, ok := event.Payload.(params.EntityGetter)
-		if !ok {
-			slog.ErrorContext(w.ctx, "invalid payload for entity type", "entity_type", event.EntityType, "payload", event.Payload)
-			return
-		}
-		entity, err := entityGetter.GetEntity()
-		if err != nil {
-			slog.ErrorContext(w.ctx, "getting entity from repository", "entity_type", event.EntityType, "payload", event.Payload, "error", err)
-			return
-		}
-		w.handleEntityEventPayload(entity, event)
+		w.handleEntityEventPayload(event)
 		return
 	case dbCommon.GithubCredentialsEntityType:
 		slog.DebugContext(w.ctx, "got github credentials payload event")
-		credentials, ok := event.Payload.(params.GithubCredentials)
-		if !ok {
-			slog.ErrorContext(w.ctx, "invalid payload for entity type", "entity_type", event.EntityType, "payload", event.Payload)
-			return
-		}
-		w.handleEntityCredentialsEventPayload(credentials, event)
+		w.handleEntityCredentialsEventPayload(event)
 	default:
 		slog.DebugContext(w.ctx, "invalid entity type; ignoring", "entity_type", event.EntityType)
 	}
 }
 
-func (w *Worker) handleEntityEventPayload(entity params.GithubEntity, event dbCommon.ChangePayload) {
+func (w *Worker) handleEntityEventPayload(event dbCommon.ChangePayload) {
+	entityGetter, ok := event.Payload.(params.EntityGetter)
+	if !ok {
+		slog.ErrorContext(w.ctx, "invalid payload for entity type", "entity_type", event.EntityType, "payload", event.Payload)
+		return
+	}
+	entity, err := entityGetter.GetEntity()
+	if err != nil {
+		slog.ErrorContext(w.ctx, "getting entity from repository", "entity_type", event.EntityType, "payload", event.Payload, "error", err)
+		return
+	}
+
 	switch event.Operation {
 	case dbCommon.UpdateOperation:
 		slog.DebugContext(w.ctx, "got update operation")
@@ -57,7 +53,13 @@ func (w *Worker) handleEntityEventPayload(entity params.GithubEntity, event dbCo
 	}
 }
 
-func (w *Worker) handleEntityCredentialsEventPayload(credentials params.GithubCredentials, event dbCommon.ChangePayload) {
+func (w *Worker) handleEntityCredentialsEventPayload(event dbCommon.ChangePayload) {
+	credentials, ok := event.Payload.(params.GithubCredentials)
+	if !ok {
+		slog.ErrorContext(w.ctx, "invalid payload for entity type", "entity_type", event.EntityType, "payload", event.Payload)
+		return
+	}
+
 	switch event.Operation {
 	case dbCommon.UpdateOperation:
 		slog.DebugContext(w.ctx, "got delete operation")