-
Notifications
You must be signed in to change notification settings - Fork 4.9k
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
[8.15](backport #40309) x-pack/filebeat/input/awss3: allow cross-regi…
…on bucket configuration (#40371) * x-pack/filebeat/input/awss3: allow cross-region bucket configuration (#40309) (cherry picked from commit e177fc5) * remove irrelevant changelog entries --------- Co-authored-by: Dan Kortschak <[email protected]>
- Loading branch information
1 parent
f218b32
commit 5054962
Showing
9 changed files
with
120 additions
and
64 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -9,6 +9,7 @@ import ( | |
| "errors" | ||
| "fmt" | ||
| "net/url" | ||
| "sync" | ||
| "time" | ||
|
|
||
| smithyhttp "github.com/aws/smithy-go/transport/http" | ||
|
|
@@ -30,6 +31,8 @@ import ( | |
| //go:generate go install github.com/golang/mock/[email protected] | ||
| //go:generate mockgen -source=interfaces.go -destination=mock_interfaces_test.go -package awss3 -mock_names=sqsAPI=MockSQSAPI,sqsProcessor=MockSQSProcessor,s3API=MockS3API,s3Pager=MockS3Pager,s3ObjectHandlerFactory=MockS3ObjectHandlerFactory,s3ObjectHandler=MockS3ObjectHandler | ||
| //go:generate mockgen -destination=mock_publisher_test.go -package=awss3 -mock_names=Client=MockBeatClient,Pipeline=MockBeatPipeline github.com/elastic/beats/v7/libbeat/beat Client,Pipeline | ||
| //go:generate go-licenser -license Elastic . | ||
| //go:generate goimports -w -local github.com/elastic . | ||
|
|
||
| // ------ | ||
| // SQS interfaces | ||
|
|
@@ -79,12 +82,12 @@ type s3API interface { | |
| } | ||
|
|
||
| type s3Getter interface { | ||
| GetObject(ctx context.Context, bucket, key string) (*s3.GetObjectOutput, error) | ||
| GetObject(ctx context.Context, region, bucket, key string) (*s3.GetObjectOutput, error) | ||
| } | ||
|
|
||
| type s3Mover interface { | ||
| CopyObject(ctx context.Context, from_bucket, to_bucket, from_key, to_key string) (*s3.CopyObjectOutput, error) | ||
| DeleteObject(ctx context.Context, bucket, key string) (*s3.DeleteObjectOutput, error) | ||
| CopyObject(ctx context.Context, region, from_bucket, to_bucket, from_key, to_key string) (*s3.CopyObjectOutput, error) | ||
| DeleteObject(ctx context.Context, region, bucket, key string) (*s3.DeleteObjectOutput, error) | ||
| } | ||
|
|
||
| type s3Lister interface { | ||
|
|
@@ -227,10 +230,23 @@ func (a *awsSQSAPI) GetQueueAttributes(ctx context.Context, attr []types.QueueAt | |
|
|
||
| type awsS3API struct { | ||
| client *s3.Client | ||
|
|
||
| // others is the set of other clients referred | ||
| // to by notifications seen by the API connection. | ||
| // The number of cached elements is limited to | ||
| // awsS3APIcacheMax. | ||
| mu sync.RWMutex | ||
| others map[string]*s3.Client | ||
| } | ||
|
|
||
| const awsS3APIcacheMax = 100 | ||
|
|
||
| func newAWSs3API(cli *s3.Client) *awsS3API { | ||
| return &awsS3API{client: cli, others: make(map[string]*s3.Client)} | ||
| } | ||
|
|
||
| func (a *awsS3API) GetObject(ctx context.Context, bucket, key string) (*s3.GetObjectOutput, error) { | ||
| getObjectOutput, err := a.client.GetObject(ctx, &s3.GetObjectInput{ | ||
| func (a *awsS3API) GetObject(ctx context.Context, region, bucket, key string) (*s3.GetObjectOutput, error) { | ||
| getObjectOutput, err := a.clientFor(region).GetObject(ctx, &s3.GetObjectInput{ | ||
| Bucket: awssdk.String(bucket), | ||
| Key: awssdk.String(key), | ||
| }, s3.WithAPIOptions( | ||
|
|
@@ -262,8 +278,8 @@ func (a *awsS3API) GetObject(ctx context.Context, bucket, key string) (*s3.GetOb | |
| return getObjectOutput, nil | ||
| } | ||
|
|
||
| func (a *awsS3API) CopyObject(ctx context.Context, from_bucket, to_bucket, from_key, to_key string) (*s3.CopyObjectOutput, error) { | ||
| copyObjectOutput, err := a.client.CopyObject(ctx, &s3.CopyObjectInput{ | ||
| func (a *awsS3API) CopyObject(ctx context.Context, region, from_bucket, to_bucket, from_key, to_key string) (*s3.CopyObjectOutput, error) { | ||
| copyObjectOutput, err := a.clientFor(region).CopyObject(ctx, &s3.CopyObjectInput{ | ||
| Bucket: awssdk.String(to_bucket), | ||
| CopySource: awssdk.String(fmt.Sprintf("%s/%s", from_bucket, from_key)), | ||
| Key: awssdk.String(to_key), | ||
|
|
@@ -274,8 +290,8 @@ func (a *awsS3API) CopyObject(ctx context.Context, from_bucket, to_bucket, from_ | |
| return copyObjectOutput, nil | ||
| } | ||
|
|
||
| func (a *awsS3API) DeleteObject(ctx context.Context, bucket, key string) (*s3.DeleteObjectOutput, error) { | ||
| deleteObjectOutput, err := a.client.DeleteObject(ctx, &s3.DeleteObjectInput{ | ||
| func (a *awsS3API) DeleteObject(ctx context.Context, region, bucket, key string) (*s3.DeleteObjectOutput, error) { | ||
| deleteObjectOutput, err := a.clientFor(region).DeleteObject(ctx, &s3.DeleteObjectInput{ | ||
| Bucket: awssdk.String(bucket), | ||
| Key: awssdk.String(key), | ||
| }) | ||
|
|
@@ -285,6 +301,49 @@ func (a *awsS3API) DeleteObject(ctx context.Context, bucket, key string) (*s3.De | |
| return deleteObjectOutput, nil | ||
| } | ||
|
|
||
| func (a *awsS3API) clientFor(region string) *s3.Client { | ||
| // Conditionally replace the client if the region of | ||
| // the request does not match the pre-prepared client. | ||
| opts := a.client.Options() | ||
| if opts.Region == region { | ||
| return a.client | ||
| } | ||
| // Use a cached client if we have already seen this region. | ||
| a.mu.RLock() | ||
| cli, ok := a.others[region] | ||
| a.mu.RUnlock() | ||
| if ok { | ||
| return cli | ||
| } | ||
|
|
||
| a.mu.Lock() | ||
| defer a.mu.Unlock() | ||
|
|
||
| // Check that another writer did not beat us here. | ||
| cli, ok = a.others[region] | ||
| if ok { | ||
| // ... they did. | ||
| return cli | ||
| } | ||
|
|
||
| // Otherwise create a new client and cache it. | ||
| opts.Region = region | ||
| cli = s3.New(opts) | ||
| // We should never be in the situation that the cache | ||
| // grows unbounded, but ensure this is the case. | ||
| if len(a.others) >= awsS3APIcacheMax { | ||
| // Do a single iteration delete to perform a | ||
| // random cache eviction. | ||
| for r := range a.others { | ||
| delete(a.others, r) | ||
| break | ||
| } | ||
| } | ||
| a.others[region] = cli | ||
|
|
||
| return cli | ||
| } | ||
|
|
||
| func (a *awsS3API) ListObjectsPaginator(bucket, prefix string) s3Pager { | ||
| pager := s3.NewListObjectsV2Paginator(a.client, &s3.ListObjectsV2Input{ | ||
| Bucket: awssdk.String(bucket), | ||
|
|
||
Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.
Oops, something went wrong.
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.