|
9 | 9 |
|
10 | 10 | . "github.com/onsi/ginkgo/v2" //nolint |
11 | 11 | . "github.com/onsi/gomega" //nolint |
| 12 | + "k8s.io/apimachinery/pkg/fields" |
12 | 13 | k8stypes "k8s.io/apimachinery/pkg/types" |
| 14 | + watchapi "k8s.io/apimachinery/pkg/watch" |
13 | 15 | "sigs.k8s.io/controller-runtime/pkg/client" |
14 | 16 |
|
15 | 17 | vmv1beta1 "github.com/VictoriaMetrics/operator/api/operator/v1beta1" |
@@ -48,12 +50,122 @@ func ExpectObjectStatus(ctx context.Context, |
48 | 50 | for _, cond := range obs.Status.Conditions { |
49 | 51 | conds = append(conds, fmt.Sprintf("type=%s,message=%q,generation=%d,status=%q", cond.Type, cond.Message, cond.ObservedGeneration, cond.Status)) |
50 | 52 | } |
| 53 | + if obs.Status.UpdateStatus == vmv1beta1.UpdateStatusFailed && status != vmv1beta1.UpdateStatusFailed { |
| 54 | + return StopTrying(fmt.Sprintf("object %q entered %q while waiting for %q: reason=%q,conditions=%s", |
| 55 | + name.Name, obs.Status.UpdateStatus, status, obs.Status.Reason, strings.Join(conds, ","))) |
| 56 | + } |
51 | 57 | return fmt.Errorf("not expected object status=%q, reason=%q,conditions=%s", obs.Status.UpdateStatus, obs.Status.Reason, strings.Join(conds, ",")) |
52 | 58 | } |
53 | 59 |
|
54 | 60 | return nil |
55 | 61 | } |
56 | 62 |
|
| 63 | +// WatchUntilStatusSeen reads events from w until it observes an object with the given name, |
| 64 | +// observedGeneration >= minGen, and target status, or the context deadline is reached. |
| 65 | +// Start the watch before triggering the action, then pass the post-action generation as minGen |
| 66 | +// to avoid matching stale pre-action ADDED events that share the same name and status. |
| 67 | +// If a failStatuses entry is seen (with obsGen >= minGen and status != targetStatus), |
| 68 | +// WatchUntilStatusSeen returns immediately with the failure reason rather than waiting for timeout. |
| 69 | +// All matched events are logged to GinkgoWriter for post-failure diagnostics. |
| 70 | +func WatchUntilStatusSeen(ctx context.Context, w watchapi.Interface, name string, minGen int64, targetStatus vmv1beta1.UpdateStatus, failStatuses ...vmv1beta1.UpdateStatus) error { |
| 71 | + for { |
| 72 | + select { |
| 73 | + case event, ok := <-w.ResultChan(): |
| 74 | + if !ok { |
| 75 | + return fmt.Errorf("watch closed before observing status %q for %q", targetStatus, name) |
| 76 | + } |
| 77 | + if event.Type == watchapi.Error { |
| 78 | + return fmt.Errorf("watch error: %v", event.Object) |
| 79 | + } |
| 80 | + obj, ok := event.Object.(client.Object) |
| 81 | + if !ok || obj.GetName() != name { |
| 82 | + continue |
| 83 | + } |
| 84 | + jsD, err := json.Marshal(obj) |
| 85 | + if err != nil { |
| 86 | + continue |
| 87 | + } |
| 88 | + type statusHolder struct { |
| 89 | + Status struct { |
| 90 | + vmv1beta1.StatusMetadata `json:",inline"` |
| 91 | + } `json:"status"` |
| 92 | + } |
| 93 | + var sh statusHolder |
| 94 | + if err := json.Unmarshal(jsD, &sh); err != nil { |
| 95 | + continue |
| 96 | + } |
| 97 | + obsGen := sh.Status.ObservedGeneration |
| 98 | + status := sh.Status.UpdateStatus |
| 99 | + fmt.Fprintf(GinkgoWriter, "[watch] %s name=%s gen=%d obsGen=%d status=%q", |
| 100 | + event.Type, name, obj.GetGeneration(), obsGen, status) |
| 101 | + if status == targetStatus && obsGen >= minGen { |
| 102 | + fmt.Fprintf(GinkgoWriter, " ✓\n") |
| 103 | + return nil |
| 104 | + } |
| 105 | + if obsGen >= minGen { |
| 106 | + for _, fs := range failStatuses { |
| 107 | + if status == fs { |
| 108 | + var conds []string |
| 109 | + for _, cond := range sh.Status.Conditions { |
| 110 | + conds = append(conds, fmt.Sprintf("type=%s,message=%q", cond.Type, cond.Message)) |
| 111 | + } |
| 112 | + fmt.Fprintf(GinkgoWriter, " (fail-fast)\n") |
| 113 | + return fmt.Errorf("object %q entered %q while waiting for %q: reason=%q conditions=%s", |
| 114 | + name, status, targetStatus, sh.Status.Reason, strings.Join(conds, ",")) |
| 115 | + } |
| 116 | + } |
| 117 | + fmt.Fprintf(GinkgoWriter, " (skipped: status %q != %q)\n", status, targetStatus) |
| 118 | + } else { |
| 119 | + fmt.Fprintf(GinkgoWriter, " (skipped: obsGen %d < minGen %d)\n", obsGen, minGen) |
| 120 | + } |
| 121 | + case <-ctx.Done(): |
| 122 | + return fmt.Errorf("timed out waiting for status %q for object %q: %w", targetStatus, name, ctx.Err()) |
| 123 | + } |
| 124 | + } |
| 125 | +} |
| 126 | + |
| 127 | +// WatchUntilStatusReached starts a watch on the named object and waits until it reaches targetStatus. |
| 128 | +// Uses minGen=0 (matches any generation), suitable when there is no preceding mutation to guard against. |
| 129 | +// Pass failStatuses to return immediately if an unexpected failure state is observed before targetStatus. |
| 130 | +func WatchUntilStatusReached(ctx context.Context, rclient client.WithWatch, list client.ObjectList, nsn k8stypes.NamespacedName, timeout time.Duration, targetStatus vmv1beta1.UpdateStatus, failStatuses ...vmv1beta1.UpdateStatus) error { |
| 131 | + listOpts := &client.ListOptions{ |
| 132 | + Namespace: nsn.Namespace, |
| 133 | + FieldSelector: fields.OneTermEqualSelector("metadata.name", nsn.Name), |
| 134 | + } |
| 135 | + watcher, err := rclient.Watch(ctx, list, listOpts) |
| 136 | + if err != nil { |
| 137 | + return err |
| 138 | + } |
| 139 | + defer watcher.Stop() |
| 140 | + watchCtx, cancel := context.WithTimeout(ctx, timeout) |
| 141 | + defer cancel() |
| 142 | + return WatchUntilStatusSeen(watchCtx, watcher, nsn.Name, 0, targetStatus, failStatuses...) |
| 143 | +} |
| 144 | + |
| 145 | +// WatchUntilDeleted reads events from w until it observes a DELETED event for the named |
| 146 | +// object, or the context deadline is reached. |
| 147 | +func WatchUntilDeleted(ctx context.Context, w watchapi.Interface, name string) error { |
| 148 | + for { |
| 149 | + select { |
| 150 | + case event, ok := <-w.ResultChan(): |
| 151 | + if !ok { |
| 152 | + return fmt.Errorf("watch closed before observing deletion of %q", name) |
| 153 | + } |
| 154 | + if event.Type == watchapi.Error { |
| 155 | + return fmt.Errorf("watch error: %v", event.Object) |
| 156 | + } |
| 157 | + if event.Type == watchapi.Deleted { |
| 158 | + obj, ok := event.Object.(client.Object) |
| 159 | + if ok && obj.GetName() == name { |
| 160 | + return nil |
| 161 | + } |
| 162 | + } |
| 163 | + case <-ctx.Done(): |
| 164 | + return fmt.Errorf("timed out waiting for deletion of object %q: %w", name, ctx.Err()) |
| 165 | + } |
| 166 | + } |
| 167 | +} |
| 168 | + |
57 | 169 | func CollectK8SResources() { |
58 | 170 | if !CurrentSpecReport().Failed() { |
59 | 171 | return |
|
0 commit comments