feat: kubedog integration with unified resource handling (#2383)

* feat: add kubedog-based resource tracking integration

Add kubedog tracking as an alternative to Helm's --wait flag with:
- Real-time deployment progress tracking
- Container log streaming
- Fine-grained resource filtering (trackKinds/skipKinds/trackResources)

Features:
- New pkg/resource package for unified manifest parsing and filtering
- New pkg/kubedog package wrapping kubedog library
- CLI flags: --track-mode, --track-timeout, --track-logs
- Helmfile YAML support for trackMode, trackTimeout, trackLogs, trackKinds, skipKinds, trackResources
- Case-insensitive kind matching for filtering
- Multi-context support with proper kubeconfig/kubeContext handling

Tracking supports: Deployment, StatefulSet, DaemonSet, Job

Resource filtering priority (highest to lowest):
1. trackResources - explicit resource whitelist
2. skipKinds - blacklist specific kinds
3. trackKinds - whitelist specific kinds

Integration:
- Disable Helm --wait when using kubedog tracking
- Track after successful Helm sync/apply
- Respect release.Namespace as fallback for resources without namespace
- Use getKubeContext() for correct cluster targeting

Tests:
- Unit tests for resource filtering and kubedog options
- Integration test with httpbin chart
- E2E snapshot tests for YAML serialization
- Documentation in docs/advanced-features.md

Signed-off-by: yxxhero <aiopsclub@163.com>

* fix: address PR #2383 review comments (round 4)

1. resource/filter.go: Skip empty whitelist entries in matchWhitelist
   - At least one field (kind/name/namespace) must be specified
   - Prevents matching all resources with empty TrackResources entries

2. config/apply.go: Add ValidateConfig for track-mode validation
   - Validate --track-mode must be 'helm' or 'kubedog'
   - Reject invalid values like --track-mode foo

3. config/sync.go: Add ValidateConfig for track-mode validation
   - Same validation as apply command
   - Ensures consistent behavior across commands

Signed-off-by: yxxhero <aiopsclub@163.com>

---------

Signed-off-by: yxxhero <aiopsclub@163.com>
This commit is contained in:
yxxhero
2026-03-02 17:15:12 +08:00
committed by GitHub
parent c70bd04f3a
commit 6e21671228
29 changed files with 1550 additions and 17 deletions
+14 -1
View File
@@ -808,7 +808,14 @@ func (a *App) loadDesiredStateFromYamlWithBaseDir(file string, baseDir string, o
valsRuntime: a.valsRuntime,
}
return ld.Load(file, op)
st, err := ld.Load(file, op)
if err != nil {
return nil, err
}
st.SetKubeconfig(a.Kubeconfig)
return st, nil
}
type helmKey struct {
@@ -1762,6 +1769,9 @@ Do you really want to apply?
HideNotes: c.HideNotes(),
TakeOwnership: c.TakeOwnership(),
SyncReleaseLabels: c.SyncReleaseLabels(),
TrackMode: c.TrackMode(),
TrackTimeout: c.TrackTimeout(),
TrackLogs: c.TrackLogs(),
}
return subst.SyncReleases(&affectedReleases, helm, c.Values(), c.Concurrency(), syncOpts)
}))
@@ -2227,6 +2237,9 @@ Do you really want to sync?
TakeOwnership: c.TakeOwnership(),
SkipSchemaValidation: c.SkipSchemaValidation(),
SyncReleaseLabels: c.SyncReleaseLabels(),
TrackMode: c.TrackMode(),
TrackTimeout: c.TrackTimeout(),
TrackLogs: c.TrackLogs(),
}
return subst.SyncReleases(&affectedReleases, helm, c.Values(), c.Concurrency(), opts)
}))
-1
View File
@@ -80,7 +80,6 @@ func TestSync(t *testing.T) {
}
syncErr := app.Sync(applyConfig{
// if we check log output, concurrency must be 1. otherwise the test becomes non-deterministic.
concurrency: tc.concurrency,
logger: logger,
skipDiffOnInstall: tc.skipDiffOnInstall,
+15
View File
@@ -2380,6 +2380,9 @@ type applyConfig struct {
takeOwnership bool
syncReleaseLabels bool
enforceNeedsAreInstalled bool
trackMode string
trackTimeout int
trackLogs bool
// template-only options
includeCRDs, skipTests bool
@@ -2590,6 +2593,18 @@ func (a applyConfig) SyncReleaseLabels() bool {
return a.syncReleaseLabels
}
func (a applyConfig) TrackMode() string {
return a.trackMode
}
func (a applyConfig) TrackTimeout() int {
return a.trackTimeout
}
func (a applyConfig) TrackLogs() bool {
return a.trackLogs
}
type depsConfig struct {
skipRepos bool
includeTransitiveNeeds bool
+7
View File
@@ -87,6 +87,10 @@ type ApplyConfigProvider interface {
DAGConfig
TrackMode() string
TrackTimeout() int
TrackLogs() bool
concurrencyConfig
interactive
loggingConfig
@@ -119,6 +123,9 @@ type SyncConfigProvider interface {
IncludeTransitiveNeeds() bool
SyncReleaseLabels() bool
TrackMode() string
TrackTimeout() int
TrackLogs() bool
DAGConfig
+11
View File
@@ -0,0 +1,11 @@
package cluster
import (
"github.com/helmfile/helmfile/pkg/resource"
)
type (
Resource = resource.Resource
FilterConfig = resource.FilterConfig
ResourceFilter = resource.ResourceFilter
)
+30
View File
@@ -1,5 +1,7 @@
package config
import "fmt"
// ApplyOptoons is the options for the apply command
type ApplyOptions struct {
// Set is a list of key value pairs to be merged into the command
@@ -75,6 +77,12 @@ type ApplyOptions struct {
TakeOwnership bool
SyncReleaseLabels bool
// TrackMode specifies whether to use 'helm' or 'kubedog' for tracking resources
TrackMode string
// TrackTimeout specifies timeout for kubedog tracking (in seconds)
TrackTimeout int
// TrackLogs enables log streaming with kubedog
TrackLogs bool
}
// NewApply creates a new Apply
@@ -280,3 +288,25 @@ func (a *ApplyImpl) TakeOwnership() bool {
func (a *ApplyImpl) SyncReleaseLabels() bool {
return a.ApplyOptions.SyncReleaseLabels
}
// TrackMode returns the track mode.
func (a *ApplyImpl) TrackMode() string {
return a.ApplyOptions.TrackMode
}
// TrackTimeout returns the track timeout.
func (a *ApplyImpl) TrackTimeout() int {
return a.ApplyOptions.TrackTimeout
}
// TrackLogs returns the track logs flag.
func (a *ApplyImpl) TrackLogs() bool {
return a.ApplyOptions.TrackLogs
}
func (a *ApplyImpl) ValidateConfig() error {
if a.ApplyOptions.TrackMode != "" && a.ApplyOptions.TrackMode != "helm" && a.ApplyOptions.TrackMode != "kubedog" {
return fmt.Errorf("--track-mode must be 'helm' or 'kubedog', got: %s", a.ApplyOptions.TrackMode)
}
return a.GlobalImpl.ValidateConfig()
}
+30
View File
@@ -1,5 +1,7 @@
package config
import "fmt"
// SyncOptions is the options for the build command
type SyncOptions struct {
// Set is the set flag
@@ -48,6 +50,12 @@ type SyncOptions struct {
TakeOwnership bool
// SyncReleaseLabels is the sync release labels flag
SyncReleaseLabels bool
// TrackMode specifies whether to use 'helm' or 'kubedog' for tracking resources
TrackMode string
// TrackTimeout specifies timeout for kubedog tracking (in seconds)
TrackTimeout int
// TrackLogs enables log streaming with kubedog
TrackLogs bool
}
// NewSyncOptions creates a new Apply
@@ -187,3 +195,25 @@ func (t *SyncImpl) TakeOwnership() bool {
func (t *SyncImpl) SyncReleaseLabels() bool {
return t.SyncOptions.SyncReleaseLabels
}
// TrackMode returns the track mode.
func (t *SyncImpl) TrackMode() string {
return t.SyncOptions.TrackMode
}
// TrackTimeout returns the track timeout.
func (t *SyncImpl) TrackTimeout() int {
return t.SyncOptions.TrackTimeout
}
// TrackLogs returns the track logs flag.
func (t *SyncImpl) TrackLogs() bool {
return t.SyncOptions.TrackLogs
}
func (t *SyncImpl) ValidateConfig() error {
if t.SyncOptions.TrackMode != "" && t.SyncOptions.TrackMode != "helm" && t.SyncOptions.TrackMode != "kubedog" {
return fmt.Errorf("--track-mode must be 'helm' or 'kubedog', got: %s", t.SyncOptions.TrackMode)
}
return t.GlobalImpl.ValidateConfig()
}
+43
View File
@@ -0,0 +1,43 @@
package kubedog
import (
"time"
"github.com/helmfile/helmfile/pkg/resource"
)
type TrackMode string
const (
TrackModeHelm TrackMode = "helm"
TrackModeKubedog TrackMode = "kubedog"
)
type TrackOptions struct {
Timeout time.Duration
Logs bool
LogsSince time.Duration
Filter *resource.FilterConfig
}
func NewTrackOptions() *TrackOptions {
return &TrackOptions{
Timeout: 5 * time.Minute,
LogsSince: 10 * time.Minute,
}
}
func (o *TrackOptions) WithTimeout(timeout time.Duration) *TrackOptions {
o.Timeout = timeout
return o
}
func (o *TrackOptions) WithLogs(logs bool) *TrackOptions {
o.Logs = logs
return o
}
func (o *TrackOptions) WithFilterConfig(config *resource.FilterConfig) *TrackOptions {
o.Filter = config
return o
}
+200
View File
@@ -0,0 +1,200 @@
package kubedog
import (
"context"
"fmt"
"os"
"strings"
"sync"
"time"
"github.com/werf/kubedog/pkg/kube"
"github.com/werf/kubedog/pkg/tracker"
"github.com/werf/kubedog/pkg/trackers/rollout/multitrack"
"go.uber.org/zap"
"k8s.io/client-go/kubernetes"
"github.com/helmfile/helmfile/pkg/resource"
)
type cacheKey struct {
kubeContext string
kubeconfig string
}
var (
kubeInitMu sync.Mutex
clientCache = make(map[cacheKey]kubernetes.Interface)
)
type Tracker struct {
logger *zap.SugaredLogger
clientSet kubernetes.Interface
trackOptions *TrackOptions
filter *resource.ResourceFilter
namespace string
}
type TrackerConfig struct {
Logger *zap.SugaredLogger
Namespace string
KubeContext string
Kubeconfig string
TrackOptions *TrackOptions
}
func NewTracker(config *TrackerConfig) (*Tracker, error) {
logger := config.Logger
if logger == nil {
logger = zap.NewNop().Sugar()
}
kubeconfig := config.Kubeconfig
if kubeconfig == "" {
kubeconfig = os.Getenv("KUBECONFIG")
}
clientSet, err := getOrCreateClient(config.KubeContext, kubeconfig)
if err != nil {
return nil, fmt.Errorf("failed to initialize kubernetes client: %w", err)
}
options := config.TrackOptions
if options == nil {
options = NewTrackOptions()
}
var filter *resource.ResourceFilter
if options.Filter != nil {
filter = resource.NewResourceFilter(options.Filter, logger)
}
return &Tracker{
logger: logger,
clientSet: clientSet,
trackOptions: options,
filter: filter,
namespace: config.Namespace,
}, nil
}
func getOrCreateClient(kubeContext, kubeconfig string) (kubernetes.Interface, error) {
key := cacheKey{
kubeContext: kubeContext,
kubeconfig: kubeconfig,
}
kubeInitMu.Lock()
defer kubeInitMu.Unlock()
if client, ok := clientCache[key]; ok {
return client, nil
}
initOpts := kube.InitOptions{
KubeConfigOptions: kube.KubeConfigOptions{
Context: kubeContext,
ConfigPath: kubeconfig,
},
}
if err := kube.Init(initOpts); err != nil {
return nil, err
}
client := kube.Kubernetes
clientCache[key] = client
return client, nil
}
func (t *Tracker) TrackResources(ctx context.Context, resources []*resource.Resource) error {
if len(resources) == 0 {
t.logger.Info("No resources to track")
return nil
}
filtered := t.filterResources(resources)
if len(filtered) == 0 {
t.logger.Info("No resources to track after filtering")
return nil
}
t.logger.Infof("Tracking %d resources with kubedog (filtered from %d total)", len(filtered), len(resources))
specs := multitrack.MultitrackSpecs{}
for _, res := range filtered {
namespace := res.Namespace
if namespace == "" {
namespace = t.namespace
}
switch strings.ToLower(res.Kind) {
case "deployment", "deploy":
specs.Deployments = append(specs.Deployments, multitrack.MultitrackSpec{
ResourceName: res.Name,
Namespace: namespace,
SkipLogs: !t.trackOptions.Logs,
})
case "statefulset", "sts":
specs.StatefulSets = append(specs.StatefulSets, multitrack.MultitrackSpec{
ResourceName: res.Name,
Namespace: namespace,
SkipLogs: !t.trackOptions.Logs,
})
case "daemonset", "ds":
specs.DaemonSets = append(specs.DaemonSets, multitrack.MultitrackSpec{
ResourceName: res.Name,
Namespace: namespace,
SkipLogs: !t.trackOptions.Logs,
})
case "job":
specs.Jobs = append(specs.Jobs, multitrack.MultitrackSpec{
ResourceName: res.Name,
Namespace: namespace,
SkipLogs: !t.trackOptions.Logs,
})
default:
t.logger.Debugf("Skipping unsupported kind %s for resource %s/%s", res.Kind, namespace, res.Name)
}
}
if len(specs.Deployments)+len(specs.StatefulSets)+len(specs.DaemonSets)+len(specs.Jobs) == 0 {
t.logger.Info("No trackable resources found (only Deployment, StatefulSet, DaemonSet, and Job are supported)")
return nil
}
opts := multitrack.MultitrackOptions{
Options: tracker.Options{
ParentContext: ctx,
Timeout: t.trackOptions.Timeout,
LogsFromTime: time.Now().Add(-t.trackOptions.LogsSince),
},
StatusProgressPeriod: 5 * time.Second,
}
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
}
func (t *Tracker) filterResources(resources []*resource.Resource) []*resource.Resource {
if t.filter == nil {
return resources
}
var result []*resource.Resource
for _, res := range resources {
if t.filter.ShouldTrack(res) {
result = append(result, res)
} else {
t.logger.Debugf("Skipping resource %s/%s (kind: %s) based on configuration", res.Namespace, res.Name, res.Kind)
}
}
return result
}
+88
View File
@@ -0,0 +1,88 @@
package kubedog
import (
"testing"
"time"
"github.com/stretchr/testify/assert"
"github.com/helmfile/helmfile/pkg/resource"
)
func TestTrackMode(t *testing.T) {
assert.Equal(t, "helm", string(TrackModeHelm))
assert.Equal(t, "kubedog", string(TrackModeKubedog))
}
func TestNewTrackOptions(t *testing.T) {
opts := NewTrackOptions()
assert.NotNil(t, opts)
assert.Equal(t, 5*time.Minute, opts.Timeout)
assert.Equal(t, false, opts.Logs)
assert.Equal(t, 10*time.Minute, opts.LogsSince)
}
func TestTrackOptions_WithTimeout(t *testing.T) {
opts := NewTrackOptions()
opts = opts.WithTimeout(10 * time.Second)
assert.Equal(t, 10*time.Second, opts.Timeout)
}
func TestTrackOptions_WithLogs(t *testing.T) {
opts := NewTrackOptions()
opts = opts.WithLogs(true)
assert.True(t, opts.Logs)
}
func TestTrackOptions_Chaining(t *testing.T) {
opts := NewTrackOptions()
opts = opts.
WithTimeout(20 * time.Second).
WithLogs(true)
assert.Equal(t, 20*time.Second, opts.Timeout)
assert.True(t, opts.Logs)
}
func TestResource(t *testing.T) {
res := &resource.Resource{
Name: "test-resource",
Namespace: "test-ns",
Kind: "deployment",
}
assert.Equal(t, "test-resource", res.Name)
assert.Equal(t, "test-ns", res.Namespace)
assert.Equal(t, "deployment", res.Kind)
}
func TestTrackerConfig(t *testing.T) {
config := &TrackerConfig{
Logger: nil,
Namespace: "test-ns",
KubeContext: "test-ctx",
Kubeconfig: "/test/kubeconfig",
TrackOptions: NewTrackOptions(),
}
assert.NotNil(t, config)
assert.Equal(t, "test-ns", config.Namespace)
assert.Equal(t, "test-ctx", config.KubeContext)
assert.Equal(t, "/test/kubeconfig", config.Kubeconfig)
assert.NotNil(t, config.TrackOptions)
}
func TestTrackOptions_WithFilterConfig(t *testing.T) {
opts := NewTrackOptions()
filter := &resource.FilterConfig{
TrackKinds: []string{"Deployment", "StatefulSet"},
SkipKinds: []string{"ConfigMap"},
}
opts = opts.WithFilterConfig(filter)
assert.NotNil(t, opts.Filter)
assert.Equal(t, []string{"Deployment", "StatefulSet"}, opts.Filter.TrackKinds)
assert.Equal(t, []string{"ConfigMap"}, opts.Filter.SkipKinds)
}
+94
View File
@@ -0,0 +1,94 @@
package resource
import (
"strings"
"go.uber.org/zap"
)
type ResourceFilter struct {
logger *zap.SugaredLogger
config *FilterConfig
skipKinds map[string]bool
trackKinds map[string]bool
}
func NewResourceFilter(config *FilterConfig, logger *zap.SugaredLogger) *ResourceFilter {
f := &ResourceFilter{
config: config,
logger: logger,
}
if config != nil {
f.skipKinds = make(map[string]bool)
for _, kind := range config.SkipKinds {
f.skipKinds[strings.ToLower(kind)] = true
}
f.trackKinds = make(map[string]bool)
for _, kind := range config.TrackKinds {
f.trackKinds[strings.ToLower(kind)] = true
}
}
return f
}
func (f *ResourceFilter) Filter(resources []Resource) []Resource {
if f.config == nil {
return resources
}
var filtered []Resource
for i := range resources {
if f.ShouldTrack(&resources[i]) {
filtered = append(filtered, resources[i])
} else if f.logger != nil {
res := resources[i]
f.logger.Debugf("Skipping resource %s/%s (kind: %s) based on configuration", res.Namespace, res.Name, res.Kind)
}
}
return filtered
}
func (f *ResourceFilter) ShouldTrack(r *Resource) bool {
if f.config == nil {
return true
}
if len(f.config.TrackResources) > 0 {
return f.matchWhitelist(r)
}
kindLower := strings.ToLower(r.Kind)
if f.skipKinds[kindLower] {
return false
}
if len(f.trackKinds) > 0 {
return f.trackKinds[kindLower]
}
return true
}
func (f *ResourceFilter) matchWhitelist(r *Resource) bool {
for _, tr := range f.config.TrackResources {
// At least one field must be specified for a match
if tr.Kind == "" && tr.Name == "" && tr.Namespace == "" {
continue
}
if tr.Kind != "" && !strings.EqualFold(tr.Kind, r.Kind) {
continue
}
if tr.Name != "" && tr.Name != r.Name {
continue
}
if tr.Namespace != "" && tr.Namespace != r.Namespace {
continue
}
return true
}
return false
}
+169
View File
@@ -0,0 +1,169 @@
package resource
import (
"testing"
"github.com/stretchr/testify/assert"
"go.uber.org/zap"
)
func TestResourceFilter_Filter(t *testing.T) {
resources := []Resource{
{Kind: "Deployment", Name: "app1", Namespace: "default"},
{Kind: "StatefulSet", Name: "db1", Namespace: "default"},
{Kind: "ConfigMap", Name: "cm1", Namespace: "default"},
{Kind: "Secret", Name: "sec1", Namespace: "kube-system"},
}
tests := []struct {
name string
config *FilterConfig
expected int
}{
{
name: "nil filter returns all",
config: nil,
expected: 4,
},
{
name: "TrackKinds whitelist",
config: &FilterConfig{
TrackKinds: []string{"Deployment", "StatefulSet"},
},
expected: 2,
},
{
name: "SkipKinds blacklist",
config: &FilterConfig{
SkipKinds: []string{"ConfigMap", "Secret"},
},
expected: 2,
},
{
name: "TrackResources whitelist by kind and namespace",
config: &FilterConfig{
TrackResources: []Resource{
{Kind: "Deployment", Namespace: "default"},
},
},
expected: 1,
},
{
name: "TrackResources whitelist by name",
config: &FilterConfig{
TrackResources: []Resource{
{Name: "app1"},
{Name: "db1"},
},
},
expected: 2,
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
filter := NewResourceFilter(tt.config, zap.NewNop().Sugar())
filtered := filter.Filter(resources)
assert.Equal(t, tt.expected, len(filtered))
})
}
}
func TestResourceFilter_ShouldTrack(t *testing.T) {
logger := zap.NewNop().Sugar()
tests := []struct {
name string
config *FilterConfig
resource *Resource
expected bool
}{
{
name: "nil config tracks all",
config: nil,
resource: &Resource{Kind: "Deployment", Name: "app1"},
expected: true,
},
{
name: "TrackKinds matches",
config: &FilterConfig{
TrackKinds: []string{"Deployment"},
},
resource: &Resource{Kind: "Deployment", Name: "app1"},
expected: true,
},
{
name: "TrackKinds no match",
config: &FilterConfig{
TrackKinds: []string{"StatefulSet"},
},
resource: &Resource{Kind: "Deployment", Name: "app1"},
expected: false,
},
{
name: "SkipKinds matches",
config: &FilterConfig{
SkipKinds: []string{"ConfigMap"},
},
resource: &Resource{Kind: "ConfigMap", Name: "cm1"},
expected: false,
},
{
name: "SkipKinds no match",
config: &FilterConfig{
SkipKinds: []string{"ConfigMap"},
},
resource: &Resource{Kind: "Deployment", Name: "app1"},
expected: true,
},
{
name: "TrackResources whitelist matches all criteria",
config: &FilterConfig{
TrackResources: []Resource{
{Kind: "Deployment", Name: "app1", Namespace: "default"},
},
},
resource: &Resource{Kind: "Deployment", Name: "app1", Namespace: "default"},
expected: true,
},
{
name: "TrackResources whitelist partial match",
config: &FilterConfig{
TrackResources: []Resource{
{Kind: "Deployment", Name: "app1"},
},
},
resource: &Resource{Kind: "Deployment", Name: "app1", Namespace: "other"},
expected: true,
},
{
name: "TrackResources whitelist no match",
config: &FilterConfig{
TrackResources: []Resource{
{Kind: "Deployment", Name: "app2"},
},
},
resource: &Resource{Kind: "Deployment", Name: "app1"},
expected: false,
},
{
name: "TrackResources takes precedence over TrackKinds",
config: &FilterConfig{
TrackResources: []Resource{
{Kind: "Deployment", Name: "app1"},
},
TrackKinds: []string{"StatefulSet"},
},
resource: &Resource{Kind: "Deployment", Name: "app1"},
expected: true,
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
filter := NewResourceFilter(tt.config, logger)
result := filter.ShouldTrack(tt.resource)
assert.Equal(t, tt.expected, result)
})
}
}
+62
View File
@@ -0,0 +1,62 @@
package resource
import (
"bytes"
"errors"
"fmt"
"io"
"go.uber.org/zap"
"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
k8syaml "k8s.io/apimachinery/pkg/util/yaml"
)
func ParseManifest(manifest []byte, defaultNamespace string, logger *zap.SugaredLogger) ([]Resource, error) {
var resources []Resource
decoder := k8syaml.NewYAMLOrJSONDecoder(bytes.NewReader(manifest), 4096)
for {
var obj unstructured.Unstructured
err := decoder.Decode(&obj)
if err != nil {
if errors.Is(err, io.EOF) {
break
}
return nil, fmt.Errorf("failed to decode manifest: %w", err)
}
if len(obj.Object) == 0 {
continue
}
kind := obj.GetKind()
if kind == "" {
if logger != nil {
logger.Debugf("Skipping resource without kind")
}
continue
}
name := obj.GetName()
if name == "" {
if logger != nil {
logger.Debugf("Skipping %s resource without name", kind)
}
continue
}
namespace := obj.GetNamespace()
if namespace == "" {
namespace = defaultNamespace
}
resources = append(resources, Resource{
Kind: kind,
Name: name,
Namespace: namespace,
})
}
return resources, nil
}
+13
View File
@@ -0,0 +1,13 @@
package resource
type Resource struct {
Kind string
Name string
Namespace string
}
type FilterConfig struct {
TrackKinds []string
SkipKinds []string
TrackResources []Resource
}
+202
View File
@@ -1,18 +1,22 @@
package state
import (
"context"
"fmt"
"os"
"path/filepath"
"slices"
"sort"
"strings"
"time"
"github.com/helmfile/chartify"
"helm.sh/helm/v4/pkg/storage/driver"
"github.com/helmfile/helmfile/pkg/helmexec"
"github.com/helmfile/helmfile/pkg/kubedog"
"github.com/helmfile/helmfile/pkg/remote"
"github.com/helmfile/helmfile/pkg/resource"
)
type Dependency struct {
@@ -165,6 +169,10 @@ func (st *HelmState) appendSuppressOutputLineRegexFlags(flags []string, release
}
func (st *HelmState) appendWaitForJobsFlags(flags []string, release *ReleaseSpec, ops *SyncOpts) []string {
if st.shouldUseKubedog(release, ops) {
return flags
}
switch {
case release.WaitForJobs != nil && *release.WaitForJobs:
flags = append(flags, "--wait-for-jobs")
@@ -173,10 +181,26 @@ func (st *HelmState) appendWaitForJobsFlags(flags []string, release *ReleaseSpec
case release.WaitForJobs == nil && st.HelmDefaults.WaitForJobs:
flags = append(flags, "--wait-for-jobs")
}
return flags
}
func (st *HelmState) shouldUseKubedog(release *ReleaseSpec, ops *SyncOpts) bool {
trackMode := release.TrackMode
if trackMode == "" && ops != nil && ops.TrackMode != "" {
trackMode = ops.TrackMode
}
if trackMode == "" {
trackMode = st.HelmDefaults.TrackMode
}
return trackMode == "kubedog"
}
func (st *HelmState) appendWaitFlags(flags []string, release *ReleaseSpec, ops *SyncOpts) []string {
if st.shouldUseKubedog(release, ops) {
return flags
}
switch {
case release.Wait != nil && *release.Wait:
flags = append(flags, "--wait")
@@ -423,3 +447,181 @@ func (st *HelmState) PrepareChartify(helm helmexec.Interface, release *ReleaseSp
return nil, clean, nil
}
func (st *HelmState) trackWithKubedog(ctx context.Context, release *ReleaseSpec, helm helmexec.Interface, ops *SyncOpts) error {
timeout := 5 * time.Minute
if release.TrackTimeout != nil && *release.TrackTimeout > 0 {
timeout = time.Duration(*release.TrackTimeout) * time.Second
} else if ops != nil && ops.TrackTimeout > 0 {
timeout = time.Duration(ops.TrackTimeout) * time.Second
}
trackLogs := release.TrackLogs != nil && *release.TrackLogs
if release.TrackLogs == nil && ops != nil {
trackLogs = ops.TrackLogs
}
filterConfig := &resource.FilterConfig{
TrackKinds: release.TrackKinds,
SkipKinds: release.SkipKinds,
TrackResources: convertTrackResources(release.TrackResources),
}
kubeContext := st.getKubeContext(release)
trackOpts := kubedog.NewTrackOptions().
WithTimeout(timeout).
WithLogs(trackLogs).
WithFilterConfig(filterConfig)
tracker, err := kubedog.NewTracker(&kubedog.TrackerConfig{
Logger: st.logger,
Namespace: release.Namespace,
KubeContext: kubeContext,
Kubeconfig: st.kubeconfig,
TrackOptions: trackOpts,
})
if err != nil {
return fmt.Errorf("failed to create kubedog tracker: %w", err)
}
resources, err := st.getReleaseResources(ctx, release, helm)
if err != nil {
return fmt.Errorf("failed to get release resources: %w", err)
}
if len(resources) == 0 {
st.logger.Infof("No trackable resources found for release %s", release.Name)
return nil
}
st.logger.Infof("Tracking %d resources from release %s with kubedog", len(resources), release.Name)
if err := tracker.TrackResources(ctx, resources); err != nil {
return fmt.Errorf("kubedog tracking failed for release %s: %w", release.Name, err)
}
return nil
}
func (st *HelmState) getReleaseResources(_ context.Context, release *ReleaseSpec, helm helmexec.Interface) ([]*resource.Resource, error) {
st.logger.Debugf("Getting resources for release %s", release.Name)
manifest, namespace, err := st.getReleaseManifest(release, helm)
if err != nil {
return nil, fmt.Errorf("failed to get release manifest: %w", err)
}
if len(manifest) == 0 {
st.logger.Infof("No manifest found for release %s", release.Name)
return nil, nil
}
defaultNs := namespace
if defaultNs == "" {
defaultNs = "default"
}
resources, err := resource.ParseManifest(manifest, defaultNs, st.logger)
if err != nil {
return nil, fmt.Errorf("failed to parse release resources from manifest: %w", err)
}
if len(resources) == 0 {
st.logger.Infof("No resources found in manifest for release %s", release.Name)
return nil, nil
}
st.logger.Infof("Found %d resources in manifest for release %s", len(resources), release.Name)
result := make([]*resource.Resource, len(resources))
for i := range resources {
result[i] = &resources[i]
}
return result, nil
}
func (st *HelmState) getReleaseManifest(release *ReleaseSpec, helm helmexec.Interface) ([]byte, string, error) {
var tempDir string
var err error
if st.tempDir != nil {
tempDir, err = st.tempDir("", "helmfile-template-")
} else {
tempDir, err = os.MkdirTemp("", "helmfile-template-")
}
if err != nil {
return nil, "", fmt.Errorf("failed to create temp directory: %w", err)
}
defer func() {
if err := os.RemoveAll(tempDir); err != nil {
st.logger.Warnf("Failed to remove temp directory %s: %v", tempDir, err)
}
}()
releaseCopy := *release
st.ApplyOverrides(&releaseCopy)
flags, files, err := st.flagsForTemplate(helm, &releaseCopy, 0, &TemplateOpts{})
if err != nil {
return nil, "", fmt.Errorf("failed to generate template flags: %w", err)
}
defer st.removeFiles(files)
flags = append(flags, "--output-dir", tempDir)
if err := helm.TemplateRelease(releaseCopy.Name, releaseCopy.ChartPathOrName(), flags...); err != nil {
return nil, "", fmt.Errorf("failed to run helm template: %w", err)
}
var manifest []byte
err = filepath.Walk(tempDir, func(path string, info os.FileInfo, err error) error {
if err != nil {
return err
}
if info.IsDir() {
return nil
}
if !strings.HasSuffix(info.Name(), ".yaml") && !strings.HasSuffix(info.Name(), ".yml") {
return nil
}
content, err := os.ReadFile(path)
if err != nil {
return fmt.Errorf("failed to read file %s: %w", path, err)
}
if len(manifest) > 0 {
manifest = append(manifest, []byte("\n---\n")...)
}
manifest = append(manifest, content...)
return nil
})
if err != nil {
return nil, "", fmt.Errorf("failed to walk template output directory: %w", err)
}
return manifest, releaseCopy.Namespace, nil
}
func convertTrackResources(resources []TrackResourceSpec) []resource.Resource {
if len(resources) == 0 {
return nil
}
result := make([]resource.Resource, len(resources))
for i, r := range resources {
result[i] = resource.Resource{
Kind: r.Kind,
Name: r.Name,
Namespace: r.Namespace,
}
}
return result
}
+46 -5
View File
@@ -2,7 +2,7 @@ package state
import (
"bytes"
"context"
gocontext "context"
"crypto/sha1"
"encoding/hex"
"errors"
@@ -129,12 +129,18 @@ type HelmState struct {
valsRuntime vals.Evaluator
kubeconfig string
// RenderedValues is the helmfile-wide values that is `.Values`
// which is accessible from within the whole helmfile go template.
// Note that this is usually computed by DesiredStateLoader from ReleaseSetSpec.Env
RenderedValues map[string]any
}
func (st *HelmState) SetKubeconfig(kubeconfig string) {
st.kubeconfig = kubeconfig
}
// SubHelmfileSpec defines the subhelmfile path and options
type SubHelmfileSpec struct {
//path or glob pattern for the sub helmfiles
@@ -226,6 +232,8 @@ type HelmSpec struct {
SyncReleaseLabels *bool `yaml:"syncReleaseLabels,omitempty"`
// TakeOwnership is true if the helmfile should take ownership of the release
TakeOwnership *bool `yaml:"takeOwnership,omitempty"`
// TrackMode specifies whether to use 'helm' or 'kubedog' for tracking resources
TrackMode string `yaml:"trackMode,omitempty"`
}
// RepositorySpec that defines values for a helm repo
@@ -254,7 +262,7 @@ type Inherit struct {
type Inherits []Inherit
// ReleaseSpec defines the structure of a helm release
// ReleaseSpec defines the configuration for a Helm release managed by helmfile.
type ReleaseSpec struct {
// Chart is the name of the chart being installed to create this release
Chart string `yaml:"chart,omitempty"`
@@ -444,8 +452,27 @@ type ReleaseSpec struct {
DeleteTimeout *int `yaml:"deleteTimeout,omitempty"`
// SyncReleaseLabels is true if the release labels should be synced with the helmfile labels
SyncReleaseLabels *bool `yaml:"syncReleaseLabels,omitempty"`
// TakeOwnership is true if the release should take ownership of the resources
// TakeOwnership is true if release should take ownership of resources
TakeOwnership *bool `yaml:"takeOwnership,omitempty"`
// TrackMode specifies whether to use 'helm' or 'kubedog' for tracking resources
TrackMode string `yaml:"trackMode,omitempty"`
// TrackTimeout specifies timeout for kubedog tracking (in seconds)
TrackTimeout *int `yaml:"trackTimeout,omitempty"`
// TrackLogs enables log streaming with kubedog
TrackLogs *bool `yaml:"trackLogs,omitempty"`
// TrackKinds is a whitelist of resource kinds to track
TrackKinds []string `yaml:"trackKinds,omitempty"`
// SkipKinds is a blacklist of resource kinds to skip tracking
SkipKinds []string `yaml:"skipKinds,omitempty"`
// TrackResources is a whitelist of specific resources to track
TrackResources []TrackResourceSpec `yaml:"trackResources,omitempty"`
}
// TrackResourceSpec specifies a resource to track
type TrackResourceSpec struct {
Kind string `yaml:"kind,omitempty"`
Name string `yaml:"name,omitempty"`
Namespace string `yaml:"namespace,omitempty"`
}
func (r *Inherits) UnmarshalYAML(unmarshal func(any) error) error {
@@ -870,6 +897,9 @@ type SyncOpts struct {
SyncArgs string
HideNotes bool
TakeOwnership bool
TrackMode string
TrackTimeout int
TrackLogs bool
}
type SyncOpt interface{ Apply(*SyncOpts) }
@@ -1095,6 +1125,11 @@ func (st *HelmState) SyncReleases(affectedReleases *AffectedReleases, helm helme
}
} else if release.UpdateStrategy == UpdateStrategyReinstallIfForbidden {
relErr = st.performSyncOrReinstallOfRelease(affectedReleases, helm, context, release, chart, m, flags...)
if relErr == nil && st.shouldUseKubedog(release, opts) {
if trackErr := st.trackWithKubedog(gocontext.Background(), release, helm, opts); trackErr != nil {
st.logger.Warnf("kubedog tracking failed for release %s: %v", release.Name, trackErr)
}
}
} else {
if err := helm.SyncRelease(context, release.Name, chart, release.Namespace, flags...); err != nil {
m.Lock()
@@ -1111,6 +1146,12 @@ func (st *HelmState) SyncReleases(affectedReleases *AffectedReleases, helm helme
} else {
release.installedVersion = installedVersion
}
if st.shouldUseKubedog(release, opts) {
if trackErr := st.trackWithKubedog(gocontext.Background(), release, helm, opts); trackErr != nil {
st.logger.Warnf("kubedog tracking failed for release %s: %v", release.Name, trackErr)
}
}
}
}
@@ -4910,7 +4951,7 @@ func (st *HelmState) acquireSharedLock(result *chartLockResult, chartPath string
result.inProcessMutex = st.getNamedRWMutex(chartPath)
result.inProcessMutex.RLock()
ctx, cancel := context.WithTimeout(context.Background(), lockTimeout)
ctx, cancel := gocontext.WithTimeout(gocontext.Background(), lockTimeout)
defer cancel()
locked, err := result.fileLock.TryRLockContext(ctx, 500*time.Millisecond)
@@ -4945,7 +4986,7 @@ func (st *HelmState) acquireExclusiveLock(result *chartLockResult, chartPath str
var lockErr error
for attempt := 1; attempt <= maxRetries; attempt++ {
ctx, cancel := context.WithTimeout(context.Background(), lockTimeout)
ctx, cancel := gocontext.WithTimeout(gocontext.Background(), lockTimeout)
locked, lockErr = result.fileLock.TryLockContext(ctx, 500*time.Millisecond)
cancel()
+6 -6
View File
@@ -38,39 +38,39 @@ func TestGenerateID(t *testing.T) {
run(testcase{
subject: "baseline",
release: ReleaseSpec{Name: "foo", Chart: "incubator/raw"},
want: "foo-values-6884949b8b",
want: "foo-values-dd88b94b8",
})
run(testcase{
subject: "different bytes content",
release: ReleaseSpec{Name: "foo", Chart: "incubator/raw"},
data: []byte(`{"k":"v"}`),
want: "foo-values-58f57b794f",
want: "foo-values-6fb7bbb95f",
})
run(testcase{
subject: "different map content",
release: ReleaseSpec{Name: "foo", Chart: "incubator/raw"},
data: map[string]any{"k": "v"},
want: "foo-values-6b6b884cc9",
want: "foo-values-56d84c9897",
})
run(testcase{
subject: "different chart",
release: ReleaseSpec{Name: "foo", Chart: "stable/envoy"},
want: "foo-values-85494c4677",
want: "foo-values-6644fc9d47",
})
run(testcase{
subject: "different name",
release: ReleaseSpec{Name: "bar", Chart: "incubator/raw"},
want: "bar-values-9d65c65f",
want: "bar-values-859cd849bf",
})
run(testcase{
subject: "specific ns",
release: ReleaseSpec{Name: "foo", Chart: "incubator/raw", Namespace: "myns"},
want: "myns-foo-values-84b69bb989",
want: "myns-foo-values-86d544f7f9",
})
for id, n := range ids {