diff options
Diffstat (limited to 'internal/server')
| -rw-r--r-- | internal/server/handlers/networkwriter.go | 176 | ||||
| -rw-r--r-- | internal/server/handlers/readcommand.go | 327 |
2 files changed, 319 insertions, 184 deletions
diff --git a/internal/server/handlers/networkwriter.go b/internal/server/handlers/networkwriter.go index f1e3bee..bb5ad1d 100644 --- a/internal/server/handlers/networkwriter.go +++ b/internal/server/handlers/networkwriter.go @@ -1,11 +1,13 @@ package handlers import ( + "bytes" "fmt" "net" "os" "github.com/mimecast/dtail/internal/io/dlog" + "github.com/mimecast/dtail/internal/io/line" "github.com/mimecast/dtail/internal/user/server" ) @@ -173,4 +175,178 @@ func (bnw *BufferedNetworkWriter) Close() error { return err } return bnw.NetworkOutputWriter.Close() +} + +// ChannelOutputWriter sends output to the server's lines channel instead of direct network +type ChannelOutputWriter struct { + linesCh chan<- *line.Line + serverMessages chan<- string + user *server.User +} + +// NewChannelOutputWriter creates a new channel output writer +func NewChannelOutputWriter(linesCh chan<- *line.Line, serverMessages chan<- string, user *server.User) *ChannelOutputWriter { + return &ChannelOutputWriter{ + linesCh: linesCh, + serverMessages: serverMessages, + user: user, + } +} + +// Write implements io.Writer interface by sending data through the lines channel +func (cow *ChannelOutputWriter) Write(data []byte) (int, error) { + if len(data) == 0 { + return 0, nil + } + + // Create a line object using the proper constructor + contentBuffer := bytes.NewBuffer(data) + lineObj := line.New(contentBuffer, 0, 100, "channelless") + + select { + case cow.linesCh <- lineObj: + return len(data), nil + default: + // Channel is full, report error + cow.sendServerMessage("Lines channel full, dropping data") + return 0, fmt.Errorf("lines channel full") + } +} + +// sendServerMessage sends a message through the existing server message channel +func (cow *ChannelOutputWriter) sendServerMessage(message string) { + if cow.serverMessages == nil { + return + } + + select { + case cow.serverMessages <- message: + // Message sent successfully + default: + // Channel full, log the issue + dlog.Server.Warn(cow.user, "Server message channel full, dropping message:", message) + } +} + +// SendLine sends a formatted line through the lines channel +func (cow *ChannelOutputWriter) SendLine(hostname, filePath string, lineNum int, content []byte) error { + // Create a line object with proper metadata + contentBuffer := bytes.NewBuffer(content) + lineObj := line.New(contentBuffer, uint64(lineNum), 100, filePath) + + select { + case cow.linesCh <- lineObj: + return nil + default: + cow.sendServerMessage(fmt.Sprintf("Lines channel full, dropping line from %s:%d", filePath, lineNum)) + return fmt.Errorf("lines channel full") + } +} + +// SendPlainLine sends a plain line through the lines channel +func (cow *ChannelOutputWriter) SendPlainLine(content []byte) error { + _, err := cow.Write(content) + return err +} + +// SendServerStat sends a server statistics message +func (cow *ChannelOutputWriter) SendServerStat(message string) { + cow.sendServerMessage(message) +} + +// SendError sends an error message +func (cow *ChannelOutputWriter) SendError(err error) { + cow.sendServerMessage(fmt.Sprintf("ERROR: %v", err)) +} + +// Close is a no-op for channel output writer +func (cow *ChannelOutputWriter) Close() error { + return nil +} + +// ServerHandlerWriter writes output directly to the server handler's lines channel +type ServerHandlerWriter struct { + server *ServerHandler + serverMessages chan<- string + user *server.User +} + +// NewServerHandlerWriter creates a new server handler writer +func NewServerHandlerWriter(serverHandler *ServerHandler, serverMessages chan<- string, user *server.User) *ServerHandlerWriter { + return &ServerHandlerWriter{ + server: serverHandler, + serverMessages: serverMessages, + user: user, + } +} + +// Write implements io.Writer interface by sending data through the server's lines channel +func (shw *ServerHandlerWriter) Write(data []byte) (int, error) { + if len(data) == 0 { + return 0, nil + } + + // Create a line object and send it through the server's lines channel + contentBuffer := bytes.NewBuffer(data) + lineObj := line.New(contentBuffer, 0, 100, "channelless") + + select { + case shw.server.lines <- lineObj: + return len(data), nil + default: + // Channel is full, report error + shw.sendServerMessage("Server lines channel full, dropping data") + return 0, fmt.Errorf("server lines channel full") + } +} + +// sendServerMessage sends a message through the existing server message channel +func (shw *ServerHandlerWriter) sendServerMessage(message string) { + if shw.serverMessages == nil { + return + } + + select { + case shw.serverMessages <- message: + // Message sent successfully + default: + // Channel full, log the issue + dlog.Server.Warn(shw.user, "Server message channel full, dropping message:", message) + } +} + +// SendLine sends a formatted line through the server's lines channel +func (shw *ServerHandlerWriter) SendLine(hostname, filePath string, lineNum int, content []byte) error { + // Create a line object with proper metadata + contentBuffer := bytes.NewBuffer(content) + lineObj := line.New(contentBuffer, uint64(lineNum), 100, filePath) + + select { + case shw.server.lines <- lineObj: + return nil + default: + shw.sendServerMessage(fmt.Sprintf("Server lines channel full, dropping line from %s:%d", filePath, lineNum)) + return fmt.Errorf("server lines channel full") + } +} + +// SendPlainLine sends a plain line through the server's lines channel +func (shw *ServerHandlerWriter) SendPlainLine(content []byte) error { + _, err := shw.Write(content) + return err +} + +// SendServerStat sends a server statistics message +func (shw *ServerHandlerWriter) SendServerStat(message string) { + shw.sendServerMessage(message) +} + +// SendError sends an error message +func (shw *ServerHandlerWriter) SendError(err error) { + shw.sendServerMessage(fmt.Sprintf("ERROR: %v", err)) +} + +// Close is a no-op for server handler writer +func (shw *ServerHandlerWriter) Close() error { + return nil }
\ No newline at end of file diff --git a/internal/server/handlers/readcommand.go b/internal/server/handlers/readcommand.go index 89bb757..59e9d1b 100644 --- a/internal/server/handlers/readcommand.go +++ b/internal/server/handlers/readcommand.go @@ -2,10 +2,10 @@ package handlers import ( "context" + "io" "os" "path/filepath" "strings" - "sync" "time" "github.com/mimecast/dtail/internal/io/dlog" @@ -32,6 +32,9 @@ func (r *readCommand) Start(ctx context.Context, ltx lcontext.LContext, argc int, args []string, retries int) { re := regex.NewNoop() + var queryStr string + + // Parse regex for non-MapReduce operations or when no aggregate exists if argc >= 4 { deserializedRegex, err := regex.Deserialize(strings.Join(args[2:], " ")) if err != nil { @@ -41,169 +44,18 @@ func (r *readCommand) Start(ctx context.Context, ltx lcontext.LContext, } re = deserializedRegex } + if argc < 3 { r.server.sendln(r.server.serverMessages, dlog.Server.Warn(r.server.user, "Unable to parse command", args, argc)) return } - // Check if channelless mode is enabled - // Note: MapReduce operations require the full channel-based aggregation infrastructure - // Note: Tail operations require continuous monitoring and real-time streaming - isMapReduceCmd := r.mode == omode.MapClient || r.isMapReduceCommand(re) - isTailCmd := r.mode == omode.TailClient - useChannelless := os.Getenv("DTAIL_USE_CHANNELLESS") == "yes" && !isMapReduceCmd && !isTailCmd - - if useChannelless { - dlog.Server.Debug("Using channelless processing mode for mode:", r.mode) - r.startChannelless(ctx, ltx, args, re, retries) - return - } - - dlog.Server.Debug("Using channel-based processing mode for mode:", r.mode) - - // In serverless mode, can also read data from pipe - // e.g.: grep foo bar.log | dmap 'from STATS select ...' - // Only read from stdin if no file is specified AND input is from pipe - if (args[1] == "" || args[1] == "-") && r.isInputFromPipe() { - dlog.Server.Debug("Reading data from stdin pipe") - // Empty file path and globID "-" represents reading from the stdin pipe. - r.read(ctx, ltx, "", "-", re) - return - } - - dlog.Server.Debug("Reading data from file(s)") - r.readGlob(ctx, ltx, args[1], re, retries) -} - -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) - - for retryCount := 0; retryCount < retries; retryCount++ { - paths, err := filepath.Glob(glob) - if err != nil { - dlog.Server.Warn(r.server.user, glob, err) - time.Sleep(retryInterval) - continue - } - - if numPaths := len(paths); numPaths == 0 { - dlog.Server.Error(r.server.user, "No such file(s) to read", glob) - r.server.sendln(r.server.serverMessages, dlog.Server.Warn(r.server.user, - "Unable to read file(s), check server logs")) - select { - case <-ctx.Done(): - return - default: - } - time.Sleep(retryInterval) - continue - } - - r.readFiles(ctx, ltx, paths, glob, re, retryInterval) - return - } - - r.server.sendln(r.server.serverMessages, dlog.Server.Warn(r.server.user, - "Giving up to read file(s)")) - return -} - -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, ltx, &wg, path, glob, re) - } - wg.Wait() -} - -func (r *readCommand) readFileIfPermissions(ctx context.Context, ltx lcontext.LContext, - 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.sendln(r.server.serverMessages, dlog.Server.Warn(r.server.user, - "Unable to read file(s), check server logs")) - return - } - r.read(ctx, ltx, path, globID, re) + // Always use channelless mode now - channel-based code is deprecated + dlog.Server.Debug("Using channelless processing mode for mode:", r.mode) + r.startChannelless(ctx, ltx, args, re, retries, queryStr) } -func (r *readCommand) read(ctx context.Context, ltx lcontext.LContext, - path, globID string, re regex.Regex) { - - dlog.Server.Info(r.server.user, "Start reading", path, globID) - var reader fs.FileReader - var limiter chan struct{} - - switch r.mode { - case omode.GrepClient, omode.CatClient: - reader = fs.NewCatFile(path, globID, r.server.serverMessages) - limiter = r.server.catLimiter - case omode.TailClient: - fallthrough - default: - reader = fs.NewTailFile(path, globID, r.server.serverMessages) - limiter = r.server.tailLimiter - } - - defer func() { - select { - case <-limiter: - default: - } - }() - - select { - case limiter <- struct{}{}: - case <-ctx.Done(): - return - default: - dlog.Server.Info("Server limit hit, queueing file", len(limiter), path) - select { - case limiter <- struct{}{}: - dlog.Server.Info("Server limit OK now, processing file", len(limiter), path) - case <-ctx.Done(): - return - } - } - - lines := r.server.lines - aggregate := r.server.aggregate - - for { - if aggregate != nil { - lines = make(chan *line.Line, 100) - aggregate.NextLinesCh <- lines - } - if err := reader.Start(ctx, ltx, lines, re); err != nil { - dlog.Server.Error(r.server.user, path, globID, err) - } - if aggregate != nil { - // Also makes aggregate to Flush - close(lines) - } - - select { - case <-ctx.Done(): - return - default: - if !reader.Retry() { - return - } - } - time.Sleep(time.Second * 2) - dlog.Server.Info(path, globID, "Reading file again") - } -} func (r *readCommand) makeGlobID(path, glob string) string { var idParts []string @@ -238,22 +90,22 @@ func (r *readCommand) isInputFromPipe() bool { // startChannelless implements channelless processing for better performance func (r *readCommand) startChannelless(ctx context.Context, ltx lcontext.LContext, - args []string, re regex.Regex, retries int) { + args []string, re regex.Regex, retries int, queryStr string) { // Handle stdin input in serverless mode if (args[1] == "" || args[1] == "-") && r.isInputFromPipe() { dlog.Server.Debug("Reading data from stdin pipe (channelless)") - r.readChannellessStdin(ctx, ltx, re) + r.readChannellessStdin(ctx, ltx, re, queryStr) return } dlog.Server.Debug("Reading data from file(s) (channelless)") - r.readGlobChannelless(ctx, ltx, args[1], re, retries) + r.readGlobChannelless(ctx, ltx, args[1], re, retries, queryStr) } // readGlobChannelless processes files using channelless approach func (r *readCommand) readGlobChannelless(ctx context.Context, ltx lcontext.LContext, - glob string, re regex.Regex, retries int) { + glob string, re regex.Regex, retries int, queryStr string) { retryInterval := time.Second * 5 glob = filepath.Clean(glob) @@ -279,7 +131,7 @@ func (r *readCommand) readGlobChannelless(ctx context.Context, ltx lcontext.LCon continue } - r.readFilesChannelless(ctx, ltx, paths, glob, re) + r.readFilesChannelless(ctx, ltx, paths, glob, re, queryStr) return } @@ -289,21 +141,26 @@ func (r *readCommand) readGlobChannelless(ctx context.Context, ltx lcontext.LCon // readFilesChannelless processes multiple files using channelless approach func (r *readCommand) readFilesChannelless(ctx context.Context, ltx lcontext.LContext, - paths []string, glob string, re regex.Regex) { - - // Create network output writer (use nil connection for now - will write to stdout) - output := NewNetworkOutputWriter(nil, r.server.serverMessages, r.server.user) + paths []string, glob string, re regex.Regex, queryStr string) { + + // Choose output writer based on server mode + var output io.Writer + if r.server.serverless { + // In serverless mode, write directly to stdout + output = os.Stdout + } else { + // In client-server mode, write to server handler lines channel + output = NewServerHandlerWriter(r.server, r.server.serverMessages, r.server.user) + } // Create appropriate processor based on mode - processor := r.createChannellessProcessor(re, ltx) + processor, needsFollowing := r.createChannellessProcessor(re, ltx, output, queryStr) // Process each file for _, path := range paths { // Generate globID just like the original system globID := r.makeGlobID(path, glob) - // Create direct processor with proper globID - directProcessor := fs.NewDirectProcessor(processor, output, globID, ltx) if !r.server.user.HasFilePermission(path, "readfiles") { dlog.Server.Error(r.server.user, "No permission to read file", path) r.server.sendln(r.server.serverMessages, dlog.Server.Warn(r.server.user, @@ -313,21 +170,59 @@ func (r *readCommand) readFilesChannelless(ctx context.Context, ltx lcontext.LCo dlog.Server.Info(r.server.user, "Start reading (channelless)", path) - if err := directProcessor.ProcessFile(ctx, path); err != nil { - dlog.Server.Error(r.server.user, path, err) - r.server.sendln(r.server.serverMessages, dlog.Server.Error(r.server.user, - "Error processing file", path, err)) + // Handle file following for tail operations + if needsFollowing { + // For aggregate processors, we need to use following with the aggregate processor + if aggregateProcessor, ok := processor.(*fs.AggregateLineProcessor); ok { + // Create a DirectProcessor wrapper that supports following + directProcessor := fs.NewDirectProcessor(aggregateProcessor, output, globID, ltx) + if err := directProcessor.ProcessFileWithTailing(ctx, path); err != nil { + dlog.Server.Error(r.server.user, path, err) + r.server.sendln(r.server.serverMessages, dlog.Server.Error(r.server.user, + "Error processing file", path, err)) + } + } else if tailProcessor, ok := processor.(*fs.TailProcessor); ok { + followingProcessor := fs.NewFollowingTailProcessor(tailProcessor, output, globID, ltx) + if err := followingProcessor.ProcessFileWithFollowing(ctx, path); err != nil { + dlog.Server.Error(r.server.user, path, err) + r.server.sendln(r.server.serverMessages, dlog.Server.Error(r.server.user, + "Error processing file", path, err)) + } + } else { + // Fallback to regular processing + directProcessor := fs.NewDirectProcessor(processor, output, globID, ltx) + if err := directProcessor.ProcessFile(ctx, path); err != nil { + dlog.Server.Error(r.server.user, path, err) + r.server.sendln(r.server.serverMessages, dlog.Server.Error(r.server.user, + "Error processing file", path, err)) + } + } + } else { + // Regular file processing + directProcessor := fs.NewDirectProcessor(processor, output, globID, ltx) + if err := directProcessor.ProcessFile(ctx, path); err != nil { + dlog.Server.Error(r.server.user, path, err) + r.server.sendln(r.server.serverMessages, dlog.Server.Error(r.server.user, + "Error processing file", path, err)) + } } } } // readChannellessStdin processes stdin using channelless approach -func (r *readCommand) readChannellessStdin(ctx context.Context, ltx lcontext.LContext, re regex.Regex) { - // Create network output writer (use nil connection for now - will write to stdout) - output := NewNetworkOutputWriter(nil, r.server.serverMessages, r.server.user) +func (r *readCommand) readChannellessStdin(ctx context.Context, ltx lcontext.LContext, re regex.Regex, queryStr string) { + // Choose output writer based on server mode + var output io.Writer + if r.server.serverless { + // In serverless mode, write directly to stdout + output = os.Stdout + } else { + // In client-server mode, write to server handler lines channel + output = NewServerHandlerWriter(r.server, r.server.serverMessages, r.server.user) + } // Create appropriate processor based on mode - processor := r.createChannellessProcessor(re, ltx) + processor, _ := r.createChannellessProcessor(re, ltx, output, queryStr) // Create direct processor with "-" as globID for stdin directProcessor := fs.NewDirectProcessor(processor, output, "-", ltx) @@ -355,23 +250,87 @@ func (r *readCommand) isMapReduceCommand(re regex.Regex) bool { } // createChannellessProcessor creates the appropriate processor based on command mode -func (r *readCommand) createChannellessProcessor(re regex.Regex, ltx lcontext.LContext) fs.LineProcessor { +func (r *readCommand) createChannellessProcessor(re regex.Regex, ltx lcontext.LContext, output io.Writer, queryStr string) (fs.LineProcessor, bool) { hostname := r.server.hostname // Use server hostname plain := r.server.plain // Use actual plain mode from server noColor := false // Enable colors by default in channelless mode - + + // If there's an existing aggregate (from a 'map' command), we need to feed data to it + // Create a lines channel and connect it to the aggregate + if r.server.aggregate != nil { + dlog.Server.Debug("Using existing aggregate, creating bridge processor for channelless mode") + // Create a lines channel for the aggregate with larger buffer + linesCh := make(chan *line.Line, 1000) + // Connect the lines channel to the aggregate + go func() { + r.server.aggregate.NextLinesCh <- linesCh + }() + + // Create a bridge processor that feeds lines to the aggregate + var bridgeProcessor fs.LineProcessor + if r.mode == omode.TailClient { + bridgeProcessor = fs.NewAggregateLineProcessorForTail(linesCh, re, hostname, ltx) + } else { + bridgeProcessor = fs.NewAggregateLineProcessor(linesCh, re, hostname, ltx) + } + + // Determine if following is needed + needsFollowing := r.mode == omode.TailClient + return bridgeProcessor, needsFollowing + } + + // No existing aggregate - check if this is a standalone MapReduce operation + isMapReduce := r.isMapReduceCommand(re) || r.mode == omode.MapClient + switch r.mode { case omode.GrepClient: - return fs.NewGrepProcessor(re, plain, noColor, hostname, ltx.BeforeContext, ltx.AfterContext, ltx.MaxCount) + if isMapReduce && queryStr != "" { + // This is a standalone MapReduce grep operation + mapProcessor, err := fs.NewMapProcessor(plain, hostname, queryStr, output) + if err != nil { + dlog.Server.Error(r.server.user, "Failed to create MapReduce processor", err) + return fs.NewGrepProcessor(re, plain, noColor, hostname, ltx.BeforeContext, ltx.AfterContext, ltx.MaxCount), false + } + return mapProcessor, false + } + return fs.NewGrepProcessor(re, plain, noColor, hostname, ltx.BeforeContext, ltx.AfterContext, ltx.MaxCount), false case omode.CatClient: - return fs.NewCatProcessor(plain, noColor, hostname) + if isMapReduce && queryStr != "" { + // This is a standalone MapReduce cat operation + mapProcessor, err := fs.NewMapProcessor(plain, hostname, queryStr, output) + if err != nil { + dlog.Server.Error(r.server.user, "Failed to create MapReduce processor", err) + return fs.NewCatProcessor(plain, noColor, hostname), false + } + return mapProcessor, false + } + return fs.NewCatProcessor(plain, noColor, hostname), false case omode.TailClient: - // For now, basic tail without follow functionality - return fs.NewTailProcessor(re, plain, noColor, hostname, false, false, 0) + if isMapReduce && queryStr != "" { + // This is a standalone MapReduce tail operation + mapProcessor, err := fs.NewMapProcessor(plain, hostname, queryStr, output) + if err != nil { + dlog.Server.Error(r.server.user, "Failed to create MapReduce processor", err) + return fs.NewTailProcessor(re, plain, noColor, hostname, true, true, 0), true + } + return mapProcessor, false + } + // Regular tail operation + return fs.NewTailProcessor(re, plain, noColor, hostname, true, true, 0), true case omode.MapClient: - return fs.NewMapProcessor(plain, hostname) + // Direct MapReduce client - should have queryStr + if queryStr != "" { + mapProcessor, err := fs.NewMapProcessor(plain, hostname, queryStr, output) + if err != nil { + dlog.Server.Error(r.server.user, "Failed to create MapReduce processor", err) + return fs.NewGrepProcessor(re, plain, noColor, hostname, ltx.BeforeContext, ltx.AfterContext, ltx.MaxCount), false + } + return mapProcessor, false + } + // Fallback + return fs.NewGrepProcessor(re, plain, noColor, hostname, ltx.BeforeContext, ltx.AfterContext, ltx.MaxCount), false default: // Default to grep behavior - return fs.NewGrepProcessor(re, plain, noColor, hostname, ltx.BeforeContext, ltx.AfterContext, ltx.MaxCount) + return fs.NewGrepProcessor(re, plain, noColor, hostname, ltx.BeforeContext, ltx.AfterContext, ltx.MaxCount), false } } |