feat: add jira_trace plugin to jira bp

backend/plugins/issue_trace/e2e/meta.go

@@ -21,9 +21,8 @@ import "github.com/apache/incubator-devlake/plugins/issue_trace/tasks"
 
 var TaskData = &tasks.TaskData{
 	Options: tasks.Options{
-		Plugin:       "jira",
-		ConnectionId: 2,
-		BoardId:      8,
+		Plugin:   "jira",
+		ScopeIds: []string{"jira:JiraBoard:2:8"},
 	},
-	BoardId: "jira:JiraBoard:2:8",
+	ScopeIds: []string{"jira:JiraBoard:2:8"},
 }

backend/plugins/issue_trace/e2e/snapshot_tables/issue_status_history.csv

@@ -1,7 +1,7 @@
 issue_id,original_status,start_date,status,is_current_status,is_first_status
 jira:JiraIssue:2:10063,Closed,2020-06-19T06:31:18.526+00:00,完成,0,0
-jira:JiraIssue:2:10063,Inbox,2020-06-12T00:13:13.360+00:00,等待,0,1
 jira:JiraIssue:2:10063,In Development,2020-06-12T03:38:39.785+00:00,开发中,0,0
+jira:JiraIssue:2:10063,Inbox,2020-06-12T00:13:13.360+00:00,等待,0,1
 jira:JiraIssue:2:10063,Open,2020-06-12T06:22:49.352+00:00,等待,0,0
 jira:JiraIssue:2:10063,Ready for Dev,2020-06-12T01:45:24.493+00:00,待开发,0,0
 jira:JiraIssue:2:10063,已完成,2021-03-28T08:06:08.715+00:00,完成,1,0
@@ -81,8 +81,8 @@ jira:JiraIssue:2:10079,Resolved,2020-07-22T07:25:29.223+00:00,完成,0,0
 jira:JiraIssue:2:10079,To Do,2020-06-12T00:24:56.048+00:00,等待,0,1
 jira:JiraIssue:2:10079,已完成,2021-03-28T08:05:54.428+00:00,完成,1,0
 jira:JiraIssue:2:10081,Closed,2020-11-10T01:43:08.611+00:00,完成,0,0
-jira:JiraIssue:2:10081,Inbox,2020-06-12T00:28:00.241+00:00,等待,0,1
 jira:JiraIssue:2:10081,In Development,2020-06-12T01:46:26.809+00:00,开发中,0,0
+jira:JiraIssue:2:10081,Inbox,2020-06-12T00:28:00.241+00:00,等待,0,1
 jira:JiraIssue:2:10081,Open,2020-06-12T05:01:20.620+00:00,等待,0,0
 jira:JiraIssue:2:10081,Ready for Dev,2020-06-12T01:46:22.354+00:00,待开发,0,0
 jira:JiraIssue:2:10081,Resolved,2020-06-12T01:46:30.363+00:00,完成,0,0

backend/plugins/issue_trace/impl/enricher.go

@@ -18,15 +18,17 @@ limitations under the License.
 package impl
 
 import (
+	"encoding/json"
+
 	"github.com/apache/incubator-devlake/core/context"
 	"github.com/apache/incubator-devlake/core/dal"
 	"github.com/apache/incubator-devlake/core/errors"
-	"github.com/apache/incubator-devlake/core/models/domainlayer/didgen"
+	coreModels "github.com/apache/incubator-devlake/core/models"
+	"github.com/apache/incubator-devlake/core/models/domainlayer/crossdomain"
 	"github.com/apache/incubator-devlake/core/plugin"
 	"github.com/apache/incubator-devlake/plugins/issue_trace/api"
 	"github.com/apache/incubator-devlake/plugins/issue_trace/models"
 	"github.com/apache/incubator-devlake/plugins/issue_trace/models/migrationscripts"
-	"github.com/apache/incubator-devlake/plugins/issue_trace/services"
 	"github.com/apache/incubator-devlake/plugins/issue_trace/tasks"
 	"github.com/mitchellh/mapstructure"
 )
@@ -38,14 +40,40 @@ var _ interface {
 	plugin.PluginInit
 	plugin.PluginTask
 	plugin.PluginModel
+	plugin.PluginMetric
 	plugin.PluginMigration
 	plugin.PluginApi
+	plugin.MetricPluginBlueprintV200
 } = (*IssueTrace)(nil)
 
 func (p IssueTrace) Name() string {
 	return "issue_trace"
 }
 
