Skip to content

Commit 06b832e

Browse files
MariemBaccariMariemBaccari
committed
[raft] Serve the admin API for the membership of Raft groups
1 parent 719b757 commit 06b832e

5 files changed

Lines changed: 283 additions & 1 deletion

File tree

‎cmds/core-service/main.go‎

Lines changed: 25 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -13,6 +13,7 @@ import (
1313
"time"
1414

1515
"cloud.google.com/go/profiler"
16+
adminserver "github.com/interuss/dss/pkg/admin/server"
1617
"github.com/interuss/dss/pkg/api"
1718
apiauxv1 "github.com/interuss/dss/pkg/api/auxv1"
1819
apiridv1 "github.com/interuss/dss/pkg/api/ridv1"
@@ -21,15 +22,19 @@ import (
2122
apiversioningv1 "github.com/interuss/dss/pkg/api/versioningv1"
2223
"github.com/interuss/dss/pkg/auth"
2324
aux "github.com/interuss/dss/pkg/aux_"
25+
auxrepos "github.com/interuss/dss/pkg/aux_/repos"
2426
auxs "github.com/interuss/dss/pkg/aux_/store"
2527
"github.com/interuss/dss/pkg/build"
2628
dsserr "github.com/interuss/dss/pkg/errors"
2729
"github.com/interuss/dss/pkg/logging"
2830
"github.com/interuss/dss/pkg/raftstore"
31+
"github.com/interuss/dss/pkg/raftstore/consensus"
32+
ridrepos "github.com/interuss/dss/pkg/rid/repos"
2933
rid_v1 "github.com/interuss/dss/pkg/rid/server/v1"
3034
rid_v2 "github.com/interuss/dss/pkg/rid/server/v2"
3135
rids "github.com/interuss/dss/pkg/rid/store"
3236
"github.com/interuss/dss/pkg/scd"
37+
scdrepos "github.com/interuss/dss/pkg/scd/repos"
3338
scds "github.com/interuss/dss/pkg/scd/store"
3439
"github.com/interuss/dss/pkg/store/params"
3540
"github.com/interuss/dss/pkg/timestamp"
@@ -60,6 +65,7 @@ var (
6065
enableMetrics = flag.Bool("enable_metrics", false, "Enable metric endpoint")
6166
enableTracing = flag.Bool("enable_tracing", false, "Enable tracing")
6267
metricsListeningAddress = flag.String("metrics_addr", ":8079", "Address and port that the for the prometheus-compatible metric service binds to and listens on for incoming connections")
68+
adminSocket = flag.String("admin_socket", "/run/dss/admin.sock", "Path of the Unix socket on which the admin API (used by dssctl) is served, when the core-service uses the raft store. Its directory is created if needed.")
6369

6470
pkFile = flag.String("public_key_files", "", "Path to public Keys to use for JWT decoding, separated by commas.")
6571
jwksEndpoint = flag.String("jwks_endpoint", "", "URL pointing to an endpoint serving JWKS")
@@ -360,6 +366,25 @@ func RunHTTPServer(ctx context.Context, ctxCanceler func(), address, locality st
360366
multiRouter.Routers = append(multiRouter.Routers, &scdV1Router)
361367
}
362368

369+
if params.GetStoreParameters().StoreType == params.RaftStoreType {
370+
raftGroups := map[string]*consensus.Consensus{
371+
"aux": auxV1Server.Store.(*raftstore.Store[auxrepos.Repository]).Consensus,
372+
"rid": ridV1Server.Store.(*raftstore.Store[ridrepos.Repository]).Consensus,
373+
}
374+
if scdV1Server != nil {
375+
raftGroups["scd"] = scdV1Server.Store.(*raftstore.Store[scdrepos.Repository]).Consensus
376+
}
377+
adminServer, err := adminserver.Start(ctx, logger.With(zap.String("component", "admin"), zap.String("socket", *adminSocket)), *adminSocket, *timeout, raftGroups)
378+
if err != nil {
379+
return stacktrace.Propagate(err, "Failed to start the admin server")
380+
}
381+
defer func() {
382+
if err := adminServer.Shutdown(context.Background()); err != nil {
383+
logger.Warn("failed to shut down admin server", zap.Error(err))
384+
}
385+
}()
386+
}
387+
363388
// the middlewares are wrapped and, therefore, executed in the opposite order
364389
handler := healthyEndpointMiddleware(logger, &multiRouter)
365390
handler = authorizer.TokenMiddleware(handler)

‎pkg/admin/server/raft_members_handler.go‎

Lines changed: 185 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,185 @@
1+
package server
2+
3+
import (
4+
"context"
5+
"maps"
6+
"slices"
7+
"strconv"
8+
9+
"github.com/interuss/dss/pkg/api"
10+
restapi "github.com/interuss/dss/pkg/api/adminv1"
11+
dsserr "github.com/interuss/dss/pkg/errors"
12+
"github.com/interuss/dss/pkg/raftstore/consensus"
13+
"github.com/interuss/stacktrace"
14+
"go.uber.org/zap"
15+
)
16+
17+
func (s *Server) GetThisRaftNode(ctx context.Context, req *restapi.GetThisRaftNodeRequest) restapi.GetThisRaftNodeResponseSet {
18+
group, err := s.raftGroup(req.GroupName)
19+
if err != nil {
20+
return restapi.GetThisRaftNodeResponseSet{Response404: &restapi.ErrorResponse{
21+
Message: dsserr.Handle(ctx, err)}}
22+
}
23+
24+
node, err := group.Node()
25+
if err != nil {
26+
return restapi.GetThisRaftNodeResponseSet{Response500: &api.InternalServerErrorBody{
27+
ErrorMessage: *dsserr.Handle(ctx, stacktrace.Propagate(err, "Could not describe this node in the %s raft group", req.GroupName))}}
28+
}
29+
return restapi.GetThisRaftNodeResponseSet{Response200: toRaftNode(node)}
30+
}
31+
32+
func (s *Server) CreateRaftGroup(ctx context.Context, req *restapi.CreateRaftGroupRequest) restapi.CreateRaftGroupResponseSet {
33+
if req.BodyParseError != nil {
34+
return restapi.CreateRaftGroupResponseSet{Response400: &restapi.ErrorResponse{
35+
Message: dsserr.Handle(ctx, stacktrace.PropagateWithCode(req.BodyParseError, dsserr.BadRequest, "Malformed params"))}}
36+
}
37+
group, err := s.raftGroup(req.Body.GroupName)
38+
if err != nil {
39+
return restapi.CreateRaftGroupResponseSet{Response404: &restapi.ErrorResponse{
40+
Message: dsserr.Handle(ctx, err)}}
41+
}
42+
43+
membership, err := group.Bootstrap()
44+
if err != nil {
45+
err = stacktrace.Propagate(err, "Could not create the %s raft group", req.Body.GroupName)
46+
switch stacktrace.GetCode(err) {
47+
case dsserr.AlreadyExists:
48+
return restapi.CreateRaftGroupResponseSet{Response409: &restapi.ErrorResponse{Message: dsserr.Handle(ctx, err)}}
49+
default:
50+
return restapi.CreateRaftGroupResponseSet{Response500: &api.InternalServerErrorBody{
51+
ErrorMessage: *dsserr.Handle(ctx, stacktrace.Propagate(err, "Got an unexpected error"))}}
52+
}
53+
}
54+
55+
s.Logger.Info("created raft group", zap.String("group", string(req.Body.GroupName)))
56+
return restapi.CreateRaftGroupResponseSet{Response200: toRaftGroup(req.Body.GroupName, membership)}
57+
}
58+
59+
func (s *Server) GetRaftGroup(ctx context.Context, req *restapi.GetRaftGroupRequest) restapi.GetRaftGroupResponseSet {
60+
group, err := s.raftGroup(req.GroupName)
61+
if err != nil {
62+
return restapi.GetRaftGroupResponseSet{Response404: &restapi.ErrorResponse{
63+
Message: dsserr.Handle(ctx, err)}}
64+
}
65+
66+
membership, err := group.Members()
67+
if err != nil {
68+
err = stacktrace.Propagate(err, "Could not get the members of the %s raft group", req.GroupName)
69+
switch stacktrace.GetCode(err) {
70+
// NotFound: this node is still waiting to create or join the group.
71+
case dsserr.NotFound:
72+
return restapi.GetRaftGroupResponseSet{Response404: &restapi.ErrorResponse{Message: dsserr.Handle(ctx, err)}}
73+
default:
74+
return restapi.GetRaftGroupResponseSet{Response500: &api.InternalServerErrorBody{
75+
ErrorMessage: *dsserr.Handle(ctx, stacktrace.Propagate(err, "Got an unexpected error"))}}
76+
}
77+
}
78+
return restapi.GetRaftGroupResponseSet{Response200: toRaftGroup(req.GroupName, membership)}
79+
}
80+
81+
func (s *Server) CreateRaftGroupMember(ctx context.Context, req *restapi.CreateRaftGroupMemberRequest) restapi.CreateRaftGroupMemberResponseSet {
82+
group, err := s.raftGroup(req.GroupName)
83+
if err != nil {
84+
return restapi.CreateRaftGroupMemberResponseSet{Response404: &restapi.ErrorResponse{
85+
Message: dsserr.Handle(ctx, err)}}
86+
}
87+
if req.BodyParseError != nil {
88+
return restapi.CreateRaftGroupMemberResponseSet{Response400: &restapi.ErrorResponse{
89+
Message: dsserr.Handle(ctx, stacktrace.PropagateWithCode(req.BodyParseError, dsserr.BadRequest, "Malformed params"))}}
90+
}
91+
92+
added, err := group.AddNode(ctx, req.Body.Url)
93+
if err != nil {
94+
err = stacktrace.Propagate(err, "Could not add node %s to the %s raft group", req.Body.Url, req.GroupName)
95+
errResp := &restapi.ErrorResponse{Message: dsserr.Handle(ctx, err)}
96+
switch stacktrace.GetCode(err) {
97+
case dsserr.BadRequest:
98+
return restapi.CreateRaftGroupMemberResponseSet{Response400: errResp}
99+
case dsserr.AlreadyExists:
100+
return restapi.CreateRaftGroupMemberResponseSet{Response409: errResp}
101+
// Unavailable: the node cannot be reached, or this node is not a member of the group.
102+
case dsserr.Unavailable:
103+
return restapi.CreateRaftGroupMemberResponseSet{Response503: errResp}
104+
default:
105+
return restapi.CreateRaftGroupMemberResponseSet{Response500: &api.InternalServerErrorBody{
106+
ErrorMessage: *dsserr.Handle(ctx, stacktrace.Propagate(err, "Got an unexpected error"))}}
107+
}
108+
}
109+
110+
s.Logger.Info("added node to raft group", zap.String("group", string(req.GroupName)), zap.Uint64("nodeID", added.ID), zap.String("url", added.URL))
111+
return restapi.CreateRaftGroupMemberResponseSet{Response200: toRaftNode(added)}
112+
}
113+
114+
func (s *Server) DeleteRaftGroupMember(ctx context.Context, req *restapi.DeleteRaftGroupMemberRequest) restapi.DeleteRaftGroupMemberResponseSet {
115+
group, err := s.raftGroup(req.GroupName)
116+
if err != nil {
117+
return restapi.DeleteRaftGroupMemberResponseSet{Response404: &restapi.ErrorResponse{
118+
Message: dsserr.Handle(ctx, err)}}
119+
}
120+
nodeID, err := parseNodeID(req.NodeId)
121+
if err != nil {
122+
return restapi.DeleteRaftGroupMemberResponseSet{Response400: &restapi.ErrorResponse{
123+
Message: dsserr.Handle(ctx, err)}}
124+
}
125+
126+
err = group.RemoveNode(ctx, nodeID)
127+
if err != nil {
128+
err = stacktrace.Propagate(err, "Could not remove node %d from the %s raft group", nodeID, req.GroupName)
129+
errResp := &restapi.ErrorResponse{Message: dsserr.Handle(ctx, err)}
130+
switch stacktrace.GetCode(err) {
131+
case dsserr.BadRequest, dsserr.Unavailable:
132+
return restapi.DeleteRaftGroupMemberResponseSet{Response400: errResp}
133+
default:
134+
return restapi.DeleteRaftGroupMemberResponseSet{Response500: &api.InternalServerErrorBody{
135+
ErrorMessage: *dsserr.Handle(ctx, stacktrace.Propagate(err, "Got an unexpected error"))}}
136+
}
137+
}
138+
139+
s.Logger.Info("removed node from raft group", zap.String("group", string(req.GroupName)), zap.Uint64("nodeID", nodeID))
140+
return restapi.DeleteRaftGroupMemberResponseSet{Response204: &api.EmptyResponseBody{}}
141+
}
142+
143+
// raftGroup returns the Raft group or a dsserr.NotFound error.
144+
func (s *Server) raftGroup(name restapi.RaftGroupName) (*consensus.Consensus, error) {
145+
group, ok := s.raftGroups[string(name)]
146+
if !ok {
147+
return nil, stacktrace.NewErrorWithCode(dsserr.NotFound, "Unknown raft group %q, expected one of %v", name, slices.Sorted(maps.Keys(s.raftGroups)))
148+
}
149+
return group, nil
150+
}
151+
152+
// parseNodeID parses a node ID formatted as a decimal string, or returns a dsserr.BadRequest error.
153+
func parseNodeID(s string) (uint64, error) {
154+
id, err := strconv.ParseUint(s, 10, 64)
155+
if err != nil || id == 0 {
156+
return 0, stacktrace.NewErrorWithCode(dsserr.BadRequest, "Invalid node ID %q, expected a non-zero decimal integer", s)
157+
}
158+
return id, nil
159+
}
160+
161+
func toRaftNode(node consensus.Node) *restapi.RaftNode {
162+
return &restapi.RaftNode{Id: strconv.FormatUint(node.ID, 10), Url: node.URL, State: string(node.State)}
163+
}
164+
165+
// toRaftGroup returns the group with its nodes: its members first, then the nodes removed from it, each by ID.
166+
func toRaftGroup(name restapi.RaftGroupName, membership consensus.Membership) *restapi.RaftGroup {
167+
result := &restapi.RaftGroup{
168+
GroupName: name,
169+
ClusterId: strconv.FormatUint(membership.ClusterID, 10),
170+
Nodes: []restapi.RaftNode{},
171+
}
172+
// The leader is unknown while an election is in progress.
173+
if membership.LeaderID != 0 {
174+
result.LeaderId = strconv.FormatUint(membership.LeaderID, 10)
175+
}
176+
ids := slices.Sorted(maps.Keys(membership.Nodes))
177+
for _, state := range []consensus.NodeState{consensus.NodeMember, consensus.NodeRemoved} {
178+
for _, id := range ids {
179+
if node := membership.Nodes[id]; node.State == state {
180+
result.Nodes = append(result.Nodes, *toRaftNode(node))
181+
}
182+
}
183+
}
184+
return result
185+
}

‎pkg/admin/server/server.go‎

Lines changed: 13 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,13 @@
1+
package server
2+
3+
import (
4+
"github.com/interuss/dss/pkg/raftstore/consensus"
5+
"go.uber.org/zap"
6+
)
7+
8+
// Server implements adminv1.Implementation.
9+
type Server struct {
10+
Logger *zap.Logger
11+
// raftGroups are the Raft groups of the DSS node, indexed by name.
12+
raftGroups map[string]*consensus.Consensus
13+
}

‎pkg/admin/server/start.go‎

Lines changed: 59 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,59 @@
1+
package server
2+
3+
import (
4+
"context"
5+
"errors"
6+
"io/fs"
7+
"net"
8+
"net/http"
9+
"os"
10+
"path/filepath"
11+
"time"
12+
13+
"github.com/interuss/dss/pkg/api"
14+
restapi "github.com/interuss/dss/pkg/api/adminv1"
15+
"github.com/interuss/dss/pkg/raftstore/consensus"
16+
"github.com/interuss/stacktrace"
17+
"go.uber.org/zap"
18+
)
19+
20+
// Start serves the admin API on the Unix socket at socketPath, with requests failing after timeout.
21+
// The directory of the socket is created if needed. A leftover socket at socketPath is replaced.
22+
func Start(ctx context.Context, logger *zap.Logger, socketPath string, timeout time.Duration, raftGroups map[string]*consensus.Consensus) (*http.Server, error) {
23+
socketDir := filepath.Dir(socketPath)
24+
if err := os.MkdirAll(socketDir, 0o700); err != nil {
25+
return nil, stacktrace.Propagate(err, "failed to create the directory of the admin socket at %s", socketDir)
26+
}
27+
28+
if err := os.Remove(socketPath); err != nil && !errors.Is(err, fs.ErrNotExist) {
29+
return nil, stacktrace.Propagate(err, "failed to remove the previous admin socket at %s", socketPath)
30+
}
31+
var listenConfig net.ListenConfig
32+
listener, err := listenConfig.Listen(ctx, "unix", socketPath)
33+
if err != nil {
34+
return nil, stacktrace.Propagate(err, "failed to listen on the admin socket at %s", socketPath)
35+
}
36+
if err := os.Chmod(socketPath, 0o600); err != nil {
37+
_ = listener.Close()
38+
return nil, stacktrace.Propagate(err, "failed to restrict the permissions of the admin socket at %s", socketPath)
39+
}
40+
41+
router := restapi.MakeAPIRouter(&Server{Logger: logger, raftGroups: raftGroups}, nil)
42+
handler := &api.MultiRouter{Routers: []api.PartialRouter{&router}}
43+
44+
server := &http.Server{
45+
Handler: http.TimeoutHandler(handler, timeout, "request timeout"),
46+
ReadHeaderTimeout: 15 * time.Second,
47+
ReadTimeout: 15 * time.Second,
48+
}
49+
50+
go func() {
51+
logger.Info("starting admin server")
52+
// The socket file is removed when the server is shut down, which closes the listener.
53+
if err := server.Serve(listener); err != nil && !errors.Is(err, http.ErrServerClosed) {
54+
logger.Error("admin server failed", zap.Error(err))
55+
}
56+
}()
57+
58+
return server, nil
59+
}

‎pkg/raftstore/README.md‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -64,7 +64,7 @@ Most of the concepts defined below come from the Raft consensus algorithm itself
6464

6565
## Membership
6666

67-
The members of a Raft group are bootstrapped from the `--raft_peers` flag on the group's very first start. Afterwards, nodes are added to and removed from a running group one at a time with [`Consensus.AddNode` and `Consensus.RemoveNode`](consensus/consensus.go), which propose a configuration change through the Raft log and block until it is applied. Each Raft group is managed independently: a DSS node is added to or removed from each of its groups with one configuration change per group.
67+
The members of a Raft group are bootstrapped from the `--raft_peers` flag on the group's very first start. Afterwards, nodes are added to and removed from a running group one at a time, through the [admin API](../../interfaces/admin/admin.yaml) served by the core-service on the Unix socket at `--admin_socket`, with [`Consensus.AddNode` and `Consensus.RemoveNode`](consensus/consensus.go), which propose a configuration change through the Raft log and block until it is applied. Each Raft group is managed independently: a DSS node is added to or removed from each of its groups with one configuration change per group.
6868

6969
- **Joining:** Starting a node that joins a running group is not supported yet, see [#1802](https://github.com/interuss/dss/issues/1802).
7070
- **Validation:** A configuration change is validated when it is proposed, to fail fast, and again by every node when it is applied. The following changes are rejected:

0 commit comments

Comments
 (0)