Add common ring configuration #4617
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
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -11,6 +11,7 @@ import ( | |
| "github.com/pkg/errors" | ||
|
|
||
| "github.com/grafana/loki/pkg/storage/chunk/cache" | ||
| "github.com/grafana/loki/pkg/util" | ||
| "github.com/grafana/loki/pkg/util/cfg" | ||
|
|
||
| loki_storage "github.com/grafana/loki/pkg/storage" | ||
|
|
@@ -76,11 +77,13 @@ func (c *ConfigWrapper) ApplyDynamicConfig() cfg.Source { | |
| r.QueryScheduler.UseSchedulerRing = true | ||
| } | ||
|
|
||
| applyPathPrefixDefaults(r, &defaults) | ||
|
|
||
| applyDynamicRingConfigs(r, &defaults) | ||
|
|
||
| if err := applyTokensFilePath(r); err != nil { | ||
| return err | ||
| } | ||
|
|
||
| if err := applyStorageConfig(r, &defaults); err != nil { | ||
| return err | ||
|
|
@@ -96,85 +99,141 @@ func (c *ConfigWrapper) ApplyDynamicConfig() cfg.Source { | |
| } | ||
| } | ||
|
|
||
| // applyDynamicRingConfigs checks the current config and, depending on the given values, reuse ring configs accordingly. | ||
| // | ||
| // 1. Gives preference to any explicit ring config set. For instance, if the user explicitly configures Distributor's ring, | ||
|
There was a problem hiding this comment. I think it should be There was a problem hiding this comment. Agreed. I can imagine users being confused due to their different rings not using default values listed in the documentation (instead, such rings would be using the ingester config, which they might not even realize). |
||
| // that config will prevail. This rule is enforced by the fact that the config file and command line args are parsed | ||
| // again after the dynamic config has been applied, so will take higher precedence. | ||
| // 2. If no explicit ring config is set, use the common ring configured if provided. | ||
| // 3. If no common ring was provided, use the memberlist config if provided. | ||
| // 4. If no common ring or memberlist were provided, use the ingester's ring configuration. | ||
|
|
||
| // When using the ingester or common ring config, the loopback interface will be appended to the end of | ||
| // the list of default interface names | ||
| func applyDynamicRingConfigs(r, defaults *ConfigWrapper) { | ||
| if reflect.DeepEqual(r.Common.Ring, defaults.Common.Ring) { | ||
| // common ring not set, so use memberlist for all rings if set | ||
| if len(r.MemberlistKV.JoinMembers) > 0 { | ||
| applyMemberlistConfig(r) | ||
DylanGuedes marked this conversation as resolved.
Show resolved
Hide resolved
|
||
| } else { | ||
| // neither common ring nor memberlist set, use ingester ring configuration for all rings | ||
| useIngesterRingConfig(r, defaults) | ||
| } | ||
| } else { | ||
| // common ring is provided, use that for all rings | ||
| useCommonRingConfig(r, defaults) | ||
| } | ||
trevorwhitney marked this conversation as resolved.
Show resolved
Hide resolved
|
||
|
|
||
| } | ||
|
|
||
| func useCommonRingConfig(r, defaults *ConfigWrapper) { | ||
| // if the default ingester config is detected, then it's safe to override with the common config | ||
| shouldOverrideIngester := reflect.DeepEqual(r.Ingester.LifecyclerConfig, defaults.Ingester.LifecyclerConfig) | ||
|
There was a problem hiding this comment. @owen-d per your comment on @DylanGuedes's original PR, would you like to see this There was a problem hiding this comment. That's a good point. I don't think it's necessary 👍 There was a problem hiding this comment. Hey, since in such a scenario the configs would be applied again, maybe this whole There was a problem hiding this comment. hmm, and just always override the ingester. that's a good point. I believe we have test coverage for that case so I'll try it out |
||
|
|
||
| // append the loopback after checking if the default is enabled, as it mutates the current | ||
| appendLoopbackInterface(r, defaults) | ||
|
|
||
| numTokens := 128 //this is the default used by the ingester and ruler | ||
|
There was a problem hiding this comment. @owen-d per your comment on @DylanGuedes's original PR, do you think we should use 512 here instead? There was a problem hiding this comment. I don't think we should be overwriting these at all. Extra fields (i.e. num_tokens) can still be reconfigured by setting the ingester ring tokens manually, but I don't think we should be overriding them here. There was a problem hiding this comment. ok, so just leave that one blank for now, I'm in agreement with that. |
||
| applyConfigToRings(r, r.Common.Ring, numTokens, shouldOverrideIngester) | ||
| } | ||
|
|
||
| //applyConfigToRings will reuse a given RingConfig everywhere else we have a ring configured. | ||
| func applyConfigToRings(r *ConfigWrapper, rc util.RingConfig, defaultNumTokens int, overrideIngester bool) { | ||
| //Ingester | ||
| if overrideIngester { | ||
| r.Ingester.LifecyclerConfig.RingConfig.KVStore = rc.KVStore | ||
| r.Ingester.LifecyclerConfig.HeartbeatPeriod = rc.HeartbeatPeriod | ||
| r.Ingester.LifecyclerConfig.RingConfig.HeartbeatTimeout = rc.HeartbeatTimeout | ||
| r.Ingester.LifecyclerConfig.TokensFilePath = rc.TokensFilePath | ||
| r.Ingester.LifecyclerConfig.RingConfig.ZoneAwarenessEnabled = rc.ZoneAwarenessEnabled | ||
| r.Ingester.LifecyclerConfig.ID = rc.InstanceID | ||
| r.Ingester.LifecyclerConfig.InfNames = rc.InstanceInterfaceNames | ||
| r.Ingester.LifecyclerConfig.Port = rc.InstancePort | ||
| r.Ingester.LifecyclerConfig.Addr = rc.InstanceAddr | ||
| r.Ingester.LifecyclerConfig.Zone = rc.InstanceZone | ||
| r.Ingester.LifecyclerConfig.ListenPort = rc.ListenPort | ||
| r.Ingester.LifecyclerConfig.ObservePeriod = rc.ObservePeriod | ||
| } | ||
|
|
||
| // Distributor | ||
| r.Distributor.DistributorRing.HeartbeatTimeout = rc.HeartbeatTimeout | ||
| r.Distributor.DistributorRing.HeartbeatPeriod = rc.HeartbeatPeriod | ||
| r.Distributor.DistributorRing.InstancePort = rc.InstancePort | ||
| r.Distributor.DistributorRing.InstanceAddr = rc.InstanceAddr | ||
| r.Distributor.DistributorRing.InstanceID = rc.InstanceID | ||
| r.Distributor.DistributorRing.InstanceInterfaceNames = rc.InstanceInterfaceNames | ||
| r.Distributor.DistributorRing.KVStore.Store = rc.KVStore.Store | ||
| r.Distributor.DistributorRing.KVStore.StoreConfig = rc.KVStore.StoreConfig | ||
|
|
||
| // Ruler | ||
| r.Ruler.Ring.HeartbeatTimeout = rc.HeartbeatTimeout | ||
| r.Ruler.Ring.HeartbeatPeriod = rc.HeartbeatPeriod | ||
| r.Ruler.Ring.InstancePort = rc.InstancePort | ||
| r.Ruler.Ring.InstanceAddr = rc.InstanceAddr | ||
| r.Ruler.Ring.InstanceID = rc.InstanceID | ||
| r.Ruler.Ring.InstanceInterfaceNames = rc.InstanceInterfaceNames | ||
| r.Ruler.Ring.NumTokens = defaultNumTokens | ||
| r.Ruler.Ring.KVStore.Store = rc.KVStore.Store | ||
| r.Ruler.Ring.KVStore.StoreConfig = rc.KVStore.StoreConfig | ||
|
|
||
| // Query Scheduler | ||
| r.QueryScheduler.SchedulerRing.HeartbeatTimeout = rc.HeartbeatTimeout | ||
| r.QueryScheduler.SchedulerRing.HeartbeatPeriod = rc.HeartbeatPeriod | ||
| r.QueryScheduler.SchedulerRing.InstancePort = rc.InstancePort | ||
| r.QueryScheduler.SchedulerRing.InstanceAddr = rc.InstanceAddr | ||
| r.QueryScheduler.SchedulerRing.InstanceID = rc.InstanceID | ||
| r.QueryScheduler.SchedulerRing.InstanceInterfaceNames = rc.InstanceInterfaceNames | ||
| r.QueryScheduler.SchedulerRing.InstanceZone = rc.InstanceZone | ||
| r.QueryScheduler.SchedulerRing.ZoneAwarenessEnabled = rc.ZoneAwarenessEnabled | ||
| r.QueryScheduler.SchedulerRing.KVStore.Store = rc.KVStore.Store | ||
| r.QueryScheduler.SchedulerRing.KVStore.StoreConfig = rc.KVStore.StoreConfig | ||
|
|
||
| // Compactor | ||
| r.CompactorConfig.CompactorRing.HeartbeatTimeout = rc.HeartbeatTimeout | ||
| r.CompactorConfig.CompactorRing.HeartbeatPeriod = rc.HeartbeatPeriod | ||
| r.CompactorConfig.CompactorRing.InstancePort = rc.InstancePort | ||
| r.CompactorConfig.CompactorRing.InstanceAddr = rc.InstanceAddr | ||
| r.CompactorConfig.CompactorRing.InstanceID = rc.InstanceID | ||
| r.CompactorConfig.CompactorRing.InstanceInterfaceNames = rc.InstanceInterfaceNames | ||
| r.CompactorConfig.CompactorRing.InstanceZone = rc.InstanceZone | ||
| r.CompactorConfig.CompactorRing.ZoneAwarenessEnabled = rc.ZoneAwarenessEnabled | ||
| r.CompactorConfig.CompactorRing.KVStore.Store = rc.KVStore.Store | ||
| r.CompactorConfig.CompactorRing.KVStore.StoreConfig = rc.KVStore.StoreConfig | ||
| } | ||
|
|
||
| // useIngesterRingConfig will override all other rings to use the ingester's ring configuration | ||
| func useIngesterRingConfig(r, defaults *ConfigWrapper) { | ||
| appendLoopbackInterface(r, defaults) | ||
|
|
||
| ingesterRingCfg := util.CortexLifecyclerConfigToRingConfig(r.Ingester.LifecyclerConfig) | ||
| numTokens := r.Ingester.LifecyclerConfig.NumTokens | ||
|
|
||
| // we're using the ingester config here, so never need to override it | ||
|
There was a problem hiding this comment. Hey, I was thinking since you are passing the edit: In conclusion, maybe we can remove the whole third param There was a problem hiding this comment. yeah, I like that idea, I'm going to look into removing the whole third param |
||
| applyConfigToRings(r, ingesterRingCfg, numTokens, false) | ||
| } | ||
|
|
||
| func applyTokensFilePath(cfg *ConfigWrapper) error { | ||
| // Ingester | ||
| f, err := tokensFile(cfg, "ingester.tokens") | ||
| if err != nil { | ||
| return err | ||
| } | ||
| cfg.Ingester.LifecyclerConfig.TokensFilePath = f | ||
|
|
||
| // Compactor | ||
| f, err = tokensFile(cfg, "compactor.tokens") | ||
| if err != nil { | ||
| return err | ||
| } | ||
| cfg.CompactorConfig.CompactorRing.TokensFilePath = f | ||
|
|
||
| // Query Scheduler | ||
| f, err = tokensFile(cfg, "scheduler.tokens") | ||
| if err != nil { | ||
| return err | ||
| } | ||
| cfg.QueryScheduler.SchedulerRing.TokensFilePath = f | ||
|
|
||
| return nil | ||
| } | ||
|
|
||
|
|
@@ -193,7 +252,7 @@ func tokensFile(cfg *ConfigWrapper, file string) (string, error) { | |
| return cfg.Common.PathPrefix + "/" + file, nil | ||
| } | ||
|
|
||
| func applyPathPrefixDefaults(r, defaults *ConfigWrapper) { | ||
| if r.Common.PathPrefix != "" { | ||
| prefix := strings.TrimSuffix(r.Common.PathPrefix, "/") | ||
|
|
||
|
|
@@ -211,26 +270,41 @@ func applyPathPrefixDefaults(r *ConfigWrapper, defaults ConfigWrapper) { | |
| } | ||
| } | ||
|
|
||
| // appendLoopbackInterface will append the loopback interface to the interface names used for the ingester ring, | ||
| // v2 frontend, and common ring config unless an explicit list of names was provided. | ||
| func appendLoopbackInterface(cfg, defaults *ConfigWrapper) { | ||
| loopbackIface, err := loki_net.LoopbackInterfaceName() | ||
| if err != nil { | ||
| return | ||
| } | ||
|
|
||
| if reflect.DeepEqual(cfg.Ingester.LifecyclerConfig.InfNames, defaults.Ingester.LifecyclerConfig.InfNames) { | ||
| cfg.Ingester.LifecyclerConfig.InfNames = append(cfg.Ingester.LifecyclerConfig.InfNames, loopbackIface) | ||
| } | ||
|
|
||
| if reflect.DeepEqual(cfg.Frontend.FrontendV2.InfNames, defaults.Frontend.FrontendV2.InfNames) { | ||
| cfg.Frontend.FrontendV2.InfNames = append(cfg.Config.Frontend.FrontendV2.InfNames, loopbackIface) | ||
| } | ||
|
|
||
| if reflect.DeepEqual(cfg.Common.Ring.InstanceInterfaceNames, defaults.Common.Ring.InstanceInterfaceNames) { | ||
| cfg.Common.Ring.InstanceInterfaceNames = append(cfg.Common.Ring.InstanceInterfaceNames, loopbackIface) | ||
| } | ||
|
|
||
| if reflect.DeepEqual(cfg.CompactorConfig.CompactorRing.InstanceInterfaceNames, defaults.CompactorConfig.CompactorRing.InstanceInterfaceNames) { | ||
| cfg.CompactorConfig.CompactorRing.InstanceInterfaceNames = append(cfg.CompactorConfig.CompactorRing.InstanceInterfaceNames, loopbackIface) | ||
| } | ||
| } | ||
|
|
||
| // applyMemberlistConfig will change the default ingester, distributor, ruler, and query scheduler ring configurations to use memberlist. | ||
| // The idea here is that if a user explicitly configured the memberlist configuration section, they probably want to be using memberlist | ||
| // for all their ring configurations. Since a user can still explicitly override a specific ring configuration | ||
trevorwhitney marked this conversation as resolved.
Show resolved
Hide resolved
|
||
| // (for example, use consul for the distributor), it seems harmless to take a guess at better defaults here. | ||
| func applyMemberlistConfig(r *ConfigWrapper) { | ||
| r.Ingester.LifecyclerConfig.RingConfig.KVStore.Store = memberlistStr | ||
| r.Distributor.DistributorRing.KVStore.Store = memberlistStr | ||
| r.Ruler.Ring.KVStore.Store = memberlistStr | ||
| r.QueryScheduler.SchedulerRing.KVStore.Store = memberlistStr | ||
| r.CompactorConfig.CompactorRing.KVStore.Store = memberlistStr | ||
| } | ||
|
|
||
| var ErrTooManyStorageConfigs = errors.New("too many storage configs provided in the common config, please only define one storage backend") | ||
|
|
||
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.
Btw, since
applyPathPrefixDefaultsis being invoked beforeapplyDynamicRingConfigs, the checkreflect.DeepEqual(r.CompactorConfig.CompactorRing, defaults.CompactorConfig.CompactorRing)will always be false I thinkedit: my bad, I meant
applyTokensFilePathinstead ofapplyPathPrefixDefaultsThere 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.
does
applyPathPrefixDefaultsmodify the compactor's ring? I see it modifying the working directory, but I'm missing the impact on the ring?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.
Sorry, I completely confused the function names 😂 I was referring to
applyTokensFilePath