Reconcile performs one full placement sync.
(ctx context.Context, _ ctrl.Request)
| 122 | |
| 123 | // Reconcile performs one full placement sync. |
| 124 | func (r *PlacementReconciler) Reconcile(ctx context.Context, _ ctrl.Request) (ctrl.Result, error) { |
| 125 | logger := log.FromContext(ctx) |
| 126 | |
| 127 | views, err := r.gather(ctx) |
| 128 | if err != nil { |
| 129 | return ctrl.Result{}, err |
| 130 | } |
| 131 | |
| 132 | tenants := make([]TenantInfo, 0, len(views)) |
| 133 | totalHRs := 0 |
| 134 | for _, v := range views { |
| 135 | tenants = append(tenants, v.info) |
| 136 | totalHRs += v.info.Weight |
| 137 | } |
| 138 | |
| 139 | readyShards, currentShards, err := r.observeShards(ctx) |
| 140 | if err != nil { |
| 141 | return ctrl.Result{}, err |
| 142 | } |
| 143 | shardCount := r.Config.EffectiveShardCount(totalHRs, len(views), currentShards) |
| 144 | |
| 145 | desired := ComputePlacement(PlacementInput{ |
| 146 | Tenants: tenants, |
| 147 | ShardCount: shardCount, |
| 148 | Pinned: r.Config.PinnedTenants, |
| 149 | RebalanceThreshold: r.Config.RebalanceThreshold, |
| 150 | CanRebalance: func(tenant string) bool { |
| 151 | return r.now().Sub(r.lastMoved[tenant]) >= rebalanceCooldown |
| 152 | }, |
| 153 | }) |
| 154 | |
| 155 | pending, errs := r.apply(ctx, views, desired, readyShards) |
| 156 | |
| 157 | r.report(views, desired, shardCount, totalHRs, pending) |
| 158 | r.pruneCooldowns(views) |
| 159 | |
| 160 | if len(errs) > 0 { |
| 161 | return ctrl.Result{}, utilerrors.NewAggregate(errs) |
| 162 | } |
| 163 | if pending > 0 { |
| 164 | logger.V(1).Info("placement sync paced", "pendingMoves", pending) |
| 165 | return ctrl.Result{RequeueAfter: requeueShort}, nil |
| 166 | } |
| 167 | return ctrl.Result{RequeueAfter: resyncPeriod}, nil |
| 168 | } |
| 169 | |
| 170 | // gather lists tenant HelmReleases and namespaces (metadata only) and groups |
| 171 | // them per tenant. |
nothing calls this directly
no test coverage detected