This is an automated email from the ASF dual-hosted git repository.
klesh pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/devlake.git
The following commit(s) were added to refs/heads/main by this push:
new 2bcc408bb fix(circleci): limit unfinished details recollection to
recent workflows (#9304) (#9309)
2bcc408bb is described below
commit 2bcc408bb169b5af83e05f4a35437c8e66fe880f
Author: Volodymyr Zahorniak <[email protected]>
AuthorDate: Tue Nov 3 09:12:35 2026 +0200
fix(circleci): limit unfinished details recollection to recent workflows
(#9304) (#9309)
The unfinished-details collectors of collectJobs and collectWorkflows
selected every
workflow in a non-terminal status. A workflow whose approval job is never
approved stays
on_hold forever, so it was collected again on every run and the raw tables
kept growing.
Select only workflows created within the last 30 days.
---
backend/plugins/circleci/e2e/job_collector_test.go | 97 ++++++++++++++++------
backend/plugins/circleci/tasks/job_collector.go | 17 ++--
backend/plugins/circleci/tasks/shared.go | 11 +++
.../plugins/circleci/tasks/workflow_collector.go | 22 +++--
4 files changed, 108 insertions(+), 39 deletions(-)
diff --git a/backend/plugins/circleci/e2e/job_collector_test.go
b/backend/plugins/circleci/e2e/job_collector_test.go
index f3988b48a..f27963e0c 100644
--- a/backend/plugins/circleci/e2e/job_collector_test.go
+++ b/backend/plugins/circleci/e2e/job_collector_test.go
@@ -21,7 +21,10 @@ import (
"reflect"
"sort"
"testing"
+ "time"
+ "github.com/apache/devlake/core/dal"
+ "github.com/apache/devlake/core/models/common"
"github.com/apache/devlake/helpers/e2ehelper"
"github.com/apache/devlake/helpers/pluginhelper/api"
"github.com/apache/devlake/plugins/circleci/impl"
@@ -30,36 +33,33 @@ import (
"github.com/stretchr/testify/assert"
)
-// TestCircleciUnfinishedJobsInputIterator is a regression test for
-// https://github.com/apache/devlake/issues/8907. The "collect unfinished job
-// details" collector builds its URL from "/v2/workflow/{{ .Input.Id }}/job"
while
-// scanning rows into a models.CircleciJob. Its input query must therefore
expose the
-// workflow id in the row's Id field; a bare "DISTINCT workflow_id" left Id
empty and
-// produced "/v2/workflow//job" (HTTP 500). This test runs the production query
-// (tasks.UnfinishedJobsInputClauses) through the real iterator and asserts
each
-// yielded row's Id is the workflow id, that results are DISTINCT, and that the
-// status/connection filters hold.
-func TestCircleciUnfinishedJobsInputIterator(t *testing.T) {
- var circleci impl.Circleci
- dataflowTester := e2ehelper.NewDataFlowTester(t, "circleci", circleci)
+const unfinishedTestProjectSlug = "github/test/repo"
- const projectSlug = "github/test/repo"
- dataflowTester.FlushTabler(&models.CircleciJob{})
+var unfinishedTestNow = time.Date(2026, 11, 2, 0, 0, 0, 0, time.UTC)
- seed := []models.CircleciJob{
- {ConnectionId: 1, WorkflowId: "wf-onhold", Id: "job-1",
ProjectSlug: projectSlug, Status: "on_hold"},
- {ConnectionId: 1, WorkflowId: "wf-onhold", Id: "job-2",
ProjectSlug: projectSlug, Status: "running"}, // same workflow -> DISTINCT
- {ConnectionId: 1, WorkflowId: "wf-queued", Id: "job-3",
ProjectSlug: projectSlug, Status: "queued"},
- {ConnectionId: 1, WorkflowId: "wf-success", Id: "job-4",
ProjectSlug: projectSlug, Status: "success"}, // terminal -> excluded
- {ConnectionId: 2, WorkflowId: "wf-otherconn", Id: "job-5",
ProjectSlug: projectSlug, Status: "on_hold"}, // other connection -> excluded
+func daysAgo(days int) *common.Iso8601Time {
+ return &common.Iso8601Time{Time: unfinishedTestNow.AddDate(0, 0, -days)}
+}
+
+func seedUnfinishedWorkflows(t *testing.T, dataflowTester
*e2ehelper.DataFlowTester) {
+ dataflowTester.FlushTabler(&models.CircleciWorkflow{})
+ seed := []models.CircleciWorkflow{
+ {ConnectionId: 1, Id: "wf-onhold", ProjectSlug:
unfinishedTestProjectSlug, Status: "on_hold", CreatedDate: daysAgo(1)},
+ {ConnectionId: 1, Id: "wf-queued", ProjectSlug:
unfinishedTestProjectSlug, Status: "running", CreatedDate: daysAgo(29)},
+ {ConnectionId: 1, Id: "wf-failing", ProjectSlug:
unfinishedTestProjectSlug, Status: "failing", CreatedDate: daysAgo(2)},
+ {ConnectionId: 1, Id: "wf-success", ProjectSlug:
unfinishedTestProjectSlug, Status: "success", CreatedDate: daysAgo(1)},
+ {ConnectionId: 1, Id: "wf-old-onhold", ProjectSlug:
unfinishedTestProjectSlug, Status: "on_hold", CreatedDate: daysAgo(31)},
+ {ConnectionId: 2, Id: "wf-otherconn", ProjectSlug:
unfinishedTestProjectSlug, Status: "on_hold", CreatedDate: daysAgo(1)},
}
for i := range seed {
assert.Nil(t, dataflowTester.Dal.Create(&seed[i]))
}
+}
- cursor, err :=
dataflowTester.Dal.Cursor(tasks.UnfinishedJobsInputClauses(1, projectSlug)...)
+func collectInputIds(t *testing.T, dataflowTester *e2ehelper.DataFlowTester,
clauses []dal.Clause, rowType reflect.Type) []string {
+ cursor, err := dataflowTester.Dal.Cursor(clauses...)
assert.Nil(t, err)
- iter, err := api.NewDalCursorIterator(dataflowTester.Dal, cursor,
reflect.TypeOf(models.CircleciJob{}))
+ iter, err := api.NewDalCursorIterator(dataflowTester.Dal, cursor,
rowType)
assert.Nil(t, err)
defer iter.Close()
@@ -67,16 +67,61 @@ func TestCircleciUnfinishedJobsInputIterator(t *testing.T) {
for iter.HasNext() {
item, err := iter.Fetch()
assert.Nil(t, err)
- job := item.(*models.CircleciJob)
- ids = append(ids, job.Id)
+ ids = append(ids,
reflect.ValueOf(item).Elem().FieldByName("Id").String())
}
sort.Strings(ids)
+ return ids
+}
+
+// TestCircleciUnfinishedJobsInputIterator is a regression test for
+// https://github.com/apache/devlake/issues/8907 and
+// https://github.com/apache/devlake/issues/9304. The "collect unfinished job
+// details" collector builds its URL from "/v2/workflow/{{ .Input.Id }}/job"
while
+// scanning rows into a models.CircleciJob. Its input query must therefore
expose the
+// workflow id in the row's Id field; a bare "DISTINCT workflow_id" left Id
empty and
+// produced "/v2/workflow//job" (HTTP 500). It must also skip workflows older
than the
+// lookback window: an approval job that is never approved stays on_hold
forever, and
+// its workflow was collected again on every run (#9304).
+func TestCircleciUnfinishedJobsInputIterator(t *testing.T) {
+ var circleci impl.Circleci
+ dataflowTester := e2ehelper.NewDataFlowTester(t, "circleci", circleci)
+
+ seedUnfinishedWorkflows(t, dataflowTester)
+ dataflowTester.FlushTabler(&models.CircleciJob{})
+ seed := []models.CircleciJob{
+ {ConnectionId: 1, WorkflowId: "wf-onhold", Id: "job-1",
ProjectSlug: unfinishedTestProjectSlug, Status: "on_hold"},
+ {ConnectionId: 1, WorkflowId: "wf-onhold", Id: "job-2",
ProjectSlug: unfinishedTestProjectSlug, Status: "running"}, // same workflow ->
DISTINCT
+ {ConnectionId: 1, WorkflowId: "wf-queued", Id: "job-3",
ProjectSlug: unfinishedTestProjectSlug, Status: "queued"},
+ {ConnectionId: 1, WorkflowId: "wf-success", Id: "job-4",
ProjectSlug: unfinishedTestProjectSlug, Status: "success"}, // terminal ->
excluded
+ {ConnectionId: 1, WorkflowId: "wf-old-onhold", Id: "job-5",
ProjectSlug: unfinishedTestProjectSlug, Status: "on_hold"}, // older than
lookback -> excluded
+ {ConnectionId: 2, WorkflowId: "wf-otherconn", Id: "job-6",
ProjectSlug: unfinishedTestProjectSlug, Status: "on_hold"}, // other
connection -> excluded
+ }
+ for i := range seed {
+ assert.Nil(t, dataflowTester.Dal.Create(&seed[i]))
+ }
+
+ clauses := tasks.UnfinishedJobsInputClauses(1,
unfinishedTestProjectSlug,
tasks.UnfinishedDetailsCreatedAfter(unfinishedTestNow))
+ ids := collectInputIds(t, dataflowTester, clauses,
reflect.TypeOf(models.CircleciJob{}))
// Distinct workflow ids for connection 1's non-terminal jobs, with Id
populated
- // (the URL template reads .Input.Id). wf-success (terminal) and
wf-otherconn
- // (connection 2) are excluded.
+ // (the URL template reads .Input.Id).
assert.Equal(t, []string{"wf-onhold", "wf-queued"}, ids)
for _, id := range ids {
assert.NotEmpty(t, id, "Input.Id must be the workflow id, not
empty (#8907)")
}
}
+
+// TestCircleciUnfinishedWorkflowsInputIterator is a regression test for
+// https://github.com/apache/devlake/issues/9304: the "collect unfinished
workflow
+// details" collector must skip workflows older than the lookback window.
+func TestCircleciUnfinishedWorkflowsInputIterator(t *testing.T) {
+ var circleci impl.Circleci
+ dataflowTester := e2ehelper.NewDataFlowTester(t, "circleci", circleci)
+
+ seedUnfinishedWorkflows(t, dataflowTester)
+
+ clauses := tasks.UnfinishedWorkflowsInputClauses(1,
unfinishedTestProjectSlug,
tasks.UnfinishedDetailsCreatedAfter(unfinishedTestNow))
+ ids := collectInputIds(t, dataflowTester, clauses,
reflect.TypeOf(models.CircleciWorkflow{}))
+
+ assert.Equal(t, []string{"wf-failing", "wf-onhold", "wf-queued"}, ids)
+}
diff --git a/backend/plugins/circleci/tasks/job_collector.go
b/backend/plugins/circleci/tasks/job_collector.go
index 2e950107a..730edaa9c 100644
--- a/backend/plugins/circleci/tasks/job_collector.go
+++ b/backend/plugins/circleci/tasks/job_collector.go
@@ -43,14 +43,16 @@ var CollectJobsMeta = plugin.SubTaskMeta{
// UnfinishedJobsInputClauses returns the DAL clauses that select the
workflows whose
// jobs are still in a non-terminal status and therefore need their job details
-// recollected by the CollectJobs "unfinished details" collector.
-func UnfinishedJobsInputClauses(connectionId uint64, projectSlug string)
[]dal.Clause {
+// recollected by the CollectJobs "unfinished details" collector. Only
workflows created
+// at or after createdAfter are selected (#9304); jobs have no created date of
their own.
+func UnfinishedJobsInputClauses(connectionId uint64, projectSlug string,
createdAfter time.Time) []dal.Clause {
return []dal.Clause{
- dal.Select("DISTINCT workflow_id AS id"), // #8907: alias to id
so {{ .Input.Id }} resolves when scanned into CircleciJob
- dal.From(&models.CircleciJob{}),
+ dal.Select("DISTINCT j.workflow_id AS id"), // #8907: alias to
id so {{ .Input.Id }} resolves when scanned into CircleciJob
+ dal.From("_tool_circleci_jobs j"),
+ dal.Join("JOIN _tool_circleci_workflows w ON w.connection_id =
j.connection_id AND w.id = j.workflow_id"),
dal.Where(
- "connection_id = ? AND project_slug = ? AND status IN
('running', 'not_running', 'queued', 'on_hold')",
- connectionId, projectSlug,
+ "j.connection_id = ? AND j.project_slug = ? AND
j.status IN ('running', 'not_running', 'queued', 'on_hold') AND w.created_date
>= ?",
+ connectionId, projectSlug, createdAfter,
),
}
}
@@ -111,7 +113,8 @@ func CollectJobs(taskCtx plugin.SubTaskContext)
errors.Error {
},
BuildInputIterator: func() (api.Iterator, errors.Error)
{
db := taskCtx.GetDal()
- cursor, err :=
db.Cursor(UnfinishedJobsInputClauses(data.Options.ConnectionId,
data.Options.ProjectSlug)...)
+ clauses :=
UnfinishedJobsInputClauses(data.Options.ConnectionId, data.Options.ProjectSlug,
UnfinishedDetailsCreatedAfter(time.Now()))
+ cursor, err := db.Cursor(clauses...)
if err != nil {
return nil, err
}
diff --git a/backend/plugins/circleci/tasks/shared.go
b/backend/plugins/circleci/tasks/shared.go
index f3be144dd..69a92db0d 100644
--- a/backend/plugins/circleci/tasks/shared.go
+++ b/backend/plugins/circleci/tasks/shared.go
@@ -31,6 +31,17 @@ import (
"github.com/apache/devlake/plugins/circleci/models"
)
+// UnfinishedDetailsLookback limits how far back the "unfinished details"
collectors look.
+// A workflow with an approval job that is never approved stays on_hold
forever; without
+// a limit it would be collected again on every run (#9304).
+const UnfinishedDetailsLookback = 30 * 24 * time.Hour
+
+// UnfinishedDetailsCreatedAfter returns the earliest created date of a
workflow that the
+// "unfinished details" collectors still collect again.
+func UnfinishedDetailsCreatedAfter(now time.Time) time.Time {
+ return now.Add(-UnfinishedDetailsLookback)
+}
+
var accountIdGen *didgen.DomainIdGenerator
var projectIdGen *didgen.DomainIdGenerator
var pipelineIdGen *didgen.DomainIdGenerator
diff --git a/backend/plugins/circleci/tasks/workflow_collector.go
b/backend/plugins/circleci/tasks/workflow_collector.go
index 7847cfe22..024917a85 100644
--- a/backend/plugins/circleci/tasks/workflow_collector.go
+++ b/backend/plugins/circleci/tasks/workflow_collector.go
@@ -42,6 +42,21 @@ var CollectWorkflowsMeta = plugin.SubTaskMeta{
DomainTypes: []string{plugin.DOMAIN_TYPE_CICD},
}
+// UnfinishedWorkflowsInputClauses returns the DAL clauses that select the
workflows that
+// are still in a non-terminal status and therefore need their details
recollected by the
+// CollectWorkflows "unfinished details" collector. Only workflows created at
or after
+// createdAfter are selected (#9304).
+func UnfinishedWorkflowsInputClauses(connectionId uint64, projectSlug string,
createdAfter time.Time) []dal.Clause {
+ return []dal.Clause{
+ dal.Select("id"),
+ dal.From(&models.CircleciWorkflow{}),
+ dal.Where(
+ "connection_id = ? AND project_slug = ? AND status IN
('running', 'on_hold', 'failing') AND created_date >= ?",
+ connectionId, projectSlug, createdAfter,
+ ),
+ }
+}
+
func CollectWorkflows(taskCtx plugin.SubTaskContext) errors.Error {
rawDataSubTaskArgs, data := CreateRawDataSubTaskArgs(taskCtx,
RAW_WORKFLOW_TABLE)
logger := taskCtx.GetLogger()
@@ -93,12 +108,7 @@ func CollectWorkflows(taskCtx plugin.SubTaskContext)
errors.Error {
AfterResponse: ignoreDeletedOrBrokenBuilds,
},
BuildInputIterator: func() (api.Iterator, errors.Error)
{
- clauses := []dal.Clause{
- dal.Select("id"),
- dal.From(&models.CircleciWorkflow{}),
- dal.Where("connection_id = ? AND
project_slug = ? AND status IN ('running', 'on_hold', 'failing')",
data.Options.ConnectionId, data.Options.ProjectSlug),
- }
-
+ clauses :=
UnfinishedWorkflowsInputClauses(data.Options.ConnectionId,
data.Options.ProjectSlug, UnfinishedDetailsCreatedAfter(time.Now()))
db := taskCtx.GetDal()
cursor, err := db.Cursor(clauses...)
if err != nil {
