wip

bootstrapper/cmd/bootstrapper/run.go

@@ -99,5 +99,4 @@ type clusterInitJoiner interface {
 type metadataAPI interface {
 	joinclient.MetadataAPI
 	initserver.MetadataAPI
-	GetLoadBalancerEndpoint(ctx context.Context) (host, port string, err error)
 }

bootstrapper/internal/joinclient/joinclient.go

@@ -188,8 +188,14 @@ func (c *JoinClient) Stop() {
 func (c *JoinClient) tryJoinWithAvailableServices() error {
 	ips, err := c.getControlPlaneIPs()
 	if err != nil {
-		return err
+		return fmt.Errorf("failed to get control plane IPs: %w", err)
+	}
+
+	ip, _, err := c.metadataAPI.GetLoadBalancerEndpoint(context.TODO())
+	if err != nil {
+		return fmt.Errorf("failed to get load balancer endpoint: %w", err)
 	}
+	ips = append(ips, ip)
 
 	if len(ips) == 0 {
 		return errors.New("no control plane IPs found")
@@ -425,6 +431,8 @@ type MetadataAPI interface {
 	List(ctx context.Context) ([]metadata.InstanceMetadata, error)
 	// Self retrieves the current instance.
 	Self(ctx context.Context) (metadata.InstanceMetadata, error)
+	// GetLoadBalancerEndpoint retrieves the load balancer endpoint.
+	GetLoadBalancerEndpoint(ctx context.Context) (host, port string, err error)
 }
 
 type encryptedDisk interface {

bootstrapper/internal/joinclient/joinclient_test.go

@@ -330,6 +330,10 @@ func (s *stubRepeaterMetadataAPI) List(_ context.Context) ([]metadata.InstanceMe
 	return s.listInstances, s.listErr
 }
 
+func (s *stubRepeaterMetadataAPI) GetLoadBalancerEndpoint(_ context.Context) (string, string, error) {
+	return "", "", nil
+}
+
 type stubMetadataAPI struct {
 	selfAnswerC chan selfAnswer
 	listAnswerC chan listAnswer
@@ -352,6 +356,10 @@ func (s *stubMetadataAPI) List(_ context.Context) ([]metadata.InstanceMetadata,
 	return answer.instances, answer.err
 }
 
+func (s *stubMetadataAPI) GetLoadBalancerEndpoint(_ context.Context) (string, string, error) {
+	return "", "", nil
+}
+
 type selfAnswer struct {
 	instance metadata.InstanceMetadata
 	err      error

cli/internal/terraform/terraform/aws/main.tf

@@ -27,13 +27,15 @@ locals {
   ports_verify       = "30081"
   ports_recovery     = "9999"
   ports_debugd       = "4000"
+  ports_join       = "30090"
   target_group_arns = {
     control-plane : flatten([
       module.load_balancer_target_bootstrapper.target_group_arn,
       module.load_balancer_target_kubernetes.target_group_arn,
       module.load_balancer_target_verify.target_group_arn,
       module.load_balancer_target_recovery.target_group_arn,
       module.load_balancer_target_konnectivity.target_group_arn,
+      module.load_balancer_target_join.target_group_arn,
       var.debug ? [module.load_balancer_target_debugd[0].target_group_arn] : [],
     ])
     worker : []
@@ -96,6 +98,7 @@ resource "aws_lb" "front_end" {
   internal           = false
   load_balancer_type = "network"
   tags               = local.tags
+  security_groups = [ aws_security_group.security_group.id ]
 
   dynamic "subnet_mapping" {
     # TODO(malt3): use for_each = toset(module.public_private_subnet.all_zones)
@@ -255,6 +258,16 @@ module "load_balancer_target_konnectivity" {
   healthcheck_protocol = "TCP"
 }
 
+module "load_balancer_target_join" {
+  source               = "./modules/load_balancer_target"
+  name                 = "${local.name}-join"
+  vpc_id               = aws_vpc.vpc.id
+  lb_arn               = aws_lb.front_end.arn
+  port                 = local.ports_join
+  tags                 = local.tags
+  healthcheck_protocol = "TCP"
+}
+
 module "instance_group" {
   source               = "./modules/instance_group"
   for_each             = var.node_groups

debugd/internal/debugd/metadata/scheduler.go

@@ -19,6 +19,7 @@ import (
 // Fetcher retrieves other debugd IPs from cloud provider metadata.
 type Fetcher interface {
 	DiscoverDebugdIPs(ctx context.Context) ([]string, error)
+	DiscoverLoadbalancerIP(ctx context.Context) (string, error)
 }
 
 // Scheduler schedules fetching of metadata using timers.
@@ -51,23 +52,35 @@ func (s *Scheduler) Start(ctx context.Context, wg *sync.WaitGroup) {
 		defer ticker.Stop()
 
 		for {
+			select {
+			case <-ctx.Done():
+				return
+			case <-ticker.C:
+			}
+
 			ips, err := s.fetcher.DiscoverDebugdIPs(ctx)
 			if err != nil {
 				s.log.With(zap.Error(err)).Warnf("Discovering debugd IPs failed")
 			}
-			if err == nil {
-				s.log.With(zap.Strings("ips", ips)).Infof("Discovered instances")
-				s.download(ctx, ips)
-				if s.deploymentDone && s.infoDone {
-					return
-				}
+
+			lbip, err := s.fetcher.DiscoverLoadbalancerIP(ctx)
+			if err != nil {
+				s.log.With(zap.Error(err)).Warnf("Discovering loadbalancer IP failed")
+			} else {
+				ips = append(ips, lbip)
 			}
 
-			select {
-			case <-ctx.Done():
+			if len(ips) == 0 {
+				s.log.With(zap.Error(err)).Warnf("No debugd IPs discovered")
+				continue
+			}
+
+			s.log.With(zap.Strings("ips", ips)).Infof("Discovered instances")
+			s.download(ctx, ips)
+			if s.deploymentDone && s.infoDone {
 				return
-			case <-ticker.C:
 			}
+
 		}
 	}()
 }

debugd/internal/debugd/metadata/scheduler_test.go

@@ -104,6 +104,10 @@ func (s *stubFetcher) DiscoverDebugdIPs(_ context.Context) ([]string, error) {
 	return s.ips, nil
 }
 
+func (s *stubFetcher) DiscoverLoadbalancerIP(_ context.Context) (string, error) {
+	return "", errors.New("not implemented")
+}
+
 type stubDownloader struct {
 	downloadDeploymentErrs   []error
 	downloadDeploymentErrIdx int

debugd/internal/debugd/server/server.go

@@ -133,9 +133,6 @@ func (s *debugdServer) UploadFiles(stream pb.Debugd_UploadFilesServer) error {
 // DownloadFiles streams the previously received files to other instances.
 func (s *debugdServer) DownloadFiles(_ *pb.DownloadFilesRequest, stream pb.Debugd_DownloadFilesServer) error {
 	s.log.Infof("Sending files to other instance")
-	if !s.transfer.CanSend() {
-		return errors.New("cannot send files at this time")
-	}
 	return s.transfer.SendFiles(stream)
 }
 

debugd/internal/debugd/server/server_test.go

@@ -228,10 +228,6 @@ func TestDownloadFiles(t *testing.T) {
 			canSend:           true,
 			wantSendFileCalls: 1,
 		},
-		"transfer is not ready for sending": {
-			request:     &pb.DownloadFilesRequest{},
-			wantRecvErr: true,
-		},
 	}
 
 	for name, tc := range testCases {

debugd/internal/filetransfer/filetransfer.go

@@ -13,6 +13,7 @@ import (
 	"io"
 	"io/fs"
 	"sync"
+	"sync/atomic"
 
 	"github.com/edgelesssys/constellation/v2/debugd/internal/debugd"
 	"github.com/edgelesssys/constellation/v2/debugd/internal/filetransfer/streamer"
@@ -33,10 +34,10 @@ type SendFilesStream interface {
 
 // FileTransferer manages sending and receiving of files.
 type FileTransferer struct {
-	mux             sync.RWMutex
+	fileMux         sync.RWMutex
 	log             *logger.Logger
 	receiveStarted  bool
-	receiveFinished bool
+	receiveFinished atomic.Bool
 	files           []FileStat
 	streamer        streamReadWriter
 	showProgress    bool
@@ -52,12 +53,15 @@ func New(log *logger.Logger, streamer streamReadWriter, showProgress bool) *File
 }
 
 // SendFiles sends files to the given stream.
+// If the FileTransferer has not received any files, an error is returned.
 func (s *FileTransferer) SendFiles(stream SendFilesStream) error {
-	s.mux.RLock()
-	defer s.mux.RUnlock()
-	if !s.receiveFinished {
+	if !s.receiveFinished.Load() {
 		return errors.New("cannot send files before receiving them")
 	}
+
+	s.fileMux.RLock()
+	defer s.fileMux.RUnlock()
+
 	for _, file := range s.files {
 		if err := s.handleFileSend(stream, file); err != nil {
 			return err
@@ -68,8 +72,8 @@ func (s *FileTransferer) SendFiles(stream SendFilesStream) error {
 
 // RecvFiles receives files from the given stream.
 func (s *FileTransferer) RecvFiles(stream RecvFilesStream) (err error) {
-	s.mux.Lock()
-	defer s.mux.Unlock()
+	s.fileMux.Lock()
+	defer s.fileMux.Unlock()
 	if err := s.startRecv(); err != nil {
 		return err
 	}
@@ -89,30 +93,23 @@ func (s *FileTransferer) RecvFiles(stream RecvFilesStream) (err error) {
 
 // GetFiles returns the a copy of the list of files that have been received.
 func (s *FileTransferer) GetFiles() []FileStat {
-	s.mux.RLock()
-	defer s.mux.RUnlock()
+	s.fileMux.RLock()
+	defer s.fileMux.RUnlock()
 	res := make([]FileStat, len(s.files))
 	copy(res, s.files)
 	return res
 }
 
 // SetFiles sets the list of files that can be sent.
+// This function is used for a sender which has not received any files through
+// this FileTransferer i.e. the CLI.
 func (s *FileTransferer) SetFiles(files []FileStat) {
-	s.mux.Lock()
-	defer s.mux.Unlock()
+	s.fileMux.Lock()
+	defer s.fileMux.Unlock()
 	res := make([]FileStat, len(files))
 	copy(res, files)
 	s.files = res
-	s.receiveFinished = true
-}
-
-// CanSend returns true if the file receive has finished.
-// This is called to determine if a debugd instance can request files from this server.
-func (s *FileTransferer) CanSend() bool {
-	s.mux.RLock()
-	defer s.mux.RUnlock()
-	ret := s.receiveFinished
-	return ret
+	s.receiveFinished.Store(true)
 }
 
 func (s *FileTransferer) handleFileSend(stream SendFilesStream, file FileStat) error {
@@ -173,7 +170,7 @@ func (s *FileTransferer) handleFileRecv(stream RecvFilesStream) (bool, error) {
 // startRecv marks the file receive as started. It returns an error if receiving has already started.
 func (s *FileTransferer) startRecv() error {
 	switch {
-	case s.receiveFinished:
+	case s.receiveFinished.Load():
 		return ErrReceiveFinished
 	case s.receiveStarted:
 		return ErrReceiveRunning
@@ -193,7 +190,7 @@ func (s *FileTransferer) abortRecv() {
 // This allows other debugd instances to request files from this server.
 func (s *FileTransferer) finishRecv() {
 	s.receiveStarted = false
-	s.receiveFinished = true
+	s.receiveFinished.Store(true)
 }
 
 // addFile adds a file to the list of received files.