diff --git a/source/extensions/dynamic_modules/BUILD b/source/extensions/dynamic_modules/BUILD index 998831dcdb7b6..72b001be389bc 100644 --- a/source/extensions/dynamic_modules/BUILD +++ b/source/extensions/dynamic_modules/BUILD @@ -86,6 +86,8 @@ go_library( "sdk/go/shared/stats_sink.go", "sdk/go/shared/stats_sink_api.go", "sdk/go/shared/types.go", + "sdk/go/shared/udp_listener_api.go", + "sdk/go/shared/udp_listener_base.go", ], cgo = True, importpath = "github.com/envoyproxy/envoy/source/extensions/dynamic_modules/sdk/go/shared", @@ -116,6 +118,7 @@ go_library( "sdk/go/abi/internal.go", "sdk/go/abi/listener.go", "sdk/go/abi/network.go", + "sdk/go/abi/udp_listener.go", "//source/extensions/dynamic_modules/abi:abi.h", ], cgo = True, diff --git a/source/extensions/dynamic_modules/sdk/go/abi/udp_listener.go b/source/extensions/dynamic_modules/sdk/go/abi/udp_listener.go new file mode 100644 index 0000000000000..dcd5aee3ca176 --- /dev/null +++ b/source/extensions/dynamic_modules/sdk/go/abi/udp_listener.go @@ -0,0 +1,422 @@ +package abi + +/* +#include +#include +#include +#include "../../../abi/abi.h" +*/ +import "C" + +import ( + "runtime" + "unsafe" + + sdk "github.com/envoyproxy/envoy/source/extensions/dynamic_modules/sdk/go" + "github.com/envoyproxy/envoy/source/extensions/dynamic_modules/sdk/go/shared" +) + +type udpListenerFilterConfigWrapper struct { + pluginFactory shared.UdpListenerFilterFactory + configHandle *dymUdpListenerConfigHandle +} + +type udpListenerFilterWrapper = dymUdpListenerFilterHandle + +var udpListenerConfigManager = newManager[udpListenerFilterConfigWrapper]() +var udpListenerPluginManager = newManager[udpListenerFilterWrapper]() + +type dymUdpListenerFilterHandle struct { + hostPluginPtr C.envoy_dynamic_module_type_udp_listener_filter_envoy_ptr + plugin shared.UdpListenerFilter + filterDestroyed bool +} + +type udpListenerAddressKind int + +const ( + udpListenerAddressPeer udpListenerAddressKind = iota + udpListenerAddressLocal +) + +func newDymUdpListenerFilterHandle( + hostPluginPtr C.envoy_dynamic_module_type_udp_listener_filter_envoy_ptr, +) *dymUdpListenerFilterHandle { + return &dymUdpListenerFilterHandle{hostPluginPtr: hostPluginPtr} +} + +func (h *dymUdpListenerFilterHandle) GetDatagramChunks() []shared.UnsafeEnvoyBuffer { + size := C.envoy_dynamic_module_callback_udp_listener_filter_get_datagram_data_chunks_size( + h.hostPluginPtr, + ) + if size == 0 { + return nil + } + result := make([]C.envoy_dynamic_module_type_envoy_buffer, size) + ok := C.envoy_dynamic_module_callback_udp_listener_filter_get_datagram_data_chunks( + h.hostPluginPtr, + unsafe.SliceData(result), + ) + if !bool(ok) { + return nil + } + chunks := envoyBufferSliceToUnsafeEnvoyBufferSlice(result) + runtime.KeepAlive(result) + return chunks +} + +func (h *dymUdpListenerFilterHandle) GetDatagramSize() uint64 { + return uint64(C.envoy_dynamic_module_callback_udp_listener_filter_get_datagram_data_size( + h.hostPluginPtr, + )) +} + +func (h *dymUdpListenerFilterHandle) SetDatagramData(data []byte) bool { + ret := C.envoy_dynamic_module_callback_udp_listener_filter_set_datagram_data( + h.hostPluginPtr, + bytesToModuleBuffer(data), + ) + runtime.KeepAlive(data) + return bool(ret) +} + +func (h *dymUdpListenerFilterHandle) getAddress( + kind udpListenerAddressKind, +) (shared.UnsafeEnvoyBuffer, uint32, bool) { + var address C.envoy_dynamic_module_type_envoy_buffer + var port C.uint32_t + var ret C.bool + switch kind { + case udpListenerAddressPeer: + ret = C.envoy_dynamic_module_callback_udp_listener_filter_get_peer_address( + h.hostPluginPtr, + &address, + &port, + ) + case udpListenerAddressLocal: + ret = C.envoy_dynamic_module_callback_udp_listener_filter_get_local_address( + h.hostPluginPtr, + &address, + &port, + ) + default: + return shared.UnsafeEnvoyBuffer{}, 0, false + } + if !bool(ret) { + return shared.UnsafeEnvoyBuffer{}, 0, false + } + if address.ptr == nil || address.length == 0 { + return shared.UnsafeEnvoyBuffer{}, uint32(port), true + } + return envoyBufferToUnsafeEnvoyBuffer(address), uint32(port), true +} + +func (h *dymUdpListenerFilterHandle) GetPeerAddress() (shared.UnsafeEnvoyBuffer, uint32, bool) { + return h.getAddress(udpListenerAddressPeer) +} + +func (h *dymUdpListenerFilterHandle) GetLocalAddress() (shared.UnsafeEnvoyBuffer, uint32, bool) { + return h.getAddress(udpListenerAddressLocal) +} + +func (h *dymUdpListenerFilterHandle) SendDatagram( + data []byte, + peerAddress string, + peerPort uint32, +) bool { + // An empty peer address tells Envoy to reuse the current datagram's sender, which it only does + // for a null buffer, so do not hand it the pointer of an empty Go string. + peerAddressBuffer := nullModuleBuffer() + if peerAddress != "" { + peerAddressBuffer = stringToModuleBuffer(peerAddress) + } + ret := C.envoy_dynamic_module_callback_udp_listener_filter_send_datagram( + h.hostPluginPtr, + bytesToModuleBuffer(data), + peerAddressBuffer, + C.uint32_t(peerPort), + ) + runtime.KeepAlive(data) + runtime.KeepAlive(peerAddress) + return bool(ret) +} + +func (h *dymUdpListenerFilterHandle) IncrementCounterValue( + id shared.MetricID, + value uint64, +) shared.MetricsResult { + return shared.MetricsResult( + C.envoy_dynamic_module_callback_udp_listener_filter_increment_counter( + h.hostPluginPtr, + C.size_t(id), + C.uint64_t(value), + ), + ) +} + +func (h *dymUdpListenerFilterHandle) SetGaugeValue( + id shared.MetricID, + value uint64, +) shared.MetricsResult { + return shared.MetricsResult( + C.envoy_dynamic_module_callback_udp_listener_filter_set_gauge( + h.hostPluginPtr, + C.size_t(id), + C.uint64_t(value), + ), + ) +} + +func (h *dymUdpListenerFilterHandle) IncrementGaugeValue( + id shared.MetricID, + value uint64, +) shared.MetricsResult { + return shared.MetricsResult( + C.envoy_dynamic_module_callback_udp_listener_filter_increment_gauge( + h.hostPluginPtr, + C.size_t(id), + C.uint64_t(value), + ), + ) +} + +func (h *dymUdpListenerFilterHandle) DecrementGaugeValue( + id shared.MetricID, + value uint64, +) shared.MetricsResult { + return shared.MetricsResult( + C.envoy_dynamic_module_callback_udp_listener_filter_decrement_gauge( + h.hostPluginPtr, + C.size_t(id), + C.uint64_t(value), + ), + ) +} + +func (h *dymUdpListenerFilterHandle) RecordHistogramValue( + id shared.MetricID, + value uint64, +) shared.MetricsResult { + return shared.MetricsResult( + C.envoy_dynamic_module_callback_udp_listener_filter_record_histogram_value( + h.hostPluginPtr, + C.size_t(id), + C.uint64_t(value), + ), + ) +} + +func (h *dymUdpListenerFilterHandle) GetWorkerIndex() uint32 { + return uint32(C.envoy_dynamic_module_callback_udp_listener_filter_get_worker_index( + h.hostPluginPtr, + )) +} + +func (h *dymUdpListenerFilterHandle) Log(level shared.LogLevel, format string, args ...any) { + hostLog(level, format, args) +} + +type dymUdpListenerConfigHandle struct { + hostConfigPtr C.envoy_dynamic_module_type_udp_listener_filter_config_envoy_ptr +} + +func (h *dymUdpListenerConfigHandle) DefineHistogram( + name string, +) (shared.MetricID, shared.MetricsResult) { + var metricID C.size_t + result := C.envoy_dynamic_module_callback_udp_listener_filter_config_define_histogram( + h.hostConfigPtr, + stringToModuleBuffer(name), + &metricID, + ) + runtime.KeepAlive(name) + return shared.MetricID(metricID), shared.MetricsResult(result) +} + +func (h *dymUdpListenerConfigHandle) DefineGauge( + name string, +) (shared.MetricID, shared.MetricsResult) { + var metricID C.size_t + result := C.envoy_dynamic_module_callback_udp_listener_filter_config_define_gauge( + h.hostConfigPtr, + stringToModuleBuffer(name), + &metricID, + ) + runtime.KeepAlive(name) + return shared.MetricID(metricID), shared.MetricsResult(result) +} + +func (h *dymUdpListenerConfigHandle) DefineCounter( + name string, +) (shared.MetricID, shared.MetricsResult) { + var metricID C.size_t + result := C.envoy_dynamic_module_callback_udp_listener_filter_config_define_counter( + h.hostConfigPtr, + stringToModuleBuffer(name), + &metricID, + ) + runtime.KeepAlive(name) + return shared.MetricID(metricID), shared.MetricsResult(result) +} + +func (h *dymUdpListenerConfigHandle) IncrementCounterValue( + id shared.MetricID, + value uint64, +) shared.MetricsResult { + return shared.MetricsResult( + C.envoy_dynamic_module_callback_udp_listener_filter_config_increment_counter( + h.hostConfigPtr, + C.size_t(id), + C.uint64_t(value), + ), + ) +} + +func (h *dymUdpListenerConfigHandle) SetGaugeValue( + id shared.MetricID, + value uint64, +) shared.MetricsResult { + return shared.MetricsResult( + C.envoy_dynamic_module_callback_udp_listener_filter_config_set_gauge( + h.hostConfigPtr, + C.size_t(id), + C.uint64_t(value), + ), + ) +} + +func (h *dymUdpListenerConfigHandle) IncrementGaugeValue( + id shared.MetricID, + value uint64, +) shared.MetricsResult { + return shared.MetricsResult( + C.envoy_dynamic_module_callback_udp_listener_filter_config_increment_gauge( + h.hostConfigPtr, + C.size_t(id), + C.uint64_t(value), + ), + ) +} + +func (h *dymUdpListenerConfigHandle) DecrementGaugeValue( + id shared.MetricID, + value uint64, +) shared.MetricsResult { + return shared.MetricsResult( + C.envoy_dynamic_module_callback_udp_listener_filter_config_decrement_gauge( + h.hostConfigPtr, + C.size_t(id), + C.uint64_t(value), + ), + ) +} + +func (h *dymUdpListenerConfigHandle) RecordHistogramValue( + id shared.MetricID, + value uint64, +) shared.MetricsResult { + return shared.MetricsResult( + C.envoy_dynamic_module_callback_udp_listener_filter_config_record_histogram_value( + h.hostConfigPtr, + C.size_t(id), + C.uint64_t(value), + ), + ) +} + +func (h *dymUdpListenerConfigHandle) Log(level shared.LogLevel, format string, args ...any) { + hostLog(level, format, args) +} + +//export envoy_dynamic_module_on_udp_listener_filter_config_new +func envoy_dynamic_module_on_udp_listener_filter_config_new( + hostConfigPtr C.envoy_dynamic_module_type_udp_listener_filter_config_envoy_ptr, + name C.envoy_dynamic_module_type_envoy_buffer, + config C.envoy_dynamic_module_type_envoy_buffer, +) C.envoy_dynamic_module_type_udp_listener_filter_config_module_ptr { + nameString := envoyBufferToStringUnsafe(name) + configBytes := envoyBufferToBytesUnsafe(config) + + configHandle := &dymUdpListenerConfigHandle{hostConfigPtr: hostConfigPtr} + factory, err := sdk.NewUdpListenerFilterFactory(configHandle, nameString, configBytes) + if err != nil { + configHandle.Log(shared.LogLevelWarn, + "Failed to load UDP listener filter configuration for %q: %v", nameString, err) + return nil + } + if factory == nil { + configHandle.Log(shared.LogLevelWarn, + "Failed to load UDP listener filter configuration for %q: UDP listener filter factory is nil", + nameString) + return nil + } + + configPtr := udpListenerConfigManager.record(&udpListenerFilterConfigWrapper{ + pluginFactory: factory, + configHandle: configHandle, + }) + return C.envoy_dynamic_module_type_udp_listener_filter_config_module_ptr(configPtr) +} + +//export envoy_dynamic_module_on_udp_listener_filter_config_destroy +func envoy_dynamic_module_on_udp_listener_filter_config_destroy( + configPtr C.envoy_dynamic_module_type_udp_listener_filter_config_module_ptr, +) { + configWrapper := udpListenerConfigManager.unwrap(unsafe.Pointer(configPtr)) + if configWrapper == nil { + return + } + configWrapper.pluginFactory.OnDestroy() + udpListenerConfigManager.remove(unsafe.Pointer(configPtr)) +} + +//export envoy_dynamic_module_on_udp_listener_filter_new +func envoy_dynamic_module_on_udp_listener_filter_new( + configPtr C.envoy_dynamic_module_type_udp_listener_filter_config_module_ptr, + hostPluginPtr C.envoy_dynamic_module_type_udp_listener_filter_envoy_ptr, +) C.envoy_dynamic_module_type_udp_listener_filter_module_ptr { + configWrapper := udpListenerConfigManager.unwrap(unsafe.Pointer(configPtr)) + if configWrapper == nil { + return nil + } + + filterWrapper := newDymUdpListenerFilterHandle(hostPluginPtr) + filterWrapper.plugin = configWrapper.pluginFactory.Create(filterWrapper) + if filterWrapper.plugin == nil { + return nil + } + filterPtr := udpListenerPluginManager.record(filterWrapper) + return C.envoy_dynamic_module_type_udp_listener_filter_module_ptr(filterPtr) +} + +//export envoy_dynamic_module_on_udp_listener_filter_on_data +func envoy_dynamic_module_on_udp_listener_filter_on_data( + filterEnvoyPtr C.envoy_dynamic_module_type_udp_listener_filter_envoy_ptr, + filterPtr C.envoy_dynamic_module_type_udp_listener_filter_module_ptr, +) C.envoy_dynamic_module_type_on_udp_listener_filter_status { + _ = filterEnvoyPtr + filterWrapper := udpListenerPluginManager.unwrap(unsafe.Pointer(filterPtr)) + if filterWrapper == nil || filterWrapper.plugin == nil || filterWrapper.filterDestroyed { + return C.envoy_dynamic_module_type_on_udp_listener_filter_status( + shared.UdpListenerFilterStatusContinue, + ) + } + return C.envoy_dynamic_module_type_on_udp_listener_filter_status( + filterWrapper.plugin.OnData(), + ) +} + +//export envoy_dynamic_module_on_udp_listener_filter_destroy +func envoy_dynamic_module_on_udp_listener_filter_destroy( + filterPtr C.envoy_dynamic_module_type_udp_listener_filter_module_ptr, +) { + filterWrapper := udpListenerPluginManager.unwrap(unsafe.Pointer(filterPtr)) + if filterWrapper == nil || filterWrapper.filterDestroyed { + return + } + filterWrapper.filterDestroyed = true + if filterWrapper.plugin != nil { + filterWrapper.plugin.OnDestroy() + } + udpListenerPluginManager.remove(unsafe.Pointer(filterPtr)) +} diff --git a/source/extensions/dynamic_modules/sdk/go/sdk.go b/source/extensions/dynamic_modules/sdk/go/sdk.go index 6f11d258a31da..6e49265772970 100644 --- a/source/extensions/dynamic_modules/sdk/go/sdk.go +++ b/source/extensions/dynamic_modules/sdk/go/sdk.go @@ -11,6 +11,7 @@ import ( var httpFilterConfigFactoryRegistry = make(map[string]shared.HttpFilterConfigFactory) var listenerFilterConfigFactoryRegistry = make(map[string]shared.ListenerFilterConfigFactory) var networkFilterConfigFactoryRegistry = make(map[string]shared.NetworkFilterConfigFactory) +var udpListenerFilterConfigFactoryRegistry = make(map[string]shared.UdpListenerFilterConfigFactory) var statSinkConfigFactoryRegistry = make(map[string]shared.StatSinkConfigFactory) // NewHttpFilterFactory creates a new plugin factory for the given plugin name and unparsed config. @@ -94,6 +95,36 @@ func RegisterNetworkFilterConfigFactories(factories map[string]shared.NetworkFil } } +// NewUdpListenerFilterFactory creates a new UDP listener filter factory for the given plugin name +// and unparsed config. +func NewUdpListenerFilterFactory(handle shared.UdpListenerFilterConfigHandle, name string, + unparsedConfig []byte) (shared.UdpListenerFilterFactory, error) { + configFactory := udpListenerFilterConfigFactoryRegistry[name] + if configFactory == nil { + return nil, fmt.Errorf("failed to get UDP listener filter config factory for %s", name) + } + return configFactory.Create(handle, unparsedConfig) +} + +// GetUdpListenerFilterConfigFactory gets the UDP listener filter config factory for the given +// plugin name. +func GetUdpListenerFilterConfigFactory(name string) shared.UdpListenerFilterConfigFactory { + return udpListenerFilterConfigFactoryRegistry[name] +} + +// RegisterUdpListenerFilterConfigFactories registers UDP listener filter config factories for +// plugins in the composer binary itself. This function MUST only be called from init() functions. +func RegisterUdpListenerFilterConfigFactories( + factories map[string]shared.UdpListenerFilterConfigFactory, +) { + for name, factory := range factories { + if _, ok := udpListenerFilterConfigFactoryRegistry[name]; ok { + panic("UDP listener filter config factory already registered: " + name) + } + udpListenerFilterConfigFactoryRegistry[name] = factory + } +} + // NewStatSink creates a new StatSink for the given sink name and unparsed config bytes. func NewStatSink(handle shared.StatSinkHandle, name string, unparsedConfig []byte) (shared.StatSink, error) { diff --git a/source/extensions/dynamic_modules/sdk/go/shared/udp_listener_api.go b/source/extensions/dynamic_modules/sdk/go/shared/udp_listener_api.go new file mode 100644 index 0000000000000..e525b5afbccd2 --- /dev/null +++ b/source/extensions/dynamic_modules/sdk/go/shared/udp_listener_api.go @@ -0,0 +1,79 @@ +package shared + +// UdpListenerFilter is the interface to implement your own UDP listener filter logic. +// +// Unlike TCP listener and network filters, a UDP listener filter instance is created once per +// listener per Envoy worker thread, not once per connection or session. OnData is then called for +// every datagram that worker receives, so any state kept on the implementation is shared across +// datagrams and across peers. Implementations do not need to be thread-safe: Envoy only ever calls +// into a given instance from its own worker thread. +type UdpListenerFilter interface { + // OnData is called when a UDP datagram is received on this worker. + // + // The datagram itself is read and modified through the UdpListenerFilterHandle passed to + // UdpListenerFilterFactory.Create. + // + // Returning UdpListenerFilterStatusContinue passes the datagram to the next UDP listener + // filter; returning UdpListenerFilterStatusStop drops it so later filters never see it. + OnData() UdpListenerFilterStatus + + // OnDestroy is called when the filter instance is being destroyed, which happens when the + // listener is drained or Envoy shuts the worker down. It should release any resources tied to + // the filter. + OnDestroy() +} + +// EmptyUdpListenerFilter provides no-op UDP listener filter hooks with default continue behavior. +type EmptyUdpListenerFilter struct{} + +// OnData implements UdpListenerFilter. +func (f *EmptyUdpListenerFilter) OnData() UdpListenerFilterStatus { + return UdpListenerFilterStatusDefault +} + +// OnDestroy implements UdpListenerFilter. +func (f *EmptyUdpListenerFilter) OnDestroy() {} + +// UdpListenerFilterFactory creates per-worker UDP listener filters. +// The implementation of this interface should be thread-safe and hold the parsed configuration. +type UdpListenerFilterFactory interface { + // Create constructs the UdpListenerFilter for one Envoy worker thread. + // + // Returning nil causes filter creation to fail, and Envoy passes datagrams through without + // invoking the module on that worker. + Create(handle UdpListenerFilterHandle) UdpListenerFilter + + // OnDestroy is called when Envoy destroys this factory, usually after configuration has been + // replaced and the listener using it has drained. + OnDestroy() +} + +// EmptyUdpListenerFilterFactory returns EmptyUdpListenerFilter instances. +type EmptyUdpListenerFilterFactory struct{} + +// Create implements UdpListenerFilterFactory. +func (f *EmptyUdpListenerFilterFactory) Create(UdpListenerFilterHandle) UdpListenerFilter { + return &EmptyUdpListenerFilter{} +} + +// OnDestroy implements UdpListenerFilterFactory. +func (f *EmptyUdpListenerFilterFactory) OnDestroy() {} + +// UdpListenerFilterConfigFactory parses configuration and returns a thread-safe filter factory. +// The implementation of this interface should be thread-safe and usually stateless. +type UdpListenerFilterConfigFactory interface { + // Create parses unparsedConfig and returns the UdpListenerFilterFactory used for this listener. + // + // Returning an error rejects the filter configuration. + Create(handle UdpListenerFilterConfigHandle, + unparsedConfig []byte) (UdpListenerFilterFactory, error) +} + +// EmptyUdpListenerFilterConfigFactory returns EmptyUdpListenerFilterFactory instances. +type EmptyUdpListenerFilterConfigFactory struct{} + +// Create implements UdpListenerFilterConfigFactory. +func (f *EmptyUdpListenerFilterConfigFactory) Create(UdpListenerFilterConfigHandle, + []byte) (UdpListenerFilterFactory, error) { + return &EmptyUdpListenerFilterFactory{}, nil +} diff --git a/source/extensions/dynamic_modules/sdk/go/shared/udp_listener_base.go b/source/extensions/dynamic_modules/sdk/go/shared/udp_listener_base.go new file mode 100644 index 0000000000000..bb2714da9d234 --- /dev/null +++ b/source/extensions/dynamic_modules/sdk/go/shared/udp_listener_base.go @@ -0,0 +1,104 @@ +package shared + +// UdpListenerFilterStatus controls whether Envoy continues UDP listener filter iteration. +type UdpListenerFilterStatus int32 + +const ( + // UdpListenerFilterStatusContinue lets Envoy pass the datagram to the next UDP listener filter. + UdpListenerFilterStatusContinue UdpListenerFilterStatus = 0 + // UdpListenerFilterStatusStop stops iteration so later filters never see this datagram. + // + // Unlike the TCP listener filter chain, there is no callback to resume iteration afterwards: + // the decision applies to the current datagram only, and the next datagram starts a fresh + // iteration. + UdpListenerFilterStatusStop UdpListenerFilterStatus = 1 + // UdpListenerFilterStatusDefault is the default UDP listener filter result. + UdpListenerFilterStatusDefault UdpListenerFilterStatus = UdpListenerFilterStatusContinue +) + +// UdpListenerFilterHandle exposes the current datagram and the UDP listener's state. +// +// The datagram accessors are only valid for the duration of an UdpListenerFilter.OnData call. +// Outside of it Envoy holds no current datagram, so they report no data. +type UdpListenerFilterHandle interface { + // GetDatagramChunks returns the current datagram payload as Envoy-owned chunks. + // + // The chunks alias Envoy memory and are only valid for the duration of the OnData call; copy + // the bytes if you need to retain them. Returns nil outside of OnData or for an empty datagram. + GetDatagramChunks() []UnsafeEnvoyBuffer + // GetDatagramSize returns the total size of the current datagram payload in bytes. + // + // Returns 0 outside of OnData. + GetDatagramSize() uint64 + // SetDatagramData replaces the entire payload of the current datagram. + // + // Passing empty data clears the payload. The provided bytes are owned by the caller for the + // duration of the call. Returns false outside of OnData. + SetDatagramData(data []byte) bool + + // GetPeerAddress returns the sender's IP address and port for the current datagram. + // + // Returns false outside of OnData, or when the peer address is not an IP address. + GetPeerAddress() (UnsafeEnvoyBuffer, uint32, bool) + // GetLocalAddress returns the local IP address and port the current datagram was received on. + // + // Returns false outside of OnData, or when the local address is not an IP address. + GetLocalAddress() (UnsafeEnvoyBuffer, uint32, bool) + + // SendDatagram sends data from the UDP listener socket to peerAddress:peerPort. + // + // An empty peerAddress reuses the current datagram's sender, which makes this an echo back to + // the client; that form only works during OnData. peerAddress must be an IP address literal, + // not a hostname. The provided bytes are owned by the caller for the duration of the call. + // + // It returns false if the address cannot be parsed or the listener has no local address to send + // from. + SendDatagram(data []byte, peerAddress string, peerPort uint32) bool + + // IncrementCounterValue increases a counter metric by value. + IncrementCounterValue(id MetricID, value uint64) MetricsResult + // SetGaugeValue sets a gauge metric to value. + SetGaugeValue(id MetricID, value uint64) MetricsResult + // IncrementGaugeValue increases a gauge metric by value. + IncrementGaugeValue(id MetricID, value uint64) MetricsResult + // DecrementGaugeValue decreases a gauge metric by value. + DecrementGaugeValue(id MetricID, value uint64) MetricsResult + // RecordHistogramValue records value in a histogram metric. + RecordHistogramValue(id MetricID, value uint64) MetricsResult + + // GetWorkerIndex returns the Envoy worker index this filter instance belongs to. + GetWorkerIndex() uint32 + + // Log writes a formatted message through Envoy's logging subsystem. + Log(level LogLevel, format string, args ...any) +} + +// UdpListenerFilterConfigHandle exposes host services during UDP listener filter config creation. +type UdpListenerFilterConfigHandle interface { + // DefineHistogram defines a histogram metric during config creation. + // + // Metrics can only be defined while the configuration is being created; afterwards Envoy + // freezes metric creation and this returns MetricsFrozen. + DefineHistogram(name string) (MetricID, MetricsResult) + // DefineGauge defines a gauge metric during config creation. + DefineGauge(name string) (MetricID, MetricsResult) + // DefineCounter defines a counter metric during config creation. + DefineCounter(name string) (MetricID, MetricsResult) + + // IncrementCounterValue increases a counter metric by value from the config context. + // + // Unlike UdpListenerFilterHandle.IncrementCounterValue, this does not require a per-worker + // filter and can be called outside of datagram processing. + IncrementCounterValue(id MetricID, value uint64) MetricsResult + // SetGaugeValue sets a gauge metric to value from the config context. + SetGaugeValue(id MetricID, value uint64) MetricsResult + // IncrementGaugeValue increases a gauge metric by value from the config context. + IncrementGaugeValue(id MetricID, value uint64) MetricsResult + // DecrementGaugeValue decreases a gauge metric by value from the config context. + DecrementGaugeValue(id MetricID, value uint64) MetricsResult + // RecordHistogramValue records value into a histogram metric from the config context. + RecordHistogramValue(id MetricID, value uint64) MetricsResult + + // Log writes a formatted message through Envoy's logging subsystem. + Log(level LogLevel, format string, args ...any) +} diff --git a/test/extensions/dynamic_modules/test_data/go/BUILD b/test/extensions/dynamic_modules/test_data/go/BUILD index 957cf6ae1505b..5c9e62ecc9711 100644 --- a/test/extensions/dynamic_modules/test_data/go/BUILD +++ b/test/extensions/dynamic_modules/test_data/go/BUILD @@ -11,3 +11,5 @@ test_program(name = "listener_integration_test") test_program(name = "network_integration_test") test_program(name = "stat_sink_integration_test") + +test_program(name = "udp_listener_integration_test") diff --git a/test/extensions/dynamic_modules/test_data/go/udp_listener_integration_test/udp_listener_integration_test.go b/test/extensions/dynamic_modules/test_data/go/udp_listener_integration_test/udp_listener_integration_test.go new file mode 100644 index 0000000000000..a6c842266accc --- /dev/null +++ b/test/extensions/dynamic_modules/test_data/go/udp_listener_integration_test/udp_listener_integration_test.go @@ -0,0 +1,149 @@ +package main + +import ( + sdk "github.com/envoyproxy/envoy/source/extensions/dynamic_modules/sdk/go" + _ "github.com/envoyproxy/envoy/source/extensions/dynamic_modules/sdk/go/abi" + "github.com/envoyproxy/envoy/source/extensions/dynamic_modules/sdk/go/shared" +) + +func init() { + sdk.RegisterUdpListenerFilterConfigFactories(map[string]shared.UdpListenerFilterConfigFactory{ + "echo_datagram": &echoDatagramConfigFactory{}, + "rewrite_datagram": &rewriteDatagramConfigFactory{}, + }) +} + +func main() {} + +// datagramPayload copies the current datagram out of the Envoy-owned chunks. +func datagramPayload(handle shared.UdpListenerFilterHandle) []byte { + chunks := handle.GetDatagramChunks() + if len(chunks) == 0 { + panic("expected at least one datagram chunk") + } + payload := make([]byte, 0, handle.GetDatagramSize()) + for _, chunk := range chunks { + payload = append(payload, chunk.ToBytes()...) + } + if uint64(len(payload)) != handle.GetDatagramSize() { + panic("chunk lengths do not add up to the datagram size") + } + return payload +} + +// echoDatagram sends the datagram straight back to its sender and stops iteration, so the +// udp_proxy filter behind it never sees the datagram and the upstream is never reached. +type echoDatagramConfigFactory struct { + shared.EmptyUdpListenerFilterConfigFactory +} + +func (f *echoDatagramConfigFactory) Create(shared.UdpListenerFilterConfigHandle, + []byte) (shared.UdpListenerFilterFactory, error) { + return &echoDatagramFactory{}, nil +} + +type echoDatagramFactory struct { + shared.EmptyUdpListenerFilterFactory +} + +func (f *echoDatagramFactory) Create(handle shared.UdpListenerFilterHandle) shared.UdpListenerFilter { + return &echoDatagramFilter{handle: handle} +} + +type echoDatagramFilter struct { + handle shared.UdpListenerFilterHandle + shared.EmptyUdpListenerFilter +} + +func (f *echoDatagramFilter) OnData() shared.UdpListenerFilterStatus { + if f.handle.GetDatagramSize() == 0 { + panic("expected a non-empty datagram") + } + payload := datagramPayload(f.handle) + + peerAddress, peerPort, ok := f.handle.GetPeerAddress() + if !ok || peerAddress.ToString() == "" || peerPort == 0 { + panic("expected a peer address") + } + if _, _, ok := f.handle.GetLocalAddress(); !ok { + panic("expected a local address") + } + + // An empty peer address reuses the current datagram's sender. + if !f.handle.SendDatagram(payload, "", 0) { + panic("failed to send datagram") + } + return shared.UdpListenerFilterStatusStop +} + +// rewriteDatagram replaces the datagram payload and lets iteration continue, so udp_proxy forwards +// the rewritten bytes upstream. +type rewriteDatagramConfigFactory struct { + shared.EmptyUdpListenerFilterConfigFactory +} + +func (f *rewriteDatagramConfigFactory) Create(handle shared.UdpListenerFilterConfigHandle, + _ []byte) (shared.UdpListenerFilterFactory, error) { + counterID, result := handle.DefineCounter("datagrams_rewritten") + if result != shared.MetricsSuccess { + panic("failed to define counter") + } + gaugeID, result := handle.DefineGauge("last_datagram_size") + if result != shared.MetricsSuccess { + panic("failed to define gauge") + } + histogramID, result := handle.DefineHistogram("datagram_size") + if result != shared.MetricsSuccess { + panic("failed to define histogram") + } + return &rewriteDatagramFactory{ + counterID: counterID, + gaugeID: gaugeID, + histogramID: histogramID, + }, nil +} + +type rewriteDatagramFactory struct { + shared.EmptyUdpListenerFilterFactory + counterID shared.MetricID + gaugeID shared.MetricID + histogramID shared.MetricID +} + +func (f *rewriteDatagramFactory) Create( + handle shared.UdpListenerFilterHandle, +) shared.UdpListenerFilter { + return &rewriteDatagramFilter{handle: handle, factory: f} +} + +type rewriteDatagramFilter struct { + handle shared.UdpListenerFilterHandle + factory *rewriteDatagramFactory + shared.EmptyUdpListenerFilter +} + +func (f *rewriteDatagramFilter) OnData() shared.UdpListenerFilterStatus { + size := f.handle.GetDatagramSize() + // Read the payload before overwriting it so the read path is exercised too. + _ = datagramPayload(f.handle) + + if !f.handle.SetDatagramData([]byte("rewritten")) { + panic("failed to set datagram data") + } + if f.handle.GetDatagramSize() != uint64(len("rewritten")) { + panic("unexpected datagram size after rewrite") + } + + if f.handle.IncrementCounterValue(f.factory.counterID, 1) != shared.MetricsSuccess { + panic("failed to increment counter") + } + if f.handle.SetGaugeValue(f.factory.gaugeID, size) != shared.MetricsSuccess { + panic("failed to set gauge") + } + if f.handle.RecordHistogramValue(f.factory.histogramID, size) != shared.MetricsSuccess { + panic("failed to record histogram value") + } + + f.handle.Log(shared.LogLevelInfo, "rewrote datagram on worker %d", f.handle.GetWorkerIndex()) + return shared.UdpListenerFilterStatusContinue +} diff --git a/test/extensions/dynamic_modules/udp/BUILD b/test/extensions/dynamic_modules/udp/BUILD index 00020bf5b81a3..8e4a2e9c10bb4 100644 --- a/test/extensions/dynamic_modules/udp/BUILD +++ b/test/extensions/dynamic_modules/udp/BUILD @@ -65,6 +65,27 @@ envoy_cc_test( ], ) +envoy_cc_test( + name = "sdk_integration_test", + srcs = ["sdk_integration_test.cc"], + data = [ + "//test/extensions/dynamic_modules/test_data/go:udp_listener_integration_test", + ], + env = {"GODEBUG": "cgocheck=0"}, + deps = [ + "//source/extensions/dynamic_modules:abi_impl", + "//source/extensions/filters/http/dynamic_modules:abi_impl", + "//source/extensions/filters/udp/dynamic_modules:config", + "//source/extensions/filters/udp/udp_proxy:config", + "//test/integration:integration_lib", + "//test/test_common:network_utility_lib", + "//test/test_common:utility_lib", + "@envoy_api//envoy/config/bootstrap/v3:pkg_cc_proto", + "@envoy_api//envoy/extensions/filters/udp/dynamic_modules/v3:pkg_cc_proto", + "@envoy_api//envoy/extensions/filters/udp/udp_proxy/v3:pkg_cc_proto", + ], +) + envoy_cc_test( name = "udp_dynamic_modules_integration_test", srcs = ["udp_dynamic_modules_integration_test.cc"], diff --git a/test/extensions/dynamic_modules/udp/sdk_integration_test.cc b/test/extensions/dynamic_modules/udp/sdk_integration_test.cc new file mode 100644 index 0000000000000..1c7a068062b09 --- /dev/null +++ b/test/extensions/dynamic_modules/udp/sdk_integration_test.cc @@ -0,0 +1,130 @@ +#include "envoy/config/bootstrap/v3/bootstrap.pb.h" +#include "envoy/extensions/filters/udp/dynamic_modules/v3/dynamic_modules.pb.h" +#include "envoy/extensions/filters/udp/udp_proxy/v3/udp_proxy.pb.h" + +#include "test/integration/integration.h" +#include "test/test_common/environment.h" +#include "test/test_common/network_utility.h" +#include "test/test_common/utility.h" + +#include "gtest/gtest.h" + +namespace Envoy { +namespace Extensions { +namespace UdpFilters { +namespace DynamicModules { +namespace { + +// Exercises the UDP listener filter SDKs end to end against a real dynamic module built from the +// per-language test_data directory. The parameter selects which SDK's module is loaded. +class DynamicModulesUdpListenerSdkIntegrationTest : public testing::TestWithParam, + public BaseIntegrationTest { +public: + DynamicModulesUdpListenerSdkIntegrationTest() + : BaseIntegrationTest(Network::Address::IpVersion::v4, + ConfigHelper::baseUdpListenerConfig()) {} + +protected: + void initializeFilter(const std::string& filter_name) { + TestEnvironment::setEnvVar( + "ENVOY_DYNAMIC_MODULES_SEARCH_PATH", + TestEnvironment::substitute("{{ test_rundir }}/test/extensions/dynamic_modules/test_data/" + + GetParam()), + 1); + + FakeUpstreamConfig::UdpConfig config; + setUdpFakeUpstream(config); + + // ConfigHelper::addListenerFilter moves the filter it adds to the front of the chain, so these + // are added back to front to end up with [dynamic_modules, udp_proxy]. The dynamic module + // filter has to run first: ActiveRawUdpListener::onDataWorker stops iterating on + // StopIteration, which is what keeps a dropped datagram away from udp_proxy. + config_helper_.addListenerFilter(R"EOF( +name: envoy.filters.udp_listener.udp_proxy +typed_config: + '@type': type.googleapis.com/envoy.extensions.filters.udp.udp_proxy.v3.UdpProxyConfig + stat_prefix: service + matcher: + on_no_match: + action: + name: route + typed_config: + '@type': type.googleapis.com/envoy.extensions.filters.udp.udp_proxy.v3.Route + cluster: cluster_0 +)EOF"); + + config_helper_.addListenerFilter(fmt::format(R"EOF( +name: envoy.filters.udp_listener.dynamic_modules +typed_config: + "@type": type.googleapis.com/envoy.extensions.filters.udp.dynamic_modules.v3.DynamicModuleUdpListenerFilter + dynamic_module_config: + name: "udp_listener_integration_test" + do_not_close: true + filter_name: "{}" +)EOF", + filter_name)); + + BaseIntegrationTest::initialize(); + } + + Network::Address::InstanceConstSharedPtr listenerAddress() { + return *Network::Utility::resolveUrl(fmt::format( + "udp://{}:{}", Network::Test::getLoopbackAddressUrlString(Network::Address::IpVersion::v4), + lookupPort("listener_0"))); + } +}; + +// Only the Go SDK ships a udp_listener_integration_test module today. Add "rust" and "cpp" here +// once the equivalent modules exist in their test_data directories. +INSTANTIATE_TEST_SUITE_P(SdkLanguages, DynamicModulesUdpListenerSdkIntegrationTest, + testing::Values("go"), + [](const testing::TestParamInfo& info) { + return info.param; + }); + +// The module reads the datagram, its peer address and its local address, sends the payload straight +// back to the sender, and stops iteration so udp_proxy never forwards it upstream. +TEST_P(DynamicModulesUdpListenerSdkIntegrationTest, EchoDatagram) { + initializeFilter("echo_datagram"); + + const std::string request = "hello"; + Network::Test::UdpSyncPeer client(Network::Address::IpVersion::v4); + client.write(request, *listenerAddress()); + + Network::UdpRecvData response; + client.recv(response); + EXPECT_EQ(request, response.buffer_->toString()); + + // StopIteration kept the datagram away from udp_proxy, so the upstream saw nothing. + Network::UdpRecvData upstream_datagram; + EXPECT_FALSE( + fake_upstreams_[0]->waitForUdpDatagram(upstream_datagram, std::chrono::milliseconds(500))); +} + +// The module replaces the datagram payload and continues iteration, so udp_proxy forwards the +// rewritten bytes upstream. It also touches every metrics callback and the worker index. +TEST_P(DynamicModulesUdpListenerSdkIntegrationTest, RewriteDatagram) { + initializeFilter("rewrite_datagram"); + + const std::string request = "hello"; + Network::Test::UdpSyncPeer client(Network::Address::IpVersion::v4); + client.write(request, *listenerAddress()); + + Network::UdpRecvData upstream_datagram; + ASSERT_TRUE(fake_upstreams_[0]->waitForUdpDatagram(upstream_datagram)); + EXPECT_EQ("rewritten", upstream_datagram.buffer_->toString()); + + // The UDP filter config scopes its metrics as "..", defaulting the + // namespace to DefaultMetricsNamespace. See DynamicModuleUdpListenerFilterConfig's constructor. + test_server_->waitForCounter("dynamicmodulescustom.rewrite_datagram.datagrams_rewritten", + testing::Eq(1)); + EXPECT_EQ( + request.size(), + test_server_->gauge("dynamicmodulescustom.rewrite_datagram.last_datagram_size")->value()); +} + +} // namespace +} // namespace DynamicModules +} // namespace UdpFilters +} // namespace Extensions +} // namespace Envoy