From 0945da8dfefcbb723eecea0e5f4eafff63398253 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Paul=20B=C3=BCtow?= Date: Sun, 26 Jan 2020 11:26:53 +0000 Subject: Introduce drun command, refactor code to use context package --- Makefile | 16 +- cmd/dcat/main.go | 19 +- cmd/dexec/main.go | 80 ---- cmd/dgrep/main.go | 18 +- cmd/dmap/main.go | 19 +- cmd/drun/main.go | 82 ++++ cmd/dserver/main.go | 22 +- cmd/dtail/main.go | 19 +- doc/examples.md | 4 +- doc/installation.md | 5 +- doc/quickstart.md | 23 +- go.mod | 1 + go.sum | 9 + internal/clients/args.go | 3 +- internal/clients/baseclient.go | 130 +++--- internal/clients/catclient.go | 20 +- internal/clients/client.go | 5 +- internal/clients/connectionmaker.go | 12 - internal/clients/execclient.go | 48 --- internal/clients/grepclient.go | 20 +- internal/clients/handlers/basehandler.go | 84 ++-- internal/clients/handlers/clienthandler.go | 11 +- internal/clients/handlers/handler.go | 12 +- internal/clients/handlers/healthhandler.go | 21 +- internal/clients/handlers/maprhandler.go | 21 +- internal/clients/handlers/withcancel.go | 24 ++ internal/clients/healthclient.go | 7 +- internal/clients/maker.go | 8 + internal/clients/maprclient.go | 52 ++- internal/clients/remote/connection.go | 116 ++---- internal/clients/runclient.go | 40 ++ internal/clients/stats.go | 8 +- internal/clients/tailclient.go | 21 +- internal/discovery/comma.go | 2 +- internal/discovery/discovery.go | 21 +- internal/discovery/file.go | 2 +- internal/fs/catfile.go | 27 -- internal/fs/filereader.go | 9 - internal/fs/lineread.go | 28 -- internal/fs/permissions/permission.go | 14 - internal/fs/permissions/permission_linux.c | 395 ------------------- internal/fs/permissions/permission_linux.go | 33 -- internal/fs/permissions/permission_linux.h | 60 --- internal/fs/permissions/permission_test.go | 112 ------ internal/fs/readfile.go | 318 --------------- internal/fs/stats.go | 69 ---- internal/fs/tailfile.go | 27 -- internal/io/fs/catfile.go | 21 + internal/io/fs/filereader.go | 14 + internal/io/fs/permissions/permission.go | 14 + internal/io/fs/permissions/permission_linux.c | 395 +++++++++++++++++++ internal/io/fs/permissions/permission_linux.go | 33 ++ internal/io/fs/permissions/permission_linux.h | 60 +++ internal/io/fs/permissions/permission_test.go | 112 ++++++ internal/io/fs/readfile.go | 307 +++++++++++++++ internal/io/fs/stats.go | 69 ++++ internal/io/fs/tailfile.go | 21 + internal/io/line/line.go | 28 ++ internal/io/logger/logger.go | 445 +++++++++++++++++++++ internal/io/run/run.go | 104 +++++ internal/logger/logger.go | 457 ---------------------- internal/mapr/aggregateset.go | 5 +- internal/mapr/client/aggregate.go | 25 +- internal/mapr/groupset.go | 5 +- internal/mapr/logformat/parser.go | 2 +- internal/mapr/query.go | 2 +- internal/mapr/server/aggregate.go | 141 ++++--- internal/mapr/wherecondition.go | 2 +- internal/omode/mode.go | 6 +- internal/pprof/pprof.go | 3 +- internal/prompt/prompt.go | 2 +- internal/server/handlers/controlhandler.go | 42 +- internal/server/handlers/handler.go | 2 - internal/server/handlers/mapcommand.go | 35 ++ internal/server/handlers/readcommand.go | 158 ++++++++ internal/server/handlers/runcommand.go | 73 ++++ internal/server/handlers/serverhandler.go | 521 +++++++++---------------- internal/server/server.go | 70 ++-- internal/server/stats.go | 10 +- internal/ssh/client/authmethods.go | 2 +- internal/ssh/client/hostkeycallback.go | 10 +- internal/ssh/server/hostkey.go | 2 +- internal/ssh/server/publickeycallback.go | 2 +- internal/ssh/ssh.go | 2 +- internal/user/name.go | 15 +- internal/user/server/user.go | 44 ++- internal/version/version.go | 22 +- samples/dtail.json.sample | 12 +- 88 files changed, 2791 insertions(+), 2601 deletions(-) delete mode 100644 cmd/dexec/main.go create mode 100644 cmd/drun/main.go delete mode 100644 internal/clients/connectionmaker.go delete mode 100644 internal/clients/execclient.go create mode 100644 internal/clients/handlers/withcancel.go create mode 100644 internal/clients/maker.go create mode 100644 internal/clients/runclient.go delete mode 100644 internal/fs/catfile.go delete mode 100644 internal/fs/filereader.go delete mode 100644 internal/fs/lineread.go delete mode 100644 internal/fs/permissions/permission.go delete mode 100644 internal/fs/permissions/permission_linux.c delete mode 100644 internal/fs/permissions/permission_linux.go delete mode 100644 internal/fs/permissions/permission_linux.h delete mode 100644 internal/fs/permissions/permission_test.go delete mode 100644 internal/fs/readfile.go delete mode 100644 internal/fs/stats.go delete mode 100644 internal/fs/tailfile.go create mode 100644 internal/io/fs/catfile.go create mode 100644 internal/io/fs/filereader.go create mode 100644 internal/io/fs/permissions/permission.go create mode 100644 internal/io/fs/permissions/permission_linux.c create mode 100644 internal/io/fs/permissions/permission_linux.go create mode 100644 internal/io/fs/permissions/permission_linux.h create mode 100644 internal/io/fs/permissions/permission_test.go create mode 100644 internal/io/fs/readfile.go create mode 100644 internal/io/fs/stats.go create mode 100644 internal/io/fs/tailfile.go create mode 100644 internal/io/line/line.go create mode 100644 internal/io/logger/logger.go create mode 100644 internal/io/run/run.go delete mode 100644 internal/logger/logger.go create mode 100644 internal/server/handlers/mapcommand.go create mode 100644 internal/server/handlers/readcommand.go create mode 100644 internal/server/handlers/runcommand.go diff --git a/Makefile b/Makefile index 3480637..c358d8e 100644 --- a/Makefile +++ b/Makefile @@ -1,29 +1,31 @@ GO ?= go all: build build: + ${GO} build -o dserver ./cmd/dserver/main.go ${GO} build -o dcat ./cmd/dcat/main.go - ${GO} build -o dexec ./cmd/dexec/main.go ${GO} build -o dgrep ./cmd/dgrep/main.go ${GO} build -o dmap ./cmd/dmap/main.go - ${GO} build -o dserver ./cmd/dserver/main.go + ${GO} build -o drun ./cmd/drun/main.go ${GO} build -o dtail ./cmd/dtail/main.go clean: - rm -v dtail dgrep dcat dmap dserver dexec 2>/dev/null + ls ./cmd/ | while read cmd; do \ + test -f $$cmd && rm $$cmd; \ + done install: build + cp -pv dserver ${GOPATH}/bin/dserver cp -pv dcat ${GOPATH}/bin/dcat - cp -pv dexec ${GOPATH}/bin/dexec cp -pv dgrep ${GOPATH}/bin/dgrep cp -pv dmap ${GOPATH}/bin/dmap - cp -pv dserver ${GOPATH}/bin/dserver + cp -pv drun ${GOPATH}/bin/drun cp -pv dtail ${GOPATH}/bin/dtail vet: find . -type d | while read dir; do \ echo ${GO} vet $$dir; \ ${GO} vet $$dir; \ - done + done lint: ${GO} get golang.org/x/lint/golint find . -type d | while read dir; do \ echo ${GOPATH}/bin/golint $$dir; \ ${GOPATH}/bin/golint $$dir; \ - done + done diff --git a/cmd/dcat/main.go b/cmd/dcat/main.go index b02d369..1ec945d 100644 --- a/cmd/dcat/main.go +++ b/cmd/dcat/main.go @@ -1,12 +1,14 @@ package main import ( + "context" "flag" + "os" "github.com/mimecast/dtail/internal/clients" "github.com/mimecast/dtail/internal/color" "github.com/mimecast/dtail/internal/config" - "github.com/mimecast/dtail/internal/logger" + "github.com/mimecast/dtail/internal/io/logger" "github.com/mimecast/dtail/internal/pprof" "github.com/mimecast/dtail/internal/user" "github.com/mimecast/dtail/internal/version" @@ -27,7 +29,6 @@ func main() { var sshPort int var trustAllHosts bool - pingTimeoutS := 60 userName := user.Name() flag.BoolVar(&debugEnable, "debug", false, "Activate debug messages") @@ -37,7 +38,6 @@ func main() { flag.BoolVar(&silentEnable, "silent", false, "Reduce output") flag.BoolVar(&trustAllHosts, "trustAllHosts", false, "Auto trust all unknown host keys") flag.IntVar(&connectionsPerCPU, "cpc", 10, "How many connections established per CPU core concurrently") - flag.IntVar(&pingTimeoutS, "pingTimeout", 10, "The server ping timeout (0 means disable pings)") flag.IntVar(&sshPort, "port", 2222, "SSH server port") flag.StringVar(&cfgFile, "cfg", "", "Config file path") flag.StringVar(&discovery, "discovery", "", "Server discovery method") @@ -54,9 +54,10 @@ func main() { version.PrintAndExit() } + ctx := context.Background() serverEnable := false - logger.Start(serverEnable, debugEnable, silentEnable, silentEnable) - defer logger.Stop() + + logger.Start(ctx, serverEnable, debugEnable, silentEnable, silentEnable) if pprofEnable || config.Common.PProfEnable { pprof.Start() @@ -67,14 +68,16 @@ func main() { ServersStr: serversStr, Discovery: discovery, UserName: userName, - Files: files, + What: files, TrustAllHosts: trustAllHosts, - PingTimeout: pingTimeoutS, } client, err := clients.NewCatClient(args) if err != nil { panic(err) } - client.Start() + + status := client.Start(ctx) + logger.Flush() + os.Exit(status) } diff --git a/cmd/dexec/main.go b/cmd/dexec/main.go deleted file mode 100644 index 7a7ab1f..0000000 --- a/cmd/dexec/main.go +++ /dev/null @@ -1,80 +0,0 @@ -package main - -import ( - "flag" - - "github.com/mimecast/dtail/internal/clients" - "github.com/mimecast/dtail/internal/color" - "github.com/mimecast/dtail/internal/config" - "github.com/mimecast/dtail/internal/logger" - "github.com/mimecast/dtail/internal/pprof" - "github.com/mimecast/dtail/internal/user" - "github.com/mimecast/dtail/internal/version" -) - -// The evil begins here. -func main() { - var cfgFile string - var connectionsPerCPU int - var debugEnable bool - var discovery string - var displayVersion bool - var command string - var noColor bool - var pprofEnable bool - var serversStr string - var silentEnable bool - var sshPort int - var trustAllHosts bool - - pingTimeoutS := 60 - userName := user.Name() - - flag.BoolVar(&debugEnable, "debug", false, "Activate debug messages") - flag.BoolVar(&displayVersion, "version", false, "Display version") - flag.BoolVar(&noColor, "noColor", false, "Disable ANSII terminal colors") - flag.BoolVar(&pprofEnable, "pprofEnable", false, "Enable pprof server") - flag.BoolVar(&silentEnable, "silent", false, "Reduce output") - flag.BoolVar(&trustAllHosts, "trustAllHosts", false, "Auto trust all unknown host keys") - flag.IntVar(&connectionsPerCPU, "cpc", 10, "How many connections established per CPU core concurrently") - flag.IntVar(&pingTimeoutS, "pingTimeout", 10, "The server ping timeout (0 means disable pings)") - flag.IntVar(&sshPort, "port", 2222, "SSH server port") - flag.StringVar(&cfgFile, "cfg", "", "Config file path") - flag.StringVar(&discovery, "discovery", "", "Server discovery method") - flag.StringVar(&command, "command", "", "Command to run") - flag.StringVar(&serversStr, "servers", "", "Remote servers to connect") - flag.StringVar(&userName, "user", userName, "Your system user name") - - flag.Parse() - - config.Read(cfgFile, sshPort) - color.Colored = !noColor - - if displayVersion { - version.PrintAndExit() - } - - serverEnable := false - logger.Start(serverEnable, debugEnable, silentEnable, silentEnable) - defer logger.Stop() - - if pprofEnable || config.Common.PProfEnable { - pprof.Start() - } - - args := clients.Args{ - ConnectionsPerCPU: connectionsPerCPU, - ServersStr: serversStr, - Discovery: discovery, - UserName: userName, - Files: files, - TrustAllHosts: trustAllHosts, - PingTimeout: pingTimeoutS, - } - - client, err := clients.NewExecClient(args) - if err != nil { - panic(err) - } - client.Start() -} diff --git a/cmd/dgrep/main.go b/cmd/dgrep/main.go index d1a7d52..74a501f 100644 --- a/cmd/dgrep/main.go +++ b/cmd/dgrep/main.go @@ -1,12 +1,14 @@ package main import ( + "context" "flag" + "os" "github.com/mimecast/dtail/internal/clients" "github.com/mimecast/dtail/internal/color" "github.com/mimecast/dtail/internal/config" - "github.com/mimecast/dtail/internal/logger" + "github.com/mimecast/dtail/internal/io/logger" "github.com/mimecast/dtail/internal/pprof" "github.com/mimecast/dtail/internal/user" "github.com/mimecast/dtail/internal/version" @@ -28,7 +30,6 @@ func main() { var sshPort int var trustAllHosts bool - pingTimeoutS := 60 userName := user.Name() flag.BoolVar(&debugEnable, "debug", false, "Activate debug messages") @@ -38,7 +39,6 @@ func main() { flag.BoolVar(&silentEnable, "silent", false, "Reduce output") flag.BoolVar(&trustAllHosts, "trustAllHosts", false, "Auto trust all unknown host keys") flag.IntVar(&connectionsPerCPU, "cpc", 10, "How many connections established per CPU core concurrently") - flag.IntVar(&pingTimeoutS, "pingTimeout", 10, "The server ping timeout (0 means disable pings)") flag.IntVar(&sshPort, "port", 2222, "SSH server port") flag.StringVar(&cfgFile, "cfg", "", "Config file path") flag.StringVar(&discovery, "discovery", "", "Server discovery method") @@ -56,9 +56,9 @@ func main() { version.PrintAndExit() } + ctx := context.Background() serverEnable := false - logger.Start(serverEnable, debugEnable, silentEnable, silentEnable) - defer logger.Stop() + logger.Start(ctx, serverEnable, debugEnable, silentEnable, silentEnable) if pprofEnable || config.Common.PProfEnable { pprof.Start() @@ -69,9 +69,8 @@ func main() { ServersStr: serversStr, Discovery: discovery, UserName: userName, - Files: files, + What: files, TrustAllHosts: trustAllHosts, - PingTimeout: pingTimeoutS, Regex: regex, } @@ -79,5 +78,8 @@ func main() { if err != nil { panic(err) } - client.Start() + + status := client.Start(ctx) + logger.Flush() + os.Exit(status) } diff --git a/cmd/dmap/main.go b/cmd/dmap/main.go index 83dad50..f3f706a 100644 --- a/cmd/dmap/main.go +++ b/cmd/dmap/main.go @@ -1,13 +1,15 @@ package main import ( + "context" "flag" + "os" - "github.com/mimecast/dtail/internal/omode" "github.com/mimecast/dtail/internal/clients" "github.com/mimecast/dtail/internal/color" "github.com/mimecast/dtail/internal/config" - "github.com/mimecast/dtail/internal/logger" + "github.com/mimecast/dtail/internal/io/logger" + "github.com/mimecast/dtail/internal/omode" "github.com/mimecast/dtail/internal/pprof" "github.com/mimecast/dtail/internal/user" "github.com/mimecast/dtail/internal/version" @@ -29,7 +31,6 @@ func main() { var sshPort int var trustAllHosts bool - pingTimeoutS := 900 userName := user.Name() flag.BoolVar(&debugEnable, "debug", false, "Activate debug messages") @@ -39,7 +40,6 @@ func main() { flag.BoolVar(&silentEnable, "silent", false, "Reduce output") flag.BoolVar(&trustAllHosts, "trustAllHosts", false, "Auto trust all unknown host keys") flag.IntVar(&connectionsPerCPU, "cpc", 10, "How many connections established per CPU core concurrently") - flag.IntVar(&pingTimeoutS, "pingTimeout", 10, "The server ping timeout (0 means disable pings)") flag.IntVar(&sshPort, "port", 2222, "SSH server port") flag.StringVar(&cfgFile, "cfg", "", "Config file path") flag.StringVar(&discovery, "discovery", "", "Server discovery method") @@ -57,10 +57,10 @@ func main() { version.PrintAndExit() } + ctx := context.Background() serverEnable := false - logger.Start(serverEnable, debugEnable, silentEnable, silentEnable) - defer logger.Stop() + logger.Start(ctx, serverEnable, debugEnable, silentEnable, silentEnable) if pprofEnable || config.Common.PProfEnable { pprof.Start() } @@ -70,9 +70,8 @@ func main() { ServersStr: serversStr, Discovery: discovery, UserName: userName, - Files: files, + What: files, TrustAllHosts: trustAllHosts, - PingTimeout: pingTimeoutS, Mode: omode.MapClient, } @@ -81,5 +80,7 @@ func main() { panic(err) } - client.Start() + status := client.Start(ctx) + logger.Flush() + os.Exit(status) } diff --git a/cmd/drun/main.go b/cmd/drun/main.go new file mode 100644 index 0000000..b1936d4 --- /dev/null +++ b/cmd/drun/main.go @@ -0,0 +1,82 @@ +package main + +import ( + "context" + "flag" + "os" + + "github.com/mimecast/dtail/internal/clients" + "github.com/mimecast/dtail/internal/color" + "github.com/mimecast/dtail/internal/config" + "github.com/mimecast/dtail/internal/io/logger" + "github.com/mimecast/dtail/internal/pprof" + "github.com/mimecast/dtail/internal/user" + "github.com/mimecast/dtail/internal/version" +) + +// The evil begins here. +func main() { + var cfgFile string + var command string + var connectionsPerCPU int + var debugEnable bool + var discovery string + var displayVersion bool + var noColor bool + var pprofEnable bool + var serversStr string + var silentEnable bool + var sshPort int + var trustAllHosts bool + + userName := user.Name() + + flag.BoolVar(&debugEnable, "debug", false, "Activate debug messages") + flag.BoolVar(&displayVersion, "version", false, "Display version") + flag.BoolVar(&noColor, "noColor", false, "Disable ANSII terminal colors") + flag.BoolVar(&pprofEnable, "pprofEnable", false, "Enable pprof server") + flag.BoolVar(&silentEnable, "silent", false, "Reduce output") + flag.BoolVar(&trustAllHosts, "trustAllHosts", false, "Auto trust all unknown host keys") + flag.IntVar(&connectionsPerCPU, "cpc", 10, "How many connections established per CPU core concurrently") + flag.IntVar(&sshPort, "port", 2222, "SSH server port") + flag.StringVar(&cfgFile, "cfg", "", "Config file path") + flag.StringVar(&command, "command", "", "Command to run") + flag.StringVar(&discovery, "discovery", "", "Server discovery method") + flag.StringVar(&serversStr, "servers", "", "Remote servers to connect") + flag.StringVar(&userName, "user", userName, "Your system user name") + + flag.Parse() + + config.Read(cfgFile, sshPort) + color.Colored = !noColor + + if displayVersion { + version.PrintAndExit() + } + + ctx := context.Background() + serverEnable := false + + logger.Start(ctx, serverEnable, debugEnable, silentEnable, silentEnable) + if pprofEnable || config.Common.PProfEnable { + pprof.Start() + } + + args := clients.Args{ + ConnectionsPerCPU: connectionsPerCPU, + ServersStr: serversStr, + Discovery: discovery, + UserName: userName, + What: command, + TrustAllHosts: trustAllHosts, + } + + client, err := clients.NewRunClient(args) + if err != nil { + panic(err) + } + + status := client.Start(ctx) + logger.Flush() + os.Exit(status) +} diff --git a/cmd/dserver/main.go b/cmd/dserver/main.go index 489910b..aa209a8 100644 --- a/cmd/dserver/main.go +++ b/cmd/dserver/main.go @@ -1,13 +1,14 @@ package main import ( + "context" "flag" "os" "time" "github.com/mimecast/dtail/internal/color" "github.com/mimecast/dtail/internal/config" - "github.com/mimecast/dtail/internal/logger" + "github.com/mimecast/dtail/internal/io/logger" "github.com/mimecast/dtail/internal/pprof" "github.com/mimecast/dtail/internal/server" "github.com/mimecast/dtail/internal/user" @@ -24,7 +25,7 @@ func main() { var shutdownAfter int var sshPort int - userName := user.Name() + user.NoRootCheck() flag.BoolVar(&debugEnable, "debug", false, "Activate debug messages") flag.BoolVar(&displayVersion, "version", false, "Display version") @@ -43,19 +44,23 @@ func main() { version.PrintAndExit() } + ctx := context.Background() + serverEnable := true silentEnable := false nothingEnable := false - logger.Start(serverEnable, debugEnable, silentEnable, nothingEnable) - defer logger.Stop() + logger.Start(ctx, serverEnable, debugEnable, silentEnable, nothingEnable) if shutdownAfter > 0 { go func() { defer os.Exit(1) logger.Info("Enabling auto shutdown timer", shutdownAfter) - time.Sleep(time.Duration(shutdownAfter) * time.Second) - logger.Info("Auto shutdown timer reached, shutting down now") + select { + case <-time.After(time.Duration(shutdownAfter) * time.Second): + logger.Info("Auto shutdown timer reached, shutting down now") + case <-ctx.Done(): + } }() } @@ -63,7 +68,8 @@ func main() { pprof.Start() } - logger.Info("Launching server", version.String(), userName) sshServer := server.New() - sshServer.Start() + status := sshServer.Start(ctx) + logger.Flush() + os.Exit(status) } diff --git a/cmd/dtail/main.go b/cmd/dtail/main.go index 1bf77c7..76070ff 100644 --- a/cmd/dtail/main.go +++ b/cmd/dtail/main.go @@ -1,13 +1,14 @@ package main import ( + "context" "flag" "os" "github.com/mimecast/dtail/internal/clients" "github.com/mimecast/dtail/internal/color" "github.com/mimecast/dtail/internal/config" - "github.com/mimecast/dtail/internal/logger" + "github.com/mimecast/dtail/internal/io/logger" "github.com/mimecast/dtail/internal/omode" "github.com/mimecast/dtail/internal/pprof" "github.com/mimecast/dtail/internal/user" @@ -32,7 +33,6 @@ func main() { var sshPort int var trustAllHosts bool - pingTimeoutS := 5 userName := user.Name() flag.BoolVar(&checkHealth, "checkHealth", false, "Only check for server health") @@ -43,7 +43,6 @@ func main() { flag.BoolVar(&silentEnable, "silent", false, "Reduce output") flag.BoolVar(&trustAllHosts, "trustAllHosts", false, "Auto trust all unknown host keys") flag.IntVar(&connectionsPerCPU, "cpc", 10, "How many connections established per CPU core concurrently") - flag.IntVar(&pingTimeoutS, "pingTimeout", 10, "The server ping timeout (0 means disable pings)") flag.IntVar(&sshPort, "port", 2222, "SSH server port") flag.StringVar(&cfgFile, "cfg", "", "Config file path") flag.StringVar(&discovery, "discovery", "", "Server discovery method") @@ -62,17 +61,18 @@ func main() { version.PrintAndExit() } + ctx := context.Background() + if checkHealth { healthClient, _ := clients.NewHealthClient(omode.HealthClient) - os.Exit(healthClient.Start()) + os.Exit(healthClient.Start(ctx)) } serverEnable := false if checkHealth { silentEnable = true } - logger.Start(serverEnable, debugEnable, silentEnable, silentEnable) - defer logger.Stop() + logger.Start(ctx, serverEnable, debugEnable, silentEnable, silentEnable) if pprofEnable || config.Common.PProfEnable { pprof.Start() @@ -83,9 +83,8 @@ func main() { ServersStr: serversStr, Discovery: discovery, UserName: userName, - Files: files, + What: files, TrustAllHosts: trustAllHosts, - PingTimeout: pingTimeoutS, Regex: regex, Mode: omode.TailClient, } @@ -104,5 +103,7 @@ func main() { } } - client.Start() + status := client.Start(ctx) + logger.Flush() + os.Exit(status) } diff --git a/doc/examples.md b/doc/examples.md index 959105c..964660a 100644 --- a/doc/examples.md +++ b/doc/examples.md @@ -25,7 +25,7 @@ To run ad-hoc mapreduce aggregations on newly written log lines you also must ad --files '/var/log/service/*.log' ``` -In order for mapreduce queries to work you have to make sure that your log format is supported by DTail. You can either use the ones which are already defined in ``mapr/logformat`` or add an extension to support a custom log format. +In order for mapreduce queries to work you have to make sure that your log format is supported by DTail. You can either use the ones which are already defined in ``internal/mapr/logformat`` or add an extension to support a custom log format. ![dtail-map](dtail-map.gif "Tail mapreduce example") @@ -62,6 +62,6 @@ To run a mapreduce aggregation over logs written in the past the ``dmap`` comman --files "/var/log/service/*.log" ``` -Remember: In order for that to work you have to make sure that your log format is supported by DTail. You can either use the ones which are already defined in ``mapr/logformat`` or add an extension to support a custom log format. +Remember: In order for that to work you have to make sure that your log format is supported by DTail. You can either use the ones which are already defined in ``internal/mapr/logformat`` or add an extension to support a custom log format. ![dmap](dmap.gif "DMap example") diff --git a/doc/installation.md b/doc/installation.md index 305eae5..a15beb1 100644 --- a/doc/installation.md +++ b/doc/installation.md @@ -3,8 +3,6 @@ DTail Installation Guide The following installation guide has been tested successfully on CentOS 7. You may need to adjust accordingly depending on the distribution you use. -This guide also assumes that you know how to use ``systemd`` and how to configure a service there. If you are unsure please consult the documentation of your distribution. - # Compile it Please check the [Quick Starting Guide](quickstart.md) for instructions how to compile DTail. It is recommended to automate the build process via your build pipeline (e.g. produce a deployable RPM via Jenkins). You don't have to use ``go get...`` to compile and install the binaries. You can also clone the repository and use ``make`` instead. @@ -12,6 +10,7 @@ Please check the [Quick Starting Guide](quickstart.md) for instructions how to c # Install it It is recommended to automate all the installation process outlined here. You could use a configuration management system such as Puppet, Chef or Ansible. However, that relies heavily on how your infrastructure is managed and is out of scope of this documentation. + 1. The ``dserver`` binary has to be installed on all machines (server boxes) involved. A good location for the binary would be ``/usr/local/bin/dserver`` with permissions set as follows: ```console @@ -72,7 +71,7 @@ Now you should be able to use DTail client like outlined in the [Quick Starting # Monitor it -To verify that DTail server is up and running and functioning as expected you should configure the Nagios check [check_dserver.sh](../samples/check_dserver.sh.sample) in your monitoring system. The check has to be executed locally on the server (e.g. via NRPE). How to configure the monitoring system in detail is out of scope of this guide, as it depends on the monitoring infrastructure used. +To verify that DTail server is up and running and functioning as expected you should configure the Nagios check [check_dserver.sh](../samples/check_dserver.sh.sample) in your monitoring system. The check has to be executed locally on the server (e.g. via NRPE). How to configure the monitoring system in detail is out of scope of this guide. ```console % ./check_dserver.sh diff --git a/doc/quickstart.md b/doc/quickstart.md index 46f7fae..7b6fbf4 100644 --- a/doc/quickstart.md +++ b/doc/quickstart.md @@ -3,13 +3,11 @@ Quick Starting Guide This is the quick starting guide. For a more sustainable setup, involving how to create a background service via ``systemd``, recommendations about automation via Jenkins and/or Puppet and health monitoring via Nagios please also follow the [Installation Guide](installation.md). -This guide assumes that you know how to generate and configure a public/private SSH key pair for secure authorization and shell access. That is out of scope of this guide. For more information please have a look at the OpenSSH documentation of your distribution. +This guide assumes that you know how to generate and configure a public/private SSH key pair for secure authorization and shell access. For more information please have a look at the OpenSSH documentation of your distribution. -This guide also assumes that you know how to install and use a Go compiler and GNU make. +# Install it -# Compile it - -To install all DTail binaries from github run: +To compile and install all DTail binaries directly from GitHub run: ```console % go get github.com/mimecast/dtail/cmd/dtail @@ -48,7 +46,9 @@ SERVER|serv-001|INFO|Binding server|0.0.0.0:2222 Make sure that your public SSH key is listed in ``~/.ssh/authorized_keys`` on all server machines involved. The private SSH key counterpart should preferably stay on your Laptop or workstation in ``~/.ssh/id_rsa`` or ``~/.ssh/id_dsa``. -DTail utilises the SSH Agent for SSH authentication. This is to avoid entering the passphrase of the private SSH key over and over again when a new SSH session is initiated from the DTail client to a new DTail server. For this the private SSH key has to be registered at the SSH Agent: +DTail relies on SSH for secure authentication and communication. The clients (all client binaries such as ``dtail``, ``dgrep`` and so on...) communicate with an auth backend via the SSH auth socket. The SSH auth socket is configured via the environment variable ``SSH_AUTH_SOCK`` which usually points to ``~/.ssh/ssh_auth_socket`` or similar (depending on your configuration it may also point to other auth backends such as GPG Agent, in which case ``SSH_AUTH_SOCK`` would point to ``~/.gnupg/S.gpg-agent.ssh`` or similar). + +Usually you would use the SSH Auth Agent. For this the private SSH key has to be registered at the SSH Agent: ```console % ssh-add ~/.ssh/id_rsa @@ -56,16 +56,17 @@ Enter passphrase for ~/.ssh/id_rsa: ********** Identity added: ~/.ssh/id_rsa (~/.ssh/id_rsa) ``` -The DTail client communicates with the SSH Agent through ``~/.ssh/ssh_auth_socket`` whenever a new session to a DTail server is established. - To test whether SSH is setup correctly you should be able to SSH into the servers with the OpenSSH client and your private SSH key through the SSH Agent without entering the private keys passphrase. The following assumes to have an OpenSSH server running on ``serv-001.lan.example.org`` and an OpenSSH client installed on your laptop or workstation. Please notice that DTail does not require to have an OpenSSH infrastructure set up but DTail uses by default the same public/private key file paths as OpenSSH. OpenSSH can be of a great help to verify that the SSH keys are configured correctly: ```console -% ssh serv-001.lan.example.org -% -% exit +workstation01 ~ % ssh serv-001.lan.example.org +serv-001 ~ % +serv-001 ~ % exit +workstation01 ~ % ``` +Please consult the OpenSSH documentation of your distribution if the test above does not work for you. + ## Run DTail client Now it is time to connect to the DTail servers through the DTail client: diff --git a/go.mod b/go.mod index bba791a..9a50e44 100644 --- a/go.mod +++ b/go.mod @@ -5,4 +5,5 @@ go 1.13 require ( github.com/DataDog/zstd v1.4.4 golang.org/x/crypto v0.0.0-20191227163750-53104e6ec876 + golang.org/x/lint v0.0.0-20200130185559-910be7a94367 // indirect ) diff --git a/go.sum b/go.sum index cbb61a3..39af4c9 100644 --- a/go.sum +++ b/go.sum @@ -1,10 +1,19 @@ github.com/DataDog/zstd v1.4.4 h1:+IawcoXhCBylN7ccwdwf8LOH2jKq7NavGpEPanrlTzE= github.com/DataDog/zstd v1.4.4/go.mod h1:1jcaCB/ufaK+sKp1NBhlGmpz41jOoPQ35bpF36t7BBo= golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w= +golang.org/x/crypto v0.0.0-20191011191535-87dc89f01550/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI= golang.org/x/crypto v0.0.0-20191227163750-53104e6ec876 h1:sKJQZMuxjOAR/Uo2LBfU90onWEf1dF4C+0hPJCc9Mpc= golang.org/x/crypto v0.0.0-20191227163750-53104e6ec876/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto= +golang.org/x/lint v0.0.0-20200130185559-910be7a94367 h1:0IiAsCRByjO2QjX7ZPkw5oU9x+n1YqRL802rjC0c3Aw= +golang.org/x/lint v0.0.0-20200130185559-910be7a94367/go.mod h1:3xt1FjdF8hUf6vQPIChWIBhFzV8gjjsPE/fR3IyQdNY= +golang.org/x/mod v0.1.1-0.20191105210325-c90efee705ee/go.mod h1:QqPTAvyqsEbceGzBzNggFXnrqF1CaUcvgkdR5Ot7KZg= golang.org/x/net v0.0.0-20190404232315-eb5bcb51f2a3/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg= +golang.org/x/net v0.0.0-20190620200207-3b0461eec859/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s= +golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY= golang.org/x/sys v0.0.0-20190412213103-97732733099d h1:+R4KGOnez64A81RvjARKc4UT5/tI9ujCIVX+P5KiHuI= golang.org/x/sys v0.0.0-20190412213103-97732733099d/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ= +golang.org/x/tools v0.0.0-20200130002326-2f3ba24bd6e7 h1:EBZoQjiKKPaLbPrbpssUfuHtwM6KV/vb4U85g/cigFY= +golang.org/x/tools v0.0.0-20200130002326-2f3ba24bd6e7/go.mod h1:TB2adYChydJhpapKDTa4BR/hXlZSLoq2Wpct/0txZ28= +golang.org/x/xerrors v0.0.0-20191011141410-1b5146add898/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= diff --git a/internal/clients/args.go b/internal/clients/args.go index 5fe0a72..dea5a9e 100644 --- a/internal/clients/args.go +++ b/internal/clients/args.go @@ -9,10 +9,9 @@ type Args struct { Mode omode.Mode ServersStr string UserName string - Files string + What string Regex string TrustAllHosts bool Discovery string ConnectionsPerCPU int - PingTimeout int } diff --git a/internal/clients/baseclient.go b/internal/clients/baseclient.go index 574ae94..b1540ea 100644 --- a/internal/clients/baseclient.go +++ b/internal/clients/baseclient.go @@ -1,13 +1,14 @@ package clients import ( + "context" "regexp" "sync" "time" "github.com/mimecast/dtail/internal/clients/remote" "github.com/mimecast/dtail/internal/discovery" - "github.com/mimecast/dtail/internal/logger" + "github.com/mimecast/dtail/internal/io/logger" "github.com/mimecast/dtail/internal/omode" "github.com/mimecast/dtail/internal/ssh/client" @@ -27,111 +28,110 @@ type baseClient struct { sshAuthMethods []gossh.AuthMethod // To deal with SSH host keys hostKeyCallback *client.HostKeyCallback - // To stop the client. - stop chan struct{} - // To indicate that the client has stopped. - stopped chan struct{} // Throttle how fast we initiate SSH connections concurrently throttleCh chan struct{} // Retry connection upon failure? retry bool - // Connection helper. - maker connectionMaker + // Connection maker helper. + maker maker } -func (c *baseClient) init(maker connectionMaker) { +func (c *baseClient) init(maker maker) { logger.Info("Initiating base client") c.maker = maker - //c.connections = make(map[string]*remote.Connection) c.sshAuthMethods, c.hostKeyCallback = client.InitSSHAuthMethods(c.TrustAllHosts, c.throttleCh) + discoveryService := discovery.New(c.Discovery, c.ServersStr, discovery.Shuffle) - // Retrieve a shuffled list of remote dtail servers. - shuffleServers := true - discoveryService := discovery.New(c.Discovery, c.ServersStr, shuffleServers) for _, server := range discoveryService.ServerList() { - c.connections = append(c.connections, c.maker.makeConnection(server, c.sshAuthMethods, c.hostKeyCallback)) + c.connections = append(c.connections, c.makeConnection(server, c.sshAuthMethods, c.hostKeyCallback)) } if _, err := regexp.Compile(c.Regex); err != nil { logger.FatalExit(c.Regex, "Can't test compile regex", err) } - // Periodically check for unknown hosts, and ask the user whether to trust them or not. - go c.hostKeyCallback.PromptAddHosts(c.stop) - - // Periodically print out connection stats to the client. c.stats = newTailStats(len(c.connections)) - go c.stats.periodicLogStats(c.throttleCh, c.stop) } -func (c *baseClient) Start() (status int) { +func (c *baseClient) Start(ctx context.Context) (status int) { + // Periodically check for unknown hosts, and ask the user whether to trust them or not. + go c.hostKeyCallback.PromptAddHosts(ctx) + // Periodically print out connection stats to the client. + go c.stats.periodicLogStats(ctx, c.throttleCh) + // Keep count of active connections active := make(chan struct{}, len(c.connections)) - var wg sync.WaitGroup - wg.Add(len(c.connections)) - + var mutex sync.Mutex for i, conn := range c.connections { go func(i int, conn *remote.Connection) { - active <- struct{}{} - defer func() { - logger.Debug(conn.Server, "Disconnected completely...") - <-active - }() - wg.Done() - - for { - conn.Start(c.throttleCh, c.stats.connectionsEstCh) - if !c.retry { - return - } - time.Sleep(time.Second * 2) - logger.Debug(conn.Server, "Reconencting") - conn = c.maker.makeConnection(conn.Server, c.sshAuthMethods, c.hostKeyCallback) - c.connections[i] = conn + connStatus := c.start(ctx, active, i, conn) + + // Update global status. + mutex.Lock() + defer mutex.Unlock() + if connStatus > status { + status = connStatus } }(i, conn) } - wg.Wait() - c.waitUntilDone(active) - + c.waitUntilDone(ctx, active) return } -func (c *baseClient) waitUntilDone(active chan struct{}) { - defer close(c.stopped) +func (c *baseClient) start(ctx context.Context, active chan struct{}, i int, conn *remote.Connection) (status int) { + // Increment connection count + active <- struct{}{} + // Derement connection count + defer func() { <-active }() - if c.Mode != omode.TailClient { - c.waitUntilZero(active) - logger.Info("All connections stopped") - return - } + for { + connCtx, cancel := conn.Handler.WithCancel(ctx) + defer cancel() - <-c.stop - logger.Info("Stopping client") - for _, conn := range c.connections { - conn.Stop() + conn.Start(connCtx, cancel, c.throttleCh, c.stats.connectionsEstCh) + // Retrieve status code from handler (dtail client will exit with that status) + status = conn.Handler.Status() + + if !c.retry { + return + } + + time.Sleep(time.Second * 2) + logger.Debug(conn.Server, "Reconnecting") + + conn = c.makeConnection(conn.Server, c.sshAuthMethods, c.hostKeyCallback) + c.connections[i] = conn } +} - c.waitUntilZero(active) +func (c *baseClient) makeConnection(server string, sshAuthMethods []gossh.AuthMethod, hostKeyCallback *client.HostKeyCallback) *remote.Connection { + conn := remote.NewConnection(server, c.UserName, sshAuthMethods, hostKeyCallback) + conn.Handler = c.maker.makeHandler(server) + conn.Commands = c.maker.makeCommands() + + return conn } -func (c *baseClient) waitUntilZero(active chan struct{}) { +func (c *baseClient) waitUntilDone(ctx context.Context, active chan struct{}) { + defer logger.Info("Terminated connection") + + // We want to have at least one active connection + <-active + // Put it back on the channel + active <- struct{}{} + + if c.Mode == omode.TailClient { + <-ctx.Done() + } + for { - logger.Debug("Active connections", len(active)) - if len(active) == 0 { + numActive := len(active) + if numActive == 0 { return } + logger.Debug("Active connections", numActive) time.Sleep(time.Second) } } - -func (c *baseClient) Stop() { - close(c.stop) - <-c.WaitC() -} - -func (c *baseClient) WaitC() <-chan struct{} { - return c.stopped -} diff --git a/internal/clients/catclient.go b/internal/clients/catclient.go index 5ea701d..7fd6bdc 100644 --- a/internal/clients/catclient.go +++ b/internal/clients/catclient.go @@ -7,11 +7,7 @@ import ( "strings" "github.com/mimecast/dtail/internal/clients/handlers" - "github.com/mimecast/dtail/internal/clients/remote" "github.com/mimecast/dtail/internal/omode" - "github.com/mimecast/dtail/internal/ssh/client" - - gossh "golang.org/x/crypto/ssh" ) // CatClient is a client for returning a whole file from the beginning to the end. @@ -31,8 +27,6 @@ func NewCatClient(args Args) (*CatClient, error) { c := CatClient{ baseClient: baseClient{ Args: args, - stop: make(chan struct{}), - stopped: make(chan struct{}), throttleCh: make(chan struct{}, args.ConnectionsPerCPU*runtime.NumCPU()), retry: false, }, @@ -43,11 +37,13 @@ func NewCatClient(args Args) (*CatClient, error) { return &c, nil } -func (c CatClient) makeConnection(server string, sshAuthMethods []gossh.AuthMethod, hostKeyCallback *client.HostKeyCallback) *remote.Connection { - conn := remote.NewConnection(server, c.UserName, sshAuthMethods, hostKeyCallback) - conn.Handler = handlers.NewClientHandler(server, c.PingTimeout) - for _, file := range strings.Split(c.Files, ",") { - conn.Commands = append(conn.Commands, fmt.Sprintf("%s %s regex %s", c.Mode.String(), file, c.Regex)) +func (c CatClient) makeHandler(server string) handlers.Handler { + return handlers.NewClientHandler(server) +} + +func (c CatClient) makeCommands() (commands []string) { + for _, file := range strings.Split(c.What, ",") { + commands = append(commands, fmt.Sprintf("%s %s regex %s", c.Mode.String(), file, c.Regex)) } - return conn + return } diff --git a/internal/clients/client.go b/internal/clients/client.go index 85d1aae..1fc5e23 100644 --- a/internal/clients/client.go +++ b/internal/clients/client.go @@ -1,7 +1,8 @@ package clients +import "context" + // Client is the interface for the end user command line client. type Client interface { - Start() int - Stop() + Start(ctx context.Context) int } diff --git a/internal/clients/connectionmaker.go b/internal/clients/connectionmaker.go deleted file mode 100644 index 0617992..0000000 --- a/internal/clients/connectionmaker.go +++ /dev/null @@ -1,12 +0,0 @@ -package clients - -import ( - "github.com/mimecast/dtail/internal/clients/remote" - "github.com/mimecast/dtail/internal/ssh/client" - - gossh "golang.org/x/crypto/ssh" -) - -type connectionMaker interface { - makeConnection(server string, sshAuthMethods []gossh.AuthMethod, hostKeyCallback *client.HostKeyCallback) *remote.Connection -} diff --git a/internal/clients/execclient.go b/internal/clients/execclient.go deleted file mode 100644 index 10bd081..0000000 --- a/internal/clients/execclient.go +++ /dev/null @@ -1,48 +0,0 @@ -package clients - -import ( - "fmt" - "runtime" - "strings" - - "github.com/mimecast/dtail/internal/clients/handlers" - "github.com/mimecast/dtail/internal/clients/remote" - "github.com/mimecast/dtail/internal/omode" - "github.com/mimecast/dtail/internal/ssh/client" - - gossh "golang.org/x/crypto/ssh" -) - -// ExecClient is a client for execute various commands on the server. -type ExecClient struct { - baseClient -} - -// NewExecClient returns a new cat client. -func NewExecClient(args Args) (*ExecClient, error) { - args.Regex = "." - args.Mode = omode.ExecClient - - c := ExecClient{ - baseClient: baseClient{ - Args: args, - stop: make(chan struct{}), - stopped: make(chan struct{}), - throttleCh: make(chan struct{}, args.ConnectionsPerCPU*runtime.NumCPU()), - retry: false, - }, - } - - c.init(c) - - return &c, nil -} - -func (c ExecClient) makeConnection(server string, sshAuthMethods []gossh.AuthMethod, hostKeyCallback *client.HostKeyCallback) *remote.Connection { - conn := remote.NewConnection(server, c.UserName, sshAuthMethods, hostKeyCallback) - conn.Handler = handlers.NewClientHandler(server, c.PingTimeout) - for _, file := range strings.Split(c.Files, ";") { - conn.Commands = append(conn.Commands, fmt.Sprintf("%s %s", c.Mode.String(), file)) - } - return conn -} diff --git a/internal/clients/grepclient.go b/internal/clients/grepclient.go index c568f63..8d11458 100644 --- a/internal/clients/grepclient.go +++ b/internal/clients/grepclient.go @@ -7,11 +7,7 @@ import ( "strings" "github.com/mimecast/dtail/internal/clients/handlers" - "github.com/mimecast/dtail/internal/clients/remote" "github.com/mimecast/dtail/internal/omode" - "github.com/mimecast/dtail/internal/ssh/client" - - gossh "golang.org/x/crypto/ssh" ) // GrepClient searches a remote file for all lines matching a regular expression. Only the matching lines are displayed. @@ -29,8 +25,6 @@ func NewGrepClient(args Args) (*GrepClient, error) { c := GrepClient{ baseClient: baseClient{ Args: args, - stop: make(chan struct{}), - stopped: make(chan struct{}), throttleCh: make(chan struct{}, args.ConnectionsPerCPU*runtime.NumCPU()), retry: false, }, @@ -41,13 +35,13 @@ func NewGrepClient(args Args) (*GrepClient, error) { return &c, nil } -func (c GrepClient) makeConnection(server string, sshAuthMethods []gossh.AuthMethod, hostKeyCallback *client.HostKeyCallback) *remote.Connection { - conn := remote.NewConnection(server, c.UserName, sshAuthMethods, hostKeyCallback) - conn.Handler = handlers.NewClientHandler(server, c.PingTimeout) +func (c GrepClient) makeHandler(server string) handlers.Handler { + return handlers.NewClientHandler(server) +} - for _, file := range strings.Split(c.Files, ",") { - conn.Commands = append(conn.Commands, fmt.Sprintf("%s %s regex %s", c.Mode.String(), file, c.Regex)) +func (c GrepClient) makeCommands() (commands []string) { + for _, file := range strings.Split(c.What, ",") { + commands = append(commands, fmt.Sprintf("%s %s regex %s", c.Mode.String(), file, c.Regex)) } - - return conn + return } diff --git a/internal/clients/handlers/basehandler.go b/internal/clients/handlers/basehandler.go index 19246f9..68b8ddc 100644 --- a/internal/clients/handlers/basehandler.go +++ b/internal/clients/handlers/basehandler.go @@ -1,60 +1,44 @@ package handlers import ( - "github.com/mimecast/dtail/internal/logger" - "errors" + "encoding/base64" "fmt" "io" + "strconv" "strings" "time" + + "github.com/mimecast/dtail/internal/io/logger" + "github.com/mimecast/dtail/internal/version" ) type baseHandler struct { + withCancel server string shellStarted bool commands chan string - pong chan struct{} receiveBuf []byte - stop chan struct{} - pingTimeout int + status int } func (h *baseHandler) Server() string { return h.server } -// Used to determine whether server is still responding to requests or not. -func (h *baseHandler) Ping() error { - if h.pingTimeout == 0 { - // Server ping disabled - return nil - } - - if err := h.SendCommand("ping"); err != nil { - return err - } - - select { - case <-h.pong: - return nil - case <-time.After(time.Duration(h.pingTimeout) * time.Second): - } - - return errors.New("Didn't receive any server pongs (ping replies)") +func (h *baseHandler) Status() int { + return h.status } -func (h *baseHandler) SendCommand(command string) error { - if command == "ping" { - logger.Trace("Sending command", h.server, command) - } else { - logger.Debug("Sending command", h.server, command) - } +// 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) select { - case h.commands <- fmt.Sprintf("%s;", command): + case h.commands <- fmt.Sprintf("protocol %s base64 %v;", version.ProtocolCompat, encoded): case <-time.After(time.Second * 5): - return errors.New("Timed out sending command " + command) - case <-h.stop: + return fmt.Errorf("Timed out sending command '%s' (base64: '%s')", command, encoded) + case <-h.ctx.Done(): } return nil @@ -81,7 +65,7 @@ func (h *baseHandler) Read(p []byte) (n int, err error) { select { case command := <-h.commands: n = copy(p, []byte(command)) - case <-h.stop: + case <-h.ctx.Done(): return 0, io.EOF } return @@ -92,6 +76,7 @@ func (h *baseHandler) handleMessageType(message string) { if len(h.receiveBuf) == 0 { return } + // Hidden server commands starti with a dot "." if h.receiveBuf[0] == '.' { h.handleHiddenMessage(message) @@ -108,6 +93,7 @@ func (h *baseHandler) handleMessageType(message string) { h.receiveBuf = h.receiveBuf[:0] return } + logger.Raw(message) h.receiveBuf = h.receiveBuf[:0] } @@ -116,19 +102,27 @@ func (h *baseHandler) handleMessageType(message string) { // to the end user. func (h *baseHandler) handleHiddenMessage(message string) { switch { - case strings.HasPrefix(message, ".pong"): - h.pong <- struct{}{} case strings.HasPrefix(message, ".syn close connection"): - h.SendCommand("ack close connection") - } -} + h.SendMessage(".ack close connection") + select { + case <-time.After(time.Second * 1): + logger.Debug("Shutting down client after timeout and sending ack to server") + h.withCancel.shutdown() + case <-h.ctx.Done(): + } -// Stop the handler. -func (h *baseHandler) Stop() { - select { - case <-h.stop: - default: - logger.Debug("Stopping base handler", h.server) - close(h.stop) + case strings.HasPrefix(message, ".run exitstatus"): + splitted := strings.Split(strings.TrimSuffix(message, "\n"), " ") + if len(splitted) != 3 { + logger.Error("Unable to retrieve exitstatus", message) + return + } + i, err := strconv.Atoi(splitted[2]) + if err != nil { + logger.Error("Unable to retrieve exitstatus", message, err) + return + } + h.status = i + logger.Debug("Retrieved exitstatus", h.status) } } diff --git a/internal/clients/handlers/clienthandler.go b/internal/clients/handlers/clienthandler.go index 4738cd3..fcd8052 100644 --- a/internal/clients/handlers/clienthandler.go +++ b/internal/clients/handlers/clienthandler.go @@ -1,7 +1,7 @@ package handlers import ( - "github.com/mimecast/dtail/internal/logger" + "github.com/mimecast/dtail/internal/io/logger" ) // ClientHandler is the basic client handler interface. @@ -10,7 +10,7 @@ type ClientHandler struct { } // NewClientHandler creates a new client handler. -func NewClientHandler(server string, pingTimeout int) *ClientHandler { +func NewClientHandler(server string) *ClientHandler { logger.Debug(server, "Creating new client handler") return &ClientHandler{ @@ -18,9 +18,10 @@ func NewClientHandler(server string, pingTimeout int) *ClientHandler { server: server, shellStarted: false, commands: make(chan string), - pong: make(chan struct{}, 1), - stop: make(chan struct{}), - pingTimeout: pingTimeout, + status: -1, + withCancel: withCancel{ + done: make(chan struct{}), + }, }, } } diff --git a/internal/clients/handlers/handler.go b/internal/clients/handlers/handler.go index 2013be0..c53ca34 100644 --- a/internal/clients/handlers/handler.go +++ b/internal/clients/handlers/handler.go @@ -1,12 +1,16 @@ package handlers -import "io" +import ( + "context" + "io" +) // Handler provides all methods which can be run on any client handler. type Handler interface { io.ReadWriter - Ping() error - Stop() - SendCommand(command string) error + SendMessage(command string) error Server() string + Status() int + WithCancel(ctx context.Context) (context.Context, context.CancelFunc) + Done() <-chan struct{} } diff --git a/internal/clients/handlers/healthhandler.go b/internal/clients/handlers/healthhandler.go index 4051e2c..9051015 100644 --- a/internal/clients/handlers/healthhandler.go +++ b/internal/clients/handlers/healthhandler.go @@ -8,6 +8,7 @@ import ( // HealthHandler implements the handler required for health checks. type HealthHandler struct { + withCancel // Buffer of incoming data from server. receiveBuf []byte // To send commands to the server. @@ -16,6 +17,7 @@ type HealthHandler struct { receive chan<- string // The remote server address server string + status int } // NewHealthHandler returns a new health check handler. @@ -24,6 +26,10 @@ func NewHealthHandler(server string, receive chan<- string) *HealthHandler { server: server, receive: receive, commands: make(chan string), + status: -1, + withCancel: withCancel{ + done: make(chan struct{}), + }, } return &h @@ -34,18 +40,13 @@ func (h *HealthHandler) Server() string { return h.server } -// Stop is not of use for health check handler. -func (h *HealthHandler) Stop() { - // Nothing done here. +// Status of the handler. +func (h *HealthHandler) Status() int { + return h.status } -// Ping is not of use for health check handler. -func (h *HealthHandler) Ping() error { - return nil -} - -// SendCommand send a DTail command to the server. -func (h *HealthHandler) SendCommand(command string) error { +// SendMessage sends a DTail command to the server. +func (h *HealthHandler) SendMessage(command string) error { select { case h.commands <- fmt.Sprintf("%s;", command): case <-time.NewTimer(time.Second * 10).C: diff --git a/internal/clients/handlers/maprhandler.go b/internal/clients/handlers/maprhandler.go index d76cdfd..874bb7d 100644 --- a/internal/clients/handlers/maprhandler.go +++ b/internal/clients/handlers/maprhandler.go @@ -1,10 +1,11 @@ package handlers import ( - "github.com/mimecast/dtail/internal/logger" + "strings" + + "github.com/mimecast/dtail/internal/io/logger" "github.com/mimecast/dtail/internal/mapr" "github.com/mimecast/dtail/internal/mapr/client" - "strings" ) // MaprHandler is the handler used on the client side for running mapreduce aggregations. @@ -16,15 +17,16 @@ type MaprHandler struct { } // NewMaprHandler returns a new mapreduce client handler. -func NewMaprHandler(server string, query *mapr.Query, globalGroup *mapr.GlobalGroupSet, pingTimeout int) *MaprHandler { +func NewMaprHandler(server string, query *mapr.Query, globalGroup *mapr.GlobalGroupSet) *MaprHandler { return &MaprHandler{ baseHandler: baseHandler{ server: server, shellStarted: false, commands: make(chan string), - pong: make(chan struct{}, 1), - stop: make(chan struct{}), - pingTimeout: pingTimeout, + status: -1, + withCancel: withCancel{ + done: make(chan struct{}), + }, }, query: query, aggregate: client.NewAggregate(server, query, globalGroup), @@ -65,10 +67,3 @@ func (h *MaprHandler) handleAggregateMessage(message string) { h.aggregate.Aggregate(parts[2:]) logger.Debug("Aggregated aggregate data", h.server, h.count) } - -// Stop stops the mapreduce client handler. -func (h *MaprHandler) Stop() { - logger.Debug("Stopping mapreduce handler", h.server) - h.aggregate.Stop() - h.baseHandler.Stop() -} diff --git a/internal/clients/handlers/withcancel.go b/internal/clients/handlers/withcancel.go new file mode 100644 index 0000000..7c9cf4e --- /dev/null +++ b/internal/clients/handlers/withcancel.go @@ -0,0 +1,24 @@ +package handlers + +import "context" + +type withCancel struct { + ctx context.Context + done chan struct{} +} + +// WithCancel sets and returns the context used. +func (w *withCancel) WithCancel(ctx context.Context) (context.Context, context.CancelFunc) { + cancelCtx, cancel := context.WithCancel(ctx) + w.ctx = cancelCtx + + return cancelCtx, cancel +} + +func (w *withCancel) Done() <-chan struct{} { + return w.done +} + +func (w *withCancel) shutdown() { + close(w.done) +} diff --git a/internal/clients/healthclient.go b/internal/clients/healthclient.go index ff13b83..7313583 100644 --- a/internal/clients/healthclient.go +++ b/internal/clients/healthclient.go @@ -1,6 +1,7 @@ package clients import ( + "context" "fmt" "runtime" "strings" @@ -39,7 +40,7 @@ func NewHealthClient(mode omode.Mode) (*HealthClient, error) { } // Start the health client. -func (c *HealthClient) Start() (status int) { +func (c *HealthClient) Start(ctx context.Context) (status int) { receive := make(chan string) throttleCh := make(chan struct{}, runtime.NumCPU()) @@ -49,8 +50,8 @@ func (c *HealthClient) Start() (status int) { conn.Handler = handlers.NewHealthHandler(c.server, receive) conn.Commands = []string{c.mode.String()} - go conn.Start(throttleCh, statsCh) - defer conn.Stop() + connCtx, cancel := conn.Handler.WithCancel(ctx) + go conn.Start(connCtx, cancel, throttleCh, statsCh) for { select { diff --git a/internal/clients/maker.go b/internal/clients/maker.go new file mode 100644 index 0000000..da9dfc9 --- /dev/null +++ b/internal/clients/maker.go @@ -0,0 +1,8 @@ +package clients + +import "github.com/mimecast/dtail/internal/clients/handlers" + +type maker interface { + makeHandler(server string) handlers.Handler + makeCommands() (commands []string) +} diff --git a/internal/clients/maprclient.go b/internal/clients/maprclient.go index 9070827..b581844 100644 --- a/internal/clients/maprclient.go +++ b/internal/clients/maprclient.go @@ -1,6 +1,7 @@ package clients import ( + "context" "errors" "fmt" "runtime" @@ -8,13 +9,9 @@ import ( "time" "github.com/mimecast/dtail/internal/clients/handlers" - "github.com/mimecast/dtail/internal/clients/remote" - "github.com/mimecast/dtail/internal/logger" + "github.com/mimecast/dtail/internal/io/logger" "github.com/mimecast/dtail/internal/mapr" "github.com/mimecast/dtail/internal/omode" - "github.com/mimecast/dtail/internal/ssh/client" - - gossh "golang.org/x/crypto/ssh" ) // MaprClient is used for running mapreduce aggregations on remote files. @@ -39,8 +36,6 @@ func NewMaprClient(args Args, queryStr string) (*MaprClient, error) { c := MaprClient{ baseClient: baseClient{ Args: args, - stop: make(chan struct{}), - stopped: make(chan struct{}), throttleCh: make(chan struct{}, args.ConnectionsPerCPU*runtime.NumCPU()), retry: args.Mode == omode.TailClient, }, @@ -70,35 +65,36 @@ func NewMaprClient(args Args, queryStr string) (*MaprClient, error) { return &c, nil } -func (c MaprClient) makeConnection(server string, sshAuthMethods []gossh.AuthMethod, hostKeyCallback *client.HostKeyCallback) *remote.Connection { - conn := remote.NewConnection(server, c.UserName, sshAuthMethods, hostKeyCallback) - conn.Handler = handlers.NewMaprHandler(conn.Server, c.query, c.globalGroup, c.PingTimeout) +// Start starts the mapreduce client. +func (c *MaprClient) Start(ctx context.Context) (status int) { + if c.query.Outfile == "" { + // Only print out periodic results if we don't write an outfile + go c.periodicPrintResults(ctx) + } - conn.Commands = append(conn.Commands, fmt.Sprintf("map %s", c.query.RawQuery)) - commandStr := "tail" + status = c.baseClient.Start(ctx) if c.additative { - commandStr = "cat" + c.recievedFinalResult() } - for _, file := range strings.Split(c.Files, ",") { - conn.Commands = append(conn.Commands, fmt.Sprintf("%s %s regex %s", commandStr, file, c.Regex)) - } + return +} - return conn +func (c MaprClient) makeHandler(server string) handlers.Handler { + return handlers.NewMaprHandler(server, c.query, c.globalGroup) } -// Start starts the mapreduce client. -func (c *MaprClient) Start() (status int) { - if c.query.Outfile == "" { - // Only print out periodic results if we don't write an outfile - go c.periodicPrintResults() - } +func (c MaprClient) makeCommands() (commands []string) { + commands = append(commands, fmt.Sprintf("map %s", c.query.RawQuery)) - status = c.baseClient.Start() + modeStr := "tail" if c.additative { - c.recievedFinalResult() + modeStr = "cat" + } + + for _, file := range strings.Split(c.What, ",") { + commands = append(commands, fmt.Sprintf("%s %s regex %s", modeStr, file, c.Regex)) } - c.baseClient.Stop() return } @@ -120,13 +116,13 @@ func (c *MaprClient) recievedFinalResult() { logger.Info(fmt.Sprintf("Wrote final mapreduce result to '%s'", c.query.Outfile)) } -func (c *MaprClient) periodicPrintResults() { +func (c *MaprClient) periodicPrintResults(ctx context.Context) { for { select { case <-time.After(c.query.Interval): logger.Info("Gathering interim mapreduce result") c.printResults() - case <-c.baseClient.stop: + case <-ctx.Done(): return } } diff --git a/internal/clients/remote/connection.go b/internal/clients/remote/connection.go index bfc7bc5..71639b1 100644 --- a/internal/clients/remote/connection.go +++ b/internal/clients/remote/connection.go @@ -1,16 +1,18 @@ package remote import ( - "github.com/mimecast/dtail/internal/clients/handlers" - "github.com/mimecast/dtail/internal/config" - "github.com/mimecast/dtail/internal/logger" - "github.com/mimecast/dtail/internal/ssh/client" + "context" "fmt" "io" "strconv" "strings" "time" + "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/ssh/client" + "golang.org/x/crypto/ssh" ) @@ -30,8 +32,6 @@ type Connection struct { Commands []string // Is it a persistent connection or a one-off? isOneOff bool - // Used to stop the connection - stop chan struct{} // To deal with SSH server host keys hostKeyCallback *client.HostKeyCallback } @@ -48,7 +48,6 @@ func NewConnection(server string, userName string, authMethods []ssh.AuthMethod, HostKeyCallback: hostKeyCallback.Wrap(), Timeout: time.Second * 3, }, - stop: make(chan struct{}), } c.initServerPort(server) @@ -64,7 +63,6 @@ func NewOneOffConnection(server string, userName string, authMethods []ssh.AuthM Auth: authMethods, HostKeyCallback: ssh.InsecureIgnoreHostKey(), }, - stop: make(chan struct{}), isOneOff: true, } @@ -90,39 +88,34 @@ func (c *Connection) initServerPort(server string) { } } -// Start the server connection. Build up SSH session and send some DTail commandc. -func (c *Connection) Start(throttleCh, statsCh chan struct{}) { +// Start the server connection. Build up SSH session and send some DTail commands. +func (c *Connection) Start(ctx context.Context, cancel context.CancelFunc, throttleCh, statsCh chan struct{}) { + // Throttle how many connections can be established concurrently (based on ch length) select { - case <-c.stop: - logger.Info(c.Server, c.port, "Disconnecting client") + case throttleCh <- struct{}{}: + defer func() { <-throttleCh }() + case <-ctx.Done(): return - default: } - // Wait for SSH connection throttler - throttleCh <- struct{}{} - - // Wait until connection has been initiated or an error occured - // during initialization. - throttleStopCh := make(chan struct{}, 2) go func() { - <-throttleStopCh - <-throttleCh - }() + defer cancel() - if err := c.dial(c.Server, c.port, throttleStopCh, statsCh); err != nil { - logger.Warn(c.Server, c.port, err) - throttleStopCh <- struct{}{} + if err := c.dial(ctx, cancel, c.Server, c.port, statsCh); err != nil { + logger.Warn(c.Server, c.port, err) - if c.hostKeyCallback.Untrusted(fmt.Sprintf("%s:%d", c.Server, c.port)) { - logger.Debug("Not trusting host, not trying to re-connect", c.Server, c.port) - return + if c.hostKeyCallback.Untrusted(fmt.Sprintf("%s:%d", c.Server, c.port)) { + logger.Debug("Not trusting host", c.Server, c.port) + return + } } - } + }() + + <-ctx.Done() } // Dail into a new SSH connection. Close connection in case of an error. -func (c *Connection) dial(host string, port int, throttleStopCh, statsCh chan struct{}) error { +func (c *Connection) dial(ctx context.Context, cancel context.CancelFunc, host string, port int, statsCh chan struct{}) error { statsCh <- struct{}{} defer func() { <-statsCh }() @@ -135,11 +128,11 @@ func (c *Connection) dial(host string, port int, throttleStopCh, statsCh chan st } defer client.Close() - return c.session(client, throttleStopCh) + return c.session(ctx, cancel, client) } // Create the SSH session. Close the session in case of an error. -func (c *Connection) session(client *ssh.Client, throttleStopCh chan<- struct{}) error { +func (c *Connection) session(ctx context.Context, cancel context.CancelFunc, client *ssh.Client) error { logger.Debug(c.Server, "session") session, err := client.NewSession() @@ -148,14 +141,10 @@ func (c *Connection) sessi