api: Change ListenBucketNotification with new API format. (#2791)

Take prefix, suffix and events as query params.

cmd/api-resources.go

@@ -38,6 +38,7 @@ func getListObjectsV1Args(values url.Values) (prefix, marker, delimiter string,
 // Parse bucket url queries for ListObjects V2.
 func getListObjectsV2Args(values url.Values) (prefix, token, startAfter, delimiter string, fetchOwner bool, maxkeys int, encodingType string) {
 	prefix = values.Get("prefix")
+	token = values.Get("continuation-token")
 	startAfter = values.Get("start-after")
 	delimiter = values.Get("delimiter")
 	if values.Get("max-keys") != "" {
@@ -45,11 +46,8 @@ func getListObjectsV2Args(values url.Values) (prefix, token, startAfter, delimit
 	} else {
 		maxkeys = maxObjectList
 	}
+	fetchOwner = values.Get("fetch-owner") == "true"
 	encodingType = values.Get("encoding-type")
-	token = values.Get("continuation-token")
-	if values.Get("fetch-owner") == "true" {
-		fetchOwner = true
-	}
 	return
 }
 
@@ -80,3 +78,11 @@ func getObjectResources(values url.Values) (uploadID string, partNumberMarker, m
 	encodingType = values.Get("encoding-type")
 	return
 }
+
+// Parse listen bucket notification resources.
+func getListenBucketNotificationResources(values url.Values) (prefix string, suffix string, events []string) {
+	prefix = values.Get("prefix")
+	suffix = values.Get("suffix")
+	events = values["events"]
+	return prefix, suffix, events
+}

cmd/api-resources_test.go

@@ -0,0 +1,190 @@
+/*
+ * Minio Cloud Storage, (C) 2016 Minio, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package cmd
+
+import (
+	"net/url"
+	"testing"
+)
+
+// Test list objects resources V2.
+func TestListObjectsV2Resources(t *testing.T) {
+	testCases := []struct {
+		values                               url.Values
+		prefix, token, startAfter, delimiter string
+		fetchOwner                           bool
+		maxKeys                              int
+		encodingType                         string
+	}{
+		{
+			values: url.Values{
+				"prefix":             []string{"photos/"},
+				"continuation-token": []string{"token"},
+				"start-after":        []string{"start-after"},
+				"delimiter":          []string{"/"},
+				"fetch-owner":        []string{"true"},
+				"max-keys":           []string{"100"},
+				"encoding-type":      []string{"gzip"},
+			},
+			prefix:       "photos/",
+			token:        "token",
+			startAfter:   "start-after",
+			delimiter:    "/",
+			fetchOwner:   true,
+			maxKeys:      100,
+			encodingType: "gzip",
+		},
+		{
+			values: url.Values{
+				"prefix":             []string{"photos/"},
+				"continuation-token": []string{"token"},
+				"start-after":        []string{"start-after"},
+				"delimiter":          []string{"/"},
+				"fetch-owner":        []string{"true"},
+				"encoding-type":      []string{"gzip"},
+			},
+			prefix:       "photos/",
+			token:        "token",
+			startAfter:   "start-after",
+			delimiter:    "/",
+			fetchOwner:   true,
+			maxKeys:      1000,
+			encodingType: "gzip",
+		},
+	}
+
+	for i, testCase := range testCases {
+		prefix, token, startAfter, delimiter, fetchOwner, maxKeys, encodingType := getListObjectsV2Args(testCase.values)
+		if prefix != testCase.prefix {
+			t.Errorf("Test %d: Expected %s, got %s", i+1, testCase.prefix, prefix)
+		}
+		if token != testCase.token {
+			t.Errorf("Test %d: Expected %s, got %s", i+1, testCase.token, token)
+		}
+		if startAfter != testCase.startAfter {
+			t.Errorf("Test %d: Expected %s, got %s", i+1, testCase.startAfter, startAfter)
+		}
+		if delimiter != testCase.delimiter {
+			t.Errorf("Test %d: Expected %s, got %s", i+1, testCase.delimiter, delimiter)
+		}
+		if fetchOwner != testCase.fetchOwner {
+			t.Errorf("Test %d: Expected %t, got %t", i+1, testCase.fetchOwner, fetchOwner)
+		}
+		if maxKeys != testCase.maxKeys {
+			t.Errorf("Test %d: Expected %d, got %d", i+1, testCase.maxKeys, maxKeys)
+		}
+		if encodingType != testCase.encodingType {
+			t.Errorf("Test %d: Expected %s, got %s", i+1, testCase.encodingType, encodingType)
+		}
+	}
+}
+
+// Test list objects resources V1.
+func TestListObjectsV1Resources(t *testing.T) {
+	testCases := []struct {
+		values                    url.Values
+		prefix, marker, delimiter string
+		maxKeys                   int
+		encodingType              string
+	}{
+		{
+			values: url.Values{
+				"prefix":        []string{"photos/"},
+				"marker":        []string{"test"},
+				"delimiter":     []string{"/"},
+				"max-keys":      []string{"100"},
+				"encoding-type": []string{"gzip"},
+			},
+			prefix:       "photos/",
+			marker:       "test",
+			delimiter:    "/",
+			maxKeys:      100,
+			encodingType: "gzip",
+		},
+		{
+			values: url.Values{
+				"prefix":        []string{"photos/"},
+				"marker":        []string{"test"},
+				"delimiter":     []string{"/"},
+				"encoding-type": []string{"gzip"},
+			},
+			prefix:       "photos/",
+			marker:       "test",
+			delimiter:    "/",
+			maxKeys:      1000,
+			encodingType: "gzip",
+		},
+	}
+
+	for i, testCase := range testCases {
+		prefix, marker, delimiter, maxKeys, encodingType := getListObjectsV1Args(testCase.values)
+		if prefix != testCase.prefix {
+			t.Errorf("Test %d: Expected %s, got %s", i+1, testCase.prefix, prefix)
+		}
+		if marker != testCase.marker {
+			t.Errorf("Test %d: Expected %s, got %s", i+1, testCase.marker, marker)
+		}
+		if delimiter != testCase.delimiter {
+			t.Errorf("Test %d: Expected %s, got %s", i+1, testCase.delimiter, delimiter)
+		}
+		if maxKeys != testCase.maxKeys {
+			t.Errorf("Test %d: Expected %d, got %d", i+1, testCase.maxKeys, maxKeys)
+		}
+		if encodingType != testCase.encodingType {
+			t.Errorf("Test %d: Expected %s, got %s", i+1, testCase.encodingType, encodingType)
+		}
+	}
+}
+
+// Validates extracting information for object resources.
+func TestGetObjectsResources(t *testing.T) {
+	testCases := []struct {
+		values                     url.Values
+		uploadID                   string
+		partNumberMarker, maxParts int
+		encodingType               string
+	}{
+		{
+			values: url.Values{
+				"uploadId":           []string{"11123-11312312311231-12313"},
+				"part-number-marker": []string{"1"},
+				"max-parts":          []string{"1000"},
+				"encoding-type":      []string{"gzip"},
+			},
+			uploadID:         "11123-11312312311231-12313",
+			partNumberMarker: 1,
+			maxParts:         1000,
+			encodingType:     "gzip",
+		},
+	}
+
+	for i, testCase := range testCases {
+		uploadID, partNumberMarker, maxParts, encodingType := getObjectResources(testCase.values)
+		if uploadID != testCase.uploadID {
+			t.Errorf("Test %d: Expected %s, got %s", i+1, testCase.uploadID, uploadID)
+		}
+		if partNumberMarker != testCase.partNumberMarker {
+			t.Errorf("Test %d: Expected %d, got %d", i+1, testCase.partNumberMarker, partNumberMarker)
+		}
+		if maxParts != testCase.maxParts {
+			t.Errorf("Test %d: Expected %d, got %d", i+1, testCase.maxParts, maxParts)
+		}
+		if encodingType != testCase.encodingType {
+			t.Errorf("Test %d: Expected %s, got %s", i+1, testCase.encodingType, encodingType)
+		}
+	}
+}

cmd/api-router.go

@@ -63,7 +63,7 @@ func registerAPIRouter(mux *router.Router, api objectAPIHandlers) {
 	// GetBucketNotification
 	bucket.Methods("GET").HandlerFunc(api.GetBucketNotificationHandler).Queries("notification", "")
 	// ListenBucketNotification
-	bucket.Methods("GET").HandlerFunc(api.ListenBucketNotificationHandler).Queries("notificationARN", "{notificationARN:.*}")
+	bucket.Methods("GET").HandlerFunc(api.ListenBucketNotificationHandler).Queries("events", "{events:.*}")
 	// ListMultipartUploads
 	bucket.Methods("GET").HandlerFunc(api.ListMultipartUploadsHandler).Queries("uploads", "")
 	// ListObjectsV2

cmd/bucket-notification-handlers.go

@@ -20,6 +20,7 @@ import (
 	"bytes"
 	"encoding/json"
 	"encoding/xml"
+	"fmt"
 	"io"
 	"net/http"
 	"path"
@@ -231,13 +232,21 @@ func (api objectAPIHandlers) ListenBucketNotificationHandler(w http.ResponseWrit
 	vars := mux.Vars(r)
 	bucket := vars["bucket"]
 
-	// Get notification ARN.
-	topicARN := r.URL.Query().Get("notificationARN")
-	if topicARN == "" {
-		writeErrorResponse(w, r, ErrARNNotification, r.URL.Path)
+	// Parse listen bucket notification resources.
+	prefix, suffix, events := getListenBucketNotificationResources(r.URL.Query())
+	if !IsValidObjectPrefix(prefix) || !IsValidObjectPrefix(suffix) {
+		writeErrorResponse(w, r, ErrFilterValueInvalid, r.URL.Path)
 		return
 	}
 
+	// Validate all the resource events.
+	for _, event := range events {
+		if errCode := checkEvent(event); errCode != ErrNone {
+			writeErrorResponse(w, r, errCode, r.URL.Path)
+			return
+		}
+	}
+
 	_, err := objAPI.GetBucketInfo(bucket)
 	if err != nil {
 		errorIf(err, "Unable to bucket info.")
@@ -245,16 +254,64 @@ func (api objectAPIHandlers) ListenBucketNotificationHandler(w http.ResponseWrit
 		return
 	}
 
-	notificationCfg := globalEventNotifier.GetBucketNotificationConfig(bucket)
-	if notificationCfg == nil {
-		writeErrorResponse(w, r, ErrARNNotification, r.URL.Path)
-		return
+	accountID := fmt.Sprintf("%d", time.Now().UTC().UnixNano())
+	accountARN := "arn:minio:sns:" + serverConfig.GetRegion() + accountID + ":listen"
+	var filterRules []filterRule
+	if prefix != "" {
+		filterRules = append(filterRules, filterRule{
+			Name:  "prefix",
+			Value: prefix,
+		})
+	}
+	if suffix != "" {
+		filterRules = append(filterRules, filterRule{
+			Name:  "suffix",
+			Value: suffix,
+		})
 	}
 
-	// Set SNS notifications only if special "listen" sns is set in bucket
-	// notification configs.
-	if !isMinioSNSConfigured(topicARN, notificationCfg.TopicConfigs) {
-		writeErrorResponse(w, r, ErrARNNotification, r.URL.Path)
+	// Fetch for existing notification configs and update topic configs.
+	nConfig := globalEventNotifier.GetBucketNotificationConfig(bucket)
+	if nConfig == nil {
+		// No notification configs found, initialize.
+		nConfig = &notificationConfig{
+			TopicConfigs: []topicConfig{{
+				TopicARN: accountARN,
+				serviceConfig: serviceConfig{
+					Events: events,
+					Filter: struct {
+						Key keyFilter `xml:"S3Key,omitempty"`
+					}{
+						Key: keyFilter{
+							FilterRules: filterRules,
+						},
+					},
+					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.
+	if err = globalEventNotifier.SetBucketNotificationConfig(bucket, nConfig); err != nil {
+		writeErrorResponse(w, r, toAPIErrorCode(err), r.URL.Path)
 		return
 	}
 
@@ -267,9 +324,9 @@ func (api objectAPIHandlers) ListenBucketNotificationHandler(w http.ResponseWrit
 	defer close(nEventCh)
 
 	// Set sns target.
-	globalEventNotifier.SetSNSTarget(topicARN, nEventCh)
+	globalEventNotifier.SetSNSTarget(accountARN, nEventCh)
 	// Remove sns listener after the writer has closed or the client disconnected.
-	defer globalEventNotifier.RemoveSNSTarget(topicARN, nEventCh)
+	defer globalEventNotifier.RemoveSNSTarget(accountARN, nEventCh)
 
 	// Start sending bucket notifications.
 	sendBucketNotification(w, nEventCh)