fix: Populate topics from Gitlab

.github/workflows/test.yml

@@ -46,3 +46,5 @@ jobs:
 
       - name: Build
         run: make build
+
+

pkg/config/config.go

@@ -135,6 +135,13 @@ type Redis struct {
 
 // Pull ..
 type Pull struct {
+	// Projects configuration
+	Projects struct {
+		OnInit          bool `default:"true" yaml:"on_init"`
+		Scheduled       bool `default:"true" yaml:"scheduled"`
+		IntervalSeconds int  `default:"1800" validate:"gte=1" yaml:"interval_seconds"`
+	} `yaml:"projects"`
+
 	// ProjectsFromWildcards configuration
 	ProjectsFromWildcards struct {
 		OnInit          bool `default:"true" yaml:"on_init"`

pkg/config/config_test.go

@@ -26,6 +26,10 @@ func TestNew(t *testing.T) {
 	c.Gitlab.BurstableRequestsPerSecond = 5
 	c.Gitlab.MaximumJobsQueueSize = 1000
 
+	c.Pull.Projects.OnInit = true
+	c.Pull.Projects.Scheduled = true
+	c.Pull.Projects.IntervalSeconds = 1800
+
 	c.Pull.ProjectsFromWildcards.OnInit = true
 	c.Pull.ProjectsFromWildcards.Scheduled = true
 	c.Pull.ProjectsFromWildcards.IntervalSeconds = 1800

pkg/controller/controller.go

@@ -55,7 +55,7 @@ func New(ctx context.Context, cfg config.Config, version string) (c Controller,
 	c.TaskController = NewTaskController(ctx, c.Redis, cfg.Gitlab.MaximumJobsQueueSize)
 	c.registerTasks()
 
-	c.Store = store.New(ctx, c.Redis, c.Config.Projects)
+	c.Store = store.New(ctx, c.Redis)
 
 	if err = c.configureGitlab(cfg.Gitlab, version); err != nil {
 		return
@@ -78,6 +78,7 @@ func (c *Controller) registerTasks() {
 		schemas.TaskTypePullEnvironmentsFromProjects: c.TaskHandlerPullEnvironmentsFromProjects,
 		schemas.TaskTypePullMetrics:                  c.TaskHandlerPullMetrics,
 		schemas.TaskTypePullProject:                  c.TaskHandlerPullProject,
+		schemas.TaskTypePullProjects:                 c.TaskHandlerPullProjects,
 		schemas.TaskTypePullProjectsFromWildcard:     c.TaskHandlerPullProjectsFromWildcard,
 		schemas.TaskTypePullProjectsFromWildcards:    c.TaskHandlerPullProjectsFromWildcards,
 		schemas.TaskTypePullRefMetrics:               c.TaskHandlerPullRefMetrics,

pkg/controller/environments_test.go

@@ -47,7 +47,7 @@ func TestPullEnvironmentsFromProject(t *testing.T) {
 }`)
 		})
 
-	p := schemas.NewProject("foo")
+	p := schemas.NewProject("foo", []string{})
 	p.Pull.Environments.Regexp = "^prod"
 	assert.NoError(t, c.PullEnvironmentsFromProject(ctx, p))
 

pkg/controller/garbage_collector.go

@@ -70,7 +70,7 @@ func (c *Controller) GarbageCollectEnvironments(ctx context.Context) error {
 	envProjects := make(map[schemas.Project]bool)
 
 	for _, env := range storedEnvironments {
-		p := schemas.NewProject(env.ProjectName)
+		p := schemas.NewProject(env.ProjectName, []string{})
 
 		projectExists, err := c.Store.ProjectExists(ctx, p.Key())
 		if err != nil {
@@ -296,7 +296,7 @@ func (c *Controller) GarbageCollectMetrics(ctx context.Context) error {
 
 		if metricLabelRefExists && !metricLabelEnvironmentExists {
 			refKey := schemas.NewRef(
-				schemas.NewProject(metricLabelProject),
+				schemas.NewProject(metricLabelProject, []string{}),
 				schemas.RefKind(m.Labels["kind"]),
 				metricLabelRef,
 			).Key()

pkg/controller/garbage_collector_test.go

@@ -14,10 +14,10 @@ import (
 )
 
 func TestGarbageCollectProjects(t *testing.T) {
-	p1 := schemas.NewProject("cfg/p1")
-	p2 := schemas.NewProject("cfg/p2")
-	p3 := schemas.NewProject("wc/p3")
-	p4 := schemas.NewProject("wc/p4")
+	p1 := schemas.NewProject("cfg/p1", []string{})
+	p2 := schemas.NewProject("cfg/p2", []string{})
+	p3 := schemas.NewProject("wc/p3", []string{})
+	p4 := schemas.NewProject("wc/p4", []string{})
 
 	ctx, c, mux, srv := newTestController(config.Config{
 		Projects: []config.Project{p1.Project},
@@ -62,7 +62,7 @@ func TestGarbageCollectEnvironments(t *testing.T) {
 			fmt.Fprint(w, `[{"name": "main"}]`)
 		})
 
-	p2 := schemas.NewProject("p2")
+	p2 := schemas.NewProject("p2", []string{})
 	p2.Pull.Environments.Enabled = true
 	p2.Pull.Environments.Regexp = "^main$"
 
@@ -103,10 +103,10 @@ func TestGarbageCollectRefs(t *testing.T) {
 			fmt.Fprint(w, `[{"name": "main"}]`)
 		})
 
-	pr1dev := schemas.NewRef(schemas.NewProject("p1"), schemas.RefKindBranch, "dev")
-	pr1main := schemas.NewRef(schemas.NewProject("p1"), schemas.RefKindBranch, "main")
+	pr1dev := schemas.NewRef(schemas.NewProject("p1", []string{}), schemas.RefKindBranch, "dev")
+	pr1main := schemas.NewRef(schemas.NewProject("p1", []string{}), schemas.RefKindBranch, "main")
 
-	p2 := schemas.NewProject("p2")
+	p2 := schemas.NewProject("p2", []string{})
 	p2.Pull.Environments.Regexp = "^main$"
 
 	pr2dev := schemas.NewRef(p2, schemas.RefKindBranch, "dev")
@@ -133,7 +133,7 @@ func TestGarbageCollectMetrics(t *testing.T) {
 	ctx, c, _, srv := newTestController(config.Config{})
 	srv.Close()
 
-	p1 := schemas.NewProject("p1")
+	p1 := schemas.NewProject("p1", []string{})
 	p1.Pull.Pipeline.Jobs.Enabled = true
 
 	ref1 := schemas.NewRef(p1, schemas.RefKindBranch, "foo")

pkg/controller/jobs_test.go

@@ -20,7 +20,7 @@ func TestPullRefPipelineJobsMetrics(t *testing.T) {
 			fmt.Fprint(w, `[{"id":1,"created_at":"2016-08-11T11:28:34.085Z","started_at":"2016-08-11T11:28:56.085Z"},{"id":2,"created_at":"2016-08-11T11:28:34.085Z","started_at":"2016-08-11T11:28:58.085Z"}]`)
 		})
 
-	p := schemas.NewProject("foo")
+	p := schemas.NewProject("foo", []string{})
 	p.Pull.Pipeline.Jobs.FromChildPipelines.Enabled = false
 
 	ref := schemas.NewRef(p, schemas.RefKindBranch, "bar")
@@ -42,7 +42,7 @@ func TestPullRefMostRecentJobsMetrics(t *testing.T) {
 		})
 
 	ref := schemas.Ref{
-		Project: schemas.NewProject("foo"),
+		Project: schemas.NewProject("foo", []string{}),
 		Name:    "bar",
 		LatestJobs: schemas.Jobs{
 			"bar": {
@@ -85,7 +85,7 @@ func TestProcessJobMetrics(t *testing.T) {
 		},
 	}
 
-	p := schemas.NewProject("foo")
+	p := schemas.NewProject("foo", []string{})
 	p.Topics = "first,second"
 	p.Pull.Pipeline.Jobs.RunnerDescription.AggregationRegexp = `foo-(.*)-bar`
 

pkg/controller/pipelines_test.go

@@ -40,7 +40,7 @@ func TestPullRefMetricsSucceed(t *testing.T) {
 		})
 
 	// Metrics pull shall succeed
-	p := schemas.NewProject("foo")
+	p := schemas.NewProject("foo", []string{})
 	p.Pull.Pipeline.Variables.Enabled = true
 	p.Pull.Pipeline.TestReports.Enabled = true
 	p.Pull.Pipeline.TestReports.TestCases.Enabled = true
@@ -130,7 +130,7 @@ func TestPullRefTestReportMetrics(t *testing.T) {
 		})
 
 	// Metrics pull shall succeed
-	p := schemas.NewProject("foo")
+	p := schemas.NewProject("foo", []string{})
 	p.Pull.Pipeline.Variables.Enabled = true
 	p.Pull.Pipeline.TestReports.Enabled = true
 	p.Pull.Pipeline.TestReports.TestCases.Enabled = true
@@ -281,7 +281,7 @@ func TestPullRefMetricsMergeRequestPipeline(t *testing.T) {
 		})
 
 	// Metrics pull shall succeed
-	p := schemas.NewProject("foo")
+	p := schemas.NewProject("foo", []string{})
 	p.Pull.Pipeline.Variables.Enabled = true
 
 	assert.NoError(t, c.PullRefMetrics(

pkg/controller/projects.go

@@ -10,33 +10,38 @@ import (
 )
 
 // PullProject ..
-func (c *Controller) PullProject(ctx context.Context, name string, pull config.ProjectPull) error {
-	gp, err := c.Gitlab.GetProject(ctx, name)
+func (c *Controller) PullProject(ctx context.Context, project config.Project) error {
+	gp, err := c.Gitlab.GetProject(ctx, project.Name)
 	if err != nil {
 		return err
 	}
 
-	p := schemas.NewProject(gp.PathWithNamespace)
-	p.Pull = pull
-
-	projectExists, err := c.Store.ProjectExists(ctx, p.Key())
+	projectExists, err := c.Store.ProjectExists(ctx, gp.Key())
 	if err != nil {
 		return err
 	}
 
+	// We need to set the project in the store regardless of it being new or not
+	// to ensure any project updates e.g. Topics are correctly reflected
+	if err := c.Store.SetProject(ctx, gp); err != nil {
+		log.WithContext(ctx).
+			WithError(err).
+			Error()
+	}
+
 	if !projectExists {
 		log.WithFields(log.Fields{
-			"project-name": p.Name,
+			"project-name": gp.Name,
 		}).Info("discovered new project")
 
-		if err := c.Store.SetProject(ctx, p); err != nil {
+		if err := c.Store.SetProject(ctx, gp); err != nil {
 			log.WithContext(ctx).
 				WithError(err).
 				Error()
 		}
 
-		c.ScheduleTask(ctx, schemas.TaskTypePullRefsFromProject, string(p.Key()), p)
-		c.ScheduleTask(ctx, schemas.TaskTypePullEnvironmentsFromProject, string(p.Key()), p)
+		c.ScheduleTask(ctx, schemas.TaskTypePullRefsFromProject, string(gp.Key()), gp)
+		c.ScheduleTask(ctx, schemas.TaskTypePullEnvironmentsFromProject, string(gp.Key()), gp)
 	}
 
 	return nil
@@ -55,6 +60,14 @@ func (c *Controller) PullProjectsFromWildcard(ctx context.Context, w config.Wild
 			return err
 		}
 
+		// We need to set the project in the store regardless of it being new or not
+		// to ensure any project updates e.g. Topics are correctly reflected
+		if err := c.Store.SetProject(ctx, p); err != nil {
+			log.WithContext(ctx).
+				WithError(err).
+				Error()
+		}
+
 		if !projectExists {
 			log.WithFields(log.Fields{
 				"wildcard-search":                  w.Search,
@@ -65,12 +78,6 @@ func (c *Controller) PullProjectsFromWildcard(ctx context.Context, w config.Wild
 				"project-name":                     p.Name,
 			}).Info("discovered new project")
 
-			if err := c.Store.SetProject(ctx, p); err != nil {
-				log.WithContext(ctx).
-					WithError(err).
-					Error()
-			}
-
 			c.ScheduleTask(ctx, schemas.TaskTypePullRefsFromProject, string(p.Key()), p)
 			c.ScheduleTask(ctx, schemas.TaskTypePullEnvironmentsFromProject, string(p.Key()), p)
 		}

pkg/controller/projects_test.go

@@ -15,19 +15,66 @@ func TestPullProjectsFromWildcard(t *testing.T) {
 	ctx, c, mux, srv := newTestController(config.Config{})
 	defer srv.Close()
 
+	topicsIdentifier := 0
+
 	mux.HandleFunc("/api/v4/projects",
 		func(w http.ResponseWriter, r *http.Request) {
-			fmt.Fprint(w, `[{"id":2,"path_with_namespace":"bar","jobs_enabled":true}]`)
+			fmt.Fprintf(w, `[{"id":2,"path_with_namespace":"bar","jobs_enabled":true,"topics":["foo%d","bar%d"]}]`, topicsIdentifier, topicsIdentifier)
+			topicsIdentifier += 1
 		})
 
 	w := config.NewWildcard()
 	assert.NoError(t, c.PullProjectsFromWildcard(ctx, w))
 
 	projects, _ := c.Store.Projects(ctx)
-	p1 := schemas.NewProject("bar")
+	p1 := schemas.NewProject("bar", []string{"foo0", "bar0"})
+	p2 := schemas.NewProject("bar", []string{"foo1", "bar1"})
+
+	expectedProjects := schemas.Projects{
+		p1.Key(): p1,
+	}
+	assert.Equal(t, expectedProjects, projects)
+
+	expectedUpdatedProjects := schemas.Projects{
+		p2.Key(): p2,
+	}
+
+	// Pull projects again, which will have topics updated
+	assert.NoError(t, c.PullProjectsFromWildcard(ctx, w))
+	projects, _ = c.Store.Projects(ctx)
+	assert.Equal(t, expectedUpdatedProjects, projects)
+}
+
+func TestPullProjects(t *testing.T) {
+	ctx, c, mux, srv := newTestController(config.Config{})
+	defer srv.Close()
+
+	topicsIdentifier := 0
+
+	mux.HandleFunc(fmt.Sprintf("/api/v4/projects/%s", "foo%2Fbar"),
+		func(w http.ResponseWriter, r *http.Request) {
+			fmt.Fprintf(w, `{"id":2,"path_with_namespace":"bar","jobs_enabled":true,"topics":["foo%d","bar%d"]}`, topicsIdentifier, topicsIdentifier)
+			topicsIdentifier += 1
+		})
+
+	w := config.NewProject("foo/bar")
+	assert.NoError(t, c.PullProject(ctx, w))
+
+	projects, _ := c.Store.Projects(ctx)
+	p1 := schemas.NewProject("bar", []string{"foo0", "bar0"})
+	p2 := schemas.NewProject("bar", []string{"foo1", "bar1"})
 
 	expectedProjects := schemas.Projects{
 		p1.Key(): p1,
 	}
 	assert.Equal(t, expectedProjects, projects)
+
+	expectedUpdatedProjects := schemas.Projects{
+		p2.Key(): p2,
+	}
+
+	// Pull projects again, which will have topics updated
+	assert.NoError(t, c.PullProject(ctx, w))
+	projects, _ = c.Store.Projects(ctx)
+	assert.Equal(t, expectedUpdatedProjects, projects)
 }

pkg/controller/refs.go

@@ -88,17 +88,19 @@ func (c *Controller) PullRefsFromProject(ctx context.Context, p schemas.Project)
 			return err
 		}
 
+		// We need to set the ref in the store regardless of it being new or not
+		// to ensure any project updates e.g. Topics are correctly reflected
+		if err = c.Store.SetRef(ctx, ref); err != nil {
+			return err
+		}
+
 		if !refExists {
 			log.WithFields(log.Fields{
 				"project-name": ref.Project.Name,
 				"ref":          ref.Name,
 				"ref-kind":     ref.Kind,
 			}).Info("discovered new ref")
 
-			if err = c.Store.SetRef(ctx, ref); err != nil {
-				return err
-			}
-
 			c.ScheduleTask(ctx, schemas.TaskTypePullRefMetrics, string(ref.Key()), ref)
 		}
 	}

pkg/controller/refs_test.go

@@ -30,7 +30,7 @@ func TestGetRefs(t *testing.T) {
 			fmt.Fprint(w, `[{"ref":"refs/merge-requests/1234/head"}]`)
 		})
 
-	p := schemas.NewProject("foo")
+	p := schemas.NewProject("foo", []string{})
 	p.Pull.Refs.Branches.Regexp = `^m`
 	p.Pull.Refs.Tags.Regexp = `^v`
 	p.Pull.Refs.MergeRequests.Enabled = true
@@ -53,11 +53,6 @@ func TestPullRefsFromProject(t *testing.T) {
 	ctx, c, mux, srv := newTestController(config.Config{})
 	defer srv.Close()
 
-	mux.HandleFunc("/api/v4/projects/foo",
-		func(w http.ResponseWriter, r *http.Request) {
-			fmt.Fprint(w, `{"name":"foo"}`)
-		})
-
 	mux.HandleFunc(fmt.Sprintf("/api/v4/projects/foo/repository/branches"),
 		func(w http.ResponseWriter, r *http.Request) {
 			fmt.Fprint(w, `[{"name":"main"},{"name":"nope"}]`)
@@ -68,14 +63,25 @@ func TestPullRefsFromProject(t *testing.T) {
 			fmt.Fprint(w, `[]`)
 		})
 
-	p1 := schemas.NewProject("foo")
+	p1 := schemas.NewProject("foo", []string{"foo", "bar"})
 	assert.NoError(t, c.PullRefsFromProject(ctx, p1))
 
 	ref1 := schemas.NewRef(p1, schemas.RefKindBranch, "main")
-	expectedRefs := schemas.Refs{
+	expectedRefs1 := schemas.Refs{
 		ref1.Key(): ref1,
 	}
 
-	projectsRefs, _ := c.Store.Refs(ctx)
-	assert.Equal(t, expectedRefs, projectsRefs)
+	projectsRefs1, _ := c.Store.Refs(ctx)
+	assert.Equal(t, expectedRefs1, projectsRefs1)
+
+	p2 := schemas.NewProject("foo", []string{"foo"})
+	assert.NoError(t, c.PullRefsFromProject(ctx, p2))
+
+	ref2 := schemas.NewRef(p2, schemas.RefKindBranch, "main")
+	expectedRefs2 := schemas.Refs{
+		ref2.Key(): ref2,
+	}
+
+	projectsRefs2, _ := c.Store.Refs(ctx)
+	assert.Equal(t, expectedRefs2, projectsRefs2)
 }