Skip to content

Commit 8f68a92

Browse files
Copilotsanketsudake
Copilot
and
sanketsudake
authored
Fix deprecation warnings by migrating AWS SDK v1 to v2 and GCP PubSub v1 to v2 (#187)
* Initial plan * Migrate AWS SDK v1 to v2 and GCP PubSub v1 to v2 to fix deprecation warnings Co-authored-by: sanketsudake <1511975+sanketsudake@users.noreply.github.com> * Clean up binary files and update .gitignore Co-authored-by: sanketsudake <1511975+sanketsudake@users.noreply.github.com> --------- Co-authored-by: copilot-swe-agent[bot] <198982749+Copilot@users.noreply.github.com> Co-authored-by: sanketsudake <1511975+sanketsudake@users.noreply.github.com>
1 parent 1d0b5f5 commit 8f68a92

File tree

8 files changed

+145
-103
lines changed

8 files changed

+145
-103
lines changed

‎.gitignore‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -8,6 +8,12 @@
88
# Test binary, built with `go test -c`
99
*.test
1010

11+
# Go build binaries by directory name
12+
aws-kinesis-http-connector/aws-kinesis-http-connector
13+
aws-sqs-http-connector/aws-sqs-http-connector
14+
aws-sqs-http-connector/test/validate/validate
15+
gcp-pubsub-http-connector/gcp-pubsub-http-connector
16+
1117
# Output of the go coverage tool, specifically when used with LiteIDE
1218
*.out
1319

‎aws-kinesis-http-connector/main.go‎

Lines changed: 24 additions & 26 deletions
Original file line numberDiff line numberDiff line change
@@ -10,8 +10,9 @@ import (
1010
"sync"
1111
"time"
1212

13-
"github.com/aws/aws-sdk-go/aws"
14-
"github.com/aws/aws-sdk-go/service/kinesis"
13+
"github.com/aws/aws-sdk-go-v2/aws"
14+
"github.com/aws/aws-sdk-go-v2/service/kinesis"
15+
"github.com/aws/aws-sdk-go-v2/service/kinesis/types"
1516

1617
"github.com/fission/keda-connectors/common"
1718

@@ -20,23 +21,23 @@ import (
2021

2122
type pullFunc func(*record) error
2223
type record struct {
23-
*kinesis.Record
24+
*types.Record
2425
shardID string
2526
millisBehindLatest *int64
2627
}
2728
type awsKinesisConnector struct {
2829
ctx context.Context
29-
client *kinesis.Kinesis
30+
client *kinesis.Client
3031
connectordata common.ConnectorMetadata
3132
logger *zap.Logger
32-
shardc chan *kinesis.Shard
33+
shardc chan *types.Shard
3334
maxRecords int64
3435
}
3536

3637
// listShards get called every 30sec to get all the shards
37-
func (conn *awsKinesisConnector) listShards() ([]*kinesis.Shard, error) {
38+
func (conn *awsKinesisConnector) listShards() ([]types.Shard, error) {
3839
// call DescribeStream to get updated shards
39-
stream, err := conn.client.DescribeStream(&kinesis.DescribeStreamInput{
40+
stream, err := conn.client.DescribeStream(context.Background(), &kinesis.DescribeStreamInput{
4041
StreamName: &conn.connectordata.Topic,
4142
})
4243
if err != nil {
@@ -65,7 +66,7 @@ func (conn *awsKinesisConnector) findNewShards() {
6566
// send only new shards
6667
_, loaded := shards.LoadOrStore(*s.ShardId, s)
6768
if !loaded {
68-
conn.shardc <- s
69+
conn.shardc <- &s
6970
}
7071
}
7172
}
@@ -82,16 +83,16 @@ func (conn *awsKinesisConnector) getIterator(shardID string, checkpoint string)
8283
if checkpoint != "" {
8384
// Start from, where we left
8485
params.StartingSequenceNumber = aws.String(checkpoint)
85-
params.ShardIteratorType = aws.String(kinesis.ShardIteratorTypeAfterSequenceNumber)
86-
iteratorOutput, err := conn.client.GetShardIteratorWithContext(conn.ctx, params)
86+
params.ShardIteratorType = types.ShardIteratorTypeAfterSequenceNumber
87+
iteratorOutput, err := conn.client.GetShardIterator(conn.ctx, params)
8788
if err != nil {
8889
return nil, err
8990
}
9091
return iteratorOutput, err
9192
}
9293
// Start from, oldest record in the shard
93-
params.ShardIteratorType = aws.String(kinesis.ShardIteratorTypeTrimHorizon)
94-
iteratorOutput, err := conn.client.GetShardIteratorWithContext(conn.ctx, params)
94+
params.ShardIteratorType = types.ShardIteratorTypeTrimHorizon
95+
iteratorOutput, err := conn.client.GetShardIterator(conn.ctx, params)
9596
if err != nil {
9697
return nil, err
9798
}
@@ -101,9 +102,9 @@ func (conn *awsKinesisConnector) getIterator(shardID string, checkpoint string)
101102
// getRecords get the data for the specific shard
102103
func (conn *awsKinesisConnector) getRecords(shardIterator *string) (*kinesis.GetRecordsOutput, error) {
103104
// get records use shard iterator for making request
104-
records, err := conn.client.GetRecords(&kinesis.GetRecordsInput{
105+
records, err := conn.client.GetRecords(context.Background(), &kinesis.GetRecordsInput{
105106
ShardIterator: shardIterator,
106-
Limit: &conn.maxRecords,
107+
Limit: aws.Int32(int32(conn.maxRecords)),
107108
})
108109
if err != nil {
109110
return nil, err
@@ -158,7 +159,7 @@ func (conn *awsKinesisConnector) pullRecords(fn pullFunc) {
158159

159160
for _, r := range resp.Records {
160161
// send records
161-
err := fn(&record{r, shardID, resp.MillisBehindLatest})
162+
err := fn(&record{&r, shardID, resp.MillisBehindLatest})
162163
checkpoints.Store(shardID, *r.SequenceNumber)
163164
if err != nil {
164165
conn.logger.Error("error in processing records",
@@ -233,7 +234,7 @@ func (conn *awsKinesisConnector) responseHandler(r *record, response string) err
233234
StreamName: aws.String(conn.connectordata.ResponseTopic), // Required
234235
SequenceNumberForOrdering: aws.String(*r.SequenceNumber),
235236
}
236-
_, err := conn.client.PutRecord(params)
237+
_, err := conn.client.PutRecord(context.Background(), params)
237238
if err != nil {
238239
return err
239240
}
@@ -250,7 +251,7 @@ func (conn *awsKinesisConnector) errorHandler(r *record, errMsg string) {
250251
SequenceNumberForOrdering: aws.String(*r.SequenceNumber),
251252
}
252253

253-
_, err := conn.client.PutRecord(params)
254+
_, err := conn.client.PutRecord(context.Background(), params)
254255
if err != nil {
255256
conn.logger.Error("failed to publish message to error topic",
256257
zap.Error(err),
@@ -277,24 +278,21 @@ func main() {
277278
ctx, cancel := context.WithCancel(context.Background())
278279
defer cancel()
279280

280-
config, err := common.GetAwsConfig()
281+
cfg, err := common.GetAwsV2Config(ctx)
281282
if err != nil {
282283
logger.Error("failed to fetch aws config", zap.Error(err))
283284
return
284285
}
285286

286-
s, err := common.CreateValidatedSession(config)
287-
if err != nil {
288-
logger.Error("not able to create the session", zap.Error(err))
289-
return
290-
}
291-
kc := kinesis.New(s)
287+
kc := kinesis.NewFromConfig(cfg)
292288
connectordata, err := common.ParseConnectorMetadata()
293289
if err != nil {
294290
logger.Error("error while parsing metadata", zap.Error(err))
295291
return
296292
}
297-
if err := kc.WaitUntilStreamExists(&kinesis.DescribeStreamInput{StreamName: &connectordata.Topic}); err != nil {
293+
// Test stream exists by describing it
294+
_, err = kc.DescribeStream(ctx, &kinesis.DescribeStreamInput{StreamName: &connectordata.Topic})
295+
if err != nil {
298296
logger.Error("not able to connect to kinesis stream", zap.Error(err))
299297
return
300298
}
@@ -306,7 +304,7 @@ func main() {
306304
cancel() // call cancellation
307305
}()
308306

309-
shardc := make(chan *kinesis.Shard, 1)
307+
shardc := make(chan *types.Shard, 1)
310308

311309
conn := awsKinesisConnector{
312310
ctx: ctx,

‎aws-sqs-http-connector/main.go‎

Lines changed: 24 additions & 25 deletions
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,7 @@
11
package main
22

33
import (
4+
"context"
45
"io"
56
"log"
67
"net/url"
@@ -11,15 +12,16 @@ import (
1112

1213
"go.uber.org/zap"
1314

14-
"github.com/aws/aws-sdk-go/aws"
15-
"github.com/aws/aws-sdk-go/service/sqs"
15+
"github.com/aws/aws-sdk-go-v2/aws"
16+
"github.com/aws/aws-sdk-go-v2/service/sqs"
17+
"github.com/aws/aws-sdk-go-v2/service/sqs/types"
1618

1719
"github.com/fission/keda-connectors/common"
1820
)
1921

2022
type awsSQSConnector struct {
2123
sqsURL *url.URL
22-
sqsClient *sqs.SQS
24+
sqsClient *sqs.Client
2325
connectordata common.ConnectorMetadata
2426
logger *zap.Logger
2527
}
@@ -67,10 +69,10 @@ func (conn awsSQSConnector) consumeMessage() {
6769
conn.logger.Info("starting to consume messages from queue", zap.String("queue", consQueueURL), zap.String("response queue", respQueueURL), zap.String("error queue", errorQueueURL))
6870

6971
for {
70-
output, err := conn.sqsClient.ReceiveMessage(&sqs.ReceiveMessageInput{
72+
output, err := conn.sqsClient.ReceiveMessage(context.Background(), &sqs.ReceiveMessageInput{
7173
QueueUrl: &consQueueURL,
72-
MaxNumberOfMessages: &maxNumberOfMessages,
73-
WaitTimeSeconds: &waitTimeSeconds,
74+
MaxNumberOfMessages: int32(maxNumberOfMessages),
75+
WaitTimeSeconds: int32(waitTimeSeconds),
7476
})
7577

7678
if err != nil {
@@ -80,7 +82,7 @@ func (conn awsSQSConnector) consumeMessage() {
8082
for _, message := range output.Messages {
8183
// Set the attributes as message header came from SQS record
8284
for k, v := range message.Attributes {
83-
headers.Add(k, *v)
85+
headers.Add(k, v)
8486
}
8587

8688
resp, err := common.HandleHTTPRequest(*message.Body, headers, conn.connectordata, conn.logger)
@@ -92,17 +94,17 @@ func (conn awsSQSConnector) consumeMessage() {
9294
conn.errorHandler(errorQueueURL, err)
9395
} else {
9496
// Generating SQS Message attribute
95-
var sqsMessageAttValue = make(map[string]*sqs.MessageAttributeValue)
97+
var sqsMessageAttValue = make(map[string]types.MessageAttributeValue)
9698
for k, v := range resp.Header {
9799
for _, d := range v {
98-
sqsMessageAttValue[k] = &sqs.MessageAttributeValue{
100+
sqsMessageAttValue[k] = types.MessageAttributeValue{
99101
DataType: aws.String("String"),
100102
StringValue: aws.String(d),
101103
}
102104
}
103105
}
104106
if success := conn.responseHandler(respQueueURL, string(body), sqsMessageAttValue); success {
105-
conn.deleteMessage(*message.ReceiptHandle, consQueueURL)
107+
conn.deleteMessage(message.ReceiptHandle, consQueueURL)
106108
}
107109
}
108110
err = resp.Body.Close()
@@ -114,10 +116,10 @@ func (conn awsSQSConnector) consumeMessage() {
114116
}
115117
}
116118

117-
func (conn awsSQSConnector) responseHandler(queueURL string, response string, messageAttValue map[string]*sqs.MessageAttributeValue) bool {
119+
func (conn awsSQSConnector) responseHandler(queueURL string, response string, messageAttValue map[string]types.MessageAttributeValue) bool {
118120
if queueURL != "" {
119-
_, err := conn.sqsClient.SendMessage(&sqs.SendMessageInput{
120-
DelaySeconds: aws.Int64(10),
121+
_, err := conn.sqsClient.SendMessage(context.Background(), &sqs.SendMessageInput{
122+
DelaySeconds: 10,
121123
MessageAttributes: messageAttValue,
122124
MessageBody: &response,
123125
QueueUrl: &queueURL,
@@ -140,8 +142,8 @@ func (conn awsSQSConnector) responseHandler(queueURL string, response string, me
140142
func (conn *awsSQSConnector) errorHandler(queueURL string, err error) {
141143
if queueURL != "" {
142144
errMsg := err.Error()
143-
_, err := conn.sqsClient.SendMessage(&sqs.SendMessageInput{
144-
DelaySeconds: aws.Int64(10),
145+
_, err := conn.sqsClient.SendMessage(context.Background(), &sqs.SendMessageInput{
146+
DelaySeconds: 10,
145147
//MessageAttributes: messageAttValue,
146148
MessageBody: &errMsg,
147149
QueueUrl: &queueURL,
@@ -162,10 +164,10 @@ func (conn *awsSQSConnector) errorHandler(queueURL string, err error) {
162164
}
163165
}
164166

165-
func (conn *awsSQSConnector) deleteMessage(id string, queueURL string) {
166-
_, err := conn.sqsClient.DeleteMessage(&sqs.DeleteMessageInput{
167+
func (conn *awsSQSConnector) deleteMessage(id *string, queueURL string) {
168+
_, err := conn.sqsClient.DeleteMessage(context.Background(), &sqs.DeleteMessageInput{
167169
QueueUrl: &queueURL,
168-
ReceiptHandle: &id,
170+
ReceiptHandle: id,
169171
})
170172

171173
if err != nil {
@@ -187,18 +189,15 @@ func main() {
187189
if err != nil {
188190
logger.Fatal("failed to parse connector metadata", zap.Error(err))
189191
}
190-
config, err := common.GetAwsConfig()
192+
193+
ctx := context.Background()
194+
cfg, err := common.GetAwsV2Config(ctx)
191195
if err != nil {
192196
logger.Error("failed to fetch aws config", zap.Error(err))
193197
return
194198
}
195199

196-
sess, err := common.CreateValidatedSession(config)
197-
if err != nil {
198-
logger.Error("not able create session using aws configuration", zap.Error(err))
199-
return
200-
}
201-
svc := sqs.New(sess)
200+
svc := sqs.NewFromConfig(cfg)
202201

203202
sqsURL, err := url.Parse(strings.TrimSuffix(os.Getenv("QUEUE_URL"), os.Getenv("TOPIC")))
204203
if err != nil {

‎aws-sqs-http-connector/test/validate/main.go‎

Lines changed: 16 additions & 18 deletions
Original file line numberDiff line numberDiff line change
@@ -1,14 +1,14 @@
11
package main
22

33
import (
4+
"context"
45
"fmt"
56
"log"
67
"os"
78
"time"
89

9-
"github.com/aws/aws-sdk-go/aws"
10-
"github.com/aws/aws-sdk-go/aws/session"
11-
"github.com/aws/aws-sdk-go/service/sqs"
10+
"github.com/aws/aws-sdk-go-v2/config"
11+
"github.com/aws/aws-sdk-go-v2/service/sqs"
1212
)
1313

1414
// comment
@@ -25,22 +25,22 @@ func main() {
2525
if endpoint == "" {
2626
log.Fatal("AWS_ENDPOINT is not set")
2727
}
28-
config := &aws.Config{
29-
Region: &region,
30-
Endpoint: &endpoint,
31-
}
32-
33-
sess, err := session.NewSession(config)
28+
ctx := context.Background()
29+
cfg, err := config.LoadDefaultConfig(ctx,
30+
config.WithRegion(region),
31+
config.WithBaseEndpoint(endpoint),
32+
)
3433
if err != nil {
35-
log.Fatal("Error while creating session", err)
34+
log.Fatal("Error while loading config", err)
3635
}
37-
svc := sqs.New(sess)
36+
37+
svc := sqs.NewFromConfig(cfg)
3838

3939
msg := "Hello Msg"
4040
url := queueURL + "/my_queue"
4141
log.Println("Sending message to queue", url)
42-
_, err = svc.SendMessage(&sqs.SendMessageInput{
43-
DelaySeconds: aws.Int64(10),
42+
_, err = svc.SendMessage(ctx, &sqs.SendMessageInput{
43+
DelaySeconds: 10,
4444
MessageBody: &msg,
4545
QueueUrl: &url,
4646
})
@@ -50,11 +50,9 @@ func main() {
5050
time.Sleep(5 * time.Second)
5151
urlRep := queueURL + "/responseTopic"
5252
log.Println("Receiving message from queue", urlRep)
53-
var maxNumberOfMessages = int64(1)
54-
var waitTimeSeconds = int64(5)
55-
output, err := svc.ReceiveMessage(&sqs.ReceiveMessageInput{
56-
MaxNumberOfMessages: &maxNumberOfMessages,
57-
WaitTimeSeconds: &waitTimeSeconds,
53+
output, err := svc.ReceiveMessage(ctx, &sqs.ReceiveMessageInput{
54+
MaxNumberOfMessages: 1,
55+
WaitTimeSeconds: 5,
5856
QueueUrl: &urlRep,
5957
})
6058
if err != nil {

0 commit comments

Comments
 (0)