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 ac0c40269 fix(tempo): only collect and extract the team of a team
scope (#9319)
ac0c40269 is described below
commit ac0c40269183dd3f6224ccbcefa91de636751900
Author: Shivam Sharma <[email protected]>
AuthorDate: Fri Nov 6 13:09:42 2026 +0530
fix(tempo): only collect and extract the team of a team scope (#9319)
collect-teams fetched every team of the Tempo instance with
connection-wide raw params, and extract_teams turned each of them into
a TempoTeam scope with ScopeConfigId 0. Since the batch save deletes
rows by raw params first, one team scope's pipeline re-created every
team scope of the connection: teams that were never selected became
scopes and the selected ones lost their scope config.
- Scope the team raw data and extracted rows per team ({ConnectionId,
TeamId}, like TempoTeam.ScopeParams and the worklog collector).
- collect-teams: request only the scope's team with teamIds, and drop
the updatedFrom filter, which GET /4/teams doesn't support, along
with the incremental collector state, so each run replaces the raw
rows instead of appending them again.
- extract_teams: skip other teams and keep the existing scope config.
Closes #9318
Generated-by: Claude Code (Claude Opus 5.5)
Signed-off-by: Shivam Sharma
<[email protected]>
Co-authored-by: Claude Opus 5.5 <[email protected]>
---
.../raw_tables/_raw_tempo_api_teams_team_71.csv | 3 +
.../snapshot_tables/_tool_tempo_teams_scopes.csv | 3 +
.../snapshot_tables/_tool_tempo_teams_team_71.csv | 3 +
backend/plugins/tempo/e2e/worklog_test.go | 29 +++++++++
backend/plugins/tempo/tasks/team_collector.go | 75 +++++++++++-----------
backend/plugins/tempo/tasks/team_collector_test.go | 45 +++++++++++++
backend/plugins/tempo/tasks/team_extractor.go | 64 ++++++++++++------
backend/plugins/tempo/tasks/team_extractor_test.go | 57 ++++++++++++++++
8 files changed, 222 insertions(+), 57 deletions(-)
diff --git
a/backend/plugins/tempo/e2e/raw_tables/_raw_tempo_api_teams_team_71.csv
b/backend/plugins/tempo/e2e/raw_tables/_raw_tempo_api_teams_team_71.csv
new file mode 100644
index 000000000..9531fce29
--- /dev/null
+++ b/backend/plugins/tempo/e2e/raw_tables/_raw_tempo_api_teams_team_71.csv
@@ -0,0 +1,3 @@
+id,params,data,url,input,created_at
+1,"{""ConnectionId"":1,""TeamId"":71}","{""self"":""https://api.tempo.io/4/teams/17"",""id"":17,""name"":""Abstract
BE Team
(renamed)"",""summary"":"""",""key"":""""}","https://api.tempo.io/4/teams?limit=50&offset=0&teamIds=71","{""self"":""https://api.tempo.io/4/teams/17"",""id"":17,""name"":""Abstract
BE Team (renamed)"",""summary"":"""",""key"":""""}","2024-02-15 10:00:00.000"
+2,"{""ConnectionId"":1,""TeamId"":71}","{""self"":""https://api.tempo.io/4/teams/71"",""id"":71,""name"":""SmartShopper
(RTL+WHS+POS)"",""summary"":""Microservice
Applications"",""key"":""""}","https://api.tempo.io/4/teams?limit=50&offset=0&teamIds=71","{""self"":""https://api.tempo.io/4/teams/71"",""id"":71,""name"":""SmartShopper
(RTL+WHS+POS)"",""summary"":""Microservice
Applications"",""key"":""""}","2024-02-15 10:00:00.000"
diff --git
a/backend/plugins/tempo/e2e/snapshot_tables/_tool_tempo_teams_scopes.csv
b/backend/plugins/tempo/e2e/snapshot_tables/_tool_tempo_teams_scopes.csv
new file mode 100644
index 000000000..248af6cbf
--- /dev/null
+++ b/backend/plugins/tempo/e2e/snapshot_tables/_tool_tempo_teams_scopes.csv
@@ -0,0 +1,3 @@
+"connection_id","team_id","scope_config_id","name","summary","key","_raw_data_params","_raw_data_table","_raw_data_id","_raw_data_remark"
+1,17,2,"Abstract BE
Team","","","{""ConnectionId"":1,""TeamId"":0}","_raw_tempo_api_teams",1,
+1,71,3,"SmartShopper (RTL+WHS)","Microservice
Applications","","{""ConnectionId"":1,""TeamId"":0}","_raw_tempo_api_teams",2,
diff --git
a/backend/plugins/tempo/e2e/snapshot_tables/_tool_tempo_teams_team_71.csv
b/backend/plugins/tempo/e2e/snapshot_tables/_tool_tempo_teams_team_71.csv
new file mode 100644
index 000000000..6c716bcb1
--- /dev/null
+++ b/backend/plugins/tempo/e2e/snapshot_tables/_tool_tempo_teams_team_71.csv
@@ -0,0 +1,3 @@
+"connection_id","team_id","scope_config_id","name","summary","key","_raw_data_params","_raw_data_table","_raw_data_id","_raw_data_remark"
+1,17,2,"Abstract BE
Team","","","{""ConnectionId"":1,""TeamId"":0}","_raw_tempo_api_teams",1,
+1,71,3,"SmartShopper (RTL+WHS+POS)","Microservice
Applications","","{""ConnectionId"":1,""TeamId"":71}","_raw_tempo_api_teams",2,
diff --git a/backend/plugins/tempo/e2e/worklog_test.go
b/backend/plugins/tempo/e2e/worklog_test.go
index 3d2d5a740..f0a29d3f8 100644
--- a/backend/plugins/tempo/e2e/worklog_test.go
+++ b/backend/plugins/tempo/e2e/worklog_test.go
@@ -102,3 +102,32 @@ func TestTeamDataFlow(t *testing.T) {
),
)
}
+
+func TestTeamScopeExtraction(t *testing.T) {
+ var plugin impl.Tempo
+ dataflowTester := e2ehelper.NewDataFlowTester(t, "tempo", plugin)
+
+ taskData := &tasks.TempoTaskData{
+ Options: &tasks.TempoOptions{
+ ConnectionId: 1,
+ TeamId: 71,
+ },
+ }
+
+ // two team scopes with scope configs; only team 71 is extracted
+
dataflowTester.ImportCsvIntoTabler("./snapshot_tables/_tool_tempo_teams_scopes.csv",
&models.TempoTeam{})
+
dataflowTester.ImportCsvIntoRawTable("./raw_tables/_raw_tempo_api_teams_team_71.csv",
"_raw_tempo_api_teams")
+ dataflowTester.Subtask(tasks.ExtractTeamsMeta, taskData)
+ dataflowTester.VerifyTable(
+ models.TempoTeam{},
+ "./snapshot_tables/_tool_tempo_teams_team_71.csv",
+ e2ehelper.ColumnWithRawData(
+ "connection_id",
+ "team_id",
+ "scope_config_id",
+ "name",
+ "summary",
+ "key",
+ ),
+ )
+}
diff --git a/backend/plugins/tempo/tasks/team_collector.go
b/backend/plugins/tempo/tasks/team_collector.go
index 4c6d0a1d0..92944df14 100644
--- a/backend/plugins/tempo/tasks/team_collector.go
+++ b/backend/plugins/tempo/tasks/team_collector.go
@@ -22,7 +22,6 @@ import (
"net/http"
"net/url"
"strconv"
- "time"
"github.com/apache/devlake/core/errors"
"github.com/apache/devlake/core/plugin"
@@ -42,45 +41,18 @@ var CollectTeamsMeta = plugin.SubTaskMeta{
func CollectTeams(taskCtx plugin.SubTaskContext) errors.Error {
data := taskCtx.GetData().(*TempoTaskData)
- apiCollector, err := api.NewStatefulApiCollector(api.RawDataSubTaskArgs{
- Ctx: taskCtx,
- Params: models.TempoApiParams{
- ConnectionId: data.Options.ConnectionId,
- },
- Table: RAW_TEAM_TABLE,
- })
- if err != nil {
- return err
- }
-
- err = apiCollector.InitCollector(api.ApiCollectorArgs{
- RawDataSubTaskArgs: api.RawDataSubTaskArgs{
- Ctx: taskCtx,
- Params: models.TempoApiParams{
- ConnectionId: data.Options.ConnectionId,
- },
- Table: RAW_TEAM_TABLE,
- },
- ApiClient: data.ApiClient,
- UrlTemplate: "teams",
- PageSize: 50,
+ // Not a stateful (incremental) collector: GET /4/teams has no
+ // "updatedFrom" filter, so an incremental run would append the same
+ // teams to the raw table again instead of refreshing them.
+ apiCollector, err := api.NewApiCollector(api.ApiCollectorArgs{
+ RawDataSubTaskArgs: teamRawDataSubTaskArgs(taskCtx,
data.Options),
+ ApiClient: data.ApiClient,
+ UrlTemplate: "teams",
+ PageSize: 50,
// No GetTotalPages: Tempo v4 pagination metadata
(PageableMetadata)
// has 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: 50}
- }
- query.Set("offset", strconv.Itoa(pager.Skip))
- query.Set("limit", strconv.Itoa(pager.Size))
-
- if apiCollector.IsIncremental() &&
apiCollector.GetSince() != nil {
- since := apiCollector.GetSince()
- query.Set("updatedFrom",
since.Format(time.RFC3339))
- }
-
- return query, nil
+ return buildTeamQuery(data.Options, reqData.Pager), nil
},
ResponseParser: func(res *http.Response) ([]json.RawMessage,
errors.Error) {
var response struct {
@@ -100,3 +72,32 @@ func CollectTeams(taskCtx plugin.SubTaskContext)
errors.Error {
return apiCollector.Execute()
}
+
+// teamRawDataSubTaskArgs scopes the raw team data and the extracted teams per
+// team, like the scope itself (TempoTeam.ScopeParams). The extracted rows are
+// the team scopes: with connection-wide params, extracting one team scope
+// deleted and re-created every team scope of the connection.
+func teamRawDataSubTaskArgs(taskCtx plugin.SubTaskContext, opts *TempoOptions)
api.RawDataSubTaskArgs {
+ return api.RawDataSubTaskArgs{
+ Ctx: taskCtx,
+ Params: models.TempoApiParams{
+ ConnectionId: opts.ConnectionId,
+ TeamId: opts.TeamId,
+ },
+ Table: RAW_TEAM_TABLE,
+ }
+}
+
+func buildTeamQuery(opts *TempoOptions, pager *api.Pager) url.Values {
+ if pager == nil {
+ pager = &api.Pager{Page: 1, Skip: 0, Size: 50}
+ }
+ query := url.Values{}
+ query.Set("offset", strconv.Itoa(pager.Skip))
+ query.Set("limit", strconv.Itoa(pager.Size))
+ if opts.TeamId != 0 {
+ // only the team of this scope, not every team of the Tempo
instance
+ query.Set("teamIds", strconv.FormatInt(opts.TeamId, 10))
+ }
+ return query
+}
diff --git a/backend/plugins/tempo/tasks/team_collector_test.go
b/backend/plugins/tempo/tasks/team_collector_test.go
new file mode 100644
index 000000000..4a9416650
--- /dev/null
+++ b/backend/plugins/tempo/tasks/team_collector_test.go
@@ -0,0 +1,45 @@
+/*
+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"
+
+ "github.com/apache/devlake/helpers/pluginhelper/api"
+ "github.com/stretchr/testify/assert"
+)
+
+func TestBuildTeamQuery(t *testing.T) {
+ pager := &api.Pager{Page: 2, Skip: 50, Size: 50}
+
+ t.Run("a team scope only fetches its own team", func(t *testing.T) {
+ q := buildTeamQuery(&TempoOptions{ConnectionId: 1, TeamId: 71},
pager)
+ assert.Equal(t, "71", q.Get("teamIds"))
+ assert.Equal(t, "50", q.Get("offset"))
+ assert.Equal(t, "50", q.Get("limit"))
+ // not a GET /4/teams parameter
+ assert.Empty(t, q.Get("updatedFrom"))
+ })
+
+ t.Run("without a team scope, fetches every team", func(t *testing.T) {
+ q := buildTeamQuery(&TempoOptions{ConnectionId: 1}, nil)
+ assert.Empty(t, q.Get("teamIds"))
+ assert.Equal(t, "0", q.Get("offset"))
+ assert.Equal(t, "50", q.Get("limit"))
+ })
+}
diff --git a/backend/plugins/tempo/tasks/team_extractor.go
b/backend/plugins/tempo/tasks/team_extractor.go
index 502100a88..aa5e50042 100644
--- a/backend/plugins/tempo/tasks/team_extractor.go
+++ b/backend/plugins/tempo/tasks/team_extractor.go
@@ -20,6 +20,7 @@ package tasks
import (
"encoding/json"
+ "github.com/apache/devlake/core/dal"
"github.com/apache/devlake/core/errors"
"github.com/apache/devlake/core/plugin"
"github.com/apache/devlake/helpers/pluginhelper/api"
@@ -46,30 +47,29 @@ type TempoTeamResponse struct {
func ExtractTeams(taskCtx plugin.SubTaskContext) errors.Error {
data := taskCtx.GetData().(*TempoTaskData)
+ // The extracted teams are the team scopes themselves, which the
extractor
+ // re-creates: keep the scope config users assigned to them.
+ clauses := []dal.Clause{dal.Where("connection_id = ?",
data.Options.ConnectionId)}
+ if data.Options.TeamId != 0 {
+ clauses = append(clauses, dal.Where("team_id = ?",
data.Options.TeamId))
+ }
+ var existingTeams []models.TempoTeam
+ err := taskCtx.GetDal().All(&existingTeams, clauses...)
+ if err != nil {
+ return err
+ }
+ scopeConfigIds := make(map[int64]uint64, len(existingTeams))
+ for _, team := range existingTeams {
+ scopeConfigIds[team.TeamId] = team.ScopeConfigId
+ }
+
extractor, err := api.NewApiExtractor(api.ApiExtractorArgs{
- RawDataSubTaskArgs: api.RawDataSubTaskArgs{
- Ctx: taskCtx,
- Params: models.TempoApiParams{
- ConnectionId: data.Options.ConnectionId,
- },
- Table: RAW_TEAM_TABLE,
- },
+ RawDataSubTaskArgs: teamRawDataSubTaskArgs(taskCtx,
data.Options),
Extract: func(row *api.RawData) ([]interface{}, errors.Error) {
- var apiTeam TempoTeamResponse
- err := errors.Convert(json.Unmarshal(row.Data,
&apiTeam))
- if err != nil {
+ team, err := extractTeam(row.Data, data.Options,
scopeConfigIds)
+ if err != nil || team == nil {
return nil, err
}
-
- // Transform to tool layer model
- team := &models.TempoTeam{
- TeamId: apiTeam.Id,
- Key: apiTeam.Key,
- Name: apiTeam.Name,
- Summary: apiTeam.Summary,
- }
- team.ConnectionId = data.Options.ConnectionId
-
return []interface{}{team}, nil
},
})
@@ -80,3 +80,27 @@ func ExtractTeams(taskCtx plugin.SubTaskContext)
errors.Error {
return extractor.Execute()
}
+
+// extractTeam returns nil for a team other than the team of this scope, so
+// that collecting one team scope doesn't add every Tempo team as a scope.
+func extractTeam(data []byte, opts *TempoOptions, scopeConfigIds
map[int64]uint64) (*models.TempoTeam, errors.Error) {
+ var apiTeam TempoTeamResponse
+ err := errors.Convert(json.Unmarshal(data, &apiTeam))
+ if err != nil {
+ return nil, err
+ }
+ if opts.TeamId != 0 && apiTeam.Id != opts.TeamId {
+ return nil, nil
+ }
+
+ // Transform to tool layer model
+ team := &models.TempoTeam{
+ TeamId: apiTeam.Id,
+ Key: apiTeam.Key,
+ Name: apiTeam.Name,
+ Summary: apiTeam.Summary,
+ }
+ team.ConnectionId = opts.ConnectionId
+ team.ScopeConfigId = scopeConfigIds[apiTeam.Id]
+ return team, nil
+}
diff --git a/backend/plugins/tempo/tasks/team_extractor_test.go
b/backend/plugins/tempo/tasks/team_extractor_test.go
new file mode 100644
index 000000000..852a16246
--- /dev/null
+++ b/backend/plugins/tempo/tasks/team_extractor_test.go
@@ -0,0 +1,57 @@
+/*
+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"
+
+ "github.com/stretchr/testify/assert"
+)
+
+func TestExtractTeam(t *testing.T) {
+ raw17 :=
[]byte(`{"self":"https://api.tempo.io/4/teams/17","id":17,"name":"Abstract BE
Team","summary":"","key":""}`)
+ raw71 :=
[]byte(`{"self":"https://api.tempo.io/4/teams/71","id":71,"name":"SmartShopper
(RTL+WHS)","summary":"Microservice Applications","key":""}`)
+ scopeConfigIds := map[int64]uint64{71: 3}
+
+ t.Run("keeps only the team of the scope", func(t *testing.T) {
+ team, err := extractTeam(raw17, &TempoOptions{ConnectionId: 1,
TeamId: 71}, scopeConfigIds)
+ assert.Nil(t, err)
+ assert.Nil(t, team)
+ })
+
+ t.Run("keeps the scope config of the team scope", func(t *testing.T) {
+ team, err := extractTeam(raw71, &TempoOptions{ConnectionId: 1,
TeamId: 71}, scopeConfigIds)
+ assert.Nil(t, err)
+ if assert.NotNil(t, team) {
+ assert.Equal(t, uint64(1), team.ConnectionId)
+ assert.Equal(t, int64(71), team.TeamId)
+ assert.Equal(t, "SmartShopper (RTL+WHS)", team.Name)
+ assert.Equal(t, "Microservice Applications",
team.Summary)
+ assert.Equal(t, uint64(3), team.ScopeConfigId)
+ }
+ })
+
+ t.Run("without a team scope, extracts every team", func(t *testing.T) {
+ team, err := extractTeam(raw17, &TempoOptions{ConnectionId: 1},
scopeConfigIds)
+ assert.Nil(t, err)
+ if assert.NotNil(t, team) {
+ assert.Equal(t, int64(17), team.TeamId)
+ assert.Equal(t, uint64(0), team.ScopeConfigId)
+ }
+ })
+}
