Improve must-gather image handling and annotation checks

pkg/cli/admin/mustgather/mustgather.go

@@ -258,13 +258,15 @@ func (o *MustGatherOptions) Complete(f kcmdutil.Factory, cmd *cobra.Command, arg
 }
 
 func (o *MustGatherOptions) completeImages() error {
+	// Resolve explicitly provided image streams
 	for _, imageStream := range o.ImageStreams {
 		if image, err := o.resolveImageStreamTagString(imageStream); err == nil {
 			o.Images = append(o.Images, image)
 		} else {
 			return fmt.Errorf("unable to resolve image stream '%v': %v", imageStream, err)
 		}
 	}
+	// If no images or --all flag is set, use default must-gather image
 	if len(o.Images) == 0 || o.AllImages {
 		var image string
 		var err error
@@ -279,14 +281,29 @@ func (o *MustGatherOptions) completeImages() error {
 		pluginImages := make(map[string]struct{})
 		var err error
 
-		pluginImages, err = o.annotatedCSVs()
+		// Annotated CSVs (via dynamic client)
+		csvGVR := schema.GroupVersionResource{
+			Group:    "operators.coreos.com",
+			Version:  "v1alpha1",
+			Resource: "clusterserviceversions",
+		}
+		uList, err := o.DynamicClient.Resource(csvGVR).List(context.TODO(), metav1.ListOptions{})
 		if err != nil {
-			return err
+			return fmt.Errorf("failed to list CSVs: %v", err)
+		}
+		for _, u := range uList.Items {
+			ann := u.GetAnnotations()
+			if v, ok := ann[mgAnnotation]; ok {
+				pluginImages[v] = struct{}{}
+			} else {
+				o.log("WARNING: CSV operator %s doesn't have the must-gather-image annotation.", u.GetName())
+			}
 		}
 
+		// Annotated ClusterOperators
 		cos, err := o.ConfigClient.ConfigV1().ClusterOperators().List(context.TODO(), metav1.ListOptions{})
 		if err != nil {
-			return err
+			return fmt.Errorf("failed to list ClusterOperators: %v", err)
 		}
 		for _, item := range cos.Items {
 			ann := item.GetAnnotations()