[Refactor][core]Data inflation when using postgres

[narrowizard]

What and why to refactor

As a software engineer, I am using devlake to collect developer data since long time ago. Recently we find that the _devlake_subtasks table occurs a data inflation. It used 750MB after a week from we upgrading devlake v1.0, but there were only 1000 records in the table.

Describe the solution you'd like

Solutions from @klesh

1. Reduce the update rate to _devlake_subtasks table when collect data.
2. Store progress info in memory firstly and write it to db in a fixed rate.

Related issues

No

Additional context

• The issue was introduced in v1.0.
• We are using postgres.

[narrowizard]

After reading some code,I organizaed a table to show when we updading the progress of sub tasks.

Trigger timing	Times	Code
Sub task started	Once a sub task	incubator-devlake/backend/core/runner/run_task.go Lines 308 to 314 in 98bc98b if progress != nil { progress <- plugin.RunningProgress{ Type: plugin.SetCurrentSubTask, SubTaskName: subtaskMeta.Name, SubTaskNumber: subtaskNumber, } }
Mannually called by sub task context	Unknown(decide by plugin implement)	incubator-devlake/backend/impls/context/default_exec_context.go Lines 68 to 97 in 98bc98b func (c *defaultExecContext) SetProgress(progressType plugin.ProgressType, current int, total int) { c.current = int64(current) c.total = total if c.progress != nil { c.progress <- plugin.RunningProgress{ Type: progressType, Current: current, Total: total, } } } func (c *defaultExecContext) IncProgress(progressType plugin.ProgressType, quantity int) { atomic.AddInt64(&c.current, int64(quantity)) current := c.current if c.progress != nil { c.progress <- plugin.RunningProgress{ Type: progressType, Current: int(current), Total: c.total, } // subtask progress may go too fast, remove old messages because they don't matter any more if progressType == plugin.SubTaskSetProgress { for len(c.progress) > 1 { <-c.progress } } } } incubator-devlake/backend/impls/context/default_subtask_context.go Lines 37 to 53 in 98bc98b func (c *DefaultSubTaskContext) SetProgress(current int, total int) { c.defaultExecContext.SetProgress(plugin.SubTaskSetProgress, current, total) if total > -1 { c.BasicRes.GetLogger().Info("total jobs: %d", c.total) } } // IncProgress FIXME ... func (c *DefaultSubTaskContext) IncProgress(quantity int) { c.defaultExecContext.IncProgress(plugin.SubTaskIncProgress, quantity) if c.LastProgressTime.IsZero() || c.LastProgressTime.Add(3*time.Second).Before(time.Now()) || c.current%1000 == 0 { c.LastProgressTime = time.Now() c.BasicRes.GetLogger().Info("finished records: %d(not exactly)", c.current) } else { c.BasicRes.GetLogger().Debug("finished records: %d", c.current) } }
Mannually called by task context	Unknown	incubator-devlake/backend/impls/context/default_task_context.go Lines 40 to 49 in 98bc98b func (c *DefaultTaskContext) SetProgress(current int, total int) { c.defaultExecContext.SetProgress(plugin.TaskSetProgress, current, total) c.BasicRes.GetLogger().Info("total step: %d", c.total) } // IncProgress FIXME ... func (c *DefaultTaskContext) IncProgress(quantity int) { c.defaultExecContext.IncProgress(plugin.TaskIncProgress, quantity) c.BasicRes.GetLogger().Info("finished step: %d / %d", c.current, c.total) }

Seems that the progress updating calling is not always under control, so we need to decrease db operation automatically.