Merge pull request #956 from containous/fix-leadership-panic

Fix leadership panic
2016-12-13 10:39:59 +01:00
parent 70f39c3326 25bab338c1
commit 59b3a392f2
5 changed files with 110 additions and 96 deletions
--- a/acme/acme.go
+++ b/acme/acme.go
@@ -11,6 +11,7 @@ import (
 	"github.com/containous/traefik/log"
 	"github.com/containous/traefik/safe"
 	"github.com/containous/traefik/types"
+	"github.com/eapache/channels"
 	"github.com/xenolf/lego/acme"
 	"golang.org/x/net/context"
 	"io/ioutil"
@@ -35,6 +36,7 @@ type ACME struct {
 	store               cluster.Store
 	challengeProvider   *challengeProvider
 	checkOnDemandDomain func(domain string) bool
+	jobs                *channels.InfiniteChannel
 }

 //Domains parse []Domain
@@ -91,6 +93,7 @@ func (a *ACME) init() error {
 		log.Warnf("ACME.StorageFile is deprecated, use ACME.Storage instead")
 		a.Storage = a.StorageFile
 	}
+	a.jobs = channels.NewInfiniteChannel()
 	return nil
 }

@@ -142,9 +145,7 @@ func (a *ACME) CreateClusterConfig(leadership *cluster.Leadership, tlsConfig *tl
 			case <-ctx.Done():
 				return
 			case <-ticker.C:
-				if err := a.renewCertificates(); err != nil {
-					log.Errorf("Error renewing ACME certificate: %s", err.Error())
-				}
+				a.renewCertificates()
 			}
 		}
 	})
@@ -205,12 +206,10 @@ func (a *ACME) CreateClusterConfig(leadership *cluster.Leadership, tlsConfig *tl
 			if err != nil {
 				return err
 			}
-			safe.Go(func() {
-				a.retrieveCertificates()
-				if err := a.renewCertificates(); err != nil {
-					log.Errorf("Error renewing ACME certificate %+v: %s", account, err.Error())
-				}
-			})
+
+			a.retrieveCertificates()
+			a.renewCertificates()
+			a.runJobs()
 		}
 		return nil
 	})
@@ -295,19 +294,14 @@ func (a *ACME) CreateLocalConfig(tlsConfig *tls.Config, checkOnDemandDomain func
 		return err
 	}

-	safe.Go(func() {
-		a.retrieveCertificates()
-		if err := a.renewCertificates(); err != nil {
-			log.Errorf("Error renewing ACME certificate %+v: %s", account, err.Error())
-		}
-	})
+	a.retrieveCertificates()
+	a.renewCertificates()
+	a.runJobs()

 	ticker := time.NewTicker(24 * time.Hour)
 	safe.Go(func() {
 		for range ticker.C {
-			if err := a.renewCertificates(); err != nil {
-				log.Errorf("Error renewing ACME certificate %+v: %s", account, err.Error())
-			}
+			a.renewCertificates()
 		}

 	})
