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 2024/04/26 09:53:21 UTC
(incubator-devlake) branch kw-7385-jira-changlog-converter-incrmode created (now 33bd1a403)
This is an automated email from the ASF dual-hosted git repository.
klesh pushed a change to branch kw-7385-jira-changlog-converter-incrmode
in repository https://gitbox.apache.org/repos/asf/incubator-devlake.git
at 33bd1a403 feat: jira issue / changelog converter supports incremental mode
This branch includes the following new commits:
new 33bd1a403 feat: jira issue / changelog converter supports incremental mode
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(incubator-devlake) 01/01: feat: jira issue / changelog converter supports incremental mode
Posted by kl...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
klesh pushed a commit to branch kw-7385-jira-changlog-converter-incrmode
in repository https://gitbox.apache.org/repos/asf/incubator-devlake.git
commit 33bd1a40333ac23974b5ed02c28f3cfb3b33b399
Author: Klesh Wong <zh...@merico.dev>
AuthorDate: Fri Apr 26 17:52:49 2024 +0800
feat: jira issue / changelog converter supports incremental mode
---
.../pluginhelper/api/api_extractor_stateful.go | 13 +--
.../pluginhelper/api/data_convertor_stateful.go | 129 +++++++++++++++++++++
.../pluginhelper/api/subtask_state_manager.go | 26 ++++-
.../jira/tasks/issue_changelog_convertor.go | 74 ++++++------
.../jira/tasks/issue_changelog_extractor.go | 7 +-
backend/plugins/jira/tasks/issue_convertor.go | 61 +++++-----
backend/plugins/jira/tasks/issue_extractor.go | 9 +-
7 files changed, 232 insertions(+), 87 deletions(-)
diff --git a/backend/helpers/pluginhelper/api/api_extractor_stateful.go b/backend/helpers/pluginhelper/api/api_extractor_stateful.go
index 42fcda261..c714d0c00 100644
--- a/backend/helpers/pluginhelper/api/api_extractor_stateful.go
+++ b/backend/helpers/pluginhelper/api/api_extractor_stateful.go
@@ -18,7 +18,6 @@ limitations under the License.
package api
import (
- "fmt"
"reflect"
"github.com/apache/incubator-devlake/core/dal"
@@ -30,7 +29,6 @@ import (
// StatefulApiExtractorArgs is a struct that contains the arguments for a stateful api extractor
type StatefulApiExtractorArgs struct {
*SubtaskCommonArgs
- Table string // subtask normally reads from a table
Extract func(row *RawData) ([]any, errors.Error)
}
@@ -60,13 +58,14 @@ func (extractor *StatefulApiExtractor) Execute() errors.Error {
// load data from database
db := extractor.GetDal()
logger := extractor.GetLogger()
- table := fmt.Sprintf("_raw_%s", extractor.Table)
+ table := extractor.GetRawDataTable()
+ params := extractor.GetRawDataParams()
if !db.HasTable(table) {
return nil
}
clauses := []dal.Clause{
dal.From(table),
- dal.Where("params = ?", extractor.Params),
+ dal.Where("params = ?", params),
dal.Orderby("id ASC"),
}
@@ -86,10 +85,10 @@ func (extractor *StatefulApiExtractor) Execute() errors.Error {
if err != nil {
return errors.Default.Wrap(err, "error running DB query")
}
- logger.Info("get data from %s where params=%s and got %d", table, extractor.Params, count)
+ logger.Info("get data from %s where params=%s and got %d", table, params, count)
defer cursor.Close()
// batch save divider
- divider := NewBatchSaveDivider(extractor.SubTaskContext, extractor.BatchSize, table, extractor.Params)
+ divider := NewBatchSaveDivider(extractor.SubTaskContext, extractor.BatchSize, table, params)
divider.SetIncrementalMode(extractor.IsIncremental())
// progress
@@ -121,8 +120,8 @@ func (extractor *StatefulApiExtractor) Execute() errors.Error {
// set raw data origin field
setRawDataOrigin(result, common.RawDataOrigin{
RawDataTable: table,
+ RawDataParams: params,
RawDataId: row.ID,
- RawDataParams: row.Params,
})
// records get saved into db when slots were max outed
err = batch.Add(result)
diff --git a/backend/helpers/pluginhelper/api/data_convertor_stateful.go b/backend/helpers/pluginhelper/api/data_convertor_stateful.go
new file mode 100644
index 000000000..ddb710a2f
--- /dev/null
+++ b/backend/helpers/pluginhelper/api/data_convertor_stateful.go
@@ -0,0 +1,129 @@
+/*
+Licensed to the Apache Software Foundation (ASF) under one or more
+contributor license agreements. See the NOTICE file distributed with
+this work for additional information regarding copyright ownership.
+The ASF licenses this file to You under the Apache License, Version 2.0
+(the "License"); you may not use this file except in compliance with
+the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package api
+
+import (
+ "reflect"
+
+ "github.com/apache/incubator-devlake/core/dal"
+ "github.com/apache/incubator-devlake/core/errors"
+ plugin "github.com/apache/incubator-devlake/core/plugin"
+)
+
+type StatefulDataConverterArgs[InputType any] struct {
+ *SubtaskCommonArgs
+ Input func(*SubtaskStateManager) (dal.Rows, errors.Error)
+ Convert func(row *InputType) ([]any, errors.Error)
+ BatchSize int
+}
+
+type StatefulDataConverter[InputType any] struct {
+ *StatefulDataConverterArgs[InputType]
+ *SubtaskStateManager
+}
+
+func NewStatefulDataConverter[
+ OptType any,
+ InputType any,
+](
+ args *StatefulDataConverterArgs[InputType],
+) (*StatefulDataConverter[InputType], errors.Error) {
+ // process args
+ if args.BatchSize == 0 {
+ args.BatchSize = 500
+ }
+ stateManager, err := NewSubtaskStateManager(args.SubtaskCommonArgs)
+ if err != nil {
+ return nil, err
+ }
+ return &StatefulDataConverter[InputType]{
+ StatefulDataConverterArgs: args,
+ SubtaskStateManager: stateManager,
+ }, nil
+}
+
+func (converter *StatefulDataConverter[InputType]) Execute() errors.Error {
+ // load data from database
+ db := converter.GetDal()
+
+ table := converter.GetRawDataTable()
+ params := converter.GetRawDataParams()
+
+ // batch save divider
+ RAW_DATA_ORIGIN := "RawDataOrigin"
+ divider := NewBatchSaveDivider(converter, converter.BatchSize, table, params)
+ divider.SetIncrementalMode(converter.IsIncremental())
+
+ // set progress
+ converter.SetProgress(0, -1)
+
+ cursor, err := converter.Input(converter.SubtaskStateManager)
+ if err != nil {
+ return err
+ }
+ defer cursor.Close()
+ ctx := converter.GetContext()
+ // iterate all rows
+ for cursor.Next() {
+ select {
+ case <-ctx.Done():
+ return errors.Convert(ctx.Err())
+ default:
+ }
+ inputRow := new(InputType)
+ err := db.Fetch(cursor, inputRow)
+ if err != nil {
+ return errors.Default.Wrap(err, "error fetching rows")
+ }
+
+ results, err := converter.Convert(inputRow)
+ if err != nil {
+ return errors.Default.Wrap(err, "error calling Converter plugin implementation")
+ }
+
+ for _, result := range results {
+ // get the batch operator for the specific type
+ batch, err := divider.ForType(reflect.TypeOf(result))
+ if err != nil {
+ return errors.Default.Wrap(err, "error getting batch from result")
+ }
+ // set raw data origin field
+ origin := reflect.ValueOf(result).Elem().FieldByName(RAW_DATA_ORIGIN)
+ if origin.IsValid() {
+ origin.Set(reflect.ValueOf(inputRow).Elem().FieldByName(RAW_DATA_ORIGIN))
+ }
+ // records get saved into db when slots were max outed
+ err = batch.Add(result)
+ if err != nil {
+ return errors.Default.Wrap(err, "error adding result to batch")
+ }
+ }
+ converter.IncProgress(1)
+ }
+
+ // save the last batches
+ err = divider.Close()
+ if err != nil {
+ return err
+ }
+ // save the incremantal state
+ return converter.SubtaskStateManager.Close()
+}
+
+// Check if DataConverter implements SubTask interface
+var _ plugin.SubTask = (*StatefulDataConverter[any])(nil)
diff --git a/backend/helpers/pluginhelper/api/subtask_state_manager.go b/backend/helpers/pluginhelper/api/subtask_state_manager.go
index 80d832cb7..82c62f19c 100644
--- a/backend/helpers/pluginhelper/api/subtask_state_manager.go
+++ b/backend/helpers/pluginhelper/api/subtask_state_manager.go
@@ -18,22 +18,37 @@ limitations under the License.
package api
import (
+ "fmt"
"time"
"github.com/apache/incubator-devlake/core/dal"
"github.com/apache/incubator-devlake/core/errors"
"github.com/apache/incubator-devlake/core/models"
plugin "github.com/apache/incubator-devlake/core/plugin"
+ "github.com/apache/incubator-devlake/core/utils"
)
// SubtaskCommonArgs is a struct that contains the common arguments for a subtask
type SubtaskCommonArgs struct {
plugin.SubTaskContext
- Params string // for filtering rows belonging to the scope (jira board, github repo) of the subtask
- SubtaskConfig string // for determining whether the subtask should run in incremental or full sync mode
+ Table string // raw table name
+ Params any // for filtering rows belonging to the scope (jira board, github repo) of the subtask
+ SubtaskConfig any // for determining whether the subtask should run in incremental or full sync mode
BatchSize int // batch size for saving data
}
+func (args *SubtaskCommonArgs) GetRawDataTable() string {
+ return fmt.Sprintf("_raw_%s", args.Table)
+}
+
+func (args *SubtaskCommonArgs) GetRawDataParams() string {
+ return utils.ToJsonString(args.Params)
+}
+
+func (args *SubtaskCommonArgs) GetSubtaskConfig() string {
+ return utils.ToJsonString(args.SubtaskConfig)
+}
+
// SubtaskStateManager manages the state of a subtask. It is used to determine whether
// the subtask should run in incremental mode or full sync mode and what time range to collect.
type SubtaskStateManager struct {
@@ -56,15 +71,16 @@ func NewSubtaskStateManager(args *SubtaskCommonArgs) (stateManager *SubtaskState
if syncPolicy == nil {
syncPolicy = &models.SyncPolicy{}
}
+ params := args.GetRawDataParams()
// load the previous state from the database
state := &models.SubtaskState{}
- err = db.First(state, dal.Where(`plugin = ? AND subtask =? AND params = ?`, plugin, subtask, args.Params))
+ err = db.First(state, dal.Where(`plugin = ? AND subtask =? AND params = ?`, plugin, subtask, params))
if err != nil {
if db.IsErrorNotFound(err) {
state = &models.SubtaskState{
Plugin: plugin,
Subtask: subtask,
- Params: args.Params,
+ Params: params,
}
err = nil
} else {
@@ -81,7 +97,7 @@ func NewSubtaskStateManager(args *SubtaskCommonArgs) (stateManager *SubtaskState
isIncremental: false,
since: syncPolicy.TimeAfter,
until: &now,
- config: args.SubtaskConfig,
+ config: utils.ToJsonString(args.SubtaskConfig),
}
// fallback to the previous timeAfter if no new value
if stateManager.since == nil {
diff --git a/backend/plugins/jira/tasks/issue_changelog_convertor.go b/backend/plugins/jira/tasks/issue_changelog_convertor.go
index 0dc8d134b..d1db33467 100644
--- a/backend/plugins/jira/tasks/issue_changelog_convertor.go
+++ b/backend/plugins/jira/tasks/issue_changelog_convertor.go
@@ -18,7 +18,6 @@ limitations under the License.
package tasks
import (
- "reflect"
"regexp"
"strconv"
"strings"
@@ -52,13 +51,12 @@ type IssueChangelogItemResult struct {
Created time.Time
}
-func ConvertIssueChangelogs(taskCtx plugin.SubTaskContext) errors.Error {
- data := taskCtx.GetData().(*JiraTaskData)
+func ConvertIssueChangelogs(subtaskCtx plugin.SubTaskContext) errors.Error {
+ data := subtaskCtx.GetData().(*JiraTaskData)
+ db := subtaskCtx.GetDal()
connectionId := data.Options.ConnectionId
boardId := data.Options.BoardId
- logger := taskCtx.GetLogger()
- db := taskCtx.GetDal()
- logger.Info("covert changelog")
+
var allStatus []models.JiraStatus
err := db.All(&allStatus, dal.Where("connection_id = ?", connectionId))
if err != nil {
@@ -68,43 +66,44 @@ func ConvertIssueChangelogs(taskCtx plugin.SubTaskContext) errors.Error {
for _, v := range allStatus {
statusMap[v.ID] = v
}
- // select all changelogs belongs to the board
- clauses := []dal.Clause{
- dal.Select("_tool_jira_issue_changelog_items.*, _tool_jira_issue_changelogs.issue_id, author_account_id, author_display_name, created"),
- dal.From("_tool_jira_issue_changelog_items"),
- dal.Join(`left join _tool_jira_issue_changelogs on (
- _tool_jira_issue_changelogs.connection_id = _tool_jira_issue_changelog_items.connection_id
- AND _tool_jira_issue_changelogs.changelog_id = _tool_jira_issue_changelog_items.changelog_id
- )`),
- dal.Join(`left join _tool_jira_board_issues on (
- _tool_jira_board_issues.connection_id = _tool_jira_issue_changelogs.connection_id
- AND _tool_jira_board_issues.issue_id = _tool_jira_issue_changelogs.issue_id
- )`),
- dal.Where("_tool_jira_issue_changelog_items.connection_id = ? AND _tool_jira_board_issues.board_id = ?", connectionId, boardId),
- }
- cursor, err := db.Cursor(clauses...)
- if err != nil {
- logger.Error(err, "")
- return err
- }
- defer cursor.Close()
+
issueIdGenerator := didgen.NewDomainIdGenerator(&models.JiraIssue{})
sprintIdGenerator := didgen.NewDomainIdGenerator(&models.JiraSprint{})
changelogIdGenerator := didgen.NewDomainIdGenerator(&models.JiraIssueChangelogItems{})
accountIdGen := didgen.NewDomainIdGenerator(&models.JiraAccount{})
- converter, err := api.NewDataConverter(api.DataConverterArgs{
- RawDataSubTaskArgs: api.RawDataSubTaskArgs{
- Ctx: taskCtx,
+
+ converter, err := api.NewStatefulDataConverter[IssueChangelogItemResult](&api.StatefulDataConverterArgs[IssueChangelogItemResult]{
+ SubtaskCommonArgs: &api.SubtaskCommonArgs{
+ SubTaskContext: subtaskCtx,
+ Table: RAW_ISSUE_TABLE,
Params: JiraApiParams{
- ConnectionId: connectionId,
- BoardId: boardId,
+ ConnectionId: data.Options.ConnectionId,
+ BoardId: data.Options.BoardId,
},
- Table: RAW_CHANGELOG_TABLE,
},
- InputRowType: reflect.TypeOf(IssueChangelogItemResult{}),
- Input: cursor,
- Convert: func(inputRow interface{}) ([]interface{}, errors.Error) {
- row := inputRow.(*IssueChangelogItemResult)
+ Input: func(stateManager *api.SubtaskStateManager) (dal.Rows, errors.Error) {
+ clauses := []dal.Clause{
+ dal.Select("_tool_jira_issue_changelog_items.*, _tool_jira_issue_changelogs.issue_id, author_account_id, author_display_name, created"),
+ dal.From("_tool_jira_issue_changelog_items"),
+ dal.Join(`left join _tool_jira_issue_changelogs on (
+ _tool_jira_issue_changelogs.connection_id = _tool_jira_issue_changelog_items.connection_id
+ AND _tool_jira_issue_changelogs.changelog_id = _tool_jira_issue_changelog_items.changelog_id
+ )`),
+ dal.Join(`left join _tool_jira_board_issues on (
+ _tool_jira_board_issues.connection_id = _tool_jira_issue_changelogs.connection_id
+ AND _tool_jira_board_issues.issue_id = _tool_jira_issue_changelogs.issue_id
+ )`),
+ dal.Where("_tool_jira_issue_changelog_items.connection_id = ? AND _tool_jira_board_issues.board_id = ?", connectionId, boardId),
+ }
+ if stateManager.IsIncremental() {
+ since := stateManager.GetSince()
+ if since != nil {
+ clauses = append(clauses, dal.Where("_tool_jira_issue_changelog_items.created_at >= ? ", since))
+ }
+ }
+ return db.Cursor(clauses...)
+ },
+ Convert: func(row *IssueChangelogItemResult) ([]interface{}, errors.Error) {
changelog := &ticket.IssueChangelogs{
DomainEntity: domainlayer.DomainEntity{Id: changelogIdGenerator.Generate(
row.ConnectionId,
@@ -149,10 +148,11 @@ func ConvertIssueChangelogs(taskCtx plugin.SubTaskContext) errors.Error {
}
}
return []interface{}{changelog}, nil
+
},
})
+
if err != nil {
- logger.Info(err.Error())
return err
}
diff --git a/backend/plugins/jira/tasks/issue_changelog_extractor.go b/backend/plugins/jira/tasks/issue_changelog_extractor.go
index 3d0cbe5f6..66d9247a6 100644
--- a/backend/plugins/jira/tasks/issue_changelog_extractor.go
+++ b/backend/plugins/jira/tasks/issue_changelog_extractor.go
@@ -22,7 +22,6 @@ import (
"github.com/apache/incubator-devlake/core/errors"
"github.com/apache/incubator-devlake/core/plugin"
- "github.com/apache/incubator-devlake/core/utils"
"github.com/apache/incubator-devlake/helpers/pluginhelper/api"
"github.com/apache/incubator-devlake/plugins/jira/models"
"github.com/apache/incubator-devlake/plugins/jira/tasks/apiv2models"
@@ -47,12 +46,12 @@ func ExtractIssueChangelogs(subtaskCtx plugin.SubTaskContext) errors.Error {
extractor, err := api.NewStatefulApiExtractor(&api.StatefulApiExtractorArgs{
SubtaskCommonArgs: &api.SubtaskCommonArgs{
SubTaskContext: subtaskCtx,
- Params: utils.ToJsonString(JiraApiParams{
+ Table: RAW_CHANGELOG_TABLE,
+ Params: JiraApiParams{
ConnectionId: data.Options.ConnectionId,
BoardId: data.Options.BoardId,
- }),
+ },
},
- Table: RAW_CHANGELOG_TABLE,
Extract: func(row *api.RawData) ([]interface{}, errors.Error) {
// process input
var input apiv2models.Input
diff --git a/backend/plugins/jira/tasks/issue_convertor.go b/backend/plugins/jira/tasks/issue_convertor.go
index e82ae3995..f30037182 100644
--- a/backend/plugins/jira/tasks/issue_convertor.go
+++ b/backend/plugins/jira/tasks/issue_convertor.go
@@ -20,7 +20,6 @@ package tasks
import (
"net/url"
"path/filepath"
- "reflect"
"strings"
"github.com/apache/incubator-devlake/core/dal"
@@ -41,48 +40,51 @@ var ConvertIssuesMeta = plugin.SubTaskMeta{
DomainTypes: []string{plugin.DOMAIN_TYPE_TICKET},
}
-func ConvertIssues(taskCtx plugin.SubTaskContext) errors.Error {
- db := taskCtx.GetDal()
- data := taskCtx.GetData().(*JiraTaskData)
-
- jiraIssue := &models.JiraIssue{}
- // select all issues belongs to the board
- clauses := []dal.Clause{
- dal.Select("_tool_jira_issues.*"),
- dal.From(jiraIssue),
- dal.Join(`left join _tool_jira_board_issues
- on _tool_jira_board_issues.issue_id = _tool_jira_issues.issue_id
- and _tool_jira_board_issues.connection_id = _tool_jira_issues.connection_id`),
- dal.Where(
- "_tool_jira_board_issues.connection_id = ? AND _tool_jira_board_issues.board_id = ?",
- data.Options.ConnectionId,
- data.Options.BoardId,
- ),
- }
- cursor, err := db.Cursor(clauses...)
+func ConvertIssues(subtaskCtx plugin.SubTaskContext) errors.Error {
+ data := subtaskCtx.GetData().(*JiraTaskData)
+ db := subtaskCtx.GetDal()
+ mappings, err := getTypeMappings(data, db)
if err != nil {
return err
}
- defer cursor.Close()
issueIdGen := didgen.NewDomainIdGenerator(&models.JiraIssue{})
accountIdGen := didgen.NewDomainIdGenerator(&models.JiraAccount{})
boardIdGen := didgen.NewDomainIdGenerator(&models.JiraBoard{})
boardId := boardIdGen.Generate(data.Options.ConnectionId, data.Options.BoardId)
- converter, err := api.NewDataConverter(api.DataConverterArgs{
- InputRowType: reflect.TypeOf(models.JiraIssue{}),
- Input: cursor,
- RawDataSubTaskArgs: api.RawDataSubTaskArgs{
- Ctx: taskCtx,
+ converter, err := api.NewStatefulDataConverter[models.JiraIssue](&api.StatefulDataConverterArgs[models.JiraIssue]{
+ SubtaskCommonArgs: &api.SubtaskCommonArgs{
+ SubTaskContext: subtaskCtx,
+ Table: RAW_ISSUE_TABLE,
Params: JiraApiParams{
ConnectionId: data.Options.ConnectionId,
BoardId: data.Options.BoardId,
},
- Table: RAW_ISSUE_TABLE,
+ SubtaskConfig: mappings,
},
- Convert: func(inputRow interface{}) ([]interface{}, errors.Error) {
- jiraIssue := inputRow.(*models.JiraIssue)
+ Input: func(stateManager *api.SubtaskStateManager) (dal.Rows, errors.Error) {
+ clauses := []dal.Clause{
+ dal.Select("_tool_jira_issues.*"),
+ dal.From("_tool_jira_issues"),
+ dal.Join(`left join _tool_jira_board_issues
+ on _tool_jira_board_issues.issue_id = _tool_jira_issues.issue_id
+ and _tool_jira_board_issues.connection_id = _tool_jira_issues.connection_id`),
+ dal.Where(
+ "_tool_jira_board_issues.connection_id = ? AND _tool_jira_board_issues.board_id = ?",
+ data.Options.ConnectionId,
+ data.Options.BoardId,
+ ),
+ }
+ if stateManager.IsIncremental() {
+ since := stateManager.GetSince()
+ if since != nil {
+ clauses = append(clauses, dal.Where("_tool_jira_issues.updated_at >= ? ", since))
+ }
+ }
+ return db.Cursor(clauses...)
+ },
+ Convert: func(jiraIssue *models.JiraIssue) ([]interface{}, errors.Error) {
issue := &ticket.Issue{
DomainEntity: domainlayer.DomainEntity{
Id: issueIdGen.Generate(jiraIssue.ConnectionId, jiraIssue.IssueId),
@@ -139,6 +141,7 @@ func ConvertIssues(taskCtx plugin.SubTaskContext) errors.Error {
return result, nil
},
})
+
if err != nil {
return err
}
diff --git a/backend/plugins/jira/tasks/issue_extractor.go b/backend/plugins/jira/tasks/issue_extractor.go
index a76acfabb..bac817402 100644
--- a/backend/plugins/jira/tasks/issue_extractor.go
+++ b/backend/plugins/jira/tasks/issue_extractor.go
@@ -26,7 +26,6 @@ import (
"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/core/utils"
"github.com/apache/incubator-devlake/helpers/pluginhelper/api"
"github.com/apache/incubator-devlake/plugins/jira/models"
"github.com/apache/incubator-devlake/plugins/jira/tasks/apiv2models"
@@ -62,13 +61,13 @@ func ExtractIssues(subtaskCtx plugin.SubTaskContext) errors.Error {
extractor, err := api.NewStatefulApiExtractor(&api.StatefulApiExtractorArgs{
SubtaskCommonArgs: &api.SubtaskCommonArgs{
SubTaskContext: subtaskCtx,
- Params: utils.ToJsonString(JiraApiParams{
+ Table: RAW_ISSUE_TABLE,
+ Params: JiraApiParams{
ConnectionId: data.Options.ConnectionId,
BoardId: data.Options.BoardId,
- }),
- SubtaskConfig: utils.ToJsonString(mappings),
+ },
+ SubtaskConfig: mappings,
},
- Table: RAW_ISSUE_TABLE,
Extract: func(row *api.RawData) ([]interface{}, errors.Error) {
return extractIssues(data, mappings, row)
},