CSI: fix topology matching logic (#24522)

Some plugins emit multiple topology segment entries for the same segment (ex. newer versions of AWS EBS) to accommodate convention changes in k8s. Check that segments are a superset instead of exactly equal to the plugin's topology segments.

.changelog/24522.txt

@@ -0,0 +1,3 @@
+```release-note:bug
+csi: Fixed a bug where drivers that emit multiple topology segments would cause placements to fail
+```

e2e/csi/input/plugin-aws-ebs-controller.nomad

@@ -25,7 +25,7 @@ job "plugin-aws-ebs-controller" {
       driver = "docker"
 
       config {
-        image = "public.ecr.aws/ebs-csi-driver/aws-ebs-csi-driver:v1.5.1"
+        image = "public.ecr.aws/ebs-csi-driver/aws-ebs-csi-driver:v1.33.0"
 
         args = [
           "controller",

e2e/csi/input/plugin-aws-ebs-nodes.nomad

@@ -22,7 +22,7 @@ job "plugin-aws-ebs-nodes" {
       driver = "docker"
 
       config {
-        image = "public.ecr.aws/ebs-csi-driver/aws-ebs-csi-driver:v1.5.1"
+        image = "public.ecr.aws/ebs-csi-driver/aws-ebs-csi-driver:v1.33.0"
 
         args = [
           "node",

nomad/structs/node.go

@@ -68,13 +68,27 @@ func (t *CSITopology) Equal(o *CSITopology) bool {
 	return maps.Equal(t.Segments, o.Segments)
 }
 
+func (t *CSITopology) Contains(o *CSITopology) bool {
+	if t == nil || o == nil {
+		return t == o
+	}
+
+	for k, ov := range o.Segments {
+		if tv, ok := t.Segments[k]; !ok || tv != ov {
+			return false
+		}
+	}
+
+	return true
+}
+
 func (t *CSITopology) MatchFound(o []*CSITopology) bool {
 	if t == nil || o == nil || len(o) == 0 {
 		return false
 	}
 
 	for _, other := range o {
-		if t.Equal(other) {
+		if t.Contains(other) {
 			return true
 		}
 	}

nomad/structs/node_test.go

@@ -163,3 +163,94 @@ func TestNodeMeta_Validate(t *testing.T) {
 		})
 	}
 }
+
+func TestCSITopology_Contains(t *testing.T) {
+	ci.Parallel(t)
+
+	cases := []struct {
+		name     string
+		this     *CSITopology
+		other    *CSITopology
+		expected bool
+	}{
+		{
+			name: "AWS EBS pre 1.27 behavior",
+			this: &CSITopology{
+				Segments: map[string]string{
+					"topology.ebs.csi.aws.com/zone": "us-east-1a",
+				},
+			},
+			other: &CSITopology{
+				Segments: map[string]string{
+					"topology.ebs.csi.aws.com/zone": "us-east-1a",
+				},
+			},
+			expected: true,
+		},
+		{
+			name: "AWS EBS post 1.27 behavior",
+			this: &CSITopology{
+				Segments: map[string]string{
+					"topology.kubernetes.io/zone":   "us-east-1a",
+					"topology.ebs.csi.aws.com/zone": "us-east-1a",
+					"kubernetes.io/os":              "linux",
+				},
+			},
+			other: &CSITopology{
+				Segments: map[string]string{
+					"topology.kubernetes.io/zone": "us-east-1a",
+				},
+			},
+			expected: true,
+		},
+		{
+			name: "other contains invalid segment value for matched key",
+			this: &CSITopology{
+				Segments: map[string]string{
+					"topology.kubernetes.io/zone":   "us-east-1a",
+					"topology.ebs.csi.aws.com/zone": "us-east-1a",
+					"kubernetes.io/os":              "linux",
+				},
+			},
+			other: &CSITopology{
+				Segments: map[string]string{
+					"topology.kubernetes.io/zone": "us-east-1a",
+					"kubernetes.io/os":            "windows",
+				},
+			},
+			expected: false,
+		},
+		{
+			name: "other contains invalid segment key",
+			this: &CSITopology{
+				Segments: map[string]string{
+					"topology.kubernetes.io/zone": "us-east-1a",
+				},
+			},
+			other: &CSITopology{
+				Segments: map[string]string{
+					"topology.kubernetes.io/zone": "us-east-1a",
+					"kubernetes.io/os":            "linux",
+				},
+			},
+			expected: false,
+		},
+		{
+			name: "other is nil",
+			this: &CSITopology{
+				Segments: map[string]string{
+					"topology.kubernetes.io/zone": "us-east-1a",
+				},
+			},
+			other:    nil,
+			expected: false,
+		},
+	}
+
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			must.Eq(t, tc.expected, tc.this.Contains(tc.other))
+		})
+	}
+
+}