feat(1-1-restore): reduces --keyspace param features to keyspace leve…

…l filtering only

This reduces functionality of `--keyspace` parameter to support only:
 - keyspace level filtering. Table level filtering (system.table_name) is forbidden.
 - only include filtering. Excluding(!) is forbidden.
 - wildcards (*) are forbidden. The only exception is when the only
   filter that is provided is a single character (*), meaning that all
   keyspaces should be included into restore process.

Refs: #4253

pkg/service/one2onerestore/model.go

@@ -3,6 +3,8 @@
 package one2onerestore
 
 import (
+	"strings"
+
 	"github.com/pkg/errors"
 	. "github.com/scylladb/scylla-manager/v3/pkg/service/backup/backupspec"
 	"github.com/scylladb/scylla-manager/v3/pkg/util/uuid"
@@ -51,12 +53,32 @@ func (t *Target) validateProperties() error {
 	if t.SourceClusterID == uuid.Nil {
 		return errors.New("source cluster id is empty")
 	}
+	if err := validateKeyspaceFilter(t.Keyspace); err != nil {
+		return errors.Wrap(err, "keyspace filter")
+	}
 	if err := validateNodesMapping(t.NodesMapping); err != nil {
 		return errors.Wrap(err, "nodes mapping")
 	}
 	return nil
 }
 
+// 1-1-restore --keyspace filter is limited to keyspaces only (e.g. keyspace.table is not supported).
+// Also exclude operations (!) are forbidden as well.
+func validateKeyspaceFilter(keyspaces []string) error {
+	for _, filter := range keyspaces {
+		if strings.Contains(filter, ".") {
+			return errors.Errorf("only keyspace level filtering is allowed, but table is provided: %s", filter)
+		}
+		if strings.HasPrefix(filter, "!") {
+			return errors.Errorf("exclude filter(!) is not supported: %s", filter)
+		}
+		if strings.Contains(filter, "*") && len(filter) != 1 {
+			return errors.Errorf("wildcard pattern(*) is not supported: %s", filter)
+		}
+	}
+	return nil
+}
+
 type dcRack struct {
 	dc   string
 	rack string

pkg/service/one2onerestore/model_test.go

@@ -121,3 +121,51 @@ func TestValidateNodesMapping(t *testing.T) {
 		})
 	}
 }
+
+func TestValidateKeyspaceFilter(t *testing.T) {
+	testCases := []struct {
+		name        string
+		keyspace    []string
+		expectedErr string
+	}{
+		{
+			name:     "Only keyspace level filtering",
+			keyspace: []string{"hello", "system"},
+		},
+		{
+			name:        "Include all * is supported (default)",
+			keyspace:    []string{"*"},
+			expectedErr: "",
+		},
+		{
+			name:        "Table level is not allowed",
+			keyspace:    []string{"hello", "system.table"},
+			expectedErr: "only keyspace level filtering is allowed, but table is provided: system.table",
+		},
+		{
+			name:        "Exclude filters are not supported",
+			keyspace:    []string{"hello", "!world"},
+			expectedErr: "exclude filter(!) is not supported: !world",
+		},
+		{
+			name:        "Wildcard patterns are not supported",
+			keyspace:    []string{"hello*"},
+			expectedErr: "wildcard pattern(*) is not supported: hello*",
+		},
+	}
+
+	for _, tc := range testCases {
+		errMsg := func(err error) string {
+			if err == nil {
+				return ""
+			}
+			return err.Error()
+		}
+		t.Run(tc.name, func(t *testing.T) {
+			err := validateKeyspaceFilter(tc.keyspace)
+			if errMsg(err) != tc.expectedErr {
+				t.Fatalf("Expected err %q, but got %q", tc.expectedErr, errMsg(err))
+			}
+		})
+	}
+}