#!/usr/libexec/isilon/isi_drive_repurpose_d

# isi_drive_repurpose_d
# =====================
#
# Migrates one diskpool at a time towards the configured state. Currently that
# configuration is 'do we use L3 for ssds?' If so, we migrate any storage SSDs
# in diskpools to L3; if not, we migrate any L3 SSDs back to storage.


import contextlib
import errno
import fcntl
import json
import os
import os.path
import random
import select
from signal import SIGHUP, SIGPIPE, SIGALRM, SIGWINCH, SIGINFO, SIGURG, \
        SIGIO, SIGUSR1, SIGUSR2, SIGINT, SIGTERM, SIGQUORUM, SIGKILL
import socket
import sys
import syslog
from syslog import LOG_ERR, LOG_INFO, LOG_CRIT, LOG_PERROR, LOG_NOWAIT, \
        LOG_USER, LOG_PID, LOG_NDELAY, LOG_DEBUG, LOG_WARNING, LOG_NOTICE
import threading
import time
import traceback


import isi.app.daemon
import isi.app.lib
import isi.app.lib.cluster
import isi.app.lib.upgrade_api
import isi.config.arrcfg
import isi.error
import isi.fs.diskpool
import isi.fs.drive_purposing
import isi.fs.job
import isi.fs.procoptions
import isi.hw.drived
import isi.misc
import isi.smartpools
import isi.smartpools.diskpools
import isi.smartpools.filepools as spfp

# XXX Remaining: User-visible progress, track completion time??? Something like:
# Upgrading X drives and Y TB took ZZZ time; queued: X' drives (Y' TB).
# Downgrading ... (similar)


AllowedUpgradeKinds = ("l3", "revertl3")
IfsVarDirectory = "/ifs/.ifsvar"
_StateDirectory = "modules/drive_repurpose_d"
StateDirectory = IfsVarDirectory + "/" + _StateDirectory
StateLockFile = StateDirectory + "/state.lk"
StateFile = StateDirectory + "/state.json"
LeaderFile = StateDirectory + "/leader.pid.lk"
StateFileNew = StateFile + ".new"
MinStateVersion = 1
StateVersion = 1
LeaderScanWait = 2.5*60  # seconds

STAGE_BEGIN = 0

KQ_VNODE_ALL = select.KQ_NOTE_DELETE | select.KQ_NOTE_WRITE | \
        select.KQ_NOTE_EXTEND | select.KQ_NOTE_LINK | \
        select.KQ_NOTE_RENAME | select.KQ_NOTE_REVOKE

Debug = False

class MigrationAborted(Exception):
    pass

#===========================================================================
# isi_drive_repurpose_d:
#   - If invoked with no arguments (typically by MCP), becomes a daemon that
#     processes diskpool upgrades
#===========================================================================



def usage():
    print "isi_drive_repurpose_d: Drive repurposing daemon"
    print ""
    print "  Usage: configure in SmartPools WebUI"
    print "    -D: Debug: run in foreground"



def main():
    global Debug
    argv = sys.argv[1:]
    while argv:
        arg = argv.pop(0)
        if arg == "-D":
            Debug = True
            continue

        usage()
        sys.exit(1)

    # MCP doesn't tolerate daemons staying in the foreground. So, we must
    # background ourselves.

    # Backgrounds, locks pidfile /var/run/isi_drive_repurpose_d.pid, and writes
    # our pid out Note: Backgrounding must happen before pretty much everything
    # else.
    if not Debug:
        isi.app.daemon.daemonize()

    os.umask(0)

    syslog.openlog("isi_drive_repurpose_d", LOG_PID | LOG_NDELAY | LOG_PERROR,
            LOG_USER)

    syslog.syslog(LOG_INFO, "Daemon starting")
    run_as_daemon()



#===========================================================================
# Auxiliary routines
#===========================================================================



def log_exc(e):
    tb = traceback.format_exc()
    syslog.syslog(LOG_WARNING, "Exception: %s: %s" % (str(e), tb))
    log = "%s: %s" % (str(e), tb)
    for line in log.strip().split("\n"):
        syslog.syslog(LOG_INFO, line)



def log_drived_exc(e, abort=False):
    tb = traceback.format_exc()
    prio = LOG_INFO
    if abort:
        prio = LOG_ERR
    syslog.syslog(prio, "Exception from isi_drive_d: %s: %s" % (str(e), tb))
    if abort:
        syslog.syslog(prio, "Critical error, aborting.")
        sys.exit(1)


def online_lnns():
    cluster = isi.app.lib.cluster.Cluster()
    online = cluster.online()

    lnns = [x.lnn() for x in online]
    return set(lnns)



def ldnum_query(*args):
    di = None
    ok = True
    try:
        di = isi.hw.drived.ldnumQuery(*args)
    except isi.hw.drived.Error as e:
        ok = False
        log_drived_exc(e)

    return (di, ok)



