fix(kinesis-output): adds new metrics for counting throttled writes

NamelessHH · Sep 18, 2018 · e60215e · e60215e
1 parent 8c72286
commit e60215e
Show file tree

Hide file tree

Showing 2 changed files with 51 additions and 15 deletions.
diff --git a/lib/output/writer/kinesis.go b/lib/output/writer/kinesis.go
@@ -102,6 +102,11 @@ type Kinesis struct {
 
 	log   log.Modular
 	stats metrics.Type
+
+	mThrottled       metrics.StatCounter
+	mThrottledF      metrics.StatCounter
+	mPartsThrottled  metrics.StatCounter
+	mPartsThrottledF metrics.StatCounter
 }
 
 // NewKinesis creates a new Amazon Kinesis writer.Type.
@@ -115,12 +120,16 @@ func NewKinesis(
 	}
 
 	k := Kinesis{
-		conf:         conf,
-		log:          log.NewModule(".output.kinesis"),
-		stats:        stats,
-		hashKey:      text.NewInterpolatedString(conf.HashKey),
-		partitionKey: text.NewInterpolatedString(conf.PartitionKey),
-		streamName:   aws.String(conf.Stream),
+		conf:             conf,
+		log:              log.NewModule(".output.kinesis"),
+		stats:            stats,
+		mPartsThrottled:  stats.GetCounter(".output.parts.send.throttled"),
+		mPartsThrottledF: stats.GetCounter(".output.kinesis.parts.send.throttled"),
+		mThrottled:       stats.GetCounter(".output.send.throttled"),
+		mThrottledF:      stats.GetCounter(".output.kinesis.send.throttled"),
+		hashKey:          text.NewInterpolatedString(conf.HashKey),
+		partitionKey:     text.NewInterpolatedString(conf.PartitionKey),
+		streamName:       aws.String(conf.Stream),
 	}
 
 	var err error
@@ -272,6 +281,10 @@ func (a *Kinesis) Write(msg types.Message) error {
 		// if throttling errors detected, pause briefly
 		l := len(failed)
 		if l > 0 {
+			a.mThrottled.Incr(1)
+			a.mThrottledF.Incr(1)
+			a.mPartsThrottled.Incr(int64(l))
+			a.mPartsThrottledF.Incr(int64(l))
 			a.log.Warnf("scheduling retry of throttled records (%d)\n", l)
 			if wait == backoff.Stop {
 				return types.ErrTimeout

diff --git a/lib/output/writer/kinesis_test.go b/lib/output/writer/kinesis_test.go
@@ -20,6 +20,17 @@ import (
 	"github.com/ory/dockertest"
 )
 
+var (
+	mockStats = metrics.DudType{}
+)
+
+var (
+	mThrottled       = mockStats.GetCounter(".output.send.throttled")
+	mThrottledF      = mockStats.GetCounter(".output.kinesis.send.throttled")
+	mPartsThrottled  = mockStats.GetCounter(".output.parts.send.throttled")
+	mPartsThrottledF = mockStats.GetCounter(".output.kinesis.parts.send.throttled")
+)
+
 type mockKinesis struct {
 	kinesisiface.KinesisAPI
 	fn func(input *kinesis.PutRecordsInput) (*kinesis.PutRecordsOutput, error)
@@ -174,9 +185,13 @@ func TestKinesisWriteChunkWithThrottling(t *testing.T) {
 				return &output, nil
 			},
 		},
-		log:          log.Noop(),
-		partitionKey: text.NewInterpolatedString("${!json_field:id}"),
-		hashKey:      text.NewInterpolatedString(""),
+		mThrottled:       mThrottled,
+		mThrottledF:      mThrottledF,
+		mPartsThrottled:  mPartsThrottled,
+		mPartsThrottledF: mPartsThrottledF,
+		log:              log.Noop(),
+		partitionKey:     text.NewInterpolatedString("${!json_field:id}"),
+		hashKey:          text.NewInterpolatedString(""),
 	}
 
 	msg := message.New(nil)
@@ -259,9 +274,13 @@ func TestKinesisWriteMessageThrottling(t *testing.T) {
 				return &output, nil
 			},
 		},
-		log:          log.Noop(),
-		partitionKey: text.NewInterpolatedString("${!json_field:id}"),
-		hashKey:      text.NewInterpolatedString(""),
+		mThrottled:       mThrottled,
+		mThrottledF:      mThrottledF,
+		mPartsThrottled:  mPartsThrottled,
+		mPartsThrottledF: mPartsThrottledF,
+		log:              log.Noop(),
+		partitionKey:     text.NewInterpolatedString("${!json_field:id}"),
+		hashKey:          text.NewInterpolatedString(""),
 	}
 
 	msg := message.New(nil)
@@ -301,9 +320,13 @@ func TestKinesisWriteBackoffMaxRetriesExceeded(t *testing.T) {
 				return &output, nil
 			},
 		},
-		log:          log.Noop(),
-		partitionKey: text.NewInterpolatedString("${!json_field:id}"),
-		hashKey:      text.NewInterpolatedString(""),
+		mThrottled:       mThrottled,
+		mThrottledF:      mThrottledF,
+		mPartsThrottled:  mPartsThrottled,
+		mPartsThrottledF: mPartsThrottledF,
+		log:              log.Noop(),
+		partitionKey:     text.NewInterpolatedString("${!json_field:id}"),
+		hashKey:          text.NewInterpolatedString(""),
 	}
 
 	msg := message.New(nil)