Syncing latest changes from upstream devel for ceph-csi

go.mod

@@ -7,9 +7,9 @@ require (
 	github.com/aws/aws-sdk-go v1.55.5
 	github.com/aws/aws-sdk-go-v2/service/sts v1.32.3
 	github.com/ceph/ceph-csi/api v0.0.0-00010101000000-000000000000
-	github.com/ceph/go-ceph v0.30.0
+	github.com/ceph/go-ceph v0.30.1-0.20241102143109-75d1af3ed638
 	github.com/container-storage-interface/spec v1.10.0
-	github.com/csi-addons/spec v0.2.1-0.20240730084235-3958a5b17d24
+	github.com/csi-addons/spec v0.2.1-0.20241104111131-27825f744db5
 	github.com/gemalto/kmip-go v0.0.10
 	github.com/golang/protobuf v1.5.4
 	github.com/google/fscrypt v0.3.6-0.20240502174735-068b9f8f5dec

go.sum

@@ -1445,8 +1445,8 @@ github.com/cenkalti/backoff/v4 v4.3.0/go.mod h1:Y3VNntkOUPxTVeUxJ/G5vcM//AlwfmyY
 github.com/census-instrumentation/opencensus-proto v0.2.1/go.mod h1:f6KPmirojxKA12rnyqOA5BBL4O983OfeGPqjHWSTneU=
 github.com/census-instrumentation/opencensus-proto v0.3.0/go.mod h1:f6KPmirojxKA12rnyqOA5BBL4O983OfeGPqjHWSTneU=
 github.com/census-instrumentation/opencensus-proto v0.4.1/go.mod h1:4T9NM4+4Vw91VeyqjLS6ao50K5bOcLKN6Q42XnYaRYw=
-github.com/ceph/go-ceph v0.30.0 h1:p/+rNnn9dUByrDhXfBFilVriRZKJghMJcts8N2wQ+ws=
-github.com/ceph/go-ceph v0.30.0/go.mod h1:OJFju/Xmtb7ihHo/aXOayw6RhVOUGNke5EwTipwaf6A=
+github.com/ceph/go-ceph v0.30.1-0.20241102143109-75d1af3ed638 h1:J/IBLjrHABhgi8okkxodSxhOPPCXhdBSt/i5OjyA9Ug=
+github.com/ceph/go-ceph v0.30.1-0.20241102143109-75d1af3ed638/go.mod h1:E2MAGzr1oBwrWpN6OPgQ8fjxqmJAwr4/0+ZixxTvs7A=
 github.com/cespare/xxhash v1.1.0/go.mod h1:XrSqR1VqqWfGrhpAt58auRo0WTKS1nRRg3ghfAqPWnc=
 github.com/cespare/xxhash/v2 v2.1.1/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs=
 github.com/cespare/xxhash/v2 v2.1.2/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs=
@@ -1504,8 +1504,8 @@ github.com/cpuguy83/go-md2man/v2 v2.0.4/go.mod h1:tgQtvFlXSQOSOSIRvRPT7W67SCa46t
 github.com/creack/pty v1.1.9/go.mod h1:oKZEueFk5CKHvIhNR5MUki03XCEU+Q6VDXinZuGJ33E=
 github.com/creack/pty v1.1.11/go.mod h1:oKZEueFk5CKHvIhNR5MUki03XCEU+Q6VDXinZuGJ33E=
 github.com/creack/pty v1.1.18/go.mod h1:MOBLtS5ELjhRRrroQr9kyvTxUAFNvYEK993ew/Vr4O4=
-github.com/csi-addons/spec v0.2.1-0.20240730084235-3958a5b17d24 h1:tJETaYbnnzlCSaqDXQzbszYyuAtG/sFzm6DargeVzJA=
-github.com/csi-addons/spec v0.2.1-0.20240730084235-3958a5b17d24/go.mod h1:Mwq4iLiUV4s+K1bszcWU6aMsR5KPsbIYzzszJ6+56vI=
+github.com/csi-addons/spec v0.2.1-0.20241104111131-27825f744db5 h1:j9NaWj5KmzEVarmsjxS/NDAhes6Uzq1qhkUGHvDlVBk=
+github.com/csi-addons/spec v0.2.1-0.20241104111131-27825f744db5/go.mod h1:Mwq4iLiUV4s+K1bszcWU6aMsR5KPsbIYzzszJ6+56vI=
 github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc h1:U9qPSI2PIWSS1VwoXQT9A3Wy9MM3WgvqSxFWenqJduM=

internal/csi-addons/networkfence/fencing.go

@@ -217,31 +217,7 @@ func isIPInCIDR(ctx context.Context, ip, cidr string) bool {
 func (ac *activeClient) fetchIP() (string, error) {
 	// example: "inst": "client.4305 172.21.9.34:0/422650892",
 	// then returning value will be 172.21.9.34
-	clientInfo := ac.Inst
-
-	// Attempt to extract the IP address using a regular expression
-	// the regular expression aims to match either a complete IPv6
-	// address or a complete IPv4 address follows by any prefix (v1 or v2)
-	// if exists
-	// (?:v[0-9]+:): this allows for an optional prefix starting with "v"
-	// followed by one or more digits and a colon.
-	// The ? outside the group makes the entire prefix section optional.
-	// (?:[0-9a-fA-F]{1,4}:){7}[0-9a-fA-F]{1,4}: this allows to check for
-	// standard IPv6 address.
-	// |: Alternation operator to allow matching either the IPv6 pattern
-	// with a prefix or the IPv4 pattern.
-	// '(?:\d+\.){3}\d+: This part matches a standard IPv4 address.
-	re := regexp.MustCompile(`(?:v[0-9]+:)?([0-9a-fA-F]{1,4}(:[0-9a-fA-F]{1,4}){7}|(?:\d+\.){3}\d+)`)
-	ipMatches := re.FindStringSubmatch(clientInfo)
-
-	if len(ipMatches) > 0 {
-		ip := net.ParseIP(ipMatches[1])
-		if ip != nil {
-			return ip.String(), nil
-		}
-	}
-
-	return "", fmt.Errorf("failed to extract IP address, incorrect format: %s", clientInfo)
+	return ParseClientIP(ac.Inst)
 }
 
 func (ac *activeClient) fetchID() (int, error) {
@@ -526,3 +502,29 @@ func (nf *NetworkFence) parseBlocklistForCIDR(ctx context.Context, blocklist, ci
 
 	return matchingHosts
 }
+
+func ParseClientIP(addr string) (string, error) {
+	// Attempt to extract the IP address using a regular expression
+	// the regular expression aims to match either a complete IPv6
+	// address or a complete IPv4 address follows by any prefix (v1 or v2)
+	// if exists
+	// (?:v[0-9]+:): this allows for an optional prefix starting with "v"
+	// followed by one or more digits and a colon.
+	// The ? outside the group makes the entire prefix section optional.
+	// (?:[0-9a-fA-F]{1,4}:){7}[0-9a-fA-F]{1,4}: this allows to check for
+	// standard IPv6 address.
+	// |: Alternation operator to allow matching either the IPv6 pattern
+	// with a prefix or the IPv4 pattern.
+	// '(?:\d+\.){3}\d+: This part matches a standard IPv4 address.
+	re := regexp.MustCompile(`(?:v[0-9]+:)?([0-9a-fA-F]{1,4}(:[0-9a-fA-F]{1,4}){7}|(?:\d+\.){3}\d+)`)
+	ipMatches := re.FindStringSubmatch(addr)
+
+	if len(ipMatches) > 0 {
+		ip := net.ParseIP(ipMatches[1])
+		if ip != nil {
+			return ip.String(), nil
+		}
+	}
+
+	return "", fmt.Errorf("failed to extract IP address, incorrect format: %s", addr)
+}

internal/csi-addons/networkfence/fencing_test.go

@@ -257,3 +257,45 @@ listed 1 entries`,
 		})
 	}
 }
+
+func TestParseClientIP(t *testing.T) {
+	t.Parallel()
+	tests := []struct {
+		name    string
+		addr    string
+		want    string
+		wantErr bool
+	}{
+		{
+			name:    "IPv4 address",
+			addr:    "10.244.0.1:0/2686266785",
+			want:    "10.244.0.1",
+			wantErr: false,
+		},
+		{
+			name:    "IPv6 address",
+			addr:    "2001:0db8:85a3:0000:0000:8a2e:0370:7334:0/2686266785",
+			want:    "2001:db8:85a3::8a2e:370:7334",
+			wantErr: false,
+		},
+		{
+			name:    "Invalid address",
+			addr:    "invalid",
+			want:    "",
+			wantErr: true,
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			t.Parallel()
+			got, err := ParseClientIP(tt.addr)
+			if (err != nil) != tt.wantErr {
+				t.Errorf("ParseClientIP() error = %v, wantErr %v", err, tt.wantErr)
+			}
+
+			if got != tt.want {
+				t.Errorf("ParseClientIP() = %v, want %v", got, tt.want)
+			}
+		})
+	}
+}

internal/csi-addons/rbd/identity.go

@@ -152,6 +152,13 @@ func (is *IdentityServer) GetCapabilities(
 						Type: identity.Capability_EncryptionKeyRotation_ENCRYPTIONKEYROTATION,
 					},
 				},
+			},
+			&identity.Capability{
+				Type: &identity.Capability_NetworkFence_{
+					NetworkFence: &identity.Capability_NetworkFence{
+						Type: identity.Capability_NetworkFence_GET_CLIENTS_TO_FENCE,
+					},
+				},
 			})
 	}
 

internal/csi-addons/rbd/network_fence.go

@@ -114,3 +114,69 @@ func (fcs *FenceControllerServer) UnfenceClusterNetwork(
 
 	return &fence.UnfenceClusterNetworkResponse{}, nil
 }
+
+// GetFenceClients fetches the ceph cluster ID and the client address that need to be fenced.
+func (fcs *FenceControllerServer) GetFenceClients(
+	ctx context.Context,
+	req *fence.GetFenceClientsRequest,
+) (*fence.GetFenceClientsResponse, error) {
+	options := req.GetParameters()
+	clusterID, err := util.GetClusterID(options)
+	if err != nil {
+		return nil, status.Error(codes.InvalidArgument, err.Error())
+	}
+
+	cr, err := util.NewUserCredentials(req.GetSecrets())
+	if err != nil {
+		return nil, status.Error(codes.InvalidArgument, err.Error())
+	}
+	defer cr.DeleteCredentials()
+
+	monitors, _ /* clusterID*/, err := util.GetMonsAndClusterID(ctx, clusterID, false)
+	if err != nil {
+		return nil, status.Error(codes.Internal, err.Error())
+	}
+
+	// Get the cluster ID of the ceph cluster.
+	conn := &util.ClusterConnection{}
+	err = conn.Connect(monitors, cr)
+	if err != nil {
+		return nil, status.Errorf(codes.Internal, "failed to connect to MONs %q: %s", monitors, err)
+	}
+	defer conn.Destroy()
+
+	fsID, err := conn.GetFSID()
+	if err != nil {
+		return nil, status.Errorf(codes.Internal, "failed to get cephfs id: %s", err)
+	}
+
+	address, err := conn.GetAddrs()
+	if err != nil {
+		return nil, status.Errorf(codes.Internal, "failed to get client address: %s", err)
+	}
+
+	// The example address we get is 10.244.0.1:0/2686266785 from
+	// which we need to extract the IP address.
+	addr, err := nf.ParseClientIP(address)
+	if err != nil {
+		return nil, status.Errorf(codes.Internal, "failed to parse client address: %s", err)
+	}
+
+	// adding /32 to the IP address to make it a CIDR block.
+	addr += "/32"
+
+	resp := &fence.GetFenceClientsResponse{
+		Clients: []*fence.ClientDetails{
+			{
+				Id: fsID,
+				Addresses: []*fence.CIDR{
+					{
+						Cidr: addr,
+					},
+				},
+			},
+		},
+	}
+
+	return resp, nil
+}

internal/journal/voljournal.go

@@ -199,6 +199,7 @@ func NewCSISnapshotJournal(suffix string) *Config {
 		cephSnapSourceKey:       "csi.source",
 		namespace:               "",
 		csiImageIDKey:           "csi.imageid",
+		csiGroupIDKey:           "csi.groupid",
 		encryptKMSKey:           "csi.volume.encryptKMS",
 		encryptionType:          "csi.volume.encryptionType",
 		ownerKey:                "csi.volume.owner",
@@ -805,7 +806,8 @@ func (conn *Connection) StoreAttribute(ctx context.Context, pool, reservedUUID,
 
 // StoreGroupID stores an groupID in omap.
 func (conn *Connection) StoreGroupID(ctx context.Context, pool, reservedUUID, groupID string) error {
-	err := conn.StoreAttribute(ctx, pool, reservedUUID, conn.config.csiGroupIDKey, groupID)
+	err := setOMapKeys(ctx, conn, pool, conn.config.namespace, conn.config.cephUUIDDirectoryPrefix+reservedUUID,
+		map[string]string{conn.config.csiGroupIDKey: groupID})
 	if err != nil {
 		return fmt.Errorf("failed to store groupID %w", err)
 	}

internal/rbd/driver/driver.go

@@ -25,6 +25,7 @@ import (
 	csiaddons "github.com/ceph/ceph-csi/internal/csi-addons/server"
 	csicommon "github.com/ceph/ceph-csi/internal/csi-common"
 	"github.com/ceph/ceph-csi/internal/rbd"
+	"github.com/ceph/ceph-csi/internal/rbd/features"
 	"github.com/ceph/ceph-csi/internal/util"
 	"github.com/ceph/ceph-csi/internal/util/k8s"
 	"github.com/ceph/ceph-csi/internal/util/log"
@@ -123,6 +124,19 @@ func (r *Driver) Run(conf *util.Config) {
 				csi.VolumeCapability_AccessMode_SINGLE_NODE_SINGLE_WRITER,
 				csi.VolumeCapability_AccessMode_SINGLE_NODE_MULTI_WRITER,
 			})
+
+		// GroupSnapGetInfo is used within the VolumeGroupSnapshot implementation
+		vgsSupported, vgsErr := features.SupportsGroupSnapGetInfo()
+		if vgsSupported {
+			r.cd.AddGroupControllerServiceCapabilities([]csi.GroupControllerServiceCapability_RPC_Type{
+				csi.GroupControllerServiceCapability_RPC_CREATE_DELETE_GET_VOLUME_GROUP_SNAPSHOT,
+			})
+		} else {
+			log.DefaultLog("not enabling VolumeGroupSnapshot service capability")
+		}
+		if vgsErr != nil {
+			log.ErrorLogMsg("failed detecting VolumeGroupSnapshot support: %v", vgsErr)
+		}
 	}
 
 	if k8s.RunsOnKubernetes() && conf.IsNodeServer {
@@ -178,6 +192,7 @@ func (r *Driver) Run(conf *util.Config) {
 		IS: r.ids,
 		CS: r.cs,
 		NS: r.ns,
+		GS: r.cs,
 	}
 	s.Start(conf.Endpoint, srv, csicommon.MiddlewareServerOptionConfig{
 		LogSlowOpInterval: conf.LogSlowOpInterval,
@@ -227,6 +242,9 @@ func (r *Driver) setupCSIAddonsServer(conf *util.Config) error {
 	}
 
 	if conf.IsNodeServer {
+		fcs := casrbd.NewFenceControllerServer()
+		r.cas.RegisterService(fcs)
+
 		rs := casrbd.NewReclaimSpaceNodeServer(r.ns.VolumeLocks)
 		r.cas.RegisterService(rs)
 

internal/rbd/group.go

@@ -44,6 +44,7 @@ func (rv *rbdVolume) AddToGroup(ctx context.Context, vg types.VolumeGroup) error
 	if err != nil {
 		return fmt.Errorf("failed to open image %q: %w", rv, err)
 	}
+	defer image.Close()
 
 	info, err := image.GetGroup()
 	if err != nil {