def bay_query(*args):
    di = None
    ok = True
    try:
        di = isi.hw.drived.bayQuery(*args)
    except isi.hw.drived.Error as e:
        ok = False
        log_drived_exc(e)

    return (di, ok)



def block_signals():
    # XXX: This makes debugging / killing w/out -9 annoying, and I'm not sure
    # it's useful in Python programs, where signals can only land in the main thread.

    # On our platform, sigprocmask(2) is equivalent to pthread_sigmask(3).
    #isi.misc.sigprocmask(isi.misc.SIG_BLOCK, [SIGHUP, SIGPIPE, SIGALRM,
    #    SIGWINCH, SIGINFO, SIGURG, SIGIO, SIGUSR1, SIGUSR2, SIGINT, SIGTERM])
    pass



LeaderPidfile = None
IAmLeader = False
def i_am_leader():  # -> bool
    global LeaderPidfile, IAmLeader

    if not IAmLeader:
        psuedo_pid = "devid:%d, pid:%d" % (isi.config.arrcfg.get_local_devid(),
                os.getpid())

        LeaderPidfile = isi.app.daemon.PidFile(filename=LeaderFile)
        try:
            LeaderPidfile.write(psuedo_pid, die_on_error=False)
            IAmLeader = True
        except IOError as ioe:
            if ioe.errno not in (errno.EACCES, errno.EAGAIN):
                raise

    return IAmLeader



def die(*msgs):
    syslog.syslog(LOG_CRIT, "Error: " + " ".join((str(msg) for msg in msgs)))
    sys.exit(1)



# Throws isi.hw.drived.Error
def diskinfo_fmt_from_lnn_ldnum(drvid):
    lnn, ldnum = drvid

    di = isi.hw.drived.ldnumQuery(ldnum, lnn)
    ds = "lnn%d:lnum%d" % (lnn, ldnum)
    return (di, ds)



@contextlib.contextmanager
def diskpool(dpid, missingok=False):
    """
    Exposes a diskpool in the DiskPoolDB via a contextmanager. Example usage:

      dpid = 15
      with diskpool(dpid) as dp:
        print dp.ssds

    Users can write changes back to the DB with dp.dpdb.write_db().
    """

    db = isi.fs.diskpool.DiskPoolDB()
    dp = db.find_entry_id(dpid)
    if dp is None and not missingok:
        die("No such diskpool %d!" % dpid)

    yield dp



def try_write_db(dpdb):  # -> Write succeeded
    """
    Given a diskpoolDB object, attempts to write the (modified) database.

    On success, returns True.
    On errors which indicate the caller should retry the modification, returns False.

    Otherwise, raises an exception.
    """

    try:
        dpdb.write_db()
        return True
    except OSError as e:
        if e.errno not in (errno.ESTALE, errno.EAGAIN, errno.ENOENT):
            raise
    except isi.error.IsiSystemError as e:
        if e.error not in (errno.ESTALE, errno.EAGAIN, errno.ENOENT):
            raise
    return False



def db_insanity_layout_evacuate(dpdb):
    for pool in dpdb.get_disk_pools():
        if not pool.layout_evacuate:
            return

    raise AssertionError("DiskpoolDB is insane -- all pools are non-writable!")



#===========================================================================
# Functions to manipulate/persist/interpret daemon state
#===========================================================================



def create_state_directory():
    if os.path.exists(StateDirectory):
        return

    if not os.path.exists(IfsVarDirectory):
        die(".ifsvar does not exist -- maybe /ifs isn't mounted?")

    try:
        os.makedirs(StateDirectory, 0755)
    except OSError as e:
        if e.errno != errno.EEXIST:
            raise



def state_read(lk):
    assert type(lk) is isi.app.lib.safelockfile
    assert not lk.closed

    state = {}
    try:
        with open(StateFile, "r") as sf:
            state = json.load(sf)
    except IOError as ioe:
        if ioe.errno != errno.ENOENT:
            raise

    if 'version' in state:
        assert MinStateVersion <= state['version'] <= StateVersion
        # XXX state version upgrade stuff goes here
        del state['version']

    if 'queue' not in state:
        state['queue'] = []

    if 'history' not in state:
        state['history'] = []

    return state



def state_write(lk, state):
    assert type(lk) is isi.app.lib.safelockfile
    assert not lk.closed

    state['version'] = StateVersion
    with open(StateFileNew, "w") as sf:
        json.dump(state, sf)

    os.rename(StateFileNew, StateFile)



@contextlib.contextmanager
def state_locked_rw():
    with isi.app.lib.safelockfile(StateLockFile, mode='w') as lkf:
        yield (lkf, state_read(lkf))



#===========================================================================
# Upgrade daemon routines
#===========================================================================



def queue_contains_l3_upgrade(q, dpid):
    for e in q:
        if e['kind'] != "l3":
            continue

        if e['dpid'] == dpid:
            return True

    return False



