replace slog with zap

TODO.md

@@ -154,3 +154,7 @@ Make it easier to debug modules by adding more logging and error handling. This
 
 Add stack tracing for typescript modules.
 
+
+## Decouple Admin API from Raft Implementation
+
+Currently, Raft Implementation is tightly coupled with the Admin API. This makes it difficult to change the Raft Implementation without changing the Admin API. Decouple the Raft Implementation from the Admin API to make it easier to change the Raft Implementation.

cmd/dgate-server/main.go

@@ -60,8 +60,14 @@ func main() {
 			fmt.Printf("Error loading config: %s\n", err)
 			os.Exit(1)
 		} else {
-			proxyState := proxy.StartProxyGateway(version, dgateConfig)
-			admin.StartAdminAPI(dgateConfig, proxyState)
+			logger, err := dgateConfig.GetLogger()
+			if err != nil {
+				fmt.Printf("Error setting up logger: %s\n", err)
+				os.Exit(1)
+			}
+			defer logger.Sync()
+			proxyState := proxy.NewProxyState(logger.Named("proxy"), dgateConfig)
+			admin.StartAdminAPI(version, dgateConfig, logger.Named("admin"), proxyState)
 			if err := proxyState.Start(); err != nil {
 				fmt.Printf("Error loading config: %s\n", err)
 				os.Exit(1)
@@ -75,7 +81,6 @@ func main() {
 			)
 			<-sigchan
 			proxyState.Stop()
-			os.Exit(1)
 		}
 	}
 }

go.mod

