Merge branch 'master' of portworx:libopenstorage/openstorage into dgo…

…el_add-rebalance-mode

SDK_CHANGELOG.md

@@ -2,6 +2,13 @@
 
 ## Releases
 
+### v0.177.0 - (12/15/2023)
+
+* Add volume spec to drive resource
+### v0.176.0 - (02/15/2024)
+
+* Add cluster domain field to StorageNode
+
 ### v0.175.0 - (12/15/2023)
 
 * Add defrag status data structure

api/api.go

@@ -1036,6 +1036,7 @@ func (s *Node) ToStorageNode() *StorageNode {
 		SecurityStatus:    s.SecurityStatus,
 		SchedulerTopology: s.SchedulerTopology,
 		NonQuorumMember:   s.NonQuorumMember,
+		ClusterDomain:     s.DomainID,
 	}
 
 	node.Disks = make(map[string]*StorageResource)

api/api.proto

@@ -236,6 +236,8 @@ message StorageResource {
   bool pool_metadata_dev = 15;
   // Cloud drive type
   string cloud_drive_type = 16;
+  // volume spec that created this drive
+  string volume_spec = 17;
 }
 
 // StoragePool groups different storage devices based on their CosType
@@ -1633,6 +1635,8 @@ message StorageNode {
   // when initializing until it reaches a point where we can definitely determine whether
   // it is a quorum member or not.
   bool non_quorum_member = 22;
+  // Name of the cluster domain where the node belongs.
+  string cluster_domain = 23;
 }
 
 // StorageCluster represents the state and information about the cluster
@@ -4111,7 +4115,7 @@ message CollectDiagsJob {
 
 // DefragJob describes a job to run defragmentation on cluster nodes
 message DefragJob {
-  // MaxDurationHours defines the time limit in hours 
+  // MaxDurationHours defines the time limit in hours
   double max_duration_hours = 1;
   // MaxNodesInParallel defines the maximum number of nodes running the defrag job in parallel
   uint32 max_nodes_in_parallel = 2;
@@ -4123,7 +4127,7 @@ message DefragJob {
   // if not provided, will run on all nodes
   // cannot coexist with IncludeNodes
   repeated string exclude_nodes = 4;
-  // NodeSelector is a list of node label `key=value` pairs separated by comma, 
+  // NodeSelector is a list of node label `key=value` pairs separated by comma,
   // which selects the nodes to be run on for the job
   // can coexist with ExcludeNodes but cannot coexist with IncludeNodes
   repeated string node_selector = 5;
@@ -4141,7 +4145,7 @@ message DefragNodeStatus {
   string running_schedule = 2;
 }
 
-// DefragNodeStatus describes the defragmentation status of a pool 
+// DefragNodeStatus describes the defragmentation status of a pool
 message DefragPoolStatus {
   // NumIterations counts the number of times the pool gets defraged
   uint32 num_iterations = 1;
@@ -5661,7 +5665,7 @@ message SdkVersion {
     // SDK version major value of this specification
     Major = 0;
     // SDK version minor value of this specification
-    Minor = 175;
+    Minor = 177;
     // SDK version patch value of this specification
     Patch = 0;
   }
@@ -6342,7 +6346,7 @@ message SdkVerifyChecksumStartRequest {
 // SdkVerifyChecksumStartResponse defines the response for a
 // SdkVerifyChecksumStartRequest.
 message SdkVerifyChecksumStartResponse {
-  // Status code representing the state of the verify checksum operation 
+  // Status code representing the state of the verify checksum operation
   VerifyChecksum.VerifyChecksumStatus status = 1;
   // Text blob containing ASCII text providing details of the operation
   string message = 2;
@@ -6377,6 +6381,3 @@ message SdkVerifyChecksumStopResponse {
   // Text blob containing ASCII text providing details of the operation
   string message = 1;
 }
-
-
-

api/server/sdk/node_test.go

@@ -22,8 +22,8 @@ import (
 	"testing"
 	"time"
 
+	"github.com/golang/mock/gomock"
 	"github.com/stretchr/testify/assert"
-    "github.com/golang/mock/gomock"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
 
@@ -156,6 +156,7 @@ func TestSdkNodeEnumerateWithFilters(t *testing.T) {
 					},
 				},
 				NonQuorumMember: true,
+				DomainID:        "blue",
 			},
 		},
 	}
@@ -174,6 +175,7 @@ func TestSdkNodeEnumerateWithFilters(t *testing.T) {
 			},
 		},
 		NonQuorumMember: true,
+		ClusterDomain:   "blue",
 	}
 
 	s.MockCluster().EXPECT().Enumerate().Return(cluster, nil).Times(1)
@@ -246,6 +248,7 @@ func TestSdkNodeInspect(t *testing.T) {
 		},
 		HWType:          api.HardwareType_VirtualMachine,
 		NonQuorumMember: true,
+		DomainID:        "blue",
 	}
 	s.MockCluster().EXPECT().Inspect(nodeid).Return(node, nil).Times(1)
 
@@ -271,6 +274,7 @@ func TestSdkNodeInspect(t *testing.T) {
 	assert.Equal(t, rn.GetStatus(), node.Status)
 	assert.Equal(t, rn.GetHWType(), node.HWType)
 	assert.Equal(t, node.NonQuorumMember, rn.NonQuorumMember)
+	assert.Equal(t, node.DomainID, rn.ClusterDomain)
 
 	// Check Disk
 	assert.Len(t, rn.GetDisks(), 2)
@@ -361,6 +365,7 @@ func TestSdkNodeInspectCurrent(t *testing.T) {
 			},
 		},
 		NonQuorumMember: true,
