[integration-test] Enable tests to run from local

Author: Prince Rachit Sinha

test/README.md

@@ -69,5 +69,4 @@ To run the tests:
      -enterprise=<true|false> \
      -gitlab=<true|false>
    ```
-3. Tests `TestUploadDownloadBlob` and `TestUploadDownloadBlobViaServer` will fail when testing locally, as they are trying to connect to cluster local resources directly. To test them use docker image instead that runs within the cluster.
-4. If you want to run specific test, add `-run <test>` before `-kubeconfig` parameter.
+3. If you want to run specific test, add `-run <test>` before `-kubeconfig` parameter.

test/pkg/integration/apis.go

@@ -17,6 +17,7 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
+	"net"
 	"net/http"
 	"net/url"
 	"strconv"
@@ -157,6 +158,39 @@ func EncryptValue(value []byte, key []byte) (data string, iv string) {
 	return
 }
 
+// Storage provides a url of the storage provider
+// it takes a url as input and creates a port forward if required
+// e.g. when minio running in gitpod cluster
+// and modifies the url to refer to the localhost instead of dns name
+func (c *ComponentAPI) Storage(connUrl string) (string, error) {
+	u, err := url.Parse(connUrl)
+	if err != nil {
+		return "", err
+	}
+	host, port, _ := net.SplitHostPort(u.Host)
+	if !strings.HasSuffix(host, ".svc.cluster.local") {
+		return connUrl, nil
+	}
+	serviceName := strings.Split(host, ".")[0]
+
+	localPort, err := getFreePort()
+	if err != nil {
+		return "", err
+	}
+
+	ctx, cancel := context.WithCancel(context.Background())
+	ready, errc := common.ForwardPortOfSvc(ctx, c.kubeconfig, c.namespace, serviceName, fmt.Sprintf("%d:%s", localPort, port))
+	select {
+	case err = <-errc:
+		cancel()
+		return "", err
+	case <-ready:
+	}
+	c.appendCloser(func() error { cancel(); return nil })
+
+	return strings.Replace(connUrl, u.Host, fmt.Sprintf("localhost:%d", localPort), 1), nil
+}
+
 // Supervisor provides a gRPC connection to a workspace's supervisor
 func (c *ComponentAPI) Supervisor(instanceID string) (grpc.ClientConnInterface, error) {
 	pod, _, err := selectPod(ComponentWorkspace, selectPodOptions{
@@ -172,7 +206,7 @@ func (c *ComponentAPI) Supervisor(instanceID string) (grpc.ClientConnInterface,
 	}
 
 	ctx, cancel := context.WithCancel(context.Background())
-	ready, errc := common.ForwardPort(ctx, c.kubeconfig, c.namespace, pod, fmt.Sprintf("%d:22999", localPort))
+	ready, errc := common.ForwardPortOfPod(ctx, c.kubeconfig, c.namespace, pod, fmt.Sprintf("%d:22999", localPort))
 	select {
 	case err = <-errc:
 		cancel()
@@ -594,7 +628,7 @@ func (c *ComponentAPI) WorkspaceManager() (wsmanapi.WorkspaceManagerClient, erro
 		}
 
 		ctx, cancel := context.WithCancel(context.Background())
-		ready, errc := common.ForwardPort(ctx, c.kubeconfig, c.namespace, pod, fmt.Sprintf("%d:8080", localPort))
+		ready, errc := common.ForwardPortOfPod(ctx, c.kubeconfig, c.namespace, pod, fmt.Sprintf("%d:8080", localPort))
 		select {
 		case err := <-errc:
 			cancel()
@@ -667,7 +701,7 @@ func (c *ComponentAPI) BlobService() (csapi.BlobServiceClient, error) {
 		}
 
 		ctx, cancel := context.WithCancel(context.Background())
-		ready, errc := common.ForwardPort(ctx, c.kubeconfig, c.namespace, pod, fmt.Sprintf("%d:8080", localPort))
+		ready, errc := common.ForwardPortOfPod(ctx, c.kubeconfig, c.namespace, pod, fmt.Sprintf("%d:8080", localPort))
 		select {
 		case err := <-errc:
 			cancel()
@@ -720,7 +754,7 @@ func (c *ComponentAPI) DB(options ...DBOpt) (*sql.DB, error) {
 	// if configured: setup local port-forward to DB pod
 	if config.ForwardPort != nil {
 		ctx, cancel := context.WithCancel(context.Background())
-		ready, errc := common.ForwardPort(ctx, c.kubeconfig, c.namespace, config.ForwardPort.PodName, fmt.Sprintf("%d:%d", config.Port, config.ForwardPort.RemotePort))
+		ready, errc := common.ForwardPortOfPod(ctx, c.kubeconfig, c.namespace, config.ForwardPort.PodName, fmt.Sprintf("%d:%d", config.Port, config.ForwardPort.RemotePort))
 		select {
 		case err := <-errc:
 			cancel()
@@ -966,7 +1000,7 @@ func (c *ComponentAPI) ImageBuilder(opts ...APIImageBuilderOpt) (imgbldr.ImageBu
 			}
 
 			ctx, cancel := context.WithCancel(context.Background())
-			ready, errc := common.ForwardPort(ctx, c.kubeconfig, c.namespace, pod, fmt.Sprintf("%d:8080", localPort))
+			ready, errc := common.ForwardPortOfPod(ctx, c.kubeconfig, c.namespace, pod, fmt.Sprintf("%d:8080", localPort))
 			select {
 			case err = <-errc:
 				cancel()
@@ -1015,7 +1049,7 @@ func (c *ComponentAPI) ContentService() (ContentService, error) {
 		}
 
 		ctx, cancel := context.WithCancel(context.Background())
-		ready, errc := common.ForwardPort(ctx, c.kubeconfig, c.namespace, pod, fmt.Sprintf("%d:8080", localPort))
+		ready, errc := common.ForwardPortOfPod(ctx, c.kubeconfig, c.namespace, pod, fmt.Sprintf("%d:8080", localPort))
 		select {
 		case err := <-errc:
 			cancel()

test/pkg/integration/common/port_forward.go

@@ -15,18 +15,28 @@ import (
 	"golang.org/x/xerrors"
 )
 
-// ForwardPort establishes a TCP port forwarding to a Kubernetes pod
+// ForwardPortOfPod establishes a TCP port forwarding to a Kubernetes pod
+func ForwardPortOfPod(ctx context.Context, kubeconfig string, namespace, name, port string) (readychan chan struct{}, errchan chan error) {
+	return forwardPort(ctx, kubeconfig, namespace, "pod", name, port)
+}
+
+// ForwardPortOfSvc establishes a TCP port forwarding to a Kubernetes service
+func ForwardPortOfSvc(ctx context.Context, kubeconfig string, namespace, name, port string) (readychan chan struct{}, errchan chan error) {
+	return forwardPort(ctx, kubeconfig, namespace, "service", name, port)
+}
+
+// forwardPort establishes a TCP port forwarding to a Kubernetes resource - pod or service
 // Uses kubectl instead of Go to use a local process that can reproduce the same behavior outside the tests
 // Since we are using kubectl directly we need to pass kubeconfig explicitly
-func ForwardPort(ctx context.Context, kubeconfig string, namespace, pod, port string) (readychan chan struct{}, errchan chan error) {
+func forwardPort(ctx context.Context, kubeconfig string, namespace, resourceType, name, port string) (readychan chan struct{}, errchan chan error) {
 	errchan = make(chan error, 1)
 	readychan = make(chan struct{}, 1)
 
 	go func() {
 		args := []string{
 			"port-forward",
 			"--address=0.0.0.0",
-			fmt.Sprintf("pod/%v", pod),
+			fmt.Sprintf("%s/%v", resourceType, name),
 			fmt.Sprintf("--namespace=%v", namespace),
 			fmt.Sprintf("--kubeconfig=%v", kubeconfig),
 			port,

test/pkg/integration/integration.go

@@ -215,7 +215,7 @@ func Instrument(component ComponentType, agentName string, namespace string, kub
 		}
 	}()
 
-	fwdReady, fwdErr := common.ForwardPort(ctx, kubeconfig, namespace, podName, strconv.Itoa(localAgentPort))
+	fwdReady, fwdErr := common.ForwardPortOfPod(ctx, kubeconfig, namespace, podName, strconv.Itoa(localAgentPort))
 	select {
 	case <-fwdReady:
 	case err := <-execErrs:

test/tests/components/content-service/content-service_test.go

@@ -9,6 +9,7 @@ import (
 	"fmt"
 	"io"
 	"net/http"
+	"net/url"
 	"strings"
 	"testing"
 	"time"
@@ -190,19 +191,27 @@ func TestUploadDownloadBlob(t *testing.T) {
 			if err != nil {
 				t.Fatal(err)
 			}
-			url := resp.Url
-			t.Logf("upload URL: %s", url)
+			originalUrl := resp.Url
+			updatedUrl, err := api.Storage(originalUrl)
+			if err != nil {
+				t.Fatalf("error resolving blob upload target url")
+			}
+			t.Logf("upload URL: %s", updatedUrl)
 
-			uploadBlob(t, url, blobContent)
+			uploadBlob(t, originalUrl, updatedUrl, blobContent)
 
 			resp2, err := bs.DownloadUrl(ctx, &content_service_api.DownloadUrlRequest{OwnerId: gitpodBuiltinUserID, Name: "test-blob"})
 			if err != nil {
 				t.Fatal(err)
 			}
-			url = resp2.Url
-			t.Logf("download URL: %s", url)
+			originalUrl = resp2.Url
+			updatedUrl, err = api.Storage(originalUrl)
+			if err != nil {
+				t.Fatalf("error resolving blob download target url")
+			}
+			t.Logf("download URL: %s", updatedUrl)
 
-			body := downloadBlob(t, url)
+			body := downloadBlob(t, originalUrl, updatedUrl)
 			if string(body) != blobContent {
 				t.Fatalf("blob content mismatch: should '%s' but is '%s'", blobContent, body)
 			}
@@ -235,21 +244,30 @@ func TestUploadDownloadBlobViaServer(t *testing.T) {
 				t.Fatalf("cannot get content blob upload URL: %q", err)
 			}
 
-			url, err := server.GetContentBlobUploadURL(ctx, "test-blob")
+			originalUrl, err := server.GetContentBlobUploadURL(ctx, "test-blob")
 			if err != nil {
 				t.Fatalf("cannot get content blob upload URL: %q", err)
 			}
-			t.Logf("upload URL: %s", url)
+			updatedUrl, err := api.Storage(originalUrl)
+			if err != nil {
+				t.Fatalf("error resolving blob upload target url")
+			}
+			t.Logf("upload URL: %s", updatedUrl)
 
-			uploadBlob(t, url, blobContent)
+			uploadBlob(t, originalUrl, updatedUrl, blobContent)
 
-			url, err = server.GetContentBlobDownloadURL(ctx, "test-blob")
+			originalUrl, err = server.GetContentBlobDownloadURL(ctx, "test-blob")
 			if err != nil {
 				t.Fatalf("cannot get content blob download URL: %q", err)
 			}
-			t.Logf("download URL: %s", url)
 
-			body := downloadBlob(t, url)
+			updatedUrl, err = api.Storage(originalUrl)
+			if err != nil {
+				t.Fatalf("error resolving blob download target url")
+			}
+			t.Logf("download URL: %s", updatedUrl)
+
+			body := downloadBlob(t, originalUrl, updatedUrl)
 			if string(body) != blobContent {
 				t.Fatalf("blob content mismatch: should '%s' but is '%s'", blobContent, body)
 			}
@@ -263,12 +281,20 @@ func TestUploadDownloadBlobViaServer(t *testing.T) {
 	testEnv.Test(t, f)
 }
 
-func uploadBlob(t *testing.T, url string, content string) {
+func uploadBlob(t *testing.T, originalUrl, updatedUrl, content string) {
+	// Always use original URL to extract the host information.
+	// This will avoid any Signature mismatch errors
+	u, err := url.Parse(originalUrl)
+	if err != nil {
+		t.Fatal(err)
+	}
 	var client = &http.Client{Timeout: time.Second * 10}
-	httpreq, err := http.NewRequest(http.MethodPut, url, strings.NewReader(content))
+	httpreq, err := http.NewRequest(http.MethodPut, updatedUrl, strings.NewReader(content))
 	if err != nil {
 		t.Fatalf("cannot create HTTP PUT request: %q", err)
 	}
+	// Add Host header
+	httpreq.Host = u.Host
 	httpresp, err := client.Do(httpreq)
 	if err != nil {
 		t.Fatalf("HTTP PUT request failed: %q", err)
@@ -282,14 +308,27 @@ func uploadBlob(t *testing.T, url string, content string) {
 	}
 }
 
-func downloadBlob(t *testing.T, url string) string {
-	httpresp, err := http.Get(url)
+func downloadBlob(t *testing.T, originalUrl, updatedUrl string) string {
+	// Always use original URL to extract the host information.
+	// This will avoid any Signature mismatch errors
+	u, err := url.Parse(originalUrl)
+	if err != nil {
+		t.Fatal(err)
+	}
+	var client = &http.Client{Timeout: time.Second * 10}
+	httpreq, err := http.NewRequest(http.MethodGet, updatedUrl, nil)
 	if err != nil {
-		t.Fatalf("HTTP GET requst failed: %q", err)
+		t.Fatalf("cannot create HTTP GET request: %q", err)
+	}
+	// Add Host header
+	httpreq.Host = u.Host
+	httpresp, err := client.Do(httpreq)
+	if err != nil {
+		t.Fatalf("HTTP GET request failed: %q", err)
 	}
 	body, err := io.ReadAll(httpresp.Body)
 	if err != nil {
-		t.Fatalf("cannot read response body of HTTP PUT: %q", err)
+		t.Fatalf("cannot read response body of HTTP GET: %q", err)
 	}
 	return string(body)
 }