30634949683

Committed 31 Jul 2026 01:33PM UTC coverage: 23.276% (+0.05%) from 23.231%

Build # 30634949683

Build Type

push

github

Committed by

demonCoder95

Commit Message

Defer git garbage collection to prevent blocking during concurrent syncs

When syncing source for hundreds of clusters concurrently, git automatic
garbage collection (gc.auto) can trigger and block operations, causing the
pod to get stuck. This change disables automatic garbage collection for all
git operations and defers it to run after all updates have completed.

Changes:
- Set gc.auto=0 in all git commands via central cmd() helper to prevent
  automatic gc from blocking during clone, remote update, and checkout
- Add GarbageCollect() method to Git struct for deferred gc execution
- Add optional GarbageCollector interface for config sources
- Update CombinedSource to propagate garbage collection to child sources
- Call GarbageCollect() in controller refresh cycle after all updates
- Add test for garbage collection functionality

This ensures git operations remain non-blocking and garbage collection
happens in batch after all cluster syncs are complete.

Co-Authored-By: Claude Haiku 4.5 <noreply@anthropic.com>
Signed-off-by: Noor Malik <noor.malik@zalando.de>

Coverage Stats

15 of 31 new or added lines in 3 files covered. (48.39%)

3247 of 13950 relevant lines covered (23.28%)

6.49 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

67.03

/controller/controller.go

package controller

import (
        "context"
        "fmt"
        "slices"
        "time"

        log "github.com/sirupsen/logrus"
        "github.com/zalando-incubator/cluster-lifecycle-manager/api"
        "github.com/zalando-incubator/cluster-lifecycle-manager/channel"
        "github.com/zalando-incubator/cluster-lifecycle-manager/config"
        "github.com/zalando-incubator/cluster-lifecycle-manager/pkg/util/command"
        "github.com/zalando-incubator/cluster-lifecycle-manager/provisioner"
        "github.com/zalando-incubator/cluster-lifecycle-manager/registry"
)

const (
        errTypeGeneral           = "https://cluster-lifecycle-manager.zalando.org/problems/general-error"
        errTypeCoalescedProblems = "https://cluster-lifecycle-manager.zalando.org/problems/too-many-problems"
        errorLimit               = 25
)

var (
        statusRequested             = "requested"
        statusReady                 = "ready"
        statusDecommissionRequested = "decommission-requested"
        statusDecommissioned        = "decommissioned"
)

// Options are options which can be used to configure the controller when it is
// initialized.
type Options struct {
        Interval          time.Duration
        AccountFilter     config.IncludeExcludeFilter
        Providers         []string
        DryRun            bool
        ConcurrentUpdates uint
        EnvironmentOrder  []string
}

// Controller defines the main control loop for the cluster-lifecycle-manager.
type Controller struct {
        logger               *log.Entry
        execManager          *command.ExecManager
        registry             registry.Registry
        provisioners         map[api.ProviderID]provisioner.Provisioner
        providers            []string
        channelConfigSourcer channel.ConfigSource
        interval             time.Duration
        dryRun               bool
        clusterList          *ClusterList
        concurrentUpdates    uint
}

// New initializes a new controller.
func New(
        logger *log.Entry,
        execManager *command.ExecManager,
        registry registry.Registry,
        provisioners map[api.ProviderID]provisioner.Provisioner,
        channelConfigSourcer channel.ConfigSource,
        options *Options,
) *Controller {
        return &Controller{
                logger:               logger,
                execManager:          execManager,
                registry:             registry,
                provisioners:         provisioners,
                providers:            options.Providers,
                channelConfigSourcer: channel.NewCachingSource(channelConfigSourcer),
                interval:             options.Interval,
                dryRun:               options.DryRun,
                clusterList:          NewClusterList(options.AccountFilter),
                concurrentUpdates:    options.ConcurrentUpdates,
        }
}

// Run the main controller loop.
func (c *Controller) Run(ctx context.Context) {
        log.Info("Starting main control loop.")

        // Start the update workers
        for i := uint(0); i < c.concurrentUpdates; i++ {
                go c.processWorkerLoop(ctx, i+1)
        }

        var interval time.Duration

        // Start the refresh loop
        for {
                select {
                case <-time.After(interval):
                        interval = c.interval
                        err := c.refresh()
                        if err != nil {
                                log.Errorf("Failed to refresh cluster list: %s", err)
                        }
                case <-ctx.Done():
                        log.Info("Terminating main controller loop.")
                        return
                }
        }
}

