func (r *VaultService) vaultServiceProcessor() {
	go func() {
		// a list of resource being watched
		var items []*watchedResource

		// the channel to receive renewal notifications on
		renewChannel := make(chan *watchedResource, 10)
		retrieveChannel := make(chan *watchedResource, 10)
		revokeChannel := make(chan *watchedResource, 10)
		statsChannel := time.NewTicker(options.statsInterval)

		for {
			select {
			// A new resource is being added to the service processor;
			//  - schedule the resource for retrieval
			case x := <-r.resourceChannel:
				glog.V(4).Infof("adding a resource into the service processor, resource: %s", x.resource)
				// step: add to the list of resources
				items = append(items, x)
				// step: push into the retrieval channel
				r.scheduleNow(x, retrieveChannel)

			// Retrieve a resource from vault
			//  - we retrieve the resource from vault
			//  - if we error attempting to retrieve the secret, we background and reschedule an attempt to add it
			//  - if ok, we grab the lease it and lease time, we setup a notification on renewal
			case x := <-retrieveChannel:
				// step: skip this resource if it's reached maxRetries
				if x.resource.maxRetries > 0 && x.resource.retries > x.resource.maxRetries {
					glog.V(4).Infof("skipping resource %s as it's failed %d/%d times", x.resource.retries, x.resource.maxRetries+1)
					break
				}

				// step: save the current lease if we have one
				leaseID := ""
				if x.secret != nil && x.secret.LeaseID != "" {
					leaseID = x.secret.LeaseID
					glog.V(10).Infof("resource: %s has a previous lease: %s", x.resource, leaseID)
				}

				err := r.get(x)
				if err != nil {
					glog.Errorf("failed to retrieve the resource: %s from vault, error: %s", x.resource, err)
					// reschedule the attempt for later
					r.scheduleIn(x, retrieveChannel, getDurationWithin(3, 10))
					x.resource.retries++
					r.upstream(VaultEvent{
						Resource: x.resource,
						Type:     EventTypeFailure,
					})
					break
				}

				glog.V(4).Infof("successfully retrieved resource: %s, leaseID: %s", x.resource, x.secret.LeaseID)
				x.resource.retries = 0

				// step: if we had a previous lease and the option is to revoke, lets throw into the revoke channel
				if leaseID != "" && x.resource.revoked {
					// step: make a rough copy
					copy := &watchedResource{
						secret: &api.Secret{
							LeaseID: x.secret.LeaseID,
						},
					}

					r.scheduleIn(copy, revokeChannel, x.resource.revokeDelay)
				}

				// step: setup a timer for renewal
				x.notifyOnRenewal(renewChannel)

				// step: update the upstream consumers
				r.upstream(VaultEvent{
					Resource: x.resource,
					Secret:   x.secret.Data,
					Type:     EventTypeSuccess,
				})

			// A watched resource is coming up for renewal
			// 	- we attempt to renew the resource from vault
			//	- if we encounter an error, we reschedule the attempt for the future
			//	- if we're ok, we update the watchedResource and we send a notification of the change upstream
			case x := <-renewChannel:
				// step: skip this resource if it's reached maxRetries
				if x.resource.maxRetries > 0 && x.resource.retries > x.resource.maxRetries {
					glog.V(4).Infof("skipping resource %s as it's failed %d/%d times", x.resource.retries, x.resource.maxRetries+1)
					break
				}

				glog.V(4).Infof("resource: %s, lease: %s up for renewal, renewable: %t, revoked: %t", x.resource,
					x.secret.LeaseID, x.resource.renewable, x.resource.revoked)

				// step: we need to check if the lease has expired?
				if time.Now().Before(x.leaseExpireTime) {
					glog.V(3).Infof("the lease on resource: %s has expired, we need to get a new lease", x.resource)
					// push into the retrieval channel and break
					r.scheduleNow(x, retrieveChannel)
					break
				}

				// step: are we renewing the resource?
				if x.resource.renewable {
					// step: is the underlining resource even renewable? - otherwise we can just grab a new lease
					if !x.secret.Renewable {
						glog.V(10).Infof("the resource: %s is not renewable, retrieving a new lease instead", x.resource)
						r.scheduleNow(x, retrieveChannel)
						break
					}

					// step: lets renew the resource
					err := r.renew(x)
					if err != nil {
						glog.Errorf("failed to renew the resource: %s for renewal, error: %s", x.resource, err)
						// reschedule the attempt for later
						r.scheduleIn(x, renewChannel, getDurationWithin(3, 10))
						x.resource.retries++
						r.upstream(VaultEvent{
							Resource: x.resource,
							Type:     EventTypeFailure,
						})
						break
					}

					glog.V(4).Infof("successfully renewed resource: %s, leaseID: %s", x.resource, x.secret.LeaseID)
					x.resource.retries = 0
				}

				// step: the option for this resource is not to renew the secret but regenerate a new secret
				if !x.resource.renewable {
					glog.V(4).Infof("resource: %s flagged as not renewable, shifting to regenerating the resource", x.resource)
					r.scheduleNow(x, retrieveChannel)
					break
				}

				// step: setup a timer for renewal
				x.notifyOnRenewal(renewChannel)

				// step: update any listener upstream
				r.upstream(VaultEvent{
					Resource: x.resource,
					Secret:   x.secret.Data,
					Type:     EventTypeSuccess,
				})

			// We receive a lease ID along on the channel, just revoke the lease when you can
			case x := <-revokeChannel:
				err := r.revoke(x.secret.LeaseID)
				if err != nil {
					glog.Errorf("failed to revoke the lease: %s, error: %s", x.secret.LeaseID, err)
				}

			// The statistics timer has gone off; we iterate the watched items and
			case <-statsChannel.C:
				glog.V(3).Infof("stats: %d resources being watched", len(items))
				for _, item := range items {
					glog.V(3).Infof("resourse: %s, lease id: %s, renewal in: %s seconds, expiration: %s",
						item.resource, item.secret.LeaseID, item.renewalTime, item.leaseExpireTime)
				}
			}
		}
	}()
}
// vaultServiceProcessor is the background routine responsible for retrieving the resources, renewing when required and
// informing those who are watching the resource that something has changed
vaultServiceProcessor is the background routine responsible for retrieving the resources, renewing when required and
informing those who are watching the resource that something has changed
// vaultServiceProcessor is the background routine responsible for retrieving the resources renewing when required and // informing those who are watching the resource that something has changed
func (r *VaultService) vaultServiceProcessor() {
	go func() {
		var items []*watchedResource
		renewChannel := make(chan *watchedResource, 10)
		retrieveChannel := make(chan *watchedResource, 10)
		revokeChannel := make(chan *watchedResource, 10)
		statsChannel := time.NewTicker(options.statsInterval)
		for {
			select {
			case x := <-r.resourceChannel:
				glog.V(4).Infof("adding a resource into the service processor, resource: %s", x.resource)
				items = append(items, x)
				r.scheduleNow(x, retrieveChannel)
			case x := <-retrieveChannel:
				if x.resource.maxRetries > 0 && x.resource.retries > x.resource.maxRetries {
					glog.V(4).Infof("skipping resource %s as it's failed %d/%d times", x.resource.retries, x.resource.maxRetries+1)
					break
				}
				leaseID := ""
				if x.secret != nil && x.secret.LeaseID != "" {
					leaseID = x.secret.LeaseID
					glog.V(10).Infof("resource: %s has a previous lease: %s", x.resource, leaseID)
				}
				err := r.get(x)
				if err != nil {
					glog.Errorf("failed to retrieve the resource: %s from vault, error: %s", x.resource, err)
					r.scheduleIn(x, retrieveChannel, getDurationWithin(3, 10))
					x.resource.retries++
					r.upstream(VaultEvent{
						Resource: x.resource,
						Type:     EventTypeFailure,
					})
					break
				}
				glog.V(4).Infof("successfully retrieved resource: %s, leaseID: %s", x.resource, x.secret.LeaseID)
				x.resource.retries = 0
				if leaseID != "" && x.resource.revoked {
					copy := &watchedResource{
						secret: &api.Secret{
							LeaseID: x.secret.LeaseID,
						},
					}
					r.scheduleIn(copy, revokeChannel, x.resource.revokeDelay)
				}
				x.notifyOnRenewal(renewChannel)
				r.upstream(VaultEvent{
					Resource: x.resource,
					Secret:   x.secret.Data,
					Type:     EventTypeSuccess,
				})
			case x := <-renewChannel:
				if x.resource.maxRetries > 0 && x.resource.retries > x.resource.maxRetries {
					glog.V(4).Infof("skipping resource %s as it's failed %d/%d times", x.resource.retries, x.resource.maxRetries+1)
					break
				}
				glog.V(4).Infof("resource: %s, lease: %s up for renewal, renewable: %t, revoked: %t", x.resource,
					x.secret.LeaseID, x.resource.renewable, x.resource.revoked)
				if time.Now().Before(x.leaseExpireTime) {
					glog.V(3).Infof("the lease on resource: %s has expired, we need to get a new lease", x.resource)
					r.scheduleNow(x, retrieveChannel)
					break
				}
				if x.resource.renewable {
					if !x.secret.Renewable {
						glog.V(10).Infof("the resource: %s is not renewable, retrieving a new lease instead", x.resource)
						r.scheduleNow(x, retrieveChannel)
						break
					}
					err := r.renew(x)
					if err != nil {
						glog.Errorf("failed to renew the resource: %s for renewal, error: %s", x.resource, err)
						r.scheduleIn(x, renewChannel, getDurationWithin(3, 10))
						x.resource.retries++
						r.upstream(VaultEvent{
							Resource: x.resource,
							Type:     EventTypeFailure,
						})
						break
					}
					glog.V(4).Infof("successfully renewed resource: %s, leaseID: %s", x.resource, x.secret.LeaseID)
					x.resource.retries = 0
				}
				if !x.resource.renewable {
					glog.V(4).Infof("resource: %s flagged as not renewable, shifting to regenerating the resource", x.resource)
					r.scheduleNow(x, retrieveChannel)
					break
				}
				x.notifyOnRenewal(renewChannel)
				r.upstream(VaultEvent{
					Resource: x.resource,
					Secret:   x.secret.Data,
					Type:     EventTypeSuccess,
				})
			case x := <-revokeChannel:
				err := r.revoke(x.secret.LeaseID)
				if err != nil {
					glog.Errorf("failed to revoke the lease: %s, error: %s", x.secret.LeaseID, err)
				}
			case <-statsChannel.C:
				glog.V(3).Infof("stats: %d resources being watched", len(items))
				for _, item := range items {
					glog.V(3).Infof("resourse: %s, lease id: %s, renewal in: %s seconds, expiration: %s",
						item.resource, item.secret.LeaseID, item.renewalTime, item.leaseExpireTime)
				}
			}
		}
	}()
}
func ( r * VaultService ) vaultServiceProcessor ( ) { go func ( ) { var items [ ] * watchedResource renewChannel := make ( chan * watchedResource , 10 ) retrieveChannel := make ( chan * watchedResource , 10 ) revokeChannel := make ( chan * watchedResource , 10 ) statsChannel := time . NewTicker ( options . statsInterval ) for { select { case x := <- r . resourceChannel : glog . V ( 4 ) . Infof ( " " , x . resource ) items = append ( items , x ) r . scheduleNow ( x , retrieveChannel ) case x := <- retrieveChannel : if x . resource . maxRetries > 0 && x . resource . retries > x . resource . maxRetries { glog . V ( 4 ) . Infof ( " " , x . resource . retries , x . resource . maxRetries + 1 ) break } leaseID := " " if x . secret != nil && x . secret . LeaseID != " " { leaseID = x . secret . LeaseID glog . V ( 10 ) . Infof ( " " , x . resource , leaseID ) } err := r . get ( x ) if err != nil { glog . Errorf ( " " , x . resource , err ) r . scheduleIn ( x , retrieveChannel , getDurationWithin ( 3 , 10 ) ) x . resource . retries ++ r . upstream ( VaultEvent { Resource : x . resource , Type : EventTypeFailure , } ) break } glog . V ( 4 ) . Infof ( " " , x . resource , x . secret . LeaseID ) x . resource . retries = 0 if leaseID != " " && x . resource . revoked { copy := & watchedResource { secret : & api . Secret { LeaseID : x . secret . LeaseID , } , } r . scheduleIn ( copy , revokeChannel , x . resource . revokeDelay ) } x . notifyOnRenewal ( renewChannel ) r . upstream ( VaultEvent { Resource : x . resource , Secret : x . secret . Data , Type : EventTypeSuccess , } ) case x := <- renewChannel : if x . resource . maxRetries > 0 && x . resource . retries > x . resource . maxRetries { glog . V ( 4 ) . Infof ( " " , x . resource . retries , x . resource . maxRetries + 1 ) break } glog . V ( 4 ) . Infof ( " " , x . resource , x . secret . LeaseID , x . resource . renewable , x . resource . revoked ) if time . Now ( ) . Before ( x . leaseExpireTime ) { glog . V ( 3 ) . Infof ( " " , x . resource ) r . scheduleNow ( x , retrieveChannel ) break } if x . resource . renewable { if ! x . secret . Renewable { glog . V ( 10 ) . Infof ( " " , x . resource ) r . scheduleNow ( x , retrieveChannel ) break } err := r . renew ( x ) if err != nil { glog . Errorf ( " " , x . resource , err ) r . scheduleIn ( x , renewChannel , getDurationWithin ( 3 , 10 ) ) x . resource . retries ++ r . upstream ( VaultEvent { Resource : x . resource , Type : EventTypeFailure , } ) break } glog . V ( 4 ) . Infof ( " " , x . resource , x . secret . LeaseID ) x . resource . retries = 0 } if ! x . resource . renewable { glog . V ( 4 ) . Infof ( " " , x . resource ) r . scheduleNow ( x , retrieveChannel ) break } x . notifyOnRenewal ( renewChannel ) r . upstream ( VaultEvent { Resource : x . resource , Secret : x . secret . Data , Type : EventTypeSuccess , } ) case x := <- revokeChannel : err := r . revoke ( x . secret . LeaseID ) if err != nil { glog . Errorf ( " " , x . secret . LeaseID , err ) } case <- statsChannel . C : glog . V ( 3 ) . Infof ( " " , len ( items ) ) for _ , item := range items { glog . V ( 3 ) . Infof ( " " , item . resource , item . secret . LeaseID , item . renewalTime , item . leaseExpireTime ) } } } } ( ) }
vaultServiceProcessor is the background routine responsible for retrieving the resources, renewing when required and
informing those who are watching the resource that something has changed
vaultServiceProcessor is the background routine responsible for retrieving the resources renewing when required and informing those who are watching the resource that something has changed
// a list of resource being watched // the channel to receive renewal notifications on // A new resource is being added to the service processor; //  - schedule the resource for retrieval // step: add to the list of resources // step: push into the retrieval channel // Retrieve a resource from vault //  - we retrieve the resource from vault //  - if we error attempting to retrieve the secret, we background and reschedule an attempt to add it //  - if ok, we grab the lease it and lease time, we setup a notification on renewal // step: skip this resource if it's reached maxRetries // step: save the current lease if we have one // reschedule the attempt for later // step: if we had a previous lease and the option is to revoke, lets throw into the revoke channel // step: make a rough copy // step: setup a timer for renewal // step: update the upstream consumers // A watched resource is coming up for renewal // 	- we attempt to renew the resource from vault //	- if we encounter an error, we reschedule the attempt for the future //	- if we're ok, we update the watchedResource and we send a notification of the change upstream // step: skip this resource if it's reached maxRetries // step: we need to check if the lease has expired? // push into the retrieval channel and break // step: are we renewing the resource? // step: is the underlining resource even renewable? - otherwise we can just grab a new lease // step: lets renew the resource // reschedule the attempt for later // step: the option for this resource is not to renew the secret but regenerate a new secret // step: setup a timer for renewal // step: update any listener upstream // We receive a lease ID along on the channel, just revoke the lease when you can // The statistics timer has gone off; we iterate the watched items and