feat: Implement support for online and local rate limiting

Task: https://pms.uniontech.com/story-view-40501.html

Author: xml

src/internal/config/config.go

@@ -91,10 +91,11 @@ type Config struct {
 	NonUnknownList     []string // 非未知来源更新list文件
 	OtherSourceList    []string // 其他类型更新list文件路径
 
-	DownloadSpeedLimitConfig string
-	lastoreDaemonStatus      LastoreDaemonStatus
-	UpdateStatus             string
-	PlatformUpdate           bool
+	DownloadSpeedLimitConfig      string
+	LocalDownloadSpeedLimitConfig string
+	lastoreDaemonStatus           LastoreDaemonStatus
+	UpdateStatus                  string
+	PlatformUpdate                bool
 
 	PlatformUrl            string // 更新接口地址
 	StartCheckRange        []int  // 开机检查更新区间
@@ -188,6 +189,7 @@ const (
 	dSettingsKeySystemSourceList                     = "system-sources"
 	dSettingsKeyNonUnknownList                       = "non-unknown-sources"
 	dSettingsKeyDownloadSpeedLimit                   = "download-speed-limit"
+	dSettingsKeyLocalDownloadSpeedLimit              = "local-download-speed-limit"
 	DSettingsKeyLastoreDaemonStatus                  = "lastore-daemon-status"
 	dSettingsKeyUpdateStatus                         = "update-status"
 	dSettingsKeyPlatformUpdate                       = "platform-update"
@@ -467,6 +469,13 @@ func getConfigFromDSettings() *Config {
 		c.DownloadSpeedLimitConfig = v.Value().(string)
 	}
 
+	v, err = c.dsLastoreManager.Value(0, dSettingsKeyLocalDownloadSpeedLimit)
+	if err != nil {
+		logger.Warning(err)
+	} else {
+		c.LocalDownloadSpeedLimitConfig = v.Value().(string)
+	}
+
 	updateLastoreDaemonStatus := func() {
 		v, err = c.dsLastoreManager.Value(0, DSettingsKeyLastoreDaemonStatus)
 		if err != nil {
@@ -899,6 +908,11 @@ func (c *Config) SetDownloadSpeedLimitConfig(config string) error {
 	return c.save(dSettingsKeyDownloadSpeedLimit, config)
 }
 
+func (c *Config) SetLocalDownloadSpeedLimitConfig(config string) error {
+	c.LocalDownloadSpeedLimitConfig = config
+	return c.save(dSettingsKeyLocalDownloadSpeedLimit, config)
+}
+
 func (c *Config) SetLastoreDaemonStatus(status LastoreDaemonStatus) error {
 	c.statusMu.Lock()
 	c.lastoreDaemonStatus = status

src/internal/system/apt/apt.go

@@ -120,19 +120,21 @@ func createCommandLine(cmdType string, cmdArgs []string) *exec.Cmd {
 	return exec.Command("apt-get", args...)
 }
 
-func newAPTCommand(cmdSet system.CommandSet, jobId string, cmdType string, fn system.Indicator, cmdArgs []string) *system.Command {
+func newAPTCommand(cmdSet system.CommandSet, jobId string, cmdType string, fn system.Indicator, deliveryFn system.DeliveryIndicator, cmdArgs []string) *system.Command {
 	cmd := createCommandLine(cmdType, cmdArgs)
 
 	// See aptCommand.Abort
 	cmd.SysProcAttr = &syscall.SysProcAttr{Setpgid: true}
 	r := &system.Command{
-		JobId:             jobId,
-		CmdSet:            cmdSet,
-		Indicator:         fn,
-		ParseJobError:     parseJobError,
-		ParseProgressInfo: parseProgressInfo,
-		Cmd:               cmd,
-		Cancelable:        true,
+		JobId:                     jobId,
+		CmdSet:                    cmdSet,
+		Indicator:                 fn,
+		DeliveryIndicator:         deliveryFn,
+		ParseJobError:             parseJobError,
+		ParseProgressInfo:         parseProgressInfo,
+		ParseDeliveryDownloadInfo: parseDeliveryDownloadInfo,
+		Cmd:                       cmd,
+		Cancelable:                true,
 	}
 	cmd.Stdout = &r.Stdout
 	cmd.Stderr = &r.Stderr

src/internal/system/apt/proxy.go

@@ -29,6 +29,7 @@ type APTSystem struct {
 	CmdSet            map[string]*system.Command
 	Indicator         system.Indicator
 	IncrementalUpdate bool
+	DeliveryIndicator system.DeliveryIndicator
 }
 
 func NewSystem(nonUnknownList []string, otherList []string, incrementalUpdate bool) system.System {
@@ -56,6 +57,53 @@ func parseProgressField(v string) (float64, error) {
 	return progress, nil
 }
 
+func parseDeliveryDownloadInfo(id, line string) (system.JobDeliveryDownloadInfo, error) {
+	if strings.Contains(line, "102 Status") {
+		var jobDeliveryDownloadInfo system.JobDeliveryDownloadInfo
+		jobDeliveryDownloadInfo.JobId = id
+		line = strings.TrimPrefix(line, "102 Status[")
+		line = strings.TrimSuffix(line, "]")
+		parts := strings.Split(line, "} {")
+		var isFinish bool
+		var err error
+		var speed int64
+		var proto string
+		for i := range parts {
+			parts[i] = strings.Trim(parts[i], "{} ")
+		}
+		for _, part := range parts {
+			kv := strings.SplitN(part, " ", 2)
+			if len(kv) != 2 {
+				continue
+			}
+			key := kv[0]
+			value := kv[1]
+
+			if key != "IsFinish" && key != "Speed" && key != "Proto" {
+				continue
+			} else if key == "IsFinish" {
+				isFinish, err = strconv.ParseBool(value)
+				if err != nil {
+					logger.Warningf("failed to parse isFinish %v", err)
+					isFinish = true
+				}
+			} else if key == "Speed" {
+				speed, _ = strconv.ParseInt(value, 10, 64)
+			} else if key == "Proto" {
+				proto = value
+			}
+		}
+		if !isFinish {
+			jobDeliveryDownloadInfo.Speed = speed
+			jobDeliveryDownloadInfo.Proto = proto
+		} else {
+			jobDeliveryDownloadInfo.Speed = -1
+		}
+		return jobDeliveryDownloadInfo, nil
+	}
+	return system.JobDeliveryDownloadInfo{JobId: id}, nil
+}
+
 func parseProgressInfo(id, line string) (system.JobProgressInfo, error) {
 	fs := strings.SplitN(line, ":", 4)
 	if len(fs) != 4 {
@@ -109,6 +157,10 @@ func (p *APTSystem) AttachIndicator(f system.Indicator) {
 	p.Indicator = f
 }
 
+func (p *APTSystem) AttachDeliveryIndicator(f system.DeliveryIndicator) {
+	p.DeliveryIndicator = f
+}
+
 func WaitDpkgLockRelease() {
 	for {
 		msg, wait := system.CheckLock("/var/lib/dpkg/lock")
@@ -268,7 +320,7 @@ func (p *APTSystem) DownloadPackages(jobId string, packages []string, environ ma
 	if err != nil {
 		return err
 	}
-	c := newAPTCommand(p, jobId, system.DownloadJobType, p.Indicator, append(packages, OptionToArgs(args)...))
+	c := newAPTCommand(p, jobId, system.DownloadJobType, p.Indicator, p.DeliveryIndicator, append(packages, OptionToArgs(args)...))
 	c.SetEnv(environ)
 	return c.Start()
 }
@@ -300,11 +352,11 @@ func (p *APTSystem) DownloadSource(jobId string, packages []string, environ map[
 		environ["DEEPIN_IMMUTABLE_UPGRADE_APT_OPTION"] = upgradeArgString
 		logger.Info("DownloadSource set env DEEPIN_IMMUTABLE_UPGRADE_APT_OPTION:", upgradeArgString)
 
-		c := newAPTCommand(p, jobId, system.IncrementalDownloadJobType, p.Indicator, cmdArgs)
+		c := newAPTCommand(p, jobId, system.IncrementalDownloadJobType, p.Indicator, p.DeliveryIndicator, cmdArgs)
 		c.SetEnv(environ)
 		return c.Start()
 	}
-	c := newAPTCommand(p, jobId, system.PrepareDistUpgradeJobType, p.Indicator, append(packages, OptionToArgs(args)...))
+	c := newAPTCommand(p, jobId, system.PrepareDistUpgradeJobType, p.Indicator, p.DeliveryIndicator, append(packages, OptionToArgs(args)...))
 	c.SetEnv(environ)
 	return c.Start()
 }
@@ -316,7 +368,7 @@ func (p *APTSystem) Remove(jobId string, packages []string, environ map[string]s
 		return err
 	}
 
-	c := newAPTCommand(p, jobId, system.RemoveJobType, p.Indicator, packages)
+	c := newAPTCommand(p, jobId, system.RemoveJobType, p.Indicator, p.DeliveryIndicator, packages)
 	environ["IMMUTABLE_DISABLE_REMOUNT"] = "false"
 	c.SetEnv(environ)
 	return safeStart(c)
@@ -328,7 +380,7 @@ func (p *APTSystem) Install(jobId string, packages []string, environ map[string]
 	if err != nil {
 		return err
 	}
-	c := newAPTCommand(p, jobId, system.InstallJobType, p.Indicator, append(OptionToArgs(args), packages...))
+	c := newAPTCommand(p, jobId, system.InstallJobType, p.Indicator, p.DeliveryIndicator, append(OptionToArgs(args), packages...))
 	environ["IMMUTABLE_DISABLE_REMOUNT"] = "false"
 	c.SetEnv(environ)
 	return safeStart(c)
@@ -365,12 +417,12 @@ func (p *APTSystem) DistUpgrade(jobId string, packages []string, environ map[str
 		environ["DEEPIN_IMMUTABLE_UPGRADE_APT_OPTION"] = upgradeArgString
 		logger.Info("DistUpgrade set env DEEPIN_IMMUTABLE_UPGRADE_APT_OPTION:", upgradeArgString)
 
-		c := newAPTCommand(p, jobId, system.IncrementalUpdateJobType, p.Indicator, cmdArgs)
+		c := newAPTCommand(p, jobId, system.IncrementalUpdateJobType, p.Indicator, p.DeliveryIndicator, cmdArgs)
 		c.SetEnv(environ)
 		return c.Start()
 	}
 
-	c := newAPTCommand(p, jobId, system.DistUpgradeJobType, p.Indicator, append(OptionToArgs(args), packages...))
+	c := newAPTCommand(p, jobId, system.DistUpgradeJobType, p.Indicator, p.DeliveryIndicator, append(OptionToArgs(args), packages...))
 	environ["IMMUTABLE_DISABLE_REMOUNT"] = "false"
 	c.SetEnv(environ)
 	return safeStart(c)
@@ -384,7 +436,7 @@ func (p *APTSystem) UpdateSource(jobId string, environ map[string]string, args m
 			logger.Warningf("Failed to update remotes: %v, %s", err, string(output))
 		}
 	}
-	c := newAPTCommand(p, jobId, system.UpdateSourceJobType, p.Indicator, OptionToArgs(args))
+	c := newAPTCommand(p, jobId, system.UpdateSourceJobType, p.Indicator, p.DeliveryIndicator, OptionToArgs(args))
 	c.AtExitFn = func() bool {
 		// 无网络时检查更新失败,exitCode为0,空间不足(不确定exit code)导致需要特殊处理
 		if c.ExitCode == system.ExitSuccess && bytes.Contains(c.Stderr.Bytes(), []byte("Some index files failed to download")) {
@@ -402,7 +454,7 @@ func (p *APTSystem) UpdateSource(jobId string, environ map[string]string, args m
 }
 
 func (p *APTSystem) Clean(jobId string) error {
-	c := newAPTCommand(p, jobId, system.CleanJobType, p.Indicator, nil)
+	c := newAPTCommand(p, jobId, system.CleanJobType, p.Indicator, p.DeliveryIndicator, nil)
 	return c.Start()
 }
 
@@ -422,7 +474,7 @@ func (p *APTSystem) AbortWithFailed(jobId string) error {
 
 func (p *APTSystem) FixError(jobId string, errType string, environ map[string]string, args map[string]string) error {
 	WaitDpkgLockRelease()
-	c := newAPTCommand(p, jobId, system.FixErrorJobType, p.Indicator, append([]string{errType}, OptionToArgs(args)...))
+	c := newAPTCommand(p, jobId, system.FixErrorJobType, p.Indicator, p.DeliveryIndicator, append([]string{errType}, OptionToArgs(args)...))
 	environ["IMMUTABLE_DISABLE_REMOUNT"] = "false"
 	c.SetEnv(environ)
 	if system.JobErrorType(errType) == system.ErrorDependenciesBroken { // 修复依赖错误的时候，会有需要卸载dde的情况，因此需要用safeStart来进行处理
@@ -655,7 +707,7 @@ func parseBackupJobError(stdErrStr string, stdOutStr string) *system.JobError {
 }
 
 func (p *APTSystem) OsBackup(jobId string) error {
-	c := newAPTCommand(p, jobId, system.BackupJobType, p.Indicator, nil)
+	c := newAPTCommand(p, jobId, system.BackupJobType, p.Indicator, p.DeliveryIndicator, nil)
 	c.ParseJobError = parseBackupJobError
 	c.ParseProgressInfo = func(id, line string) (system.JobProgressInfo, error) {
 		type info struct {

src/internal/system/command.go

@@ -34,9 +34,11 @@ type Command struct {
 
 	pipe *os.File
 
-	Indicator         Indicator
-	ParseProgressInfo ParseProgressInfo
-	ParseJobError     ParseJobError
+	Indicator                 Indicator
+	DeliveryIndicator         DeliveryIndicator
+	ParseProgressInfo         ParseProgressInfo
+	ParseJobError             ParseJobError
+	ParseDeliveryDownloadInfo ParseDeliveryDownloadInfo
 
 	Stdout   bytes.Buffer
 	Stderr   bytes.Buffer
@@ -275,17 +277,27 @@ func (c *Command) updateProgress() {
 			return
 		}
 
-		info, err := c.ParseProgressInfo(c.JobId, line)
-		if err != nil {
-			logger.Errorf("aptCommand.updateProgress %v -> %v\n", info, err)
-			c.Indicator(JobProgressInfo{
-				OnlyLog:     true,
-				OriginalLog: line,
-			})
+		if strings.Contains(line, "102 Status") {
+			deliveryInfo, err := c.ParseDeliveryDownloadInfo(c.JobId, line)
+			if err != nil {
+				logger.Errorf("aptCommand.updateProgress %v -> %v\n", deliveryInfo, err)
+				continue
+			}
+			c.DeliveryIndicator(deliveryInfo)
 			continue
+		} else {
+			info, err := c.ParseProgressInfo(c.JobId, line)
+			if err != nil {
+				logger.Errorf("aptCommand.updateProgress %v -> %v\n", info, err)
+				c.Indicator(JobProgressInfo{
+					OnlyLog:     true,
+					OriginalLog: line,
+				})
+				continue
+			}
+			info.OriginalLog = line
+			c.Cancelable = info.Cancelable
+			c.Indicator(info)
 		}
-		info.OriginalLog = line
-		c.Cancelable = info.Cancelable
-		c.Indicator(info)
 	}
 }

src/internal/system/system.go

@@ -62,6 +62,16 @@ const (
 	NotifyExpireTimeoutPrivateLong = 600000
 )
 
+type JobDeliveryDownloadInfo struct {
+	JobId          string
+	FileName       string
+	Proto          string
+	DownloadSize   int64
+	DownloadedSize int64
+	Speed          int64
+	Progress       float64
+}
+
 type JobProgressInfo struct {
 	JobId         string
 	Progress      float64
@@ -132,7 +142,9 @@ var NotSupportError = errors.New("not support operation")
 var ResourceExitError = errors.New("resource exists")
 
 type Indicator func(JobProgressInfo)
+type DeliveryIndicator func(JobDeliveryDownloadInfo)
 type ParseProgressInfo func(id, line string) (JobProgressInfo, error)
+type ParseDeliveryDownloadInfo func(id, line string) (JobDeliveryDownloadInfo, error)
 type ParseJobError func(stdErrStr string, stdOutStr string) *JobError
 
 type System interface {
@@ -146,6 +158,7 @@ type System interface {
 	Abort(jobId string) error
 	AbortWithFailed(jobId string) error
 	AttachIndicator(Indicator)
+	AttachDeliveryIndicator(DeliveryIndicator)
 	FixError(jobId string, errType string, environ map[string]string, cmdArgs map[string]string) error
 	OsBackup(jobId string) error
 	CheckSystem(jobId string, checkType string, environ map[string]string, cmdArgs map[string]string) error