From 7f3485a14d4b837916d242207ee4aabbe5cc010f Mon Sep 17 00:00:00 2001 From: Pablo BC Date: Wed, 23 Sep 2026 13:30:40 +0200 Subject: [PATCH 1/2] 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. --- .../raw_tables/_raw_tempo_api_worklogs.csv | 3 + .../_tool_tempo_worklogs_extracted.csv | 3 + backend/plugins/tempo/e2e/worklog_test.go | 30 ++++++ .../plugins/tempo/tasks/worklog_collector.go | 100 ++++++++---------- .../tempo/tasks/worklog_collector_test.go | 70 ++++++++++++ .../plugins/tempo/tasks/worklog_convertor.go | 4 + .../plugins/tempo/tasks/worklog_extractor.go | 2 + 7 files changed, 157 insertions(+), 55 deletions(-) create mode 100644 backend/plugins/tempo/e2e/raw_tables/_raw_tempo_api_worklogs.csv create mode 100644 backend/plugins/tempo/e2e/snapshot_tables/_tool_tempo_worklogs_extracted.csv create mode 100644 backend/plugins/tempo/tasks/worklog_collector_test.go 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 00000000000..9897ad35f1d --- /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 00000000000..c61ae2ab16e --- /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/worklog_test.go b/backend/plugins/tempo/e2e/worklog_test.go index 1b08af889d6..3d2d5a74039 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 528b0b8cf44..0571956ec9f 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 00000000000..d484d0d8869 --- /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 b5c62aa7bbf..81b1c9e9331 100644 --- a/backend/plugins/tempo/tasks/worklog_convertor.go +++ b/backend/plugins/tempo/tasks/worklog_convertor.go @@ -57,6 +57,9 @@ func ConvertWorklogs(taskCtx plugin.SubTaskContext) errors.Error { 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 +71,7 @@ func ConvertWorklogs(taskCtx plugin.SubTaskContext) errors.Error { Ctx: taskCtx, Params: models.TempoApiParams{ ConnectionId: connectionId, + TeamId: data.Options.TeamId, }, Table: RAW_WORKLOG_TABLE, }, diff --git a/backend/plugins/tempo/tasks/worklog_extractor.go b/backend/plugins/tempo/tasks/worklog_extractor.go index 19b82c7f4b0..b8a0427d455 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, From 0c2d8d96352420169f39c6f726d02dda612c19ba Mon Sep 17 00:00:00 2001 From: Pablo BC Date: Wed, 23 Sep 2026 13:30:40 +0200 Subject: [PATCH 2/2] fix(tempo): emit worklog issue and author ids that join the jira domain The convertor wrote issue ids as jira:JiraIssues:: (plural) while the jira plugin generates jira:JiraIssue::, and stored the author as the bare Atlassian account id instead of the jira:JiraAccount:: 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/snapshot_tables/issue_worklogs.csv | 4 +- .../plugins/tempo/tasks/worklog_convertor.go | 51 ++++--------------- 2 files changed, 11 insertions(+), 44 deletions(-) diff --git a/backend/plugins/tempo/e2e/snapshot_tables/issue_worklogs.csv b/backend/plugins/tempo/e2e/snapshot_tables/issue_worklogs.csv index a100e9640dd..417b0f122f2 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/tasks/worklog_convertor.go b/backend/plugins/tempo/tasks/worklog_convertor.go index 81b1c9e9331..bd6c4f25554 100644 --- a/backend/plugins/tempo/tasks/worklog_convertor.go +++ b/backend/plugins/tempo/tasks/worklog_convertor.go @@ -47,11 +47,6 @@ 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"), @@ -80,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) @@ -109,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, @@ -126,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 -}