refactor(circleci): make job and workflow extractors incremental (#9305)

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
+}

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
 		},

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
 		},