diff options
Diffstat (limited to 'internal')
40 files changed, 196 insertions, 112 deletions
diff --git a/internal/clients/args.go b/internal/clients/args.go index 34fcfa2..7f782f1 100644 --- a/internal/clients/args.go +++ b/internal/clients/args.go @@ -22,4 +22,5 @@ type Args struct { SSHAuthMethods []gossh.AuthMethod SSHHostKeyCallback gossh.HostKeyCallback PrivateKeyPathFile string + Quiet bool } diff --git a/internal/clients/baseclient.go b/internal/clients/baseclient.go index d8d4fde..f20156f 100644 --- a/internal/clients/baseclient.go +++ b/internal/clients/baseclient.go @@ -39,7 +39,7 @@ type baseClient struct { } func (c *baseClient) init() { - logger.Info("Initiating base client") + logger.Debug("Initiating base client") flag := regex.Default if c.Args.RegexInvert { @@ -66,11 +66,11 @@ func (c *baseClient) makeConnections(maker maker) { c.stats = newTailStats(len(c.connections)) } -func (c *baseClient) Start(ctx context.Context, statsCh <-chan struct{}) (status int) { +func (c *baseClient) Start(ctx context.Context, statsCh <-chan string) (status int) { // Periodically check for unknown hosts, and ask the user whether to trust them or not. go c.hostKeyCallback.PromptAddHosts(ctx) // Print client stats every time something on statsCh is recieved. - go c.stats.Start(ctx, c.throttleCh, statsCh) + go c.stats.Start(ctx, c.throttleCh, statsCh, c.Args.Quiet) // Keep count of active connections active := make(chan struct{}, len(c.connections)) @@ -127,7 +127,7 @@ func (c *baseClient) makeConnection(server string, sshAuthMethods []gossh.AuthMe } func (c *baseClient) waitUntilDone(ctx context.Context, active chan struct{}) { - defer logger.Info("Terminated connection") + defer logger.Debug("Terminated connection") // We want to have at least one active connection <-active diff --git a/internal/clients/catclient.go b/internal/clients/catclient.go index d8e9196..b7b6131 100644 --- a/internal/clients/catclient.go +++ b/internal/clients/catclient.go @@ -42,8 +42,9 @@ func (c CatClient) makeHandler(server string) handlers.Handler { } func (c CatClient) makeCommands() (commands []string) { + options := fmt.Sprintf("quiet=%v", c.Args.Quiet) for _, file := range strings.Split(c.What, ",") { - commands = append(commands, fmt.Sprintf("%s %s %s", c.Mode.String(), file, c.Regex.Serialize())) + commands = append(commands, fmt.Sprintf("%s:%s %s %s", c.Mode.String(), options, file, c.Regex.Serialize())) } return } diff --git a/internal/clients/client.go b/internal/clients/client.go index eb8452d..4a547e8 100644 --- a/internal/clients/client.go +++ b/internal/clients/client.go @@ -4,5 +4,5 @@ import "context" // Client is the interface for the end user command line client. type Client interface { - Start(ctx context.Context, statsCh <-chan struct{}) int + Start(ctx context.Context, statsCh <-chan string) int } diff --git a/internal/clients/grepclient.go b/internal/clients/grepclient.go index 4024083..652c31b 100644 --- a/internal/clients/grepclient.go +++ b/internal/clients/grepclient.go @@ -41,8 +41,10 @@ func (c GrepClient) makeHandler(server string) handlers.Handler { } func (c GrepClient) makeCommands() (commands []string) { + options := fmt.Sprintf("quiet=%v", c.Args.Quiet) for _, file := range strings.Split(c.What, ",") { - commands = append(commands, fmt.Sprintf("%s %s %s", c.Mode.String(), file, c.Regex.Serialize())) + commands = append(commands, fmt.Sprintf("%s:%s %s %s", c.Mode.String(), options, file, c.Regex.Serialize())) } + return } diff --git a/internal/clients/handlers/basehandler.go b/internal/clients/handlers/basehandler.go index b5045e2..f07fd90 100644 --- a/internal/clients/handlers/basehandler.go +++ b/internal/clients/handlers/basehandler.go @@ -4,7 +4,6 @@ import ( "encoding/base64" "fmt" "io" - "strconv" "strings" "time" @@ -78,6 +77,7 @@ func (h *baseHandler) Read(p []byte) (n int, err error) { case <-h.Done(): return 0, io.EOF } + return } @@ -111,21 +111,5 @@ func (h *baseHandler) handleHiddenMessage(message string) { case <-h.Done(): return } - - case strings.HasPrefix(message, ".run exitstatus"): - splitted := strings.Split(strings.TrimSuffix(message, "\n"), " ") - if len(splitted) != 3 { - logger.Error("Unable to retrieve exitstatus", message) - return - } - i, err := strconv.Atoi(splitted[2]) - if err != nil { - logger.Error("Unable to retrieve exitstatus", message, err) - return - } - logger.Debug("Retrieved exitstatus", h.status) - if i > h.status { - h.status = i - } } } diff --git a/internal/clients/handlers/healthhandler.go b/internal/clients/handlers/healthhandler.go index 95693ab..0440706 100644 --- a/internal/clients/handlers/healthhandler.go +++ b/internal/clients/handlers/healthhandler.go @@ -19,6 +19,7 @@ type HealthHandler struct { receive chan<- string // The remote server address server string + // The return status. status int } @@ -45,10 +46,12 @@ func (h *HealthHandler) Status() int { return h.status } +// Done returns done channel of the handler. func (h *HealthHandler) Done() <-chan struct{} { return h.done.Done() } +// Shutdown the handler. func (h *HealthHandler) Shutdown() { h.done.Shutdown() } diff --git a/internal/clients/maker.go b/internal/clients/maker.go index 1ba6482..d5ffd8b 100644 --- a/internal/clients/maker.go +++ b/internal/clients/maker.go @@ -4,6 +4,9 @@ import ( "github.com/mimecast/dtail/internal/clients/handlers" ) +// maker interface helps to re-use code in all DTail client implementations. +// All clients share the baseClient but have different connection handlers +// and send different commands to the DTail server. type maker interface { makeHandler(server string) handlers.Handler makeCommands() (commands []string) diff --git a/internal/clients/maprclient.go b/internal/clients/maprclient.go index 149129d..1c0c2cc 100644 --- a/internal/clients/maprclient.go +++ b/internal/clients/maprclient.go @@ -94,12 +94,12 @@ func NewMaprClient(args Args, queryStr string, maprClientMode MaprClientMode) (* } // Start starts the mapreduce client. -func (c *MaprClient) Start(ctx context.Context, statsCh <-chan struct{}) (status int) { +func (c *MaprClient) Start(ctx context.Context, statsCh <-chan string) (status int) { go c.periodicReportResults(ctx) status = c.baseClient.Start(ctx, statsCh) if c.cumulative { - logger.Info("Received final mapreduce result") + logger.Debug("Received final mapreduce result") c.reportResults() } @@ -112,6 +112,7 @@ func (c MaprClient) makeHandler(server string) handlers.Handler { func (c MaprClient) makeCommands() (commands []string) { commands = append(commands, fmt.Sprintf("map %s", c.query.RawQuery)) + options := fmt.Sprintf("quiet=%v", c.Args.Quiet) modeStr := "cat" if c.Mode == omode.TailClient { @@ -123,7 +124,7 @@ func (c MaprClient) makeCommands() (commands []string) { commands = append(commands, fmt.Sprintf("timeout %d %s %s %s", c.Timeout, modeStr, file, c.Regex.Serialize())) continue } - commands = append(commands, fmt.Sprintf("%s %s %s", modeStr, file, c.Regex.Serialize())) + commands = append(commands, fmt.Sprintf("%s:%s %s %s", modeStr, options, file, c.Regex.Serialize())) } return @@ -133,7 +134,7 @@ func (c *MaprClient) periodicReportResults(ctx context.Context) { for { select { case <-time.After(c.query.Interval): - logger.Info("Gathering interim mapreduce result") + logger.Debug("Gathering interim mapreduce result") c.reportResults() case <-ctx.Done(): return @@ -165,7 +166,7 @@ func (c *MaprClient) printResults() { } if numLines == 0 { - logger.Info("Empty result set this time...") + logger.Warn("Empty result set this time...") return } diff --git a/internal/clients/stats.go b/internal/clients/stats.go index e7eabd8..d8163d4 100644 --- a/internal/clients/stats.go +++ b/internal/clients/stats.go @@ -8,6 +8,7 @@ import ( "sync" "time" + "github.com/mimecast/dtail/internal/config" "github.com/mimecast/dtail/internal/io/logger" ) @@ -33,16 +34,18 @@ func newTailStats(connectionsTotal int) *stats { // Start starts printing client connection stats every time a signal is recieved or // connection count has changed. -func (s *stats) Start(ctx context.Context, throttleCh, statsCh <-chan struct{}) { +func (s *stats) Start(ctx context.Context, throttleCh <-chan struct{}, statsCh <-chan string, quiet bool) { var connectedLast int for { var force bool + var messages []string select { - case <-statsCh: + case message := <-statsCh: + messages = append(messages, message) force = true - case <-time.After(time.Second * 2): + case <-time.After(time.Second * 3): case <-ctx.Done(): return } @@ -52,11 +55,18 @@ func (s *stats) Start(ctx context.Context, throttleCh, statsCh <-chan struct{}) newConnections := connected - connectedLast - if connected == connectedLast && !force { + if (connected == connectedLast || quiet) && !force { continue } - logger.Info(s.statsLine(connected, newConnections, throttle)) + stats := s.statsLine(connected, newConnections, throttle) + switch force { + case true: + messages = append(messages, fmt.Sprintf("Connection stats: %s", stats)) + s.printStatsDueInterrupt(messages) + default: + logger.Info(stats) + } connectedLast = connected s.mutex.Lock() @@ -65,6 +75,15 @@ func (s *stats) Start(ctx context.Context, throttleCh, statsCh <-chan struct{}) } } +func (s *stats) printStatsDueInterrupt(messages []string) { + logger.Pause() + for _, message := range messages { + fmt.Println(fmt.Sprintf(" %s", message)) + } + time.Sleep(time.Second * time.Duration(config.InterruptTimeoutS)) + logger.Resume() +} + func (s *stats) statsLine(connected, newConnections int, throttle int) string { percConnected := percentOf(float64(s.connectionsTotal), float64(connected)) @@ -88,5 +107,6 @@ func percentOf(total float64, value float64) float64 { if total == 0 || total == value { return 100 } + return value / (total / 100.0) } diff --git a/internal/clients/tailclient.go b/internal/clients/tailclient.go index 53b5ba4..cefbaa7 100644 --- a/internal/clients/tailclient.go +++ b/internal/clients/tailclient.go @@ -29,6 +29,7 @@ func NewTailClient(args Args) (*TailClient, error) { c.init() c.makeConnections(c) + return &c, nil } @@ -37,8 +38,9 @@ func (c TailClient) makeHandler(server string) handlers.Handler { } func (c TailClient) makeCommands() (commands []string) { + options := fmt.Sprintf("quiet=%v", c.Args.Quiet) for _, file := range strings.Split(c.What, ",") { - commands = append(commands, fmt.Sprintf("%s %s %s", c.Mode.String(), file, c.Regex.Serialize())) + commands = append(commands, fmt.Sprintf("%s:%s %s %s", c.Mode.String(), options, file, c.Regex.Serialize())) } logger.Debug(commands) diff --git a/internal/color/colorfy.go b/internal/color/colorfy.go index 9ae46f5..a2beb7a 100644 --- a/internal/color/colorfy.go +++ b/internal/color/colorfy.go @@ -41,16 +41,14 @@ func paintClientStats(line string) string { // Colorfy a given line based on the line's content. func Colorfy(line string) string { - if strings.HasPrefix(line, "REMOTE") { + switch { + case strings.HasPrefix(line, "REMOTE"): return paintRemote(line) - } - if strings.HasPrefix(line, "CLIENT") && strings.Contains(line, "|stats|") { + case strings.HasPrefix(line, "CLIENT") && strings.Contains(line, "|stats|"): return paintClientStats(line) - } - if strings.Contains(line, "ERROR") { + case strings.Contains(line, "ERROR"): return Paint(Magenta, line) - } - if strings.Contains(line, "WARN") { + case strings.Contains(line, "WARN"): return Paint(Magenta, line) } diff --git a/internal/config/common.go b/internal/config/common.go index 103b390..c3e203e 100644 --- a/internal/config/common.go +++ b/internal/config/common.go @@ -2,10 +2,14 @@ package config // CommonConfig stores configuration keys shared by DTail server and client. type CommonConfig struct { - SSHPort int + // The SSH port number + SSHPort int + // Enable experimental features (mainly for dev purposes) ExperimentalFeaturesEnable bool `json:",omitempty"` - DebugEnable bool `json:",omitempty"` - TraceEnable bool `json:",omitempty"` + // Enable debug logging. Don't enable in production. + DebugEnable bool `json:",omitempty"` + // Enable trace logging. Don't enable in production. + TraceEnable bool `json:",omitempty"` // The log strategy to use, one of // stdout: only log to stdout (useful when used with systemd) // daily: create a log file for every day diff --git a/internal/config/config.go b/internal/config/config.go index dc96d6b..276ddcf 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -15,6 +15,9 @@ const ScheduleUser string = "DTAIL-SCHEDULE" // ContinuousUser is used for non-interactive continuous mapreduce queries. const ContinuousUser string = "DTAIL-CONTINUOUS" +// InterruptTimeoutS is used to terminate DTail when Ctrl+C was pressed twice within a given interval. +const InterruptTimeoutS int = 3 + // Client holds a DTail client configuration. var Client *ClientConfig diff --git a/internal/config/server.go b/internal/config/server.go index db12cec..dc0d587 100644 --- a/internal/config/server.go +++ b/internal/config/server.go @@ -61,6 +61,7 @@ type ServerConfig struct { Continuous []Continuous `json:",omitempty"` } +// ServerRelaxedAuthEnable should be used for development and testing purposes only. var ServerRelaxedAuthEnable bool // Create a new default server configuration. diff --git a/internal/discovery/comma.go b/internal/discovery/comma.go index 94276c7..4344240 100644 --- a/internal/discovery/comma.go +++ b/internal/discovery/comma.go @@ -1,8 +1,9 @@ package discovery import ( - "github.com/mimecast/dtail/internal/io/logger" "strings" + + "github.com/mimecast/dtail/internal/io/logger" ) // ServerListFromCOMMA retrieves a list of servers from comma separated input list. diff --git a/internal/discovery/file.go b/internal/discovery/file.go index c04173e..1250755 100644 --- a/internal/discovery/file.go +++ b/internal/discovery/file.go @@ -2,8 +2,9 @@ package discovery import ( "bufio" - "github.com/mimecast/dtail/internal/io/logger" "os" + + "github.com/mimecast/dtail/internal/io/logger" ) // ServerListFromFILE retrieves a list of servers from a file. diff --git a/internal/io/fs/permissions/permission_linux.go b/internal/io/fs/permissions/permission_linux.go index feae729..bbc039b 100644 --- a/internal/io/fs/permissions/permission_linux.go +++ b/internal/io/fs/permissions/permission_linux.go @@ -11,7 +11,7 @@ import ( "unsafe" ) -// To check whether user has Linux file system permissions to read a given file. +// ToRead checks whether user has Linux file system permissions to read a given file. func ToRead(user, filePath string) (bool, error) { cUser := C.CString(user) cFilePath := C.CString(filePath) diff --git a/internal/io/line/line.go b/internal/io/line/line.go index 9db93c0..715be34 100644 --- a/internal/io/line/line.go +++ b/internal/io/line/line.go @@ -15,7 +15,11 @@ type Line struct { // lines if that happens but it will signal to the client how // many log lines in % could be transmitted to the client. TransmittedPerc int - SourceID string + // Contains the unique identifier of the source log file. + // It could be the name of the log or it could be one of the parent + // directories in case multiple log files with the same basename are + // followed. + SourceID string } // Return a human readable representation of the followed line. diff --git a/internal/io/logger/logger.go b/internal/io/logger/logger.go index b7db0a7..4254eef 100644 --- a/internal/io/logger/logger.go +++ b/internal/io/logger/logger.go @@ -72,12 +72,15 @@ type buf struct { func Start(ctx context.Context, mode Modes) { Mode = mode - if Mode.Nothing { + switch { + case Mode.Nothing: return - } - - if Mode.Trace { + case Mode.Quiet: + Mode.Trace = false + Mode.Debug = false + case Mode.Trace: Mode.Debug = true + default: } strategy := logStrategy() @@ -87,7 +90,7 @@ func Start(ctx context.Context, mode Modes) { case DailyStrategy: _, err := os.Stat(config.Common.LogDir) Mode.logToFile = !os.IsNotExist(err) - Mode.logToStdout = !Mode.Server || Mode.Debug || Mode.Trace + Mode.logToStdout = !Mode.Server || Mode.Debug || Mode.Trace || Mode.Quiet case StdoutStrategy: fallthrough default: @@ -131,11 +134,14 @@ func Info(args ...interface{}) string { // Warn message logging. func Warn(args ...interface{}) string { - if Mode.Server { - return log(serverStr, warnStr, args) + if !Mode.Quiet { + if Mode.Server { + return log(serverStr, warnStr, args) + } + return log(clientStr, warnStr, args) } - return log(clientStr, warnStr, args) + return "" } // Error message logging. @@ -242,7 +248,7 @@ func log(what string, severity string, args []interface{}) string { message := strings.Join(messages, "|") write(what, severity, message) - return message + return fmt.Sprintf("%s|%s", severity, message) } // Raw message logging. diff --git a/internal/io/logger/modes.go b/internal/io/logger/modes.go index 47dadfe..8864179 100644 --- a/internal/io/logger/modes.go +++ b/internal/io/logger/modes.go @@ -6,6 +6,7 @@ type Modes struct { Trace bool Debug bool Nothing bool + Quiet bool logToStdout bool logToFile bool } diff --git a/internal/io/prompt/prompt.go b/internal/io/prompt/prompt.go index a438d33..36ebdb5 100644 --- a/internal/io/prompt/prompt.go +++ b/internal/io/prompt/prompt.go @@ -3,9 +3,10 @@ package prompt import ( "bufio" "fmt" - "github.com/mimecast/dtail/internal/io/logger" "os" "strings" + + "github.com/mimecast/dtail/internal/io/logger" ) // Answer is a user input of a prompt question. @@ -18,8 +19,7 @@ type Answer struct { Callback func() // Runs after Callback and after logging resumes EndCallback func() - - AskAgain bool + AskAgain bool } // Prompt used for interactive user input. diff --git a/internal/io/signal/signal.go b/internal/io/signal/signal.go index 3785d71..500c530 100644 --- a/internal/io/signal/signal.go +++ b/internal/io/signal/signal.go @@ -4,33 +4,38 @@ import ( "context" "os" gosignal "os/signal" + "syscall" "time" - "github.com/mimecast/dtail/internal/io/logger" + "github.com/mimecast/dtail/internal/config" ) -// StatsCh returns a channel for "please print stats" signalling. -func InterruptCh(ctx context.Context) <-chan struct{} { - sigCh := make(chan os.Signal) - gosignal.Notify(sigCh, os.Interrupt) +// InterruptCh returns a channel for "please print stats" signalling. +func InterruptCh(ctx context.Context) <-chan string { + sigIntCh := make(chan os.Signal) + gosignal.Notify(sigIntCh, os.Interrupt) - statsCh := make(chan struct{}) + sigOtherCh := make(chan os.Signal) + gosignal.Notify(sigOtherCh, syscall.SIGHUP, syscall.SIGTERM, syscall.SIGQUIT) + + statsCh := make(chan string) go func() { for { select { - case <-sigCh: + case <-sigIntCh: select { - case statsCh <- struct{}{}: - logger.Info("Hit Ctrl+C twice to exit") + case statsCh <- "Hint: Hit Ctrl+C again to exit": select { - case <-sigCh: + case <-sigIntCh: os.Exit(0) - case <-time.After(time.Second): + case <-time.After(time.Second * time.Duration(config.InterruptTimeoutS)): } default: - // Stats currently already printed. + // Stats already printed. } + case <-sigOtherCh: + os.Exit(0) case <-ctx.Done(): return } diff --git a/internal/mapr/funcs/function.go b/internal/mapr/funcs/function.go index 52aaa98..1a89c3a 100644 --- a/internal/mapr/funcs/function.go +++ b/internal/mapr/funcs/function.go @@ -12,6 +12,7 @@ type CallbackFunc func(text string) string type Function struct { // Name of the callback function Name string + // The Go-callback function to call for this DTail function. call CallbackFunc } diff --git a/internal/mapr/query.go b/internal/mapr/query.go index 7f6b63c..01852da 100644 --- a/internal/mapr/query.go +++ b/internal/mapr/query.go @@ -177,12 +177,6 @@ func (q *Query) parse(tokens []token) error { } } - // Comment out for empty table support, which is "all" log lines. - /* - if q.Table == "" { - return errors.New(invalidQuery + "Empty table specified in 'from' clause") - } - */ if len(q.Select) < 1 { return errors.New(invalidQuery + "Expected at least one field in 'select' clause but got none") } diff --git a/internal/mapr/selectcondition.go b/internal/mapr/selectcondition.go index 1882b7e..d6aa0d4 100644 --- a/internal/mapr/selectcondition.go +++ b/internal/mapr/selectcondition.go @@ -92,5 +92,6 @@ func makeSelectConditions(tokens []token) ([]selectCondition, error) { } sel = append(sel, sc) } + return sel, nil } diff --git a/internal/mapr/server/aggregate.go b/internal/mapr/server/aggregate.go index cd59b63..28bb074 100644 --- a/internal/mapr/server/aggregate.go +++ b/internal/mapr/server/aggregate.go @@ -80,6 +80,7 @@ func NewAggregate(queryStr string) (*Aggregate, error) { return &a, nil } +// Shutdown the aggregation engine. func (a *Aggregate) Shutdown() { a.Flush() a.done.Shutdown() diff --git a/internal/mapr/token.go b/internal/mapr/token.go index d337bd2..8972188 100644 --- a/internal/mapr/token.go +++ b/internal/mapr/token.go @@ -22,6 +22,7 @@ func (t token) isKeyword() bool { return true } } + return false } @@ -94,6 +95,7 @@ func tokensConsumeStr(tokens []token) ([]token, []string) { for _, token := range found { strings = append(strings, token.str) } + return tokens, strings } @@ -104,5 +106,6 @@ func tokensConsumeOptional(tokens []token, optional string) []token { if strings.ToLower(tokens[0].str) == strings.ToLower(optional) { return tokens[1:] } + return tokens } diff --git a/internal/mapr/wherecondition.go b/internal/mapr/wherecondition.go index ff1b489..7a60dba 100644 --- a/internal/mapr/wherecondition.go +++ b/internal/mapr/wherecondition.go @@ -19,6 +19,10 @@ const ( StringNe QueryOperation = iota StringContains QueryOperation = iota StringNotContains QueryOperation = iota + StringHasPrefix QueryOperation = iota + StringNotHasPrefix QueryOperation = iota + StringHasSuffix QueryOperation = iota + StringNotHasSuffix QueryOperation = iota FloatOperation QueryOperation = iota FloatEq QueryOperation = iota FloatNe QueryOperation = iota @@ -78,7 +82,17 @@ func makeWhereConditions(tokens []token) (where []whereCondition, err error) { case "contains": wc.Operation = StringContains case "lacks": + fallthrough + case "ncontains": wc.Operation = StringNotContains + case "hasprefix": + wc.Operation = StringHasPrefix + case "nhasprefix": + wc.Operation = StringNotHasPrefix + case "hassuffix": + wc.Operation = StringHasSuffix + case "nhassuffix": + wc.Operation = StringNotHasSuffix default: return wc, nil, errors.New(invalidQuery + "Unknown operation in 'where' clause: " + whereOp) } @@ -156,6 +170,7 @@ func (wc *whereCondition) floatClause(lValue float64, rValue float64) bool { default: logger.Error("Unknown float operation", lValue, wc.Operation, rValue) } + return false } @@ -169,8 +184,17 @@ func (wc *whereCondition) stringClause(lValue string, rValue string) bool { return strings.Contains(lValue, rValue) case StringNotContains: return !strings.Contains(lValue, rValue) + case StringHasPrefix: + return strings.HasPrefix(lValue, rValue) + case StringNotHasPrefix: + return !strings.HasPrefix(lValue, rValue) + case StringHasSuffix: |