[CLOUDTRUST-8743] Health check : Add kafka consumer + remove audit

healthcheck/health.go

@@ -5,7 +5,6 @@ import (
 	"net/http"
 	"time"
 
-	"github.com/cloudtrust/common-service/v2/events"
 	commonhttp "github.com/cloudtrust/common-service/v2/http"
 	log "github.com/cloudtrust/common-service/v2/log"
 	"github.com/go-kit/kit/ratelimit"
@@ -18,7 +17,7 @@ type HealthChecker interface {
 	AddHTTPEndpoint(name string, targetURL string, timeoutDuration time.Duration, expectedStatus int, cacheDuration time.Duration)
 	AddHTTPEndpoints(endpoints map[string]string, timeoutDuration time.Duration, expectedStatus int, cacheDuration time.Duration)
 	AddDatabase(name string, db HealthDatabase, cacheDuration time.Duration)
-	AddAuditEventsReporterModule(name string, reporter events.AuditEventsReporterModule, timeout time.Duration, cacheDuration time.Duration)
+	AddKafkaConsumer(name string, consumer KafkaConsumer, cacheDuration time.Duration)
 	MakeHandler(rateLimit ratelimit.Allower) http.HandlerFunc
 }
 
@@ -134,9 +133,9 @@ func (hc *healthchecker) AddDatabase(name string, db HealthDatabase, cacheDurati
 	hc.AddHealthChecker(name, newDatabaseChecker(name, db, cacheDuration, RealTimeProvider{}))
 }
 
-func (hc *healthchecker) AddAuditEventsReporterModule(name string, reporter events.AuditEventsReporterModule, timeout time.Duration, cacheDuration time.Duration) {
-	hc.logger.Info(context.Background(), "msg", "Adding audit event reporter module", "processor", name)
-	hc.AddHealthChecker(name, newAuditEventsReporterChecker(name, reporter, timeout, cacheDuration, hc.logger, RealTimeProvider{}))
+func (hc *healthchecker) AddKafkaConsumer(name string, consumer KafkaConsumer, cacheDuration time.Duration) {
+	hc.logger.Info(context.Background(), "msg", "Adding Kafka consumer", "processor", name)
+	hc.AddHealthChecker(name, newKafkaConsumerChecker(name, consumer, cacheDuration, RealTimeProvider{}))
 }
 
 // MakeHandler makes a HTTP handler that returns health check information

healthcheck/kafkaconsumer.go

@@ -0,0 +1,47 @@
+package healthcheck
+
+import (
+	"time"
+)
+
+type KafkaConsumer interface {
+	IsLive() bool
+}
+
+type kafkaConsumerChecker struct {
+	alias          string
+	consumer       KafkaConsumer
+	response       HealthStatus
+	failureCounter int
+}
+
+func newKafkaConsumerChecker(alias string, consumer KafkaConsumer, cacheDuration time.Duration, timeProvider TimeProvider) BasicChecker {
+	healthStatusType := "kafkaConsumer"
+	response := HealthStatus{Name: &alias, Type: &healthStatusType, CacheDuration: cacheDuration, TimeProvider: timeProvider}
+	response.connection("init")
+	response.stateUp()
+	return &kafkaConsumerChecker{
+		alias:          alias,
+		consumer:       consumer,
+		response:       response,
+		failureCounter: 0,
+	}
+}
+
+func (k *kafkaConsumerChecker) CheckStatus() HealthStatus {
+	if !k.response.hasExpired() {
+		return k.response
+	}
+
+	if k.consumer.IsLive() {
+		k.response.connection("consuming")
+		k.response.stateUp()
+		k.failureCounter = 0
+	} else {
+		k.response.stateDown("Kafka consumer is not live")
+		k.failureCounter++
+	}
+
+	k.response.touch()
+	return k.response
+}

healthcheck/kafkaconsumer_test.go

@@ -0,0 +1,92 @@
+package healthcheck
+
+import (
+	"testing"
+	"time"
+
+	"github.com/cloudtrust/common-service/v2/healthcheck/mock"
+	"github.com/stretchr/testify/assert"
+	"go.uber.org/mock/gomock"
+)
+
+func TestKafkaConsumerHealthCheckLiveAndCached(t *testing.T) {
+	mockCtrl := gomock.NewController(t)
+	defer mockCtrl.Finish()
+
+	mockTime := mock.NewTimeProvider(mockCtrl)
+	mockTime.EXPECT().Now().Return(testTime).AnyTimes()
+
+	consumer := mock.NewKafkaConsumer(mockCtrl)
+	consumer.EXPECT().IsLive().Return(true).Times(1)
+	checker := newKafkaConsumerChecker("alias", consumer, 10*time.Second, mockTime)
+
+	status := checker.CheckStatus()
+	assert.Equal(t, "UP", *status.State)
+	assert.Equal(t, "consuming", *status.Connection)
+	assert.Nil(t, status.Message)
+
+	status = checker.CheckStatus()
+	assert.Equal(t, "UP", *status.State)
+	assert.Equal(t, "consuming", *status.Connection)
+
+	internal := checker.(*kafkaConsumerChecker)
+	assert.Equal(t, 0, internal.failureCounter)
+}
+
+func TestKafkaConsumerHealthCheckDownAndCached(t *testing.T) {
+	mockCtrl := gomock.NewController(t)
+	defer mockCtrl.Finish()
+
+	mockTime := mock.NewTimeProvider(mockCtrl)
+	mockTime.EXPECT().Now().Return(testTime).AnyTimes()
+
+	consumer := mock.NewKafkaConsumer(mockCtrl)
+	consumer.EXPECT().IsLive().Return(false).Times(1)
+	checker := newKafkaConsumerChecker("alias", consumer, 10*time.Second, mockTime)
+
+	status := checker.CheckStatus()
+	assert.Equal(t, "DOWN", *status.State)
+	assert.Nil(t, status.Connection)
+	assert.NotNil(t, status.Message)
+	assert.Equal(t, "Kafka consumer is not live", *status.Message)
+
+	status = checker.CheckStatus()
+	assert.Equal(t, "DOWN", *status.State)
+	assert.Equal(t, "Kafka consumer is not live", *status.Message)
+
+	internal := checker.(*kafkaConsumerChecker)
+	assert.Equal(t, 1, internal.failureCounter)
+}
+
+func TestKafkaConsumerHealthCheckFailureCounterResetOnRecovery(t *testing.T) {
+	mockCtrl := gomock.NewController(t)
+	defer mockCtrl.Finish()
+
+	mockTime := mock.NewTimeProvider(mockCtrl)
+	gomock.InOrder(
+		mockTime.EXPECT().Now().Return(testTime),
+		mockTime.EXPECT().Now().Return(testTime),
+		mockTime.EXPECT().Now().Return(testTime.Add(10*time.Second).Add(time.Millisecond)),
+		mockTime.EXPECT().Now().Return(testTime.Add(10*time.Second).Add(time.Millisecond)),
+	)
+
+	consumer := mock.NewKafkaConsumer(mockCtrl)
+	gomock.InOrder(
+		consumer.EXPECT().IsLive().Return(false),
+		consumer.EXPECT().IsLive().Return(true),
+	)
+	checker := newKafkaConsumerChecker("alias", consumer, 10*time.Second, mockTime)
+
+	status := checker.CheckStatus()
+	assert.Equal(t, "DOWN", *status.State)
+	assert.Equal(t, "Kafka consumer is not live", *status.Message)
+
+	internal := checker.(*kafkaConsumerChecker)
+	assert.Equal(t, 1, internal.failureCounter)
+
+	status = checker.CheckStatus()
+	assert.Equal(t, "UP", *status.State)
+	assert.Equal(t, "consuming", *status.Connection)
+	assert.Nil(t, status.Message)
+	assert.Equal(t, 0, internal.failureCounter)
+}

healthcheck/mock_test.go

@@ -1,5 +1,5 @@
 package healthcheck
 
 //go:generate mockgen --build_flags=--mod=mod -destination=./mock/healthcheck.go -package=mock -mock_names=HealthDatabase=HealthDatabase github.com/cloudtrust/common-service/v2/healthcheck HealthDatabase
-//go:generate mockgen --build_flags=--mod=mod -destination=./mock/eventsreportermodule.go -package=mock -mock_names=AuditEventsReporterModule=AuditEventsReporterModule github.com/cloudtrust/common-service/v2/events AuditEventsReporterModule
 //go:generate mockgen --build_flags=--mod=mod -destination=./mock/timeprovider.go -package=mock -mock_names=TimeProvider=TimeProvider github.com/cloudtrust/common-service/v2/healthcheck TimeProvider
+//go:generate mockgen --build_flags=--mod=mod -destination=./mock/kafka.go -package=mock -mock_names=KafkaConsumer=KafkaConsumer github.com/cloudtrust/common-service/v2/healthcheck KafkaConsumer