Merge pull request #228 from m-lab/sandbox-update-mutex

Add a mutex for directory update.

main.go

@@ -11,17 +11,16 @@ import (
 	"runtime"
 	"time"
 
-	"github.com/m-lab/go/prometheusx"
-
 	"github.com/m-lab/annotation-service/handler"
 	"github.com/m-lab/annotation-service/manager"
 	"github.com/m-lab/go/memoryless"
+	"github.com/m-lab/go/prometheusx"
 )
 
 var (
 	updateInterval = flag.Duration("update_interval", time.Duration(24)*time.Hour, "Run the update dataset job with this frequency.")
-	minInterval    = flag.Duration("min_interval", time.Duration(23)*time.Hour, "minimum gap between 2 runs.")
-	maxInterval    = flag.Duration("max_interval", time.Duration(25)*time.Hour, "maximum gap between 2 runs.")
+	minInterval    = flag.Duration("min_interval", time.Duration(18)*time.Hour, "minimum gap between 2 runs.")
+	maxInterval    = flag.Duration("max_interval", time.Duration(26)*time.Hour, "maximum gap between 2 runs.")
 	// Create a single unified context and a cancellationMethod for said context.
 	ctx, cancelCtx = context.WithCancel(context.Background())
 )
@@ -66,6 +65,7 @@ func main() {
 	runtime.SetMutexProfileFraction(1000)
 
 	log.Print("Beginning Setup\n")
+	prometheusx.MustStartPrometheus(":9090")
 
 	go memoryless.Run(ctx, manager.MustUpdateDirectory,
 		memoryless.Config{Expected: *updateInterval, Min: *minInterval, Max: *maxInterval})
@@ -74,7 +74,6 @@ func main() {
 	http.HandleFunc("/updateDatasets", updateMaxmindDatasets)
 
 	handler.InitHandler()
-	prometheusx.MustStartPrometheus(":9090")
 	log.Print("Listening on port 8080")
 	log.Fatal(http.ListenAndServe(":8080", nil))
 }

manager/manager.go

@@ -102,6 +102,7 @@ func MustUpdateDirectory() {
 // listBuilder wraps a set of CachingLoaders, and creates a set of merged Annotators on request.
 // TODO - unit tests?
 type listBuilder struct {
+	mutex    sync.Mutex        // Prevents concurrent update and/or build
 	legacyV4 api.CachingLoader // loader for legacy v4 annotators
 	legacyV6 api.CachingLoader // loader for legacy v6 annotators
 	geolite2 api.CachingLoader // loader for geolite2 annotators
@@ -119,6 +120,9 @@ func newListBuilder(v4, v6, g2 api.CachingLoader) *listBuilder {
 
 // Update updates the (dynamic) CachingLoaders
 func (bldr *listBuilder) update() error {
+	bldr.mutex.Lock()
+	defer bldr.mutex.Unlock()
+
 	var errV4, errV6, errG2 error
 
 	wg := sync.WaitGroup{}
@@ -152,6 +156,9 @@ func (bldr *listBuilder) update() error {
 // build creates a complete list of CompositeAnnotators from the cached annotators
 // from the CachingLoaders.
 func (bldr *listBuilder) build() []api.Annotator {
+	bldr.mutex.Lock()
+	defer bldr.mutex.Unlock()
+
 	v4 := directory.SortSlice(bldr.legacyV4.Fetch())
 	v6 := directory.SortSlice(bldr.legacyV6.Fetch())