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 6cfd55b60 refactor(circleci): make job and workflow extractors
incremental (#9305) (#9313)
6cfd55b60 is described below
commit 6cfd55b60c64ac39b1cb6db3d1503ad8c876e7d6
Author: Volodymyr Zahorniak <[email protected]>
AuthorDate: Fri Nov 6 09:22:54 2026 +0200
refactor(circleci): make job and workflow extractors incremental (#9305)
(#9313)
Use api.NewStatefulApiExtractor for extractJobs and extractWorkflows, so an
incremental run reads only the raw rows added since the previous run.
---
.../circleci/e2e/extractor_incremental_test.go | 185 +++++++++++++++++++++
backend/plugins/circleci/tasks/job_extractor.go | 24 +--
.../plugins/circleci/tasks/workflow_extractor.go | 26 +--
3 files changed, 211 insertions(+), 24 deletions(-)
diff --git a/backend/plugins/circleci/e2e/extractor_incremental_test.go
b/backend/plugins/circleci/e2e/extractor_incremental_test.go
new file mode 100644
index 000000000..07ccf17d1
--- /dev/null
+++ b/backend/plugins/circleci/e2e/extractor_incremental_test.go
@@ -0,0 +1,185 @@
+/*
+Licensed to the Apache Software Foundation (ASF) under one or more
+contributor license agreements. See the NOTICE file distributed with
+this work for additional information regarding copyright ownership.
+The ASF licenses this file to You 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 e2e
+
+import (
+ "context"
+ "encoding/json"
+ "testing"
+ "time"
+
+ coremodels "github.com/apache/devlake/core/models"
+ "github.com/apache/devlake/core/models/common"
+ "github.com/apache/devlake/core/plugin"
+ "github.com/apache/devlake/core/runner"
+ "github.com/apache/devlake/helpers/e2ehelper"
+ "github.com/apache/devlake/helpers/pluginhelper/api"
+ contextimpl "github.com/apache/devlake/impls/context"
+ "github.com/apache/devlake/plugins/circleci/impl"
+ "github.com/apache/devlake/plugins/circleci/models"
+ "github.com/apache/devlake/plugins/circleci/tasks"
+ "github.com/stretchr/testify/assert"
+ "github.com/stretchr/testify/require"
+)
+
+// TestCircleciJobExtractorIncremental checks that an incremental run of
extractJobs
+// reads only the raw rows added since the previous run (#9305).
+func TestCircleciJobExtractorIncremental(t *testing.T) {
+ var circleci impl.Circleci
+ dataflowTester := e2ehelper.NewDataFlowTester(t, "circleci", circleci)
+ taskData := &tasks.CircleciTaskData{
+ Options: &tasks.CircleciOptions{
+ ConnectionId: 1,
+ ProjectSlug: "github/coldgust/coldgust.github.io",
+ },
+ }
+ beginStatefulScenario(dataflowTester)
+
dataflowTester.ImportCsvIntoRawTable("./raw_tables/_raw_circleci_api_jobs.csv",
"_raw_circleci_api_jobs")
+ dataflowTester.FlushTabler(&models.CircleciJob{})
+
+ runSubtaskPreservingState(t, dataflowTester, tasks.ExtractJobsMeta,
taskData)
+ dataflowTester.VerifyTableWithOptions(
+ models.CircleciJob{},
+ e2ehelper.TableOptions{
+ CSVRelPath:
"./snapshot_tables/_tool_circleci_jobs.csv",
+ IgnoreTypes: []interface{}{common.NoPKModel{}},
+ },
+ )
+
+ const removedJob = "7b96e45b-b10e-47a0-95d0-96580b88bdda"
+ const updatedJob = "ab8c3282-0e74-4a41-834e-152a71280bed"
+ require.NoError(t, dataflowTester.Db.Where("id = ?",
removedJob).Delete(&models.CircleciJob{}).Error)
+ appendUpdatedRawRow(t, dataflowTester, "_raw_circleci_api_jobs", 20,
map[string]any{"status": "success"})
+
+ runSubtaskPreservingState(t, dataflowTester, tasks.ExtractJobsMeta,
taskData)
+ statuses := jobStatuses(t, dataflowTester)
+ assert.Len(t, statuses, 8)
+ assert.NotContains(t, statuses, removedJob, "an incremental run must
not read old raw rows again")
+ assert.Equal(t, "success", statuses[updatedJob])
+
+ dataflowTester.Subtask(tasks.ExtractJobsMeta, taskData)
+ statuses = jobStatuses(t, dataflowTester)
+ assert.Len(t, statuses, 9)
+ assert.Contains(t, statuses, removedJob, "a full sync must read all raw
rows")
+ assert.Equal(t, "success", statuses[updatedJob], "the newest raw row
must win")
+}
+
+// TestCircleciWorkflowExtractorIncremental checks that an incremental run of
+// extractWorkflows reads only the raw rows added since the previous run
(#9305).
+func TestCircleciWorkflowExtractorIncremental(t *testing.T) {
+ var circleci impl.Circleci
+ dataflowTester := e2ehelper.NewDataFlowTester(t, "circleci", circleci)
+ taskData := &tasks.CircleciTaskData{
+ Options: &tasks.CircleciOptions{
+ ConnectionId: 1,
+ ProjectSlug: "github/coldgust/coldgust.github.io",
+ },
+ }
+ beginStatefulScenario(dataflowTester)
+
dataflowTester.ImportCsvIntoRawTable("./raw_tables/_raw_circleci_api_workflows.csv",
"_raw_circleci_api_workflows")
+ dataflowTester.FlushTabler(&models.CircleciWorkflow{})
+
+ runSubtaskPreservingState(t, dataflowTester,
tasks.ExtractWorkflowsMeta, taskData)
+ dataflowTester.VerifyTableWithOptions(
+ models.CircleciWorkflow{},
+ e2ehelper.TableOptions{
+ CSVRelPath:
"./snapshot_tables/_tool_circleci_workflows.csv",
+ IgnoreTypes: []interface{}{common.NoPKModel{}},
+ IgnoreFields: []string{"started_at", "stopped_at"},
+ },
+ )
+
+ const removedWorkflow = "89054eb2-8e85-4f5c-9a93-66d753a0e970"
+ const updatedWorkflow = "6731159f-5275-4bfa-ba70-39d343d63814"
+ require.NoError(t, dataflowTester.Db.Where("id = ?",
removedWorkflow).Delete(&models.CircleciWorkflow{}).Error)
+ appendUpdatedRawRow(t, dataflowTester, "_raw_circleci_api_workflows",
19, map[string]any{"status": "success"})
+
+ runSubtaskPreservingState(t, dataflowTester,
tasks.ExtractWorkflowsMeta, taskData)
+ statuses := workflowStatuses(t, dataflowTester)
+ assert.Len(t, statuses, 9)
+ assert.NotContains(t, statuses, removedWorkflow, "an incremental run
must not read old raw rows again")
+ assert.Equal(t, "success", statuses[updatedWorkflow])
+
+ dataflowTester.Subtask(tasks.ExtractWorkflowsMeta, taskData)
+ statuses = workflowStatuses(t, dataflowTester)
+ assert.Len(t, statuses, 10)
+ assert.Contains(t, statuses, removedWorkflow, "a full sync must read
all raw rows")
+ assert.Equal(t, "success", statuses[updatedWorkflow], "the newest raw
row must win")
+}
+
+// beginStatefulScenario removes the stored states, so that the first
extractor run is a full sync.
+func beginStatefulScenario(dataflowTester *e2ehelper.DataFlowTester) {
+ dataflowTester.FlushTabler(&coremodels.SubtaskState{})
+ dataflowTester.FlushTabler(&coremodels.CollectorLatestState{})
+}
+
+// runSubtaskPreservingState runs a subtask without the forced full sync of
DataFlowTester.Subtask.
+func runSubtaskPreservingState(t *testing.T, dataflowTester
*e2ehelper.DataFlowTester, subtaskMeta plugin.SubTaskMeta, taskData
interface{}) {
+ t.Helper()
+ subtaskCtx := contextimpl.NewStandaloneSubTaskContext(
+ context.Background(),
+ runner.CreateBasicRes(dataflowTester.Cfg, dataflowTester.Log,
dataflowTester.Db),
+ dataflowTester.Name,
+ taskData,
+ subtaskMeta.Name,
+ &coremodels.SyncPolicy{},
+ )
+ require.NoError(t, subtaskMeta.EntryPoint(subtaskCtx))
+}
+
+// appendUpdatedRawRow adds a new raw row, created now, with the data of an
existing row and the given changes.
+func appendUpdatedRawRow(t *testing.T, dataflowTester
*e2ehelper.DataFlowTester, table string, sourceId uint64, changes
map[string]any) {
+ t.Helper()
+ row := &api.RawData{}
+ require.NoError(t, dataflowTester.Db.Table(table).First(row,
sourceId).Error)
+ data := map[string]any{}
+ require.NoError(t, json.Unmarshal(row.Data, &data))
+ for k, v := range changes {
+ data[k] = v
+ }
+ var err error
+ row.Data, err = json.Marshal(data)
+ require.NoError(t, err)
+ var maxId uint64
+ require.NoError(t,
dataflowTester.Db.Table(table).Select("COALESCE(MAX(id),
0)").Row().Scan(&maxId))
+ row.ID = maxId + 1
+ row.CreatedAt = time.Now()
+ require.NoError(t, dataflowTester.Db.Table(table).Create(row).Error)
+}
+
+func jobStatuses(t *testing.T, dataflowTester *e2ehelper.DataFlowTester)
map[string]string {
+ t.Helper()
+ var jobs []models.CircleciJob
+ require.NoError(t, dataflowTester.Db.Where("connection_id = ?",
1).Find(&jobs).Error)
+ statuses := make(map[string]string, len(jobs))
+ for _, j := range jobs {
+ statuses[j.Id] = j.Status
+ }
+ return statuses
+}
+
+func workflowStatuses(t *testing.T, dataflowTester *e2ehelper.DataFlowTester)
map[string]string {
+ t.Helper()
+ var workflows []models.CircleciWorkflow
+ require.NoError(t, dataflowTester.Db.Where("connection_id = ?",
1).Find(&workflows).Error)
+ statuses := make(map[string]string, len(workflows))
+ for _, w := range workflows {
+ statuses[w.Id] = w.Status
+ }
+ return statuses
+}
diff --git a/backend/plugins/circleci/tasks/job_extractor.go
b/backend/plugins/circleci/tasks/job_extractor.go
index 8d65b46ed..4209ead43 100644
--- a/backend/plugins/circleci/tasks/job_extractor.go
+++ b/backend/plugins/circleci/tasks/job_extractor.go
@@ -36,21 +36,23 @@ var ExtractJobsMeta = plugin.SubTaskMeta{
}
func ExtractJobs(taskCtx plugin.SubTaskContext) errors.Error {
- rawDataSubTaskArgs, data := CreateRawDataSubTaskArgs(taskCtx,
RAW_JOB_TABLE)
- extractor, err := api.NewApiExtractor(api.ApiExtractorArgs{
- RawDataSubTaskArgs: *rawDataSubTaskArgs,
- Extract: func(row *api.RawData) ([]interface{}, errors.Error) {
+ data := taskCtx.GetData().(*CircleciTaskData)
+ extractor, err :=
api.NewStatefulApiExtractor(&api.StatefulApiExtractorArgs[models.CircleciJob]{
+ SubtaskCommonArgs: &api.SubtaskCommonArgs{
+ SubTaskContext: taskCtx,
+ Table: RAW_JOB_TABLE,
+ Params: models.CircleciApiParams{
+ ConnectionId: data.Options.ConnectionId,
+ ProjectSlug: data.Options.ProjectSlug,
+ },
+ },
+ Extract: func(userRes *models.CircleciJob, row *api.RawData)
([]any, errors.Error) {
input := &models.CircleciWorkflow{}
err := errors.Convert(json.Unmarshal(row.Input, input))
if err != nil {
return nil, err
}
- userRes := models.CircleciJob{}
- err = errors.Convert(json.Unmarshal(row.Data, &userRes))
- if err != nil {
- return nil, err
- }
- toolL := userRes
+ toolL := *userRes
toolL.ConnectionId = data.Options.ConnectionId
toolL.ProjectSlug = data.Options.ProjectSlug
toolL.WorkflowId = input.Id
@@ -60,7 +62,7 @@ func ExtractJobs(taskCtx plugin.SubTaskContext) errors.Error {
stopTime := userRes.StoppedAt.ToTime()
toolL.DurationSec =
float64(stopTime.Sub(startTime).Milliseconds() / 1e3)
}
- return []interface{}{
+ return []any{
&toolL,
}, nil
},
diff --git a/backend/plugins/circleci/tasks/workflow_extractor.go
b/backend/plugins/circleci/tasks/workflow_extractor.go
index b61e95069..8500d701e 100644
--- a/backend/plugins/circleci/tasks/workflow_extractor.go
+++ b/backend/plugins/circleci/tasks/workflow_extractor.go
@@ -18,8 +18,6 @@ limitations under the License.
package tasks
import (
- "encoding/json"
-
"github.com/apache/devlake/core/errors"
"github.com/apache/devlake/core/plugin"
"github.com/apache/devlake/helpers/pluginhelper/api"
@@ -37,16 +35,18 @@ var ExtractWorkflowsMeta = plugin.SubTaskMeta{
}
func ExtractWorkflows(taskCtx plugin.SubTaskContext) errors.Error {
- rawDataSubTaskArgs, data := CreateRawDataSubTaskArgs(taskCtx,
RAW_WORKFLOW_TABLE)
- extractor, err := api.NewApiExtractor(api.ApiExtractorArgs{
- RawDataSubTaskArgs: *rawDataSubTaskArgs,
- Extract: func(row *api.RawData) ([]interface{}, errors.Error) {
- userRes := models.CircleciWorkflow{}
- err := errors.Convert(json.Unmarshal(row.Data,
&userRes))
- if err != nil {
- return nil, err
- }
- toolL := userRes
+ data := taskCtx.GetData().(*CircleciTaskData)
+ extractor, err :=
api.NewStatefulApiExtractor(&api.StatefulApiExtractorArgs[models.CircleciWorkflow]{
+ SubtaskCommonArgs: &api.SubtaskCommonArgs{
+ SubTaskContext: taskCtx,
+ Table: RAW_WORKFLOW_TABLE,
+ Params: models.CircleciApiParams{
+ ConnectionId: data.Options.ConnectionId,
+ ProjectSlug: data.Options.ProjectSlug,
+ },
+ },
+ Extract: func(userRes *models.CircleciWorkflow, row
*api.RawData) ([]any, errors.Error) {
+ toolL := *userRes
toolL.ConnectionId = data.Options.ConnectionId
toolL.ProjectSlug = data.Options.ProjectSlug
if userRes.CreatedDate != nil && userRes.StoppedAt !=
nil {
@@ -57,7 +57,7 @@ func ExtractWorkflows(taskCtx plugin.SubTaskContext)
errors.Error {
toolL.DurationSec = durationSec
}
}
- return []interface{}{
+ return []any{
&toolL,
}, nil
},
