diff --git a/go.mod b/go.mod
index b5ae52d1dceb3..34129e89d5667 100644
--- a/go.mod
+++ b/go.mod
@@ -187,6 +187,7 @@ require (
 	github.com/Azure/go-autorest/logger v0.2.1 // indirect
 	github.com/Azure/go-autorest/tracing v0.6.0 // indirect
 	github.com/Azure/go-ntlmssp v0.0.0-20200615164410-66371956d46c // indirect
+	github.com/DATA-DOG/go-sqlmock v1.5.0
 	github.com/Microsoft/go-winio v0.4.17 // indirect
 	github.com/Microsoft/hcsshim v0.8.24 // indirect
 	github.com/alecthomas/participle v0.4.1 // indirect
diff --git a/go.sum b/go.sum
index 4b4441103db15..8e8845ad762a0 100644
--- a/go.sum
+++ b/go.sum
@@ -167,6 +167,8 @@ github.com/ClickHouse/clickhouse-go v1.5.4 h1:cKjXeYLNWVJIx2J1K6H2CqyRmfwVJVY1OV
 github.com/ClickHouse/clickhouse-go v1.5.4/go.mod h1:EaI/sW7Azgz9UATzd5ZdZHRUhHgv5+JMS9NSr2smCJI=
 github.com/DATA-DOG/go-sqlmock v1.3.3/go.mod h1:f/Ixk793poVmq4qj/V1dPUg2JEAKC73Q5eFN3EC/SaM=
 github.com/DATA-DOG/go-sqlmock v1.4.1/go.mod h1:f/Ixk793poVmq4qj/V1dPUg2JEAKC73Q5eFN3EC/SaM=
+github.com/DATA-DOG/go-sqlmock v1.5.0 h1:Shsta01QNfFxHCfpW6YH2STWB0MudeXXEWMr20OEh60=
+github.com/DATA-DOG/go-sqlmock v1.5.0/go.mod h1:f/Ixk793poVmq4qj/V1dPUg2JEAKC73Q5eFN3EC/SaM=
 github.com/DataDog/datadog-go v3.2.0+incompatible/go.mod h1:LButxg5PwREeZtORoXG3tL4fMGNddJ+vMq1mwgfaqoQ=
 github.com/DataDog/sketches-go v0.0.0-20190923095040-43f19ad77ff7/go.mod h1:Q5DbzQ+3AkgGwymQO7aZFNP7ns2lZKGtvRBzRXfdi60=
 github.com/DataDog/zstd v1.3.6-0.20190409195224-796139022798/go.mod h1:1jcaCB/ufaK+sKp1NBhlGmpz41jOoPQ35bpF36t7BBo=
diff --git a/plugins/inputs/mysql/mysql_test.go b/plugins/inputs/mysql/mysql_test.go
index 868c86f18b9cb..772c4c1552978 100644
--- a/plugins/inputs/mysql/mysql_test.go
+++ b/plugins/inputs/mysql/mysql_test.go
@@ -4,6 +4,8 @@ import (
 	"fmt"
 	"testing"
 
+	"github.com/DATA-DOG/go-sqlmock"
+	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 
 	"github.com/influxdata/telegraf/testutil"
@@ -21,6 +23,7 @@ func TestMysqlDefaultsToLocalIntegration(t *testing.T) {
 	var acc testutil.Accumulator
 	err := m.Gather(&acc)
 	require.NoError(t, err)
+	require.Empty(t, acc.Errors)
 
 	require.True(t, acc.HasMeasurement("mysql"))
 }
@@ -42,6 +45,7 @@ func TestMysqlMultipleInstancesIntegration(t *testing.T) {
 	var acc, acc2 testutil.Accumulator
 	err := m.Gather(&acc)
 	require.NoError(t, err)
+	require.Empty(t, acc.Errors)
 	require.True(t, acc.HasMeasurement("mysql"))
 	// acc should have global variables
 	require.True(t, acc.HasMeasurement("mysql_variables"))
@@ -52,6 +56,7 @@ func TestMysqlMultipleInstancesIntegration(t *testing.T) {
 	}
 	err = m2.Gather(&acc2)
 	require.NoError(t, err)
+	require.Empty(t, acc.Errors)
 	require.True(t, acc2.HasMeasurement("mysql"))
 	// acc2 should not have global variables
 	require.False(t, acc2.HasMeasurement("mysql_variables"))
@@ -178,6 +183,200 @@ func TestMysqlDNSAddTimeout(t *testing.T) {
 	}
 }
 
+func TestGatherGlobalVariables(t *testing.T) {
+	db, mock, err := sqlmock.New()
+	require.NoError(t, err)
+	defer db.Close()
+
+	m := Mysql{
+		Log:           testutil.Logger{},
+		MetricVersion: 2,
+	}
+	m.InitMysql()
+
+	columns := []string{"Variable_name", "Value"}
+	measurement := "mysql_variables"
+
+	type fields []*struct {
+		key         string
+		rawValue    string
+		parsedValue interface{}
+		observed    bool
+	}
+	type tags map[string]string
+	testCases := []struct {
+		name   string
+		fields fields
+		tags   tags
+	}{
+		{
+			"basic variables",
+			fields{
+				{"__test__string_variable", "text", "text", false},
+				{"__test__int_variable", "5", int64(5), false},
+				{"__test__off_variable", "OFF", int64(0), false},
+				{"__test__on_variable", "ON", int64(1), false},
+				{"__test__empty_variable", "", nil, false},
+			},
+			tags{"server": "127.0.0.1:3306"},
+		},
+		{
+			"version tag is present",
+			fields{
+				{"__test__string_variable", "text", "text", false},
+				{"version", "8.0.27-0ubuntu0.20.04.1", "8.0.27-0ubuntu0.20.04.1", false},
+			},
+			tags{"server": "127.0.0.1:3306", "version": "8.0.27-0ubuntu0.20.04.1"},
+		},
+
+		{"", fields{{"delay_key_write", "OFF", "OFF", false}}, nil},
+		{"", fields{{"delay_key_write", "ON", "ON", false}}, nil},
+		{"", fields{{"delay_key_write", "ALL", "ALL", false}}, nil},
+		{"", fields{{"enforce_gtid_consistency", "OFF", "OFF", false}}, nil},
+		{"", fields{{"enforce_gtid_consistency", "ON", "ON", false}}, nil},
+		{"", fields{{"enforce_gtid_consistency", "WARN", "WARN", false}}, nil},
+		{"", fields{{"event_scheduler", "NO", "NO", false}}, nil},
+		{"", fields{{"event_scheduler", "YES", "YES", false}}, nil},
+		{"", fields{{"event_scheduler", "DISABLED", "DISABLED", false}}, nil},
+		{"", fields{{"have_ssl", "DISABLED", int64(0), false}}, nil},
+		{"", fields{{"have_ssl", "YES", int64(1), false}}, nil},
+		{"", fields{{"have_symlink", "NO", int64(0), false}}, nil},
+		{"", fields{{"have_symlink", "DISABLED", int64(0), false}}, nil},
+		{"", fields{{"have_symlink", "YES", int64(1), false}}, nil},
+		{"", fields{{"session_track_gtids", "OFF", "OFF", false}}, nil},
+		{"", fields{{"session_track_gtids", "OWN_GTID", "OWN_GTID", false}}, nil},
+		{"", fields{{"session_track_gtids", "ALL_GTIDS", "ALL_GTIDS", false}}, nil},
+		{"", fields{{"session_track_transaction_info", "OFF", "OFF", false}}, nil},
+		{"", fields{{"session_track_transaction_info", "STATE", "STATE", false}}, nil},
+		{"", fields{{"session_track_transaction_info", "CHARACTERISTICS", "CHARACTERISTICS", false}}, nil},
+		{"", fields{{"ssl_fips_mode", "0", "0", false}}, nil}, // TODO: map this to OFF or vice versa using integers
+		{"", fields{{"ssl_fips_mode", "1", "1", false}}, nil}, // TODO: map this to ON or vice versa using integers
+		{"", fields{{"ssl_fips_mode", "2", "2", false}}, nil}, // TODO: map this to STRICT or vice versa using integers
+		{"", fields{{"ssl_fips_mode", "OFF", "OFF", false}}, nil},
+		{"", fields{{"ssl_fips_mode", "ON", "ON", false}}, nil},
+		{"", fields{{"ssl_fips_mode", "STRICT", "STRICT", false}}, nil},
+		{"", fields{{"use_secondary_engine", "OFF", "OFF", false}}, nil},
+		{"", fields{{"use_secondary_engine", "ON", "ON", false}}, nil},
+		{"", fields{{"use_secondary_engine", "FORCED", "FORCED", false}}, nil},
+		{"", fields{{"transaction_write_set_extraction", "OFF", "OFF", false}}, nil},
+		{"", fields{{"transaction_write_set_extraction", "MURMUR32", "MURMUR32", false}}, nil},
+		{"", fields{{"transaction_write_set_extraction", "XXHASH64", "XXHASH64", false}}, nil},
+		{"", fields{{"slave_skip_errors", "OFF", "OFF", false}}, nil},
+		{"", fields{{"slave_skip_errors", "0", "0", false}}, nil},
+		{"", fields{{"slave_skip_errors", "1007,1008,1050", "1007,1008,1050", false}}, nil},
+		{"", fields{{"slave_skip_errors", "all", "all", false}}, nil},
+		{"", fields{{"slave_skip_errors", "ddl_exist_errors", "ddl_exist_errors", false}}, nil},
+		{"", fields{{"gtid_mode", "OFF", int64(0), false}}, nil},
+		{"", fields{{"gtid_mode", "OFF_PERMISSIVE", int64(0), false}}, nil},
+		{"", fields{{"gtid_mode", "ON", int64(1), false}}, nil},
+		{"", fields{{"gtid_mode", "ON_PERMISSIVE", int64(1), false}}, nil},
+
+		// TODO: enable following failing tests by fixing the plugin:
+
+		// {
+		// 	"version tag with many fields", // metrics are sent in packages of 20 fields, so the tags can vary (but should not)
+		// 	fields{
+		// 		{"__test__var01", "text", "text", false},
+		// 		{"__test__var02", "text", "text", false},
+		// 		{"__test__var03", "text", "text", false},
+		// 		{"__test__var04", "text", "text", false},
+		// 		{"__test__var05", "text", "text", false},
+		// 		{"__test__var06", "text", "text", false},
+		// 		{"__test__var07", "text", "text", false},
+		// 		{"__test__var08", "text", "text", false},
+		// 		{"__test__var09", "text", "text", false},
+		// 		{"__test__var10", "text", "text", false},
+		// 		{"__test__var11", "text", "text", false},
+		// 		{"__test__var12", "text", "text", false},
+		// 		{"__test__var13", "text", "text", false},
+		// 		{"__test__var14", "text", "text", false},
+		// 		{"__test__var15", "text", "text", false},
+		// 		{"__test__var16", "text", "text", false},
+		// 		{"__test__var17", "text", "text", false},
+		// 		{"__test__var18", "text", "text", false},
+		// 		{"__test__var19", "text", "text", false},
+		// 		{"__test__var20", "text", "text", false},
+		// 		{"__test__var21", "text", "text", false},
+		// 		{"version", "8.0.27-0ubuntu0.20.04.1", "8.0.27-0ubuntu0.20.04.1", false},
+		// 	},
+		// 	tags{"server": "127.0.0.1:3306", "version": "8.0.27-0ubuntu0.20.04.1"},
+		// },
+		// {
+		// 	"misinterpreted version tags", // e.g. contains("version") also matches "conversion"
+		// 	fields{
+		// 		{"admin_tls_version", "TLSv1,TLSv1.1,TLSv1.2,TLSv1.3", "TLSv1,TLSv1.1,TLSv1.2,TLSv1.3", false},
+		// 		{"innodb_version", "8.0.27", "8.0.27", false},
+		// 		{"protocol_version", "10", "10", false},
+		// 		{"replica_type_conversions", "", "", false},
+		// 		{"server", "127.0.0.1:3306", "127.0.0.1:3306", false},
+		// 		{"slave_type_conversions", "", "", false},
+		// 		{"tls_version", "TLSv1,TLSv1.1,TLSv1.2,TLSv1.3", "TLSv1,TLSv1.1,TLSv1.2,TLSv1.3", false},
+		// 		{"version", "8.0.27-0ubuntu0.20.04.1", "8.0.27-0ubuntu0.20.04.1", false},
+		// 		{"version_comment", "(Ubuntu)", "(Ubuntu)", false},
+		// 		{"version_compile_machine", "x86_64", "x86_64", false},
+		// 		{"version_compile_os", "Linux", "Linux", false},
+		// 		{"version_compile_zlib", "1.2.11", "1.2.11", false},
+		// 	},
+		// 	tags{"server": "127.0.0.1:3306", "version": "8.0.27-0ubuntu0.20.04.1"}, // TODO: Discuss which version tags should be used as tags
+		// },
+	}
+
+	for i, testCase := range testCases {
+		if testCase.name == "" {
+			testCase.name = fmt.Sprintf("#%d", i)
+		}
+
+		// uncomment this to run only a single test case:
+		// if testCase.name != "#1" {
+		// 	continue
+		// }
+
+		rows := sqlmock.NewRows(columns)
+		for _, field := range testCase.fields {
+			rows.AddRow(field.key, field.rawValue)
+		}
+
+		mock.ExpectQuery(globalVariablesQuery).WillReturnRows(rows).RowsWillBeClosed()
+
+		acc := &testutil.Accumulator{}
+
+		err = m.gatherGlobalVariables(db, "test", acc)
+		if !assert.NoErrorf(t, err, "err on gatherGlobalVariables (test case %q)", testCase.name) {
+			continue
+		}
+
+		for _, metric := range acc.Metrics {
+			assert.Equalf(t, measurement, metric.Measurement, "wrong measurement (test case %q)", testCase.name)
+
+			if testCase.tags != nil {
+				assert.Equalf(t, testCase.tags, tags(metric.Tags), "wrong tags (test case %q)", testCase.name)
+			}
+
+			for key, value := range metric.Fields {
+				foundField := false
+
+				for _, field := range testCase.fields {
+					if field.key == key {
+						assert.Falsef(t, field.observed, "field %s observed multiple times (test case %q)", key, testCase.name)
+						assert.Equalf(t, field.parsedValue, value, "wrong value for field %s (test case %q)", key, testCase.name)
+						field.observed = true
+						foundField = true
+						break
+					}
+				}
+
+				if !assert.Truef(t, foundField, "unexpected field %s=%v (test case %q)", key, value, testCase.name) {
+					continue
+				}
+			}
+		}
+
+		for _, field := range testCase.fields {
+			assert.Truef(t, field.observed, "missing field %s=%v (test case %q)", field.key, field.parsedValue, testCase.name)
+		}
+	}
+}
+
 func TestNewNamespace(t *testing.T) {
 	testCases := []struct {
 		words     []string
diff --git a/plugins/inputs/mysql/v2/convert.go b/plugins/inputs/mysql/v2/convert.go
index b446890c9baec..28e99b79e4355 100644
--- a/plugins/inputs/mysql/v2/convert.go
+++ b/plugins/inputs/mysql/v2/convert.go
@@ -60,11 +60,11 @@ func ParseGTIDMode(value sql.RawBytes) (interface{}, error) {
 
 func ParseValue(value sql.RawBytes) (interface{}, error) {
 	if bytes.EqualFold(value, []byte("YES")) || bytes.Equal(value, []byte("ON")) {
-		return 1, nil
+		return int64(1), nil
 	}
 
 	if bytes.EqualFold(value, []byte("NO")) || bytes.Equal(value, []byte("OFF")) {
-		return 0, nil
+		return int64(0), nil
 	}
 
 	if val, err := strconv.ParseInt(string(value), 10, 64); err == nil {
@@ -99,19 +99,28 @@ var GlobalStatusConversions = map[string]ConversionFunc{
 var GlobalVariableConversions = map[string]ConversionFunc{
 	// see https://dev.mysql.com/doc/refman/5.7/en/server-system-variables.html
 	// see https://dev.mysql.com/doc/refman/8.0/en/server-system-variables.html
-	"delay_key_write":                  ParseString, // ON, OFF, ALL
-	"enforce_gtid_consistency":         ParseString, // ON, OFF, WARN
-	"event_scheduler":                  ParseString, // YES, NO, DISABLED
-	"gtid_mode":                        ParseGTIDMode,
-	"have_openssl":                     ParseBoolAsInteger, // alias for have_ssl
-	"have_ssl":                         ParseBoolAsInteger, // YES, DISABLED
-	"have_symlink":                     ParseBoolAsInteger, // YES, NO, DISABLED
-	"session_track_gtids":              ParseString,
-	"session_track_transaction_info":   ParseString,
-	"slave_skip_errors":                ParseString,
-	"ssl_fips_mode":                    ParseString,
+	"delay_key_write":                ParseString,        // ON, OFF, ALL
+	"enforce_gtid_consistency":       ParseString,        // ON, OFF, WARN
+	"event_scheduler":                ParseString,        // YES, NO, DISABLED
+	"have_openssl":                   ParseBoolAsInteger, // alias for have_ssl
+	"have_ssl":                       ParseBoolAsInteger, // YES, DISABLED
+	"have_symlink":                   ParseBoolAsInteger, // YES, NO, DISABLED
+	"session_track_gtids":            ParseString,
+	"session_track_transaction_info": ParseString,
+	"ssl_fips_mode":                  ParseString,
+	"use_secondary_engine":           ParseString,
+
+	// https://dev.mysql.com/doc/refman/5.7/en/replication-options-binary-log.html
+	// https://dev.mysql.com/doc/refman/8.0/en/replication-options-binary-log.html
 	"transaction_write_set_extraction": ParseString,
-	"use_secondary_engine":             ParseString,
+
+	// https://dev.mysql.com/doc/refman/5.7/en/replication-options-replica.html
+	// https://dev.mysql.com/doc/refman/8.0/en/replication-options-replica.html
+	"slave_skip_errors": ParseString,
+
+	// https://dev.mysql.com/doc/refman/5.7/en/replication-options-gtids.html
+	// https://dev.mysql.com/doc/refman/8.0/en/replication-options-gtids.html
+	"gtid_mode": ParseGTIDMode,
 }
 
 func ConvertGlobalStatus(key string, value sql.RawBytes) (interface{}, error) {
diff --git a/plugins/inputs/mysql/v2/convert_test.go b/plugins/inputs/mysql/v2/convert_test.go
index 95083a1e5016f..8276c9a61c28f 100644
--- a/plugins/inputs/mysql/v2/convert_test.go
+++ b/plugins/inputs/mysql/v2/convert_test.go
@@ -95,12 +95,12 @@ func TestParseValue(t *testing.T) {
 		{sql.RawBytes("123"), int64(123), ""},
 		{sql.RawBytes("abc"), "abc", ""},
 		{sql.RawBytes("10.1"), 10.1, ""},
-		{sql.RawBytes("ON"), 1, ""},
-		{sql.RawBytes("OFF"), 0, ""},
-		{sql.RawBytes("NO"), 0, ""},
-		{sql.RawBytes("YES"), 1, ""},
-		{sql.RawBytes("No"), 0, ""},
-		{sql.RawBytes("Yes"), 1, ""},
+		{sql.RawBytes("ON"), int64(1), ""},
+		{sql.RawBytes("OFF"), int64(0), ""},
+		{sql.RawBytes("NO"), int64(0), ""},
+		{sql.RawBytes("YES"), int64(1), ""},
+		{sql.RawBytes("No"), int64(0), ""},
+		{sql.RawBytes("Yes"), int64(1), ""},
 		{sql.RawBytes("-794"), int64(-794), ""},
 		{sql.RawBytes("2147483647"), int64(2147483647), ""},                       // max int32
 		{sql.RawBytes("2147483648"), int64(2147483648), ""},                       // too big for int32