waveterm/backup
0
0
Fork 0
mirror of https://github.com/wavetermdev/backup.git synced 2026-08-05 13:57:07 -07:00
Code Issues Packages Projects Releases Wiki Activity

Support running ephemeral commands (#543)

Browse Source 
* initial

* save work, starting to add backend types

* save work

* save work

* Add EphemeralWriteCloser

* Command pipes thru, triggers infinite loop

* save debugging

* fix bad merge

* save debug statements

* fixing spaghetti

* clean up code

* got cwd override working

* Add separate paths for stdout and stderr writers

* fix stdout/stderr

* env vars are now working

* revert waveshell changes

* Making EphemeralWriteCloser into a more generic BufferedPipe

* formatting

* comment

* delete unused package

* more naming changes

* add package comment

* add UsePty to EphemeralRunOpts

* document UsePty

* ensure only one downstream writer can read from the buffer

* store pointer to syncs

* remove inshellisense stuff for now

* remove debugs

* revert yarn

* remove unnecessary debugs in main-server

* more debugging removed

* revert tsconfig
This commit is contained in:
Evan Simkowitz
2024-04-02 15:46:32 -07:00
committed by GitHub
parent dc7fc2c823
commit ccc63937b6
13 changed files with 616 additions and 134 deletions
Download Patch File Download Diff File Expand all files Collapse all files
wavesrv/pkg/remote/remote.go
+85 -30
View File
@@ -19,7 +19,6 @@ import (
"strconv"
"strings"
"sync"
"sync/atomic"
"syscall"
"time"
@@ -35,6 +34,7 @@ import (
"github.com/wavetermdev/waveterm/waveshell/pkg/shexec"
"github.com/wavetermdev/waveterm/waveshell/pkg/statediff"
"github.com/wavetermdev/waveterm/waveshell/pkg/utilfn"
"github.com/wavetermdev/waveterm/wavesrv/pkg/ephemeral"
"github.com/wavetermdev/waveterm/wavesrv/pkg/scbase"
"github.com/wavetermdev/waveterm/wavesrv/pkg/scbus"
"github.com/wavetermdev/waveterm/wavesrv/pkg/scpacket"
@@ -172,13 +172,12 @@ type CommandInputSink interface {
}
type RunCmdType struct {
CK base.CommandKey
SessionId string
ScreenId string
RemotePtr sstore.RemotePtrType
RunPacket *packet.RunPacketType
Ephemeral bool
EphCancled atomic.Bool // only for Ephemeral commands, if true, then the command result should be discarded
CK base.CommandKey
SessionId string
ScreenId string
RemotePtr sstore.RemotePtrType
RunPacket *packet.RunPacketType
EphemeralOpts *ephemeral.EphemeralRunOpts
}
type ReinitCommandSink struct {
@@ -1892,8 +1891,8 @@ type RunCommandOpts struct {
NoCreateCmdPtyFile bool
// this command will not go into the DB, and will not have a ptyout file created
// forces special packet handling (sets RunCommandType.Ephemeral)
Ephemeral bool
// forces special packet handling (sets RunCommandType.EphemeralOpts)
EphemeralOpts *ephemeral.EphemeralRunOpts
}
// returns (CmdType, allow-updates-callback, err)
@@ -1924,6 +1923,10 @@ func RunCommand(ctx context.Context, rcOpts RunCommandOpts, runPacket *packet.Ru
return nil, nil, fmt.Errorf("runPacket.StatePtr should not be set, it is set in RunCommand")
}
if rcOpts.EphemeralOpts != nil {
log.Printf("[info] running ephemeral command ck: %s\n", runPacket.CK)
}
// pending state command logic
// if we are currently running a command that can change the state, we need to wait for it to finish
if rcOpts.StatePtr == nil {
@@ -1933,9 +1936,10 @@ func RunCommand(ctx context.Context, rcOpts RunCommandOpts, runPacket *packet.Ru
}
ok, existingRct := msh.testAndSetPendingStateCmd(screenId, remotePtr, newPSC)
if !ok {
if existingRct.Ephemeral {
if rcOpts.EphemeralOpts != nil {
// if the existing command is ephemeral, we cancel it and continue
existingRct.EphCancled.Store(true)
log.Printf("[warning] canceling existing ephemeral state cmd: %s\n", existingRct.CK)
rcOpts.EphemeralOpts.Canceled.Store(true)
} else {
line, _, err := sstore.GetLineCmdByLineId(ctx, screenId, existingRct.CK.GetCmdId())
return nil, nil, makePSCLineError(existingRct.CK, line, err)
@@ -1960,15 +1964,37 @@ func RunCommand(ctx context.Context, rcOpts RunCommandOpts, runPacket *packet.Ru
var err error
statePtr, err = sstore.GetRemoteStatePtr(ctx, sessionId, screenId, remotePtr)
if err != nil {
log.Printf("[error] RunCommand: cannot get remote state: %v\n", err)
return nil, nil, fmt.Errorf("cannot run command: %w", err)
}
if statePtr == nil {
log.Printf("[error] RunCommand: no valid shell state found\n")
return nil, nil, fmt.Errorf("cannot run command: no valid shell state found")
}
}
// statePtr will not be nil
runPacket.StatePtr = statePtr
currentState, err := sstore.GetFullState(ctx, *statePtr)
if rcOpts.EphemeralOpts != nil {
// Setting UsePty to false will ensure that the outputs get written to the correct file descriptors to extract stdout and stderr
runPacket.UsePty = rcOpts.EphemeralOpts.UsePty
// Ephemeral commands can override the cwd without persisting it to the DB
if rcOpts.EphemeralOpts.OverrideCwd != "" {
currentState.Cwd = rcOpts.EphemeralOpts.OverrideCwd
}
// Ephemeral commands can override the env without persisting it to the DB
if len(rcOpts.EphemeralOpts.Env) > 0 {
curEnvs := shellenv.DeclMapFromState(currentState)
for key, val := range rcOpts.EphemeralOpts.Env {
curEnvs[key] = &shellenv.DeclareDeclType{Name: key, Value: val, Args: "x"}
}
currentState.ShellVars = shellenv.SerializeDeclMap(curEnvs)
}
}
if err != nil || currentState == nil {
return nil, nil, fmt.Errorf("cannot load current remote state: %w", err)
}
@@ -2032,21 +2058,21 @@ func RunCommand(ctx context.Context, rcOpts RunCommandOpts, runPacket *packet.Ru
RunOut: nil,
RtnState: runPacket.ReturnState,
}
if !rcOpts.NoCreateCmdPtyFile && !rcOpts.Ephemeral {
if !rcOpts.NoCreateCmdPtyFile && rcOpts.EphemeralOpts == nil {
err = sstore.CreateCmdPtyFile(ctx, cmd.ScreenId, cmd.LineId, cmd.TermOpts.MaxPtySize)
if err != nil {
// TODO the cmd is running, so this is a tricky error to handle
return nil, nil, fmt.Errorf("cannot create local ptyout file for running command: %v", err)
}
}
msh.AddRunningCmd(&RunCmdType{
CK: runPacket.CK,
SessionId: sessionId,
ScreenId: screenId,
RemotePtr: remotePtr,
RunPacket: runPacket,
Ephemeral: rcOpts.Ephemeral,
})
runningCmdType := &RunCmdType{
CK: runPacket.CK,
SessionId: sessionId,
ScreenId: screenId,
RemotePtr: remotePtr,
RunPacket: runPacket,
EphemeralOpts: rcOpts.EphemeralOpts}
msh.AddRunningCmd(runningCmdType)
return cmd, func() { removeCmdWait(runPacket.CK) }, nil
}
@@ -2121,13 +2147,17 @@ func (msh *MShellProc) HandleFeInput(inputPk *scpacket.FeInputPacketType) error
func (msh *MShellProc) AddRunningCmd(rct *RunCmdType) {
msh.Lock.Lock()
defer msh.Lock.Unlock()
if rct.EphemeralOpts != nil {
log.Printf("[info] adding ephemeral running command: %s\n", rct.CK)
}
msh.RunningCmds[rct.RunPacket.CK] = rct
}
func (msh *MShellProc) GetRunningCmd(ck base.CommandKey) *RunCmdType {
msh.Lock.Lock()
defer msh.Lock.Unlock()
return msh.RunningCmds[ck]
rtn := msh.RunningCmds[ck]
return rtn
}
func (msh *MShellProc) RemoveRunningCmd(ck base.CommandKey) {
@@ -2319,14 +2349,15 @@ func (msh *MShellProc) handleCmdDonePacket(rct *RunCmdType, donePk *packet.CmdDo
}
// this will remove from RunningCmds and from PendingStateCmds
defer msh.RemoveRunningCmd(donePk.CK)
if rct.Ephemeral && rct.EphCancled.Load() {
if rct.EphemeralOpts != nil && rct.EphemeralOpts.Canceled.Load() {
log.Printf("cmddone %s (ephemeral canceled)\n", donePk.CK)
// do nothing when an ephemeral command is canceled
return
}
ctx, cancelFn := context.WithTimeout(context.Background(), 5*time.Second)
defer cancelFn()
update := scbus.MakeUpdatePacket()
if !rct.Ephemeral {
if rct.EphemeralOpts == nil {
// only update DB for non-ephemeral commands
err := sstore.UpdateCmdDoneInfo(ctx, update, donePk.CK, donePk, sstore.CmdStatusDone)
if err != nil {
@@ -2342,6 +2373,14 @@ func (msh *MShellProc) handleCmdDonePacket(rct *RunCmdType, donePk *packet.CmdDo
update.AddUpdate(*screen)
}
}
// Close the ephemeral response writer if it exists
if rct.EphemeralOpts != nil && rct.EphemeralOpts.ExpectsResponse {
log.Printf("closing ephemeral response writers\n")
defer rct.EphemeralOpts.StdoutWriter.Close()
defer rct.EphemeralOpts.StderrWriter.Close()
}
// ephemeral commands *do* update the remote state
// not all commands get a final state (only RtnState commands have this returned)
// so in those cases finalState will be nil
@@ -2360,7 +2399,7 @@ func (msh *MShellProc) handleCmdDonePacket(rct *RunCmdType, donePk *packet.CmdDo
update.AddUpdate(sstore.MakeSessionUpdateForRemote(rct.SessionId, newRI))
}
// ephemeral commands *do not* update cmd state (there is no command)
if newRI != nil && !rct.Ephemeral {
if newRI != nil && rct.EphemeralOpts == nil {
newRIStatePtr := packet.ShellStatePtr{BaseHash: newRI.StateBaseHash, DiffHashArr: newRI.StateDiffHashArr}
err = sstore.UpdateCmdRtnState(ctx, donePk.CK, newRIStatePtr)
if err != nil {
@@ -2378,10 +2417,6 @@ func (msh *MShellProc) handleCmdFinalPacket(rct *RunCmdType, finalPk *packet.Cmd
return
}
defer msh.RemoveRunningCmd(finalPk.CK)
if rct.Ephemeral {
// just remove the running command, but there is no DB state to update in this case
return
}
rtnCmd, err := sstore.GetCmdByScreenId(context.Background(), finalPk.CK.GetGroupId(), finalPk.CK.GetCmdId())
if err != nil {
log.Printf("error calling GetCmdById in handleCmdFinalPacket: %v\n", err)
@@ -2420,21 +2455,41 @@ func (msh *MShellProc) ResetDataPos(ck base.CommandKey) {
func (msh *MShellProc) handleDataPacket(rct *RunCmdType, dataPk *packet.DataPacketType, dataPosMap *utilfn.SyncMap[base.CommandKey, int64]) {
if rct == nil {
log.Printf("error handling data packet: no running cmd found %s\n", dataPk.CK)
ack := makeDataAckPacket(dataPk.CK, dataPk.FdNum, 0, fmt.Errorf("no running cmd found"))
msh.ServerProc.Input.SendPacket(ack)
return
}
realData, err := base64.StdEncoding.DecodeString(dataPk.Data64)
if err != nil {
log.Printf("error decoding data packet: %v\n", err)
ack := makeDataAckPacket(dataPk.CK, dataPk.FdNum, 0, err)
msh.ServerProc.Input.SendPacket(ack)
return
}
if rct.Ephemeral {
if rct.EphemeralOpts != nil {
// Write to the response writer if it's set
if len(realData) > 0 && rct.EphemeralOpts.ExpectsResponse {
switch dataPk.FdNum {
case 1:
_, err := rct.EphemeralOpts.StdoutWriter.Write(realData)
if err != nil {
log.Printf("*error writing to ephemeral stdout writer: %v\n", err)
}
case 2:
_, err := rct.EphemeralOpts.StderrWriter.Write(realData)
if err != nil {
log.Printf("*error writing to ephemeral stderr writer: %v\n", err)
}
default:
log.Printf("error handling data packet: invalid fdnum %d\n", dataPk.FdNum)
}
}
ack := makeDataAckPacket(dataPk.CK, dataPk.FdNum, len(realData), nil)
msh.ServerProc.Input.SendPacket(ack)
return
}
var ack *packet.DataAckPacketType
if len(realData) > 0 {
dataPos := dataPosMap.Get(dataPk.CK)