You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@devlake.apache.org by ab...@apache.org on 2024/04/26 06:20:35 UTC
(incubator-devlake) branch main updated: feat: subtask state manager (#7384)
This is an automated email from the ASF dual-hosted git repository.
abeizn pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/incubator-devlake.git
The following commit(s) were added to refs/heads/main by this push:
new afe49e109 feat: subtask state manager (#7384)
afe49e109 is described below
commit afe49e109244d60b8484c2a2956aac5a65f0fcfe
Author: Klesh Wong <zh...@merico.dev>
AuthorDate: Fri Apr 26 14:20:30 2024 +0800
feat: subtask state manager (#7384)
* feat: subtask state manager
* fix: remove commented code
---
.../20240424_add_subtask_states.go | 59 +++++++
backend/core/models/migrationscripts/register.go | 1 +
backend/core/models/subtask_state.go | 40 +++++
backend/core/utils/json.go | 9 +
.../pluginhelper/api/subtask_state_manager.go | 128 ++++++++++++++
.../pluginhelper/api/subtask_state_manager_test.go | 188 +++++++++++++++++++++
6 files changed, 425 insertions(+)
diff --git a/backend/core/models/migrationscripts/20240424_add_subtask_states.go b/backend/core/models/migrationscripts/20240424_add_subtask_states.go
new file mode 100644
index 000000000..c688ddf03
--- /dev/null
+++ b/backend/core/models/migrationscripts/20240424_add_subtask_states.go
@@ -0,0 +1,59 @@
+/*
+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 migrationscripts
+
+import (
+ "time"
+
+ "github.com/apache/incubator-devlake/core/context"
+ "github.com/apache/incubator-devlake/core/errors"
+ "github.com/apache/incubator-devlake/core/plugin"
+)
+
+var _ plugin.MigrationScript = (*addSubtaskStates)(nil)
+
+type subtaskState20240424 struct {
+ CreatedAt time.Time `json:"createdAt"`
+ UpdatedAt time.Time `json:"updatedAt"`
+ Plugin string `gorm:"primaryKey;type:varchar(50)" json:"plugin"`
+ Subtask string `gorm:"primaryKey;type:varchar(50)" json:"subtask"`
+ RawDataTable string `gorm:"primaryKey;column:raw_data_table;type:varchar(50)" json:"raw_data_table"`
+ RawDataParams string `gorm:"primaryKey;column:raw_data_params;type:varchar(255);index" json:"raw_data_params"`
+ TimeAfter *time.Time
+ LatestSuccessStart *time.Time
+}
+
+func (subtaskState20240424) TableName() string {
+ return "_devlake_subtask_states"
+}
+
+type addSubtaskStates struct{}
+
+func (script *addSubtaskStates) Up(basicRes context.BasicRes) errors.Error {
+ db := basicRes.GetDal()
+ errors.Must(db.AutoMigrate(&subtaskState20240424{}))
+ return nil
+}
+
+func (*addSubtaskStates) Version() uint64 {
+ return 20240424152734
+}
+
+func (*addSubtaskStates) Name() string {
+ return "add _devlake_subtask_states"
+}
diff --git a/backend/core/models/migrationscripts/register.go b/backend/core/models/migrationscripts/register.go
index 3c3e2e45f..b79d0bc4c 100644
--- a/backend/core/models/migrationscripts/register.go
+++ b/backend/core/models/migrationscripts/register.go
@@ -112,5 +112,6 @@ func All() []plugin.MigrationScript {
new(addStore),
new(addSubtaskField),
new(addDisplayTitleAndUrl),
+ new(addSubtaskStates),
}
}
diff --git a/backend/core/models/subtask_state.go b/backend/core/models/subtask_state.go
new file mode 100644
index 000000000..79843a64d
--- /dev/null
+++ b/backend/core/models/subtask_state.go
@@ -0,0 +1,40 @@
+/*
+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 models
+
+import (
+ "time"
+)
+
+type SubtaskState struct {
+ Plugin string `gorm:"primaryKey;type:varchar(255)" json:"plugin"`
+ Subtask string `gorm:"primaryKey;type:varchar(255)" json:"subtask"`
+ // Params is a json string to identitfy rows of a specific scope (jira board, github repo)
+ Params string `gorm:"primaryKey;type:varchar(255);index" json:"params"`
+ // PrevConfig stores the previous configuration of the subtask for determining should subtask run in Incremntal or FullSync mode
+ PrevConfig string `json:"prevConfig"`
+ // TimeAfter stores the previous timeAfter specified by the user for determining should subtask run in Incremntal or FullSync mode
+ TimeAfter *time.Time `json:"timeAfter"`
+ PrevStartedAt *time.Time `json:"prevStartedAt"`
+ CreatedAt time.Time `json:"createdAt"`
+ UpdatedAt time.Time `json:"updatedAt"`
+}
+
+func (SubtaskState) TableName() string {
+ return "_devlake_subtask_states"
+}
diff --git a/backend/core/utils/json.go b/backend/core/utils/json.go
index 557f64cec..1505dd8cd 100644
--- a/backend/core/utils/json.go
+++ b/backend/core/utils/json.go
@@ -18,6 +18,7 @@ limitations under the License.
package utils
import (
+ "encoding/json"
"fmt"
"reflect"
@@ -72,3 +73,11 @@ func Convert[T any](value any) (T, errors.Error) {
return result, nil
}
}
+
+func ToJsonString(x any) string {
+ b, err := json.Marshal(x)
+ if err != nil {
+ panic(err)
+ }
+ return string(b)
+}
diff --git a/backend/helpers/pluginhelper/api/subtask_state_manager.go b/backend/helpers/pluginhelper/api/subtask_state_manager.go
new file mode 100644
index 000000000..27d1125f1
--- /dev/null
+++ b/backend/helpers/pluginhelper/api/subtask_state_manager.go
@@ -0,0 +1,128 @@
+/*
+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 (
+ "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"
+)
+
+// 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
+ BatchSize int // batch size for saving data
+}
+
+// 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 {
+ db dal.Dal
+ state *models.SubtaskState
+ syncPolicy *models.SyncPolicy
+ isIncremental bool // tells if the subtask should run in incremental mode or full sync mode
+ since *time.Time // the start time of the time range to work on
+ until *time.Time // the end time of the time range to work on
+ config string // current configuration of the subtask for determining if the subtask should run in incremental or full sync mode
+}
+
+// NewSubtaskStateManager create a new SubtaskStateManager
+func NewSubtaskStateManager(args *SubtaskCommonArgs) (stateManager *SubtaskStateManager, err errors.Error) {
+ db := args.GetDal()
+ syncPolicy := args.SubTaskContext.TaskContext().SyncPolicy()
+ plugin := args.SubTaskContext.TaskContext().GetName()
+ subtask := args.SubTaskContext.GetName()
+ // load sync policy and make sure it is not nil
+ if syncPolicy == nil {
+ syncPolicy = &models.SyncPolicy{}
+ }
+ // 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))
+ if err != nil {
+ if db.IsErrorNotFound(err) {
+ state = &models.SubtaskState{
+ Plugin: plugin,
+ Subtask: subtask,
+ Params: args.Params,
+ }
+ err = nil
+ } else {
+ err = errors.Default.Wrap(err, "failed to load the previous subtask state")
+ return
+ }
+ }
+ // fullsync by default
+ now := time.Now()
+ stateManager = &SubtaskStateManager{
+ db: db,
+ state: state,
+ syncPolicy: syncPolicy,
+ isIncremental: false,
+ since: syncPolicy.TimeAfter,
+ until: &now,
+ config: args.SubtaskConfig,
+ }
+ // fallback to the previous timeAfter if no new value
+ if stateManager.since == nil {
+ stateManager.since = state.TimeAfter
+ }
+ // if fullsync is set or no previous success start time, we are in the full sync mode
+ if syncPolicy.FullSync || state.PrevStartedAt == nil {
+ return
+ }
+ // if timeAfter is not set or NOT before the previous vaule, we are in the incremental mode
+ if (syncPolicy.TimeAfter == nil || state.TimeAfter == nil || !syncPolicy.TimeAfter.Before(*state.TimeAfter)) &&
+ // and the previous config is the same as the current config
+ (state.PrevConfig == stateManager.config) {
+ stateManager.isIncremental = true
+ stateManager.since = state.PrevStartedAt
+ }
+ return
+}
+
+func (c *SubtaskStateManager) IsIncremental() bool {
+ return c.isIncremental
+}
+
+func (c *SubtaskStateManager) GetSince() *time.Time {
+ return c.since
+}
+
+func (c *SubtaskStateManager) GetUntil() *time.Time {
+ return c.until
+}
+
+func (c *SubtaskStateManager) Close() errors.Error {
+ // update timeAfter in the database only for fullsync mode
+ if !c.isIncremental {
+ // prefer non-nil value
+ if c.syncPolicy.TimeAfter != nil {
+ c.state.TimeAfter = c.syncPolicy.TimeAfter
+ }
+ }
+ // always update the latest success start time
+ c.state.PrevStartedAt = c.until
+ c.state.PrevConfig = c.config
+ return c.db.Update(c.state)
+}
diff --git a/backend/helpers/pluginhelper/api/subtask_state_manager_test.go b/backend/helpers/pluginhelper/api/subtask_state_manager_test.go
new file mode 100644
index 000000000..a3cc78252
--- /dev/null
+++ b/backend/helpers/pluginhelper/api/subtask_state_manager_test.go
@@ -0,0 +1,188 @@
+/*
+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 (
+ "fmt"
+ "testing"
+ "time"
+
+ "github.com/apache/incubator-devlake/core/errors"
+ "github.com/apache/incubator-devlake/core/models"
+ mockdal "github.com/apache/incubator-devlake/mocks/core/dal"
+ mockplugin "github.com/apache/incubator-devlake/mocks/core/plugin"
+ "github.com/stretchr/testify/assert"
+ "github.com/stretchr/testify/mock"
+)
+
+func TestSubtaskStateManager(t *testing.T) {
+ time0 := errors.Must1(time.Parse(time.RFC3339, "2020-01-01T00:00:00Z"))
+ time1 := errors.Must1(time.Parse(time.RFC3339, "2021-01-01T00:00:00Z"))
+ time2 := errors.Must1(time.Parse(time.RFC3339, "2022-01-01T00:00:00Z"))
+ for _, tc := range []struct {
+ name string
+ state *models.SubtaskState
+ syncPolicy *models.SyncPolicy
+ config string
+ expectedIsIncremental bool
+ expectedSince *time.Time
+ expectedNewStateTimeAfter *time.Time
+ }{
+ {
+ name: "syncPolicy has no timeAfter - First run",
+ state: &models.SubtaskState{PrevStartedAt: nil},
+ syncPolicy: &models.SyncPolicy{TimeAfter: nil},
+ expectedIsIncremental: false,
+ expectedSince: nil,
+ expectedNewStateTimeAfter: nil,
+ },
+ {
+ name: "syncPolicy has no timeAfter - Second run",
+ state: &models.SubtaskState{PrevStartedAt: &time1},
+ syncPolicy: &models.SyncPolicy{TimeAfter: nil},
+ expectedIsIncremental: true,
+ expectedSince: &time1,
+ expectedNewStateTimeAfter: nil,
+ },
+ {
+ name: "syncPolicy has no timeAfter - Third run with timeAfter specified",
+ state: &models.SubtaskState{PrevStartedAt: &time1},
+ syncPolicy: &models.SyncPolicy{TimeAfter: &time1},
+ expectedIsIncremental: true,
+ expectedSince: &time1,
+ expectedNewStateTimeAfter: nil,
+ },
+ {
+ name: "syncPolicy has timeAfter - First run",
+ state: &models.SubtaskState{PrevStartedAt: nil},
+ syncPolicy: &models.SyncPolicy{TimeAfter: &time1},
+ expectedIsIncremental: false,
+ expectedSince: &time1,
+ expectedNewStateTimeAfter: &time1,
+ },
+ {
+ name: "syncPolicy has timeAfter - Second run with a later timeAfter",
+ state: &models.SubtaskState{TimeAfter: &time1, PrevStartedAt: &time2},
+ syncPolicy: &models.SyncPolicy{TimeAfter: &time2},
+ expectedIsIncremental: true,
+ expectedSince: &time2,
+ expectedNewStateTimeAfter: &time1,
+ },
+ {
+ name: "syncPolicy has timeAfter - Third run with a earlier timeAfter",
+ state: &models.SubtaskState{TimeAfter: &time1, PrevStartedAt: &time1},
+ syncPolicy: &models.SyncPolicy{TimeAfter: &time0},
+ expectedIsIncremental: false,
+ expectedSince: &time0,
+ expectedNewStateTimeAfter: &time0,
+ },
+ {
+ name: "syncPolicy has timeAfter - Fourth run with a same timeAfter",
+ state: &models.SubtaskState{TimeAfter: &time1, PrevStartedAt: &time2},
+ syncPolicy: &models.SyncPolicy{TimeAfter: &time1},
+ expectedIsIncremental: true,
+ expectedSince: &time2,
+ expectedNewStateTimeAfter: &time1,
+ },
+ {
+ name: "config no changed",
+ state: &models.SubtaskState{TimeAfter: &time1, PrevStartedAt: &time2, PrevConfig: "hello"},
+ syncPolicy: &models.SyncPolicy{TimeAfter: &time1},
+ config: "hello",
+ expectedIsIncremental: true,
+ expectedSince: &time2,
+ expectedNewStateTimeAfter: &time1,
+ },
+ {
+ name: "Full sync - with timeAfter",
+ state: &models.SubtaskState{TimeAfter: &time1, PrevStartedAt: &time1},
+ syncPolicy: &models.SyncPolicy{FullSync: true},
+ expectedIsIncremental: false,
+ expectedSince: &time1,
+ expectedNewStateTimeAfter: &time1,
+ },
+ {
+ name: "Full sync - with newer timeAfter",
+ state: &models.SubtaskState{TimeAfter: &time1, PrevStartedAt: &time1},
+ syncPolicy: &models.SyncPolicy{TimeAfter: &time2, FullSync: true},
+ expectedIsIncremental: false,
+ expectedSince: &time2,
+ expectedNewStateTimeAfter: &time2,
+ },
+ {
+ name: "Full sync - with older timeAfter",
+ state: &models.SubtaskState{TimeAfter: &time1, PrevStartedAt: &time1},
+ syncPolicy: &models.SyncPolicy{TimeAfter: &time0, FullSync: true},
+ expectedIsIncremental: false,
+ expectedSince: &time0,
+ expectedNewStateTimeAfter: &time0,
+ },
+ {
+ name: "Full sync - without timeAfter",
+ state: &models.SubtaskState{TimeAfter: nil, PrevStartedAt: &time1},
+ syncPolicy: &models.SyncPolicy{FullSync: true},
+ expectedIsIncremental: false,
+ expectedSince: nil,
+ expectedNewStateTimeAfter: nil,
+ },
+ {
+ name: "Full sync - config changed",
+ state: &models.SubtaskState{PrevStartedAt: &time1, PrevConfig: "hello"},
+ syncPolicy: &models.SyncPolicy{},
+ config: "world",
+ expectedIsIncremental: false,
+ expectedSince: nil,
+ expectedNewStateTimeAfter: nil,
+ },
+ } {
+ started := time.Now()
+ t.Run(tc.name, func(t *testing.T) {
+ // mockBasicRes := unithelper.DummyBasicRes(func(mockDal *mockdal.Dal) {
+ mockDal := new(mockdal.Dal)
+ mockDal.On("First", mock.Anything, mock.Anything).Run(func(args mock.Arguments) {
+ dst := args.Get(0).(*models.SubtaskState)
+ *dst = *tc.state
+ }).Return(nil).Once()
+ mockDal.On("Update", mock.Anything, mock.Anything).Return(nil).Once()
+ // })
+
+ // mockBasicRes, tc.syncPolicy, "table", "params"
+ mockTaskCtx := new(mockplugin.TaskContext)
+ mockTaskCtx.On("SyncPolicy").Return(tc.syncPolicy)
+ mockTaskCtx.On("GetName").Return("test-plugin")
+ mockSubtaskCtx := new(mockplugin.SubTaskContext)
+ mockSubtaskCtx.On("TaskContext").Return(mockTaskCtx)
+ mockSubtaskCtx.On("GetName").Return("test-subtask")
+ mockSubtaskCtx.On("GetDal").Return(mockDal)
+
+ stateManager, err := NewSubtaskStateManager(&SubtaskCommonArgs{
+ SubTaskContext: mockSubtaskCtx,
+ SubtaskConfig: fmt.Sprintf("%v", tc.config),
+ })
+ assert.Nil(t, err)
+ assert.Equal(t, tc.expectedSince, stateManager.since)
+ assert.Equal(t, tc.expectedIsIncremental, stateManager.isIncremental)
+ assert.Nil(t, stateManager.Close())
+ assert.Equal(t, tc.expectedNewStateTimeAfter, stateManager.state.TimeAfter)
+ // PrevStartedAt should be updated
+ assert.GreaterOrEqual(t, stateManager.state.PrevStartedAt.Unix(), started.Unix())
+ // First and update should both be called once
+ mockDal.AssertExpectations(t)
+ })
+ }
+}