Print cloud event as json message in log (#174)

Signed-off-by: Jian Qiu <jqiu@redhat.com>

Author: qiujian16

pkg/cloudevents/generic/clients/agentclient.go

@@ -200,7 +200,12 @@ func (c *CloudEventAgentClient[T]) receive(ctx context.Context, evt cloudevents.
 
 	action, err := c.specAction(evt.Source(), eventType.CloudEventsDataType, obj)
 	if err != nil {
-		logger.Error(err, "failed to generate spec action", "event", evt)
+		if logger.V(4).Enabled() {
+			evtData, _ := evt.MarshalJSON()
+			logger.Error(err, "failed to generate spec action", "event", string(evtData))
+		} else {
+			logger.Error(err, "failed to generate spec action")
+		}
 		return
 	}
 
@@ -211,7 +216,12 @@ func (c *CloudEventAgentClient[T]) receive(ctx context.Context, evt cloudevents.
 
 	for _, handler := range handlers {
 		if err := handler(ctx, action, obj); err != nil {
-			logger.Error(err, "failed to handle spec event", "event", evt)
+			if logger.V(4).Enabled() {
+				evtData, _ := evt.MarshalJSON()
+				logger.Error(err, "failed to handle spec event", "event", string(evtData))
+			} else {
+				logger.Error(err, "failed to handle spec event")
+			}
 		}
 	}
 }

pkg/cloudevents/generic/clients/baseclient.go

@@ -140,9 +140,12 @@ func (c *baseClient) publish(ctx context.Context, evt cloudevents.Event) error {
 		return fmt.Errorf("the cloudevents client is not ready")
 	}
 
-	logger.V(2).Info("Sending event", "event", evt.Context)
 	if logger.V(5).Enabled() {
-		logger.V(5).Info("Sending event", "event", evt.String())
+		evtData, _ := evt.MarshalJSON()
+		logger.V(5).Info("Sending event", "event", string(evtData))
+	} else {
+		logger.V(2).Info("Sending event",
+			"eventType", evt.Type(), "extensions", evt.Extensions())
 	}
 	if err := c.transport.Send(ctx, evt); err != nil {
 		return err
@@ -204,9 +207,12 @@ func (c *baseClient) subscribe(ctx context.Context, receive receiveFn) {
 					if err := c.transport.Receive(receiverCtx, func(ctx context.Context, evt cloudevents.Event) {
 						receiveLogger := logging.SetLogTracingByCloudEvent(klog.FromContext(ctx), &evt)
 						ctx = klog.NewContext(ctx, receiveLogger)
-						receiveLogger.V(2).Info("Received event", "event", evt.Context)
 						if receiveLogger.V(5).Enabled() {
-							receiveLogger.V(5).Info("Received event", "event", evt.String())
+							evtData, _ := evt.MarshalJSON()
+							receiveLogger.V(5).Info("Received event", "event", string(evtData))
+						} else {
+							receiveLogger.V(2).Info("Received event",
+								"eventType", evt.Type(), "extensions", evt.Extensions())
 						}
 						receive(ctx, evt)
 					}); err != nil {

pkg/cloudevents/generic/clients/sourceclient.go

@@ -204,7 +204,12 @@ func (c *CloudEventSourceClient[T]) receive(ctx context.Context, evt cloudevents
 
 	for _, handler := range handlers {
 		if err := handler(ctx, types.StatusModified, obj); err != nil {
-			logger.Error(err, "failed to handle status event", "event", evt)
+			if logger.V(4).Enabled() {
+				evtData, _ := evt.MarshalJSON()
+				logger.Error(err, "failed to handle status event", "event", string(evtData))
+			} else {
+				logger.Error(err, "failed to handle status event")
+			}
 		}
 	}
 }

pkg/cloudevents/server/grpc/broker.go

@@ -92,7 +92,7 @@ func (bkr *GRPCBroker) Publish(ctx context.Context, pubReq *pbv1.PublishRequest)
 		return nil, status.Error(codes.InvalidArgument, fmt.Sprintf("failed to parse cloud event type %s, %v", evt.Type(), err))
 	}
 
-	logger.V(4).Info("receive the event with grpc broker", "eventContext", evt.Context)
+	logger.V(4).Info("receive the event with grpc broker", "eventType", evt.Type(), "extensions", evt.Extensions())
 
 	// handler resync request
 	if eventType.Action == types.ResyncRequestAction {
@@ -234,7 +234,8 @@ func (bkr *GRPCBroker) Subscribe(subReq *pbv1.SubscriptionRequest, subServer pbv
 		}
 
 		// send the cloudevent to the subscriber
-		logger.V(4).Info("sending the event to spec subscribers", "subID", subID, "eventContext", evt.Context)
+		logger.V(4).Info("sending the event to spec subscribers",
+			"subID", subID, "eventType", evt.Type(), "extensions", evt.Extensions())
 		select {
 		case eventCh <- pbEvt:
 		case <-subCtx.Done():
@@ -280,7 +281,8 @@ func (bkr *GRPCBroker) Subscribe(subReq *pbv1.SubscriptionRequest, subServer pbv
 //   - If the requested resource version is older than the source's current maintained resource version, the source
 //     sends the resource.
 func (bkr *GRPCBroker) respondResyncSpecRequest(ctx context.Context, eventDataType types.CloudEventsDataType, evt *cloudevents.Event) error {
-	log := klog.FromContext(ctx)
+	log := klog.FromContext(ctx).WithValues(
+		"eventDataType", eventDataType, "eventType", evt.Type(), "extensions", evt.Extensions())
 
 	resourceVersions, err := payload.DecodeSpecResyncRequest(*evt)
 	if err != nil {
@@ -304,35 +306,36 @@ func (bkr *GRPCBroker) respondResyncSpecRequest(ctx context.Context, eventDataTy
 	}
 
 	if len(objs) == 0 {
-		log.V(4).Info("no objs from the lister, do nothing", "eventContext", evt.Context)
+		log.V(4).Info("no objs from the lister, do nothing")
 		return nil
 	}
 
 	for _, obj := range objs {
 		// respond with the deleting resource regardless of the resource version
+		objLogger := log.WithValues("eventType", obj.Type(), "extensions", obj.Extensions())
 		if _, ok := obj.Extensions()[types.ExtensionDeletionTimestamp]; ok {
-			log.V(4).Info("respond spec resync request", "eventContext", evt.Context)
+			objLogger.V(4).Info("respond spec resync request")
 			err = bkr.handleRes(ctx, obj, eventDataType, "delete_request")
 			if err != nil {
-				log.Error(err, "failed to handle resync spec request")
+				objLogger.Error(err, "failed to handle resync spec request")
 			}
 			continue
 		}
 
 		lastResourceVersion := findResourceVersion(obj.ID(), resourceVersions.Versions)
 		currentResourceVersion, err := cloudeventstypes.ToInteger(obj.Extensions()[types.ExtensionResourceVersion])
 		if err != nil {
-			log.V(4).Info("ignore the event since it has a invalid resourceVersion", "eventContext", obj.Context, "error", err)
+			objLogger.V(4).Info("ignore the event since it has a invalid resourceVersion", "error", err)
 			continue
 		}
 
 		// the version of the work is not maintained on source or the source's work is newer than agent, send
 		// the newer work to agent
 		if currentResourceVersion == 0 || int64(currentResourceVersion) > lastResourceVersion {
-			log.V(4).Info("respond spec resync request", "eventContext", evt.Context)
+			objLogger.V(4).Info("respond spec resync request")
 			err := bkr.handleRes(ctx, obj, eventDataType, "update_request")
 			if err != nil {
-				log.Error(err, "failed to handle resync spec request")
+				objLogger.Error(err, "failed to handle resync spec request")
 			}
 		}
 	}
@@ -356,7 +359,7 @@ func (bkr *GRPCBroker) respondResyncSpecRequest(ctx context.Context, eventDataTy
 			NewEvent()
 
 		// send a delete event for the current resource
-		log.V(4).Info("respond spec resync request", "eventContext", evt.Context)
+		log.V(4).Info("respond spec resync request")
 		err := bkr.handleRes(ctx, &obj, eventDataType, "delete_request")
 		if err != nil {
 			log.Error(err, "failed to handle delete request")