`3b6ce7e`

admin/actions: add runner pool ops controls (S41j-4)

Authored by mfwolffe <wolffemf@dukes.jmu.edu> 22 hours ago

SHA: 3b6ce7eba2f1dc4ddb6c6bf44b884a52f3c19c2d
Parents: 6931e9a
Tree: d8dc198

5 changed files

Status	File	+	-
M	`cmd/shithubd/admin_runner.go`	488	24
M	`cmd/shithubd/admin_runner_test.go`	67	0
M	`internal/infra/metrics/actionsobserver.go`	53	5
M	`internal/infra/metrics/actionsobserver_test.go`	24	1
M	`internal/infra/metrics/metrics.go`	47	1

cmd/shithubd/admin_runner.gomodified

  	"text/tabwriter"
  	"time"
++	"github.com/jackc/pgx/v5"
  	"github.com/jackc/pgx/v5/pgtype"
  	"github.com/jackc/pgx/v5/pgxpool"
  	"github.com/spf13/cobra"
  func newAdminRunnerCmd() *cobra.Command {
  	cmd := &cobra.Command{
  		Use:   "runner",
--		Short: "Register, list, and revoke Actions runners",
++		Short: "Register and operate Actions runners",
+ 	}
  	cmd.AddCommand(newAdminRunnerRegisterCmd())
  	cmd.AddCommand(newAdminRunnerListCmd())
  	cmd.AddCommand(newAdminRunnerQueueCmd())
++	cmd.AddCommand(newAdminRunnerDrainCmd())
++	cmd.AddCommand(newAdminRunnerUndrainCmd())
++	cmd.AddCommand(newAdminRunnerRotateTokenCmd())
  	cmd.AddCommand(newAdminRunnerRevokeCmd())
++	cmd.AddCommand(newAdminRunnerCleanupStaleCmd())
  	return cmd
+ }
+ }
  func writeRunnerRegisterOutput(w io.Writer, format string, out runnerRegisterOutput) error {
++	return writeRunnerTokenOutput(w, format, "runner registered", out)
++}
++
++func writeRunnerTokenOutput(w io.Writer, format, heading string, out runnerRegisterOutput) error {
  	if format == "json" {
  		enc := json.NewEncoder(w)
  		enc.SetIndent("", "  ")
  		expires = out.TokenExpiresAt.Format(time.RFC3339)
+ 	}
  	_, err := fmt.Fprintf(w,
--		"runner registered\nid: %d\nname: %s\nlabels: %s\ncapacity: %d\ntoken_expires_at: %s\ntoken: %s\n\nStore this token now; shithub never shows it again.\n",
++		"%s\nid: %d\nname: %s\nlabels: %s\ncapacity: %d\ntoken_expires_at: %s\ntoken: %s\n\nStore this token now; shithub never shows it again.\n",
--		out.ID, out.Name, strings.Join(out.Labels, ","), out.Capacity, expires, out.Token)
++		heading, out.ID, out.Name, strings.Join(out.Labels, ","), out.Capacity, expires, out.Token)
  	return err
