upstream/helm
1
0
Fork 0
mirror of https://github.com/helm/helm.git synced 2026-05-28 04:35:48 -04:00
Code Issues Projects Releases Packages Wiki Activity Actions 3

status function

Browse source 
Signed-off-by: Austin Abro <AustinAbro321@gmail.com>
This commit is contained in:
Austin Abro 2025-01-06 13:06:54 +00:00
parent b8bdcc3a2b
commit ac9012577a
No known key found for this signature in database
GPG key ID: 92EB5159E403F9D6
2 changed files with 19 additions and 34 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

50
pkg/kube/statuswait.go
View file

@ -69,22 +69,7 @@ func (w *statusWaiter) waitForDelete(ctx context.Context, resourceList ResourceL
}
eventCh := w.sw.Watch(cancelCtx, resources, watcher.Options{})
statusCollector := collector.NewResourceStatusCollector(resources)
done := statusCollector.ListenWithObserver(eventCh, collector.ObserverFunc(
func(statusCollector *collector.ResourceStatusCollector, _ event.Event) {
rss := []*event.ResourceStatus{}
for _, rs := range statusCollector.ResourceStatuses {
if rs == nil {
continue
}
rss = append(rss, rs)
}
desired := status.NotFoundStatus
if aggregator.AggregateStatus(rss, desired) == desired {
cancel()
return
}
}),
)
done := statusCollector.ListenWithObserver(eventCh, statusObserver(cancel, status.NotFoundStatus))
<-done
if statusCollector.Error != nil {
@ -140,22 +125,7 @@ func (w *statusWaiter) wait(ctx context.Context, resourceList ResourceList, wait
}
eventCh := w.sw.Watch(cancelCtx, resources, watcher.Options{})
statusCollector := collector.NewResourceStatusCollector(resources)
done := statusCollector.ListenWithObserver(eventCh, collector.ObserverFunc(
func(statusCollector *collector.ResourceStatusCollector, _ event.Event) {
rss := []*event.ResourceStatus{}
for _, rs := range statusCollector.ResourceStatuses {
if rs == nil {
continue
}
rss = append(rss, rs)
}
desired := status.CurrentStatus
if aggregator.AggregateStatus(rss, desired) == desired {
cancel()
return
}
}),
)
done := statusCollector.ListenWithObserver(eventCh, statusObserver(cancel, status.CurrentStatus))
<-done
if statusCollector.Error != nil {
@ -177,3 +147,19 @@ func (w *statusWaiter) wait(ctx context.Context, resourceList ResourceList, wait
}
return nil
}
func statusObserver(cancel context.CancelFunc, desired status.Status) collector.ObserverFunc {
return func(statusCollector *collector.ResourceStatusCollector, _ event.Event) {
rss := []*event.ResourceStatus{}
for _, rs := range statusCollector.ResourceStatuses {
if rs == nil {
continue
}
rss = append(rss, rs)
}
if aggregator.AggregateStatus(rss, desired) == desired {
cancel()
return
}
}
}

3
pkg/kube/statuswait_test.go
View file

@ -19,7 +19,6 @@ package kube // import "helm.sh/helm/v3/pkg/kube"
import (
"context"
"errors"
"log"
"testing"
"time"
@ -246,7 +245,7 @@ func TestStatusWait(t *testing.T) {
}
kwaiter := statusWaiter{
sw: statusWatcher,
log: log.Printf,
log: t.Logf,
}
resourceList := ResourceList{}