Sync container logs

cmd/icinga-kubernetes/main.go

@@ -7,6 +7,7 @@ import (
 	"github.com/icinga/icinga-go-library/driver"
 	"github.com/icinga/icinga-go-library/logging"
 	"github.com/icinga/icinga-kubernetes/internal"
+	"github.com/icinga/icinga-kubernetes/pkg/contracts"
 	"github.com/icinga/icinga-kubernetes/pkg/schema"
 	"github.com/icinga/icinga-kubernetes/pkg/sync"
 	"github.com/okzk/sdnotify"
@@ -70,6 +71,11 @@ func main() {
 
 	g, ctx := errgroup.WithContext(ctx)
 
+	forwardUpsertPodsChannel := make(chan database.Entity)
+	forwardDeletePodsChannel := make(chan any)
+	defer close(forwardUpsertPodsChannel)
+	defer close(forwardDeletePodsChannel)
+
 	g.Go(func() error {
 		return sync.NewSync(
 			db, schema.NewNode, informers.Core().V1().Nodes().Informer(), logs.GetChildLogger("Nodes"),
@@ -82,10 +88,23 @@ func main() {
 		).Run(ctx)
 	})
 
+	podUpserts := make(chan contracts.KUpsert)
+	podDeletes := make(chan contracts.KDelete)
 	g.Go(func() error {
+		defer close(podUpserts)
+		defer close(podDeletes)
+
 		return sync.NewSync(
 			db, schema.NewPod, informers.Core().V1().Pods().Informer(), logs.GetChildLogger("Pods"),
-		).Run(ctx)
+		).Run(
+			ctx, sync.WithForwardUpserts(podUpserts), sync.WithForwardDeletes(podDeletes),
+		)
+	})
+
+	logSync := sync.NewLogSync(k, db, logs.GetChildLogger("ContainerLogs"))
+
+	g.Go(func() error {
+		return logSync.Run(ctx, podUpserts, podDeletes)
 	})
 
 	if err := g.Wait(); err != nil {

go.mod

@@ -8,8 +8,10 @@ require (
 	github.com/pkg/errors v0.9.1
 	go.uber.org/zap v1.26.0
 	golang.org/x/sync v0.5.0
-	k8s.io/apimachinery v0.28.2
-	k8s.io/client-go v0.28.2
+	k8s.io/api v0.28.4
+	k8s.io/apimachinery v0.28.4
+	k8s.io/client-go v0.28.4
+	k8s.io/metrics v0.28.4
 )
 
 require (
@@ -45,19 +47,18 @@ require (
 	github.com/ssgreg/journald v1.0.0 // indirect
 	go.uber.org/multierr v1.11.0 // indirect
 	golang.org/x/exp v0.0.0-20231110203233-9a3e6036ecaa // indirect
-	golang.org/x/net v0.16.0 // indirect
+	golang.org/x/net v0.17.0 // indirect
 	golang.org/x/oauth2 v0.8.0 // indirect
 	golang.org/x/sys v0.14.0 // indirect
 	golang.org/x/term v0.13.0 // indirect
 	golang.org/x/text v0.13.0 // indirect
 	golang.org/x/time v0.3.0 // indirect
 	golang.org/x/xerrors v0.0.0-20231012003039-104605ab7028 // indirect
 	google.golang.org/appengine v1.6.7 // indirect
-	google.golang.org/protobuf v1.30.0 // indirect
+	google.golang.org/protobuf v1.31.0 // indirect
 	gopkg.in/inf.v0 v0.9.1 // indirect
 	gopkg.in/yaml.v2 v2.4.0 // indirect
 	gopkg.in/yaml.v3 v3.0.1 // indirect
-	k8s.io/api v0.28.2 // indirect
 	k8s.io/klog/v2 v2.100.1 // indirect
 	k8s.io/kube-openapi v0.0.0-20230717233707-2695361300d9 // indirect
 	k8s.io/utils v0.0.0-20230406110748-d93618cff8a2 // indirect

go.sum

@@ -43,7 +43,6 @@ github.com/google/gofuzz v1.2.0/go.mod h1:dBl0BpW6vV/+mYPU4Po3pmUjxk6FQPldtuIdl/
 github.com/google/pprof v0.0.0-20210720184732-4bb14d4b1be1 h1:K6RDEckDVWvDI9JAJYCmNdQXq6neHJOYx3V6jnqNEec=
 github.com/google/uuid v1.4.0 h1:MtMxsa51/r9yyhkyLsVeVt0B+BGQZzpQiTQ4eHZ8bc4=
 github.com/google/uuid v1.4.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
-github.com/icinga/icinga-go-library v0.0.0-20231121080432-c03a40718ed9 h1:C+BYgMhhitPxEt9pQ/O/ssQ90mIrm7+YG4KDe9UFFBA=
 github.com/icinga/icinga-go-library v0.0.0-20231121080432-c03a40718ed9/go.mod h1:Apo85zqPgovShDWxx/TlUN/bfl+RaPviTafT666iJyw=
 github.com/imdario/mergo v0.3.6 h1:xTNEAn+kxVO7dTZGu0CegyqKZmoWFI0rF8UxjlB2d28=
 github.com/imdario/mergo v0.3.6/go.mod h1:2EnlNZ0deacrJVfApfmtdGgDfMuh/nq6Ok1EcJh5FfA=
@@ -124,8 +123,8 @@ golang.org/x/net v0.0.0-20190603091049-60506f45cf65/go.mod h1:HSz+uSET+XFnRR8LxR
 golang.org/x/net v0.0.0-20190620200207-3b0461eec859/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
 golang.org/x/net v0.0.0-20200226121028-0de0cce0169b/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
 golang.org/x/net v0.0.0-20201021035429-f5854403a974/go.mod h1:sp8m0HH+o8qH0wwXwYZr8TS3Oi6o0r6Gce1SSxlDquU=
-golang.org/x/net v0.16.0 h1:7eBu7KsSvFDtSXUIDbh3aqlK4DPsZ1rByC8PFfBThos=
-golang.org/x/net v0.16.0/go.mod h1:NxSsAGuq816PNPmqtQdLE42eU2Fs7NoRIZrHJAlaCOE=
+golang.org/x/net v0.17.0 h1:pVaXccu2ozPjCXewfr1S7xza/zcXTity9cCdXQYSjIM=
+golang.org/x/net v0.17.0/go.mod h1:NxSsAGuq816PNPmqtQdLE42eU2Fs7NoRIZrHJAlaCOE=
 golang.org/x/oauth2 v0.8.0 h1:6dkIjl3j3LtZ/O3sTgZTMsLKSftL/B8Zgq4huOIIUu8=
 golang.org/x/oauth2 v0.8.0/go.mod h1:yr7u4HXZRm1R1kBWqr/xKNqewf0plRYoB7sla+BCIXE=
 golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
@@ -165,8 +164,8 @@ google.golang.org/appengine v1.6.7 h1:FZR1q0exgwxzPzp/aF+VccGrSfxfPpkBqjIIEq3ru6
 google.golang.org/appengine v1.6.7/go.mod h1:8WjMMxjGQR8xUklV/ARdw2HLXBOI7O7uCIDZVag1xfc=
 google.golang.org/protobuf v1.26.0-rc.1/go.mod h1:jlhhOSvTdKEhbULTjvd4ARK9grFBp09yW+WbY/TyQbw=
 google.golang.org/protobuf v1.26.0/go.mod h1:9q0QmTI4eRPtz6boOQmLYwt+qCgq0jsYwAQnmE0givc=
-google.golang.org/protobuf v1.30.0 h1:kPPoIgf3TsEvrm0PFe15JQ+570QVxYzEvvHqChK+cng=
-google.golang.org/protobuf v1.30.0/go.mod h1:HV8QOd/L58Z+nl8r43ehVNZIU/HEI6OcFqwMG9pJV4I=
+google.golang.org/protobuf v1.31.0 h1:g0LDEJHgrBl9N9r17Ru3sqWhkIx2NB67okBHPwC7hs8=
+google.golang.org/protobuf v1.31.0/go.mod h1:HV8QOd/L58Z+nl8r43ehVNZIU/HEI6OcFqwMG9pJV4I=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
 gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c h1:Hei/4ADfdWqJk1ZMxUNpqntNwaWcugrBjAiHlqqRiVk=
 gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c/go.mod h1:JHkPIbrfpd72SG/EVd6muEfDQjcINNoR0C8j2r3qZ4Q=
@@ -178,16 +177,18 @@ gopkg.in/yaml.v2 v2.4.0/go.mod h1:RDklbk79AGWmwhnvt/jBztapEOGDOx6ZbXqjP6csGnQ=
 gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
 gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
 gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
-k8s.io/api v0.28.2 h1:9mpl5mOb6vXZvqbQmankOfPIGiudghwCoLl1EYfUZbw=
-k8s.io/api v0.28.2/go.mod h1:RVnJBsjU8tcMq7C3iaRSGMeaKt2TWEUXcpIt/90fjEg=
-k8s.io/apimachinery v0.28.2 h1:KCOJLrc6gu+wV1BYgwik4AF4vXOlVJPdiqn0yAWWwXQ=
-k8s.io/apimachinery v0.28.2/go.mod h1:RdzF87y/ngqk9H4z3EL2Rppv5jj95vGS/HaFXrLDApU=
-k8s.io/client-go v0.28.2 h1:DNoYI1vGq0slMBN/SWKMZMw0Rq+0EQW6/AK4v9+3VeY=
-k8s.io/client-go v0.28.2/go.mod h1:sMkApowspLuc7omj1FOSUxSoqjr+d5Q0Yc0LOFnYFJY=
+k8s.io/api v0.28.4 h1:8ZBrLjwosLl/NYgv1P7EQLqoO8MGQApnbgH8tu3BMzY=
+k8s.io/api v0.28.4/go.mod h1:axWTGrY88s/5YE+JSt4uUi6NMM+gur1en2REMR7IRj0=
+k8s.io/apimachinery v0.28.4 h1:zOSJe1mc+GxuMnFzD4Z/U1wst50X28ZNsn5bhgIIao8=
+k8s.io/apimachinery v0.28.4/go.mod h1:wI37ncBvfAoswfq626yPTe6Bz1c22L7uaJ8dho83mgg=
+k8s.io/client-go v0.28.4 h1:Np5ocjlZcTrkyRJ3+T3PkXDpe4UpatQxj85+xjaD2wY=
+k8s.io/client-go v0.28.4/go.mod h1:0VDZFpgoZfelyP5Wqu0/r/TRYcLYuJ2U1KEeoaPa1N4=
 k8s.io/klog/v2 v2.100.1 h1:7WCHKK6K8fNhTqfBhISHQ97KrnJNFZMcQvKp7gP/tmg=
 k8s.io/klog/v2 v2.100.1/go.mod h1:y1WjHnz7Dj687irZUWR/WLkLc5N1YHtjLdmgWjndZn0=
 k8s.io/kube-openapi v0.0.0-20230717233707-2695361300d9 h1:LyMgNKD2P8Wn1iAwQU5OhxCKlKJy0sHc+PcDwFB24dQ=
 k8s.io/kube-openapi v0.0.0-20230717233707-2695361300d9/go.mod h1:wZK2AVp1uHCp4VamDVgBP2COHZjqD1T68Rf0CM3YjSM=
+k8s.io/metrics v0.28.4 h1:u36fom9+6c8jX2sk8z58H0hFaIUfrPWbXIxN7GT2blk=
+k8s.io/metrics v0.28.4/go.mod h1:bBqAJxH20c7wAsTQxDXOlVqxGMdce49d7WNr1WeaLac=
 k8s.io/utils v0.0.0-20230406110748-d93618cff8a2 h1:qY1Ad8PODbnymg2pRbkyMT/ylpTrCM8P2RJ0yroCyIk=
 k8s.io/utils v0.0.0-20230406110748-d93618cff8a2/go.mod h1:OLgZIPagt7ERELqWJFomSt595RzquPNLL48iOWgYOg0=
 sigs.k8s.io/json v0.0.0-20221116044647-bc3834ca7abd h1:EDPBXCAspyGV4jQlpZSudPeMmr1bNJefnuqLsRAsHZo=

pkg/schema/log.go

@@ -0,0 +1,10 @@
+package schema
+
+type Log struct {
+	kmetaWithoutNamespace
+	Id            []byte
+	ReferenceId   []byte
+	ContainerName string
+	Time          string
+	Log           string
+}

pkg/sync/logs.go

@@ -0,0 +1,243 @@
+package sync
+
+import (
+	"bufio"
+	"context"
+	"crypto/sha1"
+	"fmt"
+	"github.com/icinga/icinga-go-library/database"
+	"github.com/icinga/icinga-go-library/logging"
+	"github.com/icinga/icinga-kubernetes/pkg/contracts"
+	"github.com/icinga/icinga-kubernetes/pkg/schema"
+	"github.com/pkg/errors"
+	"golang.org/x/sync/errgroup"
+	"io"
+	kcorev1 "k8s.io/api/core/v1"
+	kmetav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/client-go/kubernetes"
+	"slices"
+	"strings"
+	msync "sync"
+	"time"
+)
+
+// LogSync syncs logs to database. Therefore, it maintains a list
+// of pod elements to get logs from
+type LogSync struct {
+	list        []*kcorev1.Pod
+	lastChecked map[[20]byte]*kmetav1.Time
+	mutex       *msync.RWMutex
+	clientset   *kubernetes.Clientset
+	db          *database.DB
+	logger      *logging.Logger
+}
+
+// NewLogSync creates new LogSync initialized with clientset, database and logger
+func NewLogSync(clientset *kubernetes.Clientset, db *database.DB, logger *logging.Logger) *LogSync {
+	return &LogSync{
+		list:        []*kcorev1.Pod{},
+		lastChecked: make(map[[20]byte]*kmetav1.Time),
+		mutex:       &msync.RWMutex{},
+		clientset:   clientset,
+		db:          db,
+		logger:      logger,
+	}
+}
+
+// upsertStmt returns database upsert statement
+func (ls *LogSync) upsertStmt() string {
+	return fmt.Sprintf(
+		"INSERT INTO %s (%s) VALUES (%s) ON DUPLICATE KEY UPDATE %s",
+		"log",
+		"id, reference_id, container_name, time, log",
+		":id, :reference_id, :container_name, :time, :log",
+		"time=CONCAT(time, '\n', :time), log=CONCAT(log, '\n', :log)",
+	)
+}
+
+// splitTimestampsFromMessages takes a log as []byte and returns timestamps and messages as separate string slices.
+// Additionally, it updates the last checked timestamp for the log
+func (ls *LogSync) splitTimestampsFromMessages(log []byte, curContainerId [20]byte) (times []string, messages []string, err error) {
+
+	stringReader := strings.NewReader(string(log))
+	reader := bufio.NewReader(stringReader)
+
+	for {
+		line, err := reader.ReadString('\n')
+		if err != nil {
+			if err == io.EOF {
+				break
+			}
+			return nil, nil, errors.Wrap(err, "error reading log message")
+		}
+
+		messageTime, err := time.Parse("2006-01-02T15:04:05.999999999Z", strings.Split(line, " ")[0])
+		if err != nil {
+			logging.Fatal(errors.Wrap(err, "error parsing log timestamp"))
+			continue
+		}
+
+		if ls.lastChecked[curContainerId] != nil && messageTime.UnixNano() <= ls.lastChecked[curContainerId].UnixNano() {
+			continue
+		}
+
+		times = append(times, strings.Split(line, " ")[0])
+		messages = append(messages, strings.Join(strings.Split(line, " ")[1:], " "))
+	}
+
+	return times, messages, nil
+}
+
+// removeFromList removes pod from maintained list
+func (ls *LogSync) removeFromList(id database.ID) {
+	out := make([]*kcorev1.Pod, 0)
+
+	for _, element := range ls.list {
+		elementId := sha1.Sum([]byte(element.Namespace + "/" + element.Name))
+		if fmt.Sprintf("%x", elementId) != id.String() {
+			out = append(out, element)
+		}
+	}
+
+	ls.list = out
+}
+
+// MaintainList adds pods from the addChannel to the list and deletes pods from the deleteChannel from the list
+func (ls *LogSync) maintainList(ctx context.Context, upsertChannel <-chan contracts.KUpsert, deleteChannel <-chan contracts.KDelete) error {
+	g, ctx := errgroup.WithContext(ctx)
+
+	deletes := make(chan any)
+	g.Go(func() error {
+		defer close(deletes)
+
+		for {
+			select {
+			case <-ctx.Done():
+				return errors.Wrap(ctx.Err(), "context canceled maintain log sync list")
+
+			case podFromChannel, more := <-upsertChannel:
+				if !more {
+					return nil
+				}
+
+				pod := podFromChannel.KObject().(*kcorev1.Pod)
+				podIsInList := false
+
+				for _, listPod := range ls.list {
+					if listPod.UID == pod.UID {
+						podIsInList = true
+					}
+				}
+
+				if podIsInList {
+					continue
+				}
+
+				ls.mutex.RLock()
+				ls.list = append(ls.list, pod)
+				ls.mutex.RUnlock()
+
+			case podIdFromChannel, more := <-deleteChannel:
+				if !more {
+					return nil
+				}
+
+				idOfPod := podIdFromChannel.ID()
+
+				ls.mutex.RLock()
+				ls.removeFromList(idOfPod)
+				ls.mutex.RUnlock()
+
+				deletes <- idOfPod
+			}
+
+		}
+	})
+
+	g.Go(func() error {
+		return database.NewDelete(ls.db).ByColumn("reference_id").Stream(ctx, &schema.Log{}, deletes)
+	})
+
+	return g.Wait()
+}
+
+// Run starts syncing the logs to the database. Therefore, it loops over all
+// containers of each pod in the maintained list every 15 seconds.
+func (ls *LogSync) Run(ctx context.Context, upsertChannel <-chan contracts.KUpsert, deleteChannel <-chan contracts.KDelete) error {
+	ls.logger.Info("Starting sync")
+
+	g, ctx := errgroup.WithContext(ctx)
+
+	g.Go(func() error {
+		return ls.maintainList(ctx, upsertChannel, deleteChannel)
+	})
+
+	upsertStmt := ls.upsertStmt()
+	upserts := make(chan database.Entity)
+	defer close(upserts)
+
+	g.Go(func() error {
+		for {
+			for _, pod := range ls.list {
+				curPodId := sha1.Sum([]byte(pod.Namespace + "/" + pod.Name))
+				for _, container := range pod.Spec.Containers {
+					curContainerId := sha1.Sum([]byte(pod.Namespace + "/" + pod.Name + "/" + container.Name))
+					podLogOpts := kcorev1.PodLogOptions{Container: container.Name, Timestamps: true}
+
+					if ls.lastChecked[curContainerId] != nil {
+						podLogOpts.SinceTime = ls.lastChecked[curContainerId]
+					}
+
+					log, err := ls.clientset.CoreV1().Pods(pod.Namespace).GetLogs(pod.Name, &podLogOpts).Do(ctx).Raw()
+					if err != nil {
+						fmt.Println(errors.Wrap(err, "error reading container log"))
+						continue
+					}
+
+					times, messages, err := ls.splitTimestampsFromMessages(log, curContainerId)
+					if err != nil {
+						return err
+					}
+
+					if len(messages) == 0 {
+						continue
+					}
+
+					newLog := &schema.Log{
+						Id:            curContainerId[:],
+						ReferenceId:   curPodId[:],
+						ContainerName: container.Name,
+						Time:          strings.Join(times, "\n"),
+						Log:           strings.Join(messages, "\n"),
+					}
+
+					upserts <- newLog
+
+					lastTime, err := time.Parse("2006-01-02T15:04:05.999999999Z", times[len(times)-1])
+					if err != nil {
+						return errors.Wrap(err, "error parsing log time")
+					}
+
+					if !slices.Contains(ls.list, pod) {
+						continue
+					}
+
+					lastV1Time := kmetav1.Time{Time: lastTime}
+					ls.lastChecked[curContainerId] = &lastV1Time
+				}
+			}
+
+			select {
+			case <-ctx.Done():
+				return ctx.Err()
+			case <-time.After(time.Second * 15):
+			}
+		}
+	})
+
+	g.Go(func() error {
+		return database.NewUpsert(ls.db).WithStatement(upsertStmt, 5).Stream(ctx, upserts)
+	})
+
+	return g.Wait()
+}