func()

in plugins/inputs/mysql/mysql.go [809:997]

• 158 lines of code
• 30 McCabe index (conditional complexity)

func (m *Mysql) gatherGlobalStatuses(db *sql.DB, serv string, acc telegraf.Accumulator) error {
	// If user forgot the '/', add it
	if strings.HasSuffix(serv, ")") {
		serv = serv + "/"
	} else if serv == "localhost" {
		serv = ""
	}

	// run query
	rows, err := db.Query(globalStatusQuery)
	if err != nil {
		return err
	}

	// parse the DSN and save host name as a tag
	servtag := getDSNTag(serv)
	tags := map[string]string{"server": servtag}
	fields := make(map[string]interface{})
	for rows.Next() {
		var name string
		var val interface{}

		err = rows.Scan(&name, &val)
		if err != nil {
			return err
		}

		var found bool

		// iterate over mappings and gather metrics that is provided on mapping
		for _, mapped := range mappings {
			if strings.HasPrefix(name, mapped.onServer) {
				// convert numeric values to integer
				i, _ := strconv.Atoi(string(val.([]byte)))
				fields[mapped.inExport+name[len(mapped.onServer):]] = i
				found = true
			}
		}
		// Send 20 fields at a time
		if len(fields) >= 20 {
			acc.AddFields("mysql", fields, tags)
			fields = make(map[string]interface{})
		}

		if found {
			continue
		}

		// search for specific values
		switch name {
		case "Queries":
			i, err := strconv.ParseInt(string(val.([]byte)), 10, 64)
			if err != nil {
				return err
			}

			fields["queries"] = i
		case "Questions":
			i, err := strconv.ParseInt(string(val.([]byte)), 10, 64)
			if err != nil {
				return err
			}

			fields["questions"] = i
		case "Slow_queries":
			i, err := strconv.ParseInt(string(val.([]byte)), 10, 64)
			if err != nil {
				return err
			}

			fields["slow_queries"] = i
		case "Connections":
			i, err := strconv.ParseInt(string(val.([]byte)), 10, 64)
			if err != nil {
				return err
			}
			fields["connections"] = i
		case "Syncs":
			i, err := strconv.ParseInt(string(val.([]byte)), 10, 64)
			if err != nil {
				return err
			}
			fields["syncs"] = i
		case "Uptime":
			i, err := strconv.ParseInt(string(val.([]byte)), 10, 64)
			if err != nil {
				return err
			}
			fields["uptime"] = i
		}
	}
	// Send any remaining fields
	if len(fields) > 0 {
		acc.AddFields("mysql", fields, tags)
	}
	// gather connection metrics from processlist for each user
	if m.GatherProcessList {
		conn_rows, err := db.Query("SELECT user, sum(1) FROM INFORMATION_SCHEMA.PROCESSLIST GROUP BY user")

		for conn_rows.Next() {
			var user string
			var connections int64

			err = conn_rows.Scan(&user, &connections)
			if err != nil {
				return err
			}

			tags := map[string]string{"server": servtag, "user": user}
			fields := make(map[string]interface{})

			if err != nil {
				return err
			}
			fields["connections"] = connections
			acc.AddFields("mysql_users", fields, tags)
		}
	}

	// gather connection metrics from user_statistics for each user
	if m.GatherUserStatistics {
		conn_rows, err := db.Query("select user, total_connections, concurrent_connections, connected_time, busy_time, cpu_time, bytes_received, bytes_sent, binlog_bytes_written, rows_fetched, rows_updated, table_rows_read, select_commands, update_commands, other_commands, commit_transactions, rollback_transactions, denied_connections, lost_connections, access_denied, empty_queries, total_ssl_connections FROM INFORMATION_SCHEMA.USER_STATISTICS GROUP BY user")

		for conn_rows.Next() {
			var user string
			var total_connections int64
			var concurrent_connections int64
			var connected_time int64
			var busy_time int64
			var cpu_time int64
			var bytes_received int64
			var bytes_sent int64
			var binlog_bytes_written int64
			var rows_fetched int64
			var rows_updated int64
			var table_rows_read int64
			var select_commands int64
			var update_commands int64
			var other_commands int64
			var commit_transactions int64
			var rollback_transactions int64
			var denied_connections int64
			var lost_connections int64
			var access_denied int64
			var empty_queries int64
			var total_ssl_connections int64

			err = conn_rows.Scan(&user, &total_connections, &concurrent_connections,
				&connected_time, &busy_time, &cpu_time, &bytes_received, &bytes_sent, &binlog_bytes_written,
				&rows_fetched, &rows_updated, &table_rows_read, &select_commands, &update_commands, &other_commands,
				&commit_transactions, &rollback_transactions, &denied_connections, &lost_connections, &access_denied,
				&empty_queries, &total_ssl_connections,
			)

			if err != nil {
				return err
			}

			tags := map[string]string{"server": servtag, "user": user}
			fields := map[string]interface{}{
				"total_connections":      total_connections,
				"concurrent_connections": concurrent_connections,
				"connected_time":         connected_time,
				"busy_time":              busy_time,
				"cpu_time":               cpu_time,
				"bytes_received":         bytes_received,
				"bytes_sent":             bytes_sent,
				"binlog_bytes_written":   binlog_bytes_written,
				"rows_fetched":           rows_fetched,
				"rows_updated":           rows_updated,
				"table_rows_read":        table_rows_read,
				"select_commands":        select_commands,
				"update_commands":        update_commands,
				"other_commands":         other_commands,
				"commit_transactions":    commit_transactions,
				"rollback_transactions":  rollback_transactions,
				"denied_connections":     denied_connections,
				"lost_connections":       lost_connections,
				"access_denied":          access_denied,
				"empty_queries":          empty_queries,
				"total_ssl_connections":  total_ssl_connections,
			}

			acc.AddFields("mysql_user_stats", fields, tags)
		}
	}

	return nil
}