dom
/
telegraf
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

Use DescribeStreamSummary in place of ListStreams in kinesis output (#4864)

This commit is contained in:
Trevor Pounds 2018-10-18 16:05:43 -04:00 committed by Daniel Nelson
parent 7cb75ca979
commit 136a5724bd
3 changed files with 12 additions and 38 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
Gopkg.lock generated
View File

@ -145,7 +145,7 @@
revision = "f2867c24984aa53edec54a138c03db934221bdea" revision = "f2867c24984aa53edec54a138c03db934221bdea"
[[projects]] [[projects]]
digest = "1:65a05bde9b02f645c73afa61c9f6af92d94d726c81a268f45cc70218bd58de65" digest = "1:996727880e06dcf037f712c4d046e241d1b1b01844636fefb0fbaa480cfd230e"
name = "github.com/aws/aws-sdk-go" name = "github.com/aws/aws-sdk-go"
packages = [ packages = [
"aws", "aws",
@ -181,8 +181,8 @@
"service/sts", "service/sts",
] ]
pruneopts = "" pruneopts = ""
revision = "8cf662a972fa7fba8f2c1ec57648cf840e2bb401" revision = "bf8067ceb6e7f51e150c218972dccfeeed892b85"
version = "v1.14.30" version = "v1.15.54"
[[projects]] [[projects]]
branch = "master" branch = "master"

2
Gopkg.toml
View File

@ -16,7 +16,7 @@
[[constraint]] [[constraint]]
name = "github.com/aws/aws-sdk-go" name = "github.com/aws/aws-sdk-go"
version = "1.14.8" version = "1.15.54"
[[constraint]] [[constraint]]
name = "github.com/bsm/sarama-cluster" name = "github.com/bsm/sarama-cluster"

38
plugins/outputs/kinesis/kinesis.go
View File

@ -2,7 +2,6 @@ package kinesis
import ( import (
"log" "log"
"os"
"time" "time"
"github.com/aws/aws-sdk-go/aws" "github.com/aws/aws-sdk-go/aws"
@ -115,17 +114,11 @@ func (k *KinesisOutput) Description() string {
return "Configuration for the AWS Kinesis output." return "Configuration for the AWS Kinesis output."
} }
func checkstream(l []*string, s string) bool { func (k *KinesisOutput) Connect() error {
// Check if the StreamName exists in the slice returned from the ListStreams API request. if k.Partition == nil {
for _, stream := range l { log.Print("E! kinesis : Deprecated paritionkey configuration in use, please consider using outputs.kinesis.partition")
if *stream == s {
return true
}
}
return false
} }
func (k *KinesisOutput) Connect() error {
// We attempt first to create a session to Kinesis using an IAMS role, if that fails it will fall through to using // We attempt first to create a session to Kinesis using an IAMS role, if that fails it will fall through to using
// environment variables, and then Shared Credentials. // environment variables, and then Shared Credentials.
if k.Debug { if k.Debug {
@ -145,29 +138,10 @@ func (k *KinesisOutput) Connect() error {
configProvider := credentialConfig.Credentials() configProvider := credentialConfig.Credentials()
svc := kinesis.New(configProvider) svc := kinesis.New(configProvider)
KinesisParams := &kinesis.ListStreamsInput{ _, err := svc.DescribeStreamSummary(&kinesis.DescribeStreamSummaryInput{
Limit: aws.Int64(100), StreamName: aws.String(k.StreamName),
} })
resp, err := svc.ListStreams(KinesisParams)
if err != nil {
log.Printf("E! kinesis: Error in ListSteams API call : %+v \n", err)
}
if checkstream(resp.StreamNames, k.StreamName) {
if k.Debug {
log.Printf("E! kinesis: Stream Exists")
}
k.svc = svc k.svc = svc
return nil
} else {
log.Printf("E! kinesis : You have configured a StreamName %+v which does not exist. exiting.", k.StreamName)
os.Exit(1)
}
if k.Partition == nil {
log.Print("E! kinesis : Deprecated paritionkey configuration in use, please consider using outputs.kinesis.partition")
}
return err return err
} }