refactor admin, raft, and tests: fix deadlocks, slow startup, and

.github/workflows/e2e.yml

@@ -52,7 +52,7 @@ jobs:
       - run: cd functional-tests/raft_tests && goreman start &
 
       - name: Wait for server to start
-        run: sleep 5
+        run: sleep 10
 
       - name: Functional Standalone Tests
         run: |

cmd/dgate-server/main.go

@@ -58,19 +58,23 @@ func main() {
 		}
 		if dgateConfig, err := config.LoadConfig(*configPath); err != nil {
 			fmt.Printf("Error loading config: %s\n", err)
-			os.Exit(1)
+			panic(err)
 		} else {
 			logger, err := dgateConfig.GetLogger()
 			if err != nil {
 				fmt.Printf("Error setting up logger: %s\n", err)
-				os.Exit(1)
+				panic(err)
 			}
 			defer logger.Sync()
 			proxyState := proxy.NewProxyState(logger.Named("proxy"), dgateConfig)
-			admin.StartAdminAPI(version, dgateConfig, logger.Named("admin"), proxyState)
+			err = admin.StartAdminAPI(version, dgateConfig, logger.Named("admin"), proxyState)
+			if err != nil {
+				fmt.Printf("Error starting admin api: %s\n", err)
+				panic(err)
+			}
 			if err := proxyState.Start(); err != nil {
 				fmt.Printf("Error loading config: %s\n", err)
-				os.Exit(1)
+				panic(err)
 			}
 
 			sigchan := make(chan os.Signal, 1)

functional-tests/admin_tests/admin_test.sh

@@ -8,7 +8,8 @@ TEST_URL=${TEST_URL:-"http://localhost:8888"}
 
 DIR="$( cd "$( dirname "$0" )" && pwd )"
 
-# domain setup
+export DGATE_ADMIN_API=$ADMIN_URL
+
 # check if uuid is available
 if ! command -v uuid > /dev/null; then
     id=X$RANDOM-$RANDOM-$RANDOM

go.mod

@@ -26,12 +26,14 @@ require (
 	github.com/stoewer/go-strcase v1.3.0
 	github.com/stretchr/testify v1.9.0
 	github.com/urfave/cli/v2 v2.27.1
+	go.etcd.io/bbolt v1.3.10
 	go.opentelemetry.io/otel v1.26.0
 	go.opentelemetry.io/otel/exporters/prometheus v0.48.0
 	go.opentelemetry.io/otel/metric v1.26.0
 	go.opentelemetry.io/otel/sdk/metric v1.26.0
 	go.uber.org/zap v1.27.0
 	golang.org/x/net v0.21.0
+	golang.org/x/sync v0.7.0
 	golang.org/x/term v0.19.0
 )
 
@@ -80,12 +82,10 @@ require (
 	github.com/russross/blackfriday/v2 v2.1.0 // indirect
 	github.com/stretchr/objx v0.5.2 // indirect
 	github.com/xrash/smetrics v0.0.0-20201216005158-039620a65673 // indirect
-	go.etcd.io/bbolt v1.3.10 // indirect
 	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
-	golang.org/x/sync v0.7.0 // indirect
 	golang.org/x/sys v0.21.0 // indirect
 	golang.org/x/text v0.14.0 // indirect
 	google.golang.org/protobuf v1.33.0 // indirect

internal/admin/admin_fsm.go

@@ -39,6 +39,7 @@ func newAdminFSM(
 			logger.Warn("corrupted state detected", zap.ByteString("prev_state", stateBytes))
 		} else {
 			logger.Info("found state in store", zap.Any("prev_state", fsm.localState))
+			return fsm
 		}
 	}
 	return fsm
@@ -76,11 +77,10 @@ func (fsm *AdminFSM) applyLog(log *raft.Log, reload bool) (*spec.ChangeLog, erro
 }
 
 func (fsm *AdminFSM) Apply(log *raft.Log) any {
-	resps := fsm.ApplyBatch([]*raft.Log{log})
-	if len(resps) != 1 {
-		panic("apply batch not returning the correct number of responses")
+	if resps := fsm.ApplyBatch([]*raft.Log{log}); len(resps) == 1 {
+		return resps[0]
 	}
-	return resps[0]
+	panic("apply batch not returning the correct number of responses")
 }
 
 func (fsm *AdminFSM) ApplyBatch(logs []*raft.Log) []any {
@@ -115,7 +115,7 @@ func (fsm *AdminFSM) ApplyBatch(logs []*raft.Log) []any {
 				zap.Uint64("applied_index", lastLogIndex),
 			)
 		}
-		fsm.cs.SetReady(true)
+		// defer fsm.cs.SetReady(true)
 	}
 
 	return results

internal/admin/admin_raft.go

@@ -92,13 +92,9 @@ func setupRaft(
 	address := raft.ServerAddress(advertAddr)
 
 	raftHttpLogger := logger.Named("http")
-	if adminConfig.Replication.AdvertScheme != "http" && adminConfig.Replication.AdvertScheme != "https" {
-		panic(fmt.Errorf("invalid scheme: %s", adminConfig.Replication.AdvertScheme))
-	}
-
 	transport := rafthttp.NewHTTPTransport(
 		address, http.DefaultClient, raftHttpLogger,
-		adminConfig.Replication.AdvertScheme+"://(address)/raft",
+		adminConfig.Replication.AdvertScheme,
 	)
 	fsmLogger := logger.Named("fsm")
 	adminFSM := newAdminFSM(fsmLogger, configStore, cs)
@@ -110,14 +106,13 @@ func setupRaft(
 		panic(err)
 	}
 
-	cs.SetupRaft(raftNode)
-
 	// Setup raft handler
 	server.Handle("/raft/*", transport)
 
 	raftAdminLogger := logger.Named("admin")
-	raftAdmin := raftadmin.NewRaftAdminHTTPServer(
-		raftNode, raftAdminLogger, []raft.ServerAddress{address},
+	raftAdmin := raftadmin.NewServer(
+		raftNode, raftAdminLogger,
+		[]raft.ServerAddress{address},
 	)
 
 	// Setup handler for raft admin
@@ -138,6 +133,39 @@ func setupRaft(
 		util.JsonResponse(w, http.StatusOK, raftNode.Stats())
 	}))
 
+	// Setup handler for readys
+	server.Handle("/raftadmin/readyz", http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		w.Header().Set("X-Raft-State", raftNode.State().String())
+		if err := cs.WaitForChanges(nil); err != nil {
+			http.Error(w, err.Error(), http.StatusInternalServerError)
+			return
+		}
+		leaderId, leaderAddr := raftNode.LeaderWithID()
+		util.JsonResponse(w, http.StatusOK, map[string]any{
+			"status":      "ok",
+			"proxy_ready": cs.Ready(),
+			"state":       raftNode.State().String(),
+			"leader":      leaderId,
+			"leader_addr": leaderAddr,
+		})
+	}))
+
+	doer := func(req *http.Request) (*http.Response, error) {
+		req.Header.Set("User-Agent", "dgate")
+		if adminConfig.Replication.SharedKey != "" {
+			req.Header.Set("X-DGate-Shared-Key", adminConfig.Replication.SharedKey)
+		}
+		client := *http.DefaultClient
+		client.Timeout = time.Second * 10
+		return client.Do(req)
+	}
+	adminClient := raftadmin.NewClient(
+		doer, logger.Named("raft-admin-client"),
+		adminConfig.Replication.AdvertScheme,
+	)
+
+	cs.SetupRaft(raftNode, adminClient)
+
 	configFuture := raftNode.GetConfiguration()
 	if err = configFuture.Error(); err != nil {
 		panic(err)
@@ -203,33 +231,25 @@ func setupRaft(
 			if len(addresses) > 0 {
 				addresses = append(addresses, adminConfig.Replication.ClusterAddrs...)
 				retries := 0
-				doer := func(req *http.Request) (*http.Response, error) {
-					req.Header.Set("User-Agent", "dgate")
-					if adminConfig.Replication.SharedKey != "" {
-						req.Header.Set("X-DGate-Shared-Key", adminConfig.Replication.SharedKey)
-					}
-					return http.DefaultClient.Do(req)
-				}
-				adminClient := raftadmin.NewHTTPAdminClient(doer,
-					adminConfig.Replication.AdvertScheme+"://(address)/raftadmin",
-					logger.Named("raft-admin-client"),
-				)
 			RETRY:
-				for _, url := range addresses {
-					err = adminClient.VerifyLeader(context.Background(), raft.ServerAddress(url))
+				for _, addr := range addresses {
+					err = adminClient.VerifyLeader(
+						context.Background(),
+						raft.ServerAddress(addr),
+					)
 					if err != nil {
 						if err == raftadmin.ErrNotLeader {
 							continue
 						}
 						if retries > 15 {
 							logger.Error("Skipping verifying leader",
-								zap.String("url", url), zap.Error(err),
+								zap.String("url", addr), zap.Error(err),
 							)
 							continue
 						}
 						retries += 1
 						logger.Debug("Retrying verifying leader",
-							zap.String("url", url), zap.Error(err))
+							zap.String("url", addr), zap.Error(err))
 						<-time.After(3 * time.Second)
 						goto RETRY
 					}
@@ -238,10 +258,10 @@ func setupRaft(
 						logger.Info("Adding non-voter",
 							zap.String("id", raftId),
 							zap.String("leader", adminConfig.Replication.AdvertAddr),
-							zap.String("url", url),
+							zap.String("url", addr),
 						)
 						resp, err := adminClient.AddNonvoter(
-							context.Background(), raft.ServerAddress(url),
+							context.Background(), raft.ServerAddress(addr),
 							&raftadmin.AddNonvoterRequest{
 								ID:      raftId,
 								Address: adminConfig.Replication.AdvertAddr,
@@ -257,9 +277,9 @@ func setupRaft(
 						logger.Info("Adding voter: %s - leader: %s",
 							zap.String("id", raftId),
 							zap.String("leader", adminConfig.Replication.AdvertAddr),
-							zap.String("url", url),
+							zap.String("url", addr),
 						)
-						resp, err := adminClient.AddVoter(context.Background(), raft.ServerAddress(url), &raftadmin.AddVoterRequest{
+						resp, err := adminClient.AddVoter(context.Background(), raft.ServerAddress(addr), &raftadmin.AddVoterRequest{
 							ID:      raftId,
 							Address: adminConfig.Replication.AdvertAddr,
 						})

internal/admin/admin_routes_test.go

@@ -17,7 +17,9 @@ func TestAdminRoutes_configureRoutes(t *testing.T) {
 	cs.On("DocumentManager").Return(nil)
 	conf := configtest.NewTestAdminConfig()
 	if err := configureRoutes(
-		mux, "test", zap.NewNop(), cs, conf,
+		mux, "test",
+		zap.NewNop(),
+		cs, conf,
 	); err != nil {
 		t.Fatal(err)
 	}

internal/admin/changestate/change_state.go

@@ -2,6 +2,7 @@ package changestate
 
 import (
 	"github.com/dgate-io/dgate/internal/proxy"
+	"github.com/dgate-io/dgate/pkg/raftadmin"
 	"github.com/dgate-io/dgate/pkg/resources"
 	"github.com/dgate-io/dgate/pkg/spec"
 	"github.com/hashicorp/raft"
@@ -21,7 +22,7 @@ type ChangeState interface {
 	SetReady(bool)
 
 	// Replication
-	SetupRaft(*raft.Raft)
+	SetupRaft(*raft.Raft, *raftadmin.Client)
 	Raft() *raft.Raft
 
 	// Resources

internal/admin/changestate/testutil/change_state.go

@@ -7,6 +7,7 @@ import (
 	"github.com/dgate-io/dgate/internal/admin/changestate"
 	"github.com/dgate-io/dgate/pkg/resources"
 	"github.com/dgate-io/dgate/pkg/spec"
+	"github.com/dgate-io/dgate/pkg/raftadmin"
 	"github.com/hashicorp/raft"
 	"github.com/stretchr/testify/mock"
 )
@@ -71,7 +72,7 @@ func (m *MockChangeState) ReloadState(a bool, cls ...*spec.ChangeLog) error {
 }
 
 // SetupRaft implements changestate.ChangeState.
-func (m *MockChangeState) SetupRaft(*raft.Raft) {
+func (m *MockChangeState) SetupRaft(*raft.Raft, *raftadmin.Client) {
 	m.Called().Error(0)
 }
 

internal/admin/routes/collection_routes.go

@@ -63,8 +63,7 @@ func ConfigureCollectionAPI(server chi.Router, logger *zap.Logger, cs changestat
 		}
 
 		cl := spec.NewChangeLog(&collection, collection.NamespaceName, spec.AddCollectionCommand)
-		err = cs.ApplyChangeLog(cl)
-		if err != nil {
+		if err = cs.ApplyChangeLog(cl); err != nil {
 			util.JsonError(w, http.StatusInternalServerError, err.Error())
 			return
 		}
@@ -268,8 +267,7 @@ func ConfigureCollectionAPI(server chi.Router, logger *zap.Logger, cs changestat
 		}
 
 		cl := spec.NewChangeLog(&doc, doc.NamespaceName, spec.AddDocumentCommand)
-		err = cs.ApplyChangeLog(cl)
-		if err != nil {
+		if err = cs.ApplyChangeLog(cl); err != nil {
 			util.JsonError(w, http.StatusInternalServerError, err.Error())
 			return
 		}
@@ -350,8 +348,7 @@ func ConfigureCollectionAPI(server chi.Router, logger *zap.Logger, cs changestat
 			return
 		}
 		cl := spec.NewChangeLog(document, namespaceName, spec.DeleteDocumentCommand)
-		err = cs.ApplyChangeLog(cl)
-		if err != nil {
+		if err = cs.ApplyChangeLog(cl); err != nil {
 			util.JsonError(w, http.StatusInternalServerError, err.Error())
 			return
 		}

internal/admin/routes/domain_routes.go

@@ -41,8 +41,7 @@ func ConfigureDomainAPI(server chi.Router, logger *zap.Logger, cs changestate.Ch
 			domain.NamespaceName = spec.DefaultNamespace.Name
 		}
 		cl := spec.NewChangeLog(&domain, domain.NamespaceName, spec.AddDomainCommand)
-		err = cs.ApplyChangeLog(cl)
-		if err != nil {
+		if err = cs.ApplyChangeLog(cl); err != nil {
 			util.JsonError(w, http.StatusBadRequest, err.Error())
 			return
 		}

internal/admin/routes/misc_routes.go

@@ -45,11 +45,6 @@ func ConfigureHealthAPI(server chi.Router, version string, cs changestate.Change
 		w.Header().Set("Content-Type", "application/json")
 		if cs.Ready() {
 			if r := cs.Raft(); r != nil {
-				if err := cs.WaitForChanges(nil); err != nil {
-					w.WriteHeader(http.StatusServiceUnavailable)
-					w.Write([]byte(`{"status":"not ready"}`))
-					return
-				}
 				w.Header().Set("X-Raft-State", r.State().String())
 				if leaderAddr := r.Leader(); leaderAddr == "" {
 					w.WriteHeader(http.StatusServiceUnavailable)

internal/admin/routes/module_routes.go

@@ -78,8 +78,7 @@ func ConfigureModuleAPI(server chi.Router, logger *zap.Logger, cs changestate.Ch
 			mod.NamespaceName = spec.DefaultNamespace.Name
 		}
 		cl := spec.NewChangeLog(&mod, mod.NamespaceName, spec.DeleteModuleCommand)
-		err = cs.ApplyChangeLog(cl)
-		if err != nil {
+		if err = cs.ApplyChangeLog(cl); err != nil {
 			util.JsonError(w, http.StatusBadRequest, err.Error())
 			return
 		}