-
Notifications
You must be signed in to change notification settings - Fork 3.4k
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
Feature: Promtail, scrape logs from Object store #2270
Changes from all commits
82dc643
b23a90a
0024333
9e40657
e42a71c
b04890e
4091eac
87c6e0d
5485f61
f242a91
fe73650
af5d7f6
db1d9ad
95d3de1
bc2a265
a98d2e5
0457a9f
e38c55d
95a4f23
f97c6c3
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,219 @@ | ||
package objectstore | ||
|
||
import ( | ||
"crypto/tls" | ||
"crypto/x509" | ||
"encoding/json" | ||
"fmt" | ||
"net" | ||
"net/http" | ||
"os" | ||
"time" | ||
|
||
"github.com/aws/aws-sdk-go/aws" | ||
"github.com/aws/aws-sdk-go/aws/credentials" | ||
"github.com/aws/aws-sdk-go/aws/session" | ||
"github.com/aws/aws-sdk-go/service/sqs" | ||
"github.com/aws/aws-sdk-go/service/sqs/sqsiface" | ||
"github.com/pkg/errors" | ||
|
||
"github.com/grafana/loki/pkg/storage/chunk" | ||
loki_aws "github.com/grafana/loki/pkg/storage/chunk/aws" | ||
) | ||
|
||
type s3Client struct { | ||
queueURL *string | ||
svc sqsiface.SQSAPI | ||
} | ||
|
||
type s3TestEvent struct { | ||
Event string | ||
} | ||
|
||
type s3Results struct { | ||
Records []s3Record | ||
} | ||
|
||
type s3Record struct { | ||
S3 s3 | ||
EventTime string | ||
EventName string | ||
} | ||
|
||
type s3 struct { | ||
Object object | ||
} | ||
|
||
type object struct { | ||
Key string | ||
Size int64 | ||
} | ||
|
||
var ( | ||
defaultMaxRetries = 5 | ||
defaultMaxNumberOfMessages = int64(10) | ||
defaultVisibilityTimeout = int64(20) | ||
s3TestEventName = "s3:TestEvent" | ||
) | ||
|
||
func newS3Client(cfg loki_aws.S3Config, queue *string) (Client, error) { | ||
sess, err := buildSQSClient(cfg) | ||
if err != nil { | ||
return nil, err | ||
} | ||
svc := sqs.New(sess) | ||
urlResult, err := svc.GetQueueUrl(&sqs.GetQueueUrlInput{ | ||
QueueName: queue, | ||
}) | ||
if err != nil { | ||
return nil, err | ||
} | ||
queueURL := urlResult.QueueUrl | ||
return &s3Client{ | ||
queueURL: queueURL, | ||
svc: svc, | ||
}, nil | ||
} | ||
|
||
func buildSQSClient(cfg loki_aws.S3Config) (*session.Session, error) { | ||
sqsConfig := &aws.Config{} | ||
|
||
sqsConfig = sqsConfig.WithMaxRetries(defaultMaxRetries) | ||
|
||
if cfg.Region != "" { | ||
sqsConfig = sqsConfig.WithRegion(cfg.Region) | ||
} | ||
|
||
if cfg.AccessKeyID != "" && cfg.SecretAccessKey == "" || | ||
cfg.AccessKeyID == "" && cfg.SecretAccessKey != "" { | ||
return nil, errors.New("must supply both an Access Key ID and Secret Access Key or neither") | ||
} | ||
|
||
if cfg.AccessKeyID != "" && cfg.SecretAccessKey != "" { | ||
creds := credentials.NewStaticCredentials(cfg.AccessKeyID, cfg.SecretAccessKey, "") | ||
sqsConfig = sqsConfig.WithCredentials(creds) | ||
} | ||
|
||
tlsConfig := &tls.Config{ | ||
InsecureSkipVerify: cfg.HTTPConfig.InsecureSkipVerify, | ||
} | ||
|
||
if cfg.HTTPConfig.CAFile != "" { | ||
tlsConfig.RootCAs = x509.NewCertPool() | ||
data, err := os.ReadFile(cfg.HTTPConfig.CAFile) | ||
if err != nil { | ||
return nil, err | ||
} | ||
tlsConfig.RootCAs.AppendCertsFromPEM(data) | ||
} | ||
|
||
transport := http.RoundTripper(&http.Transport{ | ||
Proxy: http.ProxyFromEnvironment, | ||
DialContext: (&net.Dialer{ | ||
Timeout: 30 * time.Second, | ||
KeepAlive: 30 * time.Second, | ||
DualStack: true, | ||
}).DialContext, | ||
MaxIdleConns: 100, | ||
IdleConnTimeout: cfg.HTTPConfig.IdleConnTimeout, | ||
MaxIdleConnsPerHost: 100, | ||
TLSHandshakeTimeout: 3 * time.Second, | ||
ExpectContinueTimeout: 1 * time.Second, | ||
ResponseHeaderTimeout: cfg.HTTPConfig.ResponseHeaderTimeout, | ||
TLSClientConfig: tlsConfig, | ||
}) | ||
|
||
httpClient := &http.Client{ | ||
Transport: transport, | ||
} | ||
|
||
sqsConfig = sqsConfig.WithHTTPClient(httpClient) | ||
|
||
sess, err := session.NewSession(sqsConfig) | ||
if err != nil { | ||
return nil, errors.Wrap(err, "failed to create new s3 session") | ||
} | ||
return sess, nil | ||
} | ||
|
||
// ReceiveMessage implements Client | ||
func (s3Client *s3Client) ReceiveMessage(timeout int64) ([]messageObject, error) { | ||
msgResult, err := s3Client.svc.ReceiveMessage(&sqs.ReceiveMessageInput{ | ||
AttributeNames: []*string{ | ||
aws.String(sqs.MessageSystemAttributeNameSentTimestamp), | ||
}, | ||
MessageAttributeNames: []*string{ | ||
aws.String(sqs.QueueAttributeNameAll), | ||
}, | ||
QueueUrl: s3Client.queueURL, | ||
MaxNumberOfMessages: aws.Int64(defaultMaxNumberOfMessages), // hard coded. We might take this as user input in future versions. | ||
WaitTimeSeconds: aws.Int64(timeout), | ||
VisibilityTimeout: aws.Int64(defaultVisibilityTimeout), // hard coded. We might take this as user input in future versions. | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. If a large file takes more than 20sec to process and ack does it means another message will be sent. This means you have to run a single Promtail right ? if you run more then the position file is not shared/sync and they will both comsume the same message ? There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Yeah, to scale to process more files, running Promtail with same SQS queue would be tricky. One solution would be to create multiple SQS queues based on some pattern like |
||
}) | ||
if err != nil { | ||
return nil, err | ||
} | ||
return s3Client.processMessage(msgResult) | ||
} | ||
|
||
func (s3Client *s3Client) processMessage(messages interface{}) ([]messageObject, error) { | ||
var messageObjects []messageObject | ||
m := messages.(*sqs.ReceiveMessageOutput) | ||
for _, message := range m.Messages { | ||
content := *message.Body | ||
var sqsResult s3Results | ||
err := json.Unmarshal([]byte(content), &sqsResult) | ||
if err != nil { | ||
return nil, err | ||
} | ||
fmt.Printf("Message content %s", content) | ||
if len(sqsResult.Records) < 1 { | ||
var testEvent s3TestEvent | ||
err := json.Unmarshal([]byte(content), &testEvent) | ||
if err != nil { | ||
return nil, err | ||
} | ||
|
||
// S3 sends Test event on configuring SQS queue. | ||
// Unforutnately, it's not automatically deleted | ||
// and we have to handle it. | ||
if isTestMessage(testEvent) { | ||
if err := s3Client.acknowledgeMessage(message)(); err != nil { | ||
return nil, errors.Wrap(err, "error in deleting s3 test event message") | ||
} | ||
} | ||
continue | ||
} | ||
for _, record := range sqsResult.Records { | ||
t, err := time.Parse(time.RFC3339, record.EventTime) | ||
if err != nil { | ||
return nil, err | ||
} | ||
|
||
messageObjects = append(messageObjects, messageObject{ | ||
Object: chunk.StorageObject{ | ||
Key: record.S3.Object.Key, | ||
ModifiedAt: t, | ||
}, | ||
Acknowledge: s3Client.acknowledgeMessage(message), | ||
}) | ||
} | ||
|
||
} | ||
|
||
return messageObjects, nil | ||
} | ||
|
||
func (s3Client *s3Client) acknowledgeMessage(message *sqs.Message) ackMessage { | ||
return func() error { | ||
_, err := s3Client.svc.DeleteMessage(&sqs.DeleteMessageInput{QueueUrl: s3Client.queueURL, ReceiptHandle: message.ReceiptHandle}) | ||
if err != nil { | ||
return err | ||
} | ||
return nil | ||
} | ||
} | ||
|
||
func isTestMessage(testEvent s3TestEvent) bool { | ||
return testEvent.Event == s3TestEventName | ||
} |
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,16 @@ | ||
package objectstore | ||
|
||
import ( | ||
"github.com/grafana/loki/pkg/storage/chunk" | ||
) | ||
|
||
type Client interface { | ||
ReceiveMessage(timeout int64) ([]messageObject, error) | ||
} | ||
|
||
type ackMessage func() error | ||
|
||
type messageObject struct { | ||
Object chunk.StorageObject | ||
Acknowledge ackMessage | ||
} |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
If we want to support more cloud provider then we should do
NewS3TargetManager
.If not then we should rename the package to be
s3.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Have the check for what type of config it is here https://github.com/adityacs/loki/blob/object_storage_scrape/clients/pkg/promtail/targets/objectstore/targetmanager.go#L58
So, it should be okay to have just
NewTargetManager