Python通过librbd操作ceph

# -*- coding: utf-8 -*-
"""
@Time    : 2022/2/28 10:07
@Author  : summer
@File    : test_librbd.py
@Software: PyCharm
"""

import rados, rbd


def main():
    try:
        cluster = rados.Rados(conffile='/etc/ceph/ceph.conf')
    except TypeError as e:
        print 'Argument validation error: ', e
        raise e
    print "------------------------------"
    print "Created cluster handle."

    try:
        cluster.connect()
    except Exception as e:
        print "Connectoin error: ", e
        raise e
    finally:
        print "------------------------------"
        print "Connected to the cluster."
        print "Start Test LibRbd.\n"
        return cluster


def cluster_shutdown(cluster):
    print "------------------------------"
    print "Shutting down th handle."
    cluster.shutdown()


def cluster_stats(cluster):
    print "------------------------------"
    print "Cluster Status:"
    stats = cluster.get_cluster_stats()
    print stats
    for key, value in stats.iteritems():
        print str(key) + " -> " + str(value)


def pool_list(cluster):
    print "------------------------------"
    print "List Available Pools:"
    pools = cluster.list_pools()
    for pool in pools:
        print pool


def pool_create(cluster, pool_name):
    if cluster.pool_exists(pool_name):
        print "------------------------------"
        print "Pool carete fail. Pool name is exist."
    else:
        print "------------------------------"
        print("Create {name} Pool".format(
            name=pool_name
        ))
        cluster.create_pool(pool_name)


def pool_delete(cluster, pool_name):
    if cluster.pool_exists(pool_name):
        print "------------------------------"
        print("Delete {name} Pool".format(
            name=pool_name
        ))
        cluster.delete_pool(pool_name)
    else:
        print "Pool delete fail. Pool name is not exist."


def image_create(cluster, pool_name, image_name):
    print "------------------------------"
    print("Create {image_name} in {name} Pool".format(
        image_name=image_name,
        name=pool_name
    ))
    ioctx = cluster.open_ioctx(pool_name)
    try:
        rbd_inst = rbd.RBD()
        size = 1024 ** 3
        rbd_inst.create(ioctx, image_name, size)
        image = rbd.Image(ioctx, image_name)
        try:
            data = 'foo' * 200
            image.write(data, 0)
        except Exception as e:
            print "Write error: ", e
            raise e
        finally:
            image.close()
    finally:
        ioctx.close()


def image_list(cluster, pool_name):
    print "------------------------------"
    print("List Available Images in {pool_name} Pool".format(
        pool_name=pool_name,
    ))
    ioctx = cluster.open_ioctx(pool_name)
    try:
        rbd_inst = rbd.RBD()
        rbd_list = rbd_inst.list(ioctx)
        for r in rbd_list:
            print r
    except Exception as e:
        print "list error: ", e
        raise e
    finally:
        ioctx.close()


def image_remove(cluster, pool_name, image_name):
    print "------------------------------"
    print("Remove {image_name} in {pool_name} Pool".format(
        pool_name=pool_name,
        image_name=image_name
    ))
    ioctx = cluster.open_ioctx(pool_name)
    try:
        rbd_inst = rbd.RBD()
        rbd_inst.remove(ioctx, image_name)
    except Exception as e:
        print "remove error: ", e
        raise e
    finally:
        ioctx.close()


if __name__ == '__main__':
    # connect cluster
    cluster = main()
    cluster_stats(cluster)

    # create test pool
    pool_list(cluster)
    pool_create(cluster, 'test_rbd')
    pool_list(cluster)

    # create test image and remove it
    image_create(cluster, 'test_rbd', 'test_image')
    image_list(cluster, 'test_rbd')
    image_remove(cluster, 'test_rbd', 'test_image')
    image_list(cluster, 'test_rbd')

    # delete pool
    # pool_delete(cluster, 'test_rbd')

    # disconnect cluster
    cluster_shutdown(cluster)