test/e2e/in-memory: improve performance by fixing locking issues

Signed-off-by: Stefan Büringer [email protected]

Author: sbueringer

test/infrastructure/inmemory/internal/server/certs.go

@@ -26,13 +26,19 @@ import (
 	"sigs.k8s.io/cluster-api/util/certs"
 )
 
-// newCertAndKey builds a new cert and key signed with the given CA and with the given config.
-func newCertAndKey(caCert *x509.Certificate, caKey *rsa.PrivateKey, config *certs.Config) (*x509.Certificate, *rsa.PrivateKey, error) {
-	key, err := certs.NewPrivateKey()
+var key *rsa.PrivateKey
+
+func init() {
+	// Create a private key only once, since this is a slow operation and it is ok
+	// to reuse it for all the certificates in a test provider.
+	var err error
+	key, err = certs.NewPrivateKey()
 	if err != nil {
-		return nil, nil, errors.Wrap(err, "unable to create private key")
+		panic(errors.Wrap(err, "unable to create private key").Error())
 	}
+}
 
+func newCertAndKey(caCert *x509.Certificate, caKey *rsa.PrivateKey, config *certs.Config) (*x509.Certificate, *rsa.PrivateKey, error) {
 	cert, err := config.NewSignedCert(key, caCert, caKey)
 	if err != nil {
 		return nil, nil, errors.Wrap(err, "unable to create certificate")

test/infrastructure/inmemory/internal/server/mux.go

@@ -32,7 +32,9 @@ import (
 	"github.com/pkg/errors"
 	"golang.org/x/net/http2"
 	"golang.org/x/net/http2/h2c"
+	kerrors "k8s.io/apimachinery/pkg/util/errors"
 	"k8s.io/apimachinery/pkg/util/sets"
+	"k8s.io/apimachinery/pkg/util/wait"
 	"k8s.io/klog/v2"
 	"sigs.k8s.io/controller-runtime/pkg/log"
 
@@ -316,81 +318,107 @@ func (m *WorkloadClustersMux) initWorkloadClusterListenerWithPortLocked(wclName
 // When the first API server instance is added the serving certificates and the admin certificate
 // for tests are generated, and the listener is started.
 func (m *WorkloadClustersMux) AddAPIServer(wclName, podName string, caCert *x509.Certificate, caKey *rsa.PrivateKey) error {
-	m.lock.Lock()
-	defer m.lock.Unlock()
-
-	wcl, ok := m.workloadClusterListeners[wclName]
-	if !ok {
-		return errors.Errorf("workloadClusterListener with name %s must be initialized before adding an APIserver", wclName)
-	}
-	wcl.apiServers.Insert(podName)
-	m.log.Info("APIServer instance added to workloadClusterListener", "listenerName", wclName, "address", wcl.Address(), "podName", podName)
-
-	// TODO: check if cert/key are already set, they should match
-	wcl.apiServerCaCertificate = caCert
-	wcl.apiServerCaKey = caKey
+	// Start server
+	// Note: It is important that we unlock once the server is started. Because otherwise the server
+	// doesn't work yet as GetCertificate (which is required for the tls handshake) also requires the lock.
+	var startServerErr error
+	var wcl *WorkloadClusterListener
+	err := func() error {
+		m.lock.Lock()
+		defer m.lock.Unlock()
+
+		var ok bool
+		wcl, ok = m.workloadClusterListeners[wclName]
+		if !ok {
+			return errors.Errorf("workloadClusterListener with name %s must be initialized before adding an APIserver", wclName)
+		}
+		wcl.apiServers.Insert(podName)
+		m.log.Info("APIServer instance added to workloadClusterListener", "listenerName", wclName, "address", wcl.Address(), "podName", podName)
+
+		// TODO: check if cert/key are already set, they should match
+		wcl.apiServerCaCertificate = caCert
+		wcl.apiServerCaKey = caKey
+
+		// Generate Serving certificates for the API server instance
+		// NOTE: There is only one server certificate for all API server instances (kubeadm
+		// instead creates one for each API server pod). We don't need this because we are
+		// accessing all API servers via the same endpoint.
+		if wcl.apiServerServingCertificate == nil {
+			config := apiServerCertificateConfig(wcl.host)
+			cert, key, err := newCertAndKey(caCert, caKey, config)
+			if err != nil {
+				return errors.Wrapf(err, "failed to create serving certificate for API server %s", podName)
+			}
+
+			certificate, err := tls.X509KeyPair(certs.EncodeCertPEM(cert), certs.EncodePrivateKeyPEM(key))
+			if err != nil {
+				return errors.Wrapf(err, "failed to create X509KeyPair for API server %s", podName)
+			}
+			wcl.apiServerServingCertificate = &certificate
+		}
 
-	// Generate Serving certificates for the API server instance
-	// NOTE: There is only one server certificate for all API server instances (kubeadm
-	// instead creates one for each API server pod). We don't need this because we are
-	// accessing all API servers via the same endpoint.
-	if wcl.apiServerServingCertificate == nil {
-		config := apiServerCertificateConfig(wcl.host)
-		cert, key, err := newCertAndKey(caCert, caKey, config)
-		if err != nil {
-			return errors.Wrapf(err, "failed to create serving certificate for API server %s", podName)
+		// Generate admin certificates to be used for accessing the API server.
+		// NOTE: this is used for tests because CAPI creates its own.
+		if wcl.adminCertificate == nil {
+			config := adminClientCertificateConfig()
+			cert, key, err := newCertAndKey(caCert, caKey, config)
+			if err != nil {
+				return errors.Wrapf(err, "failed to create admin certificate for API server %s", podName)
+			}
+
+			wcl.adminCertificate = cert
+			wcl.adminKey = key
 		}
 
-		certificate, err := tls.X509KeyPair(certs.EncodeCertPEM(cert), certs.EncodePrivateKeyPEM(key))
-		if err != nil {
-			return errors.Wrapf(err, "failed to create X509KeyPair for API server %s", podName)
+		// Start the listener for the API server.
+		// NOTE: There is only one listener for all API server instances; the same listener will act
+		// as a port forward target too.
+		if wcl.listener != nil {
+			return nil
 		}
-		wcl.apiServerServingCertificate = &certificate
-	}
 
-	// Generate admin certificates to be used for accessing the API server.
-	// NOTE: this is used for tests because CAPI creates its own.
-	if wcl.adminCertificate == nil {
-		config := adminClientCertificateConfig()
-		cert, key, err := newCertAndKey(caCert, caKey, config)
+		l, err := net.Listen("tcp", wcl.HostPort())
 		if err != nil {
-			return errors.Wrapf(err, "failed to create admin certificate for API server %s", podName)
+			return errors.Wrapf(err, "failed to start WorkloadClusterListener %s, %s", wclName, wcl.HostPort())
 		}
+		wcl.listener = l
 
-		wcl.adminCertificate = cert
-		wcl.adminKey = key
-	}
-
-	// Start the listener for the API server.
-	// NOTE: There is only one listener for all API server instances; the same listener will act
-	// as a port forward target too.
-	if wcl.listener != nil {
+		go func() {
+			if startServerErr = m.muxServer.ServeTLS(wcl.listener, "", ""); startServerErr != nil && !errors.Is(startServerErr, http.ErrServerClosed) {
+				m.log.Error(startServerErr, "Failed to start WorkloadClusterListener", "listenerName", wclName, "address", wcl.Address())
+			}
+		}()
 		return nil
-	}
-
-	l, err := net.Listen("tcp", wcl.HostPort())
+	}()
 	if err != nil {
-		return errors.Wrapf(err, "failed to start WorkloadClusterListener %s, %s", wclName, wcl.HostPort())
+		return errors.Wrapf(err, "error starting server")
 	}
-	wcl.listener = l
 
-	var startErr error
-	startCh := make(chan struct{})
-	go func() {
-		startCh <- struct{}{}
-		if err := m.muxServer.ServeTLS(wcl.listener, "", ""); err != nil && !errors.Is(err, http.ErrServerClosed) {
-			startErr = err
-			m.log.Error(startErr, "Failed to start WorkloadClusterListener", "listenerName", wclName, "address", wcl.Address())
+	// Wait until the sever is working.
+	var pollErr error
+	err = wait.PollUntilContextTimeout(context.TODO(), 10*time.Millisecond, 1*time.Second, true, func(ctx context.Context) (done bool, err error) {
+		d := &net.Dialer{Timeout: 50 * time.Millisecond}
+		conn, err := tls.DialWithDialer(d, "tcp", wcl.HostPort(), &tls.Config{
+			InsecureSkipVerify: true, //nolint:gosec // config is used to connect to our own port.
+		})
+		if err != nil {
+			pollErr = fmt.Errorf("server is not reachable: %w", err)
+			return false, nil
+		}
+
+		if err := conn.Close(); err != nil {
+			pollErr = fmt.Errorf("server is not reachable: closing connection: %w", err)
+			return false, nil
 		}
-	}()
 
-	<-startCh
-	// TODO: Try to make this race condition free e.g. by checking the listener is answering.
-	// There is no guarantee ServeTLS was called after we received something on the startCh.
-	time.Sleep(100 * time.Millisecond)
+		return true, nil
+	})
+	if err != nil {
+		return kerrors.NewAggregate([]error{err, pollErr})
+	}
 
-	if startErr != nil {
-		return startErr
+	if startServerErr != nil {
+		return startServerErr
 	}
 
 	m.log.Info("WorkloadClusterListener successfully started", "listenerName", wclName, "address", wcl.Address())