From 97747ea0f3178f7f5890512d483fdccaa82846b0 Mon Sep 17 00:00:00 2001 From: Paul Buetow Date: Sat, 9 Oct 2021 21:10:29 +0300 Subject: vetting and linting and some code restyling --- Makefile | 4 +- TODO.md | 4 +- cmd/dcat/main.go | 3 +- cmd/dgrep/main.go | 4 +- cmd/dmap/main.go | 5 +- cmd/dserver/main.go | 3 +- cmd/dtail/main.go | 6 +- cmd/dtailhealthcheck/main.go | 1 - integrationtests/commons.go | 15 ++- integrationtests/dcat_test.go | 4 +- integrationtests/dgrep_test.go | 8 +- integrationtests/dmap_test.go | 50 ++++++++- integrationtests/dtail_test.go | 4 +- integrationtests/dtailhealthcheck_test.go | 10 +- internal/clients/baseclient.go | 7 +- internal/clients/catclient.go | 2 - internal/clients/connectors/serverconnection.go | 48 +++++---- internal/clients/connectors/serverless.go | 16 +-- internal/clients/grepclient.go | 5 +- internal/clients/handlers/healthhandler.go | 3 +- internal/clients/handlers/maprhandler.go | 13 ++- internal/clients/healthclient.go | 17 +-- internal/clients/maprclient.go | 13 +-- internal/clients/stats.go | 11 +- internal/clients/tailclient.go | 3 - internal/color/brush/brush.go | 8 +- internal/color/color.go | 11 +- internal/color/paint.go | 19 ++-- internal/color/table.go | 19 +++- internal/config/args.go | 6 +- internal/config/client.go | 12 +-- internal/config/config.go | 6 +- internal/config/initializer.go | 59 +++++----- internal/config/server.go | 7 +- internal/discovery/discovery.go | 13 +-- internal/done.go | 1 - internal/io/dlog/dlog.go | 22 +++- internal/io/dlog/level.go | 5 +- internal/io/dlog/loggers/factory.go | 6 +- internal/io/dlog/loggers/file.go | 17 ++- internal/io/dlog/loggers/logger.go | 1 + internal/io/dlog/loggers/strategy.go | 11 +- internal/io/fs/catfile.go | 4 +- internal/io/fs/filereader.go | 3 +- internal/io/fs/permissions/permission_linuxacl.go | 2 +- internal/io/fs/readfile.go | 29 ++--- internal/io/fs/tailfile.go | 4 +- internal/io/pool/builder.go | 3 + internal/io/pool/bytesbuffer.go | 3 + internal/io/prompt/prompt.go | 7 +- internal/io/signal/signal.go | 3 - internal/mapr/aggregateset.go | 4 - internal/mapr/client/aggregate.go | 9 +- internal/mapr/funcs/function.go | 7 +- internal/mapr/funcs/function_test.go | 21 ++-- internal/mapr/funcs/maskdigits.go | 2 - internal/mapr/globalgroupset.go | 8 -- internal/mapr/groupset.go | 9 +- internal/mapr/logformat/default.go | 5 +- internal/mapr/logformat/default_test.go | 24 +++-- internal/mapr/logformat/generickv.go | 2 +- internal/mapr/logformat/parser.go | 12 +-- internal/mapr/query.go | 17 ++- internal/mapr/query_test.go | 125 ++++++++++++++-------- internal/mapr/selectcondition.go | 12 +-- internal/mapr/server/aggregate.go | 30 +++--- internal/mapr/setclause.go | 2 - internal/mapr/setcondition.go | 15 ++- internal/mapr/token.go | 11 +- internal/mapr/whereclause.go | 10 +- internal/mapr/wherecondition.go | 24 ++--- internal/protocol/protocol.go | 1 - internal/regex/regex.go | 9 +- internal/regex/regex_test.go | 17 +-- internal/server/continuous.go | 9 +- internal/server/handlers/basehandler.go | 30 ++---- internal/server/handlers/healthhandler.go | 11 +- internal/server/handlers/mapcommand.go | 7 +- internal/server/handlers/readcommand.go | 41 ++++--- internal/server/handlers/serverhandler.go | 20 ++-- internal/server/scheduler.go | 9 +- internal/server/server.go | 47 ++++---- internal/server/stats.go | 11 +- internal/source/source.go | 14 ++- internal/ssh/client/authmethods.go | 41 ++++--- internal/ssh/client/customkeycallback.go | 3 +- internal/ssh/client/knownhostscallback.go | 19 +--- internal/ssh/server/hostkey.go | 3 +- internal/ssh/server/publickeycallback.go | 27 +++-- internal/ssh/ssh.go | 4 - internal/user/name.go | 3 - internal/user/server/user.go | 30 +++--- internal/version/version.go | 6 +- 93 files changed, 670 insertions(+), 581 deletions(-) diff --git a/Makefile b/Makefile index 6a0f828..543b29e 100644 --- a/Makefile +++ b/Makefile @@ -1,5 +1,5 @@ GO ?= go -all: build test +all: build build: dserver dcat dgrep dmap dtail dtailhealthcheck dserver: ifndef USE_ACL @@ -43,7 +43,7 @@ lint: find . -type d | while read dir; do \ echo golint $$dir; \ golint $$dir; \ - done + done | grep -F .go: test: ${GO} clean -testcache ifndef USE_ACL diff --git a/TODO.md b/TODO.md index 45ed4e3..72eb0f3 100644 --- a/TODO.md +++ b/TODO.md @@ -15,5 +15,5 @@ This is a loose list of what to do. Maybe for the next releae or maybe for a lat [ ] Manual test/adjust dtail colors [ ] More integration test colors (via dcat?) [ ] Integration test for dtail in serverless mode -[ ] Integration test for health check serverless mode -[ ] Rewrite + test health client (copy catclient) +[ ] Go through the whole source and change indentation (try not to exceed 80char line lengths by too much) +[ ] Fix the sync.Pools (they aren't concurrent as it seems and can cause a panic) diff --git a/cmd/dcat/main.go b/cmd/dcat/main.go index 662a50d..87ece9d 100644 --- a/cmd/dcat/main.go +++ b/cmd/dcat/main.go @@ -33,7 +33,8 @@ func main() { flag.BoolVar(&args.Spartan, "spartan", false, "Spartan output mode") flag.BoolVar(&args.TrustAllHosts, "trustAllHosts", false, "Trust all unknown host keys") flag.BoolVar(&displayVersion, "version", false, "Display version") - flag.IntVar(&args.ConnectionsPerCPU, "cpc", config.DefaultConnectionsPerCPU, "How many connections established per CPU core concurrently") + flag.IntVar(&args.ConnectionsPerCPU, "cpc", config.DefaultConnectionsPerCPU, + "How many connections established per CPU core concurrently") flag.IntVar(&args.SSHPort, "port", config.DefaultSSHPort, "SSH server port") flag.IntVar(&pprof, "pprof", -1, "Start PProf server this port") flag.StringVar(&args.ConfigFile, "cfg", "", "Config file path") diff --git a/cmd/dgrep/main.go b/cmd/dgrep/main.go index 529331d..576e22b 100644 --- a/cmd/dgrep/main.go +++ b/cmd/dgrep/main.go @@ -20,7 +20,6 @@ func main() { var args config.Args var displayVersion bool var grep string - userName := user.Name() flag.BoolVar(&args.NoColor, "noColor", false, "Disable ANSII terminal colors") @@ -29,7 +28,8 @@ func main() { flag.BoolVar(&args.Spartan, "spartan", false, "Spartan output mode") flag.BoolVar(&args.TrustAllHosts, "trustAllHosts", false, "Trust all unknown host keys") flag.BoolVar(&displayVersion, "version", false, "Display version") - flag.IntVar(&args.ConnectionsPerCPU, "cpc", config.DefaultConnectionsPerCPU, "How many connections established per CPU core concurrently") + flag.IntVar(&args.ConnectionsPerCPU, "cpc", config.DefaultConnectionsPerCPU, + "How many connections established per CPU core concurrently") flag.IntVar(&args.SSHPort, "port", config.DefaultSSHPort, "SSH server port") flag.StringVar(&args.ConfigFile, "cfg", "", "Config file path") flag.StringVar(&args.Discovery, "discovery", "", "Server discovery method") diff --git a/cmd/dmap/main.go b/cmd/dmap/main.go index acc1dc6..1f44076 100644 --- a/cmd/dmap/main.go +++ b/cmd/dmap/main.go @@ -19,11 +19,9 @@ import ( // The evil begins here. func main() { var displayVersion bool - args := config.Args{ Mode: omode.MapClient, } - userName := user.Name() flag.BoolVar(&args.NoColor, "noColor", false, "Disable ANSII terminal colors") @@ -31,7 +29,8 @@ func main() { flag.BoolVar(&args.Spartan, "spartan", false, "Spartan output mode") flag.BoolVar(&args.TrustAllHosts, "trustAllHosts", false, "Trust all unknown host keys") flag.BoolVar(&displayVersion, "version", false, "Display version") - flag.IntVar(&args.ConnectionsPerCPU, "cpc", config.DefaultConnectionsPerCPU, "How many connections established per CPU core concurrently") + flag.IntVar(&args.ConnectionsPerCPU, "cpc", config.DefaultConnectionsPerCPU, + "How many connections established per CPU core concurrently") flag.IntVar(&args.SSHPort, "port", config.DefaultSSHPort, "SSH server port") flag.IntVar(&args.Timeout, "timeout", 0, "Max time dtail server will collect data until disconnection") flag.StringVar(&args.ConfigFile, "cfg", "", "Config file path") diff --git a/cmd/dserver/main.go b/cmd/dserver/main.go index b4fc873..cf726cf 100644 --- a/cmd/dserver/main.go +++ b/cmd/dserver/main.go @@ -32,7 +32,8 @@ func main() { user.NoRootCheck() flag.BoolVar(&color, "color", false, "Enable ANSII terminal colors") - flag.BoolVar(&config.ServerRelaxedAuthEnable, "relaxedAuth", false, "Enable relaxced SSH auth mode (don't use in production!)") + flag.BoolVar(&config.ServerRelaxedAuthEnable, "relaxedAuth", false, + "Enable relaxced SSH auth mode (don't use in production!)") flag.BoolVar(&displayVersion, "version", false, "Display version") flag.IntVar(&args.SSHPort, "port", config.DefaultSSHPort, "SSH server port") flag.IntVar(&pprof, "pprof", -1, "Start PProf server this port") diff --git a/cmd/dtail/main.go b/cmd/dtail/main.go index adfeaa5..2863370 100644 --- a/cmd/dtail/main.go +++ b/cmd/dtail/main.go @@ -44,7 +44,8 @@ func main() { flag.BoolVar(&displayColorTable, "colorTable", false, "Show color table") flag.BoolVar(&displayWideColorTable, "wideColorTable", false, "Show a large color table") flag.BoolVar(&displayVersion, "version", false, "Display version") - flag.IntVar(&args.ConnectionsPerCPU, "cpc", config.DefaultConnectionsPerCPU, "How many connections established per CPU core concurrently") + flag.IntVar(&args.ConnectionsPerCPU, "cpc", config.DefaultConnectionsPerCPU, + "How many connections established per CPU core concurrently") flag.IntVar(&args.SSHPort, "port", config.DefaultSSHPort, "SSH server port") flag.IntVar(&args.Timeout, "timeout", 0, "Max time dtail server will collect data until disconnection") flag.IntVar(&pprof, "pprof", -1, "Start PProf server this port") @@ -93,7 +94,8 @@ func main() { dlog.Start(ctx, &wg, source.Client) if checkHealth { - fmt.Println("WARN: DTail health check has moved to separate binary dtailhealtcheck - please adjust the monitoring scripts!") + fmt.Println("WARN: DTail health check has moved to separate binary dtailhealtcheck" + + " - please adjust the monitoring scripts!") cancel() os.Exit(1) } diff --git a/cmd/dtailhealthcheck/main.go b/cmd/dtailhealthcheck/main.go index 0f37f8a..b0ba4cd 100644 --- a/cmd/dtailhealthcheck/main.go +++ b/cmd/dtailhealthcheck/main.go @@ -35,7 +35,6 @@ func main() { defer cancel() var wg sync.WaitGroup wg.Add(1) - dlog.Start(ctx, &wg, source.HealthCheck) if pprof > -1 { diff --git a/integrationtests/commons.go b/integrationtests/commons.go index f96b532..2fdbfc3 100644 --- a/integrationtests/commons.go +++ b/integrationtests/commons.go @@ -15,10 +15,12 @@ import ( ) func runCommand(t *testing.T, cmd string, args []string, stdoutFile string) (int, error) { - return runCommandContext(t, context.TODO(), cmd, args, stdoutFile) + return runCommandContext(context.TODO(), t, cmd, args, stdoutFile) } -func runCommandContext(t *testing.T, ctx context.Context, cmd string, args []string, stdoutFile string) (int, error) { +func runCommandContext(ctx context.Context, t *testing.T, cmd string, args []string, + stdoutFile string) (int, error) { + if _, err := os.Stat(cmd); err != nil { return -1, fmt.Errorf("No such binary %s, please compile first (%v)", cmd, err) } @@ -76,7 +78,8 @@ func compareFilesContents(t *testing.T, fileA, fileB string) error { return fmt.Errorf("Files differ, line '%s' is missing in one of them", line) } if countA != countB { - return fmt.Errorf("Files differ, count of line '%s' is %d in one but %d in another", line, countA, countB) + return fmt.Errorf("Files differ, count of line '%s' is %d in one but %d in another", + line, countA, countB) } } return nil @@ -92,11 +95,13 @@ func compareFilesContents(t *testing.T, fileA, fileB string) error { } // The mapreduce result can be in a different order each time (Golang maps are not sorted). - t.Log(fmt.Sprintf("Checking whether %s has same lines as file %s (ignoring line order)", fileA, fileB)) + t.Log(fmt.Sprintf("Checking whether %s has same lines as file %s (ignoring line order)", + fileA, fileB)) if err := compareMaps(a, b); err != nil { return err } - t.Log(fmt.Sprintf("Checking whether %s has same lines as file %s (ignoring line order)", fileB, fileA)) + t.Log(fmt.Sprintf("Checking whether %s has same lines as file %s (ignoring line order)", + fileB, fileA)) if err := compareMaps(b, a); err != nil { return err } diff --git a/integrationtests/dcat_test.go b/integrationtests/dcat_test.go index a164960..342ebd0 100644 --- a/integrationtests/dcat_test.go +++ b/integrationtests/dcat_test.go @@ -8,12 +8,12 @@ import ( func TestDCat(t *testing.T) { testdataFile := "dcat.txt.expected" stdoutFile := "dcat.out" + args := []string{"-spartan", testdataFile} - if _, err := runCommand(t, "../dcat", []string{"-spartan", testdataFile}, stdoutFile); err != nil { + if _, err := runCommand(t, "../dcat", args, stdoutFile); err != nil { t.Error(err) return } - if err := compareFiles(t, stdoutFile, testdataFile); err != nil { t.Error(err) return diff --git a/integrationtests/dgrep_test.go b/integrationtests/dgrep_test.go index 6a15ebd..4d54a2d 100644 --- a/integrationtests/dgrep_test.go +++ b/integrationtests/dgrep_test.go @@ -9,12 +9,12 @@ func TestDGrep(t *testing.T) { inFile := "mapr_testdata.log" stdoutFile := "dgrep.stdout.tmp" expectedStdoutFile := "dgrep.txt.expected" + args := []string{"-spartan", "--grep", "20211002-071947", inFile} - if _, err := runCommand(t, "../dgrep", []string{"-spartan", "--grep", "20211002-071947", inFile}, stdoutFile); err != nil { + if _, err := runCommand(t, "../dgrep", args, stdoutFile); err != nil { t.Error(err) return } - if err := compareFiles(t, stdoutFile, expectedStdoutFile); err != nil { t.Error(err) return @@ -26,12 +26,12 @@ func TestDGrep2(t *testing.T) { inFile := "mapr_testdata.log" stdoutFile := "dgrep2.stdout.tmp" expectedStdoutFile := "dgrep2.txt.expected" + args := []string{"-spartan", "--grep", "20211002-071947", "--invert", inFile} - if _, err := runCommand(t, "../dgrep", []string{"-spartan", "--grep", "20211002-071947", "--invert", inFile}, stdoutFile); err != nil { + if _, err := runCommand(t, "../dgrep", args, stdoutFile); err != nil { t.Error(err) return } - if err := compareFiles(t, stdoutFile, expectedStdoutFile); err != nil { t.Error(err) return diff --git a/integrationtests/dmap_test.go b/integrationtests/dmap_test.go index b512985..f5c78e0 100644 --- a/integrationtests/dmap_test.go +++ b/integrationtests/dmap_test.go @@ -14,9 +14,12 @@ func TestDMap(t *testing.T) { queryFile := fmt.Sprintf("%s.query", csvFile) expectedQueryFile := "dmap.csv.query.expected" - query := fmt.Sprintf("from STATS select count($line),last($time),avg($goroutines),min(concurrentConnections),max(lifetimeConnections) group by $hostname outfile %s", csvFile) + query := fmt.Sprintf("from STATS select count($line),last($time),"+ + "avg($goroutines),min(concurrentConnections),max(lifetimeConnections) "+ + "group by $hostname outfile %s", csvFile) + args := []string{"-query", query, inFile} - if _, err := runCommand(t, "../dmap", []string{"-query", query, inFile}, stdoutFile); err != nil { + if _, err := runCommand(t, "../dmap", args, stdoutFile); err != nil { t.Error(err) return } @@ -42,9 +45,48 @@ func TestDMap2(t *testing.T) { queryFile := fmt.Sprintf("%s.query", csvFile) expectedQueryFile := "dmap2.csv.query.expected" - query := fmt.Sprintf("from STATS select count($time),$time,max($goroutines),avg($goroutines),min($goroutines) group by $time order by count($time) outfile %s", csvFile) + query := fmt.Sprintf("from STATS select count($time),$time,max($goroutines),"+ + "avg($goroutines),min($goroutines) group by $time order by count($time) "+ + "outfile %s", csvFile) - if _, err := runCommand(t, "../dmap", []string{"-query", query, inFile}, stdoutFile); err != nil { + args := []string{"-query", query, inFile} + if _, err := runCommand(t, "../dmap", args, stdoutFile); err != nil { + t.Error(err) + return + } + if err := compareFilesContents(t, csvFile, expectedCsvFile); err != nil { + t.Error(err) + return + } + if err := compareFiles(t, queryFile, expectedQueryFile); err != nil { + t.Error(err) + return + } + + os.Remove(stdoutFile) + os.Remove(csvFile) + os.Remove(queryFile) +} + +func TestDMap3(t *testing.T) { + inFile := "mapr_testdata.log" + stdoutFile := "dmap3.stdout.tmp" + csvFile := "dmap3.csv.tmp" + expectedCsvFile := "dmap3.csv.expected" + queryFile := fmt.Sprintf("%s.query", csvFile) + expectedQueryFile := "dmap3.csv.query.expected" + + query := fmt.Sprintf("from STATS select count($time),$time,max($goroutines),"+ + "avg($goroutines),min($goroutines) group by $time order by count($time) "+ + "outfile %s", csvFile) + + // Read many input files at once. + args := []string{"-query", query} + for i := 0; i < 100; i++ { + args = append(args, inFile) + } + + if _, err := runCommand(t, "../dmap", args, stdoutFile); err != nil { t.Error(err) return } diff --git a/integrationtests/dtail_test.go b/integrationtests/dtail_test.go index 9971f1a..36eadc0 100644 --- a/integrationtests/dtail_test.go +++ b/integrationtests/dtail_test.go @@ -8,8 +8,9 @@ import ( func TestDTailColorTable(t *testing.T) { stdoutFile := "dtailcolortable.stdout.tmp" expectedStdoutFile := "dtailcolortable.expected" + args := []string{"-colorTable"} - if _, err := runCommand(t, "../dtail", []string{"-colorTable"}, stdoutFile); err != nil { + if _, err := runCommand(t, "../dtail", args, stdoutFile); err != nil { t.Error(err) return } @@ -17,6 +18,5 @@ func TestDTailColorTable(t *testing.T) { t.Error(err) return } - os.Remove(stdoutFile) } diff --git a/integrationtests/dtailhealthcheck_test.go b/integrationtests/dtailhealthcheck_test.go index 97fa5f2..d562239 100644 --- a/integrationtests/dtailhealthcheck_test.go +++ b/integrationtests/dtailhealthcheck_test.go @@ -30,9 +30,10 @@ func TestDTailHealthCheck(t *testing.T) { func TestDTailHealthCheck2(t *testing.T) { stdoutFile := "dtailhealthcheck2.stdout.tmp" expectedStdoutFile := "dtailhealthcheck2.expected" + args := []string{"--server", "example:1"} t.Log("Negative test, is supposed to exit with a critical state.") - exitCode, err := runCommand(t, "../dtailhealthcheck", []string{"--server", "example:1"}, stdoutFile) + exitCode, err := runCommand(t, "../dtailhealthcheck", args, stdoutFile) if exitCode != 2 { t.Error(fmt.Sprintf("Expected exit code '2' but got '%d': %v", exitCode, err)) return @@ -55,16 +56,17 @@ func TestDTailHealthCheck3(t *testing.T) { defer cancel() go func() { - serverArgs := []string{"--logger", "stdout", "--logLevel", "trace", "--port", "4242"} - runCommandContext(t, ctx, "../dserver", serverArgs, serverStdoutFile) + args := []string{"--logger", "stdout", "--logLevel", "trace", "--port", "4242"} + runCommandContext(ctx, t, "../dserver", args, serverStdoutFile) }() var err error + args := []string{"--server", "localhost:4242"} for i := 0; i < 30; i++ { t.Log("Waiting for dserver to start", i) time.Sleep(time.Second) var exitCode int - if exitCode, err = runCommand(t, "../dtailhealthcheck", []string{"--server", "localhost:4242"}, stdoutFile); exitCode == 0 { + if exitCode, err = runCommand(t, "../dtailhealthcheck", args, stdoutFile); exitCode == 0 { break } } diff --git a/internal/clients/baseclient.go b/internal/clients/baseclient.go index d5d7c2c..4a7bd84 100644 --- a/internal/clients/baseclient.go +++ b/internal/clients/baseclient.go @@ -64,7 +64,8 @@ func (c *baseClient) makeConnections(maker maker) { discoveryService := discovery.New(c.Discovery, c.ServersStr, discovery.Shuffle) for _, server := range discoveryService.ServerList() { - c.connections = append(c.connections, c.makeConnection(server, c.sshAuthMethods, c.hostKeyCallback)) + c.connections = append(c.connections, c.makeConnection(server, + c.sshAuthMethods, c.hostKeyCallback)) } c.stats = newTailStats(len(c.connections)) @@ -100,7 +101,9 @@ func (c *baseClient) Start(ctx context.Context, statsCh <-chan string) (status i return } -func (c *baseClient) startConnection(ctx context.Context, i int, conn connectors.Connector) (status int) { +func (c *baseClient) startConnection(ctx context.Context, i int, + conn connectors.Connector) (status int) { + for { connCtx, cancel := context.WithCancel(ctx) defer cancel() diff --git a/internal/clients/catclient.go b/internal/clients/catclient.go index 2726e7e..bd65560 100644 --- a/internal/clients/catclient.go +++ b/internal/clients/catclient.go @@ -22,7 +22,6 @@ func NewCatClient(args config.Args) (*CatClient, error) { if args.RegexStr != "" { return nil, errors.New("Can't use regex with 'cat' operating mode") } - args.Mode = omode.CatClient c := CatClient{ @@ -35,7 +34,6 @@ func NewCatClient(args config.Args) (*CatClient, error) { c.init() c.makeConnections(c) - return &c, nil } diff --git a/internal/clients/connectors/serverconnection.go b/internal/clients/connectors/serverconnection.go index 1666a79..2d7b45a 100644 --- a/internal/clients/connectors/serverconnection.go +++ b/internal/clients/connectors/serverconnection.go @@ -16,7 +16,8 @@ import ( "golang.org/x/crypto/ssh" ) -// ServerConnection represents a connection to a single remote dtail server via SSH protocol. +// ServerConnection represents a connection to a single remote dtail server via +// SSH protocol. type ServerConnection struct { server string port int @@ -28,9 +29,11 @@ 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 { - dlog.Client.Debug(server, "Creating new connection", server, handler, commands) +func NewServerConnection(server string, userName string, + authMethods []ssh.AuthMethod, hostKeyCallback client.HostKeyCallback, + handler handlers.Handler, commands []string) *ServerConnection { + dlog.Client.Debug(server, "Creating new connection", server, handler, commands) c := ServerConnection{ hostKeyCallback: hostKeyCallback, server: server, @@ -48,10 +51,12 @@ func NewServerConnection(server string, userName string, authMethods []ssh.AuthM return &c } +// Server returns the server hostname connected to. func (c *ServerConnection) Server() string { return c.server } +// Handler returns the handler used for the connection. func (c *ServerConnection) Handler() handlers.Handler { return c.handler } @@ -72,23 +77,29 @@ func (c *ServerConnection) initServerPort() { } } -func (c *ServerConnection) Start(ctx context.Context, cancel context.CancelFunc, throttleCh, statsCh chan struct{}) { +// Start the connection to the server. +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) dlog.Client.Debug(c.server, "Throttling connection", len(throttleCh), cap(throttleCh)) select { case throttleCh <- struct{}{}: case <-ctx.Done(): - dlog.Client.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 } - dlog.Client.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 { - dlog.Client.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 } @@ -107,7 +118,9 @@ 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 { +func (c *ServerConnection) dial(ctx context.Context, cancel context.CancelFunc, + throttleCh, statsCh chan struct{}) error { + dlog.Client.Debug(c.server, "Incrementing connection stats") statsCh <- struct{}{} defer func() { @@ -128,31 +141,30 @@ 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 { - dlog.Client.Debug(c.server, "Creating SSH session") +func (c *ServerConnection) session(ctx context.Context, cancel context.CancelFunc, + client *ssh.Client, throttleCh chan struct{}) error { + dlog.Client.Debug(c.server, "Creating SSH session") session, err := client.NewSession() if err != nil { return err } defer session.Close() - return c.handle(ctx, cancel, session, throttleCh) } -func (c *ServerConnection) handle(ctx context.Context, cancel context.CancelFunc, session *ssh.Session, throttleCh chan struct{}) error { - dlog.Client.Debug(c.server, "Creating handler for SSH session") +func (c *ServerConnection) handle(ctx context.Context, cancel context.CancelFunc, + session *ssh.Session, throttleCh chan struct{}) error { + dlog.Client.Debug(c.server, "Creating handler for SSH session") stdinPipe, err := session.StdinPipe() if err != nil { return err } - stdoutPipe, err := session.StdoutPipe() if err != nil { return err } - if err := session.Shell(); err != nil { return err } @@ -161,12 +173,10 @@ func (c *ServerConnection) handle(ctx context.Context, cancel context.CancelFunc io.Copy(stdinPipe, c.handler) cancel() }() - go func() { io.Copy(c.handler, stdoutPipe) cancel() }() - go func() { select { case <-c.handler.Done(): @@ -182,13 +192,13 @@ func (c *ServerConnection) handle(ctx context.Context, cancel context.CancelFunc } if !c.throttlingDone { - dlog.Client.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 } <-ctx.Done() c.handler.Shutdown() - return nil } diff --git a/internal/clients/connectors/serverless.go b/internal/clients/connectors/serverless.go index 768a5ce..2ff490a 100644 --- a/internal/clients/connectors/serverless.go +++ b/internal/clients/connectors/serverless.go @@ -18,8 +18,10 @@ type Serverless struct { userName string } -// NewServerConnection returns a new connection. -func NewServerless(userName string, handler handlers.Handler, commands []string) *Serverless { +// NewServerless starts a new serverless session. +func NewServerless(userName string, handler handlers.Handler, + commands []string) *Serverless { + dlog.Client.Debug("Creating new serverless connector", handler, commands) return &Serverless{ userName: userName, @@ -28,15 +30,20 @@ func NewServerless(userName string, handler handlers.Handler, commands []string) } } +// Server returns serverless server indicator. func (s *Serverless) Server() string { return "local(serverless)" } +// Handler returns the handler used for the serverless connection. func (s *Serverless) Handler() handlers.Handler { return s.handler } -func (s *Serverless) Start(ctx context.Context, cancel context.CancelFunc, throttleCh, statsCh chan struct{}) { +// Start the serverless connection. +func (s *Serverless) Start(ctx context.Context, cancel context.CancelFunc, + throttleCh, statsCh chan struct{}) { + dlog.Client.Debug("Starting serverless connector") go func() { defer cancel() @@ -81,13 +88,11 @@ func (s *Serverless) handle(ctx context.Context, cancel context.CancelFunc) erro dlog.Client.Trace("io.Copy(serverHandler, s.handler) => done") terminate() }() - go func() { io.Copy(s.handler, serverHandler) dlog.Client.Trace("io.Copy(s.handler, serverHandler) => done") terminate() }() - go func() { select { case <-s.handler.Done(): @@ -107,6 +112,5 @@ func (s *Serverless) handle(ctx context.Context, cancel context.CancelFunc) erro <-ctx.Done() dlog.Client.Trace("s.handler.Shutdown()") s.handler.Shutdown() - return nil } diff --git a/internal/clients/grepclient.go b/internal/clients/grepclient.go index ae21ff2..7521c67 100644 --- a/internal/clients/grepclient.go +++ b/internal/clients/grepclient.go @@ -12,7 +12,8 @@ import ( "github.com/mimecast/dtail/internal/omode" ) -// GrepClient searches a remote file for all lines matching a regular expression. Only the matching lines are displayed. +// GrepClient searches a remote file for all lines matching a regular +// expression. Only the matching lines are displayed. type GrepClient struct { baseClient } @@ -34,7 +35,6 @@ func NewGrepClient(args config.Args) (*GrepClient, error) { c.init() c.makeConnections(c) - return &c, nil } @@ -51,6 +51,5 @@ func (c GrepClient) makeCommands() (commands []string) { commands = append(commands, fmt.Sprintf("%s:%s %s %s", c.Mode.String(), c.Args.SerializeOptions(), file, regex)) } - return } diff --git a/internal/clients/handlers/healthhandler.go b/internal/clients/handlers/healthhandler.go index 10ba1f7..47b594e 100644 --- a/internal/clients/handlers/healthhandler.go +++ b/internal/clients/handlers/healthhandler.go @@ -8,7 +8,8 @@ import ( "github.com/mimecast/dtail/internal/protocol" ) -// HealthHandler is the handler used on the client side for running mapreduce aggregations. +// HealthHandler is the handler used on the client side for running mapreduce +// aggregations. type HealthHandler struct { baseHandler } diff --git a/internal/clients/handlers/maprhandler.go b/internal/clients/handlers/maprhandler.go index d1acfbd..8718b35 100644 --- a/internal/clients/handlers/maprhandler.go +++ b/internal/clients/handlers/maprhandler.go @@ -10,7 +10,8 @@ import ( "github.com/mimecast/dtail/internal/protocol" ) -// MaprHandler is the handler used on the client side for running mapreduce aggregations. +// MaprHandler is the handler used on the client side for running mapreduce +// aggregations. type MaprHandler struct { baseHandler aggregate *client.Aggregate @@ -18,7 +19,9 @@ type MaprHandler struct { } // NewMaprHandler returns a new mapreduce client handler. -func NewMaprHandler(server string, query *mapr.Query, globalGroup *mapr.GlobalGroupSet) *MaprHandler { +func NewMaprHandler(server string, query *mapr.Query, + globalGroup *mapr.GlobalGroupSet) *MaprHandler { + return &MaprHandler{ baseHandler: baseHandler{ server: server, @@ -55,12 +58,12 @@ func (h *MaprHandler) Write(p []byte) (n int, err error) { return len(p), nil } -// Handle a message received from server including mapr aggregation -// related data. +// Handle a message received from server including mapr aggregation related data. func (h *MaprHandler) handleAggregateMessage(message string) { parts := strings.SplitN(message, protocol.FieldDelimiter, 3) if len(parts) != 3 { - dlog.Client.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 { diff --git a/internal/clients/healthclient.go b/internal/clients/healthclient.go index ac1dc20..1a02827 100644 --- a/internal/clients/healthclient.go +++ b/internal/clients/healthclient.go @@ -32,7 +32,6 @@ func NewHealthClient(args config.Args) (*HealthClient, error) { c.init() c.sshAuthMethods = append(c.sshAuthMethods, gossh.Password(config.HealthUser)) c.makeConnections(c) - return &c, nil } @@ -45,28 +44,34 @@ func (c HealthClient) makeCommands() (commands []string) { return } +// Start the health client. func (c *HealthClient) Start(ctx context.Context, statsCh <-chan string) int { status := c.baseClient.Start(ctx, statsCh) switch status { case 0: if c.Serverless { - fmt.Printf("WARNING: All seems fine but the check only run in serverless mode, please specify a remote server via --server hostname:port\n") + fmt.Printf("WARNING: All seems fine but the check only run in serverless mode" + + ", please specify a remote server via --server hostname:port\n") return 1 } fmt.Printf("OK: All fine at %s :-)\n", c.ServersStr) case 2: if c.Serverless { - fmt.Printf("CRITICAL: DTail server not operating properly (using serverless connction)!\n") + fmt.Printf("CRITICAL: DTail server not operating properly (using " + + "serverless connction)!\n") return 2 } - fmt.Printf("CRITICAL: DTail server not operating properly at %s!\n", c.ServersStr) + fmt.Printf("CRITICAL: DTail server not operating properly at %s!\n", + c.ServersStr) default: if c.Serverless { - fmt.Printf("UNKNOWN: Received unknown status code %d (using serverless connection)\n", status) + fmt.Printf("UNKNOWN: Received unknown status code %d (using serverless "+ + "connection)\n", status) return status } - fmt.Printf("UNKNOWN: Received unknown status code %d from %s!\n", status, c.ServersStr) + fmt.Printf("UNKNOWN: Received unknown status code %d from %s!\n", + status, c.ServersStr) } return status diff --git a/internal/clients/maprclient.go b/internal/clients/maprclient.go index 412a219..04f258d 100644 --- a/internal/clients/maprclient.go +++ b/internal/clients/maprclient.go @@ -107,15 +107,14 @@ func (c *MaprClient) Start(ctx context.Context, statsCh <-chan string) (status i return } -// NEXT: Make this a callback function rather trying to use polymorphism to call this. -// This applies to all clients. +// NEXT: Make this a callback function rather trying to use polymorphism to call +// this. This applies to all clients. func (c MaprClient) makeHandler(server string) handlers.Handler { return handlers.NewMaprHandler(server, c.query, c.globalGroup) } func (c MaprClient) makeCommands() (commands []string) { commands = append(commands, fmt.Sprintf("map %s", c.query.RawQuery)) - modeStr := "cat" if c.Mode == omode.TailClient { modeStr = "tail" @@ -134,7 +133,6 @@ func (c MaprClient) makeCommands() (commands []string) { commands = append(commands, fmt.Sprintf("%s:%s %s %s", modeStr, c.Args.SerializeOptions(), file, regex)) } - return } @@ -155,7 +153,6 @@ func (c *MaprClient) reportResults() { c.writeResultsToOutfile() return } - c.printResults() } @@ -176,7 +173,6 @@ func (c *MaprClient) printResults() { } else { result, numRows, err = c.globalGroup.SwapOut().Result(c.query, rowsLimit) } - if err != nil { dlog.Client.FatalPanic(err) } @@ -202,8 +198,8 @@ func (c *MaprClient) printResults() { dlog.Client.Raw(rawQuery) if rowsLimit > 0 && numRows > rowsLimit { - dlog.Client.Warn(fmt.Sprintf("Got %d results but limited terminal output to %d rows! Use 'limit' clause to override!", - numRows, rowsLimit)) + dlog.Client.Warn(fmt.Sprintf("Got %d results but limited terminal output "+ + "to %d rows! Use 'limit' clause to override!", numRows, rowsLimit)) } dlog.Client.Raw(result) } @@ -215,7 +211,6 @@ func (c *MaprClient) writeResultsToOutfile() { } return } - if err := c.globalGroup.SwapOut().WriteResult(c.query); err != nil { dlog.Client.FatalPanic(err) } diff --git a/internal/clients/stats.go b/internal/clients/stats.go index fbef572..1315aea 100644 --- a/internal/clients/stats.go +++ b/internal/clients/stats.go @@ -36,9 +36,10 @@ func newTailStats(servers 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 <-chan struct{}, statsCh <-chan string, quiet bool) { - var connectedLast int +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 @@ -94,7 +95,9 @@ func (s *stats) printStatsDueInterrupt(messages []string) { dlog.Client.Resume() } -func (s *stats) statsData(connected, newConnections int, throttle int) map[string]interface{} { +func (s *stats) statsData(connected, newConnections int, + throttle int) map[string]interface{} { + percConnected := percentOf(float64(s.servers), float64(connected)) data := make(map[string]interface{}) @@ -112,7 +115,6 @@ func (s *stats) statsData(connected, newConnections int, throttle int) map[strin func (s *stats) statsLine(connected, newConnections int, throttle int) string { sb := strings.Builder{} - i := 0 for k, v := range s.statsData(connected, newConnections, throttle) { if i > 0 { @@ -123,7 +125,6 @@ func (s *stats) statsLine(connected, newConnections int, throttle int) string { sb.WriteString(fmt.Sprintf("%v", v)) i++ } - return sb.String() } diff --git a/internal/clients/tailclient.go b/internal/clients/tailclient.go index d42a0e4..35c01d4 100644 --- a/internal/clients/tailclient.go +++ b/internal/clients/tailclient.go @@ -19,7 +19,6 @@ type TailClient struct { // NewTailClient returns a new TailClient. func NewTailClient(args config.Args) (*TailClient, error) { args.Mode = omode.TailClient - c := TailClient{ baseClient: baseClient{ Args: args, @@ -30,7 +29,6 @@ func NewTailClient(args config.Args) (*TailClient, error) { c.init() c.makeConnections(c) - return &c, nil } @@ -48,6 +46,5 @@ func (c TailClient) makeCommands() (commands []string) { c.Mode.String(), c.Args.SerializeOptions(), file, regex)) } dlog.Client.Debug(commands) - return } diff --git a/internal/color/brush/brush.go b/internal/color/brush/brush.go index 82fa410..63d63d8 100644 --- a/internal/color/brush/brush.go +++ b/internal/color/brush/brush.go @@ -32,7 +32,6 @@ func paintSeverity(sb *strings.Builder, text string) bool { default: return false } - return true } @@ -87,9 +86,9 @@ func paintRemote(sb *strings.Builder, line string) { config.Client.TermColors.Remote.DelimiterAttr) color.PaintWithAttr(sb, splitted[4], - config.Client.TermColors.Remote.IdFg, - config.Client.TermColors.Remote.IdBg, - config.Client.TermColors.Remote.IdAttr) + config.Client.TermColors.Remote.IDFg, + config.Client.TermColors.Remote.IDBg, + config.Client.TermColors.Remote.IDAttr) color.PaintWithAttr(sb, protocol.FieldDelimiter, config.Client.TermColors.Remote.DelimiterFg, config.Client.TermColors.Remote.DelimiterBg, @@ -191,6 +190,5 @@ func Colorfy(line string) string { color.BgDefault, color.AttrNone) } - return sb.String() } diff --git a/internal/color/color.go b/internal/color/color.go index 5c25855..9d0bc2e 100644 --- a/internal/color/color.go +++ b/internal/color/color.go @@ -52,15 +52,19 @@ const ( AttrHidden Attribute = escape + "[8m" ) +// ColorNames is the list of all supported terminal colors. var ColorNames = []string{ "Black", "Red", "Green", "Yellow", "Blue", "Magenta", "Cyan", "White", "Default", } +// AttributeNames is the list of all supported terminal text attributes. var AttributeNames = []string{ - "Bold", "Dim", "Italic", "Underline", "Blink", "SlowBlink", "RapidBlink", "Reverse", "Hidden", "None", + "Bold", "Dim", "Italic", "Underline", "Blink", "SlowBlink", "RapidBlink", + "Reverse", "Hidden", "None", } -// ToFgColor converts a given string (e.g. from a config file) into a foreground color code. +// ToFgColor converts a given string (e.g. from a config file) into a foreground +// color code. func ToFgColor(s string) (FgColor, error) { switch strings.ToLower(s) { case "black": @@ -86,7 +90,8 @@ func ToFgColor(s string) (FgColor, error) { } } -// ToBgColor converts a given string (e.g. from a config file) into a background color code. +// ToBgColor converts a given string (e.g. from a config file) into a background +// color code. func ToBgColor(s string) (BgColor, error) { switch strings.ToLower(s) { case "black": diff --git a/internal/color/paint.go b/internal/color/paint.go index 53c9abb..7735d87 100644 --- a/internal/color/paint.go +++ b/internal/color/paint.go @@ -10,12 +10,14 @@ func PaintStr(text string, fg FgColor, bg BgColor) string { return fmt.Sprintf("%s%s%s%s%s", fg, bg, text, BgDefault, FgDefault) } -// PaintStrWithAttr paints a given text in a given foreground/background/attribute combination +// PaintStrWithAttr paints a given text in a given foreground/background/attribute +// combination func PaintStrWithAttr(text string, fg FgColor, bg BgColor, attr Attribute) string { if attr == AttrNone { return PaintStr(text, fg, bg) } - return fmt.Sprintf("%s%s%s%s%s%s%s", fg, bg, attr, text, AttrReset, BgDefault, FgDefault) + return fmt.Sprintf("%s%s%s%s%s%s%s", fg, bg, attr, text, AttrReset, + BgDefault, FgDefault) } // PaintStrFg paints a given text in a given foreground color. @@ -48,8 +50,11 @@ func Reset(sb *strings.Builder) { sb.WriteString(string(FgDefault)) } -// PaintWithAttr starts painting a given text in a given foreground/background/attribute combination. -func PaintWithAttr(sb *strings.Builder, text string, fg FgColor, bg BgColor, attr Attribute) { +// PaintWithAttr starts painting a given text in a given foreground/background/ +// attribute combination. +func PaintWithAttr(sb *strings.Builder, text string, fg FgColor, bg BgColor, + attr Attribute) { + if attr == AttrNone { Paint(sb, text, fg, bg) return @@ -63,8 +68,10 @@ func PaintWithAttr(sb *strings.Builder, text string, fg FgColor, bg BgColor, att sb.WriteString(string(FgDefault)) } -// PaintWithAttrs is similar to PaintWithAttr, but it takes multiple text attributes. -func PaintWithAttrs(sb *strings.Builder, text string, fg FgColor, bg BgColor, attrs []Attribute) { +// PaintWithAttrs is similar to PaintWithAttr, but it takes multiple attributes. +func PaintWithAttrs(sb *strings.Builder, text string, fg FgColor, bg BgColor, + attrs []Attribute) { + sb.WriteString(string(fg)) sb.WriteString(string(bg)) for _, attr := range attrs { diff --git a/internal/color/table.go b/internal/color/table.go index 2115edf..e0e4946 100644 --- a/internal/color/table.go +++ b/internal/color/table.go @@ -5,8 +5,19 @@ import ( "os" ) -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." +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." +// TablePrintAndExit prints the color table and then exits the process. func TablePrintAndExit(displaySampleParagraph bool) { for _, attr := range AttributeNames { if attr == "Hidden" || attr == "SlowBlink" { @@ -24,11 +35,13 @@ func printColorTable(attr string, displaySampleParagraph bool) { if fg == bg { continue } + bgColor, _ := ToBgColor(bg) attribute, _ := ToAttribute(attr) - - text := fmt.Sprintf(" Foreground:%10s | Background:%10s | Attribute:%10s ", fg, bg, attr) + text := fmt.Sprintf(" Foreground:%10s | Background:%10s | Attribute:%10s ", + fg, bg, attr) fmt.Print(PaintStrWithAttr(text, fgColor, bgColor, attribute)) + if displaySampleParagraph { fmt.Print("\n") fmt.Print(PaintStrWithAttr(sampleParagraph, fgColor, bgColor, attribute)) diff --git a/internal/config/args.go b/internal/config/args.go index a671ae3..f721390 100644 --- a/internal/config/args.go +++ b/internal/config/args.go @@ -74,13 +74,13 @@ func (a *Args) String() string { // 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:serverless=%v", a.Quiet, a.Spartan, a.Serverless) + return fmt.Sprintf("quiet=%v:spartan=%v:serverless=%v", a.Quiet, a.Spartan, + a.Serverless) } // DeserializeOptions deserializes the options, but into a map. func DeserializeOptions(opts []string) (map[string]string, error) { options := make(map[string]string, len(opts)) - for _, o := range opts { kv := strings.SplitN(o, "=", 2) if len(kv) != 2 { @@ -97,9 +97,7 @@ func DeserializeOptions(opts []string) (map[string]string, error) { } val = string(decoded) } - options[key] = val } - return options, nil } diff --git a/internal/config/client.go b/internal/config/client.go index ecd05c5..8227c68 100644 --- a/internal/config/client.go +++ b/internal/config/client.go @@ -15,9 +15,9 @@ type remoteTermColors struct { HostnameAttr color.Attribute HostnameBg color.BgColor HostnameFg color.FgColor - IdAttr color.Attribute - IdBg color.BgColor - IdFg color.FgColor + IDAttr color.Attribute + IDBg color.BgColor + IDFg color.FgColor StatsOkAttr color.Attribute StatsOkBg color.BgColor StatsOkFg color.FgColor @@ -124,9 +124,9 @@ func newDefaultClientConfig() *ClientConfig { HostnameAttr: color.AttrBold, HostnameBg: color.BgBlue, HostnameFg: color.FgWhite, - IdAttr: color.AttrDim, - IdBg: color.BgBlue, - IdFg: color.FgWhite, + IDAttr: color.AttrDim, + IDBg: color.BgBlue, + IDFg: color.FgWhite, StatsOkAttr: color.AttrNone, StatsOkBg: color.BgGreen, StatsOkFg: color.FgBlack, diff --git a/internal/config/config.go b/internal/config/config.go index 077a658..b99b22b 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -9,11 +9,11 @@ const ( ScheduleUser string = "DTAIL-SCHEDULE" // ContinuousUser is used for non-interactive continuous mapreduce queries. ContinuousUser string = "DTAIL-CONTINUOUS" - // InterruptTimeoutS is used to terminate DTail when Ctrl+C was pressed twice within a given interval. + // InterruptTimeoutS specifies the Ctrl+C log pause interval. InterruptTimeoutS int = 3 - // ConnectionsPerCPU controls how many connections are established concurrently as a start (slow start) + // DefaultConnectionsPerCPU controls how many connections are established concurrently. DefaultConnectionsPerCPU int = 10 - // DTailSSHServerDefaultPort is the default DServer port. + // DefaultSSHPort is the default DServer port. DefaultSSHPort int = 2222 // DefaultLogLevel specifies the default log level (obviously) DefaultLogLevel string = "INFO" diff --git a/internal/config/initializer.go b/internal/config/initializer.go index e4cbeaf..0a913db 100644 --- a/internal/config/initializer.go +++ b/internal/config/initializer.go @@ -20,13 +20,13 @@ type initializer struct { type transformCb func(*initializer, *Args, []string) error -func (c *initializer) parseConfig(args *Args) error { +func (in *initializer) parseConfig(args *Args) error { if strings.ToUpper(args.ConfigFile) == "NONE" { return nil } if args.ConfigFile != "" { - return c.parseSpecificConfig(args.ConfigFile) + return in.parseSpecificConfig(args.ConfigFile) } if homeDir, err := os.UserHomeDir(); err != nil { @@ -35,7 +35,7 @@ func (c *initializer) parseConfig(args *Args) error { paths = append(paths, fmt.Sprintf("%s/.dtail.conf", homeDir)) for _, configPath := range paths { if _, err := os.Stat(configPath); !os.IsNotExist(err) { - c.parseSpecificConfig(configPath) + in.parseSpecificConfig(configPath) } } } @@ -43,7 +43,7 @@ func (c *initializer) parseConfig(args *Args) error { return nil } -func (c *initializer) parseSpecificConfig(configFile string) error { +func (in *initializer) parseSpecificConfig(configFile string) error { fd, err := os.Open(configFile) if err != nil { return fmt.Errorf("Unable to read config file: %v", err) @@ -55,68 +55,74 @@ func (c *initializer) parseSpecificConfig(configFile string) error { return fmt.Errorf("Unable to read config file %s: %v", configFile, err) } - if err := json.Unmarshal([]byte(cfgBytes), c); err != nil { + if err := json.Unmarshal([]byte(cfgBytes), in); err != nil { return fmt.Errorf("Unable to parse config file %s: %v", configFile, err) } return nil } -func (i *initializer) transformConfig(sourceProcess source.Source, args *Args, additionalArgs []string) error { +func (in *initializer) transformConfig(sourceProcess source.Source, args *Args, + additionalArgs []string) error { switch sourceProcess { case source.Server: - return i.optimusPrime(transformServer, args, additionalArgs) + return in.optimusPrime(transformServer, args, additionalArgs) case source.Client: - return i.optimusPrime(transformClient, args, additionalArgs) + return in.optimusPrime(transformClient, args, additionalArgs) case source.HealthCheck: - return i.optimusPrime(transformHealthCheck, args, additionalArgs) + return in.optimusPrime(transformHealthCheck, args, additionalArgs) default: - return fmt.Errorf("Unable to transform config, unknown source '%s'", sourceProcess) + return fmt.Errorf("Unable to transform config, unknown source '%s'", + sourceProcess) } } -func (i *initializer) optimusPrime(sourceCb transformCb, args *Args, additionalArgs []string) error { +func (in *initializer) optimusPrime(sourceCb transformCb, args *Args, + additionalArgs []string) error { + // Copy args to config objects. + // NEXT: Maybe unify args and config structs? if args.SSHPort != DefaultSSHPort { - i.Common.SSHPort = args.SSHPort + in.Common.SSHPort = args.SSHPort } if args.LogLevel != DefaultLogLevel { - i.Common.LogLevel = args.LogLevel + in.Common.LogLevel = args.LogLevel } if args.NoColor { - i.Client.TermColorsEnable = false + in.Client.TermColorsEnable = false } if args.LogDir != "" { - i.Common.LogDir = args.LogDir + in.Common.LogDir = args.LogDir } if args.Logger != "" { - i.Common.Logger = args.Logger + in.Common.Logger = args.Logger } if args.ConnectionsPerCPU == 0 { args.ConnectionsPerCPU = DefaultConnectionsPerCPU } // Setup log directory. - if strings.Contains(i.Common.LogDir, "~/") { + if strings.Contains(in.Common.LogDir, "~/") { homeDir, err := os.UserHomeDir() if err != nil { panic(err) } - i.Common.LogDir = strings.ReplaceAll(i.Common.LogDir, "~/", fmt.Sprintf("%s/", homeDir)) + in.Common.LogDir = strings.ReplaceAll(in.Common.LogDir, "~/", + fmt.Sprintf("%s/", homeDir)) } // Source type specific transormations. - sourceCb(i, args, additionalArgs) + sourceCb(in, args, additionalArgs) // Spartan mode. if args.Spartan { args.Quiet = true args.NoColor = true - i.Client.TermColorsEnable = false + in.Client.TermColorsEnable = false if args.LogLevel == "" { args.LogLevel = "ERROR" - i.Common.LogLevel = "ERROR" + in.Common.LogLevel = "ERROR" } } // Interpret additional args as file list or as query. @@ -135,29 +141,28 @@ func (i *initializer) optimusPrime(sourceCb transformCb, args *Args, additionalA return nil } -func transformClient(i *initializer, args *Args, additionalArgs []string) error { +func transformClient(in *initializer, args *Args, additionalArgs []string) error { // Serverless mode. if args.Discovery == "" && (args.ServersStr == "" || strings.ToLower(args.ServersStr) == "serverless") { // We are not connecting to any servers. args.Serverless = true - i.Common.LogLevel = "warn" + in.Common.LogLevel = "warn" } - return nil } -func transformServer(i *initializer, args *Args, additionalArgs []string) error { +func transformServer(in *initializer, args *Args, additionalArgs []string) error { return nil } -func transformHealthCheck(i *initializer, args *Args, additionalArgs []string) error { +func transformHealthCheck(in *initializer, args *Args, additionalArgs []string) error { // Serverless mode. if args.Discovery == "" && (args.ServersStr == "" || strings.ToLower(args.ServersStr) == "serverless") { // We are not connecting to any servers. args.Serverless = true - i.Common.LogLevel = "warn" + in.Common.LogLevel = "warn" } args.TrustAllHosts = true return nil diff --git a/internal/config/server.go b/internal/config/server.go index 8bbb394..677f5ac 100644 --- a/internal/config/server.go +++ b/internal/config/server.go @@ -4,8 +4,8 @@ import ( "errors" ) -// Permissions map. Each SSH user has a list of permissions which -// log files it is allowed to follow and which ones not. +// Permissions map. Each SSH user has a list of permissions which log files it +// is allowed to follow and which ones not. type Permissions struct { // The default user permissions. Default []string @@ -68,7 +68,6 @@ var ServerRelaxedAuthEnable bool func newDefaultServerConfig() *ServerConfig { defaultPermissions := []string{"^/.*"} defaultBindAddress := "0.0.0.0" - return &ServerConfig{ SSHBindAddress: defaultBindAddress, MaxConnections: 10, @@ -89,10 +88,8 @@ func ServerUserPermissions(userName string) (permissions []string, err error) { if p, ok := Server.Permissions.Users[userName]; ok { permissions = p } - if len(permissions) == 0 { err = errors.New("Empty set of permission, user won't be able to open any files") } - return } diff --git a/internal/discovery/discovery.go b/internal/discovery/discovery.go index 83ee95e..8bb1e85 100644 --- a/internal/discovery/discovery.go +++ b/internal/discovery/discovery.go @@ -73,7 +73,6 @@ func (d *Discovery) initRegex() { regexStr := string(runes) dlog.Common.Debug("Using filter regex", regexStr) - regex, err := regexp.Compile(regexStr) if err != nil { dlog.Common.FatalPanic("Could not compile regex", regexStr, err) @@ -90,9 +89,7 @@ func (d *Discovery) ServerList() []string { if d.regex != nil { servers = d.filterList(servers) } - servers = d.dedupList(servers) - if d.order == Shuffle { servers = d.shuffleList(servers) } @@ -105,12 +102,10 @@ func (d *Discovery) serverListFromModule() []string { if d.module != "" { return d.serverListFromReflectedModule() } - if _, err := os.Stat(d.server); err == nil { // Appears to be a file name, now try to read from that file. return d.ServerListFromFILE() } - // Appears to be a list of FQDNs (or a single FQDN) return d.ServerListFromCOMMA() } @@ -120,18 +115,16 @@ func (d *Discovery) serverListFromModule() []string { // Discovery. Whereas MODULENAME must be a upeprcase string. func (d *Discovery) serverListFromReflectedModule() []string { methodName := fmt.Sprintf("ServerListFrom%s", d.module) - + // Now we are reflecting the serve discovery function by it's name. rt := reflect.TypeOf(d) reflectedMethod, ok := rt.MethodByName(methodName) if !ok { dlog.Common.FatalPanic("No such server discovery module", d.module, methodName) } - inputValues := make([]reflect.Value, 1) // Thist input value is method receiver. inputValues[0] = reflect.ValueOf(d) returnValues := reflectedMethod.Func.Call(inputValues) - // First return value is server list. return returnValues[0].Interface().([]string) } @@ -139,27 +132,23 @@ func (d *Discovery) serverListFromReflectedModule() []string { // Filter server list based on a regexp. func (d *Discovery) filterList(servers []string) (filtered []string) { dlog.Common.Debug("Filtering server list") - for _, server := range servers { if d.regex.MatchString(server) { filtered = append(filtered, server) } } - return } // Deduplicate the server list. func (d *Discovery) dedupList(servers []string) (deduped []string) { serverMap := make(map[string]struct{}, len(servers)) - for _, server := range servers { if _, ok := serverMap[server]; !ok { serverMap[server] = struct{}{} deduped = append(deduped, server) } } - dlog.Common.Debug("Deduped server list", len(servers), len(deduped)) return } diff --git a/internal/done.go b/internal/done.go index 5ea22a0..94f9289 100644 --- a/internal/done.go +++ b/internal/done.go @@ -35,7 +35,6 @@ func (d *Done) Done() <-chan struct{} { func (d *Done) Shutdown() { d.mutex.Lock() defer d.mutex.Unlock() - select { case <-d.ch: return diff --git a/internal/io/dlog/dlog.go b/internal/io/dlog/dlog.go index 28e6882..da67585 100644 --- a/internal/io/dlog/dlog.go +++ b/internal/io/dlog/dlog.go @@ -82,7 +82,7 @@ func new(sourceProcess, sourcePackage source.Source) *DLog { if err != nil { panic(err) } - strategy := loggers.GetStrategy(config.Common.LogStrategy) + strategy := loggers.NewStrategy(config.Common.LogStrategy) loggerName := config.Common.Logger level := newLevel(config.Common.LogLevel) @@ -153,6 +153,7 @@ func (d *DLog) writeArgStrings(sb *strings.Builder, args []interface{}) { } } +// FatalPanic terminates the process with a fatal error. func (d *DLog) FatalPanic(args ...interface{}) { d.log(Fatal, args) d.Flush() @@ -162,42 +163,51 @@ func (d *DLog) FatalPanic(args ...interface{}) { panic(sb.String()) } +// Fatal logs a fatal error. func (d *DLog) Fatal(args ...interface{}) string { return d.log(Fatal, args) } +// Error logging. func (d *DLog) Error(args ...interface{}) string { return d.log(Error, args) } +// Warn logs a warning message. func (d *DLog) Warn(args ...interface{}) string { return d.log(Warn, args) } +// Info logging. func (d *DLog) Info(args ...interface{}) string { return d.log(Info, args) } +// Verbose logging. func (d *DLog) Verbose(args ...interface{}) string { return d.log(Verbose, args) } +// Debug logging. func (d *DLog) Debug(args ...interface{}) string { return d.log(Debug, args) } +// Trace logging. func (d *DLog) Trace(args ...interface{}) string { _, file, line, _ := runtime.Caller(1) args = append(args, fmt.Sprintf("at %s:%d", file, line)) return d.log(Trace, args) } +// Devel used for development purpose only logging (e.g. "print" debugging). func (d *DLog) Devel(args ...interface{}) string { _, file, line, _ := runtime.Caller(1) args = append(args, fmt.Sprintf("at %s:%d", file, line)) return d.log(Devel, args) } +// Raw message logging. func (d *DLog) Raw(message string) string { if !config.Client.TermColorsEnable || !d.logger.SupportsColors() { d.logger.Log(time.Now(), message) @@ -207,6 +217,7 @@ func (d *DLog) Raw(message string) string { return message } +// Mapreduce logging. func (d *DLog) Mapreduce(table string, data map[string]interface{}) string { args := make([]interface{}, len(data)+1) @@ -251,6 +262,11 @@ func (d *DLog) Mapreduce(table string, data map[string]interface{}) string { return d.log(Info, args) } -func (d *DLog) Flush() { d.logger.Flush() } -func (d *DLog) Pause() { d.logger.Pause() } +// Flush the log buffers. +func (d *DLog) Flush() { d.logger.Flush() } + +// Pause the logging. +func (d *DLog) Pause() { d.logger.Pause() } + +// Resume the logging. func (d *DLog) Resume() { d.logger.Resume() } diff --git a/internal/io/dlog/level.go b/internal/io/dlog/level.go index 248ad83..0971094 100644 --- a/internal/io/dlog/level.go +++ b/internal/io/dlog/level.go @@ -7,6 +7,7 @@ import ( type level int +// Available log levels. const ( Fatal level = iota Error level = iota @@ -20,7 +21,8 @@ const ( All level = iota ) -var allLevels = []level{Fatal, Error, Warn, Info, Default, Verbose, Debug, Devel, Trace, All} +var allLevels = []level{Fatal, Error, Warn, Info, Default, Verbose, Debug, + Devel, Trace, All} func newLevel(l string) level { switch strings.ToLower(l) { @@ -73,6 +75,5 @@ func (l level) String() string { 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 index dda3ee6..415d7fb 100644 --- a/internal/io/dlog/loggers/factory.go +++ b/internal/io/dlog/loggers/factory.go @@ -9,11 +9,13 @@ import ( var factoryMap map[string]Logger var factoryMutex sync.Mutex +// Factory is there to retrieve a logger based on various settings. func Factory(sourceName, loggerName string, rotationStrategy Strategy) Logger { factoryMutex.Lock() defer factoryMutex.Unlock() - id := fmt.Sprintf("sourceName:%s,fileBase:%s,loggerName:%s", sourceName, rotationStrategy.FileBase, loggerName) + id := fmt.Sprintf("sourceName:%s,fileBase:%s,loggerName:%s", sourceName, + rotationStrategy.FileBase, loggerName) if factoryMap == nil { factoryMap = make(map[string]Logger) } @@ -36,10 +38,10 @@ func Factory(sourceName, loggerName string, rotationStrategy Strategy) Logger { panic(fmt.Sprintf("Unsupported logger type '%s'", loggerName)) } } - return singleton } +// FactoryRotate invokes a log rotation of all loggers. func FactoryRotate() { factoryMutex.Lock() defer factoryMutex.Unlock() diff --git a/internal/io/dlog/loggers/file.go b/internal/io/dlog/loggers/file.go index 87280fd..94824fe 100644 --- a/internal/io/dlog/loggers/file.go +++ b/internal/io/dlog/loggers/file.go @@ -12,8 +12,7 @@ import ( "github.com/mimecast/dtail/internal/config" ) -type fileWriter struct { -} +type fileWriter struct{} type fileMessageBuf struct { now time.Tim