MCPcopy
hub / github.com/fluid-cloudnative/fluid / CheckAndSyncWorkerStatus

Method CheckAndSyncWorkerStatus

pkg/ctrl/worker.go:164–237  ·  view source on GitHub ↗

CheckAndSyncWorkerStatus checks the worker statefulset's status and update it to runtime's status accordingly. It returns readyOrPartialReady to indicate if the worker statefulset is (partial) ready or not ready.

(getRuntimeFn func(client.Client) (base.RuntimeInterface, error), workerStsNamespacedName types.NamespacedName)

Source from the content-addressed store, hash-verified

162// CheckAndSyncWorkerStatus checks the worker statefulset's status and update it to runtime's status accordingly.
163// It returns readyOrPartialReady to indicate if the worker statefulset is (partial) ready or not ready.
164func (e *Helper) CheckAndSyncWorkerStatus(getRuntimeFn func(client.Client) (base.RuntimeInterface, error), workerStsNamespacedName types.NamespacedName) (readyOrPartialReady bool, err error) {
165 workers, err := GetWorkersAsStatefulset(e.client,
166 workerStsNamespacedName)
167 if err != nil {
168 if fluiderrs.IsDeprecated(err) {
169 e.log.Info("Warning: Deprecated mode is not support, so skip handling", "details", err)
170 readyOrPartialReady = true
171 return readyOrPartialReady, nil
172 }
173 return readyOrPartialReady, err
174 }
175
176 err = retry.RetryOnConflict(retry.DefaultBackoff, func() error {
177 runtime, err := getRuntimeFn(e.client)
178 if err != nil {
179 return err
180 }
181
182 oldStatus := runtime.GetStatus().DeepCopy()
183 statusToUpdate := runtime.GetStatus()
184 var expectReplicas int32
185 if workers.Spec.Replicas != nil {
186 expectReplicas = *workers.Spec.Replicas
187 } else {
188 expectReplicas = 0
189 }
190
191 statusToUpdate.DesiredWorkerNumberScheduled = expectReplicas
192 statusToUpdate.CurrentWorkerNumberScheduled = workers.Status.Replicas
193 statusToUpdate.WorkerNumberReady = workers.Status.ReadyReplicas
194 statusToUpdate.WorkerNumberAvailable = workers.Status.AvailableReplicas
195 statusToUpdate.WorkerNumberUnavailable = workers.Status.Replicas - workers.Status.AvailableReplicas
196 if statusToUpdate.WorkerNumberUnavailable < 0 {
197 statusToUpdate.WorkerNumberUnavailable = 0
198 }
199
200 phase := kubeclient.GetPhaseFromStatefulset(expectReplicas, *workers)
201 statusToUpdate.WorkerPhase = phase
202
203 var cond datav1alpha1.RuntimeCondition
204 if len(statusToUpdate.Conditions) == 0 {
205 statusToUpdate.Conditions = []datav1alpha1.RuntimeCondition{}
206 }
207 switch phase {
208 case datav1alpha1.RuntimePhaseReady:
209 readyOrPartialReady = true
210 cond = utils.NewRuntimeCondition(datav1alpha1.RuntimeWorkersReady, datav1alpha1.RuntimeWorkersReadyReason,
211 "The workers are ready.", corev1.ConditionTrue)
212 case datav1alpha1.RuntimePhasePartialReady:
213 readyOrPartialReady = true
214 cond = utils.NewRuntimeCondition(datav1alpha1.RuntimeWorkersReady, datav1alpha1.RuntimeWorkersReadyReason,
215 "The workers are partially ready.", corev1.ConditionTrue)
216 case datav1alpha1.RuntimePhaseNotReady:
217 cond = utils.NewRuntimeCondition(datav1alpha1.RuntimeWorkersReady, datav1alpha1.RuntimeWorkersReadyReason,
218 "The workers are not ready.", corev1.ConditionFalse)
219 }
220
221 if len(cond.Type) != 0 {

Callers 10

worker_test.goFile · 0.80
CheckWorkersReadyMethod · 0.80
CheckWorkersReadyMethod · 0.80
CheckWorkersReadyMethod · 0.80
CheckWorkersReadyMethod · 0.80
CheckWorkersReadyMethod · 0.80
CheckWorkersReadyMethod · 0.80
CheckWorkersReadyMethod · 0.80
CheckWorkersReadyMethod · 0.80
CheckWorkersReadyMethod · 0.80

Calls 7

GetPhaseFromStatefulsetFunction · 0.92
NewRuntimeConditionFunction · 0.92
UpdateRuntimeConditionFunction · 0.92
GetWorkersAsStatefulsetFunction · 0.85
UpdateMethod · 0.80
GetStatusMethod · 0.65
DeepCopyMethod · 0.45

Tested by

no test coverage detected