summaryrefslogtreecommitdiff
diff options
context:
space:
mode:
Diffstat
-rw-r--r--integrationtests/commandutils.go36
-rw-r--r--integrationtests/dmap_test.go42
-rw-r--r--integrationtests/dtail_test.go1
-rw-r--r--internal/clients/connectors/serverless.go1
-rw-r--r--internal/io/fs/catfile.go5
-rw-r--r--internal/io/fs/readfile.go26
-rw-r--r--internal/io/fs/tailfile.go5
-rw-r--r--internal/mapr/server/aggregate.go93
-rw-r--r--internal/server/handlers/basehandler.go1
-rw-r--r--internal/server/handlers/readcommand.go39
10 files changed, 163 insertions, 86 deletions
diff --git a/integrationtests/commandutils.go b/integrationtests/commandutils.go
index 23b9c37..d5b5987 100644
--- a/integrationtests/commandutils.go
+++ b/integrationtests/commandutils.go
@@ -12,9 +12,6 @@ import (
"time"
)
-// The exit code and the Go error of the command terminated.
-type exitPromise func() (int, error)
-
func runCommand(ctx context.Context, t *testing.T, stdoutFile, cmdStr string,
args ...string) (int, error) {
@@ -51,7 +48,7 @@ func runCommandRetry(ctx context.Context, t *testing.T, retries int, stdoutFile,
}
func startCommand(ctx context.Context, t *testing.T, cmdStr string,
- args ...string) (<-chan string, <-chan string, exitPromise, error) {
+ args ...string) (<-chan string, <-chan string, <-chan error, error) {
stdoutCh := make(chan string)
stderrCh := make(chan string)
@@ -90,10 +87,33 @@ func startCommand(ctx context.Context, t *testing.T, cmdStr string,
close(stderrCh)
}()
- return stdoutCh, stderrCh, func() (int, error) {
- err := cmd.Wait()
- return exitCodeFromError(err), err
- }, nil
+ cmdErrCh := make(chan error)
+ go func() {
+ cmdErrCh <- cmd.Wait()
+ }()
+
+ return stdoutCh, stderrCh, cmdErrCh, nil
+}
+
+func waitForCommand(ctx context.Context, t *testing.T,
+ stdoutCh, stderrCh <-chan string, cmdErrCh <-chan error) {
+
+ for {
+ select {
+ case line, ok := <-stdoutCh:
+ if ok {
+ t.Log(line)
+ }
+ case line, ok := <-stderrCh:
+ if ok {
+ t.Log(line)
+ }
+ case cmdErr := <-cmdErrCh:
+ t.Log(fmt.Sprintf("Command finished with with exit code %d: %v",
+ exitCodeFromError(cmdErr), cmdErr))
+ return
+ }
+ }
}
func exitCodeFromError(err error) int {
diff --git a/integrationtests/dmap_test.go b/integrationtests/dmap_test.go
index 7d8b8b5..2de679b 100644
--- a/integrationtests/dmap_test.go
+++ b/integrationtests/dmap_test.go
@@ -15,7 +15,6 @@ func TestDMap(t *testing.T) {
return
}
inFile := "mapr_testdata.log"
- stdoutFile := "dmap.stdout.tmp"
csvFile := "dmap.csv.tmp"
expectedCsvFile := "dmap.csv.expected"
queryFile := fmt.Sprintf("%s.query", csvFile)
@@ -25,14 +24,23 @@ func TestDMap(t *testing.T) {
"avg($goroutines),min(concurrentConnections),max(lifetimeConnections) "+
"group by $hostname outfile %s", csvFile)
- _, err := runCommand(context.TODO(), t, stdoutFile,
- "../dmap", "--query", query, inFile)
+ ctx, cancel := context.WithCancel(context.Background())
+ defer cancel()
+
+ stdoutCh, stderrCh, cmdErrCh, err := startCommand(ctx, t, "../dmap",
+ "--query", query,
+ "--logger", "stdout",
+ "--logLevel", "error",
+ "--noColor",
+ inFile)
if err != nil {
t.Error(err)
return
}
+ waitForCommand(ctx, t, stdoutCh, stderrCh, cmdErrCh)
+
if err := compareFiles(t, csvFile, expectedCsvFile); err != nil {
t.Error(err)
return
@@ -42,7 +50,6 @@ func TestDMap(t *testing.T) {
return
}
- os.Remove(stdoutFile)
os.Remove(csvFile)
os.Remove(queryFile)
}
@@ -100,16 +107,31 @@ func TestDMap3(t *testing.T) {
"avg($goroutines),min($goroutines) group by $time order by count($time) "+
"outfile %s", csvFile)
- // Read many input files at once.
- args := []string{"--logLevel", "trace", "--pprof", "localhost:8080", "--query", query}
- for i := 0; i < 100; i++ {
- args = append(args, inFile)
- }
+ ctx, cancel := context.WithCancel(context.Background())
+ defer cancel()
+
+ stdoutCh, stderrCh, cmdErrCh, err := startCommand(ctx, t, "../dmap",
+ "--query", query,
+ "--logger", "stdout",
+ "--logLevel", "info",
+ "--noColor",
+ inFile, inFile, inFile, inFile, inFile, inFile, inFile, inFile, inFile, inFile,
+ inFile, inFile, inFile, inFile, inFile, inFile, inFile, inFile, inFile, inFile,
+ inFile, inFile, inFile, inFile, inFile, inFile, inFile, inFile, inFile, inFile,
+ inFile, inFile, inFile, inFile, inFile, inFile, inFile, inFile, inFile, inFile,
+ inFile, inFile, inFile, inFile, inFile, inFile, inFile, inFile, inFile, inFile,
+ inFile, inFile, inFile, inFile, inFile, inFile, inFile, inFile, inFile, inFile,
+ inFile, inFile, inFile, inFile, inFile, inFile, inFile, inFile, inFile, inFile,
+ inFile, inFile, inFile, inFile, inFile, inFile, inFile, inFile, inFile, inFile,
+ inFile, inFile, inFile, inFile, inFile, inFile, inFile, inFile, inFile, inFile,
+ inFile, inFile, inFile, inFile, inFile, inFile, inFile, inFile, inFile, inFile)
- if _, err := runCommand(context.TODO(), t, stdoutFile, "../dmap", args...); err != nil {
+ if err != nil {
t.Error(err)
return
}
+ waitForCommand(ctx, t, stdoutCh, stderrCh, cmdErrCh)
+
if err := compareFilesContents(t, csvFile, expectedCsvFile); err != nil {
t.Error(err)
return
diff --git a/integrationtests/dtail_test.go b/integrationtests/dtail_test.go
index c6d0107..2b4d6de 100644
--- a/integrationtests/dtail_test.go
+++ b/integrationtests/dtail_test.go
@@ -45,7 +45,6 @@ func TestDTailWithServer(t *testing.T) {
return
}
- // TODO: In testmode, the client should not try to manipulate any known_hosts files.
// TODO: In testmode, never read a config file (use none for all commands)
clientCh, _, _, err := startCommand(ctx, t,
"../dtail",
diff --git a/internal/clients/connectors/serverless.go b/internal/clients/connectors/serverless.go
index 2ff490a..431247a 100644
--- a/internal/clients/connectors/serverless.go
+++ b/internal/clients/connectors/serverless.go
@@ -47,7 +47,6 @@ func (s *Serverless) Start(ctx context.Context, cancel context.CancelFunc,
dlog.Client.Debug("Starting serverless connector")
go func() {
defer cancel()
-
if err := s.handle(ctx, cancel); err != nil {
dlog.Client.Warn(err)
}
diff --git a/internal/io/fs/catfile.go b/internal/io/fs/catfile.go
index 01c15ba..e4676f3 100644
--- a/internal/io/fs/catfile.go
+++ b/internal/io/fs/catfile.go
@@ -6,9 +6,7 @@ type CatFile struct {
}
// NewCatFile returns a new file catter.
-func NewCatFile(filePath string, globID string, serverMessages chan<- string,
- limiter chan struct{}) CatFile {
-
+func NewCatFile(filePath string, globID string, serverMessages chan<- string) CatFile {
return CatFile{
readFile: readFile{
filePath: filePath,
@@ -17,7 +15,6 @@ func NewCatFile(filePath string, globID string, serverMessages chan<- string,
retry: false,
canSkipLines: false,
seekEOF: false,
- limiter: limiter,
},
}
}
diff --git a/internal/io/fs/readfile.go b/internal/io/fs/readfile.go
index 28cbe58..5815aa3 100644
--- a/internal/io/fs/readfile.go
+++ b/internal/io/fs/readfile.go
@@ -38,7 +38,6 @@ type readFile struct {
canSkipLines bool
// Seek to the EOF before processing file?
seekEOF bool
- limiter chan struct{}
}
// String returns the string representation of the readFile
@@ -66,25 +65,7 @@ func (f readFile) Retry() bool {
func (f readFile) Start(ctx context.Context, ltx lcontext.LContext,
lines chan<- line.Line, re regex.Regex) error {
- dlog.Common.Debug("readFile", f)
- defer func() {
- select {
- case <-f.limiter:
- default:
- }
- }()
-
- select {
- case f.limiter <- struct{}{}:
- default:
- select {
- case f.serverMessages <- dlog.Common.Warn(f.filePath, f.globID,
- "Server limit reached. Queuing file..."):
- case <-ctx.Done():
- return nil
- }
- f.limiter <- struct{}{}
- }
+ dlog.Common.Trace("readFile", f)
fd, err := os.Open(f.filePath)
if err != nil {
@@ -156,7 +137,9 @@ func (f readFile) makeReader(fd *os.File) (reader *bufio.Reader, err error) {
return
}
-func (f readFile) read(ctx context.Context, fd *os.File, rawLines chan *bytes.Buffer, truncate <-chan struct{}) error {
+func (f readFile) read(ctx context.Context, fd *os.File, rawLines chan *bytes.Buffer,
+ truncate <-chan struct{}) error {
+
var offset uint64
reader, err := f.makeReader(fd)
if err != nil {
@@ -250,6 +233,7 @@ func (f readFile) filterWithoutLContext(ctx context.Context, rawLines <-chan *by
return
}
if filteredLine, ok := f.transmittable(line, len(lines), cap(lines), re); ok {
+ //dlog.Common.Trace("TODO", "lines", lines, len(lines), cap(lines))
select {
case lines <- filteredLine:
case <-ctx.Done():
diff --git a/internal/io/fs/tailfile.go b/internal/io/fs/tailfile.go
index b03b45d..7a40ac4 100644
--- a/internal/io/fs/tailfile.go
+++ b/internal/io/fs/tailfile.go
@@ -6,9 +6,7 @@ type TailFile struct {
}
// NewTailFile returns a new file tailer.
-func NewTailFile(filePath string, globID string, serverMessages chan<- string,
- limiter chan struct{}) TailFile {
-
+func NewTailFile(filePath string, globID string, serverMessages chan<- string) TailFile {
return TailFile{
readFile: readFile{
filePath: filePath,
@@ -17,7 +15,6 @@ func NewTailFile(filePath string, globID string, serverMessages chan<- string,
retry: true,
canSkipLines: true,
seekEOF: true,
- limiter: limiter,
},
}
}
diff --git a/internal/mapr/server/aggregate.go b/internal/mapr/server/aggregate.go
index 97fee11..11c9ee5 100644
--- a/internal/mapr/server/aggregate.go
+++ b/internal/mapr/server/aggregate.go
@@ -20,6 +20,7 @@ type Aggregate struct {
done *internal.Done
// NextLinesCh can be used to use a new line ch.
NextLinesCh chan chan line.Line
+ linesCh chan line.Line
// Hostname of the current server (used to populate $hostname field).
hostname string
// Signals to serialize data.
@@ -113,58 +114,84 @@ func (a *Aggregate) aggregateTimer(ctx context.Context) {
}
}
+func (a *Aggregate) nextLine() (line line.Line, ok bool, noMoreChannels bool) {
+
+ dlog.Common.Trace("nextLine", "entry", line, ok, noMoreChannels)
+ select {
+ case line, ok = <-a.linesCh:
+ if !ok {
+ // Channel is closed, go to next channel.
+ select {
+ case a.linesCh = <-a.NextLinesCh:
+ default:
+ noMoreChannels = true
+ }
+ }
+ default:
+ // No new line from current lines channel. Try next one.
+ select {
+ case newLinesCh := <-a.NextLinesCh:
+ oldLinesCh := a.linesCh
+ go func() { a.NextLinesCh <- oldLinesCh }()
+ a.linesCh = newLinesCh
+ default:
+ // No new lines channel found.
+ }
+ }
+ dlog.Common.Trace("nextLine", "exit", line, ok, noMoreChannels)
+
+ return
+}
+
func (a *Aggregate) fieldsFromLines(ctx context.Context) <-chan map[string]string {
fieldsCh := make(chan map[string]string)
go func() {
defer close(fieldsCh)
- var lines chan line.Line
// Gather first lines channel (first input file)
select {
- case lines = <-a.NextLinesCh:
+ case a.linesCh = <-a.NextLinesCh:
case <-ctx.Done():
return
}
for {
select {
- case line, ok := <-lines:
- if !ok {
- select {
- case lines = <-a.NextLinesCh:
- // Have a new lines channel (e.g. new input file)
- case <-ctx.Done():
- default:
- // No new lines channel found.
- return
- }
- }
+ case <-ctx.Done():
+ return
+ default:
+ }
- maprLine := strings.TrimSpace(line.Content.String())
- fields, err := a.parser.MakeFields(maprLine)
- // Can't recycle it here yet, as field slices are still
- // TODO: Add unit test reading from multiple mapreduce files lines.
- // TODO: Add capability to recycle this bytes buffer.
- //pool.RecycleBytesBuffer(line.Content)
-
- if err != nil {
- // Should fields be ignored anyway?
- if err != logformat.ErrIgnoreFields {
- dlog.Common.Error(fields, err)
- }
- continue
- }
- if !a.query.WhereClause(fields) {
- continue
+ // Gather first lines channel (first input file)
+ line, ok, noMoreChannels := a.nextLine()
+ if !ok {
+ if noMoreChannels {
+ break
}
+ time.Sleep(time.Millisecond * 100)
+ }
+
+ maprLine := strings.TrimSpace(line.Content.String())
+ fields, err := a.parser.MakeFields(maprLine)
+ // Can't recycle it here yet, as field slices are still
+ // MAYBETODO: Add capability to recycle this bytes buffer.
+ //pool.RecycleBytesBuffer(line.Content)
- select {
- case fieldsCh <- fields:
- case <-ctx.Done():
+ if err != nil {
+ // Should fields be ignored anyway?
+ if err != logformat.ErrIgnoreFields {
+ dlog.Common.Error(fields, err)
}
+ continue
+ }
+ if !a.query.WhereClause(fields) {
+ continue
+ }
+
+ select {
+ case fieldsCh <- fields:
case <-ctx.Done():
- return
}
}
}()
diff --git a/internal/server/handlers/basehandler.go b/internal/server/handlers/basehandler.go
index 6d10d17..53bf375 100644
--- a/internal/server/handlers/basehandler.go
+++ b/internal/server/handlers/basehandler.go
@@ -114,7 +114,6 @@ func (h *baseHandler) Read(p []byte) (n int, err error) {
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
diff --git a/internal/server/handlers/readcommand.go b/internal/server/handlers/readcommand.go
index 4728a55..51077fc 100644
--- a/internal/server/handlers/readcommand.go
+++ b/internal/server/handlers/readcommand.go
@@ -109,18 +109,51 @@ func (r *readCommand) readFileIfPermissions(ctx context.Context, ltx lcontext.LC
r.readFile(ctx, ltx, path, globID, re)
}
+func (*readCommand) limit(ctx context.Context, limiter chan struct{}, message string) {
+ select {
+ case <-ctx.Done():
+ return
+ }
+}
+
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
+ var limiter chan struct{}
+
switch r.mode {
case omode.TailClient:
- reader = fs.NewTailFile(path, globID, r.server.serverMessages, r.server.tailLimiter)
+ reader = fs.NewTailFile(path, globID, r.server.serverMessages)
+ limiter = 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
default:
- reader = fs.NewTailFile(path, globID, r.server.serverMessages, r.server.tailLimiter)
+ 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
generated by cgit v1.2.3 (git 2.25.1) at 2026-06-22 10:12:47 +0000