README
¶
Golang Kinesis Consumer
Note: This repo is under active development adding Consumer Groups #42. Master should always be deployable, but there may be interface changes in master over the next few months.
Latest stable release https://github.com/harlow/kinesis-consumer/releases/tag/v0.3.2
Kinesis consumer applications written in Go. This library is intended to be a lightweight wrapper around the Kinesis API to read records, save checkpoints (with swappable backends), and gracefully recover from service timeouts/errors.
Alternate serverless options:
-
Kinesis to Firehose can be used to archive data directly to S3, Redshift, or Elasticsearch without running a consumer application.
-
Process Kinesis Streams with Golang and AWS Lambda for serverless processing and checkpoint management.
Installation
Get the package source:
$ go get github.com/harlow/kinesis-consumer
Overview
The consumer leverages a handler func that accepts a Kinesis record. The Scan method will consume all shards concurrently and call the callback func as it receives records from the stream.
Important 1: The Scan func will also poll the stream to check for new shards, it will automatically start consuming new shards added to the stream.
Important 2: The default Log, Counter, and Checkpoint are no-op which means no logs, counts, or checkpoints will be emitted when scanning the stream. See the options below to override these defaults.
import(
// ...
consumer "github.com/harlow/kinesis-consumer"
)
func main() {
var stream = flag.String("stream", "", "Stream name")
flag.Parse()
// consumer
c, err := consumer.New(*stream)
if err != nil {
log.Fatalf("consumer error: %v", err)
}
// start scan
err = c.Scan(context.TODO(), func(r *consumer.Record) error {
fmt.Println(string(r.Data))
return nil // continue scanning
})
if err != nil {
log.Fatalf("scan error: %v", err)
}
// Note: If you need to aggregate based on a specific shard
// the `ScanShard` function should be used instead.
}
ScanFunc
ScanFunc is the type of the function called for each message read from the stream. The record argument contains the original record returned from the AWS Kinesis library.
type ScanFunc func(r *Record) error
If an error is returned, scanning stops. The sole exception is when the function returns the special value SkipCheckpoint.
// continue scanning
return nil
// continue scanning, skip checkpoint
return consumer.SkipCheckpoint
// stop scanning, return error
return errors.New("my error, exit all scans")
Use context cancel to signal the scan to exit without error. For example if we wanted to gracefully exit the scan on interrupt.
// trap SIGINT, wait to trigger shutdown
signals := make(chan os.Signal, 1)
signal.Notify(signals, os.Interrupt)
// context with cancel
ctx, cancel := context.WithCancel(context.Background())
go func() {
<-signals
cancel() // call cancellation
}()
err := c.Scan(ctx, func(r *consumer.Record) error {
fmt.Println(string(r.Data))
return nil // continue scanning
})
Options
The consumer allows the following optional overrides.
Store
To record the progress of the consumer in the stream (checkpoint) we use a storage layer to persist the last sequence number the consumer has read from a particular shard. The boolean value ErrSkipCheckpoint of consumer.ScanError determines if checkpoint will be activated. ScanError is returned by the record processing callback.
This will allow consumers to re-launch and pick up at the position in the stream where they left off.
The uniq identifier for a consumer is [appName, streamName, shardID]
Note: The default storage is in-memory (no-op). Which means the scan will not persist any state and the consumer will start from the beginning of the stream each time it is re-started.
The consumer accpets a WithStore option to set the storage layer:
c, err := consumer.New(*stream, consumer.WithStore(db))
if err != nil {
log.Log("consumer error: %v", err)
}
To persist scan progress choose one of the following storage layers:
Redis
The Redis checkpoint requries App Name, and Stream Name:
import store "github.com/harlow/kinesis-consumer/store/redis"
// redis checkpoint
db, err := store.New(appName)
if err != nil {
log.Fatalf("new checkpoint error: %v", err)
}
DynamoDB
The DynamoDB checkpoint requires Table Name, App Name, and Stream Name:
import store "github.com/harlow/kinesis-consumer/store/ddb"
// ddb checkpoint
db, err := store.New(appName, tableName)
if err != nil {
log.Fatalf("new checkpoint error: %v", err)
}
// Override the Kinesis if any needs on session (e.g. assume role)
myDynamoDbClient := dynamodb.New(session.New(aws.NewConfig()))
// For versions of AWS sdk that fixed config being picked up properly, the example of
// setting region should work.
// myDynamoDbClient := dynamodb.New(session.New(aws.NewConfig()), &aws.Config{
// Region: aws.String("us-west-2"),
// })
db, err := store.New(*app, *table, checkpoint.WithDynamoClient(myDynamoDbClient))
if err != nil {
log.Fatalf("new checkpoint error: %v", err)
}
// Or we can provide your own Retryer to customize what triggers a retry inside checkpoint
// See code in examples
// ck, err := checkpoint.New(*app, *table, checkpoint.WithDynamoClient(myDynamoDbClient), checkpoint.WithRetryer(&MyRetryer{}))
To leverage the DDB checkpoint we'll also need to create a table:
Partition key: namespace
Sort key: shard_id
Postgres
The Postgres checkpoint requires Table Name, App Name, Stream Name and ConnectionString:
import store "github.com/harlow/kinesis-consumer/store/postgres"
// postgres checkpoint
db, err := store.New(app, table, connStr)
if err != nil {
log.Fatalf("new checkpoint error: %v", err)
}
To leverage the Postgres checkpoint we'll also need to create a table:
CREATE TABLE kinesis_consumer (
namespace text NOT NULL,
shard_id text NOT NULL,
sequence_number numeric NOT NULL,
CONSTRAINT kinesis_consumer_pk PRIMARY KEY (namespace, shard_id)
);
The table name has to be the same that you specify when creating the checkpoint. The primary key composed by namespace and shard_id is mandatory in order to the checkpoint run without issues and also to ensure data integrity.
Mysql
The Mysql checkpoint requires Table Name, App Name, Stream Name and ConnectionString (just like the Postgres checkpoint!):
import store "github.com/harlow/kinesis-consumer/store/mysql"
// mysql checkpoint
db, err := store.New(app, table, connStr)
if err != nil {
log.Fatalf("new checkpoint error: %v", err)
}
To leverage the Mysql checkpoint we'll also need to create a table:
CREATE TABLE kinesis_consumer (
namespace varchar(255) NOT NULL,
shard_id varchar(255) NOT NULL,
sequence_number numeric(65,0) NOT NULL,
CONSTRAINT kinesis_consumer_pk PRIMARY KEY (namespace, shard_id)
);
The table name has to be the same that you specify when creating the checkpoint. The primary key composed by namespace and shard_id is mandatory in order to the checkpoint run without issues and also to ensure data integrity.
Kinesis Client
Override the Kinesis client if there is any special config needed:
// client
client := kinesis.New(session.NewSession(aws.NewConfig()))
// consumer
c, err := consumer.New(streamName, consumer.WithClient(client))
Metrics
Add optional counter for exposing counts for checkpoints and records processed:
// counter
counter := expvar.NewMap("counters")
// consumer
c, err := consumer.New(streamName, consumer.WithCounter(counter))
The expvar package will display consumer counts:
"counters": {
"checkpoints": 3,
"records": 13005
},
Consumer starting point
Kinesis allows consumers to specify where on the stream they'd like to start consuming from. The default in this library is LATEST (Start reading just after the most recent record in the shard).
This can be adjusted by using the WithShardIteratorType option in the library:
// override starting place on stream to use TRIM_HORIZON
c, err := consumer.New(
*stream,
consumer.WithShardIteratorType(kinesis.ShardIteratorTypeTrimHorizon)
)
See AWS Docs for more options.
Logging
Logging supports the basic built-in logging library or use thrid party external one, so long as it implements the Logger interface.
For example, to use the builtin logging package, we wrap it with myLogger structure.
// A myLogger provides a minimalistic logger satisfying the Logger interface.
type myLogger struct {
logger *log.Logger
}
// Log logs the parameters to the stdlib logger. See log.Println.
func (l *myLogger) Log(args ...interface{}) {
l.logger.Println(args...)
}
The package defaults to ioutil.Discard so swallow all logs. This can be customized with the preferred logging strategy:
// logger
logger := &myLogger{
logger: log.New(os.Stdout, "consumer-example: ", log.LstdFlags),
}
// consumer
c, err := consumer.New(streamName, consumer.WithLogger(logger))
To use a more complicated logging library, e.g. apex log
type myLogger struct {
logger *log.Logger
}
func (l *myLogger) Log(args ...interface{}) {
l.logger.Infof("producer", args...)
}
func main() {
log := &myLogger{
logger: alog.Logger{
Handler: text.New(os.Stderr),
Level: alog.DebugLevel,
},
}
Examples
There are example Produder and Consumer code in /cmd directory. These should help give end-to-end examples of setting up consumers with different checkpoint strategies.
The examples run locally against Kinesis Lite.
$ kinesalite &
Produce data to the stream:
$ cat cmd/producer/users.txt | go run cmd/producer/main.go --stream myStream
Consume data from the stream:
$ go run cmd/consumer/main.go --stream myStream
Contributing
Please see CONTRIBUTING.md for more information. Thank you, contributors!
License
Copyright (c) 2015 Harlow Ward. It is free software, and may be redistributed under the terms specified in the LICENSE file.
www.hward.com · GitHub @harlow · Twitter @harlow_ward
Documentation
¶
Index ¶
- Variables
- type AllGroup
- type Consumer
- type Counter
- type Group
- type Logger
- type Option
- func WithAggregation(a bool) Option
- func WithClient(client kinesisiface.KinesisAPI) Option
- func WithCounter(counter Counter) Option
- func WithGroup(group Group) Option
- func WithLogger(logger Logger) Option
- func WithMaxRecords(n int64) Option
- func WithRecordProcessingConcurrency(concurrency int64) Option
- func WithScanInterval(d time.Duration) Option
- func WithShardIteratorType(t string) Option
- func WithStore(store Store) Option
- func WithTimestamp(t time.Time) Option
- type Record
- type ScanFunc
- type Store
Constants ¶
This section is empty.
Variables ¶
var ErrSkipCheckpoint = errors.New("skip checkpoint")
ErrSkipCheckpoint is used as a return value from ScanFunc to indicate that the current checkpoint should be skipped skipped. It is not returned as an error by any function.
Functions ¶
This section is empty.
Types ¶
type AllGroup ¶ added in v0.3.5
type AllGroup struct {
Store
// contains filtered or unexported fields
}
AllGroup is used to consume all shards from a single consumer. It caches a local list of the shards we are already processing and routinely polls the stream looking for new shards to process.
func NewAllGroup ¶ added in v0.3.5
func NewAllGroup(ksis kinesisiface.KinesisAPI, store Store, streamName string, logger Logger) *AllGroup
NewAllGroup returns an intitialized AllGroup for consuming all shards on a stream
type Consumer ¶
type Consumer struct {
// contains filtered or unexported fields
}
Consumer wraps the interaction with the Kinesis stream
func New ¶ added in v0.2.0
New creates a kinesis consumer with default settings. Use Option to override any of the optional attributes.
type Counter ¶ added in v0.2.0
Counter interface is used for exposing basic metrics from the scanner
type Group ¶ added in v0.3.5
type Group interface {
Start(ctx context.Context, shardc chan *kinesis.Shard)
GetCheckpoint(streamName, shardID string) (string, error)
SetCheckpoint(streamName, shardID, sequenceNumber string) error
}
Group interface used to manage which shard to process
type Logger ¶ added in v0.2.0
type Logger interface {
Log(...interface{})
}
A Logger is a minimal interface to as a adaptor for external logging library to consumer
type Option ¶ added in v0.2.0
type Option func(*Consumer)
Option is used to override defaults when creating a new Consumer
func WithAggregation ¶ added in v0.3.5
func WithClient ¶ added in v0.2.0
func WithClient(client kinesisiface.KinesisAPI) Option
WithClient overrides the default client
func WithCounter ¶ added in v0.2.0
WithCounter overrides the default counter
func WithLogger ¶ added in v0.2.0
WithLogger overrides the default logger
func WithMaxRecords ¶ added in v0.3.5
WithMaxRecords overrides the maximum number of records to be returned in a single GetRecords call for the consumer (specify a value of up to 10,000)
func WithRecordProcessingConcurrency ¶ added in v0.3.5
WithRecordProcessingConcurrency overrides the number of goroutines running concurrently to process each record
func WithScanInterval ¶ added in v0.3.5
WithScanInterval overrides the scan interval for the consumer
func WithShardIteratorType ¶ added in v0.3.1
WithShardIteratorType overrides the starting point for the consumer
func WithTimestamp ¶ added in v0.3.5
WithTimestamp overrides the starting point for the consumer
type Record ¶ added in v0.2.0
Record wraps the record returned from the Kinesis library and extends to include the shard id.
type ScanFunc ¶ added in v0.3.5
ScanFunc is the type of the function called for each message read from the stream. The record argument contains the original record returned from the AWS Kinesis library. If an error is returned, scanning stops. The sole exception is when the function returns the special value ErrSkipCheckpoint.
Source Files
Directories