kitsune

README

Kitsune SQS Client

Go client for Amazon Simple Queue Service with support for messages larger than the max size of an SQS message by saving to S3 and encryption using Key Management Service.

Usage

All that is needed to make a client is an aws.Config

config := aws.Config{
		Region:      &awsRegion,
		Credentials: credentials.NewSharedCredentials("", profile),
	}

The client has a default set of options which should provide a good baseline. All the options can be overwritten using ClientOptions. NB: S3 and KMS is not a part of the default configuration. The only thing thats needed is the bucket where the client should save large payloads and the KMS key to be used for encryption.

options := []kitsune.ClientOption{
		kitsune.S3Bucket("myS3Bucket"),
		kitsune.KMSKeyID("myKMSKeyID"),
		kitsune.InitialVisibilityTimeout(10),
		kitsune.MaxVisibilityTimeout(100),
	}

Pass the aws.Config and ClientOptions to New and you're good to go. New will return an error if it was unable to get an aws.Session

client, err := kitsune.New(&config, options...)

Client Options

See https://docs.aws.amazon.com/AWSSimpleQueueService/latest/SQSDeveloperGuide/welcome.html for more details on some of the options.

Option	Default	Range	Comment
delaySeconds	30s	0 - 900s	Sets how many seconds a message will be unavailable before they are pollable from SQS.
maxNumberOfMessages	10	1 - 10	Max number of messages returned when polling SQS.
initialVisibilityTimeout	60s	0 - 43,200 (12hours max)	Sets the minimum visibility timeout in seconds when calculating visibility timeout.
maxVisibilityTimeout	900s	0 - 43,200 (12hours max)	Sets max visibility timeout in seconds when calculating visibility timeout.
backoffFactor	2	No limits. But should make sense in the function used for calculating backoff	Used when calculating visibility timeout.
backoffFunction	not set	N/A	Function used for calculating next visibility timeout. One can implement one or use on of the provided functions.
waitTimeSeconds	20	1 - 20s	Number of seconds a polling call will wait for response. Remeber to enable long polling when creating the queue.
attributeNames	ApproximateReceiveCount	N/A	Determines which (AWS specific) attributes are returned when polling SQS. ApproximateReceiveCount is used for backoff and is set by default.
messageAttributeNames	The ones used for S3, KMS and compression	0 - 7 (3 - 10) attributes. 3 used by the client.	Determines which (custom) attributes are returned when polling SQS. Remeber to add here if using any custom message attributes.
s3Bucket	Not set ("")	N/A	Determines which bucket payloads will be uploaded to. Remeber that sender and receiver might use different buckets. So make sure both have appropriate permissions.
forceS3	false	N/A	All messages will be put to S3 regardless of size
kmsKeyID	Not set ("")	N/A	Sets the KMS key usedfor encryption. Remember that the key used by sender and receiver is not necessarily the same. So each side needs to have permission for all keys used when sending and receiving.
compressionEnabled	false	N/A	Payloads are gzip compressed.
kmsKeyCacheEnabled	false	N/A	If enabled keys will be kept in memory for a set duration and reused. Note that caching keys is against best practise, which is why it's disabled by default, but it can save a lot on calls to KMS.
kmsKeyCacheExpirationPeriod	5min	N/A	The duration a key in the cache will be valid if key caching is enabled
skipSQSClient	false	N/A	Used when Lambda has SQS trigger and you dont need to handle SQS communication. Dont use this if you want the Lambda to put messages on a queue (using this client).

Planned features:

• Support large payloads by using S3
• Encrypt and decrypt with KMS
  • Cache KMS key
• Compression
• Batch sending with compression, encryption and large payloads to S3
• Simple message consumer which handles message lifecycle (receive, process, delete, backoff)

Policy suggestion

Make sure the user/role has the appripriate permissions. This policy assumes there is one queue, one bucket and one key being used. It is possible to use different buckets and keys with multiple clients. Just make sure to add all the resources to the appropriate policy (assuming a sufficient policy is applied to the user/role).

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Sid": "sqsPermissions",
            "Effect": "Allow",
            "Action": [
                "sqs:SendMessage",
                "sqs:ReceiveMessage",
                "sqs:ChangeMessageVisibility",
                "sqs:DeleteMessage",
                "sqs:GetQueueUrl"
            ],
            "Resource": "<SQS arn>"
        },
        {
            "Sid": "s3Permissions",
            "Effect": "Allow",
            "Action": [
                "s3:GetObject",
                "s3:PutObject"
            ],
            "Resource": "<S3 Bucket arn>*"
        },
        {
            "Sid": "kmsPermissions",
            "Effect": "Allow",
            "Action": [
                "kms:GenerateDataKey",
                "kms:Decrypt"
            ],
            "Resource": "<KMS Key arn>"
        }
    ]
}