def queue_add_l3_downgrade(q, devid, ldnum):
    # wk => {
    #   'kind': "revertl3",
    #   'deviddrives': [
    #     {
    #       'devid': devid,
    #       'drives': [
    #         ldnum1,
    #         ...
    #       ]
    #     }
    #   }

    wk = None
    for e in q:
        if e['kind'] != "revertl3":
            continue

        wk = e
        break

    if wk is None:
        wk = {"kind": "revertl3", "deviddrives": []}
        q.append(wk)

    node = None
    for n in wk["deviddrives"]:
        if n["devid"] != devid:
            continue

        node = n
        break

    if node is None:
        node = {"devid": devid, "drives": []}
        wk["deviddrives"].append(node)

    if ldnum not in node["drives"]:
        node["drives"].append(ldnum)



# Don't log about a down node multiple times
logged_down_lnns = set()

def scan_for_work():
    while isi.app.lib.upgrade_api.upgrade_started():
        time.sleep(5)

    with state_locked_rw() as (slock, state):
        # XXX Why not just throw away the entire saved queue? We'll correct
        # anything outstanding, and it's relatively cheap.
        state['queue'] = []
        state_write(slock, state)

    pdb = isi.fs.drive_purposing.DrivePurposeDB()
    dpdb = isi.fs.diskpool.DiskPoolDB()

    # Don't try to find work if we're in the middle of an upgrade
    if not dpdb.fully_upgraded:
        return

    new_queue = []

    if Debug:
        syslog.syslog(LOG_CRIT, "XXX Scanning %s nodeinfos" % len(dpdb.node_infos))
    for (lnn, ni) in dpdb.node_infos.iteritems():
        devid = ni['devid']

        ee = pdb.find_equivalence_entry_for_devid(devid)
        if ee is None:
            # This can happen if we race purposing stuff (bug 115265). It's
            # fine, we will just catch this node on the next scan.
            continue

        # Skip nodes that don't support L3
        if pdb.devid_supports_l3(devid) != pdb.L3_SUPPORTED:
            continue

        if lnn not in online_lnns():
            if lnn not in logged_down_lnns:
                logged_down_lnns.add(lnn)
                syslog.syslog(LOG_NOTICE, "Ignoring lnn%s: offline" % lnn)
            continue

        # If the node comes back, log again if it goes down again.
        if lnn in logged_down_lnns:
            logged_down_lnns.remove(lnn)

        # Skip down, errored, etc drives
        acceptable_states = [isi.hw.drived.USR_NEW,
                isi.hw.drived.USR_PREV_USED,
                isi.hw.drived.USR_ACTIVE]
        try:
            diskinfos = [di for di in isi.hw.drived.getDrives(lnn) if
                    di.present and di.usr_state in acceptable_states]
        except EnvironmentError as enverr:
            # Can't connect to drive_d on a node? This can happen when
            # nodes go down. Skip this node.
            log_drived_exc(enverr)
            syslog.syslog(LOG_NOTICE, ("Can't scan for work on lnn%s: " +
                    "isi_drive_d is down") % lnn)
            continue

        if Debug:
            syslog.syslog(LOG_CRIT, "XXX Scanning %s drives" % len(diskinfos))
        for di in diskinfos:
            # Diskpools doesn't report SSD accurately for L3 drives
            #is_ssd = dpdb.devid_drive_is_ssd((devid, di.ldnum))

            curtype = di.usr_purpose
            if curtype not in (isi.hw.drived.USR_STORAGE,
                    isi.hw.drived.USR_L3_CACHE):
                if curtype == isi.hw.drived.USR_NO_PURPOSE:
                    continue
                syslog.syslog(LOG_WARNING, "Found unrecognized purpose drive"+\
                        "(devid:%s, lnum:%s, bay:%s, purpose:%s), skipping" % (
                            devid, di.ldnum, di.bay, curtype))
                continue

            desired_purpose = pdb.desired_purpose(devid, di)
            if desired_purpose == isi.hw.drived.USR_PURPOSE_UNKNOWN:
                continue

            cur_l3 = (curtype == isi.hw.drived.USR_L3_CACHE)
            desired_l3 = (desired_purpose == isi.hw.drived.USR_L3_CACHE)

            if cur_l3 and not desired_l3:
                # if drive is L3 and should be storage: enqueue ->storage
                # migration for (dev,drv)
                queue_add_l3_downgrade(new_queue, devid, di.ldnum)

            elif not cur_l3 and desired_l3:
                dp = dpdb.find_disk_pool_for_devid_drive((devid, di.ldnum))
                if dp is None:
                    syslog.syslog(LOG_ERR, ("STORAGE drive (devid:%s, " +
                            "lnum:%s, bay:%s) is not part of any " +
                            "DiskPool. Skipping this drive.") % (devid,
                                di.ldnum, di.bay))
                    continue

                if not dp.ssds:
                    continue

                # if drive is storage and nodepool should be L3: queue drive's diskpool
                if not queue_contains_l3_upgrade(new_queue, dp.entryid):
                    new_queue.append({'kind': "l3", 'dpid': dp.entryid})

    if new_queue:
        syslog.syslog(LOG_WARNING, "New work queue: %s" % repr(new_queue))

        with state_locked_rw() as (slock, state):
            state['queue'] = new_queue
            state_write(slock, state)



