Add health endpoint to event_display server (#5608)

* Add health endpoint to event_display server

* Add unit test for health endpoint

cmd/event_display/main.go

@@ -1,5 +1,5 @@
 /*
-Copyright 2019 The Knative Authors
+Copyright 2021 The Knative Authors
 
 Licensed under the Apache License, Version 2.0 (the "License");
 you may not use this file except in compliance with the License.
@@ -20,8 +20,10 @@ import (
 	"context"
 	"fmt"
 	"log"
+	"net/http"
 
 	cloudevents "github.com/cloudevents/sdk-go/v2"
+	cehttp "github.com/cloudevents/sdk-go/v2/protocol/http"
 )
 
 /*
@@ -47,15 +49,42 @@ Data,
   }
 */
 
+// display prints the given Event in a human-readable format.
 func display(event cloudevents.Event) {
-	fmt.Printf("☁️  cloudevents.Event\n%s", event.String())
+	fmt.Printf("☁️  cloudevents.Event\n%s", event)
 }
 
 func main() {
-	c, err := cloudevents.NewClientHTTP()
+	run(context.Background())
+}
+
+func run(ctx context.Context) {
+	c, err := cloudevents.NewClientHTTP(
+		cehttp.WithMiddleware(healthzMiddleware()),
+	)
 	if err != nil {
-		log.Fatal("Failed to create client, ", err)
+		log.Fatal("Failed to create client: ", err)
 	}
 
-	log.Fatal(c.StartReceiver(context.Background(), display))
+	if err := c.StartReceiver(ctx, display); err != nil {
+		log.Fatal("Error during receiver's runtime: ", err)
+	}
+}
+
+// HTTP path of the health endpoint used for probing the service.
+const healthzPath = "/healthz"
+
+// healthzMiddleware returns a cehttp.Middleware which exposes a health
+// endpoint by registering a handler in the multiplexer of the CloudEvents HTTP
+// client.
+func healthzMiddleware() cehttp.Middleware {
+	return func(next http.Handler) http.Handler {
+		next.(*http.ServeMux).Handle(healthzPath, http.HandlerFunc(handleHealthz))
+		return next
+	}
+}
+
+// handleHealthz is a http.Handler which responds to health requests.
+func handleHealthz(w http.ResponseWriter, _ *http.Request) {
+	w.WriteHeader(http.StatusNoContent)
 }

cmd/event_display/main_test.go

@@ -0,0 +1,86 @@
+/*
+Copyright 2021 The Knative Authors
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package main
+
+import (
+	"bytes"
+	"context"
+	"fmt"
+	"net/http"
+	"testing"
+	"time"
+)
+
+const ceClientURL = "http://localhost:8080"
+
+func TestRun_HealthEndpoint(t *testing.T) {
+	ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
+	t.Cleanup(cancel)
+
+	go run(ctx)
+	if err := waitForClient(ctx); err != nil {
+		t.Fatal("Error waiting for CloudEvents receiver:", err)
+	}
+
+	const healthzURL = ceClientURL + healthzPath
+	const expectStatusCode = http.StatusNoContent
+
+	// GET request
+	resp, err := http.Get(healthzURL)
+	if err != nil {
+		t.Fatal("Error sending GET request to health endpoint:", err)
+	}
+	if gotStatusCode := resp.StatusCode; gotStatusCode != expectStatusCode {
+		t.Error("Unexpected status code sending GET request to health endpoint:", gotStatusCode)
+	}
+
+	// POST request
+	resp, err = http.Post(healthzURL, "text/plain", new(bytes.Buffer))
+	if err != nil {
+		t.Fatal("Error sending POST request to health endpoint:", err)
+	}
+	if gotStatusCode := resp.StatusCode; gotStatusCode != expectStatusCode {
+		t.Error("Unexpected status code sending POST request to health endpoint:", gotStatusCode)
+	}
+}
+
+// waitForClient sends requests to the local CloudEvents receiver address until
+// a HTTP response is received, or until ctx is cancelled.
+func waitForClient(ctx context.Context) error {
+	httpClient := http.DefaultClient
+	var httpErr error
+
+	tick := time.Tick(5 * time.Millisecond)
+
+	for {
+		select {
+		case <-tick:
+			req, err := http.NewRequestWithContext(ctx, http.MethodHead, ceClientURL, nil)
+			if err != nil {
+				return fmt.Errorf("creating HTTP request: %w", err)
+			}
+
+			if _, httpErr = httpClient.Do(req); httpErr == nil {
+				// got a response from CloudEvents client's HTTP receiver
+				return nil
+			}
+
+		case <-ctx.Done():
+			return fmt.Errorf("context cancelled: %s. The last HTTP error was: %s", ctx.Err(), httpErr)
+		}
+	}
+}