diff --git a/docs/CONFIGURATION.md b/docs/CONFIGURATION.md
index 746fbc2b9435f..d67d5d5e47abd 100644
--- a/docs/CONFIGURATION.md
+++ b/docs/CONFIGURATION.md
@@ -410,7 +410,7 @@ Parameters that can be used with any input plugin:
 - **name_suffix**: Specifies a suffix to attach to the measurement name.
 - **tags**: A map of tags to apply to a specific input's measurements.
 - **log_level**: Override the log-level for this plugin. Possible values are
-  `error`, `warn`, `info` and `debug`.
+  `error`, `warn`, `info`, `debug` and `trace`.
 
 The [metric filtering][] parameters can be used to limit what metrics are
 emitted from the input plugin.
diff --git a/logger.go b/logger.go
index 4f8fd7cf8582e..500b04b1696cf 100644
--- a/logger.go
+++ b/logger.go
@@ -14,6 +14,8 @@ const (
 	Info
 	// Debug will log all of the above and debugging messages issued by plugins
 	Debug
+	// Trace will log all of the above and trace messages issued by plugins
+	Trace
 )
 
 func LogLevelFromString(name string) LogLevel {
@@ -26,6 +28,8 @@ func LogLevelFromString(name string) LogLevel {
 		return Info
 	case "DEBUG", "debug":
 		return Debug
+	case "TRACE", "trace":
+		return Trace
 	}
 	return None
 }
@@ -40,6 +44,8 @@ func (e LogLevel) String() string {
 		return "INFO"
 	case Debug:
 		return "DEBUG"
+	case Trace:
+		return "TRACE"
 	}
 	return "NONE"
 }
@@ -54,6 +60,8 @@ func (e LogLevel) Indicator() string {
 		return "I!"
 	case Debug:
 		return "D!"
+	case Trace:
+		return "T!"
 	}
 	return "U!"
 }
@@ -63,7 +71,7 @@ func (e LogLevel) Includes(level LogLevel) bool {
 }
 
 // Logger defines an plugin-related interface for logging.
