Oracle Linux and MySQL, Kafka and Python Flask Monitoring API

In this entry we will dig deeper into the previous post that dealt with invoking an Oracle MySQL stored procedure using Python. The focus of this entry is creating a Python API to monitor Kafka and MySQL running on an Oracle Linux VM. This API can be invoked from a User Interface that will allow the user to check the statuses of these different components.

To create a Python API that will execute the commands on your Oracle Linux or RHEL system to check the status of MySQL, Zookeeper, and Kafka, you can use the subprocess module in Python to run shell commands. 

Below is an example of how you can implement this.

Step-by-Step Implementation:

• Create Python Functions to Check Status: Each function will execute the corresponding system command using subprocess.run and return the output.
• Set Up Flask API: We'll use Flask to create a simple API that the UI can call to retrieve the status.

Python Code:

import subprocess
from flask import Flask, jsonify
app = Flask(__name__)

# Function to check MySQL status
def check_mysql_status():
try:
result = subprocess.run(['sudo', 'systemctl', 'status', 'mysqld'], capture_output=True, text=True)
return result.stdout
except subprocess.CalledProcessError as e:
return str(e)

# Function to check Zookeeper status
def check_zookeeper_status():
try:
result = subprocess.run(['sudo', 'systemctl', 'status', 'zookeeper'], capture_output=True, text=True)
return result.stdout
except subprocess.CalledProcessError as e:
return str(e)

# Function to check Kafka status
def check_kafka_status():
try:
result = subprocess.run(['sudo', 'systemctl', 'status', 'kafka'], capture_output=True, text=True)
return result.stdout
except subprocess.CalledProcessError as e:
return str(e)

# Flask API route to get MySQL status
@app.route('/status/mysql', methods=['GET'])
def get_mysql_status():
status = check_mysql_status()
return jsonify({'service': 'MySQL', 'status': status})

# Flask API route to get Zookeeper status
@app.route('/status/zookeeper', methods=['GET'])
def get_zookeeper_status():
status = check_zookeeper_status()
return jsonify({'service': 'Zookeeper', 'status': status})

# Flask API route to get Kafka status
@app.route('/status/kafka', methods=['GET'])
def get_kafka_status():
status = check_kafka_status()
return jsonify({'service': 'Kafka', 'status': status})

if __name__ == "__main__":
app.run(host='0.0.0.0', port=5000)

Explanation:

• subprocess.run: Executes the systemctl commands to check the status of MySQL, Zookeeper, and Kafka. The capture_output=True argument captures the output, while text=True ensures the output is returned as a string.
• Flask: Provides an API endpoint for each service, which the UI can call to check the respective statuses.
• Routes: Each API route (/status/mysql, /status/zookeeper, /status/kafka) responds to a GET request and returns the status of the requested service in JSON format.

Running the API:

To run the Flask API, ensure Flask is installed:

pip install Flask

To Start the Application:

python your_script_name.py

Creating the UI:

For the UI, you can use any front-end technology (HTML, React, etc.) and have buttons that call these API endpoints to display the status of each service.

For example:

• A button for MySQL could call /status/mysql.
• A button for Kafka could call /status/kafka.
• A button for Zookeeper could call /status/zookeeper.

Note on Permissions:

Ensure that the user running the Python script has the appropriate permissions to run the systemctl commands using sudo. You may need to modify the sudoers file to allow passwordless sudo for these commands.

Connecting to an Oracle MySQL Database and Invoking a Stored Procedure using Python

In this entry, we will explore how to use Python to connect to an Oracle MySQL database and invoke a stored procedure. This is particularly useful when you're looking to interact with your database programmatically—whether it's for inserting data, querying, or managing business logic through stored procedures.

We will walk through a Python script that connects to the database, generates some random test data, in this case telemetry data for school busses, and invokes a stored procedure to insert the data into multiple tables. Below is the full Python code, and we’ll break down each part of the code in detail to help you understand how it works.

This assumes that you have Oracle MySQL Running, a stored procedure to call, and the proper credentials to access your database, you can modify this script and use a stored procedure of your own!

Here's the full Python script that connects to the MySQL database, generates random test data, and calls the stored procedure InsertEvents.

Script:

import mysql.connector

from mysql.connector import errorcode

import random

import datetime

# Database connection parameters

config = {

    'user': 'yourusername',

    'password': 'yourpassword',

    'host': 'ip address of vm running mysql',

    'port': port you are using,

    'database': 'name of your mysql database',

}

# Function to generate random test data

def generate_test_data():

    asset_id = random.randint(1000, 9999)

    happened_at_time = datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')

    latitude = round(random.uniform(-90, 90), 6)

    longitude = round(random.uniform(-180, 180), 6)

    heading_degrees = random.randint(0, 360)

    accuracy_meters = round(random.uniform(0.5, 20.0), 2)

    geofence = 'TestGeofence' + str(random.randint(1, 100))

    gps_speed = round(random.uniform(0, 30), 2)

    ecu_speed = round(random.uniform(0, 30), 2)

    return (asset_id, happened_at_time, latitude, longitude, heading_degrees, accuracy_meters, geofence, gps_speed, ecu_speed)

# Function to insert event data by calling the stored procedure InsertEvents

def insert_event_data(cursor, data):

    try:

        cursor.callproc('InsertEvents', data)

        print(f"Successfully inserted event for asset_id {data[0]}")

    except mysql.connector.Error as err:

        print(f"Error: {err}")

        return False

    return True

# Main function to connect to the database and insert events

def main():

    try:

        # Connect to the database

        cnx = mysql.connector.connect(**config)

        cursor = cnx.cursor()

        # Ask the user how many events to generate

        num_events = int(input("Enter the number of events to generate: "))

        # Generate and insert the specified number of test events

        for _ in range(num_events):

            test_data = generate_test_data()

            if insert_event_data(cursor, test_data):

                cnx.commit()

            else:

                cnx.rollback()

        cursor.close()

        cnx.close()

    except mysql.connector.Error as err:

        if err.errno == errorcode.ER_ACCESS_DENIED_ERROR:

            print("Something is wrong with your username and/or password")

        elif err.errno == errorcode.ER_BAD_DB_ERROR:

            print("Database does not exist or is not reachable")

        else:

            print(err)

if __name__ == "__main__":

    main()

Inserting data using the stored procedure:

The insert_event_data() function calls the stored procedure InsertEvents and passes in the data.

cursor.callproc('InsertEvents', data): This method invokes the stored procedure InsertEvents with the data tuple as input parameters.

If the procedure is successful, a success message is printed; otherwise, an error message is displayed.

Connecting to the MySQL Database:

The main () function handles the database connection and inserts a user-specified number of events.

The mysql.connector.connect(**config) establishes a connection to the MySQL database using the provided configuration.

The script asks the user to input the number of events they wish to generate (this will prompt the user in the console).

It then generates the random data and inserts it by invoking the stored procedure. The transactions are committed or rolled back based on whether the insertion was successful.

To run the script:

Make sure your database connection parameters are correct.

Ensure that the stored procedure, like InsertEvents, is defined in your MySQL database.

Run the Python script, which will prompt you to enter the number of events to generate and insert them into the database.

Wrapping up:

This script demonstrates a simple yet powerful way to connect to an Oracle MySQL database using Python and invoke a stored procedure. With a little adjustment, you can use it for various operations, such as fetching data, updating records, or automating tasks. By leveraging Python and MySQL Connector, you can efficiently manage your data and workflows, at no cost! Remember that you can create an always free account in Oracle Cloud Infrastructure (OCI), and provision an always free Compute instance, I recommend Linux and then running MySQL and Python there, all free!