build(deps): replace werf/kubedog-for-werf-helm with werf/kubedog (#2568)

* build(deps): replace werf/kubedog-for-werf-helm with werf/kubedog Replace the fork github.com/werf/kubedog-for-werf-helm with the upstream github.com/werf/kubedog. The fork was a temporary compatibility shim; the upstream repository now includes the necessary k8s API fixes. Signed-off-by: yxxhero <aiopsclub@163.com> * fix(kubedog): initialize InformerFactory to prevent nil pointer panic The upstream werf/kubedog now requires an InformerFactory for its resource trackers (deployment, statefulset, daemonset, job, canary), but the multitrack layer still passes nil. Bypass the broken multitrack feed layer by creating resource trackers directly with a properly initialized InformerFactory. Signed-off-by: yxxhero <aiopsclub@163.com> * fix(lint): correct misspelling of canceled Signed-off-by: yxxhero <aiopsclub@163.com> --------- Signed-off-by: yxxhero <aiopsclub@163.com>
2026-09-30 16:27:38 +02:00 · 2026-05-02 08:17:48 +08:00
parent dac42105dd
commit bda57b741f
3 changed files with 262 additions and 67 deletions
@@ -9,8 +9,14 @@ import (
 	"sync"
 	"time"

-	"github.com/werf/kubedog-for-werf-helm/pkg/tracker"
-	"github.com/werf/kubedog-for-werf-helm/pkg/trackers/rollout/multitrack"
+	"github.com/werf/kubedog/pkg/informer"
+	"github.com/werf/kubedog/pkg/tracker"
+	"github.com/werf/kubedog/pkg/tracker/canary"
+	"github.com/werf/kubedog/pkg/tracker/daemonset"
+	"github.com/werf/kubedog/pkg/tracker/deployment"
+	"github.com/werf/kubedog/pkg/tracker/job"
+	"github.com/werf/kubedog/pkg/tracker/statefulset"
+	"github.com/werf/kubedog/pkg/trackers/dyntracker/util"
 	"go.uber.org/zap"
 	"k8s.io/apimachinery/pkg/api/meta"
 	"k8s.io/client-go/discovery"
@@ -20,6 +26,7 @@ import (
 	"k8s.io/client-go/rest"
 	"k8s.io/client-go/restmapper"
 	"k8s.io/client-go/tools/clientcmd"
+	watchtools "k8s.io/client-go/tools/watch"

 	"github.com/helmfile/helmfile/pkg/resource"
 )
@@ -187,6 +194,12 @@ func getOrCreateClients(kubeContext, kubeconfig string, qps float32, burst int)
 	return cache, nil
 }

+type trackTarget struct {
+	kind      string
+	name      string
+	namespace string
+}
+
 func (t *Tracker) TrackResources(ctx context.Context, resources []*resource.Resource) error {
 	if len(resources) == 0 {
 		t.logger.Info("No resources to track")
@@ -201,81 +214,269 @@ func (t *Tracker) TrackResources(ctx context.Context, resources []*resource.Reso

 	t.logger.Infof("Tracking %d resources with kubedog (filtered from %d total)", len(filtered), len(resources))

-	specs := multitrack.MultitrackSpecs{}
+	targets := t.buildTargets(filtered)
+	if len(targets) == 0 {
+		t.logger.Info("No trackable resources found (only Deployment, StatefulSet, DaemonSet, Job, and Canary are supported)")
+		return nil
+	}

-	for _, res := range filtered {
+	t.logger.Infof("Tracking breakdown: %s", t.targetSummary(targets))
+
+	watchErrCh := make(chan error, len(targets))
+	informerFactory := informer.NewConcurrentInformerFactory(
+		ctx.Done(),
+		watchErrCh,
+		t.dynamicClient,
+		informer.ConcurrentInformerFactoryOptions{},
+	)
+
+	opts := tracker.Options{
+		ParentContext: ctx,
+		Timeout:       t.trackOptions.Timeout,
+		LogsFromTime:  time.Now().Add(-t.trackOptions.LogsSince),
+	}
+
+	var wg sync.WaitGroup
+	errCh := make(chan error, len(targets))
+
+	for _, target := range targets {
+		wg.Add(1)
+		go func(tgt trackTarget) {
+			defer wg.Done()
+			if err := t.trackSingleResource(tgt, informerFactory, opts); err != nil {
+				errCh <- fmt.Errorf("%s/%s tracking failed: %w", tgt.kind, tgt.name, err)
+			}
+		}(target)
+	}
+
+	done := make(chan struct{})
+	go func() {
+		wg.Wait()
+		close(done)
+	}()
+
+	select {
+	case err := <-errCh:
+		return fmt.Errorf("tracking failed: %w", err)
+	case <-done:
+		t.logger.Info("All resources tracked successfully")
+		return nil
+	case <-ctx.Done():
+		return fmt.Errorf("tracking canceled: %w", ctx.Err())
+	}
+}
+
+func (t *Tracker) trackSingleResource(target trackTarget, informerFactory *util.Concurrent[*informer.InformerFactory], opts tracker.Options) error {
+	parentContext := opts.ParentContext
+	if parentContext == nil {
+		parentContext = context.Background()
+	}
+	ctx, cancel := watchtools.ContextWithOptionalTimeout(parentContext, opts.Timeout)
+	defer cancel()
+
+	trackErrCh := make(chan error, 1)
+	doneCh := make(chan struct{})
+
+	switch target.kind {
+	case "deploy":
+		tr := deployment.NewTracker(target.name, target.namespace, t.clientSet, informerFactory, opts)
+		go t.runDeploymentTracker(ctx, tr, trackErrCh, doneCh)
+		return t.waitDeploymentTracker(ctx, tr, trackErrCh, doneCh)
+	case "sts":
+		tr := statefulset.NewTracker(target.name, target.namespace, t.clientSet, informerFactory, opts)
+		go t.runStatefulSetTracker(ctx, tr, trackErrCh, doneCh)
+		return t.waitStatefulSetTracker(ctx, tr, trackErrCh, doneCh)
+	case "ds":
+		tr := daemonset.NewTracker(target.name, target.namespace, t.clientSet, informerFactory, opts)
+		go t.runDaemonSetTracker(ctx, tr, trackErrCh, doneCh)
+		return t.waitDaemonSetTracker(ctx, tr, trackErrCh, doneCh)
+	case "job":
+		tr := job.NewTracker(target.name, target.namespace, t.clientSet, informerFactory, opts)
+		go t.runJobTracker(ctx, tr, trackErrCh, doneCh)
+		return t.waitJobTracker(ctx, tr, trackErrCh, doneCh)
+	case "canary":
+		tr := canary.NewTracker(target.name, target.namespace, t.clientSet, t.dynamicClient, informerFactory, opts)
+		go t.runCanaryTracker(ctx, tr, trackErrCh, doneCh)
+		return t.waitCanaryTracker(ctx, tr, trackErrCh, doneCh)
+	default:
+		return fmt.Errorf("unsupported resource kind: %s", target.kind)
+	}
+}
+
+func (t *Tracker) runDeploymentTracker(ctx context.Context, tr *deployment.Tracker, errCh chan<- error, doneCh chan<- struct{}) {
+	if err := tr.Track(ctx); err != nil {
+		errCh <- err
+	} else {
+		close(doneCh)
+	}
+}
+
+func (t *Tracker) waitDeploymentTracker(ctx context.Context, tr *deployment.Tracker, trackErrCh <-chan error, doneCh <-chan struct{}) error {
+	for {
+		select {
+		case <-tr.Ready:
+			t.logger.Debugf("Deployment %s/%s is ready", tr.Namespace, tr.ResourceName)
+			return nil
+		case status := <-tr.Failed:
+			return fmt.Errorf("deployment %s/%s failed: %s", tr.Namespace, tr.ResourceName, status.FailedReason)
+		case err := <-trackErrCh:
+			return err
+		case <-doneCh:
+			return nil
+		case <-ctx.Done():
+			return fmt.Errorf("tracking canceled for deployment %s/%s: %w", tr.Namespace, tr.ResourceName, ctx.Err())
+		}
+	}
+}
+
+func (t *Tracker) runStatefulSetTracker(ctx context.Context, tr *statefulset.Tracker, errCh chan<- error, doneCh chan<- struct{}) {
+	if err := tr.Track(ctx); err != nil {
+		errCh <- err
+	} else {
+		close(doneCh)
+	}
+}
+
+func (t *Tracker) waitStatefulSetTracker(ctx context.Context, tr *statefulset.Tracker, trackErrCh <-chan error, doneCh <-chan struct{}) error {
+	for {
+		select {
+		case <-tr.Ready:
+			t.logger.Debugf("StatefulSet %s/%s is ready", tr.Namespace, tr.ResourceName)
+			return nil
+		case status := <-tr.Failed:
+			return fmt.Errorf("statefulset %s/%s failed: %s", tr.Namespace, tr.ResourceName, status.FailedReason)
+		case err := <-trackErrCh:
+			return err
+		case <-doneCh:
+			return nil
+		case <-ctx.Done():
+			return fmt.Errorf("tracking canceled for statefulset %s/%s: %w", tr.Namespace, tr.ResourceName, ctx.Err())
+		}
+	}
+}
+
+func (t *Tracker) runDaemonSetTracker(ctx context.Context, tr *daemonset.Tracker, errCh chan<- error, doneCh chan<- struct{}) {
+	if err := tr.Track(ctx); err != nil {
+		errCh <- err
+	} else {
+		close(doneCh)
+	}
+}
+
+func (t *Tracker) waitDaemonSetTracker(ctx context.Context, tr *daemonset.Tracker, trackErrCh <-chan error, doneCh <-chan struct{}) error {
+	for {
+		select {
+		case <-tr.Ready:
+			t.logger.Debugf("DaemonSet %s/%s is ready", tr.Namespace, tr.ResourceName)
+			return nil
+		case status := <-tr.Failed:
+			return fmt.Errorf("daemonset %s/%s failed: %s", tr.Namespace, tr.ResourceName, status.FailedReason)
+		case err := <-trackErrCh:
+			return err
+		case <-doneCh:
+			return nil
+		case <-ctx.Done():
+			return fmt.Errorf("tracking canceled for daemonset %s/%s: %w", tr.Namespace, tr.ResourceName, ctx.Err())
+		}
+	}
+}
+
+func (t *Tracker) runJobTracker(ctx context.Context, tr *job.Tracker, errCh chan<- error, doneCh chan<- struct{}) {
+	if err := tr.Track(ctx); err != nil {
+		errCh <- err
+	} else {
+		close(doneCh)
+	}
+}
+
+func (t *Tracker) waitJobTracker(ctx context.Context, tr *job.Tracker, trackErrCh <-chan error, doneCh <-chan struct{}) error {
+	for {
+		select {
+		case <-tr.Succeeded:
+			t.logger.Debugf("Job %s/%s succeeded", tr.Namespace, tr.ResourceName)
+			return nil
+		case status := <-tr.Failed:
+			return fmt.Errorf("job %s/%s failed: %s", tr.Namespace, tr.ResourceName, status.FailedReason)
+		case err := <-trackErrCh:
+			return err
+		case <-doneCh:
+			return nil
+		case <-ctx.Done():
+			return fmt.Errorf("tracking canceled for job %s/%s: %w", tr.Namespace, tr.ResourceName, ctx.Err())
+		}
+	}
+}
+
+func (t *Tracker) runCanaryTracker(ctx context.Context, tr *canary.Tracker, errCh chan<- error, doneCh chan<- struct{}) {
+	if err := tr.Track(ctx); err != nil {
+		errCh <- err
+	} else {
+		close(doneCh)
+	}
+}
+
+func (t *Tracker) waitCanaryTracker(ctx context.Context, tr *canary.Tracker, trackErrCh <-chan error, doneCh <-chan struct{}) error {
+	for {
+		select {
+		case <-tr.Succeeded:
+			t.logger.Debugf("Canary %s/%s succeeded", tr.Namespace, tr.ResourceName)
+			return nil
+		case status := <-tr.Failed:
+			return fmt.Errorf("canary %s/%s failed: %s", tr.Namespace, tr.ResourceName, status.FailedReason)
+		case err := <-trackErrCh:
+			return err
+		case <-doneCh:
+			return nil
+		case <-ctx.Done():
+			return fmt.Errorf("tracking canceled for canary %s/%s: %w", tr.Namespace, tr.ResourceName, ctx.Err())
+		}
+	}
+}
+
+func (t *Tracker) buildTargets(resources []*resource.Resource) []trackTarget {
+	var targets []trackTarget
+	for _, res := range resources {
 		namespace := res.Namespace
 		if namespace == "" {
 			namespace = t.namespace
 		}

+		kind := ""
 		switch strings.ToLower(res.Kind) {
 		case "deployment", "deploy":
-			specs.Deployments = append(specs.Deployments, multitrack.MultitrackSpec{
-				ResourceName: res.Name,
-				Namespace:    namespace,
-				SkipLogs:     !t.trackOptions.Logs,
-			})
+			kind = "deploy"
 		case "statefulset", "sts":
-			specs.StatefulSets = append(specs.StatefulSets, multitrack.MultitrackSpec{
-				ResourceName: res.Name,
-				Namespace:    namespace,
-				SkipLogs:     !t.trackOptions.Logs,
-			})
+			kind = "sts"
 		case "daemonset", "ds":
-			specs.DaemonSets = append(specs.DaemonSets, multitrack.MultitrackSpec{
-				ResourceName: res.Name,
-				Namespace:    namespace,
-				SkipLogs:     !t.trackOptions.Logs,
-			})
+			kind = "ds"
 		case "job":
-			specs.Jobs = append(specs.Jobs, multitrack.MultitrackSpec{
-				ResourceName: res.Name,
-				Namespace:    namespace,
-				SkipLogs:     !t.trackOptions.Logs,
-			})
+			kind = "job"
 		case "canary":
-			specs.Canaries = append(specs.Canaries, multitrack.MultitrackSpec{
-				ResourceName: res.Name,
-				Namespace:    namespace,
-				SkipLogs:     !t.trackOptions.Logs,
-			})
+			kind = "canary"
 		default:
 			t.logger.Debugf("Skipping unsupported kind %s for resource %s/%s", res.Kind, namespace, res.Name)
+			continue
 		}
+
+		targets = append(targets, trackTarget{
+			kind:      kind,
+			name:      res.Name,
+			namespace: namespace,
+		})
 	}
+	return targets
+}

-	totalResources := len(specs.Deployments) + len(specs.StatefulSets) +
-		len(specs.DaemonSets) + len(specs.Jobs) + len(specs.Canaries)
-
-	if totalResources == 0 {
-		t.logger.Info("No trackable resources found (only Deployment, StatefulSet, DaemonSet, Job, and Canary are supported)")
-		return nil
+func (t *Tracker) targetSummary(targets []trackTarget) string {
+	counts := make(map[string]int)
+	for _, tgt := range targets {
+		counts[tgt.kind]++
 	}
-
-	t.logger.Infof("Tracking breakdown: Deployments=%d, StatefulSets=%d, DaemonSets=%d, Jobs=%d, Canaries=%d",
-		len(specs.Deployments), len(specs.StatefulSets), len(specs.DaemonSets),
-		len(specs.Jobs), len(specs.Canaries))
-
-	opts := multitrack.MultitrackOptions{
-		Options: tracker.Options{
-			ParentContext: ctx,
-			Timeout:       t.trackOptions.Timeout,
-			LogsFromTime:  time.Now().Add(-t.trackOptions.LogsSince),
-		},
-		StatusProgressPeriod: 5 * time.Second,
-		DynamicClient:        t.dynamicClient,
-		DiscoveryClient:      t.discovery,
-		Mapper:               t.mapper,
+	parts := make([]string, 0, len(counts))
+	for kind, count := range counts {
+		parts = append(parts, fmt.Sprintf("%ss=%d", kind, count))
 	}
-
-	err := multitrack.Multitrack(t.clientSet, specs, opts)
-	if err != nil {
-		return fmt.Errorf("tracking failed: %w", err)
-	}
-
-	t.logger.Info("All resources tracked successfully")
-	return nil
+	return strings.Join(parts, ", ")
 }

 func (t *Tracker) filterResources(resources []*resource.Resource) []*resource.Resource {