+ }
  func newAdminRunnerListCmd() *cobra.Command {
--	return &cobra.Command{
++	var output string
++	cmd := &cobra.Command{
  		Use:   "list",
  		Short: "List registered Actions runners",
  		RunE: func(cmd *cobra.Command, _ []string) error {
++			var err error
++			output, err = normalizeRunnerOutput("admin runner list", output)
++			if err != nil {
++				return err
++			}
  			cfg, err := config.Load(nil)
  			if err != nil {
  				return err
  			if err != nil {
  				return fmt.Errorf("admin runner list: %w", err)
+ 			}
--			tw := tabwriter.NewWriter(cmd.OutOrStdout(), 0, 0, 2, ' ', 0)
++			return writeRunnerListOutput(cmd.OutOrStdout(), output, rows, time.Now().UTC())
--			_, _ = fmt.Fprintln(tw, "ID\tNAME\tSTATUS\tCAPACITY\tLABELS\tLAST_HEARTBEAT")
--			for _, r := range rows {
--				last := "never"
--				if r.LastHeartbeatAt.Valid {
--					last = r.LastHeartbeatAt.Time.Format(time.RFC3339)
--				}
--				_, _ = fmt.Fprintf(tw, "%d\t%s\t%s\t%d\t%s\t%s\n",
--					r.ID, r.Name, r.Status, r.Capacity, strings.Join(r.Labels, ","), last)
--			}
--			return tw.Flush()
  		},
+ 	}
++	cmd.Flags().StringVar(&output, "output", "text", "Output format: text or json")
++	return cmd
++}
++
++type runnerListOutputRow struct {
++	ID                      int64    `json:"id"`
++	Name                    string   `json:"name"`
++	Status                  string   `json:"status"`
++	Capacity                int32    `json:"capacity"`
++	ActiveJobCount          int32    `json:"active_job_count"`
++	Labels                  []string `json:"labels"`
++	HostName                string   `json:"host_name,omitempty"`
++	Version                 string   `json:"version,omitempty"`
++	LastHeartbeatAt         string   `json:"last_heartbeat_at,omitempty"`
++	LastHeartbeatAgeSeconds int64    `json:"last_heartbeat_age_seconds,omitempty"`
++	DrainingAt              string   `json:"draining_at,omitempty"`
++	DrainReason             string   `json:"drain_reason,omitempty"`
++	RevokedAt               string   `json:"revoked_at,omitempty"`
++	RevokedReason           string   `json:"revoked_reason,omitempty"`
++	CreatedAt               string   `json:"created_at,omitempty"`
++}
++
++func writeRunnerListOutput(w io.Writer, format string, rows []actionsdb.ListRunnersRow, now time.Time) error {
++	out := make([]runnerListOutputRow, 0, len(rows))
++	for _, row := range rows {
++		item := runnerListOutputRow{
++			ID:             row.ID,
++			Name:           row.Name,
++			Status:         string(row.Status),
++			Capacity:       row.Capacity,
++			ActiveJobCount: row.ActiveJobCount,
++			Labels:         append([]string{}, row.Labels...),
++			HostName:       row.HostName,
++			Version:        row.Version,
++		}
++		if row.LastHeartbeatAt.Valid {
++			item.LastHeartbeatAt = row.LastHeartbeatAt.Time.UTC().Format(time.RFC3339)
++			if d := now.Sub(row.LastHeartbeatAt.Time); d > 0 {
++				item.LastHeartbeatAgeSeconds = int64(d.Seconds())
++			}
++		}
++		if row.DrainingAt.Valid {
++			item.DrainingAt = row.DrainingAt.Time.UTC().Format(time.RFC3339)
++			item.DrainReason = row.DrainReason
++		}
++		if row.RevokedAt.Valid {
++			item.RevokedAt = row.RevokedAt.Time.UTC().Format(time.RFC3339)
++			item.RevokedReason = row.RevokedReason
++		}
++		if row.CreatedAt.Valid {
++			item.CreatedAt = row.CreatedAt.Time.UTC().Format(time.RFC3339)
++		}
++		out = append(out, item)
++	}
++	if format == "json" {
++		enc := json.NewEncoder(w)
++		enc.SetIndent("", "  ")
++		return enc.Encode(out)
++	}
++
++	tw := tabwriter.NewWriter(w, 0, 0, 2, ' ', 0)
++	_, _ = fmt.Fprintln(tw, "ID\tNAME\tSTATUS\tCAPACITY\tACTIVE\tLABELS\tHOST\tVERSION\tLAST_HEARTBEAT\tDRAINING\tREVOKED")
++	for _, row := range out {
++		last := "never"
++		if row.LastHeartbeatAt != "" {
++			last = row.LastHeartbeatAt
++		}
++		draining := "-"
++		if row.DrainingAt != "" {
++			draining = row.DrainingAt
++		}
++		revoked := "-"
++		if row.RevokedAt != "" {
++			revoked = row.RevokedAt
++		}
++		_, _ = fmt.Fprintf(tw, "%d\t%s\t%s\t%d\t%d\t%s\t%s\t%s\t%s\t%s\t%s\n",
++			row.ID, row.Name, row.Status, row.Capacity, row.ActiveJobCount, strings.Join(row.Labels, ","),
++			emptyDash(row.HostName), emptyDash(row.Version), last, draining, revoked)
++	}
++	return tw.Flush()
+ }
  func newAdminRunnerQueueCmd() *cobra.Command {
  	return tw.Flush()
+ }
++func newAdminRunnerDrainCmd() *cobra.Command {
++	var idRaw string
++	var reason string
++	var output string
++	cmd := &cobra.Command{
++		Use:   "drain --id <id> [--reason <text>]",
++		Short: "Stop an Actions runner from claiming new jobs",
++		RunE: func(cmd *cobra.Command, _ []string) error {
++			id, err := parseRunnerID("admin runner drain", idRaw)
++			if err != nil {
++				return err
++			}
++			output, err = normalizeRunnerOutput("admin runner drain", output)
++			if err != nil {
++				return err
++			}
++			reason, err = normalizeRunnerReason(reason, "operator requested drain")
++			if err != nil {
++				return err
++			}
++			cfg, err := config.Load(nil)
++			if err != nil {
++				return err
++			}
++			ctx, cancel := context.WithTimeout(cmd.Context(), 30*time.Second)
++			defer cancel()
++			pool, err := openAdminRunnerPool(ctx, cfg, "drain")
++			if err != nil {
++				return err
++			}
++			defer pool.Close()
++
++			row, err := actionsdb.New().SetRunnerDraining(ctx, pool, actionsdb.SetRunnerDrainingParams{
++				ID:          id,
++				DrainReason: reason,
++			})
++			if err != nil {
++				if errors.Is(err, pgx.ErrNoRows) {
++					return fmt.Errorf("admin runner drain: runner %d not found or already revoked", id)
++				}
++				return fmt.Errorf("admin runner drain: %w", err)
++			}
++			return writeRunnerStateOutput(cmd.OutOrStdout(), output, "runner draining", runnerStateOutput{
++				ID:          row.ID,
++				Name:        row.Name,
++				Status:      string(row.Status),
++				DrainingAt:  formatOptionalTime(row.DrainingAt),
++				DrainReason: row.DrainReason,
++				RevokedAt:   formatOptionalTime(row.RevokedAt),
++			})
++		},
++	}
++	cmd.Flags().StringVar(&idRaw, "id", "", "Runner id")
++	cmd.Flags().StringVar(&reason, "reason", "", "Drain reason recorded for operators")
++	cmd.Flags().StringVar(&output, "output", "text", "Output format: text or json")
++	return cmd
++}
++
++func newAdminRunnerUndrainCmd() *cobra.Command {
++	var idRaw string
++	var output string
++	cmd := &cobra.Command{
++		Use:   "undrain --id <id>",
++		Short: "Allow a drained Actions runner to claim jobs again",
++		RunE: func(cmd *cobra.Command, _ []string) error {
++			id, err := parseRunnerID("admin runner undrain", idRaw)
++			if err != nil {
++				return err
++			}
++			output, err = normalizeRunnerOutput("admin runner undrain", output)
++			if err != nil {
++				return err
++			}
++			cfg, err := config.Load(nil)
++			if err != nil {
++				return err
++			}
++			ctx, cancel := context.WithTimeout(cmd.Context(), 30*time.Second)
++			defer cancel()
++			pool, err := openAdminRunnerPool(ctx, cfg, "undrain")
++			if err != nil {
++				return err
++			}
++			defer pool.Close()
++
++			row, err := actionsdb.New().ClearRunnerDraining(ctx, pool, id)
++			if err != nil {
++				if errors.Is(err, pgx.ErrNoRows) {
++					return fmt.Errorf("admin runner undrain: runner %d not found or already revoked", id)
++				}
++				return fmt.Errorf("admin runner undrain: %w", err)
++			}
++			return writeRunnerStateOutput(cmd.OutOrStdout(), output, "runner undrained", runnerStateOutput{
++				ID:          row.ID,
++				Name:        row.Name,
++				Status:      string(row.Status),
++				DrainingAt:  formatOptionalTime(row.DrainingAt),
++				DrainReason: row.DrainReason,
++				RevokedAt:   formatOptionalTime(row.RevokedAt),
++			})
++		},
++	}
++	cmd.Flags().StringVar(&idRaw, "id", "", "Runner id")
++	cmd.Flags().StringVar(&output, "output", "text", "Output format: text or json")
++	return cmd
++}
++
++func newAdminRunnerRotateTokenCmd() *cobra.Command {
++	var idRaw string
++	var output string
++	var expiresIn time.Duration
++	cmd := &cobra.Command{
++		Use:   "rotate-token --id <id>",
++		Short: "Revoke existing registration tokens and print one replacement token",
++		RunE: func(cmd *cobra.Command, _ []string) error {
++			id, err := parseRunnerID("admin runner rotate-token", idRaw)
++			if err != nil {
++				return err
++			}
++			output, err = normalizeRunnerOutput("admin runner rotate-token", output)
++			if err != nil {
++				return err
++			}
++			if expiresIn < 0 {
++				return errors.New("admin runner rotate-token: --expires-in must be non-negative")
++			}
++			cfg, err := config.Load(nil)
++			if err != nil {
++				return err
++			}
++			ctx, cancel := context.WithTimeout(cmd.Context(), 30*time.Second)
++			defer cancel()
++			pool, err := openAdminRunnerPool(ctx, cfg, "rotate-token")
++			if err != nil {
++				return err
++			}
++			defer pool.Close()
++
++			token, tokenHash, err := runnertoken.New()
++			if err != nil {
++				return fmt.Errorf("admin runner rotate-token: mint token: %w", err)
++			}
++			var expiresAt pgtype.Timestamptz
++			var outputExpiresAt *time.Time
++			if expiresIn > 0 {
++				t := time.Now().UTC().Add(expiresIn)
++				expiresAt = pgtype.Timestamptz{Time: t, Valid: true}
++				outputExpiresAt = &t
++			}
++
++			q := actionsdb.New()
++			tx, err := pool.Begin(ctx)
++			if err != nil {
++				return fmt.Errorf("admin runner rotate-token: begin: %w", err)
++			}
++			committed := false
++			defer func() {
++				if !committed {
++					_ = tx.Rollback(ctx)
++				}
++			}()
++			runner, err := q.LockRunnerByID(ctx, tx, id)
++			if err != nil {
++				if errors.Is(err, pgx.ErrNoRows) {
++					return fmt.Errorf("admin runner rotate-token: runner %d not found", id)
++				}
++				return fmt.Errorf("admin runner rotate-token: lock runner: %w", err)
++			}
++			if runner.RevokedAt.Valid {
++				return fmt.Errorf("admin runner rotate-token: runner %d is revoked", id)
++			}
++			if err := q.RevokeAllTokensForRunner(ctx, tx, id); err != nil {
++				return fmt.Errorf("admin runner rotate-token: revoke old tokens: %w", err)
++			}
++			if _, err := q.InsertRunnerToken(ctx, tx, actionsdb.InsertRunnerTokenParams{
++				RunnerID:  runner.ID,
++				TokenHash: tokenHash,
++				ExpiresAt: expiresAt,
++			}); err != nil {
++				return fmt.Errorf("admin runner rotate-token: insert token: %w", err)
++			}
++			if err := tx.Commit(ctx); err != nil {
++				return fmt.Errorf("admin runner rotate-token: commit: %w", err)
++			}
++			committed = true
++
++			return writeRunnerTokenOutput(cmd.OutOrStdout(), output, "runner token rotated", runnerRegisterOutput{
++				ID:             runner.ID,
++				Name:           runner.Name,
++				Labels:         runner.Labels,
++				Capacity:       runner.Capacity,
++				Token:          token,
++				TokenExpiresAt: outputExpiresAt,
++			})
++		},
++	}
++	cmd.Flags().StringVar(&idRaw, "id", "", "Runner id")
++	cmd.Flags().DurationVar(&expiresIn, "expires-in", 0, "Registration token lifetime (0 means no expiration)")
++	cmd.Flags().StringVar(&output, "output", "text", "Output format: text or json")
++	return cmd
++}
++
  func newAdminRunnerRevokeCmd() *cobra.Command {
  	var idRaw string
++	var reason string
++	var output string
  	cmd := &cobra.Command{
  		Use:   "revoke --id <id>",
--		Short: "Revoke all registration tokens for an Actions runner",
++		Short: "Hard-revoke an Actions runner and all registration tokens",
  		RunE: func(cmd *cobra.Command, _ []string) error {
--			id, err := strconv.ParseInt(strings.TrimSpace(idRaw), 10, 64)
++			id, err := parseRunnerID("admin runner revoke", idRaw)
--			if err != nil || id <= 0 {
++			if err != nil {
--				return errors.New("admin runner revoke: --id must be a positive integer")
++				return err
++			}
++			output, err = normalizeRunnerOutput("admin runner revoke", output)
++			if err != nil {
++				return err
++			}
++			reason, err = normalizeRunnerReason(reason, "operator requested revoke")
++			if err != nil {
++				return err
+ 			}
  			cfg, err := config.Load(nil)
  			if err != nil {
  			defer pool.Close()
  			q := actionsdb.New()
--			runner, err := q.GetRunnerByID(ctx, pool, id)
++			tx, err := pool.Begin(ctx)
  			if err != nil {
--				return fmt.Errorf("admin runner revoke: runner %d not found", id)
++				return fmt.Errorf("admin runner revoke: begin: %w", err)
+ 			}
--			if err := q.RevokeAllTokensForRunner(ctx, pool, id); err != nil {
++			committed := false
++			defer func() {
++				if !committed {
++					_ = tx.Rollback(ctx)
++				}
++			}()
++			runner, err := q.RevokeRunner(ctx, tx, actionsdb.RevokeRunnerParams{
++				ID:            id,
++				RevokedReason: reason,
++			})
++			if err != nil {
++				if errors.Is(err, pgx.ErrNoRows) {
++					return fmt.Errorf("admin runner revoke: runner %d not found", id)
++				}
  				return fmt.Errorf("admin runner revoke: %w", err)
+ 			}
--			_, _ = fmt.Fprintf(cmd.OutOrStdout(), "runner revoked\nid: %d\nname: %s\n", runner.ID, runner.Name)
++			if err := q.RevokeAllTokensForRunner(ctx, tx, id); err != nil {
--			return nil
++				return fmt.Errorf("admin runner revoke: revoke tokens: %w", err)
++			}
++			if err := tx.Commit(ctx); err != nil {
++				return fmt.Errorf("admin runner revoke: commit: %w", err)
++			}
++			committed = true
++			metrics.ActionsRunnerRevocationsTotal.Inc()
++			return writeRunnerStateOutput(cmd.OutOrStdout(), output, "runner revoked", runnerStateOutput{
++				ID:            runner.ID,
++				Name:          runner.Name,
++				Status:        string(runner.Status),
++				DrainingAt:    formatOptionalTime(runner.DrainingAt),
++				DrainReason:   runner.DrainReason,
++				RevokedAt:     formatOptionalTime(runner.RevokedAt),
++				RevokedReason: runner.RevokedReason,
++			})
  		},
+ 	}
  	cmd.Flags().StringVar(&idRaw, "id", "", "Runner id")
++	cmd.Flags().StringVar(&reason, "reason", "", "Revocation reason recorded for operators")
++	cmd.Flags().StringVar(&output, "output", "text", "Output format: text or json")
++	return cmd
++}
++
++func newAdminRunnerCleanupStaleCmd() *cobra.Command {
++	var olderThan time.Duration
++	var output string
++	cmd := &cobra.Command{
++		Use:   "cleanup-stale",
++		Short: "Mark stale non-revoked runners offline",
++		RunE: func(cmd *cobra.Command, _ []string) error {
++			var err error
++			output, err = normalizeRunnerOutput("admin runner cleanup-stale", output)
++			if err != nil {
++				return err
++			}
++			if olderThan <= 0 {
++				return errors.New("admin runner cleanup-stale: --older-than must be positive")
++			}
++			cfg, err := config.Load(nil)
++			if err != nil {
++				return err
++			}
++			ctx, cancel := context.WithTimeout(cmd.Context(), 30*time.Second)
++			defer cancel()
++			pool, err := openAdminRunnerPool(ctx, cfg, "cleanup-stale")
++			if err != nil {
++				return err
++			}
++			defer pool.Close()
++
++			cutoff := time.Now().UTC().Add(-olderThan)
++			rows, err := actionsdb.New().MarkStaleRunnersOffline(ctx, pool, pgtype.Timestamptz{Time: cutoff, Valid: true})
++			if err != nil {
++				return fmt.Errorf("admin runner cleanup-stale: %w", err)
++			}
++			return writeRunnerCleanupOutput(cmd.OutOrStdout(), output, rows)
++		},
++	}
++	cmd.Flags().DurationVar(&olderThan, "older-than", 2*time.Minute, "Heartbeat age after which non-revoked runners are marked offline")
++	cmd.Flags().StringVar(&output, "output", "text", "Output format: text or json")
  	return cmd
+ }
++type runnerStateOutput struct {
++	ID            int64  `json:"id"`
++	Name          string `json:"name"`
++	Status        string `json:"status"`
++	DrainingAt    string `json:"draining_at,omitempty"`
++	DrainReason   string `json:"drain_reason,omitempty"`
++	RevokedAt     string `json:"revoked_at,omitempty"`
++	RevokedReason string `json:"revoked_reason,omitempty"`
++}
++
++func writeRunnerStateOutput(w io.Writer, format, heading string, out runnerStateOutput) error {
++	if format == "json" {
++		enc := json.NewEncoder(w)
++		enc.SetIndent("", "  ")
++		return enc.Encode(out)
++	}
++	_, err := fmt.Fprintf(w, "%s\nid: %d\nname: %s\nstatus: %s\ndraining_at: %s\ndrain_reason: %s\nrevoked_at: %s\nrevoked_reason: %s\n",
++		heading, out.ID, out.Name, out.Status, emptyDash(out.DrainingAt), emptyDash(out.DrainReason),
++		emptyDash(out.RevokedAt), emptyDash(out.RevokedReason))
++	return err
++}
++
++type runnerCleanupOutputRow struct {
++	ID              int64  `json:"id"`
++	Name            string `json:"name"`
++	Status          string `json:"status"`
++	LastHeartbeatAt string `json:"last_heartbeat_at,omitempty"`
++}
++
++func writeRunnerCleanupOutput(w io.Writer, format string, rows []actionsdb.MarkStaleRunnersOfflineRow) error {
++	out := make([]runnerCleanupOutputRow, 0, len(rows))
++	for _, row := range rows {
++		out = append(out, runnerCleanupOutputRow{
++			ID:              row.ID,
++			Name:            row.Name,
++			Status:          string(row.Status),
++			LastHeartbeatAt: formatOptionalTime(row.LastHeartbeatAt),
++		})
++	}
++	if format == "json" {
++		enc := json.NewEncoder(w)
++		enc.SetIndent("", "  ")
++		return enc.Encode(out)
++	}
++	tw := tabwriter.NewWriter(w, 0, 0, 2, ' ', 0)
++	_, _ = fmt.Fprintln(tw, "ID\tNAME\tSTATUS\tLAST_HEARTBEAT")
++	for _, row := range out {
++		_, _ = fmt.Fprintf(tw, "%d\t%s\t%s\t%s\n", row.ID, row.Name, row.Status, emptyDash(row.LastHeartbeatAt))
++	}
++	return tw.Flush()
++}
++
  func openAdminRunnerPool(ctx context.Context, cfg config.Config, op string) (*pgxpool.Pool, error) {
  	if cfg.DB.URL == "" {
  		return nil, fmt.Errorf("admin runner %s: DB not configured (set SHITHUB_DATABASE_URL)", op)
  	return runnerlabels.ParseCSV(raw)
+ }
++func parseRunnerID(op, raw string) (int64, error) {
++	id, err := strconv.ParseInt(strings.TrimSpace(raw), 10, 64)
++	if err != nil || id <= 0 {
++		return 0, fmt.Errorf("%s: --id must be a positive integer", op)
++	}
++	return id, nil
++}
++
++func normalizeRunnerOutput(op, output string) (string, error) {
++	output = strings.ToLower(strings.TrimSpace(output))
++	switch output {
++	case "", "text":
++		return "text", nil
++	case "json":
++		return "json", nil
++	default:
++		return "", fmt.Errorf("%s: --output must be text or json", op)
++	}
++}
++
++func normalizeRunnerReason(reason, fallback string) (string, error) {
++	reason = strings.TrimSpace(reason)
++	if reason == "" {
++		reason = fallback
++	}
++	if len(reason) > 1000 {
++		return "", errors.New("runner reason must be 1000 bytes or fewer")
++	}
++	return reason, nil
++}
++
++func formatOptionalTime(t pgtype.Timestamptz) string {
++	if !t.Valid {
++		return ""
++	}
++	return t.Time.UTC().Format(time.RFC3339)
++}
++
++func emptyDash(value string) string {
++	if strings.TrimSpace(value) == "" {
++		return "-"
++	}
++	return value
++}
++
  func init() {
  	adminCmd.AddCommand(newAdminRunnerCmd())
  	adminActionsCmd.AddCommand(newAdminRunnerCmd())

cmd/shithubd/admin_runner_test.gomodified

  		t.Fatalf("oldest seconds=%d", got[0].OldestQueuedSeconds)
+ 	}
+ }
++
++func TestWriteRunnerListOutputJSONIncludesOpsFields(t *testing.T) {
++	now := time.Date(2026, 5, 12, 16, 30, 0, 0, time.UTC)
++	rows := []actionsdb.ListRunnersRow{{
++		ID:              7,
++		Name:            "runner-a",
++		Labels:          []string{"self-hosted", "linux"},
++		Capacity:        2,
++		Status:          actionsdb.WorkflowRunnerStatusBusy,
++		LastHeartbeatAt: pgtype.Timestamptz{Time: now.Add(-75 * time.Second), Valid: true},
++		HostName:        "host-a",
++		Version:         "dev-test",
++		DrainingAt:      pgtype.Timestamptz{Time: now.Add(-30 * time.Second), Valid: true},
++		DrainReason:     "maintenance",
++		ActiveJobCount:  1,
++	}}
++	var buf bytes.Buffer
++	if err := writeRunnerListOutput(&buf, "json", rows, now); err != nil {
++		t.Fatalf("writeRunnerListOutput: %v", err)
++	}
++	var got []runnerListOutputRow
++	if err := json.Unmarshal(buf.Bytes(), &got); err != nil {
++		t.Fatalf("json.Unmarshal: %v", err)
++	}
++	if len(got) != 1 {
++		t.Fatalf("rows=%d body=%s", len(got), buf.String())
++	}
++	if got[0].HostName != "host-a" || got[0].Version != "dev-test" ||
++		got[0].ActiveJobCount != 1 || got[0].LastHeartbeatAgeSeconds != 75 ||
++		got[0].DrainingAt == "" || got[0].DrainReason != "maintenance" {
++		t.Fatalf("unexpected row: %+v", got[0])
++	}
++}
++
++func TestWriteRunnerStateOutputText(t *testing.T) {
++	var buf bytes.Buffer
++	if err := writeRunnerStateOutput(&buf, "text", "runner draining", runnerStateOutput{
++		ID:          7,
++		Name:        "runner-a",
++		Status:      "busy",
++		DrainingAt:  "2026-05-12T16:30:00Z",
++		DrainReason: "maintenance",
++	}); err != nil {
++		t.Fatalf("writeRunnerStateOutput: %v", err)
++	}
++	body := buf.String()
++	for _, want := range []string{
++		"runner draining",
++		"id: 7",
++		"name: runner-a",
++		"drain_reason: maintenance",
++	} {
++		if !strings.Contains(body, want) {
++			t.Fatalf("text output missing %q in %s", want, body)
++		}
++	}
++}
++
++func TestParseRunnerIDRejectsInvalid(t *testing.T) {
++	for _, raw := range []string{"", "0", "-1", "abc"} {
++		t.Run(raw, func(t *testing.T) {
++			if _, err := parseRunnerID("admin runner test", raw); err == nil {
++				t.Fatal("parseRunnerID returned nil error")
++			}
++		})
++	}
++}

internal/infra/metrics/actionsobserver.gomodified

  	"github.com/jackc/pgx/v5/pgxpool"
+ )
++const actionsRunnerStaleAfter = 60 * time.Second
++
  // ObserveActions starts a goroutine that periodically refreshes DB-backed
  // Actions gauges. The goroutine exits when ctx is canceled.
  func ObserveActions(ctx context.Context, pool *pgxpool.Pool, interval time.Duration) {
  func refreshActionQueueGauges(ctx context.Context, pool *pgxpool.Pool) {
  	ActionsQueueDepth.WithLabelValues("runs").Set(0)
  	ActionsQueueDepth.WithLabelValues("jobs").Set(0)
++	ActionsQueueDepthByLabels.Reset()
  	ActionsActive.WithLabelValues("runs").Set(0)
  	ActionsActive.WithLabelValues("jobs").Set(0)
  			ActionsActive.WithLabelValues(resource).Set(count)
+ 		}
+ 	}
++	rows.Close()
++
++	labelRows, err := pool.Query(ctx, `
++SELECT COALESCE(NULLIF(runs_on, ''), '(none)')::text AS labels,
++       count(*)::double precision
++FROM workflow_jobs
++WHERE status = 'queued'
++  AND cancel_requested = false
++  AND runner_id IS NULL
++GROUP BY COALESCE(NULLIF(runs_on, ''), '(none)')`)
++	if err != nil {
++		return
++	}
++	defer labelRows.Close()
++	for labelRows.Next() {
++		var labels string
++		var count float64
++		if err := labelRows.Scan(&labels, &count); err != nil {
++			return
++		}
++		ActionsQueueDepthByLabels.WithLabelValues(labels).Set(count)
++	}
+ }
  func refreshActionRunnerGauges(ctx context.Context, pool *pgxpool.Pool) {
  	ActionsRunnerHeartbeatAgeSeconds.Reset()
++	ActionsRunnerOnline.Reset()
++	ActionsRunnerDraining.Reset()
  	ActionsRunnerCapacity.Reset()
++	ActionsRunnerStaleTotal.Set(0)
  	rows, err := pool.Query(ctx, `
  SELECT name::text,
         status::text,
         capacity::double precision,
--       EXTRACT(EPOCH FROM (now() - last_heartbeat_at))::double precision AS heartbeat_age_seconds
++       COALESCE(EXTRACT(EPOCH FROM (now() - last_heartbeat_at))::double precision, -1) AS heartbeat_age_seconds,
--FROM workflow_runners
++       (draining_at IS NOT NULL)::boolean AS draining,
--WHERE last_heartbeat_at IS NOT NULL`)
++       (revoked_at IS NOT NULL)::boolean AS revoked
++FROM workflow_runners`)
  	if err != nil {
  		return
+ 	}
  	defer rows.Close()
++	var stale float64
  	for rows.Next() {
  		var name, status string
  		var capacity, age float64
--		if err := rows.Scan(&name, &status, &capacity, &age); err != nil {
++		var draining, revoked bool
++		if err := rows.Scan(&name, &status, &capacity, &age, &draining, &revoked); err != nil {
  			return
+ 		}
  		ActionsRunnerCapacity.WithLabelValues(name, status).Set(capacity)
--		ActionsRunnerHeartbeatAgeSeconds.WithLabelValues(name, status).Set(age)
++		if age >= 0 {
++			ActionsRunnerHeartbeatAgeSeconds.WithLabelValues(name, status).Set(age)
++		}
++		online := !revoked && status != "offline" && age >= 0 && age <= actionsRunnerStaleAfter.Seconds()
++		if online {
++			ActionsRunnerOnline.WithLabelValues(name).Set(1)
++		} else {
++			ActionsRunnerOnline.WithLabelValues(name).Set(0)
++		}
++		if draining {
++			ActionsRunnerDraining.WithLabelValues(name).Set(1)
++		} else {
++			ActionsRunnerDraining.WithLabelValues(name).Set(0)
++		}
++		if !revoked && status != "offline" && age > actionsRunnerStaleAfter.Seconds() {
++			stale++
++		}
+ 	}
++	ActionsRunnerStaleTotal.Set(stale)
+ }
  func refreshActionStorageGauges(ctx context.Context, pool *pgxpool.Pool) {

internal/infra/metrics/actionsobserver_test.gomodified

  		JobKey:         "build",
  		JobName:        "Build",
  		RunsOn:         `["ubuntu-latest"]`,
++		NeedsJobs:      []string{},
  		TimeoutMinutes: 30,
  		Permissions:    []byte(`{}`),
  		JobEnv:         []byte(`{}`),
  	if err != nil {
  		t.Fatalf("InsertRunner: %v", err)
  	}
--	if _, err := pool.Exec(ctx, `UPDATE workflow_runners SET status = 'busy', last_heartbeat_at = now() - interval '75 seconds' WHERE id = $1`, runner.ID); err != nil {
++	if _, err := pool.Exec(ctx, `UPDATE workflow_runners SET status = 'busy', last_heartbeat_at = now() - interval '75 seconds', draining_at = now(), drain_reason = 'maintenance' WHERE id = $1`, runner.ID); err != nil {
  		t.Fatalf("touch runner heartbeat: %v", err)
  	}
  	assertGauge(t, ActionsQueueDepth, []string{"runs"}, 1)
  	assertGauge(t, ActionsQueueDepth, []string{"jobs"}, 1)
++	assertGauge(t, ActionsQueueDepthByLabels, []string{`["ubuntu-latest"]`}, 1)
  	assertGauge(t, ActionsActive, []string{"runs"}, 0)
  	assertGauge(t, ActionsActive, []string{"jobs"}, 0)
  	assertGauge(t, ActionsRunnerCapacity, []string{"runner-a", "busy"}, 3)
++	assertGauge(t, ActionsRunnerOnline, []string{"runner-a"}, 0)
++	assertGauge(t, ActionsRunnerDraining, []string{"runner-a"}, 1)
++	assertPlainGauge(t, ActionsRunnerStaleTotal, 1)
  	if got := gaugeValue(t, ActionsRunnerHeartbeatAgeSeconds, []string{"runner-a", "busy"}); got < 60 {
  		t.Fatalf("runner heartbeat age = %v, want >= 60", got)
  	}
  func resetActionsObserverGauges() {
  	ActionsQueueDepth.Reset()
++	ActionsQueueDepthByLabels.Reset()
  	ActionsActive.Reset()
  	ActionsRunnerHeartbeatAgeSeconds.Reset()
++	ActionsRunnerOnline.Reset()
++	ActionsRunnerDraining.Reset()
++	ActionsRunnerStaleTotal.Set(0)
  	ActionsRunnerCapacity.Reset()
  	ActionsStorageObjects.Reset()
  	ActionsStorageBytes.Reset()
  	}
  	return metric.Gauge.GetValue()
  }
++
++func assertPlainGauge(t *testing.T, gauge prometheus.Gauge, want float64) {
++	t.Helper()
++	var metric dto.Metric
++	if err := gauge.Write(&metric); err != nil {
++		t.Fatalf("read gauge: %v", err)
++	}
++	if metric.Gauge == nil {
++		t.Fatal("gauge missing")
++	}
++	if got := metric.Gauge.GetValue(); got != want {
++		t.Fatalf("gauge = %v, want %v", got, want)
++	}
++}

internal/infra/metrics/metrics.gomodified

  	ActionsRunnerHeartbeatsTotal = prometheus.NewCounterVec(
  		prometheus.CounterOpts{
  			Name: "shithub_actions_runner_heartbeats_total",
--			Help: "Total runner heartbeats by result (claimed, no_job).",
++			Help: "Total runner heartbeats by result (claimed, no_job, rejected).",
  		},
  		[]string{"result"},
+ 	)
  		},
  		[]string{"resource"},
+ 	)
++	ActionsQueueDepthByLabels = prometheus.NewGaugeVec(
++		prometheus.GaugeOpts{
++			Name: "shithub_actions_queue_depth_by_labels",
++			Help: "Current queued Actions jobs by exact runs-on label expression.",
++		},
++		[]string{"labels"},
++	)
  	ActionsActive = prometheus.NewGaugeVec(
  		prometheus.GaugeOpts{
  			Name: "shithub_actions_active",
  		},
  		[]string{"resource"},
+ 	)
++	ActionsJobClaimLatencySeconds = prometheus.NewHistogram(
++		prometheus.HistogramOpts{
++			Name:    "shithub_actions_job_claim_latency_seconds",
++			Help:    "Seconds between job enqueue and runner claim.",
++			Buckets: prometheus.ExponentialBuckets(0.1, 2.5, 12),
++		},
++	)
  	ActionsRunnerHeartbeatAgeSeconds = prometheus.NewGaugeVec(
  		prometheus.GaugeOpts{
  			Name: "shithub_actions_runner_heartbeat_age_seconds",
  		},
  		[]string{"runner", "status"},
+ 	)
++	ActionsRunnerOnline = prometheus.NewGaugeVec(
++		prometheus.GaugeOpts{
++			Name: "shithub_actions_runner_online",
++			Help: "Current runner online state by runner (1 online, 0 unavailable).",
++		},
++		[]string{"runner"},
++	)
++	ActionsRunnerStaleTotal = prometheus.NewGauge(
++		prometheus.GaugeOpts{
++			Name: "shithub_actions_runner_stale_total",
++			Help: "Current count of non-revoked runners whose heartbeat is past the stale threshold.",
++		},
++	)
++	ActionsRunnerDraining = prometheus.NewGaugeVec(
++		prometheus.GaugeOpts{
++			Name: "shithub_actions_runner_draining",
++			Help: "Current runner drain state by runner (1 draining, 0 not draining).",
++		},
++		[]string{"runner"},
++	)
  	ActionsRunnerCapacity = prometheus.NewGaugeVec(
  		prometheus.GaugeOpts{
  			Name: "shithub_actions_runner_capacity",
  		},
  		[]string{"runner", "status"},
+ 	)
++	ActionsRunnerRevocationsTotal = prometheus.NewCounter(
++		prometheus.CounterOpts{
++			Name: "shithub_actions_runner_revocations_total",
++			Help: "Total Actions runner hard revocations performed by operator tooling.",
++		},
++	)
  	ActionsStorageObjects = prometheus.NewGaugeVec(
  		prometheus.GaugeOpts{
  			Name: "shithub_actions_storage_objects",
  		ActionsRunsPrunedTotal,
  		ActionsStepTimeoutsTotal,
  		ActionsQueueDepth,
++		ActionsQueueDepthByLabels,
  		ActionsActive,
++		ActionsJobClaimLatencySeconds,
  		ActionsRunnerHeartbeatAgeSeconds,
++		ActionsRunnerOnline,
++		ActionsRunnerStaleTotal,
++		ActionsRunnerDraining,
  		ActionsRunnerCapacity,
++		ActionsRunnerRevocationsTotal,
  		ActionsStorageObjects,
  		ActionsStorageBytes,
+ 	)