Merge pull request #2503 from kmlebedev/audit_log_nonblocking
Audit log force async
This commit is contained in:
@@ -197,6 +197,9 @@ func (s3opt *S3Options) startS3Server() bool {
|
||||
|
||||
if len(*s3opt.auditLogConfig) > 0 {
|
||||
s3err.InitAuditLog(*s3opt.auditLogConfig)
|
||||
if s3err.Logger != nil {
|
||||
defer s3err.Logger.Close()
|
||||
}
|
||||
}
|
||||
|
||||
if *s3opt.tlsPrivateKey != "" {
|
||||
|
||||
@@ -203,33 +203,44 @@ func (iam *IdentityAccessManagement) authRequest(r *http.Request, action Action)
|
||||
var identity *Identity
|
||||
var s3Err s3err.ErrorCode
|
||||
var found bool
|
||||
var authType string
|
||||
switch getRequestAuthType(r) {
|
||||
case authTypeStreamingSigned:
|
||||
return identity, s3err.ErrNone
|
||||
case authTypeUnknown:
|
||||
glog.V(3).Infof("unknown auth type")
|
||||
r.Header.Set(xhttp.AmzAuthType, "Unknown")
|
||||
return identity, s3err.ErrAccessDenied
|
||||
case authTypePresignedV2, authTypeSignedV2:
|
||||
glog.V(3).Infof("v2 auth type")
|
||||
identity, s3Err = iam.isReqAuthenticatedV2(r)
|
||||
authType = "SigV2"
|
||||
case authTypeSigned, authTypePresigned:
|
||||
glog.V(3).Infof("v4 auth type")
|
||||
identity, s3Err = iam.reqSignatureV4Verify(r)
|
||||
authType = "SigV4"
|
||||
case authTypePostPolicy:
|
||||
glog.V(3).Infof("post policy auth type")
|
||||
r.Header.Set(xhttp.AmzAuthType, "PostPolicy")
|
||||
return identity, s3err.ErrNone
|
||||
case authTypeJWT:
|
||||
glog.V(3).Infof("jwt auth type")
|
||||
r.Header.Set(xhttp.AmzAuthType, "Jwt")
|
||||
return identity, s3err.ErrNotImplemented
|
||||
case authTypeAnonymous:
|
||||
authType = "Anonymous"
|
||||
identity, found = iam.lookupAnonymous()
|
||||
if !found {
|
||||
r.Header.Set(xhttp.AmzAuthType, authType)
|
||||
return identity, s3err.ErrAccessDenied
|
||||
}
|
||||
default:
|
||||
return identity, s3err.ErrNotImplemented
|
||||
}
|
||||
|
||||
if len(authType) > 0 {
|
||||
r.Header.Set(xhttp.AmzAuthType, authType)
|
||||
}
|
||||
if s3Err != s3err.ErrNone {
|
||||
return identity, s3Err
|
||||
}
|
||||
@@ -250,33 +261,45 @@ func (iam *IdentityAccessManagement) authUser(r *http.Request) (*Identity, s3err
|
||||
var identity *Identity
|
||||
var s3Err s3err.ErrorCode
|
||||
var found bool
|
||||
var authType string
|
||||
switch getRequestAuthType(r) {
|
||||
case authTypeStreamingSigned:
|
||||
return identity, s3err.ErrNone
|
||||
case authTypeUnknown:
|
||||
glog.V(3).Infof("unknown auth type")
|
||||
r.Header.Set(xhttp.AmzAuthType, "Unknown")
|
||||
return identity, s3err.ErrAccessDenied
|
||||
case authTypePresignedV2, authTypeSignedV2:
|
||||
glog.V(3).Infof("v2 auth type")
|
||||
identity, s3Err = iam.isReqAuthenticatedV2(r)
|
||||
authType = "SigV2"
|
||||
case authTypeSigned, authTypePresigned:
|
||||
glog.V(3).Infof("v4 auth type")
|
||||
identity, s3Err = iam.reqSignatureV4Verify(r)
|
||||
authType = "SigV4"
|
||||
case authTypePostPolicy:
|
||||
glog.V(3).Infof("post policy auth type")
|
||||
r.Header.Set(xhttp.AmzAuthType, "PostPolicy")
|
||||
return identity, s3err.ErrNone
|
||||
case authTypeJWT:
|
||||
glog.V(3).Infof("jwt auth type")
|
||||
r.Header.Set(xhttp.AmzAuthType, "Jwt")
|
||||
return identity, s3err.ErrNotImplemented
|
||||
case authTypeAnonymous:
|
||||
authType = "Anonymous"
|
||||
identity, found = iam.lookupAnonymous()
|
||||
if !found {
|
||||
r.Header.Set(xhttp.AmzAuthType, authType)
|
||||
return identity, s3err.ErrAccessDenied
|
||||
}
|
||||
default:
|
||||
return identity, s3err.ErrNotImplemented
|
||||
}
|
||||
|
||||
if len(authType) > 0 {
|
||||
r.Header.Set(xhttp.AmzAuthType, authType)
|
||||
}
|
||||
|
||||
glog.V(3).Infof("auth error: %v", s3Err)
|
||||
if s3Err != s3err.ErrNone {
|
||||
return identity, s3Err
|
||||
|
||||
@@ -38,6 +38,7 @@ const (
|
||||
// Non-Standard S3 HTTP request constants
|
||||
const (
|
||||
AmzIdentityId = "s3-identity-id"
|
||||
AmzAuthType = "s3-auth-type"
|
||||
AmzIsAdmin = "s3-is-admin" // only set to http request header as a context
|
||||
)
|
||||
|
||||
|
||||
@@ -261,7 +261,7 @@ func (s3a *S3ApiServer) DeleteMultipleObjectsHandler(w http.ResponseWriter, r *h
|
||||
}
|
||||
if auditLog != nil {
|
||||
auditLog.Key = entryName
|
||||
s3err.PostAccessLog(auditLog)
|
||||
s3err.PostAccessLog(*auditLog)
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -48,23 +48,35 @@ type AccessLogHTTP struct {
|
||||
const tag = "s3.access"
|
||||
|
||||
var (
|
||||
Logger *fluent.Fluent
|
||||
hostname = os.Getenv("HOSTNAME")
|
||||
Logger *fluent.Fluent
|
||||
hostname = os.Getenv("HOSTNAME")
|
||||
environment = os.Getenv("ENVIRONMENT")
|
||||
)
|
||||
|
||||
func InitAuditLog(config string) {
|
||||
configContent, readErr := os.ReadFile(config)
|
||||
if readErr != nil {
|
||||
glog.Fatalf("fail to read fluent config %s : %v", config, readErr)
|
||||
glog.Errorf("fail to read fluent config %s : %v", config, readErr)
|
||||
return
|
||||
}
|
||||
var fluentConfig fluent.Config
|
||||
if err := json.Unmarshal(configContent, &fluentConfig); err != nil {
|
||||
glog.Fatalf("fail to parse fluent config %s : %v", config, err)
|
||||
fluentConfig := &fluent.Config{}
|
||||
if err := json.Unmarshal(configContent, fluentConfig); err != nil {
|
||||
glog.Errorf("fail to parse fluent config %s : %v", string(configContent), err)
|
||||
return
|
||||
}
|
||||
if len(fluentConfig.TagPrefix) == 0 && len(environment) > 0 {
|
||||
fluentConfig.TagPrefix = environment
|
||||
}
|
||||
fluentConfig.Async = true
|
||||
fluentConfig.AsyncResultCallback = func(data []byte, err error) {
|
||||
if err != nil {
|
||||
glog.Warning("Error while posting log: ", err)
|
||||
}
|
||||
}
|
||||
var err error
|
||||
Logger, err = fluent.New(fluentConfig)
|
||||
Logger, err = fluent.New(*fluentConfig)
|
||||
if err != nil {
|
||||
glog.Fatalf("fail to load fluent config: %v", err)
|
||||
glog.Errorf("fail to load fluent config: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -131,23 +143,24 @@ func GetAccessLog(r *http.Request, HTTPStatusCode int, s3errCode ErrorCode) *Acc
|
||||
if len(remoteIP) == 0 {
|
||||
remoteIP = r.RemoteAddr
|
||||
}
|
||||
hostHeader := r.Header.Get("Host")
|
||||
hostHeader := r.Header.Get("X-Forwarded-Host")
|
||||
if len(hostHeader) == 0 {
|
||||
hostHeader = r.URL.Hostname()
|
||||
hostHeader = r.Host
|
||||
}
|
||||
return &AccessLog{
|
||||
HostHeader: hostHeader,
|
||||
RequestID: r.Header.Get("X-Request-ID"),
|
||||
RemoteIP: remoteIP,
|
||||
Requester: r.Header.Get(xhttp.AmzIdentityId),
|
||||
UserAgent: r.Header.Get("UserAgent"),
|
||||
HostId: hostname,
|
||||
Bucket: bucket,
|
||||
HTTPStatus: HTTPStatusCode,
|
||||
Time: time.Now().Unix(),
|
||||
Key: key,
|
||||
Operation: getOperation(key, r),
|
||||
ErrorCode: errorCode,
|
||||
HostHeader: hostHeader,
|
||||
RequestID: r.Header.Get("X-Request-ID"),
|
||||
RemoteIP: remoteIP,
|
||||
Requester: r.Header.Get(xhttp.AmzIdentityId),
|
||||
SignatureVersion: r.Header.Get(xhttp.AmzAuthType),
|
||||
UserAgent: r.Header.Get("user-agent"),
|
||||
HostId: hostname,
|
||||
Bucket: bucket,
|
||||
HTTPStatus: HTTPStatusCode,
|
||||
Time: time.Now().Unix(),
|
||||
Key: key,
|
||||
Operation: getOperation(key, r),
|
||||
ErrorCode: errorCode,
|
||||
}
|
||||
}
|
||||
|
||||
@@ -160,11 +173,11 @@ func PostLog(r *http.Request, HTTPStatusCode int, errorCode ErrorCode) {
|
||||
}
|
||||
}
|
||||
|
||||
func PostAccessLog(log *AccessLog) {
|
||||
if Logger == nil || log == nil {
|
||||
func PostAccessLog(log AccessLog) {
|
||||
if Logger == nil || len(log.Key) == 0 {
|
||||
return
|
||||
}
|
||||
if err := Logger.Post(tag, *log); err != nil {
|
||||
if err := Logger.Post(tag, log); err != nil {
|
||||
glog.Warning("Error while posting log: ", err)
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user