-type Logger interface {
+type Logger interface { //nolint:interfacebloat // All functions are required
 	// Level returns the configured log-level of the logger
 	Level() LogLevel
 
@@ -71,10 +79,6 @@ type Logger interface {
 	Errorf(format string, args ...interface{})
 	// Error logs an error message, patterned after log.Print.
 	Error(args ...interface{})
-	// Debugf logs a debug message, patterned after log.Printf.
-	Debugf(format string, args ...interface{})
-	// Debug logs a debug message, patterned after log.Print.
-	Debug(args ...interface{})
 	// Warnf logs a warning message, patterned after log.Printf.
 	Warnf(format string, args ...interface{})
 	// Warn logs a warning message, patterned after log.Print.
@@ -83,4 +87,12 @@ type Logger interface {
 	Infof(format string, args ...interface{})
 	// Info logs an information message, patterned after log.Print.
 	Info(args ...interface{})
+	// Debugf logs a debug message, patterned after log.Printf.
+	Debugf(format string, args ...interface{})
+	// Debug logs a debug message, patterned after log.Print.
+	Debug(args ...interface{})
+	// Tracef logs a trace message, patterned after log.Printf.
+	Tracef(format string, args ...interface{})
+	// Trace logs a trace message, patterned after log.Print.
+	Trace(args ...interface{})
 }
diff --git a/logger/logger.go b/logger/logger.go
index 03d1b1ef05dfa..50840700ac85d 100644
--- a/logger/logger.go
+++ b/logger/logger.go
@@ -55,19 +55,6 @@ func New(category, name, alias string) *logger {
 	return l
 }
 
-// SetLevel changes the log-level to the given one
-func (l *logger) SetLogLevel(name string) error {
-	if name == "" {
-		return nil
-	}
-	level := telegraf.LogLevelFromString(name)
-	if level == telegraf.None {
-		return fmt.Errorf("invalid log-level %q", name)
-	}
-	l.level = &level
-	return nil
-}
-
 // SubLogger creates a new logger with the given name added as suffix
 func (l *logger) SubLogger(name string) telegraf.Logger {
 	suffix := l.suffix
@@ -118,6 +105,24 @@ func (l *logger) Level() telegraf.LogLevel {
 	return instance.level
 }
 
+// SetLevel overrides the current log-level of the logger
+func (l *logger) SetLevel(level telegraf.LogLevel) {
+	l.level = &level
+}
+
+// SetLevel changes the log-level to the given one
+func (l *logger) SetLogLevel(name string) error {
+	if name == "" {
+		return nil
+	}
+	level := telegraf.LogLevelFromString(name)
+	if level == telegraf.None {
+		return fmt.Errorf("invalid log-level %q", name)
+	}
+	l.SetLevel(level)
+	return nil
+}
+
 // Register a callback triggered when errors are about to be written to the log
 func (l *logger) RegisterErrorCallback(f func()) {
 	l.onError = append(l.onError, f)
@@ -162,6 +167,15 @@ func (l *logger) Debug(args ...interface{}) {
 	l.Print(telegraf.Debug, time.Now(), args...)
 }
 
+// Trace logging, this is suppressed on console
+func (l *logger) Tracef(format string, args ...interface{}) {
+	l.Trace(fmt.Sprintf(format, args...))
+}
+
+func (l *logger) Trace(args ...interface{}) {
+	l.Print(telegraf.Trace, time.Now(), args...)
+}
+
 func (l *logger) Print(level telegraf.LogLevel, ts time.Time, args ...interface{}) {
 	// Check if we are in early logging state and store the message in this case
 	if instance.impl == nil {
diff --git a/logger/stdlog_redirector.go b/logger/stdlog_redirector.go
index 6e9c8c2e67f4e..4c34c1d70c768 100644
--- a/logger/stdlog_redirector.go
+++ b/logger/stdlog_redirector.go
@@ -24,6 +24,8 @@ func (s *stdlogRedirector) Write(b []byte) (n int, err error) {
 
 	// Log with the given level
 	switch level {
+	case 'T':
+		s.log.Trace(string(msg))
 	case 'D':
 		s.log.Debug(string(msg))
 	case 'I':
diff --git a/plugins/common/kafka/logger.go b/plugins/common/kafka/logger.go
index 7b9bfc98b1296..0ddda7ff7bb4a 100644
--- a/plugins/common/kafka/logger.go
+++ b/plugins/common/kafka/logger.go
@@ -1,33 +1,41 @@
 package kafka
 
 import (
+	"sync"
+
 	"github.com/IBM/sarama"
 
 	"github.com/influxdata/telegraf"
 	"github.com/influxdata/telegraf/logger"
 )
 
-type Logger struct {
-}
+var (
+	log  = logger.New("sarama", "", "")
+	once sync.Once
+)
 
-// DebugLogger logs messages from sarama at the debug level.
-type DebugLogger struct {
-	Log telegraf.Logger
-}
+type debugLogger struct{}
 
-func (l *DebugLogger) Print(v ...interface{}) {
-	l.Log.Debug(v...)
+func (l *debugLogger) Print(v ...interface{}) {
+	log.Trace(v...)
 }
 
-func (l *DebugLogger) Printf(format string, v ...interface{}) {
-	l.Log.Debugf(format, v...)
+func (l *debugLogger) Printf(format string, v ...interface{}) {
+	log.Tracef(format, v...)
 }
 
-func (l *DebugLogger) Println(v ...interface{}) {
+func (l *debugLogger) Println(v ...interface{}) {
 	l.Print(v...)
 }
 
 // SetLogger configures a debug logger for kafka (sarama)
-func (k *Logger) SetLogger() {
-	sarama.Logger = &DebugLogger{Log: logger.New("sarama", "", "")}
+func SetLogger(level telegraf.LogLevel) {
+	// Set-up the sarama logger only once
+	once.Do(func() {
+		sarama.Logger = &debugLogger{}
+	})
+	// Increase the log-level if needed.
+	if !log.Level().Includes(level) {
+		log.SetLevel(level)
+	}
 }
diff --git a/plugins/inputs/kafka_consumer/kafka_consumer.go b/plugins/inputs/kafka_consumer/kafka_consumer.go
index 6e9e637bbb781..429490805c348 100644
--- a/plugins/inputs/kafka_consumer/kafka_consumer.go
+++ b/plugins/inputs/kafka_consumer/kafka_consumer.go
@@ -52,11 +52,8 @@ type KafkaConsumer struct {
 	ConsumerFetchDefault                 config.Size     `toml:"consumer_fetch_default"`
 	ConnectionStrategy                   string          `toml:"connection_strategy"`
 	ResolveCanonicalBootstrapServersOnly bool            `toml:"resolve_canonical_bootstrap_servers_only"`
-
+	Log                                  telegraf.Logger `toml:"-"`
 	kafka.ReadConfig
-	kafka.Logger
-
-	Log telegraf.Logger `toml:"-"`
 
 	ConsumerCreator ConsumerGroupCreator `toml:"-"`
 	consumer        ConsumerGroup
@@ -99,7 +96,7 @@ func (k *KafkaConsumer) SetParser(parser telegraf.Parser) {
 }
 
 func (k *KafkaConsumer) Init() error {
-	k.SetLogger()
+	kafka.SetLogger(k.Log.Level())
 
 	if k.MaxUndeliveredMessages == 0 {
 		k.MaxUndeliveredMessages = defaultMaxUndeliveredMessages
diff --git a/plugins/inputs/kafka_consumer/kafka_consumer_test.go b/plugins/inputs/kafka_consumer/kafka_consumer_test.go
index e9f75dbd0fc86..608f89b58f5cd 100644
--- a/plugins/inputs/kafka_consumer/kafka_consumer_test.go
+++ b/plugins/inputs/kafka_consumer/kafka_consumer_test.go
@@ -67,7 +67,7 @@ func TestInit(t *testing.T) {
 	}{
 		{
 			name:   "default config",
-			plugin: &KafkaConsumer{},
+			plugin: &KafkaConsumer{Log: testutil.Logger{}},
 			check: func(t *testing.T, plugin *KafkaConsumer) {
 				require.Equal(t, defaultConsumerGroup, plugin.ConsumerGroup)
 				require.Equal(t, defaultMaxUndeliveredMessages, plugin.MaxUndeliveredMessages)
diff --git a/plugins/inputs/modbus/README.md b/plugins/inputs/modbus/README.md
index 8fd7b4bb6b2aa..10a8570622218 100644
--- a/plugins/inputs/modbus/README.md
+++ b/plugins/inputs/modbus/README.md
@@ -60,9 +60,8 @@ See the [CONFIGURATION.md][CONFIGURATION.md] for more details.
   ## and "RTU" for serial connections.
   # transmission_mode = "auto"
 
-  ## Trace the connection to the modbus device as debug messages
-  ## Note: You have to enable telegraf's debug mode to see those messages!
-  # debug_connection = false
+  ## Trace the connection to the modbus device
+  # log_level = "trace"
 
   ## Define the configuration schema
   ##  |---register -- define fields per register type in the original style (only supports one slave ID)
diff --git a/plugins/inputs/modbus/modbus.go b/plugins/inputs/modbus/modbus.go
index 0f08871b8c623..0cd9634141f2e 100644
--- a/plugins/inputs/modbus/modbus.go
+++ b/plugins/inputs/modbus/modbus.go
@@ -58,7 +58,7 @@ type Modbus struct {
 	Timeout           config.Duration   `toml:"timeout"`
 	Retries           int               `toml:"busy_retries"`
 	RetriesWaitTime   config.Duration   `toml:"busy_retries_wait"`
-	DebugConnection   bool              `toml:"debug_connection"`
+	DebugConnection   bool              `toml:"debug_connection" deprecated:"1.35.0;use 'log_level' 'trace' instead"`
 	Workarounds       ModbusWorkarounds `toml:"workarounds"`
 	ConfigurationType string            `toml:"configuration_type"`
 	Log               telegraf.Logger   `toml:"-"`
@@ -261,6 +261,11 @@ func (m *Modbus) initClient() error {
 		return err
 	}
 
+	var tracelog mb.Logger
+	if m.Log.Level().Includes(telegraf.Trace) || m.DebugConnection { // for backward compatibility
+		tracelog = m
+	}
+
 	switch u.Scheme {
 	case "tcp":
 		host, port, err := net.SplitHostPort(u.Host)
@@ -271,23 +276,17 @@ func (m *Modbus) initClient() error {
 		case "", "auto", "TCP":
 			handler := mb.NewTCPClientHandler(host + ":" + port)
 			handler.Timeout = time.Duration(m.Timeout)
-			if m.DebugConnection {
-				handler.Logger = m
-			}
+			handler.Logger = tracelog
 			m.handler = handler
 		case "RTUoverTCP":
 			handler := mb.NewRTUOverTCPClientHandler(host + ":" + port)
 			handler.Timeout = time.Duration(m.Timeout)
-			if m.DebugConnection {
-				handler.Logger = m
-			}
+			handler.Logger = tracelog
 			m.handler = handler
 		case "ASCIIoverTCP":
 			handler := mb.NewASCIIOverTCPClientHandler(host + ":" + port)
 			handler.Timeout = time.Duration(m.Timeout)
-			if m.DebugConnection {
-				handler.Logger = m
-			}
+			handler.Logger = tracelog
 			m.handler = handler
 		default:
 			return fmt.Errorf("invalid transmission mode %q for %q", m.TransmissionMode, u.Scheme)
@@ -305,9 +304,7 @@ func (m *Modbus) initClient() error {
 			handler.DataBits = m.DataBits
 			handler.Parity = m.Parity
 			handler.StopBits = m.StopBits
-			if m.DebugConnection {
-				handler.Logger = m
-			}
+			handler.Logger = tracelog
 			if m.RS485 != nil {
 				handler.RS485.Enabled = true
 				handler.RS485.DelayRtsBeforeSend = time.Duration(m.RS485.DelayRtsBeforeSend)
@@ -324,9 +321,7 @@ func (m *Modbus) initClient() error {
 			handler.DataBits = m.DataBits
 			handler.Parity = m.Parity
 			handler.StopBits = m.StopBits
-			if m.DebugConnection {
-				handler.Logger = m
-			}
+			handler.Logger = tracelog
 			if m.RS485 != nil {
 				handler.RS485.Enabled = true
 				handler.RS485.DelayRtsBeforeSend = time.Duration(m.RS485.DelayRtsBeforeSend)
@@ -541,7 +536,7 @@ func (m *Modbus) collectFields(acc telegraf.Accumulator, timestamp time.Time, ta
 
 // Implement the logger interface of the modbus client
 func (m *Modbus) Printf(format string, v ...interface{}) {
-	m.Log.Debugf(format, v...)
+	m.Log.Tracef(format, v...)
 }
 
 // Add this plugin to telegraf
diff --git a/plugins/inputs/modbus/sample_general_begin.conf b/plugins/inputs/modbus/sample_general_begin.conf
index 62d2f7178487e..4a77eee586116 100644
--- a/plugins/inputs/modbus/sample_general_begin.conf
+++ b/plugins/inputs/modbus/sample_general_begin.conf
@@ -42,9 +42,8 @@
   ## and "RTU" for serial connections.
   # transmission_mode = "auto"
 
-  ## Trace the connection to the modbus device as debug messages
-  ## Note: You have to enable telegraf's debug mode to see those messages!
-  # debug_connection = false
+  ## Trace the connection to the modbus device
+  # log_level = "trace"
 
   ## Define the configuration schema
   ##  |---register -- define fields per register type in the original style (only supports one slave ID)
diff --git a/plugins/inputs/netflow/README.md b/plugins/inputs/netflow/README.md
index 23fcea6cc3269..0c3a6b4b3c373 100644
--- a/plugins/inputs/netflow/README.md
+++ b/plugins/inputs/netflow/README.md
@@ -64,10 +64,8 @@ See the [CONFIGURATION.md][CONFIGURATION.md] for more details.
   ## decoding.
   # private_enterprise_number_files = []
 
-  ## Dump incoming packets to the log
-  ## This can be helpful to debug parsing issues. Only active if
-  ## Telegraf is in debug mode.
-  # dump_packets = false
+  ## Log incoming packets for tracing issues
+  # log_level = "trace"
 ```
 
 ## Private Enterprise Number mapping
diff --git a/plugins/inputs/netflow/netflow.go b/plugins/inputs/netflow/netflow.go
index 096b127a27e0a..ad0a351e5de67 100644
--- a/plugins/inputs/netflow/netflow.go
+++ b/plugins/inputs/netflow/netflow.go
@@ -28,7 +28,7 @@ type NetFlow struct {
 	ServiceAddress string          `toml:"service_address"`
 	ReadBufferSize config.Size     `toml:"read_buffer_size"`
 	Protocol       string          `toml:"protocol"`
-	DumpPackets    bool            `toml:"dump_packets"`
+	DumpPackets    bool            `toml:"dump_packets" deprecated:"1.35.0;use 'log_level' 'trace' instead"`
 	PENFiles       []string        `toml:"private_enterprise_number_files"`
 	Log            telegraf.Logger `toml:"-"`
 
@@ -135,8 +135,8 @@ func (n *NetFlow) read(acc telegraf.Accumulator) {
 		if count < 1 {
 			continue
 		}
-		if n.DumpPackets {
-			n.Log.Debugf("raw data: %s", hex.EncodeToString(buf[:count]))
+		if n.Log.Level().Includes(telegraf.Trace) || n.DumpPackets { // for backward compatibility
+			n.Log.Tracef("raw data: %s", hex.EncodeToString(buf[:count]))
 		}
 		metrics, err := n.decoder.Decode(src.IP, buf[:count])
 		if err != nil {
diff --git a/plugins/inputs/netflow/sample.conf b/plugins/inputs/netflow/sample.conf
index b918494147129..0713fafbf4d48 100644
--- a/plugins/inputs/netflow/sample.conf
+++ b/plugins/inputs/netflow/sample.conf
@@ -24,7 +24,5 @@
   ## decoding.
   # private_enterprise_number_files = []
 
-  ## Dump incoming packets to the log
-  ## This can be helpful to debug parsing issues. Only active if
-  ## Telegraf is in debug mode.
-  # dump_packets = false
+  ## Log incoming packets for tracing issues
+  # log_level = "trace"
diff --git a/plugins/inputs/s7comm/README.md b/plugins/inputs/s7comm/README.md
index d15065bdf7bcd..f7e8f8ca61dc8 100644
--- a/plugins/inputs/s7comm/README.md
+++ b/plugins/inputs/s7comm/README.md
@@ -47,9 +47,8 @@ using the `startup_error_behavior` setting. Available values are:
   ## Timeout for requests
   # timeout = "10s"
 
-  ## Log detailed connection messages for debugging
-  ## This option only has an effect when Telegraf runs in debug mode
-  # debug_connection = false
+  ## Log detailed connection messages for tracing issues
+  # log_level = "trace"
 
   ## Metric definition(s)
   [[inputs.s7comm.metric]]
diff --git a/plugins/inputs/s7comm/s7comm.go b/plugins/inputs/s7comm/s7comm.go
index f11d27cdca1e6..ed99f2d7cca4b 100644
--- a/plugins/inputs/s7comm/s7comm.go
+++ b/plugins/inputs/s7comm/s7comm.go
@@ -8,7 +8,6 @@ import (
 	"hash/maphash"
 	"log" //nolint:depguard // Required for tracing connection issues
 	"net"
-	"os"
 	"regexp"
 	"strconv"
 	"strings"
@@ -94,7 +93,7 @@ type S7comm struct {
 	ConnectionType  string             `toml:"connection_type"`
 	BatchMaxSize    int                `toml:"pdu_size"`
 	Timeout         config.Duration    `toml:"timeout"`
-	DebugConnection bool               `toml:"debug_connection"`
+	DebugConnection bool               `toml:"debug_connection" deprecated:"1.35.0;use 'log_level' 'trace' instead"`
 	Configs         []metricDefinition `toml:"metric"`
 	Log             telegraf.Logger    `toml:"-"`
 
@@ -143,8 +142,8 @@ func (s *S7comm) Init() error {
 	// Create handler for the connection
 	s.handler = gos7.NewTCPClientHandlerWithConnectType(s.Server, s.Rack, s.Slot, connectionTypeMap[s.ConnectionType])
 	s.handler.Timeout = time.Duration(s.Timeout)
-	if s.DebugConnection {
-		s.handler.Logger = log.New(os.Stderr, "D! [inputs.s7comm] ", log.LstdFlags)
+	if s.Log.Level().Includes(telegraf.Trace) || s.DebugConnection { // for backward compatibility
+		s.handler.Logger = log.New(&tracelogger{log: s.Log}, "", 0)
 	}
 
 	// Create the requests
@@ -417,6 +416,16 @@ func fieldID(seed maphash.Seed, def metricDefinition, field metricFieldDefinitio
 	return mh.Sum64()
 }
 
+// Logger for tracing internal messages
+type tracelogger struct {
+	log telegraf.Logger
+}
+
+func (l *tracelogger) Write(b []byte) (n int, err error) {
+	l.log.Trace(string(b))
+	return len(b), nil
+}
+
 // Add this plugin to telegraf
 func init() {
 	inputs.Add("s7comm", func() telegraf.Input {
diff --git a/plugins/inputs/s7comm/sample.conf b/plugins/inputs/s7comm/sample.conf
index 1d5eb49295b43..db5aef06e03b7 100644
--- a/plugins/inputs/s7comm/sample.conf
+++ b/plugins/inputs/s7comm/sample.conf
@@ -17,9 +17,8 @@
   ## Timeout for requests
   # timeout = "10s"
 
-  ## Log detailed connection messages for debugging
-  ## This option only has an effect when Telegraf runs in debug mode
-  # debug_connection = false
+  ## Log detailed connection messages for tracing issues
+  # log_level = "trace"
 
   ## Metric definition(s)
   [[inputs.s7comm.metric]]
diff --git a/plugins/inputs/upsd/README.md b/plugins/inputs/upsd/README.md
index 96f9fc456a856..02953505662ec 100644
--- a/plugins/inputs/upsd/README.md
+++ b/plugins/inputs/upsd/README.md
@@ -42,10 +42,9 @@ See the [CONFIGURATION.md][CONFIGURATION.md] for more details.
 
   ## Dump information for debugging
   ## Allows to print the raw variables (and corresponding types) as received
-  ## from the NUT server ONCE for each UPS. The output is only available when
-  ## running Telegraf in debug-mode.
+  ## from the NUT server ONCE for each UPS.
   ## Please attach this information when reporting issues!
-  # dump_raw_variables = false
+  # log_level = "trace"
 ```
 
 ## Pitfalls
diff --git a/plugins/inputs/upsd/sample.conf b/plugins/inputs/upsd/sample.conf
index da23b9137fc07..3ec797338550f 100644
--- a/plugins/inputs/upsd/sample.conf
+++ b/plugins/inputs/upsd/sample.conf
@@ -21,7 +21,6 @@
 
   ## Dump information for debugging
   ## Allows to print the raw variables (and corresponding types) as received
-  ## from the NUT server ONCE for each UPS. The output is only available when
-  ## running Telegraf in debug-mode.
+  ## from the NUT server ONCE for each UPS.
   ## Please attach this information when reporting issues!
-  # dump_raw_variables = false
+  # log_level = "trace"
diff --git a/plugins/inputs/upsd/upsd.go b/plugins/inputs/upsd/upsd.go
index 4f1796f736343..a5cc4ac2ad14e 100644
--- a/plugins/inputs/upsd/upsd.go
+++ b/plugins/inputs/upsd/upsd.go
@@ -60,7 +60,7 @@ type Upsd struct {
 	Password   string          `toml:"password"`
 	ForceFloat bool            `toml:"force_float"`
 	Additional []string        `toml:"additional_fields"`
-	DumpRaw    bool            `toml:"dump_raw_variables"`
+	DumpRaw    bool            `toml:"dump_raw_variables" deprecated:"1.35.0;use 'log_level' 'trace' instead"`
 	Log        telegraf.Logger `toml:"-"`
 
 	filter filter.Filter
@@ -89,7 +89,7 @@ func (u *Upsd) Gather(acc telegraf.Accumulator) error {
 	if err != nil {
 		return err
 	}
-	if u.DumpRaw {
+	if u.Log.Level().Includes(telegraf.Trace) || u.DumpRaw { // for backward compatibility
 		for name, variables := range upsList {
 			// Only dump the information once per UPS
 			if u.dumped[name] {
@@ -101,7 +101,7 @@ func (u *Upsd) Gather(acc telegraf.Accumulator) error {
 				values = append(values, fmt.Sprintf("%s: %v", v.Name, v.Value))
 				types = append(types, fmt.Sprintf("%s: %v", v.Name, v.OriginalType))
 			}
-			u.Log.Debugf("Variables dump for UPS %q:\n%s\n-----\n%s", name, strings.Join(values, "\n"), strings.Join(types, "\n"))
+			u.Log.Tracef("Variables dump for UPS %q:\n%s\n-----\n%s", name, strings.Join(values, "\n"), strings.Join(types, "\n"))
 		}
 	}
 	for name, variables := range upsList {
diff --git a/plugins/outputs/kafka/kafka.go b/plugins/outputs/kafka/kafka.go
index 0b6208cd16cff..0fc430dba842f 100644
--- a/plugins/outputs/kafka/kafka.go
+++ b/plugins/outputs/kafka/kafka.go
@@ -31,16 +31,17 @@ var ValidTopicSuffixMethods = []string{
 var zeroTime = time.Unix(0, 0)
 
 type Kafka struct {
-	Brokers           []string    `toml:"brokers"`
-	Topic             string      `toml:"topic"`
-	TopicTag          string      `toml:"topic_tag"`
-	ExcludeTopicTag   bool        `toml:"exclude_topic_tag"`
-	TopicSuffix       TopicSuffix `toml:"topic_suffix"`
-	RoutingTag        string      `toml:"routing_tag"`
-	RoutingKey        string      `toml:"routing_key"`
-	ProducerTimestamp string      `toml:"producer_timestamp"`
-
+	Brokers           []string        `toml:"brokers"`
+	Topic             string          `toml:"topic"`
+	TopicTag          string          `toml:"topic_tag"`
+	ExcludeTopicTag   bool            `toml:"exclude_topic_tag"`
+	TopicSuffix       TopicSuffix     `toml:"topic_suffix"`
+	RoutingTag        string          `toml:"routing_tag"`
+	RoutingKey        string          `toml:"routing_key"`
+	ProducerTimestamp string          `toml:"producer_timestamp"`
+	Log               telegraf.Logger `toml:"-"`
 	proxy.Socks5ProxyConfig
+	kafka.WriteConfig
 
 	// Legacy TLS config options
 	// TLS client certificate
@@ -50,12 +51,6 @@ type Kafka struct {
 	// TLS certificate authority
 	CA string
 
-	kafka.WriteConfig
-
-	kafka.Logger
-
-	Log telegraf.Logger `toml:"-"`
-
 	saramaConfig *sarama.Config
 	producerFunc func(addrs []string, config *sarama.Config) (sarama.SyncProducer, error)
 	producer     sarama.SyncProducer
@@ -123,10 +118,9 @@ func (k *Kafka) SetSerializer(serializer serializers.Serializer) {
 }
 
 func (k *Kafka) Init() error {
-	k.SetLogger()
+	kafka.SetLogger(k.Log.Level())
 
-	err := ValidateTopicSuffixMethod(k.TopicSuffix.Method)
-	if err != nil {
+	if err := ValidateTopicSuffixMethod(k.TopicSuffix.Method); err != nil {
 		return err
 	}
 	config := sarama.NewConfig()
diff --git a/plugins/outputs/postgresql/postgresql_test.go b/plugins/outputs/postgresql/postgresql_test.go
index 49fa4dfc058d1..228344cb37737 100644
--- a/plugins/outputs/postgresql/postgresql_test.go
+++ b/plugins/outputs/postgresql/postgresql_test.go
@@ -182,16 +182,6 @@ func (la *LogAccumulator) Error(args ...interface{}) {
 	la.append(pgx.LogLevelError, "%v", args)
 }
 
-func (la *LogAccumulator) Debugf(format string, args ...interface{}) {
-	la.tb.Helper()
-	la.append(pgx.LogLevelDebug, format, args)
-}
-
-func (la *LogAccumulator) Debug(args ...interface{}) {
-	la.tb.Helper()
-	la.append(pgx.LogLevelDebug, "%v", args)
-}
-
 func (la *LogAccumulator) Warnf(format string, args ...interface{}) {
 	la.tb.Helper()
 	la.append(pgx.LogLevelWarn, format, args)
@@ -212,6 +202,26 @@ func (la *LogAccumulator) Info(args ...interface{}) {
 	la.append(pgx.LogLevelInfo, "%v", args)
 }
 
+func (la *LogAccumulator) Debugf(format string, args ...interface{}) {
+	la.tb.Helper()
+	la.append(pgx.LogLevelDebug, format, args)
+}
+
+func (la *LogAccumulator) Debug(args ...interface{}) {
+	la.tb.Helper()
+	la.append(pgx.LogLevelDebug, "%v", args)
+}
+
+func (la *LogAccumulator) Tracef(format string, args ...interface{}) {
+	la.tb.Helper()
+	la.append(pgx.LogLevelDebug, format, args)
+}
+
+func (la *LogAccumulator) Trace(args ...interface{}) {
+	la.tb.Helper()
+	la.append(pgx.LogLevelDebug, "%v", args)
+}
+
 var ctx = context.Background()
 
 type PostgresqlTest struct {
diff --git a/plugins/parsers/xpath/README.md b/plugins/parsers/xpath/README.md
index 798db36feb7e9..1769778e97745 100644
--- a/plugins/parsers/xpath/README.md
+++ b/plugins/parsers/xpath/README.md
@@ -136,8 +136,7 @@ XPath expressions.
   # xpath_native_types = false
 
   ## Trace empty node selections for debugging
-  ## This will only produce output in debugging mode.
-  # xpath_trace = false
+  # log_level = "trace"
 
   ## Multiple parsing sections are allowed
   [[inputs.file.xpath]]
diff --git a/plugins/parsers/xpath/parser.go b/plugins/parsers/xpath/parser.go
index 7d8dad113452c..0cf45082fc9e4 100644
--- a/plugins/parsers/xpath/parser.go
+++ b/plugins/parsers/xpath/parser.go
@@ -45,7 +45,7 @@ type Parser struct {
 	PrintDocument        bool              `toml:"xpath_print_document"`
 	AllowEmptySelection  bool              `toml:"xpath_allow_empty_selection"`
 	NativeTypes          bool              `toml:"xpath_native_types"`
-	Trace                bool              `toml:"xpath_trace"`
+	Trace                bool              `toml:"xpath_trace" deprecated:"1.35.0;use 'log_level' 'trace' instead"`
 	Configs              []Config          `toml:"xpath"`
 	DefaultMetricName    string            `toml:"-"`
 	DefaultTags          map[string]string `toml:"-"`
@@ -629,14 +629,14 @@ func (p *Parser) constructFieldName(root, node dataNode, name string, expand boo
 }
 
 func (p *Parser) debugEmptyQuery(operation string, root dataNode, initialquery string) {
-	if p.Log == nil || !p.Trace {
+	if p.Log == nil || !(p.Log.Level().Includes(telegraf.Trace) || p.Trace) { // for backward compatibility
 		return
 	}
 
 	query := initialquery
 
 	// We already know that the
-	p.Log.Debugf("got 0 nodes for query %q in %s", query, operation)
+	p.Log.Tracef("got 0 nodes for query %q in %s", query, operation)
 	for {
 		parts := splitLastPathElement(query)
 		if len(parts) < 1 {
@@ -646,10 +646,10 @@ func (p *Parser) debugEmptyQuery(operation string, root dataNode, initialquery s
 			q := parts[i]
 			nodes, err := p.document.QueryAll(root, q)
 			if err != nil {
-				p.Log.Debugf("executing query %q in %s failed: %v", q, operation, err)
+				p.Log.Tracef("executing query %q in %s failed: %v", q, operation, err)
 				return
 			}
-			p.Log.Debugf("got %d nodes for query %q in %s", len(nodes), q, operation)
+			p.Log.Tracef("got %d nodes for query %q in %s", len(nodes), q, operation)
 			if len(nodes) > 0 && nodes[0] != nil {
 				return
 			}
diff --git a/testutil/capturelog.go b/testutil/capturelog.go
index 58d1cf07fcc3d..4ed50f5fa6ace 100644
--- a/testutil/capturelog.go
+++ b/testutil/capturelog.go
@@ -15,6 +15,7 @@ const (
 	LevelWarn  = 'W'
 	LevelInfo  = 'I'
 	LevelDebug = 'D'
+	LevelTrace = 'T'
 )
 
 type Entry struct {
@@ -65,16 +66,6 @@ func (l *CaptureLogger) Error(args ...interface{}) {
 	l.loga(LevelError, args...)
 }
 
-// Debugf logs a debug message, patterned after log.Printf.
-func (l *CaptureLogger) Debugf(format string, args ...interface{}) {
-	l.logf(LevelDebug, format, args...)
-}
-
-// Debug logs a debug message, patterned after log.Print.
-func (l *CaptureLogger) Debug(args ...interface{}) {
-	l.loga(LevelDebug, args...)
-}
-
 // Warnf logs a warning message, patterned after log.Printf.
 func (l *CaptureLogger) Warnf(format string, args ...interface{}) {
 	l.logf(LevelWarn, format, args...)
@@ -95,6 +86,26 @@ func (l *CaptureLogger) Info(args ...interface{}) {
 	l.loga(LevelInfo, args...)
 }
 
+// Debugf logs a debug message, patterned after log.Printf.
+func (l *CaptureLogger) Debugf(format string, args ...interface{}) {
+	l.logf(LevelDebug, format, args...)
+}
+
+// Debug logs a debug message, patterned after log.Print.
+func (l *CaptureLogger) Debug(args ...interface{}) {
+	l.loga(LevelDebug, args...)
+}
+
+// Tracef logs a trace message, patterned after log.Printf.
+func (l *CaptureLogger) Tracef(format string, args ...interface{}) {
+	l.logf(LevelTrace, format, args...)
+}
+
+// Trace logs a trace message, patterned after log.Print.
+func (l *CaptureLogger) Trace(args ...interface{}) {
+	l.loga(LevelTrace, args...)
+}
+
 func (l *CaptureLogger) NMessages() int {
 	l.Lock()
 	defer l.Unlock()
diff --git a/testutil/log.go b/testutil/log.go
index 45fd37a3d8975..89801676d9c06 100644
--- a/testutil/log.go
+++ b/testutil/log.go
@@ -30,20 +30,6 @@ func (l Logger) Error(args ...interface{}) {
 	log.Print(append([]interface{}{"E! [" + l.Name + "] "}, args...)...)
 }
 
-// Debugf logs a debug message, patterned after log.Printf.
-func (l Logger) Debugf(format string, args ...interface{}) {
-	if !l.Quiet {
-		log.Printf("D! ["+l.Name+"] "+format, args...)
-	}
-}
-
-// Debug logs a debug message, patterned after log.Print.
-func (l Logger) Debug(args ...interface{}) {
-	if !l.Quiet {
-		log.Print(append([]interface{}{"D! [" + l.Name + "] "}, args...)...)
-	}
-}
-
 // Warnf logs a warning message, patterned after log.Printf.
 func (l Logger) Warnf(format string, args ...interface{}) {
 	log.Printf("W! ["+l.Name+"] "+format, args...)
@@ -67,3 +53,31 @@ func (l Logger) Info(args ...interface{}) {
 		log.Print(append([]interface{}{"I! [" + l.Name + "] "}, args...)...)
 	}
 }
+
+// Debugf logs a debug message, patterned after log.Printf.
+func (l Logger) Debugf(format string, args ...interface{}) {
+	if !l.Quiet {
+		log.Printf("D! ["+l.Name+"] "+format, args...)
+	}
+}
+
+// Debug logs a debug message, patterned after log.Print.
+func (l Logger) Debug(args ...interface{}) {
+	if !l.Quiet {
+		log.Print(append([]interface{}{"D! [" + l.Name + "] "}, args...)...)
+	}
+}
+
+// Tracef logs a trace message, patterned after log.Printf.
+func (l Logger) Tracef(format string, args ...interface{}) {
+	if !l.Quiet {
+		log.Printf("T! ["+l.Name+"] "+format, args...)
+	}
+}
+
+// Trace logs a trace message, patterned after log.Print.
+func (l Logger) Trace(args ...interface{}) {
+	if !l.Quiet {
+		log.Print(append([]interface{}{"T! [" + l.Name + "] "}, args...)...)
+	}
+}