diff --git a/.github/workflows/internal_tests.yaml b/.github/workflows/internal_tests.yaml index f565f772..6d1cb967 100644 --- a/.github/workflows/internal_tests.yaml +++ b/.github/workflows/internal_tests.yaml @@ -59,7 +59,7 @@ jobs: services: postgres: - image: postgres:18.1 + image: postgres:16.11 env: POSTGRES_PASSWORD: postgres ports: diff --git a/.github/workflows/planner_integration.yaml b/.github/workflows/planner_integration.yaml index 05b154ea..2500d71d 100644 --- a/.github/workflows/planner_integration.yaml +++ b/.github/workflows/planner_integration.yaml @@ -56,7 +56,7 @@ jobs: services: postgres: - image: postgres:18.1 + image: postgres:16.11 env: POSTGRES_PASSWORD: postgres ports: diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index 746972a3..5e713a1c 100644 --- a/CONTRIBUTING.md +++ b/CONTRIBUTING.md @@ -107,7 +107,7 @@ POSTGRESQL_DB_CONNECT_STRING="postgres://postgres:password@localhost:5432/postgr It assumes you have access to a Postgres server running reachable at $POSTGRESQL_DB_CONNECT_STRING. You can use `docker` to run a Postgres server locally: ```shell -docker run -d --rm --name postgres -e POSTGRES_PASSWORD=password -p 5432:5432 postgres:18.1 +docker run -d --rm --name postgres -e POSTGRES_PASSWORD=password -p 5432:5432 postgres:16.11 ``` Run `webhook-gateway` integration tests using: diff --git a/cmd/planner/main_test.go b/cmd/planner/main_test.go index a6bcaff5..63eeb8e6 100644 --- a/cmd/planner/main_test.go +++ b/cmd/planner/main_test.go @@ -25,6 +25,8 @@ import ( "github.com/stretchr/testify/require" ) +const supportedEventType = "workflow_job" + // testContext holds test configuration and dependencies. type testContext struct { t *testing.T @@ -38,58 +40,103 @@ func newTestContext(t *testing.T) *testContext { t: t, port: os.Getenv("APP_PORT"), rabbitURI: os.Getenv("RABBITMQ_CONNECT_STRING"), - queueName: "webhook-queue", + queueName: queue.DefaultQueueConfig().QueueName, } } -func TestMain_FlavorPressure(t *testing.T) { +func TestMain_IntegrationScenarios(t *testing.T) { /* - arrange: server is listening on the configured port and prepare request payload - act: send create flavor request, get flavor pressure request and publish webhook message - assert: 201 Created and 200 OK with expected pressure value updated after webhook processing + arrange: server is listening with full AMQP setup including DLQ + act: Test multiple scenarios in subtests + assert: all scenarios work correctly */ t.Setenv("APP_ADMIN_TOKEN_VALUE", "planner_v0_12345678901234567890") ctx := newTestContext(t) - ctx.declareQueue() go main() ctx.waitForHTTP("http://localhost:"+ctx.port+"/health", 10*time.Second) - - platform := "github" - labels := []string{"self-hosted", "s390x", "medium"} - priority := 100 - flavor := randString(10) - pressure := 0 - - // Test create flavor - resp := ctx.createFlavor(flavor, platform, labels, priority) - require.Equal(t, http.StatusCreated, resp, "unexpected status creating flavor") - - // Test get flavor pressure - pressures := ctx.getFlavorPressure(flavor) - ctx.assertFlavorPressureEquals(pressures, flavor, pressure) - - // Test consume webhook and reflect pressure - body := ctx.constructWebhookPayload(labels) - ctx.publishAndWaitAck(body) - - assert.Eventually(t, func() bool { - press := ctx.getFlavorPressure(flavor) - return press[flavor] > pressure - }, 2*time.Minute, 100*time.Millisecond, "expected flavor pressure to increase after webhook processing") - - // Trigger graceful shutdown and verify services stop cleanly - ctx.shutdownMain() - ctx.waitForHTTPDown("http://localhost:"+ctx.port+"/health", 30*time.Second) - - // Verify AMQP consumer is stopped by checking queue depth increases after publish - beforeDepth := ctx.getQueueDepth() - body2 := ctx.constructWebhookPayload(labels) - ctx.publishAndWaitAck(body2) - - assert.Eventually(t, func() bool { - return ctx.getQueueDepth() >= beforeDepth+1 - }, 10*time.Second, 200*time.Millisecond, "expected queue depth to increase after shutdown (consumer stopped)") + ctx.waitForQueue(queue.DefaultQueueConfig().QueueName, 10*time.Second) + + // Scenario 1: Flavor pressure updates + t.Run("flavor_pressure", func(t *testing.T) { + /* + act: send create flavor request, get flavor pressure request and publish webhook message + assert: 201 Created and 200 OK with expected pressure value updated after webhook processing + */ + + platform := "github" + labels := []string{"self-hosted", "s390x", "medium"} + priority := 100 + flavor := randString(10) + pressure := 0 + + // Test create flavor + resp := ctx.createFlavor(flavor, platform, labels, priority) + require.Equal(t, http.StatusCreated, resp, "unexpected status creating flavor") + + // Test get flavor pressure + pressures := ctx.getFlavorPressure(flavor) + ctx.assertFlavorPressureEquals(pressures, flavor, pressure) + + // Test consume webhook and reflect pressure + body := ctx.constructWebhookPayload(labels) + ctx.publish(body, supportedEventType) + + assert.Eventually(t, func() bool { + press := ctx.getFlavorPressure(flavor) + return press[flavor] > pressure + }, 2*time.Minute, 100*time.Millisecond, "expected flavor pressure to increase after webhook processing") + }) + + // Scenario 2: Dead letter queue behavior + t.Run("dead_letter_queue", func(t *testing.T) { + /* + act: + 1. unsupported event + 2. malformed message + assert: + 1. unsupported event does not appear in DLQ + 2. malformed message appears in DLQ + */ + config := queue.DefaultQueueConfig() + + // Unsupported event should be ignored + unsupportedBody := []byte(`{"action": "opened", "pull_request": {"id": 123}}`) + ctx.publish(unsupportedBody, "pull_request") + + dlqDepth := ctx.getNumMessagesInQueue(config.DeadLetterQueue) + assert.Equal(t, 0, dlqDepth) + + // Malformed message should go to DLQ (must have self-hosted label to pass filter) + malformedBody := []byte(`{"action": "queued", "workflow_job": {"id": 999, "labels": ["self-hosted"]}}`) + ctx.publish(malformedBody, supportedEventType) + + dlqMessage := ctx.consumeFromQueue(config.DeadLetterQueue, 10*time.Second) + assert.Equal(t, malformedBody, dlqMessage) + }) + + // Last scenario: Graceful shutdown + t.Run("graceful_shutdown", func(t *testing.T) { + /* + act: trigger graceful shutdown + assert: services stop cleanly and AMQP consumer stops processing messages + */ + config := queue.DefaultQueueConfig() + + labels := []string{"self-hosted", "s390x", "medium"} + + ctx.shutdownMain() + ctx.waitForHTTPDown("http://localhost:"+ctx.port+"/health", 30*time.Second) + + // Verify AMQP consumer is stopped by checking queue depth increases after publish + beforeDepth := ctx.getNumMessagesInQueue(config.QueueName) + body2 := ctx.constructWebhookPayload(labels) + ctx.publish(body2, supportedEventType) + + assert.Eventually(t, func() bool { + return ctx.getNumMessagesInQueue(config.QueueName) >= beforeDepth+1 + }, 10*time.Second, 200*time.Millisecond, "expected queue depth to increase after shutdown (consumer stopped)") + }) } // constructWebhookPayload creates a webhook payload with the given labels. @@ -114,113 +161,6 @@ func (ctx *testContext) constructWebhookPayload(labels []string) []byte { return body } -// declareQueue declares the exchange, queue, and binding for testing. -func (ctx *testContext) declareQueue() { - ctx.t.Helper() - - conn, err := amqp.Dial(ctx.rabbitURI) - require.NoError(ctx.t, err, "connect rabbitmq") - defer conn.Close() - - ch, err := conn.Channel() - require.NoError(ctx.t, err, "open channel") - defer ch.Close() - - config := queue.DefaultQueueConfig() - - err = ch.ExchangeDeclare( - config.ExchangeName, - "direct", // type - true, // durable - false, // auto-delete - false, // internal - false, // no-wait - nil, - ) - require.NoError(ctx.t, err, "declare exchange") - - _, err = ch.QueueDeclare( - ctx.queueName, - true, // durable - false, // auto-delete - false, // exclusive - false, // no-wait - nil, - ) - require.NoError(ctx.t, err, "declare queue") - - err = ch.QueueBind( - ctx.queueName, - config.RoutingKey, - config.ExchangeName, - false, // no-wait - nil, - ) - require.NoError(ctx.t, err, "bind queue") -} - -// publishAndWaitAck publishes a message to the exchange with routing key and waits for an ack. -func (ctx *testContext) publishAndWaitAck(body []byte) { - ctx.t.Helper() - - conn, err := amqp.Dial(ctx.rabbitURI) - require.NoError(ctx.t, err, "connect rabbitmq") - defer conn.Close() - - ch, err := conn.Channel() - require.NoError(ctx.t, err, "open channel") - defer ch.Close() - - err = ch.Confirm(false) - require.NoError(ctx.t, err, "enable confirms") - - confirms := ch.NotifyPublish(make(chan amqp.Confirmation, 1)) - - config := queue.DefaultQueueConfig() - err = ch.Publish( - config.ExchangeName, // publish to exchange instead of directly to queue - config.RoutingKey, // use routing key - false, false, - amqp.Publishing{ - ContentType: "application/json", - Body: body, - Headers: amqp.Table{"X-GitHub-Event": "workflow_job"}, - }, - ) - require.NoError(ctx.t, err, "publish") - - select { - case c := <-confirms: - require.True(ctx.t, c.Ack, "message not acked") - case <-time.After(10 * time.Second): - require.Fail(ctx.t, "timeout waiting for ack") - } -} - -// getQueueDepth returns the current number of messages in the queue via passive declare. -func (ctx *testContext) getQueueDepth() int { - ctx.t.Helper() - - conn, err := amqp.Dial(ctx.rabbitURI) - require.NoError(ctx.t, err, "connect rabbitmq") - defer conn.Close() - - ch, err := conn.Channel() - require.NoError(ctx.t, err, "open channel") - defer ch.Close() - - q, err := ch.QueueDeclarePassive( - ctx.queueName, - true, // durable - false, // auto-delete - false, // exclusive - false, // no-wait - nil, - ) - require.NoError(ctx.t, err, "declare passive queue") - return q.Messages -} - // shutdownMain sends SIGTERM to trigger graceful shutdown in main(). func (ctx *testContext) shutdownMain() { ctx.t.Helper() @@ -317,3 +257,116 @@ func randString(n int) string { } return string(b) } + +// publish a message to the exchange with body and eventType header +// publish waits for confirmation that the message was accepted by RabbitMQ. +// To omit the header, please pass an empty string as eventType. +func (ctx *testContext) publish(body []byte, eventType string) { + ctx.t.Helper() + + conn, err := amqp.Dial(ctx.rabbitURI) + require.NoError(ctx.t, err, "connect rabbitmq") + defer conn.Close() + + ch, err := conn.Channel() + require.NoError(ctx.t, err, "open channel") + defer ch.Close() + + err = ch.Confirm(false) + require.NoError(ctx.t, err, "enable confirms") + confirms := ch.NotifyPublish(make(chan amqp.Confirmation, 1)) + + config := queue.DefaultQueueConfig() + publishing := amqp.Publishing{ + ContentType: "application/json", + Body: body, + } + if eventType != "" { + publishing.Headers = amqp.Table{"X-GitHub-Event": eventType} + } + + err = ch.Publish(config.ExchangeName, config.RoutingKey, false, false, publishing) + require.NoError(ctx.t, err, "publish") + + select { + case c := <-confirms: + require.True(ctx.t, c.Ack, "message not acked") + case <-time.After(10 * time.Second): + require.Fail(ctx.t, "timeout waiting for ack") + } +} + +// consumeFromQueue consumes a single message from the specified queue. +func (ctx *testContext) consumeFromQueue(queueName string, timeout time.Duration) []byte { + ctx.t.Helper() + + conn, err := amqp.Dial(ctx.rabbitURI) + require.NoError(ctx.t, err, "connect rabbitmq") + defer conn.Close() + + ch, err := conn.Channel() + require.NoError(ctx.t, err, "open channel") + defer ch.Close() + + deliveryChan, err := ch.Consume(queueName, "", true, false, false, false, nil) + require.NoError(ctx.t, err, "consume from queue") + + select { + case msg := <-deliveryChan: + return msg.Body + case <-time.After(timeout): + require.Fail(ctx.t, "timeout waiting for message in "+queueName) + return nil + } +} + +// waitForQueue waits until the specified queue exists in RabbitMQ. +func (ctx *testContext) waitForQueue(queueName string, timeout time.Duration) { + ctx.t.Helper() + deadline := time.Now().Add(timeout) + for time.Now().Before(deadline) { + conn, err := amqp.Dial(ctx.rabbitURI) + if err != nil { + time.Sleep(100 * time.Millisecond) + continue + } + ch, err := conn.Channel() + if err != nil { + conn.Close() + time.Sleep(100 * time.Millisecond) + continue + } + _, err = ch.QueueDeclarePassive(queueName, true, false, false, false, nil) + ch.Close() + conn.Close() + if err == nil { + return + } + time.Sleep(100 * time.Millisecond) + } + require.Fail(ctx.t, "queue did not appear", "queue %s did not exist within %s", queueName, timeout) +} + +// getNumMessagesInQueue returns the current number of messages in the specified queue. +func (ctx *testContext) getNumMessagesInQueue(queueName string) int { + ctx.t.Helper() + + conn, err := amqp.Dial(ctx.rabbitURI) + require.NoError(ctx.t, err, "connect rabbitmq") + defer conn.Close() + + ch, err := conn.Channel() + require.NoError(ctx.t, err, "open channel") + defer ch.Close() + + q, err := ch.QueueDeclarePassive( + queueName, + true, + false, + false, + false, + nil, + ) + require.NoError(ctx.t, err, "declare passive queue") + return q.Messages +} diff --git a/internal/planner/consumer.go b/internal/planner/consumer.go index 7bb30dc5..4e80ac53 100644 --- a/internal/planner/consumer.go +++ b/internal/planner/consumer.go @@ -60,101 +60,107 @@ func parseOptionalGitHubTime(ts *github.Timestamp) *time.Time { return &t } -func parseGithubWebhookPayload(ctx context.Context, headers map[string]interface{}, body []byte) (githubWebhookJob, error) { - // Extract the event type from headers +// getWorkflowJob extracts and validates a workflow job from a webhook message. +// Returns nil for events that should be ignored (non-workflow_job, unsupported actions, non-self-hosted). +// Returns an error only for malformed data that cannot be processed. +func getWorkflowJob(ctx context.Context, headers map[string]interface{}, body []byte) (*githubWebhookJob, error) { eventTypeHeader, ok := headers[githubEventHeaderKey] if !ok { - logger.WarnContext(ctx, "missing X-GitHub-Event header, discarding message") - return githubWebhookJob{}, fmt.Errorf("missing X-GitHub-Event header") + return nil, fmt.Errorf("missing X-GitHub-Event header") } eventType, ok := eventTypeHeader.(string) if !ok { - logger.WarnContext(ctx, "X-GitHub-Event header is not a string, discarding message") - return githubWebhookJob{}, fmt.Errorf("X-GitHub-Event must be string") + return nil, fmt.Errorf("X-GitHub-Event must be string") } - // Parse the webhook using go-github library event, err := github.ParseWebHook(eventType, body) if err != nil { - return githubWebhookJob{}, fmt.Errorf("unable to parse webhook: %w", err) + return nil, fmt.Errorf("unable to parse webhook: %w", err) } - // Check if it's a workflow_job event jobEvent, ok := event.(*github.WorkflowJobEvent) if !ok { - logger.InfoContext(ctx, "event is not a workflow_job, discarding", "event_type", eventType) - return githubWebhookJob{}, fmt.Errorf("unsupported webhook event: %s", eventType) + if eventType == "workflow_job" { + logger.WarnContext(ctx, "received workflow_job in \"X-GitHub-Event\" header but payload did not parse to expected type; possible GitHub API change or library issue") + } else { + logger.DebugContext(ctx, "ignoring non-workflow_job event", "event_type", eventType) + } + return nil, nil } - // Process the workflow job event based on action action := jobEvent.GetAction() - if !slices.Contains([]string{"queued", "in_progress", "completed", "waiting"}, action) { - return githubWebhookJob{}, fmt.Errorf("unknown webhook event action: %s", action) + + if !slices.Contains([]string{"queued", "in_progress", "completed"}, action) { + logger.DebugContext(ctx, "ignoring workflow_job action", "action", action) + return nil, nil } job := jobEvent.GetWorkflowJob() if job == nil { - return githubWebhookJob{}, fmt.Errorf("missing workflow_job field in event webhook") + return nil, fmt.Errorf("missing workflow_job field in event webhook") + } + + if !isSelfHosted(job.Labels) { + repo := jobEvent.GetRepo().GetFullName() + logger.DebugContext(ctx, "ignoring non self-hosted job", "repo", repo, "labels", strings.Join(job.Labels, ",")) + return nil, nil } createdAt := job.GetCreatedAt() if createdAt.IsZero() { - return githubWebhookJob{}, fmt.Errorf("missing created_at in queued event webhook") + return nil, fmt.Errorf("missing created_at in queued event webhook") } startedAt := job.GetStartedAt() if startedAt.IsZero() && action == "in_progress" { - return githubWebhookJob{}, fmt.Errorf("missing started_at in in_progress event webhook") + return nil, fmt.Errorf("missing started_at in in_progress event webhook") } completedAt := job.GetCompletedAt() if completedAt.IsZero() && action == "completed" { - return githubWebhookJob{}, fmt.Errorf("missing completed_at in completed event webhook") + return nil, fmt.Errorf("missing completed_at in completed event webhook") } id := strconv.FormatInt(job.GetID(), 10) if id == "0" { - return githubWebhookJob{}, fmt.Errorf("missing job id in event webhook") + return nil, fmt.Errorf("missing job id in event webhook") } var rawJson map[string]interface{} err = json.Unmarshal(body, &rawJson) if err != nil { - return githubWebhookJob{}, fmt.Errorf("invalid github webhook: %v", err) - } - - dbJob := &database.Job{ - ID: id, - Platform: platform, - Labels: job.Labels, - CreatedAt: createdAt.Time, - StartedAt: parseOptionalGitHubTime(job.StartedAt), - CompletedAt: parseOptionalGitHubTime(job.CompletedAt), - Raw: map[string]interface{}{action: rawJson}, + return nil, fmt.Errorf("invalid github webhook: %v", err) } repo := jobEvent.GetRepo().GetFullName() - if repo == "" { - return githubWebhookJob{}, fmt.Errorf("missing repo full name in webhook payload") + return nil, fmt.Errorf("missing repo full name in webhook payload") } - return githubWebhookJob{ - id: id, - repo: jobEvent.GetRepo().GetFullName(), - labels: job.Labels, - action: action, - + webhookJob := &githubWebhookJob{ + id: id, + repo: repo, + labels: job.Labels, + action: action, payload: jobEvent, - - job: dbJob, - }, nil + job: &database.Job{ + ID: id, + Platform: platform, + Labels: job.Labels, + CreatedAt: createdAt.Time, + StartedAt: parseOptionalGitHubTime(job.StartedAt), + CompletedAt: parseOptionalGitHubTime(job.CompletedAt), + Raw: map[string]interface{}{action: rawJson}, + }, + } + + return webhookJob, nil } -// isSelfHosted checks if the job is intended for self-hosted runners. -func isSelfHosted(job *githubWebhookJob) bool { - for _, label := range job.labels { +// isSelfHosted checks if the job labels indicate it's for self-hosted runners. +func isSelfHosted(labels []string) bool { + for _, label := range labels { if strings.Contains(label, "self-hosted") { return true } @@ -224,30 +230,22 @@ func (c *JobConsumer) pullMessage(ctx context.Context) error { ctx, span := trace.Start(ctx, "consume webhook") defer span.End() - job, err := parseGithubWebhookPayload(ctx, msg.Headers, msg.Body) + job, err := getWorkflowJob(ctx, msg.Headers, msg.Body) if err != nil { - logger.ErrorContext(ctx, "cannot parse webhook payload", "error", err) + logger.ErrorContext(ctx, "cannot parse webhook payload, discarding to DLQ", "error", err) span.RecordError(err) c.metrics.ObserveWebhookError(ctx, platform) c.discardMessage(ctx, &msg) return err } - if !slices.Contains([]string{"queued", "in_progress", "completed"}, job.action) { - logger.InfoContext(ctx, "ignore other action types for GitHub webhook job", "action", job.action) - c.metrics.ObserveDiscardedWebhook(ctx, platform) - c.discardMessage(ctx, &msg) - return nil - } - - if !isSelfHosted(&job) { - logger.InfoContext(ctx, "ignore non self-hosted job", "repo", job.repo, "labels", strings.Join(job.labels, ",")) + if job == nil { c.metrics.ObserveDiscardedWebhook(ctx, platform) - c.discardMessage(ctx, &msg) + c.ignoreMessage(ctx, &msg) return nil } - err = c.handleMessage(ctx, &job) + err = c.handleMessage(ctx, job) if err == nil { c.consumedMessage(ctx, &msg) c.metrics.ObserveConsumedGitHubWebhook(ctx, job.payload) @@ -269,7 +267,15 @@ func (c *JobConsumer) discardMessage(ctx context.Context, delivery *amqp.Deliver oteltrace.SpanFromContext(ctx).RecordError(err) c.metrics.ObserveWebhookError(ctx, platform) } - return +} + +func (c *JobConsumer) ignoreMessage(ctx context.Context, delivery *amqp.Delivery) { + err := delivery.Ack(false) + if err != nil { + logger.ErrorContext(ctx, "cannot ignore queue message", "error", err) + oteltrace.SpanFromContext(ctx).RecordError(err) + c.metrics.ObserveWebhookError(ctx, platform) + } } func (c *JobConsumer) consumedMessage(ctx context.Context, delivery *amqp.Delivery) { diff --git a/internal/planner/consumer_test.go b/internal/planner/consumer_test.go index 2b948c96..f85778f6 100644 --- a/internal/planner/consumer_test.go +++ b/internal/planner/consumer_test.go @@ -323,16 +323,15 @@ func TestConsumer(t *testing.T) { }, }), }, - expectErr: true, + expectErr: false, checkDB: func(t *testing.T, db *fakeDB) { assert.Nil(t, db.jobs["github:16"], "job should not be inserted") }, checkMetrics: func(t *testing.T, mr *telemetry.TestMetricReader) { m := mr.Collect(t) - // 2 errors: 1 from processing error, 1 from nack failure - assert.Equal(t, 2.0, m.Counter(t, webhookErrorsMetricName)) + assert.Equal(t, 1.0, m.Counter(t, webhookErrorsMetricName)) // ack failure assert.Equal(t, 0.0, m.Counter(t, consumedWebhooksMetricName)) - assert.Equal(t, 0.0, m.Counter(t, discardedWebhooksMetricName)) + assert.Equal(t, 1.0, m.Counter(t, discardedWebhooksMetricName)) }, }, { name: "requeues when AddJob fails with other error", @@ -482,7 +481,7 @@ func TestConsumer(t *testing.T) { assert.Equal(t, 0.0, m.Counter(t, discardedWebhooksMetricName)) }, }, { - name: "discards non-workflow_job events", + name: "ignores non-workflow_job events", setupDB: func(db *fakeDB) {}, delivery: amqp.Delivery{ Headers: amqp.Table{"X-GitHub-Event": "pull_request"}, @@ -493,16 +492,15 @@ func TestConsumer(t *testing.T) { }, }), }, - expectErr: true, + expectErr: false, checkDB: func(t *testing.T, db *fakeDB) { assert.Empty(t, db.jobs, "no jobs should be inserted for non-workflow_job events") }, checkMetrics: func(t *testing.T, mr *telemetry.TestMetricReader) { m := mr.Collect(t) - // 2 errors: 1 from processing error, 1 from nack failure - assert.Equal(t, 2.0, m.Counter(t, webhookErrorsMetricName)) + assert.Equal(t, 1.0, m.Counter(t, webhookErrorsMetricName)) // ack failure assert.Equal(t, 0.0, m.Counter(t, consumedWebhooksMetricName)) - assert.Equal(t, 0.0, m.Counter(t, discardedWebhooksMetricName)) + assert.Equal(t, 1.0, m.Counter(t, discardedWebhooksMetricName)) }, }, { name: "discards non-self-hosted jobs", @@ -604,11 +602,11 @@ func TestConsumer(t *testing.T) { } } -func TestParseGithubWebhookPayload(t *testing.T) { +func TestGetWorkflowJob(t *testing.T) { /* arrange: setup context for parsing webhook payload - act: parse the GitHub webhook payload with provided headers and body - assert: check for expected errors and validate parsed job data + act: extract workflow job from GitHub webhook payload + assert: check for expected errors, nil returns, and validate parsed job data */ mk := func(m map[string]any) []byte { b, _ := json.Marshal(m); return b } @@ -618,7 +616,8 @@ func TestParseGithubWebhookPayload(t *testing.T) { body []byte expectErr bool expectErrMsg string - validateJob func(*testing.T, githubWebhookJob) + expectNil bool + validateJob func(*testing.T, *githubWebhookJob) }{{ name: "successful parsing of queued event", headers: map[string]interface{}{ @@ -636,8 +635,7 @@ func TestParseGithubWebhookPayload(t *testing.T) { "created_at": "2025-01-01T10:00:00Z", }, }), - expectErr: false, - validateJob: func(t *testing.T, job githubWebhookJob) { + validateJob: func(t *testing.T, job *githubWebhookJob) { assert.Equal(t, "100", job.id) assert.Equal(t, "canonical/test-repo", job.repo) assert.Equal(t, "queued", job.action) @@ -672,8 +670,7 @@ func TestParseGithubWebhookPayload(t *testing.T) { "id": 1, }, }), - expectErr: true, - expectErrMsg: "unsupported webhook event", + expectNil: true, }, { name: "unknown action", headers: map[string]interface{}{ @@ -690,8 +687,24 @@ func TestParseGithubWebhookPayload(t *testing.T) { "created_at": "2025-01-01T10:00:00Z", }, }), - expectErr: true, - expectErrMsg: "unknown webhook event action", + expectNil: true, + }, { + name: "non-self-hosted job", + headers: map[string]interface{}{ + "X-GitHub-Event": "workflow_job", + }, + body: mk(map[string]any{ + "action": "queued", + "repository": map[string]any{ + "full_name": "canonical/test", + }, + "workflow_job": map[string]any{ + "id": 250, + "labels": []string{"ubuntu-latest"}, // Not self-hosted + "created_at": "2025-01-01T10:00:00Z", + }, + }), + expectNil: true, }, { name: "missing workflow_job field", headers: map[string]interface{}{ @@ -800,7 +813,7 @@ func TestParseGithubWebhookPayload(t *testing.T) { expectErr: true, expectErrMsg: "missing repo full name", }, { - name: "waiting action is accepted", + name: "waiting action is ignored", headers: map[string]interface{}{ "X-GitHub-Event": "workflow_job", }, @@ -815,11 +828,7 @@ func TestParseGithubWebhookPayload(t *testing.T) { "created_at": "2025-01-01T10:00:00Z", }, }), - expectErr: false, - validateJob: func(t *testing.T, job githubWebhookJob) { - assert.Equal(t, "700", job.id) - assert.Equal(t, "waiting", job.action) - }, + expectNil: true, }, { name: "in_progress with all fields", headers: map[string]interface{}{ @@ -838,8 +847,7 @@ func TestParseGithubWebhookPayload(t *testing.T) { "started_at": "2025-01-01T10:05:00Z", }, }), - expectErr: false, - validateJob: func(t *testing.T, job githubWebhookJob) { + validateJob: func(t *testing.T, job *githubWebhookJob) { assert.Equal(t, "800", job.id) assert.Equal(t, "in_progress", job.action) assert.NotNil(t, job.job.StartedAt) @@ -864,8 +872,7 @@ func TestParseGithubWebhookPayload(t *testing.T) { "completed_at": "2025-01-01T10:15:00Z", }, }), - expectErr: false, - validateJob: func(t *testing.T, job githubWebhookJob) { + validateJob: func(t *testing.T, job *githubWebhookJob) { assert.Equal(t, "900", job.id) assert.Equal(t, "completed", job.action) assert.NotNil(t, job.job.CompletedAt) @@ -884,15 +891,20 @@ func TestParseGithubWebhookPayload(t *testing.T) { for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { ctx := context.Background() - job, err := parseGithubWebhookPayload(ctx, tt.headers, tt.body) + job, err := getWorkflowJob(ctx, tt.headers, tt.body) if tt.expectErr { assert.Error(t, err) if tt.expectErrMsg != "" { assert.Contains(t, err.Error(), tt.expectErrMsg) } + assert.Nil(t, job) + } else if tt.expectNil { + assert.NoError(t, err) + assert.Nil(t, job) } else { assert.NoError(t, err) + assert.NotNil(t, job) if tt.validateJob != nil { tt.validateJob(t, job) } diff --git a/internal/queue/config.go b/internal/queue/config.go index da334189..b1c292ea 100644 --- a/internal/queue/config.go +++ b/internal/queue/config.go @@ -7,23 +7,29 @@ package queue // Default AMQP configuration values. const ( - defaultExchangeName = "github-workflow-jobs" - defaultQueueName = "github-workflow-jobs" - defaultRoutingKey = "workflow-job" + defaultExchangeName = "github-workflow-jobs" + defaultQueueName = "github-workflow-jobs" + defaultRoutingKey = "workflow-job" + defaultDeadLetterExchange = "github-workflow-jobs-dlx" + defaultDeadLetterQueue = "github-workflow-jobs-dlq" ) // QueueConfig holds AMQP exchange, queue, and routing configuration. type QueueConfig struct { - ExchangeName string - QueueName string - RoutingKey string + ExchangeName string + QueueName string + RoutingKey string + DeadLetterExchange string + DeadLetterQueue string } // DefaultQueueConfig returns a QueueConfig with default values. func DefaultQueueConfig() QueueConfig { return QueueConfig{ - ExchangeName: defaultExchangeName, - QueueName: defaultQueueName, - RoutingKey: defaultRoutingKey, + ExchangeName: defaultExchangeName, + QueueName: defaultQueueName, + RoutingKey: defaultRoutingKey, + DeadLetterExchange: defaultDeadLetterExchange, + DeadLetterQueue: defaultDeadLetterQueue, } } diff --git a/internal/queue/consumer.go b/internal/queue/consumer.go index e96ab814..3eb5ed2d 100644 --- a/internal/queue/consumer.go +++ b/internal/queue/consumer.go @@ -111,17 +111,22 @@ func (c *AmqpConsumer) ensureAmqpChannel() error { return err } - err = c.client.declareExchange(c.config.ExchangeName) + err = c.client.ensureDeadLetterQueue(c.config.DeadLetterExchange, c.config.DeadLetterQueue, c.config.RoutingKey) if err != nil { return err } - err = c.client.declareQueue(c.config.QueueName) + err = c.client.ensureExchange(c.config.ExchangeName) if err != nil { return err } - err = c.client.bindQueue(c.config.QueueName, c.config.RoutingKey, c.config.ExchangeName) + err = c.client.ensureQueueWithDeadLetter(c.config.QueueName, c.config.DeadLetterExchange) + if err != nil { + return err + } + + err = c.client.ensureQueueBinding(c.config.QueueName, c.config.RoutingKey, c.config.ExchangeName) if err != nil { return err } diff --git a/internal/queue/consumer_test.go b/internal/queue/consumer_test.go index a5945c4e..b69be6a8 100644 --- a/internal/queue/consumer_test.go +++ b/internal/queue/consumer_test.go @@ -378,6 +378,51 @@ func TestPullQueueDeclareAndBind(t *testing.T) { assert.Equal(t, config.RoutingKey, mockAmqpChannelConsumer.boundRoutingKey, "expected routing key to be "+config.RoutingKey) } +func TestPullSetsUpDeadLetterQueue(t *testing.T) { + /* + arrange: create a consumer with no amqp channel + act: pull a message from the queue + assert: DLX exchange declared, DLQ queue declared, main queue has DLX argument + */ + msgChannel := make(chan amqp.Delivery, 1) + msgChannel <- amqp.Delivery{Body: []byte("TestMessage")} + + mockAmqpChannelConsumer := &MockAmqpChannelConsumer{ + msgChannel: msgChannel, + } + mockAmqpConnection := &MockAmqpConnectionConsumer{ + amqpChannelConsumer: mockAmqpChannelConsumer, + returnChannelOnSetup: true, + } + config := DefaultQueueConfig() + amqpConsumer := &AmqpConsumer{ + client: &Client{ + amqpChannel: nil, + amqpConnection: mockAmqpConnection, + }, + config: config, + } + + _, err := amqpConsumer.Pull(context.Background()) + + assert.NoError(t, err) + + assert.Contains(t, mockAmqpChannelConsumer.MockAmqpChannel.exchangeNames, config.DeadLetterExchange, + "expected DLX exchange to be declared") + + assert.Contains(t, mockAmqpChannelConsumer.MockAmqpChannel.queueNames, config.DeadLetterQueue, + "expected DLQ queue to be declared") + + assert.Equal(t, config.DeadLetterExchange, mockAmqpChannelConsumer.MockAmqpChannel.queueArgs["x-dead-letter-exchange"], + "expected main queue to have x-dead-letter-exchange argument") + + assert.Equal(t, 2, mockAmqpChannelConsumer.MockAmqpChannel.exchangeCount, + "expected 2 exchanges to be declared (DLX and main)") + + assert.Equal(t, 2, mockAmqpChannelConsumer.MockAmqpChannel.declareCount, + "expected 2 queues to be declared (DLQ and main)") +} + func TestPullDeclarationFailure(t *testing.T) { /* arrange: create a consumer with no amqp channel where a declaration fails diff --git a/internal/queue/deadletter.go b/internal/queue/deadletter.go new file mode 100644 index 00000000..66fcddf6 --- /dev/null +++ b/internal/queue/deadletter.go @@ -0,0 +1,29 @@ +/* + * Copyright 2026 Canonical Ltd. + * See LICENSE file for licensing details. + */ + +package queue + +import "fmt" + +// ensureDeadLetterQueue declares the dead-letter exchange and queue, and binds them together. +// This is an idempotent operation that encapsulates the RabbitMQ-specific dead-letter configuration. +func (c *Client) ensureDeadLetterQueue(dlxName, dlqName, routingKey string) error { + err := c.ensureExchange(dlxName) + if err != nil { + return fmt.Errorf("cannot declare dead-letter exchange: %w", err) + } + + err = c.ensureQueue(dlqName) + if err != nil { + return fmt.Errorf("cannot declare dead-letter queue: %w", err) + } + + err = c.ensureQueueBinding(dlqName, routingKey, dlxName) + if err != nil { + return fmt.Errorf("cannot bind dead-letter queue: %w", err) + } + + return nil +} diff --git a/internal/queue/deadletter_test.go b/internal/queue/deadletter_test.go new file mode 100644 index 00000000..14eaeb1b --- /dev/null +++ b/internal/queue/deadletter_test.go @@ -0,0 +1,78 @@ +/* + * Copyright 2026 Canonical Ltd. + * See LICENSE file for licensing details. + */ + +package queue + +import ( + "testing" + + "github.com/stretchr/testify/assert" +) + +func TestEnsureDeadLetterQueue(t *testing.T) { + /* + arrange: Create a client with mock channel. + act: Call ensureDeadLetterQueue. + assert: DLX exchange declared, DLQ queue declared, and queue bound to exchange. + */ + ch := &MockAmqpChannel{} + client := &Client{amqpChannel: ch} + + err := client.ensureDeadLetterQueue("test-dlx", "test-dlq", "test-routing-key") + + assert.NoError(t, err) + assert.Contains(t, ch.exchangeNames, "test-dlx", "DLX exchange should be declared") + assert.Contains(t, ch.queueNames, "test-dlq", "DLQ queue should be declared") + assert.Equal(t, "test-dlq", ch.boundQueue) + assert.Equal(t, "test-dlx", ch.boundExchange) + assert.Equal(t, "test-routing-key", ch.boundRoutingKey) +} + +func TestEnsureDeadLetterQueueErrors(t *testing.T) { + /* + arrange: Create a client with mock channel configured for each error scenario. + act: Call ensureDeadLetterQueue. + assert: Appropriate error returned. + */ + tests := []struct { + name string + exchangeDeclareError bool + queueDeclareError bool + bindError bool + expectErrContains string + }{ + { + name: "returns error when exchange declare fails", + exchangeDeclareError: true, + expectErrContains: "cannot declare dead-letter exchange", + }, + { + name: "returns error when queue declare fails", + queueDeclareError: true, + expectErrContains: "cannot declare dead-letter queue", + }, + { + name: "returns error when queue bind fails", + bindError: true, + expectErrContains: "cannot bind dead-letter queue", + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + ch := &MockAmqpChannel{ + exchangeDeclareError: tt.exchangeDeclareError, + queueDeclareError: tt.queueDeclareError, + bindError: tt.bindError, + } + client := &Client{amqpChannel: ch} + + err := client.ensureDeadLetterQueue("test-dlx", "test-dlq", "test-routing-key") + + assert.Error(t, err) + assert.ErrorContains(t, err, tt.expectErrContains) + }) + } +} diff --git a/internal/queue/mock_test.go b/internal/queue/mock_test.go index 6b24650e..6b84e44c 100644 --- a/internal/queue/mock_test.go +++ b/internal/queue/mock_test.go @@ -25,9 +25,14 @@ type MockAmqpChannel struct { confirmMode bool // Declaration tracking - queueName string - exchangeName string - queueDurable bool + queueName string + exchangeName string + queueDurable bool + queueArgs amqp.Table + declareCount int + exchangeNames []string + queueNames []string + exchangeCount int // Binding tracking boundQueue string @@ -92,15 +97,20 @@ func (ch *MockAmqpChannel) ExchangeDeclare(name string, _ string, _, _, _, _ boo return errors.New("exchange declare error") } ch.exchangeName = name + ch.exchangeNames = append(ch.exchangeNames, name) + ch.exchangeCount++ return nil } -func (ch *MockAmqpChannel) QueueDeclare(name string, durable, _, _, _ bool, _ amqp.Table) (amqp.Queue, error) { +func (ch *MockAmqpChannel) QueueDeclare(name string, durable, _, _, _ bool, args amqp.Table) (amqp.Queue, error) { if ch.queueDeclareError { return amqp.Queue{}, errors.New("queue declare error") } ch.queueName = name ch.queueDurable = durable + ch.queueArgs = args + ch.queueNames = append(ch.queueNames, name) + ch.declareCount++ return amqp.Queue{Name: name}, nil } diff --git a/internal/queue/producer.go b/internal/queue/producer.go index cfdda8c8..536ae43d 100644 --- a/internal/queue/producer.go +++ b/internal/queue/producer.go @@ -45,7 +45,7 @@ func (p *AmqpProducer) resetConnectionOrChannelIfNecessary() error { return err } - err = p.client.declareExchange(p.config.ExchangeName) + err = p.client.ensureExchange(p.config.ExchangeName) if err != nil { return err } diff --git a/internal/queue/types.go b/internal/queue/types.go index 6b52d0d1..cbf5e845 100644 --- a/internal/queue/types.go +++ b/internal/queue/types.go @@ -28,7 +28,7 @@ func (c *Client) resetConnection() error { } conn, err := c.connectFunc(c.uri) if err != nil { - return fmt.Errorf("failed to connect to AMQP server: %w", err) + return fmt.Errorf("cannot connect to AMQP server: %w", err) } c.amqpConnection = conn return nil @@ -42,19 +42,20 @@ func (c *Client) resetChannel() error { ch, err := c.amqpConnection.Channel() if err != nil { - return fmt.Errorf("failed to open AMQP channel: %w", err) + return fmt.Errorf("cannot open AMQP channel: %w", err) } c.amqpChannel = ch err = ch.Confirm(false) if err != nil { - return fmt.Errorf("failed to put channel in confirm mode: %w", err) + return fmt.Errorf("cannot put channel in confirm mode: %w", err) } return nil } -func (c *Client) declareExchange(exchangeName string) error { +// ensureExchange declares an exchange with given name - this is an idempotent operation +func (c *Client) ensureExchange(exchangeName string) error { err := c.amqpChannel.ExchangeDeclare( exchangeName, "direct", // use direct exchange @@ -65,12 +66,13 @@ func (c *Client) declareExchange(exchangeName string) error { nil, // no additional arguments ) if err != nil { - return fmt.Errorf("failed to declare AMQP exchange: %w", err) + return fmt.Errorf("cannot declare AMQP exchange: %w", err) } return nil } -func (c *Client) declareQueue(queueName string) error { +// ensureQueue declares a queue with given name - this is an idempotent operation +func (c *Client) ensureQueue(queueName string) error { _, err := c.amqpChannel.QueueDeclare( queueName, true, // durable @@ -80,12 +82,32 @@ func (c *Client) declareQueue(queueName string) error { nil, // arguments ) if err != nil { - return fmt.Errorf("failed to declare AMQP queue: %w", err) + return fmt.Errorf("cannot declare AMQP queue: %w", err) } return nil } -func (c *Client) bindQueue(queueName, routingKey, exchangeName string) error { +// ensureQueueWithDeadLetter declares a queue with given name and dead-letter exchange specified - this is an idempotent operation +func (c *Client) ensureQueueWithDeadLetter(queueName, dlxName string) error { + args := amqp.Table{ + "x-dead-letter-exchange": dlxName, + } + _, err := c.amqpChannel.QueueDeclare( + queueName, + true, // durable + false, // delete when unused + false, // exclusive + false, // no-wait + args, + ) + if err != nil { + return fmt.Errorf("cannot declare AMQP queue with dead-letter: %w", err) + } + return nil +} + +// ensureQueueBinding binds a queue to an exchange with the given routing key - this is an idempotent operation. +func (c *Client) ensureQueueBinding(queueName, routingKey, exchangeName string) error { err := c.amqpChannel.QueueBind( queueName, routingKey, @@ -94,7 +116,7 @@ func (c *Client) bindQueue(queueName, routingKey, exchangeName string) error { nil, // arguments ) if err != nil { - return fmt.Errorf("failed to bind AMQP queue: %w", err) + return fmt.Errorf("cannot bind AMQP queue: %w", err) } return nil }