Bump k8s deps to v1.28, kapp-ctrl to v0.49 and flux2 to >=v2.0.0

.github/workflows/kubeapps-general.yaml

@@ -52,7 +52,7 @@ env:
   # Currently, we only build the images for linux/amd64 because building cross-platform images is extremely slow...
   IMG_PLATFORMS: "linux/amd64"
   KAPP_CONTROLLER_VERSION: "v0.49.0"
-  FLUX_VERSION: "v0.37.0"
+  FLUX_VERSION: "v2.2.2"
   K8S_KIND_VERSION: "v1.27.3@sha256:3966ac761ae0136263ffdb6cfd4db23ef8a83cba8a463690e98317add2c9ba72"
   KIND_VERSION: "v0.20.0"
   KUBECTL_VERSION: "v1.27.9"

cmd/kubeapps-apis/plugins/fluxv2/packages/v1alpha1/cache/rate_limiting_queue.go

@@ -4,6 +4,7 @@
 package cache
 
 import (
+	"context"
 	"fmt"
 	"strings"
 	"sync"
@@ -109,7 +110,7 @@ func (q *rateLimitingType) WaitUntilForgotten(item string) {
 	// a call to .Forget(item).
 	// TODO: (gfichtenholt) don't do wait.PollInfinite() here, use some sensible
 	// timeout instead, and then this func will need to return an error
-	err := wait.PollInfinite(10*time.Millisecond, func() (bool, error) {
+	err := wait.PollUntilContextCancel(context.Background(), 10*time.Millisecond, true, func(ctx context.Context) (bool, error) {
 		return q.rateLimiter.NumRequeues(item) == 0, nil
 	})
 	if err != nil {

cmd/kubeapps-apis/plugins/fluxv2/packages/v1alpha1/cache/watcher_cache.go

@@ -131,7 +131,7 @@ type NamespacedResourceWatcherCacheConfig struct {
 	OnResyncFunc ResyncFunc
 
 	// These funcs are needed to manipulate API-specific objects, such as flux's
-	// sourcev1.HelmRepository, in a generic fashion
+	// sourcev1beta2.HelmRepository, in a generic fashion
 	NewObjFunc    NewObjectFunc
 	NewListFunc   NewObjectListFunc
 	ListItemsFunc GetListItemsFunc

cmd/kubeapps-apis/plugins/fluxv2/packages/v1alpha1/chart.go

@@ -12,7 +12,7 @@ import (
 	"strings"
 
 	"github.com/bufbuild/connect-go"
-	sourcev1 "github.com/fluxcd/source-controller/api/v1beta2"
+	sourcev1beta2 "github.com/fluxcd/source-controller/api/v1beta2"
 	corev1 "github.com/vmware-tanzu/kubeapps/cmd/kubeapps-apis/gen/core/packages/v1alpha1"
 	"github.com/vmware-tanzu/kubeapps/cmd/kubeapps-apis/plugins/fluxv2/packages/v1alpha1/cache"
 	"github.com/vmware-tanzu/kubeapps/cmd/kubeapps-apis/plugins/fluxv2/packages/v1alpha1/common"
@@ -27,12 +27,12 @@ import (
 	"sigs.k8s.io/yaml"
 )
 
-func (s *Server) getChartInCluster(ctx context.Context, headers http.Header, key types.NamespacedName) (*sourcev1.HelmChart, error) {
+func (s *Server) getChartInCluster(ctx context.Context, headers http.Header, key types.NamespacedName) (*sourcev1beta2.HelmChart, error) {
 	client, err := s.getClient(headers, key.Namespace)
 	if err != nil {
 		return nil, err
 	}
-	var chartObj sourcev1.HelmChart
+	var chartObj sourcev1beta2.HelmChart
 	if err = client.Get(ctx, key, &chartObj); err != nil {
 		return nil, connecterror.FromK8sError("get", "HelmChart", key.String(), err)
 	}

cmd/kubeapps-apis/plugins/fluxv2/packages/v1alpha1/chart_integration_test.go

@@ -12,7 +12,7 @@ import (
 	"testing"
 	"time"
 
-	sourcev1 "github.com/fluxcd/source-controller/api/v1beta2"
+	sourcev1beta2 "github.com/fluxcd/source-controller/api/v1beta2"
 	grpc_retry "github.com/grpc-ecosystem/go-grpc-middleware/retry"
 	corev1 "github.com/vmware-tanzu/kubeapps/cmd/kubeapps-apis/gen/core/packages/v1alpha1"
 	fluxplugin "github.com/vmware-tanzu/kubeapps/cmd/kubeapps-apis/gen/plugins/fluxv2/packages/v1alpha1"
@@ -329,12 +329,12 @@ func TestKindClusterRepoAndChartRBAC(t *testing.T) {
 	rules := map[string][]rbacv1.PolicyRule{
 		names[1].Namespace: {
 			{
-				APIGroups: []string{sourcev1.GroupVersion.Group},
+				APIGroups: []string{sourcev1beta2.GroupVersion.Group},
 				Resources: []string{fluxHelmRepositories},
 				Verbs:     []string{"get", "list"},
 			},
 			{
-				APIGroups: []string{sourcev1.GroupVersion.Group},
+				APIGroups: []string{sourcev1beta2.GroupVersion.Group},
 				Resources: []string{"helmcharts"},
 				Verbs:     []string{"get", "list"},
 			},

cmd/kubeapps-apis/plugins/fluxv2/packages/v1alpha1/chart_test.go

@@ -15,7 +15,8 @@ import (
 
 	"github.com/bufbuild/connect-go"
 	fluxmeta "github.com/fluxcd/pkg/apis/meta"
-	sourcev1 "github.com/fluxcd/source-controller/api/v1beta2"
+	sourcev1 "github.com/fluxcd/source-controller/api/v1"
+	sourcev1beta2 "github.com/fluxcd/source-controller/api/v1beta2"
 	redismock "github.com/go-redis/redismock/v8"
 	corev1 "github.com/vmware-tanzu/kubeapps/cmd/kubeapps-apis/gen/core/packages/v1alpha1"
 	"github.com/vmware-tanzu/kubeapps/cmd/kubeapps-apis/plugins/fluxv2/packages/v1alpha1/cache"
@@ -186,7 +187,7 @@ func TestGetAvailablePackageDetail(t *testing.T) {
 			}
 			defer ts2.Close()
 
-			s, mock, err := newServerWithRepos(t, []sourcev1.HelmRepository{*repo}, charts, secretObjs)
+			s, mock, err := newServerWithRepos(t, []sourcev1beta2.HelmRepository{*repo}, charts, secretObjs)
 			if err != nil {
 				t.Fatalf("%+v", err)
 			}
@@ -290,7 +291,7 @@ func TestTransientHttpFailuresAreRetriedForChartCache(t *testing.T) {
 		}
 		defer ts2.Close()
 
-		s, mock, err := newServerWithRepos(t, []sourcev1.HelmRepository{*repo}, charts, nil)
+		s, mock, err := newServerWithRepos(t, []sourcev1beta2.HelmRepository{*repo}, charts, nil)
 		if err != nil {
 			t.Fatalf("%+v", err)
 		}
@@ -467,7 +468,7 @@ func TestNonExistingRepoOrInvalidPkgVersionGetAvailablePackageDetail(t *testing.
 			}
 			defer ts2.Close()
 
-			s, mock, err := newServerWithRepos(t, []sourcev1.HelmRepository{*repo}, charts, nil)
+			s, mock, err := newServerWithRepos(t, []sourcev1beta2.HelmRepository{*repo}, charts, nil)
 			if err != nil {
 				t.Fatalf("%+v", err)
 			}
@@ -648,7 +649,7 @@ func TestGetAvailablePackageVersions(t *testing.T) {
 			}
 			defer ts.Close()
 
-			s, mock, err := newServerWithRepos(t, []sourcev1.HelmRepository{*repo}, charts, nil)
+			s, mock, err := newServerWithRepos(t, []sourcev1beta2.HelmRepository{*repo}, charts, nil)
 			if err != nil {
 				t.Fatalf("%+v", err)
 			}
@@ -735,7 +736,7 @@ func TestGetOciAvailablePackageVersions(t *testing.T) {
 				t.Fatal(err)
 			}
 
-			s, mock, err := newServerWithRepos(t, []sourcev1.HelmRepository{*repo}, tc.charts, nil)
+			s, mock, err := newServerWithRepos(t, []sourcev1beta2.HelmRepository{*repo}, tc.charts, nil)
 			if err != nil {
 				t.Fatalf("%+v", err)
 			}
@@ -978,14 +979,14 @@ func TestChartWithRelativeURL(t *testing.T) {
 		}
 	}))
 
-	repoSpec := &sourcev1.HelmRepositorySpec{
+	repoSpec := &sourcev1beta2.HelmRepositorySpec{
 		URL:      ts.URL,
 		Interval: metav1.Duration{Duration: 1 * time.Minute},
 	}
 
-	repoStatus := &sourcev1.HelmRepositoryStatus{
+	repoStatus := &sourcev1beta2.HelmRepositoryStatus{
 		Artifact: &sourcev1.Artifact{
-			Checksum:       "651f952130ea96823711d08345b85e82be011dc6",
+			Digest:         "651f952130ea96823711d08345b85e82be011dc6",
 			LastUpdateTime: metav1.Time{Time: lastUpdateTime},
 			Revision:       "651f952130ea96823711d08345b85e82be011dc6",
 		},
@@ -1002,7 +1003,7 @@ func TestChartWithRelativeURL(t *testing.T) {
 	defer ts.Close()
 
 	s, mock, err := newServerWithRepos(t,
-		[]sourcev1.HelmRepository{repo},
+		[]sourcev1beta2.HelmRepository{repo},
 		[]testSpecChartWithUrl{
 			{
 				chartID:       fmt.Sprintf("%s/airflow", repoName),
@@ -1072,7 +1073,7 @@ func TestGetOciAvailablePackageDetail(t *testing.T) {
 				t.Fatal(err)
 			}
 
-			s, mock, err := newServerWithRepos(t, []sourcev1.HelmRepository{*repo}, tc.charts, nil)
+			s, mock, err := newServerWithRepos(t, []sourcev1beta2.HelmRepository{*repo}, tc.charts, nil)
 			if err != nil {
 				t.Fatalf("%+v", err)
 			}
@@ -1128,8 +1129,8 @@ func TestGetOciAvailablePackageDetail(t *testing.T) {
 	}
 }
 
-func newChart(name, namespace string, spec *sourcev1.HelmChartSpec, status *sourcev1.HelmChartStatus) sourcev1.HelmChart {
-	helmChart := sourcev1.HelmChart{
+func newChart(name, namespace string, spec *sourcev1beta2.HelmChartSpec, status *sourcev1beta2.HelmChartStatus) sourcev1beta2.HelmChart {
+	helmChart := sourcev1beta2.HelmChart{
 		ObjectMeta: metav1.ObjectMeta{
 			Name:       name,
 			Generation: int64(1),

cmd/kubeapps-apis/plugins/fluxv2/packages/v1alpha1/common/utils.go

@@ -5,6 +5,7 @@ package common
 
 import (
 	"bytes"
+	"context"
 	"crypto/sha256"
 	"encoding/base64"
 	"encoding/json"
@@ -23,8 +24,8 @@ import (
 	"github.com/bufbuild/connect-go"
 	"github.com/docker/cli/cli/config"
 	"github.com/docker/cli/cli/config/credentials"
-	helmv2 "github.com/fluxcd/helm-controller/api/v2beta1"
-	sourcev1 "github.com/fluxcd/source-controller/api/v1beta2"
+	helmv2beta2 "github.com/fluxcd/helm-controller/api/v2beta2"
+	sourcev1beta2 "github.com/fluxcd/source-controller/api/v1beta2"
 	"github.com/go-redis/redis/v8"
 	"github.com/google/go-containerregistry/pkg/authn"
 	plugins "github.com/vmware-tanzu/kubeapps/cmd/kubeapps-apis/gen/core/plugins/v1alpha1"
@@ -67,20 +68,20 @@ func init() {
 	}
 
 	repositoriesGvr = schema.GroupVersionResource{
-		Group:    sourcev1.GroupVersion.Group,
-		Version:  sourcev1.GroupVersion.Version,
+		Group:    sourcev1beta2.GroupVersion.Group,
+		Version:  sourcev1beta2.GroupVersion.Version,
 		Resource: "helmrepositories",
 	}
 
 	chartsGvr = schema.GroupVersionResource{
-		Group:    sourcev1.GroupVersion.Group,
-		Version:  sourcev1.GroupVersion.Version,
+		Group:    sourcev1beta2.GroupVersion.Group,
+		Version:  sourcev1beta2.GroupVersion.Version,
 		Resource: "helmcharts",
 	}
 
 	releasesGvr = schema.GroupVersionResource{
-		Group:    helmv2.GroupVersion.Group,
-		Version:  helmv2.GroupVersion.Version,
+		Group:    helmv2beta2.GroupVersion.Group,
+		Version:  helmv2beta2.GroupVersion.Version,
 		Resource: "helmreleases",
 	}
 }
@@ -180,24 +181,23 @@ func NewRedisClientFromEnv(stopCh <-chan struct{}) (*redis.Client, error) {
 
 	// ref https://github.com/vmware-tanzu/kubeapps/pull/4382#discussion_r820386531
 	var redisCli *redis.Client
-	err = wait.PollImmediate(redisInitClientRetryWait, redisInitClientTimeout,
-		func() (bool, error) {
-			redisCli = redis.NewClient(&redis.Options{
-				Addr:     REDIS_ADDR,
-				Password: REDIS_PASSWORD,
-				DB:       REDIS_DB_NUM,
-			})
-
-			// ping redis to make sure client is connected
-			var pong string
-			if pong, err = redisCli.Ping(redisCli.Context()).Result(); err == nil {
-				log.Infof("Redis [PING]: %s", pong)
-				return true, nil
-			}
-			log.Infof("Waiting %s before retrying to due to %v...", redisInitClientRetryWait.String(), err)
-			return false, nil
+	err = wait.PollUntilContextTimeout(context.Background(), redisInitClientRetryWait, redisInitClientTimeout, true, func(ctx context.Context) (bool, error) {
+		redisCli = redis.NewClient(&redis.Options{
+			Addr:     REDIS_ADDR,
+			Password: REDIS_PASSWORD,
+			DB:       REDIS_DB_NUM,
 		})
 
+		// ping redis to make sure client is connected
+		var pong string
+		if pong, err = redisCli.Ping(redisCli.Context()).Result(); err == nil {
+			log.Infof("Redis [PING]: %s", pong)
+			return true, nil
+		}
+		log.Infof("Waiting %s before retrying to due to %v...", redisInitClientRetryWait.String(), err)
+		return false, nil
+	})
+
 	if err != nil {
 		return nil, fmt.Errorf("initializing redis client failed after timeout of %s was reached, error: %v", redisInitClientTimeout.String(), err)
 	}