KEP-5304: DecodeMetadataFromStream only skips metadata entries with unknown API versions

Signed-off-by: Alay Patel <alayp@nvidia.com>

staging/src/k8s.io/dynamic-resource-allocation/devicemetadata/decode.go

@@ -44,8 +44,10 @@ var (
 // for the file format); this function tries each in order and populates dest
 // with the first one it can successfully decode and convert.
 //
-// Objects that cannot be decoded or converted are skipped so that a driver
-// upgrade does not break older consumers.
+// Entries whose apiVersion is not registered in the scheme (e.g. a newer
+// version written by an upgraded driver) are skipped silently so that a
+// driver upgrade does not break older consumers. Any other failure is
+// fatal. dest is only populated on a nil return.
 //
 // dest must be a pointer to a type registered in the metadata scheme (e.g.,
 // *v1alpha1.DeviceMetadata or *metadata.DeviceMetadata). The internal type
@@ -60,26 +62,26 @@ func DecodeMetadataFromStream(decoder *json.Decoder, dest runtime.Object) error
 
 	deserializer := codecFactory.UniversalDeserializer()
 
-	var skippedErrors []string
+	var unknownVersions []string
 	for decoder.More() {
 		var raw json.RawMessage
 		if err := decoder.Decode(&raw); err != nil {
 			return fmt.Errorf("read metadata object from stream: %w", err)
 		}
-
 		obj, gvk, err := deserializer.Decode(raw, nil, nil)
 		if err != nil {
-			if gvk != nil {
-				skippedErrors = append(skippedErrors, fmt.Sprintf("%s: %v", gvk.GroupVersion(), err))
-			} else {
-				skippedErrors = append(skippedErrors, err.Error())
+			if gvk == nil || gvk.Version == "" {
+				return fmt.Errorf("decode metadata object: %w", err)
 			}
-			continue
+			if runtime.IsNotRegisteredError(err) && !scheme.IsVersionRegistered(gvk.GroupVersion()) {
+				unknownVersions = append(unknownVersions, gvk.GroupVersion().String())
+				continue
+			}
+			return fmt.Errorf("decode %s: %w", gvk.GroupVersion(), err)
 		}
 
 		if err := scheme.Convert(obj, dest, nil); err != nil {
-			skippedErrors = append(skippedErrors, fmt.Sprintf("%s: convert: %v", obj.GetObjectKind().GroupVersionKind().GroupVersion(), err))
-			continue
+			return fmt.Errorf("convert %s: %w", obj.GetObjectKind().GroupVersionKind().GroupVersion(), err)
 		}
 
 		// scheme.Convert does not propagate TypeMeta. Set it explicitly
@@ -90,10 +92,10 @@ func DecodeMetadataFromStream(decoder *json.Decoder, dest runtime.Object) error
 		}
 		return nil
 	}
-	if len(skippedErrors) > 0 {
-		return fmt.Errorf("no compatible metadata version found in stream (errors: %s)", strings.Join(skippedErrors, "; "))
+	if len(unknownVersions) == 0 {
+		return fmt.Errorf("no metadata objects found in stream")
 	}
-	return fmt.Errorf("no metadata objects found in stream")
+	return fmt.Errorf("no compatible metadata version found in stream (unknown versions: %s)", strings.Join(unknownVersions, ", "))
 }
 
 // ReadResourceClaimMetadataWithDriverName reads and decodes the metadata file

staging/src/k8s.io/dynamic-resource-allocation/devicemetadata/decode_test.go

@@ -69,8 +69,10 @@ func mustMarshal(obj interface{}) []byte {
 func TestDecodeMetadataFromStream(t *testing.T) {
 	unknownVersionJSON := `{"apiVersion":"metadata.k8s.io/v99","kind":"DeviceMetadata","metadata":{"name":"test"}}` + "\n"
 	unknownV2JSON := `{"apiVersion":"metadata.k8s.io/v100","kind":"DeviceMetadata","metadata":{"name":"test"}}` + "\n"
+	unknownVersionMissingKindJSON := `{"apiVersion":"metadata.k8s.io/v99","metadata":{"name":"test"}}` + "\n"
 	missingKindJSON := `{"apiVersion":"metadata.resource.k8s.io/v1alpha1","metadata":{"name":"test"}}` + "\n"
 	missingApiversionJSON := `{"kind":"DeviceMetadata","metadata":{"name":"test"}}` + "\n"
+	malformedV1Alpha1JSON := `{"apiVersion":"metadata.resource.k8s.io/v1alpha1","kind":"DeviceMetadata","metadata":{"name":"test"},"requests":"this-should-be-an-array"}` + "\n"
 
 	expectedV1Alpha1 := &v1alpha1.DeviceMetadata{
 		TypeMeta: metav1.TypeMeta{
@@ -150,55 +152,62 @@ func TestDecodeMetadataFromStream(t *testing.T) {
 		"missing-kind": {
 			streamInput: []byte(missingKindJSON),
 			dest:        &v1alpha1.DeviceMetadata{},
-			expectError: "no compatible metadata version found in stream",
+			expectError: "decode metadata.resource.k8s.io/v1alpha1",
 		},
 		"missing-apiversion": {
 			streamInput: []byte(missingApiversionJSON),
 			dest:        &v1alpha1.DeviceMetadata{},
-			expectError: "no compatible metadata version found in stream",
+			expectError: "decode metadata object",
+		},
+		"malformed-registered-version": {
+			streamInput: []byte(malformedV1Alpha1JSON),
+			dest:        &v1alpha1.DeviceMetadata{},
+			expectError: "decode metadata.resource.k8s.io/v1alpha1",
 		},
 		"only-unknown-versions": {
 			streamInput: []byte(unknownVersionJSON),
 			dest:        &v1alpha1.DeviceMetadata{},
-			expectError: "no compatible metadata version found in stream",
+			expectError: "no compatible metadata version found in stream (unknown versions: metadata.k8s.io/v99)",
+		},
+		"unknown-version-missing-kind": {
+			streamInput: []byte(unknownVersionMissingKindJSON),
+			dest:        &v1alpha1.DeviceMetadata{},
+			expectError: "decode metadata.k8s.io/v99",
 		},
 		"multiple-unknown-versions": {
 			streamInput: append([]byte(unknownVersionJSON), []byte(unknownV2JSON)...),
 			dest:        &v1alpha1.DeviceMetadata{},
-			expectError: "no compatible metadata version found in stream",
-		},
-		"unknown-version-then-broken": {
-			streamInput: append([]byte(unknownVersionJSON), []byte(missingKindJSON)...),
-			dest:        &v1alpha1.DeviceMetadata{},
-			expectError: "no compatible metadata version found in stream",
+			expectError: "no compatible metadata version found in stream (unknown versions: metadata.k8s.io/v99, metadata.k8s.io/v100)",
 		},
 		"known-version-then-broken": {
 			streamInput: append(validV1Alpha1JSON, []byte("{broken")...),
 			dest:        &v1alpha1.DeviceMetadata{},
 			expected:    expectedV1Alpha1,
 		},
-
-		// Forward compatibility: object-level errors are skipped so
-		// that an older consumer can reach a version it understands.
 		"skips-unknown-version": {
 			streamInput: append([]byte(unknownVersionJSON), validV1Alpha1JSON...),
 			dest:        &v1alpha1.DeviceMetadata{},
 			expected:    expectedV1Alpha1,
 		},
-		"skips-missing-kind": {
+		"malformed-registered-then-valid-is-fatal": {
+			streamInput: append([]byte(malformedV1Alpha1JSON), validV1Alpha1JSON...),
+			dest:        &v1alpha1.DeviceMetadata{},
+			expectError: "decode metadata.resource.k8s.io/v1alpha1",
+		},
+		"missing-kind-then-valid-is-fatal": {
 			streamInput: append([]byte(missingKindJSON), validV1Alpha1JSON...),
 			dest:        &v1alpha1.DeviceMetadata{},
-			expected:    expectedV1Alpha1,
+			expectError: "decode metadata.resource.k8s.io/v1alpha1",
 		},
-		"skips-missing-apiversion": {
+		"missing-apiversion-then-valid-is-fatal": {
 			streamInput: append([]byte(missingApiversionJSON), validV1Alpha1JSON...),
 			dest:        &v1alpha1.DeviceMetadata{},
-			expected:    expectedV1Alpha1,
+			expectError: "decode metadata object",
 		},
-		"skips-multiple-errors": {
-			streamInput: append(append([]byte(unknownVersionJSON), []byte(missingKindJSON)...), validV1Alpha1JSON...),
+		"unknown-version-then-malformed-is-fatal": {
+			streamInput: append([]byte(unknownVersionJSON), []byte(missingKindJSON)...),
 			dest:        &v1alpha1.DeviceMetadata{},
-			expected:    expectedV1Alpha1,
+			expectError: "decode metadata.resource.k8s.io/v1alpha1",
 		},
 	}