set-osd-service-name.py

#!/usr/bin/env python3
#
# This script sets the service name (i.e. drive group) for the given OSDs
# on the host on which it is executed.
#
# For example, to set service_name=osd.foo for OSDs 1, 2 and 3, you'd run:
#
#   ./set-osd-service-name.py osd.foo 1 2 3
#
# For each OSD specified, it will set service_name in that OSD's unit.meta
# file.  For these changes to be reflected immediately in the Ceph dashboard
# and orchestrator CLI, run `ceph orch ls --refresh`.
# 

import argparse
import json
import logging
import os
import pwd
import sys
import uuid
from typing import List, Optional
from logging import Logger


logging.basicConfig(level=logging.INFO, format='%(levelname)s: %(message)s')
logger: Logger = logging.getLogger(__name__)


def is_fsid(s: str) -> bool:
    # This is lifted from cephadm
    try:
        uuid.UUID(s)
    except ValueError:
        return False
    return True


def set_osd_service_name(new_service_name: str, osd_ids: List, fsid: Optional[str] = None) -> bool:
    if fsid is None:
        found_fsid: bool = False
        for i in os.listdir("/var/lib/ceph"):
            if is_fsid(i):
                if not found_fsid:
                    found_fsid = True
                    fsid = i
                else:
                    # Multiple FSIDs, bail out
                    logger.error("Unable to determine cluster FSID.  Please specify with --fsid.")
                    return False

    for osd_id in sorted(osd_ids):
        meta_path = f"/var/lib/ceph/{fsid}/osd.{osd_id}/unit.meta"
        if not os.path.exists(meta_path):
            logger.error(f"File does not exist: {meta_path}")
            logger.error("Perhaps the OSD ID or cluster FSID is incorrect?")
            return False
        meta_path_new = f"{meta_path}.new"
        with open(meta_path, 'r') as f:
            meta = json.load(f)
        meta["service_name"] = new_service_name if new_service_name else None
        with open(meta_path_new, 'w') as n:
            json.dump(meta, n, indent=4)
            os.fchmod(n.fileno(), 0o600)
            s_pwd = pwd.getpwnam("ceph")
            os.fchown(n.fileno(), s_pwd.pw_uid, s_pwd.pw_gid)
            os.rename(meta_path_new, meta_path)

    return True


if __name__ == "__main__":
    parser = argparse.ArgumentParser(description="Set the service name for the given OSDs.")
    parser.add_argument("service_name", help="value to set (i.e. name of OSD Service Spec / Drive Group)")
    parser.add_argument("osd_ids", metavar="ID", nargs="+", help="OSD ID (just the number, e.g.: 0, 1, etc.)")
    parser.add_argument("--fsid", help="cluster FSID (usually not necessary to specify)")
    args = parser.parse_args()
    if set_osd_service_name(args.service_name, args.osd_ids, args.fsid):
        sys.exit(0)
    else:
        sys.exit(1)