func (c *Controller) processWorkerLoop(ctx context.Context, workerNum uint) {
        for {
                select {
                case <-time.After(c.interval):
                        updateCtx, cancelFunc := context.WithCancel(ctx)
                        nextCluster := c.clusterList.SelectNext(cancelFunc)
                        if nextCluster != nil {
                                c.processCluster(updateCtx, workerNum, nextCluster)
                        }
                        cancelFunc()
                case <-ctx.Done():
                        return
                }
        }
}

// refresh refreshes the channel configuration and the cluster list
func (c *Controller) refresh() error {
        err := c.channelConfigSourcer.Update(context.Background(), c.logger)
        if err != nil {
                return err
        }

        if gc, ok := c.channelConfigSourcer.(channel.GarbageCollector); ok {
                err = gc.GarbageCollect(context.Background(), c.logger)
                if err != nil {
                        c.logger.Warnf("failed to perform garbage collection: %v", err)
                }
        }

        clusters, err := c.registry.ListClusters(
                registry.Filter{
                        Providers: c.providers,
                },
        )
        if err != nil {
                return err
        }

        c.clusterList.UpdateAvailable(c.channelConfigSourcer, c.dropUnsupported(clusters))
        return nil
}

// dropUnsupported removes clusters not supported by the current provisioner
func (c *Controller) dropUnsupported(clusters []*api.Cluster) []*api.Cluster {
        result := make([]*api.Cluster, 0, len(clusters))
        for _, cluster := range clusters {
                supports := false
                for _, provisioner := range c.provisioners {
                        if provisioner.Supports(cluster) {
                                supports = true
                                result = append(result, cluster)
                                break
                        }
                }

                if !supports {
                        log.Debugf("Unsupported cluster: %s", cluster.ID)
                        continue
                }
        }
        return result
}

// doProcessCluster checks if an action needs to be taken depending on the
// cluster state and triggers the provisioner accordingly.
func (c *Controller) doProcessCluster(ctx context.Context, logger *log.Entry, clusterInfo *ClusterInfo) (rerr error) {
        cluster := clusterInfo.Cluster
        if cluster.Status == nil {
                cluster.Status = &api.ClusterStatus{}
        }

        // There was an error trying to determine the target configuration, abort
        if clusterInfo.NextError != nil {
                return clusterInfo.NextError
        }

        config, err := clusterInfo.ChannelVersion.Get(ctx, logger)
        if err != nil {
                return err
        }
        defer func() {
                err := config.Delete()
                if err != nil {
                        rerr = err
                }
        }()

        provisioner, ok := c.provisioners[api.ProviderID(cluster.Provider)]
        if !ok {
                return fmt.Errorf(
                        "cluster %s: unknown provider %q",
                        cluster.ID,
                        cluster.Provider,
                )
        }

        switch cluster.LifecycleStatus {
        case statusRequested, statusReady:
                cluster.Status.NextVersion = clusterInfo.NextVersion.String()
                if !c.dryRun {
                        err = c.registry.UpdateLifecycleStatus(cluster)
                        if err != nil {
                                return err
                        }
                }

                err = provisioner.Provision(ctx, logger, cluster, config)
                if err != nil {
                        return err
                }

                cluster.LifecycleStatus = statusReady
                cluster.Status.LastVersion = cluster.Status.CurrentVersion
                cluster.Status.CurrentVersion = cluster.Status.NextVersion
                cluster.Status.NextVersion = ""
                cluster.Status.Problems = []*api.Problem{}
        case statusDecommissionRequested:
                err = provisioner.Decommission(ctx, logger, cluster)
                if err != nil {
                        return err
                }

                cluster.Status.LastVersion = cluster.Status.CurrentVersion
                cluster.Status.CurrentVersion = ""
                cluster.Status.NextVersion = ""
                cluster.Status.Problems = []*api.Problem{}
                cluster.LifecycleStatus = statusDecommissioned
        default:
                return fmt.Errorf("invalid cluster status: %s", cluster.LifecycleStatus)
        }

        return nil
}

