Merge pull request #2456 from wanyaoqi/feature/wyq/guest-io-throttle

feature: guest io throttle

cmd/climc/shell/servers.go

@@ -1016,4 +1016,19 @@ func init() {
 		printObject(result)
 		return nil
 	})
+
+	type ServerIoThrottle struct {
+		ID   string `help:"ID or name of VM" json:"-"`
+		BPS  int    `help:"bps(MB) of throttle" json:"bps"`
+		IOPS int    `help:"iops of throttle" json:"iops"`
+	}
+	R(&ServerIoThrottle{}, "server-io-throttle", "Guest io set throttle", func(s *mcclient.ClientSession, opts *ServerIoThrottle) error {
+		params := jsonutils.Marshal(opts)
+		result, err := modules.Servers.PerformAction(s, opts.ID, "io-throttle", params)
+		if err != nil {
+			return err
+		}
+		printObject(result)
+		return nil
+	})
 }

docs/index.yaml

@@ -231,6 +231,8 @@ paths:
     $ref: "./instance/user-data.yaml"
   /servers/{instanceId}/vnc:
     $ref: "./instance/vnc.yaml"
+  /servers/{instanceId}/io-throttle:
+    $ref: "./instance/io-throttle.yaml"
 
   /images:
     $ref: "./image/images.yaml"

docs/instance/io-throttle.yaml

@@ -0,0 +1,16 @@
+post:
+  summary: 虚拟机磁盘IO限速
+  parameters: 
+    - $ref: '../parameters/instance.yaml#/instanceId'
+    - in: body
+      name: server
+      required: true
+      schema:
+        $ref: '../schemas/instance.yaml#/InstanceIoThrottle' 
+  responses:
+    200:
+      description: 实例信息
+      schema:
+        $ref: '../schemas/instance.yaml#/InstanceResponse'
+  tags:
+    - instance

docs/schemas/instance.yaml

@@ -114,6 +114,17 @@ InstanceAddSecgroup:
       example: 38dc8e80-14e5-4629-81c1-3847320ae821
       description: 待绑定的安全组名称或ID
 
+InstanceIoThrottle:
+  type: object
+  properties:
+    iops:
+      type: integer
+      example: 1000
+      description: 最高iops
+    bps:
+      type: integer
+      example: 1000
+      description: 最高bps，单位MB
 
 InstanceRevokeSecgroup:
   type: object

pkg/compute/models/guest_actions.go

@@ -3474,3 +3474,47 @@ func (guest *SGuest) StartGuestDiskResizeTask(ctx context.Context, userCred mccl
 	task.ScheduleRun(nil)
 	return nil
 }
+
+func (self *SGuest) AllowPerformIoThrottle(ctx context.Context,
+	userCred mcclient.TokenCredential,
+	query jsonutils.JSONObject,
+	data jsonutils.JSONObject) bool {
+	return self.IsOwner(userCred) || db.IsAdminAllowPerform(userCred, self, "io-throttle")
+}
+
+func (self *SGuest) PerformIoThrottle(ctx context.Context, userCred mcclient.TokenCredential, query jsonutils.JSONObject, data jsonutils.JSONObject) (jsonutils.JSONObject, error) {
+	if self.Hypervisor != api.HYPERVISOR_KVM {
+		return nil, httperrors.NewBadRequestError("Hypervisor %s can't do io throttle", self.Hypervisor)
+	}
+	if self.Status != api.VM_RUNNING {
+		return nil, httperrors.NewServerStatusError("Cannot do io throttle in status %s", self.Status)
+	}
+	bpsMb, err := data.Int("bps")
+	if err != nil {
+		return nil, httperrors.NewMissingParameterError("bps")
+	}
+	if bpsMb < 0 {
+		return nil, httperrors.NewInputParameterError("bps must > 0")
+	}
+	iops, err := data.Int("iops")
+	if err != nil {
+		return nil, httperrors.NewMissingParameterError("iops")
+	}
+	if iops < 0 {
+		return nil, httperrors.NewInputParameterError("iops must > 0")
+	}
+	return nil, self.StartBlockIoThrottleTask(ctx, userCred, bpsMb, iops)
+}
+
+func (self *SGuest) StartBlockIoThrottleTask(ctx context.Context, userCred mcclient.TokenCredential, bpsMb, iops int64) error {
+	params := jsonutils.NewDict()
+	params.Set("bps", jsonutils.NewInt(bpsMb))
+	params.Set("iops", jsonutils.NewInt(iops))
+	task, err := taskman.TaskManager.NewTask(ctx, "GuestBlockIoThrottleTask", self, userCred, params, "", "", nil)
+	if err != nil {
+		log.Errorf("%s", err)
+		return err
+	}
+	task.ScheduleRun(nil)
+	return nil
+}

pkg/compute/tasks/guest_block_io_throttle_task.go

