MyFavMirrors/minio
1
0
Fork 0
mirror of https://github.com/minio/minio.git synced 2025-04-20 10:37:31 -04:00
Code Issues Packages Projects Releases Wiki Activity

do not panic if webhook returns an error (#15970)

Browse Source
This commit is contained in:
Harshavardhana 2022-10-30 16:45:53 -07:00 committed by GitHub
parent 0303cd8625
commit 4892a766a8
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
1 changed files with 9 additions and 6 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

15
internal/event/target/webhook.go
View File

@ -98,7 +98,8 @@ type WebhookTarget struct {
httpClient *http.Client httpClient *http.Client
store Store store Store
loggerOnce logger.LogOnce loggerOnce logger.LogOnce
quitCh chan struct{} cancel context.CancelFunc
cancelCh <-chan struct{}
} }
// ID - returns target ID. // ID - returns target ID.
@ -199,14 +200,12 @@ func (target *WebhookTarget) send(eventData event.Event) error {
resp, err := target.httpClient.Do(req) resp, err := target.httpClient.Do(req)
if err != nil { if err != nil {
target.Close()
return err return err
} }
defer resp.Body.Close() defer resp.Body.Close()
io.Copy(io.Discard, resp.Body) io.Copy(io.Discard, resp.Body)
if resp.StatusCode < 200 || resp.StatusCode > 299 { if resp.StatusCode < 200 || resp.StatusCode > 299 {
target.Close()
return fmt.Errorf("sending event failed with %v", resp.Status) return fmt.Errorf("sending event failed with %v", resp.Status)
} }
@ -242,7 +241,7 @@ func (target *WebhookTarget) Send(eventKey string) error {
// Close - does nothing and available for interface compatibility. // Close - does nothing and available for interface compatibility.
func (target *WebhookTarget) Close() error { func (target *WebhookTarget) Close() error {
close(target.quitCh) target.cancel()
return nil return nil
} }
@ -274,18 +273,21 @@ func (target *WebhookTarget) initWebhook() error {
} }
if target.store != nil { if target.store != nil {
streamEventsFromStore(target.store, target, target.quitCh, target.loggerOnce) streamEventsFromStore(target.store, target, target.cancelCh, target.loggerOnce)
} }
return nil return nil
} }
// NewWebhookTarget - creates new Webhook target. // NewWebhookTarget - creates new Webhook target.
func NewWebhookTarget(ctx context.Context, id string, args WebhookArgs, loggerOnce logger.LogOnce, transport *http.Transport) (*WebhookTarget, error) { func NewWebhookTarget(ctx context.Context, id string, args WebhookArgs, loggerOnce logger.LogOnce, transport *http.Transport) (*WebhookTarget, error) {
ctx, cancel := context.WithCancel(ctx)
var store Store var store Store
if args.QueueDir != "" { if args.QueueDir != "" {
queueDir := filepath.Join(args.QueueDir, storePrefix+"-webhook-"+id) queueDir := filepath.Join(args.QueueDir, storePrefix+"-webhook-"+id)
store = NewQueueStore(queueDir, args.QueueLimit) store = NewQueueStore(queueDir, args.QueueLimit)
if err := store.Open(); err != nil { if err := store.Open(); err != nil {
cancel()
return nil, fmt.Errorf("unable to initialize the queue store of Webhook `%s`: %w", id, err) return nil, fmt.Errorf("unable to initialize the queue store of Webhook `%s`: %w", id, err)
} }
} }
@ -296,6 +298,7 @@ func NewWebhookTarget(ctx context.Context, id string, args WebhookArgs, loggerOn
loggerOnce: loggerOnce, loggerOnce: loggerOnce,
transport: transport, transport: transport,
store: store, store: store,
quitCh: make(chan struct{}), cancel: cancel,
cancelCh: ctx.Done(),
}, nil }, nil
} }