-
Notifications
You must be signed in to change notification settings - Fork 3.7k
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.
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
Druid Server Guilds and Guild Aware Replication #10739
Conversation
* Drop will always happen on default guild to retain guild distribution. | ||
*/ | ||
@Test | ||
public void testDropMultipleGuilds() |
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.
This test fails in CI despite never failing locally. Needs to be looked into
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.
After stubbing out a load queue peon method I was able to get this test working.
This pull request has been marked as stale due to 60 days of inactivity. It will be closed in 4 weeks if no further activity occurs. If you think that's incorrect or this pull request should instead be reviewed, please simply write any comment. Even if closed, you can still revive the PR at any time or discuss it on the dev@druid.apache.org list. Thank you for your contributions. |
dont close |
This issue is no longer marked as stale. |
closing in favor of future attempt |
Fixes #9816
Description
Note: I use the term "Guild" throughout this proposal. It is an arbitrary name that I chose for the Historical grouping construct that I am proposing. A Guild would be a group of Historical servers. All servers who do not specify a Guild in runtime.properties would be assigned to a default Guild.
I am adding the idea of guilds in Druid. A guild is a logical grouping of servers. With this PR, the only use of guilds is replicant distribution across Historical Servers. The idea has been born out of the desire for HDFS like rack aware replication for bare metal deployment on-prem. For this use case, Druid Historical services are assigned a guild based on the physical rack that they live on. The coordinator uses SegmentReplicantLookup to build a lookup for segment:guild replicant count. The Coordinator has a preference for loading replicants across 2 or more guilds. It is important to note that, instead of having less replicants than specified by Druid Load Rules, the Coordinator will load replicants on the same guild if it must.
This idea can go beyond just physical racks in a data center and apply to things such as availability zones or arbitrary historical server groupings in a virtualized deployment. That is why I came up with the name "guild" instead of just saying rack explicitly.
Implementation
Configs
runtime.properties:
druid.server.guild
The STRING guild name assigned to a server. Default value applied if not specified,_default_guild
.druid.coordinator.guildReplication.on
The BOOLEAN flag telling the coordinator if it should use server guilds as a part of its coordination logic. Default value isfalse
.coordinator dynamic config:
guildReplicaitonMaxPercentOfSegmentsToMove
What % of segments moved during BalanceSegments duty should be dedicated to moving segments that are not meeting guild replication threshold of 2? This is applied against what is left over after moving segments off of decommissioning servers, if there are any. Default value is0
.emitGuildReplicationMetrics
Boolean whether or not to emit metrics for guild replication status. This PR adds one metric that is a count of segments not loaded onto multiple guilds.Code Changes
The coordinator currently builds this object each time the
HistoricalManagementDuties
run. Prior to this proposal, the object would contain two main data structures:Table<SegmentId, String, Integer> segmentsInCluster
- replicant count for each segment by tier for segments served by the cluster.Table<SegmentId, String, Integer> loadingSegments
- replicant count for each segment by tier for segments being loaded by the cluster.My proposal adds a third data structure that is specific to guild replication:
Table<SegmentId, String, Integer> historicalGuildDistribution
- replicant count for each segment by guild.This new structure is only created if guild replication is enabled. It is not worth the resources if we are not going to use it! The structure is used for quick lookup to get guild replication state for a given segment. It will be used by coordinator duties when making decisions on loading/moving/dropping segments.
When Assigning replicas, use the changes in SegmentReplicantLookup in order to split
ServerHolders
into groups of servers who are on a guild that is serving a replicant of the segment and servers who are on a guild that is not serving a replicant of the segment. If possibleLoadRule
will load replicants to the best scored server(s) from the guild(s) not serving a replicant. However, we always fallback to just loading the specified number of replicants even if replication across guilds cannot be achieved.When picking replicants to drop, we will also split the
ServerHolders
in the same way. The segments will be dropped from decommissioning servers first, then from servers on guilds with > 1 replicant of the segment, then lastly from the remaining servers serving a replicant.Add a method to the interface.
pickGuildReplicationViolatingSegmentToMove(List<ServerHolder> serverHolders, Set<String> broadcastDataSources, DruidCoordinatorRuntimeParams params);
This new method is added so we have the
SegmentReplicantLookup
information needed to pick a segment who is not replicated across > 1 guild when balancing the cluster. It is needed because we introduce the dynamic config and associated balancing phase in BalanceSegments that prioritizes the balancing of segments who are not properly replicated across guilds. We need to only pick segments that meet the requirements. The existingpickSegmentToMove
does not suffice.RandomBalancerStrategy and CostBalancerStrategy add implementations.
Adds a method for getting a
SegmentHolder
that is violator of the goal of being on > 1 guilds. This method needs to have access to theSegmentReplicantLookup
in order to quickly look up replication state of segments it is possibly selecting. It returns the first violator that it finds or null if none is found.The coordinator balancing duty gets a couple of changes. The first change is to the generic balancing that exists today. If guild replication is enabled, then we will perform the split of
ServerHolders
based on their guild replication status when looking for a server to move a segment to. Just as inLoadRule
we will do our best to make the move to a server that improves or maintains the number of guilds that hold a replicant.We also add a new phase of balancing segments. There is a dedicated move for segments off of decommissioning servers. An operator can also choose to add a dedicated move for segments that are not replicated on > 1 guild. They do this by editing the dynamic coordinator config that this proposal adds. This results in the coordinator moving a certain number of segments that are violating guild replication rules. It is an optional way for an operator to speed up the balancing of segments across guilds.
Design Choice Rationale + Alternatives
There has been some feedback that this functionality should be folded into the tier construct that already exists. I tend to disagree with that idea for multiple reasons.
So now that adapting tiering is ruled out as far as I am concerned, what else could we do? I can't think of other solutions for the configurations that are as generic as my simple guild name assignment and boolean for the coordinator on whether or not to follow guild replication logic paths. But what about the implementation details using these configurations? It could certainly be argued that the cost of the new
SegmentReplicantLookup
data structure is too steep for large clusters. However, I fail to come up with a better solution in my head. When we are choosing servers to load/drop segments from, we need a quick way to look up details on where that segment is loaded. It seems logical that this facility would require this large structure in the replicant lookup.With all this being said, I am open to improvements to my proposal. At the end of the day my motivating factors just need to be achieved. If there is a better way to achieve the simple goals required, I'm all for it.
Operational impact of deployment
- Is anything going to be deprecated or removed by this change? How will we phase out old behavior?
Nothing is deprecated or removed.
- Is there a migration path that cluster operators need to be aware of?
To go from not using guild replication, to using guild replication, there will be a simple migration path. Operators will need to assign guilds to their historical servers and restart them. They will then have to flip the coordinator config to turn on guild replication. They may choose to use the coordinator dynamic config to speed up balancing based on guild replication if they so choose.
- Will there be any effect on the ability to do a rolling upgrade, or to do a rolling downgrade if an operator wants to switch back to a previous version?
Upgrading to the first version of druid that has guild replication in it will not require any special work. Upgrading to the default configs will keep guild replication off. If they choose to use guild replication, they can perform the migration steps after the upgrade. If they choose not to use guild replication, no action is needed! Downgrading would require a pre-step by an operator if they have already turned guild replication on. They would need to turn guild replication off on the coordinator by updating their config and restarting it. That way there will not be any issues with split versions across coordinator/historical during downgrade.
- Other
This change as proposed results in additional resource utilization by the Coordinator. Extra data structures are created in
SegmentReplicantLookup
that help make replicant loading decisions based on guild distribution of segments. This may require attention from operators as far as configuring their runtime environment for the coordinator.The changed as proposed introduces a major change to the decision making process by the Coordinator. Enabling guild aware replication may result in replicant loading decisions that the coordinator would have previously considered sub-optimal. Loading replicants to different guilds to gain better uptime and maintainability standards could result in segment distribution skew that negatively impacts performance. It is a trade-off that an operator will need to carefully consider.
This PR has:
Key changed/added classes in this PR
SegmentReplicantLookup
CoordinatorDynamicConfig
BalanceSegments
LoadRule
ReserviorSegmentSampler
BalancerStrategy
interfaceDruidCoordinatorConfig
EmitClusterStatsAndMetrics
DruidServer