Support for schema id in header (#1431)

* DGS-20366 Support schema ID in header

* Use id serdes

* Minor cleanup

* Minor cleanup

* Fix comments

* Add test

* Minor renaming

* Add test

* Fix get id with response

Author: rayokota

schemaregistry/internal/rest_service.go

@@ -48,6 +48,7 @@ const (
 	Contexts                 = "/contexts"
 	SchemasBySubject         = "/schemas/ids/%d?subject=%s"
 	SubjectsAndVersionsByID  = "/schemas/ids/%d/versions"
+	SchemasByGUID            = "/schemas/guids/%s"
 	Subject                  = "/subjects"
 	Subjects                 = Subject + "/%s"
 	SubjectsNormalize        = Subject + "/%s?normalize=%t"

schemaregistry/mock_schemaregistry_client.go

@@ -19,6 +19,7 @@ package schemaregistry
 import (
 	"errors"
 	"fmt"
+	"github.com/google/uuid"
 	"net/url"
 	"reflect"
 	"sort"
@@ -67,6 +68,8 @@ type mockclient struct {
 	infoToSchemaCacheLock    sync.RWMutex
 	idToSchemaCache          map[subjectID]infoCacheEntry
 	idToSchemaCacheLock      sync.RWMutex
+	guidToSchemaCache        map[string]infoCacheEntry
+	guidToSchemaCacheLock    sync.RWMutex
 	schemaToVersionCache     map[subjectJSON]versionCacheEntry
 	schemaToVersionCacheLock sync.RWMutex
 	configCache              map[string]ServerConfig
@@ -118,7 +121,7 @@ func (c *mockclient) RegisterFullResponse(subject string, schema SchemaInfo, nor
 		return *cacheEntryVal.metadata, nil
 	}
 
-	id, err := c.getIDFromRegistry(subject, schema)
+	id, guid, err := c.getIDFromRegistry(subject, schema)
 	if err != nil {
 		return SchemaMetadata{
 			ID: -1,
@@ -127,14 +130,15 @@ func (c *mockclient) RegisterFullResponse(subject string, schema SchemaInfo, nor
 	result = SchemaMetadata{
 		SchemaInfo: schema,
 		ID:         id,
+		GUID:       guid,
 	}
 	c.infoToSchemaCacheLock.Lock()
 	c.infoToSchemaCache[cacheKey] = metadataCacheEntry{&result, false}
 	c.infoToSchemaCacheLock.Unlock()
 	return result, nil
 }
 
-func (c *mockclient) getIDFromRegistry(subject string, schema SchemaInfo) (int, error) {
+func (c *mockclient) getIDFromRegistry(subject string, schema SchemaInfo) (int, string, error) {
 	var id = -1
 	c.idToSchemaCacheLock.RLock()
 	for key, value := range c.idToSchemaCache {
@@ -144,9 +148,18 @@ func (c *mockclient) getIDFromRegistry(subject string, schema SchemaInfo) (int,
 		}
 	}
 	c.idToSchemaCacheLock.RUnlock()
+	var guid string
+	c.guidToSchemaCacheLock.RLock()
+	for key, value := range c.guidToSchemaCache {
+		if schemasEqual(*value.info, schema) {
+			guid = key
+			break
+		}
+	}
+	c.guidToSchemaCacheLock.RUnlock()
 	err := c.generateVersion(subject, schema)
 	if err != nil {
-		return -1, err
+		return -1, "", err
 	}
 	if id < 0 {
 		id = c.counter.increment()
@@ -157,8 +170,13 @@ func (c *mockclient) getIDFromRegistry(subject string, schema SchemaInfo) (int,
 		c.idToSchemaCacheLock.Lock()
 		c.idToSchemaCache[idCacheKey] = infoCacheEntry{&schema, false}
 		c.idToSchemaCacheLock.Unlock()
+
+		guid = uuid.New().String()
+		c.guidToSchemaCacheLock.Lock()
+		c.guidToSchemaCache[guid] = infoCacheEntry{&schema, false}
+		c.guidToSchemaCacheLock.Unlock()
 	}
-	return id, nil
+	return id, guid, nil
 }
 
 func (c *mockclient) generateVersion(subject string, schema SchemaInfo) error {
@@ -204,6 +222,23 @@ func (c *mockclient) GetBySubjectAndID(subject string, id int) (schema SchemaInf
 	return SchemaInfo{}, &posErr
 }
 
+// GetByGUID returns the schema identified by guid
+// Returns Schema object on success
+func (c *mockclient) GetByGUID(guid string) (schema SchemaInfo, err error) {
+	c.guidToSchemaCacheLock.RLock()
+	cacheEntryValue, ok := c.guidToSchemaCache[guid]
+	c.guidToSchemaCacheLock.RUnlock()
+	if ok {
+		return *cacheEntryValue.info, nil
+	}
+	posErr := url.Error{
+		Op:  "GET",
+		URL: c.url.String() + fmt.Sprintf(internal.SchemasByGUID, guid),
+		Err: errors.New("Schema Not Found"),
+	}
+	return SchemaInfo{}, &posErr
+}
+
 func (c *mockclient) GetSubjectsAndVersionsByID(id int) (subjectsAndVersions []SubjectAndVersion, err error) {
 	subjectsAndVersions = make([]SubjectAndVersion, 0)
 
@@ -255,10 +290,21 @@ func (c *mockclient) GetSubjectsAndVersionsByID(id int) (subjectsAndVersions []S
 
 // GetID checks if a schema has been registered with the subject. Returns ID if the registration can be found
 func (c *mockclient) GetID(subject string, schema SchemaInfo, normalize bool) (id int, err error) {
-	schemaJSON, err := schema.MarshalJSON()
+	metadata, err := c.GetIDFullResponse(subject, schema, normalize)
 	if err != nil {
 		return -1, err
 	}
+	return metadata.ID, err
+}
+
+// GetIDFullResponse checks if a schema has been registered with the subject. Returns ID if the registration can be found
+func (c *mockclient) GetIDFullResponse(subject string, schema SchemaInfo, normalize bool) (result SchemaMetadata, err error) {
+	schemaJSON, err := schema.MarshalJSON()
+	if err != nil {
+		return SchemaMetadata{
+			ID: -1,
+		}, err
+	}
 	cacheKey := subjectJSON{
 		subject: subject,
 		json:    string(schemaJSON),
@@ -270,15 +316,17 @@ func (c *mockclient) GetID(subject string, schema SchemaInfo, normalize bool) (i
 	}
 	c.infoToSchemaCacheLock.RUnlock()
 	if ok {
-		return cacheEntryVal.metadata.ID, nil
+		return *cacheEntryVal.metadata, nil
 	}
 
 	posErr := url.Error{
 		Op:  "GET",
 		URL: c.url.String() + fmt.Sprintf(internal.Subjects, url.PathEscape(subject)),
 		Err: errors.New("Subject Not found"),
 	}
-	return -1, &posErr
+	return SchemaMetadata{
+		ID: -1,
+	}, &posErr
 }
 
 // GetLatestSchemaMetadata fetches latest version registered with the provided subject
@@ -345,10 +393,28 @@ func (c *mockclient) GetSchemaMetadataIncludeDeleted(subject string, version int
 		}
 		return SchemaMetadata{}, &posErr
 	}
+	var guid string
+	c.guidToSchemaCacheLock.RLock()
+	for key, value := range c.guidToSchemaCache {
+		if schemasEqual(*value.info, info) && (!value.softDeleted || deleted) {
+			guid = key
+			break
+		}
+	}
+	c.guidToSchemaCacheLock.RUnlock()
+	if guid == "" {
+		posErr := url.Error{
+			Op:  "GET",
+			URL: c.url.String() + fmt.Sprintf(internal.Versions, url.PathEscape(subject), version),
+			Err: errors.New("Subject Not found"),
+		}
+		return SchemaMetadata{}, &posErr
+	}
 	return SchemaMetadata{
 		SchemaInfo: info,
 
 		ID:      id,
+		GUID:    guid,
 		Subject: subject,
 		Version: version,
 	}, nil

schemaregistry/rules/encryption/deks/dekregistry_client.go

@@ -397,7 +397,7 @@ func (c *client) GetDekVersion(kekName string, subject string, version int, algo
 	return dek, err
 }
 
-// GetEncryptedKeyMaterialBytes returns the EncryptedKeyMaterialBytes
+// GetDekEncryptedKeyMaterialBytes returns the EncryptedKeyMaterialBytes
 func (c *client) GetDekEncryptedKeyMaterialBytes(dek *Dek) ([]byte, error) {
 	if dek.EncryptedKeyMaterial == "" {
 		return nil, nil
@@ -416,7 +416,7 @@ func (c *client) GetDekEncryptedKeyMaterialBytes(dek *Dek) ([]byte, error) {
 	return dek.EncryptedKeyMaterialBytes, nil
 }
 
-// GetKeyMaterialBytes returns the KeyMaterialBytes
+// GetDekKeyMaterialBytes returns the KeyMaterialBytes
 func (c *client) GetDekKeyMaterialBytes(dek *Dek) ([]byte, error) {
 	if dek.KeyMaterial == "" {
 		return nil, nil
@@ -435,7 +435,7 @@ func (c *client) GetDekKeyMaterialBytes(dek *Dek) ([]byte, error) {
 	return dek.KeyMaterialBytes, nil
 }
 
-// SetKeyMaterial sets the KeyMaterial using the given bytes
+// SetDekKeyMaterial sets the KeyMaterial using the given bytes
 func (c *client) SetDekKeyMaterial(dek *Dek, keyMaterialBytes []byte) {
 	c.dekCacheLock.Lock()
 	defer c.dekCacheLock.Unlock()

schemaregistry/rules/encryption/deks/mock_dekregistry_client.go

@@ -181,7 +181,7 @@ func (c *mockclient) GetDekVersion(kekName string, subject string, version int,
 	return Dek{}, &posErr
 }
 
-// GetEncryptedKeyMaterialBytes returns the EncryptedKeyMaterialBytes
+// GetDekEncryptedKeyMaterialBytes returns the EncryptedKeyMaterialBytes
 func (c *mockclient) GetDekEncryptedKeyMaterialBytes(dek *Dek) ([]byte, error) {
 	if dek.EncryptedKeyMaterial == "" {
 		return nil, nil
@@ -200,7 +200,7 @@ func (c *mockclient) GetDekEncryptedKeyMaterialBytes(dek *Dek) ([]byte, error) {
 	return dek.EncryptedKeyMaterialBytes, nil
 }
 
-// GetKeyMaterialBytes returns the KeyMaterialBytes
+// GetDekKeyMaterialBytes returns the KeyMaterialBytes
 func (c *mockclient) GetDekKeyMaterialBytes(dek *Dek) ([]byte, error) {
 	if dek.KeyMaterial == "" {
 		return nil, nil
@@ -219,7 +219,7 @@ func (c *mockclient) GetDekKeyMaterialBytes(dek *Dek) ([]byte, error) {
 	return dek.KeyMaterialBytes, nil
 }
 
-// SetKeyMaterial sets the KeyMaterial using the given bytes
+// SetDekKeyMaterial sets the KeyMaterial using the given bytes
 func (c *mockclient) SetDekKeyMaterial(dek *Dek, keyMaterialBytes []byte) {
 	dekCacheLock.Lock()
 	defer dekCacheLock.Unlock()

schemaregistry/rules/encryption/encrypt_executor.go

@@ -597,7 +597,7 @@ func (f *FieldEncryptionExecutorTransform) Transform(ctx serde.RuleContext, fiel
 
 func prefixVersion(version int, ciphertext []byte) ([]byte, error) {
 	var buf bytes.Buffer
-	err := buf.WriteByte(serde.MagicByte)
+	err := buf.WriteByte(serde.MagicByteV0)
 	if err != nil {
 		return nil, err
 	}
@@ -615,7 +615,7 @@ func prefixVersion(version int, ciphertext []byte) ([]byte, error) {
 }
 
 func extractVersion(ciphertext []byte) (int, error) {
-	if ciphertext[0] != serde.MagicByte {
+	if ciphertext[0] != serde.MagicByteV0 {
 		return -1, fmt.Errorf("unknown magic byte")
 	}
 	version := binary.BigEndian.Uint32(ciphertext[1:5])