Documentation
¶
Index ¶
- Constants
- Variables
- func ClassifyProxyConnectResponse(status int, header http.Header, underlying error, target string) error
- func EnvoyDiagnostics(flag, details string) map[string]string
- func EnvoyResponseDetails(h http.Header) string
- func EnvoyResponseFlag(h http.Header) string
- func IsProxyError(err error) bool
- func IsProxyInfraError(err error) bool
- func MakePublisherKey(dest *models.Destination) string
- func MapEnvoyResponseFlag(flag string) string
- func NewErrDestinationPublishAttempt(err error, provider string, data map[string]interface{}) error
- func NewErrDestinationValidation(errors []ValidationErrorDetail) error
- func NewErrPublishCanceled(provider string) error
- func NewHTTPClient(config HTTPClientConfig) (*http.Client, error)
- func ObfuscateValue(value string) string
- func ProxyDialFunc(hops []*url.URL) proxychain.DialFunc
- type BaseProvider
- func (p *BaseProvider) Metadata() *metadata.ProviderMetadata
- func (p *BaseProvider) NewPublisher(additionalOpts ...BasePublisherOption) *BasePublisher
- func (p *BaseProvider) ObfuscateDestination(destination *models.Destination) *models.Destination
- func (p *BaseProvider) Preprocess(newDestination *models.Destination, originalDestination *models.Destination, ...) error
- func (p *BaseProvider) Validate(ctx context.Context, destination *models.Destination) error
- type BasePublisher
- type BasePublisherOption
- type Config
- type Delivery
- type DestinationDisplay
- type DestinationTarget
- type ErrDestinationPublishAttempt
- type ErrDestinationValidation
- type ErrProxyDestination
- type ErrProxyInfra
- type HTTPClientConfig
- type PoolSizing
- type PreprocessDestinationOpts
- type PreprocessRequest
- type Provider
- type Publisher
- type Registry
- type ValidationErrorDetail
Constants ¶
const ( // IdleConnsPerConcurrency scales the total idle pool with the delivery // worker count: at ~3s per delivery a worker revisits roughly 32 distinct // destinations within the 90s IdleConnTimeout, and slow destinations are // where reuse matters most. IdleConnsPerConcurrency = 32 // MinTotalIdleConns floors the total for low-concurrency fanout. // Concurrency bounds simultaneous requests, not distinct hosts touched // over time — a single worker at ~100ms per delivery still cycles through // ~900 destinations per idle window. MinTotalIdleConns = 512 // MaxTotalIdleConns caps the parked-FD/memory cost where the reuse hit // rate decays. The cap never binds below the concurrency level itself — // see SizeFanOutPool. MaxTotalIdleConns = 4096 // MinIdleConnsPerHost is the floor for per-host depth, matching Go's // default. DELIVERY_MAX_CONCURRENCY defaults to 1, which would otherwise // size us below stock behavior. MinIdleConnsPerHost = 2 )
Variables ¶
var ErrPublisherClosed = errors.New("publisher is closed")
Functions ¶
func ClassifyProxyConnectResponse ¶ added in v1.6.0
func ClassifyProxyConnectResponse(status int, header http.Header, underlying error, target string) error
ClassifyProxyConnectResponse maps a non-200 CONNECT response from a proxy hop into ErrProxyInfra or ErrProxyDestination. target is the host:port the CONNECT was for.
func EnvoyDiagnostics ¶ added in v1.6.0
EnvoyDiagnostics returns the diagnostics map for an Envoy-attributed destination error. Returns nil when both inputs are empty so the caller gets a properly-zero Diagnostics field (len-0 nil map).
func EnvoyResponseDetails ¶ added in v1.6.0
EnvoyResponseDetails returns the meaningful value of the x-envoy-response-code-details header (stage{reason} form when both are present, stage-only otherwise), or "" if the header is absent / empty / placeholder "-". Captured for operator-side diagnostics; never inspected for classification.
func EnvoyResponseFlag ¶ added in v1.6.0
EnvoyResponseFlag returns the meaningful value of the x-envoy-response-flags header, or "" if the header is absent / placeholder "-" / empty.
func IsProxyError ¶ added in v1.6.0
IsProxyError reports whether err carries a proxy failure from a ProxyDialFunc connection.
func IsProxyInfraError ¶ added in v1.6.0
IsProxyInfraError reports whether err is or wraps an ErrProxyInfra.
func MakePublisherKey ¶
func MakePublisherKey(dest *models.Destination) string
MakePublisherKey creates a unique key for a destination that includes type, config, credentials and delivery metadata. Any of these can be baked into the publisher when it is created, so a change in any of them must produce a new cache key.
func MapEnvoyResponseFlag ¶ added in v1.6.0
MapEnvoyResponseFlag returns the destination error code corresponding to an Envoy response flag. The output vocabulary matches the webhook provider's ClassifyNetworkError so customers see the same codes whether or not a proxy is in path. Unhandled flags fall through to "network_error" — operators should watch for that code paired with a non-empty flag in the attempt error payload as a signal that the mapping needs expansion.
Envoy response flag reference: https://www.envoyproxy.io/docs/envoy/latest/configuration/observability/access_log/usage#config-access-log-format-response-flags
func NewErrDestinationValidation ¶
func NewErrDestinationValidation(errors []ValidationErrorDetail) error
func NewErrPublishCanceled ¶ added in v0.11.0
NewErrPublishCanceled creates an error for when publish is canceled (e.g., service shutdown). This should return nil Delivery to trigger nack → requeue for another instance. See: https://github.com/hookdeck/outpost/issues/571
func NewHTTPClient ¶ added in v1.0.3
func NewHTTPClient(config HTTPClientConfig) (*http.Client, error)
NewHTTPClient builds an *http.Client from config. Free function — no provider state is involved.
func ObfuscateValue ¶
ObfuscateValue masks a sensitive value with the following rules: - For strings with length >= 10: show first 4 characters + asterisks for the rest - For strings with length < 10: replace each character with an asterisk
func ProxyDialFunc ¶ added in v1.6.0
func ProxyDialFunc(hops []*url.URL) proxychain.DialFunc
ProxyDialFunc returns a dial func that tunnels through hops, nearest first, or nil when hops is empty. A hop refusing the CONNECT is classified by ClassifyProxyConnectResponse. A CONNECT timing out passes through as a timeout, and cancellation passes through unchanged. Any other failure (a hop unreachable, TLS to a hop, a hop hanging up) is ErrProxyInfra.
Types ¶
type BaseProvider ¶
type BaseProvider struct {
// contains filtered or unexported fields
}
BaseProvider provides common functionality for all destination providers
func NewBaseProvider ¶
func NewBaseProvider(loader metadata.MetadataLoader, providerType string, opts ...BasePublisherOption) (*BaseProvider, error)
NewBaseProvider creates a new base provider with loaded metadata
func (*BaseProvider) Metadata ¶
func (p *BaseProvider) Metadata() *metadata.ProviderMetadata
Metadata returns the provider metadata
func (*BaseProvider) NewPublisher ¶ added in v0.6.1
func (p *BaseProvider) NewPublisher(additionalOpts ...BasePublisherOption) *BasePublisher
NewPublisher creates a BasePublisher with provider-configured options plus any additional options
func (*BaseProvider) ObfuscateDestination ¶
func (p *BaseProvider) ObfuscateDestination(destination *models.Destination) *models.Destination
ObfuscateDestination returns a copy of the destination with sensitive fields masked
func (*BaseProvider) Preprocess ¶
func (p *BaseProvider) Preprocess(newDestination *models.Destination, originalDestination *models.Destination, opts *PreprocessDestinationOpts) error
Preprocess is a noop by default
func (*BaseProvider) Validate ¶
func (p *BaseProvider) Validate(ctx context.Context, destination *models.Destination) error
Validate performs field-level validation using the provider's metadata
type BasePublisher ¶
type BasePublisher struct {
// contains filtered or unexported fields
}
BasePublisher provides common publisher functionality
func NewBasePublisher ¶ added in v0.6.1
func NewBasePublisher(opts ...BasePublisherOption) *BasePublisher
NewBasePublisher creates a new BasePublisher with the given options
func (*BasePublisher) FinishPublish ¶
func (p *BasePublisher) FinishPublish()
FinishPublish marks a publish operation as complete
func (*BasePublisher) MakeMetadata ¶
func (*BasePublisher) StartClose ¶
func (p *BasePublisher) StartClose()
StartClose marks publisher as closed and waits for active operations
func (*BasePublisher) StartPublish ¶
func (p *BasePublisher) StartPublish() error
StartPublish returns error if publisher is closed, otherwise adds to waitgroup
type BasePublisherOption ¶ added in v0.6.1
type BasePublisherOption func(*BasePublisher)
BasePublisherOption is a functional option for configuring BasePublisher
func WithDeliveryMetadata ¶ added in v0.8.0
func WithDeliveryMetadata(metadata map[string]string) BasePublisherOption
WithDeliveryMetadata sets static metadata to be merged with every event delivery
func WithMillisecondTimestamp ¶ added in v0.6.1
func WithMillisecondTimestamp(enabled bool) BasePublisherOption
WithMillisecondTimestamp enables millisecond-precision timestamp in metadata
type Delivery ¶
func NewFormatError ¶ added in v1.0.5
NewFormatError returns the (*Delivery, error) a publisher should return when formatting an event fails before it can be sent (e.g. an invalid key/partition template or an unparseable payload). It records a failed attempt so the failure is visible to the customer and the message is acked, instead of nacking into the DLQ.
message is the customer-facing string persisted on the attempt (ResponseData); when empty a generic default is used. The raw err is carried only in the returned error (for logs/telemetry) and is not persisted on the attempt.
func ProxyPublishResult ¶ added in v1.6.0
ProxyPublishResult is a publisher's result for a proxy failure (see IsProxyError): ErrProxyInfra returns a nil Delivery so the message is nacked, ErrProxyDestination a failed attempt with the proxy's code.
type DestinationDisplay ¶
type DestinationDisplay struct {
*models.Destination
DestinationTarget
}
DestinationDisplay represents a destination with display-specific fields
type DestinationTarget ¶
type ErrDestinationPublishAttempt ¶
func (*ErrDestinationPublishAttempt) Error ¶
func (e *ErrDestinationPublishAttempt) Error() string
type ErrDestinationValidation ¶
type ErrDestinationValidation struct {
Errors []ValidationErrorDetail `json:"errors"`
}
func (*ErrDestinationValidation) Error ¶
func (e *ErrDestinationValidation) Error() string
type ErrProxyDestination ¶ added in v1.6.0
type ErrProxyDestination struct {
Underlying error
Code string
DestHost string
Diagnostics map[string]string
}
ErrProxyDestination signals that the proxy reported a failure originating at the destination (e.g. upstream DNS lookup failed, upstream connection refused, upstream timeout). The delivery result is recorded as a normal failed attempt using Code as the classification, with response data rewritten so the customer sees a destination-attributed failure rather than proxy-attributed details.
Diagnostics is a free-form key/value map of proxy-specific signals the classification path picked up (e.g. for Envoy, "envoy_flag" and "envoy_details"). It is operator-side metadata only: surfaced in error logs and on the publish-attempt error payload, never written to the customer-visible attempt record. Whichever proxy emitted the data owns the key naming so heterogeneous proxies can coexist without colliding.
func (*ErrProxyDestination) Error ¶ added in v1.6.0
func (e *ErrProxyDestination) Error() string
func (*ErrProxyDestination) Unwrap ¶ added in v1.6.0
func (e *ErrProxyDestination) Unwrap() error
type ErrProxyInfra ¶ added in v1.6.0
ErrProxyInfra signals that a delivery failed at the proxy layer (proxy auth misconfiguration, proxy unreachable, etc.). The delivery result is nacked so the underlying message queue redelivers without recording a customer-visible failed attempt.
func (*ErrProxyInfra) Error ¶ added in v1.6.0
func (e *ErrProxyInfra) Error() string
func (*ErrProxyInfra) Unwrap ¶ added in v1.6.0
func (e *ErrProxyInfra) Unwrap() error
type HTTPClientConfig ¶
type HTTPClientConfig struct {
Timeout *time.Duration
UserAgent *string
// Proxy lists the forward proxies to go through, nearest first. Empty
// means a direct connection.
Proxy []*url.URL
// WrapTransport, if set, is invoked after a proxy has been installed on
// the *http.Transport. Callers can use it to attach proxy-specific
// concerns (e.g. OnProxyConnectResponse callbacks, response classifiers)
// without bleeding those concerns into destregistry itself. Receives the
// underlying transport plus the last hop, the one Go's transport talks
// to; returns the RoundTripper to use thereafter.
WrapTransport func(*http.Transport, *url.URL) http.RoundTripper
// Pool sizes the transport's idle connection pool. The zero value leaves
// Go's defaults in place (2 idle per host, 100 total), which is only
// appropriate for clients outside the delivery path. Use SizeFanOutPool
// or SizeSingleHostPool to derive it.
Pool PoolSizing
// OnConnection, if set, is invoked once per request with whether the
// underlying connection was reused. This is the signal that the pool
// ceiling is binding.
OnConnection func(reused bool)
}
type PoolSizing ¶ added in v1.2.0
type PoolSizing struct {
// MaxIdleConns bounds breadth — how many distinct destinations can hold a
// warm connection at all.
MaxIdleConns int
// MaxIdleConnsPerHost bounds depth — how many warm connections a single
// destination keeps.
MaxIdleConnsPerHost int
}
PoolSizing is the resolved connection pool configuration.
func SizeFanOutPool ¶ added in v1.2.0
func SizeFanOutPool(deliveryMaxConcurrency int) PoolSizing
SizeFanOutPool sizes a pool for a client that talks to arbitrarily many destination hosts (the webhook providers). Depth comes from the delivery worker pool — it caps how many deliveries can be in flight, so it caps how many connections one destination could need. Breadth scales with the same number: total = clamp(32×C, 512, max(4096, C)), the ceiling raised to C so the cap never undersizes the pool below the concurrency level.
deliveryMaxConcurrency <= 0 means "unknown"; the floors apply.
func SizeSingleHostPool ¶ added in v1.2.0
func SizeSingleHostPool(deliveryMaxConcurrency int) PoolSizing
SizeSingleHostPool sizes a pool for a client that talks to one host (the hookdeck provider). It needs depth, not breadth, so the total is the per-host value rather than a fan-out ceiling.
type PreprocessDestinationOpts ¶
type PreprocessDestinationOpts struct {
Role string
// Request holds the destination fields exactly as the caller sent them in
// the API request. On updates, newDestination carries the result of
// merge-patching the request into the stored values, so it cannot answer
// "did the caller provide this field" — the request can.
Request PreprocessRequest
}
PreprocessDestinationOpts contains options for preprocessing a destination
type PreprocessRequest ¶ added in v1.0.5
PreprocessRequest is the caller's view of the provider-owned destination fields, before any merging with stored state. Maps are nil when the request did not contain the corresponding field.
type Provider ¶
type Provider interface {
// Validate destination configuration using metadata
Validate(ctx context.Context, destination *models.Destination) error
// Create a new publisher instance
CreatePublisher(ctx context.Context, destination *models.Destination) (Publisher, error)
// Get provider metadata
Metadata() *metadata.ProviderMetadata
// ObfuscateDestination returns a copy of the destination with sensitive fields masked
ObfuscateDestination(destination *models.Destination) *models.Destination
// ComputeTarget returns a human-readable target string for the destination
ComputeTarget(destination *models.Destination) DestinationTarget
// Preprocess modifies the destination before it is stored in the DB
Preprocess(newDestination *models.Destination, originalDestination *models.Destination, opts *PreprocessDestinationOpts) error
}
Provider interface handles validation and publisher creation
type Registry ¶
type Registry interface {
// Operations
ValidateDestination(ctx context.Context, destination *models.Destination) error
PublishEvent(ctx context.Context, destination *models.Destination, event *models.Event) (*models.Attempt, error)
DisplayDestination(destination *models.Destination) (*DestinationDisplay, error)
PreprocessDestination(newDestination *models.Destination, originalDestination *models.Destination, opts *PreprocessDestinationOpts) error
// Provider management
RegisterProvider(destinationType string, provider Provider) error
ResolveProvider(destination *models.Destination) (Provider, error)
ResolvePublisher(ctx context.Context, destination *models.Destination) (Publisher, error)
// Metadata access
MetadataLoader() metadata.MetadataLoader
RetrieveProviderMetadata(providerType string) (*metadata.ProviderMetadata, error)
ListProviderMetadata() []*metadata.ProviderMetadata
}
Registry manages providers, their metadata, and publishers
type ValidationErrorDetail ¶
Source Files
¶
Directories
¶
| Path | Synopsis |
|---|---|
|
internal/destregistry/metadata/types.go
|
internal/destregistry/metadata/types.go |