Aws-sdk-go: Programmatically subscribing a queue to a topic doesn't put a confirmation token in the queue

[ijt]

Please fill out the sections below to help us address your issue.

Version of AWS SDK for Go?

53eb8b070e9a5067829fd029539966181632032a

Version of Go (go version)?

go version go1.11.2 darwin/amd64

What issue did you see?

The queue didn't get a subscription confirmation message.

Steps to reproduce

Run this:

// main.go
package main

import (
    "errors"
    "fmt"
    "log"
    "net/http"

    "github.com/aws/aws-sdk-go/aws"
    "github.com/aws/aws-sdk-go/aws/session"
    "github.com/aws/aws-sdk-go/service/sns"
    "github.com/aws/aws-sdk-go/service/sqs"
)

func main() {
    if err := makeTopicAndQueue(); err != nil {
        log.Fatalf("aws-sns-sqs: %v", err)
    }
}

func makeTopicAndQueue() error {
    sess, err := session.NewSession(&aws.Config{
        HTTPClient:  &http.Client{},
        Region:      aws.String("us-east-2"),
        Credentials: nil,
        MaxRetries:  aws.Int(0),
    })

    log.Printf("Creating an SNS topic.")
    snsClient := sns.New(sess, &aws.Config{})
    topicName := "test-topic"
    out, err := snsClient.CreateTopic(&sns.CreateTopicInput{Name: aws.String(topicName)})
    if err != nil {
        return fmt.Errorf(`creating topic "%s": %v`, topicName, err)
    }
    defer snsClient.DeleteTopic(&sns.DeleteTopicInput{TopicArn: out.TopicArn})

    log.Printf("Creating an SQS queue.")
    sqsClient := sqs.New(sess, &aws.Config{})
    subName := "test-subscription"
    out2, err := sqsClient.CreateQueue(&sqs.CreateQueueInput{QueueName: aws.String(subName)})
    if err != nil {
        return fmt.Errorf(`creating subscription queue "%s": %v`, subName, err)
    }

    log.Printf("Getting queue ARN.")
    out3, err := sqsClient.GetQueueAttributes(&sqs.GetQueueAttributesInput{
        QueueUrl:       out2.QueueUrl,
        AttributeNames: []*string{aws.String("QueueArn")},
    })
    if err != nil {
        return fmt.Errorf("getting queue ARN for %s: %v", *out2.QueueUrl, err)
    }
    qARN := out3.Attributes["QueueArn"]

    log.Printf("Subscribing the queue to the topic.")
    _, err = snsClient.Subscribe(&sns.SubscribeInput{
        TopicArn: out.TopicArn,
        Endpoint: qARN,
        Protocol: aws.String("sqs"),
    })
    if err != nil {
        return fmt.Errorf("subscribing: %v", err)
    }

    log.Printf("Getting the confirmation token from the queue.")
    out4, err := sqsClient.ReceiveMessage(&sqs.ReceiveMessageInput{
        QueueUrl: out2.QueueUrl,
    })
    if err != nil {
        return fmt.Errorf("receiving subscription confirmation message from queue: %v", err)
    }
    ms := out4.Messages
    var token *string
    switch len(ms) {
    case 0:
        return errors.New("no subscription confirmation message found in queue")
    case 1:
        m := ms[0]
        token = m.Body
    default:
        return fmt.Errorf("%d messages found in queue, want exactly 1", len(ms))
    }

    log.Printf("Using the token to finish subscribing.")
    _, err = snsClient.ConfirmSubscription(&sns.ConfirmSubscriptionInput{
        TopicArn: out.TopicArn,
        Token:    token,
    })
    if err != nil {
        return fmt.Errorf("confirming subscription: %v", err)
    }
    sqsClient.DeleteQueue(&sqs.DeleteQueueInput{QueueUrl: out2.QueueUrl})

    return nil
}

It gives the following output on my laptop:

