fix socket connection and test it in CI

Signed-off-by: dciangot <[email protected]>

CONTRIBUTING.md

@@ -31,6 +31,7 @@ you submit the pull request). We have a 3 steps process for contributions.
    understanding and appreciating your pull request, please use the template to
    explain _why_ you are making this contribution, rather than just _what_ the
    contribution entails.
+2. **Run E2E tests with success**. You can follow the steps described [here](https://intertwin-eu.github.io/interLink/docs/Developers)
 
 ### Code Review Process
 
@@ -44,7 +45,7 @@ follows:
 
 1. A maintainer will review your code and merge it if no changes are necessary.
    Your change will be merged into the repository's `main` branch.
-1. If a maintainer has feedback or questions on your changes then they will set
+2. If a maintainer has feedback or questions on your changes then they will set
    `request changes` in the review and provide an explanation.
 
 ## Using git

ci/main.go

@@ -15,19 +15,6 @@ import (
 )
 
 var (
-	interLinkPatch = `
-kind: Deployment
-metadata:
-  name: interlink
-  namespace: interlink
-spec:
-  template:
-    spec:
-      containers:
-      - name: interlink
-        image: "{{.InterLinkRef}}"
-
-`
 	virtualKubeletPatch = `
 kind: Deployment
 metadata:
@@ -39,6 +26,8 @@ spec:
       containers:
       - name: inttw-vk
         image: "{{.VirtualKubeletRef}}"
+      - name: interlink
+        image: "{{.InterLinkRef}}"
 `
 )
 
@@ -144,18 +133,8 @@ EOF`}, ContainerWithExecOpts{SkipEntrypoint: true}).
 		VirtualKubeletRef: m.VirtualKubeletRef,
 	}
 
-	interLinkCompiler, err := template.New("interlink").Parse(interLinkPatch)
-	if err != nil {
-		return nil, err
-	}
-
 	bufferIL := new(bytes.Buffer)
 
-	err = interLinkCompiler.Execute(bufferIL, patch)
-	if err != nil {
-		return nil, err
-	}
-
 	virtualKubeletCompiler, err := template.New("vk").Parse(virtualKubeletPatch)
 	if err != nil {
 		return nil, err
@@ -209,34 +188,11 @@ EOF`}, ContainerWithExecOpts{SkipEntrypoint: true}).
 	fmt.Println(vkConfig)
 
 	return m, nil
-	//maxRetries := 10
-	//retryBackoff := 10 * time.Second
-	// for i := 0; i < maxRetries; i++ {
-	// 	kubectlGetNodes, err := kubectl.WithExec([]string{"get", "nodes", "-o", "wide", "virtual-kubelet"}).Stdout(ctx)
-	// 	if err != nil {
-	// 		fmt.Println(fmt.Errorf("could not fetch nodes: %v", err))
-	// 		fmt.Println("waiting for k8s to start:", kubectlGetNodes)
-	// 		time.Sleep(retryBackoff)
-	// 		continue
-	// 	}
-	// 	if strings.Contains(kubectlGetNodes, " Ready") {
-	// 		time.Sleep(30 * time.Second)
-	// 		return m, nil
-	// 	}
-	// 	time.Sleep(retryBackoff)
-	// }
-	// kubectlAll, err := kubectl.WithExec([]string{"logs", "-n", "interlink", "-l", "nodeName=virtual-kubelet"}).Stdout(ctx)
-	// if err != nil {
-	// 	return nil, err
-	// }
-	// fmt.Println(kubectlAll)
-	//
-	// return nil, fmt.Errorf("k8s took too long to start")
 }
 
 // Returns the kubeconfig file of the k3s cluster
 func (m *Interlink) Config() *File {
-	return m.KubeConfigHost
+	return dag.K3S(m.Name).Config(true)
 }
 
 // Build interLink and virtual kubelet docker images from source

ci/manifests/interlink-config.yaml

@@ -5,7 +5,8 @@ metadata:
   namespace: interlink 
 data:
   InterLinkConfig.yaml: |
-    InterlinkAddress: "http://0.0.0.0"
+    InterlinkAddress: "unix:///var/run/interlink.socket"
+    #InterlinkAddress: "http://0.0.0.0"
     InterlinkPort: "3000"
     #SidecarURL: "http://plugin.interlink.svc.cluster.local"
     SidecarURL: "http://plugin"

ci/manifests/kustomization.yaml

@@ -2,16 +2,12 @@ resources:
 - virtual-kubelet-config.yaml
 - virtual-kubelet.yaml
 - interlink-config.yaml
-- interlink.yaml
+#- interlink.yaml
 #- plugin-k8s-config.yaml
 #- plugin.yaml
 patches:
 - path: virtual-kubelet-merge.yaml
   target:
     kind: Deployment
     labelSelector: nodeName=virtual-kubelet
-- path: interlink-merge.yaml
-  target:
-    kind: Deployment
-    labelSelector: app=interlink
 

ci/manifests/plugin-config.yaml

@@ -6,9 +6,9 @@ VerboseLogging: true
 ErrorsOnlyLogging: false
 ExportPodData: true
 # NEEDED PATH FOR GITHUB ACTIONS
-DataRootFolder: "/home/runner/work/interLink/interLink/.interlink/"
+#DataRootFolder: "/home/runner/work/interLink/interLink/.interlink/"
 # on your host use something like:
-#DataRootFolder: "/home/ubuntu/.interlink/"
+DataRootFolder: "/home/ubuntu/.interlink/"
 SbatchPath: "/usr/bin/sbatch"
 ScancelPath: "/usr/bin/scancel"
 SqueuePath: "/usr/bin/squeue"

ci/manifests/virtual-kubelet-config.yaml

