Merge pull request #1353 from kaleido-io/expose_webhook_options

feat: expose retry and HTTP options for webhooks

Author: EnriqueL8

docs/reference/types/subscription.md

@@ -116,6 +116,8 @@ nav_order: 3
 | `query` | Webhooks only: Static query params to set on the webhook request | `` |
 | `tlsConfigName` | The name of an existing TLS configuration associated to the namespace to use | `string` |
 | `input` | Webhooks only: A set of options to extract data from the first JSON input data in the incoming message. Only applies if withData=true | [`WebhookInputOptions`](#webhookinputoptions) |
+| `retry` | Webhooks only: a set of options for retrying the webhook call | [`WebhookRetryOptions`](#webhookretryoptions) |
+| `httpOptions` | Webhooks only: a set of options for HTTP | [`WebhookHTTPOptions`](#webhookhttpoptions) |
 
 ## WebhookInputOptions
 
@@ -128,4 +130,26 @@ nav_order: 3
 | `replytx` | A top-level property of the first data input, to use to dynamically set whether to pin the response (so the requester can choose) | `string` |
 
 
+## WebhookRetryOptions
+
+| Field Name | Description | Type |
+|------------|-------------|------|
+| `enabled` | Enables retry on HTTP calls, defaults to false | `bool` |
+| `count` | Number of times to retry the webhook call in case of failure | `int` |
+| `initialDelay` | Initial delay between retries when we retry the webhook call | `string` |
+| `maxDelay` | Max delay between retries when we retry the webhookcall | `string` |
+
+
+## WebhookHTTPOptions
+
+| Field Name | Description | Type |
+|------------|-------------|------|
+| `tlsHandshakeTimeout` | The max duration to hold a TLS handshake alive | `string` |
+| `requestTimeout` | The max duration to hold a TLS handshake alive | `string` |
+| `maxIdleConns` | The max number of idle connections to hold pooled | `int` |
+| `idleTimeout` | The max duration to hold a HTTP keepalive connection between calls | `string` |
+| `connectionTimeout` |  | `string` |
+| `expectContinueTimeout` | See [ExpectContinueTimeout in the Go docs](https://pkg.go.dev/net/http#Transport) | `string` |
+
+
 

docs/reference/types/wsstart.md

@@ -107,6 +107,8 @@ nav_order: 23
 | `query` | Webhooks only: Static query params to set on the webhook request | `` |
 | `tlsConfigName` | The name of an existing TLS configuration associated to the namespace to use | `string` |
 | `input` | Webhooks only: A set of options to extract data from the first JSON input data in the incoming message. Only applies if withData=true | [`WebhookInputOptions`](#webhookinputoptions) |
+| `retry` | Webhooks only: a set of options for retrying the webhook call | [`WebhookRetryOptions`](#webhookretryoptions) |
+| `httpOptions` | Webhooks only: a set of options for HTTP | [`WebhookHTTPOptions`](#webhookhttpoptions) |
 
 ## WebhookInputOptions
 
@@ -119,4 +121,26 @@ nav_order: 23
 | `replytx` | A top-level property of the first data input, to use to dynamically set whether to pin the response (so the requester can choose) | `string` |
 
 
+## WebhookRetryOptions
+
+| Field Name | Description | Type |
+|------------|-------------|------|
+| `enabled` | Enables retry on HTTP calls, defaults to false | `bool` |
+| `count` | Number of times to retry the webhook call in case of failure | `int` |
+| `initialDelay` | Initial delay between retries when we retry the webhook call | `string` |
+| `maxDelay` | Max delay between retries when we retry the webhookcall | `string` |
+
+
+## WebhookHTTPOptions
+
+| Field Name | Description | Type |
+|------------|-------------|------|
+| `tlsHandshakeTimeout` | The max duration to hold a TLS handshake alive | `string` |
+| `requestTimeout` | The max duration to hold a TLS handshake alive | `string` |
+| `maxIdleConns` | The max number of idle connections to hold pooled | `int` |
+| `idleTimeout` | The max duration to hold a HTTP keepalive connection between calls | `string` |
+| `connectionTimeout` |  | `string` |
+| `expectContinueTimeout` | See [ExpectContinueTimeout in the Go docs](https://pkg.go.dev/net/http#Transport) | `string` |
+
+
 

docs/swagger/swagger.yaml

@@ -681,22 +681,34 @@ var (
 	WSSubscriptionStatusFilter    = ffm("WSSubscriptionStatus.filter", "The subscription filter specification")
 	WSSubscriptionStatusStartTime = ffm("WSSubscriptionStatus.startTime", "The time the subscription started (reset on dynamic namespace reload)")
 
-	WebhooksOptJSON          = ffm("WebhookSubOptions.json", "Webhooks only: Whether to assume the response body is JSON, regardless of the returned Content-Type")
-	WebhooksOptReply         = ffm("WebhookSubOptions.reply", "Webhooks only: Whether to automatically send a reply event, using the body returned by the webhook")
-	WebhooksOptHeaders       = ffm("WebhookSubOptions.headers", "Webhooks only: Static headers to set on the webhook request")
-	WebhooksOptQuery         = ffm("WebhookSubOptions.query", "Webhooks only: Static query params to set on the webhook request")
-	WebhooksOptInput         = ffm("WebhookSubOptions.input", "Webhooks only: A set of options to extract data from the first JSON input data in the incoming message. Only applies if withData=true")
-	WebhooksOptFastAck       = ffm("WebhookSubOptions.fastack", "Webhooks only: When true the event will be acknowledged before the webhook is invoked, allowing parallel invocations")
-	WebhooksOptURL           = ffm("WebhookSubOptions.url", "Webhooks only: HTTP url to invoke. Can be relative if a base URL is set in the webhook plugin config")
-	WebhooksOptMethod        = ffm("WebhookSubOptions.method", "Webhooks only: HTTP method to invoke. Default=POST")
-	WebhooksOptReplyTag      = ffm("WebhookSubOptions.replytag", "Webhooks only: The tag to set on the reply message")
-	WebhooksOptReplyTx       = ffm("WebhookSubOptions.replytx", "Webhooks only: The transaction type to set on the reply message")
-	WebhooksOptTLSConfigName = ffm("WebhookSubOptions.tlsConfigName", "The name of an existing TLS configuration associated to the namespace to use")
-	WebhooksOptInputQuery    = ffm("WebhookInputOptions.query", "A top-level property of the first data input, to use for query parameters")
-	WebhooksOptInputHeaders  = ffm("WebhookInputOptions.headers", "A top-level property of the first data input, to use for headers")
-	WebhooksOptInputBody     = ffm("WebhookInputOptions.body", "A top-level property of the first data input, to use for the request body. Default is the whole first body")
-	WebhooksOptInputPath     = ffm("WebhookInputOptions.path", "A top-level property of the first data input, to use for a path to append with escaping to the webhook path")
-	WebhooksOptInputReplyTx  = ffm("WebhookInputOptions.replytx", "A top-level property of the first data input, to use to dynamically set whether to pin the response (so the requester can choose)")
+	WebhooksOptJSON                     = ffm("WebhookSubOptions.json", "Webhooks only: Whether to assume the response body is JSON, regardless of the returned Content-Type")
+	WebhooksOptReply                    = ffm("WebhookSubOptions.reply", "Webhooks only: Whether to automatically send a reply event, using the body returned by the webhook")
+	WebhooksOptHeaders                  = ffm("WebhookSubOptions.headers", "Webhooks only: Static headers to set on the webhook request")
+	WebhooksOptQuery                    = ffm("WebhookSubOptions.query", "Webhooks only: Static query params to set on the webhook request")
+	WebhooksOptInput                    = ffm("WebhookSubOptions.input", "Webhooks only: A set of options to extract data from the first JSON input data in the incoming message. Only applies if withData=true")
+	WebhooksOptFastAck                  = ffm("WebhookSubOptions.fastack", "Webhooks only: When true the event will be acknowledged before the webhook is invoked, allowing parallel invocations")
+	WebhooksOptURL                      = ffm("WebhookSubOptions.url", "Webhooks only: HTTP url to invoke. Can be relative if a base URL is set in the webhook plugin config")
+	WebhooksOptMethod                   = ffm("WebhookSubOptions.method", "Webhooks only: HTTP method to invoke. Default=POST")
+	WebhooksOptReplyTag                 = ffm("WebhookSubOptions.replytag", "Webhooks only: The tag to set on the reply message")
+	WebhooksOptReplyTx                  = ffm("WebhookSubOptions.replytx", "Webhooks only: The transaction type to set on the reply message")
+	WebhooksOptTLSConfigName            = ffm("WebhookSubOptions.tlsConfigName", "The name of an existing TLS configuration associated to the namespace to use")
+	WebhooksOptHTTPOptions              = ffm("WebhookSubOptions.httpOptions", "Webhooks only: a set of options for HTTP")
+	WebhooksOptHTTPRetry                = ffm("WebhookSubOptions.retry", "Webhooks only: a set of options for retrying the webhook call")
+	WebhooksOptInputQuery               = ffm("WebhookInputOptions.query", "A top-level property of the first data input, to use for query parameters")
+	WebhooksOptInputHeaders             = ffm("WebhookInputOptions.headers", "A top-level property of the first data input, to use for headers")
+	WebhooksOptInputBody                = ffm("WebhookInputOptions.body", "A top-level property of the first data input, to use for the request body. Default is the whole first body")
+	WebhooksOptInputPath                = ffm("WebhookInputOptions.path", "A top-level property of the first data input, to use for a path to append with escaping to the webhook path")
+	WebhooksOptInputReplyTx             = ffm("WebhookInputOptions.replytx", "A top-level property of the first data input, to use to dynamically set whether to pin the response (so the requester can choose)")
+	WebhooksOptRetryEnabled             = ffm("WebhookRetryOptions.enabled", "Enables retry on HTTP calls, defaults to false")
+	WebhooksOptRetryCount               = ffm("WebhookRetryOptions.count", "Number of times to retry the webhook call in case of failure")
+	WebhooksOptRetryInitialDelay        = ffm("WebhookRetryOptions.initialDelay", "Initial delay between retries when we retry the webhook call")
+	WebhooksOptRetryMaxDelay            = ffm("WebhookRetryOptions.maxDelay", "Max delay between retries when we retry the webhookcall")
+	WebhookOptHTTPExpectContinueTimeout = ffm("WebhookHTTPOptions.expectContinueTimeout", "See [ExpectContinueTimeout in the Go docs](https://pkg.go.dev/net/http#Transport)")
+	WebhookOptHTTPIdleTimeout           = ffm("WebhookHTTPOptions.idleTimeout", "The max duration to hold a HTTP keepalive connection between calls")
+	WebhookOptHTTPMaxIdleConns          = ffm("WebhookHTTPOptions.maxIdleConns", "The max number of idle connections to hold pooled")
+	WebhookOptHTTPConnectionTimeout     = ffm("WebhookHTTPOptions.connectionTimeout", "")
+	WebhookOptHTTPTLSHandshakeTimeout   = ffm("WebhookHTTPOptions.tlsHandshakeTimeout", "The max duration to hold a TLS handshake alive")
+	WebhookOptHTTPRequestTimeout        = ffm("WebhookHTTPOptions.requestTimeout", "The max duration to hold a TLS handshake alive")
 
 	// PublishInput field descriptions
 	PublishInputIdempotencyKey = ffm("PublishInput.idempotencyKey", "An optional identifier to allow idempotent submission of requests. Stored on the transaction uniquely within a namespace")

internal/coremsgs/en_struct_descriptions.go

@@ -1,4 +1,4 @@
-// Copyright © 2022 Kaleido, Inc.
+// Copyright © 2023 Kaleido, Inc.
 //
 // SPDX-License-Identifier: Apache-2.0
 //
@@ -379,7 +379,7 @@ func (ed *eventDispatcher) deliverEvents() {
 				data, _, err = ed.data.GetMessageDataCached(ed.ctx, event.Message)
 			}
 			if err == nil {
-				err = ed.transport.DeliveryRequest(ed.connID, ed.subscription.definition, event, data)
+				err = ed.transport.DeliveryRequest(ed.ctx, ed.connID, ed.subscription.definition, event, data)
 			}
 			if err != nil {
 				ed.deliveryResponse(&core.EventDeliveryResponse{ID: event.ID, Rejected: true})

internal/events/event_dispatcher.go

@@ -177,9 +177,9 @@ func TestEventDispatcherReadAheadOutOfOrderAcks(t *testing.T) {
 	mdm := ed.data.(*datamocks.Manager)
 
 	eventDeliveries := make(chan *core.EventDelivery)
-	deliveryRequestMock := mei.On("DeliveryRequest", mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(nil)
+	deliveryRequestMock := mei.On("DeliveryRequest", ed.ctx, mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(nil)
 	deliveryRequestMock.RunFn = func(a mock.Arguments) {
-		eventDeliveries <- a.Get(2).(*core.EventDelivery)
+		eventDeliveries <- a.Get(3).(*core.EventDelivery)
 	}
 
 	// Setup the IDs
@@ -272,9 +272,9 @@ func TestEventDispatcherNoReadAheadInOrder(t *testing.T) {
 	mei := ed.transport.(*eventsmocks.Plugin)
 
 	eventDeliveries := make(chan *core.EventDelivery)
-	deliveryRequestMock := mei.On("DeliveryRequest", mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(nil)
+	deliveryRequestMock := mei.On("DeliveryRequest", mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(nil)
 	deliveryRequestMock.RunFn = func(a mock.Arguments) {
-		eventDeliveries <- a.Get(2).(*core.EventDelivery)
+		eventDeliveries <- a.Get(3).(*core.EventDelivery)
 	}
 
 	// Setup the IDs
@@ -603,9 +603,9 @@ func TestEnrichTransactionEvents(t *testing.T) {
 	mei := ed.transport.(*eventsmocks.Plugin)
 
 	eventDeliveries := make(chan *core.EventDelivery)
-	deliveryRequestMock := mei.On("DeliveryRequest", mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(nil)
+	deliveryRequestMock := mei.On("DeliveryRequest", mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(nil)
 	deliveryRequestMock.RunFn = func(a mock.Arguments) {
-		eventDeliveries <- a.Get(2).(*core.EventDelivery)
+		eventDeliveries <- a.Get(3).(*core.EventDelivery)
 	}
 
 	// Setup the IDs
@@ -692,9 +692,9 @@ func TestEnrichBlockchainEventEvents(t *testing.T) {
 	mei := ed.transport.(*eventsmocks.Plugin)
 
 	eventDeliveries := make(chan *core.EventDelivery)
-	deliveryRequestMock := mei.On("DeliveryRequest", mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(nil)
+	deliveryRequestMock := mei.On("DeliveryRequest", mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(nil)
 	deliveryRequestMock.RunFn = func(a mock.Arguments) {
-		eventDeliveries <- a.Get(2).(*core.EventDelivery)
+		eventDeliveries <- a.Get(3).(*core.EventDelivery)
 	}
 
 	// Setup the IDs
@@ -814,7 +814,7 @@ func TestBufferedDeliveryClosedContext(t *testing.T) {
 	mdi := ed.database.(*databasemocks.Plugin)
 	mei := ed.transport.(*eventsmocks.Plugin)
 	mdi.On("GetDataRefs", mock.Anything, mock.Anything).Return(nil, nil, nil)
-	mei.On("DeliveryRequest", mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(nil)
+	mei.On("DeliveryRequest", mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(nil)
 
 	repoll, err := ed.bufferedDelivery([]core.LocallySequenced{&core.Event{ID: fftypes.NewUUID()}})
 	assert.False(t, repoll)
@@ -837,7 +837,7 @@ func TestBufferedDeliveryNackRewind(t *testing.T) {
 	mdi.On("UpdateOffset", mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(nil)
 
 	delivered := make(chan struct{})
-	deliver := mei.On("DeliveryRequest", mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(nil)
+	deliver := mei.On("DeliveryRequest", mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(nil)
 	deliver.RunFn = func(a mock.Arguments) {
 		close(delivered)
 	}
@@ -882,7 +882,7 @@ func TestBufferedDeliveryFailNack(t *testing.T) {
 	mdi.On("UpdateOffset", mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(fmt.Errorf("pop"))
 
 	failNacked := make(chan bool)
-	deliver := mei.On("DeliveryRequest", mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(fmt.Errorf("pop"))
+	deliver := mei.On("DeliveryRequest", mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(fmt.Errorf("pop"))
 	deliver.RunFn = func(a mock.Arguments) {
 		failNacked <- true
 	}

internal/events/event_dispatcher_test.go

@@ -146,7 +146,7 @@ func newTestEventManagerCommon(t *testing.T, metrics, dbconcurrency bool) *testE
 	mbi.On("VerifierType").Return(core.VerifierTypeEthAddress).Maybe()
 	mdi.On("Capabilities").Return(&database.Capabilities{Concurrency: dbconcurrency}).Maybe()
 	mev.On("SetHandler", "ns1", mock.Anything).Return(nil).Maybe()
-	mev.On("ValidateOptions", mock.Anything).Return(nil).Maybe()
+	mev.On("ValidateOptions", mock.Anything, mock.Anything).Return(nil).Maybe()
 	ns := &core.Namespace{Name: "ns1", NetworkName: "ns1"}
 	emi, err := NewEventManager(ctx, ns, mdi, mbi, mim, msh, mdm, mds, mbm, mpm, mam, msd, mmi, mom, txHelper, events, mmp, cmi)
 	em := emi.(*eventManager)

internal/events/event_manager_test.go

@@ -258,7 +258,11 @@ func (sm *subscriptionManager) parseSubscriptionDef(ctx context.Context, subDef
 		return nil, i18n.NewError(ctx, coremsgs.MsgUnknownEventTransportPlugin, subDef.Transport)
 	}
 
-	if err := transport.ValidateOptions(&subDef.Options); err != nil {
+	if subDef.Options.TLSConfigName != "" && sm.namespace.TLSConfigs[subDef.Options.TLSConfigName] != nil {
+		subDef.Options.TLSConfig = sm.namespace.TLSConfigs[subDef.Options.TLSConfigName]
+	}
+
+	if err := transport.ValidateOptions(ctx, &subDef.Options); err != nil {
 		return nil, err
 	}
 
@@ -337,10 +341,6 @@ func (sm *subscriptionManager) parseSubscriptionDef(ctx context.Context, subDef
 		}
 	}
 
-	if subDef.Options.TLSConfigName != "" && sm.namespace.TLSConfigs[subDef.Options.TLSConfigName] != nil {
-		subDef.Options.TLSConfig = sm.namespace.TLSConfigs[subDef.Options.TLSConfigName]
-	}
-
 	sub = &subscription{
 		dispatcherElection: make(chan bool, 1),
 		definition:         subDef,