Adding upstream version 7.7.0+dfsg.upstream/7.7.0+dfsg

Signed-off-by: Daniel Baumann <daniel.baumann@progress-linux.org>

7 files changed, 4200 insertions, 0 deletions

diff --git a/ansible_collections/community/mysql/plugins/modules/mysql_db.py b/ansible_collections/community/mysql/plugins/modules/mysql_db.py
new file mode 100644
index 000000000..5a8fe3e3e
--- /dev/null
+++ b/ansible_collections/community/mysql/plugins/modules/mysql_db.py
@@ -0,0 +1,763 @@
+#!/usr/bin/python
+# -*- coding: utf-8 -*-
+
+# Copyright: (c) 2012, Mark Theunissen <mark.theunissen@gmail.com>
+# Sponsored by Four Kitchens http://fourkitchens.com.
+# GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt)
+
+from __future__ import absolute_import, division, print_function
+__metaclass__ = type
+
+DOCUMENTATION = r'''
+---
+module: mysql_db
+short_description: Add or remove MySQL databases from a remote host
+description:
+- Add or remove MySQL databases from a remote host.
+options:
+  name:
+    description:
+    - Name of the database to add or remove.
+    - I(name=all) may only be provided if I(state) is C(dump) or C(import).
+    - List of databases is provided with I(state=dump), I(state=present) and I(state=absent).
+    - If I(name=all) it works like --all-databases option for mysqldump (Added in 2.0).
+    required: true
+    type: list
+    elements: str
+    aliases: [db]
+  state:
+    description:
+    - The database state.
+    type: str
+    default: present
+    choices: ['absent', 'dump', 'import', 'present']
+  collation:
+    description:
+    - Collation mode (sorting). This only applies to new table/databases and
+      does not update existing ones, this is a limitation of MySQL.
+    type: str
+    default: ''
+  encoding:
+    description:
+    - Encoding mode to use, examples include C(utf8) or C(latin1_swedish_ci),
+      at creation of database, dump or importation of sql script.
+    type: str
+    default: ''
+  target:
+    description:
+    - Location, on the remote host, of the dump file to read from or write to.
+    - Uncompressed SQL files (C(.sql)) as well as bzip2 (C(.bz2)), gzip (C(.gz)) and
+      xz (Added in 2.0) compressed files are supported.
+    type: path
+  single_transaction:
+    description:
+    - Execute the dump in a single transaction.
+    type: bool
+    default: false
+  quick:
+    description:
+    - Option used for dumping large tables.
+    type: bool
+    default: true
+  ignore_tables:
+    description:
+    - A list of table names that will be ignored in the dump
+      of the form database_name.table_name.
+    type: list
+    elements: str
+    default: []
+  hex_blob:
+    description:
+    - Dump binary columns using hexadecimal notation.
+    type: bool
+    default: false
+    version_added: '0.1.0'
+  force:
+    description:
+    - Continue dump or import even if we get an SQL error.
+    - Used only when I(state) is C(dump) or C(import).
+    type: bool
+    default: false
+    version_added: '0.1.0'
+  master_data:
+    description:
+      - Option to dump a master replication server to produce a dump file
+        that can be used to set up another server as a slave of the master.
+      - C(0) to not include master data.
+      - C(1) to generate a 'CHANGE MASTER TO' statement
+        required on the slave to start the replication process.
+      - C(2) to generate a commented 'CHANGE MASTER TO'.
+      - Can be used when I(state=dump).
+    type: int
+    choices: [0, 1, 2]
+    default: 0
+    version_added: '0.1.0'
+  skip_lock_tables:
+    description:
+      - Skip locking tables for read. Used when I(state=dump), ignored otherwise.
+    type: bool
+    default: false
+    version_added: '0.1.0'
+  dump_extra_args:
+    description:
+      - Provide additional arguments for mysqldump.
+        Used when I(state=dump) only, ignored otherwise.
+    type: str
+    version_added: '0.1.0'
+  use_shell:
+    description:
+      - Used to prevent C(Broken pipe) errors when the imported I(target) file is compressed.
+      - If C(yes), the module will internally execute commands via a shell.
+      - Used when I(state=import), ignored otherwise.
+    type: bool
+    default: false
+    version_added: '0.1.0'
+  unsafe_login_password:
+    description:
+      - If C(no), the module will safely use a shell-escaped
+        version of the I(login_password) value.
+      - It makes sense to use C(yes) only if there are special
+        symbols in the value and errors C(Access denied) occur.
+      - Used only when I(state) is C(import) or C(dump) and
+        I(login_password) is passed, ignored otherwise.
+    type: bool
+    default: false
+    version_added: '0.1.0'
+  restrict_config_file:
+    description:
+      - Read only passed I(config_file).
+      - When I(state) is C(dump) or C(import),
+        by default the module passes I(config_file) parameter
+        using C(--defaults-extra-file) command-line argument to C(mysql/mysqldump) utilities
+        under the hood that read named option file in addition to usual option files.
+      - If this behavior is undesirable, use C(yes) to read only named option file.
+    type: bool
+    default: false
+    version_added: '0.1.0'
+  check_implicit_admin:
+    description:
+      - Check if mysql allows login as root/nopassword before trying supplied credentials.
+      - If success, passed I(login_user)/I(login_password) will be ignored.
+    type: bool
+    default: false
+    version_added: '0.1.0'
+  config_overrides_defaults:
+    description:
+      - If C(yes), connection parameters from I(config_file) will override the default
+        values of I(login_host) and I(login_port) parameters.
+      - Used when I(stat) is C(present) or C(absent), ignored otherwise.
+      - It needs Python 3.5+ as the default interpreter on a target host.
+    type: bool
+    default: false
+    version_added: '0.1.0'
+  chdir:
+    description:
+    - Changes the current working directory.
+    - Can be useful, for example, when I(state=import) and a dump file contains relative paths.
+    type: path
+    version_added: '3.4.0'
+  pipefail:
+    description:
+    - Use C(bash) instead of C(sh) and add C(-o pipefail) to catch errors from the
+      mysql_dump command when I(state=import) and compression is used.
+    - The default is C(no) to prevent issues on systems without bash as a default interpreter.
+    - The default will change to C(yes) in community.mysql 4.0.0.
+    type: bool
+    default: false
+    version_added: '3.4.0'
+
+seealso:
+- module: community.mysql.mysql_info
+- module: community.mysql.mysql_variables
+- module: community.mysql.mysql_user
+- module: community.mysql.mysql_replication
+- name: MySQL command-line client reference
+  description: Complete reference of the MySQL command-line client documentation.
+  link: https://dev.mysql.com/doc/refman/8.0/en/mysql.html
+- name: mysqldump reference
+  description: Complete reference of the ``mysqldump`` client utility documentation.
+  link: https://dev.mysql.com/doc/refman/8.0/en/mysqldump.html
+- name: CREATE DATABASE reference
+  description: Complete reference of the CREATE DATABASE command documentation.
+  link: https://dev.mysql.com/doc/refman/8.0/en/create-database.html
+- name: DROP DATABASE reference
+  description: Complete reference of the DROP DATABASE command documentation.
+  link: https://dev.mysql.com/doc/refman/8.0/en/drop-database.html
+author: "Ansible Core Team"
+requirements:
+   - mysql (command line binary)
+   - mysqldump (command line binary)
+notes:
+   - Supports C(check_mode).
+   - Requires the mysql and mysqldump binaries on the remote host.
+   - This module is B(not idempotent) when I(state) is C(import),
+     and will import the dump file each time if run more than once.
+extends_documentation_fragment:
+- community.mysql.mysql
+
+'''
+
+EXAMPLES = r'''
+# If you encounter the "Please explicitly state intended protocol" error,
+# use the login_unix_socket argument
+- name: Create a new database with name 'bobdata'
+  community.mysql.mysql_db:
+    name: bobdata
+    state: present
+    login_unix_socket: /run/mysqld/mysqld.sock
+
+- name: Create new databases with names 'foo' and 'bar'
+  community.mysql.mysql_db:
+    name:
+      - foo
+      - bar
+    state: present
+
+# Copy database dump file to remote host and restore it to database 'my_db'
+- name: Copy database dump file
+  copy:
+    src: dump.sql.bz2
+    dest: /tmp
+
+- name: Restore database
+  community.mysql.mysql_db:
+    name: my_db
+    state: import
+    target: /tmp/dump.sql.bz2
+
+- name: Restore database ignoring errors
+  community.mysql.mysql_db:
+    name: my_db
+    state: import
+    target: /tmp/dump.sql.bz2
+    force: true
+
+- name: Dump multiple databases
+  community.mysql.mysql_db:
+    state: dump
+    name: db_1,db_2
+    target: /tmp/dump.sql
+
+- name: Dump multiple databases
+  community.mysql.mysql_db:
+    state: dump
+    name:
+      - db_1
+      - db_2
+    target: /tmp/dump.sql
+
+- name: Dump all databases to hostname.sql
+  community.mysql.mysql_db:
+    state: dump
+    name: all
+    target: /tmp/dump.sql
+
+- name: Dump all databases to hostname.sql including master data
+  community.mysql.mysql_db:
+    state: dump
+    name: all
+    target: /tmp/dump.sql
+    master_data: 1
+
+# Import of sql script with encoding option
+- name: >
+    Import dump.sql with specific latin1 encoding,
+    similar to mysql -u <username> --default-character-set=latin1 -p <password> < dump.sql
+  community.mysql.mysql_db:
+    state: import
+    name: all
+    encoding: latin1
+    target: /tmp/dump.sql
+
+# Dump of database with encoding option
+- name: >
+    Dump of Databse with specific latin1 encoding,
+    similar to mysqldump -u <username> --default-character-set=latin1 -p <password> <database>
+  community.mysql.mysql_db:
+    state: dump
+    name: db_1
+    encoding: latin1
+    target: /tmp/dump.sql
+
+- name: Delete database with name 'bobdata'
+  community.mysql.mysql_db:
+    name: bobdata
+    state: absent
+
+- name: Make sure there is neither a database with name 'foo', nor one with name 'bar'
+  community.mysql.mysql_db:
+    name:
+      - foo
+      - bar
+    state: absent
+
+# Dump database with argument not directly supported by this module
+# using dump_extra_args parameter
+- name: Dump databases without including triggers
+  community.mysql.mysql_db:
+    state: dump
+    name: foo
+    target: /tmp/dump.sql
+    dump_extra_args: --skip-triggers
+
+- name: Try to create database as root/nopassword first. If not allowed, pass the credentials
+  community.mysql.mysql_db:
+    check_implicit_admin: true
+    login_user: bob
+    login_password: 123456
+    name: bobdata
+    state: present
+
+- name: Dump a database with compression and catch errors from mysqldump with bash pipefail
+  community.mysql.mysql_db:
+    state: dump
+    name: foo
+    target: /tmp/dump.sql.gz
+    pipefail: true
+'''
+
+RETURN = r'''
+db:
+  description: Database names in string format delimited by white space.
+  returned: always
+  type: str
+  sample: "foo bar"
+db_list:
+  description: List of database names.
+  returned: always
+  type: list
+  sample: ["foo", "bar"]
+executed_commands:
+  description: List of commands which tried to run.
+  returned: if executed
+  type: list
+  sample: ["CREATE DATABASE acme"]
+  version_added: '0.1.0'
+'''
+
+import os
+import subprocess
+import traceback
+
+from ansible.module_utils.basic import AnsibleModule
+from ansible_collections.community.mysql.plugins.module_utils.database import mysql_quote_identifier
+from ansible_collections.community.mysql.plugins.module_utils.mysql import mysql_connect, mysql_driver, mysql_driver_fail_msg, mysql_common_argument_spec
+from ansible.module_utils.six.moves import shlex_quote
+from ansible.module_utils._text import to_native
+
+executed_commands = []
+
+# ===========================================
+# MySQL module specific support methods.
+#
+
+
+def db_exists(cursor, db):
+    res = 0
+    for each_db in db:
+        res += cursor.execute("SELECT SCHEMA_NAME FROM information_schema.SCHEMATA WHERE SCHEMA_NAME = %s", (each_db,))
+    return res == len(db)
+
+
+def db_delete(cursor, db):
+    if not db:
+        return False
+    for each_db in db:
+        query = "DROP DATABASE %s" % mysql_quote_identifier(each_db, 'database')
+        executed_commands.append(query)
+        cursor.execute(query)
+    return True
+
+
+def db_dump(module, host, user, password, db_name, target, all_databases, port,
+            config_file, socket=None, ssl_cert=None, ssl_key=None, ssl_ca=None,
+            single_transaction=None, quick=None, ignore_tables=None, hex_blob=None,
+            encoding=None, force=False, master_data=0, skip_lock_tables=False,
+            dump_extra_args=None, unsafe_password=False, restrict_config_file=False,
+            check_implicit_admin=False, pipefail=False):
+    cmd = module.get_bin_path('mysqldump', True)
+    # If defined, mysqldump demands --defaults-extra-file be the first option
+    if config_file:
+        if restrict_config_file:
+            cmd += " --defaults-file=%s" % shlex_quote(config_file)
+        else:
+            cmd += " --defaults-extra-file=%s" % shlex_quote(config_file)
+
+    if check_implicit_admin:
+        cmd += " --user=root --password=''"
+    else:
+        if user is not None:
+            cmd += " --user=%s" % shlex_quote(user)
+
+        if password is not None:
+            if not unsafe_password:
+                cmd += " --password=%s" % shlex_quote(password)
+            else:
+                cmd += " --password=%s" % password
+
+    if ssl_cert is not None:
+        cmd += " --ssl-cert=%s" % shlex_quote(ssl_cert)
+    if ssl_key is not None:
+        cmd += " --ssl-key=%s" % shlex_quote(ssl_key)
+    if ssl_ca is not None:
+        cmd += " --ssl-ca=%s" % shlex_quote(ssl_ca)
+    if force:
+        cmd += " --force"
+    if socket is not None:
+        cmd += " --socket=%s" % shlex_quote(socket)
+    else:
+        cmd += " --host=%s --port=%i" % (shlex_quote(host), port)
+
+    if all_databases:
+        cmd += " --all-databases"
+    elif len(db_name) > 1:
+        cmd += " --databases {0}".format(' '.join(db_name))
+    else:
+        cmd += " %s" % shlex_quote(' '.join(db_name))
+
+    if skip_lock_tables:
+        cmd += " --skip-lock-tables"
+    if (encoding is not None) and (encoding != ""):
+        cmd += " --default-character-set=%s" % shlex_quote(encoding)
+    if single_transaction:
+        cmd += " --single-transaction=true"
+    if quick:
+        cmd += " --quick"
+    if ignore_tables:
+        for an_ignored_table in ignore_tables:
+            cmd += " --ignore-table={0}".format(an_ignored_table)
+    if hex_blob:
+        cmd += " --hex-blob"
+    if master_data:
+        cmd += " --master-data=%s" % master_data
+    if dump_extra_args is not None:
+        cmd += " " + dump_extra_args
+
+    path = None
+    if os.path.splitext(target)[-1] == '.gz':
+        path = module.get_bin_path('gzip', True)
+    elif os.path.splitext(target)[-1] == '.bz2':
+        path = module.get_bin_path('bzip2', True)
+    elif os.path.splitext(target)[-1] == '.xz':
+        path = module.get_bin_path('xz', True)
+
+    if path:
+        cmd = '%s | %s > %s' % (cmd, path, shlex_quote(target))
+        if pipefail:
+            cmd = 'set -o pipefail && ' + cmd
+    else:
+        cmd += " > %s" % shlex_quote(target)
+
+    executed_commands.append(cmd)
+
+    if pipefail:
+        rc, stdout, stderr = module.run_command(cmd, use_unsafe_shell=True, executable='bash')
+    else:
+        rc, stdout, stderr = module.run_command(cmd, use_unsafe_shell=True)
+
+    return rc, stdout, stderr
+
+
+def db_import(module, host, user, password, db_name, target, all_databases, port, config_file,
+              socket=None, ssl_cert=None, ssl_key=None, ssl_ca=None, encoding=None, force=False,
+              use_shell=False, unsafe_password=False, restrict_config_file=False,
+              check_implicit_admin=False):
+    if not os.path.exists(target):
+        return module.fail_json(msg="target %s does not exist on the host" % target)
+
+    cmd = [module.get_bin_path('mysql', True)]
+    # --defaults-file must go first, or errors out
+    if config_file:
+        if restrict_config_file:
+            cmd.append("--defaults-file=%s" % shlex_quote(config_file))
+        else:
+            cmd.append("--defaults-extra-file=%s" % shlex_quote(config_file))
+
+    if check_implicit_admin:
+        cmd.append("--user=root --password=''")
+    else:
+        if user:
+            cmd.append("--user=%s" % shlex_quote(user))
+
+        if password:
+            if not unsafe_password:
+                cmd.append("--password=%s" % shlex_quote(password))
+            else:
+                cmd.append("--password=%s" % password)
+
+    if ssl_cert is not None:
+        cmd.append("--ssl-cert=%s" % shlex_quote(ssl_cert))
+    if ssl_key is not None:
+        cmd.append("--ssl-key=%s" % shlex_quote(ssl_key))
+    if ssl_ca is not None:
+        cmd.append("--ssl-ca=%s" % shlex_quote(ssl_ca))
+    if force:
+        cmd.append("-f")
+    if socket is not None:
+        cmd.append("--socket=%s" % shlex_quote(socket))
+    else:
+        cmd.append("--host=%s" % shlex_quote(host))
+        cmd.append("--port=%i" % port)
+    if (encoding is not None) and (encoding != ""):
+        cmd.append("--default-character-set=%s" % shlex_quote(encoding))
+    if not all_databases:
+        cmd.append("--one-database")
+        cmd.append(shlex_quote(''.join(db_name)))
+
+    comp_prog_path = None
+    if os.path.splitext(target)[-1] == '.gz':
+        comp_prog_path = module.get_bin_path('gzip', required=True)
+    elif os.path.splitext(target)[-1] == '.bz2':
+        comp_prog_path = module.get_bin_path('bzip2', required=True)
+    elif os.path.splitext(target)[-1] == '.xz':
+        comp_prog_path = module.get_bin_path('xz', required=True)
+    if comp_prog_path:
+        # The line below is for returned data only:
+        executed_commands.append('%s -dc %s | %s' % (comp_prog_path, target, cmd))
+
+        if not use_shell:
+            p1 = subprocess.Popen([comp_prog_path, '-dc', target], stdout=subprocess.PIPE, stderr=subprocess.PIPE)
+            p2 = subprocess.Popen(cmd, stdin=p1.stdout, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
+            (stdout2, stderr2) = p2.communicate()
+            p1.stdout.close()
+            p1.wait()
+
+            if p1.returncode != 0:
+                stderr1 = p1.stderr.read()
+                return p1.returncode, '', stderr1
+            else:
+                return p2.returncode, stdout2, stderr2
+        else:
+            # Used to prevent 'Broken pipe' errors that
+            # occasionaly occur when target files are compressed.
+            # FYI: passing the `shell=True` argument to p2 = subprocess.Popen()
+            # doesn't solve the problem.
+            cmd = " ".join(cmd)
+            cmd = "%s -dc %s | %s" % (comp_prog_path, shlex_quote(target), cmd)
+            rc, stdout, stderr = module.run_command(cmd, use_unsafe_shell=True)
+            return rc, stdout, stderr
+
+    else:
+        cmd = ' '.join(cmd)
+        cmd += " < %s" % shlex_quote(target)
+        executed_commands.append(cmd)
+        rc, stdout, stderr = module.run_command(cmd, use_unsafe_shell=True)
+        return rc, stdout, stderr
+
+
+def db_create(cursor, db, encoding, collation):
+    if not db:
+        return False
+    query_params = dict(enc=encoding, collate=collation)
+    res = 0
+    for each_db in db:
+        # Escape '%' since mysql cursor.execute() uses a format string
+        query = ['CREATE DATABASE %s' % mysql_quote_identifier(each_db, 'database').replace('%', '%%')]
+        if encoding:
+            query.append("CHARACTER SET %(enc)s")
+        if collation:
+            query.append("COLLATE %(collate)s")
+        query = ' '.join(query)
+        res += cursor.execute(query, query_params)
+        try:
+            executed_commands.append(cursor.mogrify(query, query_params))
+        except AttributeError:
+            executed_commands.append(cursor._executed)
+        except Exception:
+            executed_commands.append(query)
+    return res > 0
+
+
+# ===========================================
+# Module execution.
+#
+
+
+def main():
+    argument_spec = mysql_common_argument_spec()
+    argument_spec.update(
+        name=dict(type='list', required=True, aliases=['db']),
+        encoding=dict(type='str', default=''),
+        collation=dict(type='str', default=''),
+        target=dict(type='path'),
+        state=dict(type='str', default='present', choices=['absent', 'dump', 'import', 'present']),
+        single_transaction=dict(type='bool', default=False),
+        quick=dict(type='bool', default=True),
+        ignore_tables=dict(type='list', default=[]),
+        hex_blob=dict(default=False, type='bool'),
+        force=dict(type='bool', default=False),
+        master_data=dict(type='int', default=0, choices=[0, 1, 2]),
+        skip_lock_tables=dict(type='bool', default=False),
+        dump_extra_args=dict(type='str'),
+        use_shell=dict(type='bool', default=False),
+        unsafe_login_password=dict(type='bool', default=False, no_log=True),
+        restrict_config_file=dict(type='bool', default=False),
+        check_implicit_admin=dict(type='bool', default=False),
+        config_overrides_defaults=dict(type='bool', default=False),
+        chdir=dict(type='path'),
+        pipefail=dict(type='bool', default=False),
+    )
+
+    module = AnsibleModule(
+        argument_spec=argument_spec,
+        supports_check_mode=True,
+    )
+
+    if mysql_driver is None:
+        module.fail_json(msg=mysql_driver_fail_msg)
+
+    db = module.params["name"]
+    if not db:
+        module.exit_json(changed=False, db=db, db_list=[])
+    db = [each_db.strip() for each_db in db]
+
+    encoding = module.params["encoding"]
+    collation = module.params["collation"]
+    state = module.params["state"]
+    target = module.params["target"]
+    socket = module.params["login_unix_socket"]
+    login_port = module.params["login_port"]
+    if login_port < 0 or login_port > 65535:
+        module.fail_json(msg="login_port must be a valid unix port number (0-65535)")
+    ssl_cert = module.params["client_cert"]
+    ssl_key = module.params["client_key"]
+    ssl_ca = module.params["ca_cert"]
+    check_hostname = module.params["check_hostname"]
+    connect_timeout = module.params['connect_timeout']
+    config_file = module.params['config_file']
+    login_password = module.params["login_password"]
+    unsafe_login_password = module.params["unsafe_login_password"]
+    login_user = module.params["login_user"]
+    login_host = module.params["login_host"]
+    ignore_tables = module.params["ignore_tables"]
+    for a_table in ignore_tables:
+        if a_table == "":
+            module.fail_json(msg="Name of ignored table cannot be empty")
+    single_transaction = module.params["single_transaction"]
+    quick = module.params["quick"]
+    hex_blob = module.params["hex_blob"]
+    force = module.params["force"]
+    master_data = module.params["master_data"]
+    skip_lock_tables = module.params["skip_lock_tables"]
+    dump_extra_args = module.params["dump_extra_args"]
+    use_shell = module.params["use_shell"]
+    restrict_config_file = module.params["restrict_config_file"]
+    check_implicit_admin = module.params['check_implicit_admin']
+    config_overrides_defaults = module.params['config_overrides_defaults']
+    chdir = module.params['chdir']
+    pipefail = module.params['pipefail']
+
+    if chdir:
+        try:
+            os.chdir(chdir)
+        except Exception as e:
+            module.fail_json("Cannot change the current directory to %s: %s" % (chdir, e))
+
+    if len(db) > 1 and state == 'import':
+        module.fail_json(msg="Multiple databases are not supported with state=import")
+    db_name = ' '.join(db)
+
+    all_databases = False
+    if state in ['dump', 'import']:
+        if target is None:
+            module.fail_json(msg="with state=%s target is required" % state)
+        if db == ['all']:
+            all_databases = True
+    else:
+        if db == ['all']:
+            module.fail_json(msg="name is not allowed to equal 'all' unless state equals import, or dump.")
+    try:
+        cursor = None
+        if check_implicit_admin:
+            try:
+                cursor, db_conn = mysql_connect(module, 'root', '', config_file, ssl_cert, ssl_key, ssl_ca,
+                                                connect_timeout=connect_timeout, check_hostname=check_hostname,
+                                                config_overrides_defaults=config_overrides_defaults)
+            except Exception as e:
+                check_implicit_admin = False
+                pass
+
+        if not cursor:
+            cursor, db_conn = mysql_connect(module, login_user, login_password, config_file, ssl_cert, ssl_key, ssl_ca,
+                                            connect_timeout=connect_timeout, config_overrides_defaults=config_overrides_defaults,
+                                            check_hostname=check_hostname)
+    except Exception as e:
+        if os.path.exists(config_file):
+            module.fail_json(msg="unable to connect to database, check login_user and login_password are correct or %s has the credentials. "
+                                 "Exception message: %s" % (config_file, to_native(e)))
+        else:
+            module.fail_json(msg="unable to find %s. Exception message: %s" % (config_file, to_native(e)))
+
+    changed = False
+    if not os.path.exists(config_file):
+        config_file = None
+
+    existence_list = []
+    non_existence_list = []
+
+    if not all_databases:
+        for each_database in db:
+            if db_exists(cursor, [each_database]):
+                existence_list.append(each_database)
+            else:
+                non_existence_list.append(each_database)
+
+    if state == "absent":
+        if module.check_mode:
+            module.exit_json(changed=bool(existence_list), db=db_name, db_list=db)
+        try:
+            changed = db_delete(cursor, existence_list)
+        except Exception as e:
+            module.fail_json(msg="error deleting database: %s" % to_native(e))
+        module.exit_json(changed=changed, db=db_name, db_list=db, executed_commands=executed_commands)
+    elif state == "present":
+        if module.check_mode:
+            module.exit_json(changed=bool(non_existence_list), db=db_name, db_list=db)
+        changed = False
+        if non_existence_list:
+            try:
+                changed = db_create(cursor, non_existence_list, encoding, collation)
+            except Exception as e:
+                module.fail_json(msg="error creating database: %s" % to_native(e),
+                                 exception=traceback.format_exc())
+        module.exit_json(changed=changed, db=db_name, db_list=db, executed_commands=executed_commands)
+    elif state == "dump":
+        if non_existence_list and not all_databases:
+            module.fail_json(msg="Cannot dump database(s) %r - not found" % (', '.join(non_existence_list)))
+        if module.check_mode:
+            module.exit_json(changed=True, db=db_name, db_list=db)
+        rc, stdout, stderr = db_dump(module, login_host, login_user,
+                                     login_password, db, target, all_databases,
+                                     login_port, config_file, socket, ssl_cert, ssl_key,
+                                     ssl_ca, single_transaction, quick, ignore_tables,
+                                     hex_blob, encoding, force, master_data, skip_lock_tables,
+                                     dump_extra_args, unsafe_login_password, restrict_config_file,
+                                     check_implicit_admin, pipefail)
+        if rc != 0:
+            module.fail_json(msg="%s" % stderr)
+        module.exit_json(changed=True, db=db_name, db_list=db, msg=stdout,
+                         executed_commands=executed_commands)
+    elif state == "import":
+        if module.check_mode:
+            module.exit_json(changed=True, db=db_name, db_list=db)
+        if non_existence_list and not all_databases:
+            try:
+                db_create(cursor, non_existence_list, encoding, collation)
+            except Exception as e:
+                module.fail_json(msg="error creating database: %s" % to_native(e),
+                                 exception=traceback.format_exc())
+        rc, stdout, stderr = db_import(module, login_host, login_user,
+                                       login_password, db, target,
+                                       all_databases,
+                                       login_port, config_file,
+                                       socket, ssl_cert, ssl_key, ssl_ca,
+                                       encoding, force, use_shell, unsafe_login_password,
+                                       restrict_config_file, check_implicit_admin)
+        if rc != 0:
+            module.fail_json(msg="%s" % stderr)
+        module.exit_json(changed=True, db=db_name, db_list=db, msg=stdout,
+                         executed_commands=executed_commands)
+
+
+if __name__ == '__main__':
+    main()
diff --git a/ansible_collections/community/mysql/plugins/modules/mysql_info.py b/ansible_collections/community/mysql/plugins/modules/mysql_info.py
new file mode 100644
index 000000000..11b1a8003
--- /dev/null
+++ b/ansible_collections/community/mysql/plugins/modules/mysql_info.py
@@ -0,0 +1,605 @@
+#!/usr/bin/python
+# -*- coding: utf-8 -*-
+
+# Copyright: (c) 2019, Andrew Klychkov (@Andersson007) <aaklychkov@mail.ru>
+# GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt)
+
+from __future__ import absolute_import, division, print_function
+__metaclass__ = type
+
+DOCUMENTATION = r'''
+---
+module: mysql_info
+short_description: Gather information about MySQL servers
+description:
+- Gathers information about MySQL servers.
+
+options:
+  filter:
+    description:
+    - Limit the collected information by comma separated string or YAML list.
+    - Allowable values are C(version), C(databases), C(settings), C(global_status),
+      C(users), C(engines), C(master_status), C(slave_status), C(slave_hosts).
+    - By default, collects all subsets.
+    - You can use '!' before value (for example, C(!settings)) to exclude it from the information.
+    - If you pass including and excluding values to the filter, for example, I(filter=!settings,version),
+      the excluding values, C(!settings) in this case, will be ignored.
+    type: list
+    elements: str
+  login_db:
+    description:
+    - Database name to connect to.
+    - It makes sense if I(login_user) is allowed to connect to a specific database only.
+    type: str
+  exclude_fields:
+    description:
+    - List of fields which are not needed to collect.
+    - "Supports elements: C(db_size). Unsupported elements will be ignored."
+    type: list
+    elements: str
+    version_added: '0.1.0'
+  return_empty_dbs:
+    description:
+    - Includes names of empty databases to returned dictionary.
+    type: bool
+    default: false
+
+notes:
+- Calculating the size of a database might be slow, depending on the number and size of tables in it.
+  To avoid this, use I(exclude_fields=db_size).
+- Supports C(check_mode).
+
+seealso:
+- module: community.mysql.mysql_variables
+- module: community.mysql.mysql_db
+- module: community.mysql.mysql_user
+- module: community.mysql.mysql_replication
+
+author:
+- Andrew Klychkov (@Andersson007)
+- Sebastian Gumprich (@rndmh3ro)
+- Laurent Indermühle (@laurent-indermuehle)
+
+extends_documentation_fragment:
+- community.mysql.mysql
+'''
+
+EXAMPLES = r'''
+# Display info from mysql-hosts group (using creds from ~/.my.cnf to connect):
+# ansible mysql-hosts -m mysql_info
+
+# Display only databases and users info:
+# ansible mysql-hosts -m mysql_info -a 'filter=databases,users'
+
+# Display only slave status:
+# ansible standby -m mysql_info -a 'filter=slave_status'
+
+# Display all info from databases group except settings:
+# ansible databases -m mysql_info -a 'filter=!settings'
+
+# If you encounter the "Please explicitly state intended protocol" error,
+# use the login_unix_socket argument
+- name: Collect all possible information using passwordless root access
+  community.mysql.mysql_info:
+    login_user: root
+    login_unix_socket: /run/mysqld/mysqld.sock
+
+- name: Get MySQL version with non-default credentials
+  community.mysql.mysql_info:
+    login_user: mysuperuser
+    login_password: mysuperpass
+    filter: version
+
+- name: Collect all info except settings and users by root
+  community.mysql.mysql_info:
+    login_user: root
+    login_password: rootpass
+    filter: "!settings,!users"
+
+- name: Collect info about databases and version using ~/.my.cnf as a credential file
+  become: true
+  community.mysql.mysql_info:
+    filter:
+    - databases
+    - version
+
+- name: Collect info about databases and version using ~alice/.my.cnf as a credential file
+  become: true
+  community.mysql.mysql_info:
+    config_file: /home/alice/.my.cnf
+    filter:
+    - databases
+    - version
+
+- name: Collect info about databases including empty and excluding their sizes
+  become: true
+  community.mysql.mysql_info:
+    config_file: /home/alice/.my.cnf
+    filter:
+    - databases
+    exclude_fields: db_size
+    return_empty_dbs: true
+'''
+
+RETURN = r'''
+version:
+  description: Database server version.
+  returned: if not excluded by filter
+  type: dict
+  sample: { "version": { "major": 5, "minor": 5, "release": 60, "suffix": "MariaDB", "full": "5.5.60-MariaDB" } }
+  contains:
+    major:
+      description: Major server version.
+      returned: if not excluded by filter
+      type: int
+      sample: 5
+    minor:
+      description: Minor server version.
+      returned: if not excluded by filter
+      type: int
+      sample: 5
+    release:
+      description: Release server version.
+      returned: if not excluded by filter
+      type: int
+      sample: 60
+    suffix:
+      description: Server suffix, for example MySQL, MariaDB, other or none.
+      returned: if not excluded by filter
+      type: str
+      sample: "MariaDB"
+    full:
+      description: Full server version.
+      returned: if not excluded by filter
+      type: str
+      sample: "5.5.60-MariaDB"
+databases:
+  description: Information about databases.
+  returned: if not excluded by filter
+  type: dict
+  sample:
+  - { "mysql": { "size": 656594 }, "information_schema": { "size": 73728 } }
+  contains:
+    size:
+      description: Database size in bytes.
+      returned: if not excluded by filter
+      type: dict
+      sample: { 'size': 656594 }
+settings:
+  description: Global settings (variables) information.
+  returned: if not excluded by filter
+  type: dict
+  sample:
+  - { "innodb_open_files": 300, innodb_page_size": 16384 }
+global_status:
+  description: Global status information.
+  returned: if not excluded by filter
+  type: dict
+  sample:
+  - { "Innodb_buffer_pool_read_requests": 123, "Innodb_buffer_pool_reads": 32 }
+users:
+  description: Users information.
+  returned: if not excluded by filter
+  type: dict
+  sample:
+  - { "localhost": { "root": { "Alter_priv": "Y", "Alter_routine_priv": "Y" } } }
+engines:
+  description: Information about the server's storage engines.
+  returned: if not excluded by filter
+  type: dict
+  sample:
+  - { "CSV": { "Comment": "CSV storage engine", "Savepoints": "NO", "Support": "YES", "Transactions": "NO", "XA": "NO" } }
+master_status:
+  description: Master status information.
+  returned: if master
+  type: dict
+  sample:
+  - { "Binlog_Do_DB": "", "Binlog_Ignore_DB": "mysql", "File": "mysql-bin.000001", "Position": 769 }
+slave_status:
+  description: Slave status information.
+  returned: if standby
+  type: dict
+  sample:
+  - { "192.168.1.101": { "3306": { "replication_user": { "Connect_Retry": 60, "Exec_Master_Log_Pos": 769,  "Last_Errno": 0 } } } }
+slave_hosts:
+  description: Slave status information.
+  returned: if master
+  type: dict
+  sample:
+  - { "2": { "Host": "", "Master_id": 1, "Port": 3306 } }
+connector_name:
+  description: Name of the python connector used by the module. When the connector is not identified, returns C(Unknown).
+  returned: always
+  type: str
+  sample:
+  - "pymysql"
+  - "MySQLdb"
+  version_added: '3.6.0'
+connector_version:
+  description: Version of the python connector used by the module. When the connector is not identified, returns C(Unknown).
+  returned: always
+  type: str
+  sample:
+  - "1.0.2"
+  version_added: '3.6.0'
+'''
+
+from decimal import Decimal
+
+from ansible.module_utils.basic import AnsibleModule
+from ansible_collections.community.mysql.plugins.module_utils.mysql import (
+    mysql_connect,
+    mysql_common_argument_spec,
+    mysql_driver,
+    mysql_driver_fail_msg,
+    get_connector_name,
+    get_connector_version,
+)
+from ansible.module_utils.six import iteritems
+from ansible.module_utils._text import to_native
+
+
+# ===========================================
+# MySQL module specific support methods.
+#
+
+class MySQL_Info(object):
+
+    """Class for collection MySQL instance information.
+
+    Arguments:
+        module (AnsibleModule): Object of AnsibleModule class.
+        cursor (pymysql/mysql-python): Cursor class for interaction with
+            the database.
+
+    Note:
+        If you need to add a new subset:
+        1. add a new key with the same name to self.info attr in self.__init__()
+        2. add a new private method to get the information
+        3. add invocation of the new method to self.__collect()
+        4. add info about the new subset to the DOCUMENTATION block
+        5. add info about the new subset with an example to RETURN block
+    """
+
+    def __init__(self, module, cursor):
+        self.module = module
+        self.cursor = cursor
+        self.info = {
+            'version': {},
+            'databases': {},
+            'settings': {},
+            'global_status': {},
+            'engines': {},
+            'users': {},
+            'master_status': {},
+            'slave_hosts': {},
+            'slave_status': {},
+        }
+
+    def get_info(self, filter_, exclude_fields, return_empty_dbs):
+        """Get MySQL instance information based on filter_.
+
+        Arguments:
+            filter_ (list): List of collected subsets (e.g., databases, users, etc.),
+                when it is empty, return all available information.
+        """
+
+        inc_list = []
+        exc_list = []
+
+        if filter_:
+            partial_info = {}
+
+            for fi in filter_:
+                if fi.lstrip('!') not in self.info:
+                    self.module.warn('filter element: %s is not allowable, ignored' % fi)
+                    continue
+
+                if fi[0] == '!':
+                    exc_list.append(fi.lstrip('!'))
+
+                else:
+                    inc_list.append(fi)
+
+            if inc_list:
+                self.__collect(exclude_fields, return_empty_dbs, set(inc_list))
+
+                for i in self.info:
+                    if i in inc_list:
+                        partial_info[i] = self.info[i]
+
+            else:
+                not_in_exc_list = list(set(self.info) - set(exc_list))
+                self.__collect(exclude_fields, return_empty_dbs, set(not_in_exc_list))
+
+                for i in self.info:
+                    if i not in exc_list:
+                        partial_info[i] = self.info[i]
+
+            return partial_info
+
+        else:
+            self.__collect(exclude_fields, return_empty_dbs, set(self.info))
+            return self.info
+
+    def __collect(self, exclude_fields, return_empty_dbs, wanted):
+        """Collect all possible subsets."""
+        if 'version' in wanted or 'settings' in wanted:
+            self.__get_global_variables()
+
+        if 'databases' in wanted:
+            self.__get_databases(exclude_fields, return_empty_dbs)
+
+        if 'global_status' in wanted:
+            self.__get_global_status()
+
+        if 'engines' in wanted:
+            self.__get_engines()
+
+        if 'users' in wanted:
+            self.__get_users()
+
+        if 'master_status' in wanted:
+            self.__get_master_status()
+
+        if 'slave_status' in wanted:
+            self.__get_slave_status()
+
+        if 'slave_hosts' in wanted:
+            self.__get_slaves()
+
+    def __get_engines(self):
+        """Get storage engines info."""
+        res = self.__exec_sql('SHOW ENGINES')
+
+        if res:
+            for line in res:
+                engine = line['Engine']
+                self.info['engines'][engine] = {}
+
+                for vname, val in iteritems(line):
+                    if vname != 'Engine':
+                        self.info['engines'][engine][vname] = val
+
+    def __convert(self, val):
+        """Convert unserializable data."""
+        try:
+            if isinstance(val, Decimal):
+                val = float(val)
+            else:
+                val = int(val)
+
+        except ValueError:
+            pass
+
+        except TypeError:
+            pass
+
+        return val
+
+    def __get_global_variables(self):
+        """Get global variables (instance settings)."""
+        res = self.__exec_sql('SHOW GLOBAL VARIABLES')
+
+        if res:
+            for var in res:
+                self.info['settings'][var['Variable_name']] = self.__convert(var['Value'])
+
+            # version = ["5", "5," "60-MariaDB]
+            version = self.info['settings']['version'].split('.')
+
+            # full_version = "5.5.60-MariaDB"
+            full = self.info['settings']['version']
+
+            # release = "60"
+            release = version[2].split('-')[0]
+
+            # check if a suffix exists by counting the length
+            if len(version[2].split('-')) > 1:
+                # suffix = "MariaDB"
+                suffix = version[2].split('-', 1)[1]
+            else:
+                suffix = ""
+
+            self.info['version'] = dict(
+                # major = "5"
+                major=int(version[0]),
+                # minor = "5"
+                minor=int(version[1]),
+                release=int(release),
+                suffix=str(suffix),
+                full=str(full),
+            )
+
+    def __get_global_status(self):
+        """Get global status."""
+        res = self.__exec_sql('SHOW GLOBAL STATUS')
+
+        if res:
+            for var in res:
+                self.info['global_status'][var['Variable_name']] = self.__convert(var['Value'])
+
+    def __get_master_status(self):
+        """Get master status if the instance is a master."""
+        res = self.__exec_sql('SHOW MASTER STATUS')
+        if res:
+            for line in res:
+                for vname, val in iteritems(line):
+                    self.info['master_status'][vname] = self.__convert(val)
+
+    def __get_slave_status(self):
+        """Get slave status if the instance is a slave."""
+        res = self.__exec_sql('SHOW SLAVE STATUS')
+        if res:
+            for line in res:
+                host = line['Master_Host']
+                if host not in self.info['slave_status']:
+                    self.info['slave_status'][host] = {}
+
+                port = line['Master_Port']
+                if port not in self.info['slave_status'][host]:
+                    self.info['slave_status'][host][port] = {}
+
+                user = line['Master_User']
+                if user not in self.info['slave_status'][host][port]:
+                    self.info['slave_status'][host][port][user] = {}
+
+                for vname, val in iteritems(line):
+                    if vname not in ('Master_Host', 'Master_Port', 'Master_User'):
+                        self.info['slave_status'][host][port][user][vname] = self.__convert(val)
+
+    def __get_slaves(self):
+        """Get slave hosts info if the instance is a master."""
+        res = self.__exec_sql('SHOW SLAVE HOSTS')
+        if res:
+            for line in res:
+                srv_id = line['Server_id']
+                if srv_id not in self.info['slave_hosts']:
+                    self.info['slave_hosts'][srv_id] = {}
+
+                for vname, val in iteritems(line):
+                    if vname != 'Server_id':
+                        self.info['slave_hosts'][srv_id][vname] = self.__convert(val)
+
+    def __get_users(self):
+        """Get user info."""
+        res = self.__exec_sql('SELECT * FROM mysql.user')
+        if res:
+            for line in res:
+                host = line['Host']
+                if host not in self.info['users']:
+                    self.info['users'][host] = {}
+
+                user = line['User']
+                self.info['users'][host][user] = {}
+
+                for vname, val in iteritems(line):
+                    if vname not in ('Host', 'User'):
+                        self.info['users'][host][user][vname] = self.__convert(val)
+
+    def __get_databases(self, exclude_fields, return_empty_dbs):
+        """Get info about databases."""
+        if not exclude_fields:
+            query = ('SELECT table_schema AS "name", '
+                     'SUM(data_length + index_length) AS "size" '
+                     'FROM information_schema.TABLES GROUP BY table_schema')
+        else:
+            if 'db_size' in exclude_fields:
+                query = ('SELECT table_schema AS "name" '
+                         'FROM information_schema.TABLES GROUP BY table_schema')
+
+        res = self.__exec_sql(query)
+
+        if res:
+            for db in res:
+                self.info['databases'][db['name']] = {}
+
+                if not exclude_fields or 'db_size' not in exclude_fields:
+                    if db['size'] is None:
+                        db['size'] = 0
+
+                    self.info['databases'][db['name']]['size'] = int(db['size'])
+
+        # If empty dbs are not needed in the returned dict, exit from the method
+        if not return_empty_dbs:
+            return None
+
+        # Add info about empty databases (issue #65727):
+        res = self.__exec_sql('SHOW DATABASES')
+        if res:
+            for db in res:
+                if db['Database'] not in self.info['databases']:
+                    self.info['databases'][db['Database']] = {}
+
+                    if not exclude_fields or 'db_size' not in exclude_fields:
+                        self.info['databases'][db['Database']]['size'] = 0
+
+    def __exec_sql(self, query, ddl=False):
+        """Execute SQL.
+
+        Arguments:
+            ddl (bool): If True, return True or False.
+                Used for queries that don't return any rows
+                (mainly for DDL queries) (default False).
+        """
+        try:
+            self.cursor.execute(query)
+
+            if not ddl:
+                res = self.cursor.fetchall()
+                return res
+            return True
+
+        except Exception as e:
+            self.module.fail_json(msg="Cannot execute SQL '%s': %s" % (query, to_native(e)))
+        return False
+
+
+# ===========================================
+# Module execution.
+#
+
+
+def main():
+    argument_spec = mysql_common_argument_spec()
+    argument_spec.update(
+        login_db=dict(type='str'),
+        filter=dict(type='list'),
+        exclude_fields=dict(type='list'),
+        return_empty_dbs=dict(type='bool', default=False),
+    )
+
+    module = AnsibleModule(
+        argument_spec=argument_spec,
+        supports_check_mode=True,
+    )
+
+    db = module.params['login_db']
+    connect_timeout = module.params['connect_timeout']
+    login_user = module.params['login_user']
+    login_password = module.params['login_password']
+    ssl_cert = module.params['client_cert']
+    ssl_key = module.params['client_key']
+    ssl_ca = module.params['ca_cert']
+    check_hostname = module.params['check_hostname']
+    config_file = module.params['config_file']
+    filter_ = module.params['filter']
+    exclude_fields = module.params['exclude_fields']
+    return_empty_dbs = module.params['return_empty_dbs']
+
+    if filter_:
+        filter_ = [f.strip() for f in filter_]
+
+    if exclude_fields:
+        exclude_fields = set([f.strip() for f in exclude_fields])
+
+    if mysql_driver is None:
+        module.fail_json(msg=mysql_driver_fail_msg)
+
+    connector_name = get_connector_name(mysql_driver)
+    connector_version = get_connector_version(mysql_driver)
+
+    try:
+        cursor, db_conn = mysql_connect(module, login_user, login_password,
+                                        config_file, ssl_cert, ssl_key, ssl_ca, db,
+                                        check_hostname=check_hostname,
+                                        connect_timeout=connect_timeout, cursor_class='DictCursor')
+    except Exception as e:
+        msg = ('unable to connect to database using %s %s, check login_user '
+               'and login_password are correct or %s has the credentials. '
+               'Exception message: %s' % (connector_name, connector_version, config_file, to_native(e)))
+        module.fail_json(msg)
+
+    ###############################
+    # Create object and do main job
+
+    mysql = MySQL_Info(module, cursor)
+
+    module.exit_json(changed=False,
+                     connector_name=connector_name,
+                     connector_version=connector_version,
+                     **mysql.get_info(filter_, exclude_fields, return_empty_dbs))
+
+
+if __name__ == '__main__':
+    main()
diff --git a/ansible_collections/community/mysql/plugins/modules/mysql_query.py b/ansible_collections/community/mysql/plugins/modules/mysql_query.py
new file mode 100644
index 000000000..12d5a5630
--- /dev/null
+++ b/ansible_collections/community/mysql/plugins/modules/mysql_query.py
@@ -0,0 +1,284 @@
+#!/usr/bin/python
+# -*- coding: utf-8 -*-
+
+# Copyright: (c) 2020, Andrew Klychkov (@Andersson007) <aaklychkov@mail.ru>
+# GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt)
+
+from __future__ import (absolute_import, division, print_function)
+__metaclass__ = type
+
+DOCUMENTATION = r'''
+---
+module: mysql_query
+short_description: Run MySQL queries
+description:
+- Runs arbitrary MySQL queries.
+- Pay attention, the module does not support check mode!
+  All queries will be executed in autocommit mode.
+- To run SQL queries from a file, use M(community.mysql.mysql_db) module.
+version_added: '0.1.0'
+options:
+  query:
+    description:
+    - SQL query to run. Multiple queries can be passed using YAML list syntax.
+    - Must be a string or YAML list containing strings.
+    - If you use I(named_args) or I(positional_args) any C(%) will be interpreted
+      as a formatting character. All literal C(%) characters in the query should be
+      escaped as C(%%).
+    - Note that if you use the C(IF EXISTS/IF NOT EXISTS) clauses in your query
+      and C(mysqlclient) or C(PyMySQL 0.10.0+) connectors, the module will report
+      that the state has been changed even if it has not. If it is important in your
+      workflow, use the C(PyMySQL 0.9.3) connector instead.
+    type: raw
+    required: true
+  positional_args:
+    description:
+    - List of values to be passed as positional arguments to the query.
+    - Mutually exclusive with I(named_args).
+    type: list
+  named_args:
+    description:
+    - Dictionary of key-value arguments to pass to the query.
+    - Mutually exclusive with I(positional_args).
+    type: dict
+  login_db:
+    description:
+    - Name of database to connect to and run queries against.
+    type: str
+  single_transaction:
+    description:
+    - Where passed queries run in a single transaction (C(yes)) or commit them one-by-one (C(no)).
+    type: bool
+    default: false
+seealso:
+- module: community.mysql.mysql_db
+author:
+- Andrew Klychkov (@Andersson007)
+extends_documentation_fragment:
+- community.mysql.mysql
+
+'''
+
+EXAMPLES = r'''
+# If you encounter the "Please explicitly state intended protocol" error,
+# use the login_unix_socket argument
+- name: Simple select query to acme db
+  community.mysql.mysql_query:
+    login_db: acme
+    query: SELECT * FROM orders
+    login_unix_socket: /run/mysqld/mysqld.sock
+
+- name: Select query to db acme with positional arguments
+  community.mysql.mysql_query:
+    login_db: acme
+    query: SELECT * FROM acme WHERE id = %s AND story = %s
+    positional_args:
+    - 1
+    - test
+
+- name: Select query to test_db with named_args
+  community.mysql.mysql_query:
+    login_db: test_db
+    query: SELECT * FROM test WHERE id = %(id_val)s AND story = %(story_val)s
+    named_args:
+      id_val: 1
+      story_val: test
+
+- name: Run several insert queries against db test_db in single transaction
+  community.mysql.mysql_query:
+    login_db: test_db
+    query:
+    - INSERT INTO articles (id, story) VALUES (2, 'my_long_story')
+    - INSERT INTO prices (id, price) VALUES (123, '100.00')
+    single_transaction: true
+'''
+
+RETURN = r'''
+executed_queries:
+    description: List of executed queries.
+    returned: always
+    type: list
+    sample: ['SELECT * FROM bar', 'UPDATE bar SET id = 1 WHERE id = 2']
+query_result:
+    description:
+    - List of lists (sublist for each query) containing dictionaries
+      in column:value form representing returned rows.
+    returned: changed
+    type: list
+    sample: [[{"Column": "Value1"},{"Column": "Value2"}], [{"ID": 1}, {"ID": 2}]]
+rowcount:
+    description: Number of affected rows for each subquery.
+    returned: changed
+    type: list
+    sample: [5, 1]
+'''
+
+import warnings
+
+from ansible.module_utils.basic import AnsibleModule
+from ansible_collections.community.mysql.plugins.module_utils.mysql import (
+    mysql_connect,
+    mysql_common_argument_spec,
+    mysql_driver,
+    mysql_driver_fail_msg,
+)
+from ansible.module_utils._text import to_native
+
+DML_QUERY_KEYWORDS = ('INSERT', 'UPDATE', 'DELETE', 'REPLACE')
+# TRUNCATE is not DDL query but it also returns 0 rows affected:
+DDL_QUERY_KEYWORDS = ('CREATE', 'DROP', 'ALTER', 'RENAME', 'TRUNCATE')
+
+
+# ===========================================
+# Module execution.
+#
+
+def main():
+    argument_spec = mysql_common_argument_spec()
+    argument_spec.update(
+        query=dict(type='raw', required=True),
+        login_db=dict(type='str'),
+        positional_args=dict(type='list'),
+        named_args=dict(type='dict'),
+        single_transaction=dict(type='bool', default=False),
+    )
+
+    module = AnsibleModule(
+        argument_spec=argument_spec,
+        mutually_exclusive=(
+            ('positional_args', 'named_args'),
+        ),
+    )
+
+    db = module.params['login_db']
+    connect_timeout = module.params['connect_timeout']
+    login_user = module.params['login_user']
+    login_password = module.params['login_password']
+    ssl_cert = module.params['client_cert']
+    ssl_key = module.params['client_key']
+    ssl_ca = module.params['ca_cert']
+    check_hostname = module.params['check_hostname']
+    config_file = module.params['config_file']
+    query = module.params["query"]
+
+    if not isinstance(query, (str, list)):
+        module.fail_json(msg="the query option value must be a string or list, passed %s" % type(query))
+
+    if isinstance(query, str):
+        query = [query]
+
+    for elem in query:
+        if not isinstance(elem, str):
+            module.fail_json(msg="the elements in query list must be strings, passed '%s' %s" % (elem, type(elem)))
+
+    if module.params["single_transaction"]:
+        autocommit = False
+    else:
+        autocommit = True
+    # Prepare args:
+    if module.params.get("positional_args"):
+        arguments = module.params["positional_args"]
+    elif module.params.get("named_args"):
+        arguments = module.params["named_args"]
+    else:
+        arguments = None
+
+    if mysql_driver is None:
+        module.fail_json(msg=mysql_driver_fail_msg)
+
+    # Connect to DB:
+    try:
+        cursor, db_connection = mysql_connect(module, login_user, login_password,
+                                              config_file, ssl_cert, ssl_key, ssl_ca, db,
+                                              check_hostname=check_hostname,
+                                              connect_timeout=connect_timeout,
+                                              cursor_class='DictCursor', autocommit=autocommit)
+    except Exception as e:
+        module.fail_json(msg="unable to connect to database, check login_user and "
+                             "login_password are correct or %s has the credentials. "
+                             "Exception message: %s" % (config_file, to_native(e)))
+
+    # Set defaults:
+    changed = False
+
+    max_keyword_len = len(max(DML_QUERY_KEYWORDS + DDL_QUERY_KEYWORDS, key=len))
+
+    # Execute query:
+    query_result = []
+    executed_queries = []
+    rowcount = []
+
+    already_exists = False
+    for q in query:
+        try:
+            with warnings.catch_warnings():
+                warnings.filterwarnings(action='error',
+                                        message='.*already exists*',
+                                        category=mysql_driver.Warning)
+
+                try:
+                    cursor.execute(q, arguments)
+                except mysql_driver.Warning:
+                    # When something is run with IF NOT EXISTS
+                    # and there's "already exists" MySQL warning,
+                    # set the flag as True.
+                    # PyMySQL < 0.10.0 throws the warning, mysqlclient
+                    # and PyMySQL 0.10.0+ does NOT.
+                    already_exists = True
+
+        except Exception as e:
+            if not autocommit:
+                db_connection.rollback()
+
+            cursor.close()
+            module.fail_json(msg="Cannot execute SQL '%s' args [%s]: %s" % (q, arguments, to_native(e)))
+
+        try:
+            if not already_exists:
+                query_result.append([dict(row) for row in cursor.fetchall()])
+
+        except Exception as e:
+            if not autocommit:
+                db_connection.rollback()
+
+            module.fail_json(msg="Cannot fetch rows from cursor: %s" % to_native(e))
+
+        # Check DML or DDL keywords in query and set changed accordingly:
+        q = q.lstrip()[0:max_keyword_len].upper()
+        for keyword in DML_QUERY_KEYWORDS:
+            if keyword in q and cursor.rowcount > 0:
+                changed = True
+
+        for keyword in DDL_QUERY_KEYWORDS:
+            if keyword in q:
+                if already_exists:
+                    # Indicates the entity already exists
+                    changed = False
+                    already_exists = False  # Reset flag
+                else:
+                    changed = True
+        try:
+            executed_queries.append(cursor._last_executed)
+        except AttributeError:
+            # MySQLdb removed cursor._last_executed as a duplicate of cursor._executed
+            executed_queries.append(cursor._executed)
+        rowcount.append(cursor.rowcount)
+
+    # When the module run with the single_transaction == True:
+    if not autocommit:
+        db_connection.commit()
+
+    # Create dict with returned values:
+    kw = {
+        'changed': changed,
+        'executed_queries': executed_queries,
+        'query_result': query_result,
+        'rowcount': rowcount,
+    }
+
+    # Exit:
+    module.exit_json(**kw)
+
+
+if __name__ == '__main__':
+    main()
diff --git a/ansible_collections/community/mysql/plugins/modules/mysql_replication.py b/ansible_collections/community/mysql/plugins/modules/mysql_replication.py
new file mode 100644
index 000000000..33e14bc26
--- /dev/null
+++ b/ansible_collections/community/mysql/plugins/modules/mysql_replication.py
@@ -0,0 +1,654 @@
+#!/usr/bin/python
+# -*- coding: utf-8 -*-
+
+# Copyright: (c) 2013, Balazs Pocze <banyek@gawker.com>
+# Copyright: (c) 2019, Andrew Klychkov (@Andersson007) <aaklychkov@mail.ru>
+# Certain parts are taken from Mark Theunissen's mysqldb module
+# GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt)
+
+from __future__ import absolute_import, division, print_function
+__metaclass__ = type
+
+
+DOCUMENTATION = r'''
+---
+module: mysql_replication
+short_description: Manage MySQL replication
+description:
+- Manages MySQL server replication, replica, primary status, get and change primary host.
+author:
+- Balazs Pocze (@banyek)
+- Andrew Klychkov (@Andersson007)
+options:
+  mode:
+    description:
+    - Module operating mode. Could be
+      C(changeprimary) (CHANGE PRIMARY TO),
+      C(getprimary) (SHOW PRIMARY STATUS),
+      C(getreplica) (SHOW REPLICA),
+      C(startreplica) (START REPLICA),
+      C(stopreplica) (STOP REPLICA),
+      C(resetprimary) (RESET PRIMARY) - supported since community.mysql 0.1.0,
+      C(resetreplica) (RESET REPLICA),
+      C(resetreplicaall) (RESET REPLICA ALL).
+    type: str
+    choices:
+    - changeprimary
+    - getprimary
+    - getreplica
+    - startreplica
+    - stopreplica
+    - resetprimary
+    - resetreplica
+    - resetreplicaall
+    default: getreplica
+  primary_host:
+    description:
+    - Same as the C(MASTER_HOST) mysql variable.
+    type: str
+    aliases: [master_host]
+  primary_user:
+    description:
+    - Same as the C(MASTER_USER) mysql variable.
+    type: str
+    aliases: [master_user]
+  primary_password:
+    description:
+    - Same as the C(MASTER_PASSWORD) mysql variable.
+    type: str
+    aliases: [master_password]
+  primary_port:
+    description:
+    - Same as the C(MASTER_PORT) mysql variable.
+    type: int
+    aliases: [master_port]
+  primary_connect_retry:
+    description:
+    - Same as the C(MASTER_CONNECT_RETRY) mysql variable.
+    type: int
+    aliases: [master_connect_retry]
+  primary_log_file:
+    description:
+    - Same as the C(MASTER_LOG_FILE) mysql variable.
+    type: str
+    aliases: [master_log_file]
+  primary_log_pos:
+    description:
+    - Same as the C(MASTER_LOG_POS) mysql variable.
+    type: int
+    aliases: [master_log_pos]
+  relay_log_file:
+    description:
+    - Same as mysql variable.
+    type: str
+  relay_log_pos:
+    description:
+    - Same as mysql variable.
+    type: int
+  primary_ssl:
+    description:
+    - Same as the C(MASTER_SSL) mysql variable.
+    - When setting it to C(yes), the connection attempt only succeeds
+      if an encrypted connection can be established.
+    - For details, refer to
+      L(MySQL encrypted replication documentation,https://dev.mysql.com/doc/refman/8.0/en/replication-solutions-encrypted-connections.html).
+    - The default is C(false).
+    type: bool
+    aliases: [master_ssl]
+  primary_ssl_ca:
+    description:
+    - Same as the C(MASTER_SSL_CA) mysql variable.
+    - For details, refer to
+      L(MySQL encrypted replication documentation,https://dev.mysql.com/doc/refman/8.0/en/replication-solutions-encrypted-connections.html).
+    type: str
+    aliases: [master_ssl_ca]
+  primary_ssl_capath:
+    description:
+    - Same as the C(MASTER_SSL_CAPATH) mysql variable.
+    - For details, refer to
+      L(MySQL encrypted replication documentation,https://dev.mysql.com/doc/refman/8.0/en/replication-solutions-encrypted-connections.html).
+    type: str
+    aliases: [master_ssl_capath]
+  primary_ssl_cert:
+    description:
+    - Same as the C(MASTER_SSL_CERT) mysql variable.
+    - For details, refer to
+      L(MySQL encrypted replication documentation,https://dev.mysql.com/doc/refman/8.0/en/replication-solutions-encrypted-connections.html).
+    type: str
+    aliases: [master_ssl_cert]
+  primary_ssl_key:
+    description:
+    - Same as the C(MASTER_SSL_KEY) mysql variable.
+    - For details, refer to
+      L(MySQL encrypted replication documentation,https://dev.mysql.com/doc/refman/8.0/en/replication-solutions-encrypted-connections.html).
+    type: str
+    aliases: [master_ssl_key]
+  primary_ssl_cipher:
+    description:
+    - Same as the C(MASTER_SSL_CIPHER) mysql variable.
+    - Specifies a colon-separated list of one or more ciphers permitted by the replica for the replication connection.
+    - For details, refer to
+      L(MySQL encrypted replication documentation,https://dev.mysql.com/doc/refman/8.0/en/replication-solutions-encrypted-connections.html).
+    type: str
+    aliases: [master_ssl_cipher]
+  primary_ssl_verify_server_cert:
+    description:
+    - Same as mysql variable.
+    type: bool
+    default: false
+    version_added: '3.5.0'
+  primary_auto_position:
+    description:
+    - Whether the host uses GTID based replication or not.
+    - Same as the C(MASTER_AUTO_POSITION) mysql variable.
+    type: bool
+    default: false
+    aliases: [master_auto_position]
+  primary_use_gtid:
+    description:
+    - Configures the replica to use the MariaDB Global Transaction ID.
+    - C(disabled) equals MASTER_USE_GTID=no command.
+    - To find information about available values see
+      U(https://mariadb.com/kb/en/library/change-master-to/#master_use_gtid).
+    - Available since MariaDB 10.0.2.
+    choices: [current_pos, replica_pos, disabled]
+    type: str
+    version_added: '0.1.0'
+    aliases: [master_use_gtid]
+  primary_delay:
+    description:
+    - Time lag behind the primary's state (in seconds).
+    - Same as the C(MASTER_DELAY) mysql variable.
+    - Available from MySQL 5.6.
+    - For more information see U(https://dev.mysql.com/doc/refman/8.0/en/replication-delayed.html).
+    type: int
+    version_added: '0.1.0'
+    aliases: [master_delay]
+  connection_name:
+    description:
+    - Name of the primary connection.
+    - Supported from MariaDB 10.0.1.
+    - Mutually exclusive with I(channel).
+    - For more information see U(https://mariadb.com/kb/en/library/multi-source-replication/).
+    type: str
+    version_added: '0.1.0'
+  channel:
+    description:
+    - Name of replication channel.
+    - Multi-source replication is supported from MySQL 5.7.
+    - Mutually exclusive with I(connection_name).
+    - For more information see U(https://dev.mysql.com/doc/refman/8.0/en/replication-multi-source.html).
+    type: str
+    version_added: '0.1.0'
+  fail_on_error:
+    description:
+    - Fails on error when calling mysql.
+    type: bool
+    default: false
+    version_added: '0.1.0'
+
+notes:
+- If an empty value for the parameter of string type is needed, use an empty string.
+
+extends_documentation_fragment:
+- community.mysql.mysql
+
+
+seealso:
+- module: community.mysql.mysql_info
+- name: MySQL replication reference
+  description: Complete reference of the MySQL replication documentation.
+  link: https://dev.mysql.com/doc/refman/8.0/en/replication.html
+- name: MySQL encrypted replication reference.
+  description: Setting up MySQL replication to use encrypted connection.
+  link: https://dev.mysql.com/doc/refman/8.0/en/replication-solutions-encrypted-connections.html
+- name: MariaDB replication reference
+  description: Complete reference of the MariaDB replication documentation.
+  link: https://mariadb.com/kb/en/library/setting-up-replication/
+'''
+
+EXAMPLES = r'''
+# If you encounter the "Please explicitly state intended protocol" error,
+# use the login_unix_socket argument
+- name: Stop mysql replica thread
+  community.mysql.mysql_replication:
+    mode: stopreplica
+    login_unix_socket: /run/mysqld/mysqld.sock
+
+- name: Get primary binlog file name and binlog position
+  community.mysql.mysql_replication:
+    mode: getprimary
+
+- name: Change primary to primary server 192.0.2.1 and use binary log 'mysql-bin.000009' with position 4578
+  community.mysql.mysql_replication:
+    mode: changeprimary
+    primary_host: 192.0.2.1
+    primary_log_file: mysql-bin.000009
+    primary_log_pos: 4578
+
+- name: Check replica status using port 3308
+  community.mysql.mysql_replication:
+    mode: getreplica
+    login_host: ansible.example.com
+    login_port: 3308
+
+- name: On MariaDB change primary to use GTID current_pos
+  community.mysql.mysql_replication:
+    mode: changeprimary
+    primary_use_gtid: current_pos
+
+- name: Change primary to use replication delay 3600 seconds
+  community.mysql.mysql_replication:
+    mode: changeprimary
+    primary_host: 192.0.2.1
+    primary_delay: 3600
+
+- name: Start MariaDB replica with connection name primary-1
+  community.mysql.mysql_replication:
+    mode: startreplica
+    connection_name: primary-1
+
+- name: Stop replication in channel primary-1
+  community.mysql.mysql_replication:
+    mode: stopreplica
+    channel: primary-1
+
+- name: >
+    Run RESET MASTER command which will delete all existing binary log files
+    and reset the binary log index file on the primary
+  community.mysql.mysql_replication:
+    mode: resetprimary
+
+- name: Run start replica and fail the task on errors
+  community.mysql.mysql_replication:
+    mode: startreplica
+    connection_name: primary-1
+    fail_on_error: true
+
+- name: Change primary and fail on error (like when replica thread is running)
+  community.mysql.mysql_replication:
+    mode: changeprimary
+    fail_on_error: true
+
+'''
+
+RETURN = r'''
+queries:
+  description: List of executed queries which modified DB's state.
+  returned: always
+  type: list
+  sample: ["CHANGE MASTER TO MASTER_HOST='primary2.example.com',MASTER_PORT=3306"]
+  version_added: '0.1.0'
+'''
+
+import os
+import warnings
+
+from ansible.module_utils.basic import AnsibleModule
+from ansible_collections.community.mysql.plugins.module_utils.mysql import (
+    mysql_connect,
+    mysql_driver,
+    mysql_driver_fail_msg,
+    mysql_common_argument_spec,
+)
+from ansible.module_utils._text import to_native
+
+executed_queries = []
+
+
+def get_primary_status(cursor):
+    # TODO: when it's available to change on MySQL's side,
+    # change MASTER to PRIMARY using the approach from
+    # get_replica_status() function. Same for other functions.
+    cursor.execute("SHOW MASTER STATUS")
+    primarystatus = cursor.fetchone()
+    return primarystatus
+
+
+def get_replica_status(cursor, connection_name='', channel='', term='REPLICA'):
+    if connection_name:
+        query = "SHOW %s '%s' STATUS" % (term, connection_name)
+    else:
+        query = "SHOW %s STATUS" % term
+
+    if channel:
+        query += " FOR CHANNEL '%s'" % channel
+
+    cursor.execute(query)
+    replica_status = cursor.fetchone()
+    return replica_status
+
+
+def stop_replica(module, cursor, connection_name='', channel='', fail_on_error=False, term='REPLICA'):
+    if connection_name:
+        query = "STOP %s '%s'" % (term, connection_name)
+    else:
+        query = 'STOP %s' % term
+
+    if channel:
+        query += " FOR CHANNEL '%s'" % channel
+
+    try:
+        executed_queries.append(query)
+        cursor.execute(query)
+        stopped = True
+    except mysql_driver.Warning as e:
+        stopped = False
+    except Exception as e:
+        if fail_on_error:
+            module.fail_json(msg="STOP REPLICA failed: %s" % to_native(e))
+        stopped = False
+    return stopped
+
+
+def reset_replica(module, cursor, connection_name='', channel='', fail_on_error=False, term='REPLICA'):
+    if connection_name:
+        query = "RESET %s '%s'" % (term, connection_name)
+    else:
+        query = 'RESET %s' % term
+
+    if channel:
+        query += " FOR CHANNEL '%s'" % channel
+
+    try:
+        executed_queries.append(query)
+        cursor.execute(query)
+        reset = True
+    except mysql_driver.Warning as e:
+        reset = False
+    except Exception as e:
+        if fail_on_error:
+            module.fail_json(msg="RESET REPLICA failed: %s" % to_native(e))
+        reset = False
+    return reset
+
+
+def reset_replica_all(module, cursor, connection_name='', channel='', fail_on_error=False, term='REPLICA'):
+    if connection_name:
+        query = "RESET %s '%s' ALL" % (term, connection_name)
+    else:
+        query = 'RESET %s ALL' % term
+
+    if channel:
+        query += " FOR CHANNEL '%s'" % channel
+
+    try:
+        executed_queries.append(query)
+        cursor.execute(query)
+        reset = True
+    except mysql_driver.Warning as e:
+        reset = False
+    except Exception as e:
+        if fail_on_error:
+            module.fail_json(msg="RESET REPLICA ALL failed: %s" % to_native(e))
+        reset = False
+    return reset
+
+
+def reset_primary(module, cursor, fail_on_error=False):
+    query = 'RESET MASTER'
+    try:
+        executed_queries.append(query)
+        cursor.execute(query)
+        reset = True
+    except mysql_driver.Warning as e:
+        reset = False
+    except Exception as e:
+        if fail_on_error:
+            module.fail_json(msg="RESET MASTER failed: %s" % to_native(e))
+        reset = False
+    return reset
+
+
+def start_replica(module, cursor, connection_name='', channel='', fail_on_error=False, term='REPLICA'):
+    if connection_name:
+        query = "START %s '%s'" % (term, connection_name)
+    else:
+        query = 'START %s' % term
+
+    if channel:
+        query += " FOR CHANNEL '%s'" % channel
+
+    try:
+        executed_queries.append(query)
+        cursor.execute(query)
+        started = True
+    except mysql_driver.Warning as e:
+        started = False
+    except Exception as e:
+        if fail_on_error:
+            module.fail_json(msg="START REPLICA failed: %s" % to_native(e))
+        started = False
+    return started
+
+
+def changeprimary(cursor, chm, connection_name='', channel=''):
+    if connection_name:
+        query = "CHANGE MASTER '%s' TO %s" % (connection_name, ','.join(chm))
+    else:
+        query = 'CHANGE MASTER TO %s' % ','.join(chm)
+
+    if channel:
+        query += " FOR CHANNEL '%s'" % channel
+
+    executed_queries.append(query)
+    cursor.execute(query)
+
+
+def main():
+    argument_spec = mysql_common_argument_spec()
+    argument_spec.update(
+        mode=dict(type='str', default='getreplica', choices=[
+            'getprimary',
+            'getreplica',
+            'changeprimary',
+            'stopreplica',
+            'startreplica',
+            'resetprimary',
+            'resetreplica',
+            'resetreplicaall']),
+        primary_auto_position=dict(type='bool', default=False, aliases=['master_auto_position']),
+        primary_host=dict(type='str', aliases=['master_host']),
+        primary_user=dict(type='str', aliases=['master_user']),
+        primary_password=dict(type='str', no_log=True, aliases=['master_password']),
+        primary_port=dict(type='int', aliases=['master_port']),
+        primary_connect_retry=dict(type='int', aliases=['master_connect_retry']),
+        primary_log_file=dict(type='str', aliases=['master_log_file']),
+        primary_log_pos=dict(type='int', aliases=['master_log_pos']),
+        relay_log_file=dict(type='str'),
+        relay_log_pos=dict(type='int'),
+        primary_ssl=dict(type='bool', aliases=['master_ssl']),
+        primary_ssl_ca=dict(type='str', aliases=['master_ssl_ca']),
+        primary_ssl_capath=dict(type='str', aliases=['master_ssl_capath']),
+        primary_ssl_cert=dict(type='str', aliases=['master_ssl_cert']),
+        primary_ssl_key=dict(type='str', no_log=False, aliases=['master_ssl_key']),
+        primary_ssl_cipher=dict(type='str', aliases=['master_ssl_cipher']),
+        primary_ssl_verify_server_cert=dict(type='bool', default=False),
+        primary_use_gtid=dict(type='str', choices=[
+            'current_pos', 'replica_pos', 'disabled'], aliases=['master_use_gtid']),
+        primary_delay=dict(type='int', aliases=['master_delay']),
+        connection_name=dict(type='str'),
+        channel=dict(type='str'),
+        fail_on_error=dict(type='bool', default=False),
+    )
+    module = AnsibleModule(
+        argument_spec=argument_spec,
+        mutually_exclusive=[
+            ['connection_name', 'channel']
+        ],
+    )
+    mode = module.params["mode"]
+    primary_host = module.params["primary_host"]
+    primary_user = module.params["primary_user"]
+    primary_password = module.params["primary_password"]
+    primary_port = module.params["primary_port"]
+    primary_connect_retry = module.params["primary_connect_retry"]
+    primary_log_file = module.params["primary_log_file"]
+    primary_log_pos = module.params["primary_log_pos"]
+    relay_log_file = module.params["relay_log_file"]
+    relay_log_pos = module.params["relay_log_pos"]
+    primary_ssl = module.params["primary_ssl"]
+    primary_ssl_ca = module.params["primary_ssl_ca"]
+    primary_ssl_capath = module.params["primary_ssl_capath"]
+    primary_ssl_cert = module.params["primary_ssl_cert"]
+    primary_ssl_key = module.params["primary_ssl_key"]
+    primary_ssl_cipher = module.params["primary_ssl_cipher"]
+    primary_ssl_verify_server_cert = module.params["primary_ssl_verify_server_cert"]
+    primary_auto_position = module.params["primary_auto_position"]
+    ssl_cert = module.params["client_cert"]
+    ssl_key = module.params["client_key"]
+    ssl_ca = module.params["ca_cert"]
+    check_hostname = module.params["check_hostname"]
+    connect_timeout = module.params['connect_timeout']
+    config_file = module.params['config_file']
+    primary_delay = module.params['primary_delay']
+    if module.params.get("primary_use_gtid") == 'disabled':
+        primary_use_gtid = 'no'
+    else:
+        primary_use_gtid = module.params["primary_use_gtid"]
+    connection_name = module.params["connection_name"]
+    channel = module.params['channel']
+    fail_on_error = module.params['fail_on_error']
+
+    if mysql_driver is None:
+        module.fail_json(msg=mysql_driver_fail_msg)
+    else:
+        warnings.filterwarnings('error', category=mysql_driver.Warning)
+
+    login_password = module.params["login_password"]
+    login_user = module.params["login_user"]
+
+    try:
+        cursor, db_conn = mysql_connect(module, login_user, login_password, config_file,
+                                        ssl_cert, ssl_key, ssl_ca, None, cursor_class='DictCursor',
+                                        connect_timeout=connect_timeout, check_hostname=check_hostname)
+    except Exception as e:
+        if os.path.exists(config_file):
+            module.fail_json(msg="unable to connect to database, check login_user and "
+                                 "login_password are correct or %s has the credentials. "
+                                 "Exception message: %s" % (config_file, to_native(e)))
+        else:
+            module.fail_json(msg="unable to find %s. Exception message: %s" % (config_file, to_native(e)))
+
+    cursor.execute("SELECT VERSION()")
+    if 'mariadb' in cursor.fetchone()["VERSION()"].lower():
+        from ansible_collections.community.mysql.plugins.module_utils.implementations.mariadb import replication as impl
+    else:
+        from ansible_collections.community.mysql.plugins.module_utils.implementations.mysql import replication as impl
+
+    # Since MySQL 8.0.22 and MariaDB 10.5.1,
+    # "REPLICA" must be used instead of "SLAVE"
+    if impl.uses_replica_terminology(cursor):
+        replica_term = 'REPLICA'
+    else:
+        replica_term = 'SLAVE'
+        if primary_use_gtid == 'replica_pos':
+            primary_use_gtid = 'slave_pos'
+
+    if mode == 'getprimary':
+        status = get_primary_status(cursor)
+        if not isinstance(status, dict):
+            status = dict(Is_Primary=False,
+                          msg="Server is not configured as mysql primary")
+        else:
+            status['Is_Primary'] = True
+
+        module.exit_json(queries=executed_queries, **status)
+
+    elif mode == "getreplica":
+        status = get_replica_status(cursor, connection_name, channel, replica_term)
+        if not isinstance(status, dict):
+            status = dict(Is_Replica=False, msg="Server is not configured as mysql replica")
+        else:
+            status['Is_Replica'] = True
+
+        module.exit_json(queries=executed_queries, **status)
+
+    elif mode == 'changeprimary':
+        chm = []
+        result = {}
+        if primary_host is not None:
+            chm.append("MASTER_HOST='%s'" % primary_host)
+        if primary_user is not None:
+            chm.append("MASTER_USER='%s'" % primary_user)
+        if primary_password is not None:
+            chm.append("MASTER_PASSWORD='%s'" % primary_password)
+        if primary_port is not None:
+            chm.append("MASTER_PORT=%s" % primary_port)
+        if primary_connect_retry is not None:
+            chm.append("MASTER_CONNECT_RETRY=%s" % primary_connect_retry)
+        if primary_log_file is not None:
+            chm.append("MASTER_LOG_FILE='%s'" % primary_log_file)
+        if primary_log_pos is not None:
+            chm.append("MASTER_LOG_POS=%s" % primary_log_pos)
+        if primary_delay is not None:
+            chm.append("MASTER_DELAY=%s" % primary_delay)
+        if relay_log_file is not None:
+            chm.append("RELAY_LOG_FILE='%s'" % relay_log_file)
+        if relay_log_pos is not None:
+            chm.append("RELAY_LOG_POS=%s" % relay_log_pos)
+        if primary_ssl is not None:
+            if primary_ssl:
+                chm.append("MASTER_SSL=1")
+            else:
+                chm.append("MASTER_SSL=0")
+        if primary_ssl_ca is not None:
+            chm.append("MASTER_SSL_CA='%s'" % primary_ssl_ca)
+        if primary_ssl_capath is not None:
+            chm.append("MASTER_SSL_CAPATH='%s'" % primary_ssl_capath)
+        if primary_ssl_cert is not None:
+            chm.append("MASTER_SSL_CERT='%s'" % primary_ssl_cert)
+        if primary_ssl_key is not None:
+            chm.append("MASTER_SSL_KEY='%s'" % primary_ssl_key)
+        if primary_ssl_cipher is not None:
+            chm.append("MASTER_SSL_CIPHER='%s'" % primary_ssl_cipher)
+        if primary_ssl_verify_server_cert:
+            chm.append("SOURCE_SSL_VERIFY_SERVER_CERT=1")
+        if primary_auto_position:
+            chm.append("MASTER_AUTO_POSITION=1")
+        if primary_use_gtid is not None:
+            chm.append("MASTER_USE_GTID=%s" % primary_use_gtid)
+        try:
+            changeprimary(cursor, chm, connection_name, channel)
+        except mysql_driver.Warning as e:
+            result['warning'] = to_native(e)
+        except Exception as e:
+            module.fail_json(msg='%s. Query == CHANGE MASTER TO %s' % (to_native(e), chm))
+        result['changed'] = True
+        module.exit_json(queries=executed_queries, **result)
+    elif mode == "startreplica":
+        started = start_replica(module, cursor, connection_name, channel, fail_on_error, replica_term)
+        if started is True:
+            module.exit_json(msg="Replica started ", changed=True, queries=executed_queries)
+        else:
+            module.exit_json(msg="Replica already started (Or cannot be started)", changed=False, queries=executed_queries)
+    elif mode == "stopreplica":
+        stopped = stop_replica(module, cursor, connection_name, channel, fail_on_error, replica_term)
+        if stopped is True:
+            module.exit_json(msg="Replica stopped", changed=True, queries=executed_queries)
+        else:
+            module.exit_json(msg="Replica already stopped", changed=False, queries=executed_queries)
+    elif mode == 'resetprimary':
+        reset = reset_primary(module, cursor, fail_on_error)
+        if reset is True:
+            module.exit_json(msg="Primary reset", changed=True, queries=executed_queries)
+        else:
+            module.exit_json(msg="Primary already reset", changed=False, queries=executed_queries)
+    elif mode == "resetreplica":
+        reset = reset_replica(module, cursor, connection_name, channel, fail_on_error, replica_term)
+        if reset is True:
+            module.exit_json(msg="Replica reset", changed=True, queries=executed_queries)
+        else:
+            module.exit_json(msg="Replica already reset", changed=False, queries=executed_queries)
+    elif mode == "resetreplicaall":
+        reset = reset_replica_all(module, cursor, connection_name, channel, fail_on_error, replica_term)
+        if reset is True:
+            module.exit_json(msg="Replica reset", changed=True, queries=executed_queries)
+        else:
+            module.exit_json(msg="Replica already reset", changed=False, queries=executed_queries)
+
+    warnings.simplefilter("ignore")
+
+
+if __name__ == '__main__':
+    main()
diff --git a/ansible_collections/community/mysql/plugins/modules/mysql_role.py b/ansible_collections/community/mysql/plugins/modules/mysql_role.py
new file mode 100644
index 000000000..070d7939d
--- /dev/null
+++ b/ansible_collections/community/mysql/plugins/modules/mysql_role.py
@@ -0,0 +1,1095 @@
+#!/usr/bin/python
+# -*- coding: utf-8 -*-
+
+# Copyright: (c) 2021, Andrew Klychkov <aaklychkov@mail.ru>
+# GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt)
+
+from __future__ import absolute_import, division, print_function
+__metaclass__ = type
+
+DOCUMENTATION = r'''
+---
+module: mysql_role
+
+short_description: Adds, removes, or updates a MySQL role
+
+description:
+   - Adds, removes, or updates a MySQL role.
+   - Roles are supported since MySQL 8.0.0 and MariaDB 10.0.5.
+
+version_added: '2.2.0'
+
+options:
+  name:
+    description:
+      - Name of the role to add or remove.
+    type: str
+    required: true
+
+  admin:
+    description:
+      - Supported by B(MariaDB).
+      - Name of the admin user of the role (the I(login_user), by default).
+    type: str
+
+  priv:
+    description:
+      - "MySQL privileges string in the format: C(db.table:priv1,priv2)."
+      - "You can specify multiple privileges by separating each one using
+        a forward slash: C(db.table:priv/db.table:priv)."
+      - The format is based on MySQL C(GRANT) statement.
+      - Database and table names can be quoted, MySQL-style.
+      - If column privileges are used, the C(priv1,priv2) part must be
+        exactly as returned by a C(SHOW GRANT) statement. If not followed,
+        the module will always report changes. It includes grouping columns
+        by permission (C(SELECT(col1,col2)) instead of C(SELECT(col1),SELECT(col2))).
+      - Can be passed as a dictionary (see the examples).
+      - Supports GRANTs for procedures and functions
+        (see the examples for the M(community.mysql.mysql_user) module).
+    type: raw
+
+  append_privs:
+    description:
+      - Append the privileges defined by the I(priv) option to the existing ones
+        for this role instead of overwriting them. Mutually exclusive with I(subtract_privs).
+    type: bool
+    default: false
+
+  subtract_privs:
+    description:
+      - Revoke the privileges defined by the I(priv) option and keep other existing privileges.
+        If set, invalid privileges in I(priv) are ignored.
+        Mutually exclusive with I(append_privs).
+    version_added: '3.2.0'
+    type: bool
+    default: false
+
+  members:
+    description:
+      - List of members of the role.
+      - For users, use the format C(username@hostname).
+        Always specify the hostname part explicitly.
+      - For roles, use the format C(rolename).
+      - Mutually exclusive with I(admin).
+    type: list
+    elements: str
+
+  append_members:
+    description:
+      - Add members defined by the I(members) option to the existing ones
+        for this role instead of overwriting them.
+      - Mutually exclusive with the I(detach_members) and I(admin) option.
+    type: bool
+    default: false
+
+  detach_members:
+    description:
+      - Detaches members defined by the I(members) option from the role
+        instead of overwriting all the current members.
+      - Mutually exclusive with the I(append_members) and I(admin) option.
+    type: bool
+    default: false
+
+  set_default_role_all:
+    description:
+      - Is not supported by MariaDB and is silently ignored when working with MariaDB.
+      - If C(yes), runs B(SET DEFAULT ROLE ALL TO) each of the I(members) when changed.
+      - If you want to avoid this behavior, set this option to C(no) explicitly.
+    type: bool
+    default: true
+
+  state:
+    description:
+      - If C(present) and the role does not exist, creates the role.
+      - If C(present) and the role exists, does nothing or updates its attributes.
+      - If C(absent), removes the role.
+    type: str
+    choices: [ absent, present ]
+    default: present
+
+  check_implicit_admin:
+    description:
+      - Check if mysql allows login as root/nopassword before trying supplied credentials.
+      - If success, passed I(login_user)/I(login_password) will be ignored.
+    type: bool
+    default: false
+
+  members_must_exist:
+    description:
+      - When C(yes), the module fails if any user in I(members) does not exist.
+      - When C(no), users in I(members) which don't exist are simply skipped.
+    type: bool
+    default: true
+
+notes:
+  - Pay attention that the module runs C(SET DEFAULT ROLE ALL TO)
+    all the I(members) passed by default when the state has changed.
+    If you want to avoid this behavior, set I(set_default_role_all) to C(no).
+  - Supports C(check_mode).
+
+seealso:
+  - module: community.mysql.mysql_user
+  - name: MySQL role reference
+    description: Complete reference of the MySQL role documentation.
+    link: https://dev.mysql.com/doc/refman/8.0/en/create-role.html
+
+author:
+  - Andrew Klychkov (@Andersson007)
+  - Felix Hamme (@betanummeric)
+
+extends_documentation_fragment:
+  - community.mysql.mysql
+'''
+
+EXAMPLES = r'''
+# If you encounter the "Please explicitly state intended protocol" error,
+# use the login_unix_socket argument, for example, login_unix_socket: /run/mysqld/mysqld.sock
+
+# Example of a .my.cnf file content for setting a root password
+# [client]
+# user=root
+# password=n<_665{vS43y
+#
+# Example of a privileges dictionary passed through the priv option
+# priv:
+#   'mydb.*': 'INSERT,UPDATE'
+#   'anotherdb.*': 'SELECT'
+#   'yetanotherdb.*': 'ALL'
+#
+# You can also use the string format like in the community.mysql.mysql_user module, for example
+# mydb.*:INSERT,UPDATE/anotherdb.*:SELECT/yetanotherdb.*:ALL
+#
+# For more examples on how to specify privileges, refer to the community.mysql.mysql_user module
+
+# Create a role developers with all database privileges
+# and add alice and bob as members.
+# The statement 'SET DEFAULT ROLE ALL' to them will be run.
+- name: Create role developers, add members
+  community.mysql.mysql_role:
+    name: developers
+    state: present
+    priv: '*.*:ALL'
+    members:
+    - 'alice@%'
+    - 'bob@%'
+
+- name: Same as above but do not run SET DEFAULT ROLE ALL TO each member
+  community.mysql.mysql_role:
+    name: developers
+    state: present
+    priv: '*.*:ALL'
+    members:
+    - 'alice@%'
+    - 'bob@%'
+    set_default_role_all: false
+
+# Assuming that the role developers exists,
+# add john to the current members
+- name: Add members to an existing role
+  community.mysql.mysql_role:
+    name: developers
+    state: present
+    append_members: true
+    members:
+    - 'joe@localhost'
+
+# Create role readers with the SELECT privilege
+# on all tables in the fiction database
+- name: Create role developers, add members
+  community.mysql.mysql_role:
+    name: readers
+    state: present
+    priv: 'fiction.*:SELECT'
+
+# Assuming that the role readers exists,
+# add the UPDATE privilege to the role on all tables in the fiction database
+- name: Create role developers, add members
+  community.mysql.mysql_role:
+    name: readers
+    state: present
+    priv: 'fiction.*:UPDATE'
+    append_privs: true
+
+- name: Create role with the 'SELECT' and 'UPDATE' privileges in db1 and db2
+  community.mysql.mysql_role:
+    state: present
+    name: foo
+    priv:
+      'db1.*': 'SELECT,UPDATE'
+      'db2.*': 'SELECT,UPDATE'
+
+- name: Remove joe from readers
+  community.mysql.mysql_role:
+    state: present
+    name: readers
+    members:
+    - 'joe@localhost'
+    detach_members: true
+
+- name: Remove the role readers if exists
+  community.mysql.mysql_role:
+    state: absent
+    name: readers
+
+- name: Example of using login_unix_socket to connect to the server
+  community.mysql.mysql_role:
+    name: readers
+    state: present
+    login_unix_socket: /var/run/mysqld/mysqld.sock
+
+# Pay attention that the admin cannot be changed later
+# and will be ignored if a role currently exists.
+# To change members, you need to run a separate task using the admin
+# of the role as the login_user.
+- name: On MariaDB, create the role readers with alice as its admin
+  community.mysql.mysql_role:
+    state: present
+    name: readers
+    admin: 'alice@%'
+
+- name: Create the role business, add the role marketing to members
+  community.mysql.mysql_role:
+    state: present
+    name: business
+    members:
+    - marketing
+
+- name: Ensure the role foo does not have the DELETE privilege
+  community.mysql.mysql_role:
+    state: present
+    name: foo
+    subtract_privs: true
+    priv:
+      'db1.*': DELETE
+
+- name: Add some members to a role and skip not-existent users
+  community.mysql.mysql_role:
+    state: present
+    name: foo
+    append_members: true
+    members_must_exist: false
+    members:
+    - 'existing_user@localhost'
+    - 'not_existing_user@localhost'
+
+- name: Detach some members from a role and ignore not-existent users
+  community.mysql.mysql_role:
+    state: present
+    name: foo
+    detach_members: true
+    members_must_exist: false
+    members:
+    - 'existing_user@localhost'
+    - 'not_existing_user@localhost'
+'''
+
+RETURN = '''#'''
+
+from ansible.module_utils.basic import AnsibleModule
+from ansible_collections.community.mysql.plugins.module_utils.mysql import (
+    mysql_connect,
+    mysql_driver,
+    mysql_driver_fail_msg,
+    mysql_common_argument_spec
+)
+from ansible_collections.community.mysql.plugins.module_utils.user import (
+    convert_priv_dict_to_str,
+    get_impl,
+    get_mode,
+    user_mod,
+    privileges_grant,
+    privileges_unpack,
+)
+from ansible.module_utils._text import to_native
+from ansible.module_utils.six import iteritems
+
+
+def normalize_users(module, users, is_mariadb=False):
+    """Normalize passed user names.
+
+    Example of transformation:
+    ['user0'] => [('user0', '')] / ['user0'] => [('user0', '%')]
+    ['user0@host0'] => [('user0', 'host0')]
+
+    Args:
+        module (AnsibleModule): Object of the AnsibleModule class.
+        users (list): List of user names.
+        is_mariadb (bool): Flag indicating we are working with MariaDB
+
+    Returns:
+        list: List of tuples like [('user0', ''), ('user0', 'host0')].
+    """
+    normalized_users = []
+
+    for user in users:
+        try:
+            tmp = user.split('@')
+
+            if tmp[0] == '':
+                module.fail_json(msg="Member's name cannot be empty.")
+
+            if len(tmp) == 1:
+                if not is_mariadb:
+                    normalized_users.append((tmp[0], '%'))
+                else:
+                    normalized_users.append((tmp[0], ''))
+
+            elif len(tmp) == 2:
+                normalized_users.append((tmp[0], tmp[1]))
+
+        except Exception as e:
+            msg = ('Error occured while parsing the name "%s": %s. '
+                   'It must be in the format "username" or '
+                   '"username@hostname" ' % (user, to_native(e)))
+            module.fail_json(msg=msg)
+
+    return normalized_users
+
+
+class DbServer():
+    """Class to fetch information from a database.
+
+    Args:
+        module (AnsibleModule): Object of the AnsibleModule class.
+        cursor (cursor): Cursor object of a database Python connector.
+
+    Attributes:
+        module (AnsibleModule): Object of the AnsibleModule class.
+        cursor (cursor): Cursor object of a database Python connector.
+        role_impl (library): Corresponding library depending
+            on a server type (MariaDB or MySQL)
+        mariadb (bool): True if MariaDB, False otherwise.
+        roles_supported (bool): True if roles are supported, False otherwise.
+        users (set): Set of users existing in a DB in the form (username, hostname).
+    """
+    def __init__(self, module, cursor):
+        self.module = module
+        self.cursor = cursor
+        self.role_impl = self.get_implementation()
+        self.mariadb = self.role_impl.is_mariadb()
+        self.roles_supported = self.role_impl.supports_roles(self.cursor)
+        self.users = set(self.__get_users())
+
+    def is_mariadb(self):
+        """Get info whether a DB server is a MariaDB instance.
+
+        Returns:
+            self.mariadb: Attribute value.
+        """
+        return self.mariadb
+
+    def supports_roles(self):
+        """Get info whether a DB server supports roles.
+
+        Returns:
+            self.roles_supported: Attribute value.
+        """
+        return self.roles_supported
+
+    def get_implementation(self):
+        """Get a current server implementation depending on its type.
+
+        Returns:
+            library: Depending on a server type (MySQL or MariaDB).
+        """
+        self.cursor.execute("SELECT VERSION()")
+
+        if 'mariadb' in self.cursor.fetchone()[0].lower():
+            import ansible_collections.community.mysql.plugins.module_utils.implementations.mariadb.role as role_impl
+        else:
+            import ansible_collections.community.mysql.plugins.module_utils.implementations.mysql.role as role_impl
+
+        return role_impl
+
+    def check_users_in_db(self, users):
+        """Check if users exist in a database.
+
+        Args:
+            users (list): List of tuples (username, hostname) to check.
+        """
+        for user in users:
+            if user not in self.users:
+                msg = 'User / role `%s` with host `%s` does not exist' % (user[0], user[1])
+                self.module.fail_json(msg=msg)
+
+    def filter_existing_users(self, users):
+        for user in users:
+            if user in self.users:
+                yield user
+
+    def __get_users(self):
+        """Get users.
+
+        Returns:
+            list: List of tuples (username, hostname).
+        """
+        self.cursor.execute('SELECT User, Host FROM mysql.user')
+        return self.cursor.fetchall()
+
+    def get_users(self):
+        """Get set of tuples (username, hostname) existing in a DB.
+
+        Returns:
+            self.users: Attribute value.
+        """
+        return self.users
+
+    def get_grants(self, user, host):
+        """Get grants.
+
+        Args:
+            user (str): User name
+            host (str): Host name
+
+        Returns:
+            list: List of tuples like [(grant1,), (grant2,), ... ].
+        """
+        if host:
+            self.cursor.execute('SHOW GRANTS FOR %s@%s', (user, host))
+        else:
+            self.cursor.execute('SHOW GRANTS FOR %s', (user,))
+
+        return self.cursor.fetchall()
+
+
+class MySQLQueryBuilder():
+    """Class to build and return queries specific to MySQL.
+
+    Args:
+        name (str): Role name.
+        host (str): Role host.
+
+    Attributes:
+        name (str): Role name.
+        host (str): Role host.
+    """
+    def __init__(self, name, host):
+        self.name = name
+        self.host = host
+
+    def role_exists(self):
+        """Return a query to check if a role with self.name and self.host exists in a database.
+
+        Returns:
+            tuple: (query_string, tuple_containing_parameters).
+        """
+        return 'SELECT count(*) FROM mysql.user WHERE user = %s AND host = %s', (self.name, self.host)
+
+    def role_grant(self, user):
+        """Return a query to grant a role to a user or a role.
+
+        Args:
+            user (tuple): User / role to grant the role to in the form (username, hostname).
+
+        Returns:
+            tuple: (query_string, tuple_containing_parameters).
+        """
+        if user[1]:
+            return 'GRANT %s@%s TO %s@%s', (self.name, self.host, user[0], user[1])
+        else:
+            return 'GRANT %s@%s TO %s', (self.name, self.host, user[0])
+
+    def role_revoke(self, user):
+        """Return a query to revoke a role from a user or role.
+
+        Args:
+            user (tuple): User / role to revoke the role from in the form (username, hostname).
+
+        Returns:
+            tuple: (query_string, tuple_containing_parameters).
+        """
+        if user[1]:
+            return 'REVOKE %s@%s FROM %s@%s', (self.name, self.host, user[0], user[1])
+        else:
+            return 'REVOKE %s@%s FROM %s', (self.name, self.host, user[0])
+
+    def role_create(self, admin=None):
+        """Return a query to create a role.
+
+        Args:
+            admin (tuple): Admin user in the form (username, hostname).
+                Because it is not supported by MySQL, we ignore it.
+
+        Returns:
+            tuple: (query_string, tuple_containing_parameters).
+        """
+        return 'CREATE ROLE %s', (self.name,)
+
+
+class MariaDBQueryBuilder():
+    """Class to build and return queries specific to MariaDB.
+
+    Args:
+        name (str): Role name.
+
+    Attributes:
+        name (str): Role name.
+    """
+    def __init__(self, name):
+        self.name = name
+
+    def role_exists(self):
+        """Return a query to check if a role with self.name exists in a database.
+
+        Returns:
+            tuple: (query_string, tuple_containing_parameters).
+        """
+        return "SELECT count(*) FROM mysql.user WHERE user = %s AND is_role  = 'Y'", (self.name,)
+
+    def role_grant(self, user):
+        """Return a query to grant a role to a user or role.
+
+        Args:
+            user (tuple): User / role to grant the role to in the form (username, hostname).
+
+        Returns:
+            tuple: (query_string, tuple_containing_parameters).
+        """
+        if user[1]:
+            return 'GRANT %s TO %s@%s', (self.name, user[0], user[1])
+        else:
+            return 'GRANT %s TO %s', (self.name, user[0])
+
+    def role_revoke(self, user):
+        """Return a query to revoke a role from a user or role.
+
+        Args:
+            user (tuple): User / role to revoke the role from in the form (username, hostname).
+
+        Returns:
+            tuple: (query_string, tuple_containing_parameters).
+        """
+        if user[1]:
+            return 'REVOKE %s FROM %s@%s', (self.name, user[0], user[1])
+        else:
+            return 'REVOKE %s FROM %s', (self.name, user[0])
+
+    def role_create(self, admin=None):
+        """Return a query to create a role.
+
+        Args:
+            admin (tuple): Admin user in the form (username, hostname).
+
+        Returns:
+            tuple: (query_string, tuple_containing_parameters).
+        """
+        if not admin:
+            return 'CREATE ROLE %s', (self.name,)
+
+        if admin[1]:
+            return 'CREATE ROLE %s WITH ADMIN %s@%s', (self.name, admin[0], admin[1])
+        else:
+            return 'CREATE ROLE %s WITH ADMIN %s', (self.name, admin[0])
+
+
+class MySQLRoleImpl():
+    """Class to work with MySQL role implementation.
+
+    Args:
+        module (AnsibleModule): Object of the AnsibleModule class.
+        cursor (cursor): Cursor object of a database Python connector.
+        name (str): Role name.
+        host (str): Role host.
+
+    Attributes:
+        module (AnsibleModule): Object of the AnsibleModule class.
+        cursor (cursor): Cursor object of a database Python connector.
+        name (str): Role name.
+        host (str): Role host.
+    """
+    def __init__(self, module, cursor, name, host):
+        self.module = module
+        self.cursor = cursor
+        self.name = name
+        self.host = host
+
+    def set_default_role_all(self, user):
+        """Run 'SET DEFAULT ROLE ALL TO' a user.
+
+        Args:
+            user (tuple): User / role to run the command against in the form (username, hostname).
+        """
+        if user[1]:
+            self.cursor.execute('SET DEFAULT ROLE ALL TO %s@%s', (user[0], user[1]))
+        else:
+            self.cursor.execute('SET DEFAULT ROLE ALL TO %s', (user[0],))
+
+    def get_admin(self):
+        """Get a current admin of a role.
+
+        Not supported by MySQL, so ignored here.
+        """
+        pass
+
+    def set_admin(self, admin):
+        """Set an admin of a role.
+
+        Not supported by MySQL, so ignored here.
+
+        TODO: Implement the feature if this gets supported.
+
+        Args:
+            admin (tuple): Admin user of the role in the form (username, hostname).
+        """
+        pass
+
+
+class MariaDBRoleImpl():
+    """Class to work with MariaDB role implementation.
+
+    Args:
+        module (AnsibleModule): Object of the AnsibleModule class.
+        cursor (cursor): Cursor object of a database Python connector.
+        name (str): Role name.
+
+    Attributes:
+        module (AnsibleModule): Object of the AnsibleModule class.
+        cursor (cursor): Cursor object of a database Python connector.
+        name (str): Role name.
+    """
+    def __init__(self, module, cursor, name):
+        self.module = module
+        self.cursor = cursor
+        self.name = name
+
+    def set_default_role_all(self, user):
+        """Run 'SET DEFAULT ROLE ALL TO' a user.
+
+        The command is not supported by MariaDB, ignored.
+
+        Args:
+            user (tuple): User / role to run the command against in the form (username, hostname).
+        """
+        pass
+
+    def get_admin(self):
+        """Get a current admin of a role.
+
+        Returns:
+            tuple: Of the form (username, hostname).
+        """
+        query = ("SELECT User, Host FROM mysql.roles_mapping "
+                 "WHERE Role = %s and Admin_option = 'Y'")
+
+        self.cursor.execute(query, (self.name,))
+        return self.cursor.fetchone()
+
+    def set_admin(self, admin):
+        """Set an admin of a role.
+
+        TODO: Implement changing when ALTER ROLE statement to
+            change role's admin gets supported.
+
+        Args:
+            admin (tuple): Admin user of the role in the form (username, hostname).
+        """
+        admin_user = admin[0]
+        admin_host = admin[1]
+        current_admin = self.get_admin()
+
+        if (admin_user, admin_host) != current_admin:
+            msg = ('The "admin" option value and the current '
+                   'roles admin (%s@%s) don not match. Ignored. '
+                   'To change the admin, you need to drop and create the '
+                   'role again.' % (current_admin[0], current_admin[1]))
+            self.module.warn(msg)
+
+
+class Role():
+    """Class to work with MySQL role objects.
+
+    Args:
+        module (AnsibleModule): Object of the AnsibleModule class.
+        cursor (cursor): Cursor object of a database Python connector.
+        name (str): Role name.
+        server (DbServer): Object of the DbServer class.
+
+    Attributes:
+        module (AnsibleModule): Object of the AnsibleModule class.
+        cursor (cursor): Cursor object of a database Python connector.
+        name (str): Role name.
+        server (DbServer): Object of the DbServer class.
+        host (str): Role's host.
+        full_name (str): Role's full name.
+        exists (bool): Indicates if a role exists or not.
+        members (set): Set of current role's members.
+    """
+    def __init__(self, module, cursor, name, server):
+        self.module = module
+        self.cursor = cursor
+        self.name = name
+        self.server = server
+        self.is_mariadb = self.server.is_mariadb()
+
+        if self.is_mariadb:
+            self.q_builder = MariaDBQueryBuilder(self.name)
+            self.role_impl = MariaDBRoleImpl(self.module, self.cursor, self.name)
+            self.full_name = '`%s`' % self.name
+            self.host = ''
+        else:
+            self.host = '%'
+            self.q_builder = MySQLQueryBuilder(self.name, self.host)
+            self.role_impl = MySQLRoleImpl(self.module, self.cursor, self.name, self.host)
+            self.full_name = '`%s`@`%s`' % (self.name, self.host)
+
+        self.exists = self.__role_exists()
+        self.members = set()
+
+        if self.exists:
+            self.members = self.__get_members()
+
+    def __role_exists(self):
+        """Check if a role exists.
+
+        Returns:
+            bool: True if the role exists, False if it does not.
+        """
+        self.cursor.execute(*self.q_builder.role_exists())
+        return self.cursor.fetchone()[0] > 0
+
+    def add(self, users, privs, check_mode=False, admin=False,
+            set_default_role_all=True):
+        """Add a role.
+
+        Args:
+            users (list): Role members.
+            privs (str): String containing privileges.
+            check_mode (bool): If True, just checks and does nothing.
+            admin (tuple): Role's admin. Contains (username, hostname).
+            set_default_role_all (bool): If True, runs SET DEFAULT ROLE ALL TO each member.
+
+        Returns:
+            bool: True if the state has changed, False if has not.
+        """
+        if check_mode:
+            if not self.exists:
+                return True
+            return False
+
+        self.cursor.execute(*self.q_builder.role_create(admin))
+
+        if users:
+            self.update_members(users, set_default_role_all=set_default_role_all)
+
+        if privs:
+            for db_table, priv in iteritems(privs):
+                privileges_grant(self.cursor, self.name, self.host,
+                                 db_table, priv, tls_requires=None,
+                                 maria_role=self.is_mariadb)
+
+        return True
+
+    def drop(self, check_mode=False):
+        """Drop a role.
+
+        Args:
+            check_mode (bool): If True, just checks and does nothing.
+
+        Returns:
+            bool: True if the state has changed, False if has not.
+        """
+        if not self.exists:
+            return False
+
+        if check_mode and self.exists:
+            return True
+
+        self.cursor.execute('DROP ROLE %s', (self.name,))
+        return True
+
+    def update_members(self, users, check_mode=False, append_members=False,
+                       set_default_role_all=True):
+        """Add users to a role.
+
+        Args:
+            users (list): Role members.
+            check_mode (bool): If True, just checks and does nothing.
+            append_members (bool): If True, adds new members passed through users
+                not touching current members.
+            set_default_role_all (bool): If True, runs SET DEFAULT ROLE ALL TO each member.
+
+        Returns:
+            bool: True if the state has changed, False if has not.
+        """
+        if not users:
+            return False
+
+        changed = False
+        for user in users:
+            if user not in self.members:
+                if check_mode:
+                    return True
+
+                self.cursor.execute(*self.q_builder.role_grant(user))
+
+                if set_default_role_all:
+                    self.role_impl.set_default_role_all(user)
+
+                changed = True
+
+        if append_members:
+            return changed
+
+        for user in self.members:
+            if user not in users and user != ('root', 'localhost'):
+                changed = self.__remove_member(user, check_mode)
+
+        return changed
+
+    def remove_members(self, users, check_mode=False):
+        """Remove members from a role.
+
+        Args:
+            users (list): Role members.
+            check_mode (bool): If True, just checks and does nothing.
+
+        Returns:
+            bool: True if the state has changed, False if has not.
+        """
+        if not users:
+            return False
+
+        changed = False
+        for user in users:
+            if user in self.members:
+                changed = self.__remove_member(user, check_mode)
+
+        return changed
+
+    def __remove_member(self, user, check_mode=False):
+        """Remove a member from a role.
+
+        Args:
+            user (str): Role member to remove.
+            check_mode (bool): If True, just returns True and does nothing.
+
+        Returns:
+            bool: True if the state has changed, False if has not.
+        """
+        if check_mode:
+            return True
+
+        self.cursor.execute(*self.q_builder.role_revoke(user))
+
+        return True
+
+    def update(self, users, privs, check_mode=False,
+               append_privs=False, subtract_privs=False,
+               append_members=False, detach_members=False,
+               admin=False, set_default_role_all=True):
+        """Update a role.
+
+        Update a role if needed.
+
+        Todo: Implement changing of role's admin when ALTER ROLE statement
+            to do that gets supported.
+
+        Args:
+            users (list): Role members.
+            privs (str): String containing privileges.
+            check_mode (bool): If True, just checks and does nothing.
+            append_privs (bool): If True, adds new privileges passed through privs
+                not touching current privileges.
+            subtract_privs (bool): If True, revoke the privileges passed through privs
+                not touching other existing privileges.
+            append_members (bool): If True, adds new members passed through users
+                not touching current members.
+            detach_members (bool): If True, removes members passed through users from a role.
+            admin (tuple): Role's admin. Contains (username, hostname).
+            set_default_role_all (bool): If True, runs SET DEFAULT ROLE ALL TO each member.
+
+        Returns:
+            bool: True if the state has changed, False if has not.
+        """
+        changed = False
+        members_changed = False
+
+        if users:
+            if detach_members:
+                members_changed = self.remove_members(users, check_mode=check_mode)
+
+            else:
+                members_changed = self.update_members(users, check_mode=check_mode,
+                                                      append_members=append_members,
+                                                      set_default_role_all=set_default_role_all)
+
+        if privs:
+            result = user_mod(self.cursor, self.name, self.host,
+                              None, None, None, None, None, None,
+                              privs, append_privs, subtract_privs, None,
+                              self.module, role=True, maria_role=self.is_mariadb)
+            changed = result['changed']
+
+        if admin:
+            self.role_impl.set_admin(admin)
+
+        changed = changed or members_changed
+
+        return changed
+
+    def __get_members(self):
+        """Get current role's members.
+
+        Returns:
+            set: Members.
+        """
+        if self.is_mariadb:
+            self.cursor.execute('select user, host from mysql.roles_mapping where role = %s', (self.name,))
+        else:
+            self.cursor.execute('select TO_USER as user, TO_HOST as host from mysql.role_edges where FROM_USER = %s', (self.name,))
+        return set(self.cursor.fetchall())
+
+
+def main():
+    argument_spec = mysql_common_argument_spec()
+    argument_spec.update(
+        name=dict(type='str', required=True),
+        state=dict(type='str', default='present', choices=['absent', 'present']),
+        admin=dict(type='str'),
+        priv=dict(type='raw'),
+        append_privs=dict(type='bool', default=False),
+        subtract_privs=dict(type='bool', default=False),
+        members=dict(type='list', elements='str'),
+        append_members=dict(type='bool', default=False),
+        detach_members=dict(type='bool', default=False),
+        check_implicit_admin=dict(type='bool', default=False),
+        set_default_role_all=dict(type='bool', default=True),
+        members_must_exist=dict(type='bool', default=True)
+    )
+    module = AnsibleModule(
+        argument_spec=argument_spec,
+        supports_check_mode=True,
+        mutually_exclusive=(
+            ('append_members', 'detach_members'),
+            ('admin', 'members'),
+            ('admin', 'append_members'),
+            ('admin', 'detach_members'),
+            ('append_privs', 'subtract_privs'),
+        ),
+    )
+
+    login_user = module.params['login_user']
+    login_password = module.params['login_password']
+    name = module.params['name']
+    state = module.params['state']
+    admin = module.params['admin']
+    priv = module.params['priv']
+    check_implicit_admin = module.params['check_implicit_admin']
+    connect_timeout = module.params['connect_timeout']
+    config_file = module.params['config_file']
+    append_privs = module.params['append_privs']
+    subtract_privs = module.boolean(module.params['subtract_privs'])
+    members = module.params['members']
+    append_members = module.params['append_members']
+    detach_members = module.params['detach_members']
+    ssl_cert = module.params['client_cert']
+    ssl_key = module.params['client_key']
+    ssl_ca = module.params['ca_cert']
+    check_hostname = module.params['check_hostname']
+    db = ''
+    set_default_role_all = module.params['set_default_role_all']
+    members_must_exist = module.params['members_must_exist']
+
+    if priv and not isinstance(priv, (str, dict)):
+        msg = ('The "priv" parameter must be str or dict '
+               'but %s was passed' % type(priv))
+        module.fail_json(msg=msg)
+
+    if priv and isinstance(priv, dict):
+        priv = convert_priv_dict_to_str(priv)
+
+    if mysql_driver is None:
+        module.fail_json(msg=mysql_driver_fail_msg)
+
+    cursor = None
+    try:
+        if check_implicit_admin:
+            try:
+                cursor, db_conn = mysql_connect(module, 'root', '', config_file,
+                                                ssl_cert, ssl_key, ssl_ca, db,
+                                                connect_timeout=connect_timeout,
+                                                check_hostname=check_hostname,
+                                                autocommit=True)
+            except Exception:
+                pass
+
+        if not cursor:
+            cursor, db_conn = mysql_connect(module, login_user, login_password,
+                                            config_file, ssl_cert, ssl_key,
+                                            ssl_ca, db, connect_timeout=connect_timeout,
+                                            check_hostname=check_hostname,
+                                            autocommit=True)
+
+    except Exception as e:
+        module.fail_json(msg='unable to connect to database, '
+                             'check login_user and login_password '
+                             'are correct or %s has the credentials. '
+                             'Exception message: %s' % (config_file, to_native(e)))
+
+    # Set defaults
+    changed = False
+
+    get_impl(cursor)
+
+    if priv is not None:
+        try:
+            mode = get_mode(cursor)
+        except Exception as e:
+            module.fail_json(msg=to_native(e))
+
+        try:
+            priv = privileges_unpack(priv, mode, ensure_usage=not subtract_privs)
+        except Exception as e:
+            module.fail_json(msg='Invalid privileges string: %s' % to_native(e))
+
+    server = DbServer(module, cursor)
+
+    # Check if the server supports roles
+    if not server.supports_roles():
+        msg = ('Roles are not supported by the server. '
+               'Minimal versions are MySQL 8.0.0 or MariaDB 10.0.5.')
+        module.fail_json(msg=msg)
+
+    if admin:
+        if not server.is_mariadb():
+            module.fail_json(msg='The "admin" option can be used only with MariaDB.')
+
+        admin = normalize_users(module, [admin])[0]
+        server.check_users_in_db([admin])
+
+    if members:
+        members = normalize_users(module, members, server.is_mariadb())
+        if members_must_exist:
+            server.check_users_in_db(members)
+        else:
+            members = list(server.filter_existing_users(members))
+
+    # Main job starts here
+    role = Role(module, cursor, name, server)
+
+    try:
+        if state == 'present':
+            if not role.exists:
+                if subtract_privs:
+                    priv = None  # avoid granting unwanted privileges
+                if detach_members:
+                    members = None  # avoid adding unwanted members
+                changed = role.add(members, priv, module.check_mode, admin,
+                                   set_default_role_all)
+
+            else:
+                changed = role.update(members, priv, module.check_mode, append_privs, subtract_privs,
+                                      append_members, detach_members, admin,
+                                      set_default_role_all)
+
+        elif state == 'absent':
+            changed = role.drop(module.check_mode)
+
+    except Exception as e:
+        module.fail_json(msg=to_native(e))
+
+    module.exit_json(changed=changed)
+
+
+if __name__ == '__main__':
+    main()
diff --git a/ansible_collections/community/mysql/plugins/modules/mysql_user.py b/ansible_collections/community/mysql/plugins/modules/mysql_user.py
new file mode 100644
index 000000000..e87fe12db
--- /dev/null
+++ b/ansible_collections/community/mysql/plugins/modules/mysql_user.py
@@ -0,0 +1,528 @@
+#!/usr/bin/python
+# -*- coding: utf-8 -*-
+
+# Copyright: (c) 2012, Mark Theunissen <mark.theunissen@gmail.com>
+# Sponsored by Four Kitchens http://fourkitchens.com.
+# GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt)
+
+from __future__ import absolute_import, division, print_function
+__metaclass__ = type
+
+DOCUMENTATION = r'''
+---
+module: mysql_user
+short_description: Adds or removes a user from a MySQL database
+description:
+   - Adds or removes a user from a MySQL database.
+options:
+  name:
+    description:
+      - Name of the user (role) to add or remove.
+    type: str
+    required: true
+  password:
+    description:
+      - Set the user's password. Only for C(mysql_native_password) authentication.
+        For other authentication plugins see the combination of I(plugin), I(plugin_hash_string), I(plugin_auth_string).
+    type: str
+  encrypted:
+    description:
+      - Indicate that the 'password' field is a `mysql_native_password` hash.
+    type: bool
+    default: false
+  host:
+    description:
+      - The 'host' part of the MySQL username.
+    type: str
+    default: localhost
+  host_all:
+    description:
+      - Override the host option, making ansible apply changes
+        to all hostnames for a given user.
+      - This option cannot be used when creating users.
+    type: bool
+    default: false
+  priv:
+    description:
+      - "MySQL privileges string in the format: C(db.table:priv1,priv2)."
+      - "Multiple privileges can be specified by separating each one using
+        a forward slash: C(db.table1:priv/db.table2:priv)."
+      - The format is based on MySQL C(GRANT) statement.
+      - Database and table names can be quoted, MySQL-style.
+      - If column privileges are used, the C(priv1,priv2) part must be
+        exactly as returned by a C(SHOW GRANT) statement. If not followed,
+        the module will always report changes. It includes grouping columns
+        by permission (C(SELECT(col1,col2)) instead of C(SELECT(col1),SELECT(col2))).
+      - Can be passed as a dictionary (see the examples).
+      - Supports GRANTs for procedures and functions (see the examples).
+      - "Note: If you pass the same C(db.table) combination to this parameter
+        two or more times with different privileges,
+        for example, C('*.*:SELECT/*.*:SHOW VIEW'), only the last one will be applied,
+        in this example, it will be C(SHOW VIEW) respectively.
+        Use C('*.*:SELECT,SHOW VIEW') instead to apply both."
+    type: raw
+  append_privs:
+    description:
+      - Append the privileges defined by priv to the existing ones for this
+        user instead of overwriting existing ones. Mutually exclusive with I(subtract_privs).
+    type: bool
+    default: false
+  subtract_privs:
+    description:
+      - Revoke the privileges defined by the I(priv) option and keep other existing privileges.
+        If set, invalid privileges in I(priv) are ignored.
+        Mutually exclusive with I(append_privs).
+    version_added: '3.2.0'
+    type: bool
+    default: false
+  tls_requires:
+    description:
+      - Set requirement for secure transport as a dictionary of requirements (see the examples).
+      - Valid requirements are SSL, X509, SUBJECT, ISSUER, CIPHER.
+      - SUBJECT, ISSUER and CIPHER are complementary, and mutually exclusive with SSL and X509.
+      - U(https://mariadb.com/kb/en/securing-connections-for-client-and-server/#requiring-tls).
+    type: dict
+    version_added: 1.0.0
+  sql_log_bin:
+    description:
+      - Whether binary logging should be enabled or disabled for the connection.
+    type: bool
+    default: true
+  force_context:
+    description:
+      - Sets the С(mysql) system database as context for the executed statements (it will be used
+        as a database to connect to). Useful if you use binlog / replication filters in MySQL as
+        per default the statements can not be caught by a binlog / replication filter, they require
+        a database to be set to work, otherwise the replication can break down.
+      - See U(https://dev.mysql.com/doc/refman/8.0/en/replication-options-binary-log.html#option_mysqld_binlog-ignore-db)
+        for a description on how binlog filters work (filtering on the primary).
+      - See U(https://dev.mysql.com/doc/refman/8.0/en/replication-options-replica.html#option_mysqld_replicate-ignore-db)
+        for a description on how replication filters work (filtering on the replica).
+    type: bool
+    default: false
+    version_added: '3.1.0'
+  state:
+    description:
+      - Whether the user should exist.
+      - When C(absent), removes the user.
+    type: str
+    choices: [ absent, present ]
+    default: present
+  check_implicit_admin:
+    description:
+      - Check if mysql allows login as root/nopassword before trying supplied credentials.
+      - If success, passed I(login_user)/I(login_password) will be ignored.
+    type: bool
+    default: false
+  update_password:
+    description:
+      - C(always) will update passwords if they differ. This affects I(password) and the combination of I(plugin), I(plugin_hash_string), I(plugin_auth_string).
+      - C(on_create) will only set the password or the combination of I(plugin), I(plugin_hash_string), I(plugin_auth_string) for newly created users.
+      - "C(on_new_username) works like C(on_create), but it tries to reuse an existing password: If one different user
+        with the same username exists, or multiple different users with the same username and equal C(plugin) and
+        C(authentication_string) attribute, the existing C(plugin) and C(authentication_string) are used for the
+        new user instead of the I(password), I(plugin), I(plugin_hash_string) or I(plugin_auth_string) argument."
+    type: str
+    choices: [ always, on_create, on_new_username ]
+    default: always
+  plugin:
+    description:
+      - User's plugin to authenticate (``CREATE USER user IDENTIFIED WITH plugin``).
+    type: str
+    version_added: '0.1.0'
+  plugin_hash_string:
+    description:
+      - User's plugin hash string (``CREATE USER user IDENTIFIED WITH plugin AS plugin_hash_string``).
+    type: str
+    version_added: '0.1.0'
+  plugin_auth_string:
+    description:
+      - User's plugin auth_string (``CREATE USER user IDENTIFIED WITH plugin BY plugin_auth_string``).
+      - If I(plugin) is ``pam`` (MariaDB) or ``auth_pam`` (MySQL) an optional I(plugin_auth_string) can be used to choose a specific PAM service.
+    type: str
+    version_added: '0.1.0'
+  resource_limits:
+    description:
+      - Limit the user for certain server resources. Provided since MySQL 5.6 / MariaDB 10.2.
+      - "Available options are C(MAX_QUERIES_PER_HOUR: num), C(MAX_UPDATES_PER_HOUR: num),
+        C(MAX_CONNECTIONS_PER_HOUR: num), C(MAX_USER_CONNECTIONS: num), C(MAX_STATEMENT_TIME: num) (supported only for MariaDB since collection version 3.7.0)."
+      - Used when I(state=present), ignored otherwise.
+    type: dict
+    version_added: '0.1.0'
+  session_vars:
+    description:
+      - "Dictionary of session variables in form of C(variable: value) to set at the beginning of module execution."
+      - Cannot be used to set global variables, use the M(community.mysql.mysql_variables) module instead.
+    type: dict
+    version_added: '3.6.0'
+
+notes:
+   - "MySQL server installs with default I(login_user) of C(root) and no password.
+     To secure this user as part of an idempotent playbook, you must create at least two tasks:
+     1) change the root user's password, without providing any I(login_user)/I(login_password) details,
+     2) drop a C(~/.my.cnf) file containing the new root credentials.
+     Subsequent runs of the playbook will then succeed by reading the new credentials from the file."
+   - Currently, there is only support for the C(mysql_native_password) encrypted password hash module.
+   - Supports (check_mode).
+
+seealso:
+- module: community.mysql.mysql_info
+- name: MySQL access control and account management reference
+  description: Complete reference of the MySQL access control and account management documentation.
+  link: https://dev.mysql.com/doc/refman/8.0/en/access-control.html
+- name: MySQL provided privileges reference
+  description: Complete reference of the MySQL provided privileges documentation.
+  link: https://dev.mysql.com/doc/refman/8.0/en/privileges-provided.html
+
+author:
+- Jonathan Mainguy (@Jmainguy)
+- Benjamin Malynovytch (@bmalynovytch)
+- Lukasz Tomaszkiewicz (@tomaszkiewicz)
+extends_documentation_fragment:
+- community.mysql.mysql
+
+'''
+
+EXAMPLES = r'''
+# If you encounter the "Please explicitly state intended protocol" error,
+# use the login_unix_socket argument
+- name: Removes anonymous user account for localhost
+  community.mysql.mysql_user:
+    name: ''
+    host: localhost
+    state: absent
+    login_unix_socket: /run/mysqld/mysqld.sock
+
+- name: Removes all anonymous user accounts
+  community.mysql.mysql_user:
+    name: ''
+    host_all: true
+    state: absent
+
+- name: Create database user with name 'bob' and password '12345' with all database privileges
+  community.mysql.mysql_user:
+    name: bob
+    password: 12345
+    priv: '*.*:ALL'
+    state: present
+
+- name: Create database user using hashed password with all database privileges
+  community.mysql.mysql_user:
+    name: bob
+    password: '*EE0D72C1085C46C5278932678FBE2C6A782821B4'
+    encrypted: true
+    priv: '*.*:ALL'
+    state: present
+
+# Set session var wsrep_on=off before creating the user
+- name: Create database user with password and all database privileges and 'WITH GRANT OPTION'
+  community.mysql.mysql_user:
+    name: bob
+    password: 12345
+    priv: '*.*:ALL,GRANT'
+    state: present
+    session_vars:
+      wsrep_on: off
+
+- name: Create user with password, all database privileges and 'WITH GRANT OPTION' in db1 and db2
+  community.mysql.mysql_user:
+    state: present
+    name: bob
+    password: 12345dd
+    priv:
+      'db1.*': 'ALL,GRANT'
+      'db2.*': 'ALL,GRANT'
+
+# Use 'PROCEDURE' instead of 'FUNCTION' to apply GRANTs for a MySQL procedure instead.
+- name: Grant a user the right to execute a function
+  community.mysql.mysql_user:
+    name: readonly
+    password: 12345
+    priv:
+      FUNCTION my_db.my_function: EXECUTE
+    state: present
+
+- name: Modify user to require TLS connection with a valid client certificate
+  community.mysql.mysql_user:
+    name: bob
+    tls_requires:
+      x509:
+    state: present
+
+- name: Modify user to require TLS connection with a specific client certificate and cipher
+  community.mysql.mysql_user:
+    name: bob
+    tls_requires:
+      subject: '/CN=alice/O=MyDom, Inc./C=US/ST=Oregon/L=Portland'
+      cipher: 'ECDHE-ECDSA-AES256-SHA384'
+
+- name: Modify user to no longer require SSL
+  community.mysql.mysql_user:
+    name: bob
+    tls_requires:
+
+- name: Ensure no user named 'sally'@'localhost' exists, also passing in the auth credentials
+  community.mysql.mysql_user:
+    login_user: root
+    login_password: 123456
+    name: sally
+    state: absent
+
+# check_implicit_admin example
+- name: >
+    Ensure no user named 'sally'@'localhost' exists, also passing in the auth credentials.
+    If mysql allows root/nopassword login, try it without the credentials first.
+    If it's not allowed, pass the credentials
+  community.mysql.mysql_user:
+    check_implicit_admin: true
+    login_user: root
+    login_password: 123456
+    name: sally
+    state: absent
+
+- name: Ensure no user named 'sally' exists at all
+  community.mysql.mysql_user:
+    name: sally
+    host_all: true
+    state: absent
+
+- name: Specify grants composed of more than one word
+  community.mysql.mysql_user:
+    name: replication
+    password: 12345
+    priv: "*.*:REPLICATION CLIENT"
+    state: present
+
+- name: Revoke all privileges for user 'bob' and password '12345'
+  community.mysql.mysql_user:
+    name: bob
+    password: 12345
+    priv: "*.*:USAGE"
+    state: present
+
+# Example privileges string format
+# mydb.*:INSERT,UPDATE/anotherdb.*:SELECT/yetanotherdb.*:ALL
+
+- name: Example using login_unix_socket to connect to server
+  community.mysql.mysql_user:
+    name: root
+    password: abc123
+    login_unix_socket: /var/run/mysqld/mysqld.sock
+
+- name: Example of skipping binary logging while adding user 'bob'
+  community.mysql.mysql_user:
+    name: bob
+    password: 12345
+    priv: "*.*:USAGE"
+    state: present
+    sql_log_bin: false
+
+- name: Create user 'bob' authenticated with plugin 'AWSAuthenticationPlugin'
+  community.mysql.mysql_user:
+    name: bob
+    plugin: AWSAuthenticationPlugin
+    plugin_hash_string: RDS
+    priv: '*.*:ALL'
+    state: present
+
+- name: Limit bob's resources to 10 queries per hour and 5 connections per hour
+  community.mysql.mysql_user:
+    name: bob
+    resource_limits:
+      MAX_QUERIES_PER_HOUR: 10
+      MAX_CONNECTIONS_PER_HOUR: 5
+
+- name: Ensure bob does not have the DELETE privilege
+  community.mysql.mysql_user:
+    name: bob
+    subtract_privs: true
+    priv:
+      'db1.*': DELETE
+
+# Example .my.cnf file for setting the root password
+# [client]
+# user=root
+# password=n<_665{vS43y
+'''
+
+RETURN = '''#'''
+
+from ansible.module_utils.basic import AnsibleModule
+from ansible_collections.community.mysql.plugins.module_utils.database import SQLParseError
+from ansible_collections.community.mysql.plugins.module_utils.mysql import (
+    mysql_connect,
+    mysql_driver,
+    mysql_driver_fail_msg,
+    mysql_common_argument_spec,
+    set_session_vars,
+)
+from ansible_collections.community.mysql.plugins.module_utils.user import (
+    convert_priv_dict_to_str,
+    get_impl,
+    get_mode,
+    InvalidPrivsError,
+    limit_resources,
+    privileges_unpack,
+    sanitize_requires,
+    user_add,
+    user_delete,
+    user_exists,
+    user_mod,
+)
+from ansible.module_utils._text import to_native
+
+
+# ===========================================
+# Module execution.
+#
+
+
+def main():
+    argument_spec = mysql_common_argument_spec()
+    argument_spec.update(
+        user=dict(type='str', required=True, aliases=['name']),
+        password=dict(type='str', no_log=True),
+        encrypted=dict(type='bool', default=False),
+        host=dict(type='str', default='localhost'),
+        host_all=dict(type="bool", default=False),
+        state=dict(type='str', default='present', choices=['absent', 'present']),
+        priv=dict(type='raw'),
+        tls_requires=dict(type='dict'),
+        append_privs=dict(type='bool', default=False),
+        subtract_privs=dict(type='bool', default=False),
+        check_implicit_admin=dict(type='bool', default=False),
+        update_password=dict(type='str', default='always', choices=['always', 'on_create', 'on_new_username'], no_log=False),
+        sql_log_bin=dict(type='bool', default=True),
+        plugin=dict(default=None, type='str'),
+        plugin_hash_string=dict(default=None, type='str'),
+        plugin_auth_string=dict(default=None, type='str'),
+        resource_limits=dict(type='dict'),
+        force_context=dict(type='bool', default=False),
+        session_vars=dict(type='dict'),
+    )
+    module = AnsibleModule(
+        argument_spec=argument_spec,
+        supports_check_mode=True,
+        mutually_exclusive=(('append_privs', 'subtract_privs'),)
+    )
+    login_user = module.params["login_user"]
+    login_password = module.params["login_password"]
+    user = module.params["user"]
+    password = module.params["password"]
+    encrypted = module.boolean(module.params["encrypted"])
+    host = module.params["host"].lower()
+    host_all = module.params["host_all"]
+    state = module.params["state"]
+    priv = module.params["priv"]
+    tls_requires = sanitize_requires(module.params["tls_requires"])
+    check_implicit_admin = module.params["check_implicit_admin"]
+    connect_timeout = module.params["connect_timeout"]
+    config_file = module.params["config_file"]
+    append_privs = module.boolean(module.params["append_privs"])
+    subtract_privs = module.boolean(module.params['subtract_privs'])
+    update_password = module.params['update_password']
+    ssl_cert = module.params["client_cert"]
+    ssl_key = module.params["client_key"]
+    ssl_ca = module.params["ca_cert"]
+    check_hostname = module.params["check_hostname"]
+    db = ''
+    if module.params["force_context"]:
+        db = 'mysql'
+    sql_log_bin = module.params["sql_log_bin"]
+    plugin = module.params["plugin"]
+    plugin_hash_string = module.params["plugin_hash_string"]
+    plugin_auth_string = module.params["plugin_auth_string"]
+    resource_limits = module.params["resource_limits"]
+    session_vars = module.params["session_vars"]
+
+    if priv and not isinstance(priv, (str, dict)):
+        module.fail_json(msg="priv parameter must be str or dict but %s was passed" % type(priv))
+
+    if priv and isinstance(priv, dict):
+        priv = convert_priv_dict_to_str(priv)
+
+    if mysql_driver is None:
+        module.fail_json(msg=mysql_driver_fail_msg)
+
+    cursor = None
+    try:
+        if check_implicit_admin:
+            try:
+                cursor, db_conn = mysql_connect(module, "root", "", config_file, ssl_cert, ssl_key, ssl_ca, db,
+                                                connect_timeout=connect_timeout, check_hostname=check_hostname, autocommit=True)
+            except Exception:
+                pass
+
+        if not cursor:
+            cursor, db_conn = mysql_connect(module, login_user, login_password, config_file, ssl_cert, ssl_key, ssl_ca, db,
+                                            connect_timeout=connect_timeout, check_hostname=check_hostname, autocommit=True)
+    except Exception as e:
+        module.fail_json(msg="unable to connect to database, check login_user and login_password are correct or %s has the credentials. "
+                             "Exception message: %s" % (config_file, to_native(e)))
+
+    if not sql_log_bin:
+        cursor.execute("SET SQL_LOG_BIN=0;")
+
+    if session_vars:
+        set_session_vars(module, cursor, session_vars)
+
+    get_impl(cursor)
+
+    if priv is not None:
+        try:
+            mode = get_mode(cursor)
+        except Exception as e:
+            module.fail_json(msg=to_native(e))
+        priv = privileges_unpack(priv, mode, ensure_usage=not subtract_privs)
+    password_changed = False
+    if state == "present":
+        if user_exists(cursor, user, host, host_all):
+            try:
+                if update_password == "always":
+                    result = user_mod(cursor, user, host, host_all, password, encrypted,
+                                      plugin, plugin_hash_string, plugin_auth_string,
+                                      priv, append_privs, subtract_privs, tls_requires, module)
+
+                else:
+                    result = user_mod(cursor, user, host, host_all, None, encrypted,
+                                      None, None, None,
+                                      priv, append_privs, subtract_privs, tls_requires, module)
+                changed = result['changed']
+                msg = result['msg']
+                password_changed = result['password_changed']
+
+            except (SQLParseError, InvalidPrivsError, mysql_driver.Error) as e:
+                module.fail_json(msg=to_native(e))
+        else:
+            if host_all:
+                module.fail_json(msg="host_all parameter cannot be used when adding a user")
+            try:
+                if subtract_privs:
+                    priv = None  # avoid granting unwanted privileges
+                reuse_existing_password = update_password == 'on_new_username'
+                result = user_add(cursor, user, host, host_all, password, encrypted,
+                                  plugin, plugin_hash_string, plugin_auth_string,
+                                  priv, tls_requires, module.check_mode, reuse_existing_password)
+                changed = result['changed']
+                password_changed = result['password_changed']
+                if changed:
+                    msg = "User added"
+
+            except (SQLParseError, InvalidPrivsError, mysql_driver.Error) as e:
+                module.fail_json(msg=to_native(e))
+
+        if resource_limits:
+            changed = limit_resources(module, cursor, user, host, resource_limits, module.check_mode) or changed
+
+    elif state == "absent":
+        if user_exists(cursor, user, host, host_all):
+            changed = user_delete(cursor, user, host, host_all, module.check_mode)
+            msg = "User deleted"
+        else:
+            changed = False
+            msg = "User doesn't exist"
+    module.exit_json(changed=changed, user=user, msg=msg, password_changed=password_changed)
+
+
+if __name__ == '__main__':
+    main()
diff --git a/ansible_collections/community/mysql/plugins/modules/mysql_variables.py b/ansible_collections/community/mysql/plugins/modules/mysql_variables.py
new file mode 100644
index 000000000..f404d5aab
--- /dev/null
+++ b/ansible_collections/community/mysql/plugins/modules/mysql_variables.py
@@ -0,0 +1,271 @@
+#!/usr/bin/python
+# -*- coding: utf-8 -*-
+
+# Copyright: (c) 2013, Balazs Pocze <banyek@gawker.com>
+# Certain parts are taken from Mark Theunissen's mysqldb module
+# GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt)
+
+from __future__ import absolute_import, division, print_function
+__metaclass__ = type
+
+DOCUMENTATION = r'''
+---
+module: mysql_variables
+
+short_description: Manage MySQL global variables
+description:
+- Query / Set MySQL variables.
+author:
+- Balazs Pocze (@banyek)
+options:
+  variable:
+    description:
+    - Variable name to operate.
+    type: str
+    required: true
+  value:
+    description:
+    - If set, then sets variable value to this.
+    type: str
+  mode:
+    description:
+    - C(global) assigns C(value) to a global system variable which will be changed at runtime
+      but won't persist across server restarts.
+    - C(persist) assigns C(value) to a global system variable and persists it to
+      the mysqld-auto.cnf option file in the data directory
+      (the variable will survive service restarts).
+    - C(persist_only) persists C(value) to the mysqld-auto.cnf option file in the data directory
+      but without setting the global variable runtime value
+      (the value will be changed after the next service restart).
+    - Supported by MySQL 8.0 or later.
+    - For more information see U(https://dev.mysql.com/doc/refman/8.0/en/set-variable.html).
+    type: str
+    choices: ['global', 'persist', 'persist_only']
+    default: global
+    version_added: '0.1.0'
+
+notes:
+- Does not support C(check_mode).
+
+seealso:
+- module: community.mysql.mysql_info
+- name: MySQL SET command reference
+  description: Complete reference of the MySQL SET command documentation.
+  link: https://dev.mysql.com/doc/refman/8.0/en/set-statement.html
+
+extends_documentation_fragment:
+- community.mysql.mysql
+'''
+
+EXAMPLES = r'''
+# If you encounter the "Please explicitly state intended protocol" error,
+# use the login_unix_socket argument
+- name: Check for sync_binlog setting
+  community.mysql.mysql_variables:
+    variable: sync_binlog
+    login_unix_socket: /run/mysqld/mysqld.sock
+
+- name: Set read_only variable to 1 persistently
+  community.mysql.mysql_variables:
+    variable: read_only
+    value: 1
+    mode: persist
+'''
+
+RETURN = r'''
+queries:
+  description: List of executed queries which modified DB's state.
+  returned: if executed
+  type: list
+  sample: ["SET GLOBAL `read_only` = 1"]
+  version_added: '0.1.0'
+'''
+
+import os
+import warnings
+from re import match
+
+from ansible.module_utils.basic import AnsibleModule
+from ansible_collections.community.mysql.plugins.module_utils.database import SQLParseError, mysql_quote_identifier
+from ansible_collections.community.mysql.plugins.module_utils.mysql import mysql_connect, mysql_driver, mysql_driver_fail_msg, mysql_common_argument_spec
+from ansible.module_utils._text import to_native
+
+executed_queries = []
+
+
+def check_mysqld_auto(module, cursor, mysqlvar):
+    """Check variable's value in mysqld-auto.cnf."""
+    query = ("SELECT VARIABLE_VALUE "
+             "FROM performance_schema.persisted_variables "
+             "WHERE VARIABLE_NAME = %s")
+    try:
+        cursor.execute(query, (mysqlvar,))
+        res = cursor.fetchone()
+    except Exception as e:
+        if "Table 'performance_schema.persisted_variables' doesn't exist" in str(e):
+            module.fail_json(msg='Server version must be 8.0 or greater.')
+
+    if res:
+        return res[0]
+    else:
+        return None
+
+
+def typedvalue(value):
+    """
+    Convert value to number whenever possible, return same value
+    otherwise.
+
+    >>> typedvalue('3')
+    3
+    >>> typedvalue('3.0')
+    3.0
+    >>> typedvalue('foobar')
+    'foobar'
+
+    """
+    try:
+        return int(value)
+    except ValueError:
+        pass
+
+    try:
+        return float(value)
+    except ValueError:
+        pass
+
+    return value
+
+
+def getvariable(cursor, mysqlvar):
+    cursor.execute("SHOW VARIABLES WHERE Variable_name = %s", (mysqlvar,))
+    mysqlvar_val = cursor.fetchall()
+    if len(mysqlvar_val) == 1:
+        return mysqlvar_val[0][1]
+    else:
+        return None
+
+
+def setvariable(cursor, mysqlvar, value, mode='global'):
+    """ Set a global mysql variable to a given value
+
+    The DB driver will handle quoting of the given value based on its
+    type, thus numeric strings like '3.0' or '8' are illegal, they
+    should be passed as numeric literals.
+
+    """
+    if mode == 'persist':
+        query = "SET PERSIST %s = " % mysql_quote_identifier(mysqlvar, 'vars')
+    elif mode == 'global':
+        query = "SET GLOBAL %s = " % mysql_quote_identifier(mysqlvar, 'vars')
+    elif mode == 'persist_only':
+        query = "SET PERSIST_ONLY %s = " % mysql_quote_identifier(mysqlvar, 'vars')
+
+    try:
+        cursor.execute(query + "%s", (value,))
+        executed_queries.append(query + "%s" % value)
+        cursor.fetchall()
+        result = True
+    except Exception as e:
+        result = to_native(e)
+
+    return result
+
+
+def main():
+    argument_spec = mysql_common_argument_spec()
+    argument_spec.update(
+        variable=dict(type='str'),
+        value=dict(type='str'),
+        mode=dict(type='str', choices=['global', 'persist', 'persist_only'], default='global'),
+    )
+
+    module = AnsibleModule(
+        argument_spec=argument_spec
+    )
+    user = module.params["login_user"]
+    password = module.params["login_password"]
+    connect_timeout = module.params['connect_timeout']
+    ssl_cert = module.params["client_cert"]
+    ssl_key = module.params["client_key"]
+    ssl_ca = module.params["ca_cert"]
+    check_hostname = module.params["check_hostname"]
+    config_file = module.params['config_file']
+    db = 'mysql'
+
+    mysqlvar = module.params["variable"]
+    value = module.params["value"]
+    mode = module.params["mode"]
+
+    if mysqlvar is None:
+        module.fail_json(msg="Cannot run without variable to operate with")
+    if match('^[0-9A-Za-z_.]+$', mysqlvar) is None:
+        module.fail_json(msg="invalid variable name \"%s\"" % mysqlvar)
+    if mysql_driver is None:
+        module.fail_json(msg=mysql_driver_fail_msg)
+    else:
+        warnings.filterwarnings('error', category=mysql_driver.Warning)
+
+    try:
+        cursor, db_conn = mysql_connect(module, user, password, config_file, ssl_cert, ssl_key, ssl_ca, db,
+                                        connect_timeout=connect_timeout, check_hostname=check_hostname)
+    except Exception as e:
+        if os.path.exists(config_file):
+            module.fail_json(msg=("unable to connect to database, check login_user and "
+                                  "login_password are correct or %s has the credentials. "
+                                  "Exception message: %s" % (config_file, to_native(e))))
+        else:
+            module.fail_json(msg="unable to find %s. Exception message: %s" % (config_file, to_native(e)))
+
+    mysqlvar_val = None
+    var_in_mysqld_auto_cnf = None
+
+    mysqlvar_val = getvariable(cursor, mysqlvar)
+    if mysqlvar_val is None:
+        module.fail_json(msg="Variable not available \"%s\"" % mysqlvar, changed=False)
+
+    if value is None:
+        module.exit_json(msg=mysqlvar_val)
+
+    if mode in ('persist', 'persist_only'):
+        var_in_mysqld_auto_cnf = check_mysqld_auto(module, cursor, mysqlvar)
+
+        if mode == 'persist_only':
+            if var_in_mysqld_auto_cnf is None:
+                mysqlvar_val = False
+            else:
+                mysqlvar_val = var_in_mysqld_auto_cnf
+
+    # Type values before using them
+    value_wanted = typedvalue(value)
+    value_actual = typedvalue(mysqlvar_val)
+    value_in_auto_cnf = None
+    if var_in_mysqld_auto_cnf is not None:
+        value_in_auto_cnf = typedvalue(var_in_mysqld_auto_cnf)
+
+    if value_wanted == value_actual and mode in ('global', 'persist'):
+        if mode == 'persist' and value_wanted == value_in_auto_cnf:
+            module.exit_json(msg="Variable is already set to requested value globally"
+                                 "and stored into mysqld-auto.cnf file.", changed=False)
+
+        elif mode == 'global':
+            module.exit_json(msg="Variable is already set to requested value.", changed=False)
+
+    if mode == 'persist_only' and value_wanted == value_in_auto_cnf:
+        module.exit_json(msg="Variable is already stored into mysqld-auto.cnf "
+                             "with requested value.", changed=False)
+
+    try:
+        result = setvariable(cursor, mysqlvar, value_wanted, mode)
+    except SQLParseError as e:
+        result = to_native(e)
+
+    if result is True:
+        module.exit_json(msg="Variable change succeeded prev_value=%s" % value_actual,
+                         changed=True, queries=executed_queries)
+    else:
+        module.fail_json(msg=result, changed=False)
+
+
+if __name__ == '__main__':
+    main()