mirror of
https://github.com/edgelesssys/constellation.git
synced 2025-05-02 06:16:08 -04:00
join: join over lb if available (#2348)
* join: join over lb if available
This commit is contained in:
parent
df77696620
commit
2776e40df7
12 changed files with 142 additions and 62 deletions
|
@ -8,6 +8,7 @@ go_library(
|
|||
visibility = ["//disk-mapper:__subpackages__"],
|
||||
deps = [
|
||||
"//internal/cloud/metadata",
|
||||
"//internal/constants",
|
||||
"//internal/logger",
|
||||
"//internal/role",
|
||||
"//joinservice/joinproto",
|
||||
|
|
|
@ -14,10 +14,13 @@ package rejoinclient
|
|||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"net"
|
||||
"strconv"
|
||||
"time"
|
||||
|
||||
"github.com/edgelesssys/constellation/v2/internal/cloud/metadata"
|
||||
"github.com/edgelesssys/constellation/v2/internal/constants"
|
||||
"github.com/edgelesssys/constellation/v2/internal/logger"
|
||||
"github.com/edgelesssys/constellation/v2/internal/role"
|
||||
"github.com/edgelesssys/constellation/v2/joinservice/joinproto"
|
||||
|
@ -75,7 +78,7 @@ func (c *RejoinClient) Start(ctx context.Context, diskUUID string) (diskKey, mea
|
|||
defer c.log.Infof("RejoinClient stopped")
|
||||
|
||||
for {
|
||||
endpoints, err := c.getControlPlaneEndpoints()
|
||||
endpoints, err := c.getJoinEndpoints()
|
||||
if err != nil {
|
||||
c.log.With(zap.Error(err)).Errorf("Failed to get control-plane endpoints")
|
||||
} else {
|
||||
|
@ -130,19 +133,39 @@ func (c *RejoinClient) requestRejoinTicket(endpoint string) (*joinproto.IssueRej
|
|||
return joinproto.NewAPIClient(conn).IssueRejoinTicket(ctx, &joinproto.IssueRejoinTicketRequest{DiskUuid: c.diskUUID})
|
||||
}
|
||||
|
||||
// getControlPlaneEndpoints requests the available control-plane endpoints from the metadata API.
|
||||
// getJoinEndpoints requests the available control-plane endpoints from the metadata API.
|
||||
// The list is filtered to remove *this* node if it is a restarting control-plane node.
|
||||
func (c *RejoinClient) getControlPlaneEndpoints() ([]string, error) {
|
||||
// Furthermore, the load balancer's endpoint is added.
|
||||
func (c *RejoinClient) getJoinEndpoints() ([]string, error) {
|
||||
ctx, cancel := c.timeoutCtx()
|
||||
defer cancel()
|
||||
endpoints, err := metadata.JoinServiceEndpoints(ctx, c.metadataAPI)
|
||||
|
||||
joinEndpoints := []string{}
|
||||
|
||||
lbEndpoint, _, err := c.metadataAPI.GetLoadBalancerEndpoint(ctx)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
return nil, fmt.Errorf("retrieving load balancer endpoint from cloud provider: %w", err)
|
||||
}
|
||||
joinEndpoints = append(joinEndpoints, net.JoinHostPort(lbEndpoint, strconv.Itoa(constants.JoinServiceNodePort)))
|
||||
|
||||
instances, err := c.metadataAPI.List(ctx)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("retrieving instances list from cloud provider: %w", err)
|
||||
}
|
||||
|
||||
for _, instance := range instances {
|
||||
if instance.Role == role.ControlPlane {
|
||||
if instance.VPCIP != "" {
|
||||
joinEndpoints = append(joinEndpoints, net.JoinHostPort(instance.VPCIP, strconv.Itoa(constants.JoinServiceNodePort)))
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
if c.nodeInfo.Role == role.ControlPlane {
|
||||
return removeSelfFromEndpoints(c.nodeInfo.VPCIP, endpoints), nil
|
||||
return removeSelfFromEndpoints(c.nodeInfo.VPCIP, joinEndpoints), nil
|
||||
}
|
||||
return endpoints, nil
|
||||
|
||||
return joinEndpoints, nil
|
||||
}
|
||||
|
||||
// removeSelfFromEndpoints removes *this* node from the list of endpoints.
|
||||
|
@ -169,4 +192,6 @@ type grpcDialer interface {
|
|||
type metadataAPI interface {
|
||||
// List retrieves all instances belonging to the current constellation.
|
||||
List(ctx context.Context) ([]metadata.InstanceMetadata, error)
|
||||
// GetLoadBalancerEndpoint retrieves the load balancer endpoint.
|
||||
GetLoadBalancerEndpoint(ctx context.Context) (host, port string, err error)
|
||||
}
|
||||
|
|
|
@ -123,7 +123,7 @@ func TestRemoveSelfFromEndpoints(t *testing.T) {
|
|||
}
|
||||
}
|
||||
|
||||
func TestGetControlPlaneEndpoints(t *testing.T) {
|
||||
func TestGetJoinEndpoints(t *testing.T) {
|
||||
testInstances := []metadata.InstanceMetadata{
|
||||
{
|
||||
Role: role.ControlPlane,
|
||||
|
@ -154,7 +154,7 @@ func TestGetControlPlaneEndpoints(t *testing.T) {
|
|||
testCases := map[string]struct {
|
||||
nodeInfo metadata.InstanceMetadata
|
||||
meta stubMetadataAPI
|
||||
wantInstances int
|
||||
wantEndpoints int
|
||||
wantErr bool
|
||||
}{
|
||||
"worker node": {
|
||||
|
@ -163,9 +163,10 @@ func TestGetControlPlaneEndpoints(t *testing.T) {
|
|||
VPCIP: "192.0.2.1",
|
||||
},
|
||||
meta: stubMetadataAPI{
|
||||
instances: testInstances,
|
||||
instances: testInstances,
|
||||
lbEndpoint: "192.0.2.100",
|
||||
},
|
||||
wantInstances: 3,
|
||||
wantEndpoints: 4,
|
||||
},
|
||||
"control-plane node not in list": {
|
||||
nodeInfo: metadata.InstanceMetadata{
|
||||
|
@ -173,9 +174,10 @@ func TestGetControlPlaneEndpoints(t *testing.T) {
|
|||
VPCIP: "192.0.2.1",
|
||||
},
|
||||
meta: stubMetadataAPI{
|
||||
instances: testInstances,
|
||||
instances: testInstances,
|
||||
lbEndpoint: "192.0.2.100",
|
||||
},
|
||||
wantInstances: 3,
|
||||
wantEndpoints: 4,
|
||||
},
|
||||
"control-plane node in list": {
|
||||
nodeInfo: metadata.InstanceMetadata{
|
||||
|
@ -183,17 +185,28 @@ func TestGetControlPlaneEndpoints(t *testing.T) {
|
|||
VPCIP: "192.0.2.2",
|
||||
},
|
||||
meta: stubMetadataAPI{
|
||||
instances: testInstances,
|
||||
instances: testInstances,
|
||||
lbEndpoint: "192.0.2.100",
|
||||
},
|
||||
wantInstances: 2,
|
||||
wantEndpoints: 3,
|
||||
},
|
||||
"metadata error": {
|
||||
"metadata list error": {
|
||||
nodeInfo: metadata.InstanceMetadata{
|
||||
Role: role.ControlPlane,
|
||||
VPCIP: "192.0.2.1",
|
||||
},
|
||||
meta: stubMetadataAPI{
|
||||
err: errors.New("error"),
|
||||
listErr: assert.AnError,
|
||||
},
|
||||
wantErr: true,
|
||||
},
|
||||
"metadata load balancer error": {
|
||||
nodeInfo: metadata.InstanceMetadata{
|
||||
Role: role.ControlPlane,
|
||||
VPCIP: "192.0.2.1",
|
||||
},
|
||||
meta: stubMetadataAPI{
|
||||
getLoadBalancerEndpointErr: assert.AnError,
|
||||
},
|
||||
wantErr: true,
|
||||
},
|
||||
|
@ -205,13 +218,14 @@ func TestGetControlPlaneEndpoints(t *testing.T) {
|
|||
|
||||
client := New(nil, tc.nodeInfo, tc.meta, logger.NewTest(t))
|
||||
|
||||
endpoints, err := client.getControlPlaneEndpoints()
|
||||
endpoints, err := client.getJoinEndpoints()
|
||||
if tc.wantErr {
|
||||
assert.Error(err)
|
||||
} else {
|
||||
assert.NoError(err)
|
||||
assert.NotContains(endpoints, tc.nodeInfo.VPCIP)
|
||||
assert.Len(endpoints, tc.wantInstances)
|
||||
// +1 for the load balancer endpoint
|
||||
assert.Len(endpoints, tc.wantEndpoints)
|
||||
}
|
||||
})
|
||||
}
|
||||
|
@ -288,12 +302,18 @@ func TestStart(t *testing.T) {
|
|||
}
|
||||
|
||||
type stubMetadataAPI struct {
|
||||
instances []metadata.InstanceMetadata
|
||||
err error
|
||||
instances []metadata.InstanceMetadata
|
||||
lbEndpoint string
|
||||
getLoadBalancerEndpointErr error
|
||||
listErr error
|
||||
}
|
||||
|
||||
func (s stubMetadataAPI) List(context.Context) ([]metadata.InstanceMetadata, error) {
|
||||
return s.instances, s.err
|
||||
return s.instances, s.listErr
|
||||
}
|
||||
|
||||
func (s stubMetadataAPI) GetLoadBalancerEndpoint(_ context.Context) (string, string, error) {
|
||||
return s.lbEndpoint, "", s.getLoadBalancerEndpointErr
|
||||
}
|
||||
|
||||
type stubRejoinServiceAPI struct {
|
||||
|
|
Loading…
Add table
Add a link
Reference in a new issue