def try_dequeue_work():
    # If nothing is running, dequeues from the run queue and saves the work as
    # active. If nothing is found, active isn't touched.
    db = isi.fs.diskpool.DiskPoolDB()

    while True:
        with state_locked_rw() as (slock, state):
            try:
                if 'active' in state:
                    return

                if len(state['queue']) == 0:
                    return

                wk = state['queue'].pop(0)
                wk['stage'] = STAGE_BEGIN

                kind = wk['kind']
                if kind == "l3":
                    dpid = wk['dpid']

                    dp = db.find_entry_id(dpid)
                    if dp is None:
                        syslog.syslog(LOG_ERR, "Diskpool %d does not exist, skipping upgrade" % dpid)
                        state_write(slock, state)
                        continue

                wk['start'] = time.time()
                state['active'] = wk
                state_write(slock, state)

                if kind == "l3":
                    syslog.syslog(LOG_INFO, "Starting upgrade %d to %s" %
                            (wk['dpid'], kind))
                else:
                    syslog.syslog(LOG_INFO, "Starting downgrade to storage")

            except Exception as e:
                return (False, "Error queueing upgrade: %s" % str(e))



def work_update_resume(wk, stage):
    with state_locked_rw() as (slock, state):
        assert 'active' in state
        if wk['kind'] == "l3":
            assert state['active']['dpid'] == wk['dpid']
        else:
            assert (state['active']['deviddrives'] ==
                    wk['deviddrives'])

        assert state['active'].get('stage', STAGE_BEGIN) < stage

        state['active']['stage'] = stage
        wk['stage'] = stage

        state_write(slock, state)



def work_persist_finished(wk, aborted=False):
    with state_locked_rw() as (slock, state):
        assert 'active' in state
        if wk['kind'] == "l3":
            assert state['active']['dpid'] == wk['dpid']
        else:
            assert (state['active']['deviddrives'] ==
                    wk['deviddrives'])

        wk['end'] = time.time()
        wk['aborted'] = aborted
        duration = wk['end'] - wk['start']
        if duration < 0:
            duration = 0
        syslog.syslog(LOG_INFO, "Finished migration in %d seconds" % duration)
        state['history'].append(wk)

        del state['active']

        state_write(slock, state)



def try_resume():
    kind = None
    wk = None

    with state_locked_rw() as (slock, state):
        if 'active' not in state:
            return False

        wk = state['active']

    kind = wk['kind']

    assert kind in AllowedUpgradeKinds

    if kind == "l3":
        resume_l3(wk)
    elif kind == "revertl3":
        resume_revert_l3(wk)

    return True



def make_dp_kev(dpfdp):
    # Attempt to watch the lazily-created DB file. If it doesn't exist, we'll
    # watch the directory. Every time we're woken on the directory, we'll try
    # to watch the file again, until it exists.
    try:
        dpfdp[0] = os.open(isi.fs.drive_purposing.DB_FILE, os.O_RDONLY)
    except EnvironmentError as e:
        if e.errno != errno.ENOENT:
            raise

    if dpfdp[0] == -1:
        dpfdp[0] = os.open(os.path.dirname(isi.fs.drive_purposing.DB_FILE),
                os.O_RDONLY)

    return select.kevent(dpfdp[0], filter=select.KQ_FILTER_VNODE,
            fflags=KQ_VNODE_ALL)



def run_as_daemon():
    block_signals()

    isi.misc.wait_on_quorum()

    erofs = False
    while True:
        try:
            run_uncaught()
            erofs = False
        except EnvironmentError as ee:
            if ee.errno != errno.EROFS:
                raise

            if not erofs:
                syslog.syslog(LOG_ERR,
                        "Filesystem is read-only, can't procede")
                # Only log once per outage.
                erofs = True

            time.sleep(30)



