mesos/src/main/java/org/apache/hama/bsp/MesosScheduler.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 org.apache.hama.bsp;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
import org.apache.mesos.Protos;
import org.apache.mesos.Protos.*;
import org.apache.mesos.MesosSchedulerDriver;
import org.apache.mesos.Scheduler;
import org.apache.mesos.SchedulerDriver;

import java.util.*;
import java.util.concurrent.atomic.AtomicReference;

public class MesosScheduler extends TaskWorkerManager implements Scheduler {
  public static final Log LOG = LogFactory.getLog(MesosScheduler.class);

  private ResourceManager resourceManager;

  SchedulerDriver driver;

  public MesosScheduler() {
  }

  @Override
  public void init(AtomicReference<GroomServerManager> groomServerManager,
      Configuration conf) {
    super.init(groomServerManager, conf);
    start();
    resourceManager = new ResourceManager(conf, groomServerManager, driver);
  }

  private void start() {
    try {
      FrameworkInfo frameworkInfo = FrameworkInfo.newBuilder()
          .setUser("")
          // Let Mesos fill in the user.
          .setCheckpoint(conf.getBoolean("bsp.master.port", false))
          .setRole(conf.get("hama.mesos.role", "*"))
          .setName(
              "Hama: (Master Port: " + conf.get("bsp.groom.rpc.port") + ","
                  + " WebUI port: " + conf.get("bsp.http.groomserver.port")
                  + ")").build();

      String master = conf.get("hama.mesos.master", "local");
      driver = new MesosSchedulerDriver(this, frameworkInfo, master);
      driver.start();
    } catch (Exception e) {
      // If the MesosScheduler can't be loaded, the JobTracker won't be useful
      // at all, so crash it now so that the user notices.
      LOG.fatal("Failed to start MesosScheduler", e);
      System.exit(1);
    }
  }

  @Override
  public synchronized void disconnected(SchedulerDriver schedulerDriver) {
    LOG.warn("Disconnected from Mesos master.");
  }

  @Override
  public synchronized void error(SchedulerDriver schedulerDriver, String s) {
    LOG.error("Error from scheduler driver: " + s);
  }

  @Override
  public synchronized void executorLost(SchedulerDriver schedulerDriver,
      ExecutorID executorID, SlaveID slaveID, int status) {
    LOG.warn("Executor " + executorID.getValue() + " lost with status "
        + status + " on slave " + slaveID);
  }

  @Override
  public synchronized void frameworkMessage(SchedulerDriver schedulerDriver,
      ExecutorID executorID, SlaveID slaveID, byte[] bytes) {
    LOG.info("Framework Message of " + bytes.length + " bytes"
        + " from executor " + executorID.getValue() + " on slave "
        + slaveID.getValue());
  }

  @Override
  public synchronized void offerRescinded(SchedulerDriver schedulerDriver,
      OfferID offerID) {
    LOG.warn("Rescinded offer: " + offerID.getValue());
  }

  @Override
  public void registered(SchedulerDriver driver, FrameworkID frameID,
      MasterInfo masterInfo) {
    LOG.info("Registered as " + frameID.getValue() + " with master "
        + masterInfo);
  }

  @Override
  public void reregistered(SchedulerDriver arg0, MasterInfo masterInfo) {
    LOG.info("Re-registered with master " + masterInfo);
  }

  @Override
  public void resourceOffers(SchedulerDriver schedulerDriver, List<Offer> offers) {
    resourceManager.resourceOffers(schedulerDriver, offers);
  }

  @Override
  public synchronized void slaveLost(SchedulerDriver schedulerDriver,
      SlaveID slaveID) {
    LOG.warn("Slave lost: " + slaveID.getValue());
  }

  @Override
  public synchronized void statusUpdate(SchedulerDriver schedulerDriver,
      Protos.TaskStatus taskStatus) {
    LOG.info("Status update of " + taskStatus.getTaskId().getValue() + " to "
        + taskStatus.getState().name() + " with message "
        + taskStatus.getMessage());

    switch (taskStatus.getState()) {
    case TASK_FINISHED:
    case TASK_FAILED:
    case TASK_KILLED:
    case TASK_LOST:
    case TASK_STAGING:
    case TASK_STARTING:
    case TASK_RUNNING:
      break;
    default:
      LOG.error("Unexpected TaskStatus: " + taskStatus.getState().name());
      break;
    }
  }

  @Override
  public TaskWorker spawnWorker(JobInProgress jip) {
    LOG.info("Spawning worker for: " + jip);
    return resourceManager.new MesosTaskWorker(jip);
  }

}