+func (p IssueTrace) RequiredDataEntities() (data []map[string]interface{}, err errors.Error) {
+	return []map[string]interface{}{
+		{
+			"model": "issue_changelogs",
+			"requiredFields": map[string]string{
+				"column":        "type",
+				"execptedValue": "Issue",
+			},
+		},
+	}, nil
+}
+
+func (p IssueTrace) IsProjectMetric() bool {
+	return true
+}
+
+func (p IssueTrace) RunAfter() ([]string, errors.Error) {
+	return []string{}, nil
+}
+
+func (p IssueTrace) Settings() interface{} {
+	return nil
+}
+
 func (p IssueTrace) Init(basicRes context.BasicRes) errors.Error {
 	api.Init(basicRes)
 	return nil
@@ -69,32 +97,36 @@ func (p IssueTrace) SubTaskMetas() []plugin.SubTaskMeta {
 // `apiClient` is defined in `client.go` under `tasks`
 // `SprintPerformanceEnricherTaskData` is defined in `task_data.go` under `tasks`
 func (p IssueTrace) PrepareTaskData(taskCtx plugin.TaskContext, options map[string]interface{}) (interface{}, errors.Error) {
-	logger := taskCtx.GetLogger()
 	var op tasks.Options
 	err := mapstructure.Decode(options, &op)
 	if err != nil {
 		return nil, errors.Default.Wrap(err, "Failed to decode options")
 	}
-	var boardId string
-	if op.LakeBoardId != "" {
-		boardId = op.LakeBoardId
+	var scopeIds []string
+	if op.ScopeIds != nil {
+		scopeIds = op.ScopeIds
 	} else {
-		boardModel := services.GetTicketBoardModel(op.Plugin)
-		if boardModel == nil {
-			err := errors.BadInput.New("unsupported board type")
-			logger.Error(err, "")
-			return nil, err
+		db := taskCtx.GetDal()
+		pmClauses := []dal.Clause{
+			dal.From("project_mapping pm"),
+			dal.Where("pm.project_name = ? and pm.table = ?", op.ProjectName, "boards"),
+		}
+		pm := []crossdomain.ProjectMapping{}
+		err = db.All(&pm, pmClauses...)
+		if err != nil {
+			return nil, errors.Default.Wrap(err, "Failed to get project mapping")
+		}
+		for _, p := range pm {
+			scopeIds = append(scopeIds, p.RowId)
 		}
-		boardIdGen := didgen.NewDomainIdGenerator(boardModel)
-		boardId = boardIdGen.Generate(op.ConnectionId, op.BoardId)
 	}
 
 	var taskData = &tasks.TaskData{
-		Options: op,
-		BoardId: boardId,
+		Options:     op,
+		ScopeIds:    scopeIds,
+		ProjectName: op.ProjectName,
 	}
 
-	taskData.Options = op
 	return taskData, nil
 }
 
@@ -118,3 +150,30 @@ func (p IssueTrace) GetTablesInfo() []dal.Tabler {
 		&models.IssueStatusHistory{},
 	}
 }
+
+func (p IssueTrace) MakeMetricPluginPipelinePlanV200(projectName string, options json.RawMessage) (coreModels.PipelinePlan, errors.Error) {
+	op := &tasks.Options{}
+	if options != nil && string(options) != "\"\"" {
+		err := json.Unmarshal(options, op)
+		if err != nil {
+			return nil, errors.Default.WrapRaw(err)
+		}
+	}
+
+	plan := coreModels.PipelinePlan{
+		{
+			{
+				Plugin: "issue_trace",
+				Options: map[string]interface{}{
+					"projectName": projectName,
+					"scopeIds":    op.ScopeIds,
+				},
+				Subtasks: []string{
+					"ConvertIssueStatusHistory",
+					"ConvertIssueAssigneeHistory",
+				},
+			},
+		},
+	}
+	return plan, nil
+}

backend/plugins/issue_trace/tasks/issue_assignee_history_convertor.go

@@ -58,10 +58,10 @@ var ConvertIssueAssigneeHistoryMeta = plugin.SubTaskMeta{
 func ConvertIssueAssigneeHistory(taskCtx plugin.SubTaskContext) errors.Error {
 	logger := taskCtx.GetLogger()
 	options := taskCtx.GetData().(*TaskData)
-	boardId := options.BoardId
+	scopeIds := options.ScopeIds
 	db := taskCtx.GetDal()
 
-	insertor := helper.NewBatchSaveDivider(taskCtx, utils.BATCH_SIZE, rawTableIssueChangelogs, boardId)
+	insertor := helper.NewBatchSaveDivider(taskCtx, utils.BATCH_SIZE, "", "")
 	defer insertor.Close()
 	batchInsertor, err := insertor.ForType(reflect.TypeOf(&models.IssueAssigneeHistory{}))
 	if err != nil {
@@ -86,18 +86,18 @@ func ConvertIssueAssigneeHistory(taskCtx plugin.SubTaskContext) errors.Error {
 	where
 		issue_changelogs.issue_id is null
 		and assignee_id != ''
-		and board_issues.board_id = ?;
-		`, boardId)
+		and board_issues.board_id in ?;
+		`, scopeIds)
 	if err != nil {
 		logger.Error(err, "Failed to query issue assignee")
 		return err
 	}
 	defer cursorForIssuesWithoutChanglog.Close()
 	convertorForIssuesWithoutChangelog, err := helper.NewDataConverter(helper.DataConverterArgs{
 		RawDataSubTaskArgs: helper.RawDataSubTaskArgs{
-			Ctx:    taskCtx,
-			Params: boardId,
-			Table:  "issue_changelogs",
+			Ctx: taskCtx,
+			// Params: scopeId,
+			Table: "issue_changelogs",
 		},
 		InputRowType: reflect.TypeOf(AssigneeHistory{}),
 		Input:        cursorForIssuesWithoutChanglog,
@@ -152,7 +152,7 @@ func ConvertIssueAssigneeHistory(taskCtx plugin.SubTaskContext) errors.Error {
 		dal.From("issue_changelogs"),
 		dal.Join("JOIN board_issues ON issue_changelogs.issue_id = board_issues.issue_id"),
 		dal.Join("JOIN issues ON issue_changelogs.issue_id = issues.id"),
-		dal.Where("field_name = 'assignee' AND board_issues.board_id = ?", boardId),
+		dal.Where("field_name = 'assignee' AND board_issues.board_id in ?", scopeIds),
 		dal.Orderby("issue_changelogs.issue_id ASC, issue_changelogs.created_date ASC"),
 	}
 	cursorForIssueChangelogs, err := db.Cursor(clauses...)
@@ -167,9 +167,9 @@ func ConvertIssueAssigneeHistory(taskCtx plugin.SubTaskContext) errors.Error {
 
 	convertorForIssueChangelogs, err := helper.NewDataConverter(helper.DataConverterArgs{
 		RawDataSubTaskArgs: helper.RawDataSubTaskArgs{
-			Ctx:    taskCtx,
-			Params: boardId,
-			Table:  "issue_changelogs",
+			Ctx: taskCtx,
+			// Params: scopeId,
+			Table: "issue_changelogs",
 		},
 		InputRowType: reflect.TypeOf(AssigneeChangelog{}),
 		Input:        cursorForIssueChangelogs,
@@ -180,7 +180,7 @@ func ConvertIssueAssigneeHistory(taskCtx plugin.SubTaskContext) errors.Error {
 			row := inputRow.(*AssigneeChangelog)
 			if row.IssueId != currentIssue {
 				if len(currentLogs) > 0 {
-					historyRows := buildActiveAssigneeHistory(currentLogs, boardId)
+					historyRows := buildActiveAssigneeHistory(currentLogs)
 					for _, row := range historyRows {
 						err := batchInsertor.Add(row)
 						if err != nil {
@@ -206,7 +206,7 @@ func ConvertIssueAssigneeHistory(taskCtx plugin.SubTaskContext) errors.Error {
 		return err
 	}
 	if len(currentLogs) > 0 {
-		historyRows := buildActiveAssigneeHistory(currentLogs, boardId)
+		historyRows := buildActiveAssigneeHistory(currentLogs)
 		for _, row := range historyRows {
 			err := batchInsertor.Add(row)
 			if err != nil {
@@ -218,7 +218,7 @@ func ConvertIssueAssigneeHistory(taskCtx plugin.SubTaskContext) errors.Error {
 	return nil
 }
 
-func buildActiveAssigneeHistory(logs []AssigneeChangelog, boardId string) []*models.IssueAssigneeHistory {
+func buildActiveAssigneeHistory(logs []AssigneeChangelog) []*models.IssueAssigneeHistory {
 	// prepend changelog if first line has from_assignee
 	firstLine := logs[0]
 	if firstLine.FromAssignee != "" {

backend/plugins/issue_trace/tasks/issue_status_history_convertor.go

@@ -49,8 +49,6 @@ type StatusChangeLogResult struct {
 	OriginalToValue   string
 }
 
-var rawTableIssueChangelogs = "issue_changelogs"
-
 var ConvertIssueStatusHistoryMeta = plugin.SubTaskMeta{
 	Name:             "ConvertIssueStatusHistory",
 	EntryPoint:       ConvertIssueStatusHistory,
@@ -61,10 +59,10 @@ var ConvertIssueStatusHistoryMeta = plugin.SubTaskMeta{
 func ConvertIssueStatusHistory(taskCtx plugin.SubTaskContext) errors.Error {
 	logger := taskCtx.GetLogger()
 	options := taskCtx.GetData().(*TaskData)
-	boardId := options.BoardId
+	scopeIds := options.ScopeIds
 
 	db := taskCtx.GetDal()
-	inserter := helper.NewBatchSaveDivider(taskCtx, utils.BATCH_SIZE, rawTableIssueChangelogs, boardId)
+	inserter := helper.NewBatchSaveDivider(taskCtx, utils.BATCH_SIZE, "", "")
 	defer inserter.Close()
 	batchInserter, err := inserter.ForType(reflect.TypeOf(&models.IssueStatusHistory{}))
 	if err != nil {
@@ -73,15 +71,15 @@ func ConvertIssueStatusHistory(taskCtx plugin.SubTaskContext) errors.Error {
 	}
 
 	// handle issues not appeared in change logs (hasn't changed)
-	logger.Info("get issues not appeared in change logs, board %s", boardId)
+	logger.Info("get issues not appeared in change logs, board %s", scopeIds)
 	now := time.Now()
 	clauses := []dal.Clause{
 		dal.Select("issues.id AS issue_id, issues.status, issues.original_status, issues.created_date," +
 			"board_issues.board_id as _raw_data_params, issues._raw_data_table as _raw_data_table, issues._raw_data_id as _raw_data_id"),
 		dal.From("issues"),
 		dal.Join("INNER JOIN board_issues ON board_issues.issue_id = issues.id"),
 		dal.Join("LEFT JOIN issue_changelogs ON issue_changelogs.issue_id=issues.id AND issue_changelogs.field_name='status'"),
-		dal.Where("board_issues.board_id = ? AND issue_changelogs.field_name IS NULL", boardId),
+		dal.Where("board_issues.board_id in ? AND issue_changelogs.field_name IS NULL", scopeIds),
 	}
 	statusFromIssueCursor, err := db.Cursor(clauses...)
 	if err != nil {
@@ -136,7 +134,7 @@ func ConvertIssueStatusHistory(taskCtx plugin.SubTaskContext) errors.Error {
 	}
 
 	// handle issues with changelogs
-	logger.Info("get issue status change log, board %s", boardId)
+	logger.Info("get issue status change log, board %s", scopeIds)
 	clauses = []dal.Clause{
 		dal.Select("issue_changelogs.issue_id, issue_changelogs.created_date AS log_created_date, " +
 			"issue_changelogs.to_value, issue_changelogs.original_to_value, issue_changelogs.from_value, " +
@@ -145,7 +143,7 @@ func ConvertIssueStatusHistory(taskCtx plugin.SubTaskContext) errors.Error {
 		dal.From("issue_changelogs"),
 		dal.Join("INNER JOIN issues ON issues.id = issue_changelogs.issue_id"),
 		dal.Join("INNER JOIN board_issues ON board_issues.issue_id = issue_changelogs.issue_id"),
-		dal.Where("board_issues.board_id = ? AND issue_changelogs.field_name = 'status'", boardId),
+		dal.Where("board_issues.board_id in ? AND issue_changelogs.field_name = 'status'", scopeIds),
 		dal.Orderby("issue_changelogs.issue_id ASC, issue_changelogs.created_date ASC"),
 	}
 	statusFromChangelogCursor, err := db.Cursor(clauses...)
@@ -173,7 +171,7 @@ func ConvertIssueStatusHistory(taskCtx plugin.SubTaskContext) errors.Error {
 			logRow := inputRow.(*StatusChangeLogResult)
 			if logRow.IssueId != currentIssue { // reach new issue section
 				if len(currentLogs) > 0 {
-					historyRows := buildStatusHistoryRecords(currentLogs, boardId)
+					historyRows := buildStatusHistoryRecords(currentLogs)
 					for _, r := range historyRows {
 						if r.EndDate != nil {
 							var seconds int64 = 0
@@ -206,7 +204,7 @@ func ConvertIssueStatusHistory(taskCtx plugin.SubTaskContext) errors.Error {
 	}
 
 	if len(currentLogs) > 0 {
-		historyRows := buildStatusHistoryRecords(currentLogs, boardId)
+		historyRows := buildStatusHistoryRecords(currentLogs)
 		for _, r := range historyRows {
 			if r.EndDate != nil {
 				var seconds int64 = 0
@@ -225,7 +223,7 @@ func ConvertIssueStatusHistory(taskCtx plugin.SubTaskContext) errors.Error {
 	return nil
 }
 
-func buildStatusHistoryRecords(logs []*StatusChangeLogResult, boardId string) []*models.IssueStatusHistory {
+func buildStatusHistoryRecords(logs []*StatusChangeLogResult) []*models.IssueStatusHistory {
 	if len(logs) == 0 {
 		return make([]*models.IssueStatusHistory, 0)
 	}

backend/plugins/issue_trace/tasks/issue_status_history_convertor_test.go

@@ -44,7 +44,7 @@ func Test_buildStatusHistoryRecords(t *testing.T) {
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			if got := buildStatusHistoryRecords(tt.args.logs, "jira:JiraBoard:1:1"); !reflect.DeepEqual(got, tt.want) {
+			if got := buildStatusHistoryRecords(tt.args.logs); !reflect.DeepEqual(got, tt.want) {
 				t.Errorf("buildStatusHistoryRecords() = %v, want %v", got, tt.want)
 			}
 		})

backend/plugins/issue_trace/tasks/task_data.go

@@ -19,14 +19,14 @@ package tasks
 
 // Options original parameter from bp (or pipeline)
 type Options struct {
-	Plugin       string `json:"plugin"`       // jira
-	ConnectionId uint64 `json:"connectionId"` // 1
-	BoardId      uint64 `json:"boardId"`      // 68
-	LakeBoardId  string `json:"lakeBoardId"`  // jira:JiraBoard:1:68
+	Plugin      string   `json:"plugin"`   // jira
+	ScopeIds    []string `json:"scopeIds"` // 68
+	ProjectName string   `json:"projectName"`
 }
 
 // TaskData converted parameter
 type TaskData struct {
-	Options Options
-	BoardId string // jira:1:JiraBoard:68
+	Options     Options
+	ScopeIds    []string // jira:1:JiraBoard:68
+	ProjectName string
 }

backend/plugins/jira/tasks/issue_changelog_collector.go

@@ -40,7 +40,7 @@ const RAW_CHANGELOG_TABLE = "jira_api_issue_changelogs"
 var CollectIssueChangelogsMeta = plugin.SubTaskMeta{
 	Name:             "collectIssueChangelogs",
 	EntryPoint:       CollectIssueChangelogs,
-	EnabledByDefault: false,
+	EnabledByDefault: true,
 	Description:      "collect Jira Issue change logs, supports both timeFilter and diffSync.",
 	DomainTypes:      []string{plugin.DOMAIN_TYPE_TICKET, plugin.DOMAIN_TYPE_CROSS},
 }

backend/plugins/jira/tasks/issue_changelog_convertor.go

@@ -38,7 +38,7 @@ var validID = regexp.MustCompile(`[0-9]+`)
 var ConvertIssueChangelogsMeta = plugin.SubTaskMeta{
 	Name:             "convertIssueChangelogs",
 	EntryPoint:       ConvertIssueChangelogs,
-	EnabledByDefault: false,
+	EnabledByDefault: true,
 	Description:      "convert Jira Issue change logs",
 	DomainTypes:      []string{plugin.DOMAIN_TYPE_TICKET, plugin.DOMAIN_TYPE_CROSS},
 }