diff --git a/pkg/backends/alb/alb_502_test.go b/pkg/backends/alb/alb_502_test.go index 2b32fe5be..7ff8a3c10 100644 --- a/pkg/backends/alb/alb_502_test.go +++ b/pkg/backends/alb/alb_502_test.go @@ -22,8 +22,8 @@ import ( "testing" "github.com/stretchr/testify/require" - mtypes "github.com/trickstercache/trickster/v2/pkg/backends/alb/mech/types" "github.com/trickstercache/trickster/v2/pkg/backends/alb/mech/tsm" + mtypes "github.com/trickstercache/trickster/v2/pkg/backends/alb/mech/types" "github.com/trickstercache/trickster/v2/pkg/backends/alb/names" ao "github.com/trickstercache/trickster/v2/pkg/backends/alb/options" "github.com/trickstercache/trickster/v2/pkg/backends/prometheus" @@ -108,12 +108,12 @@ func TestALB502WithMultiplePrometheusBackends(t *testing.T) { pool1, _, _ := albpool.NewHealthy([]http.Handler{handler1}) albpool.WaitHealthy(t, pool1, 1) - albOpts := &ao.Options{ + albConfigs := &ao.TSMConfigs{ MechanismName: names.MechanismTSM, OutputFormat: providers.Prometheus, } - tsmMech, err := tsm.New(albOpts, types.Lookup{providers.Prometheus: prometheus.NewClient}) + tsmMech, err := tsm.New(albConfigs, types.Lookup{providers.Prometheus: prometheus.NewClient}) require.NoError(t, err) tsmPM := tsmMech.(mtypes.PoolMechanism) tsmPM.SetPool(pool1) @@ -135,12 +135,12 @@ func TestALB502WithMultiplePrometheusBackends(t *testing.T) { pool2, _, _ := albpool.NewHealthy([]http.Handler{handler1, handler2}) albpool.WaitHealthy(t, pool2, 2) - albOpts := &ao.Options{ + albConfigs := &ao.TSMConfigs{ MechanismName: names.MechanismTSM, OutputFormat: providers.Prometheus, } - tsmMech, err := tsm.New(albOpts, types.Lookup{providers.Prometheus: prometheus.NewClient}) + tsmMech, err := tsm.New(albConfigs, types.Lookup{providers.Prometheus: prometheus.NewClient}) require.NoError(t, err) tsmPM := tsmMech.(mtypes.PoolMechanism) tsmPM.SetPool(pool2) diff --git a/pkg/backends/alb/client.go b/pkg/backends/alb/client.go index 88e47bd22..afeb5cc8e 100644 --- a/pkg/backends/alb/client.go +++ b/pkg/backends/alb/client.go @@ -93,6 +93,7 @@ func NewClient(name string, o *bo.Options, router http.Handler, if err != nil { return nil, err } + c.handler = m } return c, nil diff --git a/pkg/backends/alb/mech/registry/registry.go b/pkg/backends/alb/mech/registry/registry.go index e62d1656f..1b73b93eb 100644 --- a/pkg/backends/alb/mech/registry/registry.go +++ b/pkg/backends/alb/mech/registry/registry.go @@ -40,20 +40,34 @@ var registry = []types.RegistryEntry{ var registryByName = compileSupportedByName(registry) +type funcs struct { + tsmMechanismFunc types.NewTSMMechanismFunc + mechanismFunc types.NewMechanismFunc +} + // compileSupportedByName indexes registry entries by both Name and ShortName. // Panics on duplicate registration: silently last-write-wins would mask a // configuration error that only surfaces at request time. -func compileSupportedByName(entries []types.RegistryEntry) map[types.Name]types.NewMechanismFunc { - out := make(map[types.Name]types.NewMechanismFunc, len(entries)*2) - add := func(name types.Name, fn types.NewMechanismFunc) { +func compileSupportedByName(entries []types.RegistryEntry) map[types.Name]*funcs { + out := make(map[types.Name]*funcs, len(entries)*2) + add := func(name types.Name, fns *funcs) { if _, exists := out[name]; exists { panic("alb/mech/registry: duplicate mechanism name " + name) } - out[name] = fn + out[name] = fns } + for _, entry := range entries { - add(entry.ShortName, entry.New) - add(entry.Name, entry.New) + if entry.New == nil && entry.NewTSM != nil { + fns := &funcs{tsmMechanismFunc: entry.NewTSM} + add(entry.ShortName, fns) + add(entry.Name, fns) + continue + } + + fns := &funcs{mechanismFunc: entry.New} + add(entry.ShortName, fns) + add(entry.Name, fns) } return out } @@ -62,8 +76,14 @@ func New(name types.Name, opts *options.Options, factories rt.Lookup, ) (types.Mechanism, error) { if f, ok := registryByName[name]; ok && f != nil { - return f(opts, factories) + if name == tsm.Name || name == tsm.ShortName { + tsmConfigs := options.NewTSMConfigs(opts) + return f.tsmMechanismFunc(tsmConfigs, factories) + } + + return f.mechanismFunc(opts, factories) } + return nil, errors.ErrUnsupportedMechanism } diff --git a/pkg/backends/alb/mech/tsm/time_series_merge.go b/pkg/backends/alb/mech/tsm/time_series_merge.go index e4b02c694..ebe2f2d48 100644 --- a/pkg/backends/alb/mech/tsm/time_series_merge.go +++ b/pkg/backends/alb/mech/tsm/time_series_merge.go @@ -96,23 +96,23 @@ type stripKeysSnapshot struct { } func RegistryEntry() types.RegistryEntry { - return types.RegistryEntry{Name: Name, ShortName: ShortName, New: New} + return types.RegistryEntry{Name: Name, ShortName: ShortName, NewTSM: New} } -func New(o *options.Options, factories rt.Lookup) (types.Mechanism, error) { +func New(conf *options.TSMConfigs, factories rt.Lookup) (types.Mechanism, error) { out := &handler{ - tsmOptions: o.TSMOptions, - maxCaptureBytes: o.MaxCaptureBytes, - maxFanoutCaptureBytes: o.MaxFanoutCaptureBytes, + tsmOptions: conf.TimeSeriesMergeOptions, + maxCaptureBytes: conf.MaxCaptureBytes, + maxFanoutCaptureBytes: conf.MaxFanoutCaptureBytes, } // this validates the merge configuration for the ALB client as it sets it up // First, verify the output format is a support merge provider - if !providers.IsSupportedTimeSeriesMergeProvider(o.OutputFormat) { + if !providers.IsSupportedTimeSeriesMergeProvider(conf.OutputFormat) { return nil, errors.ErrInvalidTimeSeriesMergeProvider } // next, get the factory function required to create a backend handler for the supplied format - f, ok := factories[o.OutputFormat] + f, ok := factories[conf.OutputFormat] if !ok { return nil, errors.ErrInvalidTimeSeriesMergeProvider } @@ -128,7 +128,7 @@ func New(o *options.Options, factories rt.Lookup) (types.Mechanism, error) { } // set the merge paths in the ALB client out.mergePaths = mc2.MergeablePaths() - out.outputFormat = o.OutputFormat + out.outputFormat = conf.OutputFormat return out, nil } diff --git a/pkg/backends/alb/mech/tsm/time_series_merge_ctor_test.go b/pkg/backends/alb/mech/tsm/time_series_merge_ctor_test.go index 56787eece..8354f51a9 100644 --- a/pkg/backends/alb/mech/tsm/time_series_merge_ctor_test.go +++ b/pkg/backends/alb/mech/tsm/time_series_merge_ctor_test.go @@ -35,7 +35,7 @@ func TestRegistryEntry(t *testing.T) { if entry.Name != Name || entry.ShortName != ShortName { t.Fatalf("RegistryEntry = %+v, want Name=%q ShortName=%q", entry, Name, ShortName) } - if entry.New == nil { + if entry.NewTSM == nil { t.Fatal("RegistryEntry.New is nil") } } @@ -45,7 +45,7 @@ func TestNew(t *testing.T) { t.Run("invalid output format", func(t *testing.T) { t.Parallel() - _, err := New(&options.Options{OutputFormat: "not-a-provider"}, nil) + _, err := New(&options.TSMConfigs{OutputFormat: "not-a-provider"}, nil) if !errors.Is(err, alberr.ErrInvalidTimeSeriesMergeProvider) { t.Fatalf("New() error = %v, want ErrInvalidTimeSeriesMergeProvider", err) } @@ -53,7 +53,7 @@ func TestNew(t *testing.T) { t.Run("missing factory", func(t *testing.T) { t.Parallel() - _, err := New(&options.Options{OutputFormat: providers.Prometheus}, rt.Lookup{}) + _, err := New(&options.TSMConfigs{OutputFormat: providers.Prometheus}, rt.Lookup{}) if !errors.Is(err, alberr.ErrInvalidTimeSeriesMergeProvider) { t.Fatalf("New() error = %v, want ErrInvalidTimeSeriesMergeProvider", err) } @@ -62,7 +62,7 @@ func TestNew(t *testing.T) { t.Run("valid prometheus provider", func(t *testing.T) { t.Parallel() m, err := New( - &options.Options{OutputFormat: providers.Prometheus}, + &options.TSMConfigs{OutputFormat: providers.Prometheus}, rt.Lookup{providers.Prometheus: prometheus.NewClient}, ) if err != nil { diff --git a/pkg/backends/alb/mech/types/types.go b/pkg/backends/alb/mech/types/types.go index 0191d4b7a..f17cec23b 100644 --- a/pkg/backends/alb/mech/types/types.go +++ b/pkg/backends/alb/mech/types/types.go @@ -31,6 +31,9 @@ type Name = string // provided Options type NewMechanismFunc func(*options.Options, types.Lookup) (Mechanism, error) +// NewTSMMechanismFUn defines a function that returns a Time Series Merge mechanism +type NewTSMMechanismFunc func(*options.TSMConfigs, types.Lookup) (Mechanism, error) + // Mechanism represents a specific ALB Implementation (e.g., a Round Robiner). // Pool-aware mechanisms additionally implement PoolMechanism; callers that // need to drive a pool must type-assert before invoking pool methods. @@ -54,4 +57,5 @@ type RegistryEntry struct { Name Name ShortName Name New NewMechanismFunc + NewTSM NewTSMMechanismFunc } diff --git a/pkg/backends/alb/mech/types/types_interfaces_test.go b/pkg/backends/alb/mech/types/types_interfaces_test.go index 73f8f7d9e..ef7bd77a1 100644 --- a/pkg/backends/alb/mech/types/types_interfaces_test.go +++ b/pkg/backends/alb/mech/types/types_interfaces_test.go @@ -71,7 +71,7 @@ func TestPoolMechanismMembership(t *testing.T) { return m }, false}, {"tsm", func(t *testing.T) types.Mechanism { - o := &options.Options{OutputFormat: providers.Prometheus} + o := &options.TSMConfigs{OutputFormat: providers.Prometheus} factories := rt.Lookup{providers.Prometheus: prometheus.NewClient} m, err := tsm.New(o, factories) if err != nil { diff --git a/pkg/backends/alb/options/options.go b/pkg/backends/alb/options/options.go index 7551fe1b5..f2f925391 100644 --- a/pkg/backends/alb/options/options.go +++ b/pkg/backends/alb/options/options.go @@ -141,6 +141,24 @@ var ( ErrOutputFormatOnlyForTSM = errors.New("'output_format' option is only valid for provider 'alb' and mechanism 'tsmerge'") ) +type TSMConfigs struct { + TimeSeriesMergeOptions + MaxCaptureBytes int + MaxFanoutCaptureBytes int + OutputFormat string + MechanismName string +} + +func NewTSMConfigs(o *Options) *TSMConfigs { + return &TSMConfigs{ + TimeSeriesMergeOptions: o.TSMOptions, + MaxCaptureBytes: o.MaxCaptureBytes, + MaxFanoutCaptureBytes: o.MaxFanoutCaptureBytes, + OutputFormat: o.OutputFormat, + MechanismName: o.MechanismName, + } +} + // NewErrInvalidALBOptions returns an invalid ALB Options error func NewErrInvalidALBOptions(backendName string) error { return &InvalidALBOptionsError{