def run_uncaught():
    if isi.fs.procoptions.getprocenc() != isi.fs.procoptions.ENC_UTF8:
        syslog.syslog(LOG_INFO, "Setting per-process encoding to UTF-8")
        isi.fs.procoptions.setprocoptions(enc=isi.fs.procoptions.ENC_UTF8)

    create_state_directory()

    kq = select.kqueue()
    dpfd = -1

    kq.control([make_dp_kev([dpfd])], 0, 0)
    kq.control([select.kevent(SIGHUP, filter=select.KQ_FILTER_SIGNAL)], 0, 0)

    makeuptime = None

    need_scan = True
    while True:
        made_progress = False
        if i_am_leader():
            # If I am still leader; or recently self-elected, determine what
            # needs doing and take a stab at it.
            if Debug:
                syslog.syslog(LOG_CRIT, "XXX Scanning for work")

            if need_scan:
                scan_for_work()
                need_scan = False

            if Debug:
                syslog.syslog(LOG_CRIT, "XXX Done scanning")

            prog_start_time = time.time()

            try_dequeue_work()
            made_progress = try_resume()

            prog_elapsed = min(0, time.time() - prog_start_time)

            if made_progress:
                # Only re-scan if conversion took a long time
                if prog_elapsed > 15:
                    need_scan = True
                time.sleep(15)
                continue
            else:
                need_scan = True

        # Rate-limit so we don't spin stupidly
        if makeuptime is not None:
            time.sleep(makeuptime)
            makeuptime = None

        # Nothing to do? Hang out for a while unless something interesting
        # changes.
        # - Gremlins attempt a lock coup with some frequency (to catch leader
        #   death).
        to = 15*random.random() + 5  # [5, 20) seconds
        # - Leader hangs out longer because a resume is costlier (scans for
        #   work)
        if IAmLeader:
            to = LeaderScanWait

        starttime = time.time()
        evs = kq.control(None, 5, to)
        elapsed = min(0, time.time() - starttime)

        # Allow spurious wakeups, but make up for it later...
        if IAmLeader and elapsed < LeaderScanWait:
            makeuptime = LeaderScanWait - elapsed

        # We do not care what the event(s) were, but we need to reseat the
        # vnode event. Re-scan (leader) or attempt coup (gremlins).
        for ev in evs:
            if Debug:
                syslog.syslog(LOG_CRIT, "XXX kevent(): Got event %r" % ev)

            if ev.filter == select.KQ_FILTER_VNODE:
                # Delete the event and try to re-seat it on the more specific
                # file fd. (Also, the database may have deleted the old vnode,
                # so we need to find the new vnode to watch.)
                ev.fflags = KQ_VNODE_ALL
                ev.flags = select.KQ_EV_DELETE
                kq.control([ev], 0, 0)
                os.close(ev.ident)
                kq.control([make_dp_kev([dpfd])], 0, 0)

    sys.exit(0)



def job_cancel(jid):
    try:
        isi.fs.job.JobCmd.cancel(jid, True)
    except isi.error.IsiSystemError as ie:
        # If the job already succeeded, we see EINVAL canceling it.
        if ie.error != errno.EINVAL:
            raise



def is_migration_valid(name, dpid):
    if name == "FlexProtect":
        return True

    assert name == "SP"
    assert dpid is not None

    with diskpool(dpid) as dp:
        np = dp.owner()
        if np is None:
            return False

        pdb = isi.fs.drive_purposing.DrivePurposeDB()
        return pdb.eq_entries()[np.equivalence_id]['configuration']['l3']



def je_wait(name, jid, dpid=None):  # -> Job completed (bool)
    # Wait for job to finish...
    while True:
        still_valid = is_migration_valid(name, dpid)
        if not still_valid:
            job_cancel(jid)
            raise MigrationAborted()

        JobState = isi.fs.job.JobState
        querystates = [JobState.STATE_FINISHED,
                JobState.STATE_ABORTED,
                JobState.STATE_CANCELLED_USER,
                JobState.STATE_CANCELLED_SYSTEM]

        je_events = isi.fs.job.JobHistory.query(jid, querystates, None)

        if len(je_events) > 0:
            if je_events[0].state == JobState.STATE_FINISHED:
                break

            syslog.syslog(LOG_ERR, "%s job failed to finish (cancelled or aborted)" % name)
            return False

        time.sleep(5)

    syslog.syslog(LOG_INFO, "%s job completed" % name)
    return True



def resume_job_wait(name, jidkey, jobstart, dpid=None):
    """
    Given a job name (e.g., SP, FlexProtect), a unique key name to store the
    job id in the database, and a function to start the job and return the new
    job's id if one has not already been started, this function starts a new
    job iff necessary, and waits for the job to complete.
    """

    still_valid = is_migration_valid(name, dpid)

    job_cancelled = True
    while job_cancelled:
        jid = None
        with state_locked_rw() as (slock, state):
            if jidkey in state['active']:
                jid = state['active'][jidkey]
                if still_valid:
                    syslog.syslog(LOG_INFO, "Resuming monitoring queued %s job: %d" %
                            (name, jid))
            elif still_valid:
                syslog.syslog(LOG_INFO, "Starting %s job" % name)
                jid = jobstart()
                assert jid is not None
                state['active'][jidkey] = jid
                state_write(slock, state)

        if not still_valid:
            if jid is not None:
                job_cancel(jid)
            raise MigrationAborted()

        job_cancelled = not je_wait(name, jid, dpid=dpid)
        if job_cancelled:
            # Aborted or cancelled? Need to re-enqueue a job.
            with state_locked_rw() as (slock, state):
                del state['active'][jidkey]
                state_write(slock, state)



def je_run_sp(dpid):
    resume_job_wait("SP", "resume_sp_jid",
            lambda: isi.smartpools.start_sp_job(), dpid=dpid)