Documentation

Index

• Constants
• Variables
• func ExponentialBackoff(retryCount, minBackoff, maxBackof, backoffFactor int64) int64
• func LinearBackoff(retryCount, minBackoff, maxBackof, backoffFactor int64) int64
• type Client
  • func New(awsSession *session.Session, opt ...ClientOption) (*Client, error)
  • func (c *Client) Backoff(queueName *string, message *sqs.Message) error
  • func (c *Client) ChangeMessageVisibility(queueName *string, message *sqs.Message, timeout int64) error
  • func (c *Client) DeleteMessage(queueName *string, receiptHandle *string) error
  • func (c *Client) ReceiveMessages(queueName *string) ([]*sqs.Message, error)
  • func (c *Client) ReceiveSQSEvent(event *events.SQSEvent) (*events.SQSEvent, error)
  • func (c *Client) SendMessage(queueName *string, payload []byte) error
  • func (c *Client) SendMessageWithAttributes(queueName *string, payload []byte, ...) error
• type ClientOption
  • func AttributeNames(s ...string) ClientOption
  • func BackoffFactor(b int64) ClientOption
  • func BackoffFunction(f func(int64, int64, int64, int64) int64) ClientOption
  • func CompressionEnabled(b bool) ClientOption
  • func DelaySeconds(d int64) ClientOption
  • func ForceS3(b bool) ClientOption
  • func InitialVisibilityTimeout(i int64) ClientOption
  • func KMSKeyCacheEnabled(b bool) ClientOption
  • func KMSKeyCacheExpirationPeriod(t time.Duration) ClientOption
  • func KMSKeyID(s string) ClientOption
  • func MaxNumberOfMessages(m int64) ClientOption
  • func MaxVisibilityTimeout(m int64) ClientOption
  • func MessageAttributeNames(s ...string) ClientOption
  • func S3Bucket(s string) ClientOption
  • func SkipSQSClient(b bool) ClientOption
  • func WaitTimeSeconds(w int64) ClientOption

Constants

const (
	// AttributeNameS3Bucket is an attribute name used by sender to pass the location of messages put on S3 to the receiver.
	// Receiver uses this attribute if set to fetch a message from S3.
	AttributeNameS3Bucket = "payloadBucket"

	// AttributeNameKMSKey used to pass the KMS key used to encryptData the data key.
	AttributeNameKMSKey = "kmsKey"

	// AttributeCompression is used to signal that the payload is compressed
	AttributeCompression = "compression"
)

Variables

var ErrorMaxMessageSizeExceeded = fmt.Errorf("maximum message size of %d bytes exceeded", maxMessageSize)

ErrorMaxMessageSizeExceeded is returned when the combined size of the payload and the message attributes exceeds maxMessageSize.

var ErrorMaxNumberOfAttributesExceeded = fmt.Errorf("maximum number of attributes of %d exceeded", maxNumberOfAttributes)

ErrorMaxNumberOfAttributesExceeded is returned when the number of attributes exceeds maxNumberOfAttributes.

Functions

func ExponentialBackoff

func ExponentialBackoff(retryCount, minBackoff, maxBackof, backoffFactor int64) int64

ExponentialBackoff can be configured on a client to achieve an exponential backoff strategy based on how many times the message is received.

func LinearBackoff

func LinearBackoff(retryCount, minBackoff, maxBackof, backoffFactor int64) int64

LinearBackoff can be configured on a awsSQSClient to achieve a linear backoff strategy based on how many times a message is received.

Types

type Client

type Client struct {
	// contains filtered or unexported fields
}

Client object handles communication with SQS

func New

func New(awsSession *session.Session, opt ...ClientOption) (*Client, error)

New returns a new awsSQSClient with configuration set as defined by the ClientOptions. Will create a s3Client from the aws.Config if a bucket is set. Same goes for KMS.

func (*Client) Backoff

func (c *Client) Backoff(queueName *string, message *sqs.Message) error

Backoff is used for changing message visibility based on a calculated amount of time determined by a back off function configured on the awsSQSClient.

func (*Client) ChangeMessageVisibility

func (c *Client) ChangeMessageVisibility(queueName *string, message *sqs.Message, timeout int64) error

ChangeMessageVisibility changes the visibilty of a message. Essentially putting it back in the queue and unavailable for a specified amount of time.

func (*Client) DeleteMessage

