Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Sync container logs #50

Open
wants to merge 12 commits into
base: next
Choose a base branch
from
21 changes: 20 additions & 1 deletion cmd/icinga-kubernetes/main.go
Original file line number Diff line number Diff line change
Expand Up @@ -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"
Expand Down Expand Up @@ -70,6 +71,11 @@ func main() {

g, ctx := errgroup.WithContext(ctx)

forwardUpsertPodsChannel := make(chan database.Entity)
jrauh01 marked this conversation as resolved.
Show resolved Hide resolved
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"),
jrauh01 marked this conversation as resolved.
Show resolved Hide resolved
Expand All @@ -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 {
jrauh01 marked this conversation as resolved.
Show resolved Hide resolved
return logSync.Run(ctx, podUpserts, podDeletes)
jrauh01 marked this conversation as resolved.
Show resolved Hide resolved
})

if err := g.Wait(); err != nil {
Expand Down
11 changes: 6 additions & 5 deletions go.mod
Original file line number Diff line number Diff line change
Expand Up @@ -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 (
Expand Down Expand Up @@ -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
Expand Down
23 changes: 12 additions & 11 deletions go.sum
Original file line number Diff line number Diff line change
Expand Up @@ -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=
Expand Down Expand Up @@ -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=
Expand Down Expand Up @@ -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=
Expand All @@ -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=
Expand Down
10 changes: 10 additions & 0 deletions pkg/schema/log.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,10 @@
package schema

type Log struct {
jrauh01 marked this conversation as resolved.
Show resolved Hide resolved
kmetaWithoutNamespace
Id []byte
jrauh01 marked this conversation as resolved.
Show resolved Hide resolved
ReferenceId []byte
jrauh01 marked this conversation as resolved.
Show resolved Hide resolved
ContainerName string
Time string
Log string
}
243 changes: 243 additions & 0 deletions pkg/sync/logs.go
Original file line number Diff line number Diff line change
@@ -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"
jrauh01 marked this conversation as resolved.
Show resolved Hide resolved
"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
jrauh01 marked this conversation as resolved.
Show resolved Hide resolved
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) {

jrauh01 marked this conversation as resolved.
Show resolved Hide resolved
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])
jrauh01 marked this conversation as resolved.
Show resolved Hide resolved
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])
jrauh01 marked this conversation as resolved.
Show resolved Hide resolved
messages = append(messages, strings.Join(strings.Split(line, " ")[1:], " "))
}

return times, messages, nil
jrauh01 marked this conversation as resolved.
Show resolved Hide resolved
}

// 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)
jrauh01 marked this conversation as resolved.
Show resolved Hide resolved
g.Go(func() error {
defer close(deletes)

for {
select {
case <-ctx.Done():
jrauh01 marked this conversation as resolved.
Show resolved Hide resolved
return errors.Wrap(ctx.Err(), "context canceled maintain log sync list")

case podFromChannel, more := <-upsertChannel:
jrauh01 marked this conversation as resolved.
Show resolved Hide resolved
if !more {
return nil
}

pod := podFromChannel.KObject().(*kcorev1.Pod)
jrauh01 marked this conversation as resolved.
Show resolved Hide resolved
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:
jrauh01 marked this conversation as resolved.
Show resolved Hide resolved
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()
jrauh01 marked this conversation as resolved.
Show resolved Hide resolved
upserts := make(chan database.Entity)
jrauh01 marked this conversation as resolved.
Show resolved Hide resolved
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"))
jrauh01 marked this conversation as resolved.
Show resolved Hide resolved
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
jrauh01 marked this conversation as resolved.
Show resolved Hide resolved

lastTime, err := time.Parse("2006-01-02T15:04:05.999999999Z", times[len(times)-1])
jrauh01 marked this conversation as resolved.
Show resolved Hide resolved
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):
jrauh01 marked this conversation as resolved.
Show resolved Hide resolved
}
}
})

g.Go(func() error {
return database.NewUpsert(ls.db).WithStatement(upsertStmt, 5).Stream(ctx, upserts)
})

return g.Wait()
}
Loading