Use server side checking for the deleted manifests

There are some new methods:
1. GetPodsByNamespace() used to get CVO pod
2. ListFilesInPodContainer() used to list files in a pod container, here I use a customized command to reduce files (get file content file takes long time)
3. GetFileContentInPodContainer() used to get file content from a pod container
4. ParseManifest() parse yaml file content to Manifest object.

Author: JianLi-RH

test/cvo/cvo.go

@@ -5,21 +5,16 @@ package cvo
 import (
 	"context"
 	"fmt"
-	"os"
-	"path/filepath"
-	"time"
+	"strings"
 
 	g "github.com/onsi/ginkgo/v2"
 	o "github.com/onsi/gomega"
 
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-	"k8s.io/apimachinery/pkg/util/sets"
 	"k8s.io/client-go/kubernetes"
 	"k8s.io/client-go/rest"
 
-	"github.com/openshift/library-go/pkg/manifest"
-
 	"github.com/openshift/cluster-version-operator/pkg/cvo/external/dynamicclient"
 	"github.com/openshift/cluster-version-operator/pkg/external"
 	"github.com/openshift/cluster-version-operator/test/oc"
@@ -118,50 +113,40 @@ var _ = g.Describe(`[Jira:"Cluster Version Operator"] cluster-version-operator`,
 		err = util.SkipIfNetworkRestricted(ctx, restCfg, util.FauxinnatiAPIURL)
 		o.Expect(err).NotTo(o.HaveOccurred(), "Failed to determine if cluster is network restricted")
 
-		// Initialize the ocapi.OC instance
-		g.By("Setting up oc")
-		ocClient, err := oc.NewOC(ocapi.Options{Logger: logger, Timeout: 90 * time.Second})
+		pods, err := util.GetPodsByNamespace(ctx, kubeClient, external.DefaultCVONamespace, map[string]string{"k8s-app": "cluster-version-operator"})
 		o.Expect(err).NotTo(o.HaveOccurred())
+		logger.Info("CVO pod found", "name", pods[0].Name)
 
-		g.By("Extracting manifests in the release")
 		annotation := "release.openshift.io/delete"
-		tempDir, err := os.MkdirTemp("", "OTA-42543-manifest-")
-		o.Expect(err).NotTo(o.HaveOccurred(), "create temp manifest dir failed")
-
-		authFile, err := util.GetAuthFile(context.Background(), kubeClient, "openshift-config", "pull-secret", ".dockerconfigjson")
+		manifestDir := "/release-manifests/"
+		command := []string{"find", manifestDir, "-iname", "*.yaml", "-exec", "grep", "-l", annotation, "{}", ";"}
+		files, err := util.ListFilesInPodContainer(ctx, restCfg, command, external.DefaultCVONamespace, pods[0].Name, "cluster-version-operator")
 		o.Expect(err).NotTo(o.HaveOccurred())
-		defer func() { _ = os.Remove(authFile) }()
-		manifestDir := ocapi.ReleaseExtractOptions{To: tempDir, AuthFile: authFile}
-		logger.Info(fmt.Sprintf("Extract manifests to: %s", manifestDir.To))
-		defer func() { _ = os.RemoveAll(manifestDir.To) }()
-		err = ocClient.AdmReleaseExtract(manifestDir)
-		o.Expect(err).NotTo(o.HaveOccurred(), "extracting manifests failed")
-
-		files, err := os.ReadDir(manifestDir.To)
-		o.Expect(err).NotTo(o.HaveOccurred())
-		g.By(fmt.Sprintf("Checking if getting manifests with %s on the cluster led to not-found error", annotation))
-		ignore := sets.New("release-metadata", "image-references")
-		for _, manifestFile := range files {
-			if manifestFile.IsDir() || ignore.Has(manifestFile.Name()) {
+		o.Expect(files).ToNot(o.BeEmpty(), "Expected to find files in manifests directory of CVO pod, but found none")
+
+		for _, f := range files {
+			fileContent, err := util.GetFileContentInPodContainer(ctx, restCfg, external.DefaultCVONamespace, pods[0].Name, "cluster-version-operator", f)
+			o.Expect(err).NotTo(o.HaveOccurred(), fmt.Sprintf("Failed to get content of file %s in CVO pod", f))
+			o.Expect(fileContent).ToNot(o.BeEmpty(), fmt.Sprintf("Expected to get content of file %s in CVO pod, but got empty content", f))
+
+			if !strings.Contains(fileContent, annotation) {
 				continue
 			}
-			filePath := filepath.Join(manifestDir.To, manifestFile.Name())
-			o.Expect(err).NotTo(o.HaveOccurred(), "failed to read manifest file")
-			manifests, err := manifest.ManifestsFromFiles([]string{filePath})
-			o.Expect(err).NotTo(o.HaveOccurred(), fmt.Sprintf("failed to parse manifest file: %s", filePath))
-
-			for _, ms := range manifests {
-				ann := ms.Obj.GetAnnotations()
-				if ann[annotation] != "true" {
-					continue
-				}
-				client, err := dynamicclient.New(restCfg, ms.GVK, ms.Obj.GetNamespace())
-				o.Expect(err).NotTo(o.HaveOccurred())
-				_, err = client.Get(ctx, ms.Obj.GetName(), metav1.GetOptions{})
-				o.Expect(apierrors.IsNotFound(err)).To(o.BeTrue(),
-					fmt.Sprintf("The deleted manifest should not be installed, but actually installed: manifest: %s %s in namespace %s from file %q, error: %v",
-						ms.GVK, ms.Obj.GetName(), ms.Obj.GetNamespace(), ms.OriginalFilename, err))
+
+			manifest, err := util.ParseManifest(fileContent)
+			o.Expect(err).NotTo(o.HaveOccurred(), fmt.Sprintf("Failed to parse manifest content of file %s in CVO pod", f))
+
+			ann := manifest.Obj.GetAnnotations()
+			if ann[annotation] != "true" {
+				continue
 			}
+			logger.Info("Checking file: ", f, ", GVK:", manifest.GVK.String(), ", Name: ", manifest.Obj.GetName(), ", Namespace: ", manifest.Obj.GetNamespace())
+			client, err := dynamicclient.New(restCfg, manifest.GVK, manifest.Obj.GetNamespace())
+			o.Expect(err).NotTo(o.HaveOccurred())
+			_, err = client.Get(ctx, manifest.Obj.GetName(), metav1.GetOptions{})
+			o.Expect(apierrors.IsNotFound(err)).To(o.BeTrue(),
+				fmt.Sprintf("The deleted manifest should not be installed, but actually installed: manifest: %s %s in namespace %s from file %q, error: %v",
+					manifest.GVK, manifest.Obj.GetName(), manifest.Obj.GetNamespace(), f, err))
 		}
 	})
 })

test/util/util.go

@@ -5,25 +5,30 @@ import (
 	"context"
 	"crypto/rand"
 	"fmt"
+	"io"
 	"os"
 	"path/filepath"
 	"strings"
 	"time"
 
 	g "github.com/onsi/ginkgo/v2"
 	o "github.com/onsi/gomega"
+	"github.com/pkg/errors"
 
 	corev1 "k8s.io/api/core/v1"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/labels"
 	"k8s.io/apimachinery/pkg/util/wait"
+	"k8s.io/apimachinery/pkg/util/yaml"
 	"k8s.io/client-go/kubernetes"
 	"k8s.io/client-go/kubernetes/scheme"
 	"k8s.io/client-go/rest"
 	"k8s.io/client-go/tools/clientcmd"
 	"k8s.io/client-go/tools/remotecommand"
 
+	libmanifest "github.com/openshift/library-go/pkg/manifest"
+
 	configv1 "github.com/openshift/api/config/v1"
 	clientconfigv1 "github.com/openshift/client-go/config/clientset/versioned"
 
@@ -125,6 +130,109 @@ func GetAuthFile(ctx context.Context, client kubernetes.Interface, ns string, se
 	return authFile, nil
 }
 
+// GetPodsByNamespace retrieves the list of pods in the specified namespace with the given label selector.
+func GetPodsByNamespace(ctx context.Context, client kubernetes.Interface, namespace string, labelSelector map[string]string) ([]corev1.Pod, error) {
+	podList, err := client.CoreV1().Pods(namespace).List(ctx, metav1.ListOptions{
+		LabelSelector: labels.FormatLabels(labelSelector),
+	})
+	if err != nil {
+		return nil, fmt.Errorf("error listing pods in namespace %s with label selector %v: %v", namespace, labelSelector, err)
+	}
+	return podList.Items, nil
+}
+
+// ListFilesInPodContainer executes the given command in the specified container of a pod and returns the output as a list of strings.
+func ListFilesInPodContainer(ctx context.Context, restConfig *rest.Config, command []string, namespace string, podName string, containerName string) ([]string, error) {
+	kubeClient, err := GetKubeClient(restConfig)
+	if err != nil {
+		return nil, err
+	}
+	results := []string{}
+	req := kubeClient.CoreV1().RESTClient().Post().
+		Resource("pods").
+		Name(podName).
+		Namespace(namespace).
+		SubResource("exec").
+		VersionedParams(&corev1.PodExecOptions{
+			Command:   command,
+			Container: containerName,
+			Stdout:    true,
+			Stderr:    true,
+		}, scheme.ParameterCodec)
+	exec, err := remotecommand.NewSPDYExecutor(restConfig, "POST", req.URL())
+	if err != nil {
+		return nil, fmt.Errorf("error creating executor for pod %s/%s: %v", namespace, podName, err)
+	}
+	var stdoutBuf, stderrBuf bytes.Buffer
+	err = exec.StreamWithContext(ctx, remotecommand.StreamOptions{
+		Stdout: &stdoutBuf,
+		Stderr: &stderrBuf,
+	})
+	if err != nil {
+		return nil, fmt.Errorf("error executing command in pod %s/%s: %v, stderr: %s", namespace, podName, err, stderrBuf.String())
+	}
+	files := strings.Split(stdoutBuf.String(), "\n")
+	for _, file := range files {
+		if file == "" {
+			continue
+		}
+		results = append(results, file)
+	}
+	return results, nil
+}
+
+// GetFileContentInPodContainer executes the command to read the content of a file in the specified container of a pod and returns the content as a string.
+func GetFileContentInPodContainer(ctx context.Context, restConfig *rest.Config, namespace string, podName string, containerName string, filePath string) (string, error) {
+	kubeClient, err := GetKubeClient(restConfig)
+	if err != nil {
+		return "", err
+	}
+	command := []string{"cat", filePath}
+	req := kubeClient.CoreV1().RESTClient().Post().
+		Resource("pods").
+		Name(podName).
+		Namespace(namespace).
+		SubResource("exec").
+		VersionedParams(&corev1.PodExecOptions{
+			Command:   command,
+			Container: containerName,
+			Stdout:    true,
+			Stderr:    true,
+		}, scheme.ParameterCodec)
+	exec, err := remotecommand.NewSPDYExecutor(restConfig, "POST", req.URL())
+	if err != nil {
+		return "", fmt.Errorf("error creating executor for pod %s/%s: %v", namespace, podName, err)
+	}
+	var stdoutBuf, stderrBuf bytes.Buffer
+	err = exec.StreamWithContext(ctx, remotecommand.StreamOptions{
+		Stdout: &stdoutBuf,
+		Stderr: &stderrBuf,
+	})
+	if err != nil {
+		return "", fmt.Errorf("error executing command in pod %s/%s: %v, stderr: %s", namespace, podName, err, stderrBuf.String())
+	}
+	return stdoutBuf.String(), nil
+}
+
+// ParseManifest parses the given file content as a Kubernetes manifest and returns a Manifest object.
+func ParseManifest(fileContent string) (libmanifest.Manifest, error) {
+	d := yaml.NewYAMLOrJSONDecoder(strings.NewReader(fileContent), 1024)
+	for {
+		m := libmanifest.Manifest{}
+		if err := d.Decode(&m); err != nil {
+			if err == io.EOF {
+				return m, nil
+			}
+			return m, errors.Wrapf(err, "error parsing")
+		}
+		m.Raw = bytes.TrimSpace(m.Raw)
+		if len(m.Raw) == 0 || bytes.Equal(m.Raw, []byte("null")) {
+			continue
+		}
+		return m, nil
+	}
+}
+
 // GetRestConfig loads the Kubernetes REST configuration from KUBECONFIG environment variable.
 func GetRestConfig() (*rest.Config, error) {
 	return clientcmd.NewNonInteractiveDeferredLoadingClientConfig(clientcmd.NewDefaultClientConfigLoadingRules(), &clientcmd.ConfigOverrides{}).ClientConfig()