Make addition of TopicConfig to globalEventNotifier go-routine safe (#2806)

cmd/bucket-handlers.go

@@ -331,7 +331,6 @@ func (api objectAPIHandlers) DeleteMultipleObjectsHandler(w http.ResponseWriter,
 	// Write success response.
 	writeSuccessResponse(w, encodedSuccessResponse)
 
-	if globalEventNotifier.IsBucketNotificationSet(bucket) {
 	// Notify deleted event for objects.
 	for _, dobj := range deletedObjects {
 		eventNotify(eventData{
@@ -345,7 +344,6 @@ func (api objectAPIHandlers) DeleteMultipleObjectsHandler(w http.ResponseWriter,
 			},
 		})
 	}
-	}
 }
 
 // PutBucketHandler - PUT Bucket
@@ -453,7 +451,6 @@ func (api objectAPIHandlers) PostPolicyBucketHandler(w http.ResponseWriter, r *h
 	// Write successful response.
 	writeSuccessNoContent(w)
 
-	if globalEventNotifier.IsBucketNotificationSet(bucket) {
 	// Notify object created event.
 	eventNotify(eventData{
 		Type:    ObjectCreatedPost,
@@ -463,7 +460,6 @@ func (api objectAPIHandlers) PostPolicyBucketHandler(w http.ResponseWriter, r *h
 			"sourceIPAddress": r.RemoteAddr,
 		},
 	})
-	}
 }
 
 // HeadBucketHandler - HEAD Bucket

cmd/bucket-notification-handlers.go

@@ -270,12 +270,8 @@ func (api objectAPIHandlers) ListenBucketNotificationHandler(w http.ResponseWrit
 		})
 	}
 
