// 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 notify import (	"context"	"crypto/tls"	"crypto/x509"	"errors"	"net/http"	"strconv"	"strings"	"time" 	"github.com/minio/minio/internal/config"	"github.com/minio/minio/internal/event"	"github.com/minio/minio/internal/event/target"	"github.com/minio/minio/internal/logger"	"github.com/minio/pkg/env"	xnet "github.com/minio/pkg/net") const (	formatNamespace = "namespace") // 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") // TestNotificationTargets is similar to GetNotificationTargets()// avoids explicit registration.func TestNotificationTargets(ctx context.Context, cfg config.Config, transport *http.Transport, targetIDs []event.TargetID) error {	test := true	returnOnTargetError := true	targets, err := RegisterNotificationTargets(ctx, cfg, transport, targetIDs, test, returnOnTargetError)	if err == nil {		// Close all targets since we are only testing connections.		for _, t := range targets.TargetMap() {			_ = t.Close()		}	} 	return err} // TestSubSysNotificationTargets - tests notification targets of given subsystemfunc TestSubSysNotificationTargets(ctx context.Context, cfg config.Config, transport *http.Transport, targetIDs []event.TargetID, subSys string) error {	if err := checkValidNotificationKeysForSubSys(subSys, cfg[subSys]); err != nil {		return err	}	targetList := event.NewTargetList()	targetsOffline, err := fetchSubSysTargets(ctx, cfg, transport, true, true, subSys, targetList)	if err == nil {		// Close all targets since we are only testing connections.		for _, t := range targetList.TargetMap() {			_ = t.Close()		}	} 	if targetsOffline {		return ErrTargetsOffline	} 	return err} // GetNotificationTargets registers and initializes all notification// targets, returns error if any.func GetNotificationTargets(ctx context.Context, cfg config.Config, transport *http.Transport, test bool) (*event.TargetList, error) {	returnOnTargetError := false	return RegisterNotificationTargets(ctx, cfg, transport, nil, test, returnOnTargetError)} // RegisterNotificationTargets - returns TargetList which contains enabled targets in serverConfig.// A new notification target is added like below// * Add a new target in pkg/event/target package.// * Add newly added target configuration to serverConfig.Notify.<TARGET_NAME>.// * Handle the configuration in this function to create/add into TargetList.func RegisterNotificationTargets(ctx context.Context, cfg config.Config, transport *http.Transport, targetIDs []event.TargetID, test bool, returnOnTargetError bool) (*event.TargetList, error) {	targetList, err := FetchRegisteredTargets(ctx, cfg, transport, test, returnOnTargetError)	if err != nil {		return targetList, err	} 	if test {		// Verify if user is trying to disable already configured		// notification targets, based on their target IDs		for _, targetID := range targetIDs {			if !targetList.Exists(targetID) {				return nil, config.Errorf(					"Unable to disable currently configured targets '%v'",					targetID)			}		}	} 	return targetList, nil} func fetchSubSysTargets(ctx context.Context, cfg config.Config,	transport *http.Transport, test bool, returnOnTargetError bool,	subSys string, targetList *event.TargetList,) (targetsOffline bool, err error) {	targetsOffline = false	if err := checkValidNotificationKeysForSubSys(subSys, cfg[subSys]); err != nil {		return targetsOffline, err	} 	switch subSys {	case config.NotifyAMQPSubSys:		amqpTargets, err := GetNotifyAMQP(cfg[config.NotifyAMQPSubSys])		if err != nil {			return targetsOffline, err		}		for id, args := range amqpTargets {			if !args.Enable {				continue			}			newTarget, err := target.NewAMQPTarget(id, args, ctx.Done(), logger.LogOnceIf, test)			if err != nil {				targetsOffline = true				if returnOnTargetError {					return targetsOffline, err				}				_ = newTarget.Close()			}			if err = targetList.Add(newTarget); err != nil {				logger.LogIf(context.Background(), err)				if returnOnTargetError {					return targetsOffline, err				}			}		}	case config.NotifyESSubSys:		esTargets, err := GetNotifyES(cfg[config.NotifyESSubSys], transport)		if err != nil {			return targetsOffline, err		}		for id, args := range esTargets {			if !args.Enable {				continue			}			newTarget, err := target.NewElasticsearchTarget(id, args, ctx.Done(), logger.LogOnceIf, test)			if err != nil {				targetsOffline = true				if returnOnTargetError {					return targetsOffline, err				}				_ = newTarget.Close()			}			if err = targetList.Add(newTarget); err != nil {				logger.LogIf(context.Background(), err)				if returnOnTargetError {					return targetsOffline, err				}			}		}	case config.NotifyKafkaSubSys:		kafkaTargets, err := GetNotifyKafka(cfg[config.NotifyKafkaSubSys])		if err != nil {			return targetsOffline, err		}		for id, args := range kafkaTargets {			if !args.Enable {				continue			}			args.TLS.RootCAs = transport.TLSClientConfig.RootCAs			newTarget, err := target.NewKafkaTarget(id, args, ctx.Done(), logger.LogOnceIf, test)			if err != nil {				targetsOffline = true				if returnOnTargetError {					return targetsOffline, err				}				_ = newTarget.Close()			}			if err = targetList.Add(newTarget); err != nil {				logger.LogIf(context.Background(), err)				if returnOnTargetError {					return targetsOffline, err				}			}		} 	case config.NotifyMQTTSubSys:		mqttTargets, err := GetNotifyMQTT(cfg[config.NotifyMQTTSubSys], transport.TLSClientConfig.RootCAs)		if err != nil {			return targetsOffline, err		}		for id, args := range mqttTargets {			if !args.Enable {				continue			}			args.RootCAs = transport.TLSClientConfig.RootCAs			newTarget, err := target.NewMQTTTarget(id, args, ctx.Done(), logger.LogOnceIf, test)			if err != nil {				targetsOffline = true				if returnOnTargetError {					return targetsOffline, err				}				_ = newTarget.Close()			}			if err = targetList.Add(newTarget); err != nil {				logger.LogIf(context.Background(), err)				if returnOnTargetError {					return targetsOffline, err				}			}		}	case config.NotifyMySQLSubSys:		mysqlTargets, err := GetNotifyMySQL(cfg[config.NotifyMySQLSubSys])		if err != nil {			return targetsOffline, err		}		for id, args := range mysqlTargets {			if !args.Enable {				continue			}			newTarget, err := target.NewMySQLTarget(id, args, ctx.Done(), logger.LogOnceIf, test)			if err != nil {				targetsOffline = true				if returnOnTargetError {					return targetsOffline, err				}				_ = newTarget.Close()			}			if err = targetList.Add(newTarget); err != nil {				logger.LogIf(context.Background(), err)				if returnOnTargetError {					return targetsOffline, err				}			}		}	case config.NotifyNATSSubSys:		natsTargets, err := GetNotifyNATS(cfg[config.NotifyNATSSubSys], transport.TLSClientConfig.RootCAs)		if err != nil {			return targetsOffline, err		}		for id, args := range natsTargets {			if !args.Enable {				continue			}			newTarget, err := target.NewNATSTarget(id, args, ctx.Done(), logger.LogOnceIf, test)			if err != nil {				targetsOffline = true				if returnOnTargetError {					return targetsOffline, err				}				_ = newTarget.Close()			}			if err = targetList.Add(newTarget); err != nil {				logger.LogIf(context.Background(), err)				if returnOnTargetError {					return targetsOffline, err				}			}		}	case config.NotifyNSQSubSys:		nsqTargets, err := GetNotifyNSQ(cfg[config.NotifyNSQSubSys])		if err != nil {			return targetsOffline, err		}		for id, args := range nsqTargets {			if !args.Enable {				continue			}			newTarget, err := target.NewNSQTarget(id, args, ctx.Done(), logger.LogOnceIf, test)			if err != nil {				targetsOffline = true				if returnOnTargetError {					return targetsOffline, err				}				_ = newTarget.Close()			}			if err = targetList.Add(newTarget); err != nil {				logger.LogIf(context.Background(), err)				if returnOnTargetError {					return targetsOffline, err				}			}		}	case config.NotifyPostgresSubSys:		postgresTargets, err := GetNotifyPostgres(cfg[config.NotifyPostgresSubSys])		if err != nil {			return targetsOffline, err		}		for id, args := range postgresTargets {			if !args.Enable {				continue			}			newTarget, err := target.NewPostgreSQLTarget(id, args, ctx.Done(), logger.LogOnceIf, test)			if err != nil {				targetsOffline = true				if returnOnTargetError {					return targetsOffline, err				}				_ = newTarget.Close()			}			if err = targetList.Add(newTarget); err != nil {				logger.LogIf(context.Background(), err)				if returnOnTargetError {					return targetsOffline, err				}			}		}	case config.NotifyRedisSubSys:		redisTargets, err := GetNotifyRedis(cfg[config.NotifyRedisSubSys])		if err != nil {			return targetsOffline, err		}		for id, args := range redisTargets {			if !args.Enable {				continue			}			newTarget, err := target.NewRedisTarget(id, args, ctx.Done(), logger.LogOnceIf, test)			if err != nil {				targetsOffline = true				if returnOnTargetError {					return targetsOffline, err				}				_ = newTarget.Close()			}			if err = targetList.Add(newTarget); err != nil {				logger.LogIf(context.Background(), err)				if returnOnTargetError {					return targetsOffline, err				}			}		}	case config.NotifyWebhookSubSys:		webhookTargets, err := GetNotifyWebhook(cfg[config.NotifyWebhookSubSys], transport)		if err != nil {			return targetsOffline, err		}		for id, args := range webhookTargets {			if !args.Enable {				continue			}			newTarget, err := target.NewWebhookTarget(ctx, id, args, logger.LogOnceIf, transport, test)			if err != nil {				targetsOffline = true				if returnOnTargetError {					return targetsOffline, err				}				_ = newTarget.Close()			}			if err = targetList.Add(newTarget); err != nil {				logger.LogIf(context.Background(), err)				if returnOnTargetError {					return targetsOffline, err				}			}		} 	}	return targetsOffline, nil} // FetchRegisteredTargets - Returns a set of configured TargetList// If `returnOnTargetError` is set to true, The function returns when a target initialization fails// Else, the function will return a complete TargetList irrespective of errorsfunc FetchRegisteredTargets(ctx context.Context, cfg config.Config, transport *http.Transport, test bool, returnOnTargetError bool) (_ *event.TargetList, err error) {	targetList := event.NewTargetList()	var targetsOffline bool 	defer func() {		// Automatically close all connections to targets when an error occur.		// Close all the targets if returnOnTargetError is set		// Else, close only the failed targets		if err != nil && returnOnTargetError {			for _, t := range targetList.TargetMap() {				_ = t.Close()			}		}	}() 	for _, subSys := range config.NotifySubSystems.ToSlice() {		if targetsOffline, err = fetchSubSysTargets(ctx, cfg, transport, test, returnOnTargetError, subSys, targetList); err != nil {			return targetList, err		}		if targetsOffline {			return targetList, ErrTargetsOffline		}	} 	return targetList, nil} // DefaultNotificationKVS - default notification list of kvs.var (	DefaultNotificationKVS = map[string]config.KVS{		config.NotifyAMQPSubSys:     DefaultAMQPKVS,		config.NotifyKafkaSubSys:    DefaultKafkaKVS,		config.NotifyMQTTSubSys:     DefaultMQTTKVS,		config.NotifyMySQLSubSys:    DefaultMySQLKVS,		config.NotifyNATSSubSys:     DefaultNATSKVS,		config.NotifyNSQSubSys:      DefaultNSQKVS,		config.NotifyPostgresSubSys: DefaultPostgresKVS,		config.NotifyRedisSubSys:    DefaultRedisKVS,		config.NotifyWebhookSubSys:  DefaultWebhookKVS,		config.NotifyESSubSys:       DefaultESKVS,	}) func checkValidNotificationKeysForSubSys(subSys string, tgt map[string]config.KVS) error {	validKVS, ok := DefaultNotificationKVS[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} // DefaultKakfaKVS - default KV for kafka targetvar (	DefaultKafkaKVS = config.KVS{		config.KV{			Key:   config.Enable,			Value: config.EnableOff,		},		config.KV{			Key:   target.KafkaTopic,			Value: "",		},		config.KV{			Key:   target.KafkaBrokers,			Value: "",		},		config.KV{			Key:   target.KafkaSASLUsername,			Value: "",		},		config.KV{			Key:   target.KafkaSASLPassword,			Value: "",		},		config.KV{			Key:   target.KafkaSASLMechanism,			Value: "plain",		},		config.KV{			Key:   target.KafkaClientTLSCert,			Value: "",		},		config.KV{			Key:   target.KafkaClientTLSKey,			Value: "",		},		config.KV{			Key:   target.KafkaTLSClientAuth,			Value: "0",		},		config.KV{			Key:   target.KafkaSASL,			Value: config.EnableOff,		},		config.KV{			Key:   target.KafkaTLS,			Value: config.EnableOff,		},		config.KV{			Key:   target.KafkaTLSSkipVerify,			Value: config.EnableOff,		},		config.KV{			Key:   target.KafkaQueueLimit,			Value: "0",		},		config.KV{			Key:   target.KafkaQueueDir,			Value: "",		},		config.KV{			Key:   target.KafkaVersion,			Value: "",		},

Lines 1–500 of 1846. Load more