Enable event persistence in webhook (#7614)

pkg/event/target/store.go

@@ -19,7 +19,10 @@ package target
 import (
 	"errors"
 	"fmt"
+	"net"
+	"os"
 	"strings"
+	"syscall"
 	"time"
 
 	"github.com/minio/minio/pkg/event"
@@ -76,6 +79,18 @@ func replayEvents(store Store, doneCh <-chan struct{}) <-chan string {
 	return eventKeyCh
 }
 
+// isConnResetErr - Checks for connection reset errors.
+func isConnResetErr(err error) bool {
+	if opErr, ok := err.(*net.OpError); ok {
+		if syscallErr, ok := opErr.Err.(*os.SyscallError); ok {
+			if syscallErr.Err == syscall.ECONNRESET {
+				return true
+			}
+		}
+	}
+	return false
+}
+
 // sendEvents - Reads events from the store and re-plays.
 func sendEvents(target event.Target, eventKeyCh <-chan string, doneCh <-chan struct{}) {
 	retryTimer := time.NewTimer(retryInterval)
@@ -88,7 +103,7 @@ func sendEvents(target event.Target, eventKeyCh <-chan string, doneCh <-chan str
 				break
 			}
 
-			if err != errNotConnected {
+			if err != errNotConnected && !isConnResetErr(err) {
 				panic(fmt.Errorf("target.Send() failed with '%v'", err))
 			}
 

pkg/event/target/webhook.go

@@ -28,6 +28,9 @@ import (
 	"net"
 	"net/http"
 	"net/url"
+	"os"
+	"path/filepath"
+	"syscall"
 	"time"
 
 	"github.com/minio/minio/pkg/event"
@@ -36,9 +39,11 @@ import (
 
 // WebhookArgs - Webhook target arguments.
 type WebhookArgs struct {
-	Enable   bool           `json:"enable"`
-	Endpoint xnet.URL       `json:"endpoint"`
-	RootCAs  *x509.CertPool `json:"-"`
+	Enable     bool           `json:"enable"`
+	Endpoint   xnet.URL       `json:"endpoint"`
+	RootCAs    *x509.CertPool `json:"-"`
+	QueueDir   string         `json:"queueDir"`
+	QueueLimit uint16         `json:"queueLimit"`
 }
 
 // Validate WebhookArgs fields
@@ -49,6 +54,14 @@ func (w WebhookArgs) Validate() error {
 	if w.Endpoint.IsEmpty() {
 		return errors.New("endpoint empty")
 	}
+	if w.QueueDir != "" {
+		if !filepath.IsAbs(w.QueueDir) {
+			return errors.New("queueDir path should be absolute")
+		}
+	}
+	if w.QueueLimit > maxLimit {
+		return errors.New("queueLimit should not exceed 10000")
+	}
 	return nil
 }
 
@@ -57,6 +70,7 @@ type WebhookTarget struct {
 	id         event.TargetID
 	args       WebhookArgs
 	httpClient *http.Client
+	store      Store
 }
 
 // ID - returns target ID.
@@ -64,11 +78,27 @@ func (target WebhookTarget) ID() event.TargetID {
 	return target.id
 }
 
-// Save - Sends event directly without persisting.
+// Save - saves the events to the store if queuestore is configured, which will be replayed when the wenhook connection is active.
 func (target *WebhookTarget) Save(eventData event.Event) error {
+	if target.store != nil {
+		return target.store.Put(eventData)
+	}
+	urlStr, pErr := xnet.ParseURL(target.args.Endpoint.String())
+	if pErr != nil {
+		return pErr
+	}
+	_, dErr := net.Dial("tcp", urlStr.Host)
+	if dErr != nil {
+		// To treat "connection refused" errors as errNotConnected.
+		if IsConnRefusedErr(dErr) {
+			return errNotConnected
+		}
+		return dErr
+	}
 	return target.send(eventData)
 }
 
+// send - sends an event to the webhook.
 func (target *WebhookTarget) send(eventData event.Event) error {
 	objectName, err := url.QueryUnescape(eventData.S3.Object.Key)
 	if err != nil {
@@ -104,9 +134,52 @@ func (target *WebhookTarget) send(eventData event.Event) error {
 	return nil
 }
 
-// Send - interface compatible method does no-op.
+// IsConnRefusedErr - To check for "connection refused" errors.
+func IsConnRefusedErr(err error) bool {
+	if opErr, ok := err.(*net.OpError); ok {
+		if sysErr, ok := opErr.Err.(*os.SyscallError); ok {
+			if errno, ok := sysErr.Err.(syscall.Errno); ok {
+				if errno == syscall.ECONNREFUSED {
+					return true
+				}
+			}
+		}
+	}
+	return false
+}
+
+// Send - reads an event from store and sends it to webhook.
 func (target *WebhookTarget) Send(eventKey string) error {
-	return nil
+
+	urlStr, pErr := xnet.ParseURL(target.args.Endpoint.String())
+	if pErr != nil {
+		return pErr
+	}
+	_, dErr := net.Dial("tcp", urlStr.Host)
+	if dErr != nil {
+		// To treat "connection refused" errors as errNotConnected.
+		if IsConnRefusedErr(dErr) {
+			return errNotConnected
+		}
+		return dErr
+	}
+
+	eventData, eErr := target.store.Get(eventKey)
+	if eErr != nil {
+		// The last event key in a successful batch will be sent in the channel atmost once by the replayEvents()
+		// Such events will not exist and would've been already been sent successfully.
+		if os.IsNotExist(eErr) {
+			return nil
+		}
+		return eErr
+	}
+
+	if err := target.send(eventData); err != nil {
+		return err
+	}
+
+	// Delete the event from store.
+	return target.store.Del(eventKey)
 }
 
 // Close - does nothing and available for interface compatibility.
@@ -115,8 +188,19 @@ func (target *WebhookTarget) Close() error {
 }
 
 // NewWebhookTarget - creates new Webhook target.
-func NewWebhookTarget(id string, args WebhookArgs) *WebhookTarget {
-	return &WebhookTarget{
+func NewWebhookTarget(id string, args WebhookArgs, doneCh <-chan struct{}) *WebhookTarget {
+
+	var store Store
+
+	if args.QueueDir != "" {
+		queueDir := filepath.Join(args.QueueDir, storePrefix+"-webhook-"+id)
+		store = NewQueueStore(queueDir, args.QueueLimit)
+		if oErr := store.Open(); oErr != nil {
+			return nil
+		}
+	}
+
+	target := &WebhookTarget{
 		id:   event.TargetID{ID: id, Name: "webhook"},
 		args: args,
 		httpClient: &http.Client{
@@ -131,5 +215,15 @@ func NewWebhookTarget(id string, args WebhookArgs) *WebhookTarget {
 				ExpectContinueTimeout: 2 * time.Second,
 			},
 		},
+		store: store,
 	}
+
+	if target.store != nil {
+		// Replays the events from the store.
+		eventKeyCh := replayEvents(target.store, doneCh)
+		// Start replaying events from the store.
+		go sendEvents(target, eventKeyCh, doneCh)
+	}
+
+	return target
 }