1.add http deployment method

Signed-off-by: liuminjian <[email protected]>

http/core/core.go

@@ -75,3 +75,12 @@ func ExitSuccessWithData(r *pigeon.Request, data interface{}) bool {
 	})
 	return r.Exit(200)
 }
+
+func ExitFailWithData(r *pigeon.Request, data interface{}, message string) bool {
+	r.SendJSON(pigeon.JSON{
+		"errorCode": "503",
+		"errorMsg":  message,
+		"data":      data,
+	})
+	return r.Exit(503)
+}

http/manager/bind.go

@@ -22,7 +22,10 @@
 
 package manager
 
-import "github.com/opencurve/pigeon"
+import (
+	"github.com/opencurve/pigeon"
+	"mime/multipart"
+)
 
 var METHOD_REQUEST map[string]Request
 
@@ -87,6 +90,19 @@ type DeployClusterRequest struct{}
 
 type GetClusterServicesAddrRequest struct{}
 
+type DeployClusterCmdRequest struct {
+	Command string `json:"command" binding:"required"`
+}
+
+type DeployClusterUploadRequest struct {
+	FilePath string                `json:"filepath" form:"filepath" binding:"required"`
+	File     *multipart.FileHeader `form:"file" binding:"required"`
+}
+
+type DeployClusterDownloadRequest struct {
+	FilePath string `json:"filepath" form:"filepath" binding:"required"`
+}
+
 var requests = []Request{
 	{
 		"GET",
@@ -166,4 +182,22 @@ var requests = []Request{
 		GetClusterServicesAddrRequest{},
 		GetClusterServicesAddr,
 	},
+	{
+		"POST",
+		"cluster.deploy.cmd",
+		DeployClusterCmdRequest{},
+		DeployClusterCmd,
+	},
+	{
+		"POST",
+		"cluster.deploy.upload",
+		DeployClusterUploadRequest{},
+		DeployClusterUpload,
+	},
+	{
+		"GET",
+		"cluster.deploy.download",
+		DeployClusterDownloadRequest{},
+		DeployClusterDownload,
+	},
 }

http/manager/manager.go

@@ -24,6 +24,9 @@ package manager
 
 import (
 	"fmt"
+	"github.com/opencurve/curveadm/internal/utils"
+	"io"
+	"os/exec"
 	"strings"
 
 	"github.com/opencurve/curveadm/cli/cli"
@@ -313,3 +316,48 @@ func GetClusterServicesAddr(r *pigeon.Request, ctx *Context) bool {
 	servicesAddr.Addrs = getServicesAddrFromConf(dcs, mcs)
 	return core.ExitSuccessWithData(r, servicesAddr)
 }
+
+func DeployClusterCmd(r *pigeon.Request, ctx *Context) bool {
+	data := ctx.Data.(*DeployClusterCmdRequest)
+	r.Logger().Info("DeployClusterCmd", pigeon.Field("command", data.Command))
+	cmd := exec.Command("/bin/bash", "-c", data.Command)
+	out, err := cmd.CombinedOutput()
+	if err != nil {
+		r.Logger().Warn("DeployClusterCmd failed when execute command",
+			pigeon.Field("error", err))
+		return core.ExitFailWithData(r, string(out), string(out))
+	}
+	r.Logger().Info("DeployClusterCmd", pigeon.Field("result", out))
+	return core.ExitSuccessWithData(r, string(out))
+}
+
+func DeployClusterUpload(r *pigeon.Request, ctx *Context) bool {
+	data := ctx.Data.(*DeployClusterUploadRequest)
+	r.Logger().Info("DeployClusterUpload", pigeon.Field("file", data.FilePath))
+	mf, err := data.File.Open()
+	if err != nil {
+		r.Logger().Warn("DeployClusterUpload failed when open file",
+			pigeon.Field("error", err))
+		return core.ExitFailWithData(r, err.Error(), err.Error())
+	}
+	defer mf.Close()
+	content, err := io.ReadAll(mf)
+	if err != nil {
+		r.Logger().Warn("DeployClusterUpload failed when read file",
+			pigeon.Field("error", err))
+		return core.ExitFailWithData(r, err.Error(), err.Error())
+	}
+	err = utils.WriteFile(data.FilePath, string(content), 0644)
+	if err != nil {
+		r.Logger().Warn("DeployClusterUpload failed when write file",
+			pigeon.Field("error", err))
+		return core.ExitFailWithData(r, err.Error(), err.Error())
+	}
+	return core.Exit(r, err)
+}
+
+func DeployClusterDownload(r *pigeon.Request, ctx *Context) bool {
+	data := ctx.Data.(*DeployClusterDownloadRequest)
+	r.Logger().Info("DeployClusterDownload", pigeon.Field("file", data.FilePath))
+	return r.SendFile(data.FilePath)
+}

internal/configure/hosts/hc_get.go

@@ -77,7 +77,13 @@ func (hc *HostConfig) GetForwardAgent() bool     { return hc.getBool(CONFIG_FORW
 func (hc *HostConfig) GetBecomeUser() string     { return hc.getString(CONFIG_BECOME_USER) }
 func (hc *HostConfig) GetLabels() []string       { return hc.labels }
 func (hc *HostConfig) GetEnvs() []string         { return hc.envs }
+func (hc *HostConfig) GetProtocol() string       { return hc.getString(CONFIG_PROTOCOL) }
 func (hc *HostConfig) GetSSHConfig() *module.SSHConfig {
+
+	if hc.GetProtocol() != SSH_PROTOCOL {
+		return nil
+	}
+
 	hostname := hc.GetSSHHostname()
 	if len(hostname) == 0 {
 		hostname = hc.GetHostname()
@@ -95,3 +101,17 @@ func (hc *HostConfig) GetSSHConfig() *module.SSHConfig {
 		ConnectRetries:    curveadm.GlobalCurveAdmConfig.GetSSHRetries(),
 	}
 }
+
+func (hc *HostConfig) GetHttpConfig() *module.HttpConfig {
+
+	if hc.GetProtocol() != HTTP_PROTOCOL {
+		return nil
+	}
+
+	return &module.HttpConfig{
+		Host: hc.GetHostname(),
+		Port: (uint)(hc.GetHTTPPort()),
+	}
+}
+
+func (hc *HostConfig) GetHTTPPort() int { return hc.getInt(CONFIG_HTTP_PORT) }

internal/configure/hosts/hc_item.go

@@ -32,7 +32,10 @@ import (
 )
 
 const (
-	DEFAULT_SSH_PORT = 22
+	DEFAULT_SSH_PORT  = 22
+	DEFAULT_HTTP_PORT = 8000
+	SSH_PROTOCOL      = "ssh"
+	HTTP_PROTOCOL     = "http"
 )
 
 var (
@@ -97,4 +100,18 @@ var (
 		false,
 		nil,
 	)
+
+	CONFIG_PROTOCOL = itemset.Insert(
+		"protocol",
+		comm.REQUIRE_STRING,
+		false,
+		SSH_PROTOCOL,
+	)
+
+	CONFIG_HTTP_PORT = itemset.Insert(
+		"http_port",
+		comm.REQUIRE_POSITIVE_INTEGER,
+		false,
+		DEFAULT_HTTP_PORT,
+	)
 )

internal/task/context/context.go

@@ -29,27 +29,27 @@ import (
 )
 
 type Context struct {
-	sshClient *module.SSHClient
-	module    *module.Module
-	register  *Register
+	remoteClient module.RemoteClient
+	module       *module.Module
+	register     *Register
 }
 
-func NewContext(sshClient *module.SSHClient) (*Context, error) {
+func NewContext(remoteClient module.RemoteClient) (*Context, error) {
 	return &Context{
-		sshClient: sshClient,
-		module:    module.NewModule(sshClient),
-		register:  NewRegister(),
+		remoteClient: remoteClient,
+		module:       module.NewModule(remoteClient),
+		register:     NewRegister(),
 	}, nil
 }
 
 func (ctx *Context) Close() {
-	if ctx.sshClient != nil {
-		ctx.sshClient.Client().Close()
+	if ctx.remoteClient != nil {
+		ctx.remoteClient.Close()
 	}
 }
 
-func (ctx *Context) SSHClient() *module.SSHClient {
-	return ctx.sshClient
+func (ctx *Context) RemoteClient() module.RemoteClient {
+	return ctx.remoteClient
 }
 
 func (ctx *Context) Module() *module.Module {

internal/task/step/shell.go

@@ -567,18 +567,20 @@ func (s *Scp) Execute(ctx *context.Context) error {
 		return errno.ERR_WRITE_FILE_FAILED.E(err)
 	}
 
-	config := ctx.SSHClient().Config()
-	cmd := ctx.Module().Shell().Scp(localPath, config.User, config.Host, s.RemotePath)
-	cmd.AddOption("-P %d", config.Port)
-	if !config.ForwardAgent {
-		cmd.AddOption("-i %s", config.PrivateKeyPath)
-	}
+	//config := ctx.SSHClient().Config()
+	//cmd := ctx.Module().Shell().Scp(localPath, config.User, config.Host, s.RemotePath)
+	//cmd.AddOption("-P %d", config.Port)
+	//if !config.ForwardAgent {
+	//	cmd.AddOption("-i %s", config.PrivateKeyPath)
+	//}
+
+	err = ctx.Module().File().Upload(localPath, s.RemotePath)
 
 	options := s.ExecOptions
 	options.ExecWithSudo = false
 	options.ExecInLocal = true
-	out, err := cmd.Execute(options)
-	return PostHandle(nil, nil, out, err, errno.ERR_SECURE_COPY_FILE_TO_REMOTE_FAILED)
+	//out, err := cmd.Execute(options)
+	return PostHandle(nil, nil, "", err, errno.ERR_SECURE_COPY_FILE_TO_REMOTE_FAILED)
 }
 
 func (s *Command) Execute(ctx *context.Context) error {

internal/task/task/bs/add_target.go

@@ -54,7 +54,7 @@ func NewAddTargetTask(curveadm *cli.CurveAdm, cc *configure.ClientConfig) (*task
 	}
 
 	subname := fmt.Sprintf("host=%s volume=%s", options.Host, volume)
-	t := task.NewTask("Add Target", subname, hc.GetSSHConfig())
+	t := task.NewTask("Add Target", subname, hc.GetSSHConfig(), hc.GetHttpConfig())
 
 	// add step
 	var output string

internal/task/task/bs/balance_leader.go

@@ -48,7 +48,7 @@ func NewBalanceTask(curveadm *cli.CurveAdm, dc *topology.DeployConfig) (*task.Ta
 
 	subname := fmt.Sprintf("host=%s role=%s containerId=%s",
 		dc.GetHost(), dc.GetRole(), tui.TrimContainerId(containerId))
-	t := task.NewTask("Balance Leader", subname, hc.GetSSHConfig())
+	t := task.NewTask("Balance Leader", subname, hc.GetSSHConfig(), hc.GetHttpConfig())
 
 	// add step
 	t.AddStep(&step.ContainerExec{

internal/task/task/bs/create_volume.go

@@ -74,7 +74,7 @@ func NewCreateVolumeTask(curveadm *cli.CurveAdm, cc *configure.ClientConfig) (*t
 	}
 
 	subname := fmt.Sprintf("hostname=%s image=%s", hc.GetHostname(), cc.GetContainerImage())
-	t := task.NewTask("Create Volume", subname, hc.GetSSHConfig())
+	t := task.NewTask("Create Volume", subname, hc.GetSSHConfig(), hc.GetHttpConfig())
 
 	// add step
 	var out string

internal/task/task/bs/delete_target.go

@@ -58,7 +58,7 @@ func NewDeleteTargetTask(curveadm *cli.CurveAdm, cc *client.ClientConfig) (*task
 	}
 
 	subname := fmt.Sprintf("hostname=%s tid=%s", hc.GetHostname(), options.Tid)
-	t := task.NewTask("Delete Target", subname, hc.GetSSHConfig())
+	t := task.NewTask("Delete Target", subname, hc.GetSSHConfig(), hc.GetHttpConfig())
 
 	// add step
 	var output string

internal/task/task/bs/detect_release.go

@@ -75,7 +75,7 @@ func NewDetectOSReleaseTask(curveadm *cli.CurveAdm, v interface{}) (*task.Task,
 	var success bool
 	var out string
 	subname := fmt.Sprintf("host=%s", host)
-	t := task.NewTask("Detect OS Release", subname, hc.GetSSHConfig())
+	t := task.NewTask("Detect OS Release", subname, hc.GetSSHConfig(), hc.GetHttpConfig())
 
 	// add step to task
 	t.AddStep(&step.Cat{

internal/task/task/bs/format.go

@@ -237,7 +237,7 @@ func NewFormatChunkfilePoolTask(curveadm *cli.CurveAdm, fc *configure.FormatConf
 	usagePercent := fc.GetFormatPercent()
 	subname := fmt.Sprintf("host=%s device=%s mountPoint=%s usage=%d%%",
 		fc.GetHost(), device, mountPoint, usagePercent)
-	t := task.NewTask("Start Format Chunkfile Pool", subname, hc.GetSSHConfig())
+	t := task.NewTask("Start Format Chunkfile Pool", subname, hc.GetSSHConfig(), hc.GetHttpConfig())
 
 	// add step to task
 	var oldContainerId, containerId, oldUUID string

internal/task/task/bs/format_clean.go

@@ -78,7 +78,7 @@ func NewCleanFormatTask(curveadm *cli.CurveAdm, fc *configure.FormatConfig) (*ta
 	containerName := device2ContainerName(device)
 	subname := fmt.Sprintf("host=%s device=%s mountPoint=%s containerName=%s",
 		fc.GetHost(), device, mountPoint, containerName)
-	t := task.NewTask("Clean Format Container", subname, hc.GetSSHConfig())
+	t := task.NewTask("Clean Format Container", subname, hc.GetSSHConfig(), hc.GetHttpConfig())
 
 	// add step to task
 	var out string

internal/task/task/bs/format_status.go

@@ -122,7 +122,7 @@ func NewGetFormatStatusTask(curveadm *cli.CurveAdm, fc *configure.FormatConfig)
 	// new task
 	device := fc.GetDevice()
 	subname := fmt.Sprintf("host=%s device=%s", fc.GetHost(), fc.GetDevice())
-	t := task.NewTask("Get Format Status", subname, hc.GetSSHConfig())
+	t := task.NewTask("Get Format Status", subname, hc.GetSSHConfig(), hc.GetHttpConfig())
 
 	// add step to task
 	var deviceUsage, containerStatus string

internal/task/task/bs/format_stop.go

@@ -79,7 +79,7 @@ func NewStopFormatTask(curveadm *cli.CurveAdm, fc *configure.FormatConfig) (*tas
 	containerName := device2ContainerName(device)
 	subname := fmt.Sprintf("host=%s device=%s mountPoint=%s containerName=%s",
 		fc.GetHost(), device, mountPoint, containerName)
-	t := task.NewTask("Stop Format Chunkfile Pool", subname, hc.GetSSHConfig())
+	t := task.NewTask("Stop Format Chunkfile Pool", subname, hc.GetSSHConfig(), hc.GetHttpConfig())
 
 	var oldContainerId string
 	var oldUUID string

internal/task/task/bs/install_polarfs.go

@@ -104,7 +104,7 @@ func NewInstallPolarFSTask(curveadm *cli.CurveAdm, cc *configure.ClientConfig) (
 	// new task
 	release := getRelease(curveadm)
 	subname := fmt.Sprintf("host=%s release=%s", host, release)
-	t := task.NewTask("Install PolarFS", subname, hc.GetSSHConfig())
+	t := task.NewTask("Install PolarFS", subname, hc.GetSSHConfig(), hc.GetHttpConfig())
 
 	// add step to task
 	var input, output string

internal/task/task/bs/list_targets.go

@@ -111,7 +111,7 @@ func NewListTargetsTask(curveadm *cli.CurveAdm, v interface{}) (*task.Task, erro
 	}
 
 	subname := fmt.Sprintf("host=%s", hc.GetHostname())
-	t := task.NewTask("List Targets", subname, hc.GetSSHConfig())
+	t := task.NewTask("List Targets", subname, hc.GetSSHConfig(), hc.GetHttpConfig())
 
 	// add step
 	var output string

internal/task/task/bs/map.go

@@ -79,7 +79,7 @@ func NewMapTask(curveadm *cli.CurveAdm, cc *configure.ClientConfig) (*task.Task,
 	}
 
 	subname := fmt.Sprintf("hostname=%s volume=%s:%s", hc.GetHostname(), options.User, options.Volume)
-	t := task.NewTask("Map Volume", subname, hc.GetSSHConfig())
+	t := task.NewTask("Map Volume", subname, hc.GetSSHConfig(), hc.GetHttpConfig())
 
 	// add step
 	var out string

internal/task/task/bs/start_nebd.go

@@ -146,7 +146,7 @@ func NewStartNEBDServiceTask(curveadm *cli.CurveAdm, cc *configure.ClientConfig)
 	}
 
 	subname := fmt.Sprintf("hostname=%s image=%s", hc.GetHostname(), cc.GetContainerImage())
-	t := task.NewTask("Start NEBD Service", subname, hc.GetSSHConfig())
+	t := task.NewTask("Start NEBD Service", subname, hc.GetSSHConfig(), hc.GetHttpConfig())
 
 	// add step
 	var containerId, out string

internal/task/task/bs/start_tgtd.go

@@ -66,7 +66,7 @@ func NewStartTargetDaemonTask(curveadm *cli.CurveAdm, cc *configure.ClientConfig
 
 	// new task
 	subname := fmt.Sprintf("host=%s image=%s", options.Host, cc.GetContainerImage())
-	t := task.NewTask("Start Target Daemon", subname, hc.GetSSHConfig())
+	t := task.NewTask("Start Target Daemon", subname, hc.GetSSHConfig(), hc.GetHttpConfig())
 
 	// add step to task
 	var status, containerId, out string

internal/task/task/bs/stop_tgtd.go

@@ -50,7 +50,7 @@ func NewStopTargetDaemonTask(curveadm *cli.CurveAdm, v interface{}) (*task.Task,
 
 	// new task
 	subname := fmt.Sprintf("host=%s", options.Host)
-	t := task.NewTask("Stop Target Daemon", subname, hc.GetSSHConfig())
+	t := task.NewTask("Stop Target Daemon", subname, hc.GetSSHConfig(), hc.GetHttpConfig())
 
 	// add step
 	var containerId string