[ ~/src/aws-sqs-issue ] go run main.go
2019/01/15 09:31:19 Creating an SNS topic.
2019/01/15 09:31:19 Creating an SQS queue.
2019/01/15 09:31:20 Getting queue ARN.
2019/01/15 09:31:20 Subscribing the queue to the topic.
2019/01/15 09:31:21 Getting the confirmation token from the queue.
2019/01/15 09:31:21 aws-sns-sqs: no subscription confirmation message found in queue

[diehlaws]

Hi @ijt, thanks for reaching out to us and I do apologize for the long delay in response from our end. There are a couple of things going on here that are preventing you from seeing the behavior you're expecting.

1. The code provided looks like it uses the same credential set to issue requests relating to the SQS queue and the SNS topic.

When the SNS topic and SQS queue exist in the same AWS account, subscribing the SQS queue to the SNS topic does not require confirmation so a subscription confirmation message won't be sent to the SQS queue. Likewise, as mentioned here, if the SQS queue owner creates the subscription for an SNS topic that exists in another AWS account (provided they have the appropriate IAM permissions) the subscription does not result in a confirmation message being sent to the SQS queue.

1. If the SQS queue and the SNS topic belong to different AWS accounts and the SNS topic owner is issuing the subscription request you need to grant appropriate permissions to the SQS queue to receive the subscription confirmation message.

When creating the SQS queue you need to include a policy similar to what's shown in this section of the SQS docs as an attribute. Expanding on the CreateQueue command included in your code:

    subName := "test-subscription"
    queuePolicy := `{
  "Version": "2012-10-17",
  "Id": "AllowAll",
  "Statement": [
    {
      "Sid": "Sid1550017136052",
      "Effect": "Allow",
      "Principal": "*",
      "Action": "SQS:*",
      "Resource": "arn:aws:sqs:us-east-2:123456789012:` + subName + `"
    }
  ]
}`
    out2, err := sqsClient.CreateQueue(&sqs.CreateQueueInput{
        QueueName: aws.String(subName),
        Attributes: map[string]*string{
            "Policy": &queuePolicy,
        },
    })

1. The subscription token does not exist in the format you're expecting from out4.

Once the SQS queue has been created with the appropriate policy and the subscription request results in a confirmation message being successfully received by the SQS queue, the confirmation message body contains more information than just the subscription token. The quickest way I found to address this is to unmarshal the message body before the ConfirmSubscription call to retrieve the subscription token like so:

    var dat map[string]string
    err = json.Unmarshal([]byte(*token), &dat)
    if err != nil {
        return fmt.Errorf("unmarshalling confirmation message %v", err)
    }
    subToken := dat["Token"]

    log.Printf("Using the token to finish subscribing.")
    _, err = snsClient.ConfirmSubscription(&sns.ConfirmSubscriptionInput{
        TopicArn: out.TopicArn,
        Token:    &subToken,
    })

I hope this information helps you programmatically subscribe SQS queues to SNS topics across accounts! Please be sure to let us know if you continue running into trouble with this after implementing my suggestions.

[ijt]

I tried this out. From what you said above, I gathered that I didn't need to confirm. So I wrote the following and still didn't see the messages from the SNS topic showing up in the SNS queue.

package main

import (
    "context"
    "fmt"
    "log"
    "math/rand"
    "net/http"
    "time"

    "github.com/aws/aws-sdk-go/aws"
    "github.com/aws/aws-sdk-go/aws/session"
    "github.com/aws/aws-sdk-go/service/sns"
    "github.com/aws/aws-sdk-go/service/sqs"
)

func main() {
    if err := makeTopicAndQueue(); err != nil {
        log.Fatalf("aws-sns-sqs: %v", err)
    }
}

