From 799b9b69ba08b898e13026b7ecab9f9f58580a82 Mon Sep 17 00:00:00 2001 From: Paul Buetow Date: Tue, 8 Dec 2020 14:49:41 +0000 Subject: merge develop --- internal/server/continuous.go | 2 +- internal/server/handlers/controlhandler.go | 28 +++-- internal/server/handlers/handler.go | 2 + internal/server/handlers/serverhandler.go | 182 +++++++---------------------- internal/server/scheduler.go | 2 +- internal/server/server.go | 44 +++---- 6 files changed, 78 insertions(+), 182 deletions(-) (limited to 'internal/server') diff --git a/internal/server/continuous.go b/internal/server/continuous.go index 583d136..f75c732 100644 --- a/internal/server/continuous.go +++ b/internal/server/continuous.go @@ -92,7 +92,7 @@ func (c *continuous) runJob(ctx context.Context, job config.Continuous) { } logger.Info(fmt.Sprintf("Starting job %s", job.Name)) - status := client.Start(jobCtx, make(chan struct{})) + status := client.Start(jobCtx, make(chan string)) logMessage := fmt.Sprintf("Job exited with status %d", status) if status != 0 { diff --git a/internal/server/handlers/controlhandler.go b/internal/server/handlers/controlhandler.go index daa9835..8cc5a40 100644 --- a/internal/server/handlers/controlhandler.go +++ b/internal/server/handlers/controlhandler.go @@ -1,20 +1,19 @@ package handlers import ( - "context" "fmt" "io" "os" "strings" + "github.com/mimecast/dtail/internal" "github.com/mimecast/dtail/internal/io/logger" user "github.com/mimecast/dtail/internal/user/server" ) // ControlHandler is used for control functions and health monitoring. type ControlHandler struct { - ctx context.Context - done chan struct{} + done *internal.Done hostname string payload []byte serverMessages chan string @@ -22,12 +21,11 @@ type ControlHandler struct { } // NewControlHandler returns a new control handler. -func NewControlHandler(ctx context.Context, user *user.User) (*ControlHandler, <-chan struct{}) { +func NewControlHandler(user *user.User) *ControlHandler { logger.Debug(user, "Creating control handler") h := ControlHandler{ - ctx: ctx, - done: make(chan struct{}), + done: internal.NewDone(), serverMessages: make(chan string, 10), user: user, } @@ -40,7 +38,17 @@ func NewControlHandler(ctx context.Context, user *user.User) (*ControlHandler, < s := strings.Split(fqdn, ".") h.hostname = s[0] - return &h, h.done + return &h +} + +// Shutdown the handler. +func (h *ControlHandler) Shutdown() { + h.done.Shutdown() +} + +// Done channel of the handler. +func (h *ControlHandler) Done() <-chan struct{} { + return h.done.Done() } // Read is to send data to the client via the Reader interface. @@ -51,7 +59,7 @@ func (h *ControlHandler) Read(p []byte) (n int, err error) { wholePayload := []byte(fmt.Sprintf("SERVER|%s|%s\n", h.hostname, message)) n = copy(p, wholePayload) return - case <-h.ctx.Done(): + case <-h.done.Done(): return 0, io.EOF } } @@ -63,7 +71,7 @@ func (h *ControlHandler) Write(p []byte) (n int, err error) { switch c { case ';': wholePayload := strings.TrimSpace(string(h.payload)) - h.handleCommand(h.ctx, wholePayload) + h.handleCommand(wholePayload) h.payload = nil default: @@ -75,7 +83,7 @@ func (h *ControlHandler) Write(p []byte) (n int, err error) { return } -func (h *ControlHandler) handleCommand(ctx context.Context, command string) { +func (h *ControlHandler) handleCommand(command string) { logger.Info(h.user, command) s := strings.Split(command, " ") logger.Debug(h.user, "Receiving command", command, s) diff --git a/internal/server/handlers/handler.go b/internal/server/handlers/handler.go index c42ceb9..b04e854 100644 --- a/internal/server/handlers/handler.go +++ b/internal/server/handlers/handler.go @@ -5,4 +5,6 @@ import "io" // Handler interface for server side functionality. type Handler interface { io.ReadWriter + Shutdown() + Done() <-chan struct{} } diff --git a/internal/server/handlers/serverhandler.go b/internal/server/handlers/serverhandler.go index 7017f3e..5cf8041 100644 --- a/internal/server/handlers/serverhandler.go +++ b/internal/server/handlers/serverhandler.go @@ -7,18 +7,16 @@ import ( "fmt" "io" "os" - "strconv" "strings" - "sync" "sync/atomic" "time" + "github.com/mimecast/dtail/internal" "github.com/mimecast/dtail/internal/config" "github.com/mimecast/dtail/internal/io/line" "github.com/mimecast/dtail/internal/io/logger" "github.com/mimecast/dtail/internal/mapr/server" "github.com/mimecast/dtail/internal/omode" - "github.com/mimecast/dtail/internal/server/background" user "github.com/mimecast/dtail/internal/user/server" "github.com/mimecast/dtail/internal/version" ) @@ -31,33 +29,27 @@ const ( // the Bi-directional communication between SSH client and server. // This handler implements the handler of the SSH server. type ServerHandler struct { - lines chan line.Line - regex string - aggregate *server.Aggregate - aggregatedMessages chan string - serverMessages chan string - payload []byte - hostname string - user *user.User - // TODO: Move all these channels into a separate struct for readability! + done *internal.Done + lines chan line.Line + regex string + aggregate *server.Aggregate + aggregatedMessages chan string + serverMessages chan string + payload []byte + hostname string + user *user.User catLimiter chan struct{} tailLimiter chan struct{} globalServerWaitFor chan struct{} ackCloseReceived chan struct{} - serverCtx context.Context - handlerCtx context.Context - done chan struct{} activeCommands int32 activeReaders int32 - background background.Background } // NewServerHandler returns the server handler. -func NewServerHandler(handlerCtx, serverCtx context.Context, user *user.User, catLimiter, tailLimiter, globalServerWaitFor chan struct{}, background background.Background) (*ServerHandler, <-chan struct{}) { +func NewServerHandler(user *user.User, catLimiter, tailLimiter, globalServerWaitFor chan struct{}) *ServerHandler { h := ServerHandler{ - serverCtx: serverCtx, - handlerCtx: handlerCtx, - done: make(chan struct{}), + done: internal.NewDone(), lines: make(chan line.Line, 100), serverMessages: make(chan string, 10), aggregatedMessages: make(chan string, 10), @@ -67,7 +59,6 @@ func NewServerHandler(handlerCtx, serverCtx context.Context, user *user.User, ca globalServerWaitFor: globalServerWaitFor, regex: ".", user: user, - background: background, } fqdn, err := os.Hostname() @@ -78,7 +69,17 @@ func NewServerHandler(handlerCtx, serverCtx context.Context, user *user.User, ca s := strings.Split(fqdn, ".") h.hostname = s[0] - return &h, h.done + return &h +} + +// Shutdown the handler. +func (h *ServerHandler) Shutdown() { + h.done.Shutdown() +} + +// Done channel of the handler. +func (h *ServerHandler) Done() <-chan struct{} { + return h.done.Done() } // Read is to send data to the dtail client via Reader interface. @@ -120,7 +121,7 @@ func (h *ServerHandler) Read(p []byte) (n int, err error) { case <-time.After(time.Second): // Once in a while check whether we are done. select { - case <-h.handlerCtx.Done(): + case <-h.done.Done(): return 0, io.EOF default: } @@ -134,7 +135,7 @@ func (h *ServerHandler) Write(p []byte) (n int, err error) { switch c { case ';': commandStr := strings.TrimSpace(string(h.payload)) - h.handleCommand(h.handlerCtx, commandStr) + h.handleCommand(commandStr) h.payload = nil default: h.payload = append(h.payload, c) @@ -145,9 +146,9 @@ func (h *ServerHandler) Write(p []byte) (n int, err error) { return } -func (h *ServerHandler) handleCommand(ctx context.Context, commandStr string) { +func (h *ServerHandler) handleCommand(commandStr string) { logger.Debug(h.user, commandStr) - var timeout time.Duration + ctx := context.Background() args, argc, err := h.handleProtocolVersion(strings.Split(commandStr, " ")) if err != nil { @@ -161,30 +162,18 @@ func (h *ServerHandler) handleCommand(ctx context.Context, commandStr string) { return } - args, argc, timeout, err = h.handleTimeout(args, argc) - if err != nil { - h.send(h.serverMessages, logger.Error(h.user, err)) - return - } - if h.user.Name == config.ControlUser { h.handleControlCommand(argc, args) return } - if timeout > 0 { - logger.Info(h.user, "Command with timeout context", argc, args, timeout) - commandCtx, cancel := context.WithTimeout(ctx, timeout) - go func() { - <-commandCtx.Done() - logger.Info(h.user, "Command timed out, canceling it", args, args, timeout) - cancel() - }() - h.handleUserCommand(commandCtx, argc, args, timeout) - return - } + ctx, cancel := context.WithCancel(ctx) + go func() { + <-h.done.Done() + cancel() + }() - h.handleUserCommand(ctx, argc, args, timeout) + h.handleUserCommand(ctx, argc, args) } func (h *ServerHandler) handleProtocolVersion(args []string) ([]string, int, error) { @@ -222,16 +211,6 @@ func (h *ServerHandler) handleBase64(args []string, argc int) ([]string, int, er return args, argc, nil } -func (h *ServerHandler) handleTimeout(args []string, argc int) ([]string, int, time.Duration, error) { - if argc <= 2 || args[0] != "timeout" { - // No timeout specified - return args, argc, time.Duration(0) * time.Second, nil - } - - timeout, err := strconv.Atoi(args[1]) - return args[2:], argc - 2, time.Duration(timeout) * time.Second, err -} - func (h *ServerHandler) handleControlCommand(argc int, args []string) { switch args[0] { case "debug": @@ -241,7 +220,7 @@ func (h *ServerHandler) handleControlCommand(argc int, args []string) { } } -func (h *ServerHandler) handleUserCommand(ctx context.Context, argc int, args []string, timeout time.Duration) { +func (h *ServerHandler) handleUserCommand(ctx context.Context, argc int, args []string) { logger.Debug(h.user, "handleUserCommand", argc, args) h.incrementActiveCommands() @@ -255,7 +234,7 @@ func (h *ServerHandler) handleUserCommand(ctx context.Context, argc int, args [] if h.aggregate == nil { return } - h.aggregate.Cancel() + h.aggregate.Shutdown() } } @@ -303,86 +282,6 @@ func (h *ServerHandler) handleUserCommand(ctx context.Context, argc int, args [] commandFinished() }() - case "run": - // TODO: Refactor this "run" case, move code to runcommand.go - command := newRunCommand(h) - - jobName, _ := options["jobName"] - logger.Debug(h.user, "run", options) - - if val, ok := options["background"]; ok && (val == "cancel" || val == "stop") { - if err := h.background.Cancel(h.user.Name, jobName); err != nil { - h.sendServerMessage(logger.Error(h.user, err, jobName, args)) - } else { - h.sendServerMessage(logger.Info(h.user, "job cancelled", jobName)) - } - commandFinished() - return - } - - if val, ok := options["background"]; ok && val == "list" { - h.sendServerMessage("Listing jobs") - count := 0 - for jobName := range h.background.ListJobsC(h.user.Name) { - h.sendServerMessage(jobName) - count++ - } - h.sendServerMessage(fmt.Sprintf("Found %d jobs", count)) - commandFinished() - return - } - - str, _ := options["outerArgs"] - outerArgs := strings.Split(str, " ") - - var background bool - if val, ok := options["background"]; ok && val == "start" { - background = true - } - - var wg sync.WaitGroup - wg.Add(1) - - if background { - if timeout == 0 { - // Set default background timeout. - timeout = time.Hour * 1 - } - // Use a new context based on the server context, so that background job does not get - // terminated when handler/SSH connection terminates. - commandCtx, cancel := context.WithTimeout(h.serverCtx, timeout) - - if err := h.background.Add(h.user.Name, jobName, cancel, &wg); err != nil { - h.sendServerMessage(logger.Error(h.user, err, jobName, args)) - commandFinished() - return - } - ctx = commandCtx - } - - if err := command.StartBackground(ctx, &wg, argc, args, outerArgs); err != nil { - h.sendServerMessage(logger.Error(h.user, "Unable to execute command", argc, args, err)) - commandFinished() - return - } - - // Make sure that server waits for all sub-processes to finish on shutdown - go func() { h.globalServerWaitFor <- struct{}{} }() - go func() { - wg.Wait() - <-h.globalServerWaitFor - }() - - if background { - h.sendServerMessage(logger.Info(h.user, jobName, "job started in background")) - commandFinished() - return - } - - // Command run in foreground, wait for it to complete before finishing the connection. - wg.Wait() - commandFinished() - case "ack", ".ack": h.handleAckCommand(argc, args) commandFinished() @@ -406,7 +305,7 @@ func (h *ServerHandler) handleAckCommand(argc int, args []string) { func (h *ServerHandler) send(ch chan<- string, message string) { select { case ch <- message: - case <-h.handlerCtx.Done(): + case <-h.done.Done(): } } @@ -447,7 +346,7 @@ func (h *ServerHandler) shutdown() { go func() { select { case h.serverMessageC() <- ".syn close connection": - case <-h.handlerCtx.Done(): + case <-h.done.Done(): } }() @@ -455,13 +354,10 @@ func (h *ServerHandler) shutdown() { case <-h.ackCloseReceived: case <-time.After(time.Second * 5): logger.Debug(h.user, "Shutdown timeout reached, enforcing shutdown") - case <-h.handlerCtx.Done(): + case <-h.done.Done(): } - select { - case h.done <- struct{}{}: - default: - } + h.done.Shutdown() } func (h *ServerHandler) incrementActiveCommands() { diff --git a/internal/server/scheduler.go b/internal/server/scheduler.go index 9d76a3b..a1e9e36 100644 --- a/internal/server/scheduler.go +++ b/internal/server/scheduler.go @@ -93,7 +93,7 @@ func (s *scheduler) runJobs(ctx context.Context) { defer cancel() logger.Info(fmt.Sprintf("Starting job %s", job.Name)) - status := client.Start(jobCtx, make(chan struct{})) + status := client.Start(jobCtx, make(chan string)) logMessage := fmt.Sprintf("Job exited with status %d", status) if status != 0 { diff --git a/internal/server/server.go b/internal/server/server.go index a446738..31fa85d 100644 --- a/internal/server/server.go +++ b/internal/server/server.go @@ -11,7 +11,6 @@ import ( "github.com/mimecast/dtail/internal/config" "github.com/mimecast/dtail/internal/io/logger" - "github.com/mimecast/dtail/internal/server/background" "github.com/mimecast/dtail/internal/server/handlers" "github.com/mimecast/dtail/internal/ssh/server" user "github.com/mimecast/dtail/internal/user/server" @@ -35,9 +34,8 @@ type Server struct { // Mointor log files for pattern (if configured) cont *continuous // Wait counter, e.g. there might be still subprocesses (forked by drun) to be killed. + // TODO: Remove this counter. shutdownWaitFor chan struct{} - // Background jobs - background background.Background } // New returns a new server. @@ -51,7 +49,6 @@ func New() *Server { shutdownWaitFor: make(chan struct{}, 1000), sched: newScheduler(), cont: newContinuous(), - background: background.New(), } s.sshServerConfig.PasswordCallback = s.Callback @@ -178,53 +175,46 @@ func (s *Server) handleRequests(ctx context.Context, sshConn gossh.Conn, in <-ch switch req.Type { case "shell": - handlerCtx, cancel := context.WithCancel(ctx) - var handler handlers.Handler - var done <-chan struct{} - switch user.Name { case config.ControlUser: - handler, done = handlers.NewControlHandler(handlerCtx, user) + handler = handlers.NewControlHandler(user) default: - handler, done = handlers.NewServerHandler(handlerCtx, ctx, user, s.catLimiter, s.tailLimiter, s.shutdownWaitFor, s.background) + handler = handlers.NewServerHandler(user, s.catLimiter, s.tailLimiter, s.shutdownWaitFor) } - go func() { - // Handler finished work, cancel all remaining routines - defer cancel() - - <-done - }() + terminate := func() { + handler.Shutdown() + sshConn.Close() + } go func() { // Broken pipe, cancel - defer cancel() - io.Copy(channel, handler) + terminate() }() go func() { // Broken pipe, cancel - defer cancel() - io.Copy(handler, channel) + terminate() }() go func() { - defer cancel() + select { + case <-ctx.Done(): + case <-handler.Done(): + } + terminate() + }() + go func() { if err := sshConn.Wait(); err != nil && err != io.EOF { logger.Error(user, err) } s.stats.decrementConnections() logger.Info(user, "Good bye Mister!") - }() - - go func() { - <-handlerCtx.Done() - sshConn.Close() - logger.Info(user, "Closed SSH connection") + terminate() }() // Only serving shell type -- cgit v1.2.3 From ab676c2b484225ed22765b23d8f0545088ecd610 Mon Sep 17 00:00:00 2001 From: Paul Buetow Date: Sat, 26 Dec 2020 10:48:51 +0000 Subject: code cleanup and minor refactorings --- internal/server/handlers/runcommand.go | 111 ------------------------------ internal/server/handlers/serverhandler.go | 50 +++++++------- internal/server/server.go | 24 +------ 3 files changed, 26 insertions(+), 159 deletions(-) delete mode 100644 internal/server/handlers/runcommand.go (limited to 'internal/server') diff --git a/internal/server/handlers/runcommand.go b/internal/server/handlers/runcommand.go deleted file mode 100644 index 8e5895b..0000000 --- a/internal/server/handlers/runcommand.go +++ /dev/null @@ -1,111 +0,0 @@ -package handlers - -import ( - "context" - "errors" - "fmt" - "io/ioutil" - "os" - "os/exec" - "strings" - "sync" - "time" - - "github.com/mimecast/dtail/internal/config" - "github.com/mimecast/dtail/internal/io/logger" - "github.com/mimecast/dtail/internal/io/run" -) - -type runCommand struct { - server *ServerHandler - run run.Run -} - -func newRunCommand(server *ServerHandler) runCommand { - return runCommand{ - server: server, - } -} - -func (r runCommand) StartBackground(ctx context.Context, wg *sync.WaitGroup, argc int, args, outerArgs []string) error { - if argc < 2 { - return fmt.Errorf("%s: args:%v argc:%d", commandParseWarning, args, argc) - } - - ec := make(chan int, 1) - var pid int - var err error - - command := strings.Join(args[1:], " ") - if strings.Contains(command, ";") || strings.Contains(command, "\n") { - if pid, err = r.startScript(ctx, wg, ec, command, outerArgs); err != nil { - r.server.sendServerMessage(".run exitstatus 255") - return err - } - return nil - } - - if pid, err = r.start(ctx, wg, ec, strings.TrimSpace(command), outerArgs); err != nil { - r.server.sendServerMessage(".run exitstatus 255") - return err - } - - exitCode := <-ec - r.server.sendServerMessage(fmt.Sprintf(".run exitstatus %d", exitCode)) - r.server.sendServerMessage(logger.Info(fmt.Sprintf("Process %d exited with status %d", pid, exitCode))) - - return nil -} - -func (r runCommand) startScript(ctx context.Context, wg *sync.WaitGroup, ec chan<- int, script string, outerArgs []string) (int, error) { - if _, err := os.Stat(config.Common.TmpDir); os.IsNotExist(err) { - return -1, err - } - - timestamp := time.Now().UnixNano() - scriptPath := fmt.Sprintf("%s/%s_%v.sh", config.Common.TmpDir, r.server.user.Name, timestamp) - - // TODO: On dserver startup delete all previously written scripts (there might be left overs due to a crash or so) - logger.Debug(r.server.user, "Writing temp script", scriptPath) - - script = fmt.Sprintf("#!/bin/sh\n%s", script) - if err := ioutil.WriteFile(scriptPath, []byte(script), 0700); err != nil { - return -1, err - } - - pid, err := r.start(ctx, wg, ec, scriptPath, outerArgs) - go func() { - wg.Wait() - logger.Debug("Deleting script", scriptPath) - os.Remove(scriptPath) - }() - - return pid, err -} - -func (r runCommand) start(ctx context.Context, wg *sync.WaitGroup, ec chan<- int, command string, outerArgs []string) (int, error) { - if len(command) == 0 { - return -1, errors.New("Empty command provided") - } - - splitted := strings.Split(command, " ") - path := splitted[0] - args := splitted[1:] - args = append(args, outerArgs...) - - qualifiedPath, err := exec.LookPath(path) - if err != nil { - return -1, err - } - - if !r.server.user.HasFilePermission(qualifiedPath, "runcommands") { - return -1, fmt.Errorf("No permission to execute path: %s", qualifiedPath) - } - - r.run = run.New(qualifiedPath, args) - pid, err := r.run.StartBackground(ctx, wg, ec, r.server.lines) - if err != nil { - return pid, err - } - return pid, nil -} diff --git a/internal/server/handlers/serverhandler.go b/internal/server/handlers/serverhandler.go index 5cf8041..3d1a53d 100644 --- a/internal/server/handlers/serverhandler.go +++ b/internal/server/handlers/serverhandler.go @@ -29,36 +29,34 @@ const ( // the Bi-directional communication between SSH client and server. // This handler implements the handler of the SSH server. type ServerHandler struct { - done *internal.Done - lines chan line.Line - regex string - aggregate *server.Aggregate - aggregatedMessages chan string - serverMessages chan string - payload []byte - hostname string - user *user.User - catLimiter chan struct{} - tailLimiter chan struct{} - globalServerWaitFor chan struct{} - ackCloseReceived chan struct{} - activeCommands int32 - activeReaders int32 + done *internal.Done + lines chan line.Line + regex string + aggregate *server.Aggregate + aggregatedMessages chan string + serverMessages chan string + payload []byte + hostname string + user *user.User + catLimiter chan struct{} + tailLimiter chan struct{} + ackCloseReceived chan struct{} + activeCommands int32 + activeReaders int32 } // NewServerHandler returns the server handler. -func NewServerHandler(user *user.User, catLimiter, tailLimiter, globalServerWaitFor chan struct{}) *ServerHandler { +func NewServerHandler(user *user.User, catLimiter, tailLimiter chan struct{}) *ServerHandler { h := ServerHandler{ - done: internal.NewDone(), - lines: make(chan line.Line, 100), - serverMessages: make(chan string, 10), - aggregatedMessages: make(chan string, 10), - ackCloseReceived: make(chan struct{}), - catLimiter: catLimiter, - tailLimiter: tailLimiter, - globalServerWaitFor: globalServerWaitFor, - regex: ".", - user: user, + done: internal.NewDone(), + lines: make(chan line.Line, 100), + serverMessages: make(chan string, 10), + aggregatedMessages: make(chan string, 10), + ackCloseReceived: make(chan struct{}), + catLimiter: catLimiter, + tailLimiter: tailLimiter, + regex: ".", + user: user, } fqdn, err := os.Hostname() diff --git a/internal/server/server.go b/internal/server/server.go index 31fa85d..a20737e 100644 --- a/internal/server/server.go +++ b/internal/server/server.go @@ -7,7 +7,6 @@ import ( "io" "net" "strings" - "time" "github.com/mimecast/dtail/internal/config" "github.com/mimecast/dtail/internal/io/logger" @@ -33,9 +32,6 @@ type Server struct { sched *scheduler // Mointor log files for pattern (if configured) cont *continuous - // Wait counter, e.g. there might be still subprocesses (forked by drun) to be killed. - // TODO: Remove this counter. - shutdownWaitFor chan struct{} } // New returns a new server. @@ -46,7 +42,6 @@ func New() *Server { sshServerConfig: &gossh.ServerConfig{}, catLimiter: make(chan struct{}, config.Server.MaxConcurrentCats), tailLimiter: make(chan struct{}, config.Server.MaxConcurrentTails), - shutdownWaitFor: make(chan struct{}, 1000), sched: newScheduler(), cont: newContinuous(), } @@ -80,27 +75,12 @@ func (s *Server) Start(ctx context.Context) int { go s.cont.start(ctx) go s.listenerLoop(ctx, listener) - select { - case <-ctx.Done(): - // Wait until all commands/jobs/children are no more! - s.wait() - } + <-ctx.Done() // For future use. return 0 } -func (s *Server) wait() { - for { - num := len(s.shutdownWaitFor) - logger.Debug("Waiting for stuff to finish", num) - if num <= 0 { - return - } - time.Sleep(time.Second) - } -} - func (s *Server) listenerLoop(ctx context.Context, listener net.Listener) { logger.Debug("Starting listener loop") @@ -180,7 +160,7 @@ func (s *Server) handleRequests(ctx context.Context, sshConn gossh.Conn, in <-ch case config.ControlUser: handler = handlers.NewControlHandler(user) default: - handler = handlers.NewServerHandler(user, s.catLimiter, s.tailLimiter, s.shutdownWaitFor) + handler = handlers.NewServerHandler(user, s.catLimiter, s.tailLimiter) } terminate := func() { -- cgit v1.2.3 From 60fa324cd5296b088c24d8db1d334a25ca955788 Mon Sep 17 00:00:00 2001 From: Paul Buetow Date: Sat, 26 Dec 2020 11:44:31 +0000 Subject: initial spartan mode support --- internal/server/handlers/serverhandler.go | 7 +++++++ 1 file changed, 7 insertions(+) (limited to 'internal/server') diff --git a/internal/server/handlers/serverhandler.go b/internal/server/handlers/serverhandler.go index 3d1a53d..5b948b3 100644 --- a/internal/server/handlers/serverhandler.go +++ b/internal/server/handlers/serverhandler.go @@ -43,6 +43,7 @@ type ServerHandler struct { ackCloseReceived chan struct{} activeCommands int32 activeReaders int32 + spartan bool } // NewServerHandler returns the server handler. @@ -245,6 +246,12 @@ func (h *ServerHandler) handleUserCommand(ctx context.Context, argc int, args [] commandFinished() return } + if spartan, ok := options["spartan"]; ok { + if spartan == "true" { + logger.Debug(h.user, "Enabling spartan mode") + h.spartan = true + } + } switch commandName { case "grep", "cat": -- cgit v1.2.3 From b05ae938aa6ada831d19de076861e246a03a9d7d Mon Sep 17 00:00:00 2001 From: Paul Buetow Date: Sat, 26 Dec 2020 14:00:41 +0000 Subject: rename spartan to quiet --- internal/server/handlers/serverhandler.go | 10 +++++----- 1 file changed, 5 insertions(+), 5 deletions(-) (limited to 'internal/server') diff --git a/internal/server/handlers/serverhandler.go b/internal/server/handlers/serverhandler.go index 5b948b3..3212ee1 100644 --- a/internal/server/handlers/serverhandler.go +++ b/internal/server/handlers/serverhandler.go @@ -43,7 +43,7 @@ type ServerHandler struct { ackCloseReceived chan struct{} activeCommands int32 activeReaders int32 - spartan bool + quiet bool } // NewServerHandler returns the server handler. @@ -246,10 +246,10 @@ func (h *ServerHandler) handleUserCommand(ctx context.Context, argc int, args [] commandFinished() return } - if spartan, ok := options["spartan"]; ok { - if spartan == "true" { - logger.Debug(h.user, "Enabling spartan mode") - h.spartan = true + if quiet, ok := options["quiet"]; ok { + if quiet == "true" { + logger.Debug(h.user, "Enabling quiet mode") + h.quiet = true } } -- cgit v1.2.3 From f9415259bc5d74a89801be75587da87dd37518af Mon Sep 17 00:00:00 2001 From: Paul Buetow Date: Sat, 26 Dec 2020 14:28:31 +0000 Subject: initial quiet switch --- internal/server/handlers/controlhandler.go | 4 +--- internal/server/handlers/readcommand.go | 11 +++++------ internal/server/handlers/serverhandler.go | 11 +++++++++-- 3 files changed, 15 insertions(+), 11 deletions(-) (limited to 'internal/server') diff --git a/internal/server/handlers/controlhandler.go b/internal/server/handlers/controlhandler.go index 8cc5a40..1e17c78 100644 --- a/internal/server/handlers/controlhandler.go +++ b/internal/server/handlers/controlhandler.go @@ -92,9 +92,7 @@ func (h *ControlHandler) handleCommand(command string) { case "health": h.serverMessages <- "OK: DTail SSH Server seems fine" h.serverMessages <- "done;" - case "debug": - h.serverMessages <- logger.Debug(h.user, "Receiving debug command", command, s) default: - h.serverMessages <- logger.Warn(h.user, "Received unknown control command", command, s) + h.serverMessages <- logger.Error(h.user, "Received unknown control command", command, s) } } diff --git a/internal/server/handlers/readcommand.go b/internal/server/handlers/readcommand.go index 0f9207d..5b8ce3a 100644 --- a/internal/server/handlers/readcommand.go +++ b/internal/server/handlers/readcommand.go @@ -31,14 +31,13 @@ func (r *readCommand) Start(ctx context.Context, argc int, args []string) { if argc >= 4 { deserializedRegex, err := regex.Deserialize(strings.Join(args[2:], " ")) if err != nil { - logger.Error(err) r.server.sendServerMessage(logger.Error(r.server.user, commandParseWarning, err)) return } re = deserializedRegex } if argc < 3 { - r.server.sendServerMessage(logger.Warn(r.server.user, commandParseWarning, args, argc)) + r.server.sendServerWarnMessage(logger.Warn(r.server.user, commandParseWarning, args, argc)) return } r.readGlob(ctx, args[1], re) @@ -52,7 +51,7 @@ func (r *readCommand) readGlob(ctx context.Context, glob string, re regex.Regex) for { maxRetries-- if maxRetries < 0 { - r.server.sendServerMessage(logger.Warn(r.server.user, "Giving up to read file(s)")) + r.server.sendServerWarnMessage(logger.Warn(r.server.user, "Giving up to read file(s)")) return } @@ -65,7 +64,7 @@ func (r *readCommand) readGlob(ctx context.Context, glob string, re regex.Regex) if numPaths := len(paths); numPaths == 0 { logger.Error(r.server.user, "No such file(s) to read", glob) - r.server.sendServerMessage(logger.Warn(r.server.user, "Unable to read file(s), check server logs")) + r.server.sendServerWarnMessage(logger.Warn(r.server.user, "Unable to read file(s), check server logs")) select { case <-ctx.Done(): return @@ -97,7 +96,7 @@ func (r *readCommand) readFileIfPermissions(ctx context.Context, wg *sync.WaitGr if !r.server.user.HasFilePermission(path, "readfiles") { logger.Error(r.server.user, "No permission to read file", path, globID) - r.server.sendServerMessage(logger.Warn(r.server.user, "Unable to read file(s), check server logs")) + r.server.sendServerWarnMessage(logger.Warn(r.server.user, "Unable to read file(s), check server logs")) return } @@ -161,6 +160,6 @@ func (r *readCommand) makeGlobID(path, glob string) string { return pathParts[len(pathParts)-1] } - r.server.sendServerMessage(logger.Error("Empty file path given?", path, glob)) + r.server.sendServerWarnMessage(logger.Warn("Empty file path given?", path, glob)) return "" } diff --git a/internal/server/handlers/serverhandler.go b/internal/server/handlers/serverhandler.go index 3212ee1..681598c 100644 --- a/internal/server/handlers/serverhandler.go +++ b/internal/server/handlers/serverhandler.go @@ -43,7 +43,7 @@ type ServerHandler struct { ackCloseReceived chan struct{} activeCommands int32 activeReaders int32 - quiet bool + quiet bool } // NewServerHandler returns the server handler. @@ -299,7 +299,7 @@ func (h *ServerHandler) handleUserCommand(ctx context.Context, argc int, args [] func (h *ServerHandler) handleAckCommand(argc int, args []string) { if argc < 3 { - h.sendServerMessage(logger.Warn(h.user, commandParseWarning, args, argc)) + h.sendServerWarnMessage(logger.Warn(h.user, commandParseWarning, args, argc)) return } if args[1] == "close" && args[2] == "connection" { @@ -318,6 +318,13 @@ func (h *ServerHandler) sendServerMessage(message string) { h.send(h.serverMessageC(), message) } +func (h *ServerHandler) sendServerWarnMessage(message string) { + if h.quiet { + return + } + h.send(h.serverMessageC(), message) +} + func (h *ServerHandler) serverMessageC() chan<- string { return h.serverMessages } -- cgit v1.2.3 From 94e37105c5a8c0ce22104add751e9938f239261e Mon Sep 17 00:00:00 2001 From: Paul Buetow Date: Sun, 27 Dec 2020 16:31:03 +0000 Subject: only try to read a file once in cat and grep mode but 10 times in tail mode --- internal/server/handlers/readcommand.go | 20 ++++++++------------ internal/server/handlers/serverhandler.go | 4 ++-- 2 files changed, 10 insertions(+), 14 deletions(-) (limited to 'internal/server') diff --git a/internal/server/handlers/readcommand.go b/internal/server/handlers/readcommand.go index 5b8ce3a..5bab26f 100644 --- a/internal/server/handlers/readcommand.go +++ b/internal/server/handlers/readcommand.go @@ -25,7 +25,7 @@ func newReadCommand(server *ServerHandler, mode omode.Mode) *readCommand { } } -func (r *readCommand) Start(ctx context.Context, argc int, args []string) { +func (r *readCommand) Start(ctx context.Context, argc int, args []string, retries int) { re := regex.NewNoop() if argc >= 4 { @@ -40,21 +40,14 @@ func (r *readCommand) Start(ctx context.Context, argc int, args []string) { r.server.sendServerWarnMessage(logger.Warn(r.server.user, commandParseWarning, args, argc)) return } - r.readGlob(ctx, args[1], re) + r.readGlob(ctx, args[1], re, retries) } -func (r *readCommand) readGlob(ctx context.Context, glob string, re regex.Regex) { +func (r *readCommand) readGlob(ctx context.Context, glob string, re regex.Regex, retries int) { retryInterval := time.Second * 5 glob = filepath.Clean(glob) - maxRetries := 10 - for { - maxRetries-- - if maxRetries < 0 { - r.server.sendServerWarnMessage(logger.Warn(r.server.user, "Giving up to read file(s)")) - return - } - + for retryCount := 0; retryCount < retries; retryCount++ { paths, err := filepath.Glob(glob) if err != nil { logger.Warn(r.server.user, glob, err) @@ -75,8 +68,11 @@ func (r *readCommand) readGlob(ctx context.Context, glob string, re regex.Regex) } r.readFiles(ctx, paths, glob, re, retryInterval) - break + return } + + r.server.sendServerWarnMessage(logger.Warn(r.server.user, "Giving up to read file(s)")) + return } func (r *readCommand) readFiles(ctx context.Context, paths []string, glob string, re regex.Regex, retryInterval time.Duration) { diff --git a/internal/server/handlers/serverhandler.go b/internal/server/handlers/serverhandler.go index 681598c..185e7c2 100644 --- a/internal/server/handlers/serverhandler.go +++ b/internal/server/handlers/serverhandler.go @@ -258,7 +258,7 @@ func (h *ServerHandler) handleUserCommand(ctx context.Context, argc int, args [] command := newReadCommand(h, omode.CatClient) go func() { h.incrementActiveReaders() - command.Start(ctx, argc, args) + command.Start(ctx, argc, args, 1) readerFinished() commandFinished() }() @@ -267,7 +267,7 @@ func (h *ServerHandler) handleUserCommand(ctx context.Context, argc int, args [] command := newReadCommand(h, omode.TailClient) go func() { h.incrementActiveReaders() - command.Start(ctx, argc, args) + command.Start(ctx, argc, args, 10) readerFinished() commandFinished() }() -- cgit v1.2.3