@@ -0,0 +1,40 @@
+package tasks
+
+import (
+	"context"
+	"fmt"
+
+	"yunion.io/x/jsonutils"
+	"yunion.io/x/onecloud/pkg/cloudcommon/db"
+	"yunion.io/x/onecloud/pkg/cloudcommon/db/taskman"
+	"yunion.io/x/onecloud/pkg/compute/models"
+)
+
+type GuestBlockIoThrottleTask struct {
+	SGuestBaseTask
+}
+
+func init() {
+	taskman.RegisterTask(GuestBlockIoThrottleTask{})
+}
+
+func (self *GuestBlockIoThrottleTask) OnInit(ctx context.Context, obj db.IStandaloneModel, data jsonutils.JSONObject) {
+	guest := obj.(*models.SGuest)
+	url := fmt.Sprintf("/servers/%s/io-throttle", guest.Id)
+	headers := self.GetTaskRequestHeader()
+	host := guest.GetHost()
+	self.SetStage("OnIoThrottle", nil)
+	_, err := host.Request(ctx, self.UserCred, "POST", url, headers, self.Params)
+	if err != nil {
+		self.SetStageFailed(ctx, err.Error())
+	}
+}
+
+func (self *GuestBlockIoThrottleTask) OnIoThrottle(ctx context.Context, guest *models.SGuest, data jsonutils.JSONObject) {
+	guest.SetMetadata(ctx, "io-throttle", self.Params.String(), self.UserCred)
+	self.SetStageComplete(ctx, nil)
+}
+
+func (self *GuestBlockIoThrottleTask) OnIoThrottleFailed(ctx context.Context, guest *models.SGuest, data jsonutils.JSONObject) {
+	self.SetStageFailed(ctx, data.String())
+}

pkg/hostman/guestman/guesthandlers/guesthandler.go

