S3 peers should be initialized properly (#5024)

Fixes #4991

cmd/bucket-notification-handlers.go

@@ -296,6 +296,7 @@ func (api objectAPIHandlers) ListenBucketNotificationHandler(w http.ResponseWrit
 		return
 	}
 
+	targetServer := GetLocalPeer(globalEndpoints)
 	accountID := fmt.Sprintf("%d", UTCNow().UnixNano())
 	accountARN := fmt.Sprintf(
 		"%s:%s:%s:%s-%s",
@@ -303,8 +304,9 @@ func (api objectAPIHandlers) ListenBucketNotificationHandler(w http.ResponseWrit
 		serverConfig.GetRegion(),
 		accountID,
 		snsTypeMinio,
-		globalMinioAddr,
+		targetServer,
 	)
+
 	var filterRules []filterRule
 
 	for _, prefix := range prefixes {
@@ -357,7 +359,7 @@ func (api objectAPIHandlers) ListenBucketNotificationHandler(w http.ResponseWrit
 	// nEventCh
 	lc := listenerConfig{
 		TopicConfig:  *topicCfg,
-		TargetServer: globalMinioAddr,
+		TargetServer: targetServer,
 	}
 
 	err = AddBucketListenerConfig(bucket, &lc, objAPI)

cmd/endpoint.go

@@ -419,6 +419,29 @@ func CreateEndpoints(serverAddr string, args ...string) (string, EndpointList, S
 	return serverAddr, endpoints, setupType, nil
 }
 
+// GetLocalPeer - returns local peer value, returns globalMinioAddr
+// for FS and Erasure mode. In case of distributed server return
+// the first element from the set of peers which indicate that
+// they are local. There is always one entry that is local
+// even with repeated server endpoints.
+func GetLocalPeer(endpoints EndpointList) (localPeer string) {
+	peerSet := set.NewStringSet()
+	for _, endpoint := range endpoints {
+		if endpoint.Type() != URLEndpointType {
+			continue
+		}
+		if endpoint.IsLocal && endpoint.Host != "" {
+			peerSet.Add(endpoint.Host)
+		}
+	}
+	if peerSet.IsEmpty() {
+		// If local peer is empty can happen in FS or Erasure coded mode.
+		// then set the value to globalMinioAddr instead.
+		return globalMinioAddr
+	}
+	return peerSet.ToSlice()[0]
+}
+
 // GetRemotePeers - get hosts information other than this minio service.
 func GetRemotePeers(endpoints EndpointList) []string {
 	peerSet := set.NewStringSet()

cmd/endpoint_test.go

@@ -330,6 +330,38 @@ func TestCreateEndpoints(t *testing.T) {
 	}
 }
 
+// Tests get local peer functionality, local peer is supposed to only return one entry per minio service.
+// So it means that if you have say localhost:9000 and localhost:9001 as endpointArgs then localhost:9001
+// is considered a remote service from localhost:9000 perspective.
+func TestGetLocalPeer(t *testing.T) {
+	tempGlobalMinioAddr := globalMinioAddr
+	defer func() {
+		globalMinioAddr = tempGlobalMinioAddr
+	}()
+	globalMinioAddr = ":9000"
+
+	testCases := []struct {
+		endpointArgs   []string
+		expectedResult string
+	}{
+		{[]string{"/d1", "/d2", "d3", "d4"}, ":9000"},
+		{[]string{"http://localhost:9000/d1", "http://localhost:9000/d2", "http://example.org:9000/d3", "http://example.com:9000/d4"},
+			"localhost:9000"},
+		{[]string{"http://localhost:9000/d1", "http://example.org:9000/d2", "http://example.com:9000/d3", "http://example.net:9000/d4"},
+			"localhost:9000"},
+		{[]string{"http://localhost:9000/d1", "http://localhost:9001/d2", "http://localhost:9002/d3", "http://localhost:9003/d4"},
+			"localhost:9000"},
+	}
+
+	for i, testCase := range testCases {
+		endpoints, _ := NewEndpointList(testCase.endpointArgs...)
+		remotePeer := GetLocalPeer(endpoints)
+		if remotePeer != testCase.expectedResult {
+			t.Fatalf("Test %d: expected: %v, got: %v", i+1, testCase.expectedResult, remotePeer)
+		}
+	}
+}
+
 func TestGetRemotePeers(t *testing.T) {
 	tempGlobalMinioPort := globalMinioPort
 	defer func() {

cmd/s3-peer-client.go

@@ -41,12 +41,13 @@ type s3Peers []s3Peer
 // slice. The urls slice is assumed to be non-empty and free of nil
 // values.
 func makeS3Peers(endpoints EndpointList) (s3PeerList s3Peers) {
+	localAddr := GetLocalPeer(endpoints)
 	s3PeerList = append(s3PeerList, s3Peer{
-		globalMinioAddr,
+		localAddr,
 		&localBucketMetaState{ObjectAPI: newObjectLayerFn},
 	})
 
-	hostSet := set.CreateStringSet(globalMinioAddr)
+	hostSet := set.CreateStringSet(localAddr)
 	cred := serverConfig.GetCredential()
 	serviceEndpoint := path.Join(minioReservedBucketPath, s3Path)
 	for _, host := range GetRemotePeers(endpoints) {
@@ -56,17 +57,17 @@ func makeS3Peers(endpoints EndpointList) (s3PeerList s3Peers) {
 		hostSet.Add(host)
 		s3PeerList = append(s3PeerList, s3Peer{
 			addr: host,
-			bmsClient: &remoteBucketMetaState{newAuthRPCClient(authConfig{
-				accessKey:       cred.AccessKey,
-				secretKey:       cred.SecretKey,
-				serverAddr:      host,
-				serviceEndpoint: serviceEndpoint,
-				secureConn:      globalIsSSL,
-				serviceName:     "S3",
-			})},
+			bmsClient: &remoteBucketMetaState{
+				newAuthRPCClient(authConfig{
+					accessKey:       cred.AccessKey,
+					secretKey:       cred.SecretKey,
+					serverAddr:      host,
+					serviceEndpoint: serviceEndpoint,
+					secureConn:      globalIsSSL,
+					serviceName:     "S3",
+				})},
 		})
 	}
-
 	return s3PeerList
 }
 

cmd/s3-peer-client_test.go

@@ -39,7 +39,6 @@ func TestMakeS3Peers(t *testing.T) {
 		peers      []string
 	}{
 		{"127.0.0.1:9000", mustGetNewEndpointList("/mnt/disk1"), []string{"127.0.0.1:9000"}},
-		{"127.0.0.1:9000", mustGetNewEndpointList("http://localhost:9001/d1"), []string{"127.0.0.1:9000", "localhost:9001"}},
 		{"example.org:9000", mustGetNewEndpointList("http://example.org:9000/d1", "http://example.com:9000/d1", "http://example.net:9000/d1", "http://example.edu:9000/d1"), []string{"example.org:9000", "example.com:9000", "example.edu:9000", "example.net:9000"}},
 	}