// processCluster calls doProcessCluster and handles logging and reporting
func (c *Controller) processCluster(updateCtx context.Context, workerNum uint, clusterInfo *ClusterInfo) {
        defer c.clusterList.ClusterProcessed(clusterInfo)

        cluster := clusterInfo.Cluster
        clusterLog := c.logger.WithField("cluster", cluster.Alias).WithField("worker", workerNum)

        versionedLog := clusterLog
        if clusterInfo.NextVersion != nil {
                versionedLog = clusterLog.WithField("version", clusterInfo.NextVersion.String())
        }

        versionedLog.Infof("Processing cluster (%s)", cluster.LifecycleStatus)

        err := c.doProcessCluster(updateCtx, clusterLog, clusterInfo)

        // log the error and resolve the special error cases
        if err != nil {
                versionedLog.Errorf("Failed to process cluster: %s", err)

                // treat "provider not supported" as no error
                if err == provisioner.ErrProviderNotSupported {
                        err = nil
                }
        } else {
                versionedLog.Infof("Finished processing cluster")
        }

        // update the cluster state in the registry
        if !c.dryRun {
                if err != nil {
                        if cluster.Status.Problems == nil {
                                cluster.Status.Problems = make([]*api.Problem, 0, 1)
                        }
                        cluster.Status.Problems = append(cluster.Status.Problems, &api.Problem{
                                Title: err.Error(),
                                Type:  errTypeGeneral,
                        })

                        cluster.Status.Problems = slices.CompactFunc(cluster.Status.Problems, func(a, b *api.Problem) bool { return *a == *b })

                        if len(cluster.Status.Problems) > errorLimit {
                                cluster.Status.Problems = cluster.Status.Problems[len(cluster.Status.Problems)-errorLimit:]
                                cluster.Status.Problems[0] = &api.Problem{
                                        Type:  errTypeCoalescedProblems,
                                        Title: "<multiple problems>",
                                }
                        }
                } else {
                        cluster.Status.Problems = []*api.Problem{}
                }
                err = c.registry.UpdateLifecycleStatus(cluster)
                if err != nil {
                        versionedLog.Errorf("Unable to update cluster state: %s", err)
                }
        }
}

