fix: remove Liveness and Acceptance from NodeInfo (#3785)

- fixes #3783 - Introduces NodeState type used to track NodeInfo, Connection, and Membershio. Removes the idea of Connection and Membership from data sent by compute nodes to the Requester(s) since compute nodes should not influence their Connection state or mmembership. Those are values related to the Requesters view of the network. --------- Co-authored-by: frrist <[email protected]>
bacalhau-project · Apr 18, 2024 · 3b3d8a8 · 3b3d8a8
1 parent f453e32
commit 3b3d8a8
Show file tree

Hide file tree

Showing 54 changed files with 711 additions and 913 deletions.
diff --git a/cmd/cli/agent/node.go b/cmd/cli/agent/node.go
@@ -43,7 +43,7 @@ func (o *NodeOptions) runNode(cmd *cobra.Command, _ []string) error {
 		return fmt.Errorf("could not get server node: %w", err)
 	}
 
-	writeErr := output.OutputOneNonTabular(cmd, o.OutputOpts, response.NodeInfo)
+	writeErr := output.OutputOneNonTabular(cmd, o.OutputOpts, response.NodeState)
 	if writeErr != nil {
 		return fmt.Errorf("failed to write node: %w", writeErr)
 	}

diff --git a/cmd/cli/agent/node_test.go b/cmd/cli/agent/node_test.go
@@ -5,9 +5,10 @@ package agent_test
 import (
 	"testing"
 
+	"github.com/stretchr/testify/suite"
+
 	"github.com/bacalhau-project/bacalhau/pkg/lib/marshaller"
 	"github.com/bacalhau-project/bacalhau/pkg/models"
-	"github.com/stretchr/testify/suite"
 
 	cmdtesting "github.com/bacalhau-project/bacalhau/cmd/testing"
 	"github.com/bacalhau-project/bacalhau/cmd/util/output"
@@ -25,18 +26,18 @@ func (s *NodeSuite) TestNodeJSONOutput() {
 	_, out, err := s.ExecuteTestCobraCommand("agent", "node", "--output", string(output.JSONFormat))
 	s.Require().NoError(err, "Could not request node with json output.")
 
-	nodeInfo := &models.NodeInfo{}
+	nodeInfo := &models.NodeState{}
 	err = marshaller.JSONUnmarshalWithMax([]byte(out), &nodeInfo)
 	s.Require().NoError(err, "Could not unmarshall the output into json - %+v", err)
-	s.Require().Equal(s.Node.ID, nodeInfo.ID(), "Node ID does not match in json.")
+	s.Require().Equal(s.Node.ID, nodeInfo.Info.ID(), "Node ID does not match in json.")
 }
 
 func (s *NodeSuite) TestNodeYAMLOutput() {
 	_, out, err := s.ExecuteTestCobraCommand("agent", "node")
 	s.Require().NoError(err, "Could not request node with yaml output.")
 
-	nodeInfo := &models.NodeInfo{}
+	nodeInfo := &models.NodeState{}
 	err = marshaller.YAMLUnmarshalWithMax([]byte(out), &nodeInfo)
 	s.Require().NoError(err, "Could not unmarshall the output into yaml - %+v", err)
-	s.Require().Equal(s.Node.ID, nodeInfo.ID(), "Node ID does not match in yaml.")
+	s.Require().Equal(s.Node.ID, nodeInfo.Info.ID(), "Node ID does not match in yaml.")
 }
diff --git a/cmd/cli/job/describe.go b/cmd/cli/job/describe.go
@@ -6,15 +6,16 @@ import (
 	"slices"
 	"time"
 
-	"github.com/bacalhau-project/bacalhau/pkg/lib/collections"
-	"github.com/bacalhau-project/bacalhau/pkg/models"
-	"github.com/bacalhau-project/bacalhau/pkg/util/idgen"
 	"github.com/jedib0t/go-pretty/v6/table"
 	"github.com/jedib0t/go-pretty/v6/text"
 	"github.com/samber/lo"
 	"github.com/spf13/cobra"
 	"k8s.io/kubectl/pkg/util/i18n"
 
+	"github.com/bacalhau-project/bacalhau/pkg/lib/collections"
+	"github.com/bacalhau-project/bacalhau/pkg/models"
+	"github.com/bacalhau-project/bacalhau/pkg/util/idgen"
+
 	"github.com/bacalhau-project/bacalhau/cmd/util"
 	"github.com/bacalhau-project/bacalhau/cmd/util/flags/cliflags"
 	"github.com/bacalhau-project/bacalhau/cmd/util/output"

diff --git a/cmd/cli/node/columns.go b/cmd/cli/node/columns.go
@@ -4,48 +4,49 @@ import (
 	"fmt"
 	"strings"
 
-	"github.com/bacalhau-project/bacalhau/cmd/util/output"
-	"github.com/bacalhau-project/bacalhau/pkg/model"
-	"github.com/bacalhau-project/bacalhau/pkg/models"
-	"github.com/bacalhau-project/bacalhau/pkg/util/idgen"
 	"github.com/c2h5oh/datasize"
 	"github.com/jedib0t/go-pretty/v6/table"
 	"github.com/jedib0t/go-pretty/v6/text"
 	"github.com/samber/lo"
 	"golang.org/x/exp/slices"
+
+	"github.com/bacalhau-project/bacalhau/cmd/util/output"
+	"github.com/bacalhau-project/bacalhau/pkg/model"
+	"github.com/bacalhau-project/bacalhau/pkg/models"
+	"github.com/bacalhau-project/bacalhau/pkg/util/idgen"
 )
 
-var alwaysColumns = []output.TableColumn[*models.NodeInfo]{
+var alwaysColumns = []output.TableColumn[*models.NodeState]{
 	{
 		ColumnConfig: table.ColumnConfig{Name: "id"},
-		Value:        func(node *models.NodeInfo) string { return idgen.ShortNodeID(node.ID()) },
+		Value:        func(node *models.NodeState) string { return idgen.ShortNodeID(node.Info.ID()) },
 	},
 	{
 		ColumnConfig: table.ColumnConfig{Name: "type"},
-		Value:        func(ni *models.NodeInfo) string { return ni.NodeType.String() },
+		Value:        func(ni *models.NodeState) string { return ni.Info.NodeType.String() },
 	},
 	{
 		ColumnConfig: table.ColumnConfig{Name: "approval"},
-		Value:        func(ni *models.NodeInfo) string { return ni.Approval.String() },
+		Value:        func(ni *models.NodeState) string { return ni.Membership.String() },
 	},
 	{
 		ColumnConfig: table.ColumnConfig{Name: "status"},
-		Value: func(ni *models.NodeInfo) string {
-			if ni.ComputeNodeInfo != nil {
-				return ni.State.String()
+		Value: func(ni *models.NodeState) string {
+			if ni.Info.ComputeNodeInfo != nil {
+				return ni.Connection.String()
 			}
 
 			return "" // nothing for requester nodes
 		},
 	},
 }
 
-var toggleColumns = map[string][]output.TableColumn[*models.NodeInfo]{
+var toggleColumns = map[string][]output.TableColumn[*models.NodeState]{
 	"labels": {
 		{
 			ColumnConfig: table.ColumnConfig{Name: "labels", WidthMax: 50, WidthMaxEnforcer: text.WrapSoft},
-			Value: func(ni *models.NodeInfo) string {
-				labels := lo.MapToSlice(ni.Labels, func(key, val string) string { return fmt.Sprintf("%s=%s", key, val) })
+			Value: func(ni *models.NodeState) string {
+				labels := lo.MapToSlice(ni.Info.Labels, func(key, val string) string { return fmt.Sprintf("%s=%s", key, val) })
 				slices.Sort(labels)
 				return strings.Join(labels, " ")
 			},
@@ -54,20 +55,20 @@ var toggleColumns = map[string][]output.TableColumn[*models.NodeInfo]{
 	"version": {
 		{
 			ColumnConfig: table.ColumnConfig{Name: "version"},
-			Value: func(ni *models.NodeInfo) string {
-				return ni.BacalhauVersion.GitVersion
+			Value: func(ni *models.NodeState) string {
+				return ni.Info.BacalhauVersion.GitVersion
 			},
 		},
 		{
 			ColumnConfig: table.ColumnConfig{Name: "architecture"},
-			Value: func(ni *models.NodeInfo) string {
-				return ni.BacalhauVersion.GOARCH
+			Value: func(ni *models.NodeState) string {
+				return ni.Info.BacalhauVersion.GOARCH
 			},
 		},
 		{
 			ColumnConfig: table.ColumnConfig{Name: "os"},
-			Value: func(ni *models.NodeInfo) string {
-				return ni.BacalhauVersion.GOOS
+			Value: func(ni *models.NodeState) string {
+				return ni.Info.BacalhauVersion.GOOS
 			},
 		},
 	},
@@ -123,11 +124,11 @@ func maxLen(val []string) int {
 	return lo.Max(lo.Map[string, int](val, func(item string, index int) int { return len(item) })) + 1
 }
 
-func ifComputeNode(getFromCNInfo func(*models.ComputeNodeInfo) string) func(*models.NodeInfo) string {
-	return func(ni *models.NodeInfo) string {
-		if ni.ComputeNodeInfo == nil {
+func ifComputeNode(getFromCNInfo func(*models.ComputeNodeInfo) string) func(state *models.NodeState) string {
+	return func(ni *models.NodeState) string {
+		if ni.Info.ComputeNodeInfo == nil {
 			return ""
 		}
-		return getFromCNInfo(ni.ComputeNodeInfo)
+		return getFromCNInfo(ni.Info.ComputeNodeInfo)
 	}
 }
diff --git a/cmd/cli/serve/util.go b/cmd/cli/serve/util.go
@@ -7,16 +7,17 @@ import (
 	"path/filepath"
 	"time"
 
+	pkgerrors "github.com/pkg/errors"
+	"github.com/rs/zerolog/log"
+	"github.com/samber/lo"
+	"github.com/spf13/viper"
+
 	"github.com/bacalhau-project/bacalhau/pkg/compute/store"
 	"github.com/bacalhau-project/bacalhau/pkg/compute/store/boltdb"
 	"github.com/bacalhau-project/bacalhau/pkg/jobstore"
 	boltjobstore "github.com/bacalhau-project/bacalhau/pkg/jobstore/boltdb"
 	"github.com/bacalhau-project/bacalhau/pkg/models"
 	"github.com/bacalhau-project/bacalhau/pkg/util/idgen"
-	pkgerrors "github.com/pkg/errors"
-	"github.com/rs/zerolog/log"
-	"github.com/samber/lo"
-	"github.com/spf13/viper"
 
 	"github.com/bacalhau-project/bacalhau/cmd/util/flags/configflags"
 	"github.com/bacalhau-project/bacalhau/pkg/orchestrator/transformer"
@@ -127,9 +128,9 @@ func GetRequesterConfig(ctx context.Context, createJobStore bool) (node.Requeste
 	}
 
 	if cfg.ManualNodeApproval {
-		requesterConfig.DefaultApprovalState = models.NodeApprovals.PENDING
+		requesterConfig.DefaultApprovalState = models.NodeMembership.PENDING
 	} else {
-		requesterConfig.DefaultApprovalState = models.NodeApprovals.APPROVED
+		requesterConfig.DefaultApprovalState = models.NodeMembership.APPROVED
 	}
 
 	return requesterConfig, nil

diff --git a/ops/aws/canary/lambda/cmd/alarm_slack_handler/utils.go b/ops/aws/canary/lambda/cmd/alarm_slack_handler/utils.go
@@ -2,20 +2,21 @@ package main
 
 import (
 	"encoding/json"
+	"os"
+	"strconv"
+	"time"
+
 	"github.com/aws/aws-lambda-go/events"
 	"github.com/aws/aws-sdk-go/aws"
 	"github.com/aws/aws-sdk-go/aws/session"
 	"github.com/aws/aws-sdk-go/service/secretsmanager"
 	"github.com/slack-go/slack"
-	"os"
-	"strconv"
-	"time"
 )
 
 func mustGetSlackSecret() slackSecretType {
 	secretName := os.Getenv("SLACK_SECRET_NAME")
 
-	//Create a Secrets Manager client
+	// Create a Secrets Manager client
 	sess, err := session.NewSession()
 	if err != nil {
 		panic(err)

diff --git a/pkg/compute/management_client.go b/pkg/compute/management_client.go
@@ -57,11 +57,12 @@ func NewManagementClient(params *ManagementClientParams) *ManagementClient {
 }
 
 func (m *ManagementClient) getNodeInfo(ctx context.Context) models.NodeInfo {
-	return m.nodeInfoDecorator.DecorateNodeInfo(ctx, models.NodeInfo{
+	ni := m.nodeInfoDecorator.DecorateNodeInfo(ctx, models.NodeInfo{
 		NodeID:   m.nodeID,
 		NodeType: models.NodeTypeCompute,
 		Labels:   m.labelsProvider.GetLabels(ctx),
 	})
+	return ni
 }
 
 // RegisterNode sends a registration request to the requester node. If we successfully

diff --git a/pkg/devstack/option.go b/pkg/devstack/option.go
@@ -124,7 +124,7 @@ func (o *DevStackConfig) Validate() error {
 
 func WithAutoNodeApproval() ConfigOption {
 	return func(cfg *DevStackConfig) {
-		cfg.RequesterConfig.DefaultApprovalState = models.NodeApprovals.APPROVED
+		cfg.RequesterConfig.DefaultApprovalState = models.NodeMembership.APPROVED
 	}
 }
 

diff --git a/pkg/libp2p/transport/libp2p.go b/pkg/libp2p/transport/libp2p.go
@@ -6,6 +6,13 @@ import (
 	"fmt"
 	"time"
 
+	libp2p_pubsub "github.com/libp2p/go-libp2p-pubsub"
+	"github.com/libp2p/go-libp2p/core/host"
+	basichost "github.com/libp2p/go-libp2p/p2p/host/basic"
+	routedhost "github.com/libp2p/go-libp2p/p2p/host/routed"
+	"github.com/libp2p/go-libp2p/p2p/protocol/identify"
+	"github.com/multiformats/go-multiaddr"
+
 	"github.com/bacalhau-project/bacalhau/pkg/compute"
 	pkgconfig "github.com/bacalhau-project/bacalhau/pkg/config"
 	"github.com/bacalhau-project/bacalhau/pkg/lib/validate"
@@ -18,12 +25,6 @@ import (
 	"github.com/bacalhau-project/bacalhau/pkg/system"
 	core_transport "github.com/bacalhau-project/bacalhau/pkg/transport"
 	"github.com/bacalhau-project/bacalhau/pkg/transport/bprotocol"
-	libp2p_pubsub "github.com/libp2p/go-libp2p-pubsub"
-	"github.com/libp2p/go-libp2p/core/host"
-	basichost "github.com/libp2p/go-libp2p/p2p/host/basic"
-	routedhost "github.com/libp2p/go-libp2p/p2p/host/routed"
-	"github.com/libp2p/go-libp2p/p2p/protocol/identify"
-	"github.com/multiformats/go-multiaddr"
 )
 
 const NodeInfoTopic = "bacalhau-node-info"
@@ -46,7 +47,7 @@ type Libp2pTransport struct {
 	Host              host.Host
 	computeProxy      *bprotocol.ComputeProxy
 	callbackProxy     *bprotocol.CallbackProxy
-	nodeInfoPubSub    pubsub.PubSub[models.NodeInfo]
+	nodeInfoPubSub    pubsub.PubSub[models.NodeState]
 	nodeInfoDecorator models.NodeInfoDecorator
 }
 
@@ -71,7 +72,7 @@ func NewLibp2pTransport(ctx context.Context,
 	}
 
 	// PubSub to publish node info to the network
-	nodeInfoPubSub, err := libp2p.NewPubSub[models.NodeInfo](libp2p.PubSubParams{
+	nodeInfoPubSub, err := libp2p.NewPubSub[models.NodeState](libp2p.PubSubParams{
 		Host:      libp2pHost,
 		TopicName: NodeInfoTopic,
 		PubSub:    gossipSub,
@@ -129,8 +130,8 @@ func NewLibp2pTransport(ctx context.Context,
 
 func (t *Libp2pTransport) RegisterNodeInfoConsumer(ctx context.Context, nodeInfoStore routing.NodeInfoStore) error {
 	// register consumers of node info published over gossipSub
-	nodeInfoSubscriber := pubsub.NewChainedSubscriber[models.NodeInfo](true)
-	nodeInfoSubscriber.Add(pubsub.SubscriberFunc[models.NodeInfo](nodeInfoStore.Add))
+	nodeInfoSubscriber := pubsub.NewChainedSubscriber[models.NodeState](true)
+	nodeInfoSubscriber.Add(pubsub.SubscriberFunc[models.NodeState](nodeInfoStore.Add))
 
 	return t.nodeInfoPubSub.Subscribe(ctx, nodeInfoSubscriber)
 }
@@ -181,7 +182,7 @@ func (t *Libp2pTransport) ManagementProxy() compute.ManagementEndpoint {
 }
 
 // NodeInfoPubSub returns the node info pubsub.
-func (t *Libp2pTransport) NodeInfoPubSub() pubsub.PubSub[models.NodeInfo] {
+func (t *Libp2pTransport) NodeInfoPubSub() pubsub.PubSub[models.NodeState] {
 	return t.nodeInfoPubSub
 }