Integration tests: relax testing requirements.

Prior to this commit the ct-woodpecker integration tests were a bit
flaky. The root cause is our usage of elapsed ct-woodpecker runtime as
a proxy for work completed. This is hard to do with accuracy and so (as
one example) the time required to fetch metrics output before stopping
the ct-woodpecker instance could result in another sth fetch/cert submit
cycle happening before the test completes, causing miscounts of expected
operations.

This commit changes the tests to just ensure a minimum of 1 cycles worth
of work (plus the startup cycle) occurs during testing. This is Good
Enough and will save us from considerable headaches tweaking padding
timeouts or overcomplicating the design of ct-woodpecker to expose cycle
information back to the test rig.

This commit also adds clean shutdown code to the monitors. This prevents
ghost monitor goroutines from sticking around when the parent
ct-woodpecker is stopped.

monitor/cert_submitter.go

@@ -96,6 +96,8 @@ type certSubmitter struct {
 	logURI string
 	stats  *certSubmitterStats
 
+	stopChannel chan bool
+
 	// How long to sleep between submitting certificates to the log
 	certSubmitInterval time.Duration
 	// Timeout for precert/cert submissions to the log
@@ -115,14 +117,24 @@ type certSubmitter struct {
 func (c *certSubmitter) run() {
 	go func() {
 		for {
-			c.submitCertificates()
-			c.logger.Printf("Sleeping for %s before next certificate submission\n",
-				c.certSubmitInterval)
-			c.clk.Sleep(c.certSubmitInterval)
+			select {
+			case <-c.stopChannel:
+				return
+			default:
+				c.submitCertificates()
+				c.logger.Printf("Sleeping for %s before next certificate submission\n",
+					c.certSubmitInterval)
+				c.clk.Sleep(c.certSubmitInterval)
+			}
 		}
 	}()
 }
 
+func (c *certSubmitter) stop() {
+	c.logger.Printf("Stopping %s certSubmitter", c.logURI)
+	c.stopChannel <- true
+}
+
 // submitCertificates issues a pre-certificate and a matching certificate with
 // the certSubmitter's certIssuer/certIssuerKey. If `submitPreCert` is enabled
 // then a precert is submitted. If `submitCert` is enabled then a final cert is

monitor/monitor.go

@@ -130,6 +130,7 @@ func New(opts MonitorOptions, logger *log.Logger, clk clock.Clock) (*Monitor, er
 			stats:            sthStats,
 			client:           client,
 			logURI:           opts.LogURI,
+			stopChannel:      make(chan bool, 1),
 			sthFetchInterval: opts.FetchOpts.Interval,
 			sthTimeout:       opts.FetchOpts.Timeout,
 		}
@@ -142,6 +143,7 @@ func New(opts MonitorOptions, logger *log.Logger, clk clock.Clock) (*Monitor, er
 			stats:              certStats,
 			client:             client,
 			logURI:             opts.LogURI,
+			stopChannel:        make(chan bool, 1),
 			certSubmitInterval: opts.SubmitOpts.Interval,
 			certSubmitTimeout:  opts.SubmitOpts.Timeout,
 			certIssuer:         opts.SubmitOpts.IssuerCert,
@@ -177,3 +179,13 @@ func (m *Monitor) Run() {
 		m.submitter.run()
 	}
 }
+
+func (m *Monitor) Stop() {
+	if m.fetcher != nil {
+		m.fetcher.stop()
+	}
+
+	if m.submitter != nil {
+		m.submitter.stop()
+	}
+}

monitor/sth_fetcher.go

@@ -71,6 +71,8 @@ type sthFetcher struct {
 	logURI string
 	stats  *sthFetchStats
 
+	stopChannel chan bool
+
 	// How long to sleep between fetching the log's current STH
 	sthFetchInterval time.Duration
 	// How long to wait before giving up on an STH fetch
@@ -82,13 +84,23 @@ type sthFetcher struct {
 func (f *sthFetcher) run() {
 	go func() {
 		for {
-			go f.observeSTH()
-			f.logger.Printf("Sleeping for %s before next STH check\n", f.sthFetchInterval)
-			f.clk.Sleep(f.sthFetchInterval)
+			select {
+			case <-f.stopChannel:
+				return
+			default:
+				go f.observeSTH()
+				f.logger.Printf("Sleeping for %s before next STH check\n", f.sthFetchInterval)
+				f.clk.Sleep(f.sthFetchInterval)
+			}
 		}
 	}()
 }
 
+func (f *sthFetcher) stop() {
+	f.logger.Printf("Stopping %s sthFetcher", f.logURI)
+	f.stopChannel <- true
+}
+
 // observeSTH fetches a monitored log's signed tree head (STH). The latency of
 // this operation is published to the `sth_latency` metric. The clocktime elapsed
 // since the STH's timestamp is published to the `sth_age` metric. If an error

test/cttestsrv/server.go

@@ -172,11 +172,15 @@ func (is *IntegrationSrv) addChainHandler(w http.ResponseWriter, r *http.Request
 		precert = true
 	}
 
+	is.logger.Printf("%s %s request received.", is.Addr, r.URL.Path)
+	start := time.Now()
 	sct, err := is.addChain(addChainReq.Chain, precert)
 	if err != nil {
 		http.Error(w, err.Error(), http.StatusBadRequest)
 		return
 	}
+	elapsed := time.Since(start)
+	is.logger.Printf("%s %s request completed %s later", is.Addr, r.URL.Path, elapsed)
 
 	w.WriteHeader(http.StatusOK)
 	_, _ = w.Write(sct)
@@ -196,8 +200,10 @@ func (is *IntegrationSrv) getSubmissionsHandler(w http.ResponseWriter, r *http.R
 		return
 	}
 
+	is.logger.Printf("%s %s request received.", is.Addr, r.URL.Path)
 	w.WriteHeader(http.StatusOK)
 	fmt.Fprintf(w, "%d", is.Submissions())
+	is.logger.Printf("%s %s request completed.", is.Addr, r.URL.Path)
 }
 
 // getSTHHandler processes GET requests for the CT get-sth endpoint. It returns
@@ -209,6 +215,9 @@ func (is *IntegrationSrv) getSTHHandler(w http.ResponseWriter, r *http.Request)
 		return
 	}
 
+	is.logger.Printf("%s %s request received.", is.Addr, r.URL.Path)
+	start := time.Now()
+
 	is.sleep()
 	// Track that an STH was fetched
 	atomic.AddInt64(&is.sthFetches, 1)
@@ -220,6 +229,8 @@ func (is *IntegrationSrv) getSTHHandler(w http.ResponseWriter, r *http.Request)
 		http.Error(w, err.Error(), http.StatusInternalServerError)
 		return
 	}
+	elapsed := time.Since(start)
+	is.logger.Printf("%s %s request completed %s later", is.Addr, r.URL.Path, elapsed)
 
 	w.WriteHeader(http.StatusOK)
 	fmt.Fprintf(w, "%s", response)