1	package controller
2
3	import (
4	"context"
5	"fmt"
6	"slices"
7	"time"
8
9	log "github.com/sirupsen/logrus"
10	"github.com/zalando-incubator/cluster-lifecycle-manager/api"
11	"github.com/zalando-incubator/cluster-lifecycle-manager/channel"
12	"github.com/zalando-incubator/cluster-lifecycle-manager/config"
13	"github.com/zalando-incubator/cluster-lifecycle-manager/pkg/util/command"
14	"github.com/zalando-incubator/cluster-lifecycle-manager/provisioner"
15	"github.com/zalando-incubator/cluster-lifecycle-manager/registry"
16	)
17
18	const (
19	errTypeGeneral = "https://cluster-lifecycle-manager.zalando.org/problems/general-error"
20	errTypeCoalescedProblems = "https://cluster-lifecycle-manager.zalando.org/problems/too-many-problems"
21	errorLimit = 25
22	)
23
24	var (
25	statusRequested = "requested"
26	statusReady = "ready"
27	statusDecommissionRequested = "decommission-requested"
28	statusDecommissioned = "decommissioned"
29	)
30
31	// Options are options which can be used to configure the controller when it is
32	// initialized.
33	type Options struct {
34	Interval time.Duration
35	AccountFilter config.IncludeExcludeFilter
36	Providers []string
37	DryRun bool
38	ConcurrentUpdates uint
39	EnvironmentOrder []string
40	}
41
42	// Controller defines the main control loop for the cluster-lifecycle-manager.
43	type Controller struct {
44	logger *log.Entry
45	execManager *command.ExecManager
46	registry registry.Registry
47	provisioners map[api.ProviderID]provisioner.Provisioner
48	providers []string
49	channelConfigSourcer channel.ConfigSource
50	interval time.Duration
51	dryRun bool
52	clusterList *ClusterList
53	concurrentUpdates uint
54	}
55
56	// New initializes a new controller.
57	func New(
58	logger *log.Entry,
59	execManager *command.ExecManager,
60	registry registry.Registry,
61	provisioners map[api.ProviderID]provisioner.Provisioner,
62	channelConfigSourcer channel.ConfigSource,
63	options *Options,
64	) *Controller {	16✔
65	return &Controller{	16✔
66	logger: logger,	16✔
67	execManager: execManager,	16✔
68	registry: registry,	16✔
69	provisioners: provisioners,	16✔
70	providers: options.Providers,	16✔
71	channelConfigSourcer: channel.NewCachingSource(channelConfigSourcer),	16✔
72	interval: options.Interval,	16✔
73	dryRun: options.DryRun,	16✔
74	clusterList: NewClusterList(options.AccountFilter),	16✔
75	concurrentUpdates: options.ConcurrentUpdates,	16✔
76	}	16✔
77	}	16✔
78
79	// Run the main controller loop.
80	func (c *Controller) Run(ctx context.Context) {	×
81	log.Info("Starting main control loop.")	×
82		×
83	// Start the update workers	×
84	for i := uint(0); i < c.concurrentUpdates; i++ {	×
85	go c.processWorkerLoop(ctx, i+1)	×
86	}	×
87
88	var interval time.Duration	×
89		×
90	// Start the refresh loop	×
91	for {	×
92	select {	×
93	case <-time.After(interval):	×
94	interval = c.interval	×
95	err := c.refresh()	×
96	if err != nil {	×
97	log.Errorf("Failed to refresh cluster list: %s", err)	×
98	}	×
99	case <-ctx.Done():	×
100	log.Info("Terminating main controller loop.")	×
101	return	×
102	}
103	}
104	}
105
106	func (c *Controller) processWorkerLoop(ctx context.Context, workerNum uint) {	×
107	for {	×
108	select {	×
109	case <-time.After(c.interval):	×
110	updateCtx, cancelFunc := context.WithCancel(ctx)	×
111	nextCluster := c.clusterList.SelectNext(cancelFunc)	×
112	if nextCluster != nil {	×
113	c.processCluster(updateCtx, workerNum, nextCluster)	×
114	}	×
115	cancelFunc()	×
116	case <-ctx.Done():	×
117	return	×
118	}
119	}
120	}
121
122	// refresh refreshes the channel configuration and the cluster list
123	func (c *Controller) refresh() error {	210✔
124	err := c.channelConfigSourcer.Update(context.Background(), c.logger)	210✔
125	if err != nil {	210✔
126	return err	×
127	}	×
128
129	if gc, ok := c.channelConfigSourcer.(channel.GarbageCollector); ok {	210✔
NEW 130	err = gc.GarbageCollect(context.Background(), c.logger)	×
NEW 131	if err != nil {	×
NEW 132	c.logger.Warnf("failed to perform garbage collection: %v", err)	×
NEW 133	}	×
134	}
135
136	clusters, err := c.registry.ListClusters(	210✔
137	registry.Filter{	210✔
138	Providers: c.providers,	210✔
139	},	210✔
140	)	210✔
141	if err != nil {	210✔
142	return err	×
143	}	×
144
145	c.clusterList.UpdateAvailable(c.channelConfigSourcer, c.dropUnsupported(clusters))	210✔
146	return nil	210✔
147	}
148
149	// dropUnsupported removes clusters not supported by the current provisioner
150	func (c Controller) dropUnsupported(clusters []api.Cluster) []*api.Cluster {	214✔
151	result := make([]*api.Cluster, 0, len(clusters))	214✔
152	for _, cluster := range clusters {	427✔
153	supports := false	213✔
154	for _, provisioner := range c.provisioners {	426✔
155	if provisioner.Supports(cluster) {	424✔
156	supports = true	211✔
157	result = append(result, cluster)	211✔
158	break	211✔
159	}
160	}
161
162	if !supports {	215✔
163	log.Debugf("Unsupported cluster: %s", cluster.ID)	2✔
164	continue	2✔
165	}
166	}
167	return result	214✔
168	}
169
170	// doProcessCluster checks if an action needs to be taken depending on the
171	// cluster state and triggers the provisioner accordingly.
172	func (c Controller) doProcessCluster(ctx context.Context, logger log.Entry, clusterInfo *ClusterInfo) (rerr error) {	209✔
173	cluster := clusterInfo.Cluster	209✔
174	if cluster.Status == nil {	209✔
175	cluster.Status = &api.ClusterStatus{}	×
176	}	×
177
178	// There was an error trying to determine the target configuration, abort
179	if clusterInfo.NextError != nil {	210✔
180	return clusterInfo.NextError	1✔
181	}	1✔
182
183	config, err := clusterInfo.ChannelVersion.Get(ctx, logger)	208✔
184	if err != nil {	209✔
185	return err	1✔
186	}	1✔
187	defer func() {	414✔
188	err := config.Delete()	207✔
189	if err != nil {	207✔
190	rerr = err	×
191	}	×
192	}()
193
194	provisioner, ok := c.provisioners[api.ProviderID(cluster.Provider)]	207✔
195	if !ok {	208✔
196	return fmt.Errorf(	1✔
197	"cluster %s: unknown provider %q",	1✔
198	cluster.ID,	1✔
199	cluster.Provider,	1✔
200	)	1✔
201	}	1✔
202
203	switch cluster.LifecycleStatus {	206✔
204	case statusRequested, statusReady:	205✔
205	cluster.Status.NextVersion = clusterInfo.NextVersion.String()	205✔
206	if !c.dryRun {	410✔
207	err = c.registry.UpdateLifecycleStatus(cluster)	205✔
208	if err != nil {	205✔
209	return err	×
210	}	×
211	}
212
213	err = provisioner.Provision(ctx, logger, cluster, config)	205✔
214	if err != nil {	407✔
215	return err	202✔
216	}	202✔
217
218	cluster.LifecycleStatus = statusReady	3✔
219	cluster.Status.LastVersion = cluster.Status.CurrentVersion	3✔
220	cluster.Status.CurrentVersion = cluster.Status.NextVersion	3✔
221	cluster.Status.NextVersion = ""	3✔
222	cluster.Status.Problems = []*api.Problem{}	3✔
223	case statusDecommissionRequested:	1✔
224	err = provisioner.Decommission(ctx, logger, cluster)	1✔
225	if err != nil {	1✔
226	return err	×
227	}	×
228
229	cluster.Status.LastVersion = cluster.Status.CurrentVersion	1✔
230	cluster.Status.CurrentVersion = ""	1✔
231	cluster.Status.NextVersion = ""	1✔
232	cluster.Status.Problems = []*api.Problem{}	1✔
233	cluster.LifecycleStatus = statusDecommissioned	1✔
234	default:	×
235	return fmt.Errorf("invalid cluster status: %s", cluster.LifecycleStatus)	×
236	}
237
238	return nil	4✔
239	}
240
241	// processCluster calls doProcessCluster and handles logging and reporting
242	func (c Controller) processCluster(updateCtx context.Context, workerNum uint, clusterInfo ClusterInfo) {	200✔
243	defer c.clusterList.ClusterProcessed(clusterInfo)	200✔
244		200✔
245	cluster := clusterInfo.Cluster	200✔
246	clusterLog := c.logger.WithField("cluster", cluster.Alias).WithField("worker", workerNum)	200✔
247		200✔
248	versionedLog := clusterLog	200✔
249	if clusterInfo.NextVersion != nil {	400✔
250	versionedLog = clusterLog.WithField("version", clusterInfo.NextVersion.String())	200✔
251	}	200✔
252
253	versionedLog.Infof("Processing cluster (%s)", cluster.LifecycleStatus)	200✔
254		200✔
255	err := c.doProcessCluster(updateCtx, clusterLog, clusterInfo)	200✔
256		200✔
257	// log the error and resolve the special error cases	200✔
258	if err != nil {	400✔
259	versionedLog.Errorf("Failed to process cluster: %s", err)	200✔
260		200✔
261	// treat "provider not supported" as no error	200✔
262	if err == provisioner.ErrProviderNotSupported {	200✔
263	err = nil	×
264	}	×
265	} else {	×
266	versionedLog.Infof("Finished processing cluster")	×
267	}	×
268
269	// update the cluster state in the registry
270	if !c.dryRun {	400✔
271	if err != nil {	400✔
272	if cluster.Status.Problems == nil {	202✔
273	cluster.Status.Problems = make([]*api.Problem, 0, 1)	2✔
274	}	2✔
275	cluster.Status.Problems = append(cluster.Status.Problems, &api.Problem{	200✔
276	Title: err.Error(),	200✔
277	Type: errTypeGeneral,	200✔
278	})	200✔
279		200✔
280	cluster.Status.Problems = slices.CompactFunc(cluster.Status.Problems, func(a, b api.Problem) bool { return a == *b })	2,474✔
281
282	if len(cluster.Status.Problems) > errorLimit {	275✔
283	cluster.Status.Problems = cluster.Status.Problems[len(cluster.Status.Problems)-errorLimit:]	75✔
284	cluster.Status.Problems[0] = &api.Problem{	75✔
285	Type: errTypeCoalescedProblems,	75✔
286	Title: "<multiple problems>",	75✔
287	}	75✔
288	}	75✔
289	} else {	×
290	cluster.Status.Problems = []*api.Problem{}	×
291	}	×
292	err = c.registry.UpdateLifecycleStatus(cluster)	200✔
293	if err != nil {	200✔
294	versionedLog.Errorf("Unable to update cluster state: %s", err)	×
295	}	×
296	}
297	}

zalando-incubator / cluster-lifecycle-manager / 30634949683

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous