Testing Azure MySQL database connectivity with Python

Ovidiu Borlean
2 min readNov 23, 2022
Photo by Rubaitul Azad on Unsplash

When customers complains about connectivity issues towards the Azure MySQL endpoints, we developed a simple Python script that is continuously testing respective endpoint and save the timestamped output both in stdout and local file.

Prerequisites

For running this script, we need to deploy a testing Pod on AKS cluster where we will upload our Python script along with the necessary libraries. For our testing, we used the classic nginx image.

kubectl run nginx --image=nginx
kubectl exec -it nginx -- bash

Installation

After our Nginx pod is in running state and we connect to shell, we need to initialize repositories:

apt update
mkdir /app
cd /app

cat << EOF ./mysqltest.py
import mysql.connector
from mysql.connector import errorcode
import os
import time
from datetime import datetime

def checkMySql():
try:
now = datetime.now()
dt_string = now.strftime("%d/%m/%Y %H:%M:%S")
conn = mysql.connector.connect(**config)
print(dt_string + " " + "Connection established to MySQL Server")
conn.close()
file1= open("mysqltest.log","a")
content = dt_string + " " + "Connection established to MySQL Server"
file1.write(content)
file1.write("\n")
file1.close()
except mysql.connector.Error as err:
if err.errno == errorcode.ER_ACCESS_DENIED_ERROR:
print("Something is wrong with the user name or password")
file1= open("mysqltest.log","a")
content = dt_string + " " + "Something is wrong with the user name or password"
file1.write(content)
file1.write("\n")
file1.close()
elif err.errno == errorcode.ER_BAD_DB_ERROR:
print("Database does not exist")
file1= open("mysqltest.log","a")
content = dt_string + " " + "Database does not exist"
file1.write(content)
file1.write("\n")
file1.close()
else:
print(err)
file1= open("mysqltest.log","a")
content = dt_string + " " + str(err)
file1.write(content)
file1.write("\n")
file1.close()
#else:
# cursor = conn.cursor()
# cursor.execute("DROP TABLE IF EXISTS inventory;")
# print("Finished dropping table (if existed).")
# cursor.execute("CREATE TABLE inventory (id serial PRIMARY KEY, name VARCHAR(50), quantity INTEGER);")
# print("Finished creating table.")
# # Insert some data into table
# cursor.execute("INSERT INTO inventory (name, quantity) VALUES (%s, %s);", ("banana", 150))
# print("Inserted",cursor.rowcount,"row(s) of data.")
# cursor.execute("INSERT INTO inventory (name, quantity) VALUES (%s, %s);", ("orange", 154))
# print("Inserted",cursor.rowcount,"row(s) of data.")
# cursor.execute("INSERT INTO inventory (name, quantity) VALUES (%s, %s);", ("apple", 100))
# print("Inserted",cursor.rowcount,"row(s) of data.")
# Cleanup
# conn.commit()
# cursor.close()
# conn.close()
# print("Done.")

if __name__ == '__main__':
myHostname = os.getenv('SQLSERVER')
mySQLUser = os.getenv("SQLUSER")
myPassword = os.environ.get('SQLPASS')
myDB = os.environ.get('SQLDB')
interval = os.environ.get('TIMEINTERVAL')
config = {
'host':myHostname,
'user':mySQLUser,
'password':myPassword,
'database':myDB,
'client_flags': [mysql.connector.ClientFlag.SSL],
'ssl_ca': '/home/user/mysql/DigiCertGlobalRootG2.crt.pem'
}
print("Starting MySQL Connectivity Check:...")
while True:
checkMySql()
time.sleep(int(interval))
EOF

Before running our script, we need to add MySQL connectivity details into environment variables. We can do this as follows:

export SQLSERVER="yourMySQLSERVER"
export SQLUSER="yourMySQLUser"
export SQLPASS="yourMySQLPassword"
export DBNAME="yourDBName"
export TIMEINETERVAL="5" # Desired time interval in sec between tests

GitHub Repository: https://github.com/OvidiuBorlean/azmysqltest

--

--