-	// Fetch for existing notification configs and update topic configs.
+	// Make topic configuration corresponding to this ListenBucketNotification request.
-	nConfig := globalEventNotifier.GetBucketNotificationConfig(bucket)
+	topicCfg := &topicConfig{
-	if nConfig == nil {
-		// No notification configs found, initialize.
-		nConfig = &notificationConfig{
-			TopicConfigs: []topicConfig{{
 		TopicARN: accountARN,
 		serviceConfig: serviceConfig{
 			Events: events,
@@ -288,29 +284,10 @@ func (api objectAPIHandlers) ListenBucketNotificationHandler(w http.ResponseWrit
 			},
 			ID: "sns-" + accountID,
 		},
-			}},
-		}
-	} else {
-		// Previously set notification configs found append to
-		// existing topic configs.
-		nConfig.TopicConfigs = append(nConfig.TopicConfigs, topicConfig{
-			TopicARN: accountARN,
-			serviceConfig: serviceConfig{
-				Events: events,
-				Filter: struct {
-					Key keyFilter `xml:"S3Key,omitempty"`
-				}{
-					Key: keyFilter{
-						FilterRules: filterRules,
-					},
-				},
-				ID: "sns-" + accountID,
-			},
-		})
 	}
 
-	// Save bucket notification config.
+	// Add topic config to bucket notification config.
-	if err = globalEventNotifier.SetBucketNotificationConfig(bucket, nConfig); err != nil {
+	if err = globalEventNotifier.AddTopicConfig(bucket, topicCfg); err != nil {
 		writeErrorResponse(w, r, toAPIErrorCode(err), r.URL.Path)
 		return
 	}

cmd/event-notifier.go

@@ -137,17 +137,6 @@ func (en *eventNotifier) RemoveSNSTarget(snsARN string, listenerCh chan []Notifi
 	}
 }
 
-// Returns true if bucket notification is set for the bucket, false otherwise.
-func (en *eventNotifier) IsBucketNotificationSet(bucket string) bool {
-	if en == nil {
-		return false
-	}
-	en.rwMutex.RLock()
-	defer en.rwMutex.RUnlock()
-	_, ok := en.notificationConfigs[bucket]
-	return ok
-}
-
 // Fetch bucket notification config for an input bucket.
 func (en eventNotifier) GetBucketNotificationConfig(bucket string) *notificationConfig {
 	en.rwMutex.RLock()
@@ -167,6 +156,23 @@ func (en *eventNotifier) SetBucketNotificationConfig(bucket string, notification
 	return nil
 }
 
+func (en *eventNotifier) AddTopicConfig(bucket string, topicCfg *topicConfig) error {
+	en.rwMutex.Lock()
+	defer en.rwMutex.Unlock()
+	if topicCfg == nil {
+		return errInvalidArgument
+	}
+	notificationCfg := en.notificationConfigs[bucket]
+	if notificationCfg == nil {
+		en.notificationConfigs[bucket] = &notificationConfig{
+			TopicConfigs: []topicConfig{*topicCfg},
+		}
+		return nil
+	}
+	notificationCfg.TopicConfigs = append(notificationCfg.TopicConfigs, *topicCfg)
+	return nil
+}
+
 // eventNotify notifies an event to relevant targets based on their
 // bucket notification configs.
 func eventNotify(event eventData) {

cmd/event-notifier_test.go

@@ -17,6 +17,7 @@
 package cmd
 
 import (
+	"fmt"
 	"reflect"
 	"testing"
 	"time"
@@ -64,11 +65,11 @@ func testEventNotify(obj ObjectLayer, instanceType string, t TestErrHandler) {
 		t.Errorf("Expected error to be nil, got %s", err)
 	}
 
-	if !globalEventNotifier.IsBucketNotificationSet(bucketName) {
+	nConfig := globalEventNotifier.GetBucketNotificationConfig(bucketName)
+	if nConfig == nil {
 		t.Errorf("Notification expected to be set, but notification not set.")
 	}
 
-	nConfig := globalEventNotifier.GetBucketNotificationConfig(bucketName)
 	if !reflect.DeepEqual(nConfig, &notificationConfig{}) {
 		t.Errorf("Mismatching notification configs.")
 	}
@@ -381,3 +382,69 @@ func TestListenBucketNotification(t *testing.T) {
 		break
 	}
 }
+
+func testAddTopicConfig(obj ObjectLayer, instanceType string, t TestErrHandler) {
+	root, cErr := newTestConfig("us-east-1")
+	if cErr != nil {
+		t.Fatalf("[%s] Failed to initialize test config: %v", instanceType, cErr)
+	}
+	defer removeAll(root)
+
+	if err := initEventNotifier(obj); err != nil {
+		t.Fatalf("[%s] : Failed to initialize event notifier: %v", instanceType, err)
+	}
+
+	// Make a bucket to store topicConfigs.
+	randBucket := getRandomBucketName()
+	if err := obj.MakeBucket(randBucket); err != nil {
+		t.Fatalf("[%s] : Failed to make bucket %s", instanceType, randBucket)
+	}
+
+	// Add a topicConfig to an empty notificationConfig.
+	accountID := fmt.Sprintf("%d", time.Now().UTC().UnixNano())
+	accountARN := "arn:minio:sns:" + serverConfig.GetRegion() + accountID + ":listen"
+	var filterRules []filterRule
+	filterRules = append(filterRules, filterRule{
+		Name:  "prefix",
+		Value: "minio",
+	})
+	filterRules = append(filterRules, filterRule{
+		Name:  "suffix",
+		Value: "*.jpg",
+	})
+
+	// Make topic configuration corresponding to this ListenBucketNotification request.
+	sampleTopicCfg := &topicConfig{
+		TopicARN: accountARN,
+		serviceConfig: serviceConfig{
+			Filter: struct {
+				Key keyFilter `xml:"S3Key,omitempty"`
+			}{
+				Key: keyFilter{
+					FilterRules: filterRules,
+				},
+			},
+			ID: "sns-" + accountID,
+		},
+	}
+	testCases := []struct {
+		topicCfg    *topicConfig
+		expectedErr error
+	}{
+		{sampleTopicCfg, nil},
+		{nil, errInvalidArgument},
+		{sampleTopicCfg, nil},
+	}
+
+	for i, test := range testCases {
+		err := globalEventNotifier.AddTopicConfig(randBucket, test.topicCfg)
+		if err != test.expectedErr {
+			t.Errorf("Test %d: %s failed with error %v, expected to fail with %v",
+				i+1, instanceType, err, test.expectedErr)
+		}
+	}
+}
+
+func TestAddTopicConfig(t *testing.T) {
+	ExecObjectLayerTest(t, testAddTopicConfig)
+}

cmd/object-handlers.go

@@ -378,7 +378,6 @@ func (api objectAPIHandlers) CopyObjectHandler(w http.ResponseWriter, r *http.Re
 	// write success response.
 	writeSuccessResponse(w, encodedSuccessResponse)
 
-	if globalEventNotifier.IsBucketNotificationSet(bucket) {
 	// Notify object created event.
 	eventNotify(eventData{
 		Type:    ObjectCreatedCopy,
@@ -388,7 +387,6 @@ func (api objectAPIHandlers) CopyObjectHandler(w http.ResponseWriter, r *http.Re
 			"sourceIPAddress": r.RemoteAddr,
 		},
 	})
-	}
 }
 
 // PutObjectHandler - PUT Object
@@ -482,7 +480,6 @@ func (api objectAPIHandlers) PutObjectHandler(w http.ResponseWriter, r *http.Req
 	w.Header().Set("ETag", "\""+objInfo.MD5Sum+"\"")
 	writeSuccessResponse(w, nil)
 
-	if globalEventNotifier.IsBucketNotificationSet(bucket) {
 	// Notify object created event.
 	eventNotify(eventData{
 		Type:    ObjectCreatedPut,
@@ -492,7 +489,6 @@ func (api objectAPIHandlers) PutObjectHandler(w http.ResponseWriter, r *http.Req
 			"sourceIPAddress": r.RemoteAddr,
 		},
 	})
-	}
 }
 
 /// Multipart objectAPIHandlers
@@ -834,7 +830,6 @@ func (api objectAPIHandlers) CompleteMultipartUploadHandler(w http.ResponseWrite
 	w.Write(encodedSuccessResponse)
 	w.(http.Flusher).Flush()
 
-	if globalEventNotifier.IsBucketNotificationSet(bucket) {
 	// Fetch object info for notifications.
 	objInfo, err := objectAPI.GetObjectInfo(bucket, object)
 	if err != nil {
@@ -851,7 +846,6 @@ func (api objectAPIHandlers) CompleteMultipartUploadHandler(w http.ResponseWrite
 			"sourceIPAddress": r.RemoteAddr,
 		},
 	})
-	}
 }
 
 /// Delete objectAPIHandlers
@@ -895,7 +889,6 @@ func (api objectAPIHandlers) DeleteObjectHandler(w http.ResponseWriter, r *http.
 	}
 	writeSuccessNoContent(w)
 
-	if globalEventNotifier.IsBucketNotificationSet(bucket) {
 	// Notify object deleted event.
 	eventNotify(eventData{
 		Type:   ObjectRemovedDelete,
@@ -907,5 +900,4 @@ func (api objectAPIHandlers) DeleteObjectHandler(w http.ResponseWriter, r *http.
 			"sourceIPAddress": r.RemoteAddr,
 		},
 	})
-	}
 }

cmd/web-handlers.go

@@ -428,7 +428,6 @@ func (web *webAPIHandlers) Upload(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
-	if globalEventNotifier.IsBucketNotificationSet(bucket) {
 	// Notify object created event.
 	eventNotify(eventData{
 		Type:    ObjectCreatedPut,
@@ -438,7 +437,6 @@ func (web *webAPIHandlers) Upload(w http.ResponseWriter, r *http.Request) {
 			"sourceIPAddress": r.RemoteAddr,
 		},
 	})
-	}
 }
 
 // Download - file download handler.