self-hosted/minio
self-hosted/minio
1
0
Fork 0
mirror of https://github.com/minio/minio synced 2024-10-04 15:09:36 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat: add lambda transformation functions target (#16507)

Browse source
This commit is contained in:
Harshavardhana 2023-03-07 08:12:41 -08:00 committed by GitHub
parent ee54643004
commit 901887e6bf
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
29 changed files with 2130 additions and 70 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
cmd/admin-handlers.go
View file

@ -2429,7 +2429,7 @@ func assignPoolNumbers(servers []madmin.ServerProperties) {
func fetchLambdaInfo() []map[string][]madmin.TargetIDStatus {
lambdaMap := make(map[string][]madmin.TargetIDStatus)
for _, tgt := range globalConfigTargetList.Targets() {
for _, tgt := range globalNotifyTargetList.Targets() {
targetIDStatus := make(map[string]madmin.Status)
active, _ := tgt.IsActive()
targetID := tgt.ID()

97
cmd/api-errors.go
View file

@ -42,6 +42,7 @@ import (
objectlock "github.com/minio/minio/internal/bucket/object/lock"
"github.com/minio/minio/internal/bucket/versioning"
levent "github.com/minio/minio/internal/config/lambda/event"
"github.com/minio/minio/internal/event"
"github.com/minio/minio/internal/hash"
"github.com/minio/pkg/bucket/policy"
@ -411,6 +412,10 @@ const (
ErrInvalidChecksum
// Lambda functions
ErrLambdaARNInvalid
ErrLambdaARNNotFound
apiErrCodeEnd // This is used only for the testing code
)
@ -1964,6 +1969,16 @@ var errorCodes = errorCodeMap{
Description: "Invalid checksum provided.",
HTTPStatusCode: http.StatusBadRequest,
},
ErrLambdaARNInvalid: {
Code: "LambdaARNInvalid",
Description: "The specified lambda ARN is invalid",
HTTPStatusCode: http.StatusBadRequest,
},
ErrLambdaARNNotFound: {
Code: "LambdaARNNotFound",
Description: "The specified lambda ARN does not exist",
HTTPStatusCode: http.StatusNotFound,
},
ErrPolicyAlreadyAttached: {
Code: "XMinioPolicyAlreadyAttached",
Description: "The specified policy is already attached.",
@ -1987,15 +2002,15 @@ func toAPIErrorCode(ctx context.Context, err error) (apiErr APIErrorCode) {
// Only return ErrClientDisconnected if the provided context is actually canceled.
// This way downstream context.Canceled will still report ErrOperationTimedOut
if contextCanceled(ctx) {
if ctx.Err() == context.Canceled {
return ErrClientDisconnected
}
if contextCanceled(ctx) && errors.Is(ctx.Err(), context.Canceled) {
return ErrClientDisconnected
}
switch err {
case errInvalidArgument:
apiErr = ErrAdminInvalidArgument
case errNoSuchPolicy:
apiErr = ErrAdminNoSuchPolicy
case errNoSuchUser:
apiErr = ErrAdminNoSuchUser
case errNoSuchServiceAccount:
@ -2024,6 +2039,10 @@ func toAPIErrorCode(ctx context.Context, err error) (apiErr APIErrorCode) {
apiErr = ErrAdminInvalidSecretKey
case errInvalidStorageClass:
apiErr = ErrInvalidStorageClass
case errErasureReadQuorum:
apiErr = ErrSlowDown
case errErasureWriteQuorum:
apiErr = ErrSlowDown
// SSE errors
case errInvalidEncryptionParameters:
apiErr = ErrInvalidEncryptionParameters
@ -2071,10 +2090,6 @@ func toAPIErrorCode(ctx context.Context, err error) (apiErr APIErrorCode) {
apiErr = ErrObjectLockInvalidHeaders
case objectlock.ErrMalformedXML:
apiErr = ErrMalformedXML
default:
if errors.Is(err, errNoSuchPolicy) {
apiErr = ErrAdminNoSuchPolicy
}
}
// Compression errors
@ -2089,7 +2104,7 @@ func toAPIErrorCode(ctx context.Context, err error) (apiErr APIErrorCode) {
// etcd specific errors, a key is always a bucket for us return
// ErrNoSuchBucket in such a case.
if err == dns.ErrNoEntriesFound {
if errors.Is(err, dns.ErrNoEntriesFound) {
return ErrNoSuchBucket
}
@ -2214,6 +2229,10 @@ func toAPIErrorCode(ctx context.Context, err error) (apiErr APIErrorCode) {
apiErr = ErrARNNotification
case *event.ErrARNNotFound:
apiErr = ErrARNNotification
case *levent.ErrInvalidARN:
apiErr = ErrLambdaARNInvalid
case *levent.ErrARNNotFound:
apiErr = ErrLambdaARNNotFound
case *event.ErrUnknownRegion:
apiErr = ErrRegionNotification
case *event.ErrInvalidFilterName:
@ -2277,33 +2296,17 @@ func toAPIError(ctx context.Context, err error) APIError {
}
apiErr := errorCodes.ToAPIErr(toAPIErrorCode(ctx, err))
e, ok := err.(dns.ErrInvalidBucketName)
if ok {
code := toAPIErrorCode(ctx, e)
apiErr = errorCodes.ToAPIErrWithErr(code, e)
}
if apiErr.Code == "NotImplemented" {
if e, ok := err.(NotImplemented); ok {
desc := e.Error()
if desc == "" {
desc = apiErr.Description
}
apiErr = APIError{
Code: apiErr.Code,
Description: desc,
HTTPStatusCode: apiErr.HTTPStatusCode,
}
return apiErr
switch apiErr.Code {
case "NotImplemented":
desc := fmt.Sprintf("%s (%v)", apiErr.Description, err)
apiErr = APIError{
Code: apiErr.Code,
Description: desc,
HTTPStatusCode: apiErr.HTTPStatusCode,
}
}
if apiErr.Code == "XMinioBackendDown" {
case "XMinioBackendDown":
apiErr.Description = fmt.Sprintf("%s (%v)", apiErr.Description, err)
return apiErr
}
if apiErr.Code == "InternalError" {
case "InternalError":
// If we see an internal error try to interpret
// any underlying errors if possible depending on
// their internal error types.
@ -2318,22 +2321,20 @@ func toAPIError(ctx context.Context, err error) APIError {
}
case *xml.SyntaxError:
apiErr = APIError{
Code: "MalformedXML",
Description: fmt.Sprintf("%s (%s)", errorCodes[ErrMalformedXML].Description,
e.Error()),
Code: "MalformedXML",
Description: fmt.Sprintf("%s (%s)", errorCodes[ErrMalformedXML].Description, e),
HTTPStatusCode: errorCodes[ErrMalformedXML].HTTPStatusCode,
}
case url.EscapeError:
apiErr = APIError{
Code: "XMinioInvalidObjectName",
Description: fmt.Sprintf("%s (%s)", errorCodes[ErrInvalidObjectName].Description,
e.Error()),
Code: "XMinioInvalidObjectName",
Description: fmt.Sprintf("%s (%s)", errorCodes[ErrInvalidObjectName].Description, e),
HTTPStatusCode: http.StatusBadRequest,
}
case versioning.Error:
apiErr = APIError{
Code: "IllegalVersioningConfigurationException",
Description: fmt.Sprintf("Versioning configuration specified in the request is invalid. (%s)", e.Error()),
Description: fmt.Sprintf("Versioning configuration specified in the request is invalid. (%s)", e),
HTTPStatusCode: http.StatusBadRequest,
}
case lifecycle.Error:
@ -2399,19 +2400,7 @@ func toAPIError(ctx context.Context, err error) APIError {
// Add more other SDK related errors here if any in future.
default:
//nolint:gocritic
if errors.Is(err, errMalformedEncoding) {
apiErr = APIError{
Code: "BadRequest",
Description: err.Error(),
HTTPStatusCode: http.StatusBadRequest,
}
} else if errors.Is(err, errChunkTooBig) {
apiErr = APIError{
Code: "BadRequest",
Description: err.Error(),
HTTPStatusCode: http.StatusBadRequest,
}
} else if errors.Is(err, strconv.ErrRange) {
if errors.Is(err, errMalformedEncoding) || errors.Is(err, errChunkTooBig) || errors.Is(err, strconv.ErrRange) {
apiErr = APIError{
Code: "BadRequest",
Description: err.Error(),

5
cmd/api-router.go
View file

@ -285,7 +285,10 @@ func registerAPIRouter(router *mux.Router) {
// GetObjectLegalHold
router.Methods(http.MethodGet).Path("/{object:.+}").HandlerFunc(
collectAPIStats("getobjectlegalhold", maxClients(gz(httpTraceAll(api.GetObjectLegalHoldHandler))))).Queries("legal-hold", "")
// GetObject - note gzip compression is *not* added due to Range requests.
// GetObject with lambda ARNs
router.Methods(http.MethodGet).Path("/{object:.+}").HandlerFunc(
collectAPIStats("getobject", maxClients(gz(httpTraceHdrs(api.GetObjectLambdaHandler))))).Queries("lambdaArn", "{lambdaArn:.*}")
// GetObject
router.Methods(http.MethodGet).Path("/{object:.+}").HandlerFunc(
collectAPIStats("getobject", maxClients(gz(httpTraceHdrs(api.GetObjectHandler)))))
// CopyObject

8
cmd/apierrorcode_string.go
View file

File diff suppressed because one or more lines are too long

1
cmd/common-main.go
View file

@ -661,6 +661,7 @@ func handleCommonEnvVars() {
}
u.Path = "" // remove any path component such as `/`
globalMinioEndpoint = u.String()
globalMinioEndpointURL = u
}
globalFSOSync, err = config.ParseBool(env.Get(config.EnvFSOSync, config.EnableOff))

29
cmd/config-current.go
View file

@ -37,6 +37,7 @@ import (
"github.com/minio/minio/internal/config/identity/openid"
idplugin "github.com/minio/minio/internal/config/identity/plugin"
xtls "github.com/minio/minio/internal/config/identity/tls"
"github.com/minio/minio/internal/config/lambda"
"github.com/minio/minio/internal/config/notify"
"github.com/minio/minio/internal/config/policy/opa"
polplugin "github.com/minio/minio/internal/config/policy/plugin"
@ -74,6 +75,9 @@ func initHelp() {
for k, v := range notify.DefaultNotificationKVS {
kvs[k] = v
}
for k, v := range lambda.DefaultLambdaKVS {
kvs[k] = v
}
if globalIsErasure {
kvs[config.StorageClassSubSys] = storageclass.DefaultKVS
kvs[config.HealSubSys] = heal.DefaultKVS
@ -196,6 +200,11 @@ func initHelp() {
Description: "publish bucket notifications to Redis datastores",
MultipleTargets: true,
},
config.HelpKV{
Key: config.LambdaWebhookSubSys,
Description: "manage remote lambda functions",
MultipleTargets: true,
},
config.HelpKV{
Key: config.EtcdSubSys,
Description: "persist IAM assets externally to etcd",
@ -246,6 +255,7 @@ func initHelp() {
config.NotifyRedisSubSys: notify.HelpRedis,
config.NotifyWebhookSubSys: notify.HelpWebhook,
config.NotifyESSubSys: notify.HelpES,
config.LambdaWebhookSubSys: lambda.HelpWebhook,
config.SubnetSubSys: subnet.HelpSubnet,
config.CallhomeSubSys: callhome.HelpCallhome,
}
@ -387,6 +397,13 @@ func validateSubSysConfig(s config.Config, subSys string, objAPI ObjectLayer) er
return err
}
}
if config.LambdaSubSystems.Contains(subSys) {
if err := lambda.TestSubSysLambdaTargets(GlobalContext, s, subSys, NewHTTPTransport()); err != nil {
return err
}
}
return nil
}
@ -423,6 +440,7 @@ func lookupConfigs(s config.Config, objAPI ObjectLayer) {
logger.LogIf(ctx, fmt.Errorf("Unable to initialize remote webhook DNS config %w", err))
}
if err == nil && dnsURL != "" {
bootstrapTrace("initialize remote bucket DNS store")
globalDNSConfig, err = dns.NewOperatorDNS(dnsURL,
dns.Authentication(dnsUser, dnsPass),
dns.RootCAs(globalRootCAs))
@ -437,6 +455,7 @@ func lookupConfigs(s config.Config, objAPI ObjectLayer) {
}
if etcdCfg.Enabled {
bootstrapTrace("initialize etcd store")
globalEtcdClient, err = etcd.New(etcdCfg)
if err != nil {
logger.LogIf(ctx, fmt.Errorf("Unable to initialize etcd config: %w", err))
@ -495,12 +514,18 @@ func lookupConfigs(s config.Config, objAPI ObjectLayer) {
transport := NewHTTPTransport()
bootstrapTrace("lookup the event notification targets")
globalConfigTargetList, err = notify.FetchEnabledTargets(GlobalContext, s, transport)
bootstrapTrace("initialize the event notification targets")
globalNotifyTargetList, err = notify.FetchEnabledTargets(GlobalContext, s, transport)
if err != nil {
logger.LogIf(ctx, fmt.Errorf("Unable to initialize notification target(s): %w", err))
}
bootstrapTrace("initialize the lambda targets")
globalLambdaTargetList, err = lambda.FetchEnabledTargets(GlobalContext, s, transport)
if err != nil {
logger.LogIf(ctx, fmt.Errorf("Unable to initialize lambda target(s): %w", err))
}
bootstrapTrace("applying the dynamic configuration")
// Apply dynamic config values
if err := applyDynamicConfig(ctx, objAPI, s); err != nil {

2
cmd/event-notification.go
View file

@ -101,7 +101,7 @@ func (evnot *EventNotifier) InitBucketTargets(ctx context.Context, objAPI Object
return errServerNotInitialized
}
if err := evnot.targetList.Add(globalConfigTargetList.Targets()...); err != nil {
if err := evnot.targetList.Add(globalNotifyTargetList.Targets()...); err != nil {
return err
}

7
cmd/globals.go
View file

@ -49,6 +49,7 @@ import (
xhttp "github.com/minio/minio/internal/http"
etcd "go.etcd.io/etcd/client/v3"
levent "github.com/minio/minio/internal/config/lambda/event"
"github.com/minio/minio/internal/event"
"github.com/minio/minio/internal/pubsub"
"github.com/minio/pkg/certs"
@ -174,7 +175,8 @@ var (
globalMinioConsoleHost = ""
// Holds the possible host endpoint.
globalMinioEndpoint = ""
globalMinioEndpoint = ""
globalMinioEndpointURL *xnet.URL
// globalConfigSys server config system.
globalConfigSys *ConfigSys
@ -182,7 +184,8 @@ var (
globalNotificationSys *NotificationSys
globalEventNotifier *EventNotifier
globalConfigTargetList *event.TargetList
globalNotifyTargetList *event.TargetList
globalLambdaTargetList *levent.TargetList
globalBucketMetadataSys *BucketMetadataSys
globalBucketMonitor *bandwidth.Monitor

46
cmd/metrics-v2.go
View file

@ -131,6 +131,7 @@ const (
iamSubsystem MetricSubsystem = "iam"
kmsSubsystem MetricSubsystem = "kms"
notifySubsystem MetricSubsystem = "notify"
lambdaSubsystem MetricSubsystem = "lambda"
auditSubsystem MetricSubsystem = "audit"
)
@ -1656,8 +1657,8 @@ func getNotificationMetrics() *MetricsGroup {
cacheInterval: 10 * time.Second,
}
mg.RegisterRead(func(ctx context.Context) []Metric {
stats := globalConfigTargetList.Stats()
metrics := make([]Metric, 0, 1+len(stats.TargetStats))
nstats := globalNotifyTargetList.Stats()
metrics := make([]Metric, 0, 1+len(nstats.TargetStats))
metrics = append(metrics, Metric{
Description: MetricDescription{
Namespace: minioNamespace,
@ -1666,9 +1667,9 @@ func getNotificationMetrics() *MetricsGroup {
Help: "Number of concurrent async Send calls active to all targets",
Type: gaugeMetric,
},
Value: float64(stats.CurrentSendCalls),
Value: float64(nstats.CurrentSendCalls),
})
for _, st := range stats.TargetStats {
for _, st := range nstats.TargetStats {
metrics = append(metrics, Metric{
Description: MetricDescription{
Namespace: minioNamespace,
@ -1681,6 +1682,43 @@ func getNotificationMetrics() *MetricsGroup {
Value: float64(st.CurrentQueue),
})
}
lstats := globalLambdaTargetList.Stats()
for _, st := range lstats.TargetStats {
metrics = append(metrics, Metric{
Description: MetricDescription{
Namespace: minioNamespace,
Subsystem: lambdaSubsystem,
Name: "active_requests",
Help: "Number of in progress requests",
},
VariableLabels: map[string]string{"target_id": st.ID.ID, "target_name": st.ID.Name},
Value: float64(st.ActiveRequests),
})
metrics = append(metrics, Metric{
Description: MetricDescription{
Namespace: minioNamespace,
Subsystem: lambdaSubsystem,
Name: "total_requests",
Help: "Total number of requests sent since start",
Type: counterMetric,
},
VariableLabels: map[string]string{"target_id": st.ID.ID, "target_name": st.ID.Name},
Value: float64(st.TotalRequests),
})
metrics = append(metrics, Metric{
Description: MetricDescription{
Namespace: minioNamespace,
Subsystem: lambdaSubsystem,
Name: "failed_requests",
Help: "Total number of requests that failed to send since start",
Type: counterMetric,
},
VariableLabels: map[string]string{"target_id": st.ID.ID, "target_name": st.ID.Name},
Value: float64(st.FailedRequests),
})
}
// Audit and system:
audit := logger.CurrentStats()
for id, st := range audit {

291
cmd/object-lambda-handlers.go Normal file
View file

@ -0,0 +1,291 @@
// Copyright (c) 2015-2023 MinIO, Inc.
//
// This file is part of MinIO Object Storage stack
//
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// This program is distributed in the hope that it will be useful
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU Affero General Public License for more details.
//
// You should have received a copy of the GNU Affero General Public License
// along with this program. If not, see <http://www.gnu.org/licenses/>.
package cmd
import (
"crypto/subtle"
"io"
"net/http"
"net/url"
"strings"
"time"
"github.com/klauspost/compress/gzhttp"
"github.com/lithammer/shortuuid/v4"
miniogo "github.com/minio/minio-go/v7"
"github.com/minio/minio-go/v7/pkg/credentials"
"github.com/minio/mux"
"github.com/minio/pkg/bucket/policy"
"github.com/minio/minio/internal/auth"
levent "github.com/minio/minio/internal/config/lambda/event"
xhttp "github.com/minio/minio/internal/http"
"github.com/minio/minio/internal/logger"
)
func getLambdaEventData(bucket, object string, cred auth.Credentials, r *http.Request) (levent.Event, error) {
host := globalLocalNodeName
secure := globalIsTLS
if globalMinioEndpointURL != nil {
host = globalMinioEndpointURL.Host
secure = globalMinioEndpointURL.Scheme == "https"
}
duration := time.Since(cred.Expiration)
if cred.Expiration.IsZero() {
duration = time.Hour
}
clnt, err := miniogo.New(host, &miniogo.Options{
Creds: credentials.NewStaticV4(cred.AccessKey, cred.SecretKey, cred.SessionToken),
Secure: secure,
Transport: globalRemoteTargetTransport,
Region: globalSite.Region,
})
if err != nil {
return levent.Event{}, err
}
reqParams := url.Values{}
if partNumberStr := r.Form.Get("partNumber"); partNumberStr != "" {
reqParams.Set("partNumber", partNumberStr)
}
for k := range supportedHeadGetReqParams {
if v := r.Form.Get(k); v != "" {
reqParams.Set(k, v)
}
}
extraHeaders := http.Header{}
if rng := r.Header.Get(xhttp.Range); rng != "" {
extraHeaders.Set(xhttp.Range, r.Header.Get(xhttp.Range))
}
u, err := clnt.PresignHeader(r.Context(), http.MethodGet, bucket, object, duration, reqParams, extraHeaders)
if err != nil {
return levent.Event{}, err
}
token, err := authenticateNode(cred.AccessKey, cred.SecretKey, u.RawQuery)
if err != nil {
return levent.Event{}, err
}
eventData := levent.Event{
GetObjectContext: &levent.GetObjectContext{
InputS3URL: u.String(),
OutputRoute: shortuuid.New(),
OutputToken: token,
},
UserRequest: levent.UserRequest{
URL: r.URL.String(),
Headers: r.Header.Clone(),
},
UserIdentity: levent.Identity{
Type: "IAMUser",
PrincipalID: cred.AccessKey,
AccessKeyID: cred.SecretKey,
},
}
return eventData, nil
}
var statusTextToCode = map[string]int{
"Continue": http.StatusContinue,
"Switching Protocols": http.StatusSwitchingProtocols,
"Processing": http.StatusProcessing,
"Early Hints": http.StatusEarlyHints,
"OK": http.StatusOK,
"Created": http.StatusCreated,
"Accepted": http.StatusAccepted,
"Non-Authoritative Information": http.StatusNonAuthoritativeInfo,
"No Content": http.StatusNoContent,
"Reset Content": http.StatusResetContent,
"Partial Content": http.StatusPartialContent,
"Multi-Status": http.StatusMultiStatus,
"Already Reported": http.StatusAlreadyReported,
"IM Used": http.StatusIMUsed,
"Multiple Choices": http.StatusMultipleChoices,
"Moved Permanently": http.StatusMovedPermanently,
"Found": http.StatusFound,
"See Other": http.StatusSeeOther,
"Not Modified": http.StatusNotModified,
"Use Proxy": http.StatusUseProxy,
"Temporary Redirect": http.StatusTemporaryRedirect,
"Permanent Redirect": http.StatusPermanentRedirect,
"Bad Request": http.StatusBadRequest,
"Unauthorized": http.StatusUnauthorized,
"Payment Required": http.StatusPaymentRequired,
"Forbidden": http.StatusForbidden,
"Not Found": http.StatusNotFound,
"Method Not Allowed": http.StatusMethodNotAllowed,
"Not Acceptable": http.StatusNotAcceptable,
"Proxy Authentication Required": http.StatusProxyAuthRequired,
"Request Timeout": http.StatusRequestTimeout,
"Conflict": http.StatusConflict,
"Gone": http.StatusGone,
"Length Required": http.StatusLengthRequired,
"Precondition Failed": http.StatusPreconditionFailed,
"Request Entity Too Large": http.StatusRequestEntityTooLarge,
"Request URI Too Long": http.StatusRequestURITooLong,
"Unsupported Media Type": http.StatusUnsupportedMediaType,
"Requested Range Not Satisfiable": http.StatusRequestedRangeNotSatisfiable,
"Expectation Failed": http.StatusExpectationFailed,
"I'm a teapot": http.StatusTeapot,
"Misdirected Request": http.StatusMisdirectedRequest,
"Unprocessable Entity": http.StatusUnprocessableEntity,
"Locked": http.StatusLocked,
"Failed Dependency": http.StatusFailedDependency,
"Too Early": http.StatusTooEarly,
"Upgrade Required": http.StatusUpgradeRequired,
"Precondition Required": http.StatusPreconditionRequired,
"Too Many Requests": http.StatusTooManyRequests,
"Request Header Fields Too Large": http.StatusRequestHeaderFieldsTooLarge,
"Unavailable For Legal Reasons": http.StatusUnavailableForLegalReasons,
"Internal Server Error": http.StatusInternalServerError,
"Not Implemented": http.StatusNotImplemented,
"Bad Gateway": http.StatusBadGateway,
"Service Unavailable": http.StatusServiceUnavailable,
"Gateway Timeout": http.StatusGatewayTimeout,
"HTTP Version Not Supported": http.StatusHTTPVersionNotSupported,
"Variant Also Negotiates": http.StatusVariantAlsoNegotiates,
"Insufficient Storage": http.StatusInsufficientStorage,
"Loop Detected": http.StatusLoopDetected,
"Not Extended": http.StatusNotExtended,
"Network Authentication Required": http.StatusNetworkAuthenticationRequired,
}
// StatusCode returns a HTTP Status code for the HTTP text. It returns -1
// if the text is unknown.
func StatusCode(text string) int {
if code, ok := statusTextToCode[text]; ok {
return code
}
return -1
}
func fwdHeadersToS3(h http.Header, w http.ResponseWriter) {
const trim = "x-amz-fwd-header-"
for k, v := range h {
if strings.HasPrefix(strings.ToLower(k), trim) {
w.Header()[k[len(trim):]] = v
}
}
}
func fwdStatusToAPIError(resp *http.Response) *APIError {
if status := resp.Header.Get(xhttp.AmzFwdStatus); status != "" && StatusCode(status) > -1 {
apiErr := &APIError{
HTTPStatusCode: StatusCode(status),
Description: resp.Header.Get(xhttp.AmzFwdErrorMessage),
Code: resp.Header.Get(xhttp.AmzFwdErrorCode),
}
if apiErr.HTTPStatusCode == http.StatusOK {
return nil
}
return apiErr
}
return nil
}
// GetObjectLamdbaHandler - GET Object with transformed data via lambda functions
// ----------
// This implementation of the GET operation applies lambda functions and returns the
// response generated via the lambda functions. To use this API, you must have READ access
// to the object.
func (api objectAPIHandlers) GetObjectLambdaHandler(w http.ResponseWriter, r *http.Request) {
ctx := newContext(r, w, "GetObjectLambda")
defer logger.AuditLog(ctx, w, r, mustGetClaimsFromToken(r))
objectAPI := api.ObjectAPI()
if objectAPI == nil {
writeErrorResponse(ctx, w, errorCodes.ToAPIErr(ErrServerNotInitialized), r.URL)
return
}
vars := mux.Vars(r)
bucket := vars["bucket"]
object, err := unescapePath(vars["object"])
if err != nil {
writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL)
return
}
// Check for auth type to return S3 compatible error.
cred, _, s3Error := checkRequestAuthTypeCredential(ctx, r, policy.GetObjectAction)
if s3Error != ErrNone {
writeErrorResponse(ctx, w, errorCodes.ToAPIErr(s3Error), r.URL)
return
}
target, err := globalLambdaTargetList.Lookup(r.Form.Get("lambdaArn"))
if err != nil {
writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL)
return
}
eventData, err := getLambdaEventData(bucket, object, cred, r)
if err != nil {
writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL)
return
}
resp, err := target.Send(eventData)
if err != nil {
writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL)
return
}
defer resp.Body.Close()
if eventData.GetObjectContext.OutputRoute != resp.Header.Get(xhttp.AmzRequestRoute) {
tokenErr := errorCodes.ToAPIErr(ErrInvalidRequest)
tokenErr.Description = "The request route included in the request is invalid"
writeErrorResponse(ctx, w, tokenErr, r.URL)
return
}
if subtle.ConstantTimeCompare([]byte(resp.Header.Get(xhttp.AmzRequestToken)), []byte(eventData.GetObjectContext.OutputToken)) != 1 {
tokenErr := errorCodes.ToAPIErr(ErrInvalidToken)
tokenErr.Description = "The request token included in the request is invalid"
writeErrorResponse(ctx, w, tokenErr, r.URL)
return
}
// Set all the relevant lambda forward headers if found.
fwdHeadersToS3(resp.Header, w)
if apiErr := fwdStatusToAPIError(resp); apiErr != nil {
writeErrorResponse(ctx, w, *apiErr, r.URL)
return
}
if resp.StatusCode != http.StatusOK {
writeErrorResponse(ctx, w, APIError{
Code: "LambdaFunctionError",
HTTPStatusCode: resp.StatusCode,
Description: "unexpected failure reported from lambda function",
}, r.URL)
return
}
if !globalAPIConfig.shouldGzipObjects() {
w.Header().Set(gzhttp.HeaderNoCompression, "true")
}
io.Copy(w, resp.Body)
}

15
cmd/server-startup-msg.go
View file

@ -136,6 +136,7 @@ func printServerCommonMsg(apiEndpoints []string) {
}
}
printEventNotifiers()
printLambdaTargets()
if globalBrowserEnabled {
consoleEndpointStr := strings.Join(stripStandardPorts(getConsoleEndpoints(), globalMinioConsoleHost), " ")
@ -152,6 +153,18 @@ func printObjectAPIMsg() {
logger.Info(color.Blue("\nDocumentation: ") + "https://min.io/docs/minio/linux/index.html")
}
func printLambdaTargets() {
if globalLambdaTargetList == nil || globalLambdaTargetList.Empty() {
return
}
arnMsg := color.Blue("Object Lambda ARNs: ")
for _, arn := range globalLambdaTargetList.List(globalSite.Region) {
arnMsg += color.Bold(fmt.Sprintf("%s ", arn))
}
logger.Info(arnMsg + "\n")
}
// Prints bucket notification configurations.
func printEventNotifiers() {
if globalNotificationSys == nil {
@ -168,7 +181,7 @@ func printEventNotifiers() {
arnMsg += color.Bold(fmt.Sprintf("%s ", arn))
}
logger.Info(arnMsg)
logger.Info(arnMsg + "\n")
}
// Prints startup message for command line access. Prints link to our documentation

186
docs/lambda/README.md Normal file
View file

@ -0,0 +1,186 @@
# Object Lambda
MinIO's Object Lambda implementation allows for transforming your data to serve unique data format requirements for each application. For example, a dataset created by an ecommerce application might include personally identifiable information (PII). When the same data is processed for analytics, PII should be redacted. However, if the same dataset is used for a marketing campaign, you might need to enrich the data with additional details, such as information from the customer loyalty database.
MinIO's Object Lambda, enables application developers to process data retrieved from MinIO before returning it to an application. You can register a Lambda Function target on MinIO, once successfully registered it can be used to transform the data for application GET requests on demand.
This document focuses on showing a working example on how to use Object Lambda with MinIO, you must have [MinIO deployed in your environment](https://min.io/docs/minio/linux/operations/installation.html) before you can start using external lambda functions. You also must install Python version 3.8 or later for the lambda handlers to work.
## Example Lambda handler
Install the necessary dependencies.
```sh
pip install flask requests
```
Following is an example lambda handler.
```py
from flask import Flask, request, abort, make_response
import requests
app = Flask(__name__)
@app.route('/', methods=['POST'])
def get_webhook():
if request.method == 'POST':
# obtain the request event from the 'POST' call
event = request.json
object_context = event["getObjectContext"]
# Get the presigned URL to fetch the requested
# original object from MinIO
s3_url = object_context["inputS3Url"]
# Extract the route and request token from the input context
request_route = object_context["outputRoute"]
request_token = object_context["outputToken"]
# Get the original S3 object using the presigned URL
r = requests.get(s3_url)
original_object = r.content.decode('utf-8')
# Transform all text in the original object to uppercase
# You can replace it with your custom code based on your use case
transformed_object = original_object.upper()
# Write object back to S3 Object Lambda
# response sends the transformed data
# back to MinIO and then to the user
resp = make_response(transformed_object, 200)
resp.headers['x-amz-request-route'] = request_route
resp.headers['x-amz-request-token'] = request_token
return resp
else:
abort(400)
if __name__ == '__main__':
app.run()
```
When you're writing a Lambda function for use with MinIO, the function is based on event context that MinIO provides to the Lambda function. The event context provides information about the request being made. It contains the parameters with relevant context. The fields used to create the Lambda function are as follows:
The field of `getObjectContext` means the input and output details for connections to MinIO. It has the following fields:
- `inputS3Url` – A presigned URL that the Lambda function can use to download the original object. By using a presigned URL, the Lambda function doesn't need to have MinIO credentials to retrieve the original object. This allows Lambda function to focus on transformation of the object instead of securing the credentials.
- `outputRoute` – A routing token that is added to the response headers when the Lambda function returns the transformed object. This is used by MinIO to further verify the incoming response validity.
- `outputToken` – A token added to the response headers when the Lambda function returns the transformed object. This is used by MinIO to verify the incoming response validity.
Lets start the lamdba handler.
```
python lambda_handler.py
* Serving Flask app 'webhook'
* Debug mode: off
WARNING: This is a development server. Do not use it in a production deployment. Use a production WSGI server instead.
* Running on http://127.0.0.1:5000
Press CTRL+C to quit
```
## Start MinIO with Lambda target
Register MinIO with a Lambda function, we are calling our target name as `function`, but you may call it any other friendly name of your choice.
```
MINIO_LAMBDA_WEBHOOK_ENABLE_function=on MINIO_LAMBDA_WEBHOOK_ENDPOINT_function=http://localhost:5000 minio server /data &
...
...
MinIO Object Storage Server
Copyright: 2015-2023 MinIO, Inc.
License: GNU AGPLv3 <https://www.gnu.org/licenses/agpl-3.0.html>
Version: DEVELOPMENT.2023-02-05T05-17-27Z (go1.19.4 linux/amd64)
...
...
Object Lambda ARNs: arn:minio:s3-object-lambda::function:webhook
```
### Lambda Target with Auth Token
If your lambda target expects an authorization token then you can enable it per function target as follows
```
MINIO_LAMBDA_WEBHOOK_ENABLE_function=on MINIO_LAMBDA_WEBHOOK_ENDPOINT_function=http://localhost:5000 MINIO_LAMBDA_WEBHOOK_AUTH_TOKEN="mytoken" minio server /data &
```
### Lambda Target with mTLS authentication
If your lambda target expects mTLS client you can enable it per function target as follows
```
MINIO_LAMBDA_WEBHOOK_ENABLE_function=on MINIO_LAMBDA_WEBHOOK_ENDPOINT_function=http://localhost:5000 MINIO_LAMBDA_WEBHOOK_CLIENT_CERT=client.crt MINIO_LAMBDA_WEBHOOK_CLIENT_KEY=client.key minio server /data &
```
## Create a bucket and upload some data
Create a bucket named `functionbucket`
```
mc alias set myminio/ http://localhost:9000 minioadmin minioadmin
mc mb myminio/functionbucket
```
Create a file `testobject` with some test data that will be transformed
```
cat > testobject << EOF
MinIO is a High Performance Object Storage released under GNU Affero General Public License v3.0. It is API compatible with Amazon S3 cloud storage service. Use MinIO to build high performance infrastructure for machine learning, analytics and application data workloads.
EOF
```
Upload this object to the bucket via `mc cp`
```
mc cp testobject myminio/functionbucket/
```
## Invoke Lambda transformation via PresignedGET
Following example shows how you can use [`minio-go` PresignedGetObject](https://min.io/docs/minio/linux/developers/go/API.html#presignedgetobject-ctx-context-context-bucketname-objectname-string-expiry-time-duration-reqparams-url-values-url-url-error)
```go
package main
import (
"context"
"log"
"net/url"
"time"
"fmt"
"github.com/minio/minio-go/v7"
"github.com/minio/minio-go/v7/pkg/credentials"
)
func main() {
s3Client, err := minio.New("localhost:9000", &minio.Options{
Creds: credentials.NewStaticV4("minioadmin", "minioadmin", ""),
Secure: false,
})
if err != nil {
log.Fatalln(err)
}
// Set lambda function target via `lambdaArn`
reqParams := make(url.Values)
reqParams.Set("lambdaArn", "arn:minio:s3-object-lambda::function:webhook")
// Generate presigned GET url with lambda function
presignedURL, err := s3Client.PresignedGetObject(context.Background(), "functionbucket", "testobject", time.Duration(1000)*time.Second, reqParams)
if err != nil {
log.Fatalln(err)
}
fmt.Println(presignedURL)
}
```
Use the Presigned URL via `curl` to receive the transformed object.
```
~ curl -v $(go run presigned.go)
...
...
> GET /functionbucket/testobject?X-Amz-Algorithm=AWS4-HMAC-SHA256&X-Amz-Credential=minioadmin%2F20230205%2Fus-east-1%2Fs3%2Faws4_request&X-Amz-Date=20230205T173023Z&X-Amz-Expires=1000&X-Amz-SignedHeaders=host&lambdaArn=arn%3Aminio%3As3-object-lambda%3A%3Atoupper%3Awebhook&X-Amz-Signature=d7e343f0da9d4fa2bc822c12ad2f54300ff16796a1edaa6d31f1313c8e94d5b2 HTTP/1.1
> Host: localhost:9000
> User-Agent: curl/7.81.0
> Accept: */*
>
MINIO IS A HIGH PERFORMANCE OBJECT STORAGE RELEASED UNDER GNU AFFERO GENERAL PUBLIC LICENSE V3.0. IT IS API COMPATIBLE WITH AMAZON S3 CLOUD STORAGE SERVICE. USE MINIO TO BUILD HIGH PERFORMANCE INFRASTRUCTURE FOR MACHINE LEARNING, ANALYTICS AND APPLICATION DATA WORKLOADS.
```

10
internal/config/config.go
View file

@ -137,6 +137,11 @@ const (
// Add new constants here (similar to above) if you add new fields to config.
)
// Lambda config constants.
const (
LambdaWebhookSubSys = madmin.LambdaWebhookSubSys
)
// NotifySubSystems - all notification sub-systems
var NotifySubSystems = set.CreateStringSet(
NotifyKafkaSubSys,
@ -151,6 +156,11 @@ var NotifySubSystems = set.CreateStringSet(
NotifyWebhookSubSys,
)
// LambdaSubSystems - all lambda sub-systesm
var LambdaSubSystems = set.CreateStringSet(
LambdaWebhookSubSys,
)
// LoggerSubSystems - all sub-systems related to logger
var LoggerSubSystems = set.CreateStringSet(
LoggerWebhookSubSys,

40
internal/config/lambda/config.go Normal file
View file

@ -0,0 +1,40 @@
// Copyright (c) 2015-2023 MinIO, Inc.
//
// This file is part of MinIO Object Storage stack
//
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// This program is distributed in the hope that it will be useful
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU Affero General Public License for more details.
//
// You should have received a copy of the GNU Affero General Public License
// along with this program. If not, see <http://www.gnu.org/licenses/>.
package lambda
import "github.com/minio/minio/internal/event/target"
// Config - lambda target configuration structure, holds
// information about various lambda targets.
type Config struct {
Webhook map[string]target.WebhookArgs `json:"webhook"`
}
const (
defaultTarget = "1"
)
// NewConfig - initialize lambda config.
func NewConfig() Config {
// Make sure to initialize lambda targets
cfg := Config{
Webhook: make(map[string]target.WebhookArgs),
}
cfg.Webhook[defaultTarget] = target.WebhookArgs{}
return cfg
}

62
internal/config/lambda/event/arn.go Normal file
View file

@ -0,0 +1,62 @@
// Copyright (c) 2015-2023 MinIO, Inc.
//
// This file is part of MinIO Object Storage stack
//
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// This program is distributed in the hope that it will be useful
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU Affero General Public License for more details.
//
// You should have received a copy of the GNU Affero General Public License
// along with this program. If not, see <http://www.gnu.org/licenses/>.
package event
import (
"strings"
)
// ARN - SQS resource name representation.
type ARN struct {
TargetID
region string
}
// String - returns string representation.
func (arn ARN) String() string {
if arn.TargetID.ID == "" && arn.TargetID.Name == "" && arn.region == "" {
return ""
}
return "arn:minio:s3-object-lambda:" + arn.region + ":" + arn.TargetID.String()
}
// ParseARN - parses string to ARN.
func ParseARN(s string) (*ARN, error) {
// ARN must be in the format of arn:minio:s3-object-lambda:<REGION>:<ID>:<TYPE>
if !strings.HasPrefix(s, "arn:minio:s3-object-lambda:") {
return nil, &ErrInvalidARN{s}
}
tokens := strings.Split(s, ":")
if len(tokens) != 6 {
return nil, &ErrInvalidARN{s}
}
if tokens[4] == "" || tokens[5] == "" {
return nil, &ErrInvalidARN{s}
}
return &ARN{
region: tokens[3],
TargetID: TargetID{
ID: tokens[4],
Name: tokens[5],
},
}, nil
}

72
internal/config/lambda/event/arn_test.go Normal file
View file

@ -0,0 +1,72 @@
// Copyright (c) 2015-2023 MinIO, Inc.
//
// This file is part of MinIO Object Storage stack
//
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// This program is distributed in the hope that it will be useful
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU Affero General Public License for more details.
//
// You should have received a copy of the GNU Affero General Public License
// along with this program. If not, see <http://www.gnu.org/licenses/>.
package event
import (
"testing"
)
func TestARNString(t *testing.T) {
testCases := []struct {
arn ARN
expectedResult string
}{
{ARN{}, ""},
{ARN{TargetID{"1", "webhook"}, ""}, "arn:minio:s3-object-lambda::1:webhook"},
{ARN{TargetID{"1", "webhook"}, "us-east-1"}, "arn:minio:s3-object-lambda:us-east-1:1:webhook"},
}
for i, testCase := range testCases {
result := testCase.arn.String()
if result != testCase.expectedResult {
t.Fatalf("test %v: result: expected: %v, got: %v", i+1, testCase.expectedResult, result)
}
}
}
func TestParseARN(t *testing.T) {
testCases := []struct {
s string
expectedARN *ARN
expectErr bool
}{
{"", nil, true},
{"arn:minio:s3-object-lambda:::", nil, true},
{"arn:minio:s3-object-lambda::1:webhook:remote", nil, true},
{"arn:aws:s3-object-lambda::1:webhook", nil, true},
{"arn:minio:sns::1:webhook", nil, true},
{"arn:minio:s3-object-lambda::1:webhook", &ARN{TargetID{"1", "webhook"}, ""}, false},
{"arn:minio:s3-object-lambda:us-east-1:1:webhook", &ARN{TargetID{"1", "webhook"}, "us-east-1"}, false},
}
for i, testCase := range testCases {
arn, err := ParseARN(testCase.s)
expectErr := (err != nil)
if expectErr != testCase.expectErr {
t.Fatalf("test %v: error: expected: %v, got: %v", i+1, testCase.expectErr, expectErr)
}
if !testCase.expectErr {
if *arn != *testCase.expectedARN {
t.Fatalf("test %v: data: expected: %v, got: %v", i+1, testCase.expectedARN, arn)
}
}
}
}

49
internal/config/lambda/event/errors.go Normal file
View file

@ -0,0 +1,49 @@
// Copyright (c) 2015-2023 MinIO, Inc.
//
// This file is part of MinIO Object Storage stack
//
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// This program is distributed in the hope that it will be useful
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU Affero General Public License for more details.
//
// You should have received a copy of the GNU Affero General Public License
// along with this program. If not, see <http://www.gnu.org/licenses/>.
package event
import (
"fmt"
)
// ErrUnknownRegion - unknown region error.
type ErrUnknownRegion struct {
Region string
}
func (err ErrUnknownRegion) Error() string {
return fmt.Sprintf("unknown region '%v'", err.Region)
}
// ErrARNNotFound - ARN not found error.
type ErrARNNotFound struct {
ARN ARN
}
func (err ErrARNNotFound) Error() string {
return fmt.Sprintf("ARN '%v' not found", err.ARN)
}
// ErrInvalidARN - invalid ARN error.
type ErrInvalidARN struct {
ARN string
}
func (err ErrInvalidARN) Error() string {
return fmt.Sprintf("invalid ARN '%v'", err.ARN)
}

80
internal/config/lambda/event/event.go Normal file
View file

@ -0,0 +1,80 @@
// Copyright (c) 2015-2023 MinIO, Inc.
//
// This file is part of MinIO Object Storage stack
//
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// This program is distributed in the hope that it will be useful
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU Affero General Public License for more details.
//
// You should have received a copy of the GNU Affero General Public License
// along with this program. If not, see <http://www.gnu.org/licenses/>.
package event
import "net/http"
// Identity represents access key who caused the event.
type Identity struct {
Type string `json:"type"`
PrincipalID string `json:"principalId"`
AccessKeyID string `json:"accessKeyId"`
}
// UserRequest user request headers
type UserRequest struct {
URL string `json:"url"`
Headers http.Header `json:"headers"`
}
// GetObjectContext provides the necessary details to perform
// download of the object, and return back the processed response
// to the server.
type GetObjectContext struct {
OutputRoute string `json:"outputRoute"`
OutputToken string `json:"outputToken"`
InputS3URL string `json:"inputS3Url"`
}
// Event represents lambda function event, this is undocumented in AWS S3. This
// structure bases itself on this structure but there is no binding.
//
// {
// "xAmzRequestId": "a2871150-1df5-4dc9-ad9f-3da283ca1bf3",
// "getObjectContext": {
// "outputRoute": "...",
// "outputToken": "...",
// "inputS3Url": "<presignedURL>"
// },
// "configuration": { // not useful in MinIO
// "accessPointArn": "...",
// "supportingAccessPointArn": "...",
// "payload": ""
// },
// "userRequest": {
// "url": "...",
// "headers": {
// "Host": "...",
// "X-Amz-Content-SHA256": "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855"
// }
// },
// "userIdentity": {
// "type": "IAMUser",
// "principalId": "AIDAJF5MO57RFXQCE5ZNC",
// "arn": "...",
// "accountId": "...",
// "accessKeyId": "AKIA3WNQJCXE2DYPAU7R"
// },
// "protocolVersion": "1.00"
// }
type Event struct {
ProtocolVersion string `json:"protocolVersion"`
GetObjectContext *GetObjectContext `json:"getObjectContext"`
UserIdentity Identity `json:"userIdentity"`
UserRequest UserRequest `json:"userRequest"`
}

74
internal/config/lambda/event/targetid.go Normal file
View file

@ -0,0 +1,74 @@
// Copyright (c) 2015-2023 MinIO, Inc.
//
// This file is part of MinIO Object Storage stack
//
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// This program is distributed in the hope that it will be useful
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU Affero General Public License for more details.
//
// You should have received a copy of the GNU Affero General Public License
// along with this program. If not, see <http://www.gnu.org/licenses/>.
package event
import (
"encoding/json"
"fmt"
"strings"
)
// TargetID - holds identification and name strings of notification target.
type TargetID struct {
ID string
Name string
}
// String - returns string representation.
func (tid TargetID) String() string {
return tid.ID + ":" + tid.Name
}
// ToARN - converts to ARN.
func (tid TargetID) ToARN(region string) ARN {
return ARN{TargetID: tid, region: region}
}
// MarshalJSON - encodes to JSON data.
func (tid TargetID) MarshalJSON() ([]byte, error) {
return json.Marshal(tid.String())
}
// UnmarshalJSON - decodes JSON data.
func (tid *TargetID) UnmarshalJSON(data []byte) error {
var s string
if err := json.Unmarshal(data, &s); err != nil {
return err
}
targetID, err := parseTargetID(s)
if err != nil {
return err
}
*tid = *targetID
return nil
}
// parseTargetID - parses string to TargetID.
func parseTargetID(s string) (*TargetID, error) {
tokens := strings.Split(s, ":")
if len(tokens) != 2 {
return nil, fmt.Errorf("invalid TargetID format '%v'", s)
}
return &TargetID{
ID: tokens[0],
Name: tokens[1],
}, nil
}

118
internal/config/lambda/event/targetid_test.go Normal file
View file

@ -0,0 +1,118 @@
// Copyright (c) 2015-2023 MinIO, Inc.
//
// This file is part of MinIO Object Storage stack
//
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// This program is distributed in the hope that it will be useful
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU Affero General Public License for more details.
//
// You should have received a copy of the GNU Affero General Public License
// along with this program. If not, see <http://www.gnu.org/licenses/>.
package event
import (
"reflect"
"testing"
)
func TestTargetDString(t *testing.T) {
testCases := []struct {
tid TargetID
expectedResult string
}{
{TargetID{}, ":"},
{TargetID{"1", "webhook"}, "1:webhook"},
{TargetID{"httpclient+2e33cdee-fbec-4bdd-917e-7d8e3c5a2531", "localhost:55638"}, "httpclient+2e33cdee-fbec-4bdd-917e-7d8e3c5a2531:localhost:55638"},
}
for i, testCase := range testCases {
result := testCase.tid.String()
if result != testCase.expectedResult {
t.Fatalf("test %v: result: expected: %v, got: %v", i+1, testCase.expectedResult, result)
}
}
}
func TestTargetDToARN(t *testing.T) {
tid := TargetID{"1", "webhook"}
testCases := []struct {
tid TargetID
region string
expectedARN ARN
}{
{tid, "", ARN{TargetID: tid, region: ""}},
{tid, "us-east-1", ARN{TargetID: tid, region: "us-east-1"}},
}
for i, testCase := range testCases {
arn := testCase.tid.ToARN(testCase.region)
if arn != testCase.expectedARN {
t.Fatalf("test %v: ARN: expected: %v, got: %v", i+1, testCase.expectedARN, arn)
}
}
}
func TestTargetDMarshalJSON(t *testing.T) {
testCases := []struct {
tid TargetID
expectedData []byte
expectErr bool
}{
{TargetID{}, []byte(`":"`), false},
{TargetID{"1", "webhook"}, []byte(`"1:webhook"`), false},
{TargetID{"httpclient+2e33cdee-fbec-4bdd-917e-7d8e3c5a2531", "localhost:55638"}, []byte(`"httpclient+2e33cdee-fbec-4bdd-917e-7d8e3c5a2531:localhost:55638"`), false},
}
for i, testCase := range testCases {
data, err := testCase.tid.MarshalJSON()
expectErr := (err != nil)
if expectErr != testCase.expectErr {
t.Fatalf("test %v: error: expected: %v, got: %v", i+1, testCase.expectErr, expectErr)
}
if !testCase.expectErr {
if !reflect.DeepEqual(data, testCase.expectedData) {
t.Fatalf("test %v: data: expected: %v, got: %v", i+1, string(testCase.expectedData), string(data))
}
}
}
}
func TestTargetDUnmarshalJSON(t *testing.T) {
testCases := []struct {
data []byte
expectedTargetID *TargetID
expectErr bool
}{
{[]byte(`""`), nil, true},
{[]byte(`"httpclient+2e33cdee-fbec-4bdd-917e-7d8e3c5a2531:localhost:55638"`), nil, true},
{[]byte(`":"`), &TargetID{}, false},
{[]byte(`"1:webhook"`), &TargetID{"1", "webhook"}, false},
}
for i, testCase := range testCases {
targetID := &TargetID{}
err := targetID.UnmarshalJSON(testCase.data)
expectErr := (err != nil)
if expectErr != testCase.expectErr {
t.Fatalf("test %v: error: expected: %v, got: %v", i+1, testCase.expectErr, expectErr)
}
if !testCase.expectErr {
if *targetID != *testCase.expectedTargetID {
t.Fatalf("test %v: TargetID: expected: %v, got: %v", i+1, testCase.expectedTargetID, targetID)
}
}
}
}

72
internal/config/lambda/event/targetidset.go Normal file
View file

@ -0,0 +1,72 @@
// Copyright (c) 2015-2023 MinIO, Inc.
//
// This file is part of MinIO Object Storage stack
//
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// This program is distributed in the hope that it will be useful
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU Affero General Public License for more details.
//
// You should have received a copy of the GNU Affero General Public License
// along with this program. If not, see <http://www.gnu.org/licenses/>.
package event
// TargetIDSet - Set representation of TargetIDs.
type TargetIDSet map[TargetID]struct{}
// IsEmpty returns true if the set is empty.
func (set TargetIDSet) IsEmpty() bool {
return len(set) != 0
}
// Clone - returns copy of this set.
func (set TargetIDSet) Clone() TargetIDSet {
setCopy := NewTargetIDSet()
for k, v := range set {
setCopy[k] = v
}
return setCopy
}
// add - adds TargetID to the set.
func (set TargetIDSet) add(targetID TargetID) {
set[targetID] = struct{}{}
}
// Union - returns union with given set as new set.
func (set TargetIDSet) Union(sset TargetIDSet) TargetIDSet {
nset := set.Clone()
for k := range sset {
nset.add(k)
}
return nset
}
// Difference - returns diffrence with given set as new set.
func (set TargetIDSet) Difference(sset TargetIDSet) TargetIDSet {
nset := NewTargetIDSet()
for k := range set {
if _, ok := sset[k]; !ok {
nset.add(k)
}
}
return nset
}
// NewTargetIDSet - creates new TargetID set with given TargetIDs.
func NewTargetIDSet(targetIDs ...TargetID) TargetIDSet {
set := make(TargetIDSet)
for _, targetID := range targetIDs {
set.add(targetID)
}
return set
}

110
internal/config/lambda/event/targetidset_test.go Normal file
View file

@ -0,0 +1,110 @@
// Copyright (c) 2015-2023 MinIO, Inc.
//
// This file is part of MinIO Object Storage stack
//
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// This program is distributed in the hope that it will be useful
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU Affero General Public License for more details.
//
// You should have received a copy of the GNU Affero General Public License
// along with this program. If not, see <http://www.gnu.org/licenses/>.
package event
import (
"reflect"
"testing"
)
func TestTargetIDSetClone(t *testing.T) {
testCases := []struct {
set TargetIDSet
targetIDToAdd TargetID
}{
{NewTargetIDSet(), TargetID{"1", "webhook"}},
{NewTargetIDSet(TargetID{"1", "webhook"}), TargetID{"2", "webhook"}},
{NewTargetIDSet(TargetID{"1", "webhook"}, TargetID{"2", "amqp"}), TargetID{"2", "webhook"}},
}
for i, testCase := range testCases {
result := testCase.set.Clone()
if !reflect.DeepEqual(result, testCase.set) {
t.Fatalf("test %v: result: expected: %v, got: %v", i+1, testCase.set, result)
}
result.add(testCase.targetIDToAdd)
if reflect.DeepEqual(result, testCase.set) {
t.Fatalf("test %v: result: expected: not equal, got: equal", i+1)
}
}
}
func TestTargetIDSetUnion(t *testing.T) {
testCases := []struct {
set TargetIDSet
setToAdd TargetIDSet
expectedResult TargetIDSet
}{
{NewTargetIDSet(), NewTargetIDSet(), NewTargetIDSet()},
{NewTargetIDSet(), NewTargetIDSet(TargetID{"1", "webhook"}), NewTargetIDSet(TargetID{"1", "webhook"})},
{NewTargetIDSet(TargetID{"1", "webhook"}), NewTargetIDSet(), NewTargetIDSet(TargetID{"1", "webhook"})},
{NewTargetIDSet(TargetID{"1", "webhook"}), NewTargetIDSet(TargetID{"2", "amqp"}), NewTargetIDSet(TargetID{"1", "webhook"}, TargetID{"2", "amqp"})},
{NewTargetIDSet(TargetID{"1", "webhook"}), NewTargetIDSet(TargetID{"1", "webhook"}), NewTargetIDSet(TargetID{"1", "webhook"})},
}
for i, testCase := range testCases {
result := testCase.set.Union(testCase.setToAdd)
if !reflect.DeepEqual(testCase.expectedResult, result) {
t.Fatalf("test %v: result: expected: %v, got: %v", i+1, testCase.expectedResult, result)
}
}
}
func TestTargetIDSetDifference(t *testing.T) {
testCases := []struct {
set TargetIDSet
setToRemove TargetIDSet
expectedResult TargetIDSet
}{
{NewTargetIDSet(), NewTargetIDSet(), NewTargetIDSet()},
{NewTargetIDSet(), NewTargetIDSet(TargetID{"1", "webhook"}), NewTargetIDSet()},
{NewTargetIDSet(TargetID{"1", "webhook"}), NewTargetIDSet(), NewTargetIDSet(TargetID{"1", "webhook"})},
{NewTargetIDSet(TargetID{"1", "webhook"}), NewTargetIDSet(TargetID{"2", "amqp"}), NewTargetIDSet(TargetID{"1", "webhook"})},
{NewTargetIDSet(TargetID{"1", "webhook"}), NewTargetIDSet(TargetID{"1", "webhook"}), NewTargetIDSet()},
}
for i, testCase := range testCases {
result := testCase.set.Difference(testCase.setToRemove)
if !reflect.DeepEqual(testCase.expectedResult, result) {
t.Fatalf("test %v: result: expected: %v, got: %v", i+1, testCase.expectedResult, result)
}
}
}
func TestNewTargetIDSet(t *testing.T) {
testCases := []struct {
targetIDs []TargetID
expectedResult TargetIDSet
}{
{[]TargetID{}, NewTargetIDSet()},
{[]TargetID{{"1", "webhook"}}, NewTargetIDSet(TargetID{"1", "webhook"})},
{[]TargetID{{"1", "webhook"}, {"2", "amqp"}}, NewTargetIDSet(TargetID{"1", "webhook"}, TargetID{"2", "amqp"})},
}
for i, testCase := range testCases {
result := NewTargetIDSet(testCase.targetIDs...)
if !reflect.DeepEqual(testCase.expectedResult, result) {
t.Fatalf("test %v: result: expected: %v, got: %v", i+1, testCase.expectedResult, result)
}
}
}

189
internal/config/lambda/event/targetlist.go Normal file
View file

@ -0,0 +1,189 @@
// Copyright (c) 2015-2023 MinIO, Inc.
//
// This file is part of MinIO Object Storage stack
//
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// This program is distributed in the hope that it will be useful
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU Affero General Public License for more details.
//
// You should have received a copy of the GNU Affero General Public License
// along with this program. If not, see <http://www.gnu.org/licenses/>.
package event
import (
"fmt"
"net/http"
"strings"
"sync"
)
// Target - lambda target interface
type Target interface {
ID() TargetID
IsActive() (bool, error)
Send(Event) (*http.Response, error)
Stat() TargetStat
Close() error
}
// TargetStats is a collection of stats for multiple targets.
type TargetStats struct {
TargetStats map[string]TargetStat
}
// TargetStat is the stats of a single target.
type TargetStat struct {
ID TargetID
ActiveRequests int64
TotalRequests int64
FailedRequests int64
}
// TargetList - holds list of targets indexed by target ID.
type TargetList struct {
sync.RWMutex
targets map[TargetID]Target
}
// Add - adds unique target to target list.
func (list *TargetList) Add(targets ...Target) error {
list.Lock()
defer list.Unlock()
for _, target := range targets {
if _, ok := list.targets[target.ID()]; ok {
return fmt.Errorf("target %v already exists", target.ID())
}
list.targets[target.ID()] = target
}
return nil
}
// Lookup - checks whether target by target ID exists is valid or not.
func (list *TargetList) Lookup(arnStr string) (Target, error) {
list.RLock()
defer list.RUnlock()
arn, err := ParseARN(arnStr)
if err != nil {
return nil, err
}
id, found := list.targets[arn.TargetID]
if !found {
return nil, &ErrARNNotFound{}
}
return id, nil
}
// TargetIDResult returns result of Remove/Send operation, sets err if
// any for the associated TargetID
type TargetIDResult struct {
// ID where the remove or send were initiated.
ID TargetID
// Stores any error while removing a target or while sending an event.
Err error
}
// Remove - closes and removes targets by given target IDs.
func (list *TargetList) Remove(targetIDSet TargetIDSet) {
list.Lock()
defer list.Unlock()
for id := range targetIDSet {
target, ok := list.targets[id]
if ok {
target.Close()
delete(list.targets, id)
}
}
}
// Targets - list all targets
func (list *TargetList) Targets() []Target {
if list == nil {
return []Target{}
}
list.RLock()
defer list.RUnlock()
targets := make([]Target, 0, len(list.targets))
for _, tgt := range list.targets {
targets = append(targets, tgt)
}
return targets
}
// Empty returns true if targetList is empty.
func (list *TargetList) Empty() bool {
list.RLock()
defer list.RUnlock()
return len(list.targets) == 0
}
// List - returns available target IDs.
func (list *TargetList) List(region string) []ARN {
list.RLock()
defer list.RUnlock()
keys := make([]ARN, 0, len(list.targets))
for k := range list.targets {
keys = append(keys, k.ToARN(region))
}
return keys
}
// TargetMap - returns available targets.
func (list *TargetList) TargetMap() map[TargetID]Target {
list.RLock()
defer list.RUnlock()
ntargets := make(map[TargetID]Target, len(list.targets))
for k, v := range list.targets {
ntargets[k] = v
}
return ntargets
}
// Send - sends events to targets identified by target IDs.
func (list *TargetList) Send(event Event, id TargetID) (*http.Response, error) {
list.RLock()
target, ok := list.targets[id]
list.RUnlock()
if ok {
return target.Send(event)
}
return nil, ErrARNNotFound{}
}
// Stats returns stats for targets.
func (list *TargetList) Stats() TargetStats {
t := TargetStats{}
if list == nil {
return t
}
list.RLock()
defer list.RUnlock()
t.TargetStats = make(map[string]TargetStat, len(list.targets))
for id, target := range list.targets {
t.TargetStats[strings.ReplaceAll(id.String(), ":", "_")] = target.Stat()
}
return t
}
// NewTargetList - creates TargetList.
func NewTargetList() *TargetList {
return &TargetList{targets: make(map[TargetID]Target)}
}

62
internal/config/lambda/help.go Normal file
View file

@ -0,0 +1,62 @@
// Copyright (c) 2015-2023 MinIO, Inc.
//
// This file is part of MinIO Object Storage stack
//
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// This program is distributed in the hope that it will be useful
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU Affero General Public License for more details.
//
// You should have received a copy of the GNU Affero General Public License
// along with this program. If not, see <http://www.gnu.org/licenses/>.
package lambda
import (
"github.com/minio/minio/internal/config"
"github.com/minio/minio/internal/event/target"
)
// Help template inputs for all lambda targets
var (
HelpWebhook = config.HelpKVS{
config.HelpKV{
Key: target.WebhookEndpoint,
Description: "webhook server endpoint e.g. http://localhost:8080/minio/lambda",
Type: "url",
Sensitive: true,
},
config.HelpKV{
Key: target.WebhookAuthToken,
Description: "opaque string or JWT authorization token",
Optional: true,
Type: "string",
Sensitive: true,
},
config.HelpKV{
Key: config.Comment,
Description: config.DefaultComment,
Optional: true,
Type: "sentence",
},
config.HelpKV{
Key: target.WebhookClientCert,
Description: "client cert for Webhook mTLS auth",
Optional: true,
Type: "string",
Sensitive: true,
},
config.HelpKV{
Key: target.WebhookClientKey,
Description: "client cert key for Webhook mTLS auth",
Optional: true,
Type: "string",
Sensitive: true,
},
}
)

211
internal/config/lambda/parse.go Normal file
View file

@ -0,0 +1,211 @@
// Copyright (c) 2015-2023 MinIO, Inc.
//
// This file is part of MinIO Object Storage stack
//
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// This program is distributed in the hope that it will be useful
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU Affero General Public License for more details.
//
// You should have received a copy of the GNU Affero General Public License
// along with this program. If not, see <http://www.gnu.org/licenses/>.
package lambda
import (
"context"
"errors"
"fmt"
"net/http"
"github.com/minio/minio/internal/config"
"github.com/minio/minio/internal/config/lambda/event"
"github.com/minio/minio/internal/config/lambda/target"
"github.com/minio/minio/internal/logger"
"github.com/minio/pkg/env"
xnet "github.com/minio/pkg/net"
)
// ErrTargetsOffline - Indicates single/multiple target failures.
var ErrTargetsOffline = errors.New("one or more targets are offline. Please use `mc admin info --json` to check the offline targets")
// TestSubSysLambdaTargets - tests notification targets of given subsystem
func TestSubSysLambdaTargets(ctx context.Context, cfg config.Config, subSys string, transport *http.Transport) error {
if err := checkValidLambdaKeysForSubSys(subSys, cfg[subSys]); err != nil {
return err
}
targetList, err := fetchSubSysTargets(ctx, cfg, subSys, transport)
if err != nil {
return err
}
for _, target := range targetList {
defer target.Close()
}
for _, target := range targetList {
yes, err := target.IsActive()
if err == nil && !yes {
err = ErrTargetsOffline
}
if err != nil {
return fmt.Errorf("error (%s): %w", target.ID(), err)
}
}
return nil
}
func fetchSubSysTargets(ctx context.Context, cfg config.Config, subSys string, transport *http.Transport) (targets []event.Target, err error) {
if err := checkValidLambdaKeysForSubSys(subSys, cfg[subSys]); err != nil {
return nil, err
}
if subSys == config.LambdaWebhookSubSys {
webhookTargets, err := GetLambdaWebhook(cfg[config.LambdaWebhookSubSys], transport)
if err != nil {
return nil, err
}
for id, args := range webhookTargets {
if !args.Enable {
continue
}
t, err := target.NewWebhookTarget(ctx, id, args, logger.LogOnceIf, transport)
if err != nil {
return nil, err
}
targets = append(targets, t)
}
}
return targets, nil
}
// FetchEnabledTargets - Returns a set of configured TargetList
func FetchEnabledTargets(ctx context.Context, cfg config.Config, transport *http.Transport) (*event.TargetList, error) {
targetList := event.NewTargetList()
for _, subSys := range config.LambdaSubSystems.ToSlice() {
targets, err := fetchSubSysTargets(ctx, cfg, subSys, transport)
if err != nil {
return nil, err
}
for _, t := range targets {
if err = targetList.Add(t); err != nil {
return nil, err
}
}
}
return targetList, nil
}
// DefaultLambdaKVS - default notification list of kvs.
var (
DefaultLambdaKVS = map[string]config.KVS{
config.LambdaWebhookSubSys: DefaultWebhookKVS,
}
)
// DefaultWebhookKVS - default KV for webhook config
var (
DefaultWebhookKVS = config.KVS{
config.KV{
Key: config.Enable,
Value: config.EnableOff,
},
config.KV{
Key: target.WebhookEndpoint,
Value: "",
},
config.KV{
Key: target.WebhookAuthToken,
Value: "",
},
config.KV{
Key: target.WebhookClientCert,
Value: "",
},
config.KV{
Key: target.WebhookClientKey,
Value: "",
},
}
)
func checkValidLambdaKeysForSubSys(subSys string, tgt map[string]config.KVS) error {
validKVS, ok := DefaultLambdaKVS[subSys]
if !ok {
return nil
}
for tname, kv := range tgt {
subSysTarget := subSys
if tname != config.Default {
subSysTarget = subSys + config.SubSystemSeparator + tname
}
if v, ok := kv.Lookup(config.Enable); ok && v == config.EnableOn {
if err := config.CheckValidKeys(subSysTarget, kv, validKVS); err != nil {
return err
}
}
}
return nil
}
// GetLambdaWebhook - returns a map of registered notification 'webhook' targets
func GetLambdaWebhook(webhookKVS map[string]config.KVS, transport *http.Transport) (
map[string]target.WebhookArgs, error,
) {
webhookTargets := make(map[string]target.WebhookArgs)
for k, kv := range config.Merge(webhookKVS, target.EnvWebhookEnable, DefaultWebhookKVS) {
enableEnv := target.EnvWebhookEnable
if k != config.Default {
enableEnv = enableEnv + config.Default + k
}
enabled, err := config.ParseBool(env.Get(enableEnv, kv.Get(config.Enable)))
if err != nil {
return nil, err
}
if !enabled {
continue
}
urlEnv := target.EnvWebhookEndpoint
if k != config.Default {
urlEnv = urlEnv + config.Default + k
}
url, err := xnet.ParseHTTPURL(env.Get(urlEnv, kv.Get(target.WebhookEndpoint)))
if err != nil {
return nil, err
}
authEnv := target.EnvWebhookAuthToken
if k != config.Default {
authEnv = authEnv + config.Default + k
}
clientCertEnv := target.EnvWebhookClientCert
if k != config.Default {
clientCertEnv = clientCertEnv + config.Default + k
}
clientKeyEnv := target.EnvWebhookClientKey
if k != config.Default {
clientKeyEnv = clientKeyEnv + config.Default + k
}
webhookArgs := target.WebhookArgs{
Enable: enabled,
Endpoint: *url,
Transport: transport,
AuthToken: env.Get(authEnv, kv.Get(target.WebhookAuthToken)),
ClientCert: env.Get(clientCertEnv, kv.Get(target.WebhookClientCert)),
ClientKey: env.Get(clientKeyEnv, kv.Get(target.WebhookClientKey)),
}
if err = webhookArgs.Validate(); err != nil {
return nil, err
}
webhookTargets[k] = webhookArgs
}
return webhookTargets, nil
}

51
internal/config/lambda/target/lazyinit.go Normal file
View file

@ -0,0 +1,51 @@
// Copyright (c) 2015-2023 MinIO, Inc.
//
// This file is part of MinIO Object Storage stack
//
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// This program is distributed in the hope that it will be useful
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU Affero General Public License for more details.
//
// You should have received a copy of the GNU Affero General Public License
// along with this program. If not, see <http://www.gnu.org/licenses/>.
package target
import (
"sync"
"sync/atomic"
)
// Inspired from Golang sync.Once but it is only marked
// initialized when the provided function returns nil.
type lazyInit struct {
done uint32
m sync.Mutex
}
func (l *lazyInit) Do(f func() error) error {
if atomic.LoadUint32(&l.done) == 0 {
return l.doSlow(f)
}
return nil
}
func (l *lazyInit) doSlow(f func() error) error {
l.m.Lock()
defer l.m.Unlock()
if atomic.LoadUint32(&l.done) == 0 {
if err := f(); err != nil {
return err
}
// Mark as done only when f() is successful
atomic.StoreUint32(&l.done, 1)
}
return nil
}

247
internal/config/lambda/target/webhook.go Normal file
View file

@ -0,0 +1,247 @@
// Copyright (c) 2015-2023 MinIO, Inc.
//
// This file is part of MinIO Object Storage stack
//
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// This program is distributed in the hope that it will be useful
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU Affero General Public License for more details.
//
// You should have received a copy of the GNU Affero General Public License
// along with this program. If not, see <http://www.gnu.org/licenses/>.
package target
import (
"bytes"
"context"
"crypto/tls"
"encoding/json"
"errors"
"io"
"net/http"
"strings"
"sync/atomic"
"syscall"
"time"
"github.com/minio/minio/internal/config/lambda/event"
"github.com/minio/minio/internal/logger"
"github.com/minio/pkg/certs"
xnet "github.com/minio/pkg/net"
)
// Webhook constants
const (
WebhookEndpoint = "endpoint"
WebhookAuthToken = "auth_token"
WebhookClientCert = "client_cert"
WebhookClientKey = "client_key"
EnvWebhookEnable = "MINIO_LAMBDA_WEBHOOK_ENABLE"
EnvWebhookEndpoint = "MINIO_LAMBDA_WEBHOOK_ENDPOINT"
EnvWebhookAuthToken = "MINIO_LAMBDA_WEBHOOK_AUTH_TOKEN"
EnvWebhookClientCert = "MINIO_LAMBDA_WEBHOOK_CLIENT_CERT"
EnvWebhookClientKey = "MINIO_LAMBDA_WEBHOOK_CLIENT_KEY"
)
// WebhookArgs - Webhook target arguments.
type WebhookArgs struct {
Enable bool `json:"enable"`
Endpoint xnet.URL `json:"endpoint"`
AuthToken string `json:"authToken"`
Transport *http.Transport `json:"-"`
ClientCert string `json:"clientCert"`
ClientKey string `json:"clientKey"`
}
// Validate WebhookArgs fields
func (w WebhookArgs) Validate() error {
if !w.Enable {
return nil
}
if w.Endpoint.IsEmpty() {
return errors.New("endpoint empty")
}
if w.ClientCert != "" && w.ClientKey == "" || w.ClientCert == "" && w.ClientKey != "" {
return errors.New("cert and key must be specified as a pair")
}
return nil
}
// WebhookTarget - Webhook target.
type WebhookTarget struct {
activeRequests int64
totalRequests int64
failedRequests int64
lazyInit lazyInit
id event.TargetID
args WebhookArgs
transport *http.Transport
httpClient *http.Client
loggerOnce logger.LogOnce
cancel context.CancelFunc
cancelCh <-chan struct{}
}
// ID - returns target ID.
func (target *WebhookTarget) ID() event.TargetID {
return target.id
}
// IsActive - Return true if target is up and active
func (target *WebhookTarget) IsActive() (bool, error) {
if err := target.init(); err != nil {
return false, err
}
return target.isActive()
}
// errNotConnected - indicates that the target connection is not active.
var errNotConnected = errors.New("not connected to target server/service")
func (target *WebhookTarget) isActive() (bool, error) {
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
req, err := http.NewRequestWithContext(ctx, http.MethodHead, target.args.Endpoint.String(), nil)
if err != nil {
if xnet.IsNetworkOrHostDown(err, false) {
return false, errNotConnected
}
return false, err
}
tokens := strings.Fields(target.args.AuthToken)
switch len(tokens) {
case 2:
req.Header.Set("Authorization", target.args.AuthToken)
case 1:
req.Header.Set("Authorization", "Bearer "+target.args.AuthToken)
}
resp, err := target.httpClient.Do(req)
if err != nil {
if xnet.IsNetworkOrHostDown(err, true) {
return false, errNotConnected
}
return false, err
}
io.Copy(io.Discard, resp.Body)
resp.Body.Close()
// No network failure i.e response from the target means its up
return true, nil
}
// Stat - returns lamdba webhook target statistics such as
// current calls in progress, successfully completed functions
// failed functions.
func (target *WebhookTarget) Stat() event.TargetStat {
return event.TargetStat{
ID: target.id,
ActiveRequests: atomic.LoadInt64(&target.activeRequests),
TotalRequests: atomic.LoadInt64(&target.totalRequests),
FailedRequests: atomic.LoadInt64(&target.failedRequests),
}
}
// Send - sends an event to the webhook.
func (target *WebhookTarget) Send(eventData event.Event) (resp *http.Response, err error) {
atomic.AddInt64(&target.activeRequests, 1)
defer atomic.AddInt64(&target.activeRequests, -1)
atomic.AddInt64(&target.totalRequests, 1)
defer func() {
if err != nil {
atomic.AddInt64(&target.failedRequests, 1)
}
}()
if err = target.init(); err != nil {
return nil, err
}
data, err := json.Marshal(eventData)
if err != nil {
return nil, err
}
req, err := http.NewRequest(http.MethodPost, target.args.Endpoint.String(), bytes.NewReader(data))
if err != nil {
return nil, err
}
// Verify if the authToken already contains
// <Key> <Token> like format, if this is
// already present we can blindly use the
// authToken as is instead of adding 'Bearer'
tokens := strings.Fields(target.args.AuthToken)
switch len(tokens) {
case 2:
req.Header.Set("Authorization", target.args.AuthToken)
case 1:
req.Header.Set("Authorization", "Bearer "+target.args.AuthToken)
}
req.Header.Set("Content-Type", "application/json")
return target.httpClient.Do(req)
}
// Close the target. Will cancel all active requests.
func (target *WebhookTarget) Close() error {
target.cancel()
return nil
}
func (target *WebhookTarget) init() error {
return target.lazyInit.Do(target.initWebhook)
}
// Only called from init()
func (target *WebhookTarget) initWebhook() error {
args := target.args
transport := target.transport
if args.ClientCert != "" && args.ClientKey != "" {
manager, err := certs.NewManager(context.Background(), args.ClientCert, args.ClientKey, tls.LoadX509KeyPair)
if err != nil {
return err
}
manager.ReloadOnSignal(syscall.SIGHUP) // allow reloads upon SIGHUP
transport.TLSClientConfig.GetClientCertificate = manager.GetClientCertificate
}
target.httpClient = &http.Client{Transport: transport}
yes, err := target.isActive()
if err != nil {
return err
}
if !yes {
return errNotConnected
}
return nil
}
// NewWebhookTarget - creates new Webhook target.
func NewWebhookTarget(ctx context.Context, id string, args WebhookArgs, loggerOnce logger.LogOnce, transport *http.Transport) (*WebhookTarget, error) {
ctx, cancel := context.WithCancel(ctx)
target := &WebhookTarget{
id: event.TargetID{ID: id, Name: "webhook"},
args: args,
loggerOnce: loggerOnce,
transport: transport,
cancel: cancel,
cancelCh: ctx.Done(),
}
return target, nil
}

7
internal/event/targetlist.go
View file

@ -148,7 +148,12 @@ func (list *TargetList) List() []TargetID {
func (list *TargetList) TargetMap() map[TargetID]Target {
list.RLock()
defer list.RUnlock()
return list.targets
ntargets := make(map[TargetID]Target, len(list.targets))
for k, v := range list.targets {
ntargets[k] = v
}
return ntargets
}
// Send - sends events to targets identified by target IDs.

57
internal/http/lambda-headers.go Normal file
View file

@ -0,0 +1,57 @@
// Copyright (c) 2015-2023 MinIO, Inc.
//
// This file is part of MinIO Object Storage stack
//
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// This program is distributed in the hope that it will be useful
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU Affero General Public License for more details.
//
// You should have received a copy of the GNU Affero General Public License
// along with this program. If not, see <http://www.gnu.org/licenses/>.
package http
// Object Lambda headers
const (
AmzRequestRoute = "x-amz-request-route"
AmzRequestToken = "x-amz-request-token"
AmzFwdStatus = "x-amz-fwd-status"
AmzFwdErrorCode = "x-amz-fwd-error-code"
AmzFwdErrorMessage = "x-amz-fwd-error-message"
AmzFwdHeaderAcceptRanges = "x-amz-fwd-header-accept-ranges"
AmzFwdHeaderCacheControl = "x-amz-fwd-header-Cache-Control"
AmzFwdHeaderContentDisposition = "x-amz-fwd-header-Content-Disposition"
AmzFwdHeaderContentEncoding = "x-amz-fwd-header-Content-Encoding"
AmzFwdHeaderContentLanguage = "x-amz-fwd-header-Content-Language"
AmzFwdHeaderContentRange = "x-amz-fwd-header-Content-Range"
AmzFwdHeaderContentType = "x-amz-fwd-header-Content-Type"
AmzFwdHeaderChecksumCrc32 = "x-amz-fwd-header-x-amz-checksum-crc32"
AmzFwdHeaderChecksumCrc32c = "x-amz-fwd-header-x-amz-checksum-crc32c"
AmzFwdHeaderChecksumSha1 = "x-amz-fwd-header-x-amz-checksum-sha1"
AmzFwdHeaderChecksumSha256 = "x-amz-fwd-header-x-amz-checksum-sha256"
AmzFwdHeaderDeleteMarker = "x-amz-fwd-header-x-amz-delete-marker"
AmzFwdHeaderETag = "x-amz-fwd-header-ETag"
AmzFwdHeaderExpires = "x-amz-fwd-header-Expires"
AmzFwdHeaderExpiration = "x-amz-fwd-header-x-amz-expiration"
AmzFwdHeaderLastModified = "x-amz-fwd-header-Last-Modified"
AmzFwdHeaderObjectLockMode = "x-amz-fwd-header-x-amz-object-lock-mode"
AmzFwdHeaderObjectLockLegalHold = "x-amz-fwd-header-x-amz-object-lock-legal-hold"
AmzFwdHeaderObjectLockRetainUntil = "x-amz-fwd-header-x-amz-object-lock-retain-until-date"
AmzFwdHeaderMPPartsCount = "x-amz-fwd-header-x-amz-mp-parts-count"
AmzFwdHeaderReplicationStatus = "x-amz-fwd-header-x-amz-replication-status"
AmzFwdHeaderSSE = "x-amz-fwd-header-x-amz-server-side-encryption"
AmzFwdHeaderSSEC = "x-amz-fwd-header-x-amz-server-side-encryption-customer-algorithm"
AmzFwdHeaderSSEKMSID = "x-amz-fwd-header-x-amz-server-side-encryption-aws-kms-key-id"
AmzFwdHeaderSSECMD5 = "x-amz-fwd-header-x-amz-server-side-encryption-customer-key-MD5"
AmzFwdHeaderStorageClass = "x-amz-fwd-header-x-amz-storage-class"
AmzFwdHeaderTaggingCount = "x-amz-fwd-header-x-amz-tagging-count"
AmzFwdHeaderVersionID = "x-amz-fwd-header-x-amz-version-id"
)