func makeTopicAndQueue() error {
    sess, err := session.NewSession(&aws.Config{
        HTTPClient:  &http.Client{},
        Region:      aws.String("us-east-2"),
        Credentials: nil,
        MaxRetries:  aws.Int(0),
    })

    log.Printf("Creating an SNS topic.")
    snsClient := sns.New(sess, &aws.Config{})
    topicName := "test-topic"
    out, err := snsClient.CreateTopic(&sns.CreateTopicInput{Name: aws.String(topicName)})
    if err != nil {
        return fmt.Errorf(`creating topic "%s": %v`, topicName, err)
    }
    defer snsClient.DeleteTopic(&sns.DeleteTopicInput{TopicArn: out.TopicArn})

    log.Printf("Creating an SQS queue.")
    sqsClient := sqs.New(sess, &aws.Config{})
    subName := "test-subscription"
    out2, err := sqsClient.CreateQueue(&sqs.CreateQueueInput{QueueName: aws.String(subName)})
    if err != nil {
        return fmt.Errorf(`creating subscription queue "%s": %v`, subName, err)
    }

    log.Printf("Getting queue ARN.")
    out3, err := sqsClient.GetQueueAttributes(&sqs.GetQueueAttributesInput{
        QueueUrl:       out2.QueueUrl,
        AttributeNames: []*string{aws.String("QueueArn")},
    })
    if err != nil {
        return fmt.Errorf("getting queue ARN for %s: %v", *out2.QueueUrl, err)
    }
    qARN := out3.Attributes["QueueArn"]

    log.Printf("Subscribing the queue to the topic.")
    _, err = snsClient.Subscribe(&sns.SubscribeInput{
        TopicArn: out.TopicArn,
        Endpoint: qARN,
        Protocol: aws.String("sqs"),
    })
    if err != nil {
        return fmt.Errorf("subscribing: %v", err)
    }

    log.Printf("Sending a message to the topic.")
    msg := fmt.Sprintf("sns/sqs test. random number: %d", rand.Int())
    ctx := context.Background()
    _, err = snsClient.PublishWithContext(ctx, &sns.PublishInput{
        Message:  aws.String(msg),
        TopicArn: out.TopicArn,
    })
    if err != nil {
        return fmt.Errorf("publishing: %v", err)
    }

    log.Printf("Getting the queue URL")
    req, resp := sqsClient.GetQueueUrlRequest(&sqs.GetQueueUrlInput{QueueName: aws.String(subName)})
    err = req.Send()
    if err != nil {
        return fmt.Errorf("getting queue URL: %v", err)
    }
    qURL := *resp.QueueUrl

    for _ = range time.Tick(time.Second) {
        log.Printf("Receiving the message from the queue.")
        max := int64(10)
        output, err := sqsClient.ReceiveMessageWithContext(ctx, &sqs.ReceiveMessageInput{
            QueueUrl:            aws.String(qURL),
            MaxNumberOfMessages: aws.Int64(max),
        })
        if err != nil {
            return fmt.Errorf("receiving message: %v", err)
        }
        for _, m := range output.Messages {
            log.Printf("Got message %s.", *m.Body)
            log.Printf("Stopping.")
            return nil
        }
    }

    return nil
}

Here's the result:

[ ~/src/aws-sqs-issue ] go run main.go
2019/03/04 18:54:08 Creating an SNS topic.
2019/03/04 18:54:09 Creating an SQS queue.
2019/03/04 18:54:10 Getting queue ARN.
2019/03/04 18:54:10 Subscribing the queue to the topic.
2019/03/04 18:54:10 Sending a message to the topic.
2019/03/04 18:54:10 Getting the queue URL
2019/03/04 18:54:12 Receiving the message from the queue.
2019/03/04 18:54:13 Receiving the message from the queue.
2019/03/04 18:54:14 Receiving the message from the queue.
2019/03/04 18:54:15 Receiving the message from the queue.
2019/03/04 18:54:16 Receiving the message from the queue.
2019/03/04 18:54:17 Receiving the message from the queue.
2019/03/04 18:54:18 Receiving the message from the queue.
2019/03/04 18:54:19 Receiving the message from the queue.
2019/03/04 18:54:20 Receiving the message from the queue.
2019/03/04 18:54:21 Receiving the message from the queue.
2019/03/04 18:54:22 Receiving the message from the queue.
^Csignal: interrupt

Is there a bug somewhere or am I doing something wrong?

[ijt]

The response from the subscribe call gives a subscription ARN (arn:aws:sns:us-east-2:462380225722:test-topic:839a49af-fbf8-4ca9-b575-5fb81d59a07c) implying that it doesn't need confirmation.

