add logic to garbage collect with the provide retention options

Signed-off-by: Sanskar Jaiswal <jaiswalsanskar078@gmail.com>

controllers/gitrepository_controller.go

@@ -103,12 +103,16 @@ type GitRepositoryReconciler struct {
  Storage *Storage
  ControllerName string
 
- requeueDependency time.Duration
+ requeueDependency time.Duration
+ artifactRetentionTTL time.Duration
+ artifactRetentionRecords int
 }
 
 type GitRepositoryReconcilerOptions struct {
  MaxConcurrentReconciles int
  DependencyRequeueInterval time.Duration
+ ArtifactRetentionTTL time.Duration
+ ArtifactRetentionRecords int
 }
 
 // gitRepositoryReconcileFunc is the function type for all the
@@ -121,6 +125,8 @@ func (r *GitRepositoryReconciler) SetupWithManager(mgr ctrl.Manager) error {
 
 func (r *GitRepositoryReconciler) SetupWithManagerAndOptions(mgr ctrl.Manager, opts GitRepositoryReconcilerOptions) error {
  r.requeueDependency = opts.DependencyRequeueInterval
+ r.artifactRetentionRecords = opts.ArtifactRetentionRecords
+ r.artifactRetentionTTL = opts.ArtifactRetentionTTL
 
  return ctrl.NewControllerManagedBy(mgr).
  For(&sourcev1.GitRepository{}, builder.WithPredicates(
@@ -703,13 +709,16 @@ func (r *GitRepositoryReconciler) garbageCollect(ctx context.Context, obj *sourc
  return nil
  }
  if obj.GetArtifact() != nil {
- if deleted, err := r.Storage.RemoveAllButCurrent(*obj.GetArtifact()); err != nil {
+ deleted, err := r.Storage.RemoveGarbageFiles(*obj.GetArtifact(), r.artifactRetentionRecords, r.artifactRetentionTTL)
+ if err != nil {
  return &serror.Event{
- Err: fmt.Errorf("garbage collection of old artifacts failed: %w", err),
+ Err: fmt.Errorf("garbage collection of old artifacts failed: %w", err),
+ Reason: "GarbageCollectionFailed",
  }
- } else if len(deleted) > 0 {
+ }
+ if len(deleted) > 0 {
  r.eventLogf(ctx, obj, events.EventTypeTrace, "GarbageCollectionSucceeded",
- "garbage collected old artifacts")
+ fmt.Sprintf("garbage collected %d old artifacts", len(deleted)))
  }
  }
  return nil

controllers/storage.go

@@ -26,6 +26,7 @@ import (
  "net/url"
  "os"
  "path/filepath"
+ "sort"
  "strings"
  "time"
 
@@ -145,6 +146,84 @@ func (s *Storage) RemoveAllButCurrent(artifact sourcev1.Artifact) ([]string, err
  return deletedFiles, nil
 }
 
+func (s *Storage) getGarbageFiles(artifact sourcev1.Artifact, n int, ttl time.Duration) []string {
+ localPath := s.LocalPath(artifact)
+ dir := filepath.Dir(localPath)
+ garbageFiles := []string{}
+ filesWithCreatedTs := make(map[time.Time]string)
+ // sortedPaths contain all files sorted according to their created ts.
+ sortedPaths := []string{}
+ now := time.Now().UTC()
+ _ = filepath.Walk(dir, func(path string, info os.FileInfo, err error) error {
+ createdAt := info.ModTime().UTC()
+ diff := now.Sub(createdAt)
+ // compare the time difference between now and the time at which the file was created
+ // with the provided ttl. delete if difference is greater than the ttl.
+ expired := diff > ttl
+ if path != localPath && !info.IsDir() && info.Mode()&os.ModeSymlink != os.ModeSymlink && expired {
+ garbageFiles = append(garbageFiles, path)
+ }
+ if !info.IsDir() && info.Mode()&os.ModeSymlink != os.ModeSymlink {
+ filesWithCreatedTs[info.ModTime().UTC()] = path
+ }
+ return nil
+ })
+ creationTimestamps := []time.Time{}
+ for ts := range filesWithCreatedTs {
+ creationTimestamps = append(creationTimestamps, ts)
+ }
+ // sort all timestamps in an ascending order.
+ sort.Slice(creationTimestamps, func(i, j int) bool { return creationTimestamps[i].Before(creationTimestamps[j]) })
+ for _, ts := range creationTimestamps {
+ path, ok := filesWithCreatedTs[ts]
+ if !ok {
+ return nil
+ }
+ sortedPaths = append(sortedPaths, path)
+ }
+
+ for i, path := range sortedPaths {
+ if path != localPath && !stringInSlice(path, garbageFiles) && len(sortedPaths) <= n {
+ // append path to garbageFiles and remove it from sortedPaths
+ garbageFiles = append(garbageFiles, path)
+ sortedPaths = append(sortedPaths[:i], sortedPaths[i+1:]...)
+ }
+ }
+
+ return garbageFiles
+}
+
+// RemoveGarbageFiles removes all garabge files in the artifact dir according to the provided
+// retention options.
+func (s *Storage) RemoveGarbageFiles(artifact sourcev1.Artifact, n int, ttl time.Duration) ([]string, error) {
+ garbageFiles := s.getGarbageFiles(artifact, n, ttl)
+ var errors []string
+ var deleted []string
+ if len(garbageFiles) > 0 {
+ for _, file := range garbageFiles {
+ err := os.Remove(file)
+ if err != nil {
+ errors = append(errors, err.Error())
+ } else {
+ deleted = append(deleted, file)
+ }
+ }
+ }
+ if len(errors) > 0 {
+ return deleted, fmt.Errorf("failed to remove files: %s", strings.Join(errors, " "))
+ }
+ return deleted, nil
+}
+
+func stringInSlice(a string, list []string) bool {
+ for _, b := range list {
+ if b == a {
+ return true
+ }
+ }
+ return false
+}
+
 // ArtifactExist returns a boolean indicating whether the v1beta1.Artifact exists in storage and is a regular file.
 func (s *Storage) ArtifactExist(artifact sourcev1.Artifact) bool {
  fi, err := os.Lstat(s.LocalPath(artifact))

main.go

@@ -71,21 +71,23 @@ func init() {
 
 func main() {
  var (
- metricsAddr string
- eventsAddr string
- healthAddr string
- storagePath string
- storageAddr string
- storageAdvAddr string
- concurrent int
- requeueDependency time.Duration
- watchAllNamespaces bool
- helmIndexLimit int64
- helmChartLimit int64
- helmChartFileLimit int64
- clientOptions client.Options
- logOptions logger.Options
- leaderElectionOptions leaderelection.Options
+ metricsAddr string
+ eventsAddr string
+ healthAddr string
+ storagePath string
+ storageAddr string
+ storageAdvAddr string
+ concurrent int
+ requeueDependency time.Duration
+ watchAllNamespaces bool
+ helmIndexLimit int64
+ helmChartLimit int64
+ helmChartFileLimit int64
+ clientOptions client.Options
+ logOptions logger.Options
+ leaderElectionOptions leaderelection.Options
+ artifactRetentionTTL time.Duration
+ artifactRetentionRecords int
  )
 
  flag.StringVar(&metricsAddr, "metrics-addr", envOrDefault("METRICS_ADDR", ":8080"),
@@ -110,6 +112,10 @@ func main() {
  "The max allowed size in bytes of a file in a Helm chart.")
  flag.DurationVar(&requeueDependency, "requeue-dependency", 30*time.Second,
  "The interval at which failing dependencies are reevaluated.")
+ flag.DurationVar(&artifactRetentionTTL, "artifact-retention-ttl", 30*time.Second,
+ "The duration for which artifacts be persisted in storage before being evicted.")
+ flag.IntVar(&artifactRetentionRecords, "artifact-retention-records", 2,
+ "The number of artifacts allowed to be present in storage.")
 
  clientOptions.BindFlags(flag.CommandLine)
  logOptions.BindFlags(flag.CommandLine)
@@ -174,6 +180,8 @@ func main() {
  }).SetupWithManagerAndOptions(mgr, controllers.GitRepositoryReconcilerOptions{
  MaxConcurrentReconciles: concurrent,
  DependencyRequeueInterval: requeueDependency,
+ ArtifactRetentionTTL: artifactRetentionTTL,
+ ArtifactRetentionRecords: artifactRetentionRecords,
  }); err != nil {
  setupLog.Error(err, "unable to create controller", "controller", sourcev1.GitRepositoryKind)
  os.Exit(1)