Merge pull request #86 from gabriel-samfira/add-job-tracking

Add job tracking

apiserver/controllers/controllers.go

@@ -306,3 +306,17 @@ func (a *APIController) ListProviders(w http.ResponseWriter, r *http.Request) {
 		log.Printf("failed to encode response: %q", err)
 	}
 }
+
+func (a *APIController) ListAllJobs(w http.ResponseWriter, r *http.Request) {
+	ctx := r.Context()
+	jobs, err := a.r.ListAllJobs(ctx)
+	if err != nil {
+		handleError(w, err)
+		return
+	}
+
+	w.Header().Set("Content-Type", "application/json")
+	if err := json.NewEncoder(w).Encode(jobs); err != nil {
+		log.Printf("failed to encode response: %q", err)
+	}
+}

apiserver/routers/routers.go

@@ -29,8 +29,10 @@ package routers
 //go:generate swagger generate client --target=../../ --spec=../swagger.yaml
 
 import (
+	_ "expvar" // Register the expvar handlers
 	"io"
 	"net/http"
+	_ "net/http/pprof" // Register the pprof handlers
 
 	"github.com/gorilla/mux"
 	"github.com/prometheus/client_golang/prometheus/promhttp"
@@ -54,6 +56,15 @@ func WithMetricsRouter(parentRouter *mux.Router, disableAuth bool, metricsMiddle
 	return parentRouter
 }
 
+func WithDebugServer(parentRouter *mux.Router) *mux.Router {
+	if parentRouter == nil {
+		return nil
+	}
+
+	parentRouter.PathPrefix("/debug/pprof/").Handler(http.DefaultServeMux)
+	return parentRouter
+}
+
 func NewAPIRouter(han *controllers.APIController, logWriter io.Writer, authMiddleware, initMiddleware, instanceMiddleware auth.Middleware) *mux.Router {
 	router := mux.NewRouter()
 	logMiddleware := util.NewLoggingMiddleware(logWriter)
@@ -94,6 +105,13 @@ func NewAPIRouter(han *controllers.APIController, logWriter io.Writer, authMiddl
 	apiRouter.Handle("/metrics-token/", http.HandlerFunc(han.MetricsTokenHandler)).Methods("GET", "OPTIONS")
 	apiRouter.Handle("/metrics-token", http.HandlerFunc(han.MetricsTokenHandler)).Methods("GET", "OPTIONS")
 
+	//////////
+	// Jobs //
+	//////////
+	// List all jobs
+	apiRouter.Handle("/jobs/", http.HandlerFunc(han.ListAllJobs)).Methods("GET", "OPTIONS")
+	apiRouter.Handle("/jobs", http.HandlerFunc(han.ListAllJobs)).Methods("GET", "OPTIONS")
+
 	///////////
 	// Pools //
 	///////////

cmd/garm-cli/client/client.go

@@ -183,6 +183,19 @@ func (c *Client) DeleteRunner(instanceName string) error {
 	return nil
 }
 
+func (c *Client) ListAllJobs() ([]params.Job, error) {
+	url := fmt.Sprintf("%s/api/v1/jobs", c.Config.BaseURL)
+
+	var response []params.Job
+	resp, err := c.client.R().
+		SetResult(&response).
+		Get(url)
+	if err != nil || resp.IsError() {
+		return response, c.handleError(err, resp)
+	}
+	return response, nil
+}
+
 func (c *Client) ListPoolInstances(poolID string) ([]params.Instance, error) {
 	url := fmt.Sprintf("%s/api/v1/pools/%s/instances", c.Config.BaseURL, poolID)
 

cmd/garm-cli/cmd/jobs.go

@@ -0,0 +1,79 @@
+// Copyright 2023 Cloudbase Solutions SRL
+//
+//    Licensed under the Apache License, Version 2.0 (the "License"); you may
+//    not use this file except in compliance with the License. You may obtain
+//    a copy of the License at
+//
+//         http://www.apache.org/licenses/LICENSE-2.0
+//
+//    Unless required by applicable law or agreed to in writing, software
+//    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+//    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+//    License for the specific language governing permissions and limitations
+//    under the License.
+
+package cmd
+
+import (
+	"fmt"
+	"strings"
+
+	"github.com/cloudbase/garm/params"
+	"github.com/google/uuid"
+	"github.com/jedib0t/go-pretty/v6/table"
+	"github.com/spf13/cobra"
+)
+
+// runnerCmd represents the runner command
+var jobsCmd = &cobra.Command{
+	Use:          "job",
+	SilenceUsage: true,
+	Short:        "Information about jobs",
+	Long:         `Query information about jobs.`,
+	Run:          nil,
+}
+
+var jobsListCmd = &cobra.Command{
+	Use:          "list",
+	Aliases:      []string{"ls"},
+	Short:        "List jobs",
+	Long:         `List all jobs currently recorded in the system.`,
+	SilenceUsage: true,
+	RunE: func(cmd *cobra.Command, args []string) error {
+		if needsInit {
+			return errNeedsInitError
+		}
+
+		jobs, err := cli.ListAllJobs()
+		if err != nil {
+			return err
+		}
+		formatJobs(jobs)
+		return nil
+	},
+}
+
+func formatJobs(jobs []params.Job) {
+	t := table.NewWriter()
+	header := table.Row{"ID", "Name", "Status", "Conclusion", "Runner Name", "Repository", "Requested Labels", "Locked by"}
+	t.AppendHeader(header)
+
+	for _, job := range jobs {
+		lockedBy := ""
+		repo := fmt.Sprintf("%s/%s", job.RepositoryOwner, job.RepositoryName)
+		if job.LockedBy != uuid.Nil {
+			lockedBy = job.LockedBy.String()
+		}
+		t.AppendRow(table.Row{job.ID, job.Name, job.Status, job.Conclusion, job.RunnerName, repo, strings.Join(job.Labels, " "), lockedBy})
+		t.AppendSeparator()
+	}
+	fmt.Println(t.Render())
+}
+
+func init() {
+	jobsCmd.AddCommand(
+		jobsListCmd,
+	)
+
+	rootCmd.AddCommand(jobsCmd)
+}

cmd/garm-cli/cmd/runner.go

@@ -198,11 +198,11 @@ func init() {
 
 func formatInstances(param []params.Instance) {
 	t := table.NewWriter()
-	header := table.Row{"Name", "Status", "Runner Status", "Pool ID"}
+	header := table.Row{"Nr", "Name", "Status", "Runner Status", "Pool ID"}
 	t.AppendHeader(header)
 
-	for _, inst := range param {
-		t.AppendRow(table.Row{inst.Name, inst.Status, inst.RunnerStatus, inst.PoolID})
+	for idx, inst := range param {
+		t.AppendRow(table.Row{idx + 1, inst.Name, inst.Status, inst.RunnerStatus, inst.PoolID})
 		t.AppendSeparator()
 	}
 	fmt.Println(t.Render())

cmd/garm/main.go

@@ -129,7 +129,7 @@ func main() {
 		log.Fatal(err)
 	}
 
-	runner, err := runner.NewRunner(ctx, *cfg)
+	runner, err := runner.NewRunner(ctx, *cfg, db)
 	if err != nil {
 		log.Fatalf("failed to create controller: %+v", err)
 	}
@@ -176,6 +176,11 @@ func main() {
 		router = routers.WithMetricsRouter(router, cfg.Metrics.DisableAuth, metricsMiddleware)
 	}
 
+	if cfg.Default.DebugServer {
+		log.Printf("setting up debug routes")
+		router = routers.WithDebugServer(router)
+	}
+
 	corsMw := mux.CORSMethodMiddleware(router)
 	router.Use(corsMw)
 

config/config.go

@@ -120,6 +120,7 @@ type Default struct {
 	// LogFile is the location of the log file.
 	LogFile           string `toml:"log_file,omitempty" json:"log-file"`
 	EnableLogStreamer bool   `toml:"enable_log_streamer"`
+	DebugServer       bool   `toml:"debug_server" json:"debug-server"`
 }
 
 func (d *Default) Validate() error {
@@ -337,7 +338,7 @@ func (s *SQLite) Validate() error {
 }
 
 func (s *SQLite) ConnectionString() (string, error) {
-	return s.DBFile, nil
+	return fmt.Sprintf("%s?_journal_mode=WAL&_foreign_keys=ON", s.DBFile), nil
 }
 
 // MySQL is the config entry for the mysql section

config/config_test.go

@@ -412,7 +412,7 @@ func TestGormParams(t *testing.T) {
 	dbType, uri, err := cfg.GormParams()
 	require.Nil(t, err)
 	require.Equal(t, SQLiteBackend, dbType)
-	require.Equal(t, filepath.Join(dir, "garm.db"), uri)
+	require.Equal(t, filepath.Join(dir, "garm.db?_journal_mode=WAL&_foreign_keys=ON"), uri)
 
 	cfg.DbBackend = MySQLBackend
 	cfg.MySQL = getMySQLDefaultConfig()

database/common/common.go

@@ -86,6 +86,7 @@ type PoolStore interface {
 
 	PoolInstanceCount(ctx context.Context, poolID string) (int64, error)
 	GetPoolInstanceByName(ctx context.Context, poolID string, instanceName string) (params.Instance, error)
+	FindPoolsMatchingAllTags(ctx context.Context, entityType params.PoolType, entityID string, tags []string) ([]params.Pool, error)
 }
 
 type UserStore interface {
@@ -111,6 +112,21 @@ type InstanceStore interface {
 	ListInstanceEvents(ctx context.Context, instanceID string, eventType params.EventType, eventLevel params.EventLevel) ([]params.StatusMessage, error)
 }
 
+type JobsStore interface {
+	CreateOrUpdateJob(ctx context.Context, job params.Job) (params.Job, error)
+	ListEntityJobsByStatus(ctx context.Context, entityType params.PoolType, entityID string, status params.JobStatus) ([]params.Job, error)
+	ListJobsByStatus(ctx context.Context, status params.JobStatus) ([]params.Job, error)
+	ListAllJobs(ctx context.Context) ([]params.Job, error)
+
+	GetJobByID(ctx context.Context, jobID int64) (params.Job, error)
+	DeleteJob(ctx context.Context, jobID int64) error
+	UnlockJob(ctx context.Context, jobID int64, entityID string) error
+	LockJob(ctx context.Context, jobID int64, entityID string) error
+	BreakLockJobIsQueued(ctx context.Context, jobID int64) error
+
+	DeleteCompletedJobs(ctx context.Context) error
+}
+
 //go:generate mockery --name=Store
 type Store interface {
 	RepoStore
@@ -119,6 +135,7 @@ type Store interface {
 	PoolStore
 	UserStore
 	InstanceStore
+	JobsStore
 
 	ControllerInfo() (params.ControllerInfo, error)
 	InitController() (params.ControllerInfo, error)

database/sql/controller.go

@@ -18,8 +18,8 @@ import (
 	runnerErrors "github.com/cloudbase/garm/errors"
 	"github.com/cloudbase/garm/params"
 
+	"github.com/google/uuid"
 	"github.com/pkg/errors"
-	uuid "github.com/satori/go.uuid"
 	"gorm.io/gorm"
 )
 
@@ -42,7 +42,7 @@ func (s *sqlDatabase) InitController() (params.ControllerInfo, error) {
 		return params.ControllerInfo{}, runnerErrors.NewConflictError("controller already initialized")
 	}
 
-	newID, err := uuid.NewV4()
+	newID, err := uuid.NewUUID()
 	if err != nil {
 		return params.ControllerInfo{}, errors.Wrap(err, "generating UUID")
 	}

database/sql/controller_test.go

@@ -69,5 +69,6 @@ func (s *CtrlTestSuite) TestInitControllerAlreadyInitialized() {
 }
 
 func TestCtrlTestSuite(t *testing.T) {
+	t.Parallel()
 	suite.Run(t, new(CtrlTestSuite))
 }

database/sql/enterprise.go

@@ -7,8 +7,8 @@ import (
 	"github.com/cloudbase/garm/params"
 	"github.com/cloudbase/garm/util"
 
+	"github.com/google/uuid"
 	"github.com/pkg/errors"
-	uuid "github.com/satori/go.uuid"
 	"gorm.io/datatypes"
 	"gorm.io/gorm"
 )
@@ -148,7 +148,7 @@ func (s *sqlDatabase) CreateEnterprisePool(ctx context.Context, enterpriseID str
 		Flavor:                 param.Flavor,
 		OSType:                 param.OSType,
 		OSArch:                 param.OSArch,
-		EnterpriseID:           enterprise.ID,
+		EnterpriseID:           &enterprise.ID,
 		Enabled:                param.Enabled,
 		RunnerBootstrapTimeout: param.RunnerBootstrapTimeout,
 	}
@@ -224,15 +224,15 @@ func (s *sqlDatabase) UpdateEnterprisePool(ctx context.Context, enterpriseID, po
 }
 
 func (s *sqlDatabase) FindEnterprisePoolByTags(ctx context.Context, enterpriseID string, tags []string) (params.Pool, error) {
-	pool, err := s.findPoolByTags(enterpriseID, "enterprise_id", tags)
+	pool, err := s.findPoolByTags(enterpriseID, params.EnterprisePool, tags)
 	if err != nil {
 		return params.Pool{}, errors.Wrap(err, "fetching pool")
 	}
-	return pool, nil
+	return pool[0], nil
 }
 
 func (s *sqlDatabase) ListEnterprisePools(ctx context.Context, enterpriseID string) ([]params.Pool, error) {
-	pools, err := s.getEnterprisePools(ctx, enterpriseID, "Tags", "Enterprise")
+	pools, err := s.listEntityPools(ctx, params.EnterprisePool, enterpriseID, "Tags", "Instances")
 	if err != nil {
 		return nil, errors.Wrap(err, "fetching pools")
 	}
@@ -246,7 +246,7 @@ func (s *sqlDatabase) ListEnterprisePools(ctx context.Context, enterpriseID stri
 }
 
 func (s *sqlDatabase) ListEnterpriseInstances(ctx context.Context, enterpriseID string) ([]params.Instance, error) {
-	pools, err := s.getEnterprisePools(ctx, enterpriseID, "Instances")
+	pools, err := s.listEntityPools(ctx, params.EnterprisePool, enterpriseID, "Instances", "Tags")
 	if err != nil {
 		return nil, errors.Wrap(err, "fetching enterprise")
 	}
@@ -274,7 +274,7 @@ func (s *sqlDatabase) getEnterprise(ctx context.Context, name string) (Enterpris
 }
 
 func (s *sqlDatabase) getEnterpriseByID(ctx context.Context, id string, preload ...string) (Enterprise, error) {
-	u, err := uuid.FromString(id)
+	u, err := uuid.Parse(id)
 	if err != nil {
 		return Enterprise{}, errors.Wrap(runnerErrors.ErrBadRequest, "parsing id")
 	}
@@ -315,28 +315,3 @@ func (s *sqlDatabase) getEnterprisePoolByUniqueFields(ctx context.Context, enter
 
 	return pool[0], nil
 }
-
-func (s *sqlDatabase) getEnterprisePools(ctx context.Context, enterpriseID string, preload ...string) ([]Pool, error) {
-	_, err := s.getEnterpriseByID(ctx, enterpriseID)
-	if err != nil {
-		return nil, errors.Wrap(err, "fetching enterprise")
-	}
-
-	q := s.conn
-	if len(preload) > 0 {
-		for _, item := range preload {
-			q = q.Preload(item)
-		}
-	}
-
-	var pools []Pool
-	err = q.Model(&Pool{}).Where("enterprise_id = ?", enterpriseID).
-		Omit("extra_specs").
-		Find(&pools).Error
-
-	if err != nil {
-		return nil, errors.Wrap(err, "fetching pool")
-	}
-
-	return pools, nil
-}

database/sql/enterprise_test.go

@@ -671,7 +671,7 @@ func (s *EnterpriseTestSuite) TestListEnterprisePoolsInvalidEnterpriseID() {
 	_, err := s.Store.ListEnterprisePools(context.Background(), "dummy-enterprise-id")
 
 	s.Require().NotNil(err)
-	s.Require().Equal("fetching pools: fetching enterprise: parsing id: invalid request", err.Error())
+	s.Require().Equal("fetching pools: parsing id: invalid request", err.Error())
 }
 
 func (s *EnterpriseTestSuite) TestGetEnterprisePool() {
@@ -785,7 +785,7 @@ func (s *EnterpriseTestSuite) TestListEnterpriseInstancesInvalidEnterpriseID() {
 	_, err := s.Store.ListEnterpriseInstances(context.Background(), "dummy-enterprise-id")
 
 	s.Require().NotNil(err)
-	s.Require().Equal("fetching enterprise: fetching enterprise: parsing id: invalid request", err.Error())
+	s.Require().Equal("fetching enterprise: parsing id: invalid request", err.Error())
 }
 
 func (s *EnterpriseTestSuite) TestUpdateEnterprisePool() {
@@ -811,5 +811,6 @@ func (s *EnterpriseTestSuite) TestUpdateEnterprisePoolInvalidEnterpriseID() {
 }
 
 func TestEnterpriseTestSuite(t *testing.T) {
+	t.Parallel()
 	suite.Run(t, new(EnterpriseTestSuite))
 }