From 849951be1d1a7ee9f9302006ccb187bf5b4e36f3 Mon Sep 17 00:00:00 2001 From: Paul Buetow Date: Wed, 22 Jul 2026 23:51:18 +0300 Subject: =?UTF-8?q?feat:=20DTail=20fork=20=E2=80=94=20server/client=20feat?= =?UTF-8?q?ure=20development?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Squashed development of the snonux/dtail fork's product code (internal/, cmd/) since diverging from mimecast/dtail. Major areas: - Read/output path: the former "turbo" channel-less path is now the single, default server-side read/output path for cat/grep/tail and MapReduce; the old channel-based path and its config/env toggles were removed. - MapReduce: single aggregate implementation (server + serverless) fed directly by a processor pipeline, with input-exhausted finalization via the shutdown coordinator; high-concurrency and data-race fixes. - Journal source reads (journal:unit.service) via journalctl, Linux-gated behind a journal-v1 capability. - Auth-key fast reconnect: in-memory per-user public-key cache with TTL/max-keys, registered over an authenticated session (AUTHKEY), checked before authorized_keys. - Interactive query reload (--interactive-query) with SESSION START/UPDATE generation boundaries and capability negotiation. - Client-side deadlines: --timeout / --shutdownAfter as context deadlines; follow shutdown handling. - Client logging: diagnostics-only daily log by default, opt-in payload tee via --log-payload. - Numerous correctness fixes (buffer-pool double-recycle races, EOF-sentinel leaks, glob-expansion cap, TOCTOU in CSV parsing) with accompanying unit tests. Co-Authored-By: Claude Opus 4.8 --- cmd/dcat/main.go | 52 +- cmd/dgrep/main.go | 51 +- cmd/dmap/main.go | 54 +- cmd/dserver/main.go | 60 +- cmd/dtail-tools/main.go | 60 + cmd/dtail/main.go | 112 +- cmd/dtail/main_test.go | 66 + cmd/dtailhealth/main.go | 44 +- go.mod | 10 +- go.sum | 16 +- internal/cli/authkeyflags.go | 43 + internal/cli/authkeyflags_test.go | 94 ++ internal/cli/pprof.go | 119 ++ internal/cli/pprof_test.go | 41 + internal/cli/runtime.go | 104 ++ internal/cli/runtime_test.go | 71 ++ internal/clients/baseclient.go | 213 +++- internal/clients/baseclient_retry_test.go | 280 +++++ internal/clients/catclient.go | 24 +- internal/clients/client_benchmark_test.go | 136 +++ internal/clients/connectors/connector.go | 18 + internal/clients/connectors/serverconnection.go | 248 +++- .../clients/connectors/serverconnection_test.go | 880 ++++++++++++++ internal/clients/connectors/serverless.go | 213 +++- internal/clients/connectors/sessiontransport.go | 213 ++++ internal/clients/grepclient.go | 24 +- internal/clients/handlers/basehandler.go | 319 ++++- internal/clients/handlers/basehandler_test.go | 356 ++++++ internal/clients/handlers/clienthandler.go | 15 +- internal/clients/handlers/handler.go | 6 + internal/clients/handlers/healthhandler.go | 13 +- internal/clients/handlers/maprhandler.go | 55 +- internal/clients/handlers/maprhandler_test.go | 278 +++++ internal/clients/healthclient.go | 22 +- internal/clients/interactive_control.go | 425 +++++++ internal/clients/interactive_control_test.go | 641 ++++++++++ internal/clients/maker.go | 8 + internal/clients/maprclient.go | 269 ++-- internal/clients/maprclient_test.go | 153 +++ internal/clients/query_regex.go | 30 + internal/clients/runtime_boundary.go | 219 ++++ internal/clients/runtime_boundary_test.go | 89 ++ internal/clients/session_spec.go | 14 + internal/clients/session_spec_test.go | 147 +++ internal/clients/stats.go | 28 +- internal/clients/tailclient.go | 25 +- internal/color/brush/brush.go | 30 +- internal/color/brush/brush_test.go | 90 ++ internal/color/color.go | 6 +- internal/config/args.go | 47 +- internal/config/args_test.go | 95 ++ internal/config/client.go | 41 +- internal/config/common.go | 3 + internal/config/config.go | 13 +- internal/config/env.go | 10 + internal/config/initializer.go | 87 +- internal/config/initializer_test.go | 260 ++++ internal/config/runtime.go | 18 + internal/config/server.go | 89 +- internal/ctxutil/sleep.go | 29 + internal/ctxutil/sleep_test.go | 32 + internal/discovery/comma.go | 15 +- internal/discovery/discovery.go | 34 +- internal/discovery/discovery_test.go | 176 +++ internal/io/dlog/dlog.go | 226 ++-- internal/io/dlog/dlog_test.go | 54 + internal/io/dlog/loggers/file.go | 122 +- internal/io/dlog/loggers/file_test.go | 183 +++ internal/io/dlog/loggers/fout.go | 68 +- internal/io/dlog/loggers/fout_test.go | 199 +++ internal/io/dlog/loggers/stdout.go | 86 +- internal/io/dlog/loggers/stdout_test.go | 190 +++ internal/io/dlog/rawlog_test.go | 76 ++ internal/io/dlog/rotation.go | 26 +- internal/io/dlog/rotation_test.go | 61 + internal/io/fs/catfile.go | 14 +- internal/io/fs/filereader.go | 8 +- internal/io/fs/permissions/permission.go | 1 - internal/io/fs/permissions/permission_linuxacl.go | 1 - internal/io/fs/permissions/permission_test.go | 1 - internal/io/fs/readfile.go | 238 +--- internal/io/fs/readfile_nozstd.go | 16 + internal/io/fs/readfile_processor.go | 359 ++++++ internal/io/fs/readfile_processor_optimized.go | 430 +++++++ internal/io/fs/readfile_processor_test.go | 869 +++++++++++++ internal/io/fs/readfile_zstd.go | 20 + internal/io/fs/readfilelcontext.go | 209 ---- internal/io/fs/rootedpath.go | 96 ++ internal/io/fs/rootedpath_test.go | 57 + internal/io/fs/tailfile.go | 14 +- internal/io/fs/validatedreadtarget.go | 148 +++ internal/io/fs/validatedreadtarget_test.go | 218 ++++ internal/io/journal/filter.go | 253 ++++ internal/io/journal/reader.go | 303 +++++ internal/io/journal/reader_test.go | 754 ++++++++++++ internal/io/journal/reader_unsupported.go | 45 + internal/io/journal/testhelper/mock.go | 397 ++++++ internal/io/journal/testhelper/mock_test.go | 247 ++++ internal/io/line/line.go | 18 +- internal/io/line/processor.go | 22 + internal/io/pool/bytesbuffer.go | 4 +- internal/io/pool/scanner_pool.go | 85 ++ internal/io/signal/signal.go | 45 + internal/mapr/aggregateset.go | 22 +- internal/mapr/client/aggregate.go | 73 +- internal/mapr/client/aggregate_test.go | 96 ++ internal/mapr/client/session_state.go | 95 ++ internal/mapr/client/session_state_test.go | 81 ++ internal/mapr/funcs/function.go | 64 +- internal/mapr/funcs/function_test.go | 114 +- internal/mapr/globalgroupset.go | 11 +- internal/mapr/globalgroupset_test.go | 95 ++ internal/mapr/groupset.go | 133 +- internal/mapr/groupset_avg_nan_test.go | 79 ++ internal/mapr/groupset_ordering_test.go | 125 ++ internal/mapr/groupset_percentage_test.go | 143 +++ internal/mapr/groupsetresult.go | 108 +- internal/mapr/groupsetresult_renderer_test.go | 113 ++ internal/mapr/logformat/csv.go | 97 +- internal/mapr/logformat/csv_test.go | 170 ++- internal/mapr/logformat/custom1.go | 5 +- internal/mapr/logformat/custom2.go | 5 +- internal/mapr/logformat/default.go | 239 +++- internal/mapr/logformat/default_benchmark_test.go | 44 + internal/mapr/logformat/default_test.go | 42 +- internal/mapr/logformat/delimited.go | 12 + internal/mapr/logformat/generic.go | 15 +- internal/mapr/logformat/generickv.go | 36 +- internal/mapr/logformat/mimecast.go | 5 +- internal/mapr/logformat/parser.go | 123 +- internal/mapr/logformat/parser_test.go | 69 ++ internal/mapr/logformat/variables.go | 107 ++ internal/mapr/logformat/variables_test.go | 145 +++ internal/mapr/parserfieldplan.go | 81 ++ internal/mapr/parserfieldplan_test.go | 32 + internal/mapr/query.go | 29 +- internal/mapr/query_test.go | 83 ++ internal/mapr/queryvariables.go | 92 ++ internal/mapr/queryvariables_test.go | 78 ++ internal/mapr/result_renderer.go | 34 + internal/mapr/safe_aggregateset.go | 72 ++ internal/mapr/safe_aggregateset_test.go | 147 +++ internal/mapr/selectcondition.go | 6 + internal/mapr/server/aggregate.go | 620 +++++++--- internal/mapr/server/aggregate_test.go | 714 +++++++++++ internal/mapr/server/groupkey.go | 31 + internal/mapr/server/parsername.go | 10 + internal/mapr/server/parsername_test.go | 62 + internal/mapr/token.go | 35 +- internal/mapr/token_test.go | 77 ++ internal/profiling/README.md | 320 +++++ internal/profiling/flags.go | 38 + internal/profiling/profiler.go | 227 ++++ internal/profiling/profiler_test.go | 269 ++++ internal/protocol/capabilities.go | 12 + internal/protocol/protocol.go | 6 + internal/protocol/session.go | 10 + internal/regex/bench_test.go | 111 ++ internal/regex/regex.go | 128 +- internal/regex/regex_literal_test.go | 226 ++++ internal/regex/regex_test.go | 23 + internal/server/auth_test.go | 50 + internal/server/continuous.go | 61 +- internal/server/continuous_test.go | 202 +++ internal/server/handlers/authkeycommand_test.go | 117 ++ internal/server/handlers/basehandler.go | 547 +++++++-- internal/server/handlers/basehandler_read_test.go | 295 +++++ internal/server/handlers/commandcancel_test.go | 143 +++ internal/server/handlers/commandtimeout_test.go | 150 +++ internal/server/handlers/framesize_test.go | 202 +++ internal/server/handlers/generation_output.go | 75 ++ internal/server/handlers/generation_output_test.go | 160 +++ internal/server/handlers/healthhandler.go | 28 +- internal/server/handlers/line_writer.go | 777 ++++++++++++ internal/server/handlers/line_writer_alloc_test.go | 54 + internal/server/handlers/line_writer_test.go | 1282 ++++++++++++++++++++ internal/server/handlers/lineprocessor.go | 177 +++ internal/server/handlers/lineprocessor_test.go | 82 ++ internal/server/handlers/mapcommand.go | 19 +- .../server/handlers/mapcommand_completion_test.go | 337 +++++ internal/server/handlers/mapcommand_race_test.go | 64 + internal/server/handlers/output_manager.go | 518 ++++++++ .../server/handlers/output_manager_race_test.go | 437 +++++++ internal/server/handlers/protocol_codec.go | 87 ++ internal/server/handlers/protocol_codec_test.go | 32 + internal/server/handlers/protocol_formatter.go | 40 + internal/server/handlers/readcommand.go | 513 +++++++- .../handlers/readcommand_cancellation_test.go | 65 + .../handlers/readcommand_epoch_order_test.go | 127 ++ .../server/handlers/readcommand_glob_cap_test.go | 214 ++++ .../server/handlers/readcommand_journal_test.go | 271 +++++ .../server/handlers/readcommand_semaphore_test.go | 108 ++ .../handlers/readcommand_sendmessage_test.go | 252 ++++ internal/server/handlers/readcommand_server.go | 250 ++++ internal/server/handlers/serverhandler.go | 247 +++- internal/server/handlers/sessioncommand.go | 258 ++++ internal/server/handlers/sessioncommand_test.go | 568 +++++++++ internal/server/handlers/shutdown_coordinator.go | 108 ++ internal/server/scheduler.go | 19 +- internal/server/server.go | 329 +++-- internal/server/stats.go | 63 +- internal/server/stats_test.go | 206 ++++ internal/session/spec.go | 183 +++ internal/session/spec_test.go | 117 ++ internal/ssh/client/authmethods.go | 178 ++- internal/ssh/client/authmethods_test.go | 194 +++ internal/ssh/client/customkeycallback.go | 6 +- internal/ssh/client/hostkeycallback.go | 7 +- internal/ssh/client/knownhostscallback.go | 205 +++- internal/ssh/client/knownhostscallback_test.go | 305 +++++ internal/ssh/client/simplecallback.go | 5 +- internal/ssh/server/authkeystore.go | 186 +++ internal/ssh/server/authkeystore_test.go | 178 +++ internal/ssh/server/hostkey.go | 70 +- internal/ssh/server/hostkey_test.go | 37 + internal/ssh/server/publickeycallback.go | 159 ++- internal/ssh/server/publickeycallback_test.go | 297 +++++ internal/ssh/ssh.go | 145 ++- internal/ssh/ssh_agent_test.go | 152 +++ internal/ssh/ssh_test.go | 113 ++ internal/tools/benchmark/benchmark.go | 385 ++++++ internal/tools/common/data_generator.go | 268 ++++ internal/tools/common/utils.go | 213 ++++ internal/tools/pgo/pgo.go | 1219 +++++++++++++++++++ internal/tools/pgo/pgo_test.go | 132 ++ internal/tools/profile/analyze.go | 221 ++++ internal/tools/profile/profile.go | 367 ++++++ internal/tools/profile/profile_test.go | 30 + internal/user/server/user.go | 122 +- internal/user/server/user_test.go | 203 ++++ internal/version/version.go | 15 +- 231 files changed, 32832 insertions(+), 2049 deletions(-) create mode 100644 cmd/dtail-tools/main.go create mode 100644 cmd/dtail/main_test.go create mode 100644 internal/cli/authkeyflags.go create mode 100644 internal/cli/authkeyflags_test.go create mode 100644 internal/cli/pprof.go create mode 100644 internal/cli/pprof_test.go create mode 100644 internal/cli/runtime.go create mode 100644 internal/cli/runtime_test.go create mode 100644 internal/clients/baseclient_retry_test.go create mode 100644 internal/clients/client_benchmark_test.go create mode 100644 internal/clients/connectors/serverconnection_test.go create mode 100644 internal/clients/connectors/sessiontransport.go create mode 100644 internal/clients/handlers/basehandler_test.go create mode 100644 internal/clients/handlers/maprhandler_test.go create mode 100644 internal/clients/interactive_control.go create mode 100644 internal/clients/interactive_control_test.go create mode 100644 internal/clients/maprclient_test.go create mode 100644 internal/clients/query_regex.go create mode 100644 internal/clients/runtime_boundary.go create mode 100644 internal/clients/runtime_boundary_test.go create mode 100644 internal/clients/session_spec.go create mode 100644 internal/clients/session_spec_test.go create mode 100644 internal/color/brush/brush_test.go create mode 100644 internal/config/args_test.go create mode 100644 internal/config/initializer_test.go create mode 100644 internal/config/runtime.go create mode 100644 internal/ctxutil/sleep.go create mode 100644 internal/ctxutil/sleep_test.go create mode 100644 internal/discovery/discovery_test.go create mode 100644 internal/io/dlog/dlog_test.go create mode 100644 internal/io/dlog/loggers/file_test.go create mode 100644 internal/io/dlog/loggers/fout_test.go create mode 100644 internal/io/dlog/loggers/stdout_test.go create mode 100644 internal/io/dlog/rawlog_test.go create mode 100644 internal/io/dlog/rotation_test.go create mode 100644 internal/io/fs/readfile_nozstd.go create mode 100644 internal/io/fs/readfile_processor.go create mode 100644 internal/io/fs/readfile_processor_optimized.go create mode 100644 internal/io/fs/readfile_processor_test.go create mode 100644 internal/io/fs/readfile_zstd.go delete mode 100644 internal/io/fs/readfilelcontext.go create mode 100644 internal/io/fs/rootedpath.go create mode 100644 internal/io/fs/rootedpath_test.go create mode 100644 internal/io/fs/validatedreadtarget.go create mode 100644 internal/io/fs/validatedreadtarget_test.go create mode 100644 internal/io/journal/filter.go create mode 100644 internal/io/journal/reader.go create mode 100644 internal/io/journal/reader_test.go create mode 100644 internal/io/journal/reader_unsupported.go create mode 100644 internal/io/journal/testhelper/mock.go create mode 100644 internal/io/journal/testhelper/mock_test.go create mode 100644 internal/io/line/processor.go create mode 100644 internal/io/pool/scanner_pool.go create mode 100644 internal/mapr/client/aggregate_test.go create mode 100644 internal/mapr/client/session_state.go create mode 100644 internal/mapr/client/session_state_test.go create mode 100644 internal/mapr/globalgroupset_test.go create mode 100644 internal/mapr/groupset_avg_nan_test.go create mode 100644 internal/mapr/groupset_ordering_test.go create mode 100644 internal/mapr/groupset_percentage_test.go create mode 100644 internal/mapr/groupsetresult_renderer_test.go create mode 100644 internal/mapr/logformat/default_benchmark_test.go create mode 100644 internal/mapr/logformat/delimited.go create mode 100644 internal/mapr/logformat/parser_test.go create mode 100644 internal/mapr/logformat/variables.go create mode 100644 internal/mapr/logformat/variables_test.go create mode 100644 internal/mapr/parserfieldplan.go create mode 100644 internal/mapr/parserfieldplan_test.go create mode 100644 internal/mapr/queryvariables.go create mode 100644 internal/mapr/queryvariables_test.go create mode 100644 internal/mapr/result_renderer.go create mode 100644 internal/mapr/safe_aggregateset.go create mode 100644 internal/mapr/safe_aggregateset_test.go create mode 100644 internal/mapr/server/aggregate_test.go create mode 100644 internal/mapr/server/groupkey.go create mode 100644 internal/mapr/server/parsername.go create mode 100644 internal/mapr/server/parsername_test.go create mode 100644 internal/mapr/token_test.go create mode 100644 internal/profiling/README.md create mode 100644 internal/profiling/flags.go create mode 100644 internal/profiling/profiler.go create mode 100644 internal/profiling/profiler_test.go create mode 100644 internal/protocol/capabilities.go create mode 100644 internal/protocol/session.go create mode 100644 internal/regex/bench_test.go create mode 100644 internal/regex/regex_literal_test.go create mode 100644 internal/server/auth_test.go create mode 100644 internal/server/continuous_test.go create mode 100644 internal/server/handlers/authkeycommand_test.go create mode 100644 internal/server/handlers/basehandler_read_test.go create mode 100644 internal/server/handlers/commandcancel_test.go create mode 100644 internal/server/handlers/commandtimeout_test.go create mode 100644 internal/server/handlers/framesize_test.go create mode 100644 internal/server/handlers/generation_output.go create mode 100644 internal/server/handlers/generation_output_test.go create mode 100644 internal/server/handlers/line_writer.go create mode 100644 internal/server/handlers/line_writer_alloc_test.go create mode 100644 internal/server/handlers/line_writer_test.go create mode 100644 internal/server/handlers/lineprocessor.go create mode 100644 internal/server/handlers/lineprocessor_test.go create mode 100644 internal/server/handlers/mapcommand_completion_test.go create mode 100644 internal/server/handlers/mapcommand_race_test.go create mode 100644 internal/server/handlers/output_manager.go create mode 100644 internal/server/handlers/output_manager_race_test.go create mode 100644 internal/server/handlers/protocol_codec.go create mode 100644 internal/server/handlers/protocol_codec_test.go create mode 100644 internal/server/handlers/protocol_formatter.go create mode 100644 internal/server/handlers/readcommand_cancellation_test.go create mode 100644 internal/server/handlers/readcommand_epoch_order_test.go create mode 100644 internal/server/handlers/readcommand_glob_cap_test.go create mode 100644 internal/server/handlers/readcommand_journal_test.go create mode 100644 internal/server/handlers/readcommand_semaphore_test.go create mode 100644 internal/server/handlers/readcommand_sendmessage_test.go create mode 100644 internal/server/handlers/readcommand_server.go create mode 100644 internal/server/handlers/sessioncommand.go create mode 100644 internal/server/handlers/sessioncommand_test.go create mode 100644 internal/server/handlers/shutdown_coordinator.go create mode 100644 internal/server/stats_test.go create mode 100644 internal/session/spec.go create mode 100644 internal/session/spec_test.go create mode 100644 internal/ssh/client/authmethods_test.go create mode 100644 internal/ssh/client/knownhostscallback_test.go create mode 100644 internal/ssh/server/authkeystore.go create mode 100644 internal/ssh/server/authkeystore_test.go create mode 100644 internal/ssh/server/hostkey_test.go create mode 100644 internal/ssh/server/publickeycallback_test.go create mode 100644 internal/ssh/ssh_agent_test.go create mode 100644 internal/ssh/ssh_test.go create mode 100644 internal/tools/benchmark/benchmark.go create mode 100644 internal/tools/common/data_generator.go create mode 100644 internal/tools/common/utils.go create mode 100644 internal/tools/pgo/pgo.go create mode 100644 internal/tools/pgo/pgo_test.go create mode 100644 internal/tools/profile/analyze.go create mode 100644 internal/tools/profile/profile.go create mode 100644 internal/tools/profile/profile_test.go create mode 100644 internal/user/server/user_test.go diff --git a/cmd/dcat/main.go b/cmd/dcat/main.go index a50be51..83b713c 100644 --- a/cmd/dcat/main.go +++ b/cmd/dcat/main.go @@ -3,17 +3,14 @@ package main import ( "context" "flag" + "fmt" "os" - "sync" - - "net/http" - _ "net/http" - _ "net/http/pprof" + "github.com/mimecast/dtail/internal/cli" "github.com/mimecast/dtail/internal/clients" "github.com/mimecast/dtail/internal/config" - "github.com/mimecast/dtail/internal/io/dlog" "github.com/mimecast/dtail/internal/io/signal" + "github.com/mimecast/dtail/internal/profiling" "github.com/mimecast/dtail/internal/source" "github.com/mimecast/dtail/internal/user" "github.com/mimecast/dtail/internal/version" @@ -23,56 +20,63 @@ import ( func main() { var args config.Args var displayVersion bool + var legacyAuthKeyPath string var pprof string + var profileFlags profiling.Flags userName := user.Name() flag.BoolVar(&args.NoColor, "noColor", false, "Disable ANSII terminal colors") + flag.BoolVar(&args.NoAuthKey, "no-auth-key", false, "Disable auth-key fast reconnect feature") + flag.BoolVar(&args.LogPayload, "log-payload", false, "Also tee retrieved payload into the client log file (default: file keeps diagnostics only)") flag.BoolVar(&args.Quiet, "quiet", false, "Quiet output mode") + flag.BoolVar(&args.InteractiveQuery, "interactive-query", false, "Enable interactive in-flight query control over supported sessions") flag.BoolVar(&args.Plain, "plain", false, "Plain 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.SSHAgentKeyIndex, "agentKeyIndex", -1, "SSH agent key index to use (-1 for all keys)") flag.IntVar(&args.SSHPort, "port", config.DefaultSSHPort, "SSH server port") flag.StringVar(&args.ConfigFile, "cfg", "", "Config file path") + flag.StringVar(&args.ControlTTYPath, "control-tty", "/dev/tty", "TTY device for interactive query control") flag.StringVar(&args.Discovery, "discovery", "", "Server discovery method") flag.StringVar(&args.LogDir, "logDir", "~/log", "Log dir") flag.StringVar(&args.Logger, "logger", config.DefaultClientLogger, "Logger name") flag.StringVar(&args.LogLevel, "logLevel", config.DefaultLogLevel, "Log level") - flag.StringVar(&args.SSHPrivateKeyFilePath, "key", "", "Path to private key") + cli.BindAuthKeyFlags(flag.CommandLine, &legacyAuthKeyPath, &args) flag.StringVar(&args.ServersStr, "servers", "", "Remote servers to connect") flag.StringVar(&args.UserName, "user", userName, "Your system user name") flag.StringVar(&args.What, "files", "", "File(s) to read") flag.StringVar(&pprof, "pprof", "", "Start PProf server this address") + // Add profiling flags + profiling.AddFlags(&profileFlags) + flag.Parse() + if warning := cli.ApplyAuthKeyPathCompatibility(&args, legacyAuthKeyPath, cli.FlagWasSet("auth-key-path")); warning != "" { + fmt.Fprintln(os.Stderr, warning) + } config.Setup(source.Client, &args, flag.Args()) if displayVersion { - version.PrintAndExit() + runtimeCfg := config.CurrentRuntime() + version.PrintAndExit(runtimeCfg.Client != nil && runtimeCfg.Client.TermColorsEnable) } - ctx, cancel := context.WithCancel(context.Background()) - var wg sync.WaitGroup - wg.Add(1) - dlog.Start(ctx, &wg, source.Client) - - if pprof != "" { - dlog.Client.Info("Starting PProf", pprof) - go func() { - panic(http.ListenAndServe(pprof, nil)) - }() - } + runtime := cli.NewClientRuntime(context.Background(), profileFlags, "dcat") + runtime.StartPProf(pprof) + runtime.LogStartupMetrics() client, err := clients.NewCatClient(args) if err != nil { - panic(err) + runtime.Stop() + fmt.Fprintf(os.Stderr, "unable to create dcat client: %v\n", err) + os.Exit(1) } - status := client.Start(ctx, signal.InterruptCh(ctx)) - cancel() - - wg.Wait() + status := client.Start(runtime.Context(), signal.InterruptCh(runtime.Context())) + runtime.LogShutdownMetrics() + runtime.Stop() os.Exit(status) } diff --git a/cmd/dgrep/main.go b/cmd/dgrep/main.go index 19f818b..1c2837b 100644 --- a/cmd/dgrep/main.go +++ b/cmd/dgrep/main.go @@ -3,17 +3,14 @@ package main import ( "context" "flag" + "fmt" "os" - "sync" - - "net/http" - _ "net/http" - _ "net/http/pprof" + "github.com/mimecast/dtail/internal/cli" "github.com/mimecast/dtail/internal/clients" "github.com/mimecast/dtail/internal/config" - "github.com/mimecast/dtail/internal/io/dlog" "github.com/mimecast/dtail/internal/io/signal" + "github.com/mimecast/dtail/internal/profiling" "github.com/mimecast/dtail/internal/source" "github.com/mimecast/dtail/internal/user" "github.com/mimecast/dtail/internal/version" @@ -24,12 +21,17 @@ func main() { var args config.Args var displayVersion bool var grep string + var legacyAuthKeyPath string var pprof string + var profileFlags profiling.Flags userName := user.Name() flag.BoolVar(&args.NoColor, "noColor", false, "Disable ANSII terminal colors") + flag.BoolVar(&args.NoAuthKey, "no-auth-key", false, "Disable auth-key fast reconnect feature") + flag.BoolVar(&args.LogPayload, "log-payload", false, "Also tee retrieved payload into the client log file (default: file keeps diagnostics only)") flag.BoolVar(&args.Quiet, "quiet", false, "Quiet output mode") flag.BoolVar(&args.RegexInvert, "invert", false, "Invert regex") + flag.BoolVar(&args.InteractiveQuery, "interactive-query", false, "Enable interactive in-flight query control over supported sessions") flag.BoolVar(&args.Plain, "plain", false, "Plain output mode") flag.BoolVar(&args.TrustAllHosts, "trustAllHosts", false, "Trust all unknown host keys") flag.BoolVar(&displayVersion, "version", false, "Display version") @@ -38,13 +40,15 @@ func main() { flag.IntVar(&args.LContext.AfterContext, "after", 0, "Print lines of trailing context after matching lines") flag.IntVar(&args.LContext.BeforeContext, "before", 0, "Print lines of leading context before matching lines") flag.IntVar(&args.LContext.MaxCount, "max", 0, "Stop reading file after NUM matching lines") + flag.IntVar(&args.SSHAgentKeyIndex, "agentKeyIndex", -1, "SSH agent key index to use (-1 for all keys)") flag.IntVar(&args.SSHPort, "port", config.DefaultSSHPort, "SSH server port") flag.StringVar(&args.ConfigFile, "cfg", "", "Config file path") + flag.StringVar(&args.ControlTTYPath, "control-tty", "/dev/tty", "TTY device for interactive query control") flag.StringVar(&args.Discovery, "discovery", "", "Server discovery method") flag.StringVar(&args.LogDir, "logDir", "~/log", "Log dir") flag.StringVar(&args.Logger, "logger", config.DefaultClientLogger, "Logger name") flag.StringVar(&args.LogLevel, "logLevel", config.DefaultLogLevel, "Log level") - flag.StringVar(&args.SSHPrivateKeyFilePath, "key", "", "Path to private key") + cli.BindAuthKeyFlags(flag.CommandLine, &legacyAuthKeyPath, &args) flag.StringVar(&args.RegexStr, "regex", ".", "Regular expression") flag.StringVar(&args.ServersStr, "servers", "", "Remote servers to connect") flag.StringVar(&args.UserName, "user", userName, "Your system user name") @@ -52,37 +56,38 @@ func main() { flag.StringVar(&grep, "grep", "", "Alias for -regex") flag.StringVar(&pprof, "pprof", "", "Start PProf server this address") + // Add profiling flags + profiling.AddFlags(&profileFlags) + flag.Parse() + if warning := cli.ApplyAuthKeyPathCompatibility(&args, legacyAuthKeyPath, cli.FlagWasSet("auth-key-path")); warning != "" { + fmt.Fprintln(os.Stderr, warning) + } config.Setup(source.Client, &args, flag.Args()) if displayVersion { - version.PrintAndExit() + runtimeCfg := config.CurrentRuntime() + version.PrintAndExit(runtimeCfg.Client != nil && runtimeCfg.Client.TermColorsEnable) } - ctx, cancel := context.WithCancel(context.Background()) - var wg sync.WaitGroup - wg.Add(1) - dlog.Start(ctx, &wg, source.Client) + runtime := cli.NewClientRuntime(context.Background(), profileFlags, "dgrep") if grep != "" { args.RegexStr = grep } - if pprof != "" { - dlog.Client.Info("Starting PProf", pprof) - go func() { - panic(http.ListenAndServe(pprof, nil)) - }() - } + runtime.StartPProf(pprof) + runtime.LogStartupMetrics() client, err := clients.NewGrepClient(args) if err != nil { - panic(err) + runtime.Stop() + fmt.Fprintf(os.Stderr, "unable to create dgrep client: %v\n", err) + os.Exit(1) } - status := client.Start(ctx, signal.InterruptCh(ctx)) - cancel() - - wg.Wait() + status := client.Start(runtime.Context(), signal.InterruptCh(runtime.Context())) + runtime.LogShutdownMetrics() + runtime.Stop() os.Exit(status) } diff --git a/cmd/dmap/main.go b/cmd/dmap/main.go index a8a52a2..8e980e8 100644 --- a/cmd/dmap/main.go +++ b/cmd/dmap/main.go @@ -3,18 +3,16 @@ package main import ( "context" "flag" + "fmt" "os" - "sync" - - "net/http" - _ "net/http" - _ "net/http/pprof" + "github.com/mimecast/dtail/internal/cli" "github.com/mimecast/dtail/internal/clients" "github.com/mimecast/dtail/internal/config" "github.com/mimecast/dtail/internal/io/dlog" "github.com/mimecast/dtail/internal/io/signal" "github.com/mimecast/dtail/internal/omode" + "github.com/mimecast/dtail/internal/profiling" "github.com/mimecast/dtail/internal/source" "github.com/mimecast/dtail/internal/user" "github.com/mimecast/dtail/internal/version" @@ -23,61 +21,71 @@ import ( // The evil begins here. func main() { var displayVersion bool + var legacyAuthKeyPath string var pprof string + var profileFlags profiling.Flags args := config.Args{ - Mode: omode.MapClient, + Mode: omode.MapClient, + SSHAgentKeyIndex: -1, } userName := user.Name() flag.BoolVar(&args.NoColor, "noColor", false, "Disable ANSII terminal colors") + flag.BoolVar(&args.NoAuthKey, "no-auth-key", false, "Disable auth-key fast reconnect feature") + flag.BoolVar(&args.LogPayload, "log-payload", false, "Also tee retrieved payload into the client log file (default: file keeps diagnostics only)") flag.BoolVar(&args.Quiet, "quiet", false, "Quiet output mode") + flag.BoolVar(&args.InteractiveQuery, "interactive-query", false, "Enable interactive in-flight query control over supported sessions") flag.BoolVar(&args.Plain, "plain", false, "Plain 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.SSHAgentKeyIndex, "agentKeyIndex", -1, "SSH agent key index to use (-1 for all keys)") 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") + flag.StringVar(&args.ControlTTYPath, "control-tty", "/dev/tty", "TTY device for interactive query control") flag.StringVar(&args.Discovery, "discovery", "", "Server discovery method") flag.StringVar(&args.LogDir, "logDir", "~/log", "Log dir") flag.StringVar(&args.Logger, "logger", config.DefaultClientLogger, "Logger name") flag.StringVar(&args.LogLevel, "logLevel", config.DefaultLogLevel, "Log level") - flag.StringVar(&args.SSHPrivateKeyFilePath, "key", "", "Path to private key") + cli.BindAuthKeyFlags(flag.CommandLine, &legacyAuthKeyPath, &args) flag.StringVar(&args.QueryStr, "query", "", "Map reduce query") flag.StringVar(&args.ServersStr, "servers", "", "Remote servers to connect") flag.StringVar(&args.UserName, "user", userName, "Your system user name") flag.StringVar(&args.What, "files", "", "File(s) to read") flag.StringVar(&pprof, "pprof", "", "Start PProf server this address") + // Add profiling flags + profiling.AddFlags(&profileFlags) + flag.Parse() + if warning := cli.ApplyAuthKeyPathCompatibility(&args, legacyAuthKeyPath, cli.FlagWasSet("auth-key-path")); warning != "" { + fmt.Fprintln(os.Stderr, warning) + } config.Setup(source.Client, &args, flag.Args()) if displayVersion { - version.PrintAndExit() + runtimeCfg := config.CurrentRuntime() + version.PrintAndExit(runtimeCfg.Client != nil && runtimeCfg.Client.TermColorsEnable) } - ctx, cancel := context.WithCancel(context.Background()) - var wg sync.WaitGroup - wg.Add(1) - dlog.Start(ctx, &wg, source.Client) - - if pprof != "" { - dlog.Client.Info("Starting PProf", pprof) - go func() { - panic(http.ListenAndServe(pprof, nil)) - }() - } + runtime := cli.NewClientRuntime(context.Background(), profileFlags, "dmap") + runtime.StartPProf(pprof) + runtime.LogStartupMetrics() client, err := clients.NewMaprClient(args, clients.DefaultMode) if err != nil { + runtime.Stop() dlog.Client.FatalPanic(err) } - status := client.Start(ctx, signal.InterruptCh(ctx)) - cancel() - - wg.Wait() + status := client.Start( + runtime.Context(), + signal.InterruptChWithCancel(runtime.Context(), runtime.Cancel), + ) + runtime.LogShutdownMetrics() + runtime.Stop() os.Exit(status) } diff --git a/cmd/dserver/main.go b/cmd/dserver/main.go index 3377273..936edce 100644 --- a/cmd/dserver/main.go +++ b/cmd/dserver/main.go @@ -3,15 +3,13 @@ package main import ( "context" "flag" - "net/http" - _ "net/http" - _ "net/http/pprof" "os" "os/signal" "sync" "syscall" "time" + "github.com/mimecast/dtail/internal/cli" "github.com/mimecast/dtail/internal/config" "github.com/mimecast/dtail/internal/io/dlog" "github.com/mimecast/dtail/internal/server" @@ -46,13 +44,31 @@ func main() { config.Setup(source.Server, &args, flag.Args()) if displayVersion { - version.PrintAndExit() + runtimeCfg := config.CurrentRuntime() + version.PrintAndExit(runtimeCfg.Client != nil && runtimeCfg.Client.TermColorsEnable) } - version.Print() + version.Print(false) + + // rootCtx is always cancelled on exit to ensure the internal goroutine + // spawned by context.WithCancel is released. When -shutdownAfter is set, + // ctx is replaced by a child WithTimeout context whose own cancel is also + // deferred, preventing the lostcancel leak flagged by go vet. + rootCtx, rootCancel := context.WithCancel(context.Background()) + defer rootCancel() + + ctx := rootCtx + cancel := context.CancelFunc(rootCancel) - ctx, cancel := context.WithCancel(context.Background()) if shutdownAfter > 0 { - ctx, cancel = context.WithTimeout(ctx, time.Duration(shutdownAfter)*time.Second) + // Override ctx with a timeout-bounded child; defer its cancel so the + // timeout goroutine is always cleaned up regardless of code path. + var timeoutCancel context.CancelFunc + ctx, timeoutCancel = context.WithTimeout(rootCtx, time.Duration(shutdownAfter)*time.Second) + defer timeoutCancel() + // Callers that invoke cancel() (e.g. the signal handler and post-serve + // cleanup) should trigger the timeout cancel so the server shuts down + // promptly even before the deadline fires. + cancel = timeoutCancel } sigCh := make(chan os.Signal, 10) @@ -70,16 +86,36 @@ func main() { wg.Add(1) dlog.Start(ctx, &wg, source.Server) + var pprofServer *cli.PProfServer if pprof != "" { - dlog.Client.Info("Starting PProf", pprof) - go func() { - panic(http.ListenAndServe(pprof, nil)) - }() + // Enable mutex and block profiling so the /debug/pprof/mutex and + // /debug/pprof/block endpoints actually contain samples. These rates + // are gated on --pprof so they cost nothing when profiling is off. + cli.EnableProfilingRates() + + // Assign to the outer pprofServer with '=' (declaring pprofErr + // separately) so it is NOT shadowed: the graceful Shutdown below relies + // on the outer var being non-nil to actually stop the pprof server. + var pprofErr error + pprofServer, pprofErr = cli.NewPProfServer(pprof) + if pprofErr != nil { + dlog.Client.Error("Unable to start PProf", pprofErr) + } else { + dlog.Client.Info("Starting PProf", pprofServer.Address()) + pprofServer.Start(nil) + } } - serv := server.New() + serv := server.New(config.CurrentRuntime()) status := serv.Start(ctx) cancel() + if pprofServer != nil { + shutdownCtx, shutdownCancel := context.WithTimeout(context.Background(), 5*time.Second) + if err := pprofServer.Shutdown(shutdownCtx); err != nil { + dlog.Client.Error("Unable to stop PProf", err) + } + shutdownCancel() + } wg.Wait() os.Exit(status) diff --git a/cmd/dtail-tools/main.go b/cmd/dtail-tools/main.go new file mode 100644 index 0000000..2b96a56 --- /dev/null +++ b/cmd/dtail-tools/main.go @@ -0,0 +1,60 @@ +package main + +import ( + "fmt" + "os" + + "github.com/mimecast/dtail/internal/tools/benchmark" + "github.com/mimecast/dtail/internal/tools/pgo" + "github.com/mimecast/dtail/internal/tools/profile" +) + +func main() { + if len(os.Args) < 2 { + printUsage() + os.Exit(1) + } + + command := os.Args[1] + + // Remove command from args for subcommand parsing + os.Args = append([]string{os.Args[0]}, os.Args[2:]...) + + switch command { + case "profile": + if err := profile.Run(); err != nil { + fmt.Fprintf(os.Stderr, "Error: %v\n", err) + os.Exit(1) + } + case "benchmark": + if err := benchmark.Run(); err != nil { + fmt.Fprintf(os.Stderr, "Error: %v\n", err) + os.Exit(1) + } + case "pgo": + if err := pgo.Run(); err != nil { + fmt.Fprintf(os.Stderr, "Error: %v\n", err) + os.Exit(1) + } + case "help", "-h", "--help": + printUsage() + default: + fmt.Fprintf(os.Stderr, "Unknown command: %s\n", command) + printUsage() + os.Exit(1) + } +} + +func printUsage() { + fmt.Println("dtail-tools - DTail performance analysis toolkit") + fmt.Println() + fmt.Println("Usage: dtail-tools [options]") + fmt.Println() + fmt.Println("Commands:") + fmt.Println(" profile Run profiling on dtail commands") + fmt.Println(" benchmark Run benchmarks and manage baselines") + fmt.Println(" pgo Profile-Guided Optimization for dtail commands") + fmt.Println(" help Show this help message") + fmt.Println() + fmt.Println("Run 'dtail-tools -h' for command-specific help") +} \ No newline at end of file diff --git a/cmd/dtail/main.go b/cmd/dtail/main.go index e18923a..0105709 100644 --- a/cmd/dtail/main.go +++ b/cmd/dtail/main.go @@ -4,19 +4,16 @@ import ( "context" "flag" "fmt" - "net/http" - _ "net/http" - _ "net/http/pprof" "os" - "sync" "time" + "github.com/mimecast/dtail/internal/cli" "github.com/mimecast/dtail/internal/clients" "github.com/mimecast/dtail/internal/color" "github.com/mimecast/dtail/internal/config" - "github.com/mimecast/dtail/internal/io/dlog" "github.com/mimecast/dtail/internal/io/signal" "github.com/mimecast/dtail/internal/omode" + "github.com/mimecast/dtail/internal/profiling" "github.com/mimecast/dtail/internal/source" "github.com/mimecast/dtail/internal/user" "github.com/mimecast/dtail/internal/version" @@ -30,14 +27,19 @@ func main() { var displayWideColorTable bool var displayVersion bool var grep string + var legacyAuthKeyPath string var pprof string var shutdownAfter int + var profileFlags profiling.Flags userName := user.Name() flag.BoolVar(&args.NoColor, "noColor", false, "Disable ANSII terminal colors") + flag.BoolVar(&args.NoAuthKey, "no-auth-key", false, "Disable auth-key fast reconnect feature") + flag.BoolVar(&args.LogPayload, "log-payload", false, "Also tee retrieved payload into the client log file (default: file keeps diagnostics only)") flag.BoolVar(&args.Quiet, "quiet", false, "Quiet output mode") flag.BoolVar(&args.RegexInvert, "invert", false, "Invert regex") + flag.BoolVar(&args.InteractiveQuery, "interactive-query", false, "Enable interactive in-flight query control over supported sessions") flag.BoolVar(&args.Plain, "plain", false, "Plain output mode") flag.BoolVar(&args.TrustAllHosts, "trustAllHosts", false, "Trust all unknown host keys") flag.BoolVar(&checkHealth, "checkHealth", false, "Deprecated, flag will be removed soon") @@ -49,15 +51,17 @@ func main() { flag.IntVar(&args.LContext.AfterContext, "after", 0, "Print lines of trailing context after matching lines") flag.IntVar(&args.LContext.BeforeContext, "before", 0, "Print lines of leading context before matching lines") flag.IntVar(&args.LContext.MaxCount, "max", 0, "Stop reading file after NUM matching lines") + flag.IntVar(&args.SSHAgentKeyIndex, "agentKeyIndex", -1, "SSH agent key index to use (-1 for all keys)") 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(&shutdownAfter, "shutdownAfter", 3600*24, "Shutdown after so many seconds") flag.StringVar(&args.ConfigFile, "cfg", "", "Config file path") + flag.StringVar(&args.ControlTTYPath, "control-tty", "/dev/tty", "TTY device for interactive query control") flag.StringVar(&args.Discovery, "discovery", "", "Server discovery method") flag.StringVar(&args.LogDir, "logDir", "~/log", "Log dir") flag.StringVar(&args.Logger, "logger", config.DefaultClientLogger, "Logger name") flag.StringVar(&args.LogLevel, "logLevel", config.DefaultLogLevel, "Log level") - flag.StringVar(&args.SSHPrivateKeyFilePath, "key", "", "Path to private key") + cli.BindAuthKeyFlags(flag.CommandLine, &legacyAuthKeyPath, &args) flag.StringVar(&args.QueryStr, "query", "", "Map reduce query") flag.StringVar(&args.RegexStr, "regex", ".", "Regular expression") flag.StringVar(&args.ServersStr, "servers", "", "Remote servers to connect") @@ -66,13 +70,20 @@ func main() { flag.StringVar(&grep, "grep", "", "Alias for -regex") flag.StringVar(&pprof, "pprof", "", "Start PProf server this address") + // Add profiling flags + profiling.AddFlags(&profileFlags) + flag.Parse() + if warning := cli.ApplyAuthKeyPathCompatibility(&args, legacyAuthKeyPath, cli.FlagWasSet("auth-key-path")); warning != "" { + fmt.Fprintln(os.Stderr, warning) + } if grep != "" { args.RegexStr = grep } config.Setup(source.Client, &args, flag.Args()) if displayVersion { - version.PrintAndExit() + runtimeCfg := config.CurrentRuntime() + version.PrintAndExit(runtimeCfg.Client != nil && runtimeCfg.Client.TermColorsEnable) } if !args.Plain { if displayWideColorTable { @@ -83,30 +94,26 @@ func main() { } } - ctx, cancel := context.WithCancel(context.Background()) - if shutdownAfter > 0 { - // NEXT: This does not work (auto shutdown) - ctx, cancel = context.WithTimeout(ctx, time.Duration(shutdownAfter)*time.Second) - defer cancel() - } + baseCtx, timeoutCancel := applyClientDeadlines(context.Background(), shutdownAfter, args.Timeout) - var wg sync.WaitGroup - wg.Add(1) - dlog.Start(ctx, &wg, source.Client) + runtime := cli.NewClientRuntime(baseCtx, profileFlags, "dtail") + exitWithError := func(err error) { + runtime.Stop() + timeoutCancel() + fmt.Fprintf(os.Stderr, "unable to initialize dtail client: %v\n", err) + os.Exit(1) + } if checkHealth { fmt.Println("WARN: DTail health check has moved to separate binary dtailhealth" + " - please adjust the monitoring scripts!") - cancel() + runtime.Stop() + timeoutCancel() os.Exit(1) } - if pprof != "" { - dlog.Client.Info("Starting PProf", pprof) - go func() { - panic(http.ListenAndServe(pprof, nil)) - }() - } + runtime.StartPProf(pprof) + runtime.LogStartupMetrics() var client clients.Client var err error @@ -115,17 +122,64 @@ func main() { switch args.QueryStr { case "": if client, err = clients.NewTailClient(args); err != nil { - panic(err) + exitWithError(err) } default: if client, err = clients.NewMaprClient(args, clients.DefaultMode); err != nil { - panic(err) + exitWithError(err) } } - status := client.Start(ctx, signal.InterruptCh(ctx)) - cancel() - - wg.Wait() + status := client.Start( + runtime.Context(), + signal.InterruptChWithCancel(runtime.Context(), runtime.Cancel), + ) + runtime.LogShutdownMetrics() + runtime.Stop() + timeoutCancel() os.Exit(status) } + +// applyClientDeadlines wraps ctx with the earliest of two absolute deadlines: +// the --shutdownAfter safety cap and the user's --timeout data-collection +// window. Whichever elapses first cancels the returned context, which propagates +// through the follow client's reconnect and per-connection read loops (both +// select on ctx.Done()), so client.Start returns cleanly and the process exits +// instead of auto-reconnecting for another cycle. +// +// Making --timeout a client-side deadline (rather than relying solely on the +// server closing the read) is what fixes the historical hang: the server-side +// read deadline fires at N seconds, but in tail+query mode the session stays +// alive via the map/aggregate command, so the client used to treat the closed +// read as a transient drop and reconnect indefinitely. A client-side deadline +// makes --timeout behave consistently for follows with or without --query, which +// matches the flag's help text ("Max time ... until disconnection"). +// +// The two deadlines compose naturally (context deadlines nest, so the earlier +// one wins), so they never conflict with each other. A timeout of 0 (unset) +// contributes no deadline, preserving the previous behaviour. OS signals reach +// the same context via signal.InterruptChWithCancel in main. +func applyClientDeadlines(ctx context.Context, shutdownAfter, timeout int) ( + context.Context, context.CancelFunc) { + + var cancels []context.CancelFunc + addDeadline := func(seconds int) { + if seconds <= 0 { + return + } + var cancel context.CancelFunc + ctx, cancel = context.WithTimeout(ctx, time.Duration(seconds)*time.Second) + cancels = append(cancels, cancel) + } + + addDeadline(shutdownAfter) + addDeadline(timeout) + + return ctx, func() { + // Release timers in reverse (inner first) to avoid leaking the parent + // timer while an inner context still references it. + for i := len(cancels) - 1; i >= 0; i-- { + cancels[i]() + } + } +} diff --git a/cmd/dtail/main_test.go b/cmd/dtail/main_test.go new file mode 100644 index 0000000..b7dbb77 --- /dev/null +++ b/cmd/dtail/main_test.go @@ -0,0 +1,66 @@ +package main + +import ( + "context" + "testing" + "time" +) + +// TestApplyClientDeadlines verifies that --timeout and --shutdownAfter are turned +// into client-side context deadlines and that the earliest one wins, which is +// what makes the dtail follow client exit instead of reconnecting after a +// timeout-induced disconnect (task xu0). +func TestApplyClientDeadlines(t *testing.T) { + tests := []struct { + name string + shutdownAfter int + timeout int + wantDeadline bool + // wantWithin bounds the deadline from the parent context when a deadline + // is expected. It reflects the smaller of the two configured windows. + wantWithin time.Duration + }{ + {name: "both unset: no deadline", shutdownAfter: 0, timeout: 0, wantDeadline: false}, + {name: "only shutdownAfter", shutdownAfter: 5, timeout: 0, wantDeadline: true, wantWithin: 5 * time.Second}, + {name: "only timeout", shutdownAfter: 0, timeout: 4, wantDeadline: true, wantWithin: 4 * time.Second}, + {name: "timeout smaller wins", shutdownAfter: 30, timeout: 3, wantDeadline: true, wantWithin: 3 * time.Second}, + {name: "shutdownAfter smaller wins", shutdownAfter: 2, timeout: 60, wantDeadline: true, wantWithin: 2 * time.Second}, + {name: "negative values ignored", shutdownAfter: -1, timeout: -1, wantDeadline: false}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + before := time.Now() + ctx, cancel := applyClientDeadlines(context.Background(), tt.shutdownAfter, tt.timeout) + defer cancel() + + deadline, ok := ctx.Deadline() + if ok != tt.wantDeadline { + t.Fatalf("Deadline() ok = %v, want %v", ok, tt.wantDeadline) + } + if !tt.wantDeadline { + return + } + + // The effective deadline must equal the smaller configured window + // (allow a small slack for scheduling). + gotWindow := deadline.Sub(before) + if gotWindow > tt.wantWithin+time.Second || gotWindow < tt.wantWithin-time.Second { + t.Fatalf("effective deadline window = %s, want ~%s", gotWindow, tt.wantWithin) + } + }) + } +} + +// TestApplyClientDeadlinesCancelFires ensures the returned cancel function +// actually cancels the context (releasing timers) without panicking. +func TestApplyClientDeadlinesCancelFires(t *testing.T) { + ctx, cancel := applyClientDeadlines(context.Background(), 3600, 10) + cancel() + + select { + case <-ctx.Done(): + default: + t.Fatal("context was not canceled after calling cancel()") + } +} diff --git a/cmd/dtailhealth/main.go b/cmd/dtailhealth/main.go index a0ca84e..33e84db 100644 --- a/cmd/dtailhealth/main.go +++ b/cmd/dtailhealth/main.go @@ -3,13 +3,12 @@ package main import ( "context" "flag" + "fmt" "os" "sync" + "time" - "net/http" - _ "net/http" - _ "net/http/pprof" - + "github.com/mimecast/dtail/internal/cli" "github.com/mimecast/dtail/internal/clients" "github.com/mimecast/dtail/internal/config" "github.com/mimecast/dtail/internal/io/dlog" @@ -28,11 +27,12 @@ func main() { flag.StringVar(&args.Logger, "logger", config.DefaultHealthCheckLogger, "Logger name") flag.StringVar(&args.LogLevel, "logLevel", "none", "Log level") flag.StringVar(&args.ServersStr, "server", "", "Remote server to connect") + flag.BoolVar(&args.NoAuthKey, "no-auth-key", false, "Disable auth-key fast reconnect feature") flag.StringVar(&pprof, "pprof", "", "Start PProf server this address") flag.Parse() if displayVersion { - version.PrintAndExit() + version.PrintAndExit(false) } config.Setup(source.HealthCheck, &args, flag.Args()) @@ -43,13 +43,35 @@ func main() { wg.Add(1) dlog.Start(ctx, &wg, source.HealthCheck) + var pprofServer *cli.PProfServer if pprof != "" { - dlog.Client.Info("Starting PProf", pprof) - go func() { - panic(http.ListenAndServe(pprof, nil)) - }() + pprofServer, pprofErr := cli.NewPProfServer(pprof) + if pprofErr != nil { + dlog.Client.Error("Unable to start PProf", pprofErr) + } else { + dlog.Client.Info("Starting PProf", pprofServer.Address()) + pprofServer.Start(nil) + } + } + + healthClient, err := clients.NewHealthClient(args) + status := 0 + if err != nil { + fmt.Fprintf(os.Stderr, "CRITICAL: unable to create dtailhealth client: %v\n", err) + status = 2 + } else { + status = healthClient.Start(ctx, signal.NoCh(ctx)) + } + + if pprofServer != nil { + shutdownCtx, shutdownCancel := context.WithTimeout(context.Background(), 5*time.Second) + if err := pprofServer.Shutdown(shutdownCtx); err != nil { + dlog.Client.Error("Unable to stop PProf", err) + } + shutdownCancel() } - healthClient, _ := clients.NewHealthClient(args) - os.Exit(healthClient.Start(ctx, signal.NoCh(ctx))) + cancel() + wg.Wait() + os.Exit(status) } diff --git a/go.mod b/go.mod index 5aad1c8..8a349ba 100644 --- a/go.mod +++ b/go.mod @@ -1,11 +1,11 @@ module github.com/mimecast/dtail -go 1.20 +go 1.25 require ( - github.com/DataDog/zstd v1.5.6 - golang.org/x/crypto v0.26.0 - golang.org/x/term v0.23.0 + github.com/DataDog/zstd v1.5.7 + golang.org/x/crypto v0.39.0 + golang.org/x/term v0.32.0 ) -require golang.org/x/sys v0.23.0 // indirect +require golang.org/x/sys v0.33.0 // indirect diff --git a/go.sum b/go.sum index 2149a9e..ccb17f9 100644 --- a/go.sum +++ b/go.sum @@ -1,8 +1,8 @@ -github.com/DataDog/zstd v1.5.6 h1:LbEglqepa/ipmmQJUDnSsfvA8e8IStVcGaFWDuxvGOY= -github.com/DataDog/zstd v1.5.6/go.mod h1:g4AWEaM3yOg3HYfnJ3YIawPnVdXJh9QME85blwSAmyw= -golang.org/x/crypto v0.26.0 h1:RrRspgV4mU+YwB4FYnuBoKsUapNIL5cohGAmSH3azsw= -golang.org/x/crypto v0.26.0/go.mod h1:GY7jblb9wI+FOo5y8/S2oY4zWP07AkOJ4+jxCqdqn54= -golang.org/x/sys v0.23.0 h1:YfKFowiIMvtgl1UERQoTPPToxltDeZfbj4H7dVUCwmM= -golang.org/x/sys v0.23.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA= -golang.org/x/term v0.23.0 h1:F6D4vR+EHoL9/sWAWgAR1H2DcHr4PareCbAaCo1RpuU= -golang.org/x/term v0.23.0/go.mod h1:DgV24QBUrK6jhZXl+20l6UWznPlwAHm1Q1mGHtydmSk= +github.com/DataDog/zstd v1.5.7 h1:ybO8RBeh29qrxIhCA9E8gKY6xfONU9T6G6aP9DTKfLE= +github.com/DataDog/zstd v1.5.7/go.mod h1:g4AWEaM3yOg3HYfnJ3YIawPnVdXJh9QME85blwSAmyw= +golang.org/x/crypto v0.39.0 h1:SHs+kF4LP+f+p14esP5jAoDpHU8Gu/v9lFRK6IT5imM= +golang.org/x/crypto v0.39.0/go.mod h1:L+Xg3Wf6HoL4Bn4238Z6ft6KfEpN0tJGo53AAPC632U= +golang.org/x/sys v0.33.0 h1:q3i8TbbEz+JRD9ywIRlyRAQbM0qF7hu24q3teo2hbuw= +golang.org/x/sys v0.33.0/go.mod h1:BJP2sWEmIv4KK5OTEluFJCKSidICx8ciO85XgH3Ak8k= +golang.org/x/term v0.32.0 h1:DR4lr0TjUs3epypdhTOkMmuF5CDFJ/8pOnbzMZPQ7bg= +golang.org/x/term v0.32.0/go.mod h1:uZG1FhGx848Sqfsq4/DlJr3xGGsYMu/L5GW4abiaEPQ= diff --git a/internal/cli/authkeyflags.go b/internal/cli/authkeyflags.go new file mode 100644 index 0000000..4c945d3 --- /dev/null +++ b/internal/cli/authkeyflags.go @@ -0,0 +1,43 @@ +package cli + +import ( + "flag" + + "github.com/mimecast/dtail/internal/config" +) + +const authKeyPathHelpText = "Path to auth key/private key (defaults to ~/.ssh/id_rsa via config)" + +// BindAuthKeyFlags registers the legacy and current auth-key flags. +func BindAuthKeyFlags(fs *flag.FlagSet, legacyKey *string, args *config.Args) { + fs.StringVar(legacyKey, "key", "", "Deprecated alias for -auth-key-path") + fs.StringVar(&args.SSHPrivateKeyFilePath, "auth-key-path", "", authKeyPathHelpText) +} + +// FlagWasSet reports whether the named flag was explicitly set on the command line. +func FlagWasSet(name string) bool { + var wasSet bool + flag.Visit(func(f *flag.Flag) { + if f.Name == name { + wasSet = true + } + }) + return wasSet +} + +// ApplyAuthKeyPathCompatibility copies the deprecated legacy key into args when +// the new flag was not explicitly set. If both were set, the new flag wins and +// a warning message is returned. +func ApplyAuthKeyPathCompatibility(args *config.Args, legacyKey string, authKeyPathSet bool) string { + if authKeyPathSet { + if legacyKey != "" { + return "WARN: -key is deprecated; ignoring it because -auth-key-path was also set" + } + return "" + } + + if legacyKey != "" { + args.SSHPrivateKeyFilePath = legacyKey + } + return "" +} diff --git a/internal/cli/authkeyflags_test.go b/internal/cli/authkeyflags_test.go new file mode 100644 index 0000000..64c76c3 --- /dev/null +++ b/internal/cli/authkeyflags_test.go @@ -0,0 +1,94 @@ +package cli + +import ( + "flag" + "testing" + + "github.com/mimecast/dtail/internal/config" +) + +func TestApplyAuthKeyPathCompatibilityLegacyOnly(t *testing.T) { + args := config.Args{} + + if warning := ApplyAuthKeyPathCompatibility(&args, "/tmp/legacy.pem", false); warning != "" { + t.Fatalf("unexpected warning: %q", warning) + } + + if got, want := args.SSHPrivateKeyFilePath, "/tmp/legacy.pem"; got != want { + t.Fatalf("unexpected auth key path: want %q got %q", want, got) + } +} + +func TestApplyAuthKeyPathCompatibilityPrefersExplicitAuthKeyPath(t *testing.T) { + tests := []struct { + name string + args []string + want string + warn bool + }{ + { + name: "legacy then auth", + args: []string{"-key", "/tmp/legacy.pem", "-auth-key-path", "/tmp/current.pem"}, + want: "/tmp/current.pem", + warn: true, + }, + { + name: "auth then legacy", + args: []string{"-auth-key-path", "/tmp/current.pem", "-key", "/tmp/legacy.pem"}, + want: "/tmp/current.pem", + warn: true, + }, + { + name: "explicit blank auth key path keeps blank", + args: []string{"-key", "/tmp/legacy.pem", "-auth-key-path="}, + want: "", + warn: true, + }, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + fs := flag.NewFlagSet(t.Name(), flag.ContinueOnError) + var args config.Args + var legacyKey string + + BindAuthKeyFlags(fs, &legacyKey, &args) + if err := fs.Parse(tc.args); err != nil { + t.Fatalf("Parse failed: %v", err) + } + + var authKeyPathSet bool + fs.Visit(func(f *flag.Flag) { + if f.Name == "auth-key-path" { + authKeyPathSet = true + } + }) + + warning := ApplyAuthKeyPathCompatibility(&args, legacyKey, authKeyPathSet) + if gotWarn := warning != ""; gotWarn != tc.warn { + t.Fatalf("unexpected warning presence: want %v got %v (%q)", tc.warn, gotWarn, warning) + } + if got, want := args.SSHPrivateKeyFilePath, tc.want; got != want { + t.Fatalf("unexpected auth key path: want %q got %q", want, got) + } + }) + } +} + +func TestBindAuthKeyFlagsHelpText(t *testing.T) { + fs := flag.NewFlagSet(t.Name(), flag.ContinueOnError) + var args config.Args + var legacyKey string + + BindAuthKeyFlags(fs, &legacyKey, &args) + + if got, want := fs.Lookup("key").Usage, "Deprecated alias for -auth-key-path"; got != want { + t.Fatalf("unexpected legacy flag help: want %q got %q", want, got) + } + if got, want := fs.Lookup("auth-key-path").Usage, authKeyPathHelpText; got != want { + t.Fatalf("unexpected auth-key-path help: want %q got %q", want, got) + } + if got, want := fs.Lookup("auth-key-path").DefValue, ""; got != want { + t.Fatalf("unexpected auth-key-path default: want %q got %q", want, got) + } +} diff --git a/internal/cli/pprof.go b/internal/cli/pprof.go new file mode 100644 index 0000000..57cb38c --- /dev/null +++ b/internal/cli/pprof.go @@ -0,0 +1,119 @@ +package cli + +import ( + "context" + "errors" + "net" + "net/http" + "net/http/pprof" + "runtime" + "sync" + + "github.com/mimecast/dtail/internal/io/dlog" +) + +// Mutex and block profiling rates. The /debug/pprof/mutex and +// /debug/pprof/block endpoints return empty profiles unless these runtime +// rates are enabled first; they are off by default in the Go runtime. +// +// The values below were validated during task ss0: a mutex profile fraction of +// 5 (report ~1/5 of contention events) and a block profile rate of 100000ns +// (record blocking events longer than 100µs) were sufficient to measure +// outputManager mutex contention — 12.4ms total delay across three concurrent +// 100MB server-mode dcats, i.e. NOT a bottleneck; an idle follow session +// costs 0 CPU ticks/10s because the 1ms read poll only runs while direct output is +// active and the EOF-ack drops it back to 1s polling. Documented here so the +// locking design is not re-litigated. +const ( + mutexProfileFraction = 5 + blockProfileRateNanos = 100000 +) + +// PProfServer owns a dedicated pprof HTTP server lifecycle. +type PProfServer struct { + listener net.Listener + server *http.Server + done chan struct{} +} + +// NewPProfServer creates a pprof HTTP server bound to address. +func NewPProfServer(address string) (*PProfServer, error) { + listener, err := net.Listen("tcp", address) + if err != nil { + return nil, err + } + + return &PProfServer{ + listener: listener, + server: &http.Server{ + Handler: newPProfServeMux(), + }, + done: make(chan struct{}), + }, nil +} + +// EnableProfilingRates turns on mutex and block profiling collection so that +// the /debug/pprof/mutex and /debug/pprof/block endpoints actually contain +// samples. Without this the Go runtime keeps both rates at zero and those +// endpoints report empty profiles. Only call this when pprof is enabled so it +// costs nothing in the common (no --pprof) case. +func EnableProfilingRates() { + runtime.SetMutexProfileFraction(mutexProfileFraction) + runtime.SetBlockProfileRate(blockProfileRateNanos) +} + +// Address returns the bound pprof listener address. +func (s *PProfServer) Address() string { + if s == nil || s.listener == nil { + return "" + } + return s.listener.Addr().String() +} + +// Start serves the pprof HTTP endpoints until shutdown. +func (s *PProfServer) Start(wg *sync.WaitGroup) { + if s == nil { + return + } + + if wg != nil { + wg.Add(1) + } + + go func() { + if wg != nil { + defer wg.Done() + } + defer close(s.done) + + if err := s.server.Serve(s.listener); err != nil && !errors.Is(err, http.ErrServerClosed) { + dlog.Client.Error("PProf server exited", err) + } + }() +} + +// Shutdown stops the pprof HTTP server and waits for Serve to return. +func (s *PProfServer) Shutdown(ctx context.Context) error { + if s == nil { + return nil + } + + err := s.server.Shutdown(ctx) + <-s.done + return err +} + +func newPProfServeMux() *http.ServeMux { + mux := http.NewServeMux() + mux.HandleFunc("/debug/pprof/", pprof.Index) + mux.HandleFunc("/debug/pprof/cmdline", pprof.Cmdline) + mux.HandleFunc("/debug/pprof/profile", pprof.Profile) + mux.HandleFunc("/debug/pprof/symbol", pprof.Symbol) + mux.HandleFunc("/debug/pprof/trace", pprof.Trace) + + for _, name := range []string{"allocs",