Fixed race condition with adding schedules

The multi-namespace functionality created a situation in the scheduler
where an informer would be created twice on the same namespace. This
would result in two entries in the schedule table being added, which
then resulted in only one of the schedule entries being removed. As such
"phantom scheduled events" would occur.

This patch borrows a pattern from the controllers that creates a mutex
on the informer creation function and only allows a single informer to
be created on a namespace at a given time.

Issue: [ch6933]
Issue: #1215

Author: Jonathan S. Katz

pgo-scheduler/pgo-scheduler.go

@@ -19,6 +19,7 @@ import (
 	"os"
 	"os/signal"
 	"strconv"
+	"sync"
 	"syscall"
 	"time"
 
@@ -49,6 +50,11 @@ var timeout time.Duration
 var seconds int
 var kubeClient *kubernetes.Clientset
 
+// this is used to prevent a race condition where an informer is being created
+// twice when a new scheduler-enabled ConfigMap is added.
+var informerNsMutex sync.Mutex
+var informerNamespaces map[string]struct{}
+
 func init() {
 	var err error
 	log.SetLevel(log.InfoLevel)
@@ -70,6 +76,11 @@ func init() {
 		log.Info("PGO_INSTALLATION_NAME set to " + installationName)
 	}
 
+	// set up the data structures for preventing double informers from being
+	// created
+	informerNsMutex = sync.Mutex{}
+	informerNamespaces = map[string]struct{}{}
+
 	pgoNamespace = os.Getenv(pgoNamespaceEnv)
 	if pgoNamespace == "" {
 		log.WithFields(log.Fields{}).Fatalf("Failed to get PGO_OPERATOR_NAMESPACE environment: %s", pgoNamespaceEnv)
@@ -162,6 +173,17 @@ func newKubeClient() (*kubernetes.Clientset, error) {
 }
 
 func SetupWatch(namespace string, scheduler *scheduler.Scheduler, stop chan struct{}) {
+	// don't create informer for namespace if one has already been created
+	informerNsMutex.Lock()
+	defer informerNsMutex.Unlock()
+
+	// if the namespace is already in the informer namespaces map, then exit
+	if _, ok := informerNamespaces[namespace]; ok {
+		return
+	}
+
+	informerNamespaces[namespace] = struct{}{}
+
 	watchlist := cache.NewListWatchFromClient(kubeClient.Core().RESTClient(),
 		"configmaps", namespace, fields.Everything())