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/clients/args.go | 63 -------- internal/clients/baseclient.go | 28 ++-- internal/clients/catclient.go | 12 +- internal/clients/connectors/serverconnection.go | 34 ++-- internal/clients/connectors/serverless.go | 21 ++- internal/clients/grepclient.go | 13 +- internal/clients/handlers/basehandler.go | 6 +- internal/clients/handlers/clienthandler.go | 4 +- internal/clients/handlers/maprhandler.go | 8 +- internal/clients/maprclient.go | 40 ++--- internal/clients/stats.go | 8 +- internal/clients/tailclient.go | 16 +- internal/color/table.go | 8 +- internal/config/args.go | 105 ++++++++++++ internal/config/common.go | 8 +- internal/config/config.go | 3 + internal/config/read.go | 40 ----- internal/config/server.go | 1 + internal/config/setup.go | 35 ++++ internal/discovery/comma.go | 4 +- internal/discovery/discovery.go | 18 +-- internal/discovery/file.go | 8 +- internal/io/dlog/dlog.go | 206 ++++++++++++++++++++++++ internal/io/dlog/level.go | 89 ++++++++++ internal/io/dlog/loggers/factory.go | 60 +++++++ internal/io/dlog/loggers/file.go | 156 ++++++++++++++++++ internal/io/dlog/loggers/fout.go | 46 ++++++ internal/io/dlog/loggers/logger.go | 18 +++ internal/io/dlog/loggers/none.go | 21 +++ internal/io/dlog/loggers/stdout.go | 73 +++++++++ internal/io/dlog/rotation.go | 27 ++++ internal/io/dlog/source.go | 19 +++ internal/io/dlog/strategy.go | 22 +++ internal/io/fs/permissions/permission.go | 4 +- internal/io/fs/readfile.go | 16 +- internal/io/logger/logger.go | 23 ++- internal/io/logger/modes.go | 9 +- internal/io/prompt/prompt.go | 6 +- internal/mapr/aggregateset.go | 6 +- internal/mapr/client/aggregate.go | 4 +- internal/mapr/funcs/function.go | 4 +- internal/mapr/groupset.go | 4 +- internal/mapr/logformat/default.go | 1 + internal/mapr/logformat/generickv.go | 2 +- internal/mapr/logformat/parser.go | 3 - internal/mapr/query.go | 4 - internal/mapr/server/aggregate.go | 22 +-- internal/mapr/token.go | 10 +- internal/mapr/whereclause.go | 6 +- internal/mapr/wherecondition.go | 6 +- internal/regex/regex.go | 11 +- internal/regex/regex_test.go | 16 +- 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 +- internal/ssh/client/authmethods.go | 28 ++-- internal/ssh/client/knownhostscallback.go | 14 +- internal/ssh/server/hostkey.go | 17 +- internal/ssh/server/publickeycallback.go | 12 +- internal/ssh/ssh.go | 6 +- internal/user/server/user.go | 22 +-- 65 files changed, 1236 insertions(+), 425 deletions(-) delete mode 100644 internal/clients/args.go create mode 100644 internal/config/args.go delete mode 100644 internal/config/read.go create mode 100644 internal/config/setup.go create mode 100644 internal/io/dlog/dlog.go create mode 100644 internal/io/dlog/level.go create mode 100644 internal/io/dlog/loggers/factory.go create mode 100644 internal/io/dlog/loggers/file.go create mode 100644 internal/io/dlog/loggers/fout.go create mode 100644 internal/io/dlog/loggers/logger.go create mode 100644 internal/io/dlog/loggers/none.go create mode 100644 internal/io/dlog/loggers/stdout.go create mode 100644 internal/io/dlog/rotation.go create mode 100644 internal/io/dlog/source.go create mode 100644 internal/io/dlog/strategy.go (limited to 'internal') diff --git a/internal/clients/args.go b/internal/clients/args.go deleted file mode 100644 index dc71e83..0000000 --- a/internal/clients/args.go +++ /dev/null @@ -1,63 +0,0 @@ -package clients - -import ( - "flag" - "fmt" - "strings" - - "github.com/mimecast/dtail/internal/omode" - - gossh "golang.org/x/crypto/ssh" -) - -// Args is a helper struct to summarize common client arguments. -type Args struct { - Mode omode.Mode - ServersStr string - UserName string - What string - Arguments []string - RegexStr string - RegexInvert bool - TrustAllHosts bool - Discovery string - ConnectionsPerCPU int - Timeout int - SSHAuthMethods []gossh.AuthMethod - SSHHostKeyCallback gossh.HostKeyCallback - PrivateKeyPathFile string - Quiet bool - Spartan bool - NoColor bool - Serverless bool -} - -// Transform the arguments based on certain conditions. -func (a *Args) Transform(args []string) { - // Interpret additional args as file list. - if a.What == "" { - var files []string - for _, file := range flag.Args() { - files = append(files, file) - } - a.What = strings.Join(files, ",") - } - - if a.Spartan { - a.Quiet = true - a.NoColor = true - } - - if a.Discovery == "" && a.ServersStr == "" { - a.Serverless = true - } -} - -// TransformAfterConfigFile same as Transform, but after the config file has been read. -func (a *Args) TransformAfterConfigFile() { - // TODO: Remove this method. It's not used. -} - -func (a *Args) SerializeOptions() string { - return fmt.Sprintf("quiet=%v:spartan=%v", a.Quiet, a.Spartan) -} diff --git a/internal/clients/baseclient.go b/internal/clients/baseclient.go index 5523052..fc01955 100644 --- a/internal/clients/baseclient.go +++ b/internal/clients/baseclient.go @@ -6,8 +6,9 @@ import ( "time" "github.com/mimecast/dtail/internal/clients/connectors" + "github.com/mimecast/dtail/internal/config" "github.com/mimecast/dtail/internal/discovery" - "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" "github.com/mimecast/dtail/internal/ssh/client" @@ -17,7 +18,7 @@ import ( // This is the main client data structure. type baseClient struct { - Args + config.Args // To display client side stats stats *stats // List of remote servers to connect to. @@ -39,7 +40,8 @@ type baseClient struct { } func (c *baseClient) init() { - logger.Debug("Initiating base client") + dlog.Client.Debug("Initiating base client") + dlog.Client.Debug(c.Args.String()) flag := regex.Default if c.Args.RegexInvert { @@ -47,11 +49,14 @@ func (c *baseClient) init() { } regex, err := regex.New(c.Args.RegexStr, flag) if err != nil { - logger.FatalExit(c.Regex, "invalid regex!", err, regex) + dlog.Client.FatalPanic(c.Regex, "invalid regex!", err, regex) } c.Regex = regex - logger.Debug("Regex", c.Regex) + dlog.Client.Debug("Regex", c.Regex) + if c.Args.Serverless { + return + } c.sshAuthMethods, c.hostKeyCallback = client.InitSSHAuthMethods(c.Args.SSHAuthMethods, c.Args.SSHHostKeyCallback, c.Args.TrustAllHosts, c.throttleCh, c.Args.PrivateKeyPathFile) } @@ -67,8 +72,11 @@ func (c *baseClient) makeConnections(maker maker) { } 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) + // Can be nil when serverless. + if c.hostKeyCallback != nil { + // 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, c.Args.Quiet) @@ -112,7 +120,7 @@ func (c *baseClient) start(ctx context.Context, active chan struct{}, i int, con } time.Sleep(time.Second * 2) - logger.Debug(conn.Server(), "Reconnecting") + dlog.Client.Debug(conn.Server(), "Reconnecting") conn = c.makeConnection(conn.Server(), c.sshAuthMethods, c.hostKeyCallback) c.connections[i] = conn } @@ -127,7 +135,7 @@ func (c *baseClient) makeConnection(server string, sshAuthMethods []gossh.AuthMe } func (c *baseClient) waitUntilDone(ctx context.Context, active chan struct{}) { - defer logger.Debug("Terminated connection") + defer dlog.Client.Debug("Terminated connection") // We want to have at least one active connection <-active @@ -143,7 +151,7 @@ func (c *baseClient) waitUntilDone(ctx context.Context, active chan struct{}) { if numActive == 0 { return } - logger.Debug("Active connections", numActive) + dlog.Client.Debug("Active connections", numActive) time.Sleep(time.Second) } } diff --git a/internal/clients/catclient.go b/internal/clients/catclient.go index d14cdcc..2726e7e 100644 --- a/internal/clients/catclient.go +++ b/internal/clients/catclient.go @@ -7,6 +7,8 @@ import ( "strings" "github.com/mimecast/dtail/internal/clients/handlers" + "github.com/mimecast/dtail/internal/config" + "github.com/mimecast/dtail/internal/io/dlog" "github.com/mimecast/dtail/internal/omode" ) @@ -16,7 +18,7 @@ type CatClient struct { } // NewCatClient returns a new cat client. -func NewCatClient(args Args) (*CatClient, error) { +func NewCatClient(args config.Args) (*CatClient, error) { if args.RegexStr != "" { return nil, errors.New("Can't use regex with 'cat' operating mode") } @@ -42,11 +44,13 @@ func (c CatClient) makeHandler(server string) handlers.Handler { } func (c CatClient) makeCommands() (commands []string) { + regex, err := c.Regex.Serialize() + if err != nil { + dlog.Client.FatalPanic(err) + } for _, file := range strings.Split(c.What, ",") { commands = append(commands, fmt.Sprintf("%s:%s %s %s", - c.Mode.String(), - c.Args.SerializeOptions(), - file, c.Regex.Serialize())) + c.Mode.String(), c.Args.SerializeOptions(), file, regex)) } return } diff --git a/internal/clients/connectors/serverconnection.go b/internal/clients/connectors/serverconnection.go index 0904ba1..5bc63ee 100644 --- a/internal/clients/connectors/serverconnection.go +++ b/internal/clients/connectors/serverconnection.go @@ -10,7 +10,7 @@ import ( "github.com/mimecast/dtail/internal/clients/handlers" "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/ssh/client" "golang.org/x/crypto/ssh" @@ -30,7 +30,7 @@ type ServerConnection struct { // NewServerConnection returns a new DTail SSH server connection. func NewServerConnection(server string, userName string, authMethods []ssh.AuthMethod, hostKeyCallback client.HostKeyCallback, handler handlers.Handler, commands []string) *ServerConnection { - logger.Debug(server, "Creating new connection", server, handler, commands) + dlog.Client.Debug(server, "Creating new connection", server, handler, commands) c := ServerConnection{ hostKeyCallback: hostKeyCallback, @@ -81,10 +81,10 @@ func (c *ServerConnection) initServerPort() { parts := strings.Split(c.server, ":") if len(parts) == 2 { - logger.Debug("Parsing port from hostname", parts) + dlog.Client.Debug("Parsing port from hostname", parts) port, err := strconv.Atoi(parts[1]) if err != nil { - logger.FatalExit("Unable to parse client port", c.server, parts, err) + dlog.Client.FatalPanic("Unable to parse client port", c.server, parts, err) } c.server = parts[0] c.port = port @@ -93,21 +93,21 @@ func (c *ServerConnection) initServerPort() { func (c *ServerConnection) Start(ctx context.Context, cancel context.CancelFunc, throttleCh, statsCh chan struct{}) { // Throttle how many connections can be established concurrently (based on ch length) - logger.Debug(c.server, "Throttling connection", len(throttleCh), cap(throttleCh)) + dlog.Client.Debug(c.server, "Throttling connection", len(throttleCh), cap(throttleCh)) select { case throttleCh <- struct{}{}: case <-ctx.Done(): - logger.Debug(c.server, "Not establishing connection as context is done", len(throttleCh), cap(throttleCh)) + dlog.Client.Debug(c.server, "Not establishing connection as context is done", len(throttleCh), cap(throttleCh)) return } - logger.Debug(c.server, "Throttling says that the connection can be established", len(throttleCh), cap(throttleCh)) + dlog.Client.Debug(c.server, "Throttling says that the connection can be established", len(throttleCh), cap(throttleCh)) go func() { defer func() { if !c.throttlingDone { - logger.Debug(c.server, "Unthrottling connection (1)", len(throttleCh), cap(throttleCh)) + dlog.Client.Debug(c.server, "Unthrottling connection (1)", len(throttleCh), cap(throttleCh)) c.throttlingDone = true <-throttleCh } @@ -115,9 +115,9 @@ func (c *ServerConnection) Start(ctx context.Context, cancel context.CancelFunc, }() if err := c.dial(ctx, cancel, throttleCh, statsCh); err != nil { - logger.Warn(c.server, c.port, err) + dlog.Client.Warn(c.server, c.port, err) if c.hostKeyCallback.Untrusted(fmt.Sprintf("%s:%d", c.server, c.port)) { - logger.Debug(c.server, "Not trusting host") + dlog.Client.Debug(c.server, "Not trusting host") } } }() @@ -127,14 +127,14 @@ func (c *ServerConnection) Start(ctx context.Context, cancel context.CancelFunc, // Dail into a new SSH connection. Close connection in case of an error. func (c *ServerConnection) dial(ctx context.Context, cancel context.CancelFunc, throttleCh, statsCh chan struct{}) error { - logger.Debug(c.server, "Incrementing connection stats") + dlog.Client.Debug(c.server, "Incrementing connection stats") statsCh <- struct{}{} defer func() { - logger.Debug(c.server, "Decrementing connection stats") + dlog.Client.Debug(c.server, "Decrementing connection stats") <-statsCh }() - logger.Debug(c.server, "Dialing into the connection") + dlog.Client.Debug(c.server, "Dialing into the connection") address := fmt.Sprintf("%s:%d", c.server, c.port) client, err := ssh.Dial("tcp", address, c.config) @@ -148,7 +148,7 @@ func (c *ServerConnection) dial(ctx context.Context, cancel context.CancelFunc, // Create the SSH session. Close the session in case of an error. func (c *ServerConnection) session(ctx context.Context, cancel context.CancelFunc, client *ssh.Client, throttleCh chan struct{}) error { - logger.Debug(c.server, "Creating SSH session") + dlog.Client.Debug(c.server, "Creating SSH session") session, err := client.NewSession() if err != nil { @@ -160,7 +160,7 @@ func (c *ServerConnection) session(ctx context.Context, cancel context.CancelFun } func (c *ServerConnection) handle(ctx context.Context, cancel context.CancelFunc, session *ssh.Session, throttleCh chan struct{}) error { - logger.Debug(c.server, "Creating handler for SSH session") + dlog.Client.Debug(c.server, "Creating handler for SSH session") stdinPipe, err := session.StdinPipe() if err != nil { @@ -196,12 +196,12 @@ func (c *ServerConnection) handle(ctx context.Context, cancel context.CancelFunc // Send all commands to client. for _, command := range c.commands { - logger.Debug(command) + dlog.Client.Debug(command) c.handler.SendMessage(command) } if !c.throttlingDone { - logger.Debug(c.server, "Unthrottling connection (2)", len(throttleCh), cap(throttleCh)) + dlog.Client.Debug(c.server, "Unthrottling connection (2)", len(throttleCh), cap(throttleCh)) c.throttlingDone = true <-throttleCh } diff --git a/internal/clients/connectors/serverless.go b/internal/clients/connectors/serverless.go index 0500645..c7b5f62 100644 --- a/internal/clients/connectors/serverless.go +++ b/internal/clients/connectors/serverless.go @@ -6,7 +6,7 @@ import ( "github.com/mimecast/dtail/internal/clients/handlers" "github.com/mimecast/dtail/internal/config" - "github.com/mimecast/dtail/internal/io/logger" + "github.com/mimecast/dtail/internal/io/dlog" serverHandlers "github.com/mimecast/dtail/internal/server/handlers" user "github.com/mimecast/dtail/internal/user/server" ) @@ -26,7 +26,7 @@ func NewServerless(userName string, handler handlers.Handler, commands []string) commands: commands, } - logger.Debug("Creating new serverless connector", handler, commands) + dlog.Client.Debug("Creating new serverless connector", handler, commands) return &s } @@ -43,7 +43,7 @@ func (s *Serverless) Start(ctx context.Context, cancel context.CancelFunc, throt defer cancel() if err := s.handle(ctx, cancel); err != nil { - logger.Warn(err) + dlog.Client.Warn(err) } }() @@ -51,7 +51,7 @@ func (s *Serverless) Start(ctx context.Context, cancel context.CancelFunc, throt } func (s *Serverless) handle(ctx context.Context, cancel context.CancelFunc) error { - logger.Debug("Creating server handler for a serverless session") + dlog.Client.Debug("Creating server handler for a serverless session") serverHandler := serverHandlers.NewServerHandler( user.New(s.userName, s.Server()), @@ -59,14 +59,19 @@ func (s *Serverless) handle(ctx context.Context, cancel context.CancelFunc) erro make(chan struct{}, config.Server.MaxConcurrentTails), ) + terminate := func() { + serverHandler.Shutdown() + cancel() + } + go func() { io.Copy(serverHandler, s.handler) - cancel() + terminate() }() go func() { io.Copy(s.handler, serverHandler) - cancel() + terminate() }() go func() { @@ -74,12 +79,12 @@ func (s *Serverless) handle(ctx context.Context, cancel context.CancelFunc) erro case <-s.handler.Done(): case <-ctx.Done(): } - cancel() + terminate() }() // Send all commands to client. for _, command := range s.commands { - logger.Debug(command) + dlog.Client.Debug(command) s.handler.SendMessage(command) } diff --git a/internal/clients/grepclient.go b/internal/clients/grepclient.go index ea5022b..ae21ff2 100644 --- a/internal/clients/grepclient.go +++ b/internal/clients/grepclient.go @@ -7,6 +7,8 @@ import ( "strings" "github.com/mimecast/dtail/internal/clients/handlers" + "github.com/mimecast/dtail/internal/config" + "github.com/mimecast/dtail/internal/io/dlog" "github.com/mimecast/dtail/internal/omode" ) @@ -16,7 +18,7 @@ type GrepClient struct { } // NewGrepClient creates a new grep client. -func NewGrepClient(args Args) (*GrepClient, error) { +func NewGrepClient(args config.Args) (*GrepClient, error) { if args.RegexStr == "" { return nil, errors.New("No regex specified, use '-regex' flag") } @@ -41,12 +43,13 @@ func (c GrepClient) makeHandler(server string) handlers.Handler { } func (c GrepClient) makeCommands() (commands []string) { + regex, err := c.Regex.Serialize() + if err != nil { + dlog.Client.FatalPanic(err) + } for _, file := range strings.Split(c.What, ",") { commands = append(commands, fmt.Sprintf("%s:%s %s %s", - c.Mode.String(), - c.Args.SerializeOptions(), - file, - c.Regex.Serialize())) + c.Mode.String(), c.Args.SerializeOptions(), file, regex)) } return diff --git a/internal/clients/handlers/basehandler.go b/internal/clients/handlers/basehandler.go index af1ad62..3291b43 100644 --- a/internal/clients/handlers/basehandler.go +++ b/internal/clients/handlers/basehandler.go @@ -9,7 +9,7 @@ import ( "time" "github.com/mimecast/dtail/internal" - "github.com/mimecast/dtail/internal/io/logger" + "github.com/mimecast/dtail/internal/io/dlog" "github.com/mimecast/dtail/internal/protocol" ) @@ -51,7 +51,7 @@ func (h *baseHandler) Shutdown() { // SendMessage to the server. func (h *baseHandler) SendMessage(command string) error { encoded := base64.StdEncoding.EncodeToString([]byte(command)) - logger.Debug("Sending command", h.server, command, encoded) + dlog.Client.Debug("Sending command", h.server, command, encoded) select { case h.commands <- fmt.Sprintf("protocol %s base64 %v;", protocol.ProtocolCompat, encoded): @@ -112,7 +112,7 @@ func (h *baseHandler) handleMessageType(message string) { return } - logger.Raw(message) + dlog.Client.Raw(message) } // Handle messages received from server which are not meant to be displayed diff --git a/internal/clients/handlers/clienthandler.go b/internal/clients/handlers/clienthandler.go index 2bcb038..27ac85e 100644 --- a/internal/clients/handlers/clienthandler.go +++ b/internal/clients/handlers/clienthandler.go @@ -2,7 +2,7 @@ package handlers import ( "github.com/mimecast/dtail/internal" - "github.com/mimecast/dtail/internal/io/logger" + "github.com/mimecast/dtail/internal/io/dlog" ) // ClientHandler is the basic client handler interface. @@ -12,7 +12,7 @@ type ClientHandler struct { // NewClientHandler creates a new client handler. func NewClientHandler(server string) *ClientHandler { - logger.Debug(server, "Creating new client handler") + dlog.Client.Debug(server, "Creating new client handler") return &ClientHandler{ baseHandler{ diff --git a/internal/clients/handlers/maprhandler.go b/internal/clients/handlers/maprhandler.go index 65b1454..848e7f0 100644 --- a/internal/clients/handlers/maprhandler.go +++ b/internal/clients/handlers/maprhandler.go @@ -4,7 +4,7 @@ import ( "strings" "github.com/mimecast/dtail/internal" - "github.com/mimecast/dtail/internal/io/logger" + "github.com/mimecast/dtail/internal/io/dlog" "github.com/mimecast/dtail/internal/mapr" "github.com/mimecast/dtail/internal/mapr/client" "github.com/mimecast/dtail/internal/protocol" @@ -40,7 +40,7 @@ func (h *MaprHandler) Write(p []byte) (n int, err error) { continue case protocol.MessageDelimiter: message := h.baseHandler.receiveBuf.String() - logger.Debug(message) + dlog.Client.Debug(message) if message[0] == 'A' { h.handleAggregateMessage(message) } else { @@ -60,10 +60,10 @@ func (h *MaprHandler) Write(p []byte) (n int, err error) { func (h *MaprHandler) handleAggregateMessage(message string) { parts := strings.SplitN(message, protocol.FieldDelimiter, 3) if len(parts) != 3 { - logger.Error("Unable to aggregate data", h.server, message, parts, len(parts), "expected 3 parts") + dlog.Client.Error("Unable to aggregate data", h.server, message, parts, len(parts), "expected 3 parts") return } if err := h.aggregate.Aggregate(parts[2]); err != nil { - logger.Error("Unable to aggregate data", h.server, message, err) + dlog.Client.Error("Unable to aggregate data", h.server, message, err) } } diff --git a/internal/clients/maprclient.go b/internal/clients/maprclient.go index e6ab96b..f23aa08 100644 --- a/internal/clients/maprclient.go +++ b/internal/clients/maprclient.go @@ -11,7 +11,7 @@ import ( "github.com/mimecast/dtail/internal/clients/handlers" "github.com/mimecast/dtail/internal/color" "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/mapr" "github.com/mimecast/dtail/internal/omode" ) @@ -44,14 +44,14 @@ type MaprClient struct { } // NewMaprClient returns a new mapreduce client. -func NewMaprClient(args Args, queryStr string, maprClientMode MaprClientMode) (*MaprClient, error) { +func NewMaprClient(args config.Args, queryStr string, maprClientMode MaprClientMode) (*MaprClient, error) { if queryStr == "" { return nil, errors.New("No mapreduce query specified, use '-query' flag") } query, err := mapr.NewQuery(queryStr) if err != nil { - logger.FatalExit(queryStr, "Can't parse mapr query", err) + dlog.Client.FatalPanic(queryStr, "Can't parse mapr query", err) } // Don't retry connection if in tail mode and no outfile specified. @@ -68,7 +68,7 @@ func NewMaprClient(args Args, queryStr string, maprClientMode MaprClientMode) (* cumulative = args.Mode == omode.MapClient || query.HasOutfile() } - logger.Debug("Cumulative mapreduce mode?", cumulative) + dlog.Client.Debug("Cumulative mapreduce mode?", cumulative) c := MaprClient{ baseClient: baseClient{ @@ -103,7 +103,7 @@ func (c *MaprClient) Start(ctx context.Context, statsCh <-chan string) (status i status = c.baseClient.Start(ctx, statsCh) if c.cumulative { - logger.Debug("Received final mapreduce result") + dlog.Client.Debug("Received final mapreduce result") c.reportResults() } @@ -123,15 +123,17 @@ func (c MaprClient) makeCommands() (commands []string) { } for _, file := range strings.Split(c.What, ",") { + regex, err := c.Regex.Serialize() + if err != nil { + dlog.Client.FatalPanic(err) + } if c.Timeout > 0 { - commands = append(commands, fmt.Sprintf("timeout %d %s %s %s", c.Timeout, modeStr, file, c.Regex.Serialize())) + commands = append(commands, fmt.Sprintf("timeout %d %s %s %s", c.Timeout, + modeStr, file, regex)) continue } commands = append(commands, fmt.Sprintf("%s:%s %s %s", - modeStr, - c.Args.SerializeOptions(), - file, - c.Regex.Serialize())) + modeStr, c.Args.SerializeOptions(), file, regex)) } return @@ -141,7 +143,7 @@ func (c *MaprClient) periodicReportResults(ctx context.Context) { for { select { case <-time.After(c.query.Interval): - logger.Debug("Gathering interim mapreduce result") + dlog.Client.Debug("Gathering interim mapreduce result") c.reportResults() case <-ctx.Done(): return @@ -177,17 +179,17 @@ func (c *MaprClient) printResults() { } if err != nil { - logger.FatalExit(err) + dlog.Client.FatalPanic(err) } if result == c.lastResult { - logger.Debug("Result hasn't changed compared to last time...") + dlog.Client.Debug("Result hasn't changed compared to last time...") return } c.lastResult = result if numRows == 0 { - logger.Debug("Empty result set this time...") + dlog.Client.Debug("Empty result set this time...") return } @@ -198,24 +200,24 @@ func (c *MaprClient) printResults() { config.Client.TermColors.MaprTable.RawQueryBg, config.Client.TermColors.MaprTable.RawQueryAttr) } - logger.Raw(rawQuery) + dlog.Client.Raw(rawQuery) if rowsLimit > 0 && numRows > rowsLimit { - logger.Warn(fmt.Sprintf("Got %d results but limited output to %d rows! Use 'limit' clause to override!", + dlog.Client.Warn(fmt.Sprintf("Got %d results but limited output to %d rows! Use 'limit' clause to override!", numRows, rowsLimit)) } - logger.Raw(result) + dlog.Client.Raw(result) } func (c *MaprClient) writeResultsToOutfile() { if c.cumulative { if err := c.globalGroup.WriteResult(c.query); err != nil { - logger.FatalExit(err) + dlog.Client.FatalPanic(err) } return } if err := c.globalGroup.SwapOut().WriteResult(c.query); err != nil { - logger.FatalExit(err) + dlog.Client.FatalPanic(err) } } diff --git a/internal/clients/stats.go b/internal/clients/stats.go index 6da443c..fbef572 100644 --- a/internal/clients/stats.go +++ b/internal/clients/stats.go @@ -10,7 +10,7 @@ import ( "github.com/mimecast/dtail/internal/color" "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/protocol" ) @@ -67,7 +67,7 @@ func (s *stats) Start(ctx context.Context, throttleCh <-chan struct{}, statsCh < s.printStatsDueInterrupt(messages) default: data := s.statsData(connected, newConnections, throttle) - logger.Mapreduce("STATS", data) + dlog.Client.Mapreduce("STATS", data) } connectedLast = connected @@ -78,7 +78,7 @@ func (s *stats) Start(ctx context.Context, throttleCh <-chan struct{}, statsCh < } func (s *stats) printStatsDueInterrupt(messages []string) { - logger.Pause() + dlog.Client.Pause() for i, message := range messages { if i > 0 && config.Client.TermColorsEnable { fmt.Println(color.PaintStrWithAttr(message, @@ -91,7 +91,7 @@ func (s *stats) printStatsDueInterrupt(messages []string) { fmt.Println(fmt.Sprintf(" %s", message)) } time.Sleep(time.Second * time.Duration(config.InterruptTimeoutS)) - logger.Resume() + dlog.Client.Resume() } func (s *stats) statsData(connected, newConnections int, throttle int) map[string]interface{} { diff --git a/internal/clients/tailclient.go b/internal/clients/tailclient.go index 360354b..d42a0e4 100644 --- a/internal/clients/tailclient.go +++ b/internal/clients/tailclient.go @@ -6,7 +6,8 @@ import ( "strings" "github.com/mimecast/dtail/internal/clients/handlers" - "github.com/mimecast/dtail/internal/io/logger" + "github.com/mimecast/dtail/internal/config" + "github.com/mimecast/dtail/internal/io/dlog" "github.com/mimecast/dtail/internal/omode" ) @@ -16,7 +17,7 @@ type TailClient struct { } // NewTailClient returns a new TailClient. -func NewTailClient(args Args) (*TailClient, error) { +func NewTailClient(args config.Args) (*TailClient, error) { args.Mode = omode.TailClient c := TailClient{ @@ -38,14 +39,15 @@ func (c TailClient) makeHandler(server string) handlers.Handler { } func (c TailClient) makeCommands() (commands []string) { + regex, err := c.Regex.Serialize() + if err != nil { + dlog.Client.FatalPanic(err) + } for _, file := range strings.Split(c.What, ",") { commands = append(commands, fmt.Sprintf("%s:%s %s %s", - c.Mode.String(), - c.Args.SerializeOptions(), - file, - c.Regex.Serialize())) + c.Mode.String(), c.Args.SerializeOptions(), file, regex)) } - logger.Debug(commands) + dlog.Client.Debug(commands) return } diff --git a/internal/color/table.go b/internal/color/table.go index 7ecfbca..2115edf 100644 --- a/internal/color/table.go +++ b/internal/color/table.go @@ -7,17 +7,17 @@ import ( const sampleParagraph string = "Mimecast is Making Email Safer for Business. We believe that securely operating a business in the cloud requires new levels of IT preparedness, centered around cyber resilience. This is why we unify the delivery and management of security, continuity and data protection for email via one, simple-to-use cloud platform. Thousands of organizations trust us to increase their cyber resilience preparedness, streamline compliance, reduce IT complexity and keep their business running. We give employees fast and secure access to sensitive business information, and ensure email keeps flowing in the event of an outage. Mimecast will remain committed to protecting your IT assets through constant innovation and focus on your success." -func TablePrintAndExit(useSampleParagraph bool) { +func TablePrintAndExit(displaySampleParagraph bool) { for _, attr := range AttributeNames { if attr == "Hidden" || attr == "SlowBlink" { continue } - printColorTable(attr, useSampleParagraph) + printColorTable(attr, displaySampleParagraph) } os.Exit(0) } -func printColorTable(attr string, useSampleParagraph bool) { +func printColorTable(attr string, displaySampleParagraph bool) { for _, fg := range ColorNames { fgColor, _ := ToFgColor(fg) for _, bg := range ColorNames { @@ -29,7 +29,7 @@ func printColorTable(attr string, useSampleParagraph bool) { text := fmt.Sprintf(" Foreground:%10s | Background:%10s | Attribute:%10s ", fg, bg, attr) fmt.Print(PaintStrWithAttr(text, fgColor, bgColor, attribute)) - if useSampleParagraph { + if displaySampleParagraph { fmt.Print("\n") fmt.Print(PaintStrWithAttr(sampleParagraph, fgColor, bgColor, attribute)) fmt.Print("\n") diff --git a/internal/config/args.go b/internal/config/args.go new file mode 100644 index 0000000..89e4bc9 --- /dev/null +++ b/internal/config/args.go @@ -0,0 +1,105 @@ +package config + +import ( + "flag" + "fmt" + "strings" + + "github.com/mimecast/dtail/internal/omode" + + gossh "golang.org/x/crypto/ssh" +) + +// Args is a helper struct to summarize common client arguments. +type Args struct { + Arguments []string + ConfigFile string + ConnectionsPerCPU int + Discovery string + LogLevel string + Mode omode.Mode + NoColor bool + PrivateKeyPathFile string + Quiet bool + RegexInvert bool + RegexStr string + Serverless bool + ServersStr string + Spartan bool + SSHAuthMethods []gossh.AuthMethod + SSHHostKeyCallback gossh.HostKeyCallback + SSHPort int + Timeout int + TrustAllHosts bool + UserName string + What string +} + +func (a *Args) String() string { + var sb strings.Builder + + sb.WriteString("Args(") + sb.WriteString(fmt.Sprintf("%s:%s,", "LogLevel", a.LogLevel)) + sb.WriteString(fmt.Sprintf("%s:%v,", "Arguments", a.Arguments)) + sb.WriteString(fmt.Sprintf("%s:%v,", "ConfigFile", a.ConfigFile)) + sb.WriteString(fmt.Sprintf("%s:%v,", "ConnectionsPerCPU", a.ConnectionsPerCPU)) + sb.WriteString(fmt.Sprintf("%s:%v,", "Discovery", a.Discovery)) + sb.WriteString(fmt.Sprintf("%s:%v,", "Mode", a.Mode)) + sb.WriteString(fmt.Sprintf("%s:%v,", "NoColor", a.NoColor)) + sb.WriteString(fmt.Sprintf("%s:%v,", "PrivateKeyPathFile", a.PrivateKeyPathFile)) + sb.WriteString(fmt.Sprintf("%s:%v,", "Quiet", a.Quiet)) + sb.WriteString(fmt.Sprintf("%s:%v,", "RegexInvert", a.RegexInvert)) + sb.WriteString(fmt.Sprintf("%s:%v,", "RegexStr", a.RegexStr)) + sb.WriteString(fmt.Sprintf("%s:%v,", "Serverless", a.Serverless)) + sb.WriteString(fmt.Sprintf("%s:%v,", "ServersStr", a.ServersStr)) + sb.WriteString(fmt.Sprintf("%s:%v,", "Spartan", a.Spartan)) + sb.WriteString(fmt.Sprintf("%s:%v,", "SSHAuthMethods", a.SSHAuthMethods)) + sb.WriteString(fmt.Sprintf("%s:%v,", "SSHHostKeyCallback", a.SSHHostKeyCallback)) + sb.WriteString(fmt.Sprintf("%s:%v,", "SSHPort", a.SSHPort)) + sb.WriteString(fmt.Sprintf("%s:%v,", "Timeout", a.Timeout)) + sb.WriteString(fmt.Sprintf("%s:%v,", "TrustAllHosts", a.TrustAllHosts)) + sb.WriteString(fmt.Sprintf("%s:%v,", "UserName", a.UserName)) + sb.WriteString(fmt.Sprintf("%s:%v", "What", a.What)) + sb.WriteString(")") + + return sb.String() +} + +// Based on the argument list, transform/manipulate some of the arguments. +func (a *Args) transform(args []string) { + if a.LogLevel != "" { + Common.LogLevel = a.LogLevel + } + + if a.SSHPort != 2222 { + Common.SSHPort = a.SSHPort + } + if a.NoColor { + Client.TermColorsEnable = false + } + + if a.Spartan { + a.Quiet = true + a.NoColor = true + } + + if a.Discovery == "" && a.ServersStr == "" { + a.Serverless = true + } + + // Interpret additional args as file list. + if a.What == "" { + var files []string + for _, file := range flag.Args() { + files = append(files, file) + } + a.What = strings.Join(files, ",") + } +} + +// SerializeOptions returns a string ready to be sent over the wire to the server. +func (a *Args) SerializeOptions() string { + return fmt.Sprintf("quiet=%v:spartan=%v", a.Quiet, a.Spartan) +} + +// TODO: Put the DeseializeOptions function here (move it away from the internal/server package) diff --git a/internal/config/common.go b/internal/config/common.go index c3e203e..7d45261 100644 --- a/internal/config/common.go +++ b/internal/config/common.go @@ -6,10 +6,8 @@ type CommonConfig struct { SSHPort int // Enable experimental features (mainly for dev purposes) ExperimentalFeaturesEnable 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"` + // LogLevel defines how much is logged. TODO: Adjust JSONschema + LogLevel string `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 @@ -26,8 +24,6 @@ type CommonConfig struct { func newDefaultCommonConfig() *CommonConfig { return &CommonConfig{ SSHPort: 2222, - DebugEnable: false, - TraceEnable: false, ExperimentalFeaturesEnable: false, LogDir: "log", CacheDir: "cache", diff --git a/internal/config/config.go b/internal/config/config.go index 276ddcf..2d77041 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" +// TestUser is used for unit tests and potentially also for integration tests. +const TestUser string = "DTAIL-TEST" + // InterruptTimeoutS is used to terminate DTail when Ctrl+C was pressed twice within a given interval. const InterruptTimeoutS int = 3 diff --git a/internal/config/read.go b/internal/config/read.go deleted file mode 100644 index ea358f8..0000000 --- a/internal/config/read.go +++ /dev/null @@ -1,40 +0,0 @@ -package config - -import ( - "os" -) - -// Read the DTail configuration. -func Read(configFile string, sshPort int, noColor bool) { - initializer := configInitializer{ - Common: newDefaultCommonConfig(), - Server: newDefaultServerConfig(), - Client: newDefaultClientConfig(), - } - - if configFile == "" { - configFile = "./cfg/dtail.json" - } - - if _, err := os.Stat(configFile); !os.IsNotExist(err) { - initializer.parseConfig(configFile) - } - - // Assign pointers to global variables, so that we can access the - // configuration from any place of the program. - Common = initializer.Common - Server = initializer.Server - Client = initializer.Client - - if Server.MapreduceLogFormat == "" { - Server.MapreduceLogFormat = "default" - } - - // If non-standard port specified, overwrite config - if sshPort != 2222 { - Common.SSHPort = sshPort - } - if noColor { - Client.TermColorsEnable = false - } -} diff --git a/internal/config/server.go b/internal/config/server.go index dc0d587..8bbb394 100644 --- a/internal/config/server.go +++ b/internal/config/server.go @@ -76,6 +76,7 @@ func newDefaultServerConfig() *ServerConfig { MaxConcurrentTails: 50, HostKeyFile: "./cache/ssh_host_key", HostKeyBits: 4096, + MapreduceLogFormat: "default", Permissions: Permissions{ Default: defaultPermissions, }, diff --git a/internal/config/setup.go b/internal/config/setup.go new file mode 100644 index 0000000..3c4bcc4 --- /dev/null +++ b/internal/config/setup.go @@ -0,0 +1,35 @@ +package config + +import ( + "os" +) + +const NoConfigFile string = "Don't read a config file - use defaults only" + +// Setup the DTail configuration. +func Setup(args *Args, additionalArgs []string) { + initializer := configInitializer{ + Common: newDefaultCommonConfig(), + Server: newDefaultServerConfig(), + Client: newDefaultClientConfig(), + } + + if args.ConfigFile == "" { + // TODO: Search more paths for config file (e.g. in /etc and in ~/.config/... + args.ConfigFile = "./cfg/dtail.json" + } + + if args.ConfigFile != NoConfigFile { + if _, err := os.Stat(args.ConfigFile); !os.IsNotExist(err) { + initializer.parseConfig(args.ConfigFile) + } + } + + // Assign pointers to global variables, so that we can access the + // configuration from any place of the program. + Common = initializer.Common + Server = initializer.Server + Client = initializer.Client + + args.transform(additionalArgs) +} diff --git a/internal/discovery/comma.go b/internal/discovery/comma.go index 4344240..9bea89c 100644 --- a/internal/discovery/comma.go +++ b/internal/discovery/comma.go @@ -3,11 +3,11 @@ package discovery import ( "strings" - "github.com/mimecast/dtail/internal/io/logger" + "github.com/mimecast/dtail/internal/io/dlog" ) // ServerListFromCOMMA retrieves a list of servers from comma separated input list. func (d *Discovery) ServerListFromCOMMA() []string { - logger.Debug("Retrieving server list from comma separated list", d.server) + dlog.Common.Debug("Retrieving server list from comma separated list", d.server) return strings.Split(d.server, ",") } diff --git a/internal/discovery/discovery.go b/internal/discovery/discovery.go index 3608ce7..83ee95e 100644 --- a/internal/discovery/discovery.go +++ b/internal/discovery/discovery.go @@ -9,7 +9,7 @@ import ( "strings" "time" - "github.com/mimecast/dtail/internal/io/logger" + "github.com/mimecast/dtail/internal/io/dlog" ) // ServerOrder to specify how to sort the server list. @@ -42,7 +42,7 @@ func New(method, server string, order ServerOrder) *Discovery { if strings.Contains(module, ":") { s := strings.Split(module, ":") if len(s) != 2 { - logger.FatalExit("Unable to parse discovery module", module) + dlog.Common.FatalPanic("Unable to parse discovery module", module) } module = s[0] options = s[1] @@ -72,11 +72,11 @@ func (d *Discovery) initRegex() { } regexStr := string(runes) - logger.Debug("Using filter regex", regexStr) + dlog.Common.Debug("Using filter regex", regexStr) regex, err := regexp.Compile(regexStr) if err != nil { - logger.FatalExit("Could not compile regex", regexStr, err) + dlog.Common.FatalPanic("Could not compile regex", regexStr, err) } d.regex = regex @@ -97,7 +97,7 @@ func (d *Discovery) ServerList() []string { servers = d.shuffleList(servers) } - logger.Debug("Discovered servers", len(servers), servers) + dlog.Common.Debug("Discovered servers", len(servers), servers) return servers } @@ -124,7 +124,7 @@ func (d *Discovery) serverListFromReflectedModule() []string { rt := reflect.TypeOf(d) reflectedMethod, ok := rt.MethodByName(methodName) if !ok { - logger.FatalExit("No such server discovery module", d.module, methodName) + dlog.Common.FatalPanic("No such server discovery module", d.module, methodName) } inputValues := make([]reflect.Value, 1) @@ -138,7 +138,7 @@ func (d *Discovery) serverListFromReflectedModule() []string { // Filter server list based on a regexp. func (d *Discovery) filterList(servers []string) (filtered []string) { - logger.Debug("Filtering server list") + dlog.Common.Debug("Filtering server list") for _, server := range servers { if d.regex.MatchString(server) { @@ -160,13 +160,13 @@ func (d *Discovery) dedupList(servers []string) (deduped []string) { } } - logger.Debug("Deduped server list", len(servers), len(deduped)) + dlog.Common.Debug("Deduped server list", len(servers), len(deduped)) return } // Randomly shuffle the server list. func (d *Discovery) shuffleList(servers []string) []string { - logger.Debug("Shuffling server list") + dlog.Common.Debug("Shuffling server list") r := rand.New(rand.NewSource(time.Now().Unix())) shuffled := make([]string, len(servers)) diff --git a/internal/discovery/file.go b/internal/discovery/file.go index 1250755..fb46eeb 100644 --- a/internal/discovery/file.go +++ b/internal/discovery/file.go @@ -4,16 +4,16 @@ import ( "bufio" "os" - "github.com/mimecast/dtail/internal/io/logger" + "github.com/mimecast/dtail/internal/io/dlog" ) // ServerListFromFILE retrieves a list of servers from a file. func (d *Discovery) ServerListFromFILE() (servers []string) { - logger.Debug("Retrieving server list from file", d.server) + dlog.Common.Debug("Retrieving server list from file", d.server) file, err := os.Open(d.server) if err != nil { - logger.FatalExit(d.server, err) + dlog.Common.FatalPanic(d.server, err) } defer file.Close() @@ -22,7 +22,7 @@ func (d *Discovery) ServerListFromFILE() (servers []string) { servers = append(servers, scanner.Text()) } if err := scanner.Err(); err != nil { - logger.FatalExit(d.server, err) + dlog.Common.FatalPanic(d.server, err) } return diff --git a/internal/io/dlog/dlog.go b/internal/io/dlog/dlog.go new file mode 100644 index 0000000..7282741 --- /dev/null +++ b/internal/io/dlog/dlog.go @@ -0,0 +1,206 @@ +package dlog + +import ( + "context" + "fmt" + "strings" + "sync" + "time" + + "github.com/mimecast/dtail/internal/color/brush" + "github.com/mimecast/dtail/internal/config" + "github.com/mimecast/dtail/internal/io/dlog/loggers" + "github.com/mimecast/dtail/internal/io/pool" + "github.com/mimecast/dtail/internal/protocol" +) + +// Client is the log handler for the client packages. +var Client *DLog + +// Server is the log handler for the server packages. +var Server *DLog + +// Common is the log handler for all other packages. +// TODO: Rename Common to Common +var Common *DLog + +var mutex sync.Mutex +var started bool + +// Start logger(s). +func Start(ctx context.Context, wg *sync.WaitGroup, sourceProcess source, logLevel string) { + mutex.Lock() + defer mutex.Unlock() + + if started { + Common.FatalPanic("Logger already started") + } + + level := newLevel(logLevel) + switch sourceProcess { + case CLIENT: + // This is a DTail client process running. + impl := loggers.FOUT + Client = New(CLIENT, CLIENT, impl, level) + Server = New(CLIENT, SERVER, impl, level) + Common = Client + case SERVER: + // This is a DTail server process running. + impl := loggers.FILE + Client = New(SERVER, CLIENT, impl, level) + Server = New(SERVER, SERVER, impl, level) + Common = Server + } + + var wg2 sync.WaitGroup + wg2.Add(2) + Client.start(ctx, &wg2) + Server.start(ctx, &wg2) + started = true + + go rotation(ctx) + go func() { + wg2.Wait() + wg.Done() + }() +} + +// DLog is the DTail logger. +type DLog struct { + logger loggers.Logger + // Is this a DTail server or client process logging? + sourceProcess source + // Is this a DTail server or client package logging? In serverless mode + // the client can also execute code from the server package. + sourcePackage source + // Max log level to log. + maxLevel level +} + +// New creates a new DTail logger. +func New(sourceProcess, sourcePackage source, impl loggers.Impl, maxLevel level) *DLog { + return &DLog{ + logger: loggers.Factory(sourceProcess.String(), impl), + sourceProcess: sourceProcess, + sourcePackage: sourcePackage, + maxLevel: maxLevel, + } +} + +func (d *DLog) start(ctx context.Context, wg *sync.WaitGroup) { + go func() { + defer wg.Done() + var wg2 sync.WaitGroup + wg2.Add(1) + d.logger.Start(ctx, &wg2) + <-ctx.Done() + wg2.Wait() + }() +} + +func (d *DLog) log(level level, args []interface{}) string { + if d.maxLevel < level { + return "" + } + sb := pool.BuilderBuffer.Get().(*strings.Builder) + defer pool.RecycleBuilderBuffer(sb) + now := time.Now() + + sb.WriteString(d.sourcePackage.String()) + sb.WriteString(protocol.FieldDelimiter) + sb.WriteString(now.Format("20060102-150405")) + sb.WriteString(protocol.FieldDelimiter) + sb.WriteString(level.String()) + sb.WriteString(protocol.FieldDelimiter) + d.writeArgStrings(sb, args) + + message := sb.String() + if !config.Client.TermColorsEnable || !d.logger.SupportsColors() { + d.logger.Log(now, message) + return message + } + + d.logger.LogWithColors(now, message, brush.Colorfy(message)) + return message +} + +func (d *DLog) writeArgStrings(sb *strings.Builder, args []interface{}) { + for i, arg := range args { + if i > 0 { + sb.WriteString(protocol.FieldDelimiter) + } + switch v := arg.(type) { + case string: + sb.WriteString(v) + case error: + sb.WriteString(v.Error()) + default: + sb.WriteString(fmt.Sprintf("%v", v)) + } + } +} + +func (d *DLog) FatalPanic(args ...interface{}) { + d.log(FATAL, args) + d.logger.Flush() + panic("Not recovering from this fatal error...") +} + +func (d *DLog) Fatal(args ...interface{}) string { + return d.log(FATAL, args) +} + +func (d *DLog) Error(args ...interface{}) string { + return d.log(ERROR, args) +} + +func (d *DLog) Warn(args ...interface{}) string { + return d.log(WARN, args) +} + +func (d *DLog) Info(args ...interface{}) string { + return d.log(INFO, args) +} + +func (d *DLog) Verbose(args ...interface{}) string { + return d.log(VERBOSE, args) +} + +func (d *DLog) Debug(args ...interface{}) string { + return d.log(DEBUG, args) +} + +func (d *DLog) Trace(args ...interface{}) string { + return d.log(TRACE, args) +} + +func (d *DLog) Devel(args ...interface{}) string { + return d.log(DEVEL, args) +} + +func (d *DLog) Raw(message string) string { + if !config.Client.TermColorsEnable || !d.logger.SupportsColors() { + d.logger.Log(time.Now(), message) + return message + } + + d.logger.Log(time.Now(), brush.Colorfy(message)) + return message +} + +func (d *DLog) Mapreduce(table string, data map[string]interface{}) string { + args := make([]interface{}, len(data)+1) + args[0] = fmt.Sprintf("%s:%s", "MAPREDUCE", strings.ToUpper(table)) + + i := 1 + for k, v := range data { + args[i] = fmt.Sprintf("%s=%v", k, v) + i++ + } + + return d.log(INFO, args) +} + +func (d *DLog) Flush() { d.logger.Flush() } +func (d *DLog) Pause() { d.logger.Pause() } +func (d *DLog) Resume() { d.logger.Resume() } diff --git a/internal/io/dlog/level.go b/internal/io/dlog/level.go new file mode 100644 index 0000000..84550f0 --- /dev/null +++ b/internal/io/dlog/level.go @@ -0,0 +1,89 @@ +package dlog + +import ( + "fmt" + "strings" +) + +type level int + +const ( + FATAL level = iota + ERROR level = iota + WARN level = iota + INFO level = iota + DEFAULT level = iota + VERBOSE level = iota + DEBUG level = iota + DEVEL level = iota + TRACE level = iota + ALL level = iota +) + +var allLevels = []level{ + FATAL, + ERROR, + WARN, + INFO, + DEFAULT, + VERBOSE, + DEBUG, + DEVEL, + TRACE, + ALL, +} + +func newLevel(l string) level { + switch strings.ToUpper(l) { + case "FATAL": + return FATAL + case "ERROR": + return ERROR + case "WARN": + return WARN + case "INFO": + return INFO + case "": + fallthrough + case "DEFAULT": + return DEFAULT + case "VERBOSE": + return VERBOSE + case "DEBUG": + return DEBUG + case "DEVEL": + return DEVEL + case "TRACE": + return TRACE + case "ALL": + return ALL + } + panic(fmt.Sprintf("Unknown log level %s, must be one of: %v", l, allLevels)) +} + +func (l level) String() string { + switch l { + case FATAL: + return "FATAL" + case ERROR: + return "ERROR" + case WARN: + return "WARN" + case INFO: + return "INFO" + case DEFAULT: + return "DEFAULT" + case VERBOSE: + return "VERBOSE" + case DEBUG: + return "DEBUG" + case DEVEL: + return "DEVEL" + case TRACE: + return "TRACE" + case ALL: + return "ALL" + } + + panic("Unknown log level " + fmt.Sprintf("%d", l)) +} diff --git a/internal/io/dlog/loggers/factory.go b/internal/io/dlog/loggers/factory.go new file mode 100644 index 0000000..3eb29c5 --- /dev/null +++ b/internal/io/dlog/loggers/factory.go @@ -0,0 +1,60 @@ +package loggers + +import ( + "fmt" + "sync" +) + +type Impl int + +const ( + NONE Impl = iota + STDOUT Impl = iota + FILE Impl = iota + FOUT Impl = iota +) + +var factoryMap map[string]Logger +var factoryMutex sync.Mutex + +func Factory(name string, impl Impl) Logger { + factoryMutex.Lock() + defer factoryMutex.Unlock() + + id := fmt.Sprintf("name:%s,impl:%v", name, impl) + + if factoryMap == nil { + factoryMap = make(map[string]Logger) + } + + singleton, ok := factoryMap[id] + if !ok { + switch impl { + case NONE: + singleton = none{} + case STDOUT: + singleton = newStdout() + factoryMap[id] = singleton + case FILE: + singleton = newFile() + factoryMap[id] = singleton + case FOUT: + singleton = newFout() + factoryMap[id] = singleton + } + } + + return singleton +} + +func FactoryRotate() { + factoryMutex.Lock() + defer factoryMutex.Unlock() + if factoryMap == nil { + return + } + + for _, impl := range factoryMap { + impl.Rotate() + } +} diff --git a/internal/io/dlog/loggers/file.go b/internal/io/dlog/loggers/file.go new file mode 100644 index 0000000..1c525c9 --- /dev/null +++ b/internal/io/dlog/loggers/file.go @@ -0,0 +1,156 @@ +package loggers + +import ( + "bufio" + "context" + "fmt" + "os" + "runtime" + "sync" + "time" + + "github.com/mimecast/dtail/internal/config" +) + +type fileMessageBuf struct { + now time.Time + message string +} + +type file struct { + bufferCh chan *fileMessageBuf + pauseCh chan struct{} + resumeCh chan struct{} + rotateCh chan struct{} + flushCh chan struct{} + lastDateStr string + fd *os.File + writer *bufio.Writer + mutex sync.Mutex + started bool +} + +func newFile() *file { + f := file{ + bufferCh: make(chan *fileMessageBuf, runtime.NumCPU()*100), + pauseCh: make(chan struct{}), + resumeCh: make(chan struct{}), + rotateCh: make(chan struct{}), + flushCh: make(chan struct{}), + } + f.getWriter(time.Now().Format("20060102")) + return &f +} + +func (s *file) Start(ctx context.Context, wg *sync.WaitGroup) { + s.mutex.Lock() + defer s.mutex.Unlock() + + // Logger already started from another Goroutine. + if s.started { + wg.Done() + return + } + + pause := func(ctx context.Context) { + select { + case <-s.resumeCh: + return + case <-ctx.Done(): + return + } + } + + go func() { + defer wg.Done() + + for { + select { + case m := <-s.bufferCh: + s.write(m) + case <-s.pauseCh: + pause(ctx) + case <-s.flushCh: + s.flush() + case <-ctx.Done(): + s.flush() + s.fd.Close() + return + } + } + }() + + s.started = true +} + +func (s *file) Log(now time.Time, message string) { + s.bufferCh <- &fileMessageBuf{now, message} +} + +func (s *file) LogWithColors(now time.Time, message, coloredMessage string) { + panic("Colors not supported in file logger") +} + +func (s *file) Pause() { s.pauseCh <- struct{}{} } +func (s *file) Resume() { s.resumeCh <- struct{}{} } +func (s *file) Flush() { s.flushCh <- struct{}{} } + +// TODO: Test that Rotate() actually works. +func (s *file) Rotate() { s.rotateCh <- struct{}{} } +func (file) SupportsColors() bool { return false } + +func (s *file) write(m *fileMessageBuf) { + select { + case <-s.rotateCh: + // Force re-opening the outfile. + s.lastDateStr = "" + default: + } + + writer := s.getWriter(m.now.Format("20060102")) + writer.WriteString(m.message) + writer.WriteByte('\n') +} + +func (s *file) getWriter(dateStr string) *bufio.Writer { + if s.lastDateStr == dateStr { + return s.writer + } + + if _, err := os.Stat(config.Common.LogDir); os.IsNotExist(err) { + if err = os.MkdirAll(config.Common.LogDir, 0755); err != nil { + panic(err) + } + } + + logFile := fmt.Sprintf("%s/%s.log", config.Common.LogDir, dateStr) + newFd, err := os.OpenFile(logFile, os.O_CREATE|os.O_RDWR|os.O_APPEND, 0644) + if err != nil { + panic(err) + } + + // Close old writer. + if s.fd != nil { + s.writer.Flush() + s.fd.Close() + } + + s.fd = newFd + s.writer = bufio.NewWriterSize(s.fd, 1) + s.lastDateStr = dateStr + + return s.writer +} + +func (s *file) flush() { + defer s.writer.Flush() + + for { + select { + case m := <-s.bufferCh: + s.write(m) + default: + return + } + } +} diff --git a/internal/io/dlog/loggers/fout.go b/internal/io/dlog/loggers/fout.go new file mode 100644 index 0000000..603dbe9 --- /dev/null +++ b/internal/io/dlog/loggers/fout.go @@ -0,0 +1,46 @@ +package loggers + +import ( + "context" + "sync" + "time" +) + +type fout struct { + file *file + stdout *stdout +} + +// Logs to both, a file and stdout +func newFout() *fout { + return &fout{file: newFile(), stdout: newStdout()} +} + +func (f *fout) Start(ctx context.Context, wg *sync.WaitGroup) { + go func() { + defer wg.Done() + + var wg2 sync.WaitGroup + wg2.Add(2) + f.file.Start(ctx, &wg2) + f.stdout.Start(ctx, &wg2) + wg2.Wait() + }() +} + +func (f *fout) Log(now time.Time, message string) { + f.stdout.Log(now, message) + f.file.Log(now, message) +} + +func (f *fout) LogWithColors(now time.Time, message, coloredMessage string) { + f.stdout.LogWithColors(now, "", coloredMessage) + f.file.Log(now, message) +} + +func (f *fout) Flush() { f.stdout.Flush(); f.file.Flush() } +func (f *fout) Pause() { f.stdout.Pause(); f.file.Pause() } +func (f *fout) Resume() { f.stdout.Resume(); f.file.Resume() } +func (f *fout) Rotate() { f.file.Rotate() } + +func (fout) SupportsColors() bool { return true } diff --git a/internal/io/dlog/loggers/logger.go b/internal/io/dlog/loggers/logger.go new file mode 100644 index 0000000..c88900d --- /dev/null +++ b/internal/io/dlog/loggers/logger.go @@ -0,0 +1,18 @@ +package loggers + +import ( + "context" + "sync" + "time" +) + +type Logger interface { + Log(now time.Time, message string) + LogWithColors(now time.Time, message, messageWithColors string) + Start(ctx context.Context, wg *sync.WaitGroup) + Flush() + Pause() + Resume() + Rotate() + SupportsColors() bool +} diff --git a/internal/io/dlog/loggers/none.go b/internal/io/dlog/loggers/none.go new file mode 100644 index 0000000..270027f --- /dev/null +++ b/internal/io/dlog/loggers/none.go @@ -0,0 +1,21 @@ +package loggers + +import ( + "context" + "sync" + "time" +) + +// don't log anything +type none struct{} + +func (none) Start(ctx context.Context, wg *sync.WaitGroup) { wg.Done() } +func (none) Log(now time.Time, message string) {} + +func (none) LogWithColors(now time.Time, message, coloredMessage string) {} + +func (none) Flush() {} +func (none) Pause() {} +func (none) Resume() {} +func (none) Rotate() {} +func (none) SupportsColors() bool { return false } diff --git a/internal/io/dlog/loggers/stdout.go b/internal/io/dlog/loggers/stdout.go new file mode 100644 index 0000000..9738323 --- /dev/null +++ b/internal/io/dlog/loggers/stdout.go @@ -0,0 +1,73 @@ +package loggers + +import ( + "context" + "fmt" + "sync" + "time" +) + +type stdout struct { + bufferCh chan string + pauseCh chan struct{} + resumeCh chan struct{} +} + +func newStdout() *stdout { + return &stdout{ + bufferCh: make(chan string, 100), + pauseCh: make(chan struct{}), + resumeCh: make(chan struct{}), + } +} + +func (s *stdout) Start(ctx context.Context, wg *sync.WaitGroup) { + pause := func(ctx context.Context) { + select { + case <-s.resumeCh: + return + case <-ctx.Done(): + return + } + } + + go func() { + defer wg.Done() + + for { + select { + case message := <-s.bufferCh: + fmt.Println(message) + case <-s.pauseCh: + pause(ctx) + case <-ctx.Done(): + s.Flush() + return + } + } + }() +} + +func (s *stdout) Log(now time.Time, message string) { + s.bufferCh <- message +} + +func (s *stdout) LogWithColors(now time.Time, message, coloredMessage string) { + s.bufferCh <- coloredMessage +} + +func (s *stdout) Flush() { + for { + select { + case message := <-s.bufferCh: + fmt.Println(message) + default: + return + } + } +} + +func (s *stdout) Pause() { s.pauseCh <- struct{}{} } +func (s *stdout) Resume() { s.resumeCh <- struct{}{} } +func (s *stdout) Rotate() {} +func (stdout) SupportsColors() bool { return true } diff --git a/internal/io/dlog/rotation.go b/internal/io/dlog/rotation.go new file mode 100644 index 0000000..15ce1fd --- /dev/null +++ b/internal/io/dlog/rotation.go @@ -0,0 +1,27 @@ +package dlog + +import ( + "context" + "os" + "os/signal" + "syscall" + + "github.com/mimecast/dtail/internal/io/dlog/loggers" +) + +func rotation(ctx context.Context) { + rotateCh := make(chan os.Signal, 1) + signal.Notify(rotateCh, syscall.SIGHUP) + go func() { + for { + select { + case <-rotateCh: + Common.Debug("Invoking log rotation") + loggers.FactoryRotate() + return + case <-ctx.Done(): + return + } + } + }() +} diff --git a/internal/io/dlog/source.go b/internal/io/dlog/source.go new file mode 100644 index 0000000..265885e --- /dev/null +++ b/internal/io/dlog/source.go @@ -0,0 +1,19 @@ +package dlog + +type source int + +const ( + CLIENT source = iota + SERVER source = iota +) + +func (s source) String() string { + switch s { + case CLIENT: + return "CLIENT" + case SERVER: + return "SERVER" + } + + panic("Unknown log source type") +} diff --git a/internal/io/dlog/strategy.go b/internal/io/dlog/strategy.go new file mode 100644 index 0000000..32d8298 --- /dev/null +++ b/internal/io/dlog/strategy.go @@ -0,0 +1,22 @@ +package dlog + +import "github.com/mimecast/dtail/internal/config" + +// Strategy allows to specify a log rotation strategy. +type Strategy int + +// Possible log strategies. +const ( + NormalStrategy Strategy = iota + DailyStrategy Strategy = iota + StdoutStrategy Strategy = iota +) + +func logStrategy() Strategy { + switch config.Common.LogStrategy { + case "daily": + return DailyStrategy + default: + } + return StdoutStrategy +} diff --git a/internal/io/fs/permissions/permission.go b/internal/io/fs/permissions/permission.go index cc5dd9b..bbcb74e 100644 --- a/internal/io/fs/permissions/permission.go +++ b/internal/io/fs/permissions/permission.go @@ -3,12 +3,12 @@ package permissions import ( - "github.com/mimecast/dtail/internal/io/logger" + "github.com/mimecast/dtail/internal/io/dlog" ) // ToRead is to check whether user has read permissions to a given file. func ToRead(user, filePath string) (bool, error) { // Only implemented for Linux, always expect true - logger.Warn(user, filePath, "Not performing ACL check, not supported on this platform") + dlog.Common.Warn(user, filePath, "Not performing ACL check, not supported on this platform") return true, nil } diff --git a/internal/io/fs/readfile.go b/internal/io/fs/readfile.go index ec33c60..07486a1 100644 --- a/internal/io/fs/readfile.go +++ b/internal/io/fs/readfile.go @@ -14,7 +14,7 @@ import ( "time" "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/regex" @@ -62,7 +62,7 @@ func (f readFile) Retry() bool { // Start tailing a log file. func (f readFile) Start(ctx context.Context, lines chan<- line.Line, re regex.Regex) error { - logger.Debug("readFile", f) + dlog.Common.Debug("readFile", f) defer func() { select { case <-f.limiter: @@ -74,7 +74,7 @@ func (f readFile) Start(ctx context.Context, lines chan<- line.Line, re regex.Re case f.limiter <- struct{}{}: default: select { - case f.serverMessages <- logger.Warn(f.filePath, f.globID, "Server limit reached. Queuing file..."): + case f.serverMessages <- dlog.Common.Warn(f.filePath, f.globID, "Server limit reached. Queuing file..."): case <-ctx.Done(): return nil } @@ -126,7 +126,7 @@ func (f readFile) makeReader(fd *os.File) (reader *bufio.Reader, err error) { case strings.HasSuffix(f.FilePath(), ".gz"): fallthrough case strings.HasSuffix(f.FilePath(), ".gzip"): - logger.Info(f.FilePath(), "Detected gzip compression format") + dlog.Common.Info(f.FilePath(), "Detected gzip compression format") var gzipReader *gzip.Reader gzipReader, err = gzip.NewReader(fd) if err != nil { @@ -134,7 +134,7 @@ func (f readFile) makeReader(fd *os.File) (reader *bufio.Reader, err error) { } reader = bufio.NewReader(gzipReader) case strings.HasSuffix(f.FilePath(), ".zst"): - logger.Info(f.FilePath(), "Detected zstd compression format") + dlog.Common.Info(f.FilePath(), "Detected zstd compression format") reader = bufio.NewReader(zstd.NewReader(fd)) default: reader = bufio.NewReader(fd) @@ -172,7 +172,7 @@ func (f readFile) read(ctx context.Context, fd *os.File, rawLines chan *bytes.Bu default: } if !f.seekEOF { - logger.Info(f.FilePath(), "End of file reached") + dlog.Common.Info(f.FilePath(), "End of file reached") return nil } time.Sleep(time.Millisecond * 100) @@ -201,7 +201,7 @@ func (f readFile) read(ctx context.Context, fd *os.File, rawLines chan *bytes.Bu default: if message.Len() >= lineLengthThreshold { if !warnedAboutLongLine { - f.serverMessages <- logger.Warn(f.filePath, "Long log line, splitting into multiple lines") + f.serverMessages <- dlog.Common.Warn(f.filePath, "Long log line, splitting into multiple lines") warnedAboutLongLine = true } message.WriteString("\n") @@ -268,7 +268,7 @@ func (f readFile) transmittable(lineBytes *bytes.Buffer, length, capacity int, r // Check wether log file is truncated. Returns nil if not. func (f readFile) truncated(fd *os.File) (bool, error) { - logger.De