@@ -80,6 +80,8 @@ require (
 	go.opencensus.io v0.24.0 // indirect
 	go.opentelemetry.io/otel/sdk v1.26.0 // indirect
 	go.opentelemetry.io/otel/trace v1.26.0 // indirect
+	go.uber.org/multierr v1.11.0 // indirect
+	go.uber.org/zap v1.27.0 // indirect
 	golang.org/x/sys v0.19.0 // indirect
 	golang.org/x/text v0.14.0 // indirect
 	google.golang.org/protobuf v1.33.0 // indirect

go.sum

@@ -278,6 +278,10 @@ go.opentelemetry.io/otel/sdk/metric v1.26.0 h1:cWSks5tfriHPdWFnl+qpX3P681aAYqlZH
 go.opentelemetry.io/otel/sdk/metric v1.26.0/go.mod h1:ClMFFknnThJCksebJwz7KIyEDHO+nTB6gK8obLy8RyE=
 go.opentelemetry.io/otel/trace v1.26.0 h1:1ieeAUb4y0TE26jUFrCIXKpTuVK7uJGN9/Z/2LP5sQA=
 go.opentelemetry.io/otel/trace v1.26.0/go.mod h1:4iDxvGDQuUkHve82hJJ8UqrwswHYsZuWCBllGV2U2y0=
+go.uber.org/multierr v1.11.0 h1:blXXJkSxSSfBVBlC76pxqeO+LN3aDfLQo+309xJstO0=
+go.uber.org/multierr v1.11.0/go.mod h1:20+QtiLqy0Nd6FdQB9TLXag12DsQkrbs3htMFfDN80Y=
+go.uber.org/zap v1.27.0 h1:aJMhYGrd5QSmlpLMr2MftRKl7t8J8PTZPA732ud/XR8=
+go.uber.org/zap v1.27.0/go.mod h1:GB2qFLM7cTU87MWRP2mPIjqfIDnGu+VIO4V/SdhGo2E=
 golang.org/x/crypto v0.0.0-20180904163835-0709b304e793/go.mod h1:6SG95UA2DQfeDnfUPMdvaQW0Q7yPrPDi9nlGo2tz2b4=
 golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=
 golang.org/x/crypto v0.0.0-20191011191535-87dc89f01550/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI=

internal/admin/admin_api.go

@@ -13,39 +13,56 @@ import (
 	"github.com/dgate-io/dgate/internal/admin/changestate"
 	"github.com/dgate-io/dgate/internal/config"
 	"github.com/dgate-io/dgate/pkg/util"
-
-	"log/slog"
+	"go.uber.org/zap"
 
 	"golang.org/x/net/http2"
 	"golang.org/x/net/http2/h2c"
 )
 
-func StartAdminAPI(conf *config.DGateConfig, cs changestate.ChangeState) {
+func StartAdminAPI(
+	version string, conf *config.DGateConfig,
+	logger *zap.Logger, cs changestate.ChangeState,
+) {
 	if conf.AdminConfig == nil {
-		cs.Logger().Warn("Admin API is disabled")
+		logger.Warn("Admin API is disabled")
 		return
 	}
 
-	// Start HTTP Server
 	mux := chi.NewRouter()
-	configureRoutes(mux, cs, conf)
+	configureRoutes(mux, version,
+		logger.Named("routes"), cs, conf)
+
+	// Start HTTP Server
+	go func() {
+		adminHttpLogger := logger.Named("http")
+		hostPort := fmt.Sprintf("%s:%d",
+			conf.AdminConfig.Host, conf.AdminConfig.Port)
+		logger.Info("Starting admin api on " + hostPort)
+		server := &http.Server{
+			Addr:     hostPort,
+			Handler:  mux,
+			ErrorLog: zap.NewStdLog(adminHttpLogger),
+		}
+		if err := server.ListenAndServe(); err != nil {
+			panic(err)
+		}
+	}()
 
 	// Start HTTPS Server
 	go func() {
 		if conf.AdminConfig.TLS != nil {
-			adminHttpsLogHandler := cs.Logger().
-				Handler().WithGroup("admin-https")
+			adminHttpsLog := logger.Named("https")
 			secureHostPort := fmt.Sprintf("%s:%d",
 				conf.AdminConfig.Host, conf.AdminConfig.TLS.Port)
 			secureServer := &http.Server{
 				Addr:     secureHostPort,
 				Handler:  mux,
-				ErrorLog: slog.NewLogLogger(adminHttpsLogHandler, slog.LevelInfo),
+				ErrorLog: zap.NewStdLog(adminHttpsLog),
 			}
-			cs.Logger().Info("Starting secure admin api on" + secureHostPort)
-			cs.Logger().Debug("TLS Cert",
-				"cert_file", conf.AdminConfig.TLS.CertFile,
-				"key_file", conf.AdminConfig.TLS.KeyFile,
+			logger.Info("Starting secure admin api on" + secureHostPort)
+			logger.Debug("TLS Cert",
+				zap.String("cert_file", conf.AdminConfig.TLS.CertFile),
+				zap.String("key_file", conf.AdminConfig.TLS.KeyFile),
 			)
 			if err := secureServer.ListenAndServeTLS(
 				conf.AdminConfig.TLS.CertFile,
@@ -59,13 +76,13 @@ func StartAdminAPI(conf *config.DGateConfig, cs changestate.ChangeState) {
 	// Start Test Server
 	if conf.TestServerConfig != nil {
 		if !conf.Debug {
-			cs.Logger().Warn("Test server is disabled in non-debug mode")
+			logger.Warn("Test server is disabled in non-debug mode")
 		} else {
 			go func() {
 				testHostPort := fmt.Sprintf("%s:%d",
 					conf.TestServerConfig.Host, conf.TestServerConfig.Port)
-				mux := chi.NewRouter()
-				mux.HandleFunc("/*", func(w http.ResponseWriter, r *http.Request) {
+				testMux := chi.NewRouter()
+				testMux.HandleFunc("/*", func(w http.ResponseWriter, r *http.Request) {
 					if strings.HasPrefix(r.URL.Path, "/debug") {
 						// strip /debug prefix
 						r.URL.Path = strings.TrimPrefix(r.URL.Path, "/debug")
@@ -102,13 +119,11 @@ func StartAdminAPI(conf *config.DGateConfig, cs changestate.ChangeState) {
 					util.JsonResponse(w, http.StatusOK, respMap)
 				})
 
-				testServerLogger := cs.Logger().
-					WithGroup("test-server")
-
+				testServerLogger := logger.Named("test-server")
 				testServer := &http.Server{
 					Addr:     testHostPort,
-					Handler:  mux,
-					ErrorLog: slog.NewLogLogger(testServerLogger.Handler(), slog.LevelInfo),
+					Handler:  testMux,
+					ErrorLog: zap.NewStdLog(testServerLogger),
 				}
 				if conf.TestServerConfig.EnableHTTP2 {
 					h2Server := &http2.Server{}
@@ -117,29 +132,15 @@ func StartAdminAPI(conf *config.DGateConfig, cs changestate.ChangeState) {
 						panic(err)
 					}
 					if conf.TestServerConfig.EnableH2C {
-						testServer.Handler = h2c.NewHandler(mux, h2Server)
+						testServer.Handler = h2c.NewHandler(testServer.Handler, h2Server)
 					}
 				}
-				cs.Logger().Info("Starting test server on " + testHostPort)
+				logger.Info("Starting test server on " + testHostPort)
 
 				if err := testServer.ListenAndServe(); err != nil {
 					panic(err)
 				}
 			}()
 		}
 	}
-	go func() {
-		adminHttpLogger := cs.Logger().WithGroup("admin-http")
-		hostPort := fmt.Sprintf("%s:%d",
-			conf.AdminConfig.Host, conf.AdminConfig.Port)
-		cs.Logger().Info("Starting admin api on " + hostPort)
-		server := &http.Server{
-			Addr:     hostPort,
-			Handler:  mux,
-			ErrorLog: slog.NewLogLogger(adminHttpLogger.Handler(), slog.LevelInfo),
-		}
-		if err := server.ListenAndServe(); err != nil {
-			panic(err)
-		}
-	}()
 }

internal/admin/admin_fsm.go

@@ -3,26 +3,22 @@ package admin
 import (
 	"encoding/json"
 	"io"
-	"log/slog"
 
 	"github.com/dgate-io/dgate/internal/admin/changestate"
 	"github.com/dgate-io/dgate/pkg/spec"
 	"github.com/hashicorp/raft"
+	"go.uber.org/zap"
 )
 
 type dgateAdminFSM struct {
 	cs     changestate.ChangeState
-	logger *slog.Logger
+	logger *zap.Logger
 }
 
 var _ raft.BatchingFSM = (*dgateAdminFSM)(nil)
 
-func newDGateAdminFSM(cs changestate.ChangeState) *dgateAdminFSM {
-	dgateFSMLogger := cs.Logger().WithGroup("admin-raft-fsm")
-	return &dgateAdminFSM{
-		cs:     cs,
-		logger: dgateFSMLogger,
-	}
+func newDGateAdminFSM(logger *zap.Logger, cs changestate.ChangeState) *dgateAdminFSM {
+	return &dgateAdminFSM{cs, logger}
 }
 
 func (fsm *dgateAdminFSM) isReplay(log *raft.Log) bool {
@@ -35,15 +31,13 @@ func (fsm *dgateAdminFSM) checkLast(log *raft.Log) {
 	rft := fsm.cs.Raft()
 	if !fsm.cs.Ready() && fsm.isReplay(log) {
 		fsm.logger.Info("FSM is not ready, setting ready",
-			"Index", log.Index,
-			"AIndex", rft.AppliedIndex(),
-			"LIndex", rft.LastIndex(),
+			zap.Uint64("index", log.Index),
+			zap.Uint64("applied-index", rft.AppliedIndex()),
+			zap.Uint64("last-index", rft.LastIndex()),
 		)
 		defer func() {
 			if err := fsm.cs.ReloadState(false); err != nil {
-				fsm.logger.Error("Error processing change log in FSM",
-					"error", err,
-				)
+				fsm.logger.Error("Error processing change log in FSM", zap.Error(err))
 			} else {
 				fsm.cs.SetReady()
 			}
@@ -56,12 +50,12 @@ func (fsm *dgateAdminFSM) applyLog(log *raft.Log) (*spec.ChangeLog, error) {
 	case raft.LogCommand:
 		var cl spec.ChangeLog
 		if err := json.Unmarshal(log.Data, &cl); err != nil {
-			fsm.logger.Error("Error unmarshalling change log")
+			fsm.logger.Error("Error unmarshalling change log", zap.Error(err))
 			return nil, err
 		} else if cl.Cmd.IsNoop() {
 			return nil, nil
 		} else if cl.ID == "" {
-			fsm.logger.Error("Change log ID is empty")
+			fsm.logger.Error("Change log ID is empty", zap.Error(err))
 			panic("change log ID is empty")
 		}
 		// find a way to apply only if latest index to save time
@@ -72,16 +66,17 @@ func (fsm *dgateAdminFSM) applyLog(log *raft.Log) (*spec.ChangeLog, error) {
 		servers := raft.DecodeConfiguration(log.Data).Servers
 		for i, server := range servers {
 			fsm.logger.Debug("configuration update server",
-				"address", server.Address, "index", i,
+				zap.Any("address", server.Address),
+				zap.Int("index", i),
 			)
 		}
 	case raft.LogBarrier:
 		err := fsm.cs.WaitForChanges()
 		if err != nil {
-			fsm.logger.Error("Error waiting for changes", "error", err)
+			fsm.logger.Error("Error waiting for changes", zap.Error(err))
 		}
 	default:
-		fsm.cs.Logger().Error("Unknown log type in FSM Apply")
+		fsm.logger.Error("Unknown log type in FSM Apply")
 	}
 	return nil, nil
 }
@@ -97,11 +92,11 @@ func (fsm *dgateAdminFSM) ApplyBatch(logs []*raft.Log) []any {
 	if fsm.isReplay(lastLog) {
 		rft := fsm.cs.Raft()
 		fsm.logger.Info("applying log batch logs",
-			"size", len(logs),
-			"current", lastLog.Index,
-			"applied", rft.AppliedIndex(),
-			"commit", rft.CommitIndex(),
-			"last", rft.LastIndex(),
+			zap.Int("size", len(logs)),
+			zap.Uint64("current", lastLog.Index),
+			zap.Uint64("applied", rft.AppliedIndex()),
+			zap.Uint64("commit", rft.CommitIndex()),
+			zap.Uint64("last", rft.LastIndex()),
 		)
 	}
 	cls := make([]*spec.ChangeLog, 0, len(logs))
@@ -112,7 +107,7 @@ func (fsm *dgateAdminFSM) ApplyBatch(logs []*raft.Log) []any {
 		}
 
 		if err := fsm.cs.ReloadState(true, cls...); err != nil {
-			fsm.logger.Error("Error reloading state @ FSM ApplyBatch")
+			fsm.logger.Error("Error reloading state @ FSM ApplyBatch", zap.Error(err))
 		}
 	}()