diff --git a/cmds/core-service/main.go b/cmds/core-service/main.go index 33e9e5922..19701412e 100644 --- a/cmds/core-service/main.go +++ b/cmds/core-service/main.go @@ -24,6 +24,7 @@ import ( auxs "github.com/interuss/dss/pkg/aux_/store" "github.com/interuss/dss/pkg/build" dsserr "github.com/interuss/dss/pkg/errors" + requestlocality "github.com/interuss/dss/pkg/locality" "github.com/interuss/dss/pkg/logging" "github.com/interuss/dss/pkg/rid/application" rid_v1 "github.com/interuss/dss/pkg/rid/server/v1" @@ -100,7 +101,7 @@ func createAuxServer(ctx context.Context, locality string, publicEndpoint string return nil, stacktrace.NewError("Public endpoint not set") } - auxStore, err := auxs.Init(ctx, logger, true) + auxStore, err := auxs.Init(ctx, logger, true, locality) if err != nil { return nil, err } @@ -121,7 +122,7 @@ func createAuxServer(ctx context.Context, locality string, publicEndpoint string func createRIDServers(ctx context.Context, locality string, logger *zap.Logger) (*rid_v1.Server, *rid_v2.Server, error) { - ridStore, err := rids.Init(ctx, logger, true) + ridStore, err := rids.Init(ctx, logger, true, locality) if err != nil { return nil, nil, err } @@ -151,9 +152,9 @@ func createRIDServers(ctx context.Context, locality string, logger *zap.Logger) }, nil } -func createSCDServer(ctx context.Context, logger *zap.Logger) (*scd.Server, error) { +func createSCDServer(ctx context.Context, logger *zap.Logger, locality string) (*scd.Server, error) { - scdStore, err := scds.Init(ctx, logger, true) + scdStore, err := scds.Init(ctx, logger, true, locality) if err != nil { return nil, err } @@ -350,7 +351,7 @@ func RunHTTPServer(ctx context.Context, ctxCanceler func(), address, locality st // Initialize strategic conflict detection if *enableSCD { - scdV1Server, err = createSCDServer(ctx, logger) + scdV1Server, err = createSCDServer(ctx, logger, locality) if err != nil { return stacktrace.Propagate(err, "Failed to create strategic conflict detection server") } @@ -365,6 +366,7 @@ func RunHTTPServer(ctx context.Context, ctxCanceler func(), address, locality st handler = http.TimeoutHandler(handler, *timeout, "request timeout") handler = logging.HTTPMiddleware(logger, *dumpRequests, handler) handler = timestamp.RequestTimestampMiddleware(handler) + handler = requestlocality.LocalityMiddleware(locality)(handler) if *enableMetrics || *enableTracing { // We use the default settings; the APIRouter handler will override the span value accordingly, as it has more information. diff --git a/cmds/db-manager/cleanup/evict.go b/cmds/db-manager/cleanup/evict.go index 6cdf70d62..c03c0b7a9 100644 --- a/cmds/db-manager/cleanup/evict.go +++ b/cmds/db-manager/cleanup/evict.go @@ -55,12 +55,12 @@ func evict(cmd *cobra.Command, _ []string) error { logger := logging.WithValuesFromContext(ctx, logging.Logger) - scdStore, err := scds.Init(ctx, logger, false) + scdStore, err := scds.Init(ctx, logger, false, *locality) if err != nil { return err } - ridStore, err := rids.Init(ctx, logger, false) + ridStore, err := rids.Init(ctx, logger, false, *locality) if err != nil { return err } diff --git a/pkg/aux_/store/raftstore/store.go b/pkg/aux_/store/raftstore/store.go index 3c1fd3c33..8f8a4270f 100644 --- a/pkg/aux_/store/raftstore/store.go +++ b/pkg/aux_/store/raftstore/store.go @@ -28,7 +28,7 @@ type repo struct { memRepo repos.Repository } -func Init(ctx context.Context, logger *zap.Logger) (*raftstore.Store[repos.Repository], error) { +func Init(ctx context.Context, logger *zap.Logger, locality string) (*raftstore.Store[repos.Repository], error) { params, err := auxraftparams.GetConnectParameters() if err != nil { return nil, stacktrace.Propagate(err, "failed to get aux raft parameters") @@ -40,7 +40,7 @@ func Init(ctx context.Context, logger *zap.Logger) (*raftstore.Store[repos.Repos } r := &repo{memStore: memStore, memRepo: memStore.GetRepo()} - store, err := raftstore.Init(ctx, logger.With(zap.String("service", "aux_")), params, r, nil) + store, err := raftstore.Init(ctx, logger.With(zap.String("service", "aux_")), locality, params, r, nil) if err != nil { return nil, stacktrace.Propagate(err, "failed to initialize aux raftstore") } diff --git a/pkg/aux_/store/store.go b/pkg/aux_/store/store.go index 2532842a7..93148ff80 100644 --- a/pkg/aux_/store/store.go +++ b/pkg/aux_/store/store.go @@ -19,12 +19,12 @@ import ( type Store = dssstore.Store[repos.Repository] // Init selects and initializes the aux store backend. -func Init(ctx context.Context, logger *zap.Logger, withCheckCron bool) (Store, error) { +func Init(ctx context.Context, logger *zap.Logger, withCheckCron bool, locality string) (Store, error) { switch storeType := params.GetStoreParameters().StoreType; storeType { case params.SQLStoreType: return auxsqlstore.Init(ctx, logger, withCheckCron) case params.RaftStoreType: - return auxraftstore.Init(ctx, logger) + return auxraftstore.Init(ctx, logger, locality) case params.MemStoreType: return auxmemstore.Init(ctx, logger) default: diff --git a/pkg/locality/locality.go b/pkg/locality/locality.go new file mode 100644 index 000000000..8ea51faa8 --- /dev/null +++ b/pkg/locality/locality.go @@ -0,0 +1,36 @@ +package locality + +import ( + "context" + "net/http" + + "github.com/interuss/stacktrace" +) + +type localityKey struct{} + +// MustGetRequestLocality returns the request locality from the context and panics if it is not +// present, which is a programming error. +func MustGetRequestLocality(ctx context.Context) string { + locality, ok := ctx.Value(localityKey{}).(string) + if !ok { + panic(stacktrace.NewError("request locality not present in context")) + } + + return locality +} + +// WithRequestLocality returns a new context with the given locality. +func WithRequestLocality(ctx context.Context, locality string) context.Context { + return context.WithValue(ctx, localityKey{}, locality) +} + +// LocalityMiddleware is an HTTP middleware that stamps each incoming request with this +// DSS instance's locality so that locality-dependent operations execute deterministically across nodes. +func LocalityMiddleware(locality string) func(http.Handler) http.Handler { + return func(next http.Handler) http.Handler { + return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + next.ServeHTTP(w, r.WithContext(WithRequestLocality(r.Context(), locality))) + }) + } +} diff --git a/pkg/raftstore/consensus/consensus.go b/pkg/raftstore/consensus/consensus.go index 145f53e4b..042de04d3 100644 --- a/pkg/raftstore/consensus/consensus.go +++ b/pkg/raftstore/consensus/consensus.go @@ -24,8 +24,9 @@ import ( type Consensus struct { logger *zap.Logger - nodeID uint64 - node raft.Node + locality string + nodeID uint64 + node raft.Node transport *rafthttp.Transport server *http.Server @@ -45,7 +46,7 @@ type Consensus struct { appliedIndex uint64 } -func NewConsensus(ctx context.Context, logger *zap.Logger, connectParams params.ConnectParameters, provider snapshotProvider, commitC chan<- EntryCommit) (*Consensus, error) { +func NewConsensus(ctx context.Context, logger *zap.Logger, locality string, connectParams params.ConnectParameters, provider snapshotProvider, commitC chan<- EntryCommit) (*Consensus, error) { storage, old, err := newStorage(ctx, logger.With(zap.String("component", "storage")), connectParams.DataDir, connectParams.NodeID, provider, connectParams.SnapshotCatchupEntries) if err != nil { return nil, stacktrace.Propagate(err, "failed to initialize storage") @@ -74,11 +75,11 @@ func NewConsensus(ctx context.Context, logger *zap.Logger, connectParams params. consensus := &Consensus{ logger: logging.WithValuesFromContext(ctx, logger), - nodeID: connectParams.NodeID, - node: node, - - storage: storage, - commitC: commitC, + nodeID: connectParams.NodeID, + node: node, + locality: locality, + storage: storage, + commitC: commitC, shutdownTimeout: 2 * connectParams.ElectionInterval(), serverErrC: make(chan error, 1), diff --git a/pkg/raftstore/consensus/proposal.go b/pkg/raftstore/consensus/proposal.go index 7e86f124e..eb002ced6 100644 --- a/pkg/raftstore/consensus/proposal.go +++ b/pkg/raftstore/consensus/proposal.go @@ -18,6 +18,7 @@ type EntryCommit struct { type Proposal struct { ID string `json:"id"` + Locality string `json:"locality"` NodeID uint64 `json:"node_id"` Timestamp time.Time `json:"timestamp"` RequestType RequestType `json:"request_type"` @@ -34,6 +35,7 @@ func (c *Consensus) newProposal(ctx context.Context, requestType RequestType, va return Proposal{ ID: uuid.NewString(), + Locality: c.locality, NodeID: c.nodeID, Timestamp: timestamp.UTC(), RequestType: requestType, diff --git a/pkg/raftstore/store.go b/pkg/raftstore/store.go index b01224d2a..83ef22b33 100644 --- a/pkg/raftstore/store.go +++ b/pkg/raftstore/store.go @@ -3,6 +3,7 @@ package raftstore import ( "context" + "github.com/interuss/dss/pkg/locality" "github.com/interuss/dss/pkg/logging" "github.com/interuss/dss/pkg/raftstore/consensus" raftparams "github.com/interuss/dss/pkg/raftstore/params" @@ -40,7 +41,7 @@ type Store[R any] struct { done chan struct{} } -func Init[R any](ctx context.Context, logger *zap.Logger, params raftparams.ConnectParameters, r RaftRepo[R], registry map[string]store.OperationHandler[R]) (*Store[R], error) { +func Init[R any](ctx context.Context, logger *zap.Logger, locality string, params raftparams.ConnectParameters, r RaftRepo[R], registry map[string]store.OperationHandler[R]) (*Store[R], error) { ctx, cancel := context.WithCancel(ctx) store := &Store[R]{ @@ -56,7 +57,7 @@ func Init[R any](ctx context.Context, logger *zap.Logger, params raftparams.Conn store.processCommits(ctx, commitC) }() - consensusInstance, err := consensus.NewConsensus(ctx, logger, params, r.GetSnapshot, commitC) + consensusInstance, err := consensus.NewConsensus(ctx, logger, locality, params, r.GetSnapshot, commitC) if err != nil { return nil, stacktrace.Propagate(err, "failed to initialize consensus") } @@ -115,6 +116,7 @@ func (s *Store[R]) processCommits(ctx context.Context, commitCh <-chan consensus } proposalCtx := timestamp.WithRequestTimestamp(ctx, commit.Prop.Timestamp) + proposalCtx = locality.WithRequestLocality(proposalCtx, commit.Prop.Locality) result, err := s.raftRepo.Apply(proposalCtx, commit.Prop) commit.Done <- consensus.ProposalResult{Result: result, Error: err} } diff --git a/pkg/rid/store/raftstore/store.go b/pkg/rid/store/raftstore/store.go index bc55a3dca..a5636989c 100644 --- a/pkg/rid/store/raftstore/store.go +++ b/pkg/rid/store/raftstore/store.go @@ -16,12 +16,13 @@ import ( // repo is a full implementation of rid.repos.Repository for Raft-based storage. type repo struct{} -func Init(ctx context.Context, logger *zap.Logger) (*raftstore.Store[repos.Repository], error) { +func Init(ctx context.Context, logger *zap.Logger, locality string) (*raftstore.Store[repos.Repository], error) { params, err := ridraftparams.GetConnectParameters() if err != nil { return nil, stacktrace.Propagate(err, "failed to get rid raft parameters") } - return raftstore.Init(ctx, logger.With(zap.String("service", "rid")), params, &repo{}, actions.Registry) + + return raftstore.Init(ctx, logger.With(zap.String("service", "rid")), locality, params, &repo{}, actions.Registry) } func (r *repo) GetRepo() repos.Repository { return r } diff --git a/pkg/rid/store/store.go b/pkg/rid/store/store.go index 7f2dc6c20..d9a1235c2 100644 --- a/pkg/rid/store/store.go +++ b/pkg/rid/store/store.go @@ -18,13 +18,13 @@ import ( type Store = dssstore.Store[repos.Repository] // Init selects and initializes the rid store backend. -func Init(ctx context.Context, logger *zap.Logger, withCheckCron bool) (Store, error) { +func Init(ctx context.Context, logger *zap.Logger, withCheckCron bool, locality string) (Store, error) { storeType := params.GetStoreParameters().StoreType switch storeType { case params.SQLStoreType: return ridsqlstore.Init(ctx, logger, withCheckCron) case params.RaftStoreType: - return ridraftstore.Init(ctx, logger) + return ridraftstore.Init(ctx, logger, locality) case params.MemStoreType: return ridmemstore.Init(ctx, logger) default: diff --git a/pkg/scd/store/raftstore/store.go b/pkg/scd/store/raftstore/store.go index 799d6f1dd..918499c25 100644 --- a/pkg/scd/store/raftstore/store.go +++ b/pkg/scd/store/raftstore/store.go @@ -3,37 +3,70 @@ package raftstore import ( "context" - dsserr "github.com/interuss/dss/pkg/errors" + "github.com/interuss/dss/pkg/memstore" "github.com/interuss/dss/pkg/raftstore" "github.com/interuss/dss/pkg/raftstore/consensus" "github.com/interuss/dss/pkg/scd/actions" "github.com/interuss/dss/pkg/scd/repos" + scdmemstore "github.com/interuss/dss/pkg/scd/store/memstore" scdraftparams "github.com/interuss/dss/pkg/scd/store/raftstore/params" "github.com/interuss/stacktrace" "go.uber.org/zap" ) // repo is a full implementation of scd.repos.Repository for Raft-based storage. -type repo struct{} +type repo struct { + consensus *consensus.Consensus + memStore *memstore.Store[repos.Repository] + memRepo repos.Repository +} -func Init(ctx context.Context, logger *zap.Logger) (*raftstore.Store[repos.Repository], error) { +func Init(ctx context.Context, logger *zap.Logger, locality string) (*raftstore.Store[repos.Repository], error) { params, err := scdraftparams.GetConnectParameters() if err != nil { return nil, stacktrace.Propagate(err, "failed to get scd raft parameters") } - return raftstore.Init(ctx, logger.With(zap.String("service", "scd")), params, &repo{}, actions.Registry) + + memStore, err := scdmemstore.Init(ctx, logger) + if err != nil { + return nil, stacktrace.Propagate(err, "failed to initialize scd memstore") + } + + r := &repo{memStore: memStore, memRepo: memStore.GetRepo()} + store, err := raftstore.Init(ctx, logger.With(zap.String("service", "scd")), locality, params, r, actions.Registry) + if err != nil { + return nil, stacktrace.Propagate(err, "failed to initialize scd raftstore") + } + + r.consensus = store.Consensus + + return store, nil } func (r *repo) GetRepo() repos.Repository { return r } func (r *repo) GetSnapshot() ([]byte, error) { - return nil, stacktrace.NewErrorWithCode(dsserr.NotImplemented, "not implemented yet") + return r.memStore.GetSnapshot() } -func (r *repo) RestoreFromSnapshot([]byte) error { - return stacktrace.NewErrorWithCode(dsserr.NotImplemented, "not implemented yet") +func (r *repo) RestoreFromSnapshot(data []byte) error { + return r.memStore.RestoreFromSnapshot(data) } -func (r *repo) Apply(_ context.Context, _ consensus.Proposal) (any, error) { - return nil, stacktrace.NewErrorWithCode(dsserr.NotImplemented, "not implemented yet") +func (r *repo) Apply(ctx context.Context, proposal consensus.Proposal) (any, error) { + switch proposal.RequestType { + + default: + handler, ok := actions.Registry[string(proposal.RequestType)] + if !ok { + return nil, stacktrace.NewError("unrecognized request type: %s", proposal.RequestType) + } + + request, err := handler.Decode(proposal.Value) + if err != nil { + return nil, stacktrace.Propagate(err, "failed to decode %s payload", proposal.RequestType) + } + + return handler.Execute(ctx, r.memRepo, request) + } } diff --git a/pkg/scd/store/store.go b/pkg/scd/store/store.go index 0cdc48b93..6a6b70665 100644 --- a/pkg/scd/store/store.go +++ b/pkg/scd/store/store.go @@ -18,13 +18,13 @@ import ( type Store = dssstore.Store[repos.Repository] // Init selects and initializes the scd store backend. -func Init(ctx context.Context, logger *zap.Logger, withCheckCron bool) (Store, error) { +func Init(ctx context.Context, logger *zap.Logger, withCheckCron bool, locality string) (Store, error) { storeType := params.GetStoreParameters().StoreType switch storeType { case params.SQLStoreType: return scdsqlstore.Init(ctx, logger, withCheckCron) case params.RaftStoreType: - return scdraftstore.Init(ctx, logger) + return scdraftstore.Init(ctx, logger, locality) case params.MemStoreType: return scdmemstore.Init(ctx, logger) default: