apply pushes the desired assignment out, pacing tenant reassignments and self-healing label stragglers without limit. Reassignments whose target shard is not ready are deferred. Returns the number of reassignments left for the next batch.
(ctx context.Context, views map[string]*tenantView, desired map[string]string, readyShards map[string]bool)
| 268 | // shard is not ready are deferred. Returns the number of reassignments left |
| 269 | // for the next batch. |
| 270 | func (r *PlacementReconciler) apply(ctx context.Context, views map[string]*tenantView, desired map[string]string, readyShards map[string]bool) (int, []error) { |
| 271 | logger := log.FromContext(ctx) |
| 272 | |
| 273 | names := make([]string, 0, len(views)) |
| 274 | for name := range views { |
| 275 | names = append(names, name) |
| 276 | } |
| 277 | sort.Strings(names) |
| 278 | |
| 279 | budget := movesPerSync |
| 280 | pending := 0 |
| 281 | var errs []error |
| 282 | for _, tenantNS := range names { |
| 283 | v := views[tenantNS] |
| 284 | target := desired[tenantNS] |
| 285 | if target == "" { |
| 286 | continue |
| 287 | } |
| 288 | moved := v.info.Current != target |
| 289 | if moved { |
| 290 | if budget == 0 || !readyShards[target] { |
| 291 | pending++ |
| 292 | continue |
| 293 | } |
| 294 | budget-- |
| 295 | } |
| 296 | if err := r.stamp(ctx, v, target); err != nil { |
| 297 | errs = append(errs, err) |
| 298 | continue |
| 299 | } |
| 300 | // Cooldown and metrics record successful moves only: a failed patch |
| 301 | // must neither delay the retry nor count twice. |
| 302 | if moved { |
| 303 | r.lastMoved[tenantNS] = r.now() |
| 304 | movesCounter.Inc() |
| 305 | logger.Info("reassigning tenant", "tenant", tenantNS, "from", v.info.Current, "to", target, "weight", v.info.Weight) |
| 306 | } |
| 307 | } |
| 308 | return pending, errs |
| 309 | } |
| 310 | |
| 311 | // stamp records the assignment on the tenant namespace and relabels every |
| 312 | // HelmRelease of the tenant that does not carry it yet. Patches the namespace |