cephfs: adds the implementation of client eviction

this commit adds client eviction to cephfs, based
on the IPs in cidr block, it evicts those IPs from
the network.

Signed-off-by: Riya Singhal <[email protected]>

internal/csi-addons/cephfs/network_fence.go

@@ -0,0 +1,88 @@
+/*
+Copyright 2023 The Ceph-CSI Authors.
+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 cephfs
+
+import (
+	"context"
+	"errors"
+
+	nf "github.com/ceph/ceph-csi/internal/csi-addons/networkfence"
+	"github.com/ceph/ceph-csi/internal/util"
+
+	"github.com/csi-addons/spec/lib/go/fence"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
+)
+
+// FenceControllerServer struct of cephFS CSI driver with supported methods
+// of CSI-Addons networkfence controller service spec.
+type FenceControllerServer struct {
+	*fence.UnimplementedFenceControllerServer
+}
+
+// NewFenceControllerServer creates a new FenceControllerServer which handles
+// the FenceController Service requests from the CSI-Addons specification.
+func NewFenceControllerServer() *FenceControllerServer {
+	return &FenceControllerServer{}
+}
+
+// RegisterService registers the FenceControllerServer's service
+// with the gRPC server.
+func (fcs *FenceControllerServer) RegisterService(server grpc.ServiceRegistrar) {
+	fence.RegisterFenceControllerServer(server, fcs)
+}
+
+// validateFenceClusterNetworkReq checks the sanity of FenceClusterNetworkRequest.
+func validateNetworkFenceReq(fenceClients []*fence.CIDR, options map[string]string) error {
+	if len(fenceClients) == 0 {
+		return errors.New("CIDR block cannot be empty")
+	}
+
+	if value, ok := options["clusterID"]; !ok || value == "" {
+		return errors.New("missing or empty clusterID")
+	}
+
+	return nil
+}
+
+// FenceClusterNetwork blocks access to a CIDR block by creating a network fence.
+// It evicts the IP addresses of clients, which are in CIDR block.
+func (fcs *FenceControllerServer) FenceClusterNetwork(
+	ctx context.Context,
+	req *fence.FenceClusterNetworkRequest,
+) (*fence.FenceClusterNetworkResponse, error) {
+	err := validateNetworkFenceReq(req.GetCidrs(), req.Parameters)
+	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.Internal, err.Error())
+	}
+	defer cr.DeleteCredentials()
+
+	nwFence, err := nf.NewNetworkFence(ctx, cr, req.Cidrs, req.GetParameters())
+	if err != nil {
+		return nil, status.Error(codes.Internal, err.Error())
+	}
+
+	err = nwFence.AddClientEviction(ctx)
+	if err != nil {
+		return nil, status.Errorf(codes.Internal, "failed to fence CIDR block %q: %s", nwFence.Cidr, err.Error())
+	}
+
+	return &fence.FenceClusterNetworkResponse{}, nil
+}

internal/csi-addons/networkfence/fencing.go

@@ -15,10 +15,12 @@ package networkfence
 
 import (
 	"context"
+	"encoding/json"
 	"errors"
 	"fmt"
 	"net"
 	"strings"
+	"time"
 
 	"github.com/ceph/ceph-csi/internal/util"
 	"github.com/ceph/ceph-csi/internal/util/log"
@@ -38,6 +40,11 @@ type NetworkFence struct {
 	cr       *util.Credentials
 }
 
+// activeClient represents the structure of an active client.
+type activeClient struct {
+	Inst string `json:"inst"`
+}
+
 // NewNetworkFence returns a networkFence struct object from the Network fence/unfence request.
 func NewNetworkFence(
 	ctx context.Context,
@@ -132,6 +139,94 @@ func (nf *NetworkFence) AddNetworkFence(ctx context.Context) error {
 	return nil
 }
 
+func listActiveClients(ctx context.Context, mdsDaemonID string) ([]activeClient, error) {
+	// FIXME: replace the ceph command with go-ceph API in future
+	cmd := []string{"tell", fmt.Sprintf("mds.%s", mdsDaemonID), "client", "ls"}
+	stdout, stdErr, err := util.ExecCommandWithTimeout(ctx, 5*time.Minute, "ceph", cmd...)
+	if err != nil {
+		return nil, fmt.Errorf("failed to list active clients: %w, stderr: %q", err, stdErr)
+	}
+
+	var activeClients []activeClient
+	if err := json.Unmarshal([]byte(stdout), &activeClients); err != nil {
+		return nil, fmt.Errorf("failed to unmarshal JSON: %w", err)
+	}
+
+	return activeClients, nil
+}
+
+func evictCephFSClient(ctx context.Context, clientIP string, mdsDaemonID string) error {
+	// FIXME: replace the ceph command with go-ceph API in future
+	cmd := []string{"tell", fmt.Sprintf("mds.%s", mdsDaemonID), "client", "evict", clientIP}
+	_, stdErr, err := util.ExecCommandWithTimeout(ctx, 5*time.Minute, "ceph", cmd...)
+	if err != nil {
+		return fmt.Errorf("failed to evict client %s: %w, stderr: %q", clientIP, err, stdErr)
+	}
+	log.DebugLog(ctx, "client %s has been evicted from CephFS\n", clientIP)
+
+	return nil
+}
+
+func isIPInCIDR(ctx context.Context, ip string, cidr string) bool {
+	// Parse the CIDR block
+	_, ipCidr, err := net.ParseCIDR(cidr)
+	if err != nil {
+		log.ErrorLog(ctx, "error parsing CIDR block %s: %w\n", cidr, err)
+
+		return false
+	}
+
+	// Parse the IP address
+	ipAddress := net.ParseIP(ip)
+	if ipAddress == nil {
+		log.ErrorLog(ctx, "error parsing IP address %s\n", ip)
+
+		return false
+	}
+
+	// Check if the IP address is within the CIDR block
+	return ipCidr.Contains(ipAddress)
+}
+
+func (ac *activeClient) fetchIP() (string, error) {
+	clientInfo := ac.Inst
+	parts := strings.Fields(clientInfo)
+	if len(parts) >= 2 {
+		return parts[1], nil
+	}
+	return "", errors.New("unable to extract IP address")
+}
+
+// AddClientEviction blocks access for all the IPs in the CIDR block
+// using client eviction.
+func (nf *NetworkFence) AddClientEviction(ctx context.Context) error {
+	// Fetch active clients
+	activeClients, err := listActiveClients(ctx, "0")
+	if err != nil {
+		return err
+	}
+	// Iterate through CIDR blocks and check if any active client matches
+	for _, cidr := range nf.Cidr {
+		for _, client := range activeClients {
+			clientIP, err := client.fetchIP()
+			if err != nil {
+				return fmt.Errorf("error fetching client IP: %w", err)
+			}
+			// Check if the clientIP is in the CIDR block
+			if isIPInCIDR(ctx, clientIP, cidr) {
+				// Evict the client if it matches the CIDR block
+				err := evictCephFSClient(ctx, clientIP, "0")
+				if err != nil {
+					return fmt.Errorf("error evicting client %s: %w", clientIP, err)
+				}
+				log.DebugLog(ctx, "client %s has been evicted\n", clientIP)
+			}
+		}
+	}
+
+	return nil
+}
+
 // getIPRange returns a list of IPs from the IP range
 // corresponding to a CIDR block.
 func getIPRange(cidr string) ([]string, error) {