def je_run_fp():
    resume_job_wait("FlexProtect", "resume_fp_jid",
            lambda: isi.fs.job.JobCmd.start(isi.fs.job.JobType.REPAIR_JOB,
                isi.fs.job.NULL_JOB_PRIORITY, None))



def work_persist_bays(dpid):
    with state_locked_rw() as (slock, state):
        with diskpool(dpid) as dp:
            state['active']['bays'] = []
            for bay in dp.bays:
                lnn, bayno = bay
                state['active']['bays'].append({'lnn': lnn, 'bay': bayno})

        state_write(slock, state)
        return state['active']['bays']



def work_cancel():
    with state_locked_rw() as (slock, state):
        del state['active']
        state_write(slock, state)



# Upgrades a given diskpool id
def resume_l3(wk):
    STAGE_SP_FINISHED = 1
    STAGE_SF_FINISHED = 2
    STAGE_FLEX_FINISHED = 3
    STAGE_FORMAT_FINISHED = 4

    dpid = wk['dpid']
    ssds = None


    if wk['stage'] == STAGE_BEGIN:
        syslog.syslog(LOG_INFO, "Upgrading pool with id %d to L3" % dpid)
    else:
        syslog.syslog(LOG_INFO, "Resuming upgrade of pool with id %d to L3" % dpid)


    with diskpool(dpid, missingok=True) as dp:
        if dp is None:
            syslog.syslog(LOG_WARNING, "Aborting upgrade of missing pool %d" % dpid)
            work_cancel()
            return


    if wk['stage'] < STAGE_SP_FINISHED:
        first = True
        while True:
            with diskpool(dpid) as dp:
                ssds = dp.ssds

                if first:
                    for ssd in ssds:
                        try:
                            _, ds = diskinfo_fmt_from_lnn_ldnum(ssd)
                        except isi.hw.drived.Error as e:
                            log_drived_exc(e)
                            ssds.remove(ssd)
                            syslog.syslog(LOG_NOTICE, ("Skipping ssd %s " +
                                "because we could not contact isi_drive_d") %\
                                repr(ssd))
                            continue

                        syslog.syslog(LOG_DEBUG, "Migrating pool drive " + ds)
                    first = False

                # (1) Mark pool layout_evacuate to force BAM to ignore it
                dp.layout_evacuate = True

                # Confirm at least one pool is writable...
                db_insanity_layout_evacuate(dp.dpdb)

                if try_write_db(dp.dpdb):
                    break

        # (2) Run SP or SPP rebalance job
        if len(ssds) > 0:
            try:
                je_run_sp(dpid)
            except isi.smartpools.ConfigError as ce:
                syslog.syslog(LOG_ERR, "Smartpools: " + str(ce))
                sys.exit(1)
            except MigrationAborted as e:
                while True:
                    with diskpool(dpid) as dp:
                        dp.layout_evacuate = False
                        if try_write_db(dp.dpdb):
                            break

                work_persist_finished(wk, aborted=True)
                syslog.syslog(LOG_ERR, "Aborting migration of diskpool: %d" %
                        dpid)
                return

        work_update_resume(wk, STAGE_SP_FINISHED)

    if wk['stage'] < STAGE_SF_FINISHED:
        with diskpool(dpid) as dp:
            ssds = dp.ssds

        # (3) Smartfail the pool!!! Wooo
        # Determine which drives should be smartfailed
        sf_any = False
        for ssd in ssds:
            lnn, ldnum = ssd
            try:
                di, ds = diskinfo_fmt_from_lnn_ldnum(ssd)
            except isi.hw.drived.Error as e:
                log_drived_exc(e)
                syslog.syslog(LOG_NOTICE, "Skipping ssd because " +
                    "isi_drive_d could not be contacted: %s" % repr(ssd))
                ssds.remove(ssd)
                continue

            if di.usr_purpose == isi.hw.drived.USR_L3_CACHE:
                syslog.syslog(LOG_DEBUG, "Skipping smartfail on L3_CACHE drive: " + ds)
                continue
            if di.usr_state != isi.hw.drived.USR_ACTIVE:
                syslog.syslog(LOG_DEBUG, "Skipping smartfail on offline drive: " + ds)
                continue

            syslog.syslog(LOG_INFO, "Smartfailing " + ds)
            try:
                isi.hw.drived.driveSoftfailMsg(di.handle,
                        "isi_drive_repurpose_d stage 3", lnn,
                        isi.hw.drived.FLAG_QUIET_FAILED)
            except isi.hw.drived.Error as e:
                log_drived_exc(e)
                syslog.syslog(LOG_NOTICE, ("Softfailing lnn%s:lnum%s failed " +
                    "because isi_drive_d was unreachable") % (lnn, ldnum))
                ssds.remove(ssd)
                continue

            sf_any = True

        wk['bays'] = work_persist_bays(dpid)

        if sf_any:
            work_update_resume(wk, STAGE_SF_FINISHED)
        else:
            work_update_resume(wk, STAGE_FLEX_FINISHED)

    # (5)   * Run flexprotect to GONE the SF drives
    if wk['stage'] < STAGE_FLEX_FINISHED:
        je_run_fp()
        work_update_resume(wk, STAGE_FLEX_FINISHED)

    # (6)   * Format the drives as L3 cache
    if wk['stage'] < STAGE_FORMAT_FINISHED:
        nretry = nformatted = 0
        for ssd in wk['bays']:
            lnn, bay = ssd['lnn'], ssd['bay']
            di, ok = bay_query(bay, lnn)
            if not ok:
                syslog.syslog(LOG_NOTICE, ("Can't format lnn%s:bay%s as L3 " +
                    "because isi_drive_d is not responding") % (lnn, bay))
                continue

            ds = "lnn%d:bay%d" % (lnn, bay)

            if di.usr_purpose == isi.hw.drived.USR_L3_CACHE:
                # Don't second guess a disk that is already purposed for L3
                syslog.syslog(LOG_INFO, "Drive %s is already purposed as L3, "
                    "skipped." % ds)
                nformatted += 1
            elif di.usr_state == isi.hw.drived.USR_FAILED or \
                    (di.usr_state == isi.hw.drived.USR_NEW and di.present):
                # Drive is ready to be formatted
                syslog.syslog(LOG_INFO, "Formatting %s as L3" % ds)
                try:
                    isi.hw.drived.bayStartAddNone(bay, lnn)
                    nformatted += 1
                except isi.hw.drived.Error as e:
                    log_drived_exc(e)
                    syslog.syslog(LOG_NOTICE, "Format %s as L3 failed" % ds)
                    continue
            elif di.usr_state in (isi.hw.drived.USR_ACTIVE,
                    isi.hw.drived.USR_PREPARING) and di.present:
                # The drive is already formatted
                syslog.syslog(LOG_INFO, "Drive %s has already been added" % ds)
                nformatted += 1
            elif di.usr_state in (isi.hw.drived.USR_DEPRECATED,
                    isi.hw.drived.USR_FAILED_WOR) and di.present:
                # The drive has not transitioned out of GONE yet
                syslog.syslog(LOG_INFO, "Drive %s is not ready to be added" %
                    ds)
                nretry += 1
            else:
                # The drive is in an unknown or unrecoverable state
                syslog.syslog(LOG_NOTICE,
                    "Drive %s is in state %s (%d,%d,%d), skipped" % (ds,
                    isi.hw.drived.getStateName(di.usr_state, di.present,
                    di.usr_purpose), di.usr_state, di.present, di.usr_purpose))

        # Don't make progress on the job until all bays are done formatting
        if nretry:
            syslog.syslog(LOG_INFO, "Some drives were not ready, retrying.")
            return

        # iff pool is empty, delete it.
        if nformatted == len(wk['bays']):
            while True:
                with diskpool(dpid, missingok=True) as dp:
                    if dp is None:
                        break
                    else:
                        dp.dpdb.rem_entry(dp)
                        if try_write_db(dp.dpdb):
                            break

        # Just for development painlessness, un-set layout_evacuate on nodes
        # with broken CTO receipts.
        while True:
            with diskpool(dpid, missingok=True) as dp:
                if dp is None:
                    break
                dp.layout_evacuate = False
                if try_write_db(dp.dpdb):
                    break


        work_update_resume(wk, STAGE_FORMAT_FINISHED)

    assert wk['stage'] == STAGE_FORMAT_FINISHED
    work_persist_finished(wk)
    syslog.syslog(LOG_INFO, "Completed migrating pool %d to L3" % dpid)



