diff options
Diffstat (limited to 'internal')
106 files changed, 4292 insertions, 2461 deletions
diff --git a/internal/clients/args.go b/internal/clients/args.go deleted file mode 100644 index 34fcfa2..0000000 --- a/internal/clients/args.go +++ /dev/null @@ -1,25 +0,0 @@ -package clients - -import ( - "github.com/mimecast/dtail/internal/omode" - - gossh "golang.org/x/crypto/ssh" -) - -// Args is a helper struct to summarize common client arguments. -type Args struct { - Mode omode.Mode - ServersStr string - UserName string - What string - Arguments []string - RegexStr string - RegexInvert bool - TrustAllHosts bool - Discovery string - ConnectionsPerCPU int - Timeout int - SSHAuthMethods []gossh.AuthMethod - SSHHostKeyCallback gossh.HostKeyCallback - PrivateKeyPathFile string -} diff --git a/internal/clients/baseclient.go b/internal/clients/baseclient.go index 69055a3..4a7bd84 100644 --- a/internal/clients/baseclient.go +++ b/internal/clients/baseclient.go @@ -5,10 +5,10 @@ import ( "sync" "time" - "github.com/mimecast/dtail/internal/clients/remote" + "github.com/mimecast/dtail/internal/clients/connectors" + "github.com/mimecast/dtail/internal/config" "github.com/mimecast/dtail/internal/discovery" - "github.com/mimecast/dtail/internal/io/logger" - "github.com/mimecast/dtail/internal/omode" + "github.com/mimecast/dtail/internal/io/dlog" "github.com/mimecast/dtail/internal/regex" "github.com/mimecast/dtail/internal/ssh/client" @@ -17,13 +17,13 @@ import ( // This is the main client data structure. type baseClient struct { - Args + config.Args // To display client side stats stats *stats // List of remote servers to connect to. servers []string // We have one connection per remote server. - connections []*remote.Connection + connections []connectors.Connector // SSH auth methods to use to connect to the remote servers. sshAuthMethods []gossh.AuthMethod // To deal with SSH host keys @@ -39,7 +39,7 @@ type baseClient struct { } func (c *baseClient) init() { - logger.Info("Initiating base client") + dlog.Client.Debug("Initiating base client", c.Args.String()) flag := regex.Default if c.Args.RegexInvert { @@ -47,12 +47,16 @@ func (c *baseClient) init() { } regex, err := regex.New(c.Args.RegexStr, flag) if err != nil { - logger.FatalExit(c.Regex, "invalid regex!", err, regex) + dlog.Client.FatalPanic(c.Regex, "Invalid regex!", err, regex) } c.Regex = regex - logger.Debug("Regex", c.Regex) - c.sshAuthMethods, c.hostKeyCallback = client.InitSSHAuthMethods(c.Args.SSHAuthMethods, c.Args.SSHHostKeyCallback, c.Args.TrustAllHosts, c.throttleCh, c.Args.PrivateKeyPathFile) + if c.Args.Serverless { + return + } + c.sshAuthMethods, c.hostKeyCallback = client.InitSSHAuthMethods( + c.Args.SSHAuthMethods, c.Args.SSHHostKeyCallback, c.Args.TrustAllHosts, + c.throttleCh, c.Args.PrivateKeyPathFile) } func (c *baseClient) makeConnections(maker maker) { @@ -60,26 +64,31 @@ func (c *baseClient) makeConnections(maker maker) { discoveryService := discovery.New(c.Discovery, c.ServersStr, discovery.Shuffle) for _, server := range discoveryService.ServerList() { - c.connections = append(c.connections, c.makeConnection(server, c.sshAuthMethods, c.hostKeyCallback)) + c.connections = append(c.connections, c.makeConnection(server, + c.sshAuthMethods, c.hostKeyCallback)) } c.stats = newTailStats(len(c.connections)) } func (c *baseClient) Start(ctx context.Context, statsCh <-chan string) (status int) { - // Periodically check for unknown hosts, and ask the user whether to trust them or not. - go c.hostKeyCallback.PromptAddHosts(ctx) + dlog.Client.Trace("Starting base client") + // Can be nil when serverless. + if c.hostKeyCallback != nil { + // Periodically check for unknown hosts, and ask the user whether to trust them or not. + go c.hostKeyCallback.PromptAddHosts(ctx) + } // Print client stats every time something on statsCh is recieved. - go c.stats.Start(ctx, c.throttleCh, statsCh) - // Keep count of active connections - active := make(chan struct{}, len(c.connections)) + go c.stats.Start(ctx, c.throttleCh, statsCh, c.Args.Quiet) + 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) { - connStatus := c.start(ctx, active, i, conn) - // Update global status. + for i, conn := range c.connections { + go func(i int, conn connectors.Connector) { + defer wg.Done() + connStatus := c.startConnection(ctx, i, conn) mutex.Lock() defer mutex.Unlock() if connStatus > status { @@ -88,15 +97,12 @@ func (c *baseClient) Start(ctx context.Context, statsCh <-chan string) (status i }(i, conn) } - c.waitUntilDone(ctx, active) + wg.Wait() return } -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 }() +func (c *baseClient) startConnection(ctx context.Context, i int, + conn connectors.Connector) (status int) { for { connCtx, cancel := context.WithCancel(ctx) @@ -104,46 +110,25 @@ func (c *baseClient) start(ctx context.Context, active chan struct{}, i int, con 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() + 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) + dlog.Client.Debug(conn.Server(), "Reconnecting") + conn = c.makeConnection(conn.Server(), c.sshAuthMethods, c.hostKeyCallback) c.connections[i] = conn } } -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) 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 && c.retry { - <-ctx.Done() - } - - for { - numActive := len(active) - if numActive == 0 { - return - } - logger.Debug("Active connections", numActive) - time.Sleep(time.Second) +func (c *baseClient) makeConnection(server string, sshAuthMethods []gossh.AuthMethod, + hostKeyCallback client.HostKeyCallback) connectors.Connector { + if c.Args.Serverless { + return connectors.NewServerless(c.UserName, c.maker.makeHandler(server), + c.maker.makeCommands()) } + return connectors.NewServerConnection(server, c.UserName, sshAuthMethods, + hostKeyCallback, c.maker.makeHandler(server), c.maker.makeCommands()) } diff --git a/internal/clients/catclient.go b/internal/clients/catclient.go index d8e9196..bd65560 100644 --- a/internal/clients/catclient.go +++ b/internal/clients/catclient.go @@ -7,6 +7,8 @@ import ( "strings" "github.com/mimecast/dtail/internal/clients/handlers" + "github.com/mimecast/dtail/internal/config" + "github.com/mimecast/dtail/internal/io/dlog" "github.com/mimecast/dtail/internal/omode" ) @@ -16,11 +18,10 @@ type CatClient struct { } // NewCatClient returns a new cat client. -func NewCatClient(args Args) (*CatClient, error) { +func NewCatClient(args config.Args) (*CatClient, error) { if args.RegexStr != "" { return nil, errors.New("Can't use regex with 'cat' operating mode") } - args.Mode = omode.CatClient c := CatClient{ @@ -33,7 +34,6 @@ func NewCatClient(args Args) (*CatClient, error) { c.init() c.makeConnections(c) - return &c, nil } @@ -42,8 +42,13 @@ func (c CatClient) makeHandler(server string) handlers.Handler { } func (c CatClient) makeCommands() (commands []string) { + regex, err := c.Regex.Serialize() + if err != nil { + dlog.Client.FatalPanic(err) + } for _, file := range strings.Split(c.What, ",") { - commands = append(commands, fmt.Sprintf("%s %s %s", c.Mode.String(), file, c.Regex.Serialize())) + commands = append(commands, fmt.Sprintf("%s:%s %s %s", + c.Mode.String(), c.Args.SerializeOptions(), file, regex)) } return } diff --git a/internal/clients/connectors/connector.go b/internal/clients/connectors/connector.go new file mode 100644 index 0000000..3ab6a08 --- /dev/null +++ b/internal/clients/connectors/connector.go @@ -0,0 +1,17 @@ +package connectors + +import ( + "context" + + "github.com/mimecast/dtail/internal/clients/handlers" +) + +// Connector interface. +type Connector interface { + // Start the connection. + Start(ctx context.Context, cancel context.CancelFunc, throttleCh, statsCh chan struct{}) + // Server hostname. + Server() string + // Handler for the connection. + Handler() handlers.Handler +} diff --git a/internal/clients/connectors/serverconnection.go b/internal/clients/connectors/serverconnection.go new file mode 100644 index 0000000..1df4d73 --- /dev/null +++ b/internal/clients/connectors/serverconnection.go @@ -0,0 +1,203 @@ +package connectors + +import ( + "context" + "fmt" + "io" |