diff --git a/constants.go b/constants.go index b5e8eb35080..d5e4c3e6a36 100644 --- a/constants.go +++ b/constants.go @@ -51,12 +51,6 @@ const ( ) const ( - // Component indicates a component of teleport, used for logging - Component = "component" - - // ComponentFields stores component-specific fields - ComponentFields = "fields" - // ComponentReverseTunnelServer is reverse tunnel server // that together with agent establish a bi-directional SSH revers tunnel // to bypass firewall restrictions @@ -77,7 +71,10 @@ const ( ComponentProxy = "proxy" // ComponentTunClient is a tunnel client - ComponentTunClient = "tunclient" + ComponentTunClient = "client:tunnel" + + // ComponentCachingClient is a caching auth client + ComponentCachingClient = "client:cache" // DebugEnvVar tells tests to use verbose debug output DebugEnvVar = "DEBUG" diff --git a/docker/Makefile b/docker/Makefile index ff9fab5def4..ffe4e9d0274 100644 --- a/docker/Makefile +++ b/docker/Makefile @@ -48,3 +48,7 @@ enter-two-node: .PHONY: setup-tc setup-tc: docker exec -i two-auth /bin/bash -c "tctl -c /root/go/src/github.com/gravitational/teleport/docker/two-auth.yaml create -f /root/go/src/github.com/gravitational/teleport/docker/two-tc.yaml" + +.PHONY: delete-tc +delete-tc: + docker exec -i two-auth /bin/bash -c "tctl -c /root/go/src/github.com/gravitational/teleport/docker/two-auth.yaml rm tc/one" diff --git a/docker/docker-compose.yml b/docker/docker-compose.yml index 3fbd604775e..f76d9b0fefa 100644 --- a/docker/docker-compose.yml +++ b/docker/docker-compose.yml @@ -18,6 +18,8 @@ services: networks: teleport: ipv4_address: 172.10.1.1 + aliases: + - one-lb # # one-proxy is a second xproxy of the first cluster @@ -36,6 +38,8 @@ services: networks: teleport: ipv4_address: 172.10.1.10 + aliases: + - one-lb # # two-auth is a auth server of the second cluster diff --git a/docker/two-tc.yaml b/docker/two-tc.yaml index 9921e611c42..e41b993d5d9 100644 --- a/docker/two-tc.yaml +++ b/docker/two-tc.yaml @@ -5,5 +5,5 @@ metadata: spec: enabled: true token: "bar" - tunnel_addr: one:3024 - web_proxy_addr: one:3080 + tunnel_addr: one-lb:3024 + web_proxy_addr: one-lb:3080 diff --git a/lib/auth/permissions.go b/lib/auth/permissions.go index f7c060db307..a2975511b2a 100644 --- a/lib/auth/permissions.go +++ b/lib/auth/permissions.go @@ -176,6 +176,7 @@ func GetCheckerForBuiltinRole(role teleport.Role) (services.AccessChecker, error services.NewRule(services.KindRole, services.RO()), services.NewRule(services.KindAuthServer, services.RO()), services.NewRule(services.KindReverseTunnel, services.RO()), + services.NewRule(services.KindTunnelConnection, services.RO()), }, }, }) diff --git a/lib/auth/tun.go b/lib/auth/tun.go index d17eef51869..a21d3c0eacd 100644 --- a/lib/auth/tun.go +++ b/lib/auth/tun.go @@ -76,6 +76,8 @@ type TunClient struct { // embed auth API HTTP client Client + *log.Entry + user string // static auth servers are CAs set via configuration (--auth flag) and @@ -776,6 +778,12 @@ func NewTunClient(purpose string, return nil, trace.Wrap(err) } tc := &TunClient{ + Entry: log.WithFields(log.Fields{ + trace.Component: teleport.ComponentTunClient, + trace.ComponentFields: log.Fields{ + "purpose": purpose, + }, + }), purpose: purpose, user: user, staticAuthServers: authServers, @@ -786,7 +794,7 @@ func NewTunClient(purpose string, for _, o := range opts { o(tc) } - log.Debugf("NewTunClient(%v) with auth: %v", purpose, authServers) + tc.Debugf("created, auth servers: %v", authServers) clt, err := NewClient("http://stub:0", tc.Dial) if err != nil { @@ -799,7 +807,7 @@ func NewTunClient(purpose string, cachedAuthServers, err := tc.addrStorage.GetAddresses() if err != nil { if !trace.IsNotFound(err) { - log.Warnf("unable to load the auth server cache: %s", err.Error()) + tc.Warnf("unable to load the auth server cache: %s", err.Error()) } } else { tc.setAuthServers(cachedAuthServers) @@ -828,7 +836,7 @@ func (c *TunClient) String() string { // Close releases all the resources allocated for this client func (c *TunClient) Close() error { if c != nil { - log.Debugf("%v.Close()", c) + c.Debugf("is closing") c.GetTransport().CloseIdleConnections() c.closeOnce.Do(func() { close(c.closeC) @@ -850,7 +858,7 @@ func (c *TunClient) GetDialer() AccessPointDialer { } time.Sleep(4 * time.Duration(attempt) * dialRetryInterval) } - log.Errorf("%v: ", err) + c.Error("%v", err) return nil, trace.Wrap(err) } } @@ -877,7 +885,7 @@ func (c *TunClient) GetAgent() (AgentCloser, error) { // Dial dials to Auth server's HTTP API over SSH tunnel. func (c *TunClient) Dial(network, address string) (net.Conn, error) { - log.Debugf("TunClient[%s].Dial()", c.purpose) + c.Debugf("dialing %v %v", network, address) client, err := c.getClient() if err != nil { @@ -918,7 +926,7 @@ func (c *TunClient) fetchAndSync() error { // authServersSyncLoop continuously refreshes the list of available auth servers // for this client func (c *TunClient) authServersSyncLoop() { - log.Debugf("%v: authServersSyncLoop() started", c) + c.Debugf("authServersSyncLoop started") defer c.refreshTicker.Stop() // initial fetch for quick start-ups @@ -930,7 +938,7 @@ func (c *TunClient) authServersSyncLoop() { c.fetchAndSync() // received a signal to quit? case <-c.closeC: - log.Debugf("%v: authServersSyncLoop() exited", c) + c.Debugf("authServersSyncLoop exited") return } } @@ -995,7 +1003,7 @@ func (c *TunClient) getClient() (client *ssh.Client, err error) { if len(authServers) == 0 { return nil, trace.ConnectionProblem(nil, "all auth servers are offline") } - log.Debugf("%v.authServers: %v", c, authServers) + c.Debugf("auth servers: %v", authServers) // try to connect to the 1st one who will pick up: for _, authServer := range authServers { @@ -1011,7 +1019,7 @@ func (c *TunClient) getClient() (client *ssh.Client, err error) { if trace.IsAccessDenied(err) { return nil, trace.Wrap(err) } - log.Errorf("%v.getClient() error while connecting to auth server %v: %v: throttling", c, authServer, err) + c.Errorf("getClient error while connecting to auth server %v: %v: throttling", authServer, err) c.throttleAuthServer(authServer.String()) } return nil, trace.ConnectionProblem(nil, "all auth servers are offline") @@ -1025,7 +1033,7 @@ func (c *TunClient) dialAuthServer(authServer utils.NetAddr) (sshClient *ssh.Cli } const dialRetryTimes = 1 for attempt := 0; attempt < dialRetryTimes; attempt++ { - log.Debugf("%v.Dial(to=%v, attempt=%d)", c, authServer.Addr, attempt+1) + c.Debugf("dialing %v, attempt %d", authServer.Addr, attempt+1) sshClient, err = ssh.Dial(authServer.AddrNetwork, authServer.Addr, config) // success -> get out of here if err == nil { diff --git a/lib/backend/dir/impl.go b/lib/backend/dir/impl.go index d972423ac3a..492f50a092b 100644 --- a/lib/backend/dir/impl.go +++ b/lib/backend/dir/impl.go @@ -58,6 +58,8 @@ type Backend struct { // InternalClock is a test-friendly source of current time InternalClock clockwork.Clock + + *log.Entry } func (b *Backend) Clock() clockwork.Clock { @@ -82,6 +84,12 @@ func New(params backend.Params) (backend.Backend, error) { bk := &Backend{ RootDir: rootDir, InternalClock: clockwork.NewRealClock(), + Entry: log.WithFields(log.Fields{ + trace.Component: "fs", + trace.ComponentFields: log.Fields{ + "dir": rootDir, + }, + }), } locksDir := path.Join(bk.RootDir, locksBucket) @@ -244,7 +252,7 @@ func removeFiles(dir string) error { // AcquireLock grabs a lock that will be released automatically in TTL func (bk *Backend) AcquireLock(token string, ttl time.Duration) (err error) { - log.Debugf("fs.AcquireLock(%s)", token) + bk.Debugf("AcquireLock(%s)", token) if err = backend.ValidateLockTTL(ttl); err != nil { return trace.Wrap(err) @@ -271,7 +279,7 @@ func (bk *Backend) AcquireLock(token string, ttl time.Duration) (err error) { // ReleaseLock forces lock release before TTL func (bk *Backend) ReleaseLock(token string) (err error) { - log.Debugf("fs.ReleaseLock(%s)", token) + bk.Debugf("ReleaseLock(%s)", token) if err = bk.DeleteKey([]string{locksBucket}, token); err != nil { if !os.IsNotExist(err) { diff --git a/lib/reversetunnel/agent.go b/lib/reversetunnel/agent.go index 18e47441fab..3a51c4dfba3 100644 --- a/lib/reversetunnel/agent.go +++ b/lib/reversetunnel/agent.go @@ -37,10 +37,29 @@ import ( "github.com/gravitational/teleport/lib/utils/proxy" "github.com/gravitational/trace" + "github.com/jonboulle/clockwork" log "github.com/sirupsen/logrus" "golang.org/x/crypto/ssh" ) +const ( + // agentStateConnecting is when agent is connecting to the target + // without particular purpose + agentStateConnecting = "connecting" + // agentStateDiscovering is when agent is created with a goal + // to discover one or many proxies + agentStateDiscovering = "discovering" + // agentStateConnected means that agent has connected to instance + agentStateConnected = "connected" + // agentStateDiscovered means that agent has discovered the right proxy + agentStateDiscovered = "discovered" + // agentStateMissed means that agent has connected, + // but not to the one of the instances it was targeted to discover + agentStateMissed = "missed" + // agentStateClosed is for closed agents + agentStateClosed = "closed" +) + // AgentConfig holds configuration for agent type AgentConfig struct { // Addr is target address to dial @@ -60,6 +79,12 @@ type AgentConfig struct { DiscoveryC chan *discoveryRequest // Username is the name of this client used to authenticate on SSH Username string + // DiscoverProxies is set when the agent is created in discovery mode + // and is set to connect to one of the target proxies from the list + DiscoverProxies []services.Server + // Clock is a clock passed in tests, if not set wall clock + // will be used + Clock clockwork.Clock } // CheckAndSetDefaults checks parameters and sets default values @@ -85,45 +110,93 @@ func (a *AgentConfig) CheckAndSetDefaults() error { if len(a.Username) == 0 { return trace.BadParameter("missing parameter Username") } + if a.Clock == nil { + a.Clock = clockwork.NewRealClock() + } return nil } // Agent is a reverse tunnel agent running as a part of teleport Proxies // to establish outbound reverse tunnels to remote proxies type Agent struct { + sync.RWMutex *log.Entry AgentConfig ctx context.Context cancel context.CancelFunc hostKeyCallback utils.HostKeyCallback authMethods []ssh.AuthMethod + // state is the state of this agent + state string + // stateChange records last time the state was changed + stateChange time.Time + // principals is the list of principals of the server this agent + // is currently connected to + principals []string } // NewAgent returns a new reverse tunnel agent -// Parameters: -// addr points to the remote reverse tunnel server -// remoteDomainName is the domain name of the runnel server, used only for logging -// clientName is hostid.domain (where 'domain' is local domain name) func NewAgent(cfg AgentConfig) (*Agent, error) { + if err := cfg.CheckAndSetDefaults(); err != nil { + return nil, trace.Wrap(err) + } ctx, cancel := context.WithCancel(cfg.Context) - a := &Agent{ AgentConfig: cfg, - Entry: log.WithFields(log.Fields{ - teleport.Component: teleport.ComponentReverseTunnelAgent, - teleport.ComponentFields: map[string]interface{}{ - "remote": cfg.Addr.String(), - "client": cfg.Username, - }, - }), ctx: ctx, cancel: cancel, authMethods: []ssh.AuthMethod{ssh.PublicKeys(cfg.Signers...)}, } + if len(cfg.DiscoverProxies) == 0 { + a.state = agentStateConnecting + } else { + a.state = agentStateDiscovering + } + a.Entry = log.WithFields(log.Fields{ + trace.Component: teleport.ComponentReverseTunnelAgent, + trace.ComponentFields: map[string]interface{}{ + "remote": cfg.Addr.String(), + }, + }) a.hostKeyCallback = a.checkHostSignature + + if len(a.DiscoverProxies) != 0 { + a.setState(agentStateDiscovering) + } else { + a.setState(agentStateConnecting) + } + return a, nil } +func (a *Agent) String() string { + if len(a.DiscoverProxies) == 0 { + return fmt.Sprintf("agent -> cluster %v, target %v", a.RemoteCluster, a.Addr.String()) + } + return fmt.Sprintf("agent -> cluster %v, target %v, discover %v", a.RemoteCluster, a.Addr.String(), Proxies(a.DiscoverProxies)) +} + +func (a *Agent) getLastStateChange() time.Time { + a.RLock() + defer a.RUnlock() + return a.stateChange +} + +func (a *Agent) setState(state string) { + a.Lock() + defer a.Unlock() + prev := a.state + a.Debugf("changing state %v -> %v", prev, state) + a.state = state + a.stateChange = a.Clock.Now().UTC() +} + +func (a *Agent) getState() string { + a.RLock() + defer a.RUnlock() + return a.state +} + // Close signals to close all connections func (a *Agent) Close() error { a.cancel() @@ -146,9 +219,39 @@ func (a *Agent) Wait() error { return nil } -// String returns debug-friendly -func (a *Agent) String() string { - return fmt.Sprintf("tunagent(remote=%s)", a.Addr.String()) +func (a *Agent) connectedToRightProxy() bool { + principals := a.getPrincipals() + for _, proxy := range a.DiscoverProxies { + proxyID := fmt.Sprintf("%v.%v", proxy.GetName(), a.RemoteCluster) + if _, ok := principals[proxyID]; ok { + return true + } + } + return false +} + +func (a *Agent) setPrincipals(principals []string) { + a.Lock() + defer a.Unlock() + a.principals = principals +} + +func (a *Agent) getPrincipalsList() []string { + a.RLock() + defer a.RUnlock() + out := make([]string, len(a.principals)) + copy(out, a.principals) + return out +} + +func (a *Agent) getPrincipals() map[string]struct{} { + a.RLock() + defer a.RUnlock() + out := make(map[string]struct{}, len(a.principals)) + for _, p := range a.principals { + out[p] = struct{}{} + } + return out } func (a *Agent) checkHostSignature(hostport string, remote net.Addr, key ssh.PublicKey) error { @@ -167,7 +270,7 @@ func (a *Agent) checkHostSignature(hostport string, remote net.Addr, key ssh.Pub } for _, checker := range checkers { if sshutils.KeysEqual(checker, cert.SignatureKey) { - a.Debugf("matched key %v for %v", ca.GetName(), hostport) + a.setPrincipals(cert.ValidPrincipals) return nil } } @@ -187,6 +290,16 @@ func (a *Agent) connect() (conn *ssh.Client, err error) { Timeout: defaults.DefaultDialTimeout, }) if conn != nil { + if len(a.DiscoverProxies) != 0 { + if a.connectedToRightProxy() { + a.setState(agentStateDiscovered) + } else { + a.Debugf("missed, connected to %v instead of %v", a.getPrincipalsList(), Proxies(a.DiscoverProxies)) + a.setState(agentStateMissed) + } + } else { + a.setState(agentStateConnected) + } break } } @@ -345,6 +458,7 @@ func (a *Agent) runHeartbeat(conn *ssh.Client) { } newAccesspointC := conn.HandleChannelOpen(chanAccessPoint) newTransportC := conn.HandleChannelOpen(chanTransport) + newDiscoveryC := conn.HandleChannelOpen(chanDiscovery) // send first ping right away, then start a ping timer: hb.SendRequest("ping", false, nil) @@ -392,17 +506,17 @@ func (a *Agent) runHeartbeat(conn *ssh.Client) { } go a.proxyTransport(ch, req) // new discovery request - case nch := <-newTransportC: + case nch := <-newDiscoveryC: if nch == nil { continue } - a.Debugf("transport request: %v", nch.ChannelType()) + a.Debugf("discovery request: %v", nch.ChannelType()) ch, req, err := nch.Accept() if err != nil { a.Warningf("failed to accept request: %v", err) continue } - go a.proxyTransport(ch, req) + go a.handleDiscovery(ch, req) } } } @@ -410,12 +524,17 @@ func (a *Agent) runHeartbeat(conn *ssh.Client) { // run heartbeat loop, and when it fails (probably means that a tunnel got disconnected) // keep repeating to reconnect until we're asked to stop err := heartbeatLoop() + if len(a.DiscoverProxies) != 0 { + a.setState(agentStateDiscovering) + } else { + a.setState(agentStateConnecting) + } // when this happens, this is #1 issue we have right now with Teleport. So I'm making // it EASY to see in the logs. This condition should never be permanent (like repeates // every XX seconds) if err != nil { - log.Warn(err) + a.Warn(err) } if err != nil || conn == nil { @@ -456,6 +575,8 @@ func (a *Agent) handleDiscovery(ch ssh.Channel, reqC <-chan *ssh.Request) { a.Warningf("bad payload: %v", err) return } + r.ClusterName = a.RemoteCluster + r.ClusterAddr = a.Addr select { case a.DiscoveryC <- r: case <-a.ctx.Done(): @@ -463,7 +584,6 @@ func (a *Agent) handleDiscovery(ch ssh.Channel, reqC <-chan *ssh.Request) { return default: } - req.Reply(true, []byte("thanks")) } } } diff --git a/lib/reversetunnel/agentpool.go b/lib/reversetunnel/agentpool.go index 4ccd2f74a37..289baf113a9 100644 --- a/lib/reversetunnel/agentpool.go +++ b/lib/reversetunnel/agentpool.go @@ -25,7 +25,7 @@ type AgentPool struct { sync.Mutex *log.Entry cfg AgentPoolConfig - agents map[agentKey]*Agent + agents map[agentKey][]*Agent ctx context.Context cancel context.CancelFunc discoveryC chan *discoveryRequest @@ -45,6 +45,8 @@ type AgentPoolConfig struct { HostUUID string // Context is an optional context Context context.Context + // Cluster is a cluster name + Cluster string } // CheckAndSetDefaults checks and sets defaults @@ -74,14 +76,17 @@ func NewAgentPool(cfg AgentPoolConfig) (*AgentPool, error) { } ctx, cancel := context.WithCancel(cfg.Context) pool := &AgentPool{ - agents: make(map[agentKey]*Agent), + agents: make(map[agentKey][]*Agent), cfg: cfg, ctx: ctx, cancel: cancel, discoveryC: make(chan *discoveryRequest), } pool.Entry = log.WithFields(log.Fields{ - teleport.Component: teleport.ComponentReverseTunnelAgent, + trace.Component: teleport.ComponentReverseTunnelAgent, + trace.ComponentFields: map[string]interface{}{ + "cluster": cfg.Cluster, + }, }) return pool, nil } @@ -89,6 +94,7 @@ func NewAgentPool(cfg AgentPoolConfig) (*AgentPool, error) { // Start starts the agent pool func (m *AgentPool) Start() error { go m.pollAndSyncAgents() + go m.processDiscoveryRequests() return nil } @@ -117,11 +123,38 @@ func (m *AgentPool) processDiscoveryRequests() { m.Debugf("channel closed") return } - m.Debugf("got discovery request, following proxies are not connected %v", req.Proxies) + m.tryDiscover(*req) } } } +func (m *AgentPool) tryDiscover(req discoveryRequest) { + m.Debugf("will need to discover: %v", Proxies(req.Proxies)) + m.Lock() + defer m.Unlock() + proxies := Proxies(req.Proxies) + matchKey := req.key() + var foundAgent bool + // close agents that are discovering proxies that are somehow + // different from discovery request + m.closeAgentsIf(&matchKey, func(agent *Agent) bool { + if agent.getState() != agentStateDiscovering { + return false + } + if proxies.Equal(agent.DiscoverProxies) { + foundAgent = true + agent.Debugf("is already discovering %v, nothing to do", req) + return false + } + return true + }) + + // if we haven't found any discovery agent + if !foundAgent { + m.addAgent(req.key(), req.Proxies) + } +} + // FetchAndSyncAgents executes one time fetch and sync request // (used in tests instead of polling) func (m *AgentPool) FetchAndSyncAgents() error { @@ -135,6 +168,45 @@ func (m *AgentPool) FetchAndSyncAgents() error { return nil } +func (m *AgentPool) withLock(f func()) { + m.Lock() + defer m.Unlock() + f() +} + +type matchAgentFn func(a *Agent) bool + +func (m *AgentPool) closeAgentsIf(matchKey *agentKey, matchAgent matchAgentFn) { + if matchKey != nil { + m.agents[*matchKey] = filterAndClose(m.agents[*matchKey], matchAgent) + return + } + for key, agents := range m.agents { + m.agents[key] = filterAndClose(agents, matchAgent) + } +} + +func filterAndClose(agents []*Agent, matchAgent matchAgentFn) []*Agent { + var filtered []*Agent + for i := range agents { + agent := agents[i] + if matchAgent(agent) { + agent.Debugf("pool is closing agent") + agent.Close() + } else { + filtered = append(filtered, agent) + } + } + return filtered +} + +func (m *AgentPool) closeAgents(matchKey *agentKey) { + m.closeAgentsIf(matchKey, func(*Agent) bool { + // close all agents matching the matchKey + return true + }) +} + func (m *AgentPool) pollAndSyncAgents() { ticker := time.NewTicker(defaults.ReverseTunnelAgentHeartbeatPeriod) defer ticker.Stop() @@ -142,12 +214,10 @@ func (m *AgentPool) pollAndSyncAgents() { for { select { case <-m.ctx.Done(): + m.withLock(func() { + m.closeAgents(nil) + }) m.Debugf("closing") - m.Lock() - defer m.Unlock() - for _, a := range m.agents { - a.Close() - } return case <-ticker.C: err := m.FetchAndSyncAgents() @@ -159,6 +229,31 @@ func (m *AgentPool) pollAndSyncAgents() { } } +func (m *AgentPool) addAgent(key agentKey, discoverProxies []services.Server) error { + agent, err := NewAgent(AgentConfig{ + Addr: key.addr, + RemoteCluster: key.domainName, + Username: m.cfg.HostUUID, + Signers: m.cfg.HostSigners, + Client: m.cfg.Client, + AccessPoint: m.cfg.AccessPoint, + Context: m.ctx, + DiscoveryC: m.discoveryC, + DiscoverProxies: discoverProxies, + }) + if err != nil { + return trace.Wrap(err) + } + m.Debugf("adding %v", agent) + // start the agent in a goroutine. no need to handle Start() errors: Start() will be + // retrying itself until the agent is closed + go agent.Start() + agents, _ := m.agents[key] + agents = append(agents, agent) + m.agents[key] = agents + return nil +} + func (m *AgentPool) syncAgents(tunnels []services.ReverseTunnel) error { m.Lock() defer m.Unlock() @@ -168,33 +263,24 @@ func (m *AgentPool) syncAgents(tunnels []services.ReverseTunnel) error { return trace.Wrap(err) } agentsToAdd, agentsToRemove := diffTunnels(m.agents, keys) + // remove agents from deleted reverse tunnels for _, key := range agentsToRemove { - m.Debugf("removing %v", &key) - agent := m.agents[key] - delete(m.agents, key) - agent.Close() + m.closeAgents(&key) } - + // add agents from added reverse tunnels for _, key := range agentsToAdd { - m.Debugf("adding %v", &key) - agent, err := NewAgent(AgentConfig{ - Addr: key.addr, - RemoteCluster: key.domainName, - Username: m.cfg.HostUUID, - Signers: m.cfg.HostSigners, - Client: m.cfg.Client, - AccessPoint: m.cfg.AccessPoint, - Context: m.ctx, - DiscoveryC: m.discoveryC, - }) - if err != nil { + if err := m.addAgent(key, nil); err != nil { return trace.Wrap(err) } - // start the agent in a goroutine. no need to handle Start() errors: Start() will be - // retrying itself until the agent is closed - go agent.Start() - m.agents[key] = agent } + // garbage collect agents that have connected, but to the wrong proxy + m.closeAgentsIf(nil, func(agent *Agent) bool { + if agent.getState() == agentStateMissed { + agent.Debugf("closing agent that could not discover clusters") + return true + } + return false + }) return nil } @@ -224,7 +310,7 @@ func tunnelToAgentKeys(tunnel services.ReverseTunnel) ([]agentKey, error) { return out, nil } -func diffTunnels(existingTunnels map[agentKey]*Agent, arrivedKeys map[agentKey]bool) ([]agentKey, []agentKey) { +func diffTunnels(existingTunnels map[agentKey][]*Agent, arrivedKeys map[agentKey]bool) ([]agentKey, []agentKey) { var agentsToRemove, agentsToAdd []agentKey for existingKey := range existingTunnels { if _, ok := arrivedKeys[existingKey]; !ok { // agent was removed diff --git a/lib/reversetunnel/discovery.go b/lib/reversetunnel/discovery.go index 1f3584ffecd..8d71dab6357 100644 --- a/lib/reversetunnel/discovery.go +++ b/lib/reversetunnel/discovery.go @@ -2,14 +2,57 @@ package reversetunnel import ( "encoding/json" + "fmt" + "strings" "github.com/gravitational/teleport/lib/services" + "github.com/gravitational/teleport/lib/utils" "github.com/gravitational/trace" ) type discoveryRequest struct { - Proxies []services.Server `json:"proxies"` + ClusterName string `json:"-"` + ClusterAddr utils.NetAddr `json:"-"` + Proxies []services.Server `json:"proxies"` +} + +type Proxies []services.Server + +func (proxies Proxies) String() string { + var out []string + for _, proxy := range proxies { + out = append(out, proxy.GetName()) + } + return strings.Join(out, ",") +} + +func (proxies Proxies) Equal(other []services.Server) bool { + if len(proxies) != len(other) { + return false + } + proxiesMap, otherMap := make(map[string]bool), make(map[string]bool) + for i := range proxies { + proxiesMap[proxies[i].GetName()] = true + } + for i := range other { + otherMap[other[i].GetName()] = true + } + for key := range otherMap { + if !proxiesMap[key] { + return false + } + } + return true +} + +func (r discoveryRequest) key() agentKey { + return agentKey{domainName: r.ClusterName, addr: r.ClusterAddr} +} + +func (r discoveryRequest) String() string { + return fmt.Sprintf("discovery request, cluster name: %v, address: %v, proxies: %v", + r.ClusterName, r.ClusterAddr, Proxies(r.Proxies)) } type discoveryRequestRaw struct { diff --git a/lib/reversetunnel/localsite.go b/lib/reversetunnel/localsite.go index 5a38d4871f5..e3c822f24c1 100644 --- a/lib/reversetunnel/localsite.go +++ b/lib/reversetunnel/localsite.go @@ -40,8 +40,8 @@ func newlocalSite(srv *server, domainName string, client auth.ClientI) (*localSi accessPoint: accessPoint, domainName: domainName, log: log.WithFields(log.Fields{ - teleport.Component: teleport.ComponentReverseTunnelServer, - teleport.ComponentFields: map[string]string{ + trace.Component: teleport.ComponentReverseTunnelServer, + trace.ComponentFields: map[string]string{ "cluster": domainName, }, }), diff --git a/lib/reversetunnel/peer.go b/lib/reversetunnel/peer.go index 7035fae25a5..a4f83a3f661 100644 --- a/lib/reversetunnel/peer.go +++ b/lib/reversetunnel/peer.go @@ -127,8 +127,8 @@ func newClusterPeer(srv *server, connInfo services.TunnelConnection) (*clusterPe srv: srv, connInfo: connInfo, log: log.WithFields(log.Fields{ - teleport.Component: teleport.ComponentReverseTunnelServer, - teleport.ComponentFields: map[string]string{ + trace.Component: teleport.ComponentReverseTunnelServer, + trace.ComponentFields: map[string]string{ "cluster": connInfo.GetClusterName(), }, }), diff --git a/lib/reversetunnel/remotesite.go b/lib/reversetunnel/remotesite.go index a299509abb2..c35e12af5a0 100644 --- a/lib/reversetunnel/remotesite.go +++ b/lib/reversetunnel/remotesite.go @@ -34,9 +34,9 @@ import ( "github.com/gravitational/teleport/lib/utils" "github.com/gravitational/trace" - log "github.com/sirupsen/logrus" - + "github.com/jonboulle/clockwork" "github.com/mailgun/oxy/forward" + log "github.com/sirupsen/logrus" "golang.org/x/crypto/ssh" ) @@ -57,6 +57,7 @@ type remoteSite struct { accessPoint auth.AccessPoint connInfo services.TunnelConnection ctx context.Context + clock clockwork.Clock } func (s *remoteSite) CachingAccessPoint() (auth.AccessPoint, error) { @@ -137,11 +138,10 @@ func (s *remoteSite) GetStatus() string { if err != nil { return RemoteSiteStatusOffline } - diff := time.Now().Sub(connInfo.GetLastHeartbeat()) - if diff > defaults.ReverseTunnelOfflineThreshold { - return RemoteSiteStatusOffline + if s.isOnline(connInfo) { + return RemoteSiteStatusOnline } - return RemoteSiteStatusOnline + return RemoteSiteStatusOffline } func (s *remoteSite) registerHeartbeat(t time.Time) { @@ -192,7 +192,7 @@ func (s *remoteSite) periodicSendDiscoveryRequests() { ticker := time.NewTicker(defaults.ReverseTunnelAgentHeartbeatPeriod) defer ticker.Stop() if err := s.sendDiscoveryRequest(); err != nil { - s.Warningf("failed to fetch cluster peers: %v", err) + s.Warningf("failed to send discovery: %v", err) } for { select { @@ -202,12 +202,17 @@ func (s *remoteSite) periodicSendDiscoveryRequests() { case <-ticker.C: err := s.sendDiscoveryRequest() if err != nil { - s.Warningf("could not send discovery request: %v", err) + s.Warningf("could not send discovery request: %v", trace.DebugReport(err)) } } } } +func (s *remoteSite) isOnline(conn services.TunnelConnection) bool { + diff := s.clock.Now().Sub(conn.GetLastHeartbeat()) + return diff < defaults.ReverseTunnelOfflineThreshold +} + // findDisconnectedProxies func (s *remoteSite) findDisconnectedProxies() ([]services.Server, error) { conns, err := s.srv.AccessPoint.GetTunnelConnections(s.domainName) @@ -216,7 +221,9 @@ func (s *remoteSite) findDisconnectedProxies() ([]services.Server, error) { } connected := make(map[string]bool) for _, conn := range conns { - connected[conn.GetProxyName()] = true + if s.isOnline(conn) { + connected[conn.GetProxyName()] = true + } } proxies, err := s.srv.AccessPoint.GetProxies() if err != nil { @@ -240,7 +247,7 @@ func (s *remoteSite) sendDiscoveryRequest() error { if len(disconnectedProxies) == 0 { return nil } - s.Infof("detected disconnected proxies: %v", disconnectedProxies) + s.Debugf("going to request discovery for: %v", Proxies(disconnectedProxies)) req := discoveryRequest{ Proxies: disconnectedProxies, } @@ -257,18 +264,21 @@ func (s *remoteSite) sendDiscoveryRequest() error { if err != nil { return trace.Wrap(err) } - _, err = discoveryC.SendRequest("ping", false, payload) - remoteConn.markInvalid(err) - s.Errorf("disconnecting cluster on %v, err: %v", - remoteConn.conn.RemoteAddr(), - err) - return trace.Wrap(err) + _, err = discoveryC.SendRequest("discovery", false, payload) + if err != nil { + remoteConn.markInvalid(err) + s.Errorf("disconnecting cluster on %v, err: %v", + remoteConn.conn.RemoteAddr(), + err) + return trace.Wrap(err) + } + return nil } for i := 0; i < s.connectionCount(); i++ { err := send() if err != nil { - s.Warningf("%v") + s.Warningf("%v", err) } } return nil @@ -277,8 +287,6 @@ func (s *remoteSite) sendDiscoveryRequest() error { // dialAccessPoint establishes a connection from the proxy (reverse tunnel server) // back into the client using previously established tunnel. func (s *remoteSite) dialAccessPoint(network, addr string) (net.Conn, error) { - s.Debugf("dialAccessPoint") - try := func() (net.Conn, error) { remoteConn, err := s.nextConn() if err != nil { @@ -292,7 +300,7 @@ func (s *remoteSite) dialAccessPoint(network, addr string) (net.Conn, error) { err) return nil, trace.Wrap(err) } - s.Infof("success dialing to cluster") + s.Debugf("success dialing to cluster") return utils.NewChConn(remoteConn.sshConn, ch), nil } diff --git a/lib/reversetunnel/srv.go b/lib/reversetunnel/srv.go index a9c274d99e8..d0d63e91955 100644 --- a/lib/reversetunnel/srv.go +++ b/lib/reversetunnel/srv.go @@ -37,6 +37,7 @@ import ( "github.com/gravitational/teleport/lib/utils" "github.com/gravitational/trace" + "github.com/jonboulle/clockwork" log "github.com/sirupsen/logrus" "golang.org/x/crypto/ssh" ) @@ -76,6 +77,8 @@ type server struct { // ctx is a context used for signalling and broadcast ctx context.Context + + *log.Entry } // DirectCluster is used to access cluster directly @@ -106,6 +109,9 @@ type Config struct { DirectClusters []DirectCluster // Context is a signalling context Context context.Context + // Clock is a clock used in the server, set up to + // wall clock if not set + Clock clockwork.Clock } // CheckAndSetDefaults checks parameters and sets default values @@ -126,6 +132,9 @@ func (cfg *Config) CheckAndSetDefaults() error { return trace.Wrap(err) } } + if cfg.Clock == nil { + cfg.Clock = clockwork.NewRealClock() + } return nil } @@ -146,6 +155,9 @@ func NewServer(cfg Config) (Server, error) { ctx: ctx, cancel: cancel, clusterPeers: make(map[string]*clusterPeers), + Entry: log.WithFields(log.Fields{ + trace.Component: teleport.ComponentReverseTunnelServer, + }), } for _, clusterInfo := range cfg.DirectClusters { @@ -181,17 +193,17 @@ func (s *server) periodicFetchClusterPeers() { ticker := time.NewTicker(defaults.ReverseTunnelAgentHeartbeatPeriod) defer ticker.Stop() if err := s.fetchClusterPeers(); err != nil { - log.Warningf("[TUNNEL] failed to fetch cluster peers: %v", err) + s.Warningf("failed to fetch cluster peers: %v", err) } for { select { case <-s.ctx.Done(): - log.Debugf("[TUNNEL] closing") + s.Debugf("closing") return case <-ticker.C: err := s.fetchClusterPeers() if err != nil { - log.Warningf("[TUNNEL] failed to fetch cluster peers: %v", err) + s.Warningf("failed to fetch cluster peers: %v", err) } } } @@ -266,11 +278,11 @@ func (s *server) removeClusterPeers(conns []services.TunnelConnection) { for _, conn := range conns { peers, ok := s.clusterPeers[conn.GetClusterName()] if !ok { - log.Warningf("[TUNNEL] failed to remove cluster peer, not found peers for %v", conn) + s.Warningf("failed to remove cluster peer, not found peers for %v", conn) continue } peers.removePeer(conn) - log.Debugf("[TUNNEL] removed cluster peer %v", conn) + s.Debugf("removed cluster peer %v", conn) } } @@ -338,13 +350,13 @@ func (s *server) HandleNewChan(conn net.Conn, sconn *ssh.ServerConn, nch ssh.New if ct == "session" { msg = "Cannot open new SSH session on reverse tunnel. Are you connecting to the right port?" } - log.Warningf(msg) + s.Warning(msg) nch.Reject(ssh.ConnectionFailed, msg) return } - log.Debugf("[TUNNEL] new tunnel from %s", sconn.RemoteAddr()) + s.Debugf("new tunnel from %s", sconn.RemoteAddr()) if sconn.Permissions.Extensions[extCertType] != extCertTypeHost { - log.Error(trace.BadParameter("can't retrieve certificate type in certType")) + s.Error(trace.BadParameter("can't retrieve certificate type in certType")) return } // add the incoming site (cluster) to the list of active connections: @@ -369,7 +381,7 @@ func (s *server) HandleNewChan(conn net.Conn, sconn *ssh.ServerConn, nch ssh.New func (s *server) isHostAuthority(auth ssh.PublicKey) bool { keys, err := s.getTrustedCAKeys(services.HostCA) if err != nil { - log.Errorf("failed to retrieve trusted keys, err: %v", err) + s.Errorf("failed to retrieve trusted keys, err: %v", err) return false } for _, k := range keys { @@ -385,7 +397,7 @@ func (s *server) isHostAuthority(auth ssh.PublicKey) bool { func (s *server) isUserAuthority(auth ssh.PublicKey) bool { keys, err := s.getTrustedCAKeys(services.UserCA) if err != nil { - log.Errorf("failed to retrieve trusted keys, err: %v", err) + s.Errorf("failed to retrieve trusted keys, err: %v", err) return false } for _, k := range keys { @@ -435,7 +447,7 @@ func (s *server) checkTrustedKey(CertType services.CertAuthType, domainName stri } func (s *server) keyAuth(conn ssh.ConnMetadata, key ssh.PublicKey) (*ssh.Permissions, error) { - logger := log.WithFields(log.Fields{ + logger := s.WithFields(log.Fields{ "remote": conn.RemoteAddr(), "user": conn.User(), }) @@ -533,8 +545,7 @@ func (s *server) upsertSite(conn net.Conn, sshConn *ssh.ServerConn) (*remoteSite } s.remoteSites = append(s.remoteSites, site) } - log.Infof("[TUNNEL] cluster %v connected from %v. clusters: %d", - domainName, conn.RemoteAddr(), len(s.remoteSites)) + site.Infof("connection <- %v, clusters: %d", conn.RemoteAddr(), len(s.remoteSites)) // treat first connection as a registered heartbeat, // otherwise the connection information will appear after initial // heartbeat delay @@ -668,12 +679,13 @@ func newRemoteSite(srv *server, domainName string) (*remoteSite, error) { domainName: domainName, connInfo: connInfo, Entry: log.WithFields(log.Fields{ - teleport.Component: teleport.ComponentReverseTunnelServer, - teleport.ComponentFields: map[string]string{ + trace.Component: teleport.ComponentReverseTunnelServer, + trace.ComponentFields: map[string]interface{}{ "cluster": domainName, }, }), - ctx: srv.ctx, + ctx: srv.ctx, + clock: srv.Clock, } // transport uses connection do dial out to the remote address remoteSite.transport = &http.Transport{ @@ -692,6 +704,8 @@ func newRemoteSite(srv *server, domainName string) (*remoteSite, error) { remoteSite.accessPoint = accessPoint + go remoteSite.periodicSendDiscoveryRequests() + return remoteSite, nil } diff --git a/lib/service/service.go b/lib/service/service.go index ff98c8f75e1..573be7ed98c 100644 --- a/lib/service/service.go +++ b/lib/service/service.go @@ -690,7 +690,7 @@ func (process *TeleportProcess) initProxyEndpoint(conn *Connector) error { tsrv, err := reversetunnel.NewServer( reversetunnel.Config{ - ID: conn.Identity.ID.HostUUID, + ID: process.Config.HostUUID, ListenAddr: cfg.Proxy.ReverseTunnelListenAddr, HostSigners: []ssh.Signer{conn.Identity.KeySigner}, AccessPoint: authClient, @@ -732,6 +732,7 @@ func (process *TeleportProcess) initProxyEndpoint(conn *Connector) error { Client: conn.Client, AccessPoint: authClient, HostSigners: []ssh.Signer{conn.Identity.KeySigner}, + Cluster: conn.Identity.Cert.Extensions[utils.CertExtensionAuthority], }) if err != nil { return trace.Wrap(err) diff --git a/lib/services/local/presence.go b/lib/services/local/presence.go index 0b7577db89d..e6c93f3b020 100644 --- a/lib/services/local/presence.go +++ b/lib/services/local/presence.go @@ -24,6 +24,7 @@ import ( "github.com/gravitational/teleport/lib/services" "github.com/gravitational/trace" + log "github.com/sirupsen/logrus" ) // PresenceService records and reports the presence of all components @@ -363,7 +364,11 @@ func (s *PresenceService) GetTunnelConnection(clusterName, connectionName string } return nil, trace.Wrap(err) } - return services.UnmarshalTunnelConnection(data) + conn, err := services.UnmarshalTunnelConnection(data) + if err != nil { + log.Debugf("got some problem with data: %q", string(data)) + } + return conn, err } // GetTunnelConnections returns connections for a trusted cluster diff --git a/lib/services/tunnelconn.go b/lib/services/tunnelconn.go index d8fd4ce6268..f7127bf12b2 100644 --- a/lib/services/tunnelconn.go +++ b/lib/services/tunnelconn.go @@ -3,6 +3,7 @@ package services import ( "encoding/json" "fmt" + "runtime/debug" "strings" "time" @@ -172,7 +173,7 @@ type TunnelConnectionSpecV2 struct { // ProxyName is the name of the proxy server ProxyName string `json:"proxy_name"` // LastHeartbeat is a time of the last heartbeat - LastHeartbeat time.Time `json:"last_heartbeat"` + LastHeartbeat time.Time `json:"last_heartbeat,omitempty"` } // TunnelConnectionSpecV2Schema is JSON schema for reverse tunnel spec @@ -197,7 +198,8 @@ func GetTunnelConnectionSchema() string { // sets defaults and checks the schema func UnmarshalTunnelConnection(data []byte) (TunnelConnection, error) { if len(data) == 0 { - return nil, trace.BadParameter("missing tunnel data") + debug.PrintStack() + return nil, trace.BadParameter("missing tunnel connection data") } var h ResourceHeader err := json.Unmarshal(data, &h) diff --git a/lib/srv/sshserver.go b/lib/srv/sshserver.go index e5b4054578f..fe93f48ae78 100644 --- a/lib/srv/sshserver.go +++ b/lib/srv/sshserver.go @@ -301,7 +301,7 @@ func (s *Server) getNamespace() string { return services.ProcessNamespace(s.namespace) } -func (s *Server) logFields(fields map[string]interface{}) log.Fields { +func (s *Server) logFields(fields log.Fields) log.Fields { var component string if s.proxyMode { component = teleport.ComponentProxy @@ -309,8 +309,8 @@ func (s *Server) logFields(fields map[string]interface{}) log.Fields { component = teleport.ComponentNode } return log.Fields{ - teleport.Component: component, - teleport.ComponentFields: fields, + trace.Component: component, + trace.ComponentFields: fields, } } diff --git a/lib/sshutils/server.go b/lib/sshutils/server.go index 8554fecfe21..d8f9cd1e267 100644 --- a/lib/sshutils/server.go +++ b/lib/sshutils/server.go @@ -42,6 +42,8 @@ import ( // Server is a generic implementation of an SSH server. All Teleport // services (auth, proxy, ssh) use this as a base to accept SSH connections. type Server struct { + *log.Entry + // component is a name of the facility which uses this server, // used for logging/debugging. typically it's "proxy" or "auth api", etc component string @@ -107,10 +109,13 @@ func NewServer( return nil, err } s := &Server{ - component: component, + Entry: log.WithFields(log.Fields{ + trace.Component: "ssh:" + component, + }), addr: a, newChanHandler: h, closeC: make(chan struct{}), + component: component, } s.limiter, err = limiter.NewLimiter(limiter.LimiterConfig{}) if err != nil { @@ -151,7 +156,7 @@ func SetRequestHandler(req RequestHandler) ServerOption { func SetCiphers(ciphers []string) ServerOption { return func(s *Server) error { - log.Debugf("[SSH:%v] Supported Ciphers: %q", s.component, ciphers) + s.Debugf("supported ciphers: %q", ciphers) if ciphers != nil { s.cfg.Ciphers = ciphers } @@ -161,7 +166,7 @@ func SetCiphers(ciphers []string) ServerOption { func SetKEXAlgorithms(kexAlgorithms []string) ServerOption { return func(s *Server) error { - log.Debugf("[SSH:%v] Supported KEX algorithms: %q", s.component, kexAlgorithms) + s.Debugf("supported KEX algorithms: %q", kexAlgorithms) if kexAlgorithms != nil { s.cfg.KeyExchanges = kexAlgorithms } @@ -171,7 +176,7 @@ func SetKEXAlgorithms(kexAlgorithms []string) ServerOption { func SetMACAlgorithms(macAlgorithms []string) ServerOption { return func(s *Server) error { - log.Debugf("[SSH:%v] Supported MAC algorithms: %q", s.component, macAlgorithms) + s.Debugf("supported MAC algorithms: %q", macAlgorithms) if macAlgorithms != nil { s.cfg.MACs = macAlgorithms } @@ -190,7 +195,7 @@ func (s *Server) Start() error { return err } s.listener = socket - log.Infof("[SSH:%s] listening socket: %v", s.component, socket.Addr()) + s.Infof("listening socket: %v", socket.Addr()) go s.acceptConnections() return nil } @@ -215,21 +220,21 @@ func (s *Server) Close() error { func (s *Server) acceptConnections() { defer s.notifyClosed() addr := s.Addr() - log.Infof("[SSH:%v] is listening on %v", s.component, addr) + s.Infof("listening on %v", addr) for { conn, err := s.listener.Accept() if err != nil { if s.askedToClose { - log.Infof("[SSH:%v] server %v exited", s.component, addr) + s.Infof("server %v exited", addr) s.askedToClose = false return } // our best shot to avoid excessive logging if op, ok := err.(*net.OpError); ok && !op.Timeout() { - log.Debugf("[SSH:%v] closed socket %v", s.component, op) + s.Debugf("closed socket %v", op) return } - log.Errorf("SSH:%v accept error: %T %v", s.component, err, err) + s.Errorf("accept error: %T %v", err, err) return } go s.handleConnection(conn) @@ -278,12 +283,12 @@ func (s *Server) handleConnection(conn net.Conn) { return } // Connection successfully initiated - log.Infof("[SSH:%v] new connection %v -> %v vesion: %v", - s.component, sconn.RemoteAddr(), sconn.LocalAddr(), string(sconn.ClientVersion())) + s.Debugf("incoming connection %v -> %v vesion: %v", + sconn.RemoteAddr(), sconn.LocalAddr(), string(sconn.ClientVersion())) // will be called when the connection is closed connClosed := func() { - log.Infof("[SSH:%v] closed connection", s.component) + s.Debugf("closed connection %v", sconn.RemoteAddr()) } // The keepalive ticket will ensure that SSH keepalive requests are being sent @@ -300,7 +305,7 @@ func (s *Server) handleConnection(conn net.Conn) { connClosed() return } - log.Infof("[SSH:%v] recieved out-of-band request: %+v", s.component, req) + s.Debugf("recieved out-of-band request: %+v", req) if s.reqHandler != nil { go s.reqHandler.HandleRequest(req) } diff --git a/lib/state/cachingaccesspoint.go b/lib/state/cachingaccesspoint.go index eeff683a553..dcd58756480 100644 --- a/lib/state/cachingaccesspoint.go +++ b/lib/state/cachingaccesspoint.go @@ -21,6 +21,7 @@ import ( "fmt" "time" + "github.com/gravitational/teleport" "github.com/gravitational/teleport/lib/auth" "github.com/gravitational/teleport/lib/backend" "github.com/gravitational/teleport/lib/defaults" @@ -57,6 +58,7 @@ const ( // This which can be used if the upstream AccessPoint goes offline type CachingAuthClient struct { Config + *log.Entry // ap points to the access ponit we're caching access to: ap auth.AccessPoint @@ -115,6 +117,9 @@ func NewCachingAuthClient(config Config) (*CachingAuthClient, error) { trust: local.NewCAService(config.Backend), access: local.NewAccessService(config.Backend), presence: local.NewPresenceService(config.Backend), + Entry: log.WithFields(log.Fields{ + trace.Component: teleport.ComponentCachingClient, + }), } if !cs.SkipPreload { err := cs.fetchAll() @@ -122,7 +127,7 @@ func NewCachingAuthClient(config Config) (*CachingAuthClient, error) { // we almost always get some "access denied" errors here because // not all cacheable resources are available (for example nodes do // not have access to tunnels) - log.Debugf("Auth cache: %v", err) + cs.Debugf("auth cache: %v", err) } } return cs, nil @@ -504,14 +509,14 @@ func (cs *CachingAuthClient) UpsertTunnelConnection(conn services.TunnelConnecti func (cs *CachingAuthClient) try(f func() error) error { tooSoon := cs.lastErrorTime.Add(backoffDuration).After(time.Now()) if tooSoon { - log.Warnf("Backoff: using cached value due to recent errors") + cs.Warnf("backoff: using cached value due to recent errors") return trace.ConnectionProblem(fmt.Errorf("backoff"), "backing off due to recent errors") } accessPointRequests.Inc() err := trace.ConvertSystemError(f()) if trace.IsConnectionProblem(err) { cs.lastErrorTime = time.Now() - log.Warningf("Connection Problem: failed connect to the auth servers, using local cache") + cs.Warningf("connection problem: failed connect to the auth servers, using local cache") } return err } diff --git a/lib/utils/proxy/proxy.go b/lib/utils/proxy/proxy.go index 66bd0aa49a5..5a5616b4848 100644 --- a/lib/utils/proxy/proxy.go +++ b/lib/utils/proxy/proxy.go @@ -162,20 +162,22 @@ func getProxyAddress() string { strings.ToLower(teleport.HTTPProxy), } + l := log.WithFields(log.Fields{trace.Component: "http:proxy"}) + for _, v := range envs { addr := os.Getenv(v) if addr != "" { proxyaddr, err := parse(addr) if err != nil { - log.Debugf("[HTTP PROXY] Unable to parse environment variable %q: %q.", v, addr) + l.Debugf("unable to parse environment variable %q: %q.", v, addr) continue } - log.Debugf("[HTTP PROXY] Successfully parsed environment variable %q: %q to %q", v, addr, proxyaddr) + l.Debugf("successfully parsed environment variable %q: %q to %q", v, addr, proxyaddr) return proxyaddr } } - log.Debugf("[HTTP PROXY] No valid environment variables found.") + l.Debugf("no valid environment variables found.") return "" } diff --git a/vendor/github.com/gravitational/trace/log.go b/vendor/github.com/gravitational/trace/log.go index 310a90af486..f42edcc2e7f 100644 --- a/vendor/github.com/gravitational/trace/log.go +++ b/vendor/github.com/gravitational/trace/log.go @@ -39,17 +39,26 @@ const ( LevelField = "level" // Component is a field that represents component - e.g. service or // function - Component = "component" + Component = "trace.component" + // ComponentFields is a fields compoonent + ComponentFields = "trace.fields" ) // TextFormatter is logrus-compatible formatter and adds // file and line details to every logged entry. type TextFormatter struct { + // DisableTimestamp disables timestamp output (useful when outputting to + // systemd logs) DisableTimestamp bool } // Format implements logrus.Formatter interface and adds file and line -func (tf *TextFormatter) Format(e *log.Entry) ([]byte, error) { +func (tf *TextFormatter) Format(e *log.Entry) (data []byte, err error) { + defer func() { + if r := recover(); r != nil { + err = BadParameter("recovered from panic: %v", r) + } + }() var file string if frameNo := findFrame(); frameNo != -1 { t := newTrace(frameNo, nil) @@ -183,6 +192,11 @@ func (w *writer) writeMap(m map[string]interface{}) { w.WriteString(":{") w.writeMap(val) w.WriteString(" }") + case log.Fields: + w.WriteString(key) + w.WriteString(":{") + w.writeMap(val) + w.WriteString(" }") default: w.writeKeyValue(key, val) }