feat(llm): restart action

Author: zexi

cmd/climc/shell/llm/llm.go

@@ -16,6 +16,7 @@ func init() {
 	cmd.Perform("syncstatus", new(options.LLMIdOptions))
 	cmd.BatchPerform("stop", new(options.LLMStopOptions))
 	cmd.BatchPerform("start", new(options.LLMStartOptions))
+	cmd.BatchPerform("restart", new(options.LLMRestartOptions))
 	cmd.Get("probed-models", new(options.LLMIdOptions))
 	cmd.Get("url", new(options.LLMIdOptions))
 	cmd.Custom(shell.CustomActionGet, "available-network", new(options.LLMAvailableNetworkOptions))

pkg/apis/llm/llm.go

@@ -175,3 +175,30 @@ func (info LLMMountDirInfo) ToOverlay() apis.ContainerVolumeMountDiskPostOverlay
 
 type LLMSyncStatusInput struct {
 }
+
+type LLMRestartInput struct {
+}
+
+type LLMRestartTaskInput struct {
+	LLMId         string
+	ResetDataDisk bool
+	LLMStatus     string
+	SkuId         string
+	ImageId       string
+	BackupName    string
+	Property      []string
+
+	RebindVolumeId string
+	OnlyStop       bool
+}
+
+type LLMChangeNetworkInput struct {
+	BandwidthMb   int      `json:"bandwidth_mb"`
+	WhitePrefxies []string `json:"white_prefxies"`
+}
+
+type LLMVolumeInput struct {
+	LLMId     string `json:"llm_id"`
+	VolumeId  string `json:"volume_id"`
+	AutoStart bool   `json:"auto_start"`
+}

pkg/apis/llm/llm_const.go

@@ -33,15 +33,39 @@ const (
 	/* 运行 */
 	LLM_STATUS_RUNNING = "running"
 
+	/* 开始解绑磁盘 */
+	LLM_STATUS_START_UNBIND = "start_unbind"
+	/* 正在解绑磁盘 */
+	LLM_STATUS_UNBINDING = "unbinding"
+	/* 解绑磁盘失败 */
+	LLM_STATUS_UNBIND_FAIL = "unbind_fail"
+
+	/* 开始挂载磁盘 */
+	LLM_STATUS_START_BIND = "start_bind"
+	/* 正在挂载磁盘 */
+	LLM_STATUS_BINDING = "binding"
+	/* 挂载磁盘失败 */
+	LLM_STATUS_BIND_FAIL = "bind_fail"
+
+	/* 开始重启 */
+	LLM_STATUS_START_RESTART = "start_restart"
+	/* 正在重启 */
+	LLM_STATUS_RESTARTING = "restarting"
+	/* 重启失败 */
+	LLM_STATUS_RESTART_FAILED = "restart_fail"
+
+	/* 开始删除 */
+	LLM_STATUS_START_DELETE = "start_delete"
+	/* 正在删除 */
+	LLM_STATUS_DELETING = "deleting"
+	/* 删除失败 */
+	LLM_STATUS_DELETE_FAILED = "delete_fail"
+
 	/* 删除 */
 	LLM_STATUS_DELETED = "deleted"
 
 	LLM_LLM_STATUS_NO_SERVER    = "no_server"
 	LLM_LLM_STATUS_NO_CONTAINER = "no_container"
-
-	LLM_STATUS_START_DELETE  = "start_delete"
-	LLM_STATUS_DELETING      = "deleting"
-	LLM_STATUS_DELETE_FAILED = "delete_fail"
 )
 
 type TQuickModelMethod string

pkg/apis/notify/const.go

@@ -139,6 +139,7 @@ const (
 	TOPIC_RESOURCE_VM_INTEGRITY_CHECK       = "vm_integrity"
 	TOPIC_RESOURCE_PROJECT                  = "project"
 	TOPIC_RESOURCE_CLOUDPHONE               = "cloudphone"
+	TOPIC_RESOURCE_LLM                      = "llm"
 
 	SUBSCRIBER_TYPE_ROLE     = "role"
 	SUBSCRIBER_TYPE_ROBOT    = "robot"

pkg/llm/drivers/llm_container/base_driver.go

@@ -5,6 +5,7 @@ import (
 
 	"yunion.io/x/pkg/errors"
 
+	computeapi "yunion.io/x/onecloud/pkg/apis/compute"
 	api "yunion.io/x/onecloud/pkg/apis/llm"
 	"yunion.io/x/onecloud/pkg/cloudcommon/validators"
 	"yunion.io/x/onecloud/pkg/httperrors"
@@ -94,3 +95,17 @@ func (b *baseDriver) ValidateUpdateData(ctx context.Context, userCred mcclient.T
 	input.MountedModels = mountedModels
 	return input, nil
 }
+
+func MatchContainerToUpdateByName(ctr *computeapi.SContainer, podCtrs []*computeapi.PodContainerCreateInput) (*computeapi.PodContainerCreateInput, error) {
+	ctrName := ctr.Name
+	for _, podCtr := range podCtrs {
+		if podCtr.Name == ctrName {
+			return podCtr, nil
+		}
+	}
+	return nil, errors.Wrapf(errors.ErrNotFound, "container %s not found", ctrName)
+}
+
+func (b *baseDriver) MatchContainerToUpdate(ctr *computeapi.SContainer, podCtrs []*computeapi.PodContainerCreateInput) (*computeapi.PodContainerCreateInput, error) {
+	return MatchContainerToUpdateByName(ctr, podCtrs)
+}

pkg/llm/drivers/llm_container/dify.go

@@ -172,6 +172,10 @@ func (d *dify) GetContainerSpecs(ctx context.Context, llm *models.SLLM, image *m
 	return models.GetDifyContainersByNameAndSku(llm.GetName(), sku, nil, spec.(*api.LLMSpecDify))
 }
 
+func (d *dify) MatchContainerToUpdate(ctr *computeapi.SContainer, podCtrs []*computeapi.PodContainerCreateInput) (*computeapi.PodContainerCreateInput, error) {
+	return MatchContainerToUpdateByName(ctr, podCtrs)
+}
+
 // StartLLM is a no-op for Dify; all services are started by their container entrypoints.
 func (d *dify) StartLLM(ctx context.Context, userCred mcclient.TokenCredential, llm *models.SLLM) error {
 	return nil

pkg/llm/models/llm.go

@@ -15,14 +15,18 @@ import (
 	commonapi "yunion.io/x/onecloud/pkg/apis"
 	computeapi "yunion.io/x/onecloud/pkg/apis/compute"
 	api "yunion.io/x/onecloud/pkg/apis/llm"
+	"yunion.io/x/onecloud/pkg/apis/notify"
+	notifyapi "yunion.io/x/onecloud/pkg/apis/notify"
 	"yunion.io/x/onecloud/pkg/cloudcommon/db"
 	"yunion.io/x/onecloud/pkg/cloudcommon/db/taskman"
+	"yunion.io/x/onecloud/pkg/cloudcommon/notifyclient"
 	"yunion.io/x/onecloud/pkg/httperrors"
 	"yunion.io/x/onecloud/pkg/llm/options"
 	llmutils "yunion.io/x/onecloud/pkg/llm/utils"
 	"yunion.io/x/onecloud/pkg/mcclient"
 	"yunion.io/x/onecloud/pkg/mcclient/auth"
 	"yunion.io/x/onecloud/pkg/mcclient/modules/compute"
+	baseoptions "yunion.io/x/onecloud/pkg/mcclient/options"
 	computeoptions "yunion.io/x/onecloud/pkg/mcclient/options/compute"
 	"yunion.io/x/onecloud/pkg/util/stringutils2"
 )
@@ -478,10 +482,31 @@ func (llm *SLLM) PerformStart(ctx context.Context, userCred mcclient.TokenCreden
 		return nil, errors.Wrapf(errors.ErrInvalidStatus, "llm id: %s status: %s", llm.Id, llm.Status)
 	}
 
-	if err := llm.StartStartTask(ctx, userCred, ""); err != nil {
-		return nil, errors.Wrap(err, "StartStartTask")
+	_, err := llm.GetVolume()
+	if err != nil {
+		if errors.Cause(err) == sql.ErrNoRows {
+			return nil, errors.Wrapf(errors.ErrNotSupported, "llm id: %s missing volume", llm.Id)
+		}
+		return nil, errors.Wrap(err, "GetVolume")
+	}
+	taskinput := &api.LLMRestartTaskInput{
+		LLMId:     llm.Id,
+		LLMStatus: api.LLM_STATUS_READY,
+	}
+	_, err = llm.StartRestartTask(ctx, userCred, taskinput, "")
+	if err != nil {
+		return nil, errors.Wrap(err, "StartRestartTask")
+	}
+	return nil, nil
+}
+
+func (d *SLLM) StartStartTaskInternal(ctx context.Context, userCred mcclient.TokenCredential, parentTaskId string) error {
+	d.SetStatus(ctx, userCred, computeapi.VM_STARTING, "")
+	task, err := taskman.TaskManager.NewTask(ctx, "LLMStartTask", d, userCred, nil, parentTaskId, "", nil)
+	if err != nil {
+		return errors.Wrapf(err, "NewTask")
 	}
-	return jsonutils.Marshal(nil), nil
+	return task.ScheduleRun(nil)
 }
 
 func (llm *SLLM) StartStartTask(ctx context.Context, userCred mcclient.TokenCredential, parentTaskId string) error {
@@ -504,6 +529,96 @@ func (llm *SLLM) PerformStop(ctx context.Context, userCred mcclient.TokenCredent
 	return nil, nil
 }
 
+func (llm *SLLM) ValidateRestartInput(ctx context.Context, userCred mcclient.TokenCredential, input *api.LLMRestartInput) (*api.LLMRestartTaskInput, error) {
+	if len(llm.CmpId) == 0 {
+		return nil, errors.Wrap(errors.ErrInvalidStatus, "empty cmp_id")
+	}
+
+	srv, err := llm.GetServer(ctx)
+	if err != nil {
+		return nil, errors.Wrap(err, "GetServer")
+	}
+
+	if (llm.Status != api.LLM_STATUS_READY && llm.Status != api.LLM_STATUS_RUNNING) || (srv.Status != computeapi.VM_READY && !utils.IsInArray(srv.Status, computeapi.VM_RUNNING_STATUS)) {
+		return nil, errors.Wrapf(errors.ErrInvalidStatus, "invalid llm status %s", llm.Status)
+	}
+
+	return &api.LLMRestartTaskInput{}, nil
+}
+
+func (llm *SLLM) PerformRestart(ctx context.Context, userCred mcclient.TokenCredential, query jsonutils.JSONObject, input *api.LLMRestartInput) (jsonutils.JSONObject, error) {
+	taskInput, err := llm.ValidateRestartInput(ctx, userCred, input)
+	if err != nil {
+		return nil, errors.Wrap(err, "ValidateRestartInput")
+	}
+	_, err = llm.StartRestartTask(ctx, userCred, taskInput, "")
+	if err != nil {
+		return nil, errors.Wrap(err, "StartRestartTask")
+	}
+	return nil, nil
+}
+
+func (llm *SLLM) StartRestartTask(ctx context.Context, userCred mcclient.TokenCredential, params *api.LLMRestartTaskInput, parentTaskId string) (*taskman.STask, error) {
+	key := "perform_restart"
+	if params.ResetDataDisk {
+		key = "perform_reset"
+	}
+	llm.SetStatus(ctx, userCred, api.LLM_STATUS_START_RESTART, key)
+	taskName := "LLMRestartTask"
+	if params.ResetDataDisk {
+		taskName = "LLMResetTask"
+	}
+	params.LLMId = llm.Id
+	task, err := taskman.TaskManager.NewTask(ctx, taskName, llm, userCred, jsonutils.Marshal(params).(*jsonutils.JSONDict), parentTaskId, "", nil)
+	if err != nil {
+		return nil, errors.Wrap(err, "NewTask")
+	}
+	if err := task.ScheduleRun(nil); err != nil {
+		return nil, errors.Wrap(err, "ScheduleRun")
+	}
+	return task, nil
+}
+
+func (llm *SLLM) PerformReset(ctx context.Context, userCred mcclient.TokenCredential, query jsonutils.JSONObject, input *api.LLMRestartInput) (jsonutils.JSONObject, error) {
+	taskInput, err := llm.ValidateRestartInput(ctx, userCred, input)
+	if err != nil {
+		return nil, errors.Wrap(err, "ValidateRestartInput")
+	}
+	_, err = llm.StartResetTask(ctx, userCred, taskInput, "")
+	if err != nil {
+		return nil, errors.Wrap(err, "StartRestartTask")
+	}
+	return nil, nil
+}
+
+func (llm *SLLM) StartResetTask(ctx context.Context, userCred mcclient.TokenCredential, params *api.LLMRestartTaskInput, parentTaskId string) (*taskman.STask, error) {
+	llm.SetStatus(ctx, userCred, api.LLM_STATUS_START_RESTART, "perform_reset")
+	task, err := taskman.TaskManager.NewTask(ctx, "LLMResetTask", llm, userCred, jsonutils.Marshal(params).(*jsonutils.JSONDict), parentTaskId, "", nil)
+	if err != nil {
+		return nil, errors.Wrapf(err, "NewTask")
+	}
+	if err := task.ScheduleRun(nil); err != nil {
+		return nil, errors.Wrap(err, "ScheduleRun")
+	}
+	return task, nil
+}
+
+func (llm *SLLM) NotifyRequest(ctx context.Context, userCred mcclient.TokenCredential, action notify.SAction, model jsonutils.JSONObject, success bool) {
+	obj := func(ctx context.Context, details *jsonutils.JSONDict) {}
+	if model != nil {
+		obj = func(ctx context.Context, details *jsonutils.JSONDict) {
+			details.Set("customize_details", model)
+		}
+	}
+	notifyclient.EventNotify(ctx, userCred, notifyclient.SEventNotifyParam{
+		Obj:                 llm,
+		Action:              action,
+		ObjDetailsDecorator: obj,
+		IsFail:              !success,
+		ResourceType:        notifyapi.TOPIC_RESOURCE_LLM,
+	})
+}
+
 func (llm *SLLM) StartLLMStopTask(ctx context.Context, userCred mcclient.TokenCredential, parentTaskId string) error {
 	task, err := taskman.TaskManager.NewTask(ctx, "LLMStopTask", llm, userCred, nil, parentTaskId, "", nil)
 	if err != nil {
@@ -628,3 +743,106 @@ func (man *SLLMManager) GetAvailableNetwork(ctx context.Context, userCred mcclie
 
 	return ret, nil
 }
+
+func (llm *SLLM) StartBindVolumeTask(ctx context.Context, userCred mcclient.TokenCredential, volumeId string, autoStart bool, parenentTaskId string) (*taskman.STask, error) {
+	llm.SetStatus(ctx, userCred, api.LLM_STATUS_START_BIND, "perform bind volume")
+	params := api.LLMVolumeInput{
+		LLMId:     llm.Id,
+		VolumeId:  volumeId,
+		AutoStart: autoStart,
+	}
+	task, err := taskman.TaskManager.NewTask(ctx, "LLMAttachTask", llm, userCred, jsonutils.Marshal(params).(*jsonutils.JSONDict), parenentTaskId, "", nil)
+	if err != nil {
+		return nil, errors.Wrap(err, "NewTask")
+	}
+	err = task.ScheduleRun(nil)
+	if err != nil {
+		return nil, errors.Wrap(err, "ScheduleRun")
+	}
+	return task, nil
+}
+
+func (llm *SLLM) ChangeServerNetworkConfig(ctx context.Context, bandwidth int, whitePrefixes []string, noSync bool) error {
+	s := auth.GetAdminSession(ctx, options.Options.Region)
+	params := baseoptions.BaseListOptions{}
+	params.Scope = "max"
+	limit := 0
+	params.Limit = &limit
+	serverNicObjs, err := compute.Servernetworks.ListDescendent(s, llm.CmpId, jsonutils.Marshal(params))
+	if err != nil {
+		return errors.Wrap(err, "compute.Servernetworks.ListDescendent")
+	} else if len(serverNicObjs.Data) == 0 {
+		return errors.Wrap(httperrors.ErrEmptyRequest, "compute.Servernetworks.ListDescendent")
+	}
+	gns := computeapi.GuestnetworkDetails{}
+	err = serverNicObjs.Data[0].Unmarshal(&gns)
+	if err != nil {
+		return errors.Wrap(err, "Unmarshal GuestnetworkDetails")
+	}
+	if gns.BwLimit != bandwidth {
+		// need to change bandwidth
+		params := computeapi.ServerChangeBandwidthInput{}
+		params.Mac = gns.MacAddr
+		params.Index = 0
+		params.Bandwidth = bandwidth
+		params.NoSync = &noSync
+		_, err := compute.Servers.PerformAction(s, llm.CmpId, "change-bandwidth", jsonutils.Marshal(params))
+		if err != nil {
+			return errors.Wrap(err, "compute.Servers.PerformAction change-bandwidth")
+		}
+	}
+	/*if len(adbWhitePrefixes) > 0 {
+		for _, pm := range gns.PortMappings {
+			if pm.Port == apis.PHONE_ADB_PORT {
+				// verify adb port remote ips
+				remoteIps := stringutils2.NewSortedStrings(pm.RemoteIps)
+				remoteIps2 := stringutils2.NewSortedStrings(adbWhitePrefixes)
+				if !stringutils2.Equals(remoteIps, remoteIps2) {
+					// need to update remote Ips
+					params := computeapi.GuestnetworkUpdateInput{}
+					for i := range gns.PortMappings {
+						npm := gns.PortMappings[i]
+						if gns.PortMappings[i].Port == apis.PHONE_ADB_PORT {
+							npm.RemoteIps = adbWhitePrefixes
+						}
+						params.PortMappings = append(params.PortMappings, npm)
+					}
+					_, err := compute.Servernetworks.Update(s, gns.GuestId, gns.NetworkId, nil, jsonutils.Marshal(params))
+					if err != nil {
+						return errors.Wrap(err, "Servernetworks.Update")
+					}
+				}
+				break
+			}
+		}
+	}*/
+	return nil
+}
+
+func (llm *SLLM) PerformNetConfig(
+	ctx context.Context,
+	userCred mcclient.TokenCredential,
+	query jsonutils.JSONObject,
+	input api.LLMChangeNetworkInput,
+) (jsonutils.JSONObject, error) {
+	err := llm.ChangeServerNetworkConfig(ctx, input.BandwidthMb, input.WhitePrefxies, false)
+	if err != nil {
+		return nil, errors.Wrap(err, "changeServerNetworkConfig")
+	}
+
+	if llm.BandwidthMb != input.BandwidthMb {
+		_, err := db.Update(llm, func() error {
+			llm.BandwidthMb = input.BandwidthMb
+			return nil
+		})
+		if err != nil {
+			return nil, errors.Wrap(err, "update")
+		}
+	}
+
+	return nil, nil
+}
+
+func (llm *SLLM) purgeModelList() error {
+	return GetLLMInstantModelManager().purgeModelList(llm.Id)
+}