Merge pull request #22 from threefoldtech/close-open-sockets

fix too many open connections

Author: Eslam-Nawara

cmds/identityd/ssh.go

@@ -75,6 +75,12 @@ func manageSSHKeys() error {
 				return fmt.Errorf("failed to fetch user keys: %+w", err)
 			}
 
+			defer func() {
+				if res != nil && res.Body != nil {
+					res.Body.Close()
+				}
+			}()
+
 			if res.StatusCode == http.StatusNotFound {
 				return backoff.Permanent(fmt.Errorf("failed to get user keys for user (%s): keys not found", user))
 			}

cmds/modules/api_gateway/main.go

@@ -3,10 +3,10 @@ package apigateway
 import (
 	"context"
 	"crypto/ed25519"
-	"encoding/hex"
+	// "encoding/hex"
 	"fmt"
 
-	"github.com/cenkalti/backoff/v3"
+	// "github.com/cenkalti/backoff/v3"
 	"github.com/rs/zerolog/log"
 	"github.com/threefoldtech/tfgrid-sdk-go/rmb-sdk-go/peer"
 	"github.com/threefoldtech/zbus"
@@ -49,20 +49,22 @@ func action(cli *cli.Context) error {
 	if err != nil {
 		return fmt.Errorf("fail to connect to message broker server: %w", err)
 	}
+
 	redis, err := zbus.NewRedisClient(msgBrokerCon)
 	if err != nil {
 		return fmt.Errorf("fail to connect to message broker server: %w", err)
 	}
+
 	idStub := stubs.NewIdentityManagerStub(redis)
 
 	sk := ed25519.PrivateKey(idStub.PrivateKey(cli.Context))
 	pubKey := sk.Public().(ed25519.PublicKey)
 	log.Info().Str("public key", string(pubKey)).Msg("node public key")
 
-	manager, err := environment.GetSubstrate()
-	if err != nil {
-		return fmt.Errorf("failed to create substrate manager: %w", err)
-	}
+	// manager, err := environment.GetSubstrate()
+	// if err != nil {
+	// 	return fmt.Errorf("failed to create substrate manager: %w", err)
+	// }
 
 	router := peer.NewRouter()
 	gw, err := registrar.NewRegistrarGateway(cli.Context, redis)
@@ -99,24 +101,25 @@ func action(cli *cli.Context) error {
 	}
 	api.SetupRoutes(router)
 
-	bo := backoff.NewExponentialBackOff()
-	bo.MaxElapsedTime = 0
-	backoff.Retry(func() error {
-		_, err = peer.NewPeer(
-			ctx,
-			hex.EncodeToString(sk.Seed()),
-			manager,
-			router.Serve,
-			peer.WithKeyType(peer.KeyTypeEd25519),
-			peer.WithRelay(environment.GetRelaysURLs()...),
-			peer.WithInMemoryExpiration(6*60*60), // 6 hours
-		)
-		if err != nil {
-			return fmt.Errorf("failed to start a new rmb peer: %w", err)
-		}
-
-		return nil
-	}, bo)
+	// bo := backoff.NewExponentialBackOff()
+	// bo.MaxElapsedTime = 0
+	// backoff.Retry(func() error {
+	// 	_, err = peer.NewPeer(
+	// 		ctx,
+	// 		hex.EncodeToString(sk.Seed()),
+	// 		manager,
+	// 		router.Serve,
+	// 		peer.WithKeyType(peer.KeyTypeEd25519),
+	// 		peer.WithRelay(environment.GetRelaysURLs()...),
+	// 		peer.WithInMemoryExpiration(6*60*60), // 6 hours
+	// 	)
+	// 	if err != nil {
+	// 		log.Error().Err(err).Msg("faling to start api-gateway, trying over and over again")
+	// 		return fmt.Errorf("failed to start a new rmb peer: %w", err)
+	// 	}
+	//
+	// 	return nil
+	// }, bo)
 
 	log.Info().
 		Str("broker", msgBrokerCon).

cmds/modules/vmd/main.go

@@ -47,7 +47,6 @@ var Module cli.Command = cli.Command{
 // copy files from src dir to dst dir works at one level only for
 // our specific use case
 func copyDepth1(src, dst string) error {
-
 	return filepath.Walk(src, func(path string, info fs.FileInfo, err error) error {
 		if err != nil {
 			return err
@@ -62,11 +61,15 @@ func copyDepth1(src, dst string) error {
 		if err != nil {
 			return err
 		}
+		defer srcF.Close()
+
 		dstPath := filepath.Join(dst, filepath.Base(path))
 		dstF, err := os.Create(dstPath)
 		if err != nil {
 			return err
 		}
+		defer dstF.Close()
+
 		_, err = io.Copy(dstF, srcF)
 		return err
 	})
@@ -92,7 +95,7 @@ func action(cli *cli.Context) error {
 		return errors.Wrap(err, "fail to connect to message broker server")
 	}
 
-	config, err := cache.VolatileDir(module, 50*1024*1024) //50mb volatile directory
+	config, err := cache.VolatileDir(module, 50*1024*1024) // 50mb volatile directory
 	if err != nil && !os.IsExist(err) {
 		return errors.Wrap(err, "failed to create vmd volatile storage")
 	}

go.mod

@@ -42,7 +42,7 @@ require (
 	github.com/threefoldtech/tfgrid-sdk-go/rmb-sdk-go v0.16.1-0.20241229121208-76ac3fea5e67
 	github.com/threefoldtech/tfgrid4-sdk-go/node-registrar v0.0.0-20250506110945-e1591751c92f
 	github.com/threefoldtech/zbus v1.0.1
-	github.com/threefoldtech/zosbase v0.1.9
+	github.com/threefoldtech/zosbase v0.1.10
 	github.com/urfave/cli/v2 v2.17.2-0.20221006022127-8f469abc00aa
 	github.com/vishvananda/netlink v1.1.1-0.20201029203352-d40f9887b852
 	github.com/vishvananda/netns v0.0.0-20210104183010-2eb08e3e575f // indirect

go.sum

@@ -573,8 +573,8 @@ github.com/threefoldtech/tfgrid4-sdk-go/node-registrar v0.0.0-20250506110945-e15
 github.com/threefoldtech/tfgrid4-sdk-go/node-registrar v0.0.0-20250506110945-e1591751c92f/go.mod h1:5p+3dprweeEymBRNCB8b8cdGmP36g9LGtf2wwqyqvFk=
 github.com/threefoldtech/zbus v1.0.1 h1:3KaEpyOiDYAw+lrAyoQUGIvY9BcjVRXlQ1beBRqhRNk=
 github.com/threefoldtech/zbus v1.0.1/go.mod h1:E/v/xEvG/l6z/Oj0aDkuSUXFm/1RVJkhKBwDTAIdsHo=
-github.com/threefoldtech/zosbase v0.1.9 h1:S0rB8sET5Afn/GIilU1DPLPf6uzC5sf0Ogs60FkQixY=
-github.com/threefoldtech/zosbase v0.1.9/go.mod h1:PzZ9jW1lYFgA0/F4vStP/6CIhQsCdD7DTrum3AYiAWA=
+github.com/threefoldtech/zosbase v0.1.10 h1:wRm0KLIjNUmfp92ZU/0xax/SbcFVtMIbmiHSAqFdX/w=
+github.com/threefoldtech/zosbase v0.1.10/go.mod h1:PzZ9jW1lYFgA0/F4vStP/6CIhQsCdD7DTrum3AYiAWA=
 github.com/tinylib/msgp v1.1.5 h1:2gXmtWueD2HefZHQe1QOy9HVzmFrLOVvsXwXBQ0ayy0=
 github.com/tinylib/msgp v1.1.5/go.mod h1:eQsjooMTnV42mHu917E26IogZ2930nFyBQdofk10Udg=
 github.com/tklauser/go-sysconf v0.3.12 h1:0QaGUFOdQaIVdPgfITYzaTegZvdCjmYO52cSFAEVmqU=

tools/zos-update-worker/internal/worker_test.go

@@ -41,10 +41,11 @@ func TestWorker(t *testing.T) {
 	})
 
 	t.Run("test_no_src_test", func(t *testing.T) {
-		_, err := os.Create(src + "/zos:v3.4.0-qa1.flist")
+		f, err := os.Create(src + "/zos:v3.4.0-qa1.flist")
 		if err != nil {
 			t.Error(err)
 		}
+		defer f.Close()
 
 		err = worker.updateZosVersion("testing", worker.clients["testing"])
 		if err == nil {
@@ -53,10 +54,11 @@ func TestWorker(t *testing.T) {
 	})
 
 	t.Run("test_no_src_main", func(t *testing.T) {
-		_, err = os.Create(src + "/zos:v3.1.1-rc2.flist")
+		f, err := os.Create(src + "/zos:v3.1.1-rc2.flist")
 		if err != nil {
 			t.Error(err)
 		}
+		defer f.Close()
 
 		err = worker.updateZosVersion("production", worker.clients["production"])
 		if err == nil {