Fix review

cmd/icinga-kubernetes/main.go

@@ -72,57 +72,39 @@ func main() {
 	g, ctx := errgroup.WithContext(ctx)
 
 	forwardUpsertPodsChannel := make(chan database.Entity)
-	defer close(forwardUpsertPodsChannel)
-
 	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"),
+			db, schema.NewNode, informers.Core().V1().Nodes().Informer(), logs.GetChildLogger("Nodes"),
 		).Run(ctx)
 	})
 
 	g.Go(func() error {
 		return sync.NewSync(
-			db,
-			schema.NewNamespace,
-			informers.Core().V1().Namespaces().Informer(),
-			logs.GetChildLogger("Namespaces"),
+			db, schema.NewNamespace, informers.Core().V1().Namespaces().Informer(), logs.GetChildLogger("Namespaces"),
 		).Run(ctx)
 	})
 
-	forwardUpsertPodsToLogChannel := make(chan contracts.KUpsert)
-	forwardDeletePodsToLogChannel := make(chan contracts.KDelete)
-
+	podUpserts := make(chan contracts.KUpsert)
+	podDeletes := make(chan contracts.KDelete)
 	g.Go(func() error {
-
-		defer close(forwardUpsertPodsToLogChannel)
-		defer close(forwardDeletePodsToLogChannel)
+		defer close(podUpserts)
+		defer close(podDeletes)
 
 		return sync.NewSync(
-			db,
-			schema.NewPod,
-			informers.Core().V1().Pods().Informer(),
-			logs.GetChildLogger("Pods"),
+			db, schema.NewPod, informers.Core().V1().Pods().Informer(), logs.GetChildLogger("Pods"),
 		).Run(
-			ctx,
-			sync.WithForwardUpsertToLog(forwardUpsertPodsToLogChannel),
-			sync.WithForwardDeleteToLog(forwardDeletePodsToLogChannel),
+			ctx, sync.WithForwardUpserts(podUpserts), sync.WithForwardDeletes(podDeletes),
 		)
 	})
 
 	logSync := sync.NewLogSync(k, db, logs.GetChildLogger("ContainerLogs"))
 
 	g.Go(func() error {
-		return logSync.MaintainList(ctx, forwardUpsertPodsToLogChannel, forwardDeletePodsToLogChannel)
-	})
-
-	g.Go(func() error {
-		return logSync.Run(ctx)
+		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/sync/logs.go

@@ -93,9 +93,7 @@ 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)
 		}
@@ -105,10 +103,7 @@ func (ls *LogSync) removeFromList(id database.ID) {
 }
 
 // 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, addChannel <-chan contracts.KUpsert, deleteChannel <-chan contracts.KDelete) error {
