Refactor/nats subject change

.tools/nvim/__http__/console/apps.graphql.yml

@@ -73,21 +73,10 @@ label: Create App
 query: |+ #graphql
   mutation Core_createApp($envName: String!, $app: AppIn!) {
     core_createApp(envName: $envName, app: $app) {
+      id
       metadata {
         name
       }
-      syncStatus {
-        state
-        recordVersion
-        error
-        action
-        lastSyncedAt
-      }
-      createdBy{
-        userId
-        userName
-        userEmail
-      }
     }
   }
 variables:

apps/accounts/internal/framework/framework.go

@@ -3,10 +3,12 @@ package framework
 import (
 	"context"
 	"fmt"
+	"log/slog"
+	"time"
+
 	"github.com/kloudlite/api/common"
 	"github.com/kloudlite/api/pkg/errors"
 	"github.com/kloudlite/api/pkg/nats"
-	"time"
 
 	"github.com/kloudlite/api/pkg/kv"
 	"github.com/kloudlite/api/pkg/repos"
@@ -33,7 +35,7 @@ var Module = fx.Module("framework",
 		return &fm{env: ev}
 	}),
 
-	fx.Provide(func(ev *env.Env, logger logging.Logger) (*nats.JetstreamClient, error) {
+	fx.Provide(func(ev *env.Env, logger *slog.Logger) (*nats.JetstreamClient, error) {
 		name := "accounts:jetstream-client"
 		nc, err := nats.NewClient(ev.NatsURL, nats.ClientOpts{
 			Name:   name,

apps/accounts/main.go

@@ -3,6 +3,7 @@ package main
 import (
 	"context"
 	"flag"
+	"log/slog"
 	"os"
 	"time"
 
@@ -35,6 +36,14 @@ func main() {
 			return logger
 		}),
 
+		fx.Provide(func() *slog.Logger {
+			return logging.NewSlogLogger(logging.SlogOptions{
+				ShowCaller:         true,
+				ShowDebugLogs:      isDev,
+				SetAsDefaultLogger: true,
+			})
+		}),
+
 		fx.Provide(func() (*env.Env, error) {
 			if e, err := env.LoadEnv(); err != nil {
 				return nil, errors.NewE(err)

apps/auth/internal/framework/main.go

@@ -3,6 +3,8 @@ package framework
 import (
 	"context"
 	"fmt"
+	"log/slog"
+
 	"github.com/kloudlite/api/common"
 	"github.com/kloudlite/api/pkg/errors"
 	"github.com/kloudlite/api/pkg/kv"
@@ -57,7 +59,7 @@ var Module fx.Option = fx.Module(
 
 	repos.NewMongoClientFx[*fm](),
 
-	fx.Provide(func(ev *env.Env, logger logging.Logger) (*nats.JetstreamClient, error) {
+	fx.Provide(func(ev *env.Env, logger *slog.Logger) (*nats.JetstreamClient, error) {
 		name := "auth:jetstream-client"
 		nc, err := nats.NewClient(ev.NatsURL, nats.ClientOpts{
 			Name:   name,

apps/auth/main.go

@@ -3,6 +3,7 @@ package main
 import (
 	"context"
 	"flag"
+	"log/slog"
 	"time"
 
 	"github.com/kloudlite/api/pkg/errors"
@@ -36,6 +37,15 @@ func main() {
 				return logging.New(&logging.Options{Name: "auth", Dev: isDev})
 			},
 		),
+
+		fx.Provide(func() *slog.Logger {
+			return logging.NewSlogLogger(logging.SlogOptions{
+				ShowCaller:         true,
+				ShowDebugLogs:      isDev,
+				SetAsDefaultLogger: true,
+			})
+		}),
+
 		framework.Module,
 	)
 

apps/comms/internal/framework/framework.go

@@ -3,6 +3,7 @@ package framework
 import (
 	"context"
 	"fmt"
+	"log/slog"
 	"time"
 
 	"github.com/kloudlite/api/apps/comms/internal/app"
@@ -35,7 +36,7 @@ var Module = fx.Module(
 		return &fm{ev}
 	}),
 
-	fx.Provide(func(ev *env.Env, logger logging.Logger) (*nats.Client, error) {
+	fx.Provide(func(ev *env.Env, logger *slog.Logger) (*nats.Client, error) {
 		return nats.NewClient(ev.NatsURL, nats.ClientOpts{
 			Name:   "comms",
 			Logger: logger,

apps/comms/main.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"embed"
 	"flag"
+	"log/slog"
 	"os"
 	"time"
 
@@ -30,11 +31,18 @@ func main() {
 
 	webApp := fx.New(
 		fx.NopLogger,
-		fx.Provide(
-			func() logging.Logger {
-				return logger
-			},
-		),
+		fx.Provide(func() logging.Logger {
+			return logger
+		}),
+
+		fx.Provide(func() *slog.Logger {
+			return logging.NewSlogLogger(logging.SlogOptions{
+				ShowCaller:         true,
+				ShowDebugLogs:      isDev,
+				SetAsDefaultLogger: true,
+			})
+		}),
+
 		fx.Provide(func() (*env.Env, error) {
 			return env.LoadEnv()
 		}),

apps/console/internal/app/app.go

@@ -2,6 +2,7 @@ package app
 
 import (
 	"context"
+	"log/slog"
 
 	"github.com/kloudlite/api/grpc-interfaces/kloudlite.io/rpc/console"
 	"github.com/kloudlite/api/pkg/k8s"
@@ -164,10 +165,10 @@ var Module = fx.Module("app",
 	}),
 
 	fx.Provide(func(jc *nats.JetstreamClient, ev *env.Env, logger logging.Logger) (ErrorOnApplyConsumer, error) {
-		topic := common.GetPlatformClusterMessagingTopic("*", "*", common.ConsoleReceiver, common.EventErrorOnApply)
+		topic := common.ReceiveFromAgentSubjectName(common.ReceiveFromAgentArgs{AccountName: "*", ClusterName: "*"}, common.ConsoleReceiver, common.EventErrorOnApply)
 		consumerName := "console:error-on-apply"
 		return msg_nats.NewJetstreamConsumer(context.TODO(), jc, msg_nats.JetstreamConsumerArgs{
-			Stream: ev.NatsResourceSyncStream,
+			Stream: ev.NatsReceiveFromAgentStream,
 			ConsumerConfig: msg_nats.ConsumerConfig{
 				Name:           consumerName,
 				Durable:        consumerName,
@@ -177,7 +178,7 @@ var Module = fx.Module("app",
 		})
 	}),
 
-	fx.Invoke(func(lf fx.Lifecycle, consumer ErrorOnApplyConsumer, d domain.Domain, logger logging.Logger) {
+	fx.Invoke(func(lf fx.Lifecycle, consumer ErrorOnApplyConsumer, d domain.Domain, logger *slog.Logger) {
 		lf.Append(fx.Hook{
 			OnStart: func(context.Context) error {
 				go ProcessErrorOnApply(consumer, d, logger)
@@ -190,11 +191,11 @@ var Module = fx.Module("app",
 	}),
 
 	fx.Provide(func(jc *nats.JetstreamClient, ev *env.Env, logger logging.Logger) (ResourceUpdateConsumer, error) {
-		topic := common.GetPlatformClusterMessagingTopic("*", "*", common.ConsoleReceiver, common.EventResourceUpdate)
+		topic := common.ReceiveFromAgentSubjectName(common.ReceiveFromAgentArgs{AccountName: "*", ClusterName: "*"}, common.ConsoleReceiver, common.EventResourceUpdate)
 
 		consumerName := "console:resource-updates"
 		return msg_nats.NewJetstreamConsumer(context.TODO(), jc, msg_nats.JetstreamConsumerArgs{
-			Stream: ev.NatsResourceSyncStream,
+			Stream: ev.NatsReceiveFromAgentStream,
 			ConsumerConfig: msg_nats.ConsumerConfig{
 				Name:           consumerName,
 				Durable:        consumerName,
@@ -204,7 +205,7 @@ var Module = fx.Module("app",
 		})
 	}),
 
-	fx.Invoke(func(lf fx.Lifecycle, consumer ResourceUpdateConsumer, d domain.Domain, logger logging.Logger) {
+	fx.Invoke(func(lf fx.Lifecycle, consumer ResourceUpdateConsumer, d domain.Domain, logger *slog.Logger) {
 		lf.Append(fx.Hook{
 			OnStart: func(context.Context) error {
 				go ProcessResourceUpdates(consumer, d, logger)
@@ -220,23 +221,23 @@ var Module = fx.Module("app",
 		return domain.NewSvcBindingDomain(svcBindingRepo)
 	}),
 
-	fx.Provide(func(logger logging.Logger, sbd domain.ServiceBindingDomain, ev *env.Env) *dnsHandler {
+	fx.Provide(func(logger *slog.Logger, sbd domain.ServiceBindingDomain, ev *env.Env) *dnsHandler {
 		return &dnsHandler{
 			logger:               logger,
 			serviceBindingDomain: sbd,
 			kloudliteDNSSuffix:   ev.KloudliteDNSSuffix,
 		}
 	}),
 
-	fx.Invoke(func(server *DNSServer, handler *dnsHandler, lf fx.Lifecycle, logger logging.Logger) {
+	fx.Invoke(func(server *DNSServer, handler *dnsHandler, lf fx.Lifecycle, logger *slog.Logger) {
 		lf.Append(fx.Hook{
 			OnStart: func(ctx context.Context) error {
-				logger.Infof("starting dns server at %s", server.Addr)
 				server.Handler = handler
 				go func() {
+					logger.Info("starting dns server", "at", server.Addr)
 					err := server.ListenAndServe()
 					if err != nil {
-						logger.Errorf(err, "failed to start dns server")
+						logger.Error("failed to start dns server, got", "err", err)
 						panic(err)
 					}
 				}()

apps/console/internal/app/dns-server.go

@@ -3,16 +3,17 @@ package app
 import (
 	"context"
 	"fmt"
+	"log/slog"
 	"strings"
+	"time"
 
 	"github.com/kloudlite/api/apps/console/internal/domain"
-	"github.com/kloudlite/api/pkg/logging"
 	"github.com/kloudlite/operator/pkg/errors"
 	"github.com/miekg/dns"
 )
 
 type dnsHandler struct {
-	logger               logging.Logger
+	logger               *slog.Logger
 	serviceBindingDomain domain.ServiceBindingDomain
 	kloudliteDNSSuffix   string
 }
@@ -22,8 +23,10 @@ const (
 )
 
 func (h *dnsHandler) ServeDNS(w dns.ResponseWriter, r *dns.Msg) {
-	logger := h.logger.WithKV("query", r.Question[0].Name)
-	logger.Debugf("incoming dns request")
+	logger := h.logger.With("query", r.Question[0].Name)
+	logger.Debug("INCOMING dns request")
+	start := time.Now()
+
 	msg := new(dns.Msg)
 	msg.SetReply(r)
 	msg.Authoritative = true
@@ -34,15 +37,17 @@ func (h *dnsHandler) ServeDNS(w dns.ResponseWriter, r *dns.Msg) {
 	for _, question := range r.Question {
 		answers, err := h.resolver(ctx, question.Name, question.Qtype)
 		if err != nil {
-			h.logger.Errorf(err)
+			logger.Error("FAILED to resolve dns record, got", "err", err, "question", question.Name)
 			msg.Rcode = dns.RcodeNameError
 			continue
 		}
 		msg.Answer = append(msg.Answer, answers...)
 	}
 
 	w.WriteMsg(msg)
-	logger.WithKV("answers", msg.Answer).Debugf("outgoing dns request")
+	if msg.Rcode != dns.RcodeNameError {
+		logger.Info("RESOLVED dns request", "answers", msg.Answer, "took", fmt.Sprintf("%.2fs", time.Since(start).Seconds()))
+	}
 }
 
 func (h *dnsHandler) newRR(domain string, ttl int, ip string) ([]dns.RR, error) {

apps/console/internal/app/process-error-on-apply.go

@@ -4,6 +4,9 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
+	"log/slog"
+	"sync"
+	"time"
 
 	t "github.com/kloudlite/api/apps/tenant-agent/types"
 	"github.com/kloudlite/api/pkg/errors"
@@ -13,17 +16,14 @@ import (
 	"github.com/kloudlite/api/apps/console/internal/entities"
 	msgOfficeT "github.com/kloudlite/api/apps/message-office/types"
 	fn "github.com/kloudlite/api/pkg/functions"
-	"github.com/kloudlite/api/pkg/logging"
 	"github.com/kloudlite/api/pkg/messaging"
 	msgTypes "github.com/kloudlite/api/pkg/messaging/types"
 	crdsv1 "github.com/kloudlite/operator/apis/crds/v1"
 )
 
 type ErrorOnApplyConsumer messaging.Consumer
 
-func ProcessErrorOnApply(consumer ErrorOnApplyConsumer, d domain.Domain, logger logging.Logger) {
-	counter := 0
-
+func ProcessErrorOnApply(consumer ErrorOnApplyConsumer, d domain.Domain, logger *slog.Logger) {
 	getEnvironmentResourceContext := func(ctx domain.ConsoleContext, resType entities.ResourceType, clusterName string, obj unstructured.Unstructured) (domain.ResourceContext, error) {
 		mapping, err := d.GetEnvironmentResourceMapping(ctx, resType, clusterName, obj.GetNamespace(), obj.GetName())
 		if err != nil {
@@ -35,9 +35,18 @@ func ProcessErrorOnApply(consumer ErrorOnApplyConsumer, d domain.Domain, logger
 		return newResourceContext(ctx, mapping.EnvironmentName), nil
 	}
 
+	counter := 0
+	mu := sync.Mutex{}
+
 	msgReader := func(msg *msgTypes.ConsumeMsg) error {
+		mu.Lock()
 		counter += 1
-		logger.Debugf("received message [%d]", counter)
+		mu.Unlock()
+
+		start := time.Now()
+
+		logger := logger.With("subject", msg.Subject, "counter", counter)
+		logger.Debug("INCOMING message", "counter", counter)
 
 		em, err := msgOfficeT.UnmarshalErrMessage(msg.Payload)
 		if err != nil {
@@ -50,25 +59,24 @@ func ProcessErrorOnApply(consumer ErrorOnApplyConsumer, d domain.Domain, logger
 		}
 
 		obj := unstructured.Unstructured{Object: errObj.Object}
+		gvkStr := obj.GroupVersionKind().String()
 
-		mLogger := logger.WithKV(
-			"gvk", obj.GroupVersionKind(),
-			"nn", fmt.Sprintf("%s/%s", obj.GetNamespace(), obj.GetName()),
+		mlogger := logger.With(
+			"GVK", gvkStr,
+			"NN", fmt.Sprintf("%s/%s", obj.GetNamespace(), obj.GetName()),
 			"accountName", em.AccountName,
 			"clusterName", em.ClusterName,
 		)
 
-		mLogger.Infof("received message")
+		mlogger.Info("validated message")
 		defer func() {
-			mLogger.Infof("processed message")
+			mlogger.Info("PROCESSED message", "took", fmt.Sprintf("%.2fs", time.Since(start).Seconds()))
 		}()
 
 		dctx := domain.NewConsoleContext(context.TODO(), "sys-user:apply-on-error-worker", em.AccountName)
 
 		opts := domain.UpdateAndDeleteOpts{MessageTimestamp: msg.Timestamp}
 
-		gvkStr := obj.GroupVersionKind().String()
-
 		switch gvkStr {
 		case environmentGVK.String():
 			{
@@ -204,10 +212,10 @@ func ProcessErrorOnApply(consumer ErrorOnApplyConsumer, d domain.Domain, logger
 
 	if err := consumer.Consume(msgReader, msgTypes.ConsumeOpts{
 		OnError: func(err error) error {
-			logger.Errorf(err, "received while reading messages, ignoring it")
+			logger.Error("while reading messages, got", "err", err)
 			return nil
 		},
 	}); err != nil {
-		logger.Errorf(err, "error while consuming messages")
+		logger.Error("while consuming messages, got", "err", err)
 	}
 }