Adding sensors and load balancing engine #393

Author: samyfodil

cli/node/start.go

@@ -13,6 +13,7 @@ import (
 	httpService "github.com/taubyte/tau/pkg/http"
 	auto "github.com/taubyte/tau/pkg/http-auto"
 	"github.com/taubyte/tau/pkg/kvdb"
+	"github.com/taubyte/tau/pkg/sensors"
 	commonSpecs "github.com/taubyte/tau/pkg/specs/common"
 	slices "github.com/taubyte/tau/utils/slices/string"
 )
@@ -37,6 +38,11 @@ func Start(ctx context.Context, serviceConfig *config.Node) error {
 		return err
 	}
 
+	// start sensors service
+	if serviceConfig.Sensors, err = sensors.New(serviceConfig.Node); err != nil {
+		return fmt.Errorf("new sensors service failed with: %s", err)
+	}
+
 	serviceConfig.Databases = kvdb.New(serviceConfig.Node)
 
 	// Create httpNode if needed

clients/p2p/seer/client.go

@@ -12,13 +12,16 @@ import (
 	streamClient "github.com/taubyte/tau/p2p/streams/client"
 	"github.com/taubyte/tau/p2p/streams/command"
 	"github.com/taubyte/tau/p2p/streams/command/response"
+	"github.com/taubyte/tau/pkg/sensors"
 	"github.com/taubyte/tau/utils/maps"
 
 	"github.com/taubyte/tau/services/common"
 )
 
