engine/orchestration/src/main/java/com/cloud/vm/VmWorkJobWakeupDispatcher.java

// Licensed to the Apache Software Foundation (ASF) under one
// or more contributor license agreements.  See the NOTICE file
// distributed with this work for additional information
// regarding copyright ownership.  The ASF licenses this file
// to you under the Apache License, Version 2.0 (the
// "License"); you may not use this file except in compliance
// with the License.  You may obtain a copy of the License at
//
//   http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing,
// software distributed under the License is distributed on an
// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
// KIND, either express or implied.  See the License for the
// specific language governing permissions and limitations
// under the License.

package com.cloud.vm;

import java.lang.reflect.Method;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import javax.inject.Inject;

import org.apache.log4j.Logger;

import org.apache.cloudstack.context.CallContext;
import org.apache.cloudstack.framework.jobs.AsyncJob;
import org.apache.cloudstack.framework.jobs.AsyncJobDispatcher;
import org.apache.cloudstack.framework.jobs.AsyncJobManager;
import org.apache.cloudstack.framework.jobs.dao.AsyncJobJoinMapDao;
import org.apache.cloudstack.framework.jobs.dao.VmWorkJobDao;
import org.apache.cloudstack.framework.jobs.impl.AsyncJobJoinMapVO;
import org.apache.cloudstack.framework.jobs.impl.VmWorkJobVO;

import com.cloud.user.AccountVO;
import com.cloud.user.dao.AccountDao;
import com.cloud.utils.component.AdapterBase;
import com.cloud.vm.dao.VMInstanceDao;

/**
 * Please note: VmWorkJobWakeupDispatcher is not currently in use. It is designed for event-driven based
 * job processing model.
 *
 * Current code base uses blocking calls to wait for job completion
 */
public class VmWorkJobWakeupDispatcher extends AdapterBase implements AsyncJobDispatcher {
    private static final Logger s_logger = Logger.getLogger(VmWorkJobWakeupDispatcher.class);

    @Inject
    private VmWorkJobDao _workjobDao;
    @Inject
    private AsyncJobJoinMapDao _joinMapDao;
    @Inject
    private AccountDao _accountDao;
    @Inject
    private VMInstanceDao _instanceDao;
    @Inject
    private VirtualMachineManager _vmMgr;
    @Inject
    private AsyncJobManager _asyncJobMgr;

    private final Map<String, Method> _handlerMap = new HashMap<String, Method>();

    @Override
    public void runJob(AsyncJob job) {
        try {
            List<AsyncJobJoinMapVO> joinRecords = _joinMapDao.listJoinRecords(job.getId());
            if (joinRecords.size() != 1) {
                s_logger.warn("AsyncJob-" + job.getId()
                        + " received wakeup call with un-supported joining job number: " + joinRecords.size());

                // if we fail wakeup-execution for any reason, avoid release sync-source if there is any
                job.setSyncSource(null);
                return;
            }

            AsyncJobJoinMapVO joinRecord = joinRecords.get(0);
            VmWorkJobVO joinedJob = _workjobDao.findById(joinRecord.getJoinJobId());

            Class<?> workClz = null;
            try {
                workClz = Class.forName(job.getCmd());
            } catch (ClassNotFoundException e) {
                s_logger.error("VM work class " + job.getCmd() + " is not found", e);
                return;
            }

            // get original work context information from joined job
            VmWork work = VmWorkSerializer.deserialize(workClz, joinedJob.getCmdInfo());
            assert (work != null);

            AccountVO account = _accountDao.findById(work.getAccountId());
            assert (account != null);

            VMInstanceVO vm = _instanceDao.findById(work.getVmId());
            assert (vm != null);

            CallContext.register(work.getUserId(), work.getAccountId(), job.getRelated());
            try {
                Method handler = getHandler(joinRecord.getWakeupHandler());
                if (handler != null) {
                    handler.invoke(_vmMgr);
                } else {
                    assert (false);
                    s_logger.error("Unable to find wakeup handler " + joinRecord.getWakeupHandler() +
                            " when waking up job-" + job.getId());
                }
            } finally {
                CallContext.unregister();
            }
        } catch (Throwable e) {
            s_logger.warn("Unexpected exception in waking up job-" + job.getId());

            // if we fail wakeup-execution for any reason, avoid release sync-source if there is any
            job.setSyncSource(null);
        }
    }

    private Method getHandler(String wakeupHandler) {

        synchronized (_handlerMap) {
            Class<?> clz = _vmMgr.getClass();
            Method method = _handlerMap.get(wakeupHandler);
            if (method != null)
                return method;

            try {
                method = clz.getMethod(wakeupHandler);
                method.setAccessible(true);
            } catch (SecurityException e) {
                assert (false);
                s_logger.error("Unexpected exception", e);
                return null;
            } catch (NoSuchMethodException e) {
                assert (false);
                s_logger.error("Unexpected exception", e);
                return null;
            }

            _handlerMap.put(wakeupHandler, method);
            return method;
        }
    }
}