Skip to content

Commit bd7bb17

Browse files
didildidil
committed
feat: extra logging info for queued messages
1 parent 3f1c03f commit bd7bb17

File tree

11 files changed

+84
-46
lines changed

11 files changed

+84
-46
lines changed

pkg/models/requeued_info.go renamed to pkg/models/queued_info.go

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -2,7 +2,8 @@ package models
22

33
import "time"
44

5-
type RequeuedInfo struct {
5+
type QueuedInfo struct {
6+
MessageID string
67
QueueStatus QueueStatus
78
DeliverAfter time.Time
89
}

‎pkg/server/handlers/ingest.go‎

Lines changed: 11 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -4,9 +4,11 @@ import (
44
"fmt"
55
"net/http"
66

7+
"github.com/didil/inhooks/pkg/models"
78
"github.com/go-chi/chi/v5"
89
"github.com/go-chi/chi/v5/middleware"
910
"go.uber.org/zap"
11+
"go.uber.org/zap/zapcore"
1012
)
1113

1214
func (app *App) HandleIngest(w http.ResponseWriter, r *http.Request) {
@@ -36,13 +38,21 @@ func (app *App) HandleIngest(w http.ResponseWriter, r *http.Request) {
3638
}
3739

3840
// enqueue messages
39-
err = app.messageEnqueuer.Enqueue(ctx, messages)
41+
queuedInfos, err := app.messageEnqueuer.Enqueue(ctx, messages)
4042
if err != nil {
4143
logger.Error("ingest request failed: unable to enqueue messages", zap.Error(err))
4244
app.WriteJSONErr(w, http.StatusBadRequest, reqID, fmt.Errorf("unable to enqueue data"))
4345
return
4446
}
4547

48+
for _, queuedInfo := range queuedInfos {
49+
fields := []zapcore.Field{zap.String("messageID", queuedInfo.MessageID), zap.String("queue", string(queuedInfo.QueueStatus))}
50+
if queuedInfo.QueueStatus == models.QueueStatusScheduled {
51+
fields = append(fields, zap.Time("nextAttemptAfter", queuedInfo.DeliverAfter))
52+
}
53+
logger.Info("message queued", fields...)
54+
}
55+
4656
app.WriteJSONResponse(w, http.StatusOK, JSONOK{})
4757
logger.Info("ingest request succeeded")
4858
}

‎pkg/server/handlers/ingest_test.go‎

Lines changed: 5 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -53,8 +53,11 @@ func TestUpdateLB(t *testing.T) {
5353
}
5454

5555
messageBuilder.EXPECT().FromHttp(flow, gomock.AssignableToTypeOf(&http.Request{}), gomock.AssignableToTypeOf("")).Return(messages, nil)
56-
57-
messageEnqueuer.EXPECT().Enqueue(gomock.Any(), messages).Return(nil)
56+
queuedInfos := []*models.QueuedInfo{
57+
{MessageID: messages[0].ID, QueueStatus: models.QueueStatusReady},
58+
{MessageID: messages[1].ID, QueueStatus: models.QueueStatusReady},
59+
}
60+
messageEnqueuer.EXPECT().Enqueue(gomock.Any(), messages).Return(queuedInfos, nil)
5861

5962
buf := bytes.NewBufferString(`{"id": "abc"}`)
6063

‎pkg/services/message_enqueuer.go‎

Lines changed: 34 additions & 21 deletions
Original file line numberDiff line numberDiff line change
@@ -11,7 +11,7 @@ import (
1111
)
1212

1313
type MessageEnqueuer interface {
14-
Enqueue(ctx context.Context, messages []*models.Message) error
14+
Enqueue(ctx context.Context, messages []*models.Message) ([]*models.QueuedInfo, error)
1515
}
1616

1717
func NewMessageEnqueuer(redisStore RedisStore, timeSvc TimeService) MessageEnqueuer {
@@ -26,34 +26,47 @@ type messageEnqueuer struct {
2626
timeSvc TimeService
2727
}
2828

29-
func (e *messageEnqueuer) Enqueue(ctx context.Context, messages []*models.Message) error {
29+
func (e *messageEnqueuer) Enqueue(ctx context.Context, messages []*models.Message) ([]*models.QueuedInfo, error) {
30+
queuedInfos := []*models.QueuedInfo{}
31+
3032
for _, m := range messages {
3133
queueStatus := getQueueStatus(m, e.timeSvc.Now())
3234

33-
b, err := json.Marshal(&m)
35+
err := e.redisEnqueue(ctx, m, queueStatus)
3436
if err != nil {
35-
return errors.Wrapf(err, "failed to encode message for sink: %s", m.SinkID)
37+
return nil, err
3638
}
3739

38-
mKey := messageKey(m.FlowID, m.SinkID, m.ID)
39-
qKey := queueKey(m.FlowID, m.SinkID, queueStatus)
40-
41-
switch queueStatus {
42-
case models.QueueStatusReady:
43-
err = e.redisStore.SetAndEnqueue(ctx, mKey, b, qKey, m.ID)
44-
if err != nil {
45-
return errors.Wrapf(err, "failed to set and enqueue message for sink: %s", m.SinkID)
46-
}
47-
case models.QueueStatusScheduled:
48-
err = e.redisStore.SetAndZAdd(ctx, mKey, b, qKey, m.ID, float64(m.DeliverAfter.Unix()))
49-
if err != nil {
50-
return errors.Wrapf(err, "failed to set and enqueue message for sink: %s", m.SinkID)
51-
}
52-
default:
53-
return fmt.Errorf("unexpected queue status %s", queueStatus)
54-
}
40+
queuedInfos = append(queuedInfos, &models.QueuedInfo{MessageID: m.ID, QueueStatus: queueStatus, DeliverAfter: m.DeliverAfter})
41+
}
5542

43+
return queuedInfos, nil
44+
}
45+
46+
func (e *messageEnqueuer) redisEnqueue(ctx context.Context, m *models.Message, queueStatus models.QueueStatus) error {
47+
b, err := json.Marshal(&m)
48+
if err != nil {
49+
return errors.Wrapf(err, "failed to encode message for sink: %s", m.SinkID)
5650
}
51+
52+
mKey := messageKey(m.FlowID, m.SinkID, m.ID)
53+
qKey := queueKey(m.FlowID, m.SinkID, queueStatus)
54+
55+
switch queueStatus {
56+
case models.QueueStatusReady:
57+
err = e.redisStore.SetAndEnqueue(ctx, mKey, b, qKey, m.ID)
58+
if err != nil {
59+
return errors.Wrapf(err, "failed to set and enqueue message for sink: %s", m.SinkID)
60+
}
61+
case models.QueueStatusScheduled:
62+
err = e.redisStore.SetAndZAdd(ctx, mKey, b, qKey, m.ID, float64(m.DeliverAfter.Unix()))
63+
if err != nil {
64+
return errors.Wrapf(err, "failed to set and enqueue message for sink: %s", m.SinkID)
65+
}
66+
default:
67+
return fmt.Errorf("unexpected queue status %s", queueStatus)
68+
}
69+
5770
return nil
5871
}
5972

‎pkg/services/message_enqueuer_test.go‎

Lines changed: 8 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -68,6 +68,13 @@ func TestMessageEnqueuer(t *testing.T) {
6868
Times(1).
6969
Return(nil)
7070

71-
err = messageEnqueuer.Enqueue(ctx, []*models.Message{m1, m2})
71+
queuedInfos, err := messageEnqueuer.Enqueue(ctx, []*models.Message{m1, m2})
7272
assert.NoError(t, err)
73+
74+
expectedInfos := []*models.QueuedInfo{
75+
{MessageID: m1ID, QueueStatus: models.QueueStatusReady, DeliverAfter: m1.DeliverAfter},
76+
{MessageID: m2ID, QueueStatus: models.QueueStatusScheduled, DeliverAfter: m2.DeliverAfter},
77+
}
78+
79+
assert.Equal(t, expectedInfos, queuedInfos)
7380
}

‎pkg/services/processing_results_service.go‎

Lines changed: 4 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -11,7 +11,7 @@ import (
1111
)
1212

1313
type ProcessingResultsService interface {
14-
HandleFailed(ctx context.Context, sink *models.Sink, m *models.Message, processingErr error) (*models.RequeuedInfo, error)
14+
HandleFailed(ctx context.Context, sink *models.Sink, m *models.Message, processingErr error) (*models.QueuedInfo, error)
1515
HandleOK(ctx context.Context, m *models.Message) error
1616
}
1717

@@ -27,7 +27,7 @@ func NewProcessingResultsService(timeSvc TimeService, redisStore RedisStore) Pro
2727
}
2828
}
2929

30-
func (s *processingResultsService) HandleFailed(ctx context.Context, sink *models.Sink, m *models.Message, processingErr error) (*models.RequeuedInfo, error) {
30+
func (s *processingResultsService) HandleFailed(ctx context.Context, sink *models.Sink, m *models.Message, processingErr error) (*models.QueuedInfo, error) {
3131
now := s.timeSvc.Now()
3232
m.DeliveryAttempts = append(m.DeliveryAttempts,
3333
&models.DeliveryAttempt{
@@ -68,7 +68,7 @@ func (s *processingResultsService) HandleFailed(ctx context.Context, sink *model
6868
return nil, errors.Wrapf(err, "failed to set and move to dead")
6969
}
7070

71-
return &models.RequeuedInfo{QueueStatus: models.QueueStatusDead}, nil
71+
return &models.QueuedInfo{MessageID: m.ID, QueueStatus: models.QueueStatusDead, DeliverAfter: m.DeliverAfter}, nil
7272
}
7373

7474
queueStatus := getQueueStatus(m, now)
@@ -89,7 +89,7 @@ func (s *processingResultsService) HandleFailed(ctx context.Context, sink *model
8989
return nil, fmt.Errorf("unexpected queue status %s", queueStatus)
9090
}
9191

92-
return &models.RequeuedInfo{QueueStatus: queueStatus, DeliverAfter: m.DeliverAfter}, nil
92+
return &models.QueuedInfo{MessageID: m.ID, QueueStatus: queueStatus, DeliverAfter: m.DeliverAfter}, nil
9393
}
9494

9595
func (s *processingResultsService) HandleOK(ctx context.Context, m *models.Message) error {

‎pkg/services/processing_results_service_test.go‎

Lines changed: 11 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -119,9 +119,10 @@ func TestProcessingResultsServiceHandleFailed_Dead(t *testing.T) {
119119
redisStore.EXPECT().SetAndMove(ctx, messageKey, b, sourceQueueKey, destQueueKey, mID).Return(nil)
120120

121121
s := NewProcessingResultsService(timeSvc, redisStore)
122-
requeuedInfo, err := s.HandleFailed(ctx, sink, m, processingErr)
122+
queuedInfo, err := s.HandleFailed(ctx, sink, m, processingErr)
123123
assert.NoError(t, err)
124-
assert.Equal(t, models.QueueStatusDead, requeuedInfo.QueueStatus)
124+
assert.Equal(t, mID, queuedInfo.MessageID)
125+
assert.Equal(t, models.QueueStatusDead, queuedInfo.QueueStatus)
125126
}
126127

127128
func TestProcessingResultsServiceHandleFailed_Scheduled(t *testing.T) {
@@ -181,10 +182,11 @@ func TestProcessingResultsServiceHandleFailed_Scheduled(t *testing.T) {
181182
redisStore.EXPECT().SetLRemZAdd(ctx, messageKey, b, sourceQueueKey, destQueueKey, mID, float64(mUpdated.DeliverAfter.Unix())).Return(nil)
182183

183184
s := NewProcessingResultsService(timeSvc, redisStore)
184-
requeuedInfo, err := s.HandleFailed(ctx, sink, m, processingErr)
185+
queuedInfo, err := s.HandleFailed(ctx, sink, m, processingErr)
185186
assert.NoError(t, err)
186-
assert.Equal(t, models.QueueStatusScheduled, requeuedInfo.QueueStatus)
187-
assert.Equal(t, mUpdated.DeliverAfter, requeuedInfo.DeliverAfter)
187+
assert.Equal(t, mID, queuedInfo.MessageID)
188+
assert.Equal(t, models.QueueStatusScheduled, queuedInfo.QueueStatus)
189+
assert.Equal(t, mUpdated.DeliverAfter, queuedInfo.DeliverAfter)
188190
}
189191

190192
func TestProcessingResultsServiceHandleFailed_Ready(t *testing.T) {
@@ -244,8 +246,9 @@ func TestProcessingResultsServiceHandleFailed_Ready(t *testing.T) {
244246
redisStore.EXPECT().SetAndMove(ctx, messageKey, b, sourceQueueKey, destQueueKey, mID).Return(nil)
245247

246248
s := NewProcessingResultsService(timeSvc, redisStore)
247-
requeuedInfo, err := s.HandleFailed(ctx, sink, m, processingErr)
249+
queuedInfo, err := s.HandleFailed(ctx, sink, m, processingErr)
248250
assert.NoError(t, err)
249-
assert.Equal(t, models.QueueStatusReady, requeuedInfo.QueueStatus)
250-
assert.Equal(t, mUpdated.DeliverAfter, requeuedInfo.DeliverAfter)
251+
assert.Equal(t, mID, queuedInfo.MessageID)
252+
assert.Equal(t, models.QueueStatusReady, queuedInfo.QueueStatus)
253+
assert.Equal(t, mUpdated.DeliverAfter, queuedInfo.DeliverAfter)
251254
}

‎pkg/supervisor/ready.go‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -48,11 +48,11 @@ func (s *Supervisor) FetchAndProcess(ctx context.Context, f *models.Flow, sink *
4848
processingErr := s.messageProcessor.Process(ctx, sink, m)
4949
if processingErr != nil {
5050
logger.Info("message processing failed")
51-
requeuedInfo, err := s.processingResultsSvc.HandleFailed(ctx, sink, m, processingErr)
51+
queuedInfo, err := s.processingResultsSvc.HandleFailed(ctx, sink, m, processingErr)
5252
if err != nil {
5353
return errors.Wrapf(err, "could not handle failed processing")
5454
}
55-
logger.Info("message queued after failure", zap.String("queue", string(requeuedInfo.QueueStatus)), zap.Time("nextAttemptAfter", requeuedInfo.DeliverAfter))
55+
logger.Info("message queued after failure", zap.String("queue", string(queuedInfo.QueueStatus)), zap.Time("nextAttemptAfter", queuedInfo.DeliverAfter))
5656
} else {
5757
logger.Info("message processed ok")
5858
err := s.processingResultsSvc.HandleOK(ctx, m)

‎pkg/supervisor/ready_test.go‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -102,7 +102,7 @@ func TestSupervisorFetchAndProcess_Failed(t *testing.T) {
102102

103103
messageFetcher.EXPECT().GetMessageForProcessing(ctx, appConf.Supervisor.ReadyWaitTime, flowId1, sinkID1).Return(m, nil)
104104
messageProcessor.EXPECT().Process(ctx, sink1, m).Return(processingErr)
105-
processingResultsService.EXPECT().HandleFailed(ctx, sink1, m, processingErr).Return(&models.RequeuedInfo{QueueStatus: models.QueueStatusReady}, nil)
105+
processingResultsService.EXPECT().HandleFailed(ctx, sink1, m, processingErr).Return(&models.QueuedInfo{QueueStatus: models.QueueStatusReady}, nil)
106106

107107
logger, err := zap.NewDevelopment()
108108
assert.NoError(t, err)

‎pkg/testsupport/mocks/mock_message_enqueuer.go‎

Lines changed: 4 additions & 3 deletions
Some generated files are not rendered by default. Learn more about customizing how changed files appear on GitHub.

0 commit comments

Comments
 (0)