@@ -40,20 +40,19 @@ type actionFunc func(context.Context, string, jsonutils.JSONObject) (interface{}
 var (
 	keyWords    = []string{"servers"}
 	actionFuncs = map[string]actionFunc{
-		"create":  guestCreate,
-		"deploy":  guestDeploy,
-		"start":   guestStart,
-		"stop":    guestStop,
-		"monitor": guestMonitor,
-		"sync":    guestSync,
-		"suspend": guestSuspend,
+		"create":      guestCreate,
+		"deploy":      guestDeploy,
+		"start":       guestStart,
+		"stop":        guestStop,
+		"monitor":     guestMonitor,
+		"sync":        guestSync,
+		"suspend":     guestSuspend,
+		"io-throttle": guestIoThrottle,
 
 		"snapshot":             guestSnapshot,
 		"delete-snapshot":      guestDeleteSnapshot,
 		"reload-disk-snapshot": guestReloadDiskSnapshot,
 		// "remove-statefile":     guestRemoveStatefile,
-		// "io-throttle":          guestIoThrottle,
-
 		"src-prepare-migrate":  guestSrcPrepareMigrate,
 		"dest-prepare-migrate": guestDestPrepareMigrate,
 		"live-migrate":         guestLiveMigrate,
@@ -244,6 +243,26 @@ func guestSuspend(ctx context.Context, sid string, body jsonutils.JSONObject) (i
 	return nil, nil
 }
 
+func guestIoThrottle(ctx context.Context, sid string, body jsonutils.JSONObject) (interface{}, error) {
+	guest, ok := guestman.GetGuestManager().Servers[sid]
+	if !ok {
+		return nil, httperrors.NewNotFoundError("Guest %s not found", sid)
+	}
+	if !guest.IsRunning() {
+		return nil, httperrors.NewInvalidStatusError("Not running")
+	}
+	bps, err := body.Int("bps")
+	if err != nil {
+		return nil, httperrors.NewMissingParameterError("bps")
+	}
+	iops, err := body.Int("iops")
+	if err != nil {
+		return nil, httperrors.NewMissingParameterError("iops")
+	}
+	hostutils.DelayTaskWithoutReqctx(ctx, guestman.GetGuestManager().GuestIoThrottle, &guestman.SGuestIoThrottle{sid, bps, iops})
+	return nil, nil
+}
+
 func guestSrcPrepareMigrate(ctx context.Context, sid string, body jsonutils.JSONObject) (interface{}, error) {
 	if !guestman.GetGuestManager().IsGuestExist(sid) {
 		return nil, httperrors.NewNotFoundError("Guest %s not found", sid)

pkg/hostman/guestman/guesthelper.go

@@ -105,3 +105,9 @@ type SGuestCreateFromLibvirt struct {
 	GuestDesc   *jsonutils.JSONDict
 	DisksPath   *jsonutils.JSONDict
 }
+
+type SGuestIoThrottle struct {
+	Sid  string
+	BPS  int64
+	IOPS int64
+}

pkg/hostman/guestman/guestman.go

@@ -454,6 +454,18 @@ func (m *SGuestManager) GuestSuspend(ctx context.Context, params interface{}) (j
 	return nil, nil
 }
 
+func (m *SGuestManager) GuestIoThrottle(ctx context.Context, params interface{}) (jsonutils.JSONObject, error) {
+	guestIoThrottle, ok := params.(*SGuestIoThrottle)
+	if !ok {
+		return nil, hostutils.ParamsError
+	}
+	guest := m.Servers[guestIoThrottle.Sid]
+	if guest.IsRunning() {
+		return nil, guest.BlockIoThrottle(ctx, guestIoThrottle.BPS, guestIoThrottle.IOPS)
+	}
+	return nil, httperrors.NewInvalidStatusError("Guest not running")
+}
+
 func (m *SGuestManager) SrcPrepareMigrate(ctx context.Context, params interface{}) (jsonutils.JSONObject, error) {
 	migParams, ok := params.(*SSrcPrepareMigrate)
 	if !ok {

pkg/hostman/guestman/guesttasks.go

@@ -1155,3 +1155,54 @@ func (task *SGuestHotplugCpuMemTask) onFail(reason string) {
 func (task *SGuestHotplugCpuMemTask) onSucc() {
 	hostutils.TaskComplete(task.ctx, nil)
 }
+
+type SGuestBlockIoThrottleTask struct {
+	*SKVMGuestInstance
+
+	ctx  context.Context
+	bps  int64
+	iops int64
+}
+
+func (task *SGuestBlockIoThrottleTask) Start() error {
+	task.findBlockDevices()
+	return nil
+}
+
+func (task *SGuestBlockIoThrottleTask) findBlockDevices() {
+	task.Monitor.GetBlocks(task.onBlockDriversSucc)
+}
+
+func (task *SGuestBlockIoThrottleTask) onBlockDriversSucc(res *jsonutils.JSONArray) {
+	drivers := make([]string, 0)
+	for i := 0; i < res.Length(); i++ {
+		device, err := res.GetAt(i)
+		if err == nil {
+			driver, err := device.GetString("device")
+			if err == nil {
+				if strings.HasPrefix(driver, "drive_") {
+					drivers = append(drivers, driver)
+				}
+			}
+		}
+	}
+	log.Infof("Drivers %s do io throttle bps %d iops %d", drivers, task.bps, task.iops)
+	task.doIoThrottle(drivers)
+}
+
+func (task *SGuestBlockIoThrottleTask) doIoThrottle(drivers []string) {
+	if len(drivers) == 0 {
+		hostutils.TaskComplete(task.ctx, nil)
+	} else {
+		driver := drivers[0]
+		drivers = drivers[1:]
+		_cb := func(res string) {
+			if len(res) > 0 {
+				hostutils.TaskFailed(task.ctx, res)
+			} else {
+				task.doIoThrottle(drivers)
+			}
+		}
+		task.Monitor.BlockIoThrottle(driver, task.bps, task.iops, _cb)
+	}
+}

pkg/hostman/guestman/qemu-kvm.go

@@ -1422,3 +1422,8 @@ func (s *SKVMGuestInstance) onlineResizeDisk(ctx context.Context, diskId string,
 	task := NewGuestOnlineResizeDiskTask(ctx, s, diskId, sizeMB)
 	task.Start()
 }
+
+func (s *SKVMGuestInstance) BlockIoThrottle(ctx context.Context, bps, iops int64) error {
+	task := SGuestBlockIoThrottleTask{s, ctx, bps, iops}
+	return task.Start()
+}

pkg/hostman/monitor/hmp.go

@@ -453,3 +453,8 @@ func (m *HmpMonitor) ObjectAdd(objectType string, params map[string]string, call
 	cmd := fmt.Sprintf("object_add %s,%s", objectType, strings.Join(paramsKvs, ","))
 	m.Query(cmd, callback)
 }
+
+func (m *HmpMonitor) BlockIoThrottle(driveName string, bps, iops int64, callback StringCallback) {
+	cmd := fmt.Sprintf("block_set_io_throttle %s %d 0 0 %d 0 0", driveName, bps, iops)
+	m.Query(cmd, callback)
+}

pkg/hostman/monitor/monitor.go

@@ -71,6 +71,7 @@ type Monitor interface {
 	StartNbdServer(port int, exportAllDevice, writable bool, callback StringCallback)
 
 	ResizeDisk(driveName string, sizeMB int64, callback StringCallback)
+	BlockIoThrottle(driveName string, bps, iops int64, callback StringCallback)
 }
 
 type MonitorErrorFunc func(error)

pkg/hostman/monitor/qmp.go

@@ -807,3 +807,8 @@ func (m *QmpMonitor) GeMemtSlotIndex(callback func(index int)) {
 	}
 	m.HumanMonitorCommand("info memory-devices", cb)
 }
+
+func (m *QmpMonitor) BlockIoThrottle(driveName string, bps, iops int64, callback StringCallback) {
+	cmd := fmt.Sprintf("block_set_io_throttle %s %d 0 0 %d 0 0", driveName, bps, iops)
+	m.HumanMonitorCommand(cmd, callback)
+}