Fix for #3610 All other fluxv2 plugin operations must also support UX request with cluster (#3620)

* switch from json to gob encoding when storing data in redis cache. All tests passing

* step 2

* step 3

* step 3

Author: gfichtenholt

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

@@ -363,7 +363,7 @@ func (c NamespacedResourceWatcherCache) onAddOrModify(add bool, unstructuredObj
 			log.Errorf("Failed to set value for object with key [%s] in cache due to: %v", key, err)
 			return err
 		} else {
-			log.Infof("Set value for key [%s] in cache", key)
+			log.V(4).Infof("Set value for key [%s] in cache", key)
 		}
 	}
 	return nil
@@ -456,7 +456,7 @@ func (c NamespacedResourceWatcherCache) listKeys(filters []string) ([]string, er
 			if err != nil {
 				return nil, err
 			}
-			log.Infof("listKeys: SCAN returned keys: %s, cursor: [%d]", keys, cursor)
+			log.V(4).Infof("listKeys: SCAN returned keys: %s, cursor: [%d]", keys, cursor)
 			for _, key := range keys {
 				redisKeys[key] = struct{}{}
 			}

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

@@ -37,10 +37,6 @@ import (
 //    at this point.
 // 3) run './kind-cluster-setup.sh deploy' once prior to these tests
 
-// TODO (gfichtenholt) currently core server's has broken logic inside plugins.go
-// createConfigGetterWithParams(...). Refer to my comment in there. I had to make suggested changes
-// locally to make these tests pass
-
 const (
 	// the only repo these tests use so far. This is local copy of the first few entries
 	// on "https://stefanprodan.github.io/podinfo/index.yaml" as of Sept 10 2021 with the chart
@@ -51,60 +47,72 @@ const (
 )
 
 type integrationTestCreateSpec struct {
-	testName             string
-	repoUrl              string
-	request              *corev1.CreateInstalledPackageRequest
-	expectedDetail       *corev1.InstalledPackageDetail
-	expectedPodPrefix    string
+	testName          string
+	repoUrl           string
+	request           *corev1.CreateInstalledPackageRequest
+	expectedDetail    *corev1.InstalledPackageDetail
+	expectedPodPrefix string
+	// different from expectedStatusCode due to async nature of install
 	expectInstallFailure bool
 	noCleanup            bool
-	unauthorized         bool
+	expectedStatusCode   codes.Code
 }
 
 func TestKindClusterCreateInstalledPackage(t *testing.T) {
 	fluxPluginClient := checkEnv(t)
 
 	testCases := []integrationTestCreateSpec{
 		{
-			testName:          "create test (simplest case)",
-			repoUrl:           podinfo_repo_url,
-			request:           create_request_basic,
-			expectedDetail:    expected_detail_basic,
-			expectedPodPrefix: "@TARGET_NS@-my-podinfo-",
+			testName:           "create test (simplest case)",
+			repoUrl:            podinfo_repo_url,
+			request:            create_request_basic,
+			expectedDetail:     expected_detail_basic,
+			expectedPodPrefix:  "@TARGET_NS@-my-podinfo-",
+			expectedStatusCode: codes.OK,
 		},
 		{
-			testName:          "create package (semver constraint)",
-			repoUrl:           podinfo_repo_url,
-			request:           create_request_semver_constraint,
-			expectedDetail:    expected_detail_semver_constraint,
-			expectedPodPrefix: "@TARGET_NS@-my-podinfo-2-",
+			testName:           "create package (semver constraint)",
+			repoUrl:            podinfo_repo_url,
+			request:            create_request_semver_constraint,
+			expectedDetail:     expected_detail_semver_constraint,
+			expectedPodPrefix:  "@TARGET_NS@-my-podinfo-2-",
+			expectedStatusCode: codes.OK,
 		},
 		{
-			testName:          "create package (reconcile options)",
-			repoUrl:           podinfo_repo_url,
-			request:           create_request_reconcile_options,
-			expectedDetail:    expected_detail_reconcile_options,
-			expectedPodPrefix: "@TARGET_NS@-my-podinfo-3-",
+			testName:           "create package (reconcile options)",
+			repoUrl:            podinfo_repo_url,
+			request:            create_request_reconcile_options,
+			expectedDetail:     expected_detail_reconcile_options,
+			expectedPodPrefix:  "@TARGET_NS@-my-podinfo-3-",
+			expectedStatusCode: codes.OK,
 		},
 		{
-			testName:          "create package (with values)",
-			repoUrl:           podinfo_repo_url,
-			request:           create_request_with_values,
-			expectedDetail:    expected_detail_with_values,
-			expectedPodPrefix: "@TARGET_NS@-my-podinfo-4-",
+			testName:           "create package (with values)",
+			repoUrl:            podinfo_repo_url,
+			request:            create_request_with_values,
+			expectedDetail:     expected_detail_with_values,
+			expectedPodPrefix:  "@TARGET_NS@-my-podinfo-4-",
+			expectedStatusCode: codes.OK,
 		},
 		{
 			testName:             "install fails",
 			repoUrl:              podinfo_repo_url,
 			request:              create_request_install_fails,
 			expectedDetail:       expected_detail_install_fails,
 			expectInstallFailure: true,
+			expectedStatusCode:   codes.OK,
 		},
 		{
-			testName:     "unauthorized",
-			repoUrl:      podinfo_repo_url,
-			request:      create_request_basic,
-			unauthorized: true,
+			testName:           "unauthorized",
+			repoUrl:            podinfo_repo_url,
+			request:            create_request_basic,
+			expectedStatusCode: codes.Unauthenticated,
+		},
+		{
+			testName:           "wrong cluster",
+			repoUrl:            podinfo_repo_url,
+			request:            create_request_wrong_cluster,
+			expectedStatusCode: codes.Unimplemented,
 		},
 	}
 
@@ -195,6 +203,7 @@ func TestKindClusterUpdateInstalledPackage(t *testing.T) {
 			request:      update_request_6,
 			unauthorized: true,
 		},
+		// TODO (gfichtenholt) test automatic upgrade to new version when it becomes available
 	}
 
 	grpcContext := newGrpcContext(t, "test-update-admin")
@@ -412,13 +421,13 @@ func createAndWaitForHelmRelease(t *testing.T, tc integrationTestCreateSpec, flu
 	}
 
 	ctx := grpcContext
-	if tc.unauthorized {
+	if tc.expectedStatusCode == codes.Unauthenticated {
 		ctx = context.TODO()
 	}
 	resp, err := fluxPluginClient.CreateInstalledPackage(ctx, tc.request)
-	if tc.unauthorized {
-		if status.Code(err) != codes.Unauthenticated {
-			t.Fatalf("Expected Unathenticated, got: %v", err)
+	if tc.expectedStatusCode != codes.OK {
+		if status.Code(err) != tc.expectedStatusCode {
+			t.Fatalf("Expected %v, got: %v", tc.expectedStatusCode, err)
 		}
 		return nil // done, nothing more to check
 	} else if err != nil {
@@ -481,7 +490,7 @@ func createAndWaitForHelmRelease(t *testing.T, tc integrationTestCreateSpec, flu
 }
 
 func waitUntilInstallCompletes(t *testing.T, fluxPluginClient fluxplugin.FluxV2PackagesServiceClient, grpcContext context.Context, installedPackageRef *corev1.InstalledPackageReference, expectInstallFailure bool) (actualResp *corev1.GetInstalledPackageDetailResponse) {
-	const maxWait = 25
+	const maxWait = 30
 	for i := 0; i <= maxWait; i++ {
 		resp2, err := fluxPluginClient.GetInstalledPackageDetail(
 			grpcContext,
@@ -521,6 +530,7 @@ var (
 		Name:                "my-podinfo",
 		TargetContext: &corev1.Context{
 			Namespace: "test-1",
+			Cluster:   KubeappsCluster,
 		},
 	}
 
@@ -549,6 +559,7 @@ var (
 		Name:                "my-podinfo-2",
 		TargetContext: &corev1.Context{
 			Namespace: "test-2",
+			Cluster:   KubeappsCluster,
 		},
 		PkgVersionReference: &corev1.VersionReference{
 			Version: "> 5",
@@ -580,6 +591,7 @@ var (
 		Name:                "my-podinfo-3",
 		TargetContext: &corev1.Context{
 			Namespace: "test-3",
+			Cluster:   KubeappsCluster,
 		},
 		ReconciliationOptions: &corev1.ReconciliationOptions{
 			Interval:           60,
@@ -615,6 +627,7 @@ var (
 		Name:                "my-podinfo-4",
 		TargetContext: &corev1.Context{
 			Namespace: "test-4",
+			Cluster:   KubeappsCluster,
 		},
 		Values: "{\"ui\": { \"message\": \"what we do in the shadows\" } }",
 	}
@@ -645,6 +658,7 @@ var (
 		Name:                "my-podinfo-5",
 		TargetContext: &corev1.Context{
 			Namespace: "test-5",
+			Cluster:   KubeappsCluster,
 		},
 		Values: "{\"replicaCount\": \"what we do in the shadows\" }",
 	}
@@ -682,6 +696,7 @@ var (
 		Name:                "my-podinfo-6",
 		TargetContext: &corev1.Context{
 			Namespace: "test-6",
+			Cluster:   KubeappsCluster,
 		},
 		PkgVersionReference: &corev1.VersionReference{
 			Version: "=5.2.1",
@@ -733,6 +748,7 @@ var (
 		Name:                "my-podinfo-7",
 		TargetContext: &corev1.Context{
 			Namespace: "test-7",
+			Cluster:   KubeappsCluster,
 		},
 		PkgVersionReference: &corev1.VersionReference{
 			Version: "=5.2.1",
@@ -785,6 +801,7 @@ var (
 		Name:                "my-podinfo-8",
 		TargetContext: &corev1.Context{
 			Namespace: "test-8",
+			Cluster:   KubeappsCluster,
 		},
 		PkgVersionReference: &corev1.VersionReference{
 			Version: "=5.2.1",
@@ -839,6 +856,7 @@ var (
 		Name:                "my-podinfo-9",
 		TargetContext: &corev1.Context{
 			Namespace: "test-9",
+			Cluster:   KubeappsCluster,
 		},
 		PkgVersionReference: &corev1.VersionReference{
 			Version: "=5.2.1",
@@ -892,6 +910,7 @@ var (
 		Name:                "my-podinfo-10",
 		TargetContext: &corev1.Context{
 			Namespace: "test-10",
+			Cluster:   KubeappsCluster,
 		},
 		PkgVersionReference: &corev1.VersionReference{
 			Version: "=5.2.1",
@@ -925,6 +944,7 @@ var (
 		Name:                "my-podinfo-11",
 		TargetContext: &corev1.Context{
 			Namespace: "test-11",
+			Cluster:   KubeappsCluster,
 		},
 	}
 
@@ -1000,6 +1020,7 @@ var (
 		Name:                "my-podinfo-12",
 		TargetContext: &corev1.Context{
 			Namespace: "test-12",
+			Cluster:   KubeappsCluster,
 		},
 		PkgVersionReference: &corev1.VersionReference{
 			Version: "=5.2.1",
@@ -1031,6 +1052,7 @@ var (
 		Name:                "my-podinfo-13",
 		TargetContext: &corev1.Context{
 			Namespace: "test-13",
+			Cluster:   KubeappsCluster,
 		},
 		PkgVersionReference: &corev1.VersionReference{
 			Version: "=5.2.1",
@@ -1056,4 +1078,13 @@ var (
 			"kubectl -n @TARGET_NS@ port-forward deploy/@TARGET_NS@-my-podinfo-13 8080:9898\n",
 		AvailablePackageRef: availableRef("podinfo-13/podinfo", "default"),
 	}
+
+	create_request_wrong_cluster = &corev1.CreateInstalledPackageRequest{
+		AvailablePackageRef: availableRef("podinfo-14/podinfo", "default"),
+		Name:                "my-podinfo",
+		TargetContext: &corev1.Context{
+			Namespace: "test-14",
+			Cluster:   "this is not the cluster you're looking for",
+		},
+	}
 )

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

@@ -13,8 +13,9 @@ limitations under the License.
 package main
 
 import (
+	"bytes"
 	"context"
-	"encoding/json"
+	"encoding/gob"
 	"fmt"
 	"time"
 
@@ -265,12 +266,13 @@ func onAddOrModifyRepo(key string, unstructuredRepo map[string]interface{}) (int
 			return nil, false, err
 		}
 
-		jsonBytes, err := json.Marshal(charts)
-		if err != nil {
+		// use gob encoding instead of json, it peforms much better
+		var buf bytes.Buffer
+		enc := gob.NewEncoder(&buf)
+		if err = enc.Encode(charts); err != nil {
 			return nil, false, err
 		}
-
-		return jsonBytes, true, nil
+		return buf.Bytes(), true, nil
 	} else {
 		// repo is not quite ready to be indexed - not really an error condition,
 		// just skip it eventually there will be another event when it is in ready state
@@ -280,16 +282,17 @@ func onAddOrModifyRepo(key string, unstructuredRepo map[string]interface{}) (int
 }
 
 func onGetRepo(key string, value interface{}) (interface{}, error) {
-	bytes, ok := value.([]byte)
+	b, ok := value.([]byte)
 	if !ok {
 		return nil, status.Errorf(codes.Internal, "unexpected value found in cache for key [%s]: %v", key, value)
 	}
 
+	dec := gob.NewDecoder(bytes.NewReader(b))
 	var charts []models.Chart
-	err := json.Unmarshal(bytes, &charts)
-	if err != nil {
+	if err := dec.Decode(&charts); err != nil {
 		return nil, err
 	}
+
 	return charts, nil
 }
 

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

@@ -55,6 +55,7 @@ type Server struct {
 // NewServer returns a Server automatically configured with a function to obtain
 // the k8s client config.
 func NewServer(configGetter server.KubernetesConfigGetter, kubeappsCluster string) (*Server, error) {
+	log.Infof("+fluxv2 NewServer(kubeappsCluster: [%v])", kubeappsCluster)
 	repositoriesGvr := schema.GroupVersionResource{
 		Group:    fluxGroup,
 		Version:  fluxVersion,
@@ -72,8 +73,8 @@ func NewServer(configGetter server.KubernetesConfigGetter, kubeappsCluster strin
 		return nil, err
 	} else {
 		return &Server{
-			clientGetter:       newClientGetter(configGetter),
-			actionConfigGetter: newHelmActionConfigGetter(configGetter),
+			clientGetter:       newClientGetter(configGetter, kubeappsCluster),
+			actionConfigGetter: newHelmActionConfigGetter(configGetter, kubeappsCluster),
 			cache:              cache,
 			kubeappsCluster:    kubeappsCluster,
 		}, nil
@@ -113,7 +114,8 @@ func (s *Server) GetPackageRepositories(ctx context.Context, request *v1alpha1.G
 		return nil, status.Errorf(codes.InvalidArgument, "no context provided")
 	}
 
-	if request.Context.Cluster != "" {
+	cluster := request.GetContext().GetCluster()
+	if cluster != "" && cluster != s.kubeappsCluster {
 		return nil, status.Errorf(
 			codes.Unimplemented,
 			"not supported yet: request.Context.Cluster: [%v]",
@@ -191,10 +193,14 @@ func (s *Server) GetAvailablePackageSummaries(ctx context.Context, request *core
 	if pageSize > 0 && len(packageSummaries) == int(pageSize) {
 		nextPageToken = fmt.Sprintf("%d", pageOffset+1)
 	}
+
 	return &corev1.GetAvailablePackageSummariesResponse{
 		AvailablePackageSummaries: packageSummaries,
 		NextPageToken:             nextPageToken,
 		// TODO (gfichtenholt) Categories?
+		// Just happened to notice that helm plug-in returning this.
+		// Never discussed this and the design doc appears to have a lot of back-and-forth comments
+		// about this, semantics aren't very clear
 	}, nil
 }
 
@@ -366,7 +372,8 @@ func (s *Server) CreateInstalledPackage(ctx context.Context, request *corev1.Cre
 	if request.TargetContext == nil || request.TargetContext.Namespace == "" {
 		return nil, status.Errorf(codes.InvalidArgument, "no request TargetContext namespace provided")
 	}
-	if request.TargetContext.Cluster != "" {
+	cluster := request.GetTargetContext().GetCluster()
+	if cluster != "" && cluster != s.kubeappsCluster {
 		return nil, status.Errorf(
 			codes.Unimplemented,
 			"not supported yet: request.TargetContext.Cluster: [%v]",