func (c *Client) DeleteMessage(queueName *string, receiptHandle *string) error

DeleteMessage removes a message from the queue.

func (*Client) ReceiveMessages

func (c *Client) ReceiveMessages(queueName *string) ([]*sqs.Message, error)

ReceiveMessages polls the specified queue and returns the fetched messages. If the S3 bucket attribute is set, the payload is fetched and replaces the file event in the sqs.Message body. This will not delete the object in S3. A lifecycle rule is recommended.

func (*Client) ReceiveSQSEvent

func (c *Client) ReceiveSQSEvent(event *events.SQSEvent) (*events.SQSEvent, error)

ReceiveSQSEvent unpacks payloads in a Lambda SQSEvent if compressed, encrypted or uploaded to S3 by the sennder.

func (*Client) SendMessage

func (c *Client) SendMessage(queueName *string, payload []byte) error

SendMessage sends a message to the specified queue. Convenient method for sending a message without custom attributes. This does not guarantee there will be no attributes on the message to SQS. The client might add attributes eg. for file events when the payload is uploaded to S3.

func (*Client) SendMessageWithAttributes

func (c *Client) SendMessageWithAttributes(queueName *string, payload []byte, messageAttributes map[string]*sqs.MessageAttributeValue) error

SendMessageWithAttributes sends a message to the specified queue with attributes. If the message size exceeds maximum, the payload will be uploaded to the configured S3 bucket and a file event will be sent on the SQS queue. The bucket where the message was uploaded if put on the message attributes. This means an no of attributes error can be thrown even though this function is called with less than maximum number of attributes.

type ClientOption

type ClientOption func(*options)

ClientOption sets configuration options for a awsSQSClient.

func AttributeNames

func AttributeNames(s ...string) ClientOption

AttributeNames sets the attributes to be returned when fetching messages. ApproximateReceiveCount is always returned because it is used when calculating backoff.

func BackoffFactor

func BackoffFactor(b int64) ClientOption

BackoffFactor sets the backoff factor which is a parameter used by the backoff function.

func BackoffFunction

func BackoffFunction(f func(int64, int64, int64, int64) int64) ClientOption

BackoffFunction sets the function which computes the next visibility timeout.

func CompressionEnabled

func CompressionEnabled(b bool) ClientOption

CompressionEnabled is used to enable or disable compression of payload.

func DelaySeconds

func DelaySeconds(d int64) ClientOption

DelaySeconds is used to set the DelaySeconds property on the awsSQSClient which is how many seconds the message will be unavaible once its put on a queue.

func ForceS3

func ForceS3(b bool) ClientOption

ForceS3 set if all messages should be saved to S3.

func InitialVisibilityTimeout

func InitialVisibilityTimeout(i int64) ClientOption

InitialVisibilityTimeout sets the initial time used when changing message visibility. The length of subsequent changes will be determined by strategy defined by the backoff function used.

func KMSKeyCacheEnabled

func KMSKeyCacheEnabled(b bool) ClientOption

KMSKeyCacheEnabled used to enable or disable kms key caching. Note that caching is against best practise, but might provide significant savings by reducing calls to KMS.

func KMSKeyCacheExpirationPeriod

func KMSKeyCacheExpirationPeriod(t time.Duration) ClientOption

KMSKeyCacheExpirationPeriod sets the amount of time an entry in the kms key cache will be valid.

func KMSKeyID

func KMSKeyID(s string) ClientOption

KMSKeyID sets the KMS key to be used for encryption.

func MaxNumberOfMessages

func MaxNumberOfMessages(m int64) ClientOption

MaxNumberOfMessages sets the maximum number of messages can be returned each time the awsSQSClient fetches messages.

func MaxVisibilityTimeout

func MaxVisibilityTimeout(m int64) ClientOption

MaxVisibilityTimeout sets the maxiumum time a message can be made unavailable by chaning message visibility.

func MessageAttributeNames

func MessageAttributeNames(s ...string) ClientOption

MessageAttributeNames sets the message attributes to be returned when fetching messages. ApproximateReceiveCount is always returned because it is used when calculating backoff.

func S3Bucket

func S3Bucket(s string) ClientOption

S3Bucket sets the bucket where the client should put messages which exceed max size.

func SkipSQSClient

func SkipSQSClient(b bool) ClientOption

SkipSQSClient can be used to skip creation of the SQS client. This is can be used in Lambdas with SQS trigger where handling SQS is not needed, but fetching from S3 and decrypting with KMS is.

func WaitTimeSeconds

func WaitTimeSeconds(w int64) ClientOption

WaitTimeSeconds sets the time a client will wait for messages on each call.