feat(infra): global vpn support with loadbalancer

.tools/nvim/__http__/infra/global-vpn-devices.graphql.yml

@@ -1,7 +1,7 @@
 ---
 global:
   gvpn: "default"
-  deviceName: "kloudlite-platform-device"
+  deviceName: "kloudlite-global-vpn-device"
   # deviceName: "second-device"
 ---
 label: "Create GlobalVPN Device"

apps/infra/internal/domain/byok-clusters.go

@@ -67,6 +67,10 @@ func (d *domain) CreateBYOKCluster(ctx InfraContext, cluster entities.BYOKCluste
 		cluster.GlobalVPN = DefaultGlobalVPNName
 	}
 
+	if _, err := d.ensureGlobalVPN(ctx, cluster.GlobalVPN); err != nil {
+		return nil, errors.NewE(err)
+	}
+
 	ctoken, err := d.generateClusterToken(ctx, cluster.Name)
 	if err != nil {
 		return nil, errors.NewE(err)
@@ -76,21 +80,12 @@ func (d *domain) CreateBYOKCluster(ctx InfraContext, cluster entities.BYOKCluste
 
 	cluster.MessageQueueTopicName = common.GetTenantClusterMessagingTopic(ctx.AccountName, cluster.Name)
 
-	gvpn, err := d.ensureGlobalVPN(ctx, cluster.GlobalVPN)
+	gvpnConn, err := d.ensureGlobalVPNConnection(ctx, cluster.Name, cluster.GlobalVPN, cluster.ClusterPublicEndpoint)
 	if err != nil {
 		return nil, errors.NewE(err)
 	}
 
-	clusterSvcCIDR, err := d.claimNextClusterSvcCIDR(ctx, cluster.Name, gvpn.Name)
-	if err != nil {
-		return nil, err
-	}
-
-	if _, err := d.ensureGlobalVPNConnection(ctx, cluster.Name, clusterSvcCIDR, cluster.GlobalVPN, cluster.ClusterPublicEndpoint); err != nil {
-		return nil, errors.NewE(err)
-	}
-
-	cluster.ClusterSvcCIDR = clusterSvcCIDR
+	cluster.ClusterSvcCIDR = gvpnConn.ClusterSvcCIDR
 
 	existing, err := d.clusterRepo.FindOne(ctx, repos.Filter{
 		fields.MetadataName:      cluster.Name,
@@ -178,8 +173,8 @@ func (d *domain) GetBYOKClusterSetupInstructions(ctx InfraContext, name string)
 	}
 
 	return []string{
-		fmt.Sprintf(`helm repo add kloudlite https://kloudlite.github.io/helm-charts`),
-		fmt.Sprintf(`helm repo update kloudlite`),
+		`helm repo add kloudlite https://kloudlite.github.io/helm-charts`,
+		`helm repo update kloudlite`,
 		fmt.Sprintf(`helm upgrade --install kloudlite --namespace kloudlite --create-namespace kloudlite/kloudlite-agent --version %s --set accountName="%s" --set clusterName="%s" --set clusterToken="%s" --set messageOfficeGRPCAddr="%s" --set byok.enabled=true --set helmCharts.ingressNginx.enabled=true --set helmCharts.certManager.enabled=true`, d.env.KloudliteRelease, ctx.AccountName, name, cluster.ClusterToken, d.env.MessageOfficeExternalGrpcAddr),
 	}, nil
 }
@@ -203,29 +198,14 @@ func (d *domain) DeleteBYOKCluster(ctx InfraContext, name string) error {
 		return errors.NewE(err)
 	}
 
-	if err := d.byokClusterRepo.DeleteOne(ctx, entities.UniqueBYOKClusterFilter(ctx.AccountName, name)); err != nil {
-		return errors.NewE(err)
-	}
-
 	if cluster.GlobalVPN != "" {
 		if err := d.deleteGlobalVPNConnection(ctx, cluster.Name, cluster.GlobalVPN); err != nil {
 			return errors.NewE(err)
 		}
-		if err := d.claimClusterSvcCIDRRepo.DeleteOne(ctx, repos.Filter{
-			fc.ClaimClusterSvcCIDRClaimedByCluster: cluster.Name,
-			fc.AccountName:                         ctx.AccountName,
-			fc.ClaimClusterSvcCIDRGlobalVPNName:    cluster.GlobalVPN,
-		}); err != nil {
-			return errors.NewE(err)
-		}
+	}
 
-		if _, err := d.freeClusterSvcCIDRRepo.Create(ctx, &entities.FreeClusterSvcCIDR{
-			AccountName:    ctx.AccountName,
-			GlobalVPNName:  cluster.GlobalVPN,
-			ClusterSvcCIDR: cluster.ClusterSvcCIDR,
-		}); err != nil {
-			return errors.NewE(err)
-		}
+	if err := d.byokClusterRepo.DeleteOne(ctx, entities.UniqueBYOKClusterFilter(ctx.AccountName, name)); err != nil {
+		return errors.NewE(err)
 	}
 
 	return nil

apps/infra/internal/domain/clusters.go

@@ -138,8 +138,7 @@ func (d *domain) CreateCluster(ctx InfraContext, cluster entities.Cluster) (*ent
 		cluster.GlobalVPN = fn.New(DefaultGlobalVPNName)
 	}
 
-	gvpn, err := d.ensureGlobalVPN(ctx, *cluster.GlobalVPN)
-	if err != nil {
+	if _, err := d.ensureGlobalVPN(ctx, *cluster.GlobalVPN); err != nil {
 		return nil, errors.NewE(err)
 	}
 
@@ -336,16 +335,12 @@ func (d *domain) CreateCluster(ctx InfraContext, cluster entities.Cluster) (*ent
 	cluster.Spec.AccountName = ctx.AccountName
 	cluster.SyncStatus = t.GenSyncStatus(t.SyncActionApply, 0)
 
-	clusterSvcCIDR, err := d.claimNextClusterSvcCIDR(ctx, cluster.Name, gvpn.Name)
+	gvpnConn, err := d.ensureGlobalVPNConnection(ctx, cluster.Name, *cluster.GlobalVPN, cluster.Spec.PublicDNSHost)
 	if err != nil {
-		return nil, err
-	}
-
-	if _, err := d.ensureGlobalVPNConnection(ctx, cluster.Name, clusterSvcCIDR, *cluster.GlobalVPN, cluster.Spec.PublicDNSHost); err != nil {
 		return nil, errors.NewE(err)
 	}
 
-	cluster.Spec.ClusterServiceCIDR = clusterSvcCIDR
+	cluster.Spec.ClusterServiceCIDR = gvpnConn.ClusterSvcCIDR
 
 	if err := d.k8sClient.ValidateObject(ctx, &cluster.Cluster); err != nil {
 		return nil, errors.NewE(err)
@@ -409,6 +404,7 @@ func (d *domain) syncKloudliteDeviceOnCluster(ctx InfraContext, gvpnName string)
 		Namespace:             accNs,
 		WgConfig:              wgConfig,
 		KubeReverseProxyImage: d.env.GlobalVPNKubeReverseProxyImage,
+		AuthzToken:            d.env.GlobalVPNKubeReverseProxyAuthzToken,
 	})
 	if err != nil {
 		return err

apps/infra/internal/domain/global-vpn-cluster-connection.go

@@ -25,25 +25,35 @@ const (
 	kloudliteGlobalVPNDeviceMethod = "kloudlite-global-vpn-device"
 )
 
-func (d *domain) getGlobalVPNConnectionPeers(vpns []*entities.GlobalVPNConnection) ([]wgv1.Peer, error) {
+func (d *domain) getGlobalVPNConnectionPeers(ctx InfraContext, vpns []*entities.GlobalVPNConnection) ([]wgv1.Peer, error) {
 	peers := make([]wgv1.Peer, 0, len(vpns))
 	for _, c := range vpns {
 		if c.ParsedWgParams != nil {
 			if c.ParsedWgParams.WgPublicKey == "" {
 				continue
 			}
 
-			if c.ParsedWgParams.NodePort == nil {
+			// if c.ParsedWgParams.NodePort == nil {
+			// 	d.logger.Infof("nodeport not available for gvpn %s", c.Name)
+			// 	continue
+			// }
+			if c.ParsedWgParams.PublicGatewayPort == nil || c.ParsedWgParams.PublicGatewayHosts == nil {
 				d.logger.Infof("nodeport not available for gvpn %s", c.Name)
 				continue
 			}
 
+			endpoint := fmt.Sprintf("%s:%s", c.ClusterPublicEndpoint, *c.ParsedWgParams.PublicGatewayPort)
+			if d.isBYOKCluster(ctx, c.ClusterName) {
+				endpoint = fmt.Sprintf("%s:%s", *c.ParsedWgParams.PublicGatewayHosts, *c.ParsedWgParams.PublicGatewayPort)
+			}
+
 			peers = append(peers, wgv1.Peer{
 				ClusterName: c.ClusterName,
 				IP:          c.ParsedWgParams.IP,
 				PublicKey:   c.ParsedWgParams.WgPublicKey,
-				Endpoint:    fmt.Sprintf("%s:%s", c.ClusterPublicEndpoint, *c.ParsedWgParams.NodePort),
-				AllowedIPs:  []string{c.ClusterSvcCIDR},
+				// Endpoint:    fmt.Sprintf("%s:%s", c.ClusterPublicEndpoint, *c.ParsedWgParams.NodePort),
+				Endpoint:   endpoint,
+				AllowedIPs: []string{c.ClusterSvcCIDR},
 			})
 		}
 	}
@@ -66,7 +76,7 @@ func (d *domain) reconGlobalVPNConnections(ctx InfraContext, vpnName string) err
 		return errors.NewE(err)
 	}
 
-	peers, err := d.getGlobalVPNConnectionPeers(vpns)
+	peers, err := d.getGlobalVPNConnectionPeers(ctx, vpns)
 	if err != nil {
 		return err
 	}
@@ -181,6 +191,13 @@ func (d *domain) createGlobalVPNConnection(ctx InfraContext, gvpnConn entities.G
 
 	gvpnConn.SyncStatus = t.GenSyncStatus(t.SyncActionApply, 0)
 
+	clusterSvcCIDR, err := d.claimNextClusterSvcCIDR(ctx, gvpnConn.ClusterName, gvpn.Name)
+	if err != nil {
+		return nil, err
+	}
+
+	gvpnConn.ClusterSvcCIDR = clusterSvcCIDR
+
 	gvpnDevice, err := d.createGlobalVPNDevice(ctx, entities.GlobalVPNDevice{
 		ObjectMeta: metav1.ObjectMeta{
 			Name: fmt.Sprintf("cluster-gateway-%s", gvpnConn.ClusterName),
@@ -211,26 +228,25 @@ func (d *domain) createGlobalVPNConnection(ctx InfraContext, gvpnConn entities.G
 }
 
 func (d *domain) deleteGlobalVPNConnection(ctx InfraContext, clusterName string, gvpnName string) error {
-	gvpnConn, err := d.gvpnConnRepo.FindOne(ctx, repos.Filter{
-		fields.AccountName:  ctx.AccountName,
-		fields.ClusterName:  clusterName,
-		fields.MetadataName: gvpnName,
-	})
+	gv, err := d.findGlobalVPNConnection(ctx, clusterName, gvpnName)
 	if err != nil {
-		return errors.NewE(err)
+		if !errors.OfType[errors.ErrNotFound](err) {
+			return errors.NewE(err)
+		}
 	}
-	if gvpnConn == nil {
-		return errors.Newf("no global vpn connection with name (%s) not found, for cluster (%s)", gvpnName, clusterName)
+
+	if err := d.deleteGlobalVPNDevice(ctx, gvpnName, gv.DeviceRef.Name); err != nil {
+		return errors.NewE(err)
 	}
 
-	if err := d.deleteGlobalVPNDevice(ctx, gvpnName, gvpnConn.DeviceRef.Name); err != nil {
+	if err := d.gvpnConnRepo.DeleteById(ctx, gv.Id); err != nil {
 		return errors.NewE(err)
 	}
 
 	return nil
 }
 
-func (d *domain) ensureGlobalVPNConnection(ctx InfraContext, clusterName string, clusterSvcCIDR string, groupName string, clusterPublicEndpoint string) (*entities.GlobalVPNConnection, error) {
+func (d *domain) ensureGlobalVPNConnection(ctx InfraContext, clusterName string, groupName string, clusterPublicEndpoint string) (*entities.GlobalVPNConnection, error) {
 	gvpn, err := d.gvpnConnRepo.FindOne(ctx, repos.Filter{
 		fields.AccountName:  ctx.AccountName,
 		fields.ClusterName:  clusterName,
@@ -269,7 +285,6 @@ func (d *domain) ensureGlobalVPNConnection(ctx InfraContext, clusterName string,
 		AccountName:           ctx.AccountName,
 		ClusterName:           clusterName,
 		ClusterPublicEndpoint: clusterPublicEndpoint,
-		ClusterSvcCIDR:        clusterSvcCIDR,
 		ParsedWgParams:        nil,
 	})
 }
@@ -326,7 +341,7 @@ func (d *domain) OnGlobalVPNConnectionDeleteMessage(ctx InfraContext, clusterNam
 		return errors.NewE(err)
 	}
 
-	if currRecord.DeviceRef.Name != "" {
+	if currRecord != nil && currRecord.DeviceRef.Name != "" {
 		if err := d.deleteGlobalVPNDevice(ctx, currRecord.GlobalVPNName, currRecord.DeviceRef.Name); err != nil {
 			return errors.NewE(err)
 		}

apps/infra/internal/domain/global-vpn-devices.go

@@ -81,6 +81,9 @@ func (d *domain) UpdateGlobalVPNDevice(ctx InfraContext, device entities.GlobalV
 func (d *domain) deleteGlobalVPNDevice(ctx InfraContext, gvpn string, deviceName string) error {
 	device, err := d.findGlobalVPNDevice(ctx, gvpn, deviceName)
 	if err != nil {
+		if errors.OfType[errors.ErrNotFound](err) {
+			return nil
+		}
 		return err
 	}
 
@@ -234,7 +237,7 @@ func (d *domain) getGlobalVPNDeviceWgConfig(ctx InfraContext, gvpn string, gvpnD
 		return "", err
 	}
 
-	gvpnConnPeers, err := d.getGlobalVPNConnectionPeers(gvpnConns)
+	gvpnConnPeers, err := d.getGlobalVPNConnectionPeers(ctx, gvpnConns)
 	if err != nil {
 		return "", err
 	}
@@ -304,7 +307,7 @@ func (d *domain) findGlobalVPNDevice(ctx InfraContext, gvpn string, gvpnDevice s
 	}
 
 	if device == nil {
-		return nil, errors.Newf("no global vpn device (name=%s) found", gvpnDevice)
+		return nil, errors.ErrNotFound{Message: fmt.Sprintf("no global vpn device with name=%s", gvpnDevice)}
 	}
 	return device, nil
 }

apps/infra/internal/domain/templates/global-vpn-kloudlite-device.yml.tpl

@@ -40,71 +40,42 @@ spec:
         "secret-ref": "{{.WgConfig | b64enc | sha256sum}}"
     spec:
       initContainers:
-        - name: init
-          image: busybox:1.32.0
-          command:
-            - sh
-            - -c
-            - sysctl -w net.ipv4.ip_forward=1 && sysctl -w net.ipv4.conf.all.forwarding=1
-          securityContext:
-            privileged: true
-            capabilities:
-              add:
-                - NET_ADMIN
-                - SYS_MODULE
-
-      containers:
         - image: linuxserver/wireguard
-          imagePullPolicy: Always
+          imagePullPolicy: IfNotPresent
           name: wg
-          resources:
-            limits:
-              cpu: 80m
-              memory: 100Mi
-            requests:
-              cpu: 50m
-              memory: 75Mi
+          {{- /* resources: */}}
+          {{- /*   limits: */}}
+          {{- /*     cpu: 80m */}}
+          {{- /*     memory: 100Mi */}}
+          {{- /*   requests: */}}
+          {{- /*     cpu: 50m */}}
+          {{- /*     memory: 75Mi */}}
           securityContext:
             capabilities:
               add:
                 - NET_ADMIN
-                - SYS_MODULE
-            privileged: true
+                {{- /* - SYS_MODULE */}}
+            {{- /* privileged: true */}}
+          command:
+            - wg-quick
+            - up
+            - wg0
           volumeMounts:
             - mountPath: /config/wg_confs/wg0.conf
               name: wg-config
               subPath: wg0.conf
           terminationMessagePath: /dev/termination-log
           terminationMessagePolicy: File
-
-        {{- /* - name: debug */}}
-        {{- /*   image: ghcr.io/kloudlite/hub/socat:latest */}}
-        {{- /*   imagePullPolicy: Always */}}
-        {{- /*   resources: */}}
-        {{- /*     limits: */}}
-        {{- /*       cpu: 100m */}}
-        {{- /*       memory: 100Mi */}}
-        {{- /*     requests: */}}
-        {{- /*       cpu: 100m */}}
-        {{- /*       memory: 100Mi
-        {{- /*   command: */}}
-        {{- /*     - sh */}}
-        {{- /*     - -c */}}
-        {{- /*     - |+ */}}
-        {{- /*       (socat -dd tcp4-listen:8080,fork,reuseaddr tcp4:kubectl-proxy.{{.Namespace}}.svc.example-test.local:8080 2>&1 | grep -iE --line-buffered 'listening|exiting') & */}}
-        {{- /*       pid=$! */}}
-        {{- /**/}}
-        {{- /*       trap "kill -9 $pid" EXIT SIGINT SIGTERM */}}
-        {{- /*       wait $pid */}}
-
+      containers:
         - name: kube-reverse-proxy
           image: {{.KubeReverseProxyImage}}
           args:
             - --addr
             - ":8080"
             - --proxy-addr
-            # this %s will be replaced with real cluster name by reverse proxy
             - {{ printf "kubectl-proxy.kloudlite.svc.{{.CLUSTER_NAME}}.local:8080" }}
+            - "--authz"
+            - {{.AuthzToken}}
           imagePullPolicy: "IfNotPresent"
           resources:
             limits:

apps/infra/internal/domain/templates/types.go

@@ -6,4 +6,5 @@ type GVPNKloudliteDeviceTemplateVars struct {
 	WgConfig  string
 
 	KubeReverseProxyImage string
+	AuthzToken            string
 }