move from time.sleep to an update method

go.mod

@@ -9,6 +9,7 @@ require (
 	github.com/hashicorp/vault/api v1.0.4
 	github.com/hashicorp/vault/sdk v0.1.13
 	github.com/natefinch/lumberjack v2.0.0+incompatible
+	github.com/prometheus/common v0.4.0
 	github.com/sirupsen/logrus v1.2.0
 	github.com/spf13/afero v1.1.2
 	github.com/spf13/viper v1.4.0

go.sum

@@ -2,7 +2,9 @@ cloud.google.com/go v0.26.0/go.mod h1:aQUYkXzVsufM+DwF1aE+0xfcU+56JwCaLick0ClmMT
 github.com/BurntSushi/toml v0.3.1 h1:WXkYYl6Yr3qBf1K79EBnL4mak0OimBfB0XUf9Vl28OQ=
 github.com/BurntSushi/toml v0.3.1/go.mod h1:xHWCNGjB5oqiDr8zfno3MHue2Ht5sIBksp03qcyfWMU=
 github.com/OneOfOne/xxhash v1.2.2/go.mod h1:HSdplMjZKSmBqAxg5vPj2TmRDmfkzw+cTzAElWljhcU=
+github.com/alecthomas/template v0.0.0-20160405071501-a0175ee3bccc h1:cAKDfWh5VpdgMhJosfJnn5/FoN2SRZ4p7fJNX58YPaU=
 github.com/alecthomas/template v0.0.0-20160405071501-a0175ee3bccc/go.mod h1:LOuyumcjzFXgccqObfd/Ljyb9UuFJ6TxHnclSeseNhc=
+github.com/alecthomas/units v0.0.0-20151022065526-2efee857e7cf h1:qet1QNfXsQxTZqLG4oE62mJzwPIB8+Tee4RNCL9ulrY=
 github.com/alecthomas/units v0.0.0-20151022065526-2efee857e7cf/go.mod h1:ybxpYRFXyAe+OPACYpWeL0wqObRcbAqCMya13uyzqw0=
 github.com/armon/consul-api v0.0.0-20180202201655-eb2c6b5be1b6/go.mod h1:grANhF5doyWs3UAsr3K4I6qtAmlQcZDesFNEHPZAzj8=
 github.com/armon/go-metrics v0.0.0-20180917152333-f0300d1749da/go.mod h1:Q73ZrmVTwzkszR9V5SSuryQ31EELlFMUz1kKyl939pY=
@@ -150,6 +152,7 @@ github.com/prometheus/client_golang v0.9.3/go.mod h1:/TN21ttK/J9q6uSwhBd54HahCDf
 github.com/prometheus/client_model v0.0.0-20180712105110-5c3871d89910/go.mod h1:MbSGuTsp3dbXC40dX6PRTWyKYBIrTGTE9sqQNg2J8bo=
 github.com/prometheus/client_model v0.0.0-20190129233127-fd36f4220a90/go.mod h1:xMI15A0UPsDsEKsMN9yxemIoYk6Tm2C1GtYGdfGttqA=
 github.com/prometheus/common v0.0.0-20181113130724-41aa239b4cce/go.mod h1:daVV7qP5qjZbuso7PdcryaAu0sAZbrN9i7WWcTMWvro=
+github.com/prometheus/common v0.4.0 h1:7etb9YClo3a6HjLzfl6rIQaU+FDfi0VSX39io3aQ+DM=
 github.com/prometheus/common v0.4.0/go.mod h1:TNfzLD0ON7rHzMJeJkieUDPYmFC7Snx/y86RQel1bk4=
 github.com/prometheus/procfs v0.0.0-20181005140218-185b4288413d/go.mod h1:c3At6R/oaqEKCNdg8wHV1ftS6bRYblBhIjjI8uT2IGk=
 github.com/prometheus/procfs v0.0.0-20190507164030-5867b95ac084/go.mod h1:TjEm7ze935MbeOT/UhFTIMYKhuLP4wbCsTZCD3I8kEA=
@@ -257,6 +260,7 @@ google.golang.org/grpc v1.22.0 h1:J0UbZOIrCAl+fpTOf8YLs4dJo8L/owV4LYVtAXQoPkw=
 google.golang.org/grpc v1.22.0/go.mod h1:Y5yQAOtifL1yxbo5wqy6BxZv8vAUGQwXBOALyacEbxg=
 google.golang.org/grpc v1.24.0 h1:vb/1TCsVn3DcJlQ0Gs1yB1pKI6Do2/QNwxdKqmc/b0s=
 google.golang.org/grpc v1.24.0/go.mod h1:XDChyiUovWa60DnaeDeZmSW86xtLtjtZbwvSiRnRtcA=
+gopkg.in/alecthomas/kingpin.v2 v2.2.6 h1:jMFz6MfLP0/4fUyZle81rXUoxOBFi19VUFKVDOQfozc=
 gopkg.in/alecthomas/kingpin.v2 v2.2.6/go.mod h1:FMv+mEhP44yOT+4EoQTLFTRgOQ1FBLkstjWtayDeSgw=
 gopkg.in/asn1-ber.v1 v1.0.0-20181015200546-f715ec2f112d/go.mod h1:cuepJuh7vyXfUyUwEgHQXw849cJrilpS5NeIjOWESAw=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=

internal/common/spiretrustsource.go

@@ -7,6 +7,7 @@ import (
 	"fmt"
 	"io/ioutil"
 	"os"
+	"time"
 
 	"github.com/sirupsen/logrus"
 	"github.com/spiffe/go-spiffe/workload"
@@ -18,6 +19,8 @@ type SpireTrustSource struct {
 	domainCertificates map[string][]*x509.Certificate
 	spireClients       []*workload.X509SVIDClient
 	certLocation       string
+	updateChan         chan struct{}
+	updateTimeout      time.Duration
 }
 
 type certMap struct {
@@ -40,6 +43,8 @@ func NewSpireTrustSource(domainURLs map[string]string, certLocation string) (*Sp
 		domainURLs:         domainURLs,
 		domainCertificates: make(map[string][]*x509.Certificate, 0),
 		certLocation:       certLocation,
+		updateChan:         make(chan struct{}, 0),
+		updateTimeout:      5 * time.Second,
 	}
 
 	if certLocation != "" {
@@ -145,7 +150,7 @@ func (s *SpireTrustSource) startWatchers() error {
 		s.spireClients = append(s.spireClients, client)
 
 		logrus.Infof("Starting listener for %s.\n", id)
-		go client.Start()
+		client.Start()
 	}
 	return nil
 }
@@ -156,6 +161,10 @@ func (w *watcher) UpdateX509SVIDs(svids *workload.X509SVIDs) {
 	if err != nil {
 		logrus.Warnf("Error writing to cert file: %v\n", err)
 	}
+	select {
+	case w.source.updateChan <- struct{}{}:
+	case <-time.After(w.source.updateTimeout):
+	}
 }
 
 func (w *watcher) OnError(err error) {

internal/common/spiretrustsource_test.go

@@ -12,12 +12,13 @@ import (
 	"github.com/stretchr/testify/require"
 )
 
-var (
+const (
 	certJSON = `{
 "certs": {
 	"spiffe://example.org": ["MIIB4TCCAUoCCQCfmw3vMgPS5TANBgkqhkiG9w0BAQQFADA1MQswCQYDVQQGEwJBVTETMBEGA1UECBMKU29tZS1TdGF0ZTERMA8GA1UEChMITUQ1IEluYy4wHhcNMTUxMjAzMTkyOTMyWhcNMjkwODEyMTkyOTMyWjA1MQswCQYDVQQGEwJBVTETMBEGA1UECBMKU29tZS1TdGF0ZTERMA8GA1UEChMITUQ1IEluYy4wgZ8wDQYJKoZIhvcNAQEBBQADgY0AMIGJAoGBANrq2nhLQj5mlXbpVX3QUPhfEm/vdEqPkoWtR/jRZIWm4WGfWpq/LKHJx2Pqwn+t117syN8l4U5unyAi1BJSXjBwPZNd7dXjcuJ+bRLV7FZ/iuvscfYyQQFTxan4TaJMd0x1HoNDbNbjHa02IyjjYE/r3mb/PIg+J2t5AZEh80lPAgMBAAEwDQYJKoZIhvcNAQEEBQADgYEAjGzp3K3ey/YfKHohf33yHHWd695HQxDAP+wYcs9/TAyLR+gJzJP7d18EcDDLJWVi7bhfa4EAD86di05azOh9kWSn4b3o9QYRGCSwGNnI3Zk0cwNKA49hZntKKiy22DhRk7JAHF01d6Bu3KkHkmENrtJ+zj/+159WAnUaqViorq4="]
 }
 }`
+	updateTimeout = 5 * time.Second
 )
 
 // makeX509SVIDResponse is a convenience function for generating X509 responses
@@ -67,7 +68,7 @@ func TestWriteCerts(t *testing.T) {
 	}, "vault-spire-certs.json")
 	require.NoError(t, err)
 
-	time.Sleep(1 * time.Second) // wait for watcher to get new certs
+	source.waitForUpdate(t)
 	source.Stop()
 
 	newSource, err := NewSpireTrustSource(map[string]string{}, "vault-spire-certs.json")
@@ -93,7 +94,7 @@ func TestSpireOverwrite(t *testing.T) {
 	require.NoError(t, err)
 	defer source.Stop()
 
-	time.Sleep(1 * time.Second) // wait for watcher to get new certs
+	source.waitForUpdate(t)
 	assert.Equal(t, ca.Roots(), source.TrustedCertificates()["spiffe://example.org"])
 }
 
@@ -113,13 +114,21 @@ func TestSpireReload(t *testing.T) {
 	require.NoError(t, err)
 	defer source.Stop()
 
-	time.Sleep(1 * time.Second) // wait for watcher to get new certs
+	source.waitForUpdate(t)
 	assert.Equal(t, ca.Roots(), source.TrustedCertificates()["spiffe://example.org"])
 
 	caRot := spiffetest.NewCA(t)
 	svidFooRot, keyFooRot := ca.CreateX509SVID("spiffe://example.org/foo")
 	setX509SVIDResponse(workloadAPI, caRot, svidFooRot, keyFooRot)
 
-	time.Sleep(1 * time.Second) // wait for watcher to get new certs
+	source.waitForUpdate(t)
 	assert.Equal(t, caRot.Roots(), source.TrustedCertificates()["spiffe://example.org"])
 }
+
+func (s *SpireTrustSource) waitForUpdate(t *testing.T) {
+	select {
+	case <-s.updateChan:
+	case <-time.After(updateTimeout):
+		require.Fail(t, "Timeout exceeding waiting for updates.")
+	}
+}