fix(CSI-241): disregard sync_on_close in mountmap per FS (#310)

### TL;DR

Introduced a new `AsMapKey()` method for `MountOptions` to create version-agnostic map keys.

### What changed?

- Added a new `AsMapKey()` method to the `MountOptions` struct in `mountoptions.go`.
- This method creates a version of the mount options that excludes certain options (currently only `MountOptionSyncOnClose`) to be used as a map key.
- Updated `mounter.go` to use `AsMapKey()` instead of `String()` when accessing or modifying the `mountMap`.

### How to test?
Covered by unit testing

### Why make this change?

This change allows for more consistent handling of mount options across different versions of the Weka cluster / client. By excluding version-specific options like `MountOptionSyncOnClose` from the map key, we can avoid unnecessary duplication of mounts and ensure that mounts with the same essential options are treated as identical, regardless of version-specific differences.

---

pkg/wekafs/mounter.go

@@ -45,7 +45,7 @@ func (m *wekaMounter) NewMount(fsName string, options MountOptions) *wekaMount {
 	if _, ok := m.mountMap[fsName]; !ok {
 		m.mountMap[fsName] = mountsMapPerFs{}
 	}
-	if _, ok := m.mountMap[fsName][options.String()]; !ok {
+	if _, ok := m.mountMap[fsName][options.AsMapKey()]; !ok {
 		uniqueId := getStringSha1AsB32(fsName + ":" + options.String())
 		wMount := &wekaMount{
 			kMounter:                m.kMounter,
@@ -55,10 +55,10 @@ func (m *wekaMounter) NewMount(fsName string, options MountOptions) *wekaMount {
 			mountOptions:            options,
 			allowProtocolContainers: m.allowProtocolContainers,
 		}
-		m.mountMap[fsName][options.String()] = wMount
+		m.mountMap[fsName][options.AsMapKey()] = wMount
 	}
 	m.lock.Unlock()
-	return m.mountMap[fsName][options.String()]
+	return m.mountMap[fsName][options.AsMapKey()]
 }
 
 type UnmountFunc func()
@@ -119,10 +119,10 @@ func (m *wekaMounter) unmountWithOptions(ctx context.Context, fsName string, opt
 	options.setSelinux(m.getSelinuxStatus(ctx))
 
 	log.Ctx(ctx).Trace().Strs("mount_options", opts.Strings()).Str("filesystem", fsName).Msg("Received an unmount request")
-	if mnt, ok := m.mountMap[fsName][options.String()]; ok {
+	if mnt, ok := m.mountMap[fsName][options.AsMapKey()]; ok {
 		err := mnt.decRef(ctx)
 		if err == nil {
-			if m.mountMap[fsName][options.String()].refCount <= 0 {
+			if m.mountMap[fsName][options.AsMapKey()].refCount <= 0 {
 				log.Ctx(ctx).Trace().Str("filesystem", fsName).Strs("mount_options", options.Strings()).Msg("This is a last use of this mount, removing from map")
 				delete(m.mountMap[fsName], options.String())
 			}

pkg/wekafs/mountoptions.go

@@ -145,6 +145,16 @@ func (opts MountOptions) Hash() uint32 {
 	return h.Sum32()
 }
 
+func (opts MountOptions) AsMapKey() string {
+	ret := opts
+	// TODO: if adding any other version-agnostic options, add them here
+	excludedOpts := []string{MountOptionSyncOnClose}
+	for _, o := range excludedOpts {
+		ret = ret.RemoveOption(o)
+	}
+	return ret.String()
+}
+
 func (opts MountOptions) setSelinux(selinuxSupport bool) {
 	if selinuxSupport {
 		o := newMountOptionFromString(fmt.Sprintf("fscontext=\"system_u:object_r:%s_t:s0\"", selinuxContext))