@@ -336,83 +330,87 @@ func (a *ACME) getCertificate(clientHello *tls.ClientHelloInfo) (*tls.Certificat
 }

 func (a *ACME) retrieveCertificates() {
-	log.Infof("Retrieving ACME certificates...")
-	for _, domain := range a.Domains {
-		// check if cert isn't already loaded
-		account := a.store.Get().(*Account)
-		if _, exists := account.DomainsCertificate.exists(domain); !exists {
-			domains := []string{}
-			domains = append(domains, domain.Main)
-			domains = append(domains, domain.SANs...)
-			certificateResource, err := a.getDomainsCertificates(domains)
-			if err != nil {
-				log.Errorf("Error getting ACME certificate for domain %s: %s", domains, err.Error())
-				continue
-			}
-			transaction, object, err := a.store.Begin()
-			if err != nil {
-				log.Errorf("Error creating ACME store transaction from domain %s: %s", domain, err.Error())
-				continue
-			}
-			account = object.(*Account)
-			_, err = account.DomainsCertificate.addCertificateForDomains(certificateResource, domain)
-			if err != nil {
-				log.Errorf("Error adding ACME certificate for domain %s: %s", domains, err.Error())
-				continue
-			}
+	a.jobs.In() <- func() {
+		log.Infof("Retrieving ACME certificates...")
+		for _, domain := range a.Domains {
+			// check if cert isn't already loaded
+			account := a.store.Get().(*Account)
+			if _, exists := account.DomainsCertificate.exists(domain); !exists {
+				domains := []string{}
+				domains = append(domains, domain.Main)
+				domains = append(domains, domain.SANs...)
+				certificateResource, err := a.getDomainsCertificates(domains)
+				if err != nil {
+					log.Errorf("Error getting ACME certificate for domain %s: %s", domains, err.Error())
+					continue
+				}
+				transaction, object, err := a.store.Begin()
+				if err != nil {
+					log.Errorf("Error creating ACME store transaction from domain %s: %s", domain, err.Error())
+					continue
+				}
+				account = object.(*Account)
+				_, err = account.DomainsCertificate.addCertificateForDomains(certificateResource, domain)
+				if err != nil {
+					log.Errorf("Error adding ACME certificate for domain %s: %s", domains, err.Error())
+					continue
+				}

-			if err = transaction.Commit(account); err != nil {
-				log.Errorf("Error Saving ACME account %+v: %s", account, err.Error())
-				continue
+				if err = transaction.Commit(account); err != nil {
+					log.Errorf("Error Saving ACME account %+v: %s", account, err.Error())
+					continue
+				}
 			}
 		}
+		log.Infof("Retrieved ACME certificates")
 	}
-	log.Infof("Retrieved ACME certificates")
 }

-func (a *ACME) renewCertificates() error {
-	log.Debugf("Testing certificate renew...")
-	account := a.store.Get().(*Account)
-	for _, certificateResource := range account.DomainsCertificate.Certs {
-		if certificateResource.needRenew() {
-			log.Debugf("Renewing certificate %+v", certificateResource.Domains)
-			renewedCert, err := a.client.RenewCertificate(acme.CertificateResource{
-				Domain:        certificateResource.Certificate.Domain,
-				CertURL:       certificateResource.Certificate.CertURL,
-				CertStableURL: certificateResource.Certificate.CertStableURL,
-				PrivateKey:    certificateResource.Certificate.PrivateKey,
-				Certificate:   certificateResource.Certificate.Certificate,
-			}, true)
-			if err != nil {
-				log.Errorf("Error renewing certificate: %v", err)
-				continue
-			}
-			log.Debugf("Renewed certificate %+v", certificateResource.Domains)
-			renewedACMECert := &Certificate{
-				Domain:        renewedCert.Domain,
-				CertURL:       renewedCert.CertURL,
-				CertStableURL: renewedCert.CertStableURL,
-				PrivateKey:    renewedCert.PrivateKey,
-				Certificate:   renewedCert.Certificate,
-			}
-			transaction, object, err := a.store.Begin()
-			if err != nil {
-				return err
-			}
-			account = object.(*Account)
-			err = account.DomainsCertificate.renewCertificates(renewedACMECert, certificateResource.Domains)
-			if err != nil {
-				log.Errorf("Error renewing certificate: %v", err)
-				continue
-			}
+func (a *ACME) renewCertificates() {
+	a.jobs.In() <- func() {
+		log.Debugf("Testing certificate renew...")
+		account := a.store.Get().(*Account)
+		for _, certificateResource := range account.DomainsCertificate.Certs {
+			if certificateResource.needRenew() {
+				log.Debugf("Renewing certificate %+v", certificateResource.Domains)
+				renewedCert, err := a.client.RenewCertificate(acme.CertificateResource{
+					Domain:        certificateResource.Certificate.Domain,
+					CertURL:       certificateResource.Certificate.CertURL,
+					CertStableURL: certificateResource.Certificate.CertStableURL,
+					PrivateKey:    certificateResource.Certificate.PrivateKey,
+					Certificate:   certificateResource.Certificate.Certificate,
+				}, true)
+				if err != nil {
+					log.Errorf("Error renewing certificate: %v", err)
+					continue
+				}
+				log.Debugf("Renewed certificate %+v", certificateResource.Domains)
+				renewedACMECert := &Certificate{
+					Domain:        renewedCert.Domain,
+					CertURL:       renewedCert.CertURL,
+					CertStableURL: renewedCert.CertStableURL,
+					PrivateKey:    renewedCert.PrivateKey,
+					Certificate:   renewedCert.Certificate,
+				}
+				transaction, object, err := a.store.Begin()
+				if err != nil {
+					log.Errorf("Error renewing certificate: %v", err)
+					continue
+				}
+				account = object.(*Account)
+				err = account.DomainsCertificate.renewCertificates(renewedACMECert, certificateResource.Domains)
+				if err != nil {
+					log.Errorf("Error renewing certificate: %v", err)
+					continue
+				}

-			if err = transaction.Commit(account); err != nil {
-				log.Errorf("Error Saving ACME account %+v: %s", account, err.Error())
-				continue
+				if err = transaction.Commit(account); err != nil {
+					log.Errorf("Error Saving ACME account %+v: %s", account, err.Error())
+					continue
+				}
 			}
 		}
 	}
-	return nil
 }

 func (a *ACME) buildACMEClient(account *Account) (*acme.Client, error) {
@@ -462,8 +460,9 @@ func (a *ACME) loadCertificateOnDemand(clientHello *tls.ClientHelloInfo) (*tls.C

 // LoadCertificateForDomains loads certificates from ACME for given domains
 func (a *ACME) LoadCertificateForDomains(domains []string) {
-	domains = fun.Map(types.CanonicalDomain, domains).([]string)
-	safe.Go(func() {
+	a.jobs.In() <- func() {
+		log.Debugf("LoadCertificateForDomains %s...", domains)
+		domains = fun.Map(types.CanonicalDomain, domains).([]string)
 		operation := func() error {
 			if a.client == nil {
 				return fmt.Errorf("ACME client still not built")
@@ -517,7 +516,7 @@ func (a *ACME) LoadCertificateForDomains(domains []string) {
 			log.Errorf("Error Saving ACME account %+v: %v", account, err)
 			return
 		}
-	})
+	}
 }

 func (a *ACME) getDomainsCertificates(domains []string) (*Certificate, error) {
@@ -538,3 +537,12 @@ func (a *ACME) getDomainsCertificates(domains []string) (*Certificate, error) {
 		Certificate:   certificate.Certificate,
 	}, nil
 }
+
+func (a *ACME) runJobs() {
+	safe.Go(func() {
+		for job := range a.jobs.Out() {
+			function := job.(func())
+			function()
+		}
+	})
+}
--- a/cluster/datastore.go
+++ b/cluster/datastore.go
@@ -230,21 +230,21 @@ func (s *datastoreTransaction) Commit(object Object) error {
 	s.localLock.Lock()
 	defer s.localLock.Unlock()
 	if s.dirty {
-		return fmt.Errorf("transaction already used, please begin a new one")
+		return fmt.Errorf("Transaction already used, please begin a new one")
 	}
 	s.Datastore.meta.object = object
 	err := s.Datastore.meta.Marshall()
 	if err != nil {
-		return err
+		return fmt.Errorf("Marshall error: %s", err)
 	}
 	err = s.kv.StoreConfig(s.Datastore.meta)
 	if err != nil {
-		return err
+		return fmt.Errorf("StoreConfig error: %s", err)
 	}

 	err = s.remoteLock.Unlock()
 	if err != nil {
-		return err
+		return fmt.Errorf("Unlock error: %s", err)
 	}

 	s.dirty = true
--- a/cluster/leadership.go
+++ b/cluster/leadership.go
@@ -15,7 +15,7 @@ type Leadership struct {
 	*safe.Pool
 	*types.Cluster
 	candidate *leadership.Candidate
-	leader    safe.Safe
+	leader    *safe.Safe
 	listeners []LeaderListener
 }

@@ -26,6 +26,7 @@ func NewLeadership(ctx context.Context, cluster *types.Cluster) *Leadership {
 		Cluster:   cluster,
 		candidate: leadership.NewCandidate(cluster.Store, cluster.Store.Prefix+"/leader", cluster.Node, 20*time.Second),
 		listeners: []LeaderListener{},
+		leader:    safe.New(false),
 	}
 }

--- a/glide.lock
+++ b/glide.lock
@@ -1,5 +1,5 @@
-hash: 1bbeb842ee639ccc6e2edf8cc13fc2759cb96e3d839a1aec7b7f6af4fb89c8e1
-updated: 2016-11-09T19:24:00.762904389+01:00
+hash: fa6c0ac899b3c9296d83f1d4110186b339101b800b241e08cdcd2e3d49270562
+updated: 2016-12-09T13:43:16.816754682+01:00
 imports:
 - name: github.com/abbot/go-http-auth
  version: cb4372376e1e00e9f6ab9ec142e029302c9e7140
@@ -156,6 +156,10 @@ imports:
  - store/zookeeper
 - name: github.com/donovanhide/eventsource
  version: fd1de70867126402be23c306e1ce32828455d85b
+- name: github.com/eapache/channels
+  version: 47238d5aae8c0fefd518ef2bee46290909cf8263
+- name: github.com/eapache/queue
+  version: 44cc805cf13205b55f69e14bcb69867d1ae92f98
 - name: github.com/elazarl/go-bindata-assetfs
  version: 9a6736ed45b44bf3835afeebb3034b57ed329f3e
 - name: github.com/gambol99/go-marathon
--- a/glide.yaml
+++ b/glide.yaml
@@ -109,4 +109,5 @@ import:
  subpackages:
  - daemon
 - package: github.com/google/go-github
- package: github.com/hashicorp/go-version
+- package: github.com/hashicorp/go-version
+- package: github.com/eapache/channels