Implement SnapshotMetadataService

Signed-off-by: Prasad Ghangal <[email protected]>

Author: PrasadG193

pkg/hostpath/hostpath.go

@@ -17,6 +17,7 @@ limitations under the License.
 package hostpath
 
 import (
+	"bytes"
 	"errors"
 	"fmt"
 	"io"
@@ -26,6 +27,7 @@ import (
 	"sync"
 
 	"github.com/container-storage-interface/spec/lib/go/csi"
+	"golang.org/x/net/context"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
 	"k8s.io/apimachinery/pkg/api/resource"
@@ -54,6 +56,7 @@ type hostPath struct {
 	csi.UnimplementedControllerServer
 	csi.UnimplementedNodeServer
 	csi.UnimplementedGroupControllerServer
+	csi.UnimplementedSnapshotMetadataServer
 	config Config
 
 	// gRPC calls involving any of the fields below must be serialized
@@ -80,6 +83,7 @@ type Config struct {
 	EnableTopology                bool
 	EnableVolumeExpansion         bool
 	EnableControllerModifyVolume  bool
+	EnableSnapshotMetadata        bool
 	AcceptedMutableParameterNames StringArray
 	DisableControllerExpansion    bool
 	DisableNodeExpansion          bool
@@ -130,7 +134,7 @@ func NewHostPathDriver(cfg Config) (*hostPath, error) {
 func (hp *hostPath) Run() error {
 	s := NewNonBlockingGRPCServer()
 	// hp itself implements ControllerServer, NodeServer, and IdentityServer.
-	s.Start(hp.config.Endpoint, hp, hp, hp, hp)
+	s.Start(hp.config.Endpoint, hp, hp, hp, hp, hp)
 	s.Wait()
 
 	return nil
@@ -410,3 +414,87 @@ func (hp *hostPath) createSnapshotFromVolume(vol state.Volume, file string, opts
 
 	return nil
 }
+
+func (hp *hostPath) getChangedBlockMetadata(ctx context.Context, sourcePath, targetPath string, startingOffset, blockSize int64, maxResult int32, changedBlocksChan chan<- []*csi.BlockMetadata) error {
+	klog.V(4).Infof("finding changed blocks between two files: %s, %s", sourcePath, targetPath)
+	defer close(changedBlocksChan)
+	// Open the two files
+	source, err := os.Open(sourcePath)
+	if err != nil {
+		klog.Errorf("failed to read file: %v", err)
+		return err
+	}
+	defer source.Close()
+	target, err := os.Open(targetPath)
+	if err != nil {
+		klog.Errorf("failed to read file: %v", err)
+		return err
+	}
+	defer target.Close()
+
+	// Seek to the desired offset in both files
+	_, err = source.Seek(startingOffset, 0)
+	if err != nil {
+		klog.Errorf("failed to seek file: %v", err)
+		return err
+	}
+	_, err = target.Seek(startingOffset, 0)
+	if err != nil {
+		klog.Errorf("failed to seek file: %v", err)
+		return err
+	}
+
+	// Read both files block by block and compare them
+	var blockIndex int64
+	buffer1 := make([]byte, blockSize)
+	buffer2 := make([]byte, blockSize)
+	for {
+		changedBlocks := []*csi.BlockMetadata{}
+		// Read blocks and compare them. Create the list of changed blocks metadata.
+		// Once the number of blocks reaches to maxResult, return the result and
+		// compute next batch of blocks.
+		for i := 1; i <= int(maxResult); i++ {
+			select {
+			case <-ctx.Done(): // Detect cancellation from the client
+				klog.Infof("Stream canceled by client. Exiting goroutine.")
+				return nil
+			default:
+				// Read block from source and target files
+				n1, err1 := source.Read(buffer1)
+				n2, err2 := target.Read(buffer2)
+
+				// If both files have reached EOF, exit the loop
+				if err1 == io.EOF && err2 == io.EOF {
+					klog.Infof("reached to end of the file\n")
+					return nil
+				}
+				if err1 != nil && err1 != io.EOF {
+					klog.Errorf("failed to read to buffer: %v", err)
+					return err1
+				}
+				if err2 != nil && err2 != io.EOF {
+					klog.Errorf("failed to read to buffer: %v", err)
+					return err2
+				}
+
+				// If the number of bytes read differs, the files are different
+				if n1 != n2 {
+					klog.Infof("files differ in size at block %d\n", blockIndex)
+					return nil
+				}
+
+				// Compare the two blocks and add result
+				if !bytes.Equal(buffer1[:n1], buffer2[:n2]) {
+					changedBlocks = append(changedBlocks, &csi.BlockMetadata{
+						ByteOffset: blockIndex * blockSize,
+						SizeBytes:  int64(blockSize),
+					})
+				}
+				blockIndex++
+			}
+		}
+		if len(changedBlocks) != 0 {
+			changedBlocksChan <- changedBlocks
+		}
+	}
+}

pkg/hostpath/identityserver.go

@@ -62,6 +62,13 @@ func (hp *hostPath) GetPluginCapabilities(ctx context.Context, req *csi.GetPlugi
 				},
 			},
 		},
+		{
+			Type: &csi.PluginCapability_Service_{
+				Service: &csi.PluginCapability_Service{
+					Type: csi.PluginCapability_Service_SNAPSHOT_METADATA_SERVICE,
+				},
+			},
+		},
 	}
 	if hp.config.EnableTopology {
 		caps = append(caps, &csi.PluginCapability{

pkg/hostpath/server.go

@@ -40,11 +40,11 @@ type nonBlockingGRPCServer struct {
 	cleanup func()
 }
 
-func (s *nonBlockingGRPCServer) Start(endpoint string, ids csi.IdentityServer, cs csi.ControllerServer, ns csi.NodeServer, gcs csi.GroupControllerServer) {
+func (s *nonBlockingGRPCServer) Start(endpoint string, ids csi.IdentityServer, cs csi.ControllerServer, ns csi.NodeServer, gcs csi.GroupControllerServer, sms csi.SnapshotMetadataServer) {
 
 	s.wg.Add(1)
 
-	go s.serve(endpoint, ids, cs, ns, gcs)
+	go s.serve(endpoint, ids, cs, ns, gcs, sms)
 
 	return
 }
@@ -63,7 +63,7 @@ func (s *nonBlockingGRPCServer) ForceStop() {
 	s.cleanup()
 }
 
-func (s *nonBlockingGRPCServer) serve(ep string, ids csi.IdentityServer, cs csi.ControllerServer, ns csi.NodeServer, gcs csi.GroupControllerServer) {
+func (s *nonBlockingGRPCServer) serve(ep string, ids csi.IdentityServer, cs csi.ControllerServer, ns csi.NodeServer, gcs csi.GroupControllerServer, sms csi.SnapshotMetadataServer) {
 	listener, cleanup, err := endpoint.Listen(ep)
 	if err != nil {
 		klog.Fatalf("Failed to listen: %v", err)
@@ -88,6 +88,9 @@ func (s *nonBlockingGRPCServer) serve(ep string, ids csi.IdentityServer, cs csi.
 	if gcs != nil {
 		csi.RegisterGroupControllerServer(server, gcs)
 	}
+	if sms != nil {
+		csi.RegisterSnapshotMetadataServer(server, sms)
+	}
 
 	klog.Infof("Listening for connections on address: %#v", listener.Addr())
 

pkg/hostpath/snapshotmetadataserver.go

@@ -0,0 +1,102 @@
+/*
+Copyright 2024 The Kubernetes 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 hostpath
+
+import (
+	"fmt"
+
+	"github.com/container-storage-interface/spec/lib/go/csi"
+	"github.com/kubernetes-csi/csi-driver-host-path/pkg/state"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
+	"k8s.io/klog/v2"
+)
+
+func (hp *hostPath) GetMetadataAllocated(req *csi.GetMetadataAllocatedRequest, stream csi.SnapshotMetadata_GetMetadataAllocatedServer) error {
+	return nil
+}
+
+func (hp *hostPath) GetMetadataDelta(req *csi.GetMetadataDeltaRequest, stream csi.SnapshotMetadata_GetMetadataDeltaServer) error {
+	ctx := stream.Context()
+	// Check arguments
+	baseSnapID := req.GetBaseSnapshotId()
+	targetSnapID := req.GetTargetSnapshotId()
+	if len(baseSnapID) == 0 {
+		return status.Error(codes.InvalidArgument, "BaseSnapshotID missing in request")
+	}
+	if len(targetSnapID) == 0 {
+		return status.Error(codes.InvalidArgument, "TargetSnapshotID missing in request")
+	}
+
+	// Load snapshots
+	source, err := hp.state.GetSnapshotByID(baseSnapID)
+	if err != nil {
+		return status.Error(codes.Internal, "Cannot find the source snapshot")
+	}
+	target, err := hp.state.GetSnapshotByID(targetSnapID)
+	if err != nil {
+		return status.Error(codes.Internal, "Cannot find the target snapshot")
+	}
+
+	if !source.ReadyToUse {
+		return status.Error(codes.Unavailable, fmt.Sprintf("snapshot %v is not yet ready to use", baseSnapID))
+	}
+	if !target.ReadyToUse {
+		return status.Error(codes.Unavailable, fmt.Sprintf("snapshot %v is not yet ready to use", targetSnapID))
+	}
+
+	if source.VolID != target.VolID {
+		return status.Error(codes.InvalidArgument, "Snapshots don't belong to the same Volume")
+	}
+	vol, err := hp.state.GetVolumeByID(source.VolID)
+	if err != nil {
+		return err
+	}
+	if vol.VolAccessType != state.BlockAccess {
+		return status.Error(codes.InvalidArgument, "Source volume does not have block mode access type")
+	}
+
+	changedBlocks := make(chan []*csi.BlockMetadata)
+	go func() {
+		err := hp.getChangedBlockMetadata(ctx, hp.getSnapshotPath(baseSnapID), hp.getSnapshotPath(targetSnapID), req.StartingOffset, state.BlockSizeBytes, req.MaxResults, changedBlocks)
+		if err != nil {
+			klog.Errorf("failed to get changed block metadata: %v", err)
+		}
+	}()
+
+	for {
+		select {
+		case cb, ok := <-changedBlocks:
+			if !ok {
+				klog.V(4).Info("channel closed, returning")
+				return nil
+			}
+			resp := csi.GetMetadataDeltaResponse{
+				BlockMetadataType:   csi.BlockMetadataType_FIXED_LENGTH,
+				VolumeCapacityBytes: vol.VolSize,
+				BlockMetadata:       cb,
+			}
+			if err := stream.Send(&resp); err != nil {
+				return err
+			}
+		case <-ctx.Done():
+			klog.V(4).Info("received cancellation signal, returning")
+			return nil
+		}
+	}
+	return nil
+}

pkg/state/state.go

@@ -34,6 +34,9 @@ type AccessType int
 const (
 	MountAccess AccessType = iota
 	BlockAccess
+
+	// BlockSizeBytes represents the default block size.
+	BlockSizeBytes = 4096
 )
 
 type Volume struct {