feat: add retry mechanism to kafka, fix test and add retry-go deps

conference-application/frontend-go/frontend-go_test.go

@@ -4,12 +4,13 @@ import (
 	"context"
 	"flag"
 	"fmt"
-	"github.com/stretchr/testify/assert"
-	tc "github.com/testcontainers/testcontainers-go/modules/compose"
-	"github.com/testcontainers/testcontainers-go/wait"
 	"net/http"
 	"net/http/httptest"
 	"testing"
+
+	"github.com/stretchr/testify/assert"
+	tc "github.com/testcontainers/testcontainers-go/modules/compose"
+	"github.com/testcontainers/testcontainers-go/wait"
 )
 
 func testServer() *httptest.Server {
@@ -73,17 +74,17 @@ func Test_API(t *testing.T) {
 		assert.Equal(t, http.StatusOK, resp.StatusCode)
 	})
 
-	t.Run("It should return 200 when a POST is made to '/api/events'", func(t *testing.T) {
+	t.Run("It should return 200 when a POST is made to '/api/events/'", func(t *testing.T) {
 		// arrange, act
-		resp, _ := http.Post(fmt.Sprintf("%s/api/events", ts.URL), "", nil)
+		resp, _ := http.Post(fmt.Sprintf("%s/api/events/", ts.URL), "", nil)
 
 		// assert
 		assert.Equal(t, http.StatusOK, resp.StatusCode)
 	})
 
-	t.Run("It should return 200 when a GET is made to '/api/events'", func(t *testing.T) {
+	t.Run("It should return 200 when a GET is made to '/api/events/'", func(t *testing.T) {
 		// arrange, act
-		resp, _ := http.Get(fmt.Sprintf("%s/api/events", ts.URL))
+		resp, _ := http.Get(fmt.Sprintf("%s/api/events/", ts.URL))
 
 		// assert
 		assert.Equal(t, http.StatusOK, resp.StatusCode)

conference-application/frontend-go/frontend.go

@@ -16,6 +16,7 @@ import (
 	"github.com/go-chi/chi/middleware"
 	"github.com/go-chi/chi/v5"
 	"github.com/salaboy/platforms-on-k8s/conference-application/frontend-go/api"
+	"github.com/sethvargo/go-retry"
 
 	kafka "github.com/segmentio/kafka-go"
 )
@@ -77,15 +78,19 @@ type Features struct {
 
 func main() {
 
+	ctx := context.Background()
 	r := NewChiServer()
 	log.Printf("Connecting to Kafka Instance: %s, topic: %s., group: %s", KafkaUrl, KafkaTopic, KafkaGroupId)
 	reader := getKafkaReader(KafkaUrl, KafkaTopic, KafkaGroupId)
 
-	kafkaAlive := isKafkaAlive(KafkaUrl, KafkaTopic)
-	if !kafkaAlive {
-		log.Printf("Cannot connect to Kafka, restarting until it is healthy.")
-		return
-	}
+	retry.Constant(ctx, time.Second*5, func(ctx context.Context) error {
+		kafkaAlive, err := isKafkaAlive()
+		if !kafkaAlive {
+			log.Printf("Cannot connect to Kafka, retrying until it is healthy.")
+			return retry.RetryableError(err)
+		}
+		return nil
+	})
 
 	go consumeFromKafka(reader)
 
@@ -134,11 +139,12 @@ func getKafkaReader(kafkaURL, topic, groupID string) *kafka.Reader {
 	})
 }
 
-func isKafkaAlive(kafkaURL string, topic string) bool {
-	conn, err := kafka.DialLeader(context.Background(), "tcp", kafkaURL, topic, 0)
+func isKafkaAlive() (bool, error) {
+	conn, err := kafka.DialLeader(context.Background(), "tcp", KafkaUrl, KafkaTopic, 0)
 	if err != nil {
-		panic(any(err.Error()))
+		return false, err
 	}
+
 	defer conn.Close()
 
 	brokers, err := conn.Brokers()
@@ -151,9 +157,9 @@ func isKafkaAlive(kafkaURL string, topic string) bool {
 		log.Printf("Available Broker: %v", b)
 	}
 	if len(brokers) > 0 {
-		return true
+		return true, nil
 	} else {
-		return false
+		return false, nil
 	}
 
 }
@@ -323,9 +329,17 @@ func NewChiServer() *chi.Mux {
 	r.Handle("/*", http.StripPrefix("/", fs))
 
 	// Add handlers for readiness and liveness endpoints
-	r.HandleFunc("/health/{endpoint:readiness|liveness}", func(w http.ResponseWriter, r *http.Request) {
+	r.HandleFunc("/health/liveness", func(w http.ResponseWriter, r *http.Request) {
 		json.NewEncoder(w).Encode(map[string]bool{"ok": true})
 	})
 
+	r.HandleFunc("/health/readiness", func(w http.ResponseWriter, r *http.Request) {
+		is, err := isKafkaAlive()
+		if !is || err != nil {
+			w.WriteHeader(http.StatusInternalServerError)
+			json.NewEncoder(w).Encode(map[string]bool{"ok": false})
+		}
+	})
+
 	return r
 }

conference-application/frontend-go/go.mod

@@ -123,6 +123,7 @@ require (
 	github.com/rivo/uniseg v0.2.0 // indirect
 	github.com/secure-systems-lab/go-securesystemslib v0.4.0 // indirect
 	github.com/serialx/hashring v0.0.0-20190422032157-8b2912629002 // indirect
+	github.com/sethvargo/go-retry v0.2.4 // indirect
 	github.com/shibumi/go-pathspec v1.3.0 // indirect
 	github.com/sirupsen/logrus v1.9.3 // indirect
 	github.com/spf13/cobra v1.7.0 // indirect

conference-application/frontend-go/go.sum

@@ -514,6 +514,8 @@ github.com/segmentio/kafka-go v0.4.40 h1:sszW7c0/uyv7+VcTW5trx2ZC7kMWDTxuR/6Zn8U
 github.com/segmentio/kafka-go v0.4.40/go.mod h1:naFEZc5MQKdeL3W6NkZIAn48Y6AazqjRFDhnXeg3h94=
 github.com/serialx/hashring v0.0.0-20190422032157-8b2912629002 h1:ka9QPuQg2u4LGipiZGsgkg3rJCo4iIUCy75FddM0GRQ=
 github.com/serialx/hashring v0.0.0-20190422032157-8b2912629002/go.mod h1:/yeG0My1xr/u+HZrFQ1tOQQQQrOawfyMUH13ai5brBc=
+github.com/sethvargo/go-retry v0.2.4 h1:T+jHEQy/zKJf5s95UkguisicE0zuF9y7+/vgz08Ocec=
+github.com/sethvargo/go-retry v0.2.4/go.mod h1:1afjQuvh7s4gflMObvjLPaWgluLLyhA1wmVZ6KLpICw=
 github.com/shibumi/go-pathspec v1.3.0 h1:QUyMZhFo0Md5B8zV8x2tesohbb5kfbpTi9rBnKh5dkI=
 github.com/shibumi/go-pathspec v1.3.0/go.mod h1:Xutfslp817l2I1cZvgcfeMQJG5QnU2lh5tVaaMCl3jE=
 github.com/sirupsen/logrus v1.0.6/go.mod h1:pMByvHTf9Beacp5x1UXfOR9xyW/9antXMhjMPG0dEzc=