This is an automated email from the ASF dual-hosted git repository.
klesh pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/devlake.git
The following commit(s) were added to refs/heads/main by this push:
new 48da53bb6 fix(tempo): paginate worklog collection, honour the sync
range, and emit ids that join jira (#9164)
48da53bb6 is described below
commit 48da53bb6db9f3e822f01c07186f942c4b443ea7
Author: Pablo BC <[email protected]>
AuthorDate: Thu Sep 24 15:32:17 2026 +0200
fix(tempo): paginate worklog collection, honour the sync range, and emit
ids that join jira (#9164)
* fix(tempo): paginate worklogs and honour the sync time range
The worklog collector computed the page count from metadata.total, which
the Tempo v4 API does not return (PageableMetadata only has count,
offset, limit, next and previous). The page count was therefore always
zero and only the first page (1000 worklogs) was ever collected. Per
team, the collector also ignored the sync policy and always asked for a
rolling window of the last 90 days, so it kept re-fetching the oldest
1000 worklogs of that window and never reached recent ones.
- Drop GetTotalPages so pages are fetched until one comes back short.
- Full sync starts at timeAfter; incremental runs use updatedFrom with
the last successful collection time; explicit fromDate/toDate options
still win. Same bounds for the team and the global endpoints.
- Scope raw data and collector state per team (TeamId in the params, as
TempoTeam.GetParams already declares), so team scopes of the same
connection no longer share one incremental state; the extractor tags
worklogs with their team and the convertor converts one team at a time.
* fix(tempo): emit worklog issue and author ids that join the jira domain
The convertor wrote issue ids as jira:JiraIssues:<conn>:<id> (plural)
while the jira plugin generates jira:JiraIssue:<conn>:<id>, and stored
the author as the bare Atlassian account id instead of the
jira:JiraAccount:<conn>:<id> domain id. Tempo worklogs therefore never
joined issues, accounts or the team/user mapping.
Generate both ids in the jira plugin's format. buildIssueIdMapping is
removed: both of its branches produced the same string, so the extra
scan of _tool_jira_issues had no effect.
---
.../e2e/raw_tables/_raw_tempo_api_worklogs.csv | 3 +
.../_tool_tempo_worklogs_extracted.csv | 3 +
.../tempo/e2e/snapshot_tables/issue_worklogs.csv | 4 +-
backend/plugins/tempo/e2e/worklog_test.go | 30 +++++++
backend/plugins/tempo/tasks/worklog_collector.go | 100 ++++++++++-----------
.../plugins/tempo/tasks/worklog_collector_test.go | 70 +++++++++++++++
backend/plugins/tempo/tasks/worklog_convertor.go | 55 +++---------
backend/plugins/tempo/tasks/worklog_extractor.go | 2 +
8 files changed, 168 insertions(+), 99 deletions(-)
diff --git a/backend/plugins/tempo/e2e/raw_tables/_raw_tempo_api_worklogs.csv
b/backend/plugins/tempo/e2e/raw_tables/_raw_tempo_api_worklogs.csv
new file mode 100644
index 000000000..9897ad35f
--- /dev/null
+++ b/backend/plugins/tempo/e2e/raw_tables/_raw_tempo_api_worklogs.csv
@@ -0,0 +1,3 @@
+id,params,data,url,input,created_at
+1,"{""ConnectionId"":1,""TeamId"":4}","{""tempoWorklogId"":12345,""issue"":{""id"":10001},""timeSpentSeconds"":3600,""billableSeconds"":3600,""startDate"":""2024-01-15"",""startTime"":""09:00:00"",""description"":""Working
on
task"",""author"":{""accountId"":""abc123""},""createdAt"":""2024-01-15T10:00:00Z"",""updatedAt"":""2024-01-15T10:00:00Z""}",https://api.tempo.io/4/worklogs/team/4?from=2024-01-01&limit=1000&offset=0,null,2024-01-17
10:00:00.000
+2,"{""ConnectionId"":1,""TeamId"":4}","{""tempoWorklogId"":12346,""issue"":{""id"":10002},""timeSpentSeconds"":7200,""billableSeconds"":7200,""startDate"":""2024-01-16"",""startTime"":""14:00:00"",""description"":""Internal
meeting"",""author"":{""accountId"":""712020:def-456""},""createdAt"":""2024-01-16T16:00:00Z"",""updatedAt"":""2024-01-16T16:00:00Z""}",https://api.tempo.io/4/worklogs/team/4?from=2024-01-01&limit=1000&offset=0,null,2024-01-17
10:00:00.000
diff --git
a/backend/plugins/tempo/e2e/snapshot_tables/_tool_tempo_worklogs_extracted.csv
b/backend/plugins/tempo/e2e/snapshot_tables/_tool_tempo_worklogs_extracted.csv
new file mode 100644
index 000000000..c61ae2ab1
--- /dev/null
+++
b/backend/plugins/tempo/e2e/snapshot_tables/_tool_tempo_worklogs_extracted.csv
@@ -0,0 +1,3 @@
+"connection_id","tempo_worklog_id","team_id","issue_id","author_account_id","time_spent_seconds","start_date","start_time","_raw_data_params","_raw_data_table","_raw_data_id","_raw_data_remark"
+"1","12345","4","10001","abc123","3600","2024-01-15","09:00:00","{""ConnectionId"":1,""TeamId"":4}","_raw_tempo_api_worklogs","1",""
+"1","12346","4","10002","712020:def-456","7200","2024-01-16","14:00:00","{""ConnectionId"":1,""TeamId"":4}","_raw_tempo_api_worklogs","2",""
diff --git a/backend/plugins/tempo/e2e/snapshot_tables/issue_worklogs.csv
b/backend/plugins/tempo/e2e/snapshot_tables/issue_worklogs.csv
index a100e9640..417b0f122 100644
--- a/backend/plugins/tempo/e2e/snapshot_tables/issue_worklogs.csv
+++ b/backend/plugins/tempo/e2e/snapshot_tables/issue_worklogs.csv
@@ -1,3 +1,3 @@
id,author_id,comment,time_spent_minutes,logged_date,started_date,issue_id,_raw_data_params,_raw_data_table,_raw_data_id,_raw_data_remark
-tempo:TempoWorklog:12345,abc123,Working on
task,60,2024-01-15T10:00:00.000+00:00,2024-01-15T09:00:00.000+00:00,jira:JiraIssues:1:10001,,,0,
-tempo:TempoWorklog:12346,abc123,Internal
meeting,120,2024-01-16T16:00:00.000+00:00,2024-01-16T14:00:00.000+00:00,jira:JiraIssues:1:10002,,,0,
+tempo:TempoWorklog:12345,jira:JiraAccount:1:abc123,Working on
task,60,2024-01-15T10:00:00.000+00:00,2024-01-15T09:00:00.000+00:00,jira:JiraIssue:1:10001,,,0,
+tempo:TempoWorklog:12346,jira:JiraAccount:1:abc123,Internal
meeting,120,2024-01-16T16:00:00.000+00:00,2024-01-16T14:00:00.000+00:00,jira:JiraIssue:1:10002,,,0,
diff --git a/backend/plugins/tempo/e2e/worklog_test.go
b/backend/plugins/tempo/e2e/worklog_test.go
index 1b08af889..3d2d5a740 100644
--- a/backend/plugins/tempo/e2e/worklog_test.go
+++ b/backend/plugins/tempo/e2e/worklog_test.go
@@ -47,6 +47,36 @@ func TestWorklogDataFlow(t *testing.T) {
)
}
+func TestWorklogExtraction(t *testing.T) {
+ var plugin impl.Tempo
+ dataflowTester := e2ehelper.NewDataFlowTester(t, "tempo", plugin)
+
+ taskData := &tasks.TempoTaskData{
+ Options: &tasks.TempoOptions{
+ ConnectionId: 1,
+ TeamId: 4,
+ },
+ }
+
+
dataflowTester.ImportCsvIntoRawTable("./raw_tables/_raw_tempo_api_worklogs.csv",
"_raw_tempo_api_worklogs")
+ dataflowTester.FlushTabler(&models.TempoWorklog{})
+ dataflowTester.Subtask(tasks.ExtractWorklogsMeta, taskData)
+ dataflowTester.VerifyTable(
+ models.TempoWorklog{},
+ "./snapshot_tables/_tool_tempo_worklogs_extracted.csv",
+ e2ehelper.ColumnWithRawData(
+ "connection_id",
+ "tempo_worklog_id",
+ "team_id",
+ "issue_id",
+ "author_account_id",
+ "time_spent_seconds",
+ "start_date",
+ "start_time",
+ ),
+ )
+}
+
func TestTeamDataFlow(t *testing.T) {
var plugin impl.Tempo
dataflowTester := e2ehelper.NewDataFlowTester(t, "tempo", plugin)
diff --git a/backend/plugins/tempo/tasks/worklog_collector.go
b/backend/plugins/tempo/tasks/worklog_collector.go
index 528b0b8cf..0571956ec 100644
--- a/backend/plugins/tempo/tasks/worklog_collector.go
+++ b/backend/plugins/tempo/tasks/worklog_collector.go
@@ -42,13 +42,19 @@ var CollectWorklogsMeta = plugin.SubTaskMeta{
func CollectWorklogs(taskCtx plugin.SubTaskContext) errors.Error {
data := taskCtx.GetData().(*TempoTaskData)
- apiCollector, err := api.NewStatefulApiCollector(api.RawDataSubTaskArgs{
+ // Scope raw data and collector state per team, like the scope itself
+ // (TempoTeam.GetParams): otherwise every team scope of a connection
shares
+ // one collector state, and the second team collected in a pipeline
would
+ // run incrementally from the first team's start time.
+ rawDataSubTaskArgs := api.RawDataSubTaskArgs{
Ctx: taskCtx,
Params: models.TempoApiParams{
ConnectionId: data.Options.ConnectionId,
+ TeamId: data.Options.TeamId,
},
Table: RAW_WORKLOG_TABLE,
- })
+ }
+ apiCollector, err := api.NewStatefulApiCollector(rawDataSubTaskArgs)
if err != nil {
return err
}
@@ -59,60 +65,16 @@ func CollectWorklogs(taskCtx plugin.SubTaskContext)
errors.Error {
}
err = apiCollector.InitCollector(api.ApiCollectorArgs{
- RawDataSubTaskArgs: api.RawDataSubTaskArgs{
- Ctx: taskCtx,
- Params: models.TempoApiParams{
- ConnectionId: data.Options.ConnectionId,
- },
- Table: RAW_WORKLOG_TABLE,
- },
- ApiClient: data.ApiClient,
- UrlTemplate: urlTemplate,
- PageSize: 1000,
- GetTotalPages: func(res *http.Response, args
*api.ApiCollectorArgs) (int, errors.Error) {
- var response struct {
- Metadata struct {
- Count int `json:"count"`
- Limit int `json:"limit"`
- Total int `json:"total"`
- Offset int `json:"offset"`
- } `json:"metadata"`
- }
- if err := api.UnmarshalResponse(res, &response); err !=
nil {
- return 0, err
- }
- totalPages := (response.Metadata.Total + args.PageSize
- 1) / args.PageSize
- return totalPages, nil
- },
+ RawDataSubTaskArgs: rawDataSubTaskArgs,
+ ApiClient: data.ApiClient,
+ UrlTemplate: urlTemplate,
+ PageSize: 1000,
+ // No GetTotalPages: Tempo v4 pagination metadata
(PageableMetadata)
+ // has count/offset/limit/next/previous but no total, so pages
are
+ // fetched until one comes back short.
Query: func(reqData *api.RequestData) (url.Values,
errors.Error) {
- query := url.Values{}
- pager := reqData.Pager
- if pager == nil {
- pager = &api.Pager{Page: 1, Skip: 0, Size: 1000}
- }
- query.Set("offset", strconv.Itoa(pager.Skip))
- query.Set("limit", strconv.Itoa(pager.Size))
-
- if data.Options.TeamId != 0 {
- fromDate := data.Options.FromDate
- toDate := data.Options.ToDate
- if fromDate == "" {
- since := time.Now().AddDate(0, 0, -90)
- fromDate = since.Format("2006-01-02")
- }
- if toDate == "" {
- toDate = time.Now().Format("2006-01-02")
- }
- query.Set("from", fromDate)
- query.Set("to", toDate)
- } else {
- if apiCollector.IsIncremental() &&
apiCollector.GetSince() != nil {
- since := apiCollector.GetSince()
- query.Set("updatedFrom",
since.Format(time.RFC3339))
- }
- }
-
- return query, nil
+ return buildWorklogQuery(data.Options, reqData.Pager,
+ apiCollector.IsIncremental(),
apiCollector.GetSince()), nil
},
ResponseParser: func(res *http.Response) ([]json.RawMessage,
errors.Error) {
var response struct {
@@ -132,3 +94,31 @@ func CollectWorklogs(taskCtx plugin.SubTaskContext)
errors.Error {
return apiCollector.Execute()
}
+
+// buildWorklogQuery builds the query for one page of worklogs. Explicit
+// fromDate/toDate options win; otherwise an incremental run asks for what
+// changed since the last successful collection (updatedFrom) and a full sync
+// starts at the sync policy's timeAfter (from).
+func buildWorklogQuery(opts *TempoOptions, pager *api.Pager, incremental bool,
since *time.Time) url.Values {
+ if pager == nil {
+ pager = &api.Pager{Page: 1, Skip: 0, Size: 1000}
+ }
+ query := url.Values{}
+ query.Set("offset", strconv.Itoa(pager.Skip))
+ query.Set("limit", strconv.Itoa(pager.Size))
+
+ switch {
+ case opts.FromDate != "" || opts.ToDate != "":
+ if opts.FromDate != "" {
+ query.Set("from", opts.FromDate)
+ }
+ if opts.ToDate != "" {
+ query.Set("to", opts.ToDate)
+ }
+ case since != nil && incremental:
+ query.Set("updatedFrom", since.UTC().Format(time.RFC3339))
+ case since != nil:
+ query.Set("from", since.UTC().Format("2006-01-02"))
+ }
+ return query
+}
diff --git a/backend/plugins/tempo/tasks/worklog_collector_test.go
b/backend/plugins/tempo/tasks/worklog_collector_test.go
new file mode 100644
index 000000000..d484d0d88
--- /dev/null
+++ b/backend/plugins/tempo/tasks/worklog_collector_test.go
@@ -0,0 +1,70 @@
+/*
+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 tasks
+
+import (
+ "testing"
+ "time"
+
+ "github.com/apache/devlake/helpers/pluginhelper/api"
+ "github.com/stretchr/testify/assert"
+)
+
+func TestBuildWorklogQuery(t *testing.T) {
+ pager := &api.Pager{Page: 3, Skip: 2000, Size: 1000}
+ timeAfter := time.Date(2026, 1, 28, 0, 0, 0, 0, time.UTC)
+ lastSuccess := time.Date(2026, 9, 22, 0, 0, 13, 0, time.UTC)
+
+ t.Run("pages with offset and limit", func(t *testing.T) {
+ q := buildWorklogQuery(&TempoOptions{TeamId: 4}, pager, false,
nil)
+ assert.Equal(t, "2000", q.Get("offset"))
+ assert.Equal(t, "1000", q.Get("limit"))
+ })
+
+ t.Run("full sync starts at timeAfter, without a rolling window", func(t
*testing.T) {
+ q := buildWorklogQuery(&TempoOptions{TeamId: 4}, pager, false,
&timeAfter)
+ assert.Equal(t, "2026-01-28", q.Get("from"))
+ assert.Empty(t, q.Get("to"))
+ assert.Empty(t, q.Get("updatedFrom"))
+ })
+
+ t.Run("incremental sync fetches what changed since the last success",
func(t *testing.T) {
+ q := buildWorklogQuery(&TempoOptions{TeamId: 4}, pager, true,
&lastSuccess)
+ assert.Equal(t, "2026-09-22T00:00:13Z", q.Get("updatedFrom"))
+ assert.Empty(t, q.Get("from"))
+ })
+
+ t.Run("global endpoint gets the same bounds", func(t *testing.T) {
+ q := buildWorklogQuery(&TempoOptions{}, pager, false,
&timeAfter)
+ assert.Equal(t, "2026-01-28", q.Get("from"))
+ })
+
+ t.Run("explicit fromDate/toDate options win", func(t *testing.T) {
+ opts := &TempoOptions{TeamId: 4, FromDate: "2026-03-01",
ToDate: "2026-03-31"}
+ q := buildWorklogQuery(opts, pager, true, &lastSuccess)
+ assert.Equal(t, "2026-03-01", q.Get("from"))
+ assert.Equal(t, "2026-03-31", q.Get("to"))
+ assert.Empty(t, q.Get("updatedFrom"))
+ })
+
+ t.Run("no bounds when nothing is known", func(t *testing.T) {
+ q := buildWorklogQuery(&TempoOptions{TeamId: 4}, pager, false,
nil)
+ assert.Empty(t, q.Get("from"))
+ assert.Empty(t, q.Get("updatedFrom"))
+ })
+}
diff --git a/backend/plugins/tempo/tasks/worklog_convertor.go
b/backend/plugins/tempo/tasks/worklog_convertor.go
index b5c62aa7b..bd6c4f255 100644
--- a/backend/plugins/tempo/tasks/worklog_convertor.go
+++ b/backend/plugins/tempo/tasks/worklog_convertor.go
@@ -47,16 +47,14 @@ func ConvertWorklogs(taskCtx plugin.SubTaskContext)
errors.Error {
logger.Info("converting Tempo worklogs to domain layer")
- issueIdMapping, err := buildIssueIdMapping(db, connectionId)
- if err != nil {
- return errors.Default.Wrap(err, "failed to build issue ID
mapping")
- }
-
clauses := []dal.Clause{
dal.Select("*"),
dal.From("_tool_tempo_worklogs"),
dal.Where("connection_id = ?", connectionId),
}
+ if data.Options.TeamId != 0 {
+ clauses = append(clauses, dal.Where("team_id = ?",
data.Options.TeamId))
+ }
cursor, err := db.Cursor(clauses...)
if err != nil {
return errors.Default.Wrap(err, "failed to query Tempo
worklogs")
@@ -68,6 +66,7 @@ func ConvertWorklogs(taskCtx plugin.SubTaskContext)
errors.Error {
Ctx: taskCtx,
Params: models.TempoApiParams{
ConnectionId: connectionId,
+ TeamId: data.Options.TeamId,
},
Table: RAW_WORKLOG_TABLE,
},
@@ -76,11 +75,14 @@ func ConvertWorklogs(taskCtx plugin.SubTaskContext)
errors.Error {
Convert: func(inputRow interface{}) ([]interface{},
errors.Error) {
tempoWorklog := inputRow.(*models.TempoWorklog)
- domainIssueId := ""
- if domainId, ok :=
issueIdMapping[tempoWorklog.IssueId]; ok {
- domainIssueId = domainId
- } else {
- domainIssueId =
fmt.Sprintf("jira:JiraIssues:%d:%d", connectionId, tempoWorklog.IssueId)
+ // Must match the domain ids the jira plugin generates
with didgen
+ // for JiraIssue / JiraAccount (plugins may not import
each other),
+ // so worklogs join issues and accounts. Like before,
this assumes
+ // the Tempo connection id equals the Jira connection
id.
+ domainIssueId := fmt.Sprintf("jira:JiraIssue:%d:%d",
connectionId, tempoWorklog.IssueId)
+ authorId := ""
+ if tempoWorklog.AuthorAccountId != "" {
+ authorId =
fmt.Sprintf("jira:JiraAccount:%d:%s", connectionId,
tempoWorklog.AuthorAccountId)
}
domainWorklogId := fmt.Sprintf("tempo:TempoWorklog:%d",
tempoWorklog.TempoWorklogId)
@@ -105,7 +107,7 @@ func ConvertWorklogs(taskCtx plugin.SubTaskContext)
errors.Error {
Id: domainWorklogId,
},
IssueId: domainIssueId,
- AuthorId: tempoWorklog.AuthorAccountId,
+ AuthorId: authorId,
TimeSpentMinutes: timeSpentMinutes,
StartedDate: startedDate,
LoggedDate: loggedDate,
@@ -122,34 +124,3 @@ func ConvertWorklogs(taskCtx plugin.SubTaskContext)
errors.Error {
return converter.Execute()
}
-
-func buildIssueIdMapping(db dal.Dal, connectionId uint64) (map[int64]string,
errors.Error) {
- mapping := make(map[int64]string)
-
- if !db.HasTable("_tool_jira_issues") {
- return mapping, nil
- }
-
- clauses := []dal.Clause{
- dal.Select("issue_id"),
- dal.From("_tool_jira_issues"),
- dal.Where("connection_id = ?", connectionId),
- }
-
- rows, err := db.Cursor(clauses...)
- if err != nil {
- return nil, errors.Default.Wrap(err, "failed to query jira
issues")
- }
- defer rows.Close()
-
- for rows.Next() {
- var issueId uint64
- if err := rows.Scan(&issueId); err != nil {
- return nil, errors.Default.Wrap(err, "failed to scan
issue")
- }
- domainId := fmt.Sprintf("jira:JiraIssues:%d:%d", connectionId,
issueId)
- mapping[int64(issueId)] = domainId
- }
-
- return mapping, nil
-}
diff --git a/backend/plugins/tempo/tasks/worklog_extractor.go
b/backend/plugins/tempo/tasks/worklog_extractor.go
index 19b82c7f4..b8a0427d4 100644
--- a/backend/plugins/tempo/tasks/worklog_extractor.go
+++ b/backend/plugins/tempo/tasks/worklog_extractor.go
@@ -61,6 +61,7 @@ func ExtractWorklogs(taskCtx plugin.SubTaskContext)
errors.Error {
Ctx: taskCtx,
Params: models.TempoApiParams{
ConnectionId: data.Options.ConnectionId,
+ TeamId: data.Options.TeamId,
},
Table: RAW_WORKLOG_TABLE,
},
@@ -74,6 +75,7 @@ func ExtractWorklogs(taskCtx plugin.SubTaskContext)
errors.Error {
worklog := &models.TempoWorklog{
ConnectionId: data.Options.ConnectionId,
TempoWorklogId: apiWorklog.TempoWorklogId,
+ TeamId: data.Options.TeamId,
IssueId: apiWorklog.Issue.Id,
TimeSpentSeconds: apiWorklog.TimeSpentSeconds,
BillableSeconds: apiWorklog.BillableSeconds,