+		DomainID:        "blue",
 	}
 
 	cluster := api.Cluster{
@@ -393,6 +398,7 @@ func TestSdkNodeInspectCurrent(t *testing.T) {
 	assert.Equal(t, rn.GetStatus(), node.Status)
 	assert.Equal(t, rn.GetHWType(), node.HWType)
 	assert.Equal(t, node.NonQuorumMember, rn.NonQuorumMember)
+	assert.Equal(t, node.DomainID, rn.ClusterDomain)
 
 	// Check Disk
 	assert.Len(t, rn.GetDisks(), 1)

cluster/manager/manager.go

@@ -289,6 +289,7 @@ func (c *ClusterManager) getNodeEntry(nodeID string, clustDBRef *cluster.Cluster
 			n.HWType = v.HWType
 			n.SecurityStatus = v.SecurityStatus
 			n.NonQuorumMember = v.NonQuorumMember
+			n.DomainID = v.ClusterDomain
 		} else {
 			logrus.Warnf("Could not query NodeID %v", nodeID)
 			// Node entry won't be refreshed form DB, will use the "offline" original
@@ -1157,7 +1158,7 @@ func (c *ClusterManager) waitForQuorum(exist bool) error {
 	return nil
 }
 
-func (c *ClusterManager) initializeCluster(db kvdb.Kvdb, selfClusterDomain string) (
+func (c *ClusterManager) initializeCluster(db kvdb.Kvdb) (
 	*cluster.ClusterInfo,
 	error,
 ) {
@@ -1220,10 +1221,9 @@ func (c *ClusterManager) initListeners(
 	db kvdb.Kvdb,
 	nodeExists *bool,
 	nodeInitialized bool,
-	selfClusterDomain string,
 ) (uint64, *cluster.ClusterInfo, error) {
 	// Initialize the cluster if required
-	clusterInfo, err := c.initializeCluster(db, selfClusterDomain)
+	clusterInfo, err := c.initializeCluster(db)
 	if err != nil {
 		return 0, nil, err
 	}
@@ -1328,13 +1328,11 @@ func (c *ClusterManager) initializeAndStartHeartbeat(
 	kvdb kvdb.Kvdb,
 	exist *bool,
 	nodeInitialized bool,
-	selfClusterDomain string,
 ) (uint64, *cluster.ClusterInfo, error) {
 	lastIndex, clusterInfo, err := c.initListeners(
 		kvdb,
 		exist,
 		nodeInitialized,
-		selfClusterDomain,
 	)
 	if err != nil {
 		return 0, nil, err
@@ -1475,6 +1473,7 @@ func (c *ClusterManager) StartWithConfiguration(
 	c.gossipPort = gossipPort
 	c.selfNode.GossipPort = gossipPort
 	c.selfClusterDomain = selfClusterDomain
+	c.selfNode.DomainID = selfClusterDomain
 	if err != nil {
 		logrus.Errorf("Failed to get external IP address for mgt/data interfaces: %s.",
 			err)
@@ -1527,7 +1526,6 @@ func (c *ClusterManager) StartWithConfiguration(
 		kv,
 		&exist,
 		nodeInitialized,
-		selfClusterDomain,
 	)
 	if err != nil {
 		return err
@@ -1643,6 +1641,7 @@ func (c *ClusterManager) nodes(clusterDB *cluster.ClusterInfo) []*api.Node {
 			node.NodeLabels = n.NodeLabels
 			node.SecurityStatus = n.SecurityStatus
 			node.NonQuorumMember = n.NonQuorumMember
+			node.DomainID = n.ClusterDomain
 		}
 		nodes = append(nodes, &node)
 	}

csi/node.go

@@ -364,8 +364,9 @@ func (s *OsdCsiServer) NodeGetVolumeStats(ctx context.Context, req *csi.NodeGetV
 	}
 
 	var attachPathMatch bool
+	sharedPath := fmt.Sprintf("%s/%s", api.SharedVolExportPrefix, id)
 	for _, attachPath := range vol.AttachPath {
-		if attachPath == path {
+		if attachPath == path || attachPath == sharedPath {
 			attachPathMatch = true
 		}
 	}

csi/node_test.go

@@ -1067,8 +1067,9 @@ func TestNodeGetVolumeStats(t *testing.T) {
 	used := int64(1 * 1024 * 1024)
 	available := size - used
 	id := "myvol123"
+	sharedPath := fmt.Sprintf("%s/%s", api.SharedVolExportPrefix, id) // "/var/lib/osd/pxns/myvol123"
 	vol := &api.Volume{
-		AttachPath: []string{"/test"},
+		AttachPath: []string{"/test", sharedPath},
 		Id:         id,
 		Locator: &api.VolumeLocator{
 			Name: id,
@@ -1104,6 +1105,18 @@ func TestNodeGetVolumeStats(t *testing.T) {
 	assert.Equal(t, false, resp.VolumeCondition.Abnormal)
 	assert.Equal(t, "Volume status is up", resp.VolumeCondition.Message)
 
+	// Get VolumeStats - shared volume
+	resp, err = c.NodeGetVolumeStats(
+		context.Background(),
+		&csi.NodeGetVolumeStatsRequest{VolumeId: id, VolumePath: sharedPath})
+	assert.NoError(t, err)
+	assert.Equal(t, 1, len(resp.Usage))
+	assert.Equal(t, size, resp.Usage[0].Total)
+	assert.Equal(t, used, resp.Usage[0].Used)
+	assert.Equal(t, available, resp.Usage[0].Available)
+	assert.Equal(t, false, resp.VolumeCondition.Abnormal)
+	assert.Equal(t, "Volume status is up", resp.VolumeCondition.Message)
+
 	// Get VolumeStats - down
 	vol.Status = api.VolumeStatus_VOLUME_STATUS_DOWN
 	resp, err = c.NodeGetVolumeStats(