You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@devlake.apache.org by kl...@apache.org on 2023/09/13 09:09:31 UTC
[incubator-devlake] branch main updated: refactor: syncPolicy timeAfter (#6064)
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/incubator-devlake.git
The following commit(s) were added to refs/heads/main by this push:
new 738837303 refactor: syncPolicy timeAfter (#6064)
738837303 is described below
commit 73883730362d16148cee167cb0cd5f13edc28f62
Author: abeizn <zi...@merico.dev>
AuthorDate: Wed Sep 13 17:09:20 2023 +0800
refactor: syncPolicy timeAfter (#6064)
* refactor: syncPolicy timeAfter
* fix: cobra command timeAfter and some ci issues
* fix: some ci issues
* fix: collectorWithState.TimeAfter and get syncpolicy from pipeline
* fix: use manager.TimeAfter instead of syncPolicy.TimeAfter
---------
Co-authored-by: Klesh Wong <zh...@merico.dev>
---
backend/core/models/blueprint.go | 4 +--
...dd_full_sync.go => 20230911_add_sync_policy.go} | 33 +++++++++++++---------
backend/core/models/migrationscripts/register.go | 2 +-
backend/core/models/pipeline.go | 6 ++--
backend/core/plugin/plugin_blueprint.go | 1 -
backend/core/runner/directrun.go | 15 ++++++++--
backend/core/runner/run_task.go | 5 +---
.../pluginhelper/api/api_collector_with_state.go | 15 ++++++----
backend/plugins/ae/ae.go | 3 +-
backend/plugins/bamboo/api/blueprint_V200_test.go | 3 +-
backend/plugins/bamboo/api/blueprint_v200.go | 5 ++--
backend/plugins/bamboo/bamboo.go | 3 +-
backend/plugins/bamboo/impl/impl.go | 3 +-
.../plugins/bitbucket/api/blueprint_V200_test.go | 3 +-
backend/plugins/bitbucket/api/blueprint_v200.go | 8 +-----
backend/plugins/bitbucket/bitbucket.go | 5 ++--
backend/plugins/bitbucket/impl/impl.go | 17 ++---------
backend/plugins/bitbucket/tasks/api_common.go | 9 +++---
backend/plugins/bitbucket/tasks/issue_collector.go | 2 +-
.../bitbucket/tasks/issue_comment_collector.go | 2 +-
.../plugins/bitbucket/tasks/pipeline_collector.go | 2 +-
.../bitbucket/tasks/pipeline_steps_collector.go | 2 +-
backend/plugins/bitbucket/tasks/pr_collector.go | 2 +-
.../bitbucket/tasks/pr_comment_collector.go | 2 +-
.../plugins/bitbucket/tasks/pr_commit_collector.go | 5 ++--
backend/plugins/bitbucket/tasks/task_data.go | 4 ---
backend/plugins/circleci/api/blueprint200.go | 8 +-----
backend/plugins/circleci/circleci.go | 3 +-
backend/plugins/circleci/impl/impl.go | 14 +--------
backend/plugins/circleci/tasks/task_data.go | 4 ---
backend/plugins/customize/customize.go | 4 ++-
backend/plugins/dbt/dbt.go | 3 +-
backend/plugins/dora/dora.go | 3 +-
backend/plugins/feishu/feishu.go | 4 ++-
backend/plugins/gitee/gitee.go | 3 +-
backend/plugins/github/api/blueprint_V200_test.go | 3 +-
backend/plugins/github/api/blueprint_v200.go | 8 +-----
backend/plugins/github/github.go | 5 ++--
backend/plugins/github/impl/impl.go | 13 +--------
backend/plugins/github/tasks/cicd_job_collector.go | 2 +-
backend/plugins/github/tasks/cicd_run_collector.go | 1 -
backend/plugins/github/tasks/comment_collector.go | 7 +++--
backend/plugins/github/tasks/commit_collector.go | 7 +++--
backend/plugins/github/tasks/event_collector.go | 1 -
backend/plugins/github/tasks/issue_collector.go | 7 +++--
backend/plugins/github/tasks/pr_collector.go | 1 -
.../plugins/github/tasks/pr_commit_collector.go | 2 +-
.../plugins/github/tasks/pr_review_collector.go | 2 +-
.../github/tasks/pr_review_comment_collector.go | 7 +++--
backend/plugins/github/tasks/task_data.go | 3 --
backend/plugins/github_graphql/impl/impl.go | 9 ------
backend/plugins/github_graphql/plugin_main.go | 5 ++--
.../github_graphql/tasks/issue_collector.go | 3 +-
.../plugins/github_graphql/tasks/job_collector.go | 2 +-
.../plugins/github_graphql/tasks/pr_collector.go | 3 +-
backend/plugins/gitlab/api/blueprint_V200_test.go | 3 +-
backend/plugins/gitlab/api/blueprint_v200.go | 9 ++----
backend/plugins/gitlab/gitlab.go | 5 ++--
backend/plugins/gitlab/impl/impl.go | 16 +----------
backend/plugins/gitlab/tasks/issue_collector.go | 2 +-
backend/plugins/gitlab/tasks/job_collector.go | 1 -
backend/plugins/gitlab/tasks/mr_collector.go | 2 +-
.../plugins/gitlab/tasks/mr_commit_collector.go | 2 +-
.../plugins/gitlab/tasks/mr_detail_collector.go | 2 +-
backend/plugins/gitlab/tasks/mr_note_collector.go | 2 +-
backend/plugins/gitlab/tasks/pipeline_collector.go | 2 +-
.../gitlab/tasks/pipeline_detail_collector.go | 2 +-
backend/plugins/gitlab/tasks/task_data.go | 4 ---
backend/plugins/icla/plugin_main.go | 3 +-
backend/plugins/jenkins/api/blueprint_v200.go | 10 +------
backend/plugins/jenkins/api/blueprint_v200_test.go | 4 +--
backend/plugins/jenkins/impl/impl.go | 16 ++---------
backend/plugins/jenkins/jenkins.go | 5 ++--
backend/plugins/jenkins/tasks/build_collector.go | 1 -
backend/plugins/jenkins/tasks/stage_collector.go | 6 ++--
backend/plugins/jenkins/tasks/task_data.go | 11 +++-----
backend/plugins/jira/api/blueprint_v200.go | 9 +-----
backend/plugins/jira/api/blueprint_v200_test.go | 4 +--
backend/plugins/jira/impl/impl.go | 14 ++-------
backend/plugins/jira/jira.go | 6 ++--
.../jira/tasks/development_panel_collector.go | 2 +-
.../jira/tasks/issue_changelog_collector.go | 2 +-
backend/plugins/jira/tasks/issue_collector.go | 16 ++++++++---
.../plugins/jira/tasks/issue_comment_collector.go | 2 +-
backend/plugins/jira/tasks/remotelink_collector.go | 2 +-
backend/plugins/jira/tasks/task_data.go | 7 ++---
backend/plugins/jira/tasks/worklog_collector.go | 2 +-
backend/plugins/pagerduty/api/blueprint_v200.go | 10 ++-----
backend/plugins/pagerduty/impl/impl.go | 18 +++---------
backend/plugins/pagerduty/pager_duty.go | 3 +-
.../plugins/pagerduty/tasks/incidents_collector.go | 1 -
backend/plugins/pagerduty/tasks/task_data.go | 8 ++----
backend/plugins/refdiff/refdiff.go | 3 +-
backend/plugins/slack/slack.go | 3 +-
backend/plugins/sonarqube/api/blueprint_v200.go | 4 +--
.../plugins/sonarqube/api/blueprint_v200_test.go | 4 +--
backend/plugins/sonarqube/impl/impl.go | 3 +-
backend/plugins/sonarqube/sonarqube.go | 3 +-
backend/plugins/starrocks/starrocks.go | 3 +-
backend/plugins/tapd/api/blueprint_v200.go | 8 +-----
backend/plugins/tapd/api/blueprint_v200_test.go | 3 +-
backend/plugins/tapd/impl/impl.go | 12 +-------
backend/plugins/tapd/tapd.go | 5 ++--
.../plugins/tapd/tasks/bug_changelog_collector.go | 7 +++--
backend/plugins/tapd/tasks/bug_collector.go | 11 ++++----
backend/plugins/tapd/tasks/bug_commit_collector.go | 9 +++---
backend/plugins/tapd/tasks/iteration_collector.go | 7 +++--
backend/plugins/tapd/tasks/story_bug_collector.go | 7 +++--
.../tapd/tasks/story_changelog_collector.go | 7 +++--
backend/plugins/tapd/tasks/story_collector.go | 7 +++--
.../plugins/tapd/tasks/story_commit_collector.go | 9 +++---
.../plugins/tapd/tasks/task_changelog_collector.go | 8 +++---
backend/plugins/tapd/tasks/task_collector.go | 8 +++---
.../plugins/tapd/tasks/task_commit_collector.go | 9 +++---
backend/plugins/tapd/tasks/task_data.go | 2 --
backend/plugins/tapd/tasks/worklog_collector.go | 8 +++---
backend/plugins/teambition/api/blueprint200.go | 8 +-----
backend/plugins/teambition/impl/impl.go | 15 ++--------
backend/plugins/teambition/tasks/task_data.go | 5 ++--
backend/plugins/teambition/teambition.go | 3 +-
backend/plugins/trello/api/blueprint_v200.go | 10 ++-----
backend/plugins/trello/impl/impl.go | 3 +-
backend/plugins/trello/trello.go | 3 +-
backend/plugins/zentao/api/blueprint_V200_test.go | 3 +-
backend/plugins/zentao/api/blueprint_v200.go | 9 +-----
backend/plugins/zentao/impl/impl.go | 3 +-
.../plugins/zentao/tasks/bug_commits_collector.go | 2 +-
.../zentao/tasks/story_commits_collector.go | 2 +-
.../plugins/zentao/tasks/task_commits_collector.go | 2 +-
backend/plugins/zentao/tasks/task_data.go | 3 --
backend/plugins/zentao/zentao.go | 3 +-
backend/server/api/blueprints/blueprints.go | 1 -
backend/server/services/blueprint.go | 33 ++++++++--------------
backend/server/services/blueprint_makeplan_v200.go | 5 ++--
.../services/blueprint_makeplan_v200_test.go | 5 ++--
backend/server/services/pipeline_helper.go | 3 +-
.../services/remote/plugin/plugin_extensions.go | 1 -
backend/server/services/remote/run/run.go | 2 +-
backend/test/helper/api.go | 6 ++--
139 files changed, 297 insertions(+), 495 deletions(-)
diff --git a/backend/core/models/blueprint.go b/backend/core/models/blueprint.go
index 371b97210..b9ff2a457 100644
--- a/backend/core/models/blueprint.go
+++ b/backend/core/models/blueprint.go
@@ -37,13 +37,11 @@ type Blueprint struct {
Enable bool `json:"enable"`
CronConfig string `json:"cronConfig" format:"* * * * *" example:"0 0 * * 1"`
IsManual bool `json:"isManual"`
- SkipOnFail bool `json:"skipOnFail"`
BeforePlan PipelinePlan `json:"beforePlan" gorm:"serializer:encdec"`
AfterPlan PipelinePlan `json:"afterPlan" gorm:"serializer:encdec"`
- TimeAfter *time.Time `json:"timeAfter"`
- FullSync bool `json:"fullSync"`
Labels []string `json:"labels" gorm:"-"`
Connections []*BlueprintConnection `json:"connections" gorm:"-"`
+ SyncPolicy `gorm:"embedded"`
common.Model `swaggerignore:"true"`
}
diff --git a/backend/core/models/migrationscripts/20230907_add_full_sync.go b/backend/core/models/migrationscripts/20230911_add_sync_policy.go
similarity index 59%
rename from backend/core/models/migrationscripts/20230907_add_full_sync.go
rename to backend/core/models/migrationscripts/20230911_add_sync_policy.go
index 293ac69a9..c9e91bf59 100644
--- a/backend/core/models/migrationscripts/20230907_add_full_sync.go
+++ b/backend/core/models/migrationscripts/20230911_add_sync_policy.go
@@ -18,41 +18,46 @@ limitations under the License.
package migrationscripts
import (
+ "time"
+
"github.com/apache/incubator-devlake/core/context"
"github.com/apache/incubator-devlake/core/errors"
"github.com/apache/incubator-devlake/helpers/migrationhelper"
)
-type addFullSyncToBlueprint struct {
- FullSync bool `json:"fullSync"`
+type addSyncPolicyToBlueprint struct {
+ FullSync bool `json:"fullSync"`
+ SkipCollectors bool `json:"skipCollectors"`
}
-func (*addFullSyncToBlueprint) TableName() string {
+func (*addSyncPolicyToBlueprint) TableName() string {
return "_devlake_blueprints"
}
-type addFullSyncToPipeline struct {
- FullSync bool `json:"fullSync"`
+type addSyncPolicyToPipeline struct {
+ FullSync bool `json:"fullSync"`
+ SkipCollectors bool `json:"skipCollectors"`
+ TimeAfter *time.Time `json:"timeAfter"`
}
-func (*addFullSyncToPipeline) TableName() string {
+func (*addSyncPolicyToPipeline) TableName() string {
return "_devlake_pipelines"
}
-type addFullSync struct{}
+type addSyncPolicy struct{}
-func (*addFullSync) Up(basicRes context.BasicRes) errors.Error {
+func (*addSyncPolicy) Up(basicRes context.BasicRes) errors.Error {
return migrationhelper.AutoMigrateTables(
basicRes,
- &addFullSyncToBlueprint{},
- &addFullSyncToPipeline{},
+ &addSyncPolicyToBlueprint{},
+ &addSyncPolicyToPipeline{},
)
}
-func (*addFullSync) Version() uint64 {
- return 20230907000041
+func (*addSyncPolicy) Version() uint64 {
+ return 20230911000041
}
-func (*addFullSync) Name() string {
- return "add full_sync to _devlake_blueprints and _devlake_pipelines table"
+func (*addSyncPolicy) Name() string {
+ return "add sync policy to _devlake_blueprints and _devlake_pipelines table"
}
diff --git a/backend/core/models/migrationscripts/register.go b/backend/core/models/migrationscripts/register.go
index 7c86af0d4..3874e9d91 100644
--- a/backend/core/models/migrationscripts/register.go
+++ b/backend/core/models/migrationscripts/register.go
@@ -93,6 +93,6 @@ func All() []plugin.MigrationScript {
new(dropTapStateTable),
new(addCICDDeploymentsTable),
new(normalizeBpSettings),
- new(addFullSync),
+ new(addSyncPolicy),
}
}
diff --git a/backend/core/models/pipeline.go b/backend/core/models/pipeline.go
index 1bbb21086..4a0c944d9 100644
--- a/backend/core/models/pipeline.go
+++ b/backend/core/models/pipeline.go
@@ -69,8 +69,7 @@ type Pipeline struct {
SpentSeconds int `json:"spentSeconds"`
Stage int `json:"stage"`
Labels []string `json:"labels" gorm:"-"`
- SkipOnFail bool `json:"skipOnFail"`
- FullSync bool `json:"fullSync"`
+ SyncPolicy `gorm:"embedded"`
}
// We use a 2D array because the request body must be an array of a set of tasks
@@ -79,9 +78,8 @@ type NewPipeline struct {
Name string `json:"name"`
Plan PipelinePlan `json:"plan" swaggertype:"array,string" example:"please check api /pipelines/<PLUGIN_NAME>/pipeline-plan"`
Labels []string `json:"labels"`
- SkipOnFail bool `json:"skipOnFail"`
- FullSync bool `json:"fullSync"`
BlueprintId uint64
+ SyncPolicy `gorm:"embedded"`
}
func (Pipeline) TableName() string {
diff --git a/backend/core/plugin/plugin_blueprint.go b/backend/core/plugin/plugin_blueprint.go
index db3412540..492e54d43 100644
--- a/backend/core/plugin/plugin_blueprint.go
+++ b/backend/core/plugin/plugin_blueprint.go
@@ -72,7 +72,6 @@ type DataSourcePluginBlueprintV200 interface {
MakeDataSourcePipelinePlanV200(
connectionId uint64,
scopes []*models.BlueprintScope,
- syncPolicy *models.SyncPolicy,
) (models.PipelinePlan, []Scope, errors.Error)
}
diff --git a/backend/core/runner/directrun.go b/backend/core/runner/directrun.go
index 7578d8f39..8d596651d 100644
--- a/backend/core/runner/directrun.go
+++ b/backend/core/runner/directrun.go
@@ -26,6 +26,7 @@ import (
"os/signal"
"runtime"
"syscall"
+ "time"
"github.com/apache/incubator-devlake/core/models"
"github.com/apache/incubator-devlake/core/models/migrationscripts"
@@ -47,7 +48,7 @@ func RunCmd(cmd *cobra.Command) {
// args: command line arguments
// pluginTask: specific built-in plugin, for example: feishu, jira...
// options: plugin config
-func DirectRun(cmd *cobra.Command, args []string, pluginTask plugin.PluginTask, options map[string]interface{}) {
+func DirectRun(cmd *cobra.Command, args []string, pluginTask plugin.PluginTask, options map[string]interface{}, timeAfter string) {
basicRes := CreateAppBasicRes()
tasks, err := cmd.Flags().GetStringSlice("subtasks")
if err != nil {
@@ -84,13 +85,23 @@ func DirectRun(cmd *cobra.Command, args []string, pluginTask plugin.PluginTask,
Options: options,
Subtasks: tasks,
}
+ parsedTimeAfter := time.Time{}
+ syncPolicy := models.SyncPolicy{}
+ if timeAfter != "" {
+ parsedTimeAfter, err = time.Parse(time.RFC3339, timeAfter)
+ if err != nil {
+ panic(err)
+ }
+ syncPolicy.TimeAfter = &parsedTimeAfter
+ }
+
err = RunPluginSubTasks(
ctx,
basicRes,
task,
pluginTask,
nil,
- nil,
+ &syncPolicy,
)
if err != nil {
panic(err)
diff --git a/backend/core/runner/run_task.go b/backend/core/runner/run_task.go
index 009e4f951..4cc5eb251 100644
--- a/backend/core/runner/run_task.go
+++ b/backend/core/runner/run_task.go
@@ -53,9 +53,6 @@ func RunTask(
if err := db.First(dbPipeline, dal.Where("id = ? ", task.PipelineId)); err != nil {
return err
}
- syncPolicy := &models.SyncPolicy{}
- syncPolicy.SkipOnFail = dbPipeline.SkipOnFail
- syncPolicy.FullSync = dbPipeline.FullSync
logger, err := getTaskLogger(basicRes.GetLogger(), task)
if err != nil {
@@ -140,7 +137,7 @@ func RunTask(
basicRes.ReplaceLogger(logger),
task,
progress,
- syncPolicy,
+ &dbPipeline.SyncPolicy,
)
return err
}
diff --git a/backend/helpers/pluginhelper/api/api_collector_with_state.go b/backend/helpers/pluginhelper/api/api_collector_with_state.go
index 43f48f689..813a75f73 100644
--- a/backend/helpers/pluginhelper/api/api_collector_with_state.go
+++ b/backend/helpers/pluginhelper/api/api_collector_with_state.go
@@ -42,7 +42,7 @@ type ApiCollectorStateManager struct {
}
// NewStatefulApiCollector create a new ApiCollectorStateManager
-func NewStatefulApiCollector(args RawDataSubTaskArgs, timeAfter *time.Time) (*ApiCollectorStateManager, errors.Error) {
+func NewStatefulApiCollector(args RawDataSubTaskArgs) (*ApiCollectorStateManager, errors.Error) {
db := args.Ctx.GetDal()
rawDataSubTask, err := NewRawDataSubTask(args)
@@ -61,6 +61,11 @@ func NewStatefulApiCollector(args RawDataSubTaskArgs, timeAfter *time.Time) (*Ap
return nil, errors.Default.Wrap(err, "failed to load JiraLatestCollectorMeta")
}
}
+ var timeAfter *time.Time
+ syncPolicy := args.Ctx.TaskContext().SyncPolicy()
+ if syncPolicy != nil && syncPolicy.TimeAfter != nil {
+ timeAfter = syncPolicy.TimeAfter
+ }
return &ApiCollectorStateManager{
RawDataSubTaskArgs: args,
LatestState: latestState,
@@ -73,16 +78,16 @@ func NewStatefulApiCollector(args RawDataSubTaskArgs, timeAfter *time.Time) (*Ap
func (m *ApiCollectorStateManager) IsIncremental() bool {
prevSyncTime := m.LatestState.LatestSuccessStart
prevTimeAfter := m.LatestState.TimeAfter
- currTimeAfter := m.TimeAfter
syncPolicy := m.Ctx.TaskContext().SyncPolicy()
-
if syncPolicy != nil && syncPolicy.FullSync {
return false
}
+
if prevSyncTime == nil {
return false
}
// if we cleared the timeAfter, or moved timeAfter back in time, we should do a full sync
+ currTimeAfter := syncPolicy.TimeAfter
if currTimeAfter != nil {
return prevTimeAfter == nil || !currTimeAfter.Before(*prevTimeAfter)
}
@@ -123,6 +128,7 @@ func (m *ApiCollectorStateManager) Execute() errors.Error {
db := m.Ctx.GetDal()
m.LatestState.LatestSuccessStart = &m.ExecuteStart
m.LatestState.TimeAfter = m.TimeAfter
+
return db.CreateOrUpdate(&m.LatestState)
}
@@ -154,12 +160,11 @@ func NewStatefulApiCollectorForFinalizableEntity(args FinalizableApiCollectorArg
Options: args.Options,
Params: args.Params,
Table: args.Table,
- }, args.TimeAfter)
+ })
if err != nil {
return nil, err
}
- // // prepare the basic variables
var isIncremental = manager.IsIncremental()
var createdAfter *time.Time
if isIncremental {
diff --git a/backend/plugins/ae/ae.go b/backend/plugins/ae/ae.go
index 59fa374c3..4d0d4a238 100644
--- a/backend/plugins/ae/ae.go
+++ b/backend/plugins/ae/ae.go
@@ -30,12 +30,13 @@ func main() {
aeCmd := &cobra.Command{Use: "ae"}
connectionId := aeCmd.Flags().Uint64P("Connection-id", "c", 0, "ae connection id")
projectId := aeCmd.Flags().IntP("project-id", "p", 0, "ae project id")
+ timeAfter := aeCmd.Flags().StringP("time-after", "a", "", "collect data that are created after specified time, ie 2006-01-02T15:04:05Z")
_ = aeCmd.MarkFlagRequired("project-id")
aeCmd.Run = func(cmd *cobra.Command, args []string) {
runner.DirectRun(cmd, args, PluginEntry, map[string]interface{}{
"connectionId": *connectionId,
"projectId": *projectId,
- })
+ }, *timeAfter)
}
runner.RunCmd(aeCmd)
}
diff --git a/backend/plugins/bamboo/api/blueprint_V200_test.go b/backend/plugins/bamboo/api/blueprint_V200_test.go
index 5fb5c8a86..bd3d1abe3 100644
--- a/backend/plugins/bamboo/api/blueprint_V200_test.go
+++ b/backend/plugins/bamboo/api/blueprint_V200_test.go
@@ -48,7 +48,6 @@ func TestMakeDataSourcePipelinePlanV200(t *testing.T) {
const testScopeConfigName string = "bamboo scope config"
const testProxy string = ""
- syncPolicy := &coreModels.SyncPolicy{}
bpScopes := []*coreModels.BlueprintScope{
{
ScopeId: testKey,
@@ -148,7 +147,7 @@ func TestMakeDataSourcePipelinePlanV200(t *testing.T) {
})
Init(basicRes, mockMeta)
- plans, scopes, err := MakePipelinePlanV200(testSubTaskMeta, testConnectionID, bpScopes, syncPolicy)
+ plans, scopes, err := MakePipelinePlanV200(testSubTaskMeta, testConnectionID, bpScopes)
assert.Equal(t, err, nil)
assert.Equal(t, expectPlans, plans)
diff --git a/backend/plugins/bamboo/api/blueprint_v200.go b/backend/plugins/bamboo/api/blueprint_v200.go
index bbd2a3ae8..1b252a2eb 100644
--- a/backend/plugins/bamboo/api/blueprint_v200.go
+++ b/backend/plugins/bamboo/api/blueprint_v200.go
@@ -36,7 +36,6 @@ func MakePipelinePlanV200(
subtaskMetas []plugin.SubTaskMeta,
connectionId uint64,
scope []*coreModels.BlueprintScope,
- syncPolicy *coreModels.SyncPolicy,
) (coreModels.PipelinePlan, []plugin.Scope, errors.Error) {
var err errors.Error
connection := new(models.BambooConnection)
@@ -50,7 +49,7 @@ func MakePipelinePlanV200(
return nil, nil, err
}
- pp, err := makePipelinePlanV200(subtaskMetas, scope, connection, syncPolicy)
+ pp, err := makePipelinePlanV200(subtaskMetas, scope, connection)
if err != nil {
return nil, nil, err
}
@@ -84,7 +83,7 @@ func makeScopeV200(connectionId uint64, scopes []*coreModels.BlueprintScope) ([]
func makePipelinePlanV200(
subtaskMetas []plugin.SubTaskMeta,
scopes []*coreModels.BlueprintScope,
- connection *models.BambooConnection, syncPolicy *coreModels.SyncPolicy,
+ connection *models.BambooConnection,
) (coreModels.PipelinePlan, errors.Error) {
plans := make(coreModels.PipelinePlan, 0, len(scopes))
for _, scope := range scopes {
diff --git a/backend/plugins/bamboo/bamboo.go b/backend/plugins/bamboo/bamboo.go
index c89d0bc63..625ea5484 100644
--- a/backend/plugins/bamboo/bamboo.go
+++ b/backend/plugins/bamboo/bamboo.go
@@ -31,6 +31,7 @@ func main() {
bambooCmd := &cobra.Command{Use: "bamboo"}
connectionId := bambooCmd.Flags().Uint64P("Connection-id", "c", 0, "bamboo connection id")
projectKey := bambooCmd.Flags().StringP("project-key", "p", "", "bamboo project key")
+ timeAfter := bambooCmd.Flags().StringP("time-after", "a", "", "collect data that are created after specified time, ie 2006-01-02T15:04:05Z")
deploymentPattern := bambooCmd.Flags().StringP("deployment", "", "", "deployment pattern")
productionPattern := bambooCmd.Flags().StringP("production", "", "", "production pattern")
_ = bambooCmd.MarkFlagRequired("project-key")
@@ -42,7 +43,7 @@ func main() {
"deploymentPattern": *deploymentPattern,
"productionPattern": *productionPattern,
},
- })
+ }, *timeAfter)
}
runner.RunCmd(bambooCmd)
}
diff --git a/backend/plugins/bamboo/impl/impl.go b/backend/plugins/bamboo/impl/impl.go
index 69d77b3f4..529089a1e 100644
--- a/backend/plugins/bamboo/impl/impl.go
+++ b/backend/plugins/bamboo/impl/impl.go
@@ -68,9 +68,8 @@ func (p Bamboo) ScopeConfig() dal.Tabler {
func (p Bamboo) MakeDataSourcePipelinePlanV200(
connectionId uint64,
scopes []*coreModels.BlueprintScope,
- syncPolicy *coreModels.SyncPolicy,
) (coreModels.PipelinePlan, []plugin.Scope, errors.Error) {
- return api.MakePipelinePlanV200(p.SubTaskMetas(), connectionId, scopes, syncPolicy)
+ return api.MakePipelinePlanV200(p.SubTaskMetas(), connectionId, scopes)
}
func (p Bamboo) GetTablesInfo() []dal.Tabler {
diff --git a/backend/plugins/bitbucket/api/blueprint_V200_test.go b/backend/plugins/bitbucket/api/blueprint_V200_test.go
index 0a878c2e1..eb9d3534f 100644
--- a/backend/plugins/bitbucket/api/blueprint_V200_test.go
+++ b/backend/plugins/bitbucket/api/blueprint_V200_test.go
@@ -68,10 +68,9 @@ func TestMakeDataSourcePipelinePlanV200(t *testing.T) {
}
bpScopes := make([]*coreModels.BlueprintScope, 0)
bpScopes = append(bpScopes, bs)
- syncPolicy := &coreModels.SyncPolicy{}
plan := make(coreModels.PipelinePlan, len(bpScopes))
- plan, err = makeDataSourcePipelinePlanV200(nil, plan, bpScopes, connection, syncPolicy)
+ plan, err = makeDataSourcePipelinePlanV200(nil, plan, bpScopes, connection)
assert.Nil(t, err)
scopes, err := makeScopesV200(bpScopes, connection)
assert.Nil(t, err)
diff --git a/backend/plugins/bitbucket/api/blueprint_v200.go b/backend/plugins/bitbucket/api/blueprint_v200.go
index e438a4a68..d1b562103 100644
--- a/backend/plugins/bitbucket/api/blueprint_v200.go
+++ b/backend/plugins/bitbucket/api/blueprint_v200.go
@@ -19,7 +19,6 @@ package api
import (
"net/url"
- "time"
"github.com/apache/incubator-devlake/core/errors"
coreModels "github.com/apache/incubator-devlake/core/models"
@@ -39,7 +38,6 @@ func MakeDataSourcePipelinePlanV200(
subtaskMetas []plugin.SubTaskMeta,
connectionId uint64,
bpScopes []*coreModels.BlueprintScope,
- syncPolicy *coreModels.SyncPolicy,
) (coreModels.PipelinePlan, []plugin.Scope, errors.Error) {
// get the connection info for url
connection := &models.BitbucketConnection{}
@@ -49,7 +47,7 @@ func MakeDataSourcePipelinePlanV200(
}
plan := make(coreModels.PipelinePlan, len(bpScopes))
- plan, err = makeDataSourcePipelinePlanV200(subtaskMetas, plan, bpScopes, connection, syncPolicy)
+ plan, err = makeDataSourcePipelinePlanV200(subtaskMetas, plan, bpScopes, connection)
if err != nil {
return nil, nil, err
}
@@ -66,7 +64,6 @@ func makeDataSourcePipelinePlanV200(
plan coreModels.PipelinePlan,
bpScopes []*coreModels.BlueprintScope,
connection *models.BitbucketConnection,
- syncPolicy *coreModels.SyncPolicy,
) (coreModels.PipelinePlan, errors.Error) {
for i, bpScope := range bpScopes {
stage := plan[i]
@@ -101,9 +98,6 @@ func makeDataSourcePipelinePlanV200(
ConnectionId: repo.ConnectionId,
FullName: repo.BitbucketId,
}
- if syncPolicy.TimeAfter != nil {
- op.TimeAfter = syncPolicy.TimeAfter.Format(time.RFC3339)
- }
options, err := tasks.EncodeTaskOptions(op)
if err != nil {
return nil, err
diff --git a/backend/plugins/bitbucket/bitbucket.go b/backend/plugins/bitbucket/bitbucket.go
index fa36b7ad3..c578cd3d1 100644
--- a/backend/plugins/bitbucket/bitbucket.go
+++ b/backend/plugins/bitbucket/bitbucket.go
@@ -31,7 +31,7 @@ func main() {
cmd := &cobra.Command{Use: "bitbucket"}
connectionId := cmd.Flags().Uint64P("connectionId", "c", 0, "bitbucket connection id")
fullName := cmd.Flags().StringP("fullName", "n", "", "bitbucket id: owner/repo")
- timeAfter := cmd.Flags().StringP("timeAfter", "a", "", "collect data that are created after specified time, ie 2006-05-06T07:08:09Z")
+ timeAfter := cmd.Flags().StringP("timeAfter", "a", "", "collect data that are created after specified time, ie 2006-01-02T15:04:05Z")
deploymentPattern := cmd.Flags().StringP("deployment", "", "", "deployment pattern")
productionPattern := cmd.Flags().StringP("production", "", "", "production pattern")
_ = cmd.MarkFlagRequired("connectionId")
@@ -41,12 +41,11 @@ func main() {
runner.DirectRun(cmd, args, PluginEntry, map[string]interface{}{
"connectionId": *connectionId,
"fullName": *fullName,
- "timeAfter": *timeAfter,
"scopeConfig": map[string]string{
"deploymentPattern": *deploymentPattern,
"productionPattern": *productionPattern,
},
- })
+ }, *timeAfter)
}
runner.RunCmd(cmd)
}
diff --git a/backend/plugins/bitbucket/impl/impl.go b/backend/plugins/bitbucket/impl/impl.go
index 275e97377..57c243418 100644
--- a/backend/plugins/bitbucket/impl/impl.go
+++ b/backend/plugins/bitbucket/impl/impl.go
@@ -19,7 +19,6 @@ package impl
import (
"fmt"
- "time"
"github.com/apache/incubator-devlake/core/context"
"github.com/apache/incubator-devlake/core/dal"
@@ -164,13 +163,6 @@ func (p Bitbucket) PrepareTaskData(taskCtx plugin.TaskContext, options map[strin
return nil, err
}
- var timeAfter time.Time
- if op.TimeAfter != "" {
- timeAfter, err = errors.Convert01(time.Parse(time.RFC3339, op.TimeAfter))
- if err != nil {
- return nil, errors.BadInput.Wrap(err, "invalid value for `timeAfter`")
- }
- }
regexEnricher := helper.NewRegexEnricher()
if err := regexEnricher.TryAdd(devops.DEPLOYMENT, op.DeploymentPattern); err != nil {
return nil, errors.BadInput.Wrap(err, "invalid value for `deploymentPattern`")
@@ -183,10 +175,6 @@ func (p Bitbucket) PrepareTaskData(taskCtx plugin.TaskContext, options map[strin
ApiClient: apiClient,
RegexEnricher: regexEnricher,
}
- if !timeAfter.IsZero() {
- taskData.TimeAfter = &timeAfter
- logger.Debug("collect data updated timeAfter %s", timeAfter)
- }
return taskData, nil
}
@@ -201,9 +189,8 @@ func (p Bitbucket) MigrationScripts() []plugin.MigrationScript {
func (p Bitbucket) MakeDataSourcePipelinePlanV200(
connectionId uint64,
- scopes []*coreModels.BlueprintScope,
- syncPolicy *coreModels.SyncPolicy) (pp coreModels.PipelinePlan, sc []plugin.Scope, err errors.Error) {
- return api.MakeDataSourcePipelinePlanV200(p.SubTaskMetas(), connectionId, scopes, syncPolicy)
+ scopes []*coreModels.BlueprintScope) (pp coreModels.PipelinePlan, sc []plugin.Scope, err errors.Error) {
+ return api.MakeDataSourcePipelinePlanV200(p.SubTaskMetas(), connectionId, scopes)
}
func (p Bitbucket) ApiResources() map[string]map[string]plugin.ApiResourceHandler {
diff --git a/backend/plugins/bitbucket/tasks/api_common.go b/backend/plugins/bitbucket/tasks/api_common.go
index 5aa43a0fe..284df3df2 100644
--- a/backend/plugins/bitbucket/tasks/api_common.go
+++ b/backend/plugins/bitbucket/tasks/api_common.go
@@ -20,15 +20,16 @@ package tasks
import (
"encoding/json"
"fmt"
- "github.com/apache/incubator-devlake/core/dal"
- "github.com/apache/incubator-devlake/core/errors"
- plugin "github.com/apache/incubator-devlake/core/plugin"
- "github.com/apache/incubator-devlake/helpers/pluginhelper/api"
"io"
"net/http"
"net/url"
"reflect"
"time"
+
+ "github.com/apache/incubator-devlake/core/dal"
+ "github.com/apache/incubator-devlake/core/errors"
+ plugin "github.com/apache/incubator-devlake/core/plugin"
+ "github.com/apache/incubator-devlake/helpers/pluginhelper/api"
)
type BitbucketApiParams struct {
diff --git a/backend/plugins/bitbucket/tasks/issue_collector.go b/backend/plugins/bitbucket/tasks/issue_collector.go
index ef5f56fd5..a5f2e4a41 100644
--- a/backend/plugins/bitbucket/tasks/issue_collector.go
+++ b/backend/plugins/bitbucket/tasks/issue_collector.go
@@ -35,7 +35,7 @@ var CollectApiIssuesMeta = plugin.SubTaskMeta{
func CollectApiIssues(taskCtx plugin.SubTaskContext) errors.Error {
rawDataSubTaskArgs, data := CreateRawDataSubTaskArgs(taskCtx, RAW_ISSUE_TABLE)
- collectorWithState, err := helper.NewStatefulApiCollector(*rawDataSubTaskArgs, data.TimeAfter)
+ collectorWithState, err := helper.NewStatefulApiCollector(*rawDataSubTaskArgs)
if err != nil {
return err
}
diff --git a/backend/plugins/bitbucket/tasks/issue_comment_collector.go b/backend/plugins/bitbucket/tasks/issue_comment_collector.go
index 8f85ceace..8e0036c4a 100644
--- a/backend/plugins/bitbucket/tasks/issue_comment_collector.go
+++ b/backend/plugins/bitbucket/tasks/issue_comment_collector.go
@@ -36,7 +36,7 @@ var CollectApiIssueCommentsMeta = plugin.SubTaskMeta{
func CollectApiIssueComments(taskCtx plugin.SubTaskContext) errors.Error {
rawDataSubTaskArgs, data := CreateRawDataSubTaskArgs(taskCtx, RAW_ISSUE_COMMENTS_TABLE)
- collectorWithState, err := helper.NewStatefulApiCollector(*rawDataSubTaskArgs, data.TimeAfter)
+ collectorWithState, err := helper.NewStatefulApiCollector(*rawDataSubTaskArgs)
if err != nil {
return err
}
diff --git a/backend/plugins/bitbucket/tasks/pipeline_collector.go b/backend/plugins/bitbucket/tasks/pipeline_collector.go
index 58a5f6a4c..e25861936 100644
--- a/backend/plugins/bitbucket/tasks/pipeline_collector.go
+++ b/backend/plugins/bitbucket/tasks/pipeline_collector.go
@@ -35,7 +35,7 @@ var CollectApiPipelinesMeta = plugin.SubTaskMeta{
func CollectApiPipelines(taskCtx plugin.SubTaskContext) errors.Error {
rawDataSubTaskArgs, data := CreateRawDataSubTaskArgs(taskCtx, RAW_PIPELINE_TABLE)
- collectorWithState, err := helper.NewStatefulApiCollector(*rawDataSubTaskArgs, data.TimeAfter)
+ collectorWithState, err := helper.NewStatefulApiCollector(*rawDataSubTaskArgs)
if err != nil {
return err
}
diff --git a/backend/plugins/bitbucket/tasks/pipeline_steps_collector.go b/backend/plugins/bitbucket/tasks/pipeline_steps_collector.go
index b03fdfefe..99cc1e39b 100644
--- a/backend/plugins/bitbucket/tasks/pipeline_steps_collector.go
+++ b/backend/plugins/bitbucket/tasks/pipeline_steps_collector.go
@@ -38,7 +38,7 @@ var CollectPipelineStepsMeta = plugin.SubTaskMeta{
func CollectPipelineSteps(taskCtx plugin.SubTaskContext) errors.Error {
rawDataSubTaskArgs, data := CreateRawDataSubTaskArgs(taskCtx, RAW_PIPELINE_STEPS_TABLE)
- collectorWithState, err := helper.NewStatefulApiCollector(*rawDataSubTaskArgs, data.TimeAfter)
+ collectorWithState, err := helper.NewStatefulApiCollector(*rawDataSubTaskArgs)
if err != nil {
return err
}
diff --git a/backend/plugins/bitbucket/tasks/pr_collector.go b/backend/plugins/bitbucket/tasks/pr_collector.go
index aab2c6b13..7379cc1e4 100644
--- a/backend/plugins/bitbucket/tasks/pr_collector.go
+++ b/backend/plugins/bitbucket/tasks/pr_collector.go
@@ -38,7 +38,7 @@ var CollectApiPullRequestsMeta = plugin.SubTaskMeta{
func CollectApiPullRequests(taskCtx plugin.SubTaskContext) errors.Error {
rawDataSubTaskArgs, data := CreateRawDataSubTaskArgs(taskCtx, RAW_PULL_REQUEST_TABLE)
- collectorWithState, err := helper.NewStatefulApiCollector(*rawDataSubTaskArgs, data.TimeAfter)
+ collectorWithState, err := helper.NewStatefulApiCollector(*rawDataSubTaskArgs)
if err != nil {
return err
}
diff --git a/backend/plugins/bitbucket/tasks/pr_comment_collector.go b/backend/plugins/bitbucket/tasks/pr_comment_collector.go
index 70bd44433..c4d539f85 100644
--- a/backend/plugins/bitbucket/tasks/pr_comment_collector.go
+++ b/backend/plugins/bitbucket/tasks/pr_comment_collector.go
@@ -36,7 +36,7 @@ var CollectApiPrCommentsMeta = plugin.SubTaskMeta{
func CollectApiPullRequestsComments(taskCtx plugin.SubTaskContext) errors.Error {
rawDataSubTaskArgs, data := CreateRawDataSubTaskArgs(taskCtx, RAW_PULL_REQUEST_COMMENTS_TABLE)
- collectorWithState, err := helper.NewStatefulApiCollector(*rawDataSubTaskArgs, data.TimeAfter)
+ collectorWithState, err := helper.NewStatefulApiCollector(*rawDataSubTaskArgs)
if err != nil {
return err
}
diff --git a/backend/plugins/bitbucket/tasks/pr_commit_collector.go b/backend/plugins/bitbucket/tasks/pr_commit_collector.go
index 14c53d5d2..9d99da39c 100644
--- a/backend/plugins/bitbucket/tasks/pr_commit_collector.go
+++ b/backend/plugins/bitbucket/tasks/pr_commit_collector.go
@@ -19,10 +19,11 @@ package tasks
import (
"fmt"
+ "net/url"
+
"github.com/apache/incubator-devlake/core/errors"
"github.com/apache/incubator-devlake/core/plugin"
helper "github.com/apache/incubator-devlake/helpers/pluginhelper/api"
- "net/url"
)
const RAW_PULL_REQUEST_COMMITS_TABLE = "bitbucket_api_pull_request_commits"
@@ -37,7 +38,7 @@ var CollectApiPrCommitsMeta = plugin.SubTaskMeta{
func CollectApiPullRequestCommits(taskCtx plugin.SubTaskContext) errors.Error {
rawDataSubTaskArgs, data := CreateRawDataSubTaskArgs(taskCtx, RAW_PULL_REQUEST_COMMITS_TABLE)
- collectorWithState, err := helper.NewStatefulApiCollector(*rawDataSubTaskArgs, data.TimeAfter)
+ collectorWithState, err := helper.NewStatefulApiCollector(*rawDataSubTaskArgs)
if err != nil {
return err
}
diff --git a/backend/plugins/bitbucket/tasks/task_data.go b/backend/plugins/bitbucket/tasks/task_data.go
index 6ecb87689..235b05080 100644
--- a/backend/plugins/bitbucket/tasks/task_data.go
+++ b/backend/plugins/bitbucket/tasks/task_data.go
@@ -18,8 +18,6 @@ limitations under the License.
package tasks
import (
- "time"
-
"github.com/apache/incubator-devlake/core/errors"
"github.com/apache/incubator-devlake/helpers/pluginhelper/api"
"github.com/apache/incubator-devlake/plugins/bitbucket/models"
@@ -29,7 +27,6 @@ type BitbucketOptions struct {
ConnectionId uint64 `json:"connectionId" mapstructure:"connectionId,omitempty"`
Tasks []string `json:"tasks,omitempty" mapstructure:",omitempty"`
FullName string `json:"fullName" mapstructure:"fullName"`
- TimeAfter string `json:"timeAfter" mapstructure:"timeAfter,omitempty"`
ScopeConfigId uint64 `json:"scopeConfigId" mapstructure:"scopeConfigId,omitempty"`
*models.BitbucketScopeConfig `mapstructure:"scopeConfig,omitempty" json:"scopeConfig"`
}
@@ -37,7 +34,6 @@ type BitbucketOptions struct {
type BitbucketTaskData struct {
Options *BitbucketOptions
ApiClient *api.ApiAsyncClient
- TimeAfter *time.Time
RegexEnricher *api.RegexEnricher
}
diff --git a/backend/plugins/circleci/api/blueprint200.go b/backend/plugins/circleci/api/blueprint200.go
index 9f5a8d55b..57bb1e6ed 100644
--- a/backend/plugins/circleci/api/blueprint200.go
+++ b/backend/plugins/circleci/api/blueprint200.go
@@ -19,7 +19,6 @@ package api
import (
"fmt"
- "time"
"github.com/apache/incubator-devlake/core/models/domainlayer/devops"
"github.com/apache/incubator-devlake/plugins/circleci/models"
@@ -38,10 +37,9 @@ func MakeDataSourcePipelinePlanV200(
subtaskMetas []plugin.SubTaskMeta,
connectionId uint64,
bpScopes []*coreModels.BlueprintScope,
- syncPolicy *coreModels.SyncPolicy,
) (coreModels.PipelinePlan, []plugin.Scope, errors.Error) {
plan := make(coreModels.PipelinePlan, len(bpScopes))
- plan, err := makeDataSourcePipelinePlanV200(subtaskMetas, plan, bpScopes, connectionId, syncPolicy)
+ plan, err := makeDataSourcePipelinePlanV200(subtaskMetas, plan, bpScopes, connectionId)
if err != nil {
return nil, nil, err
}
@@ -58,7 +56,6 @@ func makeDataSourcePipelinePlanV200(
plan coreModels.PipelinePlan,
bpScopes []*coreModels.BlueprintScope,
connectionId uint64,
- syncPolicy *coreModels.SyncPolicy,
) (coreModels.PipelinePlan, errors.Error) {
for i, bpScope := range bpScopes {
stage := plan[i]
@@ -73,9 +70,6 @@ func makeDataSourcePipelinePlanV200(
}
options["projectSlug"] = project.Slug
options["connectionId"] = connectionId
- if syncPolicy.TimeAfter != nil {
- options["timeAfter"] = syncPolicy.TimeAfter.Format(time.RFC3339)
- }
// get scope config from db
_, scopeConfig, err := scopeHelper.DbHelper().GetScopeAndConfig(connectionId, bpScope.ScopeId)
diff --git a/backend/plugins/circleci/circleci.go b/backend/plugins/circleci/circleci.go
index 7466bd490..cf63d5a30 100644
--- a/backend/plugins/circleci/circleci.go
+++ b/backend/plugins/circleci/circleci.go
@@ -31,12 +31,13 @@ func main() {
cmd := &cobra.Command{Use: "CircleCi"}
connectionId := cmd.Flags().Uint64P("connection", "c", 0, "CircleCi connection id")
projectSlug := cmd.Flags().StringP("project", "p", "", "CircleCi project slug")
+ timeAfter := cmd.Flags().StringP("timeAfter", "a", "", "collect data that are created after specified time, ie 2006-01-02T15:04:05Z")
cmd.Run = func(cmd *cobra.Command, args []string) {
runner.DirectRun(cmd, args, PluginEntry, map[string]interface{}{
"connectionId": *connectionId,
"projectSlug": *projectSlug,
- })
+ }, *timeAfter)
}
runner.RunCmd(cmd)
diff --git a/backend/plugins/circleci/impl/impl.go b/backend/plugins/circleci/impl/impl.go
index 6d7a97601..df9d9605b 100644
--- a/backend/plugins/circleci/impl/impl.go
+++ b/backend/plugins/circleci/impl/impl.go
@@ -19,7 +19,6 @@ package impl
import (
"fmt"
- "time"
"github.com/apache/incubator-devlake/core/context"
"github.com/apache/incubator-devlake/core/dal"
@@ -90,9 +89,8 @@ func (p Circleci) SubTaskMetas() []plugin.SubTaskMeta {
func (p Circleci) MakeDataSourcePipelinePlanV200(
connectionId uint64,
scopes []*coreModels.BlueprintScope,
- syncPolicy *coreModels.SyncPolicy,
) (pp coreModels.PipelinePlan, sc []plugin.Scope, err errors.Error) {
- return api.MakeDataSourcePipelinePlanV200(p.SubTaskMetas(), connectionId, scopes, syncPolicy)
+ return api.MakeDataSourcePipelinePlanV200(p.SubTaskMetas(), connectionId, scopes)
}
func (p Circleci) PrepareTaskData(taskCtx plugin.TaskContext, options map[string]interface{}) (interface{}, errors.Error) {
@@ -119,16 +117,6 @@ func (p Circleci) PrepareTaskData(taskCtx plugin.TaskContext, options map[string
Options: op,
ApiClient: apiClient,
}
- var createdDateAfter time.Time
- if op.TimeAfter != "" {
- createdDateAfter, err = errors.Convert01(time.Parse(time.RFC3339, op.TimeAfter))
- if err != nil {
- return nil, errors.BadInput.Wrap(err, "invalid value for `createdDateAfter`")
- }
- }
- if !createdDateAfter.IsZero() {
- taskData.TimeAfter = &createdDateAfter
- }
// fallback to project scope config
if op.ScopeConfigId == 0 {
project := &models.CircleciProject{}
diff --git a/backend/plugins/circleci/tasks/task_data.go b/backend/plugins/circleci/tasks/task_data.go
index 6f5e3daf1..5c3028694 100644
--- a/backend/plugins/circleci/tasks/task_data.go
+++ b/backend/plugins/circleci/tasks/task_data.go
@@ -18,8 +18,6 @@ limitations under the License.
package tasks
import (
- "time"
-
"github.com/apache/incubator-devlake/core/errors"
helper "github.com/apache/incubator-devlake/helpers/pluginhelper/api"
"github.com/apache/incubator-devlake/plugins/circleci/models"
@@ -29,7 +27,6 @@ type CircleciOptions struct {
ConnectionId uint64 `json:"connectionId"`
ProjectSlug string `json:"projectSlug"`
PageSize uint64 `mapstruct:"pageSize"`
- TimeAfter string `json:"timeAfter" mapstructure:"timeAfter,omitempty"`
ScopeConfigId uint64 `json:"scopeConfigId" mapstructure:"scopeConfigId,omitempty"`
ScopeConfig *models.CircleciScopeConfig `json:"scopeConfig" mapstructure:"scopeConfig,omitempty"`
}
@@ -37,7 +34,6 @@ type CircleciOptions struct {
type CircleciTaskData struct {
Options *CircleciOptions
ApiClient *helper.ApiAsyncClient
- TimeAfter *time.Time
RegexEnricher *helper.RegexEnricher
}
diff --git a/backend/plugins/customize/customize.go b/backend/plugins/customize/customize.go
index 1211b39e3..e4771424b 100644
--- a/backend/plugins/customize/customize.go
+++ b/backend/plugins/customize/customize.go
@@ -29,6 +29,8 @@ var PluginEntry impl.Customize //nolint
// standalone mode for debugging
func main() {
cmd := &cobra.Command{Use: "customize"}
+ timeAfter := cmd.Flags().StringP("timeAfter", "a", "", "collect data that are created after specified time, ie 2006-01-02T15:04:05Z")
+
cmd.Run = func(c *cobra.Command, args []string) {
runner.DirectRun(c, args, PluginEntry, map[string]interface{}{"transformationRules": []tasks.MappingRules{{
Table: "issues",
@@ -36,7 +38,7 @@ func main() {
RawDataParams: "{\"ConnectionId\":1,\"BoardId\":8}",
Mapping: map[string]string{"x_test": "fields.timespent"},
}},
- })
+ }, *timeAfter)
}
runner.RunCmd(cmd)
}
diff --git a/backend/plugins/dbt/dbt.go b/backend/plugins/dbt/dbt.go
index 7cc225245..ffb435ddc 100644
--- a/backend/plugins/dbt/dbt.go
+++ b/backend/plugins/dbt/dbt.go
@@ -51,6 +51,7 @@ func main() {
projectVars["event_min_id"] = "7581"
projectVars["event_max_id"] = "7582"
dbtCmd.Flags().StringToStringVarP(&projectVars, "projectVars", "v", projectVars, "dbt provides variables to provide data to models for compilation.")
+ timeAfter := dbtCmd.Flags().StringP("timeAfter", "a", "", "collect data that are created after specified time, ie 2006-01-02T15:04:05Z")
dbtCmd.Run = func(cmd *cobra.Command, args []string) {
projectVarsConvert := make(map[string]interface{}, len(projectVars))
@@ -76,7 +77,7 @@ func main() {
"defer": *deferFlag,
"noDefer": *noDefer,
"fullRefresh": *fullRefresh,
- })
+ }, *timeAfter)
}
runner.RunCmd(dbtCmd)
}
diff --git a/backend/plugins/dora/dora.go b/backend/plugins/dora/dora.go
index a4ee412d3..a0b9a198a 100644
--- a/backend/plugins/dora/dora.go
+++ b/backend/plugins/dora/dora.go
@@ -31,11 +31,12 @@ func main() {
cmd := &cobra.Command{Use: "dora"}
projectName := cmd.Flags().StringP("projectName", "p", "", "project name")
+ timeAfter := cmd.Flags().StringP("timeAfter", "a", "", "collect data that are created after specified time, ie 2006-01-02T15:04:05Z")
cmd.Run = func(cmd *cobra.Command, args []string) {
runner.DirectRun(cmd, args, PluginEntry, map[string]interface{}{
"projectName": *projectName,
- })
+ }, *timeAfter)
}
runner.RunCmd(cmd)
}
diff --git a/backend/plugins/feishu/feishu.go b/backend/plugins/feishu/feishu.go
index c5c055c94..f203735aa 100644
--- a/backend/plugins/feishu/feishu.go
+++ b/backend/plugins/feishu/feishu.go
@@ -30,13 +30,15 @@ func main() {
cmd := &cobra.Command{Use: "feishu"}
connectionId := cmd.Flags().Uint64P("connectionId", "c", 0, "feishu connection id")
numOfDaysToCollect := cmd.Flags().IntP("numOfDaysToCollect", "n", 8, "feishu collect days")
+ timeAfter := cmd.Flags().StringP("timeAfter", "a", "", "collect data that are created after specified time, ie 2006-01-02T15:04:05Z")
+
_ = cmd.MarkFlagRequired("connectionId")
_ = cmd.MarkFlagRequired("numOfDaysToCollect")
cmd.Run = func(cmd *cobra.Command, args []string) {
runner.DirectRun(cmd, args, PluginEntry, map[string]interface{}{
"connectionId": *connectionId,
"numOfDaysToCollect": *numOfDaysToCollect,
- })
+ }, *timeAfter)
}
runner.RunCmd(cmd)
}
diff --git a/backend/plugins/gitee/gitee.go b/backend/plugins/gitee/gitee.go
index 9fbc67207..060734002 100644
--- a/backend/plugins/gitee/gitee.go
+++ b/backend/plugins/gitee/gitee.go
@@ -30,6 +30,7 @@ func main() {
connectionId := cmd.Flags().Uint64P("connectionId", "c", 0, "gitee connection id")
owner := cmd.Flags().StringP("owner", "o", "", "gitee owner")
repo := cmd.Flags().StringP("repo", "r", "", "gitee repo")
+ timeAfter := cmd.Flags().StringP("timeAfter", "a", "", "collect data that are created after specified time, ie 2006-01-02T15:04:05Z")
_ = cmd.MarkFlagRequired("connectionId")
_ = cmd.MarkFlagRequired("owner")
_ = cmd.MarkFlagRequired("repo")
@@ -58,7 +59,7 @@ func main() {
"issueTypeBug": *issueTypeBug,
"issueTypeIncident": *issueTypeIncident,
"issueTypeRequirement": *issueTypeRequirement,
- })
+ }, *timeAfter)
}
runner.RunCmd(cmd)
}
diff --git a/backend/plugins/github/api/blueprint_V200_test.go b/backend/plugins/github/api/blueprint_V200_test.go
index 718bcb5b2..96c4c5ce0 100644
--- a/backend/plugins/github/api/blueprint_V200_test.go
+++ b/backend/plugins/github/api/blueprint_V200_test.go
@@ -69,10 +69,9 @@ func TestMakeDataSourcePipelinePlanV200(t *testing.T) {
}
bpScopes := make([]*coreModels.BlueprintScope, 0)
bpScopes = append(bpScopes, bs)
- syncPolicy := &coreModels.SyncPolicy{}
plan := make(coreModels.PipelinePlan, len(bpScopes))
- plan, err = makeDataSourcePipelinePlanV200(nil, plan, bpScopes, connection, syncPolicy)
+ plan, err = makeDataSourcePipelinePlanV200(nil, plan, bpScopes, connection)
assert.Nil(t, err)
scopes, err := makeScopesV200(bpScopes, connection)
assert.Nil(t, err)
diff --git a/backend/plugins/github/api/blueprint_v200.go b/backend/plugins/github/api/blueprint_v200.go
index e405e52c3..b6b508930 100644
--- a/backend/plugins/github/api/blueprint_v200.go
+++ b/backend/plugins/github/api/blueprint_v200.go
@@ -25,7 +25,6 @@ import (
"net/http"
"net/url"
"strings"
- "time"
"github.com/apache/incubator-devlake/core/dal"
"github.com/apache/incubator-devlake/core/errors"
@@ -47,7 +46,6 @@ func MakeDataSourcePipelinePlanV200(
subtaskMetas []plugin.SubTaskMeta,
connectionId uint64,
bpScopes []*coreModels.BlueprintScope,
- syncPolicy *coreModels.SyncPolicy,
) (coreModels.PipelinePlan, []plugin.Scope, errors.Error) {
// get the connection info for url
connection := &models.GithubConnection{}
@@ -64,7 +62,7 @@ func MakeDataSourcePipelinePlanV200(
}
plan := make(coreModels.PipelinePlan, len(bpScopes))
- plan, err = makeDataSourcePipelinePlanV200(subtaskMetas, plan, bpScopes, connection, syncPolicy)
+ plan, err = makeDataSourcePipelinePlanV200(subtaskMetas, plan, bpScopes, connection)
if err != nil {
return nil, nil, err
}
@@ -81,7 +79,6 @@ func makeDataSourcePipelinePlanV200(
plan coreModels.PipelinePlan,
bpScopes []*coreModels.BlueprintScope,
connection *models.GithubConnection,
- syncPolicy *coreModels.SyncPolicy,
) (coreModels.PipelinePlan, errors.Error) {
for i, bpScope := range bpScopes {
stage := plan[i]
@@ -117,9 +114,6 @@ func makeDataSourcePipelinePlanV200(
GithubId: githubRepo.GithubId,
Name: githubRepo.FullName,
}
- if syncPolicy.TimeAfter != nil {
- op.TimeAfter = syncPolicy.TimeAfter.Format(time.RFC3339)
- }
options, err := tasks.EncodeTaskOptions(op)
if err != nil {
return nil, err
diff --git a/backend/plugins/github/github.go b/backend/plugins/github/github.go
index 5dd7f8c03..70d2c7830 100644
--- a/backend/plugins/github/github.go
+++ b/backend/plugins/github/github.go
@@ -32,7 +32,7 @@ func main() {
connectionId := cmd.Flags().Uint64P("connectionId", "c", 0, "github connection id")
owner := cmd.Flags().StringP("owner", "o", "", "github owner")
repo := cmd.Flags().StringP("repo", "r", "", "github repo")
- timeAfter := cmd.Flags().StringP("timeAfter", "a", "", "collect data that are created after specified time, ie 2006-05-06T07:08:09Z")
+ timeAfter := cmd.Flags().StringP("timeAfter", "a", "", "collect data that are created after specified time, ie 2006-01-02T15:04:05Z")
_ = cmd.MarkFlagRequired("connectionId")
_ = cmd.MarkFlagRequired("owner")
_ = cmd.MarkFlagRequired("repo")
@@ -54,7 +54,6 @@ func main() {
"connectionId": *connectionId,
"owner": *owner,
"repo": *repo,
- "timeAfter": *timeAfter,
"scopeConfig": map[string]interface{}{
"prType": *prType,
"prComponent": *prComponent,
@@ -68,7 +67,7 @@ func main() {
"deploymentPattern": *deploymentPattern,
"productionPattern": *productionPattern,
},
- })
+ }, *timeAfter)
}
runner.RunCmd(cmd)
}
diff --git a/backend/plugins/github/impl/impl.go b/backend/plugins/github/impl/impl.go
index 12481fe0e..2a2d0d4ed 100644
--- a/backend/plugins/github/impl/impl.go
+++ b/backend/plugins/github/impl/impl.go
@@ -19,7 +19,6 @@ package impl
import (
"fmt"
- "time"
"github.com/apache/incubator-devlake/helpers/pluginhelper/subtaskmeta/sorter"
@@ -159,15 +158,6 @@ func (p Github) PrepareTaskData(taskCtx plugin.TaskContext, options map[string]i
RegexEnricher: regexEnricher,
}
- if op.TimeAfter != "" {
- var timeAfter time.Time
- timeAfter, err = errors.Convert01(time.Parse(time.RFC3339, op.TimeAfter))
- if err != nil {
- return nil, errors.BadInput.Wrap(err, "invalid value for `timeAfter`")
- }
- taskData.TimeAfter = &timeAfter
- logger.Debug("collect data updated timeAfter %s", timeAfter)
- }
return taskData, nil
}
@@ -226,9 +216,8 @@ func (p Github) ApiResources() map[string]map[string]plugin.ApiResourceHandler {
func (p Github) MakeDataSourcePipelinePlanV200(
connectionId uint64,
scopes []*coreModels.BlueprintScope,
- syncPolicy *coreModels.SyncPolicy,
) (pp coreModels.PipelinePlan, sc []plugin.Scope, err errors.Error) {
- return api.MakeDataSourcePipelinePlanV200(p.SubTaskMetas(), connectionId, scopes, syncPolicy)
+ return api.MakeDataSourcePipelinePlanV200(p.SubTaskMetas(), connectionId, scopes)
}
func (p Github) Close(taskCtx plugin.TaskContext) errors.Error {
diff --git a/backend/plugins/github/tasks/cicd_job_collector.go b/backend/plugins/github/tasks/cicd_job_collector.go
index 787476c31..73fa2ab0d 100644
--- a/backend/plugins/github/tasks/cicd_job_collector.go
+++ b/backend/plugins/github/tasks/cicd_job_collector.go
@@ -59,7 +59,7 @@ func CollectJobs(taskCtx plugin.SubTaskContext) errors.Error {
Name: data.Options.Name,
},
Table: RAW_JOB_TABLE,
- }, data.TimeAfter)
+ })
if err != nil {
return err
}
diff --git a/backend/plugins/github/tasks/cicd_run_collector.go b/backend/plugins/github/tasks/cicd_run_collector.go
index 4d9158fcf..f29958575 100644
--- a/backend/plugins/github/tasks/cicd_run_collector.go
+++ b/backend/plugins/github/tasks/cicd_run_collector.go
@@ -78,7 +78,6 @@ func CollectRuns(taskCtx plugin.SubTaskContext) errors.Error {
Table: RAW_RUN_TABLE,
},
ApiClient: data.ApiClient,
- TimeAfter: data.TimeAfter,
CollectNewRecordsByList: helper.FinalizableApiCollectorListArgs{
PageSize: PAGE_SIZE,
Concurrency: 10,
diff --git a/backend/plugins/github/tasks/comment_collector.go b/backend/plugins/github/tasks/comment_collector.go
index 402bfa4e8..21e57f853 100644
--- a/backend/plugins/github/tasks/comment_collector.go
+++ b/backend/plugins/github/tasks/comment_collector.go
@@ -53,12 +53,13 @@ func CollectApiComments(taskCtx plugin.SubTaskContext) errors.Error {
Name: data.Options.Name,
},
Table: RAW_COMMENTS_TABLE,
- }, data.TimeAfter)
+ })
if err != nil {
return err
}
incremental := collectorWithState.IsIncremental()
+ syncPolicy := taskCtx.TaskContext().SyncPolicy()
err = collectorWithState.InitCollector(helper.ApiCollectorArgs{
ApiClient: data.ApiClient,
PageSize: 100,
@@ -68,11 +69,11 @@ func CollectApiComments(taskCtx plugin.SubTaskContext) errors.Error {
Query: func(reqData *helper.RequestData) (url.Values, errors.Error) {
query := url.Values{}
query.Set("state", "all")
- if data.TimeAfter != nil {
+ if syncPolicy != nil && syncPolicy.TimeAfter != nil {
// Note that `since` is for filtering records by the `updated` time
// which is not ideal for semantic reasons and would result in slightly more records than expected.
// But we have no choice since it is the only available field we could exploit from the API.
- query.Set("since", data.TimeAfter.String())
+ query.Set("since", syncPolicy.TimeAfter.String())
}
// if incremental == true, we overwrite it
if incremental {
diff --git a/backend/plugins/github/tasks/commit_collector.go b/backend/plugins/github/tasks/commit_collector.go
index 33e005331..c8ed10780 100644
--- a/backend/plugins/github/tasks/commit_collector.go
+++ b/backend/plugins/github/tasks/commit_collector.go
@@ -53,12 +53,13 @@ func CollectApiCommits(taskCtx plugin.SubTaskContext) errors.Error {
Name: data.Options.Name,
},
Table: RAW_COMMIT_TABLE,
- }, data.TimeAfter)
+ })
if err != nil {
return err
}
incremental := collectorWithState.IsIncremental()
+ syncPolicy := taskCtx.TaskContext().SyncPolicy()
err = collectorWithState.InitCollector(helper.ApiCollectorArgs{
ApiClient: data.ApiClient,
PageSize: 100,
@@ -79,8 +80,8 @@ func CollectApiCommits(taskCtx plugin.SubTaskContext) errors.Error {
Query: func(reqData *helper.RequestData) (url.Values, errors.Error) {
query := url.Values{}
query.Set("state", "all")
- if data.TimeAfter != nil {
- query.Set("since", data.TimeAfter.String())
+ if syncPolicy != nil && syncPolicy.TimeAfter != nil {
+ query.Set("since", syncPolicy.TimeAfter.String())
}
if incremental {
query.Set("since", collectorWithState.LatestState.LatestSuccessStart.String())
diff --git a/backend/plugins/github/tasks/event_collector.go b/backend/plugins/github/tasks/event_collector.go
index 7652f068f..491cc021c 100644
--- a/backend/plugins/github/tasks/event_collector.go
+++ b/backend/plugins/github/tasks/event_collector.go
@@ -69,7 +69,6 @@ func CollectApiEvents(taskCtx plugin.SubTaskContext) errors.Error {
Table: RAW_EVENTS_TABLE,
},
ApiClient: data.ApiClient,
- TimeAfter: data.TimeAfter, // set to nil to disable timeFilter
CollectNewRecordsByList: helper.FinalizableApiCollectorListArgs{
PageSize: 100,
Concurrency: 10,
diff --git a/backend/plugins/github/tasks/issue_collector.go b/backend/plugins/github/tasks/issue_collector.go
index f88942319..551081c2d 100644
--- a/backend/plugins/github/tasks/issue_collector.go
+++ b/backend/plugins/github/tasks/issue_collector.go
@@ -53,12 +53,13 @@ func CollectApiIssues(taskCtx plugin.SubTaskContext) errors.Error {
Name: data.Options.Name,
},
Table: RAW_ISSUE_TABLE,
- }, data.TimeAfter)
+ })
if err != nil {
return err
}
incremental := collectorWithState.IsIncremental()
+ syncPolicy := taskCtx.TaskContext().SyncPolicy()
err = collectorWithState.InitCollector(helper.ApiCollectorArgs{
ApiClient: data.ApiClient,
PageSize: 100,
@@ -79,8 +80,8 @@ func CollectApiIssues(taskCtx plugin.SubTaskContext) errors.Error {
Query: func(reqData *helper.RequestData) (url.Values, errors.Error) {
query := url.Values{}
query.Set("state", "all")
- if data.TimeAfter != nil {
- query.Set("since", data.TimeAfter.String())
+ if syncPolicy != nil && syncPolicy.TimeAfter != nil {
+ query.Set("since", syncPolicy.TimeAfter.String())
}
if incremental {
query.Set("since", collectorWithState.LatestState.LatestSuccessStart.String())
diff --git a/backend/plugins/github/tasks/pr_collector.go b/backend/plugins/github/tasks/pr_collector.go
index f94e96539..a83cd3085 100644
--- a/backend/plugins/github/tasks/pr_collector.go
+++ b/backend/plugins/github/tasks/pr_collector.go
@@ -72,7 +72,6 @@ func CollectApiPullRequests(taskCtx plugin.SubTaskContext) errors.Error {
Table: RAW_PULL_REQUEST_TABLE,
},
ApiClient: data.ApiClient,
- TimeAfter: data.TimeAfter, // set to nil to disable timeFilter
CollectNewRecordsByList: helper.FinalizableApiCollectorListArgs{
PageSize: 100,
Concurrency: 10,
diff --git a/backend/plugins/github/tasks/pr_commit_collector.go b/backend/plugins/github/tasks/pr_commit_collector.go
index 30ac310e8..0a209ff2e 100644
--- a/backend/plugins/github/tasks/pr_commit_collector.go
+++ b/backend/plugins/github/tasks/pr_commit_collector.go
@@ -68,7 +68,7 @@ func CollectApiPullRequestCommits(taskCtx plugin.SubTaskContext) errors.Error {
Name: data.Options.Name,
},
Table: RAW_PR_COMMIT_TABLE,
- }, data.TimeAfter)
+ })
if err != nil {
return err
}
diff --git a/backend/plugins/github/tasks/pr_review_collector.go b/backend/plugins/github/tasks/pr_review_collector.go
index 268f565ce..18bbde6b3 100644
--- a/backend/plugins/github/tasks/pr_review_collector.go
+++ b/backend/plugins/github/tasks/pr_review_collector.go
@@ -60,7 +60,7 @@ func CollectApiPullRequestReviews(taskCtx plugin.SubTaskContext) errors.Error {
Name: data.Options.Name,
},
Table: RAW_PR_REVIEW_TABLE,
- }, data.TimeAfter)
+ })
if err != nil {
return err
}
diff --git a/backend/plugins/github/tasks/pr_review_comment_collector.go b/backend/plugins/github/tasks/pr_review_comment_collector.go
index 48661a335..bd03b032b 100644
--- a/backend/plugins/github/tasks/pr_review_comment_collector.go
+++ b/backend/plugins/github/tasks/pr_review_comment_collector.go
@@ -56,12 +56,13 @@ func CollectPrReviewComments(taskCtx plugin.SubTaskContext) errors.Error {
Name: data.Options.Name,
},
Table: RAW_PR_REVIEW_COMMENTS_TABLE,
- }, data.TimeAfter)
+ })
if err != nil {
return err
}
incremental := collectorWithState.IsIncremental()
+ syncPolicy := taskCtx.TaskContext().SyncPolicy()
err = collectorWithState.InitCollector(helper.ApiCollectorArgs{
ApiClient: data.ApiClient,
PageSize: 100,
@@ -76,11 +77,11 @@ func CollectPrReviewComments(taskCtx plugin.SubTaskContext) errors.Error {
UrlTemplate: "repos/{{ .Params.Name }}/pulls/comments",
Query: func(reqData *helper.RequestData) (url.Values, errors.Error) {
query := url.Values{}
- if data.TimeAfter != nil {
+ if syncPolicy != nil && syncPolicy.TimeAfter != nil {
// Note that `since` is for filtering records by the `updated` time
// which is not ideal for semantic reasons and would result in slightly more records than expected.
// But we have no choice since it is the only available field we could exploit from the API.
- query.Set("since", data.TimeAfter.String())
+ query.Set("since", syncPolicy.TimeAfter.String())
}
// if incremental == true, we overwrite it
if incremental {
diff --git a/backend/plugins/github/tasks/task_data.go b/backend/plugins/github/tasks/task_data.go
index 8dcc777f8..714dbbd95 100644
--- a/backend/plugins/github/tasks/task_data.go
+++ b/backend/plugins/github/tasks/task_data.go
@@ -20,7 +20,6 @@ package tasks
import (
"fmt"
"strings"
- "time"
"github.com/apache/incubator-devlake/core/errors"
helper "github.com/apache/incubator-devlake/helpers/pluginhelper/api"
@@ -31,7 +30,6 @@ type GithubOptions struct {
ConnectionId uint64 `json:"connectionId" mapstructure:"connectionId,omitempty"`
ScopeConfigId uint64 `json:"scopeConfigId" mapstructure:"scopeConfigId,omitempty"`
GithubId int `json:"githubId" mapstructure:"githubId,omitempty"`
- TimeAfter string `json:"timeAfter" mapstructure:"timeAfter,omitempty"`
Owner string `json:"owner" mapstructure:"owner,omitempty"`
Repo string `json:"repo" mapstructure:"repo,omitempty"`
Name string `json:"name" mapstructure:"name,omitempty"`
@@ -42,7 +40,6 @@ type GithubTaskData struct {
Options *GithubOptions
ApiClient *helper.ApiAsyncClient
GraphqlClient *helper.GraphqlAsyncClient
- TimeAfter *time.Time
RegexEnricher *helper.RegexEnricher
}
diff --git a/backend/plugins/github_graphql/impl/impl.go b/backend/plugins/github_graphql/impl/impl.go
index a1ef349a5..cf39093a3 100644
--- a/backend/plugins/github_graphql/impl/impl.go
+++ b/backend/plugins/github_graphql/impl/impl.go
@@ -232,15 +232,6 @@ func (p GithubGraphql) PrepareTaskData(taskCtx plugin.TaskContext, options map[s
GraphqlClient: graphqlClient,
RegexEnricher: regexEnricher,
}
- if op.TimeAfter != "" {
- var timeAfter time.Time
- timeAfter, err = errors.Convert01(time.Parse(time.RFC3339, op.TimeAfter))
- if err != nil {
- return nil, errors.BadInput.Wrap(err, "invalid value for `timeAfter`")
- }
- taskData.TimeAfter = &timeAfter
- logger.Debug("collect data updated timeAfter %s", timeAfter)
- }
return taskData, nil
}
diff --git a/backend/plugins/github_graphql/plugin_main.go b/backend/plugins/github_graphql/plugin_main.go
index e9edf721b..af79257b0 100644
--- a/backend/plugins/github_graphql/plugin_main.go
+++ b/backend/plugins/github_graphql/plugin_main.go
@@ -32,7 +32,7 @@ func main() {
connectionId := cmd.Flags().Uint64P("connectionId", "c", 0, "github connection id")
owner := cmd.Flags().StringP("owner", "o", "", "github owner")
repo := cmd.Flags().StringP("repo", "r", "", "github repo")
- timeAfter := cmd.Flags().StringP("timeAfter", "a", "", "collect data that are updated/created after specified time, ie 2006-05-06T07:08:09Z")
+ timeAfter := cmd.Flags().StringP("timeAfter", "a", "", "collect data that are created after specified time, ie 2006-01-02T15:04:05Z")
_ = cmd.MarkFlagRequired("connectionId")
_ = cmd.MarkFlagRequired("owner")
_ = cmd.MarkFlagRequired("repo")
@@ -42,8 +42,7 @@ func main() {
"connectionId": *connectionId,
"owner": *owner,
"repo": *repo,
- "timeAfter": *timeAfter,
- })
+ }, *timeAfter)
}
runner.RunCmd(cmd)
}
diff --git a/backend/plugins/github_graphql/tasks/issue_collector.go b/backend/plugins/github_graphql/tasks/issue_collector.go
index 768822494..7645db8d9 100644
--- a/backend/plugins/github_graphql/tasks/issue_collector.go
+++ b/backend/plugins/github_graphql/tasks/issue_collector.go
@@ -104,13 +104,12 @@ func CollectIssue(taskCtx plugin.SubTaskContext) errors.Error {
Name: data.Options.Name,
},
Table: RAW_ISSUES_TABLE,
- }, data.TimeAfter)
+ })
if err != nil {
return err
}
incremental := collectorWithState.IsIncremental()
-
err = collectorWithState.InitGraphQLCollector(helper.GraphqlCollectorArgs{
GraphqlClient: data.GraphqlClient,
PageSize: 100,
diff --git a/backend/plugins/github_graphql/tasks/job_collector.go b/backend/plugins/github_graphql/tasks/job_collector.go
index 805a86c5e..095276c4a 100644
--- a/backend/plugins/github_graphql/tasks/job_collector.go
+++ b/backend/plugins/github_graphql/tasks/job_collector.go
@@ -111,7 +111,7 @@ func CollectGraphqlJobs(taskCtx plugin.SubTaskContext) errors.Error {
Name: data.Options.Name,
},
Table: RAW_GRAPHQL_JOBS_TABLE,
- }, data.TimeAfter)
+ })
if err != nil {
return err
}
diff --git a/backend/plugins/github_graphql/tasks/pr_collector.go b/backend/plugins/github_graphql/tasks/pr_collector.go
index cc854c7f0..afd8ff0e8 100644
--- a/backend/plugins/github_graphql/tasks/pr_collector.go
+++ b/backend/plugins/github_graphql/tasks/pr_collector.go
@@ -155,13 +155,12 @@ func CollectPr(taskCtx plugin.SubTaskContext) errors.Error {
Name: data.Options.Name,
},
Table: RAW_PRS_TABLE,
- }, data.TimeAfter)
+ })
if err != nil {
return err
}
incremental := collectorWithState.IsIncremental()
-
err = collectorWithState.InitGraphQLCollector(api.GraphqlCollectorArgs{
GraphqlClient: data.GraphqlClient,
PageSize: 10,
diff --git a/backend/plugins/gitlab/api/blueprint_V200_test.go b/backend/plugins/gitlab/api/blueprint_V200_test.go
index 99bc27062..aa366b73b 100644
--- a/backend/plugins/gitlab/api/blueprint_V200_test.go
+++ b/backend/plugins/gitlab/api/blueprint_V200_test.go
@@ -52,7 +52,6 @@ func TestMakeDataSourcePipelinePlanV200(t *testing.T) {
const testScopeConfigName string = "gitlab scope config"
const testProxy string = ""
- syncPolicy := &coreModels.SyncPolicy{}
bpScopes := []*coreModels.BlueprintScope{
{
ScopeId: strconv.Itoa(testID),
@@ -201,7 +200,7 @@ func TestMakeDataSourcePipelinePlanV200(t *testing.T) {
})
Init(mockRes, mockMeta)
- plans, scopes, err := MakePipelinePlanV200(testSubTaskMeta, testConnectionID, bpScopes, syncPolicy)
+ plans, scopes, err := MakePipelinePlanV200(testSubTaskMeta, testConnectionID, bpScopes)
assert.Equal(t, err, nil)
assert.Equal(t, expectPlans, plans)
diff --git a/backend/plugins/gitlab/api/blueprint_v200.go b/backend/plugins/gitlab/api/blueprint_v200.go
index d7197cb5d..7b6fba3cb 100644
--- a/backend/plugins/gitlab/api/blueprint_v200.go
+++ b/backend/plugins/gitlab/api/blueprint_v200.go
@@ -24,7 +24,6 @@ import (
"net/http"
"net/url"
"strconv"
- "time"
"github.com/apache/incubator-devlake/plugins/gitlab/tasks"
@@ -46,7 +45,6 @@ func MakePipelinePlanV200(
subtaskMetas []plugin.SubTaskMeta,
connectionId uint64,
scope []*coreModels.BlueprintScope,
- syncPolicy *coreModels.SyncPolicy,
) (coreModels.PipelinePlan, []plugin.Scope, errors.Error) {
var err errors.Error
connection := new(models.GitlabConnection)
@@ -60,7 +58,7 @@ func MakePipelinePlanV200(
return nil, nil, err
}
- pp, err := makePipelinePlanV200(subtaskMetas, scope, connection, syncPolicy)
+ pp, err := makePipelinePlanV200(subtaskMetas, scope, connection)
if err != nil {
return nil, nil, err
}
@@ -114,7 +112,7 @@ func makeScopeV200(connectionId uint64, scopes []*coreModels.BlueprintScope) ([]
func makePipelinePlanV200(
subtaskMetas []plugin.SubTaskMeta,
scopes []*coreModels.BlueprintScope,
- connection *models.GitlabConnection, syncPolicy *coreModels.SyncPolicy,
+ connection *models.GitlabConnection,
) (coreModels.PipelinePlan, errors.Error) {
plans := make(coreModels.PipelinePlan, 0, 3*len(scopes))
for _, scope := range scopes {
@@ -137,9 +135,6 @@ func makePipelinePlanV200(
options["connectionId"] = connection.ID
options["projectId"] = intScopeId
options["scopeConfigId"] = scopeConfig.ID
- if syncPolicy.TimeAfter != nil {
- options["timeAfter"] = syncPolicy.TimeAfter.Format(time.RFC3339)
- }
// construct subtasks
subtasks, err := helper.MakePipelinePlanSubtasks(subtaskMetas, scopeConfig.Entities)
diff --git a/backend/plugins/gitlab/gitlab.go b/backend/plugins/gitlab/gitlab.go
index 5a6d2701c..187585315 100644
--- a/backend/plugins/gitlab/gitlab.go
+++ b/backend/plugins/gitlab/gitlab.go
@@ -31,7 +31,7 @@ func main() {
cmd := &cobra.Command{Use: "gitlab"}
projectId := cmd.Flags().IntP("project-id", "p", 0, "gitlab project id")
connectionId := cmd.Flags().Uint64P("connection-id", "c", 0, "gitlab connection id")
- timeAfter := cmd.Flags().StringP("timeAfter", "a", "", "collect data that are created after specified time, ie 2006-05-06T07:08:09Z")
+ timeAfter := cmd.Flags().StringP("time-after", "a", "", "collect data that are created after specified time, ie 2006-01-02T15:04:05Z")
_ = cmd.MarkFlagRequired("project-id")
_ = cmd.MarkFlagRequired("connection-id")
@@ -50,7 +50,6 @@ func main() {
runner.DirectRun(cmd, args, PluginEntry, map[string]interface{}{
"projectId": *projectId,
"connectionId": *connectionId,
- "timeAfter": *timeAfter,
"scopeConfig": map[string]interface{}{
"prType": *prType,
"prComponent": *prComponent,
@@ -64,7 +63,7 @@ func main() {
"deploymentPattern": *deploymentPattern,
"productionPattern": *productionPattern,
},
- })
+ }, *timeAfter)
}
runner.RunCmd(cmd)
diff --git a/backend/plugins/gitlab/impl/impl.go b/backend/plugins/gitlab/impl/impl.go
index 36c7c2ac2..1233802cd 100644
--- a/backend/plugins/gitlab/impl/impl.go
+++ b/backend/plugins/gitlab/impl/impl.go
@@ -19,7 +19,6 @@ package impl
import (
"fmt"
- "time"
"github.com/apache/incubator-devlake/helpers/pluginhelper/subtaskmeta/sorter"
@@ -77,9 +76,8 @@ func (p Gitlab) ScopeConfig() dal.Tabler {
func (p Gitlab) MakeDataSourcePipelinePlanV200(
connectionId uint64,
scopes []*coreModels.BlueprintScope,
- syncPolicy *coreModels.SyncPolicy,
) (coreModels.PipelinePlan, []plugin.Scope, errors.Error) {
- return api.MakePipelinePlanV200(p.SubTaskMetas(), connectionId, scopes, syncPolicy)
+ return api.MakePipelinePlanV200(p.SubTaskMetas(), connectionId, scopes)
}
func (p Gitlab) GetTablesInfo() []dal.Tabler {
@@ -151,14 +149,6 @@ func (p Gitlab) PrepareTaskData(taskCtx plugin.TaskContext, options map[string]i
return nil, err
}
- var timeAfter time.Time
- if op.TimeAfter != "" {
- timeAfter, err = errors.Convert01(time.Parse(time.RFC3339, op.TimeAfter))
- if err != nil {
- return nil, errors.BadInput.Wrap(err, "invalid value for `timeAfter`")
- }
- }
-
if op.ProjectId != 0 {
var scope *models.GitlabProject
// support v100 & advance mode
@@ -212,10 +202,6 @@ func (p Gitlab) PrepareTaskData(taskCtx plugin.TaskContext, options map[string]i
RegexEnricher: regexEnricher,
}
- if !timeAfter.IsZero() {
- taskData.TimeAfter = &timeAfter
- logger.Debug("collect data updated timeAfter %s", timeAfter)
- }
return &taskData, nil
}
diff --git a/backend/plugins/gitlab/tasks/issue_collector.go b/backend/plugins/gitlab/tasks/issue_collector.go
index a8b51c5c6..0829f9cde 100644
--- a/backend/plugins/gitlab/tasks/issue_collector.go
+++ b/backend/plugins/gitlab/tasks/issue_collector.go
@@ -46,7 +46,7 @@ var CollectApiIssuesMeta = plugin.SubTaskMeta{
func CollectApiIssues(taskCtx plugin.SubTaskContext) errors.Error {
rawDataSubTaskArgs, data := CreateRawDataSubTaskArgs(taskCtx, RAW_ISSUE_TABLE)
- collectorWithState, err := helper.NewStatefulApiCollector(*rawDataSubTaskArgs, data.TimeAfter)
+ collectorWithState, err := helper.NewStatefulApiCollector(*rawDataSubTaskArgs)
if err != nil {
return err
}
diff --git a/backend/plugins/gitlab/tasks/job_collector.go b/backend/plugins/gitlab/tasks/job_collector.go
index 6a7475784..755f8800b 100644
--- a/backend/plugins/gitlab/tasks/job_collector.go
+++ b/backend/plugins/gitlab/tasks/job_collector.go
@@ -55,7 +55,6 @@ func CollectApiJobs(taskCtx plugin.SubTaskContext) errors.Error {
collector, err := helper.NewStatefulApiCollectorForFinalizableEntity(helper.FinalizableApiCollectorArgs{
RawDataSubTaskArgs: *rawDataSubTaskArgs,
ApiClient: data.ApiClient,
- TimeAfter: data.TimeAfter, // set to nil to disable timeFilter
CollectNewRecordsByList: helper.FinalizableApiCollectorListArgs{
PageSize: 100,
Concurrency: 10,
diff --git a/backend/plugins/gitlab/tasks/mr_collector.go b/backend/plugins/gitlab/tasks/mr_collector.go
index 125262e60..c55fe760e 100644
--- a/backend/plugins/gitlab/tasks/mr_collector.go
+++ b/backend/plugins/gitlab/tasks/mr_collector.go
@@ -43,7 +43,7 @@ var CollectApiMergeRequestsMeta = plugin.SubTaskMeta{
func CollectApiMergeRequests(taskCtx plugin.SubTaskContext) errors.Error {
rawDataSubTaskArgs, data := CreateRawDataSubTaskArgs(taskCtx, RAW_MERGE_REQUEST_TABLE)
- collectorWithState, err := helper.NewStatefulApiCollector(*rawDataSubTaskArgs, data.TimeAfter)
+ collectorWithState, err := helper.NewStatefulApiCollector(*rawDataSubTaskArgs)
if err != nil {
return err
}
diff --git a/backend/plugins/gitlab/tasks/mr_commit_collector.go b/backend/plugins/gitlab/tasks/mr_commit_collector.go
index f84a58f1d..2fbbdd48a 100644
--- a/backend/plugins/gitlab/tasks/mr_commit_collector.go
+++ b/backend/plugins/gitlab/tasks/mr_commit_collector.go
@@ -40,7 +40,7 @@ var CollectApiMrCommitsMeta = plugin.SubTaskMeta{
func CollectApiMergeRequestsCommits(taskCtx plugin.SubTaskContext) errors.Error {
rawDataSubTaskArgs, data := CreateRawDataSubTaskArgs(taskCtx, RAW_MERGE_REQUEST_COMMITS_TABLE)
- collectorWithState, err := helper.NewStatefulApiCollector(*rawDataSubTaskArgs, data.TimeAfter)
+ collectorWithState, err := helper.NewStatefulApiCollector(*rawDataSubTaskArgs)
if err != nil {
return err
}
diff --git a/backend/plugins/gitlab/tasks/mr_detail_collector.go b/backend/plugins/gitlab/tasks/mr_detail_collector.go
index 0b3971d3f..5b9af48dc 100644
--- a/backend/plugins/gitlab/tasks/mr_detail_collector.go
+++ b/backend/plugins/gitlab/tasks/mr_detail_collector.go
@@ -44,7 +44,7 @@ var CollectApiMergeRequestDetailsMeta = plugin.SubTaskMeta{
func CollectApiMergeRequestDetails(taskCtx plugin.SubTaskContext) errors.Error {
rawDataSubTaskArgs, data := CreateRawDataSubTaskArgs(taskCtx, RAW_MERGE_REQUEST_DETAIL_TABLE)
- collectorWithState, err := helper.NewStatefulApiCollector(*rawDataSubTaskArgs, data.TimeAfter)
+ collectorWithState, err := helper.NewStatefulApiCollector(*rawDataSubTaskArgs)
if err != nil {
return err
}
diff --git a/backend/plugins/gitlab/tasks/mr_note_collector.go b/backend/plugins/gitlab/tasks/mr_note_collector.go
index a4d5f40ff..d6fde5751 100644
--- a/backend/plugins/gitlab/tasks/mr_note_collector.go
+++ b/backend/plugins/gitlab/tasks/mr_note_collector.go
@@ -40,7 +40,7 @@ var CollectApiMrNotesMeta = plugin.SubTaskMeta{
func CollectApiMergeRequestsNotes(taskCtx plugin.SubTaskContext) errors.Error {
rawDataSubTaskArgs, data := CreateRawDataSubTaskArgs(taskCtx, RAW_MERGE_REQUEST_NOTES_TABLE)
- collectorWithState, err := helper.NewStatefulApiCollector(*rawDataSubTaskArgs, data.TimeAfter)
+ collectorWithState, err := helper.NewStatefulApiCollector(*rawDataSubTaskArgs)
if err != nil {
return err
}
diff --git a/backend/plugins/gitlab/tasks/pipeline_collector.go b/backend/plugins/gitlab/tasks/pipeline_collector.go
index 1ef244887..c22cc872a 100644
--- a/backend/plugins/gitlab/tasks/pipeline_collector.go
+++ b/backend/plugins/gitlab/tasks/pipeline_collector.go
@@ -44,7 +44,7 @@ var CollectApiPipelinesMeta = plugin.SubTaskMeta{
func CollectApiPipelines(taskCtx plugin.SubTaskContext) errors.Error {
rawDataSubTaskArgs, data := CreateRawDataSubTaskArgs(taskCtx, RAW_PIPELINE_TABLE)
- collectorWithState, err := helper.NewStatefulApiCollector(*rawDataSubTaskArgs, data.TimeAfter)
+ collectorWithState, err := helper.NewStatefulApiCollector(*rawDataSubTaskArgs)
if err != nil {
return err
}
diff --git a/backend/plugins/gitlab/tasks/pipeline_detail_collector.go b/backend/plugins/gitlab/tasks/pipeline_detail_collector.go
index 856659adb..a37807b8c 100644
--- a/backend/plugins/gitlab/tasks/pipeline_detail_collector.go
+++ b/backend/plugins/gitlab/tasks/pipeline_detail_collector.go
@@ -45,7 +45,7 @@ var CollectApiPipelineDetailsMeta = plugin.SubTaskMeta{
func CollectApiPipelineDetails(taskCtx plugin.SubTaskContext) errors.Error {
rawDataSubTaskArgs, data := CreateRawDataSubTaskArgs(taskCtx, RAW_PIPELINE_DETAILS_TABLE)
- collectorWithState, err := helper.NewStatefulApiCollector(*rawDataSubTaskArgs, data.TimeAfter)
+ collectorWithState, err := helper.NewStatefulApiCollector(*rawDataSubTaskArgs)
if err != nil {
return err
}
diff --git a/backend/plugins/gitlab/tasks/task_data.go b/backend/plugins/gitlab/tasks/task_data.go
index a49d11389..45e4abe3b 100644
--- a/backend/plugins/gitlab/tasks/task_data.go
+++ b/backend/plugins/gitlab/tasks/task_data.go
@@ -18,8 +18,6 @@ limitations under the License.
package tasks
import (
- "time"
-
"github.com/apache/incubator-devlake/core/errors"
helper "github.com/apache/incubator-devlake/helpers/pluginhelper/api"
"github.com/apache/incubator-devlake/plugins/gitlab/models"
@@ -29,7 +27,6 @@ type GitlabOptions struct {
ConnectionId uint64 `mapstructure:"connectionId" json:"connectionId"`
ProjectId int `mapstructure:"projectId" json:"projectId"`
ScopeConfigId uint64 `mapstructure:"scopeConfigId" json:"scopeConfigId"`
- TimeAfter string `mapstructure:"timeAfter" json:"timeAfter"`
ScopeConfig *models.GitlabScopeConfig `mapstructure:"scopeConfig" json:"scopeConfig"`
}
@@ -37,7 +34,6 @@ type GitlabTaskData struct {
Options *GitlabOptions
ApiClient *helper.ApiAsyncClient
ProjectCommit *models.GitlabProjectCommit
- TimeAfter *time.Time
RegexEnricher *helper.RegexEnricher
}
diff --git a/backend/plugins/icla/plugin_main.go b/backend/plugins/icla/plugin_main.go
index b7183ed76..3af83c70a 100644
--- a/backend/plugins/icla/plugin_main.go
+++ b/backend/plugins/icla/plugin_main.go
@@ -29,8 +29,9 @@ var PluginEntry impl.Icla //nolint
// standalone mode for debugging
func main() {
cmd := &cobra.Command{Use: "icla"}
+ timeAfter := cmd.Flags().StringP("timeAfter", "a", "", "collect data that are created after specified time, ie 2006-01-02T15:04:05Z")
cmd.Run = func(cmd *cobra.Command, args []string) {
- runner.DirectRun(cmd, args, PluginEntry, map[string]interface{}{})
+ runner.DirectRun(cmd, args, PluginEntry, map[string]interface{}{}, *timeAfter)
}
runner.RunCmd(cmd)
}
diff --git a/backend/plugins/jenkins/api/blueprint_v200.go b/backend/plugins/jenkins/api/blueprint_v200.go
index 7dca09a03..10940fb62 100644
--- a/backend/plugins/jenkins/api/blueprint_v200.go
+++ b/backend/plugins/jenkins/api/blueprint_v200.go
@@ -18,8 +18,6 @@ limitations under the License.
package api
import (
- "time"
-
"github.com/apache/incubator-devlake/core/errors"
coreModels "github.com/apache/incubator-devlake/core/models"
"github.com/apache/incubator-devlake/core/models/domainlayer"
@@ -35,10 +33,9 @@ func MakeDataSourcePipelinePlanV200(
subtaskMetas []plugin.SubTaskMeta,
connectionId uint64,
bpScopes []*coreModels.BlueprintScope,
- syncPolicy *coreModels.SyncPolicy,
) (coreModels.PipelinePlan, []plugin.Scope, errors.Error) {
plan := make(coreModels.PipelinePlan, len(bpScopes))
- plan, err := makeDataSourcePipelinePlanV200(subtaskMetas, plan, bpScopes, connectionId, syncPolicy)
+ plan, err := makeDataSourcePipelinePlanV200(subtaskMetas, plan, bpScopes, connectionId)
if err != nil {
return nil, nil, err
}
@@ -55,7 +52,6 @@ func makeDataSourcePipelinePlanV200(
plan coreModels.PipelinePlan,
bpScopes []*coreModels.BlueprintScope,
connectionId uint64,
- syncPolicy *coreModels.SyncPolicy,
) (coreModels.PipelinePlan, errors.Error) {
for i, bpScope := range bpScopes {
stage := plan[i]
@@ -68,10 +64,6 @@ func makeDataSourcePipelinePlanV200(
options["scopeId"] = bpScope.ScopeId
options["connectionId"] = connectionId
- if syncPolicy.TimeAfter != nil {
- options["timeAfter"] = syncPolicy.TimeAfter.Format(time.RFC3339)
- }
-
// get scope config from db
_, scopeConfig, err := scopeHelper.DbHelper().GetScopeAndConfig(connectionId, bpScope.ScopeId)
if err != nil {
diff --git a/backend/plugins/jenkins/api/blueprint_v200_test.go b/backend/plugins/jenkins/api/blueprint_v200_test.go
index bfd61f573..561cc6c56 100644
--- a/backend/plugins/jenkins/api/blueprint_v200_test.go
+++ b/backend/plugins/jenkins/api/blueprint_v200_test.go
@@ -42,14 +42,14 @@ func TestMakeDataSourcePipelinePlanV200(t *testing.T) {
bs := &coreModels.BlueprintScope{
ScopeId: "a/b/ccc",
}
- syncPolicy := &coreModels.SyncPolicy{}
+
bpScopes := make([]*coreModels.BlueprintScope, 0)
bpScopes = append(bpScopes, bs)
mockBasicRes(t)
plan := make(coreModels.PipelinePlan, len(bpScopes))
- plan, err = makeDataSourcePipelinePlanV200(nil, plan, bpScopes, 1, syncPolicy)
+ plan, err = makeDataSourcePipelinePlanV200(nil, plan, bpScopes, 1)
assert.Nil(t, err)
scopes, err := makeScopesV200(bpScopes, 1)
assert.Nil(t, err)
diff --git a/backend/plugins/jenkins/impl/impl.go b/backend/plugins/jenkins/impl/impl.go
index 01970c5dc..d0807fd64 100644
--- a/backend/plugins/jenkins/impl/impl.go
+++ b/backend/plugins/jenkins/impl/impl.go
@@ -20,7 +20,6 @@ package impl
import (
"fmt"
"strings"
- "time"
coreModels "github.com/apache/incubator-devlake/core/models"
@@ -135,13 +134,6 @@ func (p Jenkins) PrepareTaskData(taskCtx plugin.TaskContext, options map[string]
return nil, err
}
- var timeAfter time.Time
- if op.TimeAfter != "" {
- timeAfter, err = errors.Convert01(time.Parse(time.RFC3339, op.TimeAfter))
- if err != nil {
- return nil, errors.BadInput.Wrap(err, "invalid value for `timeAfter`")
- }
- }
regexEnricher := helper.NewRegexEnricher()
if err := regexEnricher.TryAdd(devops.DEPLOYMENT, op.ScopeConfig.DeploymentPattern); err != nil {
return nil, errors.BadInput.Wrap(err, "invalid value for `deploymentPattern`")
@@ -155,10 +147,7 @@ func (p Jenkins) PrepareTaskData(taskCtx plugin.TaskContext, options map[string]
Connection: connection,
RegexEnricher: regexEnricher,
}
- if !timeAfter.IsZero() {
- taskData.TimeAfter = &timeAfter
- logger.Debug("collect data created from %s", timeAfter)
- }
+
return taskData, nil
}
@@ -173,9 +162,8 @@ func (p Jenkins) MigrationScripts() []plugin.MigrationScript {
func (p Jenkins) MakeDataSourcePipelinePlanV200(
connectionId uint64,
scopes []*coreModels.BlueprintScope,
- syncPolicy *coreModels.SyncPolicy,
) (pp coreModels.PipelinePlan, sc []plugin.Scope, err errors.Error) {
- return api.MakeDataSourcePipelinePlanV200(p.SubTaskMetas(), connectionId, scopes, syncPolicy)
+ return api.MakeDataSourcePipelinePlanV200(p.SubTaskMetas(), connectionId, scopes)
}
func (p Jenkins) ApiResources() map[string]map[string]plugin.ApiResourceHandler {
diff --git a/backend/plugins/jenkins/jenkins.go b/backend/plugins/jenkins/jenkins.go
index 140e1167a..fff9039b9 100644
--- a/backend/plugins/jenkins/jenkins.go
+++ b/backend/plugins/jenkins/jenkins.go
@@ -30,15 +30,14 @@ func main() {
connectionId := jenkinsCmd.Flags().Uint64P("connection", "c", 1, "jenkins connection id")
jobFullName := jenkinsCmd.Flags().StringP("jobFullName", "j", "", "jenkins job full name")
deployTagPattern := jenkinsCmd.Flags().String("deployTagPattern", "(?i)deploy", "deploy tag name")
- timeAfter := jenkinsCmd.Flags().StringP("timeAfter", "a", "", "collect data that are created after specified time, ie 2006-05-06T07:08:09Z")
+ timeAfter := jenkinsCmd.Flags().StringP("timeAfter", "a", "", "collect data that are created after specified time, ie 2006-01-02T15:04:05Z")
jenkinsCmd.Run = func(cmd *cobra.Command, args []string) {
runner.DirectRun(cmd, args, PluginEntry, map[string]interface{}{
"connectionId": *connectionId,
"jobFullName": *jobFullName,
"deployTagPattern": *deployTagPattern,
- "timeAfter": *timeAfter,
- })
+ }, *timeAfter)
}
runner.RunCmd(jenkinsCmd)
}
diff --git a/backend/plugins/jenkins/tasks/build_collector.go b/backend/plugins/jenkins/tasks/build_collector.go
index 290d3a3e6..56924f350 100644
--- a/backend/plugins/jenkins/tasks/build_collector.go
+++ b/backend/plugins/jenkins/tasks/build_collector.go
@@ -66,7 +66,6 @@ func CollectApiBuilds(taskCtx plugin.SubTaskContext) errors.Error {
Table: RAW_BUILD_TABLE,
},
ApiClient: data.ApiClient,
- TimeAfter: data.TimeAfter,
CollectNewRecordsByList: helper.FinalizableApiCollectorListArgs{
PageSize: 100,
Concurrency: 10,
diff --git a/backend/plugins/jenkins/tasks/stage_collector.go b/backend/plugins/jenkins/tasks/stage_collector.go
index e6c4d1b30..6c8478750 100644
--- a/backend/plugins/jenkins/tasks/stage_collector.go
+++ b/backend/plugins/jenkins/tasks/stage_collector.go
@@ -54,9 +54,9 @@ func CollectApiStages(taskCtx plugin.SubTaskContext) errors.Error {
dal.Where(`tjb.connection_id = ? and tjb.job_path = ? and tjb.job_name = ? and tjb.class = ?`,
data.Options.ConnectionId, data.Options.JobPath, data.Options.JobName, "WorkflowRun"),
}
- timeAfter := data.TimeAfter
- if timeAfter != nil {
- clauses = append(clauses, dal.Where(`tjb.start_time >= ?`, timeAfter))
+ syncPolicy := taskCtx.TaskContext().SyncPolicy()
+ if syncPolicy != nil && syncPolicy.TimeAfter != nil {
+ clauses = append(clauses, dal.Where(`tjb.start_time >= ?`, syncPolicy.TimeAfter))
}
cursor, err := db.Cursor(clauses...)
diff --git a/backend/plugins/jenkins/tasks/task_data.go b/backend/plugins/jenkins/tasks/task_data.go
index 3be060df7..d34b4e6ac 100644
--- a/backend/plugins/jenkins/tasks/task_data.go
+++ b/backend/plugins/jenkins/tasks/task_data.go
@@ -19,7 +19,6 @@ package tasks
import (
"strings"
- "time"
"github.com/apache/incubator-devlake/core/errors"
"github.com/apache/incubator-devlake/helpers/pluginhelper/api"
@@ -30,11 +29,10 @@ type JenkinsApiParams models.JenkinsApiParams
type JenkinsOptions struct {
ConnectionId uint64 `json:"connectionId"`
ScopeId string
- ScopeConfigId uint64 `json:"scopeConfigId"`
- JobFullName string `json:"jobFullName"` // "path1/path2/job name"
- JobName string `json:"jobName"` // "job name"
- JobPath string `json:"jobPath"` // "job/path1/job/path2"
- TimeAfter string
+ ScopeConfigId uint64 `json:"scopeConfigId"`
+ JobFullName string `json:"jobFullName"` // "path1/path2/job name"
+ JobName string `json:"jobName"` // "job name"
+ JobPath string `json:"jobPath"` // "job/path1/job/path2"
Tasks []string `json:"tasks,omitempty"`
ScopeConfig *models.JenkinsScopeConfig `mapstructure:"scopeConfig" json:"scopeConfig"`
}
@@ -43,7 +41,6 @@ type JenkinsTaskData struct {
Options *JenkinsOptions
ApiClient *api.ApiAsyncClient
Connection *models.JenkinsConnection
- TimeAfter *time.Time
RegexEnricher *api.RegexEnricher
}
diff --git a/backend/plugins/jira/api/blueprint_v200.go b/backend/plugins/jira/api/blueprint_v200.go
index 20be17970..d67b86b98 100644
--- a/backend/plugins/jira/api/blueprint_v200.go
+++ b/backend/plugins/jira/api/blueprint_v200.go
@@ -18,8 +18,6 @@ limitations under the License.
package api
import (
- "time"
-
"github.com/apache/incubator-devlake/core/errors"
coreModels "github.com/apache/incubator-devlake/core/models"
"github.com/apache/incubator-devlake/core/models/domainlayer"
@@ -35,10 +33,9 @@ func MakeDataSourcePipelinePlanV200(
subtaskMetas []plugin.SubTaskMeta,
connectionId uint64,
bpScopes []*coreModels.BlueprintScope,
- syncPolicy *coreModels.SyncPolicy,
) (coreModels.PipelinePlan, []plugin.Scope, errors.Error) {
plan := make(coreModels.PipelinePlan, len(bpScopes))
- plan, err := makeDataSourcePipelinePlanV200(subtaskMetas, plan, bpScopes, connectionId, syncPolicy)
+ plan, err := makeDataSourcePipelinePlanV200(subtaskMetas, plan, bpScopes, connectionId)
if err != nil {
return nil, nil, err
}
@@ -55,7 +52,6 @@ func makeDataSourcePipelinePlanV200(
plan coreModels.PipelinePlan,
bpScopes []*coreModels.BlueprintScope,
connectionId uint64,
- syncPolicy *coreModels.SyncPolicy,
) (coreModels.PipelinePlan, errors.Error) {
for i, bpScope := range bpScopes {
stage := plan[i]
@@ -66,9 +62,6 @@ func makeDataSourcePipelinePlanV200(
options := make(map[string]interface{})
options["scopeId"] = bpScope.ScopeId
options["connectionId"] = connectionId
- if syncPolicy.TimeAfter != nil {
- options["timeAfter"] = syncPolicy.TimeAfter.Format(time.RFC3339)
- }
// get scope config from db
_, scopeConfig, err := scopeHelper.DbHelper().GetScopeAndConfig(connectionId, bpScope.ScopeId)
diff --git a/backend/plugins/jira/api/blueprint_v200_test.go b/backend/plugins/jira/api/blueprint_v200_test.go
index 76b9306f6..81c44186b 100644
--- a/backend/plugins/jira/api/blueprint_v200_test.go
+++ b/backend/plugins/jira/api/blueprint_v200_test.go
@@ -42,13 +42,13 @@ func TestMakeDataSourcePipelinePlanV200(t *testing.T) {
bs := &coreModels.BlueprintScope{
ScopeId: "10",
}
- syncPolicy := &coreModels.SyncPolicy{}
+
bpScopes := make([]*coreModels.BlueprintScope, 0)
bpScopes = append(bpScopes, bs)
plan := make(coreModels.PipelinePlan, len(bpScopes))
mockBasicRes(t)
- plan, err = makeDataSourcePipelinePlanV200(nil, plan, bpScopes, uint64(1), syncPolicy)
+ plan, err = makeDataSourcePipelinePlanV200(nil, plan, bpScopes, uint64(1))
assert.Nil(t, err)
scopes, err := makeScopesV200(bpScopes, uint64(1))
assert.Nil(t, err)
diff --git a/backend/plugins/jira/impl/impl.go b/backend/plugins/jira/impl/impl.go
index 432a4142a..20e6a052f 100644
--- a/backend/plugins/jira/impl/impl.go
+++ b/backend/plugins/jira/impl/impl.go
@@ -20,7 +20,6 @@ package impl
import (
"fmt"
"net/http"
- "time"
"github.com/apache/incubator-devlake/core/context"
"github.com/apache/incubator-devlake/core/dal"
@@ -250,24 +249,15 @@ func (p Jira) PrepareTaskData(taskCtx plugin.TaskContext, options map[string]int
ApiClient: jiraApiClient,
JiraServerInfo: *info,
}
- if op.TimeAfter != "" {
- var timeAfter time.Time
- timeAfter, err = errors.Convert01(time.Parse(time.RFC3339, op.TimeAfter))
- if err != nil {
- return nil, errors.BadInput.Wrap(err, "invalid value for `timeAfter`")
- }
- taskData.TimeAfter = &timeAfter
- logger.Debug("collect data created from %s", timeAfter)
- }
+
return taskData, nil
}
func (p Jira) MakeDataSourcePipelinePlanV200(
connectionId uint64,
scopes []*coreModels.BlueprintScope,
- syncPolicy *coreModels.SyncPolicy,
) (pp coreModels.PipelinePlan, sc []plugin.Scope, err errors.Error) {
- return api.MakeDataSourcePipelinePlanV200(p.SubTaskMetas(), connectionId, scopes, syncPolicy)
+ return api.MakeDataSourcePipelinePlanV200(p.SubTaskMetas(), connectionId, scopes)
}
func (p Jira) RootPkgPath() string {
diff --git a/backend/plugins/jira/jira.go b/backend/plugins/jira/jira.go
index 208b4c82e..f22cc9179 100644
--- a/backend/plugins/jira/jira.go
+++ b/backend/plugins/jira/jira.go
@@ -30,15 +30,15 @@ func main() {
cmd := &cobra.Command{Use: "jira"}
connectionId := cmd.Flags().Uint64P("connection", "c", 0, "jira connection id")
boardId := cmd.Flags().Uint64P("board", "b", 0, "jira board id")
+ timeAfter := cmd.Flags().StringP("timeAfter", "a", "", "collect data that are created after specified time, ie 2006-01-02T15:04:05Z")
_ = cmd.MarkFlagRequired("connection")
_ = cmd.MarkFlagRequired("board")
- timeAfter := cmd.Flags().StringP("timeAfter", "a", "", "collect data that are created after specified time, ie 2006-05-06T07:08:09Z")
+
cmd.Run = func(c *cobra.Command, args []string) {
runner.DirectRun(c, args, PluginEntry, map[string]interface{}{
"connectionId": *connectionId,
"boardId": *boardId,
- "timeAfter": *timeAfter,
- })
+ }, *timeAfter)
}
runner.RunCmd(cmd)
}
diff --git a/backend/plugins/jira/tasks/development_panel_collector.go b/backend/plugins/jira/tasks/development_panel_collector.go
index 44d65f627..278052cdc 100644
--- a/backend/plugins/jira/tasks/development_panel_collector.go
+++ b/backend/plugins/jira/tasks/development_panel_collector.go
@@ -61,7 +61,7 @@ func CollectDevelopmentPanel(taskCtx plugin.SubTaskContext) errors.Error {
},
Table: RAW_DEVELOPMENT_PANEL,
- }, data.TimeAfter)
+ })
if err != nil {
return err
}
diff --git a/backend/plugins/jira/tasks/issue_changelog_collector.go b/backend/plugins/jira/tasks/issue_changelog_collector.go
index 99f95ffa6..7b5fc7674 100644
--- a/backend/plugins/jira/tasks/issue_changelog_collector.go
+++ b/backend/plugins/jira/tasks/issue_changelog_collector.go
@@ -60,7 +60,7 @@ func CollectIssueChangelogs(taskCtx plugin.SubTaskContext) errors.Error {
BoardId: data.Options.BoardId,
},
Table: RAW_CHANGELOG_TABLE,
- }, data.TimeAfter)
+ })
if err != nil {
return err
}
diff --git a/backend/plugins/jira/tasks/issue_collector.go b/backend/plugins/jira/tasks/issue_collector.go
index 19f9259c9..a6cf3b8ea 100644
--- a/backend/plugins/jira/tasks/issue_collector.go
+++ b/backend/plugins/jira/tasks/issue_collector.go
@@ -20,13 +20,14 @@ package tasks
import (
"encoding/json"
"fmt"
- "github.com/apache/incubator-devlake/core/dal"
- "github.com/apache/incubator-devlake/plugins/jira/models"
"io"
"net/http"
"net/url"
"time"
+ "github.com/apache/incubator-devlake/core/dal"
+ "github.com/apache/incubator-devlake/plugins/jira/models"
+
"github.com/apache/incubator-devlake/core/errors"
"github.com/apache/incubator-devlake/core/plugin"
"github.com/apache/incubator-devlake/helpers/pluginhelper/api"
@@ -61,7 +62,7 @@ func CollectIssues(taskCtx plugin.SubTaskContext) errors.Error {
Table store raw data
*/
Table: RAW_ISSUE_TABLE,
- }, data.TimeAfter)
+ })
if err != nil {
return err
}
@@ -76,7 +77,14 @@ func CollectIssues(taskCtx plugin.SubTaskContext) errors.Error {
} else {
logger.Info("got user's timezone: %v", loc.String())
}
- jql := buildJQL(data.TimeAfter, collectorWithState.LatestState.LatestSuccessStart, incremental, loc)
+
+ jql := ""
+ syncPolicy := taskCtx.TaskContext().SyncPolicy()
+ if syncPolicy != nil && syncPolicy.TimeAfter != nil {
+ jql = buildJQL(syncPolicy.TimeAfter, collectorWithState.LatestState.LatestSuccessStart, incremental, loc)
+ } else {
+ jql = buildJQL(nil, collectorWithState.LatestState.LatestSuccessStart, incremental, loc)
+ }
err = collectorWithState.InitCollector(api.ApiCollectorArgs{
ApiClient: data.ApiClient,
diff --git a/backend/plugins/jira/tasks/issue_comment_collector.go b/backend/plugins/jira/tasks/issue_comment_collector.go
index c7879d4e2..fb0d2168c 100644
--- a/backend/plugins/jira/tasks/issue_comment_collector.go
+++ b/backend/plugins/jira/tasks/issue_comment_collector.go
@@ -60,7 +60,7 @@ func CollectIssueComments(taskCtx plugin.SubTaskContext) errors.Error {
BoardId: data.Options.BoardId,
},
Table: RAW_ISSUE_COMMENT_TABLE,
- }, data.TimeAfter)
+ })
if err != nil {
return err
}
diff --git a/backend/plugins/jira/tasks/remotelink_collector.go b/backend/plugins/jira/tasks/remotelink_collector.go
index e3b141053..158640d5c 100644
--- a/backend/plugins/jira/tasks/remotelink_collector.go
+++ b/backend/plugins/jira/tasks/remotelink_collector.go
@@ -59,7 +59,7 @@ func CollectRemotelinks(taskCtx plugin.SubTaskContext) errors.Error {
BoardId: data.Options.BoardId,
},
Table: RAW_REMOTELINK_TABLE,
- }, data.TimeAfter)
+ })
if err != nil {
return err
}
diff --git a/backend/plugins/jira/tasks/task_data.go b/backend/plugins/jira/tasks/task_data.go
index 13c8f00c2..3a9b01d26 100644
--- a/backend/plugins/jira/tasks/task_data.go
+++ b/backend/plugins/jira/tasks/task_data.go
@@ -19,7 +19,6 @@ package tasks
import (
"fmt"
- "time"
"github.com/apache/incubator-devlake/core/errors"
"github.com/apache/incubator-devlake/helpers/pluginhelper/api"
@@ -27,9 +26,8 @@ import (
)
type JiraOptions struct {
- ConnectionId uint64 `json:"connectionId"`
- BoardId uint64 `json:"boardId"`
- TimeAfter string
+ ConnectionId uint64 `json:"connectionId"`
+ BoardId uint64 `json:"boardId"`
ScopeConfig *models.JiraScopeConfig `json:"scopeConfig"`
ScopeId string
ScopeConfigId uint64
@@ -39,7 +37,6 @@ type JiraOptions struct {
type JiraTaskData struct {
Options *JiraOptions
ApiClient *api.ApiAsyncClient
- TimeAfter *time.Time
JiraServerInfo models.JiraServerInfo
}
diff --git a/backend/plugins/jira/tasks/worklog_collector.go b/backend/plugins/jira/tasks/worklog_collector.go
index bd536eb47..d0feab119 100644
--- a/backend/plugins/jira/tasks/worklog_collector.go
+++ b/backend/plugins/jira/tasks/worklog_collector.go
@@ -52,7 +52,7 @@ func CollectWorklogs(taskCtx plugin.SubTaskContext) errors.Error {
BoardId: data.Options.BoardId,
},
Table: RAW_WORKLOGS_TABLE,
- }, data.TimeAfter)
+ })
if err != nil {
return err
}
diff --git a/backend/plugins/pagerduty/api/blueprint_v200.go b/backend/plugins/pagerduty/api/blueprint_v200.go
index c213e4755..a1c8042fe 100644
--- a/backend/plugins/pagerduty/api/blueprint_v200.go
+++ b/backend/plugins/pagerduty/api/blueprint_v200.go
@@ -18,8 +18,6 @@ limitations under the License.
package api
import (
- "time"
-
"github.com/apache/incubator-devlake/core/errors"
coreModels "github.com/apache/incubator-devlake/core/models"
"github.com/apache/incubator-devlake/core/models/domainlayer"
@@ -36,7 +34,6 @@ func MakeDataSourcePipelinePlanV200(
subtaskMetas []plugin.SubTaskMeta,
connectionId uint64,
bpScopes []*coreModels.BlueprintScope,
- syncPolicy *coreModels.SyncPolicy,
) (coreModels.PipelinePlan, []plugin.Scope, errors.Error) {
// get the connection info for url
connection := &models.PagerDutyConnection{}
@@ -46,7 +43,7 @@ func MakeDataSourcePipelinePlanV200(
}
plan := make(coreModels.PipelinePlan, len(bpScopes))
- plan, err = makeDataSourcePipelinePlanV200(subtaskMetas, plan, bpScopes, connection, syncPolicy)
+ plan, err = makeDataSourcePipelinePlanV200(subtaskMetas, plan, bpScopes, connection)
if err != nil {
return nil, nil, err
}
@@ -63,7 +60,6 @@ func makeDataSourcePipelinePlanV200(
plan coreModels.PipelinePlan,
bpScopes []*coreModels.BlueprintScope,
connection *models.PagerDutyConnection,
- syncPolicy *coreModels.SyncPolicy,
) (coreModels.PipelinePlan, errors.Error) {
for i, bpScope := range bpScopes {
// get board and scope config from db
@@ -77,9 +73,7 @@ func makeDataSourcePipelinePlanV200(
ServiceId: service.Id,
ServiceName: service.Name,
}
- if syncPolicy.TimeAfter != nil {
- op.TimeAfter = syncPolicy.TimeAfter.Format(time.RFC3339)
- }
+
var options map[string]any
options, err = tasks.EncodeTaskOptions(op)
if err != nil {
diff --git a/backend/plugins/pagerduty/impl/impl.go b/backend/plugins/pagerduty/impl/impl.go
index 14737971d..273e578b3 100644
--- a/backend/plugins/pagerduty/impl/impl.go
+++ b/backend/plugins/pagerduty/impl/impl.go
@@ -19,7 +19,6 @@ package impl
import (
"fmt"
- "time"
"github.com/apache/incubator-devlake/core/context"
"github.com/apache/incubator-devlake/core/dal"
@@ -108,14 +107,7 @@ func (p PagerDuty) PrepareTaskData(taskCtx plugin.TaskContext, options map[strin
if err != nil {
return nil, errors.Default.Wrap(err, "unable to get Pagerduty connection by the given connection ID")
}
- var timeAfter *time.Time
- if op.TimeAfter != "" {
- convertedTime, err := errors.Convert01(time.Parse(time.RFC3339, op.TimeAfter))
- if err != nil {
- return nil, errors.BadInput.Wrap(err, fmt.Sprintf("invalid value for `timeAfter`: %s", timeAfter))
- }
- timeAfter = &convertedTime
- }
+
client, err := helper.NewApiClient(taskCtx.GetContext(), connection.Endpoint, map[string]string{
"Authorization": fmt.Sprintf("Token %s", connection.Token),
}, 0, connection.Proxy, taskCtx)
@@ -127,9 +119,8 @@ func (p PagerDuty) PrepareTaskData(taskCtx plugin.TaskContext, options map[strin
return nil, err
}
return &tasks.PagerDutyTaskData{
- Options: op,
- TimeAfter: timeAfter,
- Client: asyncClient,
+ Options: op,
+ Client: asyncClient,
}, nil
}
@@ -177,9 +168,8 @@ func (p PagerDuty) ApiResources() map[string]map[string]plugin.ApiResourceHandle
func (p PagerDuty) MakeDataSourcePipelinePlanV200(
connectionId uint64,
scopes []*coreModels.BlueprintScope,
- syncPolicy *coreModels.SyncPolicy,
) (coreModels.PipelinePlan, []plugin.Scope, errors.Error) {
- return api.MakeDataSourcePipelinePlanV200(p.SubTaskMetas(), connectionId, scopes, syncPolicy)
+ return api.MakeDataSourcePipelinePlanV200(p.SubTaskMetas(), connectionId, scopes)
}
func (p PagerDuty) Close(taskCtx plugin.TaskContext) errors.Error {
diff --git a/backend/plugins/pagerduty/pager_duty.go b/backend/plugins/pagerduty/pager_duty.go
index 57c0f9dd8..34145fbd2 100644
--- a/backend/plugins/pagerduty/pager_duty.go
+++ b/backend/plugins/pagerduty/pager_duty.go
@@ -29,6 +29,7 @@ var PluginEntry impl.PagerDuty //nolint
// standalone mode for debugging
func main() {
cmd := &cobra.Command{Use: "pagerduty"}
+ timeAfter := cmd.Flags().StringP("timeAfter", "a", "", "collect data that are created after specified time, ie 2006-01-02T15:04:05Z")
// TODO add your cmd flag if necessary
// yourFlag := cmd.Flags().IntP("yourFlag", "y", 8, "TODO add description here")
@@ -37,7 +38,7 @@ func main() {
cmd.Run = func(cmd *cobra.Command, args []string) {
runner.DirectRun(cmd, args, PluginEntry, map[string]interface{}{
// TODO add more custom params here
- })
+ }, *timeAfter)
}
runner.RunCmd(cmd)
}
diff --git a/backend/plugins/pagerduty/tasks/incidents_collector.go b/backend/plugins/pagerduty/tasks/incidents_collector.go
index d8cae2f1f..91068bf5b 100644
--- a/backend/plugins/pagerduty/tasks/incidents_collector.go
+++ b/backend/plugins/pagerduty/tasks/incidents_collector.go
@@ -69,7 +69,6 @@ func CollectIncidents(taskCtx plugin.SubTaskContext) errors.Error {
collector, err := api.NewStatefulApiCollectorForFinalizableEntity(api.FinalizableApiCollectorArgs{
RawDataSubTaskArgs: args,
ApiClient: data.Client,
- TimeAfter: data.TimeAfter,
CollectNewRecordsByList: api.FinalizableApiCollectorListArgs{
PageSize: 100,
GetNextPageCustomData: func(prevReqData *api.RequestData, prevPageResponse *http.Response) (interface{}, errors.Error) {
diff --git a/backend/plugins/pagerduty/tasks/task_data.go b/backend/plugins/pagerduty/tasks/task_data.go
index 894628073..5f456f725 100644
--- a/backend/plugins/pagerduty/tasks/task_data.go
+++ b/backend/plugins/pagerduty/tasks/task_data.go
@@ -18,8 +18,6 @@ limitations under the License.
package tasks
import (
- "time"
-
"github.com/apache/incubator-devlake/core/errors"
"github.com/apache/incubator-devlake/helpers/pluginhelper/api"
"github.com/apache/incubator-devlake/plugins/pagerduty/models"
@@ -27,7 +25,6 @@ import (
type PagerDutyOptions struct {
ConnectionId uint64 `json:"connectionId"`
- TimeAfter string `json:"time_after,omitempty"`
ServiceId string `json:"service_id,omitempty"`
ServiceName string `json:"service_name,omitempty"`
Tasks []string `json:"tasks,omitempty"`
@@ -35,9 +32,8 @@ type PagerDutyOptions struct {
}
type PagerDutyTaskData struct {
- Options *PagerDutyOptions
- TimeAfter *time.Time
- Client api.RateLimitedApiClient
+ Options *PagerDutyOptions
+ Client api.RateLimitedApiClient
}
func (p *PagerDutyOptions) GetParams() any {
diff --git a/backend/plugins/refdiff/refdiff.go b/backend/plugins/refdiff/refdiff.go
index 2c246f19b..0f2f50f70 100644
--- a/backend/plugins/refdiff/refdiff.go
+++ b/backend/plugins/refdiff/refdiff.go
@@ -38,6 +38,7 @@ func main() {
tagsOrder := refdiffCmd.Flags().StringP("tags-order", "d", "", "tags order")
projectName := refdiffCmd.Flags().StringP("project-name", "P", "", "project name")
+ timeAfter := refdiffCmd.Flags().StringP("time-after", "a", "", "collect data that are created after specified time, ie 2006-01-02T15:04:05Z")
// _ = refdiffCmd.MarkFlagRequired("repo-id")
//_ = refdiffCmd.MarkFlagRequired("new-ref")
@@ -59,7 +60,7 @@ func main() {
"tagsLimit": *tagsLimit,
"tagsOrder": *tagsOrder,
"projectName": *projectName,
- })
+ }, *timeAfter)
}
runner.RunCmd(refdiffCmd)
}
diff --git a/backend/plugins/slack/slack.go b/backend/plugins/slack/slack.go
index 01355bb29..19734bf96 100644
--- a/backend/plugins/slack/slack.go
+++ b/backend/plugins/slack/slack.go
@@ -29,11 +29,12 @@ var PluginEntry impl.Slack
func main() {
cmd := &cobra.Command{Use: "slack"}
connectionId := cmd.Flags().Uint64P("connectionId", "c", 0, "slack connection id")
+ timeAfter := cmd.Flags().StringP("timeAfter", "a", "", "collect data that are created after specified time, ie 2006-01-02T15:04:05Z")
_ = cmd.MarkFlagRequired("connectionId")
cmd.Run = func(cmd *cobra.Command, args []string) {
runner.DirectRun(cmd, args, PluginEntry, map[string]interface{}{
"connectionId": *connectionId,
- })
+ }, *timeAfter)
}
runner.RunCmd(cmd)
}
diff --git a/backend/plugins/sonarqube/api/blueprint_v200.go b/backend/plugins/sonarqube/api/blueprint_v200.go
index 95f886c20..b1148e530 100644
--- a/backend/plugins/sonarqube/api/blueprint_v200.go
+++ b/backend/plugins/sonarqube/api/blueprint_v200.go
@@ -38,10 +38,9 @@ func MakeDataSourcePipelinePlanV200(
subtaskMetas []plugin.SubTaskMeta,
connectionId uint64,
bpScopes []*coreModels.BlueprintScope,
- syncPolicy *coreModels.SyncPolicy,
) (coreModels.PipelinePlan, []plugin.Scope, errors.Error) {
plan := make(coreModels.PipelinePlan, len(bpScopes))
- plan, err := makeDataSourcePipelinePlanV200(subtaskMetas, plan, bpScopes, connectionId, syncPolicy)
+ plan, err := makeDataSourcePipelinePlanV200(subtaskMetas, plan, bpScopes, connectionId)
if err != nil {
return nil, nil, err
}
@@ -58,7 +57,6 @@ func makeDataSourcePipelinePlanV200(
plan coreModels.PipelinePlan,
bpScopes []*coreModels.BlueprintScope,
connectionId uint64,
- syncPolicy *coreModels.SyncPolicy,
) (coreModels.PipelinePlan, errors.Error) {
for i, bpScope := range bpScopes {
stage := plan[i]
diff --git a/backend/plugins/sonarqube/api/blueprint_v200_test.go b/backend/plugins/sonarqube/api/blueprint_v200_test.go
index fd1937119..093355e8a 100644
--- a/backend/plugins/sonarqube/api/blueprint_v200_test.go
+++ b/backend/plugins/sonarqube/api/blueprint_v200_test.go
@@ -41,11 +41,11 @@ func TestMakeDataSourcePipelinePlanV200(t *testing.T) {
bs := &coreModels.BlueprintScope{
ScopeId: "f5a50c63-2e8f-4107-9014-853f6f467757",
}
- syncPolicy := &coreModels.SyncPolicy{}
+
bpScopes := make([]*coreModels.BlueprintScope, 0)
bpScopes = append(bpScopes, bs)
plan := make(coreModels.PipelinePlan, len(bpScopes))
- plan, err = makeDataSourcePipelinePlanV200(nil, plan, bpScopes, uint64(1), syncPolicy)
+ plan, err = makeDataSourcePipelinePlanV200(nil, plan, bpScopes, uint64(1))
assert.Nil(t, err)
basicRes = NewMockBasicRes()
diff --git a/backend/plugins/sonarqube/impl/impl.go b/backend/plugins/sonarqube/impl/impl.go
index ac2743104..28496b180 100644
--- a/backend/plugins/sonarqube/impl/impl.go
+++ b/backend/plugins/sonarqube/impl/impl.go
@@ -200,9 +200,8 @@ func (p Sonarqube) ApiResources() map[string]map[string]plugin.ApiResourceHandle
func (p Sonarqube) MakeDataSourcePipelinePlanV200(
connectionId uint64,
scopes []*coreModels.BlueprintScope,
- syncPolicy *coreModels.SyncPolicy,
) (pp coreModels.PipelinePlan, sc []plugin.Scope, err errors.Error) {
- return api.MakeDataSourcePipelinePlanV200(p.SubTaskMetas(), connectionId, scopes, syncPolicy)
+ return api.MakeDataSourcePipelinePlanV200(p.SubTaskMetas(), connectionId, scopes)
}
func (p Sonarqube) Close(taskCtx plugin.TaskContext) errors.Error {
diff --git a/backend/plugins/sonarqube/sonarqube.go b/backend/plugins/sonarqube/sonarqube.go
index 64b9363f3..739ca2a76 100644
--- a/backend/plugins/sonarqube/sonarqube.go
+++ b/backend/plugins/sonarqube/sonarqube.go
@@ -30,6 +30,7 @@ func main() {
cmd := &cobra.Command{Use: "sonarqube"}
connectionId := cmd.Flags().Uint64P("connectionId", "c", 0, "sonarqube connection id")
projectKey := cmd.Flags().StringP("projectKey", "p", "", "sonarqube projectKey")
+ timeAfter := cmd.Flags().StringP("timeAfter", "a", "", "collect data that are created after specified time, ie 2006-01-02T15:04:05Z")
_ = cmd.MarkFlagRequired("connectionId")
//_ = cmd.MarkFlagRequired("projectKey")
@@ -37,7 +38,7 @@ func main() {
runner.DirectRun(cmd, args, PluginEntry, map[string]interface{}{
"connectionId": *connectionId,
"projectKey": *projectKey,
- })
+ }, *timeAfter)
}
runner.RunCmd(cmd)
}
diff --git a/backend/plugins/starrocks/starrocks.go b/backend/plugins/starrocks/starrocks.go
index a6c10dd55..c942e8e92 100644
--- a/backend/plugins/starrocks/starrocks.go
+++ b/backend/plugins/starrocks/starrocks.go
@@ -50,6 +50,7 @@ func main() {
_ = cmd.MarkFlagRequired("batch_size")
extra := cmd.Flags().StringP("extra", "e", "", "StarRocks create table sql extra")
orderBy := cmd.Flags().StringP("order_by", "o", "", "Source tables order by, default is primary key")
+ timeAfter := cmd.Flags().StringP("time_after", "a", "", "collect data that are created after specified time, ie 2006-01-02T15:04:05Z")
cmd.Run = func(cmd *cobra.Command, args []string) {
runner.DirectRun(cmd, args, PluginEntry, map[string]interface{}{
"source_type": sourceType,
@@ -66,7 +67,7 @@ func main() {
"batch_size": batchSize,
"extra": extra,
"order_by": orderBy,
- })
+ }, *timeAfter)
}
runner.RunCmd(cmd)
}
diff --git a/backend/plugins/tapd/api/blueprint_v200.go b/backend/plugins/tapd/api/blueprint_v200.go
index 41aa2cfa4..8ab9cb0cb 100644
--- a/backend/plugins/tapd/api/blueprint_v200.go
+++ b/backend/plugins/tapd/api/blueprint_v200.go
@@ -19,7 +19,6 @@ package api
import (
"strconv"
- "time"
"github.com/apache/incubator-devlake/core/errors"
coreModels "github.com/apache/incubator-devlake/core/models"
@@ -36,10 +35,9 @@ func MakeDataSourcePipelinePlanV200(
subtaskMetas []plugin.SubTaskMeta,
connectionId uint64,
bpScopes []*coreModels.BlueprintScope,
- syncPolicy *coreModels.SyncPolicy,
) (coreModels.PipelinePlan, []plugin.Scope, errors.Error) {
plan := make(coreModels.PipelinePlan, len(bpScopes))
- plan, err := makeDataSourcePipelinePlanV200(subtaskMetas, plan, bpScopes, connectionId, syncPolicy)
+ plan, err := makeDataSourcePipelinePlanV200(subtaskMetas, plan, bpScopes, connectionId)
if err != nil {
return nil, nil, err
}
@@ -56,7 +54,6 @@ func makeDataSourcePipelinePlanV200(
plan coreModels.PipelinePlan,
bpScopes []*coreModels.BlueprintScope,
connectionId uint64,
- syncPolicy *coreModels.SyncPolicy,
) (coreModels.PipelinePlan, errors.Error) {
for i, bpScope := range bpScopes {
stage := plan[i]
@@ -71,9 +68,6 @@ func makeDataSourcePipelinePlanV200(
}
options["workspaceId"] = intNum
options["connectionId"] = connectionId
- if syncPolicy.TimeAfter != nil {
- options["timeAfter"] = syncPolicy.TimeAfter.Format(time.RFC3339)
- }
_, scopeConfig, err := scopeHelper.DbHelper().GetScopeAndConfig(connectionId, bpScope.ScopeId)
if err != nil {
diff --git a/backend/plugins/tapd/api/blueprint_v200_test.go b/backend/plugins/tapd/api/blueprint_v200_test.go
index dac1d9a43..bb3a17eec 100644
--- a/backend/plugins/tapd/api/blueprint_v200_test.go
+++ b/backend/plugins/tapd/api/blueprint_v200_test.go
@@ -42,13 +42,12 @@ func TestMakeDataSourcePipelinePlanV200(t *testing.T) {
bs := &coreModels.BlueprintScope{
ScopeId: "10",
}
- syncPolicy := &coreModels.SyncPolicy{}
bpScopes := make([]*coreModels.BlueprintScope, 0)
bpScopes = append(bpScopes, bs)
plan := make(coreModels.PipelinePlan, len(bpScopes))
mockBasicRes(t)
- plan, err = makeDataSourcePipelinePlanV200(nil, plan, bpScopes, uint64(1), syncPolicy)
+ plan, err = makeDataSourcePipelinePlanV200(nil, plan, bpScopes, uint64(1))
assert.Nil(t, err)
scopes, err := makeScopesV200(bpScopes, uint64(1))
assert.Nil(t, err)
diff --git a/backend/plugins/tapd/impl/impl.go b/backend/plugins/tapd/impl/impl.go
index f72a81a4c..04ec1d62d 100644
--- a/backend/plugins/tapd/impl/impl.go
+++ b/backend/plugins/tapd/impl/impl.go
@@ -255,24 +255,14 @@ func (p Tapd) PrepareTaskData(taskCtx plugin.TaskContext, options map[string]int
ApiClient: tapdApiClient,
Connection: connection,
}
- if op.TimeAfter != "" {
- var timeAfter time.Time
- timeAfter, err = errors.Convert01(time.Parse(time.RFC3339, op.TimeAfter))
- if err != nil {
- return nil, errors.BadInput.Wrap(err, "invalid value for `timeAfter`")
- }
- taskData.TimeAfter = &timeAfter
- logger.Debug("collect data updated timeAfter %s", timeAfter)
- }
return taskData, nil
}
func (p Tapd) MakeDataSourcePipelinePlanV200(
connectionId uint64,
scopes []*coreModels.BlueprintScope,
- syncPolicy *coreModels.SyncPolicy,
) (pp coreModels.PipelinePlan, sc []plugin.Scope, err errors.Error) {
- return api.MakeDataSourcePipelinePlanV200(p.SubTaskMetas(), connectionId, scopes, syncPolicy)
+ return api.MakeDataSourcePipelinePlanV200(p.SubTaskMetas(), connectionId, scopes)
}
func (p Tapd) RootPkgPath() string {
diff --git a/backend/plugins/tapd/tapd.go b/backend/plugins/tapd/tapd.go
index 4d61d67ef..c7d88316c 100644
--- a/backend/plugins/tapd/tapd.go
+++ b/backend/plugins/tapd/tapd.go
@@ -32,7 +32,7 @@ func main() {
workspaceId := cmd.Flags().Uint64P("workspace", "w", 0, "tapd workspace id")
companyId := cmd.Flags().Uint64P("company", "o", 0, "tapd company id")
pageSize := cmd.Flags().Uint64P("pageSize", "p", 0, "tapd page size")
- timeAfter := cmd.Flags().StringP("timeAfter", "a", "", "collect data that are created after specified time, ie 2006-05-06T07:08:09Z")
+ timeAfter := cmd.Flags().StringP("timeAfter", "a", "", "collect data that are created after specified time, ie 2006-01-02T15:04:05Z")
err := cmd.MarkFlagRequired("connection")
if err != nil {
panic(err)
@@ -47,9 +47,8 @@ func main() {
"connectionId": *connectionId,
"workspaceId": *workspaceId,
"companyId": *companyId,
- "timeAfter": *timeAfter,
"pageSize": *pageSize,
- })
+ }, *timeAfter)
//
// cfg := config.GetConfig()
// log := logger.Global.Nested(cmd.Use)
diff --git a/backend/plugins/tapd/tasks/bug_changelog_collector.go b/backend/plugins/tapd/tasks/bug_changelog_collector.go
index 32992091c..b967cfa78 100644
--- a/backend/plugins/tapd/tasks/bug_changelog_collector.go
+++ b/backend/plugins/tapd/tasks/bug_changelog_collector.go
@@ -34,11 +34,12 @@ func CollectBugChangelogs(taskCtx plugin.SubTaskContext) errors.Error {
rawDataSubTaskArgs, data := CreateRawDataSubTaskArgs(taskCtx, RAW_BUG_CHANGELOG_TABLE)
logger := taskCtx.GetLogger()
logger.Info("collect storyChangelogs")
- collectorWithState, err := helper.NewStatefulApiCollector(*rawDataSubTaskArgs, data.TimeAfter)
+ collectorWithState, err := helper.NewStatefulApiCollector(*rawDataSubTaskArgs)
if err != nil {
return err
}
incremental := collectorWithState.IsIncremental()
+ syncPolicy := taskCtx.TaskContext().SyncPolicy()
err = collectorWithState.InitCollector(helper.ApiCollectorArgs{
Incremental: incremental,
@@ -51,10 +52,10 @@ func CollectBugChangelogs(taskCtx plugin.SubTaskContext) errors.Error {
query.Set("page", fmt.Sprintf("%v", reqData.Pager.Page))
query.Set("limit", fmt.Sprintf("%v", reqData.Pager.Size))
query.Set("order", "created desc")
- if data.TimeAfter != nil {
+ if syncPolicy != nil && syncPolicy.TimeAfter != nil {
query.Set("created",
fmt.Sprintf(">%s",
- data.TimeAfter.In(data.Options.CstZone).Format("2006-01-02")))
+ syncPolicy.TimeAfter.In(data.Options.CstZone).Format("2006-01-02")))
}
if incremental {
query.Set("created",
diff --git a/backend/plugins/tapd/tasks/bug_collector.go b/backend/plugins/tapd/tasks/bug_collector.go
index 530c5ff04..3265d320f 100644
--- a/backend/plugins/tapd/tasks/bug_collector.go
+++ b/backend/plugins/tapd/tasks/bug_collector.go
@@ -19,10 +19,11 @@ package tasks
import (
"fmt"
+ "net/url"
+
"github.com/apache/incubator-devlake/core/errors"
"github.com/apache/incubator-devlake/core/plugin"
helper "github.com/apache/incubator-devlake/helpers/pluginhelper/api"
- "net/url"
)
const RAW_BUG_TABLE = "tapd_api_bugs"
@@ -33,12 +34,12 @@ func CollectBugs(taskCtx plugin.SubTaskContext) errors.Error {
rawDataSubTaskArgs, data := CreateRawDataSubTaskArgs(taskCtx, RAW_BUG_TABLE)
logger := taskCtx.GetLogger()
logger.Info("collect bugs")
- collectorWithState, err := helper.NewStatefulApiCollector(*rawDataSubTaskArgs, data.TimeAfter)
+ collectorWithState, err := helper.NewStatefulApiCollector(*rawDataSubTaskArgs)
if err != nil {
return err
}
incremental := collectorWithState.IsIncremental()
-
+ syncPolicy := taskCtx.TaskContext().SyncPolicy()
err = collectorWithState.InitCollector(helper.ApiCollectorArgs{
Incremental: incremental,
ApiClient: data.ApiClient,
@@ -51,10 +52,10 @@ func CollectBugs(taskCtx plugin.SubTaskContext) errors.Error {
query.Set("limit", fmt.Sprintf("%v", reqData.Pager.Size))
query.Set("fields", "labels")
query.Set("order", "created asc")
- if data.TimeAfter != nil {
+ if syncPolicy != nil && syncPolicy.TimeAfter != nil {
query.Set("modified",
fmt.Sprintf(">%s",
- data.TimeAfter.In(data.Options.CstZone).Format("2006-01-02")))
+ syncPolicy.TimeAfter.In(data.Options.CstZone).Format("2006-01-02")))
}
if incremental {
query.Set("modified",
diff --git a/backend/plugins/tapd/tasks/bug_commit_collector.go b/backend/plugins/tapd/tasks/bug_commit_collector.go
index 26a8e1e11..97a1c2937 100644
--- a/backend/plugins/tapd/tasks/bug_commit_collector.go
+++ b/backend/plugins/tapd/tasks/bug_commit_collector.go
@@ -20,14 +20,15 @@ package tasks
import (
"encoding/json"
"fmt"
+ "net/http"
+ "net/url"
+ "reflect"
+
"github.com/apache/incubator-devlake/core/dal"
"github.com/apache/incubator-devlake/core/errors"
"github.com/apache/incubator-devlake/core/plugin"
"github.com/apache/incubator-devlake/helpers/pluginhelper/api"
"github.com/apache/incubator-devlake/plugins/tapd/models"
- "net/http"
- "net/url"
- "reflect"
)
const RAW_BUG_COMMIT_TABLE = "tapd_api_bug_commits"
@@ -37,7 +38,7 @@ var _ plugin.SubTaskEntryPoint = CollectBugCommits
func CollectBugCommits(taskCtx plugin.SubTaskContext) errors.Error {
rawDataSubTaskArgs, data := CreateRawDataSubTaskArgs(taskCtx, RAW_BUG_COMMIT_TABLE)
db := taskCtx.GetDal()
- collectorWithState, err := api.NewStatefulApiCollector(*rawDataSubTaskArgs, data.TimeAfter)
+ collectorWithState, err := api.NewStatefulApiCollector(*rawDataSubTaskArgs)
if err != nil {
return err
}
diff --git a/backend/plugins/tapd/tasks/iteration_collector.go b/backend/plugins/tapd/tasks/iteration_collector.go
index 7e18030f9..fe21fbc4f 100644
--- a/backend/plugins/tapd/tasks/iteration_collector.go
+++ b/backend/plugins/tapd/tasks/iteration_collector.go
@@ -36,11 +36,12 @@ func CollectIterations(taskCtx plugin.SubTaskContext) errors.Error {
rawDataSubTaskArgs, data := CreateRawDataSubTaskArgs(taskCtx, RAW_ITERATION_TABLE)
logger := taskCtx.GetLogger()
logger.Info("collect iterations")
- collectorWithState, err := api.NewStatefulApiCollector(*rawDataSubTaskArgs, data.TimeAfter)
+ collectorWithState, err := api.NewStatefulApiCollector(*rawDataSubTaskArgs)
if err != nil {
return err
}
incremental := collectorWithState.IsIncremental()
+ syncPolicy := taskCtx.TaskContext().SyncPolicy()
err = collectorWithState.InitCollector(api.ApiCollectorArgs{
Incremental: incremental,
ApiClient: data.ApiClient,
@@ -53,10 +54,10 @@ func CollectIterations(taskCtx plugin.SubTaskContext) errors.Error {
query.Set("page", fmt.Sprintf("%v", reqData.Pager.Page))
query.Set("limit", fmt.Sprintf("%v", reqData.Pager.Size))
query.Set("order", "created asc")
- if data.TimeAfter != nil {
+ if syncPolicy != nil && syncPolicy.TimeAfter != nil {
query.Set("modified",
fmt.Sprintf(">%s",
- data.TimeAfter.In(data.Options.CstZone).Format("2006-01-02")))
+ syncPolicy.TimeAfter.In(data.Options.CstZone).Format("2006-01-02")))
}
if incremental {
query.Set("modified",
diff --git a/backend/plugins/tapd/tasks/story_bug_collector.go b/backend/plugins/tapd/tasks/story_bug_collector.go
index 0115399f5..b9ac18dfa 100644
--- a/backend/plugins/tapd/tasks/story_bug_collector.go
+++ b/backend/plugins/tapd/tasks/story_bug_collector.go
@@ -19,13 +19,14 @@ package tasks
import (
"fmt"
+ "net/url"
+ "reflect"
+
"github.com/apache/incubator-devlake/core/dal"
"github.com/apache/incubator-devlake/core/errors"
"github.com/apache/incubator-devlake/core/plugin"
"github.com/apache/incubator-devlake/helpers/pluginhelper/api"
"github.com/apache/incubator-devlake/plugins/tapd/models"
- "net/url"
- "reflect"
)
const RAW_STORY_BUG_TABLE = "tapd_api_story_bugs"
@@ -35,7 +36,7 @@ var _ plugin.SubTaskEntryPoint = CollectStoryBugs
func CollectStoryBugs(taskCtx plugin.SubTaskContext) errors.Error {
rawDataSubTaskArgs, data := CreateRawDataSubTaskArgs(taskCtx, RAW_STORY_BUG_TABLE)
db := taskCtx.GetDal()
- collectorWithState, err := api.NewStatefulApiCollector(*rawDataSubTaskArgs, data.TimeAfter)
+ collectorWithState, err := api.NewStatefulApiCollector(*rawDataSubTaskArgs)
if err != nil {
return err
}
diff --git a/backend/plugins/tapd/tasks/story_changelog_collector.go b/backend/plugins/tapd/tasks/story_changelog_collector.go
index c7ef61407..a0bad43bd 100644
--- a/backend/plugins/tapd/tasks/story_changelog_collector.go
+++ b/backend/plugins/tapd/tasks/story_changelog_collector.go
@@ -32,13 +32,14 @@ var _ plugin.SubTaskEntryPoint = CollectStoryChangelogs
func CollectStoryChangelogs(taskCtx plugin.SubTaskContext) errors.Error {
rawDataSubTaskArgs, data := CreateRawDataSubTaskArgs(taskCtx, RAW_STORY_CHANGELOG_TABLE)
- collectorWithState, err := helper.NewStatefulApiCollector(*rawDataSubTaskArgs, data.TimeAfter)
+ collectorWithState, err := helper.NewStatefulApiCollector(*rawDataSubTaskArgs)
if err != nil {
return err
}
logger := taskCtx.GetLogger()
logger.Info("collect storyChangelogs")
incremental := collectorWithState.IsIncremental()
+ syncPolicy := taskCtx.TaskContext().SyncPolicy()
err = collectorWithState.InitCollector(helper.ApiCollectorArgs{
Incremental: incremental,
@@ -51,10 +52,10 @@ func CollectStoryChangelogs(taskCtx plugin.SubTaskContext) errors.Error {
query.Set("page", fmt.Sprintf("%v", reqData.Pager.Page))
query.Set("limit", fmt.Sprintf("%v", reqData.Pager.Size))
query.Set("order", "created asc")
- if data.TimeAfter != nil {
+ if syncPolicy != nil && syncPolicy.TimeAfter != nil {
query.Set("created",
fmt.Sprintf(">%s",
- data.TimeAfter.In(data.Options.CstZone).Format("2006-01-02")))
+ syncPolicy.TimeAfter.In(data.Options.CstZone).Format("2006-01-02")))
}
if incremental {
query.Set("created",
diff --git a/backend/plugins/tapd/tasks/story_collector.go b/backend/plugins/tapd/tasks/story_collector.go
index e6df82388..e88dd87ac 100644
--- a/backend/plugins/tapd/tasks/story_collector.go
+++ b/backend/plugins/tapd/tasks/story_collector.go
@@ -34,11 +34,12 @@ func CollectStorys(taskCtx plugin.SubTaskContext) errors.Error {
rawDataSubTaskArgs, data := CreateRawDataSubTaskArgs(taskCtx, RAW_STORY_TABLE)
logger := taskCtx.GetLogger()
logger.Info("collect stories")
- collectorWithState, err := helper.NewStatefulApiCollector(*rawDataSubTaskArgs, data.TimeAfter)
+ collectorWithState, err := helper.NewStatefulApiCollector(*rawDataSubTaskArgs)
if err != nil {
return err
}
incremental := collectorWithState.IsIncremental()
+ syncPolicy := taskCtx.TaskContext().SyncPolicy()
err = collectorWithState.InitCollector(helper.ApiCollectorArgs{
Incremental: incremental,
ApiClient: data.ApiClient,
@@ -51,10 +52,10 @@ func CollectStorys(taskCtx plugin.SubTaskContext) errors.Error {
query.Set("limit", fmt.Sprintf("%v", reqData.Pager.Size))
query.Set("fields", "labels")
query.Set("order", "created asc")
- if data.TimeAfter != nil {
+ if syncPolicy != nil && syncPolicy.TimeAfter != nil {
query.Set("modified",
fmt.Sprintf(">%s",
- data.TimeAfter.In(data.Options.CstZone).Format("2006-01-02")))
+ syncPolicy.TimeAfter.In(data.Options.CstZone).Format("2006-01-02")))
}
if incremental {
query.Set("modified",
diff --git a/backend/plugins/tapd/tasks/story_commit_collector.go b/backend/plugins/tapd/tasks/story_commit_collector.go
index 1b0f63505..b69c6fa96 100644
--- a/backend/plugins/tapd/tasks/story_commit_collector.go
+++ b/backend/plugins/tapd/tasks/story_commit_collector.go
@@ -20,14 +20,15 @@ package tasks
import (
"encoding/json"
"fmt"
+ "net/http"
+ "net/url"
+ "reflect"
+
"github.com/apache/incubator-devlake/core/dal"
"github.com/apache/incubator-devlake/core/errors"
"github.com/apache/incubator-devlake/core/plugin"
"github.com/apache/incubator-devlake/helpers/pluginhelper/api"
"github.com/apache/incubator-devlake/plugins/tapd/models"
- "net/http"
- "net/url"
- "reflect"
)
const RAW_STORY_COMMIT_TABLE = "tapd_api_story_commits"
@@ -37,7 +38,7 @@ var _ plugin.SubTaskEntryPoint = CollectStoryCommits
func CollectStoryCommits(taskCtx plugin.SubTaskContext) errors.Error {
rawDataSubTaskArgs, data := CreateRawDataSubTaskArgs(taskCtx, RAW_STORY_COMMIT_TABLE)
db := taskCtx.GetDal()
- collectorWithState, err := api.NewStatefulApiCollector(*rawDataSubTaskArgs, data.TimeAfter)
+ collectorWithState, err := api.NewStatefulApiCollector(*rawDataSubTaskArgs)
if err != nil {
return err
}
diff --git a/backend/plugins/tapd/tasks/task_changelog_collector.go b/backend/plugins/tapd/tasks/task_changelog_collector.go
index 7f52ea28c..9a457c7bd 100644
--- a/backend/plugins/tapd/tasks/task_changelog_collector.go
+++ b/backend/plugins/tapd/tasks/task_changelog_collector.go
@@ -32,14 +32,14 @@ var _ plugin.SubTaskEntryPoint = CollectTaskChangelogs
func CollectTaskChangelogs(taskCtx plugin.SubTaskContext) errors.Error {
rawDataSubTaskArgs, data := CreateRawDataSubTaskArgs(taskCtx, RAW_TASK_CHANGELOG_TABLE)
- collectorWithState, err := helper.NewStatefulApiCollector(*rawDataSubTaskArgs, data.TimeAfter)
+ collectorWithState, err := helper.NewStatefulApiCollector(*rawDataSubTaskArgs)
if err != nil {
return err
}
logger := taskCtx.GetLogger()
logger.Info("collect taskChangelogs")
incremental := collectorWithState.IsIncremental()
-
+ syncPolicy := taskCtx.TaskContext().SyncPolicy()
err = collectorWithState.InitCollector(helper.ApiCollectorArgs{
Incremental: incremental,
ApiClient: data.ApiClient,
@@ -51,10 +51,10 @@ func CollectTaskChangelogs(taskCtx plugin.SubTaskContext) errors.Error {
query.Set("page", fmt.Sprintf("%v", reqData.Pager.Page))
query.Set("limit", fmt.Sprintf("%v", reqData.Pager.Size))
query.Set("order", "created asc")
- if data.TimeAfter != nil {
+ if syncPolicy != nil && syncPolicy.TimeAfter != nil {
query.Set("created",
fmt.Sprintf(">%s",
- data.TimeAfter.In(data.Options.CstZone).Format("2006-01-02")))
+ syncPolicy.TimeAfter.In(data.Options.CstZone).Format("2006-01-02")))
}
if incremental {
query.Set("created",
diff --git a/backend/plugins/tapd/tasks/task_collector.go b/backend/plugins/tapd/tasks/task_collector.go
index f0da6d10b..36d505e5c 100644
--- a/backend/plugins/tapd/tasks/task_collector.go
+++ b/backend/plugins/tapd/tasks/task_collector.go
@@ -34,12 +34,12 @@ func CollectTasks(taskCtx plugin.SubTaskContext) errors.Error {
rawDataSubTaskArgs, data := CreateRawDataSubTaskArgs(taskCtx, RAW_TASK_TABLE)
logger := taskCtx.GetLogger()
logger.Info("collect tasks")
- collectorWithState, err := helper.NewStatefulApiCollector(*rawDataSubTaskArgs, data.TimeAfter)
+ collectorWithState, err := helper.NewStatefulApiCollector(*rawDataSubTaskArgs)
if err != nil {
return err
}
incremental := collectorWithState.IsIncremental()
-
+ syncPolicy := taskCtx.TaskContext().SyncPolicy()
collector, err := helper.NewApiCollector(helper.ApiCollectorArgs{
RawDataSubTaskArgs: *rawDataSubTaskArgs,
Incremental: incremental,
@@ -53,10 +53,10 @@ func CollectTasks(taskCtx plugin.SubTaskContext) errors.Error {
query.Set("limit", fmt.Sprintf("%v", reqData.Pager.Size))
query.Set("fields", "labels")
query.Set("order", "created asc")
- if data.TimeAfter != nil {
+ if syncPolicy != nil && syncPolicy.TimeAfter != nil {
query.Set("modified",
fmt.Sprintf(">%s",
- data.TimeAfter.In(data.Options.CstZone).Format("2006-01-02")))
+ syncPolicy.TimeAfter.In(data.Options.CstZone).Format("2006-01-02")))
}
if incremental {
query.Set("modified",
diff --git a/backend/plugins/tapd/tasks/task_commit_collector.go b/backend/plugins/tapd/tasks/task_commit_collector.go
index 1a73a870c..88da97d6e 100644
--- a/backend/plugins/tapd/tasks/task_commit_collector.go
+++ b/backend/plugins/tapd/tasks/task_commit_collector.go
@@ -20,14 +20,15 @@ package tasks
import (
"encoding/json"
"fmt"
+ "net/http"
+ "net/url"
+ "reflect"
+
"github.com/apache/incubator-devlake/core/dal"
"github.com/apache/incubator-devlake/core/errors"
"github.com/apache/incubator-devlake/core/plugin"
"github.com/apache/incubator-devlake/helpers/pluginhelper/api"
"github.com/apache/incubator-devlake/plugins/tapd/models"
- "net/http"
- "net/url"
- "reflect"
)
const RAW_TASK_COMMIT_TABLE = "tapd_api_task_commits"
@@ -37,7 +38,7 @@ var _ plugin.SubTaskEntryPoint = CollectTaskCommits
func CollectTaskCommits(taskCtx plugin.SubTaskContext) errors.Error {
rawDataSubTaskArgs, data := CreateRawDataSubTaskArgs(taskCtx, RAW_TASK_COMMIT_TABLE)
db := taskCtx.GetDal()
- collectorWithState, err := api.NewStatefulApiCollector(*rawDataSubTaskArgs, data.TimeAfter)
+ collectorWithState, err := api.NewStatefulApiCollector(*rawDataSubTaskArgs)
if err != nil {
return err
}
diff --git a/backend/plugins/tapd/tasks/task_data.go b/backend/plugins/tapd/tasks/task_data.go
index fc892f0fe..16b077bbb 100644
--- a/backend/plugins/tapd/tasks/task_data.go
+++ b/backend/plugins/tapd/tasks/task_data.go
@@ -29,7 +29,6 @@ type TapdOptions struct {
ConnectionId uint64 `mapstruct:"connectionId"`
WorkspaceId uint64 `mapstruct:"workspaceId"`
PageSize uint64 `mapstruct:"pageSize"`
- TimeAfter string `json:"timeAfter" mapstructure:"timeAfter,omitempty"`
CstZone *time.Location
ScopeConfigId uint64
ScopeConfig *models.TapdScopeConfig `json:"scopeConfig"`
@@ -38,7 +37,6 @@ type TapdOptions struct {
type TapdTaskData struct {
Options *TapdOptions
ApiClient *helper.ApiAsyncClient
- TimeAfter *time.Time
Connection *models.TapdConnection
}
diff --git a/backend/plugins/tapd/tasks/worklog_collector.go b/backend/plugins/tapd/tasks/worklog_collector.go
index 076b682fb..845fbe161 100644
--- a/backend/plugins/tapd/tasks/worklog_collector.go
+++ b/backend/plugins/tapd/tasks/worklog_collector.go
@@ -34,12 +34,12 @@ func CollectWorklogs(taskCtx plugin.SubTaskContext) errors.Error {
rawDataSubTaskArgs, data := CreateRawDataSubTaskArgs(taskCtx, RAW_WORKLOG_TABLE)
logger := taskCtx.GetLogger()
logger.Info("collect worklogs")
- collectorWithState, err := helper.NewStatefulApiCollector(*rawDataSubTaskArgs, data.TimeAfter)
+ collectorWithState, err := helper.NewStatefulApiCollector(*rawDataSubTaskArgs)
if err != nil {
return err
}
incremental := collectorWithState.IsIncremental()
-
+ syncPolicy := taskCtx.TaskContext().SyncPolicy()
err = collectorWithState.InitCollector(helper.ApiCollectorArgs{
Incremental: incremental,
ApiClient: data.ApiClient,
@@ -51,10 +51,10 @@ func CollectWorklogs(taskCtx plugin.SubTaskContext) errors.Error {
query.Set("page", fmt.Sprintf("%v", reqData.Pager.Page))
query.Set("limit", fmt.Sprintf("%v", reqData.Pager.Size))
query.Set("order", "created asc")
- if data.TimeAfter != nil {
+ if syncPolicy != nil && syncPolicy.TimeAfter != nil {
query.Set("modified",
fmt.Sprintf(">%s",
- data.TimeAfter.In(data.Options.CstZone).Format("2006-01-02")))
+ syncPolicy.TimeAfter.In(data.Options.CstZone).Format("2006-01-02")))
}
if incremental {
query.Set("modified",
diff --git a/backend/plugins/teambition/api/blueprint200.go b/backend/plugins/teambition/api/blueprint200.go
index 6ee599725..4c5611b7c 100644
--- a/backend/plugins/teambition/api/blueprint200.go
+++ b/backend/plugins/teambition/api/blueprint200.go
@@ -19,7 +19,6 @@ package api
import (
"fmt"
- "time"
"github.com/apache/incubator-devlake/plugins/teambition/models"
@@ -37,10 +36,9 @@ func MakeDataSourcePipelinePlanV200(
subtaskMetas []plugin.SubTaskMeta,
connectionId uint64,
bpScopes []*coreModels.BlueprintScope,
- syncPolicy *coreModels.SyncPolicy,
) (coreModels.PipelinePlan, []plugin.Scope, errors.Error) {
plan := make(coreModels.PipelinePlan, len(bpScopes))
- plan, err := makeDataSourcePipelinePlanV200(subtaskMetas, plan, bpScopes, connectionId, syncPolicy)
+ plan, err := makeDataSourcePipelinePlanV200(subtaskMetas, plan, bpScopes, connectionId)
if err != nil {
return nil, nil, err
}
@@ -57,7 +55,6 @@ func makeDataSourcePipelinePlanV200(
plan coreModels.PipelinePlan,
bpScopes []*coreModels.BlueprintScope,
connectionId uint64,
- syncPolicy *coreModels.SyncPolicy,
) (coreModels.PipelinePlan, errors.Error) {
for i, bpScope := range bpScopes {
stage := plan[i]
@@ -68,9 +65,6 @@ func makeDataSourcePipelinePlanV200(
options := make(map[string]interface{})
options["projectId"] = bpScope.ScopeId
options["connectionId"] = connectionId
- if syncPolicy.TimeAfter != nil {
- options["timeAfter"] = syncPolicy.TimeAfter.Format(time.RFC3339)
- }
subtasks, err := helper.MakePipelinePlanSubtasks(subtaskMetas, plugin.DOMAIN_TYPES)
if err != nil {
diff --git a/backend/plugins/teambition/impl/impl.go b/backend/plugins/teambition/impl/impl.go
index 2aeb2f11a..410dfd173 100644
--- a/backend/plugins/teambition/impl/impl.go
+++ b/backend/plugins/teambition/impl/impl.go
@@ -19,7 +19,6 @@ package impl
import (
"fmt"
- "time"
"github.com/apache/incubator-devlake/core/context"
"github.com/apache/incubator-devlake/core/dal"
@@ -122,9 +121,8 @@ func (p Teambition) SubTaskMetas() []plugin.SubTaskMeta {
func (p Teambition) MakeDataSourcePipelinePlanV200(
connectionId uint64,
scopes []*coreModels.BlueprintScope,
- syncPolicy *coreModels.SyncPolicy,
) (pp coreModels.PipelinePlan, sc []plugin.Scope, err errors.Error) {
- return api.MakeDataSourcePipelinePlanV200(p.SubTaskMetas(), connectionId, scopes, syncPolicy)
+ return api.MakeDataSourcePipelinePlanV200(p.SubTaskMetas(), connectionId, scopes)
}
func (p Teambition) PrepareTaskData(taskCtx plugin.TaskContext, options map[string]interface{}) (interface{}, errors.Error) {
@@ -152,16 +150,7 @@ func (p Teambition) PrepareTaskData(taskCtx plugin.TaskContext, options map[stri
ApiClient: apiClient,
TenantId: connection.TenantId,
}
- var createdDateAfter time.Time
- if op.TimeAfter != "" {
- createdDateAfter, err = errors.Convert01(time.Parse(time.RFC3339, op.TimeAfter))
- if err != nil {
- return nil, errors.BadInput.Wrap(err, "invalid value for `createdDateAfter`")
- }
- }
- if !createdDateAfter.IsZero() {
- taskData.TimeAfter = &createdDateAfter
- }
+
return taskData, nil
}
diff --git a/backend/plugins/teambition/tasks/task_data.go b/backend/plugins/teambition/tasks/task_data.go
index 6f4f30747..961f1f33d 100644
--- a/backend/plugins/teambition/tasks/task_data.go
+++ b/backend/plugins/teambition/tasks/task_data.go
@@ -18,16 +18,16 @@ limitations under the License.
package tasks
import (
+ "time"
+
"github.com/apache/incubator-devlake/core/errors"
helper "github.com/apache/incubator-devlake/helpers/pluginhelper/api"
- "time"
)
type TeambitionOptions struct {
ConnectionId uint64 `json:"connectionId"`
ProjectId string `json:"projectId"`
PageSize uint64 `mapstruct:"pageSize"`
- TimeAfter string `json:"timeAfter" mapstructure:"timeAfter,omitempty"`
CstZone *time.Location
TransformationRules TransformationRules `json:"transformationRules"`
}
@@ -35,7 +35,6 @@ type TeambitionOptions struct {
type TeambitionTaskData struct {
Options *TeambitionOptions
ApiClient *helper.ApiAsyncClient
- TimeAfter *time.Time
TenantId string
}
diff --git a/backend/plugins/teambition/teambition.go b/backend/plugins/teambition/teambition.go
index fde97b831..86b4b26c6 100644
--- a/backend/plugins/teambition/teambition.go
+++ b/backend/plugins/teambition/teambition.go
@@ -31,12 +31,13 @@ func main() {
cmd := &cobra.Command{Use: "teambition"}
connectionId := cmd.Flags().Uint64P("connection", "c", 0, "teambition connection id")
projectId := cmd.Flags().StringP("project", "p", "", "teambition project id")
+ timeAfter := cmd.Flags().StringP("timeAfter", "a", "", "collect data that are created after specified time, ie 2006-01-02T15:04:05Z")
cmd.Run = func(cmd *cobra.Command, args []string) {
runner.DirectRun(cmd, args, PluginEntry, map[string]interface{}{
"connectionId": *connectionId,
"projectId": *projectId,
- })
+ }, *timeAfter)
}
runner.RunCmd(cmd)
diff --git a/backend/plugins/trello/api/blueprint_v200.go b/backend/plugins/trello/api/blueprint_v200.go
index 4329e37df..836e5cdd3 100644
--- a/backend/plugins/trello/api/blueprint_v200.go
+++ b/backend/plugins/trello/api/blueprint_v200.go
@@ -18,8 +18,6 @@ limitations under the License.
package api
import (
- "time"
-
"github.com/apache/incubator-devlake/core/models/domainlayer"
"github.com/apache/incubator-devlake/core/models/domainlayer/ticket"
@@ -38,7 +36,6 @@ func MakePipelinePlanV200(
subtaskMetas []plugin.SubTaskMeta,
connectionId uint64,
scope []*coreModels.BlueprintScope,
- syncPolicy *coreModels.SyncPolicy,
) (coreModels.PipelinePlan, []plugin.Scope, errors.Error) {
scopes, err := makeScopeV200(connectionId, scope)
if err != nil {
@@ -46,7 +43,7 @@ func MakePipelinePlanV200(
}
plan := make(coreModels.PipelinePlan, len(scope))
- plan, err = makePipelinePlanV200(subtaskMetas, plan, scope, connectionId, syncPolicy)
+ plan, err = makePipelinePlanV200(subtaskMetas, plan, scope, connectionId)
if err != nil {
return nil, nil, err
}
@@ -81,7 +78,7 @@ func makePipelinePlanV200(
subtaskMetas []plugin.SubTaskMeta,
plan coreModels.PipelinePlan,
scopes []*coreModels.BlueprintScope,
- connectionId uint64, syncPolicy *coreModels.SyncPolicy,
+ connectionId uint64,
) (coreModels.PipelinePlan, errors.Error) {
for i, scope := range scopes {
stage := plan[i]
@@ -93,9 +90,6 @@ func makePipelinePlanV200(
options := make(map[string]interface{})
options["connectionId"] = connectionId
options["scopeId"] = scope.ScopeId
- if syncPolicy.TimeAfter != nil {
- options["timeAfter"] = syncPolicy.TimeAfter.Format(time.RFC3339)
- }
_, scopeConfig, err := scopeHelper.DbHelper().GetScopeAndConfig(connectionId, scope.ScopeId)
if err != nil {
diff --git a/backend/plugins/trello/impl/impl.go b/backend/plugins/trello/impl/impl.go
index 39b177a0a..2b2edaaea 100644
--- a/backend/plugins/trello/impl/impl.go
+++ b/backend/plugins/trello/impl/impl.go
@@ -195,9 +195,8 @@ func (p Trello) ApiResources() map[string]map[string]plugin.ApiResourceHandler {
func (p Trello) MakeDataSourcePipelinePlanV200(
connectionId uint64,
scopes []*coreModels.BlueprintScope,
- syncPolicy *coreModels.SyncPolicy,
) (coreModels.PipelinePlan, []plugin.Scope, errors.Error) {
- return api.MakePipelinePlanV200(p.SubTaskMetas(), connectionId, scopes, syncPolicy)
+ return api.MakePipelinePlanV200(p.SubTaskMetas(), connectionId, scopes)
}
func (p Trello) Close(taskCtx plugin.TaskContext) errors.Error {
diff --git a/backend/plugins/trello/trello.go b/backend/plugins/trello/trello.go
index f2510de1c..d82067006 100644
--- a/backend/plugins/trello/trello.go
+++ b/backend/plugins/trello/trello.go
@@ -30,13 +30,14 @@ func main() {
cmd := &cobra.Command{Use: "trello"}
connectionId := cmd.Flags().Uint64P("connection", "c", 0, "trello connection id")
boardId := cmd.Flags().StringP("board", "b", "", "trello board id")
+ timeAfter := cmd.Flags().StringP("timeAfter", "a", "", "collect data that are created after specified time, ie 2006-01-02T15:04:05Z")
_ = cmd.MarkFlagRequired("connection")
_ = cmd.MarkFlagRequired("board")
cmd.Run = func(c *cobra.Command, args []string) {
runner.DirectRun(c, args, PluginEntry, map[string]interface{}{
"connectionId": *connectionId,
"boardId": *boardId,
- })
+ }, *timeAfter)
}
runner.RunCmd(cmd)
}
diff --git a/backend/plugins/zentao/api/blueprint_V200_test.go b/backend/plugins/zentao/api/blueprint_V200_test.go
index b379b7e3d..2cd62a269 100644
--- a/backend/plugins/zentao/api/blueprint_V200_test.go
+++ b/backend/plugins/zentao/api/blueprint_V200_test.go
@@ -71,10 +71,9 @@ func TestMakeDataSourcePipelinePlanV200(t *testing.T) {
}*/
bpScopes := make([]*coreModels.BlueprintScope, 0)
bpScopes = append(bpScopes, bs)
- syncPolicy := &coreModels.SyncPolicy{}
plan := make(coreModels.PipelinePlan, len(bpScopes))
- plan, scopes, err := makePipelinePlanV200(nil, plan, bpScopes, connection, syncPolicy)
+ plan, scopes, err := makePipelinePlanV200(nil, plan, bpScopes, connection)
assert.Nil(t, err)
expectPlan := coreModels.PipelinePlan{
diff --git a/backend/plugins/zentao/api/blueprint_v200.go b/backend/plugins/zentao/api/blueprint_v200.go
index b316cff43..d86105021 100644
--- a/backend/plugins/zentao/api/blueprint_v200.go
+++ b/backend/plugins/zentao/api/blueprint_v200.go
@@ -18,8 +18,6 @@ limitations under the License.
package api
import (
- "time"
-
"github.com/apache/incubator-devlake/core/errors"
coreModels "github.com/apache/incubator-devlake/core/models"
"github.com/apache/incubator-devlake/core/models/domainlayer"
@@ -36,7 +34,6 @@ func MakeDataSourcePipelinePlanV200(
subtaskMetas []plugin.SubTaskMeta,
connectionId uint64,
bpScopes []*coreModels.BlueprintScope,
- syncPolicy *coreModels.SyncPolicy,
) (coreModels.PipelinePlan, []plugin.Scope, errors.Error) {
// get the connection info for url
connection := &models.ZentaoConnection{}
@@ -46,7 +43,7 @@ func MakeDataSourcePipelinePlanV200(
}
plan := make(coreModels.PipelinePlan, len(bpScopes))
- plan, scopes, err := makePipelinePlanV200(subtaskMetas, plan, bpScopes, connection, syncPolicy)
+ plan, scopes, err := makePipelinePlanV200(subtaskMetas, plan, bpScopes, connection)
if err != nil {
return nil, nil, err
}
@@ -59,7 +56,6 @@ func makePipelinePlanV200(
plan coreModels.PipelinePlan,
bpScopes []*coreModels.BlueprintScope,
connection *models.ZentaoConnection,
- syncPolicy *coreModels.SyncPolicy,
) (coreModels.PipelinePlan, []plugin.Scope, errors.Error) {
domainScopes := make([]plugin.Scope, 0)
for i, bpScope := range bpScopes {
@@ -111,9 +107,6 @@ func makePipelinePlanV200(
}
}*/
- if syncPolicy.TimeAfter != nil {
- op.TimeAfter = syncPolicy.TimeAfter.Format(time.RFC3339)
- }
options, err := tasks.EncodeTaskOptions(op)
if err != nil {
return nil, nil, err
diff --git a/backend/plugins/zentao/impl/impl.go b/backend/plugins/zentao/impl/impl.go
index cbea91b34..9f2f760cf 100644
--- a/backend/plugins/zentao/impl/impl.go
+++ b/backend/plugins/zentao/impl/impl.go
@@ -283,9 +283,8 @@ func (p Zentao) ApiResources() map[string]map[string]plugin.ApiResourceHandler {
func (p Zentao) MakeDataSourcePipelinePlanV200(
connectionId uint64,
scopes []*coreModels.BlueprintScope,
- syncPolicy *coreModels.SyncPolicy,
) (pp coreModels.PipelinePlan, sc []plugin.Scope, err errors.Error) {
- return api.MakeDataSourcePipelinePlanV200(p.SubTaskMetas(), connectionId, scopes, syncPolicy)
+ return api.MakeDataSourcePipelinePlanV200(p.SubTaskMetas(), connectionId, scopes)
}
func (p Zentao) Close(taskCtx plugin.TaskContext) errors.Error {
diff --git a/backend/plugins/zentao/tasks/bug_commits_collector.go b/backend/plugins/zentao/tasks/bug_commits_collector.go
index 63109bdfa..526ffc8a4 100644
--- a/backend/plugins/zentao/tasks/bug_commits_collector.go
+++ b/backend/plugins/zentao/tasks/bug_commits_collector.go
@@ -56,7 +56,7 @@ func CollectBugCommits(taskCtx plugin.SubTaskContext) errors.Error {
Ctx: taskCtx,
Options: data.Options,
Table: RAW_BUG_COMMITS_TABLE,
- }, data.TimeAfter)
+ })
if err != nil {
return err
}
diff --git a/backend/plugins/zentao/tasks/story_commits_collector.go b/backend/plugins/zentao/tasks/story_commits_collector.go
index 1c1552d66..117fdc7d9 100644
--- a/backend/plugins/zentao/tasks/story_commits_collector.go
+++ b/backend/plugins/zentao/tasks/story_commits_collector.go
@@ -51,7 +51,7 @@ func CollectStoryCommits(taskCtx plugin.SubTaskContext) errors.Error {
Ctx: taskCtx,
Options: data.Options,
Table: RAW_STORY_COMMITS_TABLE,
- }, data.TimeAfter)
+ })
if err != nil {
return err
}
diff --git a/backend/plugins/zentao/tasks/task_commits_collector.go b/backend/plugins/zentao/tasks/task_commits_collector.go
index 0d6bacd84..4d4c54feb 100644
--- a/backend/plugins/zentao/tasks/task_commits_collector.go
+++ b/backend/plugins/zentao/tasks/task_commits_collector.go
@@ -50,7 +50,7 @@ func CollectTaskCommits(taskCtx plugin.SubTaskContext) errors.Error {
Ctx: taskCtx,
Options: data.Options,
Table: RAW_TASK_COMMITS_TABLE,
- }, data.TimeAfter)
+ })
if err != nil {
return err
}
diff --git a/backend/plugins/zentao/tasks/task_data.go b/backend/plugins/zentao/tasks/task_data.go
index 9cf09292b..26bc9734c 100644
--- a/backend/plugins/zentao/tasks/task_data.go
+++ b/backend/plugins/zentao/tasks/task_data.go
@@ -19,7 +19,6 @@ package tasks
import (
"fmt"
- "time"
"github.com/apache/incubator-devlake/core/dal"
"github.com/apache/incubator-devlake/core/errors"
@@ -37,7 +36,6 @@ type ZentaoOptions struct {
ConnectionId uint64 `json:"connectionId"`
ProjectId int64 `json:"projectId" mapstructure:"projectId"`
// TODO not support now
- TimeAfter string `json:"timeAfter" mapstructure:"timeAfter,omitempty"`
ScopeConfigId uint64 `json:"scopeConfigId" mapstructure:"scopeConfigId,omitempty"`
ScopeConfig *models.ZentaoScopeConfig `json:"scopeConfig" mapstructure:"scopeConfig,omitempty"`
}
@@ -53,7 +51,6 @@ type ZentaoTaskData struct {
Options *ZentaoOptions
RemoteDb dal.Dal
- TimeAfter *time.Time
ProjectName string
Stories map[int64]struct{}
Tasks map[int64]struct{}
diff --git a/backend/plugins/zentao/zentao.go b/backend/plugins/zentao/zentao.go
index fcbbeaa55..fb1713c82 100644
--- a/backend/plugins/zentao/zentao.go
+++ b/backend/plugins/zentao/zentao.go
@@ -33,13 +33,14 @@ func main() {
connectionId := cmd.Flags().Uint64P("connectionId", "c", 0, "zentao connection id")
productId := cmd.Flags().IntP("productId", "o", 8, "product id")
projectId := cmd.Flags().IntP("projectId", "p", 8, "project id")
+ timeAfter := cmd.Flags().StringP("timeAfter", "a", "", "collect data that are created after specified time, ie 2006-01-02T15:04:05Z")
cmd.Run = func(cmd *cobra.Command, args []string) {
runner.DirectRun(cmd, args, PluginEntry, map[string]interface{}{
"connectionId": *connectionId,
"productId": *productId,
"projectId": *projectId,
- })
+ }, *timeAfter)
}
runner.RunCmd(cmd)
}
diff --git a/backend/server/api/blueprints/blueprints.go b/backend/server/api/blueprints/blueprints.go
index 389247abb..b4f7410de 100644
--- a/backend/server/api/blueprints/blueprints.go
+++ b/backend/server/api/blueprints/blueprints.go
@@ -193,7 +193,6 @@ func Trigger(c *gin.Context) {
return
}
}
-
pipeline, err := services.TriggerBlueprint(id, syncPolicy)
if err != nil {
shared.ApiOutputError(c, errors.Default.Wrap(err, "error triggering blueprint"))
diff --git a/backend/server/services/blueprint.go b/backend/server/services/blueprint.go
index 29246b43f..7176ddabc 100644
--- a/backend/server/services/blueprint.go
+++ b/backend/server/services/blueprint.go
@@ -53,12 +53,7 @@ type BlueprintJob struct {
func (bj BlueprintJob) Run() {
blueprint := bj.Blueprint
- syncPolicy := &models.SyncPolicy{
- TimeAfter: blueprint.TimeAfter,
- FullSync: blueprint.FullSync,
- SkipOnFail: blueprint.SkipOnFail,
- }
- pipeline, err := createPipelineByBlueprint(blueprint, syncPolicy)
+ pipeline, err := createPipelineByBlueprint(blueprint, &blueprint.SyncPolicy)
if err == ErrEmptyPlan {
blueprintLog.Info("Empty plan, blueprint id:[%d] blueprint name:[%s]", blueprint.ID, blueprint.Name)
return
@@ -274,9 +269,6 @@ func ReloadBlueprints(c *cron.Cron) errors.Error {
func createPipelineByBlueprint(blueprint *models.Blueprint, syncPolicy *models.SyncPolicy) (*models.Pipeline, errors.Error) {
var plan models.PipelinePlan
var err errors.Error
- if syncPolicy != nil && syncPolicy.FullSync {
- blueprint.FullSync = true
- }
if blueprint.Mode == models.BLUEPRINT_MODE_NORMAL {
plan, err = MakePlanForBlueprint(blueprint, syncPolicy)
if err != nil {
@@ -286,18 +278,13 @@ func createPipelineByBlueprint(blueprint *models.Blueprint, syncPolicy *models.S
} else {
plan = blueprint.Plan
}
+
newPipeline := models.NewPipeline{}
newPipeline.Plan = plan
newPipeline.Name = blueprint.Name
newPipeline.BlueprintId = blueprint.ID
newPipeline.Labels = blueprint.Labels
- newPipeline.SkipOnFail = blueprint.SkipOnFail
-
- if syncPolicy != nil && syncPolicy.FullSync {
- newPipeline.FullSync = true
- } else {
- newPipeline.FullSync = blueprint.FullSync
- }
+ newPipeline.SyncPolicy = blueprint.SyncPolicy
// if the plan is empty, we should not create the pipeline
var shouldCreatePipeline bool
@@ -326,11 +313,6 @@ func createPipelineByBlueprint(blueprint *models.Blueprint, syncPolicy *models.S
// MakePlanForBlueprint generates pipeline plan by version
func MakePlanForBlueprint(blueprint *models.Blueprint, syncPolicy *models.SyncPolicy) (models.PipelinePlan, errors.Error) {
- if syncPolicy != nil {
- syncPolicy.TimeAfter = blueprint.TimeAfter
- syncPolicy.FullSync = blueprint.FullSync
- }
-
var plan models.PipelinePlan
// load project metric plugins and convert it to a map
metrics := make(map[string]json.RawMessage)
@@ -344,7 +326,11 @@ func MakePlanForBlueprint(blueprint *models.Blueprint, syncPolicy *models.SyncPo
metrics[projectMetric.PluginName] = json.RawMessage(projectMetric.PluginOption)
}
}
- plan, err := GeneratePlanJsonV200(blueprint.ProjectName, syncPolicy, blueprint.Connections, metrics)
+ skipCollectors := false
+ if syncPolicy != nil && syncPolicy.SkipCollectors {
+ skipCollectors = true
+ }
+ plan, err := GeneratePlanJsonV200(blueprint.ProjectName, blueprint.Connections, metrics, skipCollectors)
if err != nil {
return nil, err
}
@@ -388,5 +374,8 @@ func TriggerBlueprint(id uint64, syncPolicy *models.SyncPolicy) (*models.Pipelin
logger.Error(err, "GetBlueprint, id: %d", id)
return nil, err
}
+ blueprint.SkipCollectors = syncPolicy.SkipCollectors
+ blueprint.FullSync = syncPolicy.FullSync
+
return createPipelineByBlueprint(blueprint, syncPolicy)
}
diff --git a/backend/server/services/blueprint_makeplan_v200.go b/backend/server/services/blueprint_makeplan_v200.go
index b4f1bfff7..69e11bfc2 100644
--- a/backend/server/services/blueprint_makeplan_v200.go
+++ b/backend/server/services/blueprint_makeplan_v200.go
@@ -30,9 +30,9 @@ import (
// GeneratePlanJsonV200 generates pipeline plan according v2.0.0 definition
func GeneratePlanJsonV200(
projectName string,
- syncPolicy *coreModels.SyncPolicy,
connections []*coreModels.BlueprintConnection,
metrics map[string]json.RawMessage,
+ skipCollectors bool,
) (coreModels.PipelinePlan, errors.Error) {
var err errors.Error
// make plan for data-source coreModels fist. generate plan for each
@@ -56,7 +56,6 @@ func GeneratePlanJsonV200(
sourcePlans[i], pluginScopes, err = pluginBp.MakeDataSourcePipelinePlanV200(
connection.ConnectionId,
connection.Scopes,
- syncPolicy,
)
if err != nil {
return nil, err
@@ -72,7 +71,7 @@ func GeneratePlanJsonV200(
}
// skip collectors
- if syncPolicy != nil && syncPolicy.SkipCollectors {
+ if skipCollectors {
for i, plan := range sourcePlans {
for j, stage := range plan {
for k, task := range stage {
diff --git a/backend/server/services/blueprint_makeplan_v200_test.go b/backend/server/services/blueprint_makeplan_v200_test.go
index 093b9ea69..22b53b532 100644
--- a/backend/server/services/blueprint_makeplan_v200_test.go
+++ b/backend/server/services/blueprint_makeplan_v200_test.go
@@ -36,7 +36,6 @@ func TestMakePlanV200(t *testing.T) {
githubName := "TestMakePlanV200-github" // mimic github
// mock github plugin as a data source plugin
githubConnId := uint64(1)
- syncPolicy := &coreModels.SyncPolicy{}
githubScopes := []*coreModels.BlueprintScope{
{ScopeId: "github:GithubRepo:1:123"},
{ScopeId: "github:GithubRepo:1:321"},
@@ -56,7 +55,7 @@ func TestMakePlanV200(t *testing.T) {
&ticket.Board{DomainEntity: domainlayer.DomainEntity{Id: "github:GithubRepo:1:123"}, Name: "apache/incubator-devlake"},
}
github := new(mockplugin.CompositeDataSourcePluginBlueprintV200)
- github.On("MakeDataSourcePipelinePlanV200", githubConnId, githubScopes, syncPolicy).Return(githubOutputPlan, githubOutputScopes, nil)
+ github.On("MakeDataSourcePipelinePlanV200", githubConnId, githubScopes).Return(githubOutputPlan, githubOutputScopes, nil)
// mock dora plugin as a metric plugin
doraName := "TestMakePlanV200-dora"
@@ -97,7 +96,7 @@ func TestMakePlanV200(t *testing.T) {
doraName: nil,
}
- plan, err := GeneratePlanJsonV200(projectName, syncPolicy, connections, metrics)
+ plan, err := GeneratePlanJsonV200(projectName, connections, metrics, false)
assert.Nil(t, err)
assert.Equal(t, expectedPlan, plan)
diff --git a/backend/server/services/pipeline_helper.go b/backend/server/services/pipeline_helper.go
index a681608e0..11c00e94d 100644
--- a/backend/server/services/pipeline_helper.go
+++ b/backend/server/services/pipeline_helper.go
@@ -61,8 +61,7 @@ func CreateDbPipeline(newPipeline *models.NewPipeline) (pipeline *models.Pipelin
Message: "",
SpentSeconds: 0,
Plan: newPipeline.Plan,
- SkipOnFail: newPipeline.SkipOnFail,
- FullSync: newPipeline.FullSync,
+ SyncPolicy: newPipeline.SyncPolicy,
}
if newPipeline.BlueprintId != 0 {
dbPipeline.BlueprintId = newPipeline.BlueprintId
diff --git a/backend/server/services/remote/plugin/plugin_extensions.go b/backend/server/services/remote/plugin/plugin_extensions.go
index 09e070383..5f1f4612c 100644
--- a/backend/server/services/remote/plugin/plugin_extensions.go
+++ b/backend/server/services/remote/plugin/plugin_extensions.go
@@ -43,7 +43,6 @@ func (p remoteMetricPlugin) MakeMetricPluginPipelinePlanV200(projectName string,
func (p remoteDatasourcePlugin) MakeDataSourcePipelinePlanV200(
connectionId uint64,
bpScopes []*coreModels.BlueprintScope,
- syncPolicy *coreModels.SyncPolicy,
) (coreModels.PipelinePlan, []plugin.Scope, errors.Error) {
connection := p.connectionTabler.New()
err := p.connHelper.FirstById(connection, connectionId)
diff --git a/backend/server/services/remote/run/run.go b/backend/server/services/remote/run/run.go
index 81c656bc1..d9b2c63eb 100644
--- a/backend/server/services/remote/run/run.go
+++ b/backend/server/services/remote/run/run.go
@@ -59,7 +59,7 @@ func main() {
}
options["connectionId"] = *connectionId
- runner.DirectRun(cmd, args, plugin, options)
+ runner.DirectRun(cmd, args, plugin, options, "")
}
runner.RunCmd(cmd)
}
diff --git a/backend/test/helper/api.go b/backend/test/helper/api.go
index 5caf817d0..9099f90ad 100644
--- a/backend/test/helper/api.go
+++ b/backend/test/helper/api.go
@@ -86,8 +86,10 @@ func (d *DevlakeClient) CreateBasicBlueprintV2(name string, config *BlueprintV2C
Enable: true,
CronConfig: "manual",
IsManual: true,
- SkipOnFail: config.SkipOnFail,
- Labels: []string{"test-label"},
+ SyncPolicy: models.SyncPolicy{
+ SkipOnFail: config.SkipOnFail,
+ },
+ Labels: []string{"test-label"},
Connections: []*models.BlueprintConnection{
config.Connection,
},