@ -208,6 +208,9 @@ type Agent struct {
// proxyManager is the proxy process manager for managed Connect proxies.
// proxyManager is the proxy process manager for managed Connect proxies.
proxyManager * proxy . Manager
proxyManager * proxy . Manager
// proxyLock protects proxy information in the local state from concurrent modification
proxyLock sync . Mutex
} }
func New ( c * config . RuntimeConfig ) ( * Agent , error ) { func New ( c * config . RuntimeConfig ) ( * Agent , error ) {
@ -1616,16 +1619,21 @@ func (a *Agent) purgeService(serviceID string) error {
type persistedProxy struct { type persistedProxy struct {
ProxyToken string
ProxyToken string
Proxy * structs . ConnectManagedProxy
Proxy * structs . ConnectManagedProxy
// Set to true when the proxy information originated from the agents configuration
// as opposed to API registration.
FromFile bool
} }
// persistProxy saves a proxy definition to a JSON file in the data dir
// persistProxy saves a proxy definition to a JSON file in the data dir
func ( a * Agent ) persistProxy ( proxy * local . ManagedProxy ) error { func ( a * Agent ) persistProxy ( proxy * local . ManagedProxy , FromFile bool ) error {
proxyPath := filepath . Join ( a . config . DataDir , proxyDir ,
proxyPath := filepath . Join ( a . config . DataDir , proxyDir ,
stringHash ( proxy . Proxy . ProxyService . ID ) )
stringHash ( proxy . Proxy . ProxyService . ID ) )
wrapped := persistedProxy {
wrapped := persistedProxy {
ProxyToken : proxy . ProxyToken ,
ProxyToken : proxy . ProxyToken ,
Proxy : proxy . Proxy ,
Proxy : proxy . Proxy ,
FromFile : FromFile ,
}
}
encoded , err := json . Marshal ( wrapped )
encoded , err := json . Marshal ( wrapped )
if err != nil {
if err != nil {
@ -2076,7 +2084,9 @@ func (a *Agent) RemoveCheck(checkID types.CheckID, persist bool) error {
return nil
return nil
} }
// AddProxy adds a new local Connect Proxy instance to be managed by the agent.
// addProxyLocked adds a new local Connect Proxy instance to be managed by the agent.
//
// This assumes that the agent's proxyLock is already held
//
//
// It REQUIRES that the service that is being proxied is already present in the
// It REQUIRES that the service that is being proxied is already present in the
// local state. Note that this is only used for agent-managed proxies so we can
// local state. Note that this is only used for agent-managed proxies so we can
@ -2091,7 +2101,7 @@ func (a *Agent) RemoveCheck(checkID types.CheckID, persist bool) error {
// definitions from disk; new proxies must leave it blank to get a new token
// definitions from disk; new proxies must leave it blank to get a new token
// assigned. We need to restore from disk to enable to continue authenticating
// assigned. We need to restore from disk to enable to continue authenticating
// running proxies that already had that credential injected.
// running proxies that already had that credential injected.
func ( a * Agent ) AddProxy ( proxy * structs . ConnectManagedProxy , persist bool , func ( a * Agent ) addProxyLocked ( proxy * structs . ConnectManagedProxy , persist , FromFile bool ,
restoredProxyToken string ) error {
restoredProxyToken string ) error {
// Lookup the target service token in state if there is one.
// Lookup the target service token in state if there is one.
token := a . State . ServiceToken ( proxy . TargetServiceID )
token := a . State . ServiceToken ( proxy . TargetServiceID )
@ -2143,11 +2153,33 @@ func (a *Agent) AddProxy(proxy *structs.ConnectManagedProxy, persist bool,
// Persist the proxy
// Persist the proxy
if persist && a . config . DataDir != "" {
if persist && a . config . DataDir != "" {
return a . persistProxy ( proxyState
return a . persistProxy ( proxyState , FromFile )
}
}
return nil
return nil
} }
// addProxyLocked adds a new local Connect Proxy instance to be managed by the agent.
//
// It REQUIRES that the service that is being proxied is already present in the
// local state. Note that this is only used for agent-managed proxies so we can
// ensure that we always make this true. For externally managed and registered
// proxies we explicitly allow the proxy to be registered first to make
// bootstrap ordering of a new service simpler but the same is not true here
// since this is only ever called when setting up a _managed_ proxy which was
// registered as part of a service registration either from config or HTTP API
// call.
//
// The restoredProxyToken argument should only be used when restoring proxy
// definitions from disk; new proxies must leave it blank to get a new token
// assigned. We need to restore from disk to enable to continue authenticating
// running proxies that already had that credential injected.
func ( a * Agent ) AddProxy ( proxy * structs . ConnectManagedProxy , persist , FromFile bool ,
restoredProxyToken string ) error {
a . proxyLock . Lock ( )
defer a . proxyLock . Unlock ( )
return a . addProxyLocked ( proxy , persist , FromFile , restoredProxyToken )
}
// resolveProxyCheckAddress returns the best address to use for a TCP check of
// resolveProxyCheckAddress returns the best address to use for a TCP check of
// the proxy's public listener. It expects the input to already have default
// the proxy's public listener. It expects the input to already have default
// values populated by applyProxyConfigDefaults. It may return an empty string
// values populated by applyProxyConfigDefaults. It may return an empty string
@ -2290,8 +2322,10 @@ func (a *Agent) applyProxyDefaults(proxy *structs.ConnectManagedProxy) error {
return nil
return nil
} }
// RemoveProxy stops and removes a local proxy instance.
// removeProxyLocked stops and removes a local proxy instance.
func ( a * Agent ) RemoveProxy ( proxyID string , persist bool ) error { //
// It is assumed that this function is called while holding the proxyLock already
func ( a * Agent ) removeProxyLocked ( proxyID string , persist bool ) error {
// Validate proxyID
// Validate proxyID
if proxyID == "" {
if proxyID == "" {
return fmt . Errorf ( "proxyID missing" )
return fmt . Errorf ( "proxyID missing" )
@ -2316,6 +2350,13 @@ func (a *Agent) RemoveProxy(proxyID string, persist bool) error {
return nil
return nil
} }
// RemoveProxy stops and removes a local proxy instance.
func ( a * Agent ) RemoveProxy ( proxyID string , persist bool ) error {
a . proxyLock . Lock ( )
defer a . proxyLock . Unlock ( )
return a . removeProxyLocked ( proxyID , persist )
}
// verifyProxyToken takes a token and attempts to verify it against the
// verifyProxyToken takes a token and attempts to verify it against the
// targetService name. If targetProxy is specified, then the local proxy token
// targetService name. If targetProxy is specified, then the local proxy token
// must exactly match the given proxy ID. cert, config, etc.).
// must exactly match the given proxy ID. cert, config, etc.).
@ -2782,33 +2823,21 @@ func (a *Agent) unloadChecks() error {
return nil
return nil
} }
// loadProxies will load connect proxy definitions from configuration and
// loadPersistedProxies will load connect proxy definitions from their
// persisted definitions on disk, and load them into the local agent.
// persisted state on disk and return a slice of them
func ( a * Agent ) loadProxies ( conf * config . RuntimeConfig ) error { //
for _ , svc := range conf . Services {
// This does not add them to the local
if svc . Connect != nil {
func ( a * Agent ) loadPersistedProxies ( ) ( map [ string ] persistedProxy , error ) {
proxy , err := svc . ConnectManagedProxy ( )
persistedProxies := make ( map [ string ] persistedProxy )
if err != nil {
return fmt . Errorf ( "failed adding proxy: %s" , err )
}
if proxy == nil {
continue
}
if err := a . AddProxy ( proxy , false , "" ) ; err != nil {
return fmt . Errorf ( "failed adding proxy: %s" , err )
}
}
}
// Load any persisted proxies
proxyDir := filepath . Join ( a . config . DataDir , proxyDir )
proxyDir := filepath . Join ( a . config . DataDir , proxyDir )
files , err := ioutil . ReadDir ( proxyDir )
files , err := ioutil . ReadDir ( proxyDir )
if err != nil {
if err != nil {
if os . IsNotExist ( err ) {
if ! os . IsNotExist ( err ) {
return nil
return nil , fmt . Errorf ( "Failed reading proxies dir %q: %s" , proxyDir , err )
}
}
return fmt . Errorf ( "Failed reading proxies dir %q: %s" , proxyDir , err )
}
}
for _ , fi := range files {
for _ , fi := range files {
// Skip all dirs
// Skip all dirs
if fi . IsDir ( ) {
if fi . IsDir ( ) {
@ -2817,55 +2846,93 @@ func (a *Agent) loadProxies(conf *config.RuntimeConfig) error {
// Skip all partially written temporary files
// Skip all partially written temporary files
if strings . HasSuffix ( fi . Name ( ) , "tmp" ) {
if strings . HasSuffix ( fi . Name ( ) , "tmp" ) {
a . logger . Printf ( "[WARN] agent: Ignoring temporary proxy file %v" , fi . Name ( ) )
return nil , fmt . Errorf ( "Ignoring temporary proxy file %v" , fi . Name ( ) )
continue
}
}
// Open the file for reading
// Open the file for reading
file := filepath . Join ( proxyDir , fi . Name ( ) )
file := filepath . Join ( proxyDir , fi . Name ( ) )
fh , err := os . Open ( file )
fh , err := os . Open ( file )
if err != nil {
if err != nil {
return fmt . Errorf ( "failed opening proxy file %q: %s" , file , err )
return nil , fmt . Errorf ( "failed opening proxy file %q: %s" , file , err )
}
}
// Read the contents into a buffer
// Read the contents into a buffer
buf , err := ioutil . ReadAll ( fh )
buf , err := ioutil . ReadAll ( fh )
fh . Close ( )
fh . Close ( )
if err != nil {
if err != nil {
return fmt . Errorf ( "failed reading proxy file %q: %s" , file , err )
return nil , fmt . Errorf ( "failed reading proxy file %q: %s" , file , err )
}
}
// Try decoding the proxy definition
// Try decoding the proxy definition
var p persistedProxy
var p persistedProxy
if err := json . Unmarshal ( buf , & p ) ; err != nil {
if err := json . Unmarshal ( buf , & p ) ; err != nil {
a . logger . Printf ( "[ERR] agent: Failed decoding proxy file %q: %s" , file , err )
return nil , fmt . Errorf ( "Failed decoding proxy file %q: %s" , file , err )
continue
}
}
proxyID := p . Proxy . ProxyService . ID
svcID := p . Proxy . TargetServiceID
persistedProxies [ svcID ] = p
}
return persistedProxies , nil
}
// loadProxies will load connect proxy definitions from configuration and
// persisted definitions on disk, and load them into the local agent.
func ( a * Agent ) loadProxies ( conf * config . RuntimeConfig ) error {
a . proxyLock . Lock ( )
defer a . proxyLock . Unlock ( )
persistedProxies , persistenceErr := a . loadPersistedProxies ( )
if a . State . Proxy ( proxyID ) != nil {
for _ , svc := range conf . Services {
// Purge previously persisted proxy. This allows config to be preferred
if svc . Connect != nil {
// over services persisted from the API.
proxy , err := svc . ConnectManagedProxy ( )
a . logger . Printf ( "[DEBUG] agent: proxy %q exists, not restoring from %q" ,
if err != nil {
proxyID , file )
return fmt . Errorf ( "failed adding proxy: %s" , err )
}
if proxy == nil {
continue
}
restoredToken := ""
if persisted , ok := persistedProxies [ proxy . TargetServiceID ] ; ok {
restoredToken = persisted . ProxyToken
}
if err := a . addProxyLocked ( proxy , true , true , restoredToken ) ; err != nil {
return fmt . Errorf ( "failed adding proxy: %s" , err )
}
}
}
for _ , persisted := range persistedProxies {
proxyID := persisted . Proxy . ProxyService . ID
if persisted . FromFile && a . State . Proxy ( proxyID ) == nil {
// Purge proxies that were configured previously but are no longer in the config
a . logger . Printf ( "[DEBUG] agent: purging stale persisted proxy %q" , proxyID )
if err := a . purgeProxy ( proxyID ) ; err != nil {
if err := a . purgeProxy ( proxyID ) ; err != nil {
return fmt . Errorf ( "failed purging proxy %q: %s" , proxyID , err )
return fmt . Errorf ( "failed purging proxy %q: % v ", proxyID , err )
}
}
} else {
} else if ! persisted . FromFile {
a . logger . Printf ( "[DEBUG] agent: restored proxy definition %q from %q" ,
if a . State . Proxy ( proxyID ) == nil {
proxyID , file )
a . logger . Printf ( "[DEBUG] agent: restored proxy definition %q" , proxyID )
if err := a . AddProxy ( p . Proxy , false , p . ProxyToken ) ; err != nil {
if err := a . addProxyLocked ( persisted . Proxy , false , false , persisted . ProxyToken ) ; err != nil {
return fmt . Errorf ( "failed adding proxy %q: %s" , proxyID , err )
return fmt . Errorf ( "failed adding proxy %q: %v" , proxyID , err )
}
} else {
a . logger . Printf ( "[WARN] agent: proxy definition %q was overwritten by a proxy definition within a config file" , proxyID )
}
}
}
}
}
}
return nil
return persistenceErr
} }
// unloadProxies will deregister all proxies known to the local agent.
// unloadProxies will deregister all proxies known to the local agent.
func ( a * Agent ) unloadProxies ( ) error { func ( a * Agent ) unloadProxies ( ) error {
a . proxyLock . Lock ( )
defer a . proxyLock . Unlock ( )
for id := range a . State . Proxies ( ) {
for id := range a . State . Proxies ( ) {
if err := a . RemoveProxy ( id , false ) ; err != nil {
if err := a . removeProxyLocked ( id , false ) ; err != nil {
return fmt . Errorf ( "Failed deregistering proxy '%s': %s" , id , err )
return fmt . Errorf ( "Failed deregistering proxy '%s': %s" , id , err )
}
}
}
}
Matt Keeler
6 years ago