-func New(ctx context.Context, node peer.Node) (client iface.Client, err error) {
-	c := &Client{}
+func New(ctx context.Context, node peer.Node, registry *sensors.Registry) (client iface.Client, err error) {
+	c := &Client{
+		sensors: registry,
+	}
 	c.client, err = streamClient.New(node, common.SeerProtocol)
 	if err != nil {
 		logger.Error("API client creation failed: %s", err)

clients/p2p/seer/dream/init.go

@@ -15,5 +15,5 @@ func init() {
 }
 
 func createSeerClient(node peer.Node, config *common.ClientConfig) (common.Client, error) {
-	return seer.New(node.Context(), node)
+	return seer.New(node.Context(), node, nil)
 }

clients/p2p/seer/heartbeat.go

@@ -12,6 +12,8 @@ import (
 )
 
 func (u *Usage) Heartbeat(usage *iface.UsageData, hostname, nodeId, clientNodeId string, signature []byte) (response.Response, error) {
+	logger.Debug("Heartbeat", "sending usage", usage)
+
 	usageData, err := cbor.Marshal(usage)
 	if err != nil {
 		return nil, err
@@ -21,6 +23,7 @@ func (u *Usage) Heartbeat(usage *iface.UsageData, hostname, nodeId, clientNodeId
 	if err != nil {
 		return nil, fmt.Errorf("calling heartbeat send failed with: %w", err)
 	}
+
 	return resp, nil
 }
 

clients/p2p/seer/types.go

@@ -6,13 +6,15 @@ import (
 	peerCore "github.com/libp2p/go-libp2p/core/peer"
 	iface "github.com/taubyte/tau/core/services/seer"
 	client "github.com/taubyte/tau/p2p/streams/client"
+	"github.com/taubyte/tau/pkg/sensors"
 )
 
 var _ iface.Client = &Client{}
 
 type Client struct {
 	client   *client.Client
 	services iface.Services
+	sensors  *sensors.Registry
 	peers    []peerCore.ID
 }
 

clients/p2p/seer/usage/all_usage.go

@@ -26,9 +26,10 @@ func GetUsage() (usage iface.UsageData, err error) {
 	}
 
 	usage = iface.UsageData{
-		Memory: memory,
-		Cpu:    cpu,
-		Disk:   disk,
+		Memory:       memory,
+		Cpu:          cpu,
+		Disk:         disk,
+		CustomValues: make(map[string]float64),
 	}
 
 	return

clients/p2p/seer/usage_client.go

@@ -49,6 +49,13 @@ func (u *Usage) updateUsage(hostname, nodeId, clientNodeId string, signature []b
 		return nil, fmt.Errorf("getting usage of hostname `%s` failed with: %s", hostname, err)
 	}
 
+	// get custom values
+	if u.sensors != nil {
+		for _, sensor := range u.sensors.List() {
+			usageData.CustomValues[sensor.Name] = sensor.Value
+		}
+	}
+
 	resp, err := u.Heartbeat(&usageData, hostname, nodeId, clientNodeId, signature)
 	if err != nil {
 		return nil, err

config/services.go

@@ -11,6 +11,7 @@ import (
 	seerIface "github.com/taubyte/tau/core/services/seer"
 	"github.com/taubyte/tau/p2p/peer"
 	http "github.com/taubyte/tau/pkg/http"
+	"github.com/taubyte/tau/pkg/sensors"
 )
 
 var (
@@ -56,6 +57,8 @@ type Node struct {
 
 	Http http.Service
 
+	Sensors *sensors.Service
+
 	EnableHTTPS bool
 	Verbose     bool
 	DevMode     bool
@@ -69,6 +72,13 @@ type DomainValidation struct {
 	PublicKey  []byte
 }
 
+func (config *Node) SensorsRegistry() *sensors.Registry {
+	if config.Sensors != nil {
+		return config.Sensors.Registry()
+	}
+	return nil
+}
+
 func (config *Node) Validate() error {
 	if config == nil {
 		config = &Node{}

core/services/seer/types.go

@@ -29,6 +29,7 @@ type UsageReturn struct {
 	FreeDisk      int
 	UsedDisk      int
 	AvailableDisk int
+	CustomValues  map[string]float64
 }
 
 type Cpu struct {
@@ -48,9 +49,10 @@ type Cpu struct {
 }
 
 type UsageData struct {
-	Memory Memory `cbor:"3,keyasint"`
-	Cpu    Cpu    `cbor:"5,keyasint"`
-	Disk   Disk   `cbor:"7,keyasint"`
+	Memory       Memory             `cbor:"3,keyasint"`
+	Cpu          Cpu                `cbor:"5,keyasint"`
+	Disk         Disk               `cbor:"7,keyasint"`
+	CustomValues map[string]float64 `cbor:"9,keyasint,omitempty"`
 }
 
 type Memory struct {

core/services/seer/usage_map.go

@@ -0,0 +1,80 @@
+package seer
+
+import "strings"
+
+func (u *UsageData) ToMap() map[string]any {
+	result := map[string]any{
+		"memory": map[string]any{
+			"used":  u.Memory.Used,
+			"total": u.Memory.Total,
+			"free":  u.Memory.Free,
+		},
+		"cpu": map[string]any{
+			"total":     u.Cpu.Total,
+			"count":     u.Cpu.Count,
+			"user":      u.Cpu.User,
+			"nice":      u.Cpu.Nice,
+			"system":    u.Cpu.System,
+			"idle":      u.Cpu.Idle,
+			"iowait":    u.Cpu.Iowait,
+			"irq":       u.Cpu.Irq,
+			"softirq":   u.Cpu.Softirq,
+			"steal":     u.Cpu.Steal,
+			"guest":     u.Cpu.Guest,
+			"guestNice": u.Cpu.GuestNice,
+			"statCount": u.Cpu.StatCount,
+		},
+		"disk": map[string]any{
+			"total":     u.Disk.Total,
+			"free":      u.Disk.Free,
+			"used":      u.Disk.Used,
+			"available": u.Disk.Available,
+		},
+	}
+
+	custom := convertCustomValuesToNestedMap(u.CustomValues)
+	if custom != nil {
+		result["custom"] = custom
+	}
+
+	return result
+}
+
+/*
+convertCustomValuesToNestedMap converts a flat map with dot-separated keys
+into a nested map structure where '.' is treated as a hierarchical separator.
+For example: {"sensor.network.bytes": 123.45} becomes
+{"sensor": {"network": {"bytes": 123.45}}}
+*/
+func convertCustomValuesToNestedMap(customValues map[string]float64) map[string]any {
+	if len(customValues) == 0 {
+		return nil
+	}
+
+	result := make(map[string]any)
+	for key, value := range customValues {
+		parts := strings.Split(key, ".")
+		cur := result
+		for i, part := range parts {
+			if i == len(parts)-1 {
+				cur[part] = value
+			} else {
+				if next, ok := cur[part]; ok {
+					if nextMap, ok := next.(map[string]any); ok {
+						cur = nextMap
+					} else {
+						newMap := make(map[string]any)
+						cur[part] = newMap
+						cur = newMap
+					}
+				} else {
+					newMap := make(map[string]any)
+					cur[part] = newMap
+					cur = newMap
+				}
+			}
+		}
+	}
+
+	return result
+}