However, when I try to edit the subscription attributes in the console (SNS dashboard | Topics | Other subscription options | Edit subscription attributes), it shows something wrong.

[diehlaws]

Thanks for the additional information. Unfortunately point #2 in my previous response was partially incorrect - when creating an SQS queue without specifying a Queue Access policy, only the owner of the queue has permission to send messages to this queue. I do apologize for the oversight on this point.

When a message is published to an SNS topic that has an SQS queue subscribed to it, SNS uses a different account than the one that owns the SQS queue and the SNS topic to send the message to the queue. This means that, even if the SQS queue and the SNS topic are on the same account, you need to specify an SQS Queue Access policy that allows messages to be sent to this queue as long as the source ARN matches the ARN of the SNS topic to which the SQS queue is subscribed. The final part of such a policy was not reflected in the sample policy I provided previously, following is an updated sample policy that represents this more accurately.

{
  "Version": "2012-10-17",
  "Id": "AllowSNS",
  "Statement": [
    {
      "Sid": "Sid1234567890123",
      "Effect": "Allow",
      "Principal": {
        "AWS": "*"
      },
      "Action": "SQS:SendMessage",
      "Resource": "arn:aws:sqs:us-east-2:123456789012:my-queue",
      "Condition": {
        "ArnEquals": {
          "aws:SourceArn": "arn:aws:sns:us-east-2:123456789012:my-topic"
        }
      }
    }
  ]
}

This policy can be set at the time the queue is being created as I showed in my previous response, or in a SetQueueAttributes call after GetQueueAttributes in your code sample. The latter would retrieve the ARNs for the SQS queue and SNS topic from the corresponding responses for the CreateTopic and GetQueueAttributes calls rather than having to hard-code the AWS account ID in the policy as my previous example showed:

    queuePolicy := `{
"Version": "2012-10-17",
"Id": "AllowQueue",
"Statement": [
{
"Sid": "MySQSPolicy001",
"Effect": "Allow",
"Principal": {
"AWS": "*"
},
"Action": "sqs:SendMessage",
"Resource": "` + *qARN + `",
"Condition": {
"ArnEquals": {
"aws:SourceArn": "` + *out.TopicArn + `"
}
}
}
]
}`
    log.Printf("Adding queue access policy to queue.")
    _, err = sqsClient.SetQueueAttributes(&sqs.SetQueueAttributesInput{
        Attributes: map[string]*string{
            "Policy": &queuePolicy,
        },
        QueueUrl: out2.QueueUrl,
    })
    if err != nil {
        return fmt.Errorf("setting policy: %v", err)
    }

As far as the validation error shown in your last screenshot is concerned, testing on both the web console and the AWS SDK for Go on my end did not show this behavior. Given that the subscription ID shows a subscription ARN rather than 'PendingConfirmation' the SQS queue should be successfully subscribed to the SNS topic (unless you selected a different subscription to open the 'Edit subscription attributes' window than the one included at the bottom of the screenshot). If you continue seeing this error when subscribing SQS queues to SNS topics after setting the Queue Access Policy I suggest reaching out to our Premium Support team by opening a new support case under the SNS service as they will have more in-depth knowledge on the service as well as access to internal tools that can give them more information about your SNS topic and the corresponding subscription to better troubleshoot this behavior.

[ijt]

Okay, this works. Thank you so much for your help. I'm posting the final working example here for future reference in case anyone else runs into this:

package main

import (
    "context"
    "fmt"
    "log"
    "math/rand"
    "net/http"
    "time"

    "github.com/aws/aws-sdk-go/aws"
    "github.com/aws/aws-sdk-go/aws/session"
    "github.com/aws/aws-sdk-go/service/sns"
    "github.com/aws/aws-sdk-go/service/sqs"
)

func main() {
    if err := makeTopicAndQueue(); err != nil {
        log.Fatalf("aws-sns-sqs: %v", err)
    }
}

