From c2522ffb59514443816a96386c16bb7527cbe57c Mon Sep 17 00:00:00 2001 From: Paul Buetow Date: Sat, 21 Aug 2021 14:54:24 +0300 Subject: read files bytewise for more control of whats happening - change transport protocol for more control over newlines --- internal/server/handlers/serverhandler.go | 15 +++++++++------ 1 file changed, 9 insertions(+), 6 deletions(-) (limited to 'internal/server') diff --git a/internal/server/handlers/serverhandler.go b/internal/server/handlers/serverhandler.go index 185e7c2..23e3aeb 100644 --- a/internal/server/handlers/serverhandler.go +++ b/internal/server/handlers/serverhandler.go @@ -17,8 +17,8 @@ import ( "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/protocol" user "github.com/mimecast/dtail/internal/user/server" - "github.com/mimecast/dtail/internal/version" ) const ( @@ -92,24 +92,27 @@ func (h *ServerHandler) Read(p []byte) (n int, err error) { } if message[0] == '.' { // Handle hidden message (don't display to the user, interpreted by dtail client) - wholePayload := []byte(fmt.Sprintf("%s\n", message)) + wholePayload := []byte(fmt.Sprintf("%s%b", message, protocol.MessageDelimiter)) n = copy(p, wholePayload) return } // Handle normal server message (display to the user) - wholePayload := []byte(fmt.Sprintf("SERVER|%s|%s\n", h.hostname, message)) + wholePayload := []byte(fmt.Sprintf("SERVER|%s|%s%b", h.hostname, message, protocol.MessageDelimiter)) n = copy(p, wholePayload) return case message := <-h.aggregatedMessages: // Send mapreduce-aggregated data as a message. - data := fmt.Sprintf("AGGREGATE➔%s➔%s\n", h.hostname, message) + data := fmt.Sprintf("AGGREGATE%s%s%s%s%b", + protocol.AggregateDelimiter, h.hostname, + protocol.AggregateDelimiter, message, protocol.MessageDelimiter) wholePayload := []byte(data) n = copy(p, wholePayload) return case line := <-h.lines: + //fmt.Printf("<<<%d,%s>>>\n", len(line.Content), line.Content) // Send normal file content data as a message. serverInfo := []byte(fmt.Sprintf("REMOTE|%s|%3d|%v|%s|", h.hostname, line.TransmittedPerc, line.Count, line.SourceID)) @@ -182,8 +185,8 @@ func (h *ServerHandler) handleProtocolVersion(args []string) ([]string, int, err return args, argc, errors.New("unable to determine protocol version") } - if args[1] != version.ProtocolCompat { - err := fmt.Errorf("server with protocol version '%s' but client with '%s', please update DTail", version.ProtocolCompat, args[1]) + if args[1] != protocol.ProtocolCompat { + err := fmt.Errorf("server with protocol version '%s' but client with '%s', please update DTail", protocol.ProtocolCompat, args[1]) return args, argc, err } -- cgit v1.2.3 From 9883a190109623b64e6d311dc2b462a6eae68003 Mon Sep 17 00:00:00 2001 From: Paul Buetow Date: Sun, 22 Aug 2021 10:07:00 +0300 Subject: introduces the protocol package --- internal/server/handlers/controlhandler.go | 3 ++- internal/server/handlers/serverhandler.go | 6 +++--- 2 files changed, 5 insertions(+), 4 deletions(-) (limited to 'internal/server') diff --git a/internal/server/handlers/controlhandler.go b/internal/server/handlers/controlhandler.go index 1e17c78..a217b40 100644 --- a/internal/server/handlers/controlhandler.go +++ b/internal/server/handlers/controlhandler.go @@ -8,6 +8,7 @@ import ( "github.com/mimecast/dtail/internal" "github.com/mimecast/dtail/internal/io/logger" + "github.com/mimecast/dtail/internal/protocol" user "github.com/mimecast/dtail/internal/user/server" ) @@ -56,7 +57,7 @@ func (h *ControlHandler) Read(p []byte) (n int, err error) { for { select { case message := <-h.serverMessages: - wholePayload := []byte(fmt.Sprintf("SERVER|%s|%s\n", h.hostname, message)) + wholePayload := []byte(fmt.Sprintf("SERVER|%s|%s%b", h.hostname, message, protocol.MessageDelimiter)) n = copy(p, wholePayload) return case <-h.done.Done(): diff --git a/internal/server/handlers/serverhandler.go b/internal/server/handlers/serverhandler.go index 23e3aeb..9541a34 100644 --- a/internal/server/handlers/serverhandler.go +++ b/internal/server/handlers/serverhandler.go @@ -92,13 +92,13 @@ func (h *ServerHandler) Read(p []byte) (n int, err error) { } if message[0] == '.' { // Handle hidden message (don't display to the user, interpreted by dtail client) - wholePayload := []byte(fmt.Sprintf("%s%b", message, protocol.MessageDelimiter)) + wholePayload := []byte(fmt.Sprintf("%s%s", message, string(protocol.MessageDelimiter))) n = copy(p, wholePayload) return } // Handle normal server message (display to the user) - wholePayload := []byte(fmt.Sprintf("SERVER|%s|%s%b", h.hostname, message, protocol.MessageDelimiter)) + wholePayload := []byte(fmt.Sprintf("SERVER|%s|%s%s", h.hostname, message, string(protocol.MessageDelimiter))) n = copy(p, wholePayload) return @@ -112,7 +112,7 @@ func (h *ServerHandler) Read(p []byte) (n int, err error) { return case line := <-h.lines: - //fmt.Printf("<<<%d,%s>>>\n", len(line.Content), line.Content) + //fmt.Printf("\t<<<%d,%s>>>\n", len(line.Content), line.Content) // Send normal file content data as a message. serverInfo := []byte(fmt.Sprintf("REMOTE|%s|%3d|%v|%s|", h.hostname, line.TransmittedPerc, line.Count, line.SourceID)) -- cgit v1.2.3 From 6d727b9bdbc387c8a5c34406a2c4de9140face38 Mon Sep 17 00:00:00 2001 From: Paul Buetow Date: Sat, 28 Aug 2021 19:36:46 +0100 Subject: use a byte.Buffer in the file reader --- internal/server/handlers/serverhandler.go | 9 +++++---- 1 file changed, 5 insertions(+), 4 deletions(-) (limited to 'internal/server') diff --git a/internal/server/handlers/serverhandler.go b/internal/server/handlers/serverhandler.go index 9541a34..62f3c2b 100644 --- a/internal/server/handlers/serverhandler.go +++ b/internal/server/handlers/serverhandler.go @@ -15,6 +15,7 @@ import ( "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/io/pool" "github.com/mimecast/dtail/internal/mapr/server" "github.com/mimecast/dtail/internal/omode" "github.com/mimecast/dtail/internal/protocol" @@ -114,10 +115,10 @@ func (h *ServerHandler) Read(p []byte) (n int, err error) { case line := <-h.lines: //fmt.Printf("\t<<<%d,%s>>>\n", len(line.Content), line.Content) // Send normal file content data as a message. - serverInfo := []byte(fmt.Sprintf("REMOTE|%s|%3d|%v|%s|", - h.hostname, line.TransmittedPerc, line.Count, line.SourceID)) - wholePayload := append(serverInfo, line.Content[:]...) - n = copy(p, wholePayload) + payload := []byte(fmt.Sprintf("REMOTE|%s|%3d|%v|%s|%s", + h.hostname, line.TransmittedPerc, line.Count, line.SourceID, line.Content.String())) + n = copy(p, payload) + pool.RecycleBytesBuffer(line.Content) return case <-time.After(time.Second): -- cgit v1.2.3 From 8c2e94030d0e31289c35fcfb56499707fd4a7ccd Mon Sep 17 00:00:00 2001 From: Paul Buetow Date: Sat, 28 Aug 2021 20:10:54 +0100 Subject: make use of more buffers on server side --- internal/server/handlers/serverhandler.go | 40 +++++++++++++++++++++---------- 1 file changed, 27 insertions(+), 13 deletions(-) (limited to 'internal/server') diff --git a/internal/server/handlers/serverhandler.go b/internal/server/handlers/serverhandler.go index 62f3c2b..f5aefa2 100644 --- a/internal/server/handlers/serverhandler.go +++ b/internal/server/handlers/serverhandler.go @@ -1,6 +1,7 @@ package handlers import ( + "bytes" "context" "encoding/base64" "errors" @@ -93,31 +94,44 @@ func (h *ServerHandler) Read(p []byte) (n int, err error) { } if message[0] == '.' { // Handle hidden message (don't display to the user, interpreted by dtail client) - wholePayload := []byte(fmt.Sprintf("%s%s", message, string(protocol.MessageDelimiter))) - n = copy(p, wholePayload) + payload := []byte(fmt.Sprintf("%s%s", message, string(protocol.MessageDelimiter))) + n = copy(p, payload) return } // Handle normal server message (display to the user) - wholePayload := []byte(fmt.Sprintf("SERVER|%s|%s%s", h.hostname, message, string(protocol.MessageDelimiter))) - n = copy(p, wholePayload) + payload := []byte(fmt.Sprintf("SERVER|%s|%s%s", h.hostname, message, string(protocol.MessageDelimiter))) + n = copy(p, payload) return case message := <-h.aggregatedMessages: // Send mapreduce-aggregated data as a message. - data := fmt.Sprintf("AGGREGATE%s%s%s%s%b", - protocol.AggregateDelimiter, h.hostname, - protocol.AggregateDelimiter, message, protocol.MessageDelimiter) - wholePayload := []byte(data) - n = copy(p, wholePayload) + buf := pool.BytesBuffer.Get().(*bytes.Buffer) + buf.WriteString("AGGREGATE") + buf.WriteString(protocol.AggregateDelimiter) + buf.WriteString(h.hostname) + buf.WriteString(protocol.AggregateDelimiter) + buf.WriteString(message) + buf.WriteByte(protocol.MessageDelimiter) + n = copy(p, buf.Bytes()) + pool.RecycleBytesBuffer(buf) return case line := <-h.lines: - //fmt.Printf("\t<<<%d,%s>>>\n", len(line.Content), line.Content) - // Send normal file content data as a message. - payload := []byte(fmt.Sprintf("REMOTE|%s|%3d|%v|%s|%s", - h.hostname, line.TransmittedPerc, line.Count, line.SourceID, line.Content.String())) + buf := pool.BytesBuffer.Get().(*bytes.Buffer) + buf.WriteString("REMOTE") + buf.WriteByte(protocol.FieldDelimiter) + buf.WriteString(h.hostname) + buf.WriteByte(protocol.FieldDelimiter) + buf.WriteString(fmt.Sprintf("%3d", line.TransmittedPerc)) + buf.WriteByte(protocol.FieldDelimiter) + buf.WriteString(fmt.Sprintf("%v", line.Count)) + buf.WriteByte(protocol.FieldDelimiter) + buf.WriteString(line.SourceID) + buf.WriteByte(protocol.FieldDelimiter) + payload := append(buf.Bytes(), line.Content.Bytes()...) n = copy(p, payload) + pool.RecycleBytesBuffer(buf) pool.RecycleBytesBuffer(line.Content) return -- cgit v1.2.3 From 23982f331c2154a66b86d596226c24454fd06be5 Mon Sep 17 00:00:00 2001 From: Paul Buetow Date: Sat, 28 Aug 2021 20:26:32 +0100 Subject: 1. Major performance gain by not checking for file truncation aftter each bytes read. 2. Introduce field separator to the protocol package. --- 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 f5aefa2..14fc5d0 100644 --- a/internal/server/handlers/serverhandler.go +++ b/internal/server/handlers/serverhandler.go @@ -120,15 +120,15 @@ func (h *ServerHandler) Read(p []byte) (n int, err error) { case line := <-h.lines: buf := pool.BytesBuffer.Get().(*bytes.Buffer) buf.WriteString("REMOTE") - buf.WriteByte(protocol.FieldDelimiter) + buf.WriteString(protocol.FieldDelimiter) buf.WriteString(h.hostname) - buf.WriteByte(protocol.FieldDelimiter) + buf.WriteString(protocol.FieldDelimiter) buf.WriteString(fmt.Sprintf("%3d", line.TransmittedPerc)) - buf.WriteByte(protocol.FieldDelimiter) + buf.WriteString(protocol.FieldDelimiter) buf.WriteString(fmt.Sprintf("%v", line.Count)) - buf.WriteByte(protocol.FieldDelimiter) + buf.WriteString(protocol.FieldDelimiter) buf.WriteString(line.SourceID) - buf.WriteByte(protocol.FieldDelimiter) + buf.WriteString(protocol.FieldDelimiter) payload := append(buf.Bytes(), line.Content.Bytes()...) n = copy(p, payload) pool.RecycleBytesBuffer(buf) -- cgit v1.2.3 From cc89d3fb8be2465af276d7ef03ea2a8affd87b2e Mon Sep 17 00:00:00 2001 From: Paul Buetow Date: Mon, 6 Sep 2021 13:48:55 +0300 Subject: Print out client/server update notice even from dtail server 4 to dtail client 3. --- internal/server/handlers/serverhandler.go | 30 +++++++++++++++++++++++------- 1 file changed, 23 insertions(+), 7 deletions(-) (limited to 'internal/server') diff --git a/internal/server/handlers/serverhandler.go b/internal/server/handlers/serverhandler.go index 14fc5d0..14f46a3 100644 --- a/internal/server/handlers/serverhandler.go +++ b/internal/server/handlers/serverhandler.go @@ -8,6 +8,7 @@ import ( "fmt" "io" "os" + "strconv" "strings" "sync/atomic" "time" @@ -167,9 +168,9 @@ func (h *ServerHandler) handleCommand(commandStr string) { logger.Debug(h.user, commandStr) ctx := context.Background() - args, argc, err := h.handleProtocolVersion(strings.Split(commandStr, " ")) + args, argc, add, err := h.handleProtocolVersion(strings.Split(commandStr, " ")) if err != nil { - h.send(h.serverMessages, logger.Error(h.user, err)) + h.send(h.serverMessages, logger.Error(h.user, err)+add) return } @@ -193,19 +194,34 @@ func (h *ServerHandler) handleCommand(commandStr string) { h.handleUserCommand(ctx, argc, args) } -func (h *ServerHandler) handleProtocolVersion(args []string) ([]string, int, error) { +func (h *ServerHandler) handleProtocolVersion(args []string) ([]string, int, string, error) { argc := len(args) + var add string if argc <= 2 || args[0] != "protocol" { - return args, argc, errors.New("unable to determine protocol version") + return args, argc, add, errors.New("unable to determine protocol version") } if args[1] != protocol.ProtocolCompat { - err := fmt.Errorf("server with protocol version '%s' but client with '%s', please update DTail", protocol.ProtocolCompat, args[1]) - return args, argc, err + clientCompat, _ := strconv.Atoi(args[1]) + serverCompat, _ := strconv.Atoi(protocol.ProtocolCompat) + if clientCompat <= 3 { + // Protocol version 3 or lower expect a newline as message separator + // One day (after 2 major versions) this exception may be removed! + add = "\n" + } + + toUpdate := "client" + if clientCompat > serverCompat { + toUpdate = "server" + } + + err := fmt.Errorf("DTail server protocol version '%s' does not match client protocol version '%s', please update DTail %s!", + protocol.ProtocolCompat, args[1], toUpdate) + return args, argc, add, err } - return args[2:], argc - 2, nil + return args[2:], argc - 2, add, nil } func (h *ServerHandler) handleBase64(args []string, argc int) ([]string, int, error) { -- cgit v1.2.3 From f74a9e4b35feb8c07d8a70b5a581088a0a59889d Mon Sep 17 00:00:00 2001 From: Paul Buetow Date: Tue, 7 Sep 2021 10:01:32 +0300 Subject: Produce MAPREDUCE lines, can aggregate these via default log format --- internal/server/stats.go | 12 ++++++++---- 1 file changed, 8 insertions(+), 4 deletions(-) (limited to 'internal/server') diff --git a/internal/server/stats.go b/internal/server/stats.go index ac579ad..3e8c71d 100644 --- a/internal/server/stats.go +++ b/internal/server/stats.go @@ -50,10 +50,14 @@ func (s *stats) logServerStats() { s.mutex.Lock() defer s.mutex.Unlock() - currentConnections := fmt.Sprintf("currentConnections=%d", s.currentConnections) - lifetimeConnections := fmt.Sprintf("lifetimeConnections=%d", s.lifetimeConnections) - goroutines := fmt.Sprintf("goroutines=%d", runtime.NumGoroutine()) - logger.Info("stats", currentConnections, lifetimeConnections, goroutines) + data := make(map[string]interface{}) + data["currentConnections"] = s.currentConnections + data["lifetimeConnections"] = s.lifetimeConnections + data["goroutines"] = runtime.NumGoroutine() + data["cgocalls"] = runtime.NumCgoCall() + data["cpu"] = runtime.NumCPU() + + logger.Mapreduce("STATS", data) } func (s *stats) serverLimitExceeded() error { -- cgit v1.2.3 From 16dc57e1e1c28e9d762424e596223a980770e059 Mon Sep 17 00:00:00 2001 From: Paul Buetow Date: Wed, 8 Sep 2021 19:10:50 +0300 Subject: mapreduce tables are in colors now too --- internal/server/handlers/controlhandler.go | 3 +- internal/server/handlers/serverhandler.go | 118 ++++++++++++++--------------- 2 files changed, 59 insertions(+), 62 deletions(-) (limited to 'internal/server') diff --git a/internal/server/handlers/controlhandler.go b/internal/server/handlers/controlhandler.go index a217b40..1e17c78 100644 --- a/internal/server/handlers/controlhandler.go +++ b/internal/server/handlers/controlhandler.go @@ -8,7 +8,6 @@ import ( "github.com/mimecast/dtail/internal" "github.com/mimecast/dtail/internal/io/logger" - "github.com/mimecast/dtail/internal/protocol" user "github.com/mimecast/dtail/internal/user/server" ) @@ -57,7 +56,7 @@ func (h *ControlHandler) Read(p []byte) (n int, err error) { for { select { case message := <-h.serverMessages: - wholePayload := []byte(fmt.Sprintf("SERVER|%s|%s%b", h.hostname, message, protocol.MessageDelimiter)) + wholePayload := []byte(fmt.Sprintf("SERVER|%s|%s\n", h.hostname, message)) n = copy(p, wholePayload) return case <-h.done.Done(): diff --git a/internal/server/handlers/serverhandler.go b/internal/server/handlers/serverhandler.go index 14f46a3..e74e686 100644 --- a/internal/server/handlers/serverhandler.go +++ b/internal/server/handlers/serverhandler.go @@ -38,7 +38,6 @@ type ServerHandler struct { aggregate *server.Aggregate aggregatedMessages chan string serverMessages chan string - payload []byte hostname string user *user.User catLimiter chan struct{} @@ -47,6 +46,8 @@ type ServerHandler struct { activeCommands int32 activeReaders int32 quiet bool + readBuf bytes.Buffer + writeBuf bytes.Buffer } // NewServerHandler returns the server handler. @@ -86,77 +87,74 @@ func (h *ServerHandler) Done() <-chan struct{} { // Read is to send data to the dtail client via Reader interface. func (h *ServerHandler) Read(p []byte) (n int, err error) { - for { - select { - case message := <-h.serverMessages: - if len(message) == 0 { - logger.Warn(h.user, "Empty message recieved") - return - } - if message[0] == '.' { - // Handle hidden message (don't display to the user, interpreted by dtail client) - payload := []byte(fmt.Sprintf("%s%s", message, string(protocol.MessageDelimiter))) - n = copy(p, payload) - return - } - - // Handle normal server message (display to the user) - payload := []byte(fmt.Sprintf("SERVER|%s|%s%s", h.hostname, message, string(protocol.MessageDelimiter))) - n = copy(p, payload) - return + defer h.readBuf.Reset() - case message := <-h.aggregatedMessages: - // Send mapreduce-aggregated data as a message. - buf := pool.BytesBuffer.Get().(*bytes.Buffer) - buf.WriteString("AGGREGATE") - buf.WriteString(protocol.AggregateDelimiter) - buf.WriteString(h.hostname) - buf.WriteString(protocol.AggregateDelimiter) - buf.WriteString(message) - buf.WriteByte(protocol.MessageDelimiter) - n = copy(p, buf.Bytes()) - pool.RecycleBytesBuffer(buf) + select { + case message := <-h.serverMessages: + if message[0] == '.' { + // Handle hidden message (don't display to the user, interpreted by dtail client) + h.readBuf.WriteString(message) + h.readBuf.WriteByte(protocol.MessageDelimiter) + n = copy(p, h.readBuf.Bytes()) return + } - case line := <-h.lines: - buf := pool.BytesBuffer.Get().(*bytes.Buffer) - buf.WriteString("REMOTE") - buf.WriteString(protocol.FieldDelimiter) - buf.WriteString(h.hostname) - buf.WriteString(protocol.FieldDelimiter) - buf.WriteString(fmt.Sprintf("%3d", line.TransmittedPerc)) - buf.WriteString(protocol.FieldDelimiter) - buf.WriteString(fmt.Sprintf("%v", line.Count)) - buf.WriteString(protocol.FieldDelimiter) - buf.WriteString(line.SourceID) - buf.WriteString(protocol.FieldDelimiter) - payload := append(buf.Bytes(), line.Content.Bytes()...) - n = copy(p, payload) - pool.RecycleBytesBuffer(buf) - pool.RecycleBytesBuffer(line.Content) + // Handle normal server message (display to the user) + h.readBuf.WriteString("SERVER") + h.readBuf.WriteString(protocol.FieldDelimiter) + h.readBuf.WriteString(h.hostname) + h.readBuf.WriteString(protocol.FieldDelimiter) + h.readBuf.WriteString(message) + h.readBuf.WriteByte(protocol.MessageDelimiter) + n = copy(p, h.readBuf.Bytes()) + + case message := <-h.aggregatedMessages: + // Send mapreduce-aggregated data as a message. + h.readBuf.WriteString("AGGREGATE") + h.readBuf.WriteString(protocol.FieldDelimiter) + h.readBuf.WriteString(h.hostname) + h.readBuf.WriteString(protocol.FieldDelimiter) + h.readBuf.WriteString(message) + h.readBuf.WriteByte(protocol.MessageDelimiter) + n = copy(p, h.readBuf.Bytes()) + + case line := <-h.lines: + h.readBuf.WriteString("REMOTE") + h.readBuf.WriteString(protocol.FieldDelimiter) + h.readBuf.WriteString(h.hostname) + h.readBuf.WriteString(protocol.FieldDelimiter) + h.readBuf.WriteString(fmt.Sprintf("%3d", line.TransmittedPerc)) + h.readBuf.WriteString(protocol.FieldDelimiter) + h.readBuf.WriteString(fmt.Sprintf("%v", line.Count)) + h.readBuf.WriteString(protocol.FieldDelimiter) + h.readBuf.WriteString(line.SourceID) + h.readBuf.WriteString(protocol.FieldDelimiter) + h.readBuf.WriteString(line.Content.String()) + h.readBuf.WriteByte(protocol.MessageDelimiter) + n = copy(p, h.readBuf.Bytes()) + pool.RecycleBytesBuffer(line.Content) + + case <-time.After(time.Second): + // Once in a while check whether we are done. + select { + case <-h.done.Done(): + err = io.EOF return - - case <-time.After(time.Second): - // Once in a while check whether we are done. - select { - case <-h.done.Done(): - return 0, io.EOF - default: - } + default: } } + return } // Write is to receive data from the dtail client via Writer interface. func (h *ServerHandler) Write(p []byte) (n int, err error) { - for _, c := range p { - switch c { + for _, b := range p { + switch b { case ';': - commandStr := strings.TrimSpace(string(h.payload)) - h.handleCommand(commandStr) - h.payload = nil + h.handleCommand(string(h.writeBuf.Bytes())) + h.writeBuf.Reset() default: - h.payload = append(h.payload, c) + h.writeBuf.WriteByte(b) } } -- cgit v1.2.3 From 2ebe7e9d63ba62c6f19749c39fe0a577d86ca775 Mon Sep 17 00:00:00 2001 From: Paul Buetow Date: Sun, 12 Sep 2021 19:04:42 +0300 Subject: bugfix: dmap skipped the last couple of mapreduce lines --- internal/server/handlers/readcommand.go | 15 ++++-- internal/server/handlers/serverhandler.go | 86 +++++++++++-------------------- 2 files changed, 40 insertions(+), 61 deletions(-) (limited to 'internal/server') diff --git a/internal/server/handlers/readcommand.go b/internal/server/handlers/readcommand.go index 5bab26f..69dd4a5 100644 --- a/internal/server/handlers/readcommand.go +++ b/internal/server/handlers/readcommand.go @@ -8,6 +8,7 @@ import ( "time" "github.com/mimecast/dtail/internal/io/fs" + "github.com/mimecast/dtail/internal/io/line" "github.com/mimecast/dtail/internal/io/logger" "github.com/mimecast/dtail/internal/omode" "github.com/mimecast/dtail/internal/regex" @@ -113,16 +114,20 @@ func (r *readCommand) readFile(ctx context.Context, path, globID string, re rege } lines := r.server.lines - - // Plug in mappreduce engine - if r.server.aggregate != nil { - lines = r.server.aggregate.Lines - } + aggregate := r.server.aggregate for { + if aggregate != nil { + lines = make(chan line.Line, 100) + aggregate.NextLinesCh <- lines + } if err := reader.Start(ctx, lines, re); err != nil { logger.Error(r.server.user, path, globID, err) } + if aggregate != nil { + // Also makes aggregate to Flush + close(lines) + } select { case <-ctx.Done(): diff --git a/internal/server/handlers/serverhandler.go b/internal/server/handlers/serverhandler.go index e74e686..ed19412 100644 --- a/internal/server/handlers/serverhandler.go +++ b/internal/server/handlers/serverhandler.go @@ -32,36 +32,35 @@ 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 - hostname string - user *user.User - catLimiter chan struct{} - tailLimiter chan struct{} - ackCloseReceived chan struct{} - activeCommands int32 - activeReaders int32 - quiet bool - readBuf bytes.Buffer - writeBuf bytes.Buffer + done *internal.Done + lines chan line.Line + regex string + aggregate *server.Aggregate + maprMessages chan string + serverMessages chan string + hostname string + user *user.User + catLimiter chan struct{} + tailLimiter chan struct{} + ackCloseReceived chan struct{} + activeCommands int32 + quiet bool + readBuf bytes.Buffer + writeBuf bytes.Buffer } // NewServerHandler returns the server handler. 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, - regex: ".", - user: user, + done: internal.NewDone(), + lines: make(chan line.Line, 100), + serverMessages: make(chan string, 10), + maprMessages: make(chan string, 10), + ackCloseReceived: make(chan struct{}), + catLimiter: catLimiter, + tailLimiter: tailLimiter, + regex: ".", + user: user, } fqdn, err := os.Hostname() @@ -108,7 +107,7 @@ func (h *ServerHandler) Read(p []byte) (n int, err error) { h.readBuf.WriteByte(protocol.MessageDelimiter) n = copy(p, h.readBuf.Bytes()) - case message := <-h.aggregatedMessages: + case message := <-h.maprMessages: // Send mapreduce-aggregated data as a message. h.readBuf.WriteString("AGGREGATE") h.readBuf.WriteString(protocol.FieldDelimiter) @@ -260,14 +259,6 @@ func (h *ServerHandler) handleUserCommand(ctx context.Context, argc int, args [] h.shutdown() } } - readerFinished := func() { - if h.decrementActiveReaders() == 0 { - if h.aggregate == nil { - return - } - h.aggregate.Shutdown() - } - } splitted := strings.Split(args[0], ":") commandName := splitted[0] @@ -289,18 +280,14 @@ func (h *ServerHandler) handleUserCommand(ctx context.Context, argc int, args [] case "grep", "cat": command := newReadCommand(h, omode.CatClient) go func() { - h.incrementActiveReaders() command.Start(ctx, argc, args, 1) - readerFinished() commandFinished() }() case "tail": command := newReadCommand(h, omode.TailClient) go func() { - h.incrementActiveReaders() command.Start(ctx, argc, args, 10) - readerFinished() commandFinished() }() @@ -315,7 +302,7 @@ func (h *ServerHandler) handleUserCommand(ctx context.Context, argc int, args [] h.aggregate = aggregate go func() { - command.Start(ctx, h.aggregatedMessages) + command.Start(ctx, h.maprMessages) commandFinished() }() @@ -361,15 +348,11 @@ func (h *ServerHandler) serverMessageC() chan<- string { return h.serverMessages } -func (h *ServerHandler) flush() { - logger.Debug(h.user, "flush()") - - if h.aggregate != nil { - h.aggregate.Flush() - } +func (h *ServerHandler) flushMessages() { + logger.Debug(h.user, "flushMessages()") unsentMessages := func() int { - return len(h.lines) + len(h.serverMessages) + len(h.aggregatedMessages) + return len(h.lines) + len(h.serverMessages) + len(h.maprMessages) } for i := 0; i < 3; i++ { if unsentMessages() == 0 { @@ -385,7 +368,7 @@ func (h *ServerHandler) flush() { func (h *ServerHandler) shutdown() { logger.Debug(h.user, "shutdown()") - h.flush() + h.flushMessages() go func() { select { @@ -413,15 +396,6 @@ func (h *ServerHandler) decrementActiveCommands() int32 { return atomic.LoadInt32(&h.activeCommands) } -func (h *ServerHandler) incrementActiveReaders() { - atomic.AddInt32(&h.activeReaders, 1) -} - -func (h *ServerHandler) decrementActiveReaders() int32 { - atomic.AddInt32(&h.activeReaders, -1) - return atomic.LoadInt32(&h.activeReaders) -} - func readOptions(opts []string) (map[string]string, error) { options := make(map[string]string, len(opts)) -- cgit v1.2.3 From 6506e20f6c80f4acb7434eb9dd14f784a67189cd Mon Sep 17 00:00:00 2001 From: Paul Buetow Date: Sat, 18 Sep 2021 14:41:25 +0300 Subject: add spartan mode --- internal/server/handlers/serverhandler.go | 31 +++++++++++++++++++++---------- 1 file changed, 21 insertions(+), 10 deletions(-) (limited to 'internal/server') diff --git a/internal/server/handlers/serverhandler.go b/internal/server/handlers/serverhandler.go index ed19412..2f3b73b 100644 --- a/internal/server/handlers/serverhandler.go +++ b/internal/server/handlers/serverhandler.go @@ -45,6 +45,7 @@ type ServerHandler struct { ackCloseReceived chan struct{} activeCommands int32 quiet bool + spartan bool readBuf bytes.Buffer writeBuf bytes.Buffer } @@ -118,16 +119,18 @@ func (h *ServerHandler) Read(p []byte) (n int, err error) { n = copy(p, h.readBuf.Bytes()) case line := <-h.lines: - h.readBuf.WriteString("REMOTE") - h.readBuf.WriteString(protocol.FieldDelimiter) - h.readBuf.WriteString(h.hostname) - h.readBuf.WriteString(protocol.FieldDelimiter) - h.readBuf.WriteString(fmt.Sprintf("%3d", line.TransmittedPerc)) - h.readBuf.WriteString(protocol.FieldDelimiter) - h.readBuf.WriteString(fmt.Sprintf("%v", line.Count)) - h.readBuf.WriteString(protocol.FieldDelimiter) - h.readBuf.WriteString(line.SourceID) - h.readBuf.WriteString(protocol.FieldDelimiter) + if !h.spartan { + h.readBuf.WriteString("REMOTE") + h.readBuf.WriteString(protocol.FieldDelimiter) + h.readBuf.WriteString(h.hostname) + h.readBuf.WriteString(protocol.FieldDelimiter) + h.readBuf.WriteString(fmt.Sprintf("%3d", line.TransmittedPerc)) + h.readBuf.WriteString(protocol.FieldDelimiter) + h.readBuf.WriteString(fmt.Sprintf("%v", line.Count)) + h.readBuf.WriteString(protocol.FieldDelimiter) + h.readBuf.WriteString(line.SourceID) + h.readBuf.WriteString(protocol.FieldDelimiter) + } h.readBuf.WriteString(line.Content.String()) h.readBuf.WriteByte(protocol.MessageDelimiter) n = copy(p, h.readBuf.Bytes()) @@ -275,6 +278,12 @@ func (h *ServerHandler) handleUserCommand(ctx context.Context, argc int, args [] h.quiet = true } } + if spartan, ok := options["spartan"]; ok { + if spartan == "true" { + logger.Debug(h.user, "Enabling spartan mode") + h.spartan = true + } + } switch commandName { case "grep", "cat": @@ -397,6 +406,7 @@ func (h *ServerHandler) decrementActiveCommands() int32 { } func readOptions(opts []string) (map[string]string, error) { + logger.Debug("Parsing options", opts) options := make(map[string]string, len(opts)) for _, o := range opts { @@ -416,6 +426,7 @@ func readOptions(opts []string) (map[string]string, error) { val = string(decoded) } + logger.Debug("Setting option", key, val) options[key] = val } -- cgit v1.2.3 From abeac87aec44249bf67f1b0eca471a31086265ca Mon Sep 17 00:00:00 2001 From: Paul Buetow Date: Sat, 18 Sep 2021 19:27:50 +0300 Subject: fix auto reconnect --- internal/server/handlers/serverhandler.go | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) (limited to 'internal/server') diff --git a/internal/server/handlers/serverhandler.go b/internal/server/handlers/serverhandler.go index 2f3b73b..4820476 100644 --- a/internal/server/handlers/serverhandler.go +++ b/internal/server/handlers/serverhandler.go @@ -331,7 +331,11 @@ func (h *ServerHandler) handleAckCommand(argc int, args []string) { return } if args[1] == "close" && args[2] == "connection" { - close(h.ackCloseReceived) + select { + case <-h.ackCloseReceived: + default: + close(h.ackCloseReceived) + } } } -- cgit v1.2.3 From fe3e68afd99d8ea246be52893730f987e138ec24 Mon Sep 17 00:00:00 2001 From: Paul Buetow Date: Sun, 19 Sep 2021 13:22:59 +0300 Subject: move args to config package logger package rewrite as dlog --- internal/server/continuous.go | 21 ++++++------ internal/server/handlers/controlhandler.go | 12 +++---- internal/server/handlers/readcommand.go | 26 +++++++-------- internal/server/handlers/serverhandler.go | 46 +++++++++++++------------- internal/server/scheduler.go | 22 ++++++------- internal/server/server.go | 52 +++++++++++++++--------------- internal/server/stats.go | 6 ++-- 7 files changed, 92 insertions(+), 93 deletions(-) (limited to 'internal/server') diff --git a/internal/server/continuous.go b/internal/server/continuous.go index f75c732..5f4c454 100644 --- a/internal/server/continuous.go +++ b/internal/server/continuous.go @@ -8,9 +8,8 @@ import ( "github.com/mimecast/dtail/internal/clients" "github.com/mimecast/dtail/internal/config" - "github.com/mimecast/dtail/internal/io/logger" + "github.com/mimecast/dtail/internal/io/dlog" "github.com/mimecast/dtail/internal/omode" - gossh "golang.org/x/crypto/ssh" ) @@ -22,7 +21,7 @@ func newContinuous() *continuous { } func (c *continuous) start(ctx context.Context) { - logger.Info("Starting continuous job runner after 10s") + dlog.Server.Info("Starting continuous job runner after 10s") time.Sleep(time.Second * 10) c.runJobs(ctx) @@ -31,7 +30,7 @@ func (c *continuous) start(ctx context.Context) { func (c *continuous) runJobs(ctx context.Context) { for _, job := range config.Server.Continuous { if !job.Enable { - logger.Debug(job.Name, "Not running job as not enabled") + dlog.Server.Debug(job.Name, "Not running job as not enabled") continue } @@ -51,7 +50,7 @@ func (c *continuous) runJobs(ctx context.Context) { } func (c *continuous) runJob(ctx context.Context, job config.Continuous) { - logger.Debug(job.Name, "Processing job") + dlog.Server.Debug(job.Name, "Processing job") files := fillDates(job.Files) outfile := fillDates(job.Outfile) @@ -61,7 +60,7 @@ func (c *continuous) runJob(ctx context.Context, job config.Continuous) { servers = config.Server.SSHBindAddress } - args := clients.Args{ + args := config.Args{ ConnectionsPerCPU: 10, Discovery: job.Discovery, ServersStr: servers, @@ -75,7 +74,7 @@ func (c *continuous) runJob(ctx context.Context, job config.Continuous) { query := fmt.Sprintf("%s outfile %s", job.Query, outfile) client, err := clients.NewMaprClient(args, query, clients.NonCumulativeMode) if err != nil { - logger.Error(fmt.Sprintf("Unable to create job %s", job.Name), err) + dlog.Server.Error(fmt.Sprintf("Unable to create job %s", job.Name), err) return } @@ -85,21 +84,21 @@ func (c *continuous) runJob(ctx context.Context, job config.Continuous) { if job.RestartOnDayChange { go func() { if c.waitForDayChange(ctx) { - logger.Info(fmt.Sprintf("Canceling job %s due to day change", job.Name)) + dlog.Server.Info(fmt.Sprintf("Canceling job %s due to day change", job.Name)) cancel() } }() } - logger.Info(fmt.Sprintf("Starting job %s", job.Name)) + dlog.Server.Info(fmt.Sprintf("Starting job %s", job.Name)) status := client.Start(jobCtx, make(chan string)) logMessage := fmt.Sprintf("Job exited with status %d", status) if status != 0 { - logger.Warn(logMessage) + dlog.Server.Warn(logMessage) return } - logger.Info(logMessage) + dlog.Server.Info(logMessage) } func (c *continuous) waitForDayChange(ctx context.Context) bool { diff --git a/internal/server/handlers/controlhandler.go b/internal/server/handlers/controlhandler.go index 1e17c78..ae70675 100644 --- a/internal/server/handlers/controlhandler.go +++ b/internal/server/handlers/controlhandler.go @@ -7,7 +7,7 @@ import ( "strings" "github.com/mimecast/dtail/internal" - "github.com/mimecast/dtail/internal/io/logger" + "github.com/mimecast/dtail/internal/io/dlog" user "github.com/mimecast/dtail/internal/user/server" ) @@ -22,7 +22,7 @@ type ControlHandler struct { // NewControlHandler returns a new control handler. func NewControlHandler(user *user.User) *ControlHandler { - logger.Debug(user, "Creating control handler") + dlog.Server.Debug(user, "Creating control handler") h := ControlHandler{ done: internal.NewDone(), @@ -32,7 +32,7 @@ func NewControlHandler(user *user.User) *ControlHandler { fqdn, err := os.Hostname() if err != nil { - logger.FatalExit(err) + dlog.Server.FatalPanic(err) } s := strings.Split(fqdn, ".") @@ -84,15 +84,15 @@ func (h *ControlHandler) Write(p []byte) (n int, err error) { } func (h *ControlHandler) handleCommand(command string) { - logger.Info(h.user, command) + dlog.Server.Info(h.user, command) s := strings.Split(command, " ") - logger.Debug(h.user, "Receiving command", command, s) + dlog.Server.Debug(h.user, "Receiving command", command, s) switch s[0] { case "health": h.serverMessages <- "OK: DTail SSH Server seems fine" h.serverMessages <- "done;" default: - h.serverMessages <- logger.Error(h.user, "Received unknown control command", command, s) + h.serverMessages <- dlog.Server.Error(h.user, "Received unknown control command", command, s) } } diff --git a/internal/server/handlers/readcommand.go b/internal/server/handlers/readcommand.go index 69dd4a5..60ad2a0 100644 --- a/internal/server/handlers/readcommand.go +++ b/internal/server/handlers/readcommand.go @@ -9,7 +9,7 @@ import ( "github.com/mimecast/dtail/internal/io/fs" "github.com/mimecast/dtail/internal/io/line" - "github.com/mimecast/dtail/internal/io/logger" + "github.com/mimecast/dtail/internal/io/dlog" "github.com/mimecast/dtail/internal/omode" "github.com/mimecast/dtail/internal/regex" ) @@ -32,13 +32,13 @@ func (r *readCommand) Start(ctx context.Context, argc int, args []string, retrie if argc >= 4 { deserializedRegex, err := regex.Deserialize(strings.Join(args[2:], " ")) if err != nil { - r.server.sendServerMessage(logger.Error(r.server.user, commandParseWarning, err)) + r.server.sendServerMessage(dlog.Server.Error(r.server.user, commandParseWarning, err)) return } re = deserializedRegex } if argc < 3 { - r.server.sendServerWarnMessage(logger.Warn(r.server.user, commandParseWarning, args, argc)) + r.server.sendServerWarnMessage(dlog.Server.Warn(r.server.user, commandParseWarning, args, argc)) return } r.readGlob(ctx, args[1], re, retries) @@ -51,14 +51,14 @@ func (r *readCommand) readGlob(ctx context.Context, glob string, re regex.Regex, for retryCount := 0; retryCount < retries; retryCount++ { paths, err := filepath.Glob(glob) if err != nil { - logger.Warn(r.server.user, glob, err) + dlog.Server.Warn(r.server.user, glob, err) time.Sleep(retryInterval) continue } if numPaths := len(paths); numPaths == 0 { - logger.Error(r.server.user, "No such file(s) to read", glob) - r.server.sendServerWarnMessage(logger.Warn(r.server.user, "Unable to read file(s), check server logs")) + dlog.Server.Error(r.server.user, "No such file(s) to read", glob) + r.server.sendServerWarnMessage(dlog.Server.Warn(r.server.user, "Unable to read file(s), check server logs")) select { case <-ctx.Done(): return @@ -72,7 +72,7 @@ func (r *readCommand) readGlob(ctx context.Context, glob string, re regex.Regex, return } - r.server.sendServerWarnMessage(logger.Warn(r.server.user, "Giving up to read file(s)")) + r.server.sendServerWarnMessage(dlog.Server.Warn(r.server.user, "Giving up to read file(s)")) return } @@ -92,8 +92,8 @@ func (r *readCommand) readFileIfPermissions(ctx context.Context, wg *sync.WaitGr globID := r.makeGlobID(path, glob) if !r.server.user.HasFilePermission(path, "readfiles") { - logger.Error(r.server.user, "No permission to read file", path, globID) - r.server.sendServerWarnMessage(logger.Warn(r.server.user, "Unable to read file(s), check server logs")) + dlog.Server.Error(r.server.user, "No permission to read file", path, globID) + r.server.sendServerWarnMessage(dlog.Server.Warn(r.server.user, "Unable to read file(s), check server logs")) return } @@ -101,7 +101,7 @@ func (r *readCommand) readFileIfPermissions(ctx context.Context, wg *sync.WaitGr } func (r *readCommand) readFile(ctx context.Context, path, globID string, re regex.Regex) { - logger.Info(r.server.user, "Start reading file", path, globID) + dlog.Server.Info(r.server.user, "Start reading file", path, globID) var reader fs.FileReader switch r.mode { @@ -122,7 +122,7 @@ func (r *readCommand) readFile(ctx context.Context, path, globID string, re rege aggregate.NextLinesCh <- lines } if err := reader.Start(ctx, lines, re); err != nil { - logger.Error(r.server.user, path, globID, err) + dlog.Server.Error(r.server.user, path, globID, err) } if aggregate != nil { // Also makes aggregate to Flush @@ -139,7 +139,7 @@ func (r *readCommand) readFile(ctx context.Context, path, globID string, re rege } time.Sleep(time.Second * 2) - logger.Info(path, globID, "Reading file again") + dlog.Server.Info(path, globID, "Reading file again") } } @@ -161,6 +161,6 @@ func (r *readCommand) makeGlobID(path, glob string) string { return pathParts[len(pathParts)-1] } - r.server.sendServerWarnMessage(logger.Warn("Empty file path given?", path, glob)) + r.server.sendServerWarnMessage(dlog.Server.Warn("Empty file path given?", path, glob)) return "" } diff --git a/internal/server/handlers/serverhandler.go b/internal/server/handlers/serverhandler.go index 4820476..b664566 100644 --- a/internal/server/handlers/serverhandler.go +++ b/internal/server/handlers/serverhandler.go @@ -16,7 +16,7 @@ import ( "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/io/dlog" "github.com/mimecast/dtail/internal/io/pool" "github.com/mimecast/dtail/internal/mapr/server" "github.com/mimecast/dtail/internal/omode" @@ -66,7 +66,7 @@ func NewServerHandler(user *user.User, catLimiter, tailLimiter chan struct{}) *S fqdn, err := os.Hostname() if err != nil { - logger.FatalExit(err) + dlog.Server.FatalPanic(err) } s := strings.Split(fqdn, ".") @@ -165,18 +165,18 @@ func (h *ServerHandler) Write(p []byte) (n int, err error) { } func (h *ServerHandler) handleCommand(commandStr string) { - logger.Debug(h.user, commandStr) + dlog.Server.Debug(h.user, commandStr) ctx := context.Background() args, argc, add, err := h.handleProtocolVersion(strings.Split(commandStr, " ")) if err != nil { - h.send(h.serverMessages, logger.Error(h.user, err)+add) + h.send(h.serverMessages, dlog.Server.Error(h.user, err)+add) return } args, argc, err = h.handleBase64(args, argc) if err != nil { - h.send(h.serverMessages, logger.Error(h.user, err)) + h.send(h.serverMessages, dlog.Server.Error(h.user, err)) return } @@ -239,7 +239,7 @@ func (h *ServerHandler) handleBase64(args []string, argc int) ([]string, int, er args = strings.Split(decodedStr, " ") argc = len(decodedStr) - logger.Trace(h.user, "Base64 decoded received command", decodedStr, argc, args) + dlog.Server.Trace(h.user, "Base64 decoded received command", decodedStr, argc, args) return args, argc, nil } @@ -247,14 +247,14 @@ func (h *ServerHandler) handleBase64(args []string, argc int) ([]string, int, er func (h *ServerHandler) handleControlCommand(argc int, args []string) { switch args[0] { case "debug": - h.send(h.serverMessages, logger.Debug(h.user, "Receiving debug command", argc, args)) + h.send(h.serverMessages, dlog.Server.Debug(h.user, "Receiving debug command", argc, args)) default: - logger.Warn(h.user, "Received unknown control command", argc, args) + dlog.Server.Warn(h.user, "Received unknown control command", argc, args) } } func (h *ServerHandler) handleUserCommand(ctx context.Context, argc int, args []string) { - logger.Debug(h.user, "handleUserCommand", argc, args) + dlog.Server.Debug(h.user, "handleUserCommand", argc, args) h.incrementActiveCommands() commandFinished := func() { @@ -268,19 +268,19 @@ func (h *ServerHandler) handleUserCommand(ctx context.Context, argc int, args [] options, err := readOptions(splitted[1:]) if err != nil { - h.sendServerMessage(logger.Error(h.user, err)) + h.sendServerMessage(dlog.Server.Error(h.user, err)) commandFinished() return } if quiet, ok := options["quiet"]; ok { if quiet == "true" { - logger.Debug(h.user, "Enabling quiet mode") + dlog.Server.Debug(h.user, "Enabling quiet mode") h.quiet = true } } if spartan, ok := options["spartan"]; ok { if spartan == "true" { - logger.Debug(h.user, "Enabling spartan mode") + dlog.Server.Debug(h.user, "Enabling spartan mode") h.spartan = true } } @@ -304,7 +304,7 @@ func (h *ServerHandler) handleUserCommand(ctx context.Context, argc int, args [] command, aggregate, err := newMapCommand(h, argc, args) if err != nil { h.sendServerMessage(err.Error()) - logger.Error(h.user, err) + dlog.Server.Error(h.user, err) commandFinished() return } @@ -320,14 +320,14 @@ func (h *ServerHandler) handleUserCommand(ctx context.Context, argc int, args [] commandFinished() default: - h.sendServerMessage(logger.Error(h.user, "Received unknown user command", commandName, argc, args, options)) + h.sendServerMessage(dlog.Server.Error(h.user, "Received unknown user command", commandName, argc, args, options)) commandFinished() } } func (h *ServerHandler) handleAckCommand(argc int, args []string) { if argc < 3 { - h.sendServerWarnMessage(logger.Warn(h.user, commandParseWarning, args, argc)) + h.sendServerWarnMessage(dlog.Server.Warn(h.user, commandParseWarning, args, argc)) return } if args[1] == "close" && args[2] == "connection" { @@ -362,25 +362,25 @@ func (h *ServerHandler) serverMessageC() chan<- string { } func (h *ServerHandler) flushMessages() { - logger.Debug(h.user, "flushMessages()") + dlog.Server.Debug(h.user, "flushMessages()") unsentMessages := func() int { return len(h.lines) + len(h.serverMessages) + len(h.maprMessages) } for i := 0; i < 3; i++ { if unsentMessages() == 0 { - logger.Debug(h.user, "All lines sent") + dlog.Server.Debug(h.user, "All lines sent") return } - logger.Debug(h.user, "Still lines to be sent") + dlog.Server.Debug(h.user, "Still lines to be sent") time.Sleep(time.Second) } - logger.Warn(h.user, "Some lines remain unsent", unsentMessages()) + dlog.Server.Warn(h.user, "Some lines remain unsent", unsentMessages()) } func (h *ServerHandler) shutdown() { - logger.Debug(h.user, "shutdown()") + dlog.Server.Debug(h.user, "shutdown()") h.flushMessages() go func() { @@ -393,7 +393,7 @@ func (h *ServerHandler) shutdown() { select { case <-h.ackCloseReceived: case <-time.After(time.Second * 5): - logger.Debug(h.user, "Shutdown timeout reached, enforcing shutdown") + dlog.Server.Debug(h.user, "Shutdown timeout reached, enforcing shutdown") case <-h.done.Done(): } @@ -410,7 +410,7 @@ func (h *ServerHandler) decrementActiveCommands() int32 { } func readOptions(opts []string) (map[string]string, error) { - logger.Debug("Parsing options", opts) + dlog.Server.Debug("Parsing options", opts) options := make(map[string]string, len(opts)) for _, o := range opts { @@ -430,7 +430,7 @@ func readOptions(opts []string) (map[string]string, error) { val = string(decoded) } - logger.Debug("Setting option", key, val) + dlog.Server.Debug("Setting option", key, val) options[key] = val } diff --git a/internal/server/scheduler.go b/internal/server/scheduler.go index a1e9e36..f474cc8 100644 --- a/internal/server/scheduler.go +++ b/internal/server/scheduler.go @@ -10,7 +10,7 @@ import ( "github.com/mimecast/dtail/internal/clients" "github.com/mimecast/dtail/internal/config" - "github.com/mimecast/dtail/internal/io/logger" + "github.com/mimecast/dtail/internal/io/dlog" "github.com/mimecast/dtail/internal/omode" gossh "golang.org/x/crypto/ssh" @@ -24,7 +24,7 @@ func newScheduler() *scheduler { } func (s *scheduler) start(ctx context.Context) { - logger.Info("Starting scheduled job runner after 10s") + dlog.Server.Info("Starting scheduled job runner after 10s") // First run after just 10s! time.Sleep(time.Second * 10) s.runJobs(ctx) @@ -42,18 +42,18 @@ func (s *scheduler) start(ctx context.Context) { func (s *scheduler) runJobs(ctx context.Context) { for _, job := range config.Server.Schedule { if !job.Enable { - logger.Debug(job.Name, "Not running job as not enabled") + dlog.Server.Debug(job.Name, "Not running job as not enabled") continue } hour, err := strconv.Atoi(time.Now().Format("15")) if err != nil { - logger.Error(job.Name, "Unable to create job", err) + dlog.Server.Error(job.Name, "Unable to create job", err) continue } if hour < job.TimeRange[0] || hour >= job.TimeRange[1] { - logger.Debug(job.Name, "Not running job out of time range") + dlog.Server.Debug(job.Name, "Not running job out of time range") continue } @@ -62,7 +62,7 @@ func (s *scheduler) runJobs(ctx context.Context) { _, err = os.Stat(outfile) if !os.IsNotExist(err) { - logger.Debug(job.Name, "Not running job as outfile already exists", outfile) + dlog.Server.Debug(job.Name, "Not running job as outfile already exists", outfile) continue } @@ -71,7 +71,7 @@ func (s *scheduler) runJobs(ctx context.Context) { servers = config.Server.SSHBindAddress } - args := clients.Args{ + args := config.Args{ ConnectionsPerCPU: 10, Discovery: job.Discovery, ServersStr: servers, @@ -85,21 +85,21 @@ func (s *scheduler) runJobs(ctx context.Context) { query := fmt.Sprintf("%s outfile %s", job.Query, outfile) client, err := clients.NewMaprClient(args, query, clients.CumulativeMode) if err != nil { - logger.Error(fmt.Sprintf("Unable to create job %s", job.Name), err) + dlog.Server.Error(fmt.Sprintf("Unable to create job %s", job.Name), err) continue } jobCtx, cancel := context.WithCancel(ctx) defer cancel() - logger.Info(fmt.Sprintf("Starting job %s", job.Name)) + dlog.Server.Info(fmt.Sprintf("Starting job %s", job.Name)) status := client.Start(jobCtx, make(chan string)) logMessage := fmt.Sprintf("Job exited with status %d", status) if status != 0 { - logger.Warn(logMessage) + dlog.Server.Warn(logMessage) continue } - logger.Info(logMessage) + dlog.Server.Info(logMessage) } } diff --git a/internal/server/server.go b/internal/server/server.go index a20737e..a8f541b 100644 --- a/internal/server/server.go +++ b/internal/server/server.go @@ -9,7 +9,7 @@ import ( "strings" "github.com/mimecast/dtail/internal/config" - "github.com/mimecast/dtail/internal/io/logger" + "github.com/mimecast/dtail/internal/io/dlog" "github.com/mimecast/dtail/internal/server/handlers" "github.com/mimecast/dtail/internal/ssh/server" user "github.com/mimecast/dtail/internal/user/server" @@ -36,7 +36,7 @@ type Server struct { // New returns a new server. func New() *Server { - logger.Info("Creating server", version.String()) + dlog.Server.Info("Creating server", version.String()) s := Server{ sshServerConfig: &gossh.ServerConfig{}, @@ -51,7 +51,7 @@ func New() *Server { private, err := gossh.ParsePrivateKey(server.PrivateHostKey()) if err != nil { - logger.FatalExit(err) + dlog.Server.FatalPanic(err) } s.sshServerConfig.AddHostKey(private) @@ -60,14 +60,14 @@ func New() *Server { // Start the server. func (s *Server) Start(ctx context.Context) int { - logger.Info("Starting server") + dlog.Server.Info("Starting server") bindAt := fmt.Sprintf("%s:%d", config.Server.SSHBindAddress, config.Common.SSHPort) - logger.Info("Binding server", bindAt) + dlog.Server.Info("Binding server", bindAt) listener, err := net.Listen("tcp", bindAt) if err != nil { - logger.FatalExit("Failed to open listening TCP socket", err) + dlog.Server.FatalPanic("Failed to open listening TCP socket", err) } go s.stats.start(ctx) @@ -82,7 +82,7 @@ func (s *Server) Start(ctx context.Context) int { } func (s *Server) listenerLoop(ctx context.Context, listener net.Listener) { - logger.Debug("Starting listener loop") + dlog.Server.Debug("Starting listener loop") for { conn, err := listener.Accept() // Blocking @@ -92,12 +92,12 @@ func (s *Server) listenerLoop(ctx context.Context, listener net.Listener) { return default: } - logger.Error("Failed to accept incoming connection", err) + dlog.Server.Error("Failed to accept incoming connection", err) continue } if err := s.stats.serverLimitExceeded(); err != nil { - logger.Error(err) + dlog.Server.Error(err) conn.Close() continue } @@ -107,11 +107,11 @@ func (s *Server) listenerLoop(ctx context.Context, listener net.Listener) { } func (s *Server) handleConnection(ctx context.Context, conn net.Conn) { - logger.Info("Handling connection") + dlog.Server.Info("Handling connection") sshConn, chans, reqs, err := gossh.NewServerConn(conn, s.sshServerConfig) if err != nil { - logger.Error("Something just happened", err) + dlog.Server.Error("Something just happened", err) return } @@ -125,29 +125,29 @@ func (s *Server) handleConnection(ctx context.Context, conn net.Conn) { func (s *Server) handleChannel(ctx context.Context, sshConn gossh.Conn, newChannel gossh.NewChannel) { user := user.New(sshConn.User(), sshConn.RemoteAddr().String()) - logger.Info(user, "Invoking channel handler") + dlog.Server.Info(user, "Invoking channel handler") if newChannel.ChannelType() != "session" { err := errors.New("Don'w allow other channel types than session") - logger.Error(user, err) + dlog.Server.Error(user, err) newChannel.Reject(gossh.Prohibited, err.Error()) return } channel, requests, err := newChannel.Accept() if err != nil { - logger.Error(user, "Could not accept channel", err) + dlog.Server.Error(user, "Could not accept channel", err) return } if err := s.handleRequests(ctx, sshConn, requests, channel, user); err != nil { - logger.Error(user, err) + dlog.Server.Error(user, err) sshConn.Close() } } func (s *Server) handleRequests(ctx context.Context, sshConn gossh.Conn, in <-chan *gossh.Request, channel gossh.Channel, user *user.User) error { - logger.Info(user, "Invoking request handler") + dlog.Server.Info(user, "Invoking request handler") for req := range in { var payload = struct{ Value string }{} @@ -190,10 +190,10 @@ func (s *Server) handleRequests(ctx context.Context, sshConn gossh.Conn, in <-ch go func() { if err := sshConn.Wait(); err != nil && err != io.EOF { - logger.Error(user, err) + dlog.Server.Error(user, err) } s.stats.decrementConnections() - logger.Info(user, "Good bye Mister!") + dlog.Server.Info(user, "Good bye Mister!") terminate() }() @@ -216,7 +216,7 @@ func (s *Server) Callback(c gossh.ConnMetadata, authPayload []byte) (*gossh.Perm user := user.New(c.User(), c.RemoteAddr().String()) if config.ServerRelaxedAuthEnable { - logger.Fatal(user, "Granting permissions via relaxed-auth") + dlog.Server.Fatal(user, "Granting permissions via relaxed-auth") return nil, nil } @@ -228,20 +228,20 @@ func (s *Server) Callback(c gossh.ConnMetadata, authPayload []byte) (*gossh.Perm switch user.Name { case config.ControlUser: if authInfo == config.ControlUser { - logger.Debug(user, "Granting permissions to control user") + dlog.Server.Debug(user, "Granting permissions to control user") return nil, nil } case config.ScheduleUser: for _, job := range config.Server.Schedule { if s.backgroundCanSSH(user, authInfo, remoteIP, job.Name, job.AllowFrom) { - logger.Debug(user, "Granting SSH connection") + dlog.Server.Debug(user, "Granting SSH connection") return nil, nil } } case config.ContinuousUser: for _, job := range config.Server.Continuous { if s.backgroundCanSSH(user, authInfo, remoteIP, job.Name, job.AllowFrom) { - logger.Debug(user, "Granting SSH connection") + dlog.Server.Debug(user, "Granting SSH connection") return nil, nil } } @@ -252,22 +252,22 @@ func (s *Server) Callback(c gossh.ConnMetadata, authPayload []byte) (*gossh.Perm } func (s *Server) backgroundCanSSH(user *user.User, jobName, remoteIP, allowedJobName string, allowFrom []string) bool { - logger.Debug("backgroundCanSSH", user, jobName, remoteIP, allowedJobName, allowFrom) + dlog.Server.Debug("backgroundCanSSH", user, jobName, remoteIP, allowedJobName, allowFrom) if jobName != allowedJobName { - logger.Debug(user, jobName, "backgroundCanSSH", "Job name does not match, skipping to next one...", allowedJobName) + dlog.Server.Debug(user, jobName, "backgroundCanSSH", "Job name does not match, skipping to next one...", allowedJobName) return false } for _, myAddr := range allowFrom { ips, err := net.LookupIP(myAddr) if err != nil { - logger.Debug(user, jobName, "backgroundCanSSH", "Unable to lookup IP address for allowed hosts lookup, skipping to next one...", myAddr, err) + dlog.Server.Debug(user, jobName, "backgroundCanSSH", "Unable to lookup IP address for allowed hosts lookup, skipping to next one...", myAddr, err) continue } for _, ip := range ips { - logger.Debug(user, jobName, "backgroundCanSSH", "Comparing IP addresses", remoteIP, ip.String()) + dlog.Server.Debug(user, jobName, "backgroundCanSSH", "Comparing IP addresses", remoteIP, ip.String()) if remoteIP == ip.String() { return true } diff --git a/internal/server/stats.go b/internal/server/stats.go index 3e8c71d..8583318 100644 --- a/internal/server/stats.go +++ b/internal/server/stats.go @@ -8,7 +8,7 @@ import ( "time" "github.com/mimecast/dtail/internal/config" - "github.com/mimecast/dtail/internal/io/logger" + "github.com/mimecast/dtail/internal/io/dlog" ) // Used to collect and display various server stats. @@ -41,7 +41,7 @@ func (s *stats) hasConnections() bool { s.mutex.Unlock() has := currentConnections > 0 - logger.Info("stats", "Server with open connections?", has, currentConnections) + dlog.Server.Info("stats", "Server with open connections?", has, currentConnections) return has } @@ -57,7 +57,7 @@ func (s *stats) logServerStats() { data["cgocalls"] = runtime.NumCgoCall() data["cpu"] = runtime.NumCPU() - logger.Mapreduce("STATS", data) + dlog.Server.Mapreduce("STATS", data) } func (s *stats) serverLimitExceeded() error { -- cgit v1.2.3 From fcaa94c7453efa0d74e330128c0f5c2cde8f11b3 Mon Sep 17 00:00:00 2001 From: Paul Buetow Date: Sun, 26 Sep 2021 16:42:47 +0300 Subject: refactor config reader - also looks in additional search paths for config file unless NONE is specified --- internal/server/continuous.go | 2 +- internal/server/handlers/readcommand.go | 2 +- internal/server/scheduler.go | 2 +- internal/server/server.go | 12 ++++++++++-- 4 files changed, 13 insertions(+), 5 deletions(-) (limited to 'internal/server') diff --git a/internal/server/continuous.go b/internal/server/continuous.go index 5f4c454..87c8889 100644 --- a/internal/server/continuous.go +++ b/internal/server/continuous.go @@ -61,7 +61,7 @@ func (c *continuous) runJob(ctx context.Context, job config.Continuous) { } args := config.Args{ - ConnectionsPerCPU: 10, + ConnectionsPerCPU: config.DefaultConnectionsPerCPU, Discovery: job.Discovery, ServersStr: servers, What: files, diff --git a/internal/server/handlers/readcommand.go b/internal/server/handlers/readcommand.go index 60ad2a0..c76ae2a 100644 --- a/internal/server/handlers/readcommand.go +++ b/internal/server/handlers/readcommand.go @@ -7,9 +7,9 @@ import ( "sync" "time" + "github.com/mimecast/dtail/internal/io/dlog" "github.com/mimecast/dtail/internal/io/fs" "github.com/mimecast/dtail/internal/io/line" - "github.com/mimecast/dtail/internal/io/dlog" "github.com/mimecast/dtail/internal/omode" "github.com/mimecast/dtail/internal/regex" ) diff --git a/internal/server/scheduler.go b/internal/server/scheduler.go index f474cc8..64e6573 100644 --- a/internal/server/scheduler.go +++ b/internal/server/scheduler.go @@ -72,7 +72,7 @@ func (s *scheduler) runJobs(ctx context.Context) { } args := config.Args{ - ConnectionsPerCPU: 10, + ConnectionsPerCPU: config.DefaultConnectionsPerCPU, Discovery: job.Discovery, ServersStr: servers, What: files, diff --git a/internal/server/server.go b/internal/server/server.go index a8f541b..d1cd57d 100644 --- a/internal/server/server.go +++ b/internal/server/server.go @@ -124,7 +124,12 @@ func (s *Server) handleConnection(ctx context.Context, conn net.Conn) { } func (s *Server) handleChannel(ctx context.Context, sshConn gossh.Conn, newChannel gossh.NewChannel) { - user := user.New(sshConn.User(), sshConn.RemoteAddr().String()) + user, err := user.New(sshConn.User(), sshConn.RemoteAddr().String()) + if err != nil { + dlog.Server.Error(user, err) + newChannel.Reject(gossh.Prohibited, err.Error()) + return + } dlog.Server.Info(user, "Invoking channel handler") if newChannel.ChannelType() != "session" { @@ -213,7 +218,10 @@ func (s *Server) handleRequests(ctx context.Context, sshConn gossh.Conn, in <-ch // Callback for SSH authentication. func (s *Server) Callback(c gossh.ConnMetadata, authPayload []byte) (*gossh.Permissions, error) { - user := user.New(c.User(), c.RemoteAddr().String()) + user, err := user.New(c.User(), c.RemoteAddr().String()) + if err != nil { + return nil, err + } if config.ServerRelaxedAuthEnable { dlog.Server.Fatal(user, "Granting permissions via relaxed-auth") -- cgit v1.2.3 From 764ef99a3d779a0db1fb60679292af52425ba2f6 Mon Sep 17 00:00:00 2001 From: Paul Buetow Date: Sat, 2 Oct 2021 10:46:47 +0300 Subject: add more default fields to MAPREDUCE --- internal/server/stats.go | 4 ---- 1 file changed, 4 deletions(-) (limited to 'internal/server') diff --git a/internal/server/stats.go b/internal/server/stats.go index 8583318..c07634d 100644 --- a/internal/server/stats.go +++ b/internal/server/stats.go @@ -3,7 +3,6 @@ package server import ( "context" "fmt" - "runtime" "sync" "time" @@ -53,9 +52,6 @@ func (s *stats) logServerStats() { data := make(map[string]interface{}) data["currentConnections"] = s.currentConnections data["lifetimeConnections"] = s.lifetimeConnections - data["goroutines"] = runtime.NumGoroutine() - data["cgocalls"] = runtime.NumCgoCall() - data["cpu"] = runtime.NumCPU() dlog.Server.Mapreduce("STATS", data) } -- cgit v1.2.3 From 12c79f68bb5bda6673819d7b754820ecfe6d08ff Mon Sep 17 00:00:00 2001 From: Paul Buetow Date: Sat, 2 Oct 2021 11:54:07 +0300 Subject: reduce logging in serverless mode --- internal/server/handlers/readcommand.go | 12 ++-- internal/server/handlers/serverhandler.go | 93 ++++++++++--------------------- 2 files changed, 36 insertions(+), 69 deletions(-) (limited to 'internal/server') diff --git a/internal/server/handlers/readcommand.go b/internal/server/handlers/readcommand.go index c76ae2a..6579018 100644 --- a/internal/server/handlers/readcommand.go +++ b/internal/server/handlers/readcommand.go @@ -32,13 +32,13 @@ func (r *readCommand) Start(ctx context.Context, argc int, args []string, retrie if argc >= 4 { deserializedRegex, err := regex.Deserialize(strings.Join(args[2:], " ")) if err != nil { - r.server.sendServerMessage(dlog.Server.Error(r.server.user, commandParseWarning, err)) + r.server.send(r.server.serverMessages, dlog.Server.Error(r.server.user, commandParseWarning, err)) return } re = deserializedRegex } if argc < 3 { - r.server.sendServerWarnMessage(dlog.Server.Warn(r.server.user, commandParseWarning, args, argc)) + r.server.send(r.server.serverMessages, dlog.Server.Warn(r.server.user, commandParseWarning, args, argc)) return } r.readGlob(ctx, args[1], re, retries) @@ -58,7 +58,7 @@ func (r *readCommand) readGlob(ctx context.Context, glob string, re regex.Regex, if numPaths := len(paths); numPaths == 0 { dlog.Server.Error(r.server.user, "No such file(s) to read", glob) - r.server.sendServerWarnMessage(dlog.Server.Warn(r.server.user, "Unable to read file(s), check server logs")) + r.server.send(r.server.serverMessages, dlog.Server.Warn(r.server.user, "Unable to read file(s), check server logs")) select { case <-ctx.Done(): return @@ -72,7 +72,7 @@ func (r *readCommand) readGlob(ctx context.Context, glob string, re regex.Regex, return } - r.server.sendServerWarnMessage(dlog.Server.Warn(r.server.user, "Giving up to read file(s)")) + r.server.send(r.server.serverMessages, dlog.Server.Warn(r.server.user, "Giving up to read file(s)")) return } @@ -93,7 +93,7 @@ func (r *readCommand) readFileIfPermissions(ctx context.Context, wg *sync.WaitGr if !r.server.user.HasFilePermission(path, "readfiles") { dlog.Server.Error(r.server.user, "No permission to read file", path, globID) - r.server.sendServerWarnMessage(dlog.Server.Warn(r.server.user, "Unable to read file(s), check server logs")) + r.server.send(r.server.serverMessages, dlog.Server.Warn(r.server.user, "Unable to read file(s), check server logs")) return } @@ -161,6 +161,6 @@ func (r *readCommand) makeGlobID(path, glob string) string { return pathParts[len(pathParts)-1] } - r.server.sendServerWarnMessage(dlog.Server.Warn("Empty file path given?", path, glob)) + r.server.send(r.server.serverMessages, dlog.Server.Warn("Empty file path given?", path, glob)) return "" } diff --git a/internal/server/handlers/serverhandler.go b/internal/server/handlers/serverhandler.go index b664566..ace2626 100644 --- a/internal/server/handlers/serverhandler.go +++ b/internal/server/handlers/serverhandler.go @@ -15,8 +15,8 @@ import ( "github.com/mimecast/dtail/internal" "github.com/mimecast/dtail/internal/config" - "github.com/mimecast/dtail/internal/io/line" "github.com/mimecast/dtail/internal/io/dlog" + "github.com/mimecast/dtail/internal/io/line" "github.com/mimecast/dtail/internal/io/pool" "github.com/mimecast/dtail/internal/mapr/server" "github.com/mimecast/dtail/internal/omode" @@ -46,6 +46,7 @@ type ServerHandler struct { activeCommands int32 quiet bool spartan bool + serverless bool readBuf bytes.Buffer writeBuf bytes.Buffer } @@ -99,6 +100,12 @@ func (h *ServerHandler) Read(p []byte) (n int, err error) { return } + if h.serverless { + // In serverless mode we have logged the server message already via the + // dlog logger, no need to send the message again to the client part. + return + } + // Handle normal server message (display to the user) h.readBuf.WriteString("SERVER") h.readBuf.WriteString(protocol.FieldDelimiter) @@ -266,23 +273,24 @@ func (h *ServerHandler) handleUserCommand(ctx context.Context, argc int, args [] splitted := strings.Split(args[0], ":") commandName := splitted[0] - options, err := readOptions(splitted[1:]) + options, err := config.DeserializeOptions(splitted[1:]) if err != nil { - h.sendServerMessage(dlog.Server.Error(h.user, err)) + h.send(h.serverMessages, dlog.Server.Error(h.user, err)) commandFinished() return } - if quiet, ok := options["quiet"]; ok { - if quiet == "true" { - dlog.Server.Debug(h.user, "Enabling quiet mode") - h.quiet = true - } + + if quiet, _ := options["quiet"]; quiet == "true" { + dlog.Server.Debug(h.user, "Enabling quiet mode") + h.quiet = true } - if spartan, ok := options["spartan"]; ok { - if spartan == "true" { - dlog.Server.Debug(h.user, "Enabling spartan mode") - h.spartan = true - } + if spartan, _ := options["spartan"]; spartan == "true" { + dlog.Server.Debug(h.user, "Enabling spartan mode") + h.spartan = true + } + if serverless, _ := options["serverless"]; serverless == "true" { + dlog.Server.Debug(h.user, "Enabling serverless mode") + h.serverless = true } switch commandName { @@ -303,7 +311,7 @@ func (h *ServerHandler) handleUserCommand(ctx context.Context, argc int, args [] case "map": command, aggregate, err := newMapCommand(h, argc, args) if err != nil { - h.sendServerMessage(err.Error()) + h.send(h.serverMessages, err.Error()) dlog.Server.Error(h.user, err) commandFinished() return @@ -320,14 +328,16 @@ func (h *ServerHandler) handleUserCommand(ctx context.Context, argc int, args [] commandFinished() default: - h.sendServerMessage(dlog.Server.Error(h.user, "Received unknown user command", commandName, argc, args, options)) + h.send(h.serverMessages, dlog.Server.Error(h.user, "Received unknown user command", commandName, argc, args, options)) commandFinished() } } func (h *ServerHandler) handleAckCommand(argc int, args []string) { if argc < 3 { - h.sendServerWarnMessage(dlog.Server.Warn(h.user, commandParseWarning, args, argc)) + if !h.quiet { + h.send(h.serverMessages, dlog.Server.Warn(h.user, commandParseWarning, args, argc)) + } return } if args[1] == "close" && args[2] == "connection" { @@ -346,23 +356,8 @@ func (h *ServerHandler) send(ch chan<- string, message string) { } } -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 -} - -func (h *ServerHandler) flushMessages() { - dlog.Server.Debug(h.user, "flushMessages()") +func (h *ServerHandler) flush() { + dlog.Server.Debug(h.user, "flush()") unsentMessages := func() int { return len(h.lines) + len(h.serverMessages) + len(h.maprMessages) @@ -381,11 +376,11 @@ func (h *ServerHandler) flushMessages() { func (h *ServerHandler) shutdown() { dlog.Server.Debug(h.user, "shutdown()") - h.flushMessages() + h.flush() go func() { select { - case h.serverMessageC() <- ".syn close connection": + case h.serverMessages <- ".syn close connection": case <-h.done.Done(): } }() @@ -408,31 +403,3 @@ func (h *ServerHandler) decrementActiveCommands() int32 { atomic.AddInt32(&h.activeCommands, -1) return atomic.LoadInt32(&h.activeCommands) } - -func readOptions(opts []string) (map[string]string, error) { - dlog.Server.Debug("Parsing options", opts) - options := make(map[string]string, len(opts)) - - for _, o := range opts { - kv := strings.SplitN(o, "=", 2) - if len(kv) != 2 { - return options, fmt.Errorf("Unable to parse options: %v", kv) - } - key := kv[0] - val := kv[1] - - if strings.HasPrefix(val, "base64%") { - s := strings.SplitN(val, "%", 2) - decoded, err := base64.StdEncoding.DecodeString(s[1]) - if err != nil { - return options, err - } - val = string(decoded) - } - - dlog.Server.Debug("Setting option", key, val) - options[key] = val - } - - return options, nil -} -- cgit v1.2.3 From f70622f307629a2542ea5eb128dea8c1043d3a40 Mon Sep 17 00:00:00 2001 From: Paul Buetow Date: Tue, 5 Oct 2021 10:00:38 +0300 Subject: more on this --- internal/server/handlers/basehandler.go | 283 +++++++++++++++++++++++++ internal/server/handlers/readcommand.go | 4 +- internal/server/handlers/serverhandler.go | 332 +++--------------------------- 3 files changed, 319 insertions(+), 300 deletions(-) create mode 100644 internal/server/handlers/basehandler.go (limited to 'internal/server') diff --git a/internal/server/handlers/basehandler.go b/internal/server/handlers/basehandler.go new file mode 100644 index 0000000..12fb2b3 --- /dev/null +++ b/internal/server/handlers/basehandler.go @@ -0,0 +1,283 @@ +package handlers + +import ( + "bytes" + "context" + "encoding/base64" + "errors" + "fmt" + "io" + "strconv" + "strings" + "sync/atomic" + "time" + + "github.com/mimecast/dtail/internal" + "github.com/mimecast/dtail/internal/io/dlog" + "github.com/mimecast/dtail/internal/io/line" + "github.com/mimecast/dtail/internal/io/pool" + "github.com/mimecast/dtail/internal/mapr/server" + "github.com/mimecast/dtail/internal/protocol" + user "github.com/mimecast/dtail/internal/user/server" +) + +type handleCommandCb func(context.Context, int, []string) + +type baseHandler struct { + done *internal.Done + handleCommandCb handleCommandCb + lines chan line.Line + aggregate *server.Aggregate + maprMessages chan string + serverMessages chan string + hostname string + user *user.User + ackCloseReceived chan struct{} + activeCommands int32 + quiet bool + spartan bool + serverless bool + readBuf bytes.Buffer + writeBuf bytes.Buffer +} + +// Shutdown the handler. +func (h *baseHandler) Shutdown() { + h.done.Shutdown() +} + +// Done channel of the handler. +func (h *baseHandler) Done() <-chan struct{} { + return h.done.Done() +} + +// Read is to send data to the dtail client via Reader interface. +func (h *baseHandler) Read(p []byte) (n int, err error) { + defer h.readBuf.Reset() + + select { + case message := <-h.serverMessages: + if message[0] == '.' { + // Handle hidden message (don't display to the user, interpreted by dtail client) + h.readBuf.WriteString(message) + h.readBuf.WriteByte(protocol.MessageDelimiter) + n = copy(p, h.readBuf.Bytes()) + return + } + + if h.serverless { + // In serverless mode we have logged the server message already via the + // dlog logger, no need to send the message again to the client part. + return + } + + // Handle normal server message (display to the user) + h.readBuf.WriteString("SERVER") + h.readBuf.WriteString(protocol.FieldDelimiter) + h.readBuf.WriteString(h.hostname) + h.readBuf.WriteString(protocol.FieldDelimiter) + h.readBuf.WriteString(message) + h.readBuf.WriteByte(protocol.MessageDelimiter) + n = copy(p, h.readBuf.Bytes()) + + case message := <-h.maprMessages: + // Send mapreduce-aggregated data as a message. + h.readBuf.WriteString("AGGREGATE") + h.readBuf.WriteString(protocol.FieldDelimiter) + h.readBuf.WriteString(h.hostname) + h.readBuf.WriteString(protocol.FieldDelimiter) + h.readBuf.WriteString(message) + h.readBuf.WriteByte(protocol.MessageDelimiter) + n = copy(p, h.readBuf.Bytes()) + + case line := <-h.lines: + if !h.spartan { + h.readBuf.WriteString("REMOTE") + h.readBuf.WriteString(protocol.FieldDelimiter) + h.readBuf.WriteString(h.hostname) + h.readBuf.WriteString(protocol.FieldDelimiter) + h.readBuf.WriteString(fmt.Sprintf("%3d", line.TransmittedPerc)) + h.readBuf.WriteString(protocol.FieldDelimiter) + h.readBuf.WriteString(fmt.Sprintf("%v", line.Count)) + h.readBuf.WriteString(protocol.FieldDelimiter) + h.readBuf.WriteString(line.SourceID) + h.readBuf.WriteString(protocol.FieldDelimiter) + } + h.readBuf.WriteString(line.Content.String()) + h.readBuf.WriteByte(protocol.MessageDelimiter) + n = copy(p, h.readBuf.Bytes()) + pool.RecycleBytesBuffer(line.Content) + + case <-time.After(time.Second): + // Once in a while check whether we are done. + select { + case <-h.done.Done(): + err = io.EOF + return + default: + } + } + return +} + +// Write is to receive data from the dtail client via Writer interface. +func (h *baseHandler) Write(p []byte) (n int, err error) { + for _, b := range p { + switch b { + case ';': + h.handleCommand(string(h.writeBuf.Bytes())) + h.writeBuf.Reset() + default: + h.writeBuf.WriteByte(b) + } + } + + n = len(p) + return +} + +func (h *baseHandler) handleCommand(commandStr string) { + dlog.Server.Debug(h.user, commandStr) + + args, argc, add, err := h.handleProtocolVersion(strings.Split(commandStr, " ")) + if err != nil { + h.send(h.serverMessages, dlog.Server.Error(h.user, err)+add) + return + } + + args, argc, err = h.handleBase64(args, argc) + if err != nil { + h.send(h.serverMessages, dlog.Server.Error(h.user, err)) + return + } + + ctx, cancel := context.WithCancel(context.Background()) + go func() { + <-h.done.Done() + cancel() + }() + + h.handleCommandCb(ctx, argc, args) +} + +func (h *baseHandler) handleProtocolVersion(args []string) ([]string, int, string, error) { + argc := len(args) + var add string + + if argc <= 2 || args[0] != "protocol" { + return args, argc, add, errors.New("unable to determine protocol version") + } + + if args[1] != protocol.ProtocolCompat { + clientCompat, _ := strconv.Atoi(args[1]) + serverCompat, _ := strconv.Atoi(protocol.ProtocolCompat) + if clientCompat <= 3 { + // Protocol version 3 or lower expect a newline as message separator + // One day (after 2 major versions) this exception may be removed! + add = "\n" + } + + toUpdate := "client" + if clientCompat > serverCompat { + toUpdate = "server" + } + + err := fmt.Errorf("DTail server protocol version '%s' does not match client protocol version '%s', please update DTail %s!", + protocol.ProtocolCompat, args[1], toUpdate) + return args, argc, add, err + } + + return args[2:], argc - 2, add, nil +} + +func (h *baseHandler) handleBase64(args []string, argc int) ([]string, int, error) { + err := errors.New("Unable to decode client message, DTail server and client versions may not be compatible") + + if argc != 2 || args[0] != "base64" { + return args, argc, err + } + + decoded, err := base64.StdEncoding.DecodeString(args[1]) + if err != nil { + return args, argc, err + } + decodedStr := string(decoded) + + args = strings.Split(decodedStr, " ") + argc = len(decodedStr) + dlog.Server.Trace(h.user, "Base64 decoded received command", decodedStr, argc, args) + + return args, argc, nil +} + +func (h *baseHandler) handleAckCommand(argc int, args []string) { + if argc < 3 { + if !h.quiet { + h.send(h.serverMessages, dlog.Server.Warn(h.user, "Unable to parse command", args, argc)) + } + return + } + if args[1] == "close" && args[2] == "connection" { + select { + case <-h.ackCloseReceived: + default: + close(h.ackCloseReceived) + } + } +} + +func (h *baseHandler) send(ch chan<- string, message string) { + select { + case ch <- message: + case <-h.done.Done(): + } +} + +func (h *baseHandler) flush() { + dlog.Server.Debug(h.user, "flush()") + + numUnsentMessages := func() int { + return len(h.lines) + len(h.serverMessages) + len(h.maprMessages) + } + + for i := 0; i < 3; i++ { + if numUnsentMessages() == 0 { + dlog.Server.Debug(h.user, "All lines sent") + return + } + dlog.Server.Debug(h.user, "Still lines to be sent") + time.Sleep(time.Second) + } + + dlog.Server.Warn(h.user, "Some lines remain unsent", numUnsentMessages()) +} + +func (h *baseHandler) shutdown() { + dlog.Server.Debug(h.user, "shutdown()") + h.flush() + + go func() { + select { + case h.serverMessages <- ".syn close connection": + case <-h.done.Done(): + } + }() + + select { + case <-h.ackCloseReceived: + case <-time.After(time.Second * 5): + dlog.Server.Debug(h.user, "Shutdown timeout reached, enforcing shutdown") + case <-h.done.Done(): + } + + h.done.Shutdown() +} + +func (h *baseHandler) incrementActiveCommands() { + atomic.AddInt32(&h.activeCommands, 1) +} + +func (h *baseHandler) decrementActiveCommands() int32 { + atomic.AddInt32(&h.activeCommands, -1) + return atomic.LoadInt32(&h.activeCommands) +} diff --git a/internal/server/handlers/readcommand.go b/internal/server/handlers/readcommand.go index 6579018..abc44c7 100644 --- a/internal/server/handlers/readcommand.go +++ b/internal/server/handlers/readcommand.go @@ -32,13 +32,13 @@ func (r *readCommand) Start(ctx context.Context, argc int, args []string, retrie if argc >= 4 { deserializedRegex, err := regex.Deserialize(strings.Join(args[2:], " ")) if err != nil { - r.server.send(r.server.serverMessages, dlog.Server.Error(r.server.user, commandParseWarning, err)) + r.server.send(r.server.serverMessages, dlog.Server.Error(r.server.user, "Unable to parse command", err)) return } re = deserializedRegex } if argc < 3 { - r.server.send(r.server.serverMessages, dlog.Server.Warn(r.server.user, commandParseWarning, args, argc)) + r.server.send(r.server.serverMessages, dlog.Server.Warn(r.server.user, "Unable to parse command", args, argc)) return } r.readGlob(ctx, args[1], re, retries) diff --git a/internal/server/handlers/serverhandler.go b/internal/server/handlers/serverhandler.go index ace2626..2ec4fbf 100644 --- a/internal/server/handlers/serverhandler.go +++ b/internal/server/handlers/serverhandler.go @@ -1,69 +1,60 @@ package handlers import ( - "bytes" "context" - "encoding/base64" - "errors" - "fmt" - "io" "os" - "strconv" "strings" - "sync/atomic" - "time" "github.com/mimecast/dtail/internal" "github.com/mimecast/dtail/internal/config" "github.com/mimecast/dtail/internal/io/dlog" "github.com/mimecast/dtail/internal/io/line" - "github.com/mimecast/dtail/internal/io/pool" - "github.com/mimecast/dtail/internal/mapr/server" "github.com/mimecast/dtail/internal/omode" - "github.com/mimecast/dtail/internal/protocol" user "github.com/mimecast/dtail/internal/user/server" ) -const ( - commandParseWarning string = "Unable to parse command" -) - // ServerHandler implements the Reader and Writer interfaces to handle // 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 - maprMessages chan string - serverMessages chan string - hostname string - user *user.User - catLimiter chan struct{} - tailLimiter chan struct{} - ackCloseReceived chan struct{} - activeCommands int32 - quiet bool - spartan bool - serverless bool - readBuf bytes.Buffer - writeBuf bytes.Buffer + baseHandler + catLimiter chan struct{} + tailLimiter chan struct{} + regex string + /* + done *internal.Done + lines chan line.Line + aggregate *server.Aggregate + maprMessages chan string + serverMessages chan string + hostname string + user *user.User + ackCloseReceived chan struct{} + activeCommands int32 + quiet bool + spartan bool + serverless bool + readBuf bytes.Buffer + writeBuf bytes.Buffer + */ } // NewServerHandler returns the server handler. 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), - maprMessages: make(chan string, 10), - ackCloseReceived: make(chan struct{}), - catLimiter: catLimiter, - tailLimiter: tailLimiter, - regex: ".", - user: user, - } + baseHandler: baseHandler{ + done: internal.NewDone(), + lines: make(chan line.Line, 100), + serverMessages: make(chan string, 10), + maprMessages: make(chan string, 10), + ackCloseReceived: make(chan struct{}), + user: user, + }, + catLimiter: catLimiter, + tailLimiter: tailLimiter, + regex: ".", + } + h.handleCommandCb = h.handleUserCommand fqdn, err := os.Hostname() if err != nil { @@ -76,192 +67,8 @@ func NewServerHandler(user *user.User, catLimiter, tailLimiter chan struct{}) *S 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. -func (h *ServerHandler) Read(p []byte) (n int, err error) { - defer h.readBuf.Reset() - - select { - case message := <-h.serverMessages: - if message[0] == '.' { - // Handle hidden message (don't display to the user, interpreted by dtail client) - h.readBuf.WriteString(message) - h.readBuf.WriteByte(protocol.MessageDelimiter) - n = copy(p, h.readBuf.Bytes()) - return - } - - if h.serverless { - // In serverless mode we have logged the server message already via the - // dlog logger, no need to send the message again to the client part. - return - } - - // Handle normal server message (display to the user) - h.readBuf.WriteString("SERVER") - h.readBuf.WriteString(protocol.FieldDelimiter) - h.readBuf.WriteString(h.hostname) - h.readBuf.WriteString(protocol.FieldDelimiter) - h.readBuf.WriteString(message) - h.readBuf.WriteByte(protocol.MessageDelimiter) - n = copy(p, h.readBuf.Bytes()) - - case message := <-h.maprMessages: - // Send mapreduce-aggregated data as a message. - h.readBuf.WriteString("AGGREGATE") - h.readBuf.WriteString(protocol.FieldDelimiter) - h.readBuf.WriteString(h.hostname) - h.readBuf.WriteString(protocol.FieldDelimiter) - h.readBuf.WriteString(message) - h.readBuf.WriteByte(protocol.MessageDelimiter) - n = copy(p, h.readBuf.Bytes()) - - case line := <-h.lines: - if !h.spartan { - h.readBuf.WriteString("REMOTE") - h.readBuf.WriteString(protocol.FieldDelimiter) - h.readBuf.WriteString(h.hostname) - h.readBuf.WriteString(protocol.FieldDelimiter) - h.readBuf.WriteString(fmt.Sprintf("%3d", line.TransmittedPerc)) - h.readBuf.WriteString(protocol.FieldDelimiter) - h.readBuf.WriteString(fmt.Sprintf("%v", line.Count)) - h.readBuf.WriteString(protocol.FieldDelimiter) - h.readBuf.WriteString(line.SourceID) - h.readBuf.WriteString(protocol.FieldDelimiter) - } - h.readBuf.WriteString(line.Content.String()) - h.readBuf.WriteByte(protocol.MessageDelimiter) - n = copy(p, h.readBuf.Bytes()) - pool.RecycleBytesBuffer(line.Content) - - case <-time.After(time.Second): - // Once in a while check whether we are done. - select { - case <-h.done.Done(): - err = io.EOF - return - default: - } - } - return -} - -// Write is to receive data from the dtail client via Writer interface. -func (h *ServerHandler) Write(p []byte) (n int, err error) { - for _, b := range p { - switch b { - case ';': - h.handleCommand(string(h.writeBuf.Bytes())) - h.writeBuf.Reset() - default: - h.writeBuf.WriteByte(b) - } - } - - n = len(p) - return -} - -func (h *ServerHandler) handleCommand(commandStr string) { - dlog.Server.Debug(h.user, commandStr) - ctx := context.Background() - - args, argc, add, err := h.handleProtocolVersion(strings.Split(commandStr, " ")) - if err != nil { - h.send(h.serverMessages, dlog.Server.Error(h.user, err)+add) - return - } - - args, argc, err = h.handleBase64(args, argc) - if err != nil { - h.send(h.serverMessages, dlog.Server.Error(h.user, err)) - return - } - - if h.user.Name == config.ControlUser { - h.handleControlCommand(argc, args) - return - } - - ctx, cancel := context.WithCancel(ctx) - go func() { - <-h.done.Done() - cancel() - }() - - h.handleUserCommand(ctx, argc, args) -} - -func (h *ServerHandler) handleProtocolVersion(args []string) ([]string, int, string, error) { - argc := len(args) - var add string - - if argc <= 2 || args[0] != "protocol" { - return args, argc, add, errors.New("unable to determine protocol version") - } - - if args[1] != protocol.ProtocolCompat { - clientCompat, _ := strconv.Atoi(args[1]) - serverCompat, _ := strconv.Atoi(protocol.ProtocolCompat) - if clientCompat <= 3 { - // Protocol version 3 or lower expect a newline as message separator - // One day (after 2 major versions) this exception may be removed! - add = "\n" - } - - toUpdate := "client" - if clientCompat > serverCompat { - toUpdate = "server" - } - - err := fmt.Errorf("DTail server protocol version '%s' does not match client protocol version '%s', please update DTail %s!", - protocol.ProtocolCompat, args[1], toUpdate) - return args, argc, add, err - } - - return args[2:], argc - 2, add, nil -} - -func (h *ServerHandler) handleBase64(args []string, argc int) ([]string, int, error) { - err := errors.New("Unable to decode client message, DTail server and client versions may not be compatible") - - if argc != 2 || args[0] != "base64" { - return args, argc, err - } - - decoded, err := base64.StdEncoding.DecodeString(args[1]) - if err != nil { - return args, argc, err - } - decodedStr := string(decoded) - - args = strings.Split(decodedStr, " ") - argc = len(decodedStr) - dlog.Server.Trace(h.user, "Base64 decoded received command", decodedStr, argc, args) - - return args, argc, nil -} - -func (h *ServerHandler) handleControlCommand(argc int, args []string) { - switch args[0] { - case "debug": - h.send(h.serverMessages, dlog.Server.Debug(h.user, "Receiving debug command", argc, args)) - default: - dlog.Server.Warn(h.user, "Received unknown control command", argc, args) - } -} - func (h *ServerHandler) handleUserCommand(ctx context.Context, argc int, args []string) { - dlog.Server.Debug(h.user, "handleUserCommand", argc, args) + dlog.Server.Debug(h.user, "Handling user command", argc, args) h.incrementActiveCommands() commandFinished := func() { @@ -332,74 +139,3 @@ func (h *ServerHandler) handleUserCommand(ctx context.Context, argc int, args [] commandFinished() } } - -func (h *ServerHandler) handleAckCommand(argc int, args []string) { - if argc < 3 { - if !h.quiet { - h.send(h.serverMessages, dlog.Server.Warn(h.user, commandParseWarning, args, argc)) - } - return - } - if args[1] == "close" && args[2] == "connection" { - select { - case <-h.ackCloseReceived: - default: - close(h.ackCloseReceived) - } - } -} - -func (h *ServerHandler) send(ch chan<- string, message string) { - select { - case ch <- message: - case <-h.done.Done(): - } -} - -func (h *ServerHandler) flush() { - dlog.Server.Debug(h.user, "flush()") - - unsentMessages := func() int { - return len(h.lines) + len(h.serverMessages) + len(h.maprMessages) - } - for i := 0; i < 3; i++ { - if unsentMessages() == 0 { - dlog.Server.Debug(h.user, "All lines sent") - return - } - dlog.Server.Debug(h.user, "Still lines to be sent") - time.Sleep(time.Second) - } - - dlog.Server.Warn(h.user, "Some lines remain unsent", unsentMessages()) -} - -func (h *ServerHandler) shutdown() { - dlog.Server.Debug(h.user, "shutdown()") - h.flush() - - go func() { - select { - case h.serverMessages <- ".syn close connection": - case <-h.done.Done(): - } - }() - - select { - case <-h.ackCloseReceived: - case <-time.After(time.Second * 5): - dlog.Server.Debug(h.user, "Shutdown timeout reached, enforcing shutdown") - case <-h.done.Done(): - } - - h.done.Shutdown() -} - -func (h *ServerHandler) incrementActiveCommands() { - atomic.AddInt32(&h.activeCommands, 1) -} - -func (h *ServerHandler) decrementActiveCommands() int32 { - atomic.AddInt32(&h.activeCommands, -1) - return atomic.LoadInt32(&h.activeCommands) -} -- cgit v1.2.3 From 9f395a03f25941d8ed98ec43035688daa1e8877f Mon Sep 17 00:00:00 2001 From: Paul Buetow Date: Tue, 5 Oct 2021 22:39:58 +0300 Subject: more on this --- internal/server/handlers/basehandler.go | 2 +- internal/server/handlers/serverhandler.go | 16 ---------------- 2 files changed, 1 insertion(+), 17 deletions(-) (limited to 'internal/server') diff --git a/internal/server/handlers/basehandler.go b/internal/server/handlers/basehandler.go index 12fb2b3..b683578 100644 --- a/internal/server/handlers/basehandler.go +++ b/internal/server/handlers/basehandler.go @@ -242,7 +242,7 @@ func (h *baseHandler) flush() { for i := 0; i < 3; i++ { if numUnsentMessages() == 0 { - dlog.Server.Debug(h.user, "All lines sent") + dlog.Server.Debug(h.user, "All lines sent", fmt.Sprintf("%p", h)) return } dlog.Server.Debug(h.user, "Still lines to be sent") diff --git a/internal/server/handlers/serverhandler.go b/internal/server/handlers/serverhandler.go index 2ec4fbf..25cb8ba 100644 --- a/internal/server/handlers/serverhandler.go +++ b/internal/server/handlers/serverhandler.go @@ -21,22 +21,6 @@ type ServerHandler struct { catLimiter chan struct{} tailLimiter chan struct{} regex string - /* - done *internal.Done - lines chan line.Line - aggregate *server.Aggregate - maprMessages chan string - serverMessages chan string - hostname string - user *user.User - ackCloseReceived chan struct{} - activeCommands int32 - quiet bool - spartan bool - serverless bool - readBuf bytes.Buffer - writeBuf bytes.Buffer - */ } // NewServerHandler returns the server handler. -- cgit v1.2.3 From fab5dc3e70434ea0abc7a0976487a1973b662331 Mon Sep 17 00:00:00 2001 From: Paul Buetow Date: Wed, 6 Oct 2021 09:50:41 +0300 Subject: enable faster shutdown - useful for dgrep/dmap and dcat commands --- internal/server/handlers/basehandler.go | 20 ++++-- internal/server/handlers/controlhandler.go | 98 ------------------------------ internal/server/handlers/healthhandler.go | 58 ++++++++++++++++++ internal/server/handlers/serverhandler.go | 16 ++--- internal/server/server.go | 10 +-- 5 files changed, 82 insertions(+), 120 deletions(-) delete mode 100644 internal/server/handlers/controlhandler.go create mode 100644 internal/server/handlers/healthhandler.go (limited to 'internal/server') diff --git a/internal/server/handlers/basehandler.go b/internal/server/handlers/basehandler.go index b683578..4fa8f00 100644 --- a/internal/server/handlers/basehandler.go +++ b/internal/server/handlers/basehandler.go @@ -13,6 +13,7 @@ import ( "time" "github.com/mimecast/dtail/internal" + "github.com/mimecast/dtail/internal/config" "github.com/mimecast/dtail/internal/io/dlog" "github.com/mimecast/dtail/internal/io/line" "github.com/mimecast/dtail/internal/io/pool" @@ -21,7 +22,7 @@ import ( user "github.com/mimecast/dtail/internal/user/server" ) -type handleCommandCb func(context.Context, int, []string) +type handleCommandCb func(context.Context, int, []string, string, map[string]string) type baseHandler struct { done *internal.Done @@ -157,7 +158,16 @@ func (h *baseHandler) handleCommand(commandStr string) { cancel() }() - h.handleCommandCb(ctx, argc, args) + splitted := strings.Split(args[0], ":") + commandName := splitted[0] + + options, err := config.DeserializeOptions(splitted[1:]) + if err != nil { + h.send(h.serverMessages, dlog.Server.Error(h.user, err)) + return + } + + h.handleCommandCb(ctx, argc, args, commandName, options) } func (h *baseHandler) handleProtocolVersion(args []string) ([]string, int, string, error) { @@ -234,19 +244,19 @@ func (h *baseHandler) send(ch chan<- string, message string) { } func (h *baseHandler) flush() { - dlog.Server.Debug(h.user, "flush()") + dlog.Server.Trace(h.user, "flush()") numUnsentMessages := func() int { return len(h.lines) + len(h.serverMessages) + len(h.maprMessages) } - for i := 0; i < 3; i++ { + for i := 0; i < 10; i++ { if numUnsentMessages() == 0 { dlog.Server.Debug(h.user, "All lines sent", fmt.Sprintf("%p", h)) return } dlog.Server.Debug(h.user, "Still lines to be sent") - time.Sleep(time.Second) + time.Sleep(time.Millisecond * 10) } dlog.Server.Warn(h.user, "Some lines remain unsent", numUnsentMessages()) diff --git a/internal/server/handlers/controlhandler.go b/internal/server/handlers/controlhandler.go deleted file mode 100644 index ae70675..0000000 --- a/internal/server/handlers/controlhandler.go +++ /dev/null @@ -1,98 +0,0 @@ -package handlers - -import ( - "fmt" - "io" - "os" - "strings" - - "github.com/mimecast/dtail/internal" - "github.com/mimecast/dtail/internal/io/dlog" - user "github.com/mimecast/dtail/internal/user/server" -) - -// ControlHandler is used for control functions and health monitoring. -type ControlHandler struct { - done *internal.Done - hostname string - payload []byte - serverMessages chan string - user *user.User -} - -// NewControlHandler returns a new control handler. -func NewControlHandler(user *user.User) *ControlHandler { - dlog.Server.Debug(user, "Creating control handler") - - h := ControlHandler{ - done: internal.NewDone(), - serverMessages: make(chan string, 10), - user: user, - } - - fqdn, err := os.Hostname() - if err != nil { - dlog.Server.FatalPanic(err) - } - - s := strings.Split(fqdn, ".") - h.hostname = s[0] - - 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. -func (h *ControlHandler) Read(p []byte) (n int, err error) { - for { - select { - case message := <-h.serverMessages: - wholePayload := []byte(fmt.Sprintf("SERVER|%s|%s\n", h.hostname, message)) - n = copy(p, wholePayload) - return - case <-h.done.Done(): - return 0, io.EOF - } - } -} - -// Write is to read data to the client via the Writer interface. -func (h *ControlHandler) Write(p []byte) (n int, err error) { - for _, c := range p { - switch c { - case ';': - wholePayload := strings.TrimSpace(string(h.payload)) - h.handleCommand(wholePayload) - h.payload = nil - - default: - h.payload = append(h.payload, c) - } - } - - n = len(p) - return -} - -func (h *ControlHandler) handleCommand(command string) { - dlog.Server.Info(h.user, command) - s := strings.Split(command, " ") - dlog.Server.Debug(h.user, "Receiving command", command, s) - - switch s[0] { - case "health": - h.serverMessages <- "OK: DTail SSH Server seems fine" - h.serverMessages <- "done;" - default: - h.serverMessages <- dlog.Server.Error(h.user, "Received unknown control command", command, s) - } -} diff --git a/internal/server/handlers/healthhandler.go b/internal/server/handlers/healthhandler.go new file mode 100644 index 0000000..3f3b932 --- /dev/null +++ b/internal/server/handlers/healthhandler.go @@ -0,0 +1,58 @@ +package handlers + +import ( + "context" + "os" + "strings" + + "github.com/mimecast/dtail/internal" + "github.com/mimecast/dtail/internal/io/dlog" + "github.com/mimecast/dtail/internal/io/line" + user "github.com/mimecast/dtail/internal/user/server" +) + +// HealthHandler is for the remote health check. +type HealthHandler struct { + baseHandler +} + +// NewHealthHandler returns the server handler. +func NewHealthHandler(user *user.User) *HealthHandler { + dlog.Server.Debug(user, "Creating new server health handler") + h := HealthHandler{ + baseHandler: baseHandler{ + done: internal.NewDone(), + lines: make(chan line.Line, 100), + serverMessages: make(chan string, 10), + maprMessages: make(chan string, 10), + ackCloseReceived: make(chan struct{}), + user: user, + }, + } + h.handleCommandCb = h.handleHealthCommand + + fqdn, err := os.Hostname() + if err != nil { + dlog.Server.FatalPanic(err) + } + + s := strings.Split(fqdn, ".") + h.hostname = s[0] + + return &h +} + +func (h *HealthHandler) handleHealthCommand(ctx context.Context, argc int, args []string, + commandName string, options map[string]string) { + dlog.Server.Debug(h.user, "Handling health command", argc, args) + + switch commandName { + case "health": + h.send(h.serverMessages, "OK: DTail SSH Server seems fine") + case "ack", ".ack": + h.handleAckCommand(argc, args) + default: + h.send(h.serverMessages, dlog.Server.Error(h.user, "Received unknown health command", commandName, argc, args)) + } + h.shutdown() +} diff --git a/internal/server/handlers/serverhandler.go b/internal/server/handlers/serverhandler.go index 25cb8ba..aaffe14 100644 --- a/internal/server/handlers/serverhandler.go +++ b/internal/server/handlers/serverhandler.go @@ -6,7 +6,6 @@ import ( "strings" "github.com/mimecast/dtail/internal" - "github.com/mimecast/dtail/internal/config" "github.com/mimecast/dtail/internal/io/dlog" "github.com/mimecast/dtail/internal/io/line" "github.com/mimecast/dtail/internal/omode" @@ -25,6 +24,7 @@ type ServerHandler struct { // NewServerHandler returns the server handler. func NewServerHandler(user *user.User, catLimiter, tailLimiter chan struct{}) *ServerHandler { + dlog.Server.Debug(user, "Creating new server handler") h := ServerHandler{ baseHandler: baseHandler{ done: internal.NewDone(), @@ -51,7 +51,9 @@ func NewServerHandler(user *user.User, catLimiter, tailLimiter chan struct{}) *S return &h } -func (h *ServerHandler) handleUserCommand(ctx context.Context, argc int, args []string) { +func (h *ServerHandler) handleUserCommand(ctx context.Context, argc int, args []string, + commandName string, options map[string]string) { + dlog.Server.Debug(h.user, "Handling user command", argc, args) h.incrementActiveCommands() @@ -61,16 +63,6 @@ func (h *ServerHandler) handleUserCommand(ctx context.Context, argc int, args [] } } - splitted := strings.Split(args[0], ":") - commandName := splitted[0] - - options, err := config.DeserializeOptions(splitted[1:]) - if err != nil { - h.send(h.serverMessages, dlog.Server.Error(h.user, err)) - commandFinished() - return - } - if quiet, _ := options["quiet"]; quiet == "true" { dlog.Server.Debug(h.user, "Enabling quiet mode") h.quiet = true diff --git a/internal/server/server.go b/internal/server/server.go index d1cd57d..b3d4bff 100644 --- a/internal/server/server.go +++ b/internal/server/server.go @@ -162,8 +162,8 @@ func (s *Server) handleRequests(ctx context.Context, sshConn gossh.Conn, in <-ch case "shell": var handler handlers.Handler switch user.Name { - case config.ControlUser: - handler = handlers.NewControlHandler(user) + case config.HealthUser: + handler = handlers.NewHealthHandler(user) default: handler = handlers.NewServerHandler(user, s.catLimiter, s.tailLimiter) } @@ -234,9 +234,9 @@ func (s *Server) Callback(c gossh.ConnMetadata, authPayload []byte) (*gossh.Perm remoteIP := splitted[0] switch user.Name { - case config.ControlUser: - if authInfo == config.ControlUser { - dlog.Server.Debug(user, "Granting permissions to control user") + case config.HealthUser: + if authInfo == config.HealthUser { + dlog.Server.Debug(user, "Granting permissions to health user") return nil, nil } case config.ScheduleUser: -- cgit v1.2.3 From 7306afe9ab073c424ddca0ddc57950f237948118 Mon Sep 17 00:00:00 2001 From: Paul Buetow Date: Wed, 6 Oct 2021 10:55:50 +0300 Subject: move health check to separate client binary --- internal/server/handlers/healthhandler.go | 4 ++-- internal/server/handlers/serverhandler.go | 2 +- 2 files changed, 3 insertions(+), 3 deletions(-) (limited to 'internal/server') diff --git a/internal/server/handlers/healthhandler.go b/internal/server/handlers/healthhandler.go index 3f3b932..347ff66 100644 --- a/internal/server/handlers/healthhandler.go +++ b/internal/server/handlers/healthhandler.go @@ -48,8 +48,8 @@ func (h *HealthHandler) handleHealthCommand(ctx context.Context, argc int, args switch commandName { case "health": - h.send(h.serverMessages, "OK: DTail SSH Server seems fine") - case "ack", ".ack": + h.send(h.serverMessages, "OK") + case ".ack": h.handleAckCommand(argc, args) default: h.send(h.serverMessages, dlog.Server.Error(h.user, "Received unknown health command", commandName, argc, args)) diff --git a/internal/server/handlers/serverhandler.go b/internal/server/handlers/serverhandler.go index aaffe14..aed8956 100644 --- a/internal/server/handlers/serverhandler.go +++ b/internal/server/handlers/serverhandler.go @@ -106,7 +106,7 @@ func (h *ServerHandler) handleUserCommand(ctx context.Context, argc int, args [] commandFinished() }() - case "ack", ".ack": + case ".ack": h.handleAckCommand(argc, args) commandFinished() -- cgit v1.2.3 From 2d7ddbeae8286373ac19787dc7dde598a7cb0598 Mon Sep 17 00:00:00 2001 From: Paul Buetow Date: Fri, 8 Oct 2021 11:43:43 +0300 Subject: refactor --- internal/server/continuous.go | 4 ++-- internal/server/handlers/basehandler.go | 2 +- internal/server/scheduler.go | 4 ++-- 3 files changed, 5 insertions(+), 5 deletions(-) (limited to 'internal/server') diff --git a/internal/server/continuous.go b/internal/server/continuous.go index 87c8889..5f84afc 100644 --- a/internal/server/continuous.go +++ b/internal/server/continuous.go @@ -71,8 +71,8 @@ func (c *continuous) runJob(ctx context.Context, job config.Continuous) { args.SSHAuthMethods = append(args.SSHAuthMethods, gossh.Password(job.Name)) - query := fmt.Sprintf("%s outfile %s", job.Query, outfile) - client, err := clients.NewMaprClient(args, query, clients.NonCumulativeMode) + args.QueryStr = fmt.Sprintf("%s outfile %s", job.Query, outfile) + client, err := clients.NewMaprClient(args, clients.NonCumulativeMode) if err != nil { dlog.Server.Error(fmt.Sprintf("Unable to create job %s", job.Name), err) return diff --git a/internal/server/handlers/basehandler.go b/internal/server/handlers/basehandler.go index 4fa8f00..f73f82e 100644 --- a/internal/server/handlers/basehandler.go +++ b/internal/server/handlers/basehandler.go @@ -252,7 +252,7 @@ func (h *baseHandler) flush() { for i := 0; i < 10; i++ { if numUnsentMessages() == 0 { - dlog.Server.Debug(h.user, "All lines sent", fmt.Sprintf("%p", h)) + dlog.Server.Debug(h.user, "ALL lines sent", fmt.Sprintf("%p", h)) return } dlog.Server.Debug(h.user, "Still lines to be sent") diff --git a/internal/server/scheduler.go b/internal/server/scheduler.go index 64e6573..ccb2225 100644 --- a/internal/server/scheduler.go +++ b/internal/server/scheduler.go @@ -82,8 +82,8 @@ func (s *scheduler) runJobs(ctx context.Context) { args.SSHAuthMethods = append(args.SSHAuthMethods, gossh.Password(job.Name)) - query := fmt.Sprintf("%s outfile %s", job.Query, outfile) - client, err := clients.NewMaprClient(args, query, clients.CumulativeMode) + args.QueryStr = fmt.Sprintf("%s outfile %s", job.Query, outfile) + client, err := clients.NewMaprClient(args, clients.CumulativeMode) if err != nil { dlog.Server.Error(fmt.Sprintf("Unable to create job %s", job.Name), err) continue -- cgit v1.2.3 From 97747ea0f3178f7f5890512d483fdccaa82846b0 Mon Sep 17 00:00:00 2001 From: Paul Buetow Date: Sat, 9 Oct 2021 21:10:29 +0300 Subject: vetting and linting and some code restyling --- internal/server/continuous.go | 9 +----- internal/server/handlers/basehandler.go | 30 ++++++++------------ internal/server/handlers/healthhandler.go | 11 ++++---- internal/server/handlers/mapcommand.go | 7 ++--- internal/server/handlers/readcommand.go | 41 ++++++++++++++++----------- internal/server/handlers/serverhandler.go | 20 ++++++------- internal/server/scheduler.go | 9 +----- internal/server/server.go | 47 ++++++++++++++----------------- internal/server/stats.go | 11 +++----- 9 files changed, 79 insertions(+), 106 deletions(-) (limited to 'internal/server') diff --git a/internal/server/continuous.go b/internal/server/continuous.go index 5f84afc..93b3fcb 100644 --- a/internal/server/continuous.go +++ b/internal/server/continuous.go @@ -13,8 +13,7 @@ import ( gossh "golang.org/x/crypto/ssh" ) -type continuous struct { -} +type continuous struct{} func newContinuous() *continuous { return &continuous{} @@ -23,7 +22,6 @@ func newContinuous() *continuous { func (c *continuous) start(ctx context.Context) { dlog.Server.Info("Starting continuous job runner after 10s") time.Sleep(time.Second * 10) - c.runJobs(ctx) } @@ -33,7 +31,6 @@ func (c *continuous) runJobs(ctx context.Context) { dlog.Server.Debug(job.Name, "Not running job as not enabled") continue } - go func(job config.Continuous) { c.runJob(ctx, job) for { @@ -54,7 +51,6 @@ func (c *continuous) runJob(ctx context.Context, job config.Continuous) { files := fillDates(job.Files) outfile := fillDates(job.Outfile) - servers := strings.Join(job.Servers, ",") if servers == "" { servers = config.Server.SSHBindAddress @@ -70,7 +66,6 @@ func (c *continuous) runJob(ctx context.Context, job config.Continuous) { } args.SSHAuthMethods = append(args.SSHAuthMethods, gossh.Password(job.Name)) - args.QueryStr = fmt.Sprintf("%s outfile %s", job.Query, outfile) client, err := clients.NewMaprClient(args, clients.NonCumulativeMode) if err != nil { @@ -80,7 +75,6 @@ func (c *continuous) runJob(ctx context.Context, job config.Continuous) { jobCtx, cancel := context.WithCancel(ctx) defer cancel() - if job.RestartOnDayChange { go func() { if c.waitForDayChange(ctx) { @@ -93,7 +87,6 @@ func (c *continuous) runJob(ctx context.Context, job config.Continuous) { dlog.Server.Info(fmt.Sprintf("Starting job %s", job.Name)) status := client.Start(jobCtx, make(chan string)) logMessage := fmt.Sprintf("Job exited with status %d", status) - if status != 0 { dlog.Server.Warn(logMessage) return diff --git a/internal/server/handlers/basehandler.go b/internal/server/handlers/basehandler.go index f73f82e..847e8f9 100644 --- a/internal/server/handlers/basehandler.go +++ b/internal/server/handlers/basehandler.go @@ -37,7 +37,7 @@ type baseHandler struct { activeCommands int32 quiet bool spartan bool - serverless bool + serverless int32 readBuf bytes.Buffer writeBuf bytes.Buffer } @@ -59,16 +59,14 @@ func (h *baseHandler) Read(p []byte) (n int, err error) { select { case message := <-h.serverMessages: if message[0] == '.' { - // Handle hidden message (don't display to the user, interpreted by dtail client) + // Handle hidden message (don't display to the user) h.readBuf.WriteString(message) h.readBuf.WriteByte(protocol.MessageDelimiter) n = copy(p, h.readBuf.Bytes()) return } - if h.serverless { - // In serverless mode we have logged the server message already via the - // dlog logger, no need to send the message again to the client part. + if h.serverless > 0 { return } @@ -132,7 +130,6 @@ func (h *baseHandler) Write(p []byte) (n int, err error) { h.writeBuf.WriteByte(b) } } - n = len(p) return } @@ -145,13 +142,11 @@ func (h *baseHandler) handleCommand(commandStr string) { h.send(h.serverMessages, dlog.Server.Error(h.user, err)+add) return } - args, argc, err = h.handleBase64(args, argc) if err != nil { h.send(h.serverMessages, dlog.Server.Error(h.user, err)) return } - ctx, cancel := context.WithCancel(context.Background()) go func() { <-h.done.Done() @@ -160,7 +155,6 @@ func (h *baseHandler) handleCommand(commandStr string) { splitted := strings.Split(args[0], ":") commandName := splitted[0] - options, err := config.DeserializeOptions(splitted[1:]) if err != nil { h.send(h.serverMessages, dlog.Server.Error(h.user, err)) @@ -191,8 +185,8 @@ func (h *baseHandler) handleProtocolVersion(args []string) ([]string, int, strin if clientCompat > serverCompat { toUpdate = "server" } - - err := fmt.Errorf("DTail server protocol version '%s' does not match client protocol version '%s', please update DTail %s!", + err := fmt.Errorf("the DTail server protocol version '%s' does not match "+ + "client protocol version '%s', please update DTail %s", protocol.ProtocolCompat, args[1], toUpdate) return args, argc, add, err } @@ -201,8 +195,8 @@ func (h *baseHandler) handleProtocolVersion(args []string) ([]string, int, strin } func (h *baseHandler) handleBase64(args []string, argc int) ([]string, int, error) { - err := errors.New("Unable to decode client message, DTail server and client versions may not be compatible") - + err := errors.New("unable to decode client message, DTail server and client " + + "versions may not be compatible") if argc != 2 || args[0] != "base64" { return args, argc, err } @@ -215,7 +209,8 @@ func (h *baseHandler) handleBase64(args []string, argc int) ([]string, int, erro args = strings.Split(decodedStr, " ") argc = len(decodedStr) - dlog.Server.Trace(h.user, "Base64 decoded received command", decodedStr, argc, args) + dlog.Server.Trace(h.user, "Base64 decoded received command", + decodedStr, argc, args) return args, argc, nil } @@ -223,7 +218,8 @@ func (h *baseHandler) handleBase64(args []string, argc int) ([]string, int, erro func (h *baseHandler) handleAckCommand(argc int, args []string) { if argc < 3 { if !h.quiet { - h.send(h.serverMessages, dlog.Server.Warn(h.user, "Unable to parse command", args, argc)) + h.send(h.serverMessages, dlog.Server.Warn(h.user, + "Unable to parse command", args, argc)) } return } @@ -245,11 +241,9 @@ func (h *baseHandler) send(ch chan<- string, message string) { func (h *baseHandler) flush() { dlog.Server.Trace(h.user, "flush()") - numUnsentMessages := func() int { return len(h.lines) + len(h.serverMessages) + len(h.maprMessages) } - for i := 0; i < 10; i++ { if numUnsentMessages() == 0 { dlog.Server.Debug(h.user, "ALL lines sent", fmt.Sprintf("%p", h)) @@ -258,7 +252,6 @@ func (h *baseHandler) flush() { dlog.Server.Debug(h.user, "Still lines to be sent") time.Sleep(time.Millisecond * 10) } - dlog.Server.Warn(h.user, "Some lines remain unsent", numUnsentMessages()) } @@ -279,7 +272,6 @@ func (h *baseHandler) shutdown() { dlog.Server.Debug(h.user, "Shutdown timeout reached, enforcing shutdown") case <-h.done.Done(): } - h.done.Shutdown() } diff --git a/internal/server/handlers/healthhandler.go b/internal/server/handlers/healthhandler.go index 347ff66..8d6c400 100644 --- a/internal/server/handlers/healthhandler.go +++ b/internal/server/handlers/healthhandler.go @@ -35,24 +35,23 @@ func NewHealthHandler(user *user.User) *HealthHandler { if err != nil { dlog.Server.FatalPanic(err) } - s := strings.Split(fqdn, ".") h.hostname = s[0] - return &h } -func (h *HealthHandler) handleHealthCommand(ctx context.Context, argc int, args []string, - commandName string, options map[string]string) { - dlog.Server.Debug(h.user, "Handling health command", argc, args) +func (h *HealthHandler) handleHealthCommand(ctx context.Context, argc int, + args []string, commandName string, options map[string]string) { + dlog.Server.Debug(h.user, "Handling health command", argc, args) switch commandName { case "health": h.send(h.serverMessages, "OK") case ".ack": h.handleAckCommand(argc, args) default: - h.send(h.serverMessages, dlog.Server.Error(h.user, "Received unknown health command", commandName, argc, args)) + h.send(h.serverMessages, dlog.Server.Error(h.user, + "Received unknown health command", commandName, argc, args)) } h.shutdown() } diff --git a/internal/server/handlers/mapcommand.go b/internal/server/handlers/mapcommand.go index c3e600e..65e0ed8 100644 --- a/internal/server/handlers/mapcommand.go +++ b/internal/server/handlers/mapcommand.go @@ -14,18 +14,17 @@ type mapCommand struct { } // NewMapCommand returns a new server side mapreduce command. -func newMapCommand(serverHandler *ServerHandler, argc int, args []string) (mapCommand, *server.Aggregate, error) { - m := mapCommand{server: serverHandler} +func newMapCommand(serverHandler *ServerHandler, argc int, + args []string) (mapCommand, *server.Aggregate, error) { + m := mapCommand{server: serverHandler} queryStr := strings.Join(args[1:], " ") aggregate, err := server.NewAggregate(queryStr) if err != nil { return m, nil, err } - m.aggregate = aggregate return m, aggregate, nil - } func (m mapCommand) Start(ctx context.Context, aggregatedMessages chan<- string) { diff --git a/internal/server/handlers/readcommand.go b/internal/server/handlers/readcommand.go index abc44c7..384e966 100644 --- a/internal/server/handlers/readcommand.go +++ b/internal/server/handlers/readcommand.go @@ -26,25 +26,30 @@ func newReadCommand(server *ServerHandler, mode omode.Mode) *readCommand { } } -func (r *readCommand) Start(ctx context.Context, argc int, args []string, retries int) { - re := regex.NewNoop() +func (r *readCommand) Start(ctx context.Context, argc int, args []string, + retries int) { + re := regex.NewNoop() if argc >= 4 { deserializedRegex, err := regex.Deserialize(strings.Join(args[2:], " ")) if err != nil { - r.server.send(r.server.serverMessages, dlog.Server.Error(r.server.user, "Unable to parse command", err)) + r.server.send(r.server.serverMessages, dlog.Server.Error(r.server.user, + "Unable to parse command", err)) return } re = deserializedRegex } if argc < 3 { - r.server.send(r.server.serverMessages, dlog.Server.Warn(r.server.user, "Unable to parse command", args, argc)) + r.server.send(r.server.serverMessages, dlog.Server.Warn(r.server.user, + "Unable to parse command", args, argc)) return } r.readGlob(ctx, args[1], re, retries) } -func (r *readCommand) readGlob(ctx context.Context, glob string, re regex.Regex, retries int) { +func (r *readCommand) readGlob(ctx context.Context, glob string, re regex.Regex, + retries int) { + retryInterval := time.Second * 5 glob = filepath.Clean(glob) @@ -58,7 +63,8 @@ func (r *readCommand) readGlob(ctx context.Context, glob string, re regex.Regex, if numPaths := len(paths); numPaths == 0 { dlog.Server.Error(r.server.user, "No such file(s) to read", glob) - r.server.send(r.server.serverMessages, dlog.Server.Warn(r.server.user, "Unable to read file(s), check server logs")) + r.server.send(r.server.serverMessages, dlog.Server.Warn(r.server.user, + "Unable to read file(s), check server logs")) select { case <-ctx.Done(): return @@ -72,31 +78,33 @@ func (r *readCommand) readGlob(ctx context.Context, glob string, re regex.Regex, return } - r.server.send(r.server.serverMessages, dlog.Server.Warn(r.server.user, "Giving up to read file(s)")) + r.server.send(r.server.serverMessages, dlog.Server.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) { +func (r *readCommand) readFiles(ctx context.Context, paths []string, glob string, + re regex.Regex, retryInterval time.Duration) { + var wg sync.WaitGroup wg.Add(len(paths)) - for _, path := range paths { go r.readFileIfPermissions(ctx, &wg, path, glob, re) } - wg.Wait() } -func (r *readCommand) readFileIfPermissions(ctx context.Context, wg *sync.WaitGroup, path, glob string, re regex.Regex) { +func (r *readCommand) readFileIfPermissions(ctx context.Context, + wg *sync.WaitGroup, path, glob string, re regex.Regex) { + defer wg.Done() globID := r.makeGlobID(path, glob) - if !r.server.user.HasFilePermission(path, "readfiles") { dlog.Server.Error(r.server.user, "No permission to read file", path, globID) - r.server.send(r.server.serverMessages, dlog.Server.Warn(r.server.user, "Unable to read file(s), check server logs")) + r.server.send(r.server.serverMessages, dlog.Server.Warn(r.server.user, + "Unable to read file(s), check server logs")) return } - r.readFile(ctx, path, globID, re) } @@ -137,7 +145,6 @@ func (r *readCommand) readFile(ctx context.Context, path, globID string, re rege return } } - time.Sleep(time.Second * 2) dlog.Server.Info(path, globID, "Reading file again") } @@ -156,11 +163,11 @@ func (r *readCommand) makeGlobID(path, glob string) string { if len(idParts) > 0 { return strings.Join(idParts, "/") } - if len(pathParts) > 0 { return pathParts[len(pathParts)-1] } - r.server.send(r.server.serverMessages, dlog.Server.Warn("Empty file path given?", path, glob)) + r.server.send(r.server.serverMessages, + dlog.Server.Warn("Empty file path given?", path, glob)) return "" } diff --git a/internal/server/handlers/serverhandler.go b/internal/server/handlers/serverhandler.go index aed8956..f12d590 100644 --- a/internal/server/handlers/serverhandler.go +++ b/internal/server/handlers/serverhandler.go @@ -4,6 +4,7 @@ import ( "context" "os" "strings" + "sync/atomic" "github.com/mimecast/dtail/internal" "github.com/mimecast/dtail/internal/io/dlog" @@ -23,7 +24,9 @@ type ServerHandler struct { } // NewServerHandler returns the server handler. -func NewServerHandler(user *user.User, catLimiter, tailLimiter chan struct{}) *ServerHandler { +func NewServerHandler(user *user.User, catLimiter, + tailLimiter chan struct{}) *ServerHandler { + dlog.Server.Debug(user, "Creating new server handler") h := ServerHandler{ baseHandler: baseHandler{ @@ -51,11 +54,10 @@ func NewServerHandler(user *user.User, catLimiter, tailLimiter chan struct{}) *S return &h } -func (h *ServerHandler) handleUserCommand(ctx context.Context, argc int, args []string, - commandName string, options map[string]string) { +func (h *ServerHandler) handleUserCommand(ctx context.Context, argc int, + args []string, commandName string, options map[string]string) { dlog.Server.Debug(h.user, "Handling user command", argc, args) - h.incrementActiveCommands() commandFinished := func() { if h.decrementActiveCommands() == 0 { @@ -73,7 +75,7 @@ func (h *ServerHandler) handleUserCommand(ctx context.Context, argc int, args [] } if serverless, _ := options["serverless"]; serverless == "true" { dlog.Server.Debug(h.user, "Enabling serverless mode") - h.serverless = true + atomic.AddInt32(&h.serverless, 1) } switch commandName { @@ -83,14 +85,12 @@ func (h *ServerHandler) handleUserCommand(ctx context.Context, argc int, args [] command.Start(ctx, argc, args, 1) commandFinished() }() - case "tail": command := newReadCommand(h, omode.TailClient) go func() { command.Start(ctx, argc, args, 10) commandFinished() }() - case "map": command, aggregate, err := newMapCommand(h, argc, args) if err != nil { @@ -99,19 +99,17 @@ func (h *ServerHandler) handleUserCommand(ctx context.Context, argc int, args [] commandFinished() return } - h.aggregate = aggregate go func() { command.Start(ctx, h.maprMessages) commandFinished() }() - case ".ack": h.handleAckCommand(argc, args) commandFinished() - default: - h.send(h.serverMessages, dlog.Server.Error(h.user, "Received unknown user command", commandName, argc, args, options)) + h.send(h.serverMessages, dlog.Server.Error(h.user, + "Received unknown user command", commandName, argc, args, options)) commandFinished() } } diff --git a/internal/server/scheduler.go b/internal/server/scheduler.go index ccb2225..0ba65f7 100644 --- a/internal/server/scheduler.go +++ b/internal/server/scheduler.go @@ -16,8 +16,7 @@ import ( gossh "golang.org/x/crypto/ssh" ) -type scheduler struct { -} +type scheduler struct{} func newScheduler() *scheduler { return &scheduler{} @@ -28,7 +27,6 @@ func (s *scheduler) start(ctx context.Context) { // First run after just 10s! time.Sleep(time.Second * 10) s.runJobs(ctx) - for { select { case <-time.After(time.Minute): @@ -45,13 +43,11 @@ func (s *scheduler) runJobs(ctx context.Context) { dlog.Server.Debug(job.Name, "Not running job as not enabled") continue } - hour, err := strconv.Atoi(time.Now().Format("15")) if err != nil { dlog.Server.Error(job.Name, "Unable to create job", err) continue } - if hour < job.TimeRange[0] || hour >= job.TimeRange[1] { dlog.Server.Debug(job.Name, "Not running job out of time range") continue @@ -59,7 +55,6 @@ func (s *scheduler) runJobs(ctx context.Context) { files := fillDates(job.Files) outfile := fillDates(job.Outfile) - _, err = os.Stat(outfile) if !os.IsNotExist(err) { dlog.Server.Debug(job.Name, "Not running job as outfile already exists", outfile) @@ -70,7 +65,6 @@ func (s *scheduler) runJobs(ctx context.Context) { if servers == "" { servers = config.Server.SSHBindAddress } - args := config.Args{ ConnectionsPerCPU: config.DefaultConnectionsPerCPU, Discovery: job.Discovery, @@ -81,7 +75,6 @@ func (s *scheduler) runJobs(ctx context.Context) { } args.SSHAuthMethods = append(args.SSHAuthMethods, gossh.Password(job.Name)) - args.QueryStr = fmt.Sprintf("%s outfile %s", job.Query, outfile) client, err := clients.NewMaprClient(args, clients.CumulativeMode) if err != nil { diff --git a/internal/server/server.go b/internal/server/server.go index b3d4bff..0cb5e27 100644 --- a/internal/server/server.go +++ b/internal/server/server.go @@ -24,9 +24,9 @@ type Server struct { stats stats // SSH server configuration. sshServerConfig *gossh.ServerConfig - // To control the max amount of concurrent cats (which can cause a lot of I/O on the server) + // To control the max amount of concurrent cats. catLimiter chan struct{} - // To control the max amount of concurrent tails + // To control the max amount of concurrent tails. tailLimiter chan struct{} // To run scheduled tasks (if configured) sched *scheduler @@ -61,7 +61,6 @@ func New() *Server { // Start the server. func (s *Server) Start(ctx context.Context) int { dlog.Server.Info("Starting server") - bindAt := fmt.Sprintf("%s:%d", config.Server.SSHBindAddress, config.Common.SSHPort) dlog.Server.Info("Binding server", bindAt) @@ -76,14 +75,12 @@ func (s *Server) Start(ctx context.Context) int { go s.listenerLoop(ctx, listener) <-ctx.Done() - // For future use. return 0 } func (s *Server) listenerLoop(ctx context.Context, listener net.Listener) { dlog.Server.Debug("Starting listener loop") - for { conn, err := listener.Accept() // Blocking if err != nil { @@ -101,7 +98,6 @@ func (s *Server) listenerLoop(ctx context.Context, listener net.Listener) { conn.Close() continue } - go s.handleConnection(ctx, conn) } } @@ -116,22 +112,23 @@ func (s *Server) handleConnection(ctx context.Context, conn net.Conn) { } s.stats.incrementConnections() - go gossh.DiscardRequests(reqs) for newChannel := range chans { go s.handleChannel(ctx, sshConn, newChannel) } } -func (s *Server) handleChannel(ctx context.Context, sshConn gossh.Conn, newChannel gossh.NewChannel) { +func (s *Server) handleChannel(ctx context.Context, sshConn gossh.Conn, + newChannel gossh.NewChannel) { + user, err := user.New(sshConn.User(), sshConn.RemoteAddr().String()) if err != nil { dlog.Server.Error(user, err) newChannel.Reject(gossh.Prohibited, err.Error()) return } - dlog.Server.Info(user, "Invoking channel handler") + dlog.Server.Info(user, "Invoking channel handler") if newChannel.ChannelType() != "session" { err := errors.New("Don'w allow other channel types than session") dlog.Server.Error(user, err) @@ -151,9 +148,10 @@ func (s *Server) handleChannel(ctx context.Context, sshConn gossh.Conn, newChann } } -func (s *Server) handleRequests(ctx context.Context, sshConn gossh.Conn, in <-chan *gossh.Request, channel gossh.Channel, user *user.User) error { - dlog.Server.Info(user, "Invoking request handler") +func (s *Server) handleRequests(ctx context.Context, sshConn gossh.Conn, + in <-chan *gossh.Request, channel gossh.Channel, user *user.User) error { + dlog.Server.Info(user, "Invoking request handler") for req := range in { var payload = struct{ Value string }{} gossh.Unmarshal(req.Payload, &payload) @@ -167,7 +165,6 @@ func (s *Server) handleRequests(ctx context.Context, sshConn gossh.Conn, in <-ch default: handler = handlers.NewServerHandler(user, s.catLimiter, s.tailLimiter) } - terminate := func() { handler.Shutdown() sshConn.Close() @@ -178,13 +175,11 @@ func (s *Server) handleRequests(ctx context.Context, sshConn gossh.Conn, in <-ch io.Copy(channel, handler) terminate() }() - go func() { // Broken pipe, cancel io.Copy(handler, channel) terminate() }() - go func() { select { case <-ctx.Done(): @@ -192,7 +187,6 @@ func (s *Server) handleRequests(ctx context.Context, sshConn gossh.Conn, in <-ch } terminate() }() - go func() { if err := sshConn.Wait(); err != nil && err != io.EOF { dlog.Server.Error(user, err) @@ -204,20 +198,19 @@ func (s *Server) handleRequests(ctx context.Context, sshConn gossh.Conn, in <-ch // Only serving shell type req.Reply(true, nil) - default: req.Reply(false, nil) - return fmt.Errorf("Closing SSH connection as unknown request recieved|%s|%v", req.Type, payload.Value) } } - return nil } // Callback for SSH authentication. -func (s *Server) Callback(c gossh.ConnMetadata, authPayload []byte) (*gossh.Permissions, error) { +func (s *Server) Callback(c gossh.ConnMetadata, + authPayload []byte) (*gossh.Permissions, error) { + user, err := user.New(c.User(), c.RemoteAddr().String()) if err != nil { return nil, err @@ -229,7 +222,6 @@ func (s *Server) Callback(c gossh.ConnMetadata, authPayload []byte) (*gossh.Perm } authInfo := string(authPayload) - splitted := strings.Split(c.RemoteAddr().String(), ":") remoteIP := splitted[0] @@ -259,23 +251,26 @@ func (s *Server) Callback(c gossh.ConnMetadata, authPayload []byte) (*gossh.Perm return nil, fmt.Errorf("user %s not authorized", user) } -func (s *Server) backgroundCanSSH(user *user.User, jobName, remoteIP, allowedJobName string, allowFrom []string) bool { - dlog.Server.Debug("backgroundCanSSH", user, jobName, remoteIP, allowedJobName, allowFrom) +func (s *Server) backgroundCanSSH(user *user.User, jobName, remoteIP, + allowedJobName string, allowFrom []string) bool { + dlog.Server.Debug("backgroundCanSSH", user, jobName, remoteIP, allowedJobName, allowFrom) if jobName != allowedJobName { - dlog.Server.Debug(user, jobName, "backgroundCanSSH", "Job name does not match, skipping to next one...", allowedJobName) + dlog.Server.Debug(user, jobName, "backgroundCanSSH", + "Job name does not match, skipping to next one...", allowedJobName) return false } for _, myAddr := range allowFrom { ips, err := net.LookupIP(myAddr) if err != nil { - dlog.Server.Debug(user, jobName, "backgroundCanSSH", "Unable to lookup IP address for allowed hosts lookup, skipping to next one...", myAddr, err) + dlog.Server.Debug(user, jobName, "backgroundCanSSH", "Unable to lookup IP "+ + "address for allowed hosts lookup, skipping to next one...", myAddr, err) continue } - for _, ip := range ips { - dlog.Server.Debug(user, jobName, "backgroundCanSSH", "Comparing IP addresses", remoteIP, ip.String()) + dlog.Server.Debug(user, jobName, "backgroundCanSSH", "Comparing IP addresses", + remoteIP, ip.String()) if remoteIP == ip.String() { return true } diff --git a/internal/server/stats.go b/internal/server/stats.go index c07634d..99a644a 100644 --- a/internal/server/stats.go +++ b/internal/server/stats.go @@ -19,7 +19,6 @@ type stats struct { func (s *stats) incrementConnections() { defer s.logServerStats() - s.mutex.Lock() s.currentConnections++ s.lifetimeConnections++ @@ -28,7 +27,6 @@ func (s *stats) incrementConnections() { func (s *stats) decrementConnections() { defer s.logServerStats() - s.mutex.Lock() s.currentConnections-- s.mutex.Unlock() @@ -40,8 +38,8 @@ func (s *stats) hasConnections() bool { s.mutex.Unlock() has := currentConnections > 0 - dlog.Server.Info("stats", "Server with open connections?", has, currentConnections) - + dlog.Server.Info("stats", "Server with open connections?", + has, currentConnections) return has } @@ -52,7 +50,6 @@ func (s *stats) logServerStats() { data := make(map[string]interface{}) data["currentConnections"] = s.currentConnections data["lifetimeConnections"] = s.lifetimeConnections - dlog.Server.Mapreduce("STATS", data) } @@ -61,9 +58,9 @@ func (s *stats) serverLimitExceeded() error { defer s.mutex.Unlock() if s.currentConnections >= config.Server.MaxConnections { - return fmt.Errorf("Exceeded max allowed concurrent connections of %d", config.Server.MaxConnections) + return fmt.Errorf("Exceeded max allowed concurrent connections of %d", + config.Server.MaxConnections) } - return nil } -- cgit v1.2.3 From f44792c9102488774c9993b080f35c65287a64b1 Mon Sep 17 00:00:00 2001 From: Paul Buetow Date: Sun, 10 Oct 2021 14:02:12 +0300 Subject: add another dmap test - reading 100 source files at once fix a data race when reading multiple files on one server from the same session at once --- internal/server/handlers/basehandler.go | 40 ++++++++++++++++++++++++++----- internal/server/handlers/healthhandler.go | 2 +- internal/server/handlers/serverhandler.go | 18 ++------------ 3 files changed, 37 insertions(+), 23 deletions(-) (limited to 'internal/server') diff --git a/internal/server/handlers/basehandler.go b/internal/server/handlers/basehandler.go index 847e8f9..d814cc9 100644 --- a/internal/server/handlers/basehandler.go +++ b/internal/server/handlers/basehandler.go @@ -9,6 +9,7 @@ import ( "io" "strconv" "strings" + "sync" "sync/atomic" "time" @@ -22,7 +23,7 @@ import ( user "github.com/mimecast/dtail/internal/user/server" ) -type handleCommandCb func(context.Context, int, []string, string, map[string]string) +type handleCommandCb func(context.Context, int, []string, string) type baseHandler struct { done *internal.Done @@ -35,11 +36,15 @@ type baseHandler struct { user *user.User ackCloseReceived chan struct{} activeCommands int32 - quiet bool - spartan bool - serverless int32 readBuf bytes.Buffer writeBuf bytes.Buffer + + // Some global options + sync primitives required. + once sync.Once + mutex sync.Mutex + quiet bool + spartan bool + serverless bool } // Shutdown the handler. @@ -66,7 +71,7 @@ func (h *baseHandler) Read(p []byte) (n int, err error) { return } - if h.serverless > 0 { + if h.serverless { return } @@ -160,8 +165,9 @@ func (h *baseHandler) handleCommand(commandStr string) { h.send(h.serverMessages, dlog.Server.Error(h.user, err)) return } + h.setOptions(options) - h.handleCommandCb(ctx, argc, args, commandName, options) + h.handleCommandCb(ctx, argc, args, commandName) } func (h *baseHandler) handleProtocolVersion(args []string) ([]string, int, string, error) { @@ -232,6 +238,28 @@ func (h *baseHandler) handleAckCommand(argc int, args []string) { } } +func (h *baseHandler) setOptions(options map[string]string) { + // We have to make sure that this block is executed only once. + h.mutex.Lock() + defer h.mutex.Unlock() + // We can read the options only once, will cause a data race otherwise if + // changed multiple times for multiple incoming commands. + h.once.Do(func() { + if quiet, _ := options["quiet"]; quiet == "true" { + dlog.Server.Debug(h.user, "Enabling quiet mode") + h.quiet = true + } + if spartan, _ := options["spartan"]; spartan == "true" { + dlog.Server.Debug(h.user, "Enabling spartan mode") + h.spartan = true + } + if serverless, _ := options["serverless"]; serverless == "true" { + dlog.Server.Debug(h.user, "Enabling serverless mode") + h.serverless = true + } + }) +} + func (h *baseHandler) send(ch chan<- string, message string) { select { case ch <- message: diff --git a/internal/server/handlers/healthhandler.go b/internal/server/handlers/healthhandler.go index 8d6c400..0425696 100644 --- a/internal/server/handlers/healthhandler.go +++ b/internal/server/handlers/healthhandler.go @@ -41,7 +41,7 @@ func NewHealthHandler(user *user.User) *HealthHandler { } func (h *HealthHandler) handleHealthCommand(ctx context.Context, argc int, - args []string, commandName string, options map[string]string) { + args []string, commandName string) { dlog.Server.Debug(h.user, "Handling health command", argc, args) switch commandName { diff --git a/internal/server/handlers/serverhandler.go b/internal/server/handlers/serverhandler.go index f12d590..52c4570 100644 --- a/internal/server/handlers/serverhandler.go +++ b/internal/server/handlers/serverhandler.go @@ -4,7 +4,6 @@ import ( "context" "os" "strings" - "sync/atomic" "github.com/mimecast/dtail/internal" "github.com/mimecast/dtail/internal/io/dlog" @@ -55,7 +54,7 @@ func NewServerHandler(user *user.User, catLimiter, } func (h *ServerHandler) handleUserCommand(ctx context.Context, argc int, - args []string, commandName string, options map[string]string) { + args []string, commandName string) { dlog.Server.Debug(h.user, "Handling user command", argc, args) h.incrementActiveCommands() @@ -65,19 +64,6 @@ func (h *ServerHandler) handleUserCommand(ctx context.Context, argc int, } } - if quiet, _ := options["quiet"]; quiet == "true" { - dlog.Server.Debug(h.user, "Enabling quiet mode") - h.quiet = true - } - if spartan, _ := options["spartan"]; spartan == "true" { - dlog.Server.Debug(h.user, "Enabling spartan mode") - h.spartan = true - } - if serverless, _ := options["serverless"]; serverless == "true" { - dlog.Server.Debug(h.user, "Enabling serverless mode") - atomic.AddInt32(&h.serverless, 1) - } - switch commandName { case "grep", "cat": command := newReadCommand(h, omode.CatClient) @@ -109,7 +95,7 @@ func (h *ServerHandler) handleUserCommand(ctx context.Context, argc int, commandFinished() default: h.send(h.serverMessages, dlog.Server.Error(h.user, - "Received unknown user command", commandName, argc, args, options)) + "Received unknown user command", commandName, argc, args)) commandFinished() } } -- cgit v1.2.3 From 7b873100d94ddc3c698a620cb83b61dcb2074303 Mon Sep 17 00:00:00 2001 From: Paul Buetow Date: Wed, 13 Oct 2021 09:00:03 +0300 Subject: add another dcat integration test - catting 100 files at once --- internal/server/handlers/basehandler.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) (limited to 'internal/server') diff --git a/internal/server/handlers/basehandler.go b/internal/server/handlers/basehandler.go index d814cc9..6bc8268 100644 --- a/internal/server/handlers/basehandler.go +++ b/internal/server/handlers/basehandler.go @@ -63,7 +63,7 @@ func (h *baseHandler) Read(p []byte) (n int, err error) { select { case message := <-h.serverMessages: - if message[0] == '.' { + if len(message) > 0 && message[0] == '.' { // Handle hidden message (don't display to the user) h.readBuf.WriteString(message) h.readBuf.WriteByte(protocol.MessageDelimiter) -- cgit v1.2.3 From 1dead22129a26e4f532e68c2c63fe4122b519506 Mon Sep 17 00:00:00 2001 From: Paul Buetow Date: Wed, 13 Oct 2021 21:10:28 +0300 Subject: Merging grep context from master --- internal/server/handlers/basehandler.go | 12 ++++++------ internal/server/handlers/healthhandler.go | 5 +++-- internal/server/handlers/readcommand.go | 30 ++++++++++++++++-------------- internal/server/handlers/serverhandler.go | 9 +++++---- 4 files changed, 30 insertions(+), 26 deletions(-) (limited to 'internal/server') diff --git a/internal/server/handlers/basehandler.go b/internal/server/handlers/basehandler.go index 6bc8268..c25f85a 100644 --- a/internal/server/handlers/basehandler.go +++ b/internal/server/handlers/basehandler.go @@ -18,12 +18,13 @@ import ( "github.com/mimecast/dtail/internal/io/dlog" "github.com/mimecast/dtail/internal/io/line" "github.com/mimecast/dtail/internal/io/pool" + "github.com/mimecast/dtail/internal/lcontext" "github.com/mimecast/dtail/internal/mapr/server" "github.com/mimecast/dtail/internal/protocol" user "github.com/mimecast/dtail/internal/user/server" ) -type handleCommandCb func(context.Context, int, []string, string) +type handleCommandCb func(context.Context, lcontext.LContext, int, []string, string) type baseHandler struct { done *internal.Done @@ -160,14 +161,13 @@ func (h *baseHandler) handleCommand(commandStr string) { splitted := strings.Split(args[0], ":") commandName := splitted[0] - options, err := config.DeserializeOptions(splitted[1:]) + options, ltx, err := config.DeserializeOptions(splitted[1:]) if err != nil { h.send(h.serverMessages, dlog.Server.Error(h.user, err)) return } - h.setOptions(options) - - h.handleCommandCb(ctx, argc, args, commandName) + h.handleOptions(options) + h.handleCommandCb(ctx, ltx, argc, args, commandName) } func (h *baseHandler) handleProtocolVersion(args []string) ([]string, int, string, error) { @@ -238,7 +238,7 @@ func (h *baseHandler) handleAckCommand(argc int, args []string) { } } -func (h *baseHandler) setOptions(options map[string]string) { +func (h *baseHandler) handleOptions(options map[string]string) { // We have to make sure that this block is executed only once. h.mutex.Lock() defer h.mutex.Unlock() diff --git a/internal/server/handlers/healthhandler.go b/internal/server/handlers/healthhandler.go index 0425696..6dd9872 100644 --- a/internal/server/handlers/healthhandler.go +++ b/internal/server/handlers/healthhandler.go @@ -8,6 +8,7 @@ import ( "github.com/mimecast/dtail/internal" "github.com/mimecast/dtail/internal/io/dlog" "github.com/mimecast/dtail/internal/io/line" + "github.com/mimecast/dtail/internal/lcontext" user "github.com/mimecast/dtail/internal/user/server" ) @@ -40,8 +41,8 @@ func NewHealthHandler(user *user.User) *HealthHandler { return &h } -func (h *HealthHandler) handleHealthCommand(ctx context.Context, argc int, - args []string, commandName string) { +func (h *HealthHandler) handleHealthCommand(ctx context.Context, + ltx lcontext.LContext, argc int, args []string, commandName string) { dlog.Server.Debug(h.user, "Handling health command", argc, args) switch commandName { diff --git a/internal/server/handlers/readcommand.go b/internal/server/handlers/readcommand.go index 384e966..4728a55 100644 --- a/internal/server/handlers/readcommand.go +++ b/internal/server/handlers/readcommand.go @@ -10,6 +10,7 @@ import ( "github.com/mimecast/dtail/internal/io/dlog" "github.com/mimecast/dtail/internal/io/fs" "github.com/mimecast/dtail/internal/io/line" + "github.com/mimecast/dtail/internal/lcontext" "github.com/mimecast/dtail/internal/omode" "github.com/mimecast/dtail/internal/regex" ) @@ -26,8 +27,8 @@ func newReadCommand(server *ServerHandler, mode omode.Mode) *readCommand { } } -func (r *readCommand) Start(ctx context.Context, argc int, args []string, - retries int) { +func (r *readCommand) Start(ctx context.Context, ltx lcontext.LContext, + argc int, args []string, retries int) { re := regex.NewNoop() if argc >= 4 { @@ -44,11 +45,11 @@ func (r *readCommand) Start(ctx context.Context, argc int, args []string, "Unable to parse command", args, argc)) return } - r.readGlob(ctx, args[1], re, retries) + r.readGlob(ctx, ltx, args[1], re, retries) } -func (r *readCommand) readGlob(ctx context.Context, glob string, re regex.Regex, - retries int) { +func (r *readCommand) readGlob(ctx context.Context, ltx lcontext.LContext, + glob string, re regex.Regex, retries int) { retryInterval := time.Second * 5 glob = filepath.Clean(glob) @@ -74,7 +75,7 @@ func (r *readCommand) readGlob(ctx context.Context, glob string, re regex.Regex, continue } - r.readFiles(ctx, paths, glob, re, retryInterval) + r.readFiles(ctx, ltx, paths, glob, re, retryInterval) return } @@ -83,18 +84,18 @@ func (r *readCommand) readGlob(ctx context.Context, glob string, re regex.Regex, return } -func (r *readCommand) readFiles(ctx context.Context, paths []string, glob string, - re regex.Regex, retryInterval time.Duration) { +func (r *readCommand) readFiles(ctx context.Context, ltx lcontext.LContext, + paths []string, glob string, re regex.Regex, retryInterval time.Duration) { var wg sync.WaitGroup wg.Add(len(paths)) for _, path := range paths { - go r.readFileIfPermissions(ctx, &wg, path, glob, re) + go r.readFileIfPermissions(ctx, ltx, &wg, path, glob, re) } wg.Wait() } -func (r *readCommand) readFileIfPermissions(ctx context.Context, +func (r *readCommand) readFileIfPermissions(ctx context.Context, ltx lcontext.LContext, wg *sync.WaitGroup, path, glob string, re regex.Regex) { defer wg.Done() @@ -105,12 +106,13 @@ func (r *readCommand) readFileIfPermissions(ctx context.Context, "Unable to read file(s), check server logs")) return } - r.readFile(ctx, path, globID, re) + r.readFile(ctx, ltx, path, globID, re) } -func (r *readCommand) readFile(ctx context.Context, path, globID string, re regex.Regex) { - dlog.Server.Info(r.server.user, "Start reading file", path, globID) +func (r *readCommand) readFile(ctx context.Context, ltx lcontext.LContext, + path, globID string, re regex.Regex) { + dlog.Server.Info(r.server.user, "Start reading file", path, globID) var reader fs.FileReader switch r.mode { case omode.TailClient: @@ -129,7 +131,7 @@ func (r *readCommand) readFile(ctx context.Context, path, globID string, re rege lines = make(chan line.Line, 100) aggregate.NextLinesCh <- lines } - if err := reader.Start(ctx, lines, re); err != nil { + if err := reader.Start(ctx, ltx, lines, re); err != nil { dlog.Server.Error(r.server.user, path, globID, err) } if aggregate != nil { diff --git a/internal/server/handlers/serverhandler.go b/internal/server/handlers/serverhandler.go index 52c4570..36574a9 100644 --- a/internal/server/handlers/serverhandler.go +++ b/internal/server/handlers/serverhandler.go @@ -8,6 +8,7 @@ import ( "github.com/mimecast/dtail/internal" "github.com/mimecast/dtail/internal/io/dlog" "github.com/mimecast/dtail/internal/io/line" + "github.com/mimecast/dtail/internal/lcontext" "github.com/mimecast/dtail/internal/omode" user "github.com/mimecast/dtail/internal/user/server" ) @@ -53,8 +54,8 @@ func NewServerHandler(user *user.User, catLimiter, return &h } -func (h *ServerHandler) handleUserCommand(ctx context.Context, argc int, - args []string, commandName string) { +func (h *ServerHandler) handleUserCommand(ctx context.Context, ltx lcontext.LContext, + argc int, args []string, commandName string) { dlog.Server.Debug(h.user, "Handling user command", argc, args) h.incrementActiveCommands() @@ -68,13 +69,13 @@ func (h *ServerHandler) handleUserCommand(ctx context.Context, argc int, case "grep", "cat": command := newReadCommand(h, omode.CatClient) go func() { - command.Start(ctx, argc, args, 1) + command.Start(ctx, ltx, argc, args, 1) commandFinished() }() case "tail": command := newReadCommand(h, omode.TailClient) go func() { - command.Start(ctx, argc, args, 10) + command.Start(ctx, ltx, argc, args, 10) commandFinished() }() case "map": -- cgit v1.2.3 From 06ece112c0dd20c0c211c538216fe64ebe4045c9 Mon Sep 17 00:00:00 2001 From: Paul Buetow Date: Thu, 14 Oct 2021 20:10:55 +0300 Subject: add dgrep context integration tests --- internal/server/handlers/basehandler.go | 2 ++ 1 file changed, 2 insertions(+) (limited to 'internal/server') diff --git a/internal/server/handlers/basehandler.go b/internal/server/handlers/basehandler.go index c25f85a..934f2bc 100644 --- a/internal/server/handlers/basehandler.go +++ b/internal/server/handlers/basehandler.go @@ -159,6 +159,8 @@ func (h *baseHandler) handleCommand(commandStr string) { cancel() }() + dlog.Server.Trace(args) + dlog.Server.Trace(args[0]) splitted := strings.Split(args[0], ":") commandName := splitted[0] options, ltx, err := config.DeserializeOptions(splitted[1:]) -- cgit v1.2.3 From 698fb76b98c46c677abe13fdc93afc6c4f38c39e Mon Sep 17 00:00:00 2001 From: Paul Buetow Date: Thu, 14 Oct 2021 20:55:35 +0300 Subject: refactor --- internal/server/handlers/basehandler.go | 15 ++++++++++----- 1 file changed, 10 insertions(+), 5 deletions(-) (limited to 'internal/server') diff --git a/internal/server/handlers/basehandler.go b/internal/server/handlers/basehandler.go index 934f2bc..6d10d17 100644 --- a/internal/server/handlers/basehandler.go +++ b/internal/server/handlers/basehandler.go @@ -159,11 +159,16 @@ func (h *baseHandler) handleCommand(commandStr string) { cancel() }() - dlog.Server.Trace(args) - dlog.Server.Trace(args[0]) - splitted := strings.Split(args[0], ":") - commandName := splitted[0] - options, ltx, err := config.DeserializeOptions(splitted[1:]) + parts := strings.Split(args[0], ":") + commandName := parts[0] + + // Either no options or empty options provided. + if len(parts) == 1 || len(parts[1]) == 0 { + h.handleCommandCb(ctx, lcontext.LContext{}, argc, args, commandName) + return + } + + options, ltx, err := config.DeserializeOptions(parts[1:]) if err != nil { h.send(h.serverMessages, dlog.Server.Error(h.user, err)) return -- cgit v1.2.3