patch: Improve listing GitLab API resources

cmd/catchup.go

@@ -11,12 +11,12 @@ import (
 	"sync"
 	"syscall"
 
-	"github.com/cluttrdev/cli"
+	"github.com/alitto/pond"
 	"github.com/oklog/run"
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/prometheus/client_golang/prometheus/collectors"
 
-	"github.com/cluttrdev/gitlab-exporter/pkg/worker"
+	"github.com/cluttrdev/cli"
 
 	"github.com/cluttrdev/gitlab-exporter/internal/config"
 	"github.com/cluttrdev/gitlab-exporter/internal/exporter"
@@ -104,25 +104,11 @@ func (c *CatchUpConfig) Exec(ctx context.Context, args []string) error {
 
 	g := &run.Group{}
 
-	pool := worker.NewWorkerPool(42)
-	{ // worker pool
-		ctx, cancel := context.WithCancel(context.Background())
-
-		g.Add(func() error { // execute
-			slog.Info("Starting worker pool")
-			pool.Start(ctx)
-			<-ctx.Done()
-			return ctx.Err()
-		}, func(err error) { // interrupt
-			defer cancel()
-			slog.Info("Stopping worker pool...")
-			pool.Stop()
-			slog.Info("Stopping worker pool... done")
-		})
-	}
-
 	if len(cfg.Projects) > 0 { // jobs
-		ctx, cancel := context.WithCancel(context.Background())
+		ctxJobs, cancelJobs := context.WithCancel(context.Background())
+
+		slog.Info("Starting worker pool")
+		pool := pond.New(42, 1024, pond.Context(ctxJobs))
 
 		g.Add(func() error { // execute
 			var wg sync.WaitGroup
@@ -149,8 +135,12 @@ func (c *CatchUpConfig) Exec(ctx context.Context, args []string) error {
 			return nil
 		}, func(err error) { // interrupt
 			slog.Info("Cancelling jobs...")
-			cancel()
-			<-ctx.Done()
+			cancelJobs()
+
+			slog.Info("Stopping worker pool...")
+			pool.StopAndWait()
+			slog.Info("Stopping worker pool... done")
+
 			slog.Info("Cancelling jobs... done")
 		})
 	} else {

cmd/fetch_pipeline.go

@@ -89,8 +89,8 @@ func (c *FetchPipelineConfig) Exec(ctx context.Context, args []string) error {
 			FetchSections: c.fetchSections,
 		}
 
-		phr := <-glc.GetPipelineHierarchy(ctx, projectID, pipelineID, opt)
-		if err := phr.Error; err != nil {
+		phr, err := glc.GetPipelineHierarchy(ctx, projectID, pipelineID, opt)
+		if err != nil {
 			return fmt.Errorf("error fetching pipeline hierarchy: %w", err)
 		}
 		ph := phr.PipelineHierarchy

cmd/run.go

@@ -16,6 +16,7 @@ import (
 	"sync"
 	"syscall"
 
+	"github.com/alitto/pond"
 	"github.com/oklog/run"
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/prometheus/client_golang/prometheus/collectors"
@@ -24,8 +25,6 @@ import (
 
 	"github.com/cluttrdev/cli"
 
-	"github.com/cluttrdev/gitlab-exporter/pkg/worker"
-
 	"github.com/cluttrdev/gitlab-exporter/internal/config"
 	"github.com/cluttrdev/gitlab-exporter/internal/exporter"
 	"github.com/cluttrdev/gitlab-exporter/internal/gitlab"
@@ -134,6 +133,7 @@ func (c *RunConfig) Exec(ctx context.Context, _ []string) error {
 	}
 
 	// gather projects from config
+	slog.Info("Resolving projects to export...")
 	projects, err := resolveProjects(ctx, cfg, gitlabclient)
 	if err != nil {
 		return fmt.Errorf("error resolving projects: %w", err)
@@ -155,25 +155,12 @@ func (c *RunConfig) Exec(ctx context.Context, _ []string) error {
 
 	g := &run.Group{}
 
-	pool := worker.NewWorkerPool(42)
-	{ // worker pool
-		ctx, cancel := context.WithCancel(context.Background())
-
-		g.Add(func() error { // execute
-			slog.Info("Starting worker pool")
-			pool.Start(ctx)
-			<-ctx.Done()
-			return ctx.Err()
-		}, func(err error) { // interrupt
-			defer cancel()
-			slog.Info("Stopping worker pool...")
-			pool.Stop()
-			slog.Info("Stopping worker pool... done")
-		})
-	}
-
 	if len(projects) > 0 { // jobs
-		ctx, cancel := context.WithCancel(context.Background())
+		slog.Info(fmt.Sprintf("Found %d projects to export", len(projects)))
+		ctxJobs, cancelJobs := context.WithCancel(context.Background())
+
+		slog.Info("Starting worker pool")
+		pool := pond.New(42, 1024, pond.Context(ctxJobs))
 
 		g.Add(func() error { // execute
 			var wg sync.WaitGroup
@@ -189,7 +176,7 @@ func (c *RunConfig) Exec(ctx context.Context, _ []string) error {
 					wg.Add(1)
 					go func() {
 						defer wg.Done()
-						job.Run(ctx)
+						job.Run(ctxJobs)
 					}()
 				}
 
@@ -203,16 +190,15 @@ func (c *RunConfig) Exec(ctx context.Context, _ []string) error {
 				wg.Add(1)
 				go func() {
 					defer wg.Done()
-					job.Run(ctx)
+					job.Run(ctxJobs)
 				}()
 			}
 
 			wg.Wait()
 			return nil
 		}, func(err error) { // interrupt
 			slog.Info("Cancelling jobs...")
-			cancel()
-			<-ctx.Done()
+			cancelJobs()
 			slog.Info("Cancelling jobs... done")
 		})
 	} else {
@@ -296,43 +282,47 @@ func serveHTTP(cfg config.HTTP, reg *prometheus.Registry) (func() error, func(er
 }
 
 func resolveProjects(ctx context.Context, cfg config.Config, glab *gitlab.Client) ([]config.Project, error) {
-	pm := make(map[int64]config.Project)
+	projectConfigs := make(map[int64]config.Project)
 
 	opt := gitlab.ListNamespaceProjectsOptions{}
 	for _, namespace := range cfg.Namespaces {
 		opt.Kind = namespace.Kind
-		opt.Visibility = (*gitlab.VisibilityValue)(&namespace.Visibility)
+		opt.Visibility = (*_gitlab.VisibilityValue)(&namespace.Visibility)
 		opt.WithShared = namespace.WithShared
 		opt.IncludeSubgroups = namespace.IncludeSubgroups
 
-		ps, err := glab.ListNamespaceProjects(ctx, namespace.Id, opt)
-		if err != nil {
-			return nil, err
-		}
-
-		for _, p := range ps {
-			pm[p.Id] = config.Project{
-				ProjectSettings: namespace.ProjectSettings,
-				Id:              p.Id,
+		err := glab.ListNamespaceProjects(ctx, namespace.Id, opt, func(projects []*_gitlab.Project) bool {
+			for _, project := range projects {
+				projectID := int64(project.ID)
+				projectConfigs[projectID] = config.Project{
+					ProjectSettings: namespace.ProjectSettings,
+					Id:              projectID,
+				}
 			}
-		}
 
-		for _, pid := range namespace.ExcludeProjects {
-			p, _, err := glab.Client().Projects.GetProject(pid, nil, _gitlab.WithContext(ctx))
-			if err != nil {
-				return nil, fmt.Errorf("error getting project %q: %w", pid, err)
+			for _, pid := range namespace.ExcludeProjects {
+				p, _, err := glab.Client().Projects.GetProject(pid, nil, _gitlab.WithContext(ctx))
+				if err != nil {
+					slog.Error("error getting namespace project", "namespace_id", namespace.Id, "project", pid, "error", err)
+					return false
+				}
+				delete(projectConfigs, int64(p.ID))
 			}
-			delete(pm, int64(p.ID))
+
+			return true
+		})
+		if err != nil {
+			return nil, err
 		}
 	}
 
 	// overwrite with explicitly configured projects
 	for _, p := range cfg.Projects {
-		pm[p.Id] = p
+		projectConfigs[p.Id] = p
 	}
 
-	projects := make([]config.Project, 0, len(pm))
-	for _, p := range pm {
+	projects := make([]config.Project, 0, len(projectConfigs))
+	for _, p := range projectConfigs {
 		projects = append(projects, p)
 	}
 

configs/gitlab-exporter.yaml

@@ -38,11 +38,11 @@ project_defaults:
       # requires fetching entire job logs to parse for embedded metrics.
       enabled: true
 
-      note_events: true
-
     mergerequests:
       # Whether or not to export merge request data
       enabled: true
+
+      note_events: true
 
   catch_up:
     # Whether to export data from past pipelines at startup.

go.mod

@@ -3,6 +3,7 @@ module github.com/cluttrdev/gitlab-exporter
 go 1.22
 
 require (
+	github.com/alitto/pond v1.9.2
 	github.com/cluttrdev/cli v0.0.0-20240318153739-b60d492ac8ff
 	github.com/creasty/defaults v1.7.0
 	github.com/google/go-cmp v0.6.0

go.sum

@@ -33,6 +33,8 @@ cloud.google.com/go/storage v1.10.0/go.mod h1:FLPqc6j+Ki4BU591ie1oL6qBQGu2Bl/tZ9
 dmitri.shuralyov.com/gpu/mtl v0.0.0-20190408044501-666a987793e9/go.mod h1:H6x//7gZCb22OMCxBHrMx7a5I7Hp++hsVxbQ4BYO7hU=
 github.com/BurntSushi/toml v0.3.1/go.mod h1:xHWCNGjB5oqiDr8zfno3MHue2Ht5sIBksp03qcyfWMU=
 github.com/BurntSushi/xgb v0.0.0-20160522181843-27f122750802/go.mod h1:IVnqGOEym/WlBOVXweHU+Q+/VP0lqqI8lqeDx9IjBqo=
+github.com/alitto/pond v1.9.2 h1:9Qb75z/scEZVCoSU+osVmQ0I0JOeLfdTDafrbcJ8CLs=
+github.com/alitto/pond v1.9.2/go.mod h1:xQn3P/sHTYcU/1BR3i86IGIrilcrGC2LiS+E2+CJWsI=
 github.com/beorn7/perks v1.0.1 h1:VlbKKnNfV8bJzeqoa4cOKqO6bYr3WgKZxO8Z16+hsOM=
 github.com/beorn7/perks v1.0.1/go.mod h1:G2ZrVWU2WbWT9wwq4/hrbKbnv/1ERSJQ0ibhJ6rlkpw=
 github.com/census-instrumentation/opencensus-proto v0.2.1/go.mod h1:f6KPmirojxKA12rnyqOA5BBL4O983OfeGPqjHWSTneU=