func makeTopicAndQueue() error {
    sess, err := session.NewSession(&aws.Config{
        HTTPClient:  &http.Client{},
        Region:      aws.String("us-east-2"),
        Credentials: nil,
        MaxRetries:  aws.Int(0),
    })

    log.Printf("Creating an SNS topic")
    snsClient := sns.New(sess, &aws.Config{})
    topicName := "test-topic"
    out, err := snsClient.CreateTopic(&sns.CreateTopicInput{Name: aws.String(topicName)})
    if err != nil {
        return fmt.Errorf(`creating topic "%s": %v`, topicName, err)
    }
    defer snsClient.DeleteTopic(&sns.DeleteTopicInput{TopicArn: out.TopicArn})

    log.Printf("Creating an SQS queue")
    sqsClient := sqs.New(sess, &aws.Config{})
    subName := "test-subscription"
    out2, err := sqsClient.CreateQueue(&sqs.CreateQueueInput{QueueName: aws.String(subName)})
    if err != nil {
        return fmt.Errorf(`creating subscription queue "%s": %v`, subName, err)
    }

    log.Printf("Getting queue ARN")
    out3, err := sqsClient.GetQueueAttributes(&sqs.GetQueueAttributesInput{
        QueueUrl:       out2.QueueUrl,
        AttributeNames: []*string{aws.String("QueueArn")},
    })
    if err != nil {
        return fmt.Errorf("getting queue ARN for %s: %v", *out2.QueueUrl, err)
    }
    qARN := out3.Attributes["QueueArn"]

    log.Printf("Adding queue access policy to queue.")
    queuePolicy := `{
"Version": "2012-10-17",
"Id": "AllowQueue",
"Statement": [
{
"Sid": "MySQSPolicy001",
"Effect": "Allow",
"Principal": {
"AWS": "*"
},
"Action": "sqs:SendMessage",
"Resource": "` + *qARN + `",
"Condition": {
"ArnEquals": {
"aws:SourceArn": "` + *out.TopicArn + `"
}
}
}
]
}`
    _, err = sqsClient.SetQueueAttributes(&sqs.SetQueueAttributesInput{
        Attributes: map[string]*string{
            "Policy": &queuePolicy,
        },
        QueueUrl: out2.QueueUrl,
    })
    if err != nil {
        return fmt.Errorf("setting policy: %v", err)
    }

    log.Printf("Subscribing the queue to the topic")
    subResp, err := snsClient.Subscribe(&sns.SubscribeInput{
        Attributes: map[string]*string{"RawMessageDelivery": aws.String("true")},
        TopicArn:   out.TopicArn,
        Endpoint:   qARN,
        Protocol:   aws.String("sqs"),
    })
    if err != nil {
        return fmt.Errorf("subscribing: %v", err)
    }
    log.Printf("Subscription arn: '%s'", *subResp.SubscriptionArn)

    log.Printf("Sending a message to the topic")
    msg := fmt.Sprintf("sns/sqs test. random number: %d", rand.Int())
    ctx := context.Background()
    _, err = snsClient.PublishWithContext(ctx, &sns.PublishInput{
        Message:  aws.String(msg),
        TopicArn: out.TopicArn,
    })
    if err != nil {
        return fmt.Errorf("publishing: %v", err)
    }

    log.Printf("Getting the queue URL")
    req, resp := sqsClient.GetQueueUrlRequest(&sqs.GetQueueUrlInput{QueueName: aws.String(subName)})
    err = req.Send()
    if err != nil {
        return fmt.Errorf("getting queue URL: %v", err)
    }
    log.Printf("Queue URL: %s", *resp.QueueUrl)

    for _ = range time.Tick(time.Second) {
        log.Printf("Receiving the message from the queue")
        max := int64(10)
        output, err := sqsClient.ReceiveMessageWithContext(ctx, &sqs.ReceiveMessageInput{
            QueueUrl:            resp.QueueUrl,
            MaxNumberOfMessages: aws.Int64(max),
        })
        if err != nil {
            return fmt.Errorf("receiving message: %v", err)
        }
        for _, m := range output.Messages {
            log.Printf("Got message %s", *m.Body)
            log.Printf("Stopping")
            return nil
        }
    }

    return nil
}