@ -19,7 +19,6 @@ package cmd
import (
"bytes"
"context"
"encoding/json"
"encoding/xml"
"fmt"
"io"
@ -590,24 +589,6 @@ func (sys *NotificationSys) PutBucketNotification(ctx context.Context, bucketNam
} ( )
}
// ListenBucketNotification - calls ListenBucketNotification RPC call on all peers.
func ( sys * NotificationSys ) ListenBucketNotification ( ctx context . Context , bucketName string ,
eventNames [ ] event . Name , pattern string , targetID event . TargetID , localPeer xnet . Host ) {
go func ( ) {
ng := WithNPeers ( len ( sys . peerClients ) )
for idx , client := range sys . peerClients {
if client == nil {
continue
}
client := client
ng . Go ( ctx , func ( ) error {
return client . ListenBucketNotification ( bucketName , eventNames , pattern , targetID , localPeer )
} , idx , * client . host )
}
ng . Wait ( )
} ( )
}
// AddNotificationTargetsFromConfig - adds notification targets from server config.
func ( sys * NotificationSys ) AddNotificationTargetsFromConfig ( cfg config . Config ) error {
targetList , err := notify . GetNotificationTargets ( cfg , GlobalServiceDoneCh , NewCustomHTTPTransport ( ) )
@ -660,82 +641,6 @@ func (sys *NotificationSys) RemoteTargetExist(bucketName string, targetID event.
return ok
}
// ListenBucketNotificationArgs - listen bucket notification RPC arguments.
type ListenBucketNotificationArgs struct {
BucketName string ` json:"-" `
EventNames [ ] event . Name ` json:"eventNames" `
Pattern string ` json:"pattern" `
TargetID event . TargetID ` json:"targetId" `
Addr xnet . Host ` json:"addr" `
}
// initListeners - initializes PeerREST clients available in listener.json.
func ( sys * NotificationSys ) initListeners ( ctx context . Context , objAPI ObjectLayer , bucketName string ) error {
// listener.json is available/applicable only in DistXL mode.
if ! globalIsDistXL {
return nil
}
// Construct path to listener.json for the given bucket.
configFile := path . Join ( bucketConfigPrefix , bucketName , bucketListenerConfig )
configData , e := readConfig ( ctx , objAPI , configFile )
if e != nil && ! IsErrIgnored ( e , errDiskNotFound , errConfigNotFound ) {
return e
}
listenerList := [ ] ListenBucketNotificationArgs { }
if configData != nil {
if err := json . Unmarshal ( configData , & listenerList ) ; err != nil {
logger . LogIf ( ctx , err )
return err
}
}
if len ( listenerList ) == 0 {
// Nothing to initialize for empty listener list.
return nil
}
for _ , args := range listenerList {
found , err := isLocalHost ( args . Addr . Name , args . Addr . Port . String ( ) , args . Addr . Port . String ( ) )
if err != nil {
logger . GetReqInfo ( ctx ) . AppendTags ( "host" , args . Addr . String ( ) )
logger . LogIf ( ctx , err )
return err
}
if found {
// As this function is called at startup, skip HTTP listener to this host.
continue
}
client , err := newPeerRESTClient ( & args . Addr )
if err != nil {
return fmt . Errorf ( "unable to find PeerHost by address %v in listener.json for bucket %v" , args . Addr , bucketName )
}
exist , err := client . RemoteTargetExist ( bucketName , args . TargetID )
if err != nil {
logger . GetReqInfo ( ctx ) . AppendTags ( "targetID" , args . TargetID . Name )
logger . LogIf ( ctx , err )
return err
}
if ! exist {
// Skip previously connected HTTP listener which is not found in remote peer.
continue
}
target := NewPeerRESTClientTarget ( bucketName , args . TargetID , client )
rulesMap := event . NewRulesMap ( args . EventNames , args . Pattern , target . ID ( ) )
if err = sys . AddRemoteTarget ( bucketName , target , rulesMap ) ; err != nil {
logger . GetReqInfo ( ctx ) . AppendTags ( "targetName" , target . id . Name )
logger . LogIf ( ctx , err )
return err
}
}
return nil
}
// Loads notification policies for all buckets into NotificationSys.
func ( sys * NotificationSys ) load ( buckets [ ] BucketInfo , objAPI ObjectLayer ) error {
for _ , bucket := range buckets {
@ -751,9 +656,6 @@ func (sys *NotificationSys) load(buckets []BucketInfo, objAPI ObjectLayer) error
continue
}
sys . AddRulesMap ( bucket . Name , config . ToRulesMap ( ) )
if err = sys . initListeners ( ctx , objAPI , bucket . Name ) ; err != nil {
return err
}
}
return nil
}
@ -1283,7 +1185,6 @@ func (args eventArgs) ToEvent() event.Event {
}
func sendEvent ( args eventArgs ) {
// remove sensitive encryption entries in metadata.
switch {
case crypto . IsEncrypted ( args . Object . UserDefined ) :
@ -1297,16 +1198,15 @@ func sendEvent(args eventArgs) {
crypto . RemoveSensitiveEntries ( args . Object . UserDefined )
crypto . RemoveInternalEntries ( args . Object . UserDefined )
if globalHTTPListen . HasSubscribers ( ) {
globalHTTPListen . Publish ( args . ToEvent ( ) )
return
}
// globalNotificationSys is not initialized in gateway mode.
if globalNotificationSys == nil {
return
}
if globalHTTPListen . HasSubscribers ( ) {
globalHTTPListen . Publish ( args . ToEvent ( ) )
}
notifyCh := globalNotificationSys . Send ( args )
go func ( ) {
for _ , err := range notifyCh {
@ -1345,92 +1245,3 @@ func saveNotificationConfig(ctx context.Context, objAPI ObjectLayer, bucketName
configFile := path . Join ( bucketConfigPrefix , bucketName , bucketNotificationConfig )
return saveConfig ( ctx , objAPI , configFile , data )
}
// SaveListener - saves HTTP client currently listening for events to listener.json.
func SaveListener ( objAPI ObjectLayer , bucketName string , eventNames [ ] event . Name , pattern string , targetID event . TargetID , addr xnet . Host ) error {
// listener.json is available/applicable only in DistXL mode.
if ! globalIsDistXL {
return nil
}
ctx := logger . SetReqInfo ( context . Background ( ) , & logger . ReqInfo { BucketName : bucketName } )
// Construct path to listener.json for the given bucket.
configFile := path . Join ( bucketConfigPrefix , bucketName , bucketListenerConfig )
configData , err := readConfig ( ctx , objAPI , configFile )
if err != nil && ! IsErrIgnored ( err , errDiskNotFound , errConfigNotFound ) {
return err
}
listenerList := [ ] ListenBucketNotificationArgs { }
if configData != nil {
if err = json . Unmarshal ( configData , & listenerList ) ; err != nil {
logger . LogIf ( ctx , err )
return err
}
}
listenerList = append ( listenerList , ListenBucketNotificationArgs {
EventNames : eventNames ,
Pattern : pattern ,
TargetID : targetID ,
Addr : addr ,
} )
data , err := json . Marshal ( listenerList )
if err != nil {
logger . LogIf ( ctx , err )
return err
}
return saveConfig ( ctx , objAPI , configFile , data )
}
// RemoveListener - removes HTTP client currently listening for events from listener.json.
func RemoveListener ( objAPI ObjectLayer , bucketName string , targetID event . TargetID , addr xnet . Host ) error {
// listener.json is available/applicable only in DistXL mode.
if ! globalIsDistXL {
return nil
}
ctx := logger . SetReqInfo ( context . Background ( ) , & logger . ReqInfo { BucketName : bucketName } )
// Construct path to listener.json for the given bucket.
configFile := path . Join ( bucketConfigPrefix , bucketName , bucketListenerConfig )
configData , err := readConfig ( ctx , objAPI , configFile )
if err != nil && ! IsErrIgnored ( err , errDiskNotFound , errConfigNotFound ) {
return err
}
listenerList := [ ] ListenBucketNotificationArgs { }
if configData != nil {
if err = json . Unmarshal ( configData , & listenerList ) ; err != nil {
logger . LogIf ( ctx , err )
return err
}
}
if len ( listenerList ) == 0 {
// Nothing to remove.
return nil
}
activeListenerList := [ ] ListenBucketNotificationArgs { }
for _ , args := range listenerList {
if args . TargetID == targetID && args . Addr . Equal ( addr ) {
// Skip if matches
continue
}
activeListenerList = append ( activeListenerList , args )
}
data , err := json . Marshal ( activeListenerList )
if err != nil {
logger . LogIf ( ctx , err )
return err
}
return saveConfig ( ctx , objAPI , configFile , data )
}
Harshavardhana
5 years ago
committed by