nodes now publish heartbeats (#181)

* nodes now publish heartbeats

* removing superfluous field

control-api/events.go

@@ -5,6 +5,7 @@ const (
 	AgentStoppedEventType    = "agent_stopped"
 	NodeStartedEventType     = "node_started"
 	NodeStoppedEventType     = "node_stopped"
+	HeartbeatEventType       = "heartbeat"
 	WorkloadStartedEventType = "workload_started" // FIXME-- should this be WorkloadDeployed?
 	WorkloadStoppedEventType = "workload_stopped" // FIXME-- should this be in addition to WorkloadUndeployed (likely yes, in case of something bad happening...)
 	// FIXME-- where is WorkloadDeployedEventType? (likely just need to rename WorkloadStartedEventType -> WorkloadDeployedEventType)
@@ -40,3 +41,11 @@ type NodeStoppedEvent struct {
 	Id       string `json:"id"`
 	Graceful bool   `json:"graceful"`
 }
+
+type HeartbeatEvent struct {
+	Version         string            `json:"version"`
+	NodeId          string            `json:"node_id"`
+	Uptime          string            `json:"uptime"`
+	Tags            map[string]string `json:"tags,omitempty"`
+	RunningMachines int               `json:"running_machines"`
+}

internal/node/node.go

@@ -24,11 +24,14 @@ import (
 	"github.com/synadia-io/nex/internal/node/observability"
 )
 
-const defaultNatsStoreDir = "pnats"
-const defaultPidFilepath = "/var/run/nex.pid"
-
-const runloopSleepInterval = 100 * time.Millisecond
-const runloopTickInterval = 2500 * time.Millisecond
+const (
+	systemNamespace      = "system"
+	defaultNatsStoreDir  = "pnats"
+	defaultPidFilepath   = "/var/run/nex.pid"
+	runloopSleepInterval = 100 * time.Millisecond
+	runloopTickInterval  = 2500 * time.Millisecond
+	heartbeatInterval    = 30 * time.Second
+)
 
 // Nex node process
 type Node struct {
@@ -109,6 +112,8 @@ func (n *Node) Start() {
 
 	_ = n.publishNodeStarted()
 
+	go n.emitHeartbeats()
+
 	timer := time.NewTicker(runloopTickInterval)
 	defer timer.Stop()
 
@@ -173,6 +178,17 @@ func (n *Node) createPid() error {
 	return nil
 }
 
+func (n *Node) emitHeartbeats() {
+	ticker := time.NewTicker(heartbeatInterval)
+	for range ticker.C {
+		n.publishHeartbeat()
+
+		if n.closing > 0 {
+			ticker.Stop()
+		}
+	}
+}
+
 func (n *Node) generateKeypair() error {
 	var err error
 
@@ -318,6 +334,34 @@ func (n *Node) loadNodeConfig() error {
 	return nil
 }
 
+func (n *Node) publishHeartbeat() error {
+	machines, err := n.manager.RunningWorkloads()
+	if err != nil {
+		n.log.Error("Failed to query running machines during heartbeat", slog.Any("error", err))
+		return nil
+	}
+
+	now := time.Now().UTC()
+
+	evt := controlapi.HeartbeatEvent{
+		NodeId:          n.publicKey,
+		Version:         Version(),
+		Uptime:          myUptime(now.Sub(n.startedAt)),
+		RunningMachines: len(machines),
+		Tags:            n.config.Tags,
+	}
+
+	cloudevent := cloudevents.NewEvent()
+	cloudevent.SetSource(n.publicKey)
+	cloudevent.SetID(uuid.NewString())
+	cloudevent.SetTime(now)
+	cloudevent.SetType(controlapi.HeartbeatEventType)
+	cloudevent.SetDataContentType(cloudevents.ApplicationJSON)
+	_ = cloudevent.SetData(evt)
+
+	return PublishCloudEvent(n.nc, systemNamespace, cloudevent, n.log)
+}
+
 func (n *Node) publishNodeStarted() error {
 	nodeStart := controlapi.NodeStartedEvent{
 		Version: VERSION,