From 0bdfa7bcb6b4b409313a61355deb127028140cf4 Mon Sep 17 00:00:00 2001 From: Mariem Baccari Date: Thu, 15 Oct 2026 12:32:26 +0200 Subject: [PATCH] [raft] Implement admin API client for cluster membership --- pkg/admin/client/client.go | 154 +++++++++++++++++++++++++++++++++++++ 1 file changed, 154 insertions(+) create mode 100644 pkg/admin/client/client.go diff --git a/pkg/admin/client/client.go b/pkg/admin/client/client.go new file mode 100644 index 000000000..ee912f5ca --- /dev/null +++ b/pkg/admin/client/client.go @@ -0,0 +1,154 @@ +package client + +import ( + "context" + "net" + "net/http" + "time" + + restapi "github.com/interuss/dss/pkg/api/adminv1" + dsserr "github.com/interuss/dss/pkg/errors" + "github.com/interuss/stacktrace" +) + +// statusCodes maps the HTTP status of the error responses of the admin API to the code of the error of the DSS node. +var statusCodes = map[int]stacktrace.ErrorCode{ + http.StatusBadRequest: dsserr.BadRequest, + http.StatusNotFound: dsserr.NotFound, + http.StatusConflict: dsserr.AlreadyExists, + http.StatusServiceUnavailable: dsserr.Unavailable, +} + +type Group struct { + client *restapi.Client + socketPath string + name restapi.RaftGroupName +} + +func NewRaftGroup(socketPath string, name string, timeout time.Duration) (*Group, error) { + if socketPath == "" { + return nil, stacktrace.NewError("the path of the Unix socket of the admin API is required") + } + + // The requests are sent over the Unix socket whatever their URL, whose host is only used in the Host header. + transport := &http.Transport{ + DialContext: func(ctx context.Context, _, _ string) (net.Conn, error) { + var d net.Dialer + return d.DialContext(ctx, "unix", socketPath) + }, + } + return &Group{ + client: &restapi.Client{BaseURL: "http://localhost", HTTPClient: &http.Client{Transport: transport, Timeout: timeout}}, + socketPath: socketPath, + name: restapi.RaftGroupName(name), + }, nil +} + +func (g *Group) Node(ctx context.Context) (*restapi.RaftNode, error) { + resp, err := g.client.GetThisRaftNode(ctx, g.name) + if err != nil { + return nil, stacktrace.Propagate(err, "failed to call the admin API at %s", g.socketPath) + } + switch { + case resp.Response200 != nil: + return resp.Response200, nil + case resp.Response404 != nil: + return nil, responseError(http.StatusNotFound, resp.Response404.Message) + case resp.Response500 != nil: + return nil, responseError(http.StatusInternalServerError, &resp.Response500.ErrorMessage) + default: + return nil, stacktrace.NewError("no response received from the admin API") + } +} + +func (g *Group) Bootstrap(ctx context.Context) (*restapi.RaftGroup, error) { + resp, err := g.client.CreateRaftGroup(ctx, &restapi.CreateRaftGroupParameters{GroupName: g.name}) + if err != nil { + return nil, stacktrace.Propagate(err, "failed to call the admin API at %s", g.socketPath) + } + switch { + case resp.Response200 != nil: + return resp.Response200, nil + case resp.Response400 != nil: + return nil, responseError(http.StatusBadRequest, resp.Response400.Message) + case resp.Response404 != nil: + return nil, responseError(http.StatusNotFound, resp.Response404.Message) + case resp.Response409 != nil: + return nil, responseError(http.StatusConflict, resp.Response409.Message) + case resp.Response500 != nil: + return nil, responseError(http.StatusInternalServerError, &resp.Response500.ErrorMessage) + default: + return nil, stacktrace.NewError("no response received from the admin API") + } +} + +func (g *Group) Members(ctx context.Context) (*restapi.RaftGroup, error) { + resp, err := g.client.GetRaftGroup(ctx, g.name) + if err != nil { + return nil, stacktrace.Propagate(err, "failed to call the admin API at %s", g.socketPath) + } + switch { + case resp.Response200 != nil: + return resp.Response200, nil + case resp.Response404 != nil: + return nil, responseError(http.StatusNotFound, resp.Response404.Message) + case resp.Response500 != nil: + return nil, responseError(http.StatusInternalServerError, &resp.Response500.ErrorMessage) + default: + return nil, stacktrace.NewError("no response received from the admin API") + } +} + +func (g *Group) AddNode(ctx context.Context, peerURL string) (*restapi.RaftNode, error) { + resp, err := g.client.CreateRaftGroupMember(ctx, g.name, &restapi.CreateRaftGroupMemberParameters{Url: peerURL}) + if err != nil { + return nil, stacktrace.Propagate(err, "failed to call the admin API at %s", g.socketPath) + } + switch { + case resp.Response200 != nil: + return resp.Response200, nil + case resp.Response400 != nil: + return nil, responseError(http.StatusBadRequest, resp.Response400.Message) + case resp.Response404 != nil: + return nil, responseError(http.StatusNotFound, resp.Response404.Message) + case resp.Response409 != nil: + return nil, responseError(http.StatusConflict, resp.Response409.Message) + case resp.Response503 != nil: + return nil, responseError(http.StatusServiceUnavailable, resp.Response503.Message) + case resp.Response500 != nil: + return nil, responseError(http.StatusInternalServerError, &resp.Response500.ErrorMessage) + default: + return nil, stacktrace.NewError("no response received from the admin API") + } +} + +func (g *Group) RemoveNode(ctx context.Context, nodeID string) error { + resp, err := g.client.DeleteRaftGroupMember(ctx, g.name, nodeID) + if err != nil { + return stacktrace.Propagate(err, "failed to call the admin API at %s", g.socketPath) + } + switch { + case resp.Response204 != nil: + return nil + case resp.Response400 != nil: + return responseError(http.StatusBadRequest, resp.Response400.Message) + case resp.Response404 != nil: + return responseError(http.StatusNotFound, resp.Response404.Message) + case resp.Response500 != nil: + return responseError(http.StatusInternalServerError, &resp.Response500.ErrorMessage) + default: + return stacktrace.NewError("no response received from the admin API") + } +} + +// responseError returns the error described by an error response of the admin API, with the code of statusCodes. +func responseError(status int, message *string) error { + text := http.StatusText(status) + if message != nil && *message != "" { + text += ": " + *message + } + if code, ok := statusCodes[status]; ok { + return stacktrace.NewErrorWithCode(code, "%d %s", status, text) + } + return stacktrace.NewError("%d %s", status, text) +}