Use VClusterOps API to get node details in podfacts (#769)

This adds a new fetcher in podfacts to fetch node details from the
running database inside the pod. This new fetcher will call VClusterOps
API which will send an HTTP request to the database. The new fetcher
will be enabled when VclusterOps annotation is set, and Vertica server
version is not older than v24.3.0. The new fetcher should be faster and
more reliable than the old fetcher which will execute vsql within the
pod.

api/v1/version.go

@@ -81,6 +81,8 @@ const (
 	ScrutinizeDBPasswdInSecretMinVersion = "v24.2.0"
 	// Starting in v24.3.0, sandboxing a subcluster with the operator is supported
 	SandboxSupportedMinVersion = "v24.3.0"
+	// Starting in v24.3.0, we call vclusterops API to get node details instead of executing vsql within the pod
+	FetchNodeDetailsWithVclusterOpsMinVersion = "v24.3.0"
 )
 
 // GetVerticaVersionStr returns the vertica version, in string form, that is stored

go.mod

@@ -16,7 +16,7 @@ require (
 	github.com/onsi/gomega v1.24.2
 	github.com/pkg/errors v0.9.1
 	github.com/prometheus/client_golang v1.14.0
-	github.com/vertica/vcluster v1.2.1-0.20240401123316-045acf30a006
+	github.com/vertica/vcluster v1.2.1-0.20240411151456-29d822a8fa93
 	github.com/vertica/vertica-sql-go v1.1.1
 	go.uber.org/zap v1.25.0
 	golang.org/x/text v0.14.0

go.sum

@@ -322,8 +322,8 @@ github.com/stretchr/testify v1.8.4 h1:CcVxjf3Q8PM0mHUKJCdn+eZZtm5yQwehR5yeSVQQcU
 github.com/theckman/yacspin v0.13.12 h1:CdZ57+n0U6JMuh2xqjnjRq5Haj6v1ner2djtLQRzJr4=
 github.com/theckman/yacspin v0.13.12/go.mod h1:Rd2+oG2LmQi5f3zC3yeZAOl245z8QOvrH4OPOJNZxLg=
 github.com/tonglil/buflogr v1.0.1 h1:WXFZLKxLfqcVSmckwiMCF8jJwjIgmStJmg63YKRF1p0=
-github.com/vertica/vcluster v1.2.1-0.20240401123316-045acf30a006 h1:oAE7MEsNphvHIh3uhOY4P84rrOsoOV3O/YquZLuc+oQ=
-github.com/vertica/vcluster v1.2.1-0.20240401123316-045acf30a006/go.mod h1:zkTLy1hF6LzTeWmXiGYLFzihbi397WTQhPZ9xX4B+FY=
+github.com/vertica/vcluster v1.2.1-0.20240411151456-29d822a8fa93 h1:8FnB36cLWSPq0iSQg+w8JkJsdkE7WnTQZ5PV1hXn0sw=
+github.com/vertica/vcluster v1.2.1-0.20240411151456-29d822a8fa93/go.mod h1:zkTLy1hF6LzTeWmXiGYLFzihbi397WTQhPZ9xX4B+FY=
 github.com/vertica/vertica-sql-go v1.1.1 h1:sZYijzBbvdAbJcl4cYlKjR+Eh/X1hGKzukWuhh8PjvI=
 github.com/vertica/vertica-sql-go v1.1.1/go.mod h1:fGr44VWdEvL+f+Qt5LkKLOT7GoxaWdoUCnPBU9h6t04=
 github.com/yuin/goldmark v1.1.25/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74=

pkg/catalog/fetch_node_details_vcluster.go

@@ -0,0 +1,63 @@
+/*
+ (c) Copyright [2021-2024] Open Text.
+ Licensed under the Apache License, Version 2.0 (the "License");
+ You may not use this file except in compliance with the License.
+ You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing, software
+ distributed under the License is distributed on an "AS IS" BASIS,
+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ See the License for the specific language governing permissions and
+ limitations under the License.
+*/
+
+package catalog
+
+import (
+	"context"
+	"strconv"
+
+	"github.com/vertica/vcluster/vclusterops"
+	"github.com/vertica/vertica-kubernetes/pkg/vadmin"
+	"github.com/vertica/vertica-kubernetes/pkg/vadmin/opts/fetchnodedetails"
+)
+
+// FetchNodeDetails returns details for a node, including its state, shard subscriptions, and depot details
+func (v *VCluster) FetchNodeDetails(ctx context.Context) (nodeDetails *NodeDetails, err error) {
+	vclusterOps := vadmin.MakeVClusterOps(v.Log, v.VDB, v.Client, v.Password, v.EVRec, vadmin.SetupVClusterOps)
+	opts := []fetchnodedetails.Option{
+		fetchnodedetails.WithInitiator(v.PodIP),
+	}
+	vnodeDetails, err := vclusterOps.FetchNodeDetails(ctx, opts...)
+	if err != nil {
+		return nil, err
+	}
+	nodeDetails = &NodeDetails{}
+	nodeDetails.parseVNodeDetails(&vnodeDetails)
+	return nodeDetails, nil
+}
+
+// parseVNodeDetails will parse node details returned by vclusterOps API
+func (nodeDetails *NodeDetails) parseVNodeDetails(vnodeDetails *vclusterops.NodeDetails) {
+	nodeDetails.Name = vnodeDetails.Name
+	nodeDetails.State = vnodeDetails.State
+	nodeDetails.SubclusterOid = strconv.FormatUint(vnodeDetails.SubclusterID, 10)
+	nodeDetails.ReadOnly = vnodeDetails.IsReadOnly
+	nodeDetails.SandboxName = vnodeDetails.SandboxName
+	nodeDetails.ShardSubscriptions = int(vnodeDetails.NumberShardSubscriptions)
+	// The shard subscriptions we get from vcluster includes the replica shard.
+	// We decrement that by one to account for that. We want to know when there
+	// are 0 shard subscriptions in order to drive a shard rebalance.
+	if nodeDetails.ShardSubscriptions > 0 {
+		nodeDetails.ShardSubscriptions--
+	}
+	for _, storageLoc := range vnodeDetails.StorageLocList {
+		if storageLoc.UsageType == "DEPOT" {
+			nodeDetails.MaxDepotSize = int(storageLoc.MaxSize)
+			nodeDetails.DepotDiskPercentSize = storageLoc.DiskPercent
+			break
+		}
+	}
+}

pkg/catalog/fetch_node_details_vcluster_test.go

@@ -0,0 +1,86 @@
+/*
+ (c) Copyright [2021-2024] Open Text.
+ Licensed under the Apache License, Version 2.0 (the "License");
+ You may not use this file except in compliance with the License.
+ You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing, software
+ distributed under the License is distributed on an "AS IS" BASIS,
+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ See the License for the specific language governing permissions and
+ limitations under the License.
+*/
+
+package catalog
+
+import (
+	. "github.com/onsi/ginkgo/v2"
+	. "github.com/onsi/gomega"
+	"github.com/vertica/vcluster/vclusterops"
+)
+
+var _ = Describe("nodedetailsvcluster", func() {
+	It("should parse node details from vclusterOps API correctly", func() {
+		vnodeDetails := vclusterops.NodeDetails{
+			NodeState: vclusterops.NodeState{
+				Name:                     "v_test_db_node0001",
+				ID:                       45035996273704992,
+				Address:                  "192.168.1.101",
+				State:                    "UP",
+				Database:                 "test_db",
+				IsPrimary:                false,
+				IsReadOnly:               false,
+				CatalogPath:              "/data/test_db/v_test_db_node0001_catalog/Catalog",
+				DataPath:                 []string{"/data/test_db/v_test_db_node0001_data"},
+				DepotPath:                "/data/test_db/v_test_db_node0001_depot",
+				SubclusterName:           "sc1",
+				SubclusterID:             45035996273704988,
+				LastMsgFromNodeAt:        "2024-04-05T15:23:32.016281-04",
+				DownSince:                "",
+				Version:                  "v24.3.0-a0efe9ba3abb08d9e6472ffc29c8e0949b5998d2",
+				SandboxName:              "sandbox1",
+				NumberShardSubscriptions: 3,
+			},
+			StorageLocations: vclusterops.StorageLocations{
+				StorageLocList: []vclusterops.StorageLocation{
+					{
+						Name:        "__location_0_v_test_db_node0001",
+						ID:          45035996273705024,
+						Label:       "",
+						UsageType:   "DATA,TEMP",
+						Path:        "/data/test_db/v_test_db_node0001_data",
+						SharingType: "NONE",
+						MaxSize:     0,
+						DiskPercent: "",
+						HasCatalog:  false,
+						Retired:     false,
+					},
+					{
+						Name:        "__location_1_v_test_db_node0001",
+						ID:          45035996273705166,
+						Label:       "auto-data-depot",
+						UsageType:   "DEPOT",
+						Path:        "/data/test_db/v_test_db_node0001_depot",
+						SharingType: "NONE",
+						MaxSize:     8215897325568,
+						DiskPercent: "60%",
+						HasCatalog:  false,
+						Retired:     false,
+					},
+				},
+			},
+		}
+		nodeDetails := &NodeDetails{}
+		nodeDetails.parseVNodeDetails(&vnodeDetails)
+		Expect(nodeDetails.Name).Should(Equal("v_test_db_node0001"))
+		Expect(nodeDetails.State).Should(Equal("UP"))
+		Expect(nodeDetails.SubclusterOid).Should(Equal("45035996273704988"))
+		Expect(nodeDetails.ReadOnly).Should(BeFalse())
+		Expect(nodeDetails.SandboxName).Should(Equal("sandbox1"))
+		Expect(nodeDetails.ShardSubscriptions).Should(Equal(2))
+		Expect(nodeDetails.MaxDepotSize).Should(Equal(8215897325568))
+		Expect(nodeDetails.DepotDiskPercentSize).Should(Equal("60%"))
+	})
+})

pkg/catalog/fetch_node_state_vsql.go → pkg/catalog/fetch_node_details_vsql.go

@@ -18,22 +18,52 @@ package catalog
 import (
 	"context"
 	"fmt"
+	"strconv"
 	"strings"
 
 	vapi "github.com/vertica/vertica-kubernetes/api/v1"
 )
 
-func (v *VSQL) FetchNodeState(ctx context.Context) (*NodeInfo, error) {
+// FetchNodeDetails returns details for a node, including its state, shard subscriptions, and depot details
+func (v *VSQL) FetchNodeDetails(ctx context.Context) (nodeDetails *NodeDetails, err error) {
+	nodeDetails = &NodeDetails{}
 	sql := v.buildFetchNodeStateQuery()
-	stdout, err := v.queryNodeStatus(ctx, sql)
+	stdout, err := v.executeSQL(ctx, sql)
 	if err != nil {
 		// Skip parsing that happens next
 		return nil, err
 	}
-	return parseNodeState(stdout)
+	err = nodeDetails.parseNodeState(stdout)
+	if err != nil {
+		return nil, err
+	}
+
+	sql = v.buildFetchShardSubscriptionsQuery()
+	stdout, err = v.executeSQL(ctx, sql)
+	if err != nil {
+		// Skip parsing that happens next
+		return nil, err
+	}
+	err = nodeDetails.parseShardSubscriptions(stdout)
+	if err != nil {
+		return nil, err
+	}
+
+	sql = v.buildFetchDepotDetailsQuery()
+	stdout, err = v.executeSQL(ctx, sql)
+	if err != nil {
+		// Skip parsing that happens next
+		return nil, err
+	}
+	err = nodeDetails.parseDepotDetails(stdout)
+	if err != nil {
+		return nil, err
+	}
+
+	return nodeDetails, nil
 }
 
-// buildFetchNodeStateQuery constructs the query to get the node state
+// buildFetchNodeStateQuery constructs a sql query to get the node state
 func (v *VSQL) buildFetchNodeStateQuery() string {
 	// The first two columns are just for informational purposes.
 	cols := "n.node_name, node_state"
@@ -69,20 +99,31 @@ func (v *VSQL) buildFetchNodeStateQuery() string {
 	return sql
 }
 
-// queryNodeStatus will query the nodes system table for the following info:
-// node name, node is up, read-only state, subcluster oid and sandbox name.
+// buildFetchShardSubscriptionsQuery constructs a sql query to get shard subscriptions
+func (v *VSQL) buildFetchShardSubscriptionsQuery() string {
+	return fmt.Sprintf("select count(*) from v_catalog.node_subscriptions where node_name = '%s' and shard_name != 'replica'",
+		v.VNodeName)
+}
+
+// buildFetchDepotDetailsQuery constructs a sql query to get depot details
+func (v *VSQL) buildFetchDepotDetailsQuery() string {
+	return fmt.Sprintf("select max_size, disk_percent from storage_locations "+
+		"where location_usage = 'DEPOT' and node_name = '%s'", v.VNodeName)
+}
+
+// executeSQL will run a sql query through vsql in the pod.
 // It assumes the database exists and the pod is running.
-func (v *VSQL) queryNodeStatus(ctx context.Context, sql string) (string, error) {
+func (v *VSQL) executeSQL(ctx context.Context, sql string) (string, error) {
 	cmd := []string{"-tAc", sql}
 	stdout, _, err := v.PRunner.ExecVSQL(ctx, v.PodName, v.ExecContainerName, cmd...)
 	return stdout, err
 }
 
 // parseNodeState will parse query output from node state
-func parseNodeState(stdout string) (*NodeInfo, error) {
+func (nodeDetails *NodeDetails) parseNodeState(stdout string) error {
 	// For testing purposes we early out with no error if there is no output
 	if stdout == "" {
-		return nil, nil
+		return nil
 	}
 	// The stdout comes in the form like this:
 	// v_vertdb_node0001|UP|41231232423|t|sandbox1
@@ -99,20 +140,56 @@ func parseNodeState(stdout string) (*NodeInfo, error) {
 	const MinExpectedCols = 3
 	if len(cols) < MinExpectedCols {
 		err = fmt.Errorf("expected at least %d columns from node query but only got %d", MinExpectedCols, len(cols))
-		return nil, err
+		return err
 	}
-	ninf := &NodeInfo{}
-	ninf.SubclusterOid = cols[2]
+	nodeDetails.SubclusterOid = cols[2]
 	// Read-only can be missing on versions that don't support that state.
 	// Return false in those cases.
 	if len(cols) > MinExpectedCols {
-		ninf.ReadOnly = cols[3] == "t"
+		nodeDetails.ReadOnly = cols[3] == "t"
 		// sandbox can be missing on versions that don't support that state
 		if len(cols) > MinExpectedCols+1 {
-			ninf.SandboxName = cols[4]
+			nodeDetails.SandboxName = cols[4]
 		}
 	} else {
-		ninf.ReadOnly = false
+		nodeDetails.ReadOnly = false
+	}
+	return nil
+}
+
+// parseShardSubscriptions will parse the query output of shard subscriptions
+func (nodeDetails *NodeDetails) parseShardSubscriptions(op string) error {
+	// For testing purposes we early out with no error if there is no output
+	if op == "" {
+		return nil
+	}
+
+	lines := strings.Split(op, "\n")
+	subs, err := strconv.Atoi(lines[0])
+	if err != nil {
+		return err
+	}
+	nodeDetails.ShardSubscriptions = subs
+	return nil
+}
+
+// parseDepotDetails will parse the query output of depot details
+func (nodeDetails *NodeDetails) parseDepotDetails(op string) error {
+	// For testing purposes, return without error if there is no output
+	if op == "" {
+		return nil
+	}
+	lines := strings.Split(op, "\n")
+	cols := strings.Split(lines[0], "|")
+	const ExpectedCols = 2
+	if len(cols) != ExpectedCols {
+		return fmt.Errorf("expected %d columns from storage_locations query but only got %d", ExpectedCols, len(cols))
+	}
+	var err error
+	nodeDetails.MaxDepotSize, err = strconv.Atoi(cols[0])
+	if err != nil {
+		return err
 	}
-	return ninf, nil
+	nodeDetails.DepotDiskPercentSize = cols[1]
+	return nil
 }