summaryrefslogtreecommitdiff
path: root/internal/server/handlers/readcommand.go
diff options
context:
space:
mode:
Diffstat (limited to 'internal/server/handlers/readcommand.go')
-rw-r--r--internal/server/handlers/readcommand.go135
1 files changed, 97 insertions, 38 deletions
diff --git a/internal/server/handlers/readcommand.go b/internal/server/handlers/readcommand.go
index b659c06..85f5b2d 100644
--- a/internal/server/handlers/readcommand.go
+++ b/internal/server/handlers/readcommand.go
@@ -2,13 +2,15 @@ package handlers
import (
"context"
+ "os"
"path/filepath"
"strings"
"sync"
"time"
+ "github.com/mimecast/dtail/internal/io/dlog"
"github.com/mimecast/dtail/internal/io/fs"
- "github.com/mimecast/dtail/internal/io/logger"
+ "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,39 +28,56 @@ func newReadCommand(server *ServerHandler, mode omode.Mode) *readCommand {
}
}
-func (r *readCommand) Start(ctx context.Context, lContext lcontext.LContext, argc int, args []string, retries int) {
- re := regex.NewNoop()
+func (r *readCommand) Start(ctx context.Context, ltx lcontext.LContext,
+ 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.sendServerMessage(logger.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.sendServerWarnMessage(logger.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
+ }
+
+ // In serverless mode, can also read data from pipe
+ // e.g.: grep foo bar.log | dmap 'from STATS select ...'
+ if 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
}
- r.readGlob(ctx, lContext, args[1], re, retries)
+
+ dlog.Server.Debug("Reading data from file(s)")
+ r.readGlob(ctx, ltx, args[1], re, retries)
}
-func (r *readCommand) readGlob(ctx context.Context, lContext lcontext.LContext, 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)
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.send(r.server.serverMessages, dlog.Server.Warn(r.server.user,
+ "Unable to read file(s), check server logs"))
select {
case <-ctx.Done():
return
@@ -68,61 +87,93 @@ func (r *readCommand) readGlob(ctx context.Context, lContext lcontext.LContext,
continue
}
- r.readFiles(ctx, lContext, paths, glob, re, retryInterval)
+ r.readFiles(ctx, ltx, paths, glob, re, retryInterval)
return
}
- r.server.sendServerWarnMessage(logger.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, lContext lcontext.LContext, 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, lContext, &wg, path, glob, re)
+ go r.readFileIfPermissions(ctx, ltx, &wg, path, glob, re)
}
-
wg.Wait()
}
-func (r *readCommand) readFileIfPermissions(ctx context.Context, lContext lcontext.LContext, wg *sync.WaitGroup, path, glob string, re regex.Regex) {
+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") {
- 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.send(r.server.serverMessages, dlog.Server.Warn(r.server.user,
+ "Unable to read file(s), check server logs"))
return
}
-
- r.readFile(ctx, lContext, path, globID, re)
+ r.read(ctx, ltx, path, globID, re)
}
-func (r *readCommand) readFile(ctx context.Context, lContext lcontext.LContext, path, globID string, re regex.Regex) {
- logger.Info(r.server.user, "Start reading file", path, globID)
+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.TailClient:
- reader = fs.NewTailFile(path, globID, r.server.serverMessages, r.server.tailLimiter)
case omode.GrepClient, omode.CatClient:
- reader = fs.NewCatFile(path, globID, r.server.serverMessages, r.server.catLimiter)
+ 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, r.server.tailLimiter)
+ reader = fs.NewTailFile(path, globID, r.server.serverMessages)
+ limiter = r.server.tailLimiter
}
- lines := r.server.lines
+ defer func() {
+ select {
+ case <-limiter:
+ default:
+ }
+ }()
- // Plug in mappreduce engine
- if r.server.aggregate != nil {
- lines = r.server.aggregate.Lines
+ 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 err := reader.Start(ctx, lContext, lines, re); err != nil {
- logger.Error(r.server.user, path, globID, err)
+ 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 {
@@ -133,9 +184,8 @@ func (r *readCommand) readFile(ctx context.Context, lContext lcontext.LContext,
return
}
}
-
time.Sleep(time.Second * 2)
- logger.Info(path, globID, "Reading file again")
+ dlog.Server.Info(path, globID, "Reading file again")
}
}
@@ -152,11 +202,20 @@ 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.sendServerWarnMessage(logger.Warn("Empty file path given?", path, glob))
+ r.server.send(r.server.serverMessages,
+ dlog.Server.Warn("Empty file path given?", path, glob))
return ""
}
+
+func (r *readCommand) isInputFromPipe() bool {
+ if !r.server.serverless {
+ // Can read from pipe only in serverless mode.
+ return false
+ }
+ fileInfo, _ := os.Stdin.Stat()
+ return fileInfo.Mode()&os.ModeCharDevice == 0
+}
generated by cgit v1.2.3 (git 2.25.1) at 2026-08-08 11:21:52 +0000