buildCopyPartitionRequest returns the partition copy request. Currently this is a naive implementation which just checks the existence of pindexes among the known node lists and then sort them based on the partition state, rack and number of total partitions/load on those nodes.
(pindexName string, stats *CopyPartitionStats,
mgr *cbgt.Manager, stopCh chan struct{})
| 91 | // of pindexes among the known node lists and then sort them based on the |
| 92 | // partition state, rack and number of total partitions/load on those nodes. |
| 93 | func buildCopyPartitionRequest(pindexName string, stats *CopyPartitionStats, |
| 94 | mgr *cbgt.Manager, stopCh chan struct{}) (*CopyPartitionRequest, error) { |
| 95 | planPIndexes, _, err := cbgt.CfgGetPlanPIndexes(mgr.Cfg()) |
| 96 | if err != nil { |
| 97 | return nil, fmt.Errorf("pindex_copy_request:"+ |
| 98 | " CfgGetPlanPIndexes err: %v", err) |
| 99 | } |
| 100 | if planPIndexes == nil { |
| 101 | return nil, fmt.Errorf("pindex_copy_request:" + |
| 102 | " skipped on nil planPIndexes") |
| 103 | } |
| 104 | |
| 105 | var planPIndex *cbgt.PlanPIndex |
| 106 | var exists bool |
| 107 | if planPIndex, exists = planPIndexes.PlanPIndexes[pindexName]; !exists || |
| 108 | len(planPIndex.Nodes) == 0 { |
| 109 | return nil, fmt.Errorf("pindex_copy_request:"+ |
| 110 | " missing pindex: %s in planPIndexes", pindexName) |
| 111 | } |
| 112 | nodeDefs, _, err := cbgt.CfgGetNodeDefs(mgr.Cfg(), cbgt.NODE_DEFS_KNOWN) |
| 113 | if err != nil { |
| 114 | return nil, err |
| 115 | } |
| 116 | |
| 117 | formerPrimary := "" |
| 118 | // get all the potential source nodes except self. |
| 119 | uuids := make([]string, len(planPIndex.Nodes)) |
| 120 | for uuid, node := range planPIndex.Nodes { |
| 121 | if node.Priority == 0 && node.CanRead && node.CanWrite { |
| 122 | formerPrimary = uuid |
| 123 | } |
| 124 | // if the current node is the primary |
| 125 | // then no file copying is needed. |
| 126 | if mgr.UUID() == formerPrimary { |
| 127 | atomic.StoreInt32(&stats.TotCopyPartitionSkipped, int32(1)) |
| 128 | return nil, nil |
| 129 | } |
| 130 | // skip self. |
| 131 | if mgr.UUID() == uuid { |
| 132 | continue |
| 133 | } |
| 134 | |
| 135 | uuids = append(uuids, uuid) |
| 136 | } |
| 137 | |
| 138 | // check explicitly whether the requested pindex resides in those |
| 139 | // nodes, and if so gives back a set of potential source node uuids. |
| 140 | sourceUUIDs := getNodesHostingPIndex(uuids, pindexName, |
| 141 | mgr.Options()["authType"], nodeDefs) |
| 142 | if len(sourceUUIDs) == 0 { |
| 143 | atomic.StoreInt32(&stats.TotCopyPartitionSkipped, int32(1)) |
| 144 | return nil, nil |
| 145 | } |
| 146 | |
| 147 | return buildCopyPartitionRequestUtil(planPIndexes, pindexName, stats, |
| 148 | nodeDefs, sourceUUIDs, formerPrimary, mgr, stopCh) |
| 149 | } |
| 150 |
no test coverage detected