OP#792 initial commit after fork

internal/logger/target/console/console.go

@@ -0,0 +1,160 @@
+// Copyright (c) 2015-2021 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 console
+
+import (
+	"encoding/json"
+	"fmt"
+	"io"
+	"strconv"
+	"strings"
+
+	"github.com/minio/madmin-go/v3/logger/log"
+	"github.com/minio/minio/internal/color"
+	"github.com/minio/minio/internal/logger"
+)
+
+// Target implements loggerTarget to send log
+// in plain or json format to the standard output.
+type Target struct {
+	output io.Writer
+}
+
+// Validate - validate if the tty can be written to
+func (c *Target) Validate() error {
+	return nil
+}
+
+// Endpoint returns the backend endpoint
+func (c *Target) Endpoint() string {
+	return ""
+}
+
+func (c *Target) String() string {
+	return "console"
+}
+
+// Send log message 'e' to console
+func (c *Target) Send(e any) error {
+	entry, ok := e.(log.Entry)
+	if !ok {
+		return fmt.Errorf("Uexpected log entry structure %#v", e)
+	}
+	if logger.IsJSON() {
+		logJSON, err := json.Marshal(&entry)
+		if err != nil {
+			return err
+		}
+		fmt.Fprintln(c.output, string(logJSON))
+		return nil
+	}
+
+	if entry.Level == logger.EventKind {
+		fmt.Fprintln(c.output, entry.Message)
+		return nil
+	}
+
+	traceLength := len(entry.Trace.Source)
+	trace := make([]string, traceLength)
+
+	// Add a sequence number and formatting for each stack trace
+	// No formatting is required for the first entry
+	for i, element := range entry.Trace.Source {
+		trace[i] = fmt.Sprintf("%8v: %s", traceLength-i, element)
+	}
+
+	tagString := ""
+	for key, value := range entry.Trace.Variables {
+		if value != "" {
+			if tagString != "" {
+				tagString += ", "
+			}
+			tagString += fmt.Sprintf("%s=%#v", key, value)
+		}
+	}
+
+	var apiString string
+	if entry.API != nil {
+		apiString = "API: " + entry.API.Name
+		if entry.API.Args != nil {
+			args := ""
+			if entry.API.Args.Bucket != "" {
+				args = args + "bucket=" + entry.API.Args.Bucket
+			}
+			if entry.API.Args.Object != "" {
+				args = args + ", object=" + entry.API.Args.Object
+			}
+			if entry.API.Args.VersionID != "" {
+				args = args + ", versionId=" + entry.API.Args.VersionID
+			}
+			if len(entry.API.Args.Objects) > 0 {
+				args = args + ", multiObject=true, numberOfObjects=" + strconv.Itoa(len(entry.API.Args.Objects))
+			}
+			if len(args) > 0 {
+				apiString += "(" + args + ")"
+			}
+		}
+	} else {
+		apiString = "INTERNAL"
+	}
+	timeString := "Time: " + entry.Time.Format(logger.TimeFormat)
+
+	var deploymentID string
+	if entry.DeploymentID != "" {
+		deploymentID = "\nDeploymentID: " + entry.DeploymentID
+	}
+
+	var requestID string
+	if entry.RequestID != "" {
+		requestID = "\nRequestID: " + entry.RequestID
+	}
+
+	var remoteHost string
+	if entry.RemoteHost != "" {
+		remoteHost = "\nRemoteHost: " + entry.RemoteHost
+	}
+
+	var host string
+	if entry.Host != "" {
+		host = "\nHost: " + entry.Host
+	}
+
+	var userAgent string
+	if entry.UserAgent != "" {
+		userAgent = "\nUserAgent: " + entry.UserAgent
+	}
+
+	if len(entry.Trace.Variables) > 0 {
+		tagString = "\n       " + tagString
+	}
+
+	msg := color.RedBold(entry.Trace.Message)
+	output := fmt.Sprintf("\n%s\n%s%s%s%s%s%s\nError: %s%s\n%s",
+		apiString, timeString, deploymentID, requestID, remoteHost, host, userAgent,
+		msg, tagString, strings.Join(trace, "\n"))
+
+	fmt.Fprintln(c.output, output)
+	return nil
+}
+
+// New initializes a new logger target
+// which prints log directly in the standard
+// output.
+func New(w io.Writer) *Target {
+	return &Target{output: w}
+}

internal/logger/target/http/http.go