@@ -5,14 +5,14 @@ metadata:
   namespace: interlink 
 data:
   InterLinkConfig.yaml: |
-    InterlinkURL: http://interlink.interlink.svc.cluster.local
-    InterlinkPort: 3000 
+    InterlinkURL: unix:///var/run/interlink.socket
+    #InterlinkPort: 3000 
     ExportPodData: true
     VerboseLogging: true
     ErrorsOnlyLogging: false
     ServiceAccount: "virtual-kubelet"
     Namespace: interlink 
-    VKTokenFile: /dev/null 
+    VKTokenFile: ""
     CPU: "100"
     Memory: "128Gi"
     Pods: "100"

ci/manifests/virtual-kubelet.yaml

@@ -18,6 +18,18 @@ spec:
       automountServiceAccountToken: true
       serviceAccountName: virtual-kubelet 
       containers:
+      - name: interlink
+        image: "ghcr.io/intertwin-eu/interlink/interlink"
+        imagePullPolicy: Always
+        env:
+          - name: INTERLINKCONFIGPATH
+            value: "/etc/interlink/InterLinkConfig.yaml"
+        volumeMounts:
+          - name: il-config
+            mountPath: /etc/interlink/InterLinkConfig.yaml
+            subPath: InterLinkConfig.yaml
+          - name: sockets
+            mountPath: /var/run/
       - name: inttw-vk
         image: "ghcr.io/intertwin-eu/interlink/virtual-kubelet-inttw" 
         imagePullPolicy: Always
@@ -36,8 +48,20 @@ spec:
         - name: config
           mountPath: /etc/interlink/InterLinkConfig.yaml
           subPath: InterLinkConfig.yaml
+        - name: sockets 
+          mountPath: /var/run/
       volumes:
       - name: config
         configMap:
           # Provide the name of the ConfigMap you want to mount.
           name: virtual-kubelet-config
+      - name: sockets
+        hostPath: 
+          path: /var/run
+          type: Directory
+      - name: il-config
+        configMap:
+          # Provide the name of the ConfigMap you want to mount.
+          name: interlink-config
+      - name: sockets
+        emptyDir: {}  

cmd/interlink/main.go

@@ -5,9 +5,12 @@ import (
 	"crypto/tls"
 	"flag"
 	"fmt"
+	"net"
 	"net/http"
 	"os"
+	"os/signal"
 	"strings"
+	"syscall"
 	"time"
 
 	"github.com/sirupsen/logrus"
@@ -157,15 +160,39 @@ func main() {
 	interLinkEndpoint := ""
 	if strings.HasPrefix(interLinkConfig.InterlinkAddress, "unix://") {
 		interLinkEndpoint = interLinkConfig.InterlinkAddress
+
+		// Create a Unix domain socket and listen for incoming connections.
+		socket, err := net.Listen("unix", strings.Replace(interLinkEndpoint, "unix://", "", -1))
+		if err != nil {
+			panic(err)
+		}
+
+		// Cleanup the sockfile.
+		c := make(chan os.Signal, 1)
+		signal.Notify(c, os.Interrupt, syscall.SIGTERM)
+		go func() {
+			<-c
+			os.Remove(interLinkEndpoint)
+			os.Exit(1)
+		}()
+		server := http.Server{
+			Handler: mutex,
+		}
+
+		log.G(ctx).Info(socket)
+
+		if err := server.Serve(socket); err != nil {
+			log.G(ctx).Fatal(err)
+		}
 	} else if strings.HasPrefix(interLinkConfig.InterlinkAddress, "http://") {
 		interLinkEndpoint = strings.Replace(interLinkConfig.InterlinkAddress, "http://", "", -1) + ":" + interLinkConfig.Interlinkport
-	} else {
-		log.G(ctx).Fatal("Sidecar URL should either start per unix:// or http://")
-	}
 
-	err = http.ListenAndServe(interLinkEndpoint, mutex)
+		err = http.ListenAndServe(interLinkEndpoint, mutex)
 
-	if err != nil {
-		log.G(ctx).Fatal(err)
+		if err != nil {
+			log.G(ctx).Fatal(err)
+		}
+	} else {
+		log.G(ctx).Fatal("Sidecar URL should either start per unix:// or http://")
 	}
 }

cmd/virtual-kubelet/main.go

@@ -24,6 +24,7 @@ import (
 	"os"
 	"path"
 	"strconv"
+	"strings"
 	"time"
 
 	// "k8s.io/client-go/rest"
@@ -185,6 +186,8 @@ func main() {
 	}
 	log.L = logruslogger.FromLogrus(logrus.NewEntry(logger))
 
+	log.G(ctx).Info("Config dump", interLinkConfig)
+
 	if os.Getenv("ENABLE_TRACING") == "1" {
 		shutdown, err := initProvider(ctx)
 		if err != nil {
@@ -207,6 +210,19 @@ func main() {
 
 	http.DefaultTransport.(*http.Transport).TLSClientConfig = &tls.Config{InsecureSkipVerify: true}
 
+	if strings.HasPrefix(interLinkConfig.InterlinkURL, "unix://") {
+		// Dial the Unix socket
+		interLinkEndpoint := strings.Replace(interLinkConfig.InterlinkURL, "unix://", "", -1)
+		conn, err := net.Dial("unix", interLinkEndpoint)
+		if err != nil {
+			panic(err)
+		}
+
+		http.DefaultTransport.(*http.Transport).DialContext = func(_ context.Context, _, _ string) (net.Conn, error) {
+			return conn, nil
+		}
+	}
+
 	dport, err := strconv.ParseInt(os.Getenv("KUBELET_PORT"), 10, 32)
 	if err != nil {
 		log.G(ctx).Fatal(err)