-
-	ls.logger.Info("Starting maintain 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)
@@ -120,13 +115,12 @@ func (ls *LogSync) MaintainList(ctx context.Context, addChannel <-chan contracts
 			case <-ctx.Done():
 				return errors.Wrap(ctx.Err(), "context canceled maintain log sync list")
 
-			case podFromChannel, more := <-addChannel:
+			case podFromChannel, more := <-upsertChannel:
 				if !more {
 					return nil
 				}
 
 				pod := podFromChannel.KObject().(*kcorev1.Pod)
-
 				podIsInList := false
 
 				for _, listPod := range ls.list {
@@ -169,27 +163,25 @@ func (ls *LogSync) MaintainList(ctx context.Context, addChannel <-chan contracts
 
 // 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) error {
-
+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)
 
-	upsertStmt := ls.upsertStmt()
+	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 {

pkg/sync/spreader.go

@@ -5,11 +5,11 @@ import (
 	"sync/atomic"
 )
 
-// ChannelSpreader takes a channel of type T and fans it out to an array of other channels of type T
+// ChannelSpreader takes a channel of type T and fans it out to an array of other addedChannels of type T
 type ChannelSpreader[T any] struct {
 	channelToBreak  <-chan T
 	createdChannels []chan<- T
-	channels        []chan<- T
+	addedChannels   []chan<- T
 	started         atomic.Bool
 }
 
@@ -20,9 +20,9 @@ func NewChannelSpreader[T any](channelToBreak <-chan T) *ChannelSpreader[T] {
 	}
 }
 
-// NewChannel returns and adds new output channel to the list of created channels
+// NewChannel returns and adds new output channel to the list of created addedChannels
 func (cs *ChannelSpreader[T]) NewChannel() <-chan T {
-	if cs.started.Load() == true {
+	if cs.started.Load() {
 		panic("ChannelSpreader already started")
 	}
 
@@ -32,18 +32,17 @@ func (cs *ChannelSpreader[T]) NewChannel() <-chan T {
 	return channel
 }
 
-// AddChannel adds given output channel to the list of added channels
+// AddChannel adds given output channel to the list of added addedChannels
 func (cs *ChannelSpreader[T]) AddChannel(channel chan<- T) {
-	if cs.started.Load() == true {
+	if cs.started.Load() {
 		panic("ChannelSpreader already started")
 	}
 
-	cs.channels = append(cs.channels, channel)
+	cs.addedChannels = append(cs.addedChannels, channel)
 }
 
-// Run combines the lists and starts fanning out the channel to the channels from the list
+// Run combines the lists and starts fanning out the channel to the addedChannels from the list
 func (cs *ChannelSpreader[T]) Run(ctx context.Context) error {
-
 	cs.started.Store(true)
 
 	defer func() {
@@ -52,7 +51,7 @@ func (cs *ChannelSpreader[T]) Run(ctx context.Context) error {
 		}
 	}()
 
-	cs.channels = append(cs.channels, cs.createdChannels...)
+	channels := append(cs.addedChannels, cs.createdChannels...)
 
 	for {
 		select {
@@ -61,14 +60,13 @@ func (cs *ChannelSpreader[T]) Run(ctx context.Context) error {
 				return nil
 			}
 
-			for _, channel := range cs.channels {
+			for _, channel := range channels {
 				select {
 				case channel <- spread:
 				case <-ctx.Done():
 					return ctx.Err()
 				}
 			}
-
 		case <-ctx.Done():
 			return ctx.Err()
 		}

pkg/sync/sync.go

@@ -41,33 +41,33 @@ func NewSync(
 	return s
 }
 
-func WithForwardUpsertToLog(channel chan<- contracts.KUpsert) SyncOption {
+func WithForwardUpserts(channel chan<- contracts.KUpsert) SyncOption {
 	return func(options *SyncOptions) {
-		options.forwardUpsertToLogChannel = channel
+		options.forwardUpserts = channel
 	}
 }
 
-func WithForwardDeleteToLog(channel chan<- contracts.KDelete) SyncOption {
+func WithForwardDeletes(channel chan<- contracts.KDelete) SyncOption {
 	return func(options *SyncOptions) {
-		options.forwardDeleteToLogChannel = channel
+		options.forwardDeletes = channel
 	}
 }
 
 type SyncOption func(options *SyncOptions)
 
 type SyncOptions struct {
-	forwardUpsertToLogChannel chan<- contracts.KUpsert
-	forwardDeleteToLogChannel chan<- contracts.KDelete
+	forwardUpserts chan<- contracts.KUpsert
+	forwardDeletes chan<- contracts.KDelete
 }
 
-func NewOptionStorage(execOptions ...SyncOption) *SyncOptions {
-	optionStorage := &SyncOptions{}
+func NewSyncOptions(options ...SyncOption) *SyncOptions {
+	syncOptions := &SyncOptions{}
 
-	for _, option := range execOptions {
-		option(optionStorage)
+	for _, option := range options {
+		option(syncOptions)
 	}
 
-	return optionStorage
+	return syncOptions
 }
 
 func (s *sync) Run(ctx context.Context, execOptions ...SyncOption) error {
@@ -98,7 +98,7 @@ func (s *sync) Run(ctx context.Context, execOptions ...SyncOption) error {
 
 	s.factory().GetResourceVersion()
 
-	syncOptions := NewOptionStorage(execOptions...)
+	syncOptions := NewSyncOptions(execOptions...)
 
 	// init upsert channel spreader
 	multiplexUpsertChannel := make(chan contracts.KUpsert)
@@ -108,8 +108,8 @@ func (s *sync) Run(ctx context.Context, execOptions ...SyncOption) error {
 
 	upsertChannel := multiplexUpsert.NewChannel()
 
-	if syncOptions.forwardUpsertToLogChannel != nil {
-		multiplexUpsert.AddChannel(syncOptions.forwardUpsertToLogChannel)
+	if syncOptions.forwardUpserts != nil {
+		multiplexUpsert.AddChannel(syncOptions.forwardUpserts)
 	}
 
 	// run upsert channel spreader
@@ -182,8 +182,8 @@ func (s *sync) Run(ctx context.Context, execOptions ...SyncOption) error {
 
 	deleteChannel := multiplexDelete.NewChannel()
 
-	if syncOptions.forwardDeleteToLogChannel != nil {
-		multiplexDelete.AddChannel(syncOptions.forwardDeleteToLogChannel)
+	if syncOptions.forwardDeletes != nil {
+		multiplexDelete.AddChannel(syncOptions.forwardDeletes)
 	}
 
 	// run delete channel spreader