openshift: add functions for getting pod logs

Signed-off-by: Caleb Xu <caxu@redhat.com>

Author: caxu-rh

internal/openshift/openshift.go

@@ -1,8 +1,10 @@
 package openshift
 
 import (
+	"bytes"
 	"context"
 	"fmt"
+	"io"
 	"maps"
 
 	"github.com/go-logr/logr"
@@ -18,19 +20,22 @@ import (
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	apiruntime "k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/client-go/kubernetes"
 	crclient "sigs.k8s.io/controller-runtime/pkg/client"
 )
 
 type openshiftClient struct {
-	Client crclient.Client
+	Client       crclient.Client
+	K8sInterface kubernetes.Interface
 }
 
 // NewClient provides a wrapper around the passed in client in
 // order to present convenience functions for each of the object
 // types that are interacted with.
-func NewClient(client crclient.Client) Client {
+func NewClient(client crclient.Client, k8sInterface kubernetes.Interface) Client {
 	var osclient Client = &openshiftClient{
-		Client: client,
+		Client:       client,
+		K8sInterface: k8sInterface,
 	}
 	return osclient
 }
@@ -546,3 +551,82 @@ func (oe *openshiftClient) GetDeploymentPods(ctx context.Context, name string, n
 
 	return podList.Items, nil
 }
+
+// GetPod can return an ErrNotFound
+func (oe *openshiftClient) GetPod(ctx context.Context, name string, namespace string) (*corev1.Pod, error) {
+	logger := logr.FromContextOrDiscard(ctx)
+
+	logger.V(log.TRC).Info("fetching pod", "namespace", namespace, "name", name)
+	pod := corev1.Pod{}
+	err := oe.Client.Get(ctx, crclient.ObjectKey{
+		Name:      name,
+		Namespace: namespace,
+	}, &pod)
+	if apierrors.IsNotFound(err) {
+		return nil, fmt.Errorf("could not retrieve pod: %s/%s: %w: %v", namespace, name, ErrNotFound, err)
+	}
+	if err != nil {
+		return nil, fmt.Errorf("could not retrieve pod: %s/%s: %v", namespace, name, err)
+	}
+	return &pod, nil
+}
+
+func (oe *openshiftClient) getContainerLogs(ctx context.Context, namespace string, pod string, container string) (*bytes.Buffer, error) {
+	podLogOpts := corev1.PodLogOptions{
+		Container: container,
+	}
+	req := oe.K8sInterface.CoreV1().Pods(namespace).GetLogs(pod, &podLogOpts)
+	logs, err := req.Stream(ctx)
+	if err != nil {
+		return nil, fmt.Errorf("failed to open log stream: %s/%s/%s: %v", namespace, pod, container, err)
+	}
+	defer logs.Close()
+	buf := new(bytes.Buffer)
+	_, err = io.Copy(buf, logs)
+	if err != nil {
+		return nil, fmt.Errorf("failed to copy log stream: %s/%s/%s: %v", namespace, pod, container, err)
+	}
+
+	return buf, nil
+}
+
+// GetPodLogs can return an ErrNotFound
+func (oe *openshiftClient) GetPodLogs(ctx context.Context, name string, namespace string) (map[string]*bytes.Buffer, error) {
+	logger := logr.FromContextOrDiscard(ctx)
+
+	pod, err := oe.GetPod(ctx, name, namespace)
+	if err != nil {
+		return nil, err
+	}
+
+	results := map[string]*bytes.Buffer{}
+
+	for _, container := range pod.Spec.InitContainers {
+		buf, err := oe.getContainerLogs(ctx, namespace, name, container.Name)
+		if err != nil {
+			logger.V(log.TRC).Info("failed to get pod logs", "error", err)
+			continue
+		}
+		results[container.Name] = buf
+	}
+
+	for _, container := range pod.Spec.Containers {
+		buf, err := oe.getContainerLogs(ctx, namespace, name, container.Name)
+		if err != nil {
+			logger.V(log.TRC).Info("failed to get pod logs", "error", err)
+			continue
+		}
+		results[container.Name] = buf
+	}
+
+	for _, container := range pod.Spec.EphemeralContainers {
+		buf, err := oe.getContainerLogs(ctx, namespace, name, container.Name)
+		if err != nil {
+			logger.V(log.TRC).Info("failed to get pod logs", "error", err)
+			continue
+		}
+		results[container.Name] = buf
+	}
+
+	return results, nil
+}

internal/openshift/openshift_test.go

@@ -12,7 +12,8 @@ import (
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	apiruntime "k8s.io/apimachinery/pkg/runtime"
-	"sigs.k8s.io/controller-runtime/pkg/client/fake"
+	fakecg "k8s.io/client-go/kubernetes/fake"
+	fakecr "sigs.k8s.io/controller-runtime/pkg/client/fake"
 )
 
 var _ = Describe("OpenShift Engine", func() {
@@ -152,12 +153,13 @@ var _ = Describe("OpenShift Engine", func() {
 		scheme := apiruntime.NewScheme()
 		Expect(AddSchemes(scheme)).To(Succeed())
 		Expect(appsv1.AddToScheme(scheme)).To(Succeed())
-		cl := fake.NewClientBuilder().
+		cl := fakecr.NewClientBuilder().
 			WithScheme(scheme).
 			WithObjects(&csv, &deployment1, &deployment2, &deployment3).
 			WithLists(&pods, &isList).
 			Build()
-		oc = NewClient(cl)
+		cls := fakecg.NewClientset()
+		oc = NewClient(cl, cls)
 	})
 	Context("Namespaces", func() {
 		It("should exercise Namespaces", func() {
@@ -439,4 +441,24 @@ var _ = Describe("OpenShift Engine", func() {
 			Expect(pods).To(BeNil())
 		})
 	})
+	Context("Pods", func() {
+		It("should get a Pod", func() {
+			pod, err := oc.GetPod(context.TODO(), "pod1", "testns")
+			Expect(err).ToNot(HaveOccurred())
+			Expect(pod).ToNot(BeNil())
+		})
+		It("should error if Pod doesn't exist", func() {
+			pod, err := oc.GetPod(context.TODO(), "pod3", "badns")
+			Expect(err).To(HaveOccurred())
+			Expect(err).To(MatchError(ErrNotFound))
+			Expect(pod).To(BeNil())
+		})
+		It("should get Pod logs", func() {
+			logs, err := oc.GetPodLogs(context.TODO(), "pod1", "testns")
+			Expect(err).ToNot(HaveOccurred())
+			Expect(len(logs)).To(Equal(2))
+			Expect(logs["cont1"].String()).To(Equal("fake logs"))
+			Expect(logs["cont2"].String()).To(Equal("fake logs"))
+		})
+	})
 })

internal/openshift/types.go

@@ -1,6 +1,7 @@
 package openshift
 
 import (
+	"bytes"
 	"context"
 
 	operatorsv1 "github.com/operator-framework/api/pkg/operators/v1"
@@ -59,4 +60,6 @@ type Client interface {
 	DeleteRoleBinding(ctx context.Context, name string, namespace string) error
 	GetDeployment(ctx context.Context, name string, namespace string) (*appsv1.Deployment, error)
 	GetDeploymentPods(ctx context.Context, name string, namespace string) ([]corev1.Pod, error)
+	GetPod(ctx context.Context, name string, namespace string) (*corev1.Pod, error)
+	GetPodLogs(ctx context.Context, name string, namespace string) (map[string]*bytes.Buffer, error)
 }

internal/policy/operator/deployable_by_olm.go

@@ -30,6 +30,7 @@ import (
 	apiruntime "k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/apimachinery/pkg/util/validation"
+	"k8s.io/client-go/kubernetes"
 	ctrl "sigs.k8s.io/controller-runtime"
 	crclient "sigs.k8s.io/controller-runtime/pkg/client"
 )
@@ -61,6 +62,7 @@ type DeployableByOlmCheck struct {
 
 	openshiftClient     openshift.Client
 	client              crclient.Client
+	k8sClientset        kubernetes.Interface
 	csvReady            bool
 	validImages         bool
 	csvTimeout          time.Duration
@@ -96,12 +98,20 @@ func (p *DeployableByOlmCheck) initClient() error {
 	}
 
 	p.client = client
+
+	k8sClientset, err := kubernetes.NewForConfig(kubeconfig)
+	if err != nil {
+		return fmt.Errorf("could not get k8s clientset: %w", err)
+	}
+
+	p.k8sClientset = k8sClientset
+
 	return nil
 }
 
 func (p *DeployableByOlmCheck) initOpenShiftEngine() {
 	if p.openshiftClient == nil {
-		p.openshiftClient = openshift.NewClient(p.client)
+		p.openshiftClient = openshift.NewClient(p.client, p.k8sClientset)
 	}
 }