[raft/aux] Add aux Raftstore

Author: MariemBaccari

pkg/aux_/store/raftstore/dss.go

@@ -2,25 +2,50 @@ package raftstore
 
 import (
 	"context"
+	"strconv"
 
 	auxmodels "github.com/interuss/dss/pkg/aux_/models"
-	dsserr "github.com/interuss/dss/pkg/errors"
+	"github.com/interuss/dss/pkg/raftstore/consensus"
+	raftparams "github.com/interuss/dss/pkg/raftstore/params"
 	"github.com/interuss/stacktrace"
 )
 
-// SaveOwnMetadata returns nil instead of dsserr.NotImplemented because it is needed to allow the server to startup.
-func (r *repo) SaveOwnMetadata(_ context.Context, locality string, publicEndpoint string) error {
-	return nil
+func (r *repo) SaveOwnMetadata(ctx context.Context, locality string, publicEndpoint string) error {
+	p, err := consensus.NewProposal(ctx, storeID, saveOwnMetadata, saveOwnMetadataPayload{
+		Locality:       locality,
+		PublicEndpoint: publicEndpoint,
+	}, false, nil)
+	if err != nil {
+		return stacktrace.Propagate(err, "failed to build %s proposal", saveOwnMetadata)
+	}
+	_, err = r.consensus.ProposeValue(ctx, p)
+	return stacktrace.Propagate(err, "failed to propose %s", saveOwnMetadata)
 }
 
-func (r *repo) GetDSSMetadata(_ context.Context) ([]*auxmodels.DSSMetadata, error) {
-	return nil, stacktrace.NewErrorWithCode(dsserr.NotImplemented, "GetDSSMetadata not implemented for raftstore")
+func (r *repo) GetDSSMetadata(ctx context.Context) ([]*auxmodels.DSSMetadata, error) {
+	p, err := consensus.NewProposal(ctx, storeID, getDSSMetadata, nil, true, nil)
+	if err != nil {
+		return nil, stacktrace.Propagate(err, "failed to build %s proposal", getDSSMetadata)
+	}
+	result, err := r.consensus.ProposeValue(ctx, p)
+	if err != nil {
+		return nil, stacktrace.Propagate(err, "failed to propose %s", getDSSMetadata)
+	}
+	if result == nil {
+		return nil, nil
+	}
+	return result.([]*auxmodels.DSSMetadata), nil
 }
 
-func (r *repo) RecordHeartbeat(_ context.Context, heartbeat auxmodels.Heartbeat) error {
-	return stacktrace.NewErrorWithCode(dsserr.NotImplemented, "RecordHeartbeat not implemented for raftstore")
+func (r *repo) RecordHeartbeat(ctx context.Context, heartbeat auxmodels.Heartbeat) error {
+	p, err := consensus.NewProposal(ctx, storeID, recordHeartbeat, heartbeat, false, nil)
+	if err != nil {
+		return stacktrace.Propagate(err, "failed to build %s proposal", recordHeartbeat)
+	}
+	_, err = r.consensus.ProposeValue(ctx, p)
+	return stacktrace.Propagate(err, "failed to propose %s", recordHeartbeat)
 }
 
 func (r *repo) GetDSSAirspaceRepresentationID(_ context.Context) (string, error) {
-	return "", stacktrace.NewErrorWithCode(dsserr.NotImplemented, "GetDSSAirspaceRepresentationID not implemented for raftstore")
+	return strconv.Itoa(int(raftparams.GetClusterID())), nil
 }

pkg/aux_/store/raftstore/store.go

@@ -2,26 +2,63 @@ package raftstore
 
 import (
 	"context"
+	"encoding/json"
 
+	auxmodels "github.com/interuss/dss/pkg/aux_/models"
 	"github.com/interuss/dss/pkg/aux_/repos"
+	auxmemstore "github.com/interuss/dss/pkg/aux_/store/memstore"
 	dsserr "github.com/interuss/dss/pkg/errors"
 	"github.com/interuss/dss/pkg/raftstore"
 	"github.com/interuss/dss/pkg/raftstore/consensus"
 	"github.com/interuss/stacktrace"
 	"go.uber.org/zap"
 )
 
+const storeID = "aux_"
+
+const (
+	saveOwnMetadata raftstore.RequestType = "saveOwnMetadata"
+	getDSSMetadata  raftstore.RequestType = "getDSSMetadata"
+	recordHeartbeat raftstore.RequestType = "recordHeartbeat"
+)
+
+type saveOwnMetadataPayload struct {
+	Locality       string `json:"locality"`
+	PublicEndpoint string `json:"public_endpoint"`
+}
+
 // repo is a full implementation of aux_.repos.Repository for Raft-based storage.
-type repo struct{}
+type repo struct {
+	consensus *consensus.Consensus
+	mem       repos.Repository
+}
 
 func Init(ctx context.Context, logger *zap.Logger) (*raftstore.Store[repos.Repository], error) {
-	store, _, err := raftstore.Init(ctx, logger, "aux_", &repo{})
-	return store, err
+	memStore, err := auxmemstore.Init(ctx, logger)
+	if err != nil {
+		return nil, stacktrace.Propagate(err, "failed to initialize aux memstore")
+	}
+
+	mem, err := memStore.Interact(ctx)
+	if err != nil {
+		return nil, stacktrace.Propagate(err, "failed to obtain aux memstore repository")
+	}
+
+	r := &repo{mem: mem}
+	store, consensus, err := raftstore.Init(ctx, logger, storeID, r)
+	if err != nil {
+		return nil, stacktrace.Propagate(err, "failed to initialize aux raftstore")
+	}
+
+	r.consensus = consensus
+	return store, nil
 }
 
 func (r *repo) GetRepo() repos.Repository { return r }
 
-func (r *repo) IsReadOnly(_ raftstore.RequestType) bool { return false }
+func (r *repo) IsReadOnly(requestType raftstore.RequestType) bool {
+	return requestType == getDSSMetadata
+}
 
 func (r *repo) GetSnapshot() ([]byte, error) {
 	return nil, stacktrace.NewErrorWithCode(dsserr.NotImplemented, "not implemented yet")
@@ -31,6 +68,28 @@ func (r *repo) RestoreFromSnapshot([]byte) error {
 	return stacktrace.NewErrorWithCode(dsserr.NotImplemented, "not implemented yet")
 }
 
-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 raftstore.RequestType(proposal.RequestType) {
+	case saveOwnMetadata:
+		var p saveOwnMetadataPayload
+		if err := json.Unmarshal(proposal.Value, &p); err != nil {
+			return nil, stacktrace.Propagate(err, "failed to unmarshal %s payload", saveOwnMetadata)
+		}
+
+		return nil, r.mem.SaveOwnMetadata(ctx, p.Locality, p.PublicEndpoint)
+
+	case getDSSMetadata:
+		return r.mem.GetDSSMetadata(ctx)
+
+	case recordHeartbeat:
+		var hb auxmodels.Heartbeat
+		if err := json.Unmarshal(proposal.Value, &hb); err != nil {
+			return nil, stacktrace.Propagate(err, "failed to unmarshal %s payload", recordHeartbeat)
+		}
+
+		return nil, r.mem.RecordHeartbeat(ctx, hb)
+
+	default:
+		return nil, stacktrace.NewError("unknown request type: %q", proposal.RequestType)
+	}
 }

pkg/raftstore/params/params.go

@@ -145,3 +145,7 @@ func init() {
 func GetConnectParameters() ConnectParameters {
 	return connectParameters
 }
+
+func GetClusterID() uint64 {
+	return connectParameters.ClusterID
+}