Repository: threefoldtech/zos
Pull request #1594 (2022-02-17 13:51:09)

Peers


Patch:
diff --git a/cmds/modules/noded/stats.go b/cmds/modules/noded/stats.go
index c7f29bcf..7ceab223 100644
--- a/cmds/modules/noded/stats.go
+++ b/cmds/modules/noded/stats.go
@@ -36,44 +36,46 @@ func reportStatistics(ctx context.Context, redis string, cl zbus.Client) error {
 	if err != nil {
 		return errors.Wrap(err, "couldn't get an rmb bus instance")
 	}
-	tc := time.NewTicker(ReportInterval)
 	updateCounter := CyclesToUpdate
 	extended, err := environment.GetConfig()
 	if err != nil {
 		return err
 	}
 	for {
-		select {
-		case <-ctx.Done():
-			return nil
-		case <-tc.C:
-			if updateCounter == 0 {
-				extended, err = environment.GetConfig()
-				if err != nil {
-					log.Error().Err(err).Msg("couldn't get twins to report to")
-				}
-				updateCounter = CyclesToUpdate
+		if updateCounter == 0 {
+			extended, err = environment.GetConfig()
+			if err != nil {
+				log.Error().Err(err).Msg("couldn't get twins to report to")
 			}
-			updateCounter--
+			updateCounter = CyclesToUpdate
+		}
+		updateCounter--
+		version := stubs.NewVersionMonitorStub(cl).GetVersion(ctx2).String()
 
-			version := stubs.NewVersionMonitorStub(cl).GetVersion(ctx2).String()
+		// TODO: .Current should return error
+		current := stats.Current(ctx)
+		report := client.NodeStatus{
+			Current:    current,
+			Total:      total,
+			ZosVersion: version,
+			Hypervisor: hypervisor,
+		}
 
-			// TODO: .Current should return error
-			current := stats.Current(ctx)
-			report := client.NodeStatus{
-				Current:    current,
-				Total:      total,
-				ZosVersion: version,
-				Hypervisor: hypervisor,
-			}
-			for _, twin := range extended.Monitor {
-				go func(twinID uint32) {
-					cl := client.NewProxyClient(twinID, bus)
-					if err := sendStatisticsReport(ctx, cl, report); err != nil {
-						log.Error().Err(err).Uint32("twin", twinID).Msg("couldn't send report to twin")
-					}
-				}(twin)
-			}
+		for _, twin := range extended.Monitor {
+			go func(twinID uint32) {
+				log.Debug().Uint32("twin", twinID).Msg("sending status update to twin")
+				cl := client.NewProxyClient(twinID, bus)
+				if err := sendStatisticsReport(ctx, cl, report); err != nil {
+					log.Error().Err(err).Uint32("twin", twinID).Msg("couldn't send report to twin")
+				}
+			}(twin)
+		}
+
+		select {
+		case <-ctx.Done():
+			return nil
+		case <-time.After(ReportInterval):
+			continue
 		}
 	}
 }
diff --git a/pkg/environment/config.go b/pkg/environment/config.go
new file mode 100644
index 00000000..382a6cd4
--- /dev/null
+++ b/pkg/environment/config.go
@@ -0,0 +1,103 @@
+package environment
+
+import (
+	"encoding/json"
+	"fmt"
+	"io/ioutil"
+	"net/http"
+	"sort"
+	"strings"
+
+	"github.com/pkg/errors"
+	"github.com/rs/zerolog/log"
+)
+
+// Config is configuration set by the organization
+type Config struct {
+	// Monitor is a list of twins that need to updated continuesly
+	// with node free capacity and status.
+	Monitor   []uint32 `json:"monitor"`
+	Yggdrasil struct {
+		Peers []string `json:"peers"`
+	} `json:"yggdrasil"`
+}
+
+// Merge, updates current config with cfg merging and override config
+// based on some update rules.
+func (c *Config) Merge(cfg Config) {
+	c.Monitor = uniqueUint32(append(c.Monitor, cfg.Monitor...))
+	c.Yggdrasil.Peers = uniqueStr(append(c.Yggdrasil.Peers, cfg.Yggdrasil.Peers...))
+	// sort peers for easier comparison
+	sort.Strings(c.Yggdrasil.Peers)
+}
+
+// GetConfig returns extend config for specific run mode
+func GetConfig() (base Config, err error) {
+	env, err := Get()
+	if err != nil {
+		return
+	}
+	base, err = getConfig(env.RunningMode, BaseExtendedURL)
+	if err != nil {
+		return
+	}
+	if env.ExtendedConfigURL != "" {
+		custom, err := getConfig(env.RunningMode, env.ExtendedConfigURL)
+		if err != nil {
+			log.Error().Err(err).Msg("fetching the config from the env config-url failed")
+		}
+		base.Merge(custom)
+	}
+
+	return base, nil
+}
+
+func uniqueUint32(slice []uint32) []uint32 {
+	keys := make(map[uint32]struct{})
+	list := slice[:0]
+	for _, entry := range slice {
+		if _, exists := keys[entry]; !exists {
+			keys[entry] = struct{}{}
+			list = append(list, entry)
+		}
+	}
+	return list
+}
+
+func uniqueStr(slice []string) []string {
+	keys := make(map[string]struct{})
+	list := slice[:0]
+	for _, entry := range slice {
+		if _, exists := keys[entry]; !exists {
+			keys[entry] = struct{}{}
+			list = append(list, entry)
+		}
+	}
+	return list
+}
+
+func getConfig(run RunningMode, url string) (ext Config, err error) {
+	if !strings.HasSuffix(url, "/") {
+		url += "/"
+	}
+	u := url + fmt.Sprintf("%s.json", run)
+
+	response, err := http.Get(u)
+	if err != nil {
+		return ext, err
+	}
+
+	defer func() {
+		_, _ = ioutil.ReadAll(response.Body)
+	}()
+
+	if response.StatusCode != http.StatusOK {
+		return ext, fmt.Errorf("failed to get extended config: %s", response.Status)
+	}
+
+	if err := json.NewDecoder(response.Body).Decode(&ext); err != nil {
+		return ext, errors.Wrap(err, "failed to decode extended settings")
+	}
+
+	return
+}
diff --git a/pkg/environment/environment.go b/pkg/environment/environment.go
index b4bb756e..54d1153b 100644
--- a/pkg/environment/environment.go
+++ b/pkg/environment/environment.go
@@ -1,16 +1,10 @@
 package environment
 
 import (
-	"encoding/json"
-	"fmt"
-	"io/ioutil"
-	"net/http"
 	"os"
 	"strconv"
-	"strings"
 
 	"github.com/pkg/errors"
-	"github.com/rs/zerolog/log"
 	"github.com/threefoldtech/substrate-client"
 	"github.com/threefoldtech/zos/pkg"
 
@@ -50,13 +44,6 @@ type Environment struct {
 	ExtendedConfigURL string
 }
 
-// Extended is configuration set by the organization
-type Extended struct {
-	// Monitor is a list of twins that need to updated continuesly
-	// with node free capacity and status.
-	Monitor []uint32 `json:"monitor"`
-}
-
 // RunningMode type
 type RunningMode string
 
@@ -139,16 +126,21 @@ func (e *Environment) GetSubstrate() (*substrate.Substrate, error) {
 }
 
 func getEnvironmentFromParams(params kernel.Params) (Environment, error) {
-	var runmode []string
 	var env Environment
+	runmode := ""
+	if modes, ok := params.Get("runmode"); ok {
+		if len(modes) >= 1 {
+			runmode = modes[0]
+		}
+	} else {
+		runmode = os.Getenv("ZOS_RUNMODE")
+	}
 
-	runmode, found := params.Get("runmode")
-	if !found || len(runmode) < 1 {
-		// Fallback to default production mode
-		runmode = []string{string(RunningMain)}
+	if len(runmode) == 0 {
+		runmode = string(RunningMain)
 	}
 
-	switch RunningMode(runmode[0]) {
+	switch RunningMode(runmode) {
 	case RunningDev:
 		env = envDev
 	case RunningTest:
@@ -222,60 +214,3 @@ func getEnvironmentFromParams(params kernel.Params) (Environment, error) {
 
 	return env, nil
 }
-
-// GetConfig returns extend config for specific run mode
-func GetConfig() (ext Extended, err error) {
-	env, err := Get()
-	if err != nil {
-		return
-	}
-	ext, err = getConfig(env.RunningMode, BaseExtendedURL)
-	if err != nil {
-		return
-	}
-	if env.ExtendedConfigURL != "" {
-		config2, err := getConfig(env.RunningMode, env.ExtendedConfigURL)
-		if err != nil {
-			log.Error().Err(err).Msg("fetching the config from the env config-url failed")
-		}
-		ext.Monitor = unique(append(ext.Monitor, config2.Monitor...))
-	}
-	return ext, nil
-}
-
-func unique(intSlice []uint32) []uint32 {
-	keys := make(map[uint32]struct{})
-	list := []uint32{}
-	for _, entry := range intSlice {
-		if _, exists := keys[entry]; !exists {
-			keys[entry] = struct{}{}
-			list = append(list, entry)
-		}
-	}
-	return list
-}
-
-func getConfig(run RunningMode, url string) (ext Extended, err error) {
-	if !strings.HasSuffix(url, "/") {
-		url += "/"
-	}
-	u := url + fmt.Sprintf("%s.json", run)
-
-	response, err := http.Get(u)
-	if err != nil {
-		return ext, err
-	}
-
-	defer func() {
-		_, _ = ioutil.ReadAll(response.Body)
-	}()
-	if response.StatusCode != http.StatusOK {
-		return ext, fmt.Errorf("failed to get extended config: %s", response.Status)
-	}
-
-	if err := json.NewDecoder(response.Body).Decode(&ext); err != nil {
-		return ext, errors.Wrap(err, "failed to decode extended settings")
-	}
-
-	return
-}
diff --git a/pkg/network/yggdrasil/config.go b/pkg/network/yggdrasil/config.go
index afd2e467..dc355a07 100644
--- a/pkg/network/yggdrasil/config.go
+++ b/pkg/network/yggdrasil/config.go
@@ -40,16 +40,28 @@ func (n *NodeConfig) Address() (net.IP, error) {
 	return ip, nil
 }
 
-func (n *NodeConfig) FindPeers(ctx context.Context, top uint, filter ...Filter) error {
-	pl := fetchPeerList()
-	peersUp, err := pl.Ups(filter...)
+func (n *NodeConfig) FindPeers(ctx context.Context, filter ...Filter) error {
+	// fetching a peer list goes as this
+	// - Always include the list of peers from
+	zos, err := fetchZosYggList()
+	if err != nil {
+		return errors.Wrap(err, "failed to get zos public peer list")
+	}
+
+	zos, err = zos.Ups(filter...)
+	if err != nil {
+		return errors.Wrap(err, "failed to filter out peer list")
+	}
+
+	pub := fetchPubYggList()
+	pub, err = pub.Ups(filter...)
 	if err != nil {
 		return errors.Wrap(err, "failed to get peers list")
 	}
 
-	log.Info().Int("count", len(peersUp)).Msg("found yggdrasil up peers")
-	endpoints := make([]string, len(peersUp))
-	for i, p := range peersUp {
+	log.Info().Int("count", len(pub)).Msg("found yggdrasil up peers")
+	endpoints := make([]string, len(pub))
+	for i, p := range pub {
 		endpoints[i] = p.Endpoint
 	}
 
@@ -61,7 +73,11 @@ func (n *NodeConfig) FindPeers(ctx context.Context, top uint, filter ...Filter)
 
 	// select the best 3 public peers
 	var peers []string
-	for i := 0; i < int(top); i++ {
+	for _, peer := range zos {
+		peers = append(peers, peer.Endpoint)
+	}
+
+	for i := 0; i < 3; i++ {
 		peers = append(peers, results[i].Endpoint)
 		log.Info().Str("endpoint", results[i].Endpoint).Msg("yggdrasill public peer selected")
 	}
diff --git a/pkg/network/yggdrasil/public_peerlist.go b/pkg/network/yggdrasil/public_peerlist.go
index 2edb1a9c..b932a4e3 100644
--- a/pkg/network/yggdrasil/public_peerlist.go
+++ b/pkg/network/yggdrasil/public_peerlist.go
@@ -5,8 +5,11 @@ import (
 	"net"
 	"net/http"
 	"net/url"
+	"time"
 
+	"github.com/cenkalti/backoff"
 	"github.com/rs/zerolog/log"
+	"github.com/threefoldtech/zos/pkg/environment"
 )
 
 //PeerListFallback is an hardcoded list of public yggdrasil node
@@ -30,8 +33,6 @@ var PeerListFallback = Peers{
 type NodeInfo struct {
 	Endpoint   string `json:"-"`
 	Up         bool   `json:"up"`
-	BoxPubKey  string `json:"key"`
-	LastSeen   int    `json:"last_seen"`
 	ProtoMinor int    `json:"proto_minor"`
 }
 
@@ -72,8 +73,53 @@ func IPV4Only(ip net.IP) bool {
 	return ip.To4() != nil
 }
 
-// FetchPeerList download the list of public yggdrasil peer from https://publicpeers.neilalexander.dev/publicnodes.json
-func FetchPeerList() (Peers, error) {
+func fetchZosYggList() (Peers, error) {
+	cfg, err := environment.GetConfig()
+	if err != nil {
+		return nil, err
+	}
+
+	var peers Peers
+	for _, peer := range cfg.Yggdrasil.Peers {
+		peers = append(peers, NodeInfo{
+			Endpoint:   peer,
+			Up:         true,
+			ProtoMinor: 4,
+		})
+	}
+
+	return peers, nil
+}
+
+func fetchPubYggList() Peers {
+	// Try to fetch public peer
+	// If we failed to do so, use the fallback hardcoded peer list
+	var pl Peers
+
+	// Do not retry more than 4 times
+	bo := backoff.WithMaxRetries(backoff.NewConstantBackOff(time.Second), 4)
+
+	fetchPeerList := func() error {
+		p, err := FetchPubYggPeerList()
+		if err != nil {
+			log.Debug().Err(err).Msg("failed to fetch yggdrasil peers")
+			return err
+		}
+		pl = p
+		return nil
+	}
+
+	err := backoff.Retry(fetchPeerList, bo)
+	if err != nil {
+		log.Error().Err(err).Msg("failed to read yggdrasil public peer list online, using fallback")
+		pl = PeerListFallback
+	}
+
+	return pl
+}
+
+// FetchPubYggPeerList download the list of public yggdrasil peer from https://publicpeers.neilalexander.dev/publicnodes.json
+func FetchPubYggPeerList() (Peers, error) {
 	//pl := PeerList{}
 	pl := map[string]map[string]NodeInfo{}
 
diff --git a/pkg/network/yggdrasil/utils.go b/pkg/network/yggdrasil/utils.go
index 78143f93..4765ffa3 100644
--- a/pkg/network/yggdrasil/utils.go
+++ b/pkg/network/yggdrasil/utils.go
@@ -3,41 +3,12 @@ package yggdrasil
 import (
 	"context"
 	"crypto/ed25519"
-	"time"
 
-	"github.com/cenkalti/backoff"
 	"github.com/pkg/errors"
 	"github.com/rs/zerolog/log"
 	"github.com/threefoldtech/zos/pkg/zinit"
 )
 
-func fetchPeerList() Peers {
-	// Try to fetch public peer
-	// If we failed to do so, use the fallback hardcoded peer list
-	var pl Peers
-
-	// Do not retry more than 4 times
-	bo := backoff.WithMaxRetries(backoff.NewConstantBackOff(time.Second), 4)
-
-	fetchPeerList := func() error {
-		p, err := FetchPeerList()
-		if err != nil {
-			log.Debug().Err(err).Msg("failed to fetch yggdrasil peers")
-			return err
-		}
-		pl = p
-		return nil
-	}
-
-	err := backoff.Retry(fetchPeerList, bo)
-	if err != nil {
-		log.Error().Err(err).Msg("failed to read yggdrasil public peer list online, using fallback")
-		pl = PeerListFallback
-	}
-
-	return pl
-}
-
 func EnsureYggdrasil(ctx context.Context, privateKey ed25519.PrivateKey, ns YggdrasilNamespace) (*YggServer, error) {
 	// Filter out all the nodes from the same
 	// segment so we do not just connect locally
@@ -52,12 +23,12 @@ func EnsureYggdrasil(ctx context.Context, privateKey ed25519.PrivateKey, ns Yggd
 			ranges = append(ranges, ip)
 		}
 	}
-
+	log.Info().Msgf("filtering out peers from ranges: %s", ranges)
 	filter := Exclude(ranges)
 	z := zinit.Default()
 
 	cfg := GenerateConfig(privateKey)
-	if err := cfg.FindPeers(ctx, 4, filter); err != nil {
+	if err := cfg.FindPeers(ctx, filter); err != nil {
 		return nil, err
 	}
 


Test patch:
diff --git a/pkg/environment/config_test.go b/pkg/environment/config_test.go
new file mode 100644
index 00000000..bdd99120
--- /dev/null
+++ b/pkg/environment/config_test.go
@@ -0,0 +1,35 @@
+package environment
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestConfigMerge(t *testing.T) {
+	var cfg Config
+	cfg.Monitor = []uint32{0, 1}
+	cfg.Yggdrasil.Peers = []string{"a", "b"}
+
+	var ex Config
+	ex.Monitor = []uint32{1, 2}
+	ex.Yggdrasil.Peers = []string{"c", "b"}
+
+	cfg.Merge(ex)
+
+	require.Equal(t, []string{"a", "b", "c"}, cfg.Yggdrasil.Peers)
+	require.Len(t, cfg.Monitor, 3)
+}
+
+func TestConfigMergeEmpty(t *testing.T) {
+	var cfg Config
+	cfg.Monitor = []uint32{0, 1}
+	cfg.Yggdrasil.Peers = []string{"a", "b"}
+
+	var ex Config
+
+	cfg.Merge(ex)
+
+	require.Equal(t, []string{"a", "b"}, cfg.Yggdrasil.Peers)
+	require.Len(t, cfg.Monitor, 2)
+}
diff --git a/pkg/network/latency/latency_test.go b/pkg/network/latency/latency_test.go
index e71b0777..87b369fb 100644
--- a/pkg/network/latency/latency_test.go
+++ b/pkg/network/latency/latency_test.go
@@ -34,7 +34,7 @@ func TestLatencySorter(t *testing.T) {
 }
 
 func TestYggPeering(t *testing.T) {
-	pl, err := yggdrasil.FetchPeerList()
+	pl, err := yggdrasil.FetchPubYggPeerList()
 	require.NoError(t, err)
 
 	peersUp, err := pl.Ups()
diff --git a/pkg/network/yggdrasil/public_peerlist_test.go b/pkg/network/yggdrasil/public_peerlist_test.go
index 0a2cd01e..d141e0d3 100644
--- a/pkg/network/yggdrasil/public_peerlist_test.go
+++ b/pkg/network/yggdrasil/public_peerlist_test.go
@@ -8,7 +8,7 @@ import (
 )
 
 func TestFetchPeerList(t *testing.T) {
-	pl, err := FetchPeerList()
+	pl, err := FetchPubYggPeerList()
 	require.NoError(t, err)
 	peers, err := pl.Ups()
 	require.NoError(t, err)
@@ -18,7 +18,7 @@ func TestFetchPeerList(t *testing.T) {
 }
 
 func TestUps(t *testing.T) {
-	pl, err := FetchPeerList()
+	pl, err := FetchPubYggPeerList()
 	require.NoError(t, err)
 
 	peers, err := pl.Ups()