# Downgrade a set of (devid, ldnum) pairs to storage from L3
def resume_revert_l3(wk):
    STAGE_BAYS_PERSISTED = 1
    STAGE_STOP_L3 = 2
    STAGE_REVERTL3_FINISHED = 3

    dds = []
    for node in wk['deviddrives']:
        for drv in node['drives']:
            dds.append((node['devid'], drv))

    if wk['stage'] == STAGE_BEGIN:
        syslog.syslog(LOG_INFO, "Downgrading %d SSDs to storage" % len(dds))
    else:
        syslog.syslog(LOG_INFO, "Resuming L3 downgrade")

    arrcfg = isi.config.arrcfg.config_load()
    devid_to_lnn = {}
    for d in arrcfg.devices:
        if d.lnn:
            devid_to_lnn[d.devid] = d.lnn

    lnns = set([ devid_to_lnn[dd[0]] for dd in dds ])

    if wk['stage'] < STAGE_BAYS_PERSISTED:
        bays = []

        for dd in dds:
            devid, ldnum = dd
            lnn = devid_to_lnn[devid]
            di, ok = ldnum_query(ldnum, lnn)
            if not ok:
                syslog.syslog(LOG_NOTICE, "Skipping lnn%s:ld%s" % (lnn, ldnum))
                continue

            bay = di.bay

            bays.append((lnn, bay))

        wk['bays'] = bays
        with state_locked_rw() as (slock, state):
            state['active']['bays'] = bays
            state_write(slock, state)
        work_update_resume(wk, STAGE_BAYS_PERSISTED)

    if 'bays' not in wk:
        syslog.syslog(LOG_ERR, "Skipping broken wk (missing 'bays'): %r" % wk)
        work_persist_finished(wk, aborted=True)
        return
    bays = wk['bays']

    if wk['stage'] < STAGE_STOP_L3:
        diskinfos = {}
        skipped_lnns = set()

        for lnn in lnns:
            # Don't fail drives on offline nodes
            if lnn not in online_lnns():
                syslog.syslog(LOG_NOTICE, ("Skipping lnn%s as it appears " +
                    "offline") % lnn)
                skipped_lnns.add(lnn)
                continue

            try:
                diskinfos[lnn] = isi.hw.drived.getDrives(lnn)
            except EnvironmentError as e:
                log_drived_exc(e)
                skipped_lnns.add(lnn)
                syslog.syslog(LOG_NOTICE, "Skipping lnn%s" % lnn)

        for (i, (lnn, bay)) in enumerate(bays):
            if lnn in skipped_lnns:
                syslog.syslog(LOG_NOTICE, ("Skipping lnn%s:bay%s because " +
                    "node is offline") % (lnn, bay))
                with state_locked_rw() as (slock, state):
                    state['active']['bays'].remove((lnn, bay))
                    state_write(slock, state)
                wk['bays'].remove((lnn, bay))
                continue

            dilist = filter(lambda x: x.bay == bay and x.present, diskinfos[lnn])
            assert len(dilist) == 1

            di = dilist[0]
            if (di.usr_purpose == isi.hw.drived.USR_L3_CACHE and di.present and
                    di.usr_state == isi.hw.drived.USR_ACTIVE):
                syslog.syslog(LOG_INFO, "Smartfailing L3 drive lnn%s:bay%s" % (lnn, bay))
                try:
                    isi.hw.drived.driveSoftfailMsg(di.handle,
                            "isi_drive_repurpose_d stop_l3", lnn,
                            isi.hw.drived.FLAG_QUIET_FAILED)
                except EnvironmentError as e:
                    log_drived_exc(e)
                    syslog.syslog(LOG_NOTICE, "Skipping drive we couldn't " +
                            "smartfail")
                    with state_locked_rw() as (slock, state):
                        state['active']['bays'].remove((lnn, bay))
                        state_write(slock, state)
                    wk['bays'].remove((lnn, bay))

        work_update_resume(wk, STAGE_STOP_L3)
        time.sleep(2)

    if wk['stage'] < STAGE_REVERTL3_FINISHED:
        nretry = 0
        for (lnn, bay) in bays:
            di, ok = bay_query(bay, lnn)
            if not ok:
                syslog.syslog(LOG_WARNING, ("Leaving lnn%s:bay%s in REPLACE " +
                    "because isi_drive_d is not responding") % (lnn, bay))
                continue

            ds = "lnn%d:bay%d" % (lnn, bay)

            if di.present and di.usr_state == isi.hw.drived.USR_FAILED:
                # Drive is ready to be formatted
                syslog.syslog(LOG_INFO, "Adding %s as storage" % ds)
                try:
                    isi.hw.drived.bayStartAddNone(bay, lnn)
                except EnvironmentError as e:
                    log_drived_exc(e)
                    syslog.syslog(LOG_WARNING, ("Leaving %s in " +
                        "REPLACE because isi_drive_d is not responding") % ds)
            elif di.usr_state in (isi.hw.drived.USR_ACTIVE,
                    isi.hw.drived.USR_PREPARING) and di.present:
                # The drive is already formatted
                syslog.syslog(LOG_INFO, "Drive %s has already been added" % ds)
            elif di.present and di.usr_state in (isi.hw.drived.USR_DEPRECATED,
                    isi.hw.drived.USR_FAILED_WOR):
                # The drive has not transitioned out of GONE yet
                syslog.syslog(LOG_INFO, "Drive %s is not ready yet" % ds)
                nretry += 1
            else:
                # The drive is in an unknown or unrecoverable state
                syslog.syslog(LOG_NOTICE,
                    "Drive %s is in state %s (%d,%d,%d), skipped" % (ds,
                    isi.hw.drived.getStateName(di.usr_state, di.present,
                    di.usr_purpose), di.usr_state, di.present, di.usr_purpose))

        # Don't make progress on the job until all bays are done formatting
        if nretry:
            syslog.syslog(LOG_INFO, "Some drives were not ready, retrying.")
            return

        work_update_resume(wk, STAGE_REVERTL3_FINISHED)

    assert wk['stage'] == STAGE_REVERTL3_FINISHED
    work_persist_finished(wk)
    syslog.syslog(LOG_INFO, "Completed reverting some L3 drives to storage")


if __name__ == "__main__":
    try:
        main()
    except Exception as e:
        log_exc(e)
        raise
