Delivery Format: set kn-ingress-format ce extension on imc ingress

[Cali0707]

Problem
To support dispatching events from the InMemoryChannel with the correct format when a user sets the spec.delivery.format to ingress, we need to set a cloudevent extension of kn-ingress-format to specify which format was used. This should be set to structured if the input event was structured, and binary if the input event was in binary. However, since this event may have been forwarded from earlier ingress, we need to only set the kn-ingress-format extension if it is not already present on the event.

Persona:
Which persona is this feature for?

Exit Criteria
Unit tests verifying that:

1. If the extension is not present it is set correctly for both structured and binary events
2. If the extension is already present, it is not changed for both structured and binary events

Time Estimate (optional):
How many developer-days do you think this may take to resolve? 1-2

Additional context (optional)
Add any other context about the feature request here.

• The parent issue is: Delivery Format: support specifying the event format #8057
• The channel ingress handler code is here:

  eventing/pkg/channel/event_receiver.go

  Lines 177 to 278 in 5f6713a

  	func (r *EventReceiver) ServeHTTP(response nethttp.ResponseWriter, request *nethttp.Request) {
  	ctx := request.Context()
  	if r.withContext != nil {
  	ctx = r.withContext(ctx)
  	}
  	response.Header().Set("Allow", "POST, OPTIONS")
  	if request.Method == nethttp.MethodOptions {
  	response.Header().Set("WebHook-Allowed-Origin", "*") // Accept from any Origin:
  	response.Header().Set("WebHook-Allowed-Rate", "*") // Unlimited requests/minute
  	response.WriteHeader(nethttp.StatusOK)
  	return
  	}
  	if request.Method != nethttp.MethodPost {
  	response.WriteHeader(nethttp.StatusMethodNotAllowed)
  	return
  	}
  	// The response status codes:
  	// 202 - the event was sent to subscribers
  	// 400 - the request was malformed
  	// 404 - the request was for an unknown channel
  	// 500 - an error occurred processing the request
  	args := ReportArgs{}
  	var channel ChannelReference
  	var err error
  	// prefer using pathToChannelFunc if available
  	if r.pathToChannelFunc != nil {
  	channel, err = r.pathToChannelFunc(request.URL.Path)
  	} else {
  	if request.URL.Path != "/" {
  	response.WriteHeader(nethttp.StatusBadRequest)
  	return
  	}
  	channel, err = r.hostToChannelFunc(request.Host)
  	}
  	if err != nil {
  	switch err.(type) {
  	case UnknownHostError:
  	response.WriteHeader(nethttp.StatusNotFound)
  	case BadRequestError:
  	response.WriteHeader(nethttp.StatusBadRequest)
  	default:
  	response.WriteHeader(nethttp.StatusInternalServerError)
  	}
  	r.logger.Info("Could not extract channel", zap.Error(err))
  	ReportEventCountMetricsForDispatchError(err, r.reporter, &args)
  	return
  	}
  	r.logger.Debug("Request mapped to channel", zap.String("channel", channel.String()))
  	args.Ns = channel.Namespace
  	if request.TLS != nil {
  	args.EventScheme = "https"
  	} else {
  	args.EventScheme = "http"
  	}
  	event, err := http.NewEventFromHTTPRequest(request)
  	if err != nil {
  	r.logger.Warn("failed to extract event from request", zap.Error(err))
  	response.WriteHeader(nethttp.StatusBadRequest)
  	_ = r.reporter.ReportEventCount(&args, nethttp.StatusBadRequest)
  	return
  	}
  	// run validation for the extracted event
  	if err := event.Validate(); err != nil {
  	r.logger.Warn("failed to validate extracted event", zap.Error(err))
  	response.WriteHeader(nethttp.StatusBadRequest)
  	return
  	}
  	/// Here we do the OIDC audience verification
  	features := feature.FromContext(ctx)
  	if features.IsOIDCAuthentication() {
  	r.logger.Debug("OIDC authentication is enabled")
  	err = r.tokenVerifier.VerifyJWTFromRequest(ctx, request, &r.audience, response)
  	if err != nil {
  	r.logger.Warn("Error when validating the JWT token in the request", zap.Error(err))
  	return
  	}
  	r.logger.Debug("Request contained a valid JWT. Continuing...")
  	}
  	err = r.receiverFunc(request.Context(), channel, *event, utils.PassThroughHeaders(request.Header))
  	if err != nil {
  	if _, ok := err.(*UnknownChannelError); ok {
  	response.WriteHeader(nethttp.StatusNotFound)
  	} else {
  	r.logger.Info("Error in receiver", zap.Error(err))
  	response.WriteHeader(nethttp.StatusInternalServerError)
  	}
  	return
  	}
  	response.WriteHeader(nethttp.StatusAccepted)
  	}

[Cali0707]

/good-first-issue
/triage accepted

[knative-prow]

@Cali0707:
This request has been marked as suitable for new contributors.

Please ensure the request meets the requirements listed here.

If this request no longer meets these requirements, the label can be removed
by commenting with the /remove-good-first-issue command.

In response to this:

/good-first-issue
/triage accepted

Instructions for interacting with me using PR comments are available here. If you have questions or suggestions related to my behavior, please file an issue against the kubernetes-sigs/prow repository.

[EraKin575]

/assign

[EraKin575]

/unassign

[Cali0707]

/close