文前说明

作为码农中的一员，需要不断的学习，我工作之余将一些分析总结和学习笔记写成博客与大家一起交流，也希望采用这种方式记录自己的学习之旅。

本文仅供学习交流使用，侵权必删。
不用于商业目的，转载请注明出处。

分析整理的版本为 Ovirt 4.2.3 版本。

1. checkDBConnectivity

• 检测数据库连接
  • 配置文件中读取数据库连接超时时间 ENGINE_DB_CONNECTION_TIMEOUT。未超时且未执行过数据库连接检测，则执行 CheckDBConnection 存储过程（查询一次数据库，保持数据库连接）。

private void loadDbFacadeConfig() {
        final EngineLocalConfig config = EngineLocalConfig.getInstance();
        try {
            connectionTimeout = config.getInteger("ENGINE_DB_CONNECTION_TIMEOUT");
            checkInterval = config.getInteger("ENGINE_DB_CHECK_INTERVAL");
        }
        catch (Exception exception) {
            log.warn("Can't load connection checking parameters of DB facade, "
                            + "will continue using the default values. Error: {}",
                exception.getMessage());
            log.debug("Exception", exception);
        }
    }

while (!dbUp && System.currentTimeMillis() < expectedTimeout) {
            try {
                dbUp = dbConnectionUtil.checkDBConnection();

CREATE
        OR replace FUNCTION CheckDBConnection ()
RETURNS SETOF INT IMMUTABLE AS $PROCEDURE$

BEGIN
        RETURN QUERY

        SELECT 1;
END;$PROCEDURE$

2. initialize

• 进行 engine 的初始化。

2.1 注入 engine 线程池对象

• 通过 init 方法创建线程池 EngineThreadPool 对象。
  • 通过配置文件设置了池的核心线程数，最大线程数和队列长度。
    • ENGINE_THREAD_POOL_MIN_SIZE 核心线程数。
    • ENGINE_THREAD_POOL_MAX_SIZE 最大线程数。
    • ENGINE_THREAD_POOL_QUEUE_SIZE 队列大小。

serviceLoader.load(EngineThreadPools.class);

ThreadPoolUtil.setExecutorService(
                new InternalThreadExecutor(
                        "EngineThreadPool",
                        threadFactory,
                        EngineLocalConfig.getInstance().getInteger("ENGINE_THREAD_POOL_MIN_SIZE"),
                        EngineLocalConfig.getInstance().getInteger("ENGINE_THREAD_POOL_MAX_SIZE"),
                        EngineLocalConfig.getInstance().getInteger("ENGINE_THREAD_POOL_QUEUE_SIZE")));

• 通过不同的 @ThreadPools 注解类型的不同，生产不同的线程池对象。
  • 各线程池对象通过在 ovirt-engine.xml 文件中配置 JSR-236 EE 并发应用程序进行生产（参数配置等）。

关键字	说明
ThreadPoolType.CoCo	协调 Command 执行的线程池
HostUpdatesChecker	节点升级检查的线程池
EngineScheduledThreadPool	engine 调度的线程池
EngineThreadMonitoringThreadPool	对 engine 线程进行监控的线程池

<managed-scheduled-executor-services>
          <managed-scheduled-executor-service
              name="default"
              jndi-name="java:jboss/ee/concurrency/scheduler/default"
              context-service="default"
              thread-factory="default"
              hung-task-threshold="60000"
              core-threads="5"
              keepalive-time="3000"/>
          <managed-scheduled-executor-service
              name="engineScheduledThreadPool"
              long-running-tasks="true"
              jndi-name="java:jboss/ee/concurrency/scheduler/engineScheduledThreadPool"
              context-service="default"
              thread-factory="engineScheduled"
              core-threads="{{ config.getinteger('ENGINE_SCHEDULED_THREAD_POOL_SIZE') }}"
              keepalive-time="5000"
              reject-policy="RETRY_ABORT" />
          <managed-scheduled-executor-service
              name="engineThreadMonitoringThreadPool"
              long-running-tasks="true"
              jndi-name="java:jboss/ee/concurrency/scheduler/engineThreadMonitoringThreadPool"
              context-service="default"
              thread-factory="engineThreadMonitoring"
              core-threads="1"
              keepalive-time="5000"
              reject-policy="RETRY_ABORT" />
        </managed-scheduled-executor-services>

@Qualifier
@Retention(RetentionPolicy.RUNTIME)
@Target({ElementType.FIELD, ElementType.TYPE, ElementType.METHOD })
public @interface ThreadPools {

    enum ThreadPoolType {
        CoCo,
        HostUpdatesChecker,
        EngineScheduledThreadPool,
        EngineThreadMonitoringThreadPool}

    ThreadPoolType value();
}

2.2 注入对 engine 的线程进行监控的线程池对象

@Inject
@ThreadPools(ThreadPools.ThreadPoolType.EngineThreadMonitoringThreadPool)
private ManagedScheduledExecutorService executor;

• 该对象是一个调度执行对象
  • 通过配置文件 THREAD_POOL_MONITORING_INTERVAL_IN_SECONDS 设置调度频率。
    • 当上述值小于等于 0 时，停止线程监控。
  • 在调度中循环所有的线程 ID，获取线程相关信息，并输出各线程执行状态，线程池状态，队列状态等信息到日志中。

private void monitorEngineThreadPools() {
        try {
            threadPoolInfoMap.clear();
            Arrays.asList(threadMXBean.getAllThreadIds())
                    .stream()
                    .forEach(threadId -> processThread(threadMXBean.getThreadInfo(threadId)));
            threadPoolInfoMap.entrySet()
                    .stream()
                    .forEach(entry -> log.info(entry.getValue().toString()));
        } catch (Exception ex) {
            log.info("Error fetching thread pools data: {}", ex.getMessage());
            log.debug("Exception", ex);
        }
}

状态码	状态	说明
0	NEW	新建状态
1	RUNNABLE	正在运行状态
2	BLOCKED	阻塞状态
3	WAITING	等待状态
4	TIMED_WAITING	超时等待状态
5	TERMINATED	终止状态

2.3 engine 启动前保存 HostEngine 所运行的主机信息

serviceLoader.load(PreviousHostedEngineHost.class);

List<VM> vms = vmDao.getVmsByOrigins(Arrays.asList(OriginType.HOSTED_ENGINE, OriginType.MANAGED_HOSTED_ENGINE));
if (vms != null && !vms.isEmpty()) {
     previousHostId = vms.iterator().next().getRunOnVds();
}
log.debug("Hosted engine VM was running prior to restart on host '{}'", previousHostId);

2.4 开始执行任务调度

for (SchedulerUtil taskScheduler : taskSchedulers) {
    log.info("Started task scheduler {}", taskScheduler);
}

• 调度工厂初始化。

public void setup() {
        try {
            SchedulerFactory sf = new StdSchedulerFactory();
            sched = sf.getScheduler();
            sched.start();
            sched.getListenerManager()
                    .addJobListener(new FixedDelayJobListener(this), jobGroupEquals(Scheduler.DEFAULT_GROUP));
        } catch (SchedulerException se) {
            log.error("there is a problem with the underlying Scheduler: {}", se.getMessage());
            log.debug("Exception", se);
        }
}

2.5 注入 engine 的缓存管理

serviceLoader.load(CacheManager.class);

• 这里使用了 infinispan。

public static final String TIMEOUT_BASE = "timeout-base";

@Resource(lookup = "java:jboss/infinispan/ovirt-engine")
private CacheContainer cacheContainer;
private static Cache<String, String> cache;

@PostConstruct
private void init() {
    cache = cacheContainer.getCache(TIMEOUT_BASE);
}

public static Cache<String, String> getTimeoutBaseCache() {
    return cache;
}

2.6 加载 ConfigValue 系统操作项

Config.setConfigUtils(new DBConfigUtils());

• 从数据库中 vdc_option 表中读取系统操作项，放入缓存。

public void refresh() {
        _vdcOptionCache.clear();
        List<VdcOption> list = moveDependentToEnd(getVdcOptionDao().getAll());
        for (VdcOption option : list) {
            try {
                if (!_vdcOptionCache.containsKey(option.getOptionName()) ||
                        !_vdcOptionCache.get(option.getOptionName()).containsKey(option.getVersion()) ||
                        isReloadable(option.getOptionName())) {
                    updateOption(option);
                }
            } catch (NoSuchFieldException e) {
                log.error("Not refreshing field '{}': does not exist in class {}.", option.getOptionName(),
                        ConfigValues.class.getSimpleName());
            }
        }
}

2.7 初始化操作系统库

• 重置 dwh_osinfo 表中数据。

private void initOsRepository() {
        OsInfoPreferencesLoader.INSTANCE.init(FileSystems.getDefault().getPath(EngineLocalConfig.getInstance().getEtcDir().getAbsolutePath(), Config.<String>getValue(ConfigValues.OsRepositoryConfDir)));
        OsRepositoryImpl.INSTANCE.init(OsInfoPreferencesLoader.INSTANCE.getPreferences());
        OsRepository osRepository = OsRepositoryImpl.INSTANCE;
        SimpleDependencyInjector.getInstance().bind(OsRepository.class, osRepository);
        osInfoDao.populateDwhOsInfo(osRepository.getOsNames());
}

2.8 注入并初始化 Mac 池

@Inject
private MacPoolPerCluster macPoolPerCluster;
......
macPoolPerCluster.logFreeMacs();

• 输出剩余 Mac 日志

public void logFreeMacs() {
        macPools.values()
                .stream()
                .forEach(macPool -> log.info("Mac pool {} has {} available free macs",
                        macPool.getId(),
                        macPool.getAvailableMacsCount()));
    }

2.9 执行补偿机制

@Inject
private CommandCompensator compensator;
......
compensator.compensate();

• 根据不同的补偿快照类型，执行不同的回滚操作。

public void compensate() {
        // get all command snapshot entries
        List<KeyValue> commandSnapshots = businessEntitySnapshotDao.getAllCommands();
        for (KeyValue commandSnapshot : commandSnapshots) {
            // create an instance of the related command by its class name and command id
            try {
                compensate((Guid) commandSnapshot.getKey(), (String) commandSnapshot.getValue(), null);
            } catch (RuntimeException e) {
                log.error(
                        "Failed to run compensation on startup for Command '{}', Command Id '{}': {}",
                        commandSnapshot.getValue(),
                        commandSnapshot.getKey(),
                        e.getMessage());
                log.error("Exception", e);
            }
            log.info("Running compensation on startup for Command '{}', Command Id '{}'",
                    commandSnapshot.getValue(),
                    commandSnapshot.getKey());
        }
}

public enum SnapshotType {
        DELETED_OR_UPDATED_ENTITY,
        NEW_ENTITY_ID,
        CHANGED_STATUS_ONLY,
        UPDATED_ONLY_ENTITY,
        TRANSIENT_ENTITY
}

2.10 注入并初始化 CPU 架构相关信息

serviceLoader.load(CpuFlagsManagerHandler.class);

public void initDictionaries() {
    log.info("Start initializing dictionaries");
    managersDictionary.clear();
    for (Version ver : Config.<HashSet<Version>> getValue(ConfigValues.SupportedClusterLevels)) {
         managersDictionary.put(ver, new CpuFlagsManager(ver));
    }
    log.info("Finished initializing dictionaries");
}

2.11 注入并初始化清除日志管理调度

serviceLoader.load(AuditLogCleanupManager.class);

• 注入 engine 调度池对象

@Inject
@ThreadPools(ThreadPools.ThreadPoolType.EngineScheduledThreadPool)
private ManagedScheduledExecutorService executor;

• 初始化清除日志管理
  • AuditLogCleanupTime 系统操作参数配置日志清除的时间（crontab 格式）。
  • AuditLogAgingThreshold 审计日志阈值，定义删除多少天之前的日志。

private void init() {
    log.info("Start initializing {}", getClass().getSimpleName());
    Calendar calendar = new GregorianCalendar();
    Date auditLogCleanupTime = Config.<DateTime> getValue(ConfigValues.AuditLogCleanupTime);
    calendar.setTimeInMillis(auditLogCleanupTime.getTime());
    String cronExpression = String.format("%d %d %d * * ?", calendar.get(Calendar.SECOND), calendar.get(Calendar.MINUTE), calendar.get(Calendar.HOUR_OF_DAY));
    log.info("Setting audit cleanup manager to run at '{}'", cronExpression);
    executor.schedule(this::cleanup, new EngineCronTrigger(cronExpression));
    log.info("Finished initializing {}", getClass().getSimpleName());
}

DateTime latestTimeToKeep = DateTime.getNow().addDays(Config.<Integer>getValue(ConfigValues.AuditLogAgingThreshold) * -1);
auditLogDao.removeAllBeforeDate(latestTimeToKeep);

2.12 注入并初始化标签管理

serviceLoader.load(TagsDirector.class);

protected void init() {
    log.info("Start initializing {}", getClass().getSimpleName());
    tagsMapByID.clear();
    tagsMapByName.clear();
    Tags root = new Tags("root", null, true, ROOT_TAG_ID, "root");
    addTagToHash(root);
    addChildren(root);
    log.info("Finished initializing {}", getClass().getSimpleName());
}

2.13 注入并初始化 ISO 域同步器

serviceLoader.load(IsoDomainListSynchronizer.class);

2.14 初始化操作系统库搜索依赖关系

private void initSearchDependencies() {
   SimpleDependencyInjector.getInstance().bind(new OsValueAutoCompleter(SimpleDependencyInjector.getInstance().get(OsRepository.class).getUniqueOsNames()));
}

2.15 初始化一些处理程序

• 标签处理程序 tagsHandler。
• 虚拟机处理程序 VmHandler。
• 主机处理程序 VdsHandler。
• 模板处理程序 VmTemplateHandler。

private void initHandlers() {
     BaseConditionFieldAutoCompleter.tagsHandler = tagsDirector;
     serviceLoader.load(VmHandler.class);
     serviceLoader.load(VdsHandler.class);
     serviceLoader.load(VmTemplateHandler.class);
     log.info("Completed initializing handlers");
}

2.16 初始化虚拟机属性工具

• 初始化一下预定义属性、用户自定义属性等。

private void initVmPropertiesUtils() {
      VmPropertiesUtils vmPropertiesUtils = VmPropertiesUtils.getInstance();
      SimpleDependencyInjector.getInstance().bind(VmPropertiesUtils.class, vmPropertiesUtils);
}

2.17 加载错误日志资源文件（国际化）

final String AppErrorsFileName = "bundles/AppErrors.properties";
final String VdsErrorsFileName = "bundles/VdsmErrors.properties";
errorsTranslator = new ErrorTranslatorImpl(AppErrorsFileName, VdsErrorsFileName);
vdsErrorsTranslator = new ErrorTranslatorImpl(VdsErrorsFileName);

2.18 注入并初始化任务库

@Inject
private JobRepository jobRepository;

• 删除之前运行失败的日志。
• 将运行状态的日志置为未知状态。

private void initJobRepository() {
        try {
            jobRepository.finalizeJobs();
        } catch (Exception e) {
            log.error("Failed to finalize running Jobs", e);
        }
}

2.19 注入并初始化任务清除调度

serviceLoader.load(JobRepositoryCleanupManager.class);

• 注入 engine 调度线程池

@Inject
@ThreadPools(ThreadPools.ThreadPoolType.EngineScheduledThreadPool)
private ManagedScheduledExecutorService executor;

• 设置成功任务的清除周期与失败任务的清除周期。
  • SucceededJobCleanupTimeInMinutes 成功任务的清除周期。
  • FailedJobCleanupTimeInMinutes 失败任务的清除周期。
  • JobCleanupRateInMinutes 调度任务的频率。

@PostConstruct
public void initialize() {
    log.info("Start initializing {}", getClass().getSimpleName());
    succeededJobTime = Config.<Integer> getValue(ConfigValues.SucceededJobCleanupTimeInMinutes);
    failedJobTime = Config.<Integer> getValue(ConfigValues.FailedJobCleanupTimeInMinutes);

    long cleanupFrequency = Config.<Long> getValue(ConfigValues.JobCleanupRateInMinutes);
    executor.scheduleWithFixedDelay(this::cleanCompletedJob,
          cleanupFrequency,
          cleanupFrequency,
          TimeUnit.MINUTES);
   log.info("Finished initializing {}", getClass().getSimpleName());
}

2.20 注入并初始化主机的自动恢复管理

serviceLoader.load(AutoRecoveryManager.class);

• 注入 engine 调度线程池

@Inject
@ThreadPools(ThreadPools.ThreadPoolType.EngineScheduledThreadPool)
private ManagedScheduledExecutorService executor;

• 初始化自动恢复主机调度
  • AutoRecoverySchedule 尝试自动恢复主机的时间（crontab 格式）。

void initialize() {
    log.info("Start initializing {}", getClass().getSimpleName());
    executor.schedule(this::recover, new EngineCronTrigger(Config.getValue(ConfigValues.AutoRecoverySchedule)));
    log.info("Finished initializing {}", getClass().getSimpleName());
}

• 过滤掉网络有问题的主机，尝试自动激活主机。
• 尝试连接存储域。

public void recoverImpl() {
        check(vdsDao,
                ActionType.ActivateVds,
                arg -> {
                    final VdsActionParameters params = new VdsActionParameters(arg.getId());
                    params.setRunSilent(true);
                    return params;
                }, list -> {
                    List<VDS> filtered = new ArrayList<>(list.size());
                    List<VdsNetworkInterface> nics;

                    for (VDS vds : list) {
                        if (vds.getNonOperationalReason() == NonOperationalReason.NETWORK_INTERFACE_IS_DOWN) {
                            backend.getResourceManager().runVdsCommand(VDSCommandType.GetStats,
                                    new VdsIdAndVdsVDSCommandParametersBase(vds));
                            nics = vds.getInterfaces();
                        } else {
                            nics = interfaceDao.getAllInterfacesForVds(vds.getId());
                        }

                        Map<String, Set<String>> problematicNics =
                                NetworkMonitoringHelper.determineProblematicNics(nics,
                                        networkDao.getAllForCluster(vds.getClusterId()));
                        if (problematicNics.isEmpty()) {
                            filtered.add(vds);
                        }
                    }
                    return filtered;
        }, "hosts");
        check(storageDomainDao,
                ActionType.ConnectDomainToStorage,
                arg -> {
                    final StorageDomainPoolParametersBase params = new StorageDomainPoolParametersBase(
                            arg.getId(), arg.getStoragePoolId());
                    params.setRunSilent(true);
                    return params;
                }, list -> list, "storage domains");
}

2.21 初始化任务日志资源文件（国际化）

private void initExecutionMessageDirector() {
     try {
        ExecutionMessageDirector.getInstance().initialize(ExecutionMessageDirector.EXECUTION_MESSAGES_FILE_PATH);
     } catch (RuntimeException e) {
        log.error("Failed to initialize ExecutionMessageDirector", e);
     }
}

public void initialize(String bundleBaseName) {
        log.info("Start initializing {}", getClass().getSimpleName());
        ResourceBundle bundle = ResourceBundle.getBundle(bundleBaseName);
        final int jobMessagePrefixLength = JOB_MESSAGE_PREFIX.length();
        final int stepMessagePrefixLength = STEP_MESSAGE_PREFIX.length();

        for (String key : bundle.keySet()) {

            if (key.startsWith(JOB_MESSAGE_PREFIX)) {
                addMessage(key, bundle.getString(key), jobMessages, ActionType.class, jobMessagePrefixLength);
            } else if (key.startsWith(STEP_MESSAGE_PREFIX)) {
                addMessage(key, bundle.getString(key), stepMessages, StepEnum.class, stepMessagePrefixLength);
            } else {
                log.error("The message key '{}' cannot be categorized since not started with '{}' nor '{}'",
                        key,
                        JOB_MESSAGE_PREFIX,
                        STEP_MESSAGE_PREFIX);
            }
        }
        log.info("Finished initializing {}", getClass().getSimpleName());
}

2.22 保存 engine 启动时间

_startedAt = DateTime.getNow();
......
@Override
public DateTime getStartedAt() {
    return _startedAt;
}

2.23 初始化虚拟机池监控（预启动）

serviceLoader.load(VmPoolMonitor.class);

• 注入 engine 调度线程池对象

@Inject
@ThreadPools(ThreadPools.ThreadPoolType.EngineScheduledThreadPool)
private ManagedScheduledExecutorService schedulerService;

• 初始化
  • VmPoolMonitorIntervalInMinutes 预启动执行周期。

private void init() {
        vmPoolMonitorIntervalInMinutes = Config.<Long>getValue(ConfigValues.VmPoolMonitorIntervalInMinutes);
        poolMonitoringJob =
                schedulerService.scheduleWithFixedDelay(
                        this::managePrestartedVmsInAllVmPools,
                        vmPoolMonitorIntervalInMinutes,
                        vmPoolMonitorIntervalInMinutes,
                        TimeUnit.MINUTES);
 }

• 虚拟机池中虚拟机进行预启动
  • VmPoolMonitorBatchSize 虚拟机池中虚拟机总数量。
  • VmPoolMonitorMaxAttempts 预启动失败总次数。

private void managePrestartedVmsInAllVmPools() {
        lock.lock();
        try {
            vmPoolDao.getAll()
                    .stream()
                    .filter(pool -> pool.getPrestartedVms() > 0)
                    .forEach(this::managePrestartedVmsInPool);
        } catch (Throwable t) {
            log.error("Exception managing prestarted VMs in all VM pools: {}", ExceptionUtils.getRootCauseMessage(t));
            log.debug("Exception", t);
        } finally {
            lock.unlock();
        }
}

2.24 初始化高可用自动启动虚拟机

serviceLoader.load(HaAutoStartVmsRunner.class);

• 注入 engine 调度线程池

@Inject
@ThreadPools(ThreadPools.ThreadPoolType.EngineScheduledThreadPool)
private ManagedScheduledExecutorService executor;

• 初始化
  • AutoStartVmsRunnerIntervalInSeconds 自动启动虚拟机的周期。
  • RetryToRunAutoStartVmIntervalInSeconds 尝试启动虚拟机次数。
  • DelayToRunAutoStartVmIntervalInSeconds 尝试启动各虚拟机之间的间隔时间。

@PostConstruct
    private void init() {
        autoStartVmsToRestart = new CopyOnWriteArraySet<>(getInitialVmsToStart());

        long autoStartVmsRunnerIntervalInSeconds =
                Config.<Long>getValue(ConfigValues.AutoStartVmsRunnerIntervalInSeconds);
        executor.scheduleWithFixedDelay(
                this::startFailedAutoStartVms,
                autoStartVmsRunnerIntervalInSeconds,
                autoStartVmsRunnerIntervalInSeconds,
                TimeUnit.SECONDS);
}

2.25 初始化配额管理

serviceLoader.load(QuotaManager.class);

• 注入 engine 调度线程池

@Inject
@ThreadPools(ThreadPools.ThreadPoolType.EngineScheduledThreadPool)
private ManagedScheduledExecutorService executor;

• 初始化
  • QuotaCacheIntervalInMinutes 配额调度周期。

@PostConstruct
    private void init() {
        long quotaCacheIntervalInMinutes = Config.<Long>getValue(ConfigValues.QuotaCacheIntervalInMinutes);
        executor.scheduleWithFixedDelay(this::updateQuotaCache,
                1,
                quotaCacheIntervalInMinutes,
                TimeUnit.MINUTES
        );
}

• 更新配额缓存

private synchronized void updateQuotaCacheImpl() {
        if (!isCacheUpdateNeeded()) {
            return;
        }

        log.debug("Updating Quota Cache...");
        long timeStart = System.currentTimeMillis();
        List<Quota> allQuotaIncludingConsumption = getQuotaDao().getAllQuotaIncludingConsumption();

        if (allQuotaIncludingConsumption.isEmpty()) {
            return;
        }

        HashMap<Guid, Map<Guid, Quota>> newStoragePoolQuotaMap = new HashMap<>();
        HashMap<Guid, Guid> newDefaultQuotaIdMap = new HashMap<>();

        for (Quota quota : allQuotaIncludingConsumption) {
            if (!newStoragePoolQuotaMap.containsKey(quota.getStoragePoolId())) {
                newStoragePoolQuotaMap.put(quota.getStoragePoolId(), new HashMap<>());
            }
            newStoragePoolQuotaMap.get(quota.getStoragePoolId()).put(quota.getId(), quota);

            if (quota.isDefault()) {
                newDefaultQuotaIdMap.put(quota.getStoragePoolId(), quota.getId());
            }
        }

        lock.writeLock().lock();
        try {
            storagePoolQuotaMap = newStoragePoolQuotaMap;
            storagePoolDefaultQuotaIdMap = newDefaultQuotaIdMap;
        } finally {
            lock.writeLock().unlock();
        }
        long timeEnd = System.currentTimeMillis();
        log.info("Quota Cache updated. ({} msec)", timeEnd-timeStart);
}

2.26 虚拟机迁移线程监控

• 虚拟机迁移监控主要使用到了 Reactive Streams（响应式流）。

serviceLoader.load(VmMigrationProgressMonitoring.class);

• 注入资源管理对象

@Inject
private ResourceManager resourceManager;

• 组装运行主机与虚拟机的对应关系

@PostConstruct
private void init() {
        log.info("Start initializing {}", getClass().getSimpleName());
        populateVdsAndVmsList();

        // Populate the VDS dictionary
        final List<VDS> allVdsList = hostDao.getAll();
        for (VDS curVds : allVdsList) {
            addVds(curVds, true, false);
        }

        log.info("Finished initializing {}", getClass().getSimpleName());
}

private void populateVdsAndVmsList() {
        final List<VmDynamic> vms = vmDynamicDao.getAll();
        vdsAndVmsList.putAll(vms.stream()
                .filter(vm -> !vm.getStatus().isNotRunning() && vm.getRunOnVds() != null)
                .collect(Collectors.groupingBy(VmDynamic::getRunOnVds,
                        Collectors.mapping(VmDynamic::getId, Collectors.toSet()))));
}

2.27 初始化与主机的可信认证明

• 生成可信任证明缓存

private void initAttestation() {
        List<Cluster> clusters = clusterDao.getTrustedClusters();
        List<VDS> trustedVdsList = new ArrayList<>();
        List<String> trustedVdsNames = new ArrayList<>();

        if (clusters == null || clusters.size() == 0) {
            return;
        }
        for (Cluster cluster : clusters) {
            List<VDS> hostsInCluster = vdsDao.getAllForClusterWithStatus(cluster.getId(), VDSStatus.Up);
            if (hostsInCluster != null) {
                trustedVdsList.addAll(hostsInCluster);
            }
        }

        for (VDS vds : trustedVdsList) {
            trustedVdsNames.add(vds.getHostName());
            setNonOperational(NonOperationalReason.UNINITIALIZED, vds);
        }

        try {
            AttestThread attestThread = new AttestThread(trustedVdsNames);
            attestThread.start();//start a thread to attest the hosts
        } catch (Exception e) {
            log.error("Failed to initialize attestation cache", e);
        }
 }

2.28 初始化预定义图标

private void updatePredefinedIcons() {
    serviceLoader.load(IconLoader.class);
}

• 初始化
  • 加载操作系统图标。
  • 确认默认系统图标是否存在。
  • 更新虚拟机默认图标。
  • 更新虚拟机状态图标。

@PostConstruct
private void init() {
    loadIconsToDatabase();
    ensureDefaultOsIconExists();
    updateVmIconDefaultsTable();
    updateVmStaticTable();
}

2.29 清除图标

• 删除所有未使用的图标

private void iconCleanup() {
    vmIconDao.removeAllUnusedIcons();
}

2.30 engine 的扩展管理

EngineExtensionsManager.getInstance().engineInitialize();

• 判断 engine 扩展目录是否存在，读取扩展属性文件

for (File directory : EngineLocalConfig.getInstance().getExtensionsDirectories()) {
            if (!directory.exists()) {
                log.warn("The directory '{}' cotaning configuration files does not exist.",
                        directory.getAbsolutePath());
            } else {

                // The order of the files inside the directory is relevant, as the objects are created in
                // the same order
                // that
                // the files are processed, so it is better to sort them so that objects will always be
                // created in the
                // same
                // order regardless of how the filesystem decides to store the entries of the directory:
                File[] files = directory.listFiles();
                if (files != null) {
                    sort(files);
                    for (File file : files) {
                        if (file.getName().endsWith(".properties")) {
                            try {
                                load(file);
                            } catch (Exception ex) {
                                log.error("Could not load extension based on configuration file '{}'. Please check the configuration file is valid. Exception message is: {}",
                                        file.getAbsolutePath(),
                                        ex.getMessage());
                                log.debug("", ex);
                            }
                        }
                    }
                }
            }
}

[root@rhvm423 extensions.d]# pwd
/etc/ovirt-engine/extensions.d
[root@rhvm423 extensions.d]# ll
total 8
-rw-------. 1 ovirt ovirt 543 May 22 19:09 internal-authn.properties
-rw-------. 1 ovirt ovirt 443 May 22 19:09 internal-authz.properties

• 这里的配置文件中，主要配置了认证与授权的相关扩展。
  • AuthN 系统主要用于认证 （Authentication），决定谁访问了系统。
  • AuthZ 系统主要用于授权 （Authorization），决定访问者具有什么样的权限。
  • internal-authz.properties 内容类似。

[root@rhvm423 extensions.d]# cat internal-authn.properties 
ovirt.engine.extension.name = internal-authn
ovirt.engine.extension.bindings.method = jbossmodule
ovirt.engine.extension.binding.jbossmodule.module = org.ovirt.engine.extension.aaa.jdbc
ovirt.engine.extension.binding.jbossmodule.class = org.ovirt.engine.extension.aaa.jdbc.binding.api.AuthnExtension
ovirt.engine.extension.provides = org.ovirt.engine.api.extensions.aaa.Authn
ovirt.engine.aaa.authn.profile.name = internal
ovirt.engine.aaa.authn.authz.plugin = internal-authz
config.datasource.file = /etc/ovirt-engine/aaa/internal.properties

• /etc/ovirt-engine/aaa/internal.properties 配置文件中，定义了认证功能的数据库的定义。
  • 使用了 aaa_jdbc 表空间。
    • users 表中存放了用户数据信息。
    • settings 表中存放了系统对于用户操作的相关设置（如密码错误次数、历史密码要求个数等）。

2.31 初始化认证配置库

• 认证配置文件是 AuthN 和 AuthZ 扩展的组合。
• 登录到系统的用户通过 AuthN 扩展进行身份验证，然后在 AuthZ 扩展中查找详细信息。

AuthenticationProfileRepository.getInstance();

2.32 执行进程信息记录

AcctUtils.reportReason(Acct.ReportReason.STARTUP, "Starting up engine");

@Override
public void invoke(ExtMap input, ExtMap output) {
        input.putIfAbsent(Base.InvokeKeys.CONTEXT, context);

        dumpMap("Invoke Input", input);
        ClassLoader savedClassLoader = Thread.currentThread().getContextClassLoader();
        try {
            Thread.currentThread().setContextClassLoader(classLoader);
            proxied.invoke(input, output);
        } catch (Throwable e) {
            output.mput(
                Base.InvokeKeys.RESULT,
                Base.InvokeResult.FAILED
            ).mput(
                Base.InvokeKeys.MESSAGE,
                String.format(
                    "Exception: %s: %s",
                    e.getClass(),
                    e.getMessage()
                )
            ).mput(
                ExtensionsManager.CAUSE_OUTPUT_KEY,
                e
            );
        } finally {
            Thread.currentThread().setContextClassLoader(savedClassLoader);
        }
        dumpMap("Invoke Output", output);
}