forked from cockroachdb/cockroach
-
Notifications
You must be signed in to change notification settings - Fork 0
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
migration: break down pkg across a few files
We can separate out the `Helper`, `Migration`, and various utilities into their own files. We'll add tests for individual components in future commits; the physical separation here sets the foundation for doing so (prototyped in cockroachdb#57445). This commit is purely code movement. Release note: None
- Loading branch information
1 parent
5a53b00
commit fd6bd4f
Showing
5 changed files
with
178 additions
and
151 deletions.
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 |
|---|---|---|
| @@ -0,0 +1,121 @@ | ||
| // Copyright 2020 The Cockroach Authors. | ||
| // | ||
| // Use of this software is governed by the Business Source License | ||
| // included in the file licenses/BSL.txt. | ||
| // | ||
| // As of the Change Date specified in that file, in accordance with | ||
| // the Business Source License, use of this software will be governed | ||
| // by the Apache License, Version 2.0, included in the file | ||
| // licenses/APL.txt. | ||
|
|
||
| package migration | ||
|
|
||
| import ( | ||
| "context" | ||
|
|
||
| "github.com/cockroachdb/cockroach/pkg/roachpb" | ||
| "github.com/cockroachdb/cockroach/pkg/rpc" | ||
| "github.com/cockroachdb/cockroach/pkg/server/serverpb" | ||
| "github.com/cockroachdb/cockroach/pkg/util/log" | ||
| "github.com/cockroachdb/errors" | ||
| ) | ||
|
|
||
| // Helper captures all the primitives required to fully specify a migration. | ||
| type Helper struct { | ||
| *Manager | ||
| } | ||
|
|
||
| // RequiredNodeIDs returns the node IDs for all nodes that are currently part of | ||
| // the cluster (i.e. they haven't been decommissioned away). Migrations have the | ||
| // pre-requisite that all required nodes are up and running so that we're able | ||
| // to execute all relevant node-level operations on them. If any of the nodes | ||
| // are found to be unavailable, an error is returned. | ||
| func (h *Helper) RequiredNodeIDs(ctx context.Context) ([]roachpb.NodeID, error) { | ||
| var nodeIDs []roachpb.NodeID | ||
| ls, err := h.nl.GetLivenessesFromKV(ctx) | ||
| if err != nil { | ||
| return nil, err | ||
| } | ||
| for _, l := range ls { | ||
| if l.Membership.Decommissioned() { | ||
| continue | ||
| } | ||
| live, err := h.nl.IsLive(l.NodeID) | ||
| if err != nil { | ||
| return nil, err | ||
| } | ||
| if !live { | ||
| return nil, errors.Newf("n%d required, but unavailable", l.NodeID) | ||
| } | ||
| nodeIDs = append(nodeIDs, l.NodeID) | ||
| } | ||
| return nodeIDs, nil | ||
| } | ||
|
|
||
| // EveryNode invokes the given closure (named by the informational parameter op) | ||
| // across every node in the cluster. The mechanism for ensuring that we've done | ||
| // so, while accounting for the possibility of new nodes being added to the | ||
| // cluster in the interim, is provided by the following structure: | ||
| // (a) We'll retrieve the list of node IDs for all nodes in the system | ||
| // (b) For each node, we'll invoke the closure | ||
| // (c) We'll retrieve the list of node IDs again to account for the | ||
| // possibility of a new node being added during (b) | ||
| // (d) If there any discrepancies between the list retrieved in (a) | ||
| // and (c), we'll invoke the closure each node again | ||
| // (e) We'll continue to loop around until the node ID list stabilizes | ||
| // | ||
| // By the time EveryNode returns, we'll have thus invoked the closure against | ||
| // every node in the cluster. | ||
| // | ||
| // To consider one example of how this primitive is used, let's consider our use | ||
| // of it to bump the cluster version. After we return, given all nodes in the | ||
| // cluster will have their cluster versions bumped, and future node additions | ||
| // will observe the latest version (through the join RPC). This lets us author | ||
| // migrations that can assume that a certain version gate has been enabled on | ||
| // all nodes in the cluster, and will always be enabled for any new nodes in the | ||
| // system. | ||
| // | ||
| // It may be possible however that right after we return, a new node may join. | ||
| // This means that some migrations may have to be split up into two version | ||
| // bumps: one that phases out the old version (i.e. stops creation of stale data | ||
| // or behavior) and a clean-up version, which removes any vestiges of the stale | ||
| // data/behavior, and which, when active, ensures that the old data has vanished | ||
| // from the system. This is similar in spirit to how schema changes are split up | ||
| // into multiple smaller steps that are carried out sequentially. | ||
| func (h *Helper) EveryNode( | ||
| ctx context.Context, op string, fn func(context.Context, serverpb.MigrationClient) error, | ||
| ) error { | ||
| nodeIDs, err := h.RequiredNodeIDs(ctx) | ||
| if err != nil { | ||
| return err | ||
| } | ||
|
|
||
| for { | ||
| // TODO(irfansharif): We can/should send out these RPCs in parallel. | ||
| log.Infof(ctx, "executing op=%s on nodes=%s", op, nodeIDs) | ||
| for _, nodeID := range nodeIDs { | ||
| conn, err := h.dialer.Dial(ctx, nodeID, rpc.DefaultClass) | ||
| if err != nil { | ||
| return err | ||
| } | ||
| client := serverpb.NewMigrationClient(conn) | ||
| if err := fn(ctx, client); err != nil { | ||
| return err | ||
| } | ||
| } | ||
|
|
||
| curNodeIDs, err := h.RequiredNodeIDs(ctx) | ||
| if err != nil { | ||
| return err | ||
| } | ||
|
|
||
| if !identical(nodeIDs, curNodeIDs) { | ||
| nodeIDs = curNodeIDs | ||
| continue | ||
| } | ||
|
|
||
| break | ||
| } | ||
|
|
||
| return nil | ||
| } |
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