@@ -0,0 +1,633 @@
+// Copyright (c) 2015-2024 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
+
+import (
+	"bytes"
+	"context"
+	"errors"
+	"fmt"
+	"net/http"
+	"net/url"
+	"os"
+	"path/filepath"
+	"strconv"
+	"strings"
+	"sync"
+	"sync/atomic"
+	"time"
+
+	jsoniter "github.com/json-iterator/go"
+	xhttp "github.com/minio/minio/internal/http"
+	xioutil "github.com/minio/minio/internal/ioutil"
+	types "github.com/minio/minio/internal/logger/target/loggertypes"
+	"github.com/minio/minio/internal/once"
+	"github.com/minio/minio/internal/store"
+	xnet "github.com/minio/pkg/v3/net"
+	"github.com/valyala/bytebufferpool"
+)
+
+const (
+
+	// maxWorkers is the maximum number of concurrent http loggers
+	maxWorkers = 16
+
+	// maxWorkers is the maximum number of concurrent batch http loggers
+	maxWorkersWithBatchEvents = 4
+
+	// the suffix for the configured queue dir where the logs will be persisted.
+	httpLoggerExtension = ".http.log"
+)
+
+const (
+	statusOffline = iota
+	statusOnline
+	statusClosed
+)
+
+var (
+	logChBuffers = make(map[string]chan any)
+	logChLock    = sync.Mutex{}
+)
+
+// Config http logger target
+type Config struct {
+	Enabled     bool              `json:"enabled"`
+	Name        string            `json:"name"`
+	UserAgent   string            `json:"userAgent"`
+	Endpoint    *xnet.URL         `json:"endpoint"`
+	AuthToken   string            `json:"authToken"`
+	ClientCert  string            `json:"clientCert"`
+	ClientKey   string            `json:"clientKey"`
+	BatchSize   int               `json:"batchSize"`
+	QueueSize   int               `json:"queueSize"`
+	QueueDir    string            `json:"queueDir"`
+	MaxRetry    int               `json:"maxRetry"`
+	RetryIntvl  time.Duration     `json:"retryInterval"`
+	Proxy       string            `json:"string"`
+	Transport   http.RoundTripper `json:"-"`
+	HTTPTimeout time.Duration     `json:"httpTimeout"`
+
+	// Custom logger
+	LogOnceIf func(ctx context.Context, err error, id string, errKind ...any) `json:"-"`
+}
+
+// Target implements logger.Target and sends the json
+// format of a log entry to the configured http endpoint.
+// An internal buffer of logs is maintained but when the
+// buffer is full, new logs are just ignored and an error
+// is returned to the caller.
+type Target struct {
+	totalMessages  atomic.Int64
+	failedMessages atomic.Int64
+	status         atomic.Int32
+
+	// Worker control
+	workers    atomic.Int64
+	maxWorkers int64
+
+	// workerStartMu sync.Mutex
+	lastStarted time.Time
+
+	wg sync.WaitGroup
+
+	// Channel of log entries.
+	// Reading logCh must hold read lock on logChMu (to avoid read race)
+	// Sending a value on logCh must hold read lock on logChMu (to avoid closing)
+	logCh   chan any
+	logChMu sync.RWMutex
+
+	// If this webhook is being re-configured we will
+	// assign the new webhook target to this field.
+	// The Send() method will then re-direct entries
+	// to the new target when the current one
+	// has been set to status "statusClosed".
+	// Once the glogal target slice has been migrated
+	// the current target will stop receiving entries.
+	migrateTarget *Target
+
+	// Number of events per HTTP send to webhook target
+	// this is ideally useful only if your endpoint can
+	// support reading multiple events on a stream for example
+	// like : Splunk HTTP Event collector, if you are unsure
+	// set this to '1'.
+	batchSize   int
+	payloadType string
+
+	// store to persist and replay the logs to the target
+	// to avoid missing events when the target is down.
+	store          store.Store[any]
+	storeCtxCancel context.CancelFunc
+
+	initQueueOnce once.Init
+
+	config      Config
+	client      *http.Client
+	httpTimeout time.Duration
+}
+
+// Name returns the name of the target
+func (h *Target) Name() string {
+	return "minio-http-" + h.config.Name
+}
+
+// Type - returns type of the target
+func (h *Target) Type() types.TargetType {
+	return types.TargetHTTP
+}
+
+// Endpoint returns the backend endpoint
+func (h *Target) Endpoint() string {
+	return h.config.Endpoint.String()
+}
+
+func (h *Target) String() string {
+	return h.config.Name
+}
+
+// IsOnline returns true if the target is reachable using a cached value
+func (h *Target) IsOnline(ctx context.Context) bool {
+	return h.status.Load() == statusOnline
+}
+
+// Stats returns the target statistics.
+func (h *Target) Stats() types.TargetStats {
+	h.logChMu.RLock()
+	queueLength := len(h.logCh)
+	h.logChMu.RUnlock()
+	stats := types.TargetStats{
+		TotalMessages:  h.totalMessages.Load(),
+		FailedMessages: h.failedMessages.Load(),
+		QueueLength:    queueLength,
+	}
+
+	return stats
+}
+
+// AssignMigrateTarget assigns a target
+// which will eventually replace the current target.
+func (h *Target) AssignMigrateTarget(migrateTgt *Target) {
+	h.migrateTarget = migrateTgt
+}
+
+// Init validate and initialize the http target
+func (h *Target) Init(ctx context.Context) (err error) {
+	if h.config.QueueDir != "" {
+		return h.initQueueOnce.DoWithContext(ctx, h.initDiskStore)
+	}
+	return h.initQueueOnce.DoWithContext(ctx, h.initMemoryStore)
+}
+
+func (h *Target) initDiskStore(ctx context.Context) (err error) {
+	ctx, cancel := context.WithCancel(ctx)
+	h.storeCtxCancel = cancel
+	h.lastStarted = time.Now()
+	go h.startQueueProcessor(ctx, true)
+
+	queueStore := store.NewQueueStore[any](
+		filepath.Join(h.config.QueueDir, h.Name()),
+		uint64(h.config.QueueSize),
+		httpLoggerExtension,
+	)
+
+	if err := queueStore.Open(); err != nil {
+		return fmt.Errorf("unable to initialize the queue store of %s webhook: %w", h.Name(), err)
+	}
+
+	h.store = queueStore
+	store.StreamItems(h.store, h, ctx.Done(), h.config.LogOnceIf)
+
+	return nil
+}
+
+func (h *Target) initMemoryStore(ctx context.Context) (err error) {
+	ctx, cancel := context.WithCancel(ctx)
+	h.storeCtxCancel = cancel
+	h.lastStarted = time.Now()
+	go h.startQueueProcessor(ctx, true)
+	return nil
+}
+
+func (h *Target) send(ctx context.Context, payload []byte, payloadCount int, payloadType string, timeout time.Duration) (err error) {
+	defer func() {
+		if err != nil {
+			if xnet.IsNetworkOrHostDown(err, false) {
+				h.status.Store(statusOffline)
+			}
+			h.failedMessages.Add(int64(payloadCount))
+		} else {
+			h.status.Store(statusOnline)
+		}
+	}()
+
+	ctx, cancel := context.WithTimeout(ctx, timeout)
+	defer cancel()
+
+	req, err := http.NewRequestWithContext(ctx, http.MethodPost,
+		h.Endpoint(), bytes.NewReader(payload))
+	if err != nil {
+		return fmt.Errorf("invalid configuration for '%s'; %v", h.Endpoint(), err)
+	}
+	if payloadType != "" {
+		req.Header.Set(xhttp.ContentType, payloadType)
+	}
+	req.Header.Set(xhttp.WebhookEventPayloadCount, strconv.Itoa(payloadCount))
+	req.Header.Set(xhttp.MinIOVersion, xhttp.GlobalMinIOVersion)
+	req.Header.Set(xhttp.MinioDeploymentID, xhttp.GlobalDeploymentID)
+
+	// Set user-agent to indicate MinIO release
+	// version to the configured log endpoint
+	req.Header.Set("User-Agent", h.config.UserAgent)
+
+	if h.config.AuthToken != "" {
+		req.Header.Set("Authorization", h.config.AuthToken)
+	}
+
+	resp, err := h.client.Do(req)
+	if err != nil {
+		return fmt.Errorf("%s returned '%w', please check your endpoint configuration", h.Endpoint(), err)
+	}
+
+	// Drain any response.
+	xhttp.DrainBody(resp.Body)
+
+	if resp.StatusCode >= 200 && resp.StatusCode <= 299 {
+		// accepted HTTP status codes.
+		return nil
+	} else if resp.StatusCode == http.StatusForbidden {
+		return fmt.Errorf("%s returned '%s', please check if your auth token is correctly set", h.Endpoint(), resp.Status)
+	}
+	return fmt.Errorf("%s returned '%s', please check your endpoint configuration", h.Endpoint(), resp.Status)
+}
+
+func (h *Target) startQueueProcessor(ctx context.Context, mainWorker bool) {
+	h.logChMu.RLock()
+	if h.logCh == nil {
+		h.logChMu.RUnlock()
+		return
+	}
+	h.logChMu.RUnlock()
+
+	h.workers.Add(1)
+	defer h.workers.Add(-1)
+
+	h.wg.Add(1)
+	defer h.wg.Done()
+
+	entries := make([]any, 0)
+	name := h.Name()
+
+	defer func() {
+		// re-load the global buffer pointer
+		// in case it was modified by a new target.
+		logChLock.Lock()
+		currentGlobalBuffer, ok := logChBuffers[name]
+		logChLock.Unlock()
+		if !ok {
+			return
+		}
+
+		for _, v := range entries {
+			select {
+			case currentGlobalBuffer <- v:
+			default:
+			}
+		}
+
+		if mainWorker {
+		drain:
+			for {
+				select {
+				case v, ok := <-h.logCh:
+					if !ok {
+						break drain
+					}
+
+					currentGlobalBuffer <- v
+				default:
+					break drain
+				}
+			}
+		}
+	}()
+
+	lastBatchProcess := time.Now()
+
+	buf := bytebufferpool.Get()
+	enc := jsoniter.ConfigCompatibleWithStandardLibrary.NewEncoder(buf)
+	defer bytebufferpool.Put(buf)
+
+	isDirQueue := h.config.QueueDir != ""
+
+	// globalBuffer is always created or adjusted
+	// before this method is launched.
+	logChLock.Lock()
+	globalBuffer := logChBuffers[name]
+	logChLock.Unlock()
+
+	ticker := time.NewTicker(time.Second)
+	defer ticker.Stop()
+
+	var count int
+	for {
+		var (
+			ok    bool
+			entry any
+		)
+
+		if count < h.batchSize {
+			tickered := false
+			select {
+			case <-ticker.C:
+				tickered = true
+			case entry = <-globalBuffer:
+			case entry, ok = <-h.logCh:
+				if !ok {
+					return
+				}
+			case <-ctx.Done():
+				return
+			}
+
+			if !tickered {
+				h.totalMessages.Add(1)
+				if !isDirQueue {
+					if err := enc.Encode(&entry); err != nil {
+						h.config.LogOnceIf(
+							ctx,
+							fmt.Errorf("unable to encode webhook log entry, err  '%w' entry: %v\n", err, entry),
+							h.Name(),
+						)
+						h.failedMessages.Add(1)
+						continue
+					}
+				} else {
+					entries = append(entries, entry)
+				}
+				count++
+			}
+
+			if len(h.logCh) > 0 || len(globalBuffer) > 0 || count == 0 {
+				// there is something in the log queue
+				// process it first, even if we tickered
+				// first, or we have not received any events
+				// yet, still wait on it.
+				continue
+			}
+
+			// If we are doing batching, we should wait
+			// at least for a second, before sending.
+			// Even if there is nothing in the queue.
+			if h.batchSize > 1 && time.Since(lastBatchProcess) < time.Second {
+				continue
+			}
+		}
+
+		// if we have reached the count send at once
+		// or we have crossed last second before batch was sent, send at once
+		lastBatchProcess = time.Now()
+
+		var retries int
+		retryIntvl := h.config.RetryIntvl
+		if retryIntvl <= 0 {
+			retryIntvl = 3 * time.Second
+		}
+
+		maxRetries := h.config.MaxRetry
+
+	retry:
+		// If the channel reaches above half capacity
+		// we spawn more workers. The workers spawned
+		// from this main worker routine will exit
+		// once the channel drops below half capacity
+		// and when it's been at least 30 seconds since
+		// we launched a new worker.
+		if mainWorker && len(h.logCh) > cap(h.logCh)/2 {
+			nWorkers := h.workers.Load()
+			if nWorkers < h.maxWorkers {
+				if time.Since(h.lastStarted).Milliseconds() > 10 {
+					h.lastStarted = time.Now()
+					go h.startQueueProcessor(ctx, false)
+				}
+			}
+		}
+
+		var err error
+		if !isDirQueue {
+			err = h.send(ctx, buf.Bytes(), count, h.payloadType, h.httpTimeout)
+		} else {
+			_, err = h.store.PutMultiple(entries)
+		}
+
+		if err != nil {
+			if errors.Is(err, context.Canceled) {
+				return
+			}
+
+			h.config.LogOnceIf(
+				context.Background(),
+				fmt.Errorf("unable to send audit/log entry(s) to '%s' err '%w': %d", name, err, count),
+				name,
+			)
+
+			time.Sleep(retryIntvl)
+			if maxRetries == 0 {
+				goto retry
+			}
+			retries++
+			if retries <= maxRetries {
+				goto retry
+			}
+		}
+
+		entries = make([]any, 0)
+		count = 0
+		if !isDirQueue {
+			buf.Reset()
+		}
+
+		if !mainWorker && len(h.logCh) < cap(h.logCh)/2 {
+			if time.Since(h.lastStarted).Seconds() > 30 {
+				return
+			}
+		}
+	}
+}
+
+// CreateOrAdjustGlobalBuffer will create or adjust the global log entry buffers
+// which are used to migrate log entries between old and new targets.
+func CreateOrAdjustGlobalBuffer(currentTgt *Target, newTgt *Target) {
+	logChLock.Lock()
+	defer logChLock.Unlock()
+
+	requiredCap := currentTgt.config.QueueSize + (currentTgt.config.BatchSize * int(currentTgt.maxWorkers))
+	currentCap := 0
+	name := newTgt.Name()
+
+	currentBuff, ok := logChBuffers[name]
+	if !ok {
+		logChBuffers[name] = make(chan any, requiredCap)
+		currentCap = requiredCap
+	} else {
+		currentCap = cap(currentBuff)
+		requiredCap += len(currentBuff)
+	}
+
+	if requiredCap > currentCap {
+		logChBuffers[name] = make(chan any, requiredCap)
+
+		if len(currentBuff) > 0 {
+		drain:
+			for {
+				select {
+				case v, ok := <-currentBuff:
+					if !ok {
+						break drain
+					}
+					logChBuffers[newTgt.Name()] <- v
+				default:
+					break drain
+				}
+			}
+		}
+	}
+}
+
+// New initializes a new logger target which
+// sends log over http to the specified endpoint
+func New(config Config) (*Target, error) {
+	maxWorkers := maxWorkers
+	if config.BatchSize > 100 {
+		maxWorkers = maxWorkersWithBatchEvents
+	} else if config.BatchSize <= 0 {
+		config.BatchSize = 1
+	}
+
+	h := &Target{
+		logCh:       make(chan any, config.QueueSize),
+		config:      config,
+		batchSize:   config.BatchSize,
+		maxWorkers:  int64(maxWorkers),
+		httpTimeout: config.HTTPTimeout,
+	}
+	h.status.Store(statusOffline)
+
+	if config.BatchSize > 1 {
+		h.payloadType = ""
+	} else {
+		h.payloadType = "application/json"
+	}
+
+	// If proxy available, set the same
+	if h.config.Proxy != "" {
+		proxyURL, _ := url.Parse(h.config.Proxy)
+		transport := h.config.Transport
+		if tr, ok := transport.(*http.Transport); ok {
+			ctransport := tr.Clone()
+			ctransport.Proxy = http.ProxyURL(proxyURL)
+			h.config.Transport = ctransport
+		}
+	}
+
+	h.client = &http.Client{Transport: h.config.Transport}
+	return h, nil
+}
+
+// SendFromStore - reads the log from store and sends it to webhook.
+func (h *Target) SendFromStore(key store.Key) (err error) {
+	var eventData []byte
+	eventData, err = h.store.GetRaw(key)
+	if err != nil {
+		if os.IsNotExist(err) {
+			return nil
+		}
+		return err
+	}
+
+	count := 1
+	v := strings.Split(key.Name, ":")
+	if len(v) == 2 {
+		count, err = strconv.Atoi(v[0])
+		if err != nil {
+			return err
+		}
+	}
+
+	if err := h.send(context.Background(), eventData, count, h.payloadType, h.httpTimeout); err != nil {
+		return err
+	}
+
+	// Delete the event from store.
+	return h.store.Del(key)
+}
+
+// Send the log message 'entry' to the http target.
+// Messages are queued in the disk if the store is enabled
+// If Cancel has been called the message is ignored.
+func (h *Target) Send(ctx context.Context, entry any) error {
+	if h.status.Load() == statusClosed {
+		if h.migrateTarget != nil {
+			return h.migrateTarget.Send(ctx, entry)
+		}
+		return nil
+	}
+
+	h.logChMu.RLock()
+	defer h.logChMu.RUnlock()
+	if h.logCh == nil {
+		// We are closing...
+		return nil
+	}
+
+	select {
+	case h.logCh <- entry:
+		h.totalMessages.Add(1)
+	case <-ctx.Done():
+		// return error only for context timedout.
+		if errors.Is(ctx.Err(), context.DeadlineExceeded) {
+			return ctx.Err()
+		}
+		return nil
+	default:
+		h.totalMessages.Add(1)
+		h.failedMessages.Add(1)
+		return errors.New("log buffer full")
+	}
+
+	return nil
+}
+
+// Cancel - cancels the target.
+// All queued messages are flushed and the function returns afterwards.
+// All messages sent to the target after this function has been called will be dropped.
+func (h *Target) Cancel() {
+	h.status.Store(statusClosed)
+	h.storeCtxCancel()
+
+	// Wait for messages to be sent...
+	h.wg.Wait()
+
+	// Set logch to nil and close it.
+	// This will block all Send operations,
+	// and finish the existing ones.
+	// All future ones will be discarded.
+	h.logChMu.Lock()
+	xioutil.SafeClose(h.logCh)
+	h.logCh = nil
+	h.logChMu.Unlock()
+}

internal/logger/target/kafka/kafka.go

@@ -0,0 +1,404 @@
+// 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 kafka
+
+import (
+	"context"
+	"crypto/tls"
+	"crypto/x509"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"log"
+	"os"
+	"path/filepath"
+	"sync"
+	"sync/atomic"
+	"time"
+
+	"github.com/IBM/sarama"
+	saramatls "github.com/IBM/sarama/tools/tls"
+
+	xioutil "github.com/minio/minio/internal/ioutil"
+	types "github.com/minio/minio/internal/logger/target/loggertypes"
+	"github.com/minio/minio/internal/once"
+	"github.com/minio/minio/internal/store"
+	xnet "github.com/minio/pkg/v3/net"
+)
+
+// the suffix for the configured queue dir where the logs will be persisted.
+const kafkaLoggerExtension = ".kafka.log"
+
+const (
+	statusClosed = iota
+	statusOffline
+	statusOnline
+)
+
+// Config - kafka target arguments.
+type Config struct {
+	Enabled bool        `json:"enable"`
+	Brokers []xnet.Host `json:"brokers"`
+	Topic   string      `json:"topic"`
+	Version string      `json:"version"`
+	TLS     struct {
+		Enable        bool               `json:"enable"`
+		RootCAs       *x509.CertPool     `json:"-"`
+		SkipVerify    bool               `json:"skipVerify"`
+		ClientAuth    tls.ClientAuthType `json:"clientAuth"`
+		ClientTLSCert string             `json:"clientTLSCert"`
+		ClientTLSKey  string             `json:"clientTLSKey"`
+	} `json:"tls"`
+	SASL struct {
+		Enable    bool   `json:"enable"`
+		User      string `json:"username"`
+		Password  string `json:"password"`
+		Mechanism string `json:"mechanism"`
+	} `json:"sasl"`
+	// Queue store
+	QueueSize int    `json:"queueSize"`
+	QueueDir  string `json:"queueDir"`
+
+	// Custom logger
+	LogOnce func(ctx context.Context, err error, id string, errKind ...any) `json:"-"`
+}
+
+// Target - Kafka target.
+type Target struct {
+	status int32
+
+	totalMessages  int64
+	failedMessages int64
+
+	wg sync.WaitGroup
+
+	// Channel of log entries.
+	// Reading logCh must hold read lock on logChMu (to avoid read race)
+	// Sending a value on logCh must hold read lock on logChMu (to avoid closing)
+	logCh   chan any
+	logChMu sync.RWMutex
+
+	// store to persist and replay the logs to the target
+	// to avoid missing events when the target is down.
+	store          store.Store[any]
+	storeCtxCancel context.CancelFunc
+
+	initKafkaOnce      once.Init
+	initQueueStoreOnce once.Init
+
+	client   sarama.Client
+	producer sarama.SyncProducer
+	kconfig  Config
+	config   *sarama.Config
+}
+
+func (h *Target) validate() error {
+	if len(h.kconfig.Brokers) == 0 {
+		return errors.New("no broker address found")
+	}
+	for _, b := range h.kconfig.Brokers {
+		if _, err := xnet.ParseHost(b.String()); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+// Name returns the name of the target
+func (h *Target) Name() string {
+	return "minio-kafka-audit"
+}
+
+// Endpoint - return kafka target
+func (h *Target) Endpoint() string {
+	return "kafka"
+}
+
+// String - kafka string
+func (h *Target) String() string {
+	return "kafka"
+}
+
+// Stats returns the target statistics.
+func (h *Target) Stats() types.TargetStats {
+	h.logChMu.RLock()
+	queueLength := len(h.logCh)
+	h.logChMu.RUnlock()
+
+	return types.TargetStats{
+		TotalMessages:  atomic.LoadInt64(&h.totalMessages),
+		FailedMessages: atomic.LoadInt64(&h.failedMessages),
+		QueueLength:    queueLength,
+	}
+}
+
+// Init initialize kafka target
+func (h *Target) Init(ctx context.Context) error {
+	if !h.kconfig.Enabled {
+		return nil
+	}
+	if err := h.validate(); err != nil {
+		return err
+	}
+	if h.kconfig.QueueDir != "" {
+		if err := h.initQueueStoreOnce.DoWithContext(ctx, h.initQueueStore); err != nil {
+			return err
+		}
+		return h.initKafkaOnce.Do(h.init)
+	}
+	if err := h.init(); err != nil {
+		return err
+	}
+	go h.startKafkaLogger()
+	return nil
+}
+
+func (h *Target) initQueueStore(ctx context.Context) (err error) {
+	queueDir := filepath.Join(h.kconfig.QueueDir, h.Name())
+	queueStore := store.NewQueueStore[any](queueDir, uint64(h.kconfig.QueueSize), kafkaLoggerExtension)
+	if err = queueStore.Open(); err != nil {
+		return fmt.Errorf("unable to initialize the queue store of %s webhook: %w", h.Name(), err)
+	}
+	ctx, cancel := context.WithCancel(ctx)
+	h.store = queueStore
+	h.storeCtxCancel = cancel
+	store.StreamItems(h.store, h, ctx.Done(), h.kconfig.LogOnce)
+	return err
+}
+
+func (h *Target) startKafkaLogger() {
+	h.logChMu.RLock()
+	logCh := h.logCh
+	if logCh != nil {
+		// We are not allowed to add when logCh is nil
+		h.wg.Add(1)
+		defer h.wg.Done()
+	}
+	h.logChMu.RUnlock()
+
+	if logCh == nil {
+		return
+	}
+
+	// Create a routine which sends json logs received
+	// from an internal channel.
+	for entry := range logCh {
+		h.logEntry(entry)
+	}
+}
+
+func (h *Target) logEntry(entry any) {
+	atomic.AddInt64(&h.totalMessages, 1)
+	if err := h.send(entry); err != nil {
+		atomic.AddInt64(&h.failedMessages, 1)
+		h.kconfig.LogOnce(context.Background(), err, h.kconfig.Topic)
+	}
+}
+
+func (h *Target) send(entry any) error {
+	if err := h.initKafkaOnce.Do(h.init); err != nil {
+		return err
+	}
+	logJSON, err := json.Marshal(&entry)
+	if err != nil {
+		return err
+	}
+	msg := sarama.ProducerMessage{
+		Topic: h.kconfig.Topic,
+		Value: sarama.ByteEncoder(logJSON),
+	}
+	_, _, err = h.producer.SendMessage(&msg)
+	if err != nil {
+		atomic.StoreInt32(&h.status, statusOffline)
+	} else {
+		atomic.StoreInt32(&h.status, statusOnline)
+	}
+	return err
+}
+
+// Init initialize kafka target
+func (h *Target) init() error {
+	if os.Getenv("_MINIO_KAFKA_DEBUG") != "" {
+		sarama.DebugLogger = log.Default()
+	}
+
+	sconfig := sarama.NewConfig()
+	if h.kconfig.Version != "" {
+		kafkaVersion, err := sarama.ParseKafkaVersion(h.kconfig.Version)
+		if err != nil {
+			return err
+		}
+		sconfig.Version = kafkaVersion
+	}
+
+	sconfig.Net.KeepAlive = 60 * time.Second
+	sconfig.Net.SASL.User = h.kconfig.SASL.User
+	sconfig.Net.SASL.Password = h.kconfig.SASL.Password
+	initScramClient(h.kconfig, sconfig) // initializes configured scram client.
+	sconfig.Net.SASL.Enable = h.kconfig.SASL.Enable
+
+	tlsConfig, err := saramatls.NewConfig(h.kconfig.TLS.ClientTLSCert, h.kconfig.TLS.ClientTLSKey)
+	if err != nil {
+		return err
+	}
+
+	sconfig.Net.TLS.Enable = h.kconfig.TLS.Enable
+	sconfig.Net.TLS.Config = tlsConfig
+	sconfig.Net.TLS.Config.InsecureSkipVerify = h.kconfig.TLS.SkipVerify
+	sconfig.Net.TLS.Config.ClientAuth = h.kconfig.TLS.ClientAuth
+	sconfig.Net.TLS.Config.RootCAs = h.kconfig.TLS.RootCAs
+
+	// These settings are needed to ensure that kafka client doesn't hang on brokers
+	// refer https://github.com/IBM/sarama/issues/765#issuecomment-254333355
+	sconfig.Producer.Retry.Max = 2
+	sconfig.Producer.Retry.Backoff = (10 * time.Second)
+	sconfig.Producer.Return.Successes = true
+	sconfig.Producer.Return.Errors = true
+	sconfig.Producer.RequiredAcks = 1
+	sconfig.Producer.Timeout = (10 * time.Second)
+	sconfig.Net.ReadTimeout = (10 * time.Second)
+	sconfig.Net.DialTimeout = (10 * time.Second)
+	sconfig.Net.WriteTimeout = (10 * time.Second)
+	sconfig.Metadata.Retry.Max = 1
+	sconfig.Metadata.Retry.Backoff = (10 * time.Second)
+	sconfig.Metadata.RefreshFrequency = (15 * time.Minute)
+
+	h.config = sconfig
+
+	var brokers []string
+	for _, broker := range h.kconfig.Brokers {
+		brokers = append(brokers, broker.String())
+	}
+
+	client, err := sarama.NewClient(brokers, sconfig)
+	if err != nil {
+		return err
+	}
+
+	producer, err := sarama.NewSyncProducerFromClient(client)
+	if err != nil {
+		return err
+	}
+	h.client = client
+	h.producer = producer
+
+	if len(h.client.Brokers()) > 0 {
+		// Refer https://github.com/IBM/sarama/issues/1341
+		atomic.StoreInt32(&h.status, statusOnline)
+	}
+
+	return nil
+}
+
+// IsOnline returns true if the target is online.
+func (h *Target) IsOnline(_ context.Context) bool {
+	return atomic.LoadInt32(&h.status) == statusOnline
+}
+
+// Send log message 'e' to kafka target.
+func (h *Target) Send(ctx context.Context, entry any) error {
+	if h.store != nil {
+		// save the entry to the queue store which will be replayed to the target.
+		_, err := h.store.Put(entry)
+		return err
+	}
+	h.logChMu.RLock()
+	defer h.logChMu.RUnlock()
+	if h.logCh == nil {
+		// We are closing...
+		return nil
+	}
+
+	select {
+	case h.logCh <- entry:
+	case <-ctx.Done():
+		// return error only for context timedout.
+		if errors.Is(ctx.Err(), context.DeadlineExceeded) {
+			return ctx.Err()
+		}
+		return nil
+	default:
+		// log channel is full, do not wait and return
+		// an error immediately to the caller
+		atomic.AddInt64(&h.totalMessages, 1)
+		atomic.AddInt64(&h.failedMessages, 1)
+		return errors.New("log buffer full")
+	}
+	return nil
+}
+
+// SendFromStore - reads the log from store and sends it to kafka.
+func (h *Target) SendFromStore(key store.Key) (err error) {
+	auditEntry, err := h.store.Get(key)
+	if err != nil {
+		if os.IsNotExist(err) {
+			return nil
+		}
+		return err
+	}
+	atomic.AddInt64(&h.totalMessages, 1)
+	err = h.send(auditEntry)
+	if err != nil {
+		atomic.AddInt64(&h.failedMessages, 1)
+		return err
+	}
+	// Delete the event from store.
+	return h.store.Del(key)
+}
+
+// Cancel - cancels the target
+func (h *Target) Cancel() {
+	// If queuestore is configured, cancel it's context to
+	// stop the replay go-routine.
+	if h.store != nil {
+		h.storeCtxCancel()
+	}
+
+	// Set logch to nil and close it.
+	// This will block all Send operations,
+	// and finish the existing ones.
+	// All future ones will be discarded.
+	h.logChMu.Lock()
+	xioutil.SafeClose(h.logCh)
+	h.logCh = nil
+	h.logChMu.Unlock()
+
+	if h.producer != nil {
+		h.producer.Close()
+		h.client.Close()
+	}
+
+	// Wait for messages to be sent...
+	h.wg.Wait()
+}
+
+// New initializes a new logger target which
+// sends log over http to the specified endpoint
+func New(config Config) *Target {
+	target := &Target{
+		logCh:   make(chan any, config.QueueSize),
+		kconfig: config,
+		status:  statusOffline,
+	}
+	return target
+}
+
+// Type - returns type of the target
+func (h *Target) Type() types.TargetType {
+	return types.TargetKafka
+}

internal/logger/target/kafka/kafka_scram_client_contrib.go

@@ -0,0 +1,87 @@
+// Copyright (c) 2015-2021 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 kafka
+
+import (
+	"crypto/sha512"
+	"strings"
+
+	"github.com/IBM/sarama"
+	"github.com/xdg/scram"
+
+	"github.com/minio/minio/internal/hash/sha256"
+)
+
+func initScramClient(cfg Config, config *sarama.Config) {
+	switch strings.ToLower(cfg.SASL.Mechanism) {
+	case "sha512":
+		config.Net.SASL.SCRAMClientGeneratorFunc = func() sarama.SCRAMClient { return &XDGSCRAMClient{HashGeneratorFcn: KafkaSHA512} }
+		config.Net.SASL.Mechanism = sarama.SASLMechanism(sarama.SASLTypeSCRAMSHA512)
+	case "sha256":
+		config.Net.SASL.SCRAMClientGeneratorFunc = func() sarama.SCRAMClient { return &XDGSCRAMClient{HashGeneratorFcn: KafkaSHA256} }
+		config.Net.SASL.Mechanism = sarama.SASLMechanism(sarama.SASLTypeSCRAMSHA256)
+	default:
+		// default to PLAIN
+		config.Net.SASL.Mechanism = sarama.SASLMechanism(sarama.SASLTypePlaintext)
+	}
+}
+
+// KafkaSHA256 is a function that returns a crypto/sha256 hasher and should be used
+// to create Client objects configured for SHA-256 hashing.
+var KafkaSHA256 scram.HashGeneratorFcn = sha256.New
+
+// KafkaSHA512 is a function that returns a crypto/sha512 hasher and should be used
+// to create Client objects configured for SHA-512 hashing.
+var KafkaSHA512 scram.HashGeneratorFcn = sha512.New
+
+// XDGSCRAMClient implements the client-side of an authentication
+// conversation with a server.  A new conversation must be created for
+// each authentication attempt.
+type XDGSCRAMClient struct {
+	*scram.Client
+	*scram.ClientConversation
+	scram.HashGeneratorFcn
+}
+
+// Begin constructs a SCRAM client component based on a given hash.Hash
+// factory receiver.  This constructor will normalize the username, password
+// and authzID via the SASLprep algorithm, as recommended by RFC-5802.  If
+// SASLprep fails, the method returns an error.
+func (x *XDGSCRAMClient) Begin(userName, password, authzID string) (err error) {
+	x.Client, err = x.NewClient(userName, password, authzID)
+	if err != nil {
+		return err
+	}
+	x.ClientConversation = x.NewConversation()
+	return nil
+}
+
+// Step takes a string provided from a server (or just an empty string for the
+// very first conversation step) and attempts to move the authentication
+// conversation forward.  It returns a string to be sent to the server or an
+// error if the server message is invalid.  Calling Step after a conversation
+// completes is also an error.
+func (x *XDGSCRAMClient) Step(challenge string) (response string, err error) {
+	response, err = x.ClientConversation.Step(challenge)
+	return response, err
+}
+
+// Done returns true if the conversation is completed or has errored.
+func (x *XDGSCRAMClient) Done() bool {
+	return x.ClientConversation.Done()
+}

internal/logger/target/loggertypes/targettype_string.go

@@ -0,0 +1,26 @@
+// Code generated by "stringer -type=TargetType -trimprefix=Target types.go"; DO NOT EDIT.
+
+package loggertypes
+
+import "strconv"
+
+func _() {
+	// An "invalid array index" compiler error signifies that the constant values have changed.
+	// Re-run the stringer command to generate them again.
+	var x [1]struct{}
+	_ = x[TargetConsole-1]
+	_ = x[TargetHTTP-2]
+	_ = x[TargetKafka-3]
+}
+
+const _TargetType_name = "ConsoleHTTPKafka"
+
+var _TargetType_index = [...]uint8{0, 7, 11, 16}
+
+func (i TargetType) String() string {
+	i -= 1
+	if i >= TargetType(len(_TargetType_index)-1) {
+		return "TargetType(" + strconv.FormatInt(int64(i+1), 10) + ")"
+	}
+	return _TargetType_name[_TargetType_index[i]:_TargetType_index[i+1]]
+}

internal/logger/target/loggertypes/types.go

@@ -0,0 +1,43 @@
+// Copyright (c) 2015-2022 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 loggertypes
+
+// TargetType indicates type of the target e.g. console, http, kafka
+type TargetType uint8
+
+//go:generate stringer -type=TargetType -trimprefix=Target $GOFILE
+
+// Constants for target types
+const (
+	_ TargetType = iota
+	TargetConsole
+	TargetHTTP
+	TargetKafka
+)
+
+// TargetStats contains statistics for a target.
+type TargetStats struct {
+	// QueueLength is the queue length if any messages are queued.
+	QueueLength int
+
+	// TotalMessages is the total number of messages sent in the lifetime of the target
+	TotalMessages int64
+
+	// FailedMessages should log message count that failed to send.
+	FailedMessages int64
+}

internal/logger/target/testlogger/testlogger.go

@@ -0,0 +1,170 @@
+// 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 testlogger contains an autoregistering logger that can be used to capture logging events
+// for individual tests.
+// This package should only be included by test files.
+// To enable logging for a test, use:
+//
+//	func TestSomething(t *testing.T) {
+//		defer testlogger.T.SetLogTB(t)()
+//
+// This cannot be used for parallel tests.
+package testlogger
+
+import (
+	"context"
+	"fmt"
+	"os"
+	"strings"
+	"sync/atomic"
+	"testing"
+
+	"github.com/minio/madmin-go/v3/logger/log"
+	"github.com/minio/minio/internal/logger"
+	types "github.com/minio/minio/internal/logger/target/loggertypes"
+)
+
+const (
+	logMessage = iota
+	errorMessage
+	fatalMessage
+)
+
+// T is the test logger.
+var T = &testLogger{}
+
+func init() {
+	logger.AddSystemTarget(context.Background(), T)
+}
+
+type testLogger struct {
+	current atomic.Pointer[testing.TB]
+	action  atomic.Int32
+}
+
+// SetLogTB will set the logger to output to tb.
+// Call the returned function to disable logging.
+func (t *testLogger) SetLogTB(tb testing.TB) func() {
+	return t.setTB(tb, logMessage)
+}
+
+// SetErrorTB will set the logger to output to tb.Error.
+// Call the returned function to disable logging.
+func (t *testLogger) SetErrorTB(tb testing.TB) func() {
+	return t.setTB(tb, errorMessage)
+}
+
+// SetFatalTB will set the logger to output to tb.Panic.
+// Call the returned function to disable logging.
+func (t *testLogger) SetFatalTB(tb testing.TB) func() {
+	return t.setTB(tb, fatalMessage)
+}
+
+func (t *testLogger) setTB(tb testing.TB, action int32) func() {
+	old := t.action.Swap(action)
+	t.current.Store(&tb)
+	return func() {
+		t.current.Store(nil)
+		t.action.Store(old)
+	}
+}
+
+func (t *testLogger) String() string {
+	tb := t.current.Load()
+	if tb != nil {
+		tbb := *tb
+		return tbb.Name()
+	}
+	return ""
+}
+
+func (t *testLogger) Endpoint() string {
+	return ""
+}
+
+func (t *testLogger) Stats() types.TargetStats {
+	return types.TargetStats{}
+}
+
+func (t *testLogger) Init(ctx context.Context) error {
+	return nil
+}
+
+func (t *testLogger) IsOnline(ctx context.Context) bool {
+	return t.current.Load() != nil
+}
+
+func (t *testLogger) Cancel() {
+	t.current.Store(nil)
+}
+
+func (t *testLogger) Send(ctx context.Context, entry any) error {
+	tb := t.current.Load()
+	var logf func(format string, args ...any)
+	if tb != nil {
+		tbb := *tb
+		tbb.Helper()
+		switch t.action.Load() {
+		case errorMessage:
+			logf = tbb.Errorf
+		case fatalMessage:
+			logf = tbb.Fatalf
+		default:
+			logf = tbb.Logf
+		}
+	} else {
+		switch t.action.Load() {
+		case errorMessage:
+			logf = func(format string, args ...any) {
+				fmt.Fprintf(os.Stderr, format+"\n", args...)
+			}
+		case fatalMessage:
+			logf = func(format string, args ...any) {
+				fmt.Fprintf(os.Stderr, format+"\n", args...)
+			}
+			defer os.Exit(1)
+		default:
+			logf = func(format string, args ...any) {
+				fmt.Fprintf(os.Stdout, format+"\n", args...)
+			}
+		}
+	}
+
+	switch v := entry.(type) {
+	case log.Entry:
+		if v.Trace == nil {
+			logf("%s: %s", v.Level, v.Message)
+		} else {
+			msg := fmt.Sprintf("%s: %+v", v.Level, v.Trace.Message)
+			for i, m := range v.Trace.Source {
+				if i == 0 && strings.Contains(m, "logger.go:") {
+					continue
+				}
+				msg += fmt.Sprintf("\n%s", m)
+			}
+			logf("%s", msg)
+		}
+	default:
+		logf("%+v (%T)", v, v)
+	}
+	return nil
+}
+
+func (t *testLogger) Type() types.TargetType {
+	return types.TargetConsole
+}