using resoruce id as the work metadata name

Signed-off-by: Wei Liu <liuweixa@redhat.com>

Author: skeeey

cmd/maestro/server/grpc_broker.go

@@ -10,7 +10,6 @@ import (
 	"time"
 
 	ce "github.com/cloudevents/sdk-go/v2"
-	cetypes "github.com/cloudevents/sdk-go/v2/types"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/credentials"
 	"google.golang.org/grpc/keepalive"
@@ -24,11 +23,14 @@ import (
 	servergrpc "open-cluster-management.io/sdk-go/pkg/cloudevents/server/grpc"
 
 	"github.com/openshift-online/maestro/pkg/api"
+	"github.com/openshift-online/maestro/pkg/client/cloudevents"
 	"github.com/openshift-online/maestro/pkg/dao"
 	"github.com/openshift-online/maestro/pkg/event"
 	"github.com/openshift-online/maestro/pkg/services"
 )
 
+const source = "maestro"
+
 type resourceHandler func(res *api.Resource) error
 
 // subscriber defines a subscriber that can receive and handle resource spec.
@@ -299,63 +301,18 @@ func (s *GRPCBroker) PredicateEvent(ctx context.Context, eventID string) (bool,
 
 // decodeResourceStatus translates a CloudEvent into a resource containing the status JSON map.
 func decodeResourceStatus(evt *ce.Event) (*api.Resource, error) {
-	evtExtensions := evt.Context.GetExtensions()
-
-	clusterName, err := cetypes.ToString(evtExtensions[types.ExtensionClusterName])
-	if err != nil {
-		return nil, fmt.Errorf("failed to get clustername extension: %v", err)
-	}
-
-	resourceID, err := cetypes.ToString(evtExtensions[types.ExtensionResourceID])
-	if err != nil {
-		return nil, fmt.Errorf("failed to get resourceid extension: %v", err)
-	}
-
-	resourceVersion, err := cetypes.ToInteger(evtExtensions[types.ExtensionResourceVersion])
-	if err != nil {
-		return nil, fmt.Errorf("failed to get resourceversion extension: %v", err)
-	}
-
-	status, err := api.CloudEventToJSONMap(evt)
-	if err != nil {
-		return nil, fmt.Errorf("failed to convert cloudevent to resource status: %v", err)
-	}
-
-	resource := &api.Resource{
-		Source:       evt.Source(),
-		ConsumerName: clusterName,
-		Version:      resourceVersion,
-		Meta: api.Meta{
-			ID: resourceID,
-		},
-		Status: status,
-	}
-
-	return resource, nil
+	codec := cloudevents.NewCodec(source)
+	return codec.Decode(evt)
 }
 
 // encodeResourceSpec translates a resource spec JSON map into a CloudEvent.
 func encodeResourceSpec(resource *api.Resource) (*ce.Event, error) {
-	evt, err := api.JSONMAPToCloudEvent(resource.Payload)
-	if err != nil {
-		return nil, fmt.Errorf("failed to convert resource payload to cloudevent: %v", err)
-	}
-
 	eventType := types.CloudEventsType{
 		CloudEventsDataType: workpayload.ManifestBundleEventDataType,
 		SubResource:         types.SubResourceSpec,
 		Action:              types.EventAction("create_request"),
 	}
-	evt.SetType(eventType.String())
-	evt.SetSource("maestro")
-	// TODO set resource.Source with a new extension attribute if the agent needs
-	evt.SetExtension(types.ExtensionResourceID, resource.ID)
-	evt.SetExtension(types.ExtensionResourceVersion, int64(resource.Version))
-	evt.SetExtension(types.ExtensionClusterName, resource.ConsumerName)
-
-	if !resource.GetDeletionTimestamp().IsZero() {
-		evt.SetExtension(types.ExtensionDeletionTimestamp, resource.GetDeletionTimestamp().Time)
-	}
 
-	return evt, nil
+	codec := cloudevents.NewCodec(source)
+	return codec.Encode(source, eventType, resource)
 }

docs/maestro.md

@@ -35,6 +35,12 @@ The Maestro Server includes various components to fulfill its functions, as illu
 
 ![maestro-mqtt-pub-dataflow](./images/maestro-mqtt-pub-dataflow.png)
 
+1. The Consumer (e.g. ClustersService) use Maestro GRPCSourceClient create a ManifestWork
+2. The GRPCSourceClient publish the ManifestWork using CloudEvents
+3. The Maestro server 
+4. The Maestro server 
+5. The Maestro agent
+
 ### Maestro Subscribe Resource Status with MQTT
 
 ![maestro-mqtt-sub-dataflow](./images/maestro-mqtt-sub-dataflow.png)

pkg/client/cloudevents/codec.go

@@ -7,6 +7,7 @@ import (
 	cloudevents "github.com/cloudevents/sdk-go/v2"
 	cloudeventstypes "github.com/cloudevents/sdk-go/v2/types"
 	"github.com/google/uuid"
+
 	workpayload "open-cluster-management.io/sdk-go/pkg/cloudevents/clients/work/payload"
 	cegeneric "open-cluster-management.io/sdk-go/pkg/cloudevents/generic"
 	cetypes "open-cluster-management.io/sdk-go/pkg/cloudevents/generic/types"
@@ -31,6 +32,14 @@ func (codec *Codec) EventDataType() cetypes.CloudEventsDataType {
 }
 
 func (codec *Codec) Encode(source string, eventType cetypes.CloudEventsType, res *api.Resource) (*cloudevents.Event, error) {
+	// use resource id as the resource payload metadata name to ensure the payload metadata name is
+	// unique on the agent side
+	if err := resetPayloadMetadataNameWithResID(res); err != nil {
+		return nil, err
+	}
+
+	// converts a resource payload to a CloudEvent
+	// If the resource payload has metadata the event will have the metadata extension
 	evt, err := api.JSONMAPToCloudEvent(res.Payload)
 	if err != nil {
 		return nil, fmt.Errorf("failed to convert resource payload to cloudevent: %v", err)
@@ -107,3 +116,20 @@ func (codec *Codec) Decode(evt *cloudevents.Event) (*api.Resource, error) {
 
 	return resource, nil
 }
+
+func resetPayloadMetadataNameWithResID(res *api.Resource) error {
+	metadata, ok := res.Payload[cetypes.ExtensionWorkMeta]
+	if !ok {
+		// the resource payload does not have metadata, do nothing
+		// the agent will using resource id as the payload metadata name
+		return nil
+	}
+
+	metaObj, ok := metadata.(map[string]interface{})
+	if !ok {
+		return fmt.Errorf("unsupported object type %T", metadata)
+	}
+	metaObj["name"] = res.ID
+	res.Payload[cetypes.ExtensionWorkMeta] = metaObj
+	return nil
+}