Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

[jaeger-v2] Streamline storage initialization #5171

Merged
merged 1 commit into from
Feb 6, 2024
Merged

[jaeger-v2] Streamline storage initialization #5171

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
71 changes: 50 additions & 21 deletions cmd/jaeger/internal/extension/jaegerstorage/extension.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -13,8 +13,10 @@ import (
"go.opentelemetry.io/collector/extension"
"go.uber.org/zap"

memoryCfg "github.com/jaegertracing/jaeger/pkg/memory/config"
"github.com/jaegertracing/jaeger/pkg/metrics"
"github.com/jaegertracing/jaeger/plugin/storage/badger"
badgerCfg "github.com/jaegertracing/jaeger/plugin/storage/badger"
"github.com/jaegertracing/jaeger/plugin/storage/memory"
"github.com/jaegertracing/jaeger/storage"
)
Expand Down Expand Up @@ -60,35 +62,62 @@ func newStorageExt(config *Config, otel component.TelemetrySettings) *storageExt
}
}

func (s *storageExt) Start(ctx context.Context, host component.Host) error {
for name, mem := range s.config.Memory {
if _, ok := s.factories[name]; ok {
return fmt.Errorf("duplicate memory storage name %s", name)
}
s.factories[name] = memory.NewFactoryWithConfig(
mem,
metrics.NullFactory,
s.logger.With(zap.String("storage_name", name)),
)
}
type starter[Config any, Factory storage.Factory] struct {
ext *storageExt
storageKind string
cfg map[string]Config
builder func(Config, metrics.Factory, *zap.Logger) (Factory, error)
}

for name, b := range s.config.Badger {
if _, ok := s.factories[name]; ok {
return fmt.Errorf("duplicate badger storage name %s", name)
func (s *starter[Config, Factory]) build(ctx context.Context, host component.Host) error {
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Internal state isn't persisted between method calls so all methods are effectively read-only; we could assert this by using value receivers instead of pointer receivers.

Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

true, but makes no difference in this context since the code runs exactly once. Don't want to reset the approval just for this fix :-)

for name, cfg := range s.cfg {
if _, ok := s.ext.factories[name]; ok {
return fmt.Errorf("duplicate %s storage name %s", s.storageKind, name)
}
var err error
factory, err := badger.NewFactoryWithConfig(
b,
factory, err := s.builder(
cfg,
metrics.NullFactory,
s.logger.With(zap.String("storage_name", name)),
s.ext.logger.With(zap.String("storage_name", name)),
)
if err != nil {
return fmt.Errorf("failed to initialize badger storage: %w", err)
return fmt.Errorf("failed to initialize %s storage %s: %w", s.storageKind, name, err)
}
s.factories[name] = factory
s.ext.factories[name] = factory
}
return nil
}

// TODO add support for other backends
func (s *storageExt) Start(ctx context.Context, host component.Host) error {
memStarter := &starter[memoryCfg.Configuration, *memory.Factory]{
ext: s,
storageKind: "memory",
cfg: s.config.Memory,
// memory factory does not return an error, so need to wrap it
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Why not change memory.NewFactoryWithConfig to return the error? It doesn't look like it's used anywhere else, so it shouldn't be a large breaking change.

Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Because I hate functions that are declared with errors but always return nil. The caller either has to ignore the error, or it has no easy way of unit testing the error handing path (which will never be used). Lose-lose. I think if the function can never return an error it should never declare it as a possibility (unless you're dealing with an interface where there's no choice).

builder: func(
cfg memoryCfg.Configuration,
metricsFactory metrics.Factory,
logger *zap.Logger,
) (*memory.Factory, error) {
return memory.NewFactoryWithConfig(cfg, metricsFactory, logger), nil
},
}
badgerStarter := &starter[badgerCfg.NamespaceConfig, *badger.Factory]{
ext: s,
storageKind: "badger",
cfg: s.config.Badger,
builder: badger.NewFactoryWithConfig,
}

builders := []func(ctx context.Context, host component.Host) error{
memStarter.build,
badgerStarter.build,
// TODO add support for other backends
}
for _, builder := range builders {
if err := builder(ctx, host); err != nil {
return err
}
}
return nil
}

Expand Down
163 changes: 67 additions & 96 deletions cmd/jaeger/internal/extension/jaegerstorage/extension_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -24,11 +24,6 @@ import (
"github.com/jaegertracing/jaeger/storage/spanstore"
)

const (
memstoreName = "memstore"
badgerName = "badgerstore"
)

type storageHost struct {
t *testing.T
storageExtension component.Component
Expand Down Expand Up @@ -79,143 +74,119 @@ func (e errorFactory) Close() error {
func TestStorageExtensionConfigError(t *testing.T) {
config := createDefaultConfig().(*Config)
err := config.Validate()
require.EqualError(t, err, fmt.Sprintf("%s: no storage type present in config", ID))
require.ErrorContains(t, err, "no storage type present in config")
}

func TestStorageExtensionStartTwiceError(t *testing.T) {
ctx := context.Background()

storageExtension := makeStorageExtension(t, memstoreName)

host := componenttest.NewNopHost()
err := storageExtension.Start(ctx, host)
require.Error(t, err)
require.EqualError(t, err, fmt.Sprintf("duplicate memory storage name %s", memstoreName))
func TestStorageExtensionNameConflict(t *testing.T) {
storageExtension := makeStorageExtenion(t, &Config{
Memory: map[string]memoryCfg.Configuration{
"foo": {MaxTraces: 10000},
},
Badger: map[string]badgerCfg.NamespaceConfig{
"foo": {},
},
})
err := storageExtension.Start(context.Background(), componenttest.NewNopHost())
require.ErrorContains(t, err, "duplicate")
}

func TestStorageFactoryBadHostError(t *testing.T) {
makeStorageExtension(t, memstoreName)

host := componenttest.NewNopHost()
_, err := GetStorageFactory(memstoreName, host)
require.Error(t, err)
require.EqualError(t, err, fmt.Sprintf("cannot find extension '%s' (make sure it's defined earlier in the config)", ID))
_, err := GetStorageFactory("something", host)
require.ErrorContains(t, err, "cannot find extension")
}

func TestStorageFactoryBadNameError(t *testing.T) {
storageExtension := makeStorageExtension(t, memstoreName)

host := storageHost{t: t, storageExtension: storageExtension}
const badMemstoreName = "test"

_, err := GetStorageFactory(badMemstoreName, host)
require.Error(t, err)
require.EqualError(t, err, fmt.Sprintf("cannot find storage '%s' declared with '%s' extension", badMemstoreName, ID))
host := storageHost{t: t, storageExtension: startStorageExtension(t, "foo")}
_, err := GetStorageFactory("bar", host)
require.ErrorContains(t, err, "cannot find storage 'bar'")
}

func TestStorageFactoryBadShutdownError(t *testing.T) {
shutdownError := fmt.Errorf("shutdown error")
storageExtension := storageExt{
factories: make(map[string]storage.Factory),
factories: map[string]storage.Factory{
"foo": errorFactory{closeErr: shutdownError},
},
}
badFactoryError := fmt.Errorf("error factory")
storageExtension.factories[memstoreName] = errorFactory{closeErr: badFactoryError}

err := storageExtension.Shutdown(context.Background())
require.ErrorIs(t, err, badFactoryError)
require.ErrorIs(t, err, shutdownError)
}

func TestStorageExtension(t *testing.T) {
storageExtension := makeStorageExtension(t, memstoreName)

host := storageHost{t: t, storageExtension: storageExtension}

_, err := GetStorageFactory(memstoreName, host)
const name = "foo"
host := storageHost{t: t, storageExtension: startStorageExtension(t, name)}
f, err := GetStorageFactory(name, host)
require.NoError(t, err)
require.NotNil(t, f)
}

func TestBadgerStorageExtension(t *testing.T) {
ctx := context.Background()
telemetrySettings := component.TelemetrySettings{
Logger: zap.NewNop(),
TracerProvider: nooptrace.NewTracerProvider(),
MeterProvider: noopmetric.NewMeterProvider(),
}

config := &Config{
storageExtension := makeStorageExtenion(t, &Config{
Badger: map[string]badgerCfg.NamespaceConfig{
badgerName: {
"foo": {
Ephemeral: true,
MaintenanceInterval: 5,
MetricsUpdateInterval: 10,
},
},
}

require.NoError(t, config.Validate())

extensionFactory := NewFactory()
storageExtension, err := extensionFactory.CreateExtension(ctx, extension.CreateSettings{
ID: ID,
TelemetrySettings: telemetrySettings,
BuildInfo: component.NewDefaultBuildInfo(),
}, config)
require.NoError(t, err)

host := componenttest.NewNopHost()
err = storageExtension.Start(ctx, host)
})
ctx := context.Background()
err := storageExtension.Start(ctx, componenttest.NewNopHost())
require.NoError(t, err)

err = storageExtension.Start(ctx, host)
t.Cleanup(func() { require.NoError(t, storageExtension.Shutdown(ctx)) })
require.Error(t, err)
require.EqualError(t, err, fmt.Sprintf("duplicate badger storage name %s", badgerName))
require.NoError(t, storageExtension.Shutdown(ctx))
}

func TestBadgerStorageExtensionError(t *testing.T) {
ctx := context.Background()
factory, _ := badgerCfg.NewFactoryWithConfig(badgerCfg.NamespaceConfig{}, metrics.NullFactory, zap.NewNop())
ext := storageExt{
config: &Config{
Badger: map[string]badgerCfg.NamespaceConfig{
badgerName: {},
ext := makeStorageExtenion(t, &Config{
Badger: map[string]badgerCfg.NamespaceConfig{
"foo": {
KeyDirectory: "/bad/path",
ValueDirectory: "/bad/path",
},
},
logger: zap.NewNop(),
factories: map[string]storage.Factory{"badger": factory},
}
err := ext.Start(ctx, componenttest.NewNopHost())
require.Error(t, err)
require.EqualError(t, err, "failed to initialize badger storage: Error Creating Dir: \"\" error: mkdir : no such file or directory")
})
err := ext.Start(context.Background(), componenttest.NewNopHost())
require.ErrorContains(t, err, "failed to initialize badger storage")
require.ErrorContains(t, err, "/bad/path")
}

func makeStorageExtension(t *testing.T, memstoreName string) component.Component {
extensionFactory := NewFactory()

ctx := context.Background()
telemetrySettings := component.TelemetrySettings{
func noopTelemetrySettings() component.TelemetrySettings {
return component.TelemetrySettings{
Logger: zap.L(),
TracerProvider: nooptrace.NewTracerProvider(),
MeterProvider: noopmetric.NewMeterProvider(),
}
}

func makeStorageExtenion(t *testing.T, config *Config) component.Component {
extensionFactory := NewFactory()
ctx := context.Background()
storageExtension, err := extensionFactory.CreateExtension(ctx,
extension.CreateSettings{
ID: ID,
TelemetrySettings: noopTelemetrySettings(),
BuildInfo: component.NewDefaultBuildInfo(),
},
config,
)
require.NoError(t, err)
return storageExtension
}

func startStorageExtension(t *testing.T, memstoreName string) component.Component {
config := &Config{
Memory: map[string]memoryCfg.Configuration{
memstoreName: {MaxTraces: 10000},
},
}
err := config.Validate()
require.NoError(t, err)

storageExtension, err := extensionFactory.CreateExtension(ctx, extension.CreateSettings{
ID: ID,
TelemetrySettings: telemetrySettings,
BuildInfo: component.NewDefaultBuildInfo(),
}, config)
require.NoError(t, err)
require.NoError(t, config.Validate())

host := componenttest.NewNopHost()
err = storageExtension.Start(ctx, host)
storageExtension := makeStorageExtenion(t, config)
err := storageExtension.Start(context.Background(), componenttest.NewNopHost())
require.NoError(t, err)
t.Cleanup(func() { require.NoError(t, storageExtension.Shutdown(ctx)) })

t.Cleanup(func() {
require.NoError(t, storageExtension.Shutdown(context.Background()))
})
return storageExtension
}
Loading