mirror of
https://github.com/edgelesssys/constellation.git
synced 2025-09-18 20:14:48 -04:00
Simplify joinproto
This commit is contained in:
katexochen
Paul Meyer
parent
dc9e8e75df
commit
15adba9235
14 changed files with 606 additions and 1180 deletions
|
|
@ -16,7 +16,7 @@ import (
|
|||
"github.com/edgelesssys/constellation/internal/cloud/metadata"
|
||||
"github.com/edgelesssys/constellation/internal/constants"
|
||||
"github.com/edgelesssys/constellation/internal/file"
|
||||
activationproto "github.com/edgelesssys/constellation/joinservice/joinproto"
|
||||
"github.com/edgelesssys/constellation/joinservice/joinproto"
|
||||
"github.com/spf13/afero"
|
||||
"go.uber.org/zap"
|
||||
"google.golang.org/grpc"
|
||||
|
|
@ -167,10 +167,13 @@ func (c *JoinClient) tryJoinAtAvailableServices() error {
|
|||
}
|
||||
|
||||
for _, ip := range ips {
|
||||
err = c.join(net.JoinHostPort(ip, strconv.Itoa(constants.ActivationServiceNodePort)))
|
||||
err = c.join(net.JoinHostPort(ip, strconv.Itoa(constants.JoinServiceNodePort)))
|
||||
if err == nil {
|
||||
return nil
|
||||
}
|
||||
if isUnrecoverable(err) {
|
||||
return err
|
||||
}
|
||||
}
|
||||
|
||||
return err
|
||||
|
|
@ -182,78 +185,27 @@ func (c *JoinClient) join(serviceEndpoint string) error {
|
|||
|
||||
conn, err := c.dialer.Dial(ctx, serviceEndpoint)
|
||||
if err != nil {
|
||||
c.log.Info("join service unreachable", zap.String("endpoint", serviceEndpoint), zap.Error(err))
|
||||
return fmt.Errorf("dialing join service endpoint: %v", err)
|
||||
c.log.Info("Join service unreachable", zap.String("endpoint", serviceEndpoint), zap.Error(err))
|
||||
return fmt.Errorf("dialing join service endpoint: %w", err)
|
||||
}
|
||||
defer conn.Close()
|
||||
|
||||
protoClient := activationproto.NewAPIClient(conn)
|
||||
|
||||
switch c.role {
|
||||
case role.Worker:
|
||||
return c.joinAsWorkerNode(ctx, protoClient)
|
||||
case role.ControlPlane:
|
||||
return c.joinAsControlPlaneNode(ctx, protoClient)
|
||||
default:
|
||||
return fmt.Errorf("cannot activate as %s", role.Unknown)
|
||||
protoClient := joinproto.NewAPIClient(conn)
|
||||
req := &joinproto.IssueJoinTicketRequest{
|
||||
DiskUuid: c.diskUUID,
|
||||
NodeName: c.nodeName,
|
||||
IsControlPlane: c.role == role.ControlPlane,
|
||||
}
|
||||
}
|
||||
|
||||
func (c *JoinClient) joinAsWorkerNode(ctx context.Context, client activationproto.APIClient) error {
|
||||
req := &activationproto.ActivateWorkerNodeRequest{
|
||||
DiskUuid: c.diskUUID,
|
||||
NodeName: c.nodeName,
|
||||
}
|
||||
resp, err := client.ActivateWorkerNode(ctx, req)
|
||||
ticket, err := protoClient.IssueJoinTicket(ctx, req)
|
||||
if err != nil {
|
||||
c.log.Info("Failed to activate as worker node", zap.Error(err))
|
||||
return fmt.Errorf("activating worker node: %w", err)
|
||||
c.log.Info("Issuing join ticket failed", zap.String("endpoint", serviceEndpoint), zap.Error(err))
|
||||
return fmt.Errorf("issuing join ticket: %w", err)
|
||||
}
|
||||
c.log.Info("Activation at AaaS succeeded")
|
||||
|
||||
return c.startNodeAndJoin(
|
||||
ctx,
|
||||
resp.StateDiskKey,
|
||||
resp.OwnerId,
|
||||
resp.ClusterId,
|
||||
resp.KubeletKey,
|
||||
resp.KubeletCert,
|
||||
resp.ApiServerEndpoint,
|
||||
resp.Token,
|
||||
resp.DiscoveryTokenCaCertHash,
|
||||
"",
|
||||
)
|
||||
return c.startNodeAndJoin(ctx, ticket)
|
||||
}
|
||||
|
||||
func (c *JoinClient) joinAsControlPlaneNode(ctx context.Context, client activationproto.APIClient) error {
|
||||
req := &activationproto.ActivateControlPlaneNodeRequest{
|
||||
DiskUuid: c.diskUUID,
|
||||
NodeName: c.nodeName,
|
||||
}
|
||||
resp, err := client.ActivateControlPlaneNode(ctx, req)
|
||||
if err != nil {
|
||||
c.log.Info("Failed to activate as control plane node", zap.Error(err))
|
||||
return fmt.Errorf("activating control plane node: %w", err)
|
||||
}
|
||||
c.log.Info("Activation at AaaS succeeded")
|
||||
|
||||
return c.startNodeAndJoin(
|
||||
ctx,
|
||||
resp.StateDiskKey,
|
||||
resp.OwnerId,
|
||||
resp.ClusterId,
|
||||
resp.KubeletKey,
|
||||
resp.KubeletCert,
|
||||
resp.ApiServerEndpoint,
|
||||
resp.Token,
|
||||
resp.DiscoveryTokenCaCertHash,
|
||||
resp.CertificateKey,
|
||||
)
|
||||
}
|
||||
|
||||
func (c *JoinClient) startNodeAndJoin(ctx context.Context, diskKey, ownerID, clusterID, kubeletKey, kubeletCert []byte, endpoint, token,
|
||||
discoveryCACertHash, certKey string,
|
||||
) (retErr error) {
|
||||
func (c *JoinClient) startNodeAndJoin(ctx context.Context, ticket *joinproto.IssueJoinTicketResponse) (retErr error) {
|
||||
// If an error occurs in this func, the client cannot continue.
|
||||
defer func() {
|
||||
if retErr != nil {
|
||||
|
|
@ -268,25 +220,25 @@ func (c *JoinClient) startNodeAndJoin(ctx context.Context, diskKey, ownerID, clu
|
|||
return errors.New("node is already being initialized")
|
||||
}
|
||||
|
||||
if err := c.updateDiskPassphrase(string(diskKey)); err != nil {
|
||||
if err := c.updateDiskPassphrase(string(ticket.StateDiskKey)); err != nil {
|
||||
return fmt.Errorf("updating disk passphrase: %w", err)
|
||||
}
|
||||
|
||||
state := nodestate.NodeState{
|
||||
Role: c.role,
|
||||
OwnerID: ownerID,
|
||||
ClusterID: clusterID,
|
||||
OwnerID: ticket.OwnerId,
|
||||
ClusterID: ticket.ClusterId,
|
||||
}
|
||||
if err := state.ToFile(c.fileHandler); err != nil {
|
||||
return fmt.Errorf("persisting node state: %w", err)
|
||||
}
|
||||
|
||||
btd := &kubeadm.BootstrapTokenDiscovery{
|
||||
APIServerEndpoint: endpoint,
|
||||
Token: token,
|
||||
CACertHashes: []string{discoveryCACertHash},
|
||||
APIServerEndpoint: ticket.ApiServerEndpoint,
|
||||
Token: ticket.ApiServerEndpoint,
|
||||
CACertHashes: []string{ticket.DiscoveryTokenCaCertHash},
|
||||
}
|
||||
if err := c.joiner.JoinCluster(ctx, btd, certKey, c.role); err != nil {
|
||||
if err := c.joiner.JoinCluster(ctx, btd, ticket.CertificateKey, c.role); err != nil {
|
||||
return fmt.Errorf("joining Kubernetes cluster: %w", err)
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -17,6 +17,7 @@ import (
|
|||
"github.com/edgelesssys/constellation/internal/grpc/atlscredentials"
|
||||
"github.com/edgelesssys/constellation/internal/grpc/dialer"
|
||||
"github.com/edgelesssys/constellation/internal/grpc/testdialer"
|
||||
"github.com/edgelesssys/constellation/joinservice/joinproto"
|
||||
activationproto "github.com/edgelesssys/constellation/joinservice/joinproto"
|
||||
"github.com/spf13/afero"
|
||||
"github.com/stretchr/testify/assert"
|
||||
|
|
@ -56,7 +57,7 @@ func TestClient(t *testing.T) {
|
|||
selfAnswer{err: someErr},
|
||||
selfAnswer{instance: self},
|
||||
listAnswer{instances: peers},
|
||||
activateWorkerNodeAnswer{},
|
||||
issueJoinTicketAnswer{},
|
||||
},
|
||||
clusterJoiner: &stubClusterJoiner{},
|
||||
nodeLock: nodelock.New(),
|
||||
|
|
@ -70,7 +71,7 @@ func TestClient(t *testing.T) {
|
|||
selfAnswer{instance: metadata.InstanceMetadata{Name: "node-1"}},
|
||||
selfAnswer{instance: self},
|
||||
listAnswer{instances: peers},
|
||||
activateWorkerNodeAnswer{},
|
||||
issueJoinTicketAnswer{},
|
||||
},
|
||||
clusterJoiner: &stubClusterJoiner{},
|
||||
nodeLock: nodelock.New(),
|
||||
|
|
@ -84,7 +85,7 @@ func TestClient(t *testing.T) {
|
|||
listAnswer{err: someErr},
|
||||
listAnswer{err: someErr},
|
||||
listAnswer{instances: peers},
|
||||
activateWorkerNodeAnswer{},
|
||||
issueJoinTicketAnswer{},
|
||||
},
|
||||
clusterJoiner: &stubClusterJoiner{},
|
||||
nodeLock: nodelock.New(),
|
||||
|
|
@ -98,7 +99,7 @@ func TestClient(t *testing.T) {
|
|||
listAnswer{},
|
||||
listAnswer{},
|
||||
listAnswer{instances: peers},
|
||||
activateWorkerNodeAnswer{},
|
||||
issueJoinTicketAnswer{},
|
||||
},
|
||||
clusterJoiner: &stubClusterJoiner{},
|
||||
nodeLock: nodelock.New(),
|
||||
|
|
@ -109,11 +110,11 @@ func TestClient(t *testing.T) {
|
|||
apiAnswers: []any{
|
||||
selfAnswer{instance: self},
|
||||
listAnswer{instances: peers},
|
||||
activateWorkerNodeAnswer{err: someErr},
|
||||
issueJoinTicketAnswer{err: someErr},
|
||||
listAnswer{instances: peers},
|
||||
activateWorkerNodeAnswer{err: someErr},
|
||||
issueJoinTicketAnswer{err: someErr},
|
||||
listAnswer{instances: peers},
|
||||
activateWorkerNodeAnswer{},
|
||||
issueJoinTicketAnswer{},
|
||||
},
|
||||
clusterJoiner: &stubClusterJoiner{},
|
||||
nodeLock: nodelock.New(),
|
||||
|
|
@ -146,13 +147,13 @@ func TestClient(t *testing.T) {
|
|||
}
|
||||
|
||||
serverCreds := atlscredentials.New(nil, nil)
|
||||
activationServer := grpc.NewServer(grpc.Creds(serverCreds))
|
||||
activationAPI := newStubActivationServiceAPI()
|
||||
activationproto.RegisterAPIServer(activationServer, activationAPI)
|
||||
port := strconv.Itoa(constants.ActivationServiceNodePort)
|
||||
joinServer := grpc.NewServer(grpc.Creds(serverCreds))
|
||||
joinserviceAPI := newStubJoinServiceAPI()
|
||||
joinproto.RegisterAPIServer(joinServer, joinserviceAPI)
|
||||
port := strconv.Itoa(constants.JoinServiceNodePort)
|
||||
listener := netDialer.GetListener(net.JoinHostPort("192.0.2.3", port))
|
||||
go activationServer.Serve(listener)
|
||||
defer activationServer.GracefulStop()
|
||||
go joinServer.Serve(listener)
|
||||
defer joinServer.GracefulStop()
|
||||
|
||||
client.Start()
|
||||
|
||||
|
|
@ -162,8 +163,8 @@ func TestClient(t *testing.T) {
|
|||
metadataAPI.selfAnswerC <- a
|
||||
case listAnswer:
|
||||
metadataAPI.listAnswerC <- a
|
||||
case activateWorkerNodeAnswer:
|
||||
activationAPI.activateWorkerNodeAnswerC <- a
|
||||
case issueJoinTicketAnswer:
|
||||
joinserviceAPI.issueJoinTicketAnswerC <- a
|
||||
}
|
||||
clock.Step(time.Second)
|
||||
}
|
||||
|
|
@ -267,44 +268,29 @@ type listAnswer struct {
|
|||
err error
|
||||
}
|
||||
|
||||
type stubActivationServiceAPI struct {
|
||||
activateWorkerNodeAnswerC chan activateWorkerNodeAnswer
|
||||
activateControlPlaneNodeAnswerC chan activateControlPlaneNodeAnswer
|
||||
type stubJoinServiceAPI struct {
|
||||
issueJoinTicketAnswerC chan issueJoinTicketAnswer
|
||||
|
||||
activationproto.UnimplementedAPIServer
|
||||
joinproto.UnimplementedAPIServer
|
||||
}
|
||||
|
||||
func newStubActivationServiceAPI() *stubActivationServiceAPI {
|
||||
return &stubActivationServiceAPI{
|
||||
activateWorkerNodeAnswerC: make(chan activateWorkerNodeAnswer),
|
||||
func newStubJoinServiceAPI() *stubJoinServiceAPI {
|
||||
return &stubJoinServiceAPI{
|
||||
issueJoinTicketAnswerC: make(chan issueJoinTicketAnswer),
|
||||
}
|
||||
}
|
||||
|
||||
func (s *stubActivationServiceAPI) ActivateWorkerNode(_ context.Context, _ *activationproto.ActivateWorkerNodeRequest,
|
||||
) (*activationproto.ActivateWorkerNodeResponse, error) {
|
||||
answer := <-s.activateWorkerNodeAnswerC
|
||||
func (s *stubJoinServiceAPI) IssueJoinTicket(_ context.Context, _ *joinproto.IssueJoinTicketRequest,
|
||||
) (*joinproto.IssueJoinTicketResponse, error) {
|
||||
answer := <-s.issueJoinTicketAnswerC
|
||||
if answer.resp == nil {
|
||||
answer.resp = &activationproto.ActivateWorkerNodeResponse{}
|
||||
answer.resp = &activationproto.IssueJoinTicketResponse{}
|
||||
}
|
||||
return answer.resp, answer.err
|
||||
}
|
||||
|
||||
func (s *stubActivationServiceAPI) ActivateControlPlaneNode(_ context.Context, _ *activationproto.ActivateControlPlaneNodeRequest,
|
||||
) (*activationproto.ActivateControlPlaneNodeResponse, error) {
|
||||
answer := <-s.activateControlPlaneNodeAnswerC
|
||||
if answer.resp == nil {
|
||||
answer.resp = &activationproto.ActivateControlPlaneNodeResponse{}
|
||||
}
|
||||
return answer.resp, answer.err
|
||||
}
|
||||
|
||||
type activateWorkerNodeAnswer struct {
|
||||
resp *activationproto.ActivateWorkerNodeResponse
|
||||
err error
|
||||
}
|
||||
|
||||
type activateControlPlaneNodeAnswer struct {
|
||||
resp *activationproto.ActivateControlPlaneNodeResponse
|
||||
type issueJoinTicketAnswer struct {
|
||||
resp *joinproto.IssueJoinTicketResponse
|
||||
err error
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -214,7 +214,7 @@ func NewActivationDaemonset(csp, measurementsJSON, idJSON string) *activationDae
|
|||
Protocol: k8s.ProtocolTCP,
|
||||
Port: constants.ActivationServicePort,
|
||||
TargetPort: intstr.IntOrString{IntVal: constants.ActivationServicePort},
|
||||
NodePort: constants.ActivationServiceNodePort,
|
||||
NodePort: constants.JoinServiceNodePort,
|
||||
},
|
||||
},
|
||||
Selector: map[string]string{
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue