Add option to save pod container logs (#232)

* add logic go get pod container logs

* split into files

* save files in controller

* refactor method

Author: bakito

pkg/config/types.go

@@ -5,6 +5,7 @@ import (
 	"fmt"
 	"net/http"
 	"os"
+	"path/filepath"
 	"strings"
 	"time"
 
@@ -39,6 +40,8 @@ type Config struct {
 	CallbackServicePort int    `json:"callbackServicePort"`
 	// LeaderElectionResourceLock resource lock type. if empty default (resourcelock.ConfigMapsLeasesResourceLock) is used
 	LeaderElectionResourceLock string `json:"leaderElectionResourceLock,omitempty"`
+	// SavePodLog if enabled, pod logs are saved along other with other job files
+	SavePodLog bool `json:"savePodLog"`
 
 	Namespace      string         `json:"-"`
 	JobPodTemplate string         `json:"-"`
@@ -69,6 +72,14 @@ func (cfg *Config) ReportDirExistsChecker() healthz.Checker {
 	}
 }
 
+func (cfg Config) MkReportDir(executionID string) error {
+	return os.MkdirAll(filepath.Join(cfg.ReportDirectory, executionID), 0o755)
+}
+
+func (cfg Config) ReportFileName(executionID string, name string) string {
+	return filepath.Join(cfg.ReportDirectory, executionID, name)
+}
+
 // Metrics config
 type Metrics struct {
 	Port   int               `json:"port"`

pkg/controller/controller.go

@@ -1,16 +1,21 @@
 package controller
 
 import (
+	"bytes"
 	"context"
 	"errors"
+	"fmt"
+	"io"
+	"io/ioutil"
 
 	"github.com/bakito/batch-job-controller/pkg/lifecycle"
+	"github.com/go-logr/logr"
 	corev1 "k8s.io/api/core/v1"
 	k8serrors "k8s.io/apimachinery/pkg/api/errors"
-	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/client-go/kubernetes"
+	corev1client "k8s.io/client-go/kubernetes/typed/core/v1"
 	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/client"
-	"sigs.k8s.io/controller-runtime/pkg/event"
 	"sigs.k8s.io/controller-runtime/pkg/handler"
 	"sigs.k8s.io/controller-runtime/pkg/log"
 	"sigs.k8s.io/controller-runtime/pkg/reconcile"
@@ -27,11 +32,17 @@ const (
 // PodReconciler reconciler
 type PodReconciler struct {
 	client.Client
+	coreClient corev1client.CoreV1Interface
 	Controller lifecycle.Controller
 }
 
 // SetupWithManager setup
 func (r *PodReconciler) SetupWithManager(mgr ctrl.Manager) error {
+	clientset, err := kubernetes.NewForConfig(mgr.GetConfig())
+	if err != nil {
+		return err
+	}
+	r.coreClient = clientset.CoreV1()
 	return ctrl.NewControllerManagedBy(mgr).
 		For(&corev1.Pod{}).
 		Watches(&source.Kind{Type: &corev1.Pod{}}, &handler.EnqueueRequestForObject{}).
@@ -59,40 +70,60 @@ func (r *PodReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.R
 	executionID := pod.GetLabels()[LabelExecutionID]
 	node := pod.Spec.NodeName
 
-	switch pod.Status.Phase {
-	case corev1.PodSucceeded:
-		err = r.Controller.PodTerminated(executionID, node, pod.Status.Phase)
-	case corev1.PodFailed:
-		err = r.Controller.PodTerminated(executionID, node, pod.Status.Phase)
-	}
-	if err != nil {
-		if !errors.Is(err, &lifecycle.ExecutionIDNotFound{}) {
-			podLog.Error(err, "unexpected error")
-			return reconcile.Result{}, err
+	if pod.Status.Phase == corev1.PodSucceeded || pod.Status.Phase == corev1.PodFailed {
+		if r.Controller.Config().SavePodLog {
+			r.savePodLogs(ctx, pod, podLog, executionID)
+		}
+		if err := r.Controller.PodTerminated(executionID, node, pod.Status.Phase); err != nil {
+			if !errors.Is(err, &lifecycle.ExecutionIDNotFound{}) {
+				podLog.Error(err, "unexpected error")
+				return reconcile.Result{}, err
+			}
 		}
 	}
 
 	return reconcile.Result{}, nil
 }
 
-type podPredicate struct{}
-
-func (podPredicate) Create(e event.CreateEvent) bool {
-	return matches(e.Object)
+func (r *PodReconciler) savePodLogs(ctx context.Context, pod *corev1.Pod, podLog logr.Logger, executionID string) {
+	for _, c := range pod.Spec.Containers {
+		clog := podLog.WithValues("container", c.Name)
+		if l, err := r.getPodLog(ctx, pod.Namespace, pod.Name, c.Name); err != nil {
+			clog.Info("could not get log of container")
+		} else {
+			if fileName, err := r.savePodLog(executionID, c.Name, l); err != nil {
+				clog.Error(err, "error saving container log file")
+			} else {
+				clog.WithValues("name", fileName).Info("saved container log file")
+			}
+		}
+	}
 }
 
-func (podPredicate) Update(e event.UpdateEvent) bool {
-	return matches(e.ObjectNew)
-}
+func (r *PodReconciler) getPodLog(ctx context.Context, namespace string, name string, containerName string) (string, error) {
+	podLogOpts := corev1.PodLogOptions{
+		Container: containerName,
+	}
+	req := r.coreClient.Pods(namespace).GetLogs(name, &podLogOpts)
+	podLogs, err := req.Stream(ctx)
+	if err != nil {
+		return "", err
+	}
+	defer func() { _ = podLogs.Close() }()
 
-func (podPredicate) Delete(e event.DeleteEvent) bool {
-	return matches(e.Object)
-}
+	buf := new(bytes.Buffer)
+	if _, err = io.Copy(buf, podLogs); err != nil {
+		return "", err
+	}
+	str := buf.String()
 
-func (podPredicate) Generic(e event.GenericEvent) bool {
-	return matches(e.Object)
+	return str, nil
 }
 
-func matches(m metav1.Object) bool {
-	return m.GetLabels()[LabelExecutionID] != "" && m.GetLabels()[LabelOwner] != ""
+func (r *PodReconciler) savePodLog(executionID string, name string, data string) (string, error) {
+	if err := r.Controller.Config().MkReportDir(executionID); err != nil {
+		return "", err
+	}
+	fileName := r.Controller.Config().ReportFileName(executionID, fmt.Sprintf("container-%s.log", name))
+	return fileName, ioutil.WriteFile(fileName, []byte(data), 0o600)
 }

pkg/controller/controller_test.go

@@ -3,18 +3,24 @@ package controller
 import (
 	"context"
 	"fmt"
+	"os"
 
+	"github.com/bakito/batch-job-controller/pkg/config"
 	mock_client "github.com/bakito/batch-job-controller/pkg/mocks/client"
 	mock_lifecycle "github.com/bakito/batch-job-controller/pkg/mocks/lifecycle"
 	mock_logr "github.com/bakito/batch-job-controller/pkg/mocks/logr"
+	"github.com/bakito/batch-job-controller/pkg/test"
 	"github.com/go-logr/logr"
 	gm "github.com/golang/mock/gomock"
+	"github.com/google/uuid"
 	. "github.com/onsi/ginkgo/v2"
 	. "github.com/onsi/gomega"
 	corev1 "k8s.io/api/core/v1"
 	k8serrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime/schema"
+	"k8s.io/client-go/kubernetes/fake"
+	corev1client "k8s.io/client-go/kubernetes/typed/core/v1"
 	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/event"
@@ -62,8 +68,12 @@ var _ = Describe("Controller", func() {
 			mockClient     *mock_client.MockClient
 			mockSink       *mock_logr.MockLogSink
 			ctx            context.Context
+			coreClient     corev1client.CoreV1Interface
+			cfg            config.Config
+			executionID    string
 		)
 		BeforeEach(func() {
+			executionID = uuid.NewString()
 			mockCtrl = gm.NewController(GinkgoT())
 			mockController = mock_lifecycle.NewMockController(mockCtrl)
 			mockClient = mock_client.NewMockClient(mockCtrl)
@@ -72,9 +82,23 @@ var _ = Describe("Controller", func() {
 			mockSink.EXPECT().Init(gm.Any())
 			mockSink.EXPECT().Enabled(gm.Any()).AnyTimes().Return(true)
 			ctx = log.IntoContext(context.TODO(), logr.New(mockSink))
+
+			coreClient = fake.NewSimpleClientset().CoreV1()
+
+			tmp, err := test.TempDir(executionID)
+			Ω(err).ShouldNot(HaveOccurred())
+			cfg = config.Config{
+				ReportDirectory: tmp,
+			}
+
 			r = &PodReconciler{}
 			r.Controller = mockController
 			r.Client = mockClient
+			r.coreClient = coreClient
+
+			DeferCleanup(func() error {
+				return os.RemoveAll(cfg.ReportDirectory)
+			})
 		})
 		It("should not find an entry", func() {
 			mockSink.EXPECT().WithValues(gm.Any()).Return(mockSink)
@@ -95,13 +119,22 @@ var _ = Describe("Controller", func() {
 			Ω(result).ShouldNot(BeNil())
 			Ω(result.Requeue).Should(BeFalse())
 		})
-		It("should update controller on pod succeeded", func() {
-			mockSink.EXPECT().WithValues(gm.Any()).Return(mockSink)
+		It("should update controller on pod succeeded with logs of 2 containers", func() {
+			cfg.SavePodLog = true
+			mockController.EXPECT().Config().Return(cfg).AnyTimes()
+			mockSink.EXPECT().WithValues(gm.Any()).Return(mockSink).AnyTimes()
+			mockSink.EXPECT().Info(gm.Any(), gm.Any()).AnyTimes()
 			mockClient.EXPECT().Get(gm.Any(), gm.Any(), gm.AssignableToTypeOf(&corev1.Pod{})).
 				Do(func(ctx context.Context, key client.ObjectKey, pod *corev1.Pod) error {
 					pod.Status = corev1.PodStatus{
 						Phase: corev1.PodSucceeded,
 					}
+					pod.Spec = corev1.PodSpec{
+						Containers: []corev1.Container{
+							{Name: "container-a"},
+							{Name: "container-b"},
+						},
+					}
 					return nil
 				})
 			mockController.EXPECT().PodTerminated(gm.Any(), gm.Any(), corev1.PodSucceeded)
@@ -112,6 +145,7 @@ var _ = Describe("Controller", func() {
 			Ω(result.Requeue).Should(BeFalse())
 		})
 		It("should update controller on pod failed", func() {
+			mockController.EXPECT().Config().Return(cfg)
 			mockSink.EXPECT().WithValues(gm.Any()).Return(mockSink)
 			mockClient.EXPECT().Get(gm.Any(), gm.Any(), gm.AssignableToTypeOf(&corev1.Pod{})).
 				Do(func(ctx context.Context, key client.ObjectKey, pod *corev1.Pod) error {
@@ -128,6 +162,7 @@ var _ = Describe("Controller", func() {
 			Ω(result.Requeue).Should(BeFalse())
 		})
 		It("should return error on update controller error", func() {
+			mockController.EXPECT().Config().Return(cfg)
 			mockSink.EXPECT().WithValues(gm.Any()).Return(mockSink)
 			mockSink.EXPECT().Error(gm.Any(), gm.Any())
 			mockClient.EXPECT().Get(gm.Any(), gm.Any(), gm.AssignableToTypeOf(&corev1.Pod{})).

pkg/controller/predicate.go

@@ -0,0 +1,28 @@
+package controller
+
+import (
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"sigs.k8s.io/controller-runtime/pkg/event"
+)
+
+type podPredicate struct{}
+
+func (podPredicate) Create(e event.CreateEvent) bool {
+	return matches(e.Object)
+}
+
+func (podPredicate) Update(e event.UpdateEvent) bool {
+	return matches(e.ObjectNew)
+}
+
+func (podPredicate) Delete(e event.DeleteEvent) bool {
+	return matches(e.Object)
+}
+
+func (podPredicate) Generic(e event.GenericEvent) bool {
+	return matches(e.Object)
+}
+
+func matches(m metav1.Object) bool {
+	return m.GetLabels()[LabelExecutionID] != "" && m.GetLabels()[LabelOwner] != ""
+}

pkg/http/middleware.go

@@ -12,7 +12,7 @@ const (
 
 func (s *PostServer) middleware(ctx *gin.Context) {
 	if s.Controller != nil {
-		if !s.Controller.Has(nodeAndID(ctx)) && !s.DevMode {
+		if !s.Controller.Has(nodeAndID(ctx)) && !s.Config.DevMode {
 			ctx.String(http.StatusNotAcceptable, errorMiddlewareNotAcceptable)
 			ctx.Abort()
 			return

pkg/http/postserver-file.go

@@ -18,13 +18,13 @@ func (s *PostServer) postFile(ctx *gin.Context) {
 }
 
 func (s *PostServer) saveFormFilesCallback(ctx *gin.Context, postLog logr.Logger, executionID string, node string, file *multipart.FileHeader) error {
-	if err := s.mkdir(executionID); err != nil {
+	if err := s.Config.MkReportDir(executionID); err != nil {
 		ctx.String(http.StatusInternalServerError, err.Error())
 		postLog.Error(err, "error creating upload directory")
 		return err
 	}
 
-	err := ctx.SaveUploadedFile(file, filepath.Join(s.ReportPath, executionID, fmt.Sprintf("%s-%s", node, file.Filename)))
+	err := ctx.SaveUploadedFile(file, s.Config.ReportFileName(executionID, fmt.Sprintf("%s-%s", node, file.Filename)))
 	if err != nil {
 		ctx.String(http.StatusInternalServerError, err.Error())
 		postLog.Error(err, "error saving file")

pkg/http/postserver.go

@@ -4,8 +4,6 @@ import (
 	"fmt"
 	"io/ioutil"
 	"net/http/pprof"
-	"os"
-	"path/filepath"
 
 	"github.com/bakito/batch-job-controller/pkg/config"
 	"github.com/bakito/batch-job-controller/pkg/lifecycle"
@@ -41,9 +39,9 @@ func GenericAPIServer(port int, cfg *config.Config) manager.Runnable {
 			Kind:    "internal",
 			Handler: r,
 			Log:     ctrl.Log.WithName("api-server"),
+			Config:  cfg,
 		},
-		ReportPath: cfg.ReportDirectory,
-		DevMode:    cfg.DevMode,
+		Config: cfg,
 	}
 
 	rep := r.Group(CallbackBasePath)
@@ -84,8 +82,7 @@ func SetupProfiling(r *gin.Engine) {
 type PostServer struct {
 	*Server
 	Controller    lifecycle.Controller
-	ReportPath    string
-	DevMode       bool
+	Config        *config.Config
 	EventRecorder record.EventRecorder
 	Client        client.Reader
 }
@@ -118,17 +115,13 @@ func nodeAndID(ctx *gin.Context) (string, string) {
 
 // SaveFile save a received file
 func (s *PostServer) SaveFile(executionID, name string, data []byte) (string, error) {
-	if err := s.mkdir(executionID); err != nil {
+	if err := s.Config.MkReportDir(executionID); err != nil {
 		return "", err
 	}
-	fileName := filepath.Join(s.ReportPath, executionID, name)
+	fileName := s.Config.ReportFileName(executionID, name)
 	return fileName, ioutil.WriteFile(fileName, data, 0o600)
 }
 
-func (s *PostServer) mkdir(executionID string) error {
-	return os.MkdirAll(filepath.Join(s.ReportPath, executionID), 0o755)
-}
-
 // Name the name of the server
 func (s *PostServer) Name() string {
 	return "api-server"