[WIP] Refactor storage factories to hold one configuration #6156
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -36,16 +36,15 @@ import ( | |
| ) | ||
|
|
||
| const ( | ||
| PrimaryNamespace = "es" | ||
| ArchiveNamespace = "es-archive" | ||
| ) | ||
|
|
||
| var ( // interface comformance checks | ||
|
There was a problem hiding this comment. need to be very careful about removing ArchiveFactory interface, because query service uses it via runtime cast, so unless we have integration tests (many of them disable archive tests as I recall) you can introduce a breaking change There was a problem hiding this comment. @yurishkuro yep - noted. I was thinking to avoid breaking changes we can remove the ArchiveFactory within this PR and use an archive storage factory wherever needed. There was a problem hiding this comment. https://github.com/jaegertracing/jaeger/blob/main/plugin/storage/integration/elasticsearch_test.go#L165 - ES doesn't skip the archive test |
||
| _ storage.Factory = (*Factory)(nil) | ||
| _ io.Closer = (*Factory)(nil) | ||
| _ plugin.Configurable = (*Factory)(nil) | ||
| _ storage.Purger = (*Factory)(nil) | ||
| ) | ||
|
|
||
| // Factory implements storage.Factory for Elasticsearch backend. | ||
|
|
@@ -58,26 +57,25 @@ type Factory struct { | |
|
|
||
| newClientFn func(c *config.Configuration, logger *zap.Logger, metricsFactory metrics.Factory) (es.Client, error) | ||
|
|
||
| config *config.Configuration | ||
|
|
||
| client atomic.Pointer[es.Client] | ||
|
|
||
| watchers []*fswatcher.FSWatcher | ||
| } | ||
|
|
||
| // NewFactory creates a new Factory. | ||
| func NewFactory(namespace string) *Factory { | ||
| return &Factory{ | ||
| Options: NewOptions(namespace), | ||
| newClientFn: config.NewClient, | ||
| tracer: otel.GetTracerProvider(), | ||
| } | ||
| } | ||
|
|
||
| func NewFactoryWithConfig( | ||
| cfg config.Configuration, | ||
| namespace string, | ||
mahadzaryab1 marked this conversation as resolved.
Show resolved
Hide resolved
|
||
| metricsFactory metrics.Factory, | ||
| logger *zap.Logger, | ||
| ) (*Factory, error) { | ||
|
|
@@ -88,19 +86,12 @@ func NewFactoryWithConfig( | |
| defaultConfig := DefaultConfig() | ||
| cfg.ApplyDefaults(&defaultConfig) | ||
|
|
||
| f := NewFactory(namespace) | ||
| f.configureFromOptions(&Options{ | ||
| Primary: namespaceConfig{ | ||
| Configuration: cfg, | ||
| namespace: namespace, | ||
| }, | ||
| }) | ||
| err := f.Initialize(metricsFactory, logger) | ||
| if err != nil { | ||
|
|
@@ -123,96 +114,55 @@ func (f *Factory) InitFromViper(v *viper.Viper, _ *zap.Logger) { | |
| // configureFromOptions configures factory from Options struct. | ||
| func (f *Factory) configureFromOptions(o *Options) { | ||
| f.Options = o | ||
| f.config = f.Options.GetPrimary() | ||
|
There was a problem hiding this comment.
|
||
| } | ||
|
|
||
| // Initialize implements storage.Factory. | ||
| func (f *Factory) Initialize(metricsFactory metrics.Factory, logger *zap.Logger) error { | ||
| f.metricsFactory, f.logger = metricsFactory, logger | ||
|
|
||
| primaryClient, err := f.newClientFn(f.config, logger, metricsFactory) | ||
| if err != nil { | ||
| return fmt.Errorf("failed to create primary Elasticsearch client: %w", err) | ||
| } | ||
| f.client.Store(&primaryClient) | ||
|
|
||
| if f.config.Authentication.BasicAuthentication.PasswordFilePath != "" { | ||
| primaryWatcher, err := fswatcher.New([]string{f.config.Authentication.BasicAuthentication.PasswordFilePath}, f.onPrimaryPasswordChange, f.logger) | ||
| if err != nil { | ||
| return fmt.Errorf("failed to create watcher for primary ES client's password: %w", err) | ||
| } | ||
| f.watchers = append(f.watchers, primaryWatcher) | ||
| } | ||
|
|
||
| return nil | ||
| } | ||
|
|
||
| func (f *Factory) getPrimaryClient() es.Client { | ||
| if c := f.client.Load(); c != nil { | ||
| return *c | ||
| } | ||
| return nil | ||
| } | ||
|
|
||
| // CreateSpanReader implements storage.Factory | ||
| func (f *Factory) CreateSpanReader() (spanstore.Reader, error) { | ||
| return createSpanReader(f.getPrimaryClient, f.config, f.metricsFactory, f.logger, f.tracer) | ||
| } | ||
|
|
||
| // CreateSpanWriter implements storage.Factory | ||
| func (f *Factory) CreateSpanWriter() (spanstore.Writer, error) { | ||
| return createSpanWriter(f.getPrimaryClient, f.config, false, f.metricsFactory, f.logger) | ||
| } | ||
|
|
||
| // CreateDependencyReader implements storage.Factory | ||
| func (f *Factory) CreateDependencyReader() (dependencystore.Reader, error) { | ||
| return createDependencyReader(f.getPrimaryClient, f.config, f.logger) | ||
| } | ||
|
|
||
| func createSpanReader( | ||
| clientFn func() es.Client, | ||
| cfg *config.Configuration, | ||
| mFactory metrics.Factory, | ||
| logger *zap.Logger, | ||
| tp trace.TracerProvider, | ||
|
|
@@ -229,7 +179,7 @@ func createSpanReader( | |
| ServiceIndex: cfg.Indices.Services, | ||
| TagDotReplacement: cfg.Tags.DotReplacement, | ||
| UseReadWriteAliases: cfg.UseReadWriteAliases, | ||
| Archive: cfg.IsArchive, | ||
| RemoteReadClusters: cfg.RemoteReadClusters, | ||
| Logger: logger, | ||
| MetricsFactory: mFactory, | ||
|
|
@@ -287,16 +237,16 @@ func (f *Factory) CreateSamplingStore(int /* maxBuckets */) (samplingstore.Store | |
| params := esSampleStore.Params{ | ||
| Client: f.getPrimaryClient, | ||
| Logger: f.logger, | ||
| IndexPrefix: f.config.Indices.IndexPrefix, | ||
| IndexDateLayout: f.config.Indices.Sampling.DateLayout, | ||
| IndexRolloverFrequency: config.RolloverFrequencyAsNegativeDuration(f.config.Indices.Sampling.RolloverFrequency), | ||
| Lookback: f.config.AdaptiveSamplingLookback, | ||
| MaxDocCount: f.config.MaxDocCount, | ||
| } | ||
| store := esSampleStore.NewSamplingStore(params) | ||
|
|
||
| if f.config.CreateIndexTemplates && !f.config.UseILM { | ||
| mappingBuilder := mappingBuilderFromConfig(f.config) | ||
| samplingMapping, err := mappingBuilder.GetSamplingMappings() | ||
| if err != nil { | ||
| return nil, err | ||
|
|
@@ -344,19 +294,12 @@ func (f *Factory) Close() error { | |
| errs = append(errs, w.Close()) | ||
| } | ||
| errs = append(errs, f.getPrimaryClient().Close()) | ||
|
|
||
| return errors.Join(errs...) | ||
| } | ||
|
|
||
| func (f *Factory) onPrimaryPasswordChange() { | ||
| f.onClientPasswordChange(f.config, &f.client) | ||
| } | ||
|
|
||
| func (f *Factory) onClientPasswordChange(cfg *config.Configuration, client *atomic.Pointer[es.Client]) { | ||
|
|
||
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -118,7 +118,7 @@ func (*Factory) getFactoryOfType(factoryType string) (storage.Factory, error) { | |
| case cassandraStorageType: | ||
| return cassandra.NewFactory(), nil | ||
| case elasticsearchStorageType, opensearchStorageType: | ||
| return es.NewFactory(es.PrimaryNamespace), nil | ||
|
There was a problem hiding this comment. @yurishkuro can you take another look at the changes? I minimized the code movements to simply remove the There was a problem hiding this comment. see #6156 (comment) The query service instantiates just a single factory and casts it to ArchiveFactory. With your changes (which are in the right direction, but of insufficient scope) it never gets a chance to create archive CLI flags, because that has to happen via different factories. There was a problem hiding this comment. @yurishkuro is the archive factory only needed for the query service? |
||
| case memoryStorageType: | ||
| return memory.NewFactory(), nil | ||
| case kafkaStorageType: | ||
|
|
||
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
is this still used? Let's add some comments, because this is not the first time I have this question, and the flag doesn't really make sense to me.