From 2755be315d06f06f425867c70b39e185d2a934c9 Mon Sep 17 00:00:00 2001 From: sawka Date: Tue, 5 Jul 2022 16:54:49 -0700 Subject: [PATCH] checkpoint for migrating to remoteinstance --- cmd/main-server.go | 28 ++++-------- db/migrations/000001_init.down.sql | 2 +- db/migrations/000001_init.up.sql | 11 +++-- pkg/remote/remote.go | 70 +++++++++++++++++++++-------- pkg/sstore/dbops.go | 4 +- pkg/sstore/sstore.go | 71 +++++++++++++++++++++--------- 6 files changed, 121 insertions(+), 65 deletions(-) diff --git a/cmd/main-server.go b/cmd/main-server.go index a1884287..2e2f2cca 100644 --- a/cmd/main-server.go +++ b/cmd/main-server.go @@ -377,6 +377,7 @@ func WriteJsonSuccess(w http.ResponseWriter, data interface{}) { type runCommandResponse struct { Line *sstore.LineType `json:"line"` + Cmd *sstore.CmdType `json:"cmd"` } func HandleRunCommand(w http.ResponseWriter, r *http.Request) { @@ -400,16 +401,16 @@ func HandleRunCommand(w http.ResponseWriter, r *http.Request) { WriteJsonError(w, fmt.Errorf("invalid sessionid '%s': %w", commandPk.SessionId, err)) return } - line, err := ProcessFeCommandPacket(r.Context(), &commandPk) + resp, err := ProcessFeCommandPacket(r.Context(), &commandPk) if err != nil { WriteJsonError(w, err) return } - WriteJsonSuccess(w, &runCommandResponse{Line: line}) + WriteJsonSuccess(w, resp) return } -func ProcessFeCommandPacket(ctx context.Context, pk *scpacket.FeCommandPacketType) (*sstore.LineType, error) { +func ProcessFeCommandPacket(ctx context.Context, pk *scpacket.FeCommandPacketType) (*runCommandResponse, error) { commandStr := strings.TrimSpace(pk.CmdStr) if commandStr == "" { return nil, fmt.Errorf("invalid emtpty command") @@ -420,14 +421,14 @@ func ProcessFeCommandPacket(ctx context.Context, pk *scpacket.FeCommandPacketTyp if err != nil { return nil, err } - return rtnLine, nil + return &runCommandResponse{Line: rtnLine}, nil } if strings.HasPrefix(commandStr, "cd ") { newDir := strings.TrimSpace(commandStr[3:]) cdPacket := packet.MakeCdPacket() - cdPacket.PacketId = uuid.New().String() + cdPacket.ReqId = uuid.New().String() cdPacket.Dir = newDir - localRemote := remote.GetRemote("local") + localRemote := remote.GetRemoteById(pk.RemoteState.RemoteId) if localRemote != nil { localRemote.Input.SendPacket(cdPacket) } @@ -437,19 +438,8 @@ func ProcessFeCommandPacket(ctx context.Context, pk *scpacket.FeCommandPacketTyp if err != nil { return nil, err } - runPacket := packet.MakeRunPacket() - runPacket.CK = base.MakeCommandKey(pk.SessionId, rtnLine.CmdId) - runPacket.Cwd = pk.RemoteState.Cwd - runPacket.Env = nil - runPacket.Command = commandStr - fmt.Printf("run-packet %v\n", runPacket) - go func() { - localRemote := remote.GetRemote("local") - if localRemote != nil { - localRemote.Input.SendPacket(runPacket) - } - }() - return rtnLine, nil + err = remote.RunCommand(pk, rtnLine.CmdId) + return &runCommandResponse{Line: rtnLine}, nil } // /api/start-session diff --git a/db/migrations/000001_init.down.sql b/db/migrations/000001_init.down.sql index b9110317..6887fccd 100644 --- a/db/migrations/000001_init.down.sql +++ b/db/migrations/000001_init.down.sql @@ -1,6 +1,6 @@ DROP TABLE session; DROP TABLE window; -DROP TABLE session_remote; +DROP TABLE remote_instance; DROP TABLE line; DROP TABLE remote; DROP TABLE session_cmd; diff --git a/db/migrations/000001_init.up.sql b/db/migrations/000001_init.up.sql index 622a3b6f..2ad0b2ec 100644 --- a/db/migrations/000001_init.up.sql +++ b/db/migrations/000001_init.up.sql @@ -14,13 +14,14 @@ CREATE TABLE window ( ); CREATE UNIQUE INDEX window_name_unique ON window(sessionid, name); -CREATE TABLE session_remote ( +CREATE TABLE remote_instance ( + riid varchar(36) PRIMARY KEY, + name varchar(50) NOT NULL, sessionid varchar(36) NOT NULL, windowid varchar(36) NOT NULL, - remotename varchar(50) NOT NULL, remoteid varchar(36) NOT NULL, - cwd varchar(300) NOT NULL, - PRIMARY KEY (sessionid, windowid, remotename) + sessionscope boolean NOT NULL, + state json NOT NULL ); CREATE TABLE line ( @@ -55,7 +56,9 @@ CREATE TABLE remote ( CREATE TABLE session_cmd ( sessionid varchar(36) NOT NULL, cmdid varchar(36) NOT NULL, + rsid varchar(36) NOT NULL, remoteid varchar(36) NOT NULL, + remotestate json NOT NULL, status varchar(10) NOT NULL, startts bigint NOT NULL, pid int NOT NULL, diff --git a/pkg/remote/remote.go b/pkg/remote/remote.go index 98634dcd..e9896958 100644 --- a/pkg/remote/remote.go +++ b/pkg/remote/remote.go @@ -6,12 +6,14 @@ import ( "io" "os" "os/exec" + "strings" "sync" "time" "github.com/scripthaus-dev/mshell/pkg/base" "github.com/scripthaus-dev/mshell/pkg/packet" "github.com/scripthaus-dev/mshell/pkg/shexec" + "github.com/scripthaus-dev/sh2-server/pkg/scpacket" "github.com/scripthaus-dev/sh2-server/pkg/sstore" ) @@ -29,15 +31,15 @@ var GlobalStore *Store type Store struct { Lock *sync.Mutex - Map map[string]*MShellProc + Map map[string]*MShellProc // key=remoteid } type RemoteState struct { - RemoteType string `json:"remotetype"` - RemoteId string `json:"remoteid"` - RemoteName string `json:"remotename"` - Status string `json:"status"` - Cwd string `json:"cwd"` + RemoteType string `json:"remotetype"` + RemoteId string `json:"remoteid"` + RemoteName string `json:"remotename"` + Status string `json:"status"` + DefaultState *sstore.RemoteState `json:"defaultstate"` } type MShellProc struct { @@ -57,8 +59,8 @@ type MShellProc struct { } type RpcEntry struct { - PacketId string - RespCh chan packet.RpcPacketType + ReqId string + RespCh chan packet.RpcResponsePacketType } func LoadRemotes(ctx context.Context) error { @@ -72,7 +74,7 @@ func LoadRemotes(ctx context.Context) error { } for _, remote := range allRemotes { msh := MakeMShell(remote) - GlobalStore.Map[remote.RemoteName] = msh + GlobalStore.Map[remote.RemoteId] = msh if remote.AutoConnect { go msh.Launch() } @@ -80,10 +82,21 @@ func LoadRemotes(ctx context.Context) error { return nil } -func GetRemote(name string) *MShellProc { +func GetRemoteByName(name string) *MShellProc { GlobalStore.Lock.Lock() defer GlobalStore.Lock.Unlock() - return GlobalStore.Map[name] + for _, msh := range GlobalStore.Map { + if msh.Remote.RemoteName == name { + return msh + } + } + return nil +} + +func GetRemoteById(remoteId string) *MShellProc { + GlobalStore.Lock.Lock() + defer GlobalStore.Lock.Unlock() + return GlobalStore.Map[remoteId] } func GetAllRemoteState() []RemoteState { @@ -99,7 +112,7 @@ func GetAllRemoteState() []RemoteState { Status: proc.Status, } if proc.InitPk != nil { - state.Cwd = proc.InitPk.HomeDir + state.DefaultState = &sstore.RemoteState{Cwd: proc.InitPk.HomeDir} } rtn = append(rtn, state) } @@ -177,17 +190,37 @@ func (msh *MShellProc) IsConnected() bool { return msh.Status == StatusConnected } -func (runner *MShellProc) PacketRpc(pk packet.RpcPacketType, timeout time.Duration) (packet.RpcPacketType, error) { +func RunCommand(pk *scpacket.FeCommandPacketType, cmdId string) error { + msh := GetRemoteById(pk.RemoteState.RemoteId) + if msh == nil { + return fmt.Errorf("no remote id=%s found", pk.RemoteState.RemoteId) + } + if !msh.IsConnected() { + return fmt.Errorf("remote '%s' is not connected", msh.Remote.RemoteName) + } + runPacket := packet.MakeRunPacket() + runPacket.CK = base.MakeCommandKey(pk.SessionId, cmdId) + runPacket.Cwd = pk.RemoteState.Cwd + runPacket.Env = nil + runPacket.Command = strings.TrimSpace(pk.CmdStr) + fmt.Printf("run-packet %v\n", runPacket) + go func() { + msh.Input.SendPacket(runPacket) + }() + return nil +} + +func (runner *MShellProc) PacketRpc(pk packet.RpcPacketType, timeout time.Duration) (packet.RpcResponsePacketType, error) { if !runner.IsConnected() { return nil, fmt.Errorf("runner is not connected") } if pk == nil { return nil, fmt.Errorf("PacketRpc passed nil packet") } - id := pk.GetPacketId() - respCh := make(chan packet.RpcPacketType) + id := pk.GetReqId() + respCh := make(chan packet.RpcResponsePacketType) runner.WithLock(func() { - runner.RpcMap[id] = &RpcEntry{PacketId: id, RespCh: respCh} + runner.RpcMap[id] = &RpcEntry{ReqId: id, RespCh: respCh} }) defer runner.WithLock(func() { delete(runner.RpcMap, id) @@ -217,8 +250,9 @@ func (runner *MShellProc) ProcessPackets() { } }) for pk := range runner.Output.MainCh { - if rpcPk, ok := pk.(packet.RpcPacketType); ok { - rpcId := rpcPk.GetPacketId() + fmt.Printf("MSH> %s\n", packet.AsString(pk)) + if rpcPk, ok := pk.(packet.RpcResponsePacketType); ok { + rpcId := rpcPk.GetResponseId() runner.WithLock(func() { entry := runner.RpcMap[rpcId] if entry == nil { diff --git a/pkg/sstore/dbops.go b/pkg/sstore/dbops.go index 1becc4a7..71b13d0c 100644 --- a/pkg/sstore/dbops.go +++ b/pkg/sstore/dbops.go @@ -108,7 +108,7 @@ func GetSessionById(ctx context.Context, id string) (*SessionType, error) { rtnSession = &session query = `SELECT sessionid, windowid, name, curremote, version FROM window WHERE sessionid = ?` tx.SelectWrap(&session.Windows, query, session.SessionId) - query = `SELECT * FROM session_remote WHERE sessionid = ?` + query = `SELECT * FROM remote_instance WHERE sessionid = ?` tx.SelectWrap(&session.Remotes, query, session.SessionId) return nil }) @@ -149,7 +149,7 @@ func GetWindowLines(ctx context.Context, sessionId string, windowId string) ([]* return lines, nil } -// also creates window, and sessionremote +// also creates window func InsertSessionWithName(ctx context.Context, sessionName string) error { if sessionName == "" { return fmt.Errorf("invalid session name '%s'", sessionName) diff --git a/pkg/sstore/sstore.go b/pkg/sstore/sstore.go index e0cfb2d3..967f8dc4 100644 --- a/pkg/sstore/sstore.go +++ b/pkg/sstore/sstore.go @@ -2,6 +2,9 @@ package sstore import ( "context" + "database/sql/driver" + "encoding/json" + "fmt" "log" "path" "sync" @@ -48,11 +51,11 @@ func GetDB() (*sqlx.DB, error) { } type SessionType struct { - SessionId string `json:"sessionid"` - Name string `json:"name"` - Windows []*WindowType `json:"windows"` - Cmds []*CmdType `json:"cmds"` - Remotes []*SessionRemote `json:"remotes"` + SessionId string `json:"sessionid"` + Name string `json:"name"` + Windows []*WindowType `json:"windows"` + Cmds []*CmdType `json:"cmds"` + Remotes []*RemoteInstance `json:"remotes"` } type WindowType struct { @@ -64,12 +67,36 @@ type WindowType struct { Version int `json:"version"` } -type SessionRemote struct { - SessionId string `json:"sessionid"` - WindowId string `json:"windowid"` - RemoteId string `json"remoteid"` - RemoteName string `json:"name"` - Cwd string `json:"cwd"` +type RemoteState struct { + Cwd string `json:"cwd"` +} + +func (s *RemoteState) Scan(val interface{}) error { + if strVal, ok := val.(string); ok { + if strVal == "" { + return nil + } + err := json.Unmarshal([]byte(strVal), s) + if err != nil { + return err + } + return nil + } + return fmt.Errorf("cannot scan '%T' into RemoteState", val) +} + +func (s *RemoteState) Value() (driver.Value, error) { + return json.Marshal(s) +} + +type RemoteInstance struct { + RIId string `json:"riid"` + Name string `json:"name"` + SessionId string `json:"sessionid"` + WindowId string `json:"windowid"` + RemoteId string `json"remoteid"` + SessionScope bool `json:"sessionscope"` + State RemoteState `json:"state"` } type LineType struct { @@ -101,16 +128,18 @@ type RemoteType struct { } type CmdType struct { - RowId int64 `json:"rowid"` - SessionId string `json:"sessionid"` - CmdId string `json:"cmdid"` - RemoteId string `json:"remoteid"` - Status string `json:"status"` - StartTs int64 `json:"startts"` - DoneTs int64 `json:"donets"` - Pid int `json:"pid"` - RunnerPid int `json:"runnerpid"` - ExitCode int `json:"exitcode"` + RowId int64 `json:"rowid"` + SessionId string `json:"sessionid"` + CmdId string `json:"cmdid"` + RSId string `json:"rsid"` + RemoteId string `json:"remoteid"` + RemoteState string `json:"remotestate"` + Status string `json:"status"` + StartTs int64 `json:"startts"` + DoneTs int64 `json:"donets"` + Pid int `json:"pid"` + RunnerPid int `json:"runnerpid"` + ExitCode int `json:"exitcode"` RunOut packet.PacketType `json:"runout"` }