mockgcp: implement implied field mask in pubsub topic

Author: justinsb

mockgcp/common/fields/impliedfieldmask.go

@@ -0,0 +1,23 @@
+package fields
+
+import (
+	"context"
+	"sort"
+
+	"google.golang.org/protobuf/proto"
+	"google.golang.org/protobuf/reflect/protoreflect"
+)
+
+// ComputeImpliedFieldMask computes the implied field mask when it is omitted.
+// According to AIP 134: this is "equivalent to all fields that are populated (have a non-empty value)."
+// https://google.aip.dev/134
+func ComputeImpliedFieldMask(ctx context.Context, req proto.Message) []string {
+	var fieldMask []string
+	req.ProtoReflect().Range(func(fd protoreflect.FieldDescriptor, v protoreflect.Value) bool {
+		fieldName := string(fd.Name())
+		fieldMask = append(fieldMask, fieldName)
+		return true
+	})
+	sort.Strings(fieldMask)
+	return fieldMask
+}

mockgcp/mockpubsub/topic.go

@@ -23,7 +23,9 @@ import (
 	"google.golang.org/grpc/status"
 	"google.golang.org/protobuf/proto"
 	"google.golang.org/protobuf/types/known/emptypb"
+	"k8s.io/klog/v2"
 
+	"github.com/GoogleCloudPlatform/k8s-config-connector/mockgcp/common/fields"
 	"github.com/GoogleCloudPlatform/k8s-config-connector/mockgcp/common/projects"
 	pb "github.com/GoogleCloudPlatform/k8s-config-connector/mockgcp/generated/mockgcp/pubsub/v1"
 	"github.com/GoogleCloudPlatform/k8s-config-connector/mockgcp/pkg/storage"
@@ -86,6 +88,8 @@ func (s *publisherService) populateDefaultsForTopic(obj *pb.Topic) {
 }
 
 func (s *publisherService) UpdateTopic(ctx context.Context, req *pb.UpdateTopicRequest) (*pb.Topic, error) {
+	log := klog.FromContext(ctx)
+
 	reqName := req.Topic.Name
 	name, err := s.parseTopicName(reqName)
 	if err != nil {
@@ -100,23 +104,53 @@ func (s *publisherService) UpdateTopic(ctx context.Context, req *pb.UpdateTopicR
 	updated := ProtoClone(existing)
 	updated.Name = name.String()
 
-	// Required. The update mask applies to the resource.
 	paths := req.GetUpdateMask().GetPaths()
 	if len(paths) == 0 {
-		return nil, status.Errorf(codes.InvalidArgument, "update_mask is required")
+		// https://google.aip.dev/134
+		// field mask must be optional, and the service must treat an omitted field mask as an implied field mask equivalent to all fields that are populated (have a non-empty value).
+
+		paths = fields.ComputeImpliedFieldMask(ctx, req.GetTopic())
+		log.Info("computed implied field_mask", "paths", paths)
 	}
+
 	// TODO: Some sort of helper for fieldmask?
 	for _, path := range paths {
 		switch path {
-		// case "description":
-		// 	updated.Description = req.GetTopic().GetDescription()
-		// case "labels":
-		// 	updated.Labels = req.GetTopic().GetLabels()
+		case "name":
+			if updated.Name != req.GetTopic().GetName() {
+				return nil, status.Errorf(codes.InvalidArgument, "name is immutable")
+			}
+		case "labels":
+			updated.Labels = req.GetTopic().GetLabels()
+		case "schema_settings":
+			updated.SchemaSettings = req.GetTopic().GetSchemaSettings()
+		case "message_retention_duration":
+			updated.MessageRetentionDuration = req.GetTopic().GetMessageRetentionDuration()
 		default:
 			return nil, status.Errorf(codes.InvalidArgument, "update_mask path %q not valid", path)
 		}
 	}
 
+	// Note that if we try to pass e.g. 10d to message_retention_duration we get this:
+	// {
+	//   "error": {
+	//     "code": 400,
+	//     "message": "Invalid value at 'topic.message_retention_duration' (type.googleapis.com/google.protobuf.Duration), Field 'messageRetentionDuration', Illegal duration format; duration must end with 's'",
+	//     "status": "INVALID_ARGUMENT",
+	//     "details": [
+	//       {
+	//         "@type": "type.googleapis.com/google.rpc.BadRequest",
+	//         "fieldViolations": [
+	//           {
+	//             "field": "topic.message_retention_duration",
+	//             "description": "Invalid value at 'topic.message_retention_duration' (type.googleapis.com/google.protobuf.Duration), Field 'messageRetentionDuration', Illegal duration format; duration must end with 's'"
+	//           }
+	//         ]
+	//       }
+	//     ]
+	//   }
+	// }
+
 	if err := s.storage.Update(ctx, fqn, updated); err != nil {
 		return nil, err
 	}