import org.apache.mesos.v1.Protos; //导入依赖的package包/类
public static FrameworkID getFrameworkId() {
    JuiceFramework frameworkInfo = null;
    try {
        frameworkInfo = daoUtils.queryFramework(MESOS_FRAMEWORK_TAG);
    } catch (DataBaseException ex) {
        log.error("get frameworkId from db failed {cause : db operation error}!");
        throw ex;
    }

    if (null == frameworkInfo || StringUtils.isBlank(frameworkInfo.getFrameworkId())) {
        return genFrameworkId();
    } else {
        log.info("get framework id from db, id : " + frameworkInfo.getFrameworkId());
        return Protos.FrameworkID.newBuilder().setValue(frameworkInfo.getFrameworkId()).build();
    }
}
 

import org.apache.mesos.v1.Protos; //导入依赖的package包/类
private static void addTask(@NotNull Map<Long, String> killMap, @NotNull Protos.AgentID agentId, @NotNull Task task, @NotNull List<Protos.TaskInfo> tasks, boolean isRetryTask, Address address) {
    boolean isToKilled = false;
    try {
        isToKilled = killMap.containsKey(task.getTaskId());
        //update db set taskAgentRel
        AuxiliaryService.updateTask(killMap, task.getTaskId(), agentId.getValue(), isToKilled, address);
        if(!isToKilled){
            tasks.add(task.getTask(agentId));
            log.info("resourceAllocation --> add task : " + task.getTaskId());
        }
    } catch (Exception e) {
        if(!isToKilled) {
            taskReserve(task, isRetryTask);
        }
    }
}
 

import org.apache.mesos.v1.Protos; //导入依赖的package包/类
public @NotNull
Protos.TaskInfo getTask(
        final @NotNull Protos.AgentID agentID
) {
    Protos.TaskInfo.Builder taskInfoBuilder = Protos.TaskInfo.newBuilder();
    taskInfoBuilder.setName(taskName);
    taskInfoBuilder.setTaskId(Protos.TaskID.newBuilder().setValue(generateTaskNameId(taskName, taskId, retry)));
    taskInfoBuilder.setAgentId(agentID);
    taskInfoBuilder.addAllResources(resources.protos());

    if(null != command && StringUtils.isNotBlank(command.getValue())) {
        //  run shell
        return taskInfoBuilder.setCommand(command.protos(true)).build();
    } else {
        //  run docker
        return taskInfoBuilder.setContainer(container.protos()).setCommand(command.protos(false)).build();
    }
}
 

import org.apache.mesos.v1.Protos; //导入依赖的package包/类
public @NotNull Protos.ContainerInfo protos() {

        Protos.ContainerInfo.Builder builder
                = Protos.ContainerInfo.newBuilder()
                .setDocker(docker.protos())
                .setType(getType());

        if(null != volumes && !volumes.isEmpty()) {
            volumes.forEach(
                    volume -> {
                        builder.addVolumes(Protos.Volume.newBuilder()
                                .setContainerPath(volume.getContainerPath())
                                .setHostPath(volume.getHostPath())
                                .setMode(volume.getMode())
                                .build());
                    }
            );
        }

        return builder.build();
    }
 

import org.apache.mesos.v1.Protos; //导入依赖的package包/类
public @NotNull Protos.CommandInfo protos(boolean isShell) {

        Protos.CommandInfo.Builder builder = Protos.CommandInfo.newBuilder();
        if(isShell) {
            builder.setValue(value);
        } else {
            builder.setShell(false);
            if (null != args && !args.isEmpty()) {
                builder.addAllArguments(args);
            }
        }

        if(null != uris && !uris.isEmpty()) {
            builder.addAllUris(uris.stream().map(uri -> Protos.CommandInfo.URI.newBuilder().setValue(uri).build()).collect(Collectors.toList()));
        }

        if(null != envs && !envs.isEmpty()) {
            Protos.Environment.Builder envBuilder = Protos.Environment.newBuilder();
            envBuilder.addAllVariables(envs.stream().map(env -> Protos.Environment.Variable.newBuilder().setName(env.getName()).setValue(env.getValue()).build()).collect(Collectors.toList()));
            builder.setEnvironment(envBuilder).build();
        }

        return builder.build();
    }
 

import org.apache.mesos.v1.Protos; //导入依赖的package包/类
/**
 * Utility method to more succinctly construct a {@link Call Call} of type {@link Type#ACKNOWLEDGE ACKNOWLEDGE}.
 * <p>
 *
 * @param frameworkId    The {@link Protos.FrameworkID} to be set on the {@link Call}
 * @param uuid           The {@link Protos.TaskStatus#getUuid() uuid} from the
 *                       {@link org.apache.mesos.v1.scheduler.Protos.Event.Update#getStatus() TaskStatus} received from Mesos.
 * @param agentId        The {@link Protos.TaskStatus#getAgentId() agentId} from the
 *                       {@link org.apache.mesos.v1.scheduler.Protos.Event.Update#getStatus() TaskStatus} received from Mesos.
 * @param taskId         The {@link Protos.TaskStatus#getTaskId() taskId} from the
 *                       {@link org.apache.mesos.v1.scheduler.Protos.Event.Update#getStatus() TaskStatus} received from Mesos.
 * @return  A {@link Call} with a configured {@link Acknowledge}.
 */
@NotNull
public static Call ackUpdate(
    @NotNull final Protos.FrameworkID frameworkId,
    @NotNull final ByteString uuid,
    @NotNull final Protos.AgentID agentId,
    @NotNull final Protos.TaskID taskId
) {
    return newBuilder()
        .setFrameworkId(frameworkId)
        .setType(Type.ACKNOWLEDGE)
        .setAcknowledge(
            Acknowledge.newBuilder()
                .setUuid(uuid)
                .setAgentId(agentId)
                .setTaskId(taskId)
                .build()
        )
        .build();
}
 

import org.apache.mesos.v1.Protos; //导入依赖的package包/类
/**
 * Utility method to more succinctly construct a {@link Call Call} of type {@link Type#SUBSCRIBE SUBSCRIBE}.
 * <p>
 *
 * @param frameworkId               The frameworkId to set on the {@link Protos.FrameworkInfo FrameworkInfo} and
 *                                  {@link Call Call} messages.
 * @param user                      The user to set on the {@link Protos.FrameworkInfo FrameworkInfo} message.
 * @param frameworkName             The name to set on the {@link Protos.FrameworkInfo FrameworkInfo} message.
 * @param failoverTimeoutSeconds    The failoverTimeoutSeconds to set on the
 *                                  {@link Protos.FrameworkInfo FrameworkInfo} message.
 * @return An {@link Call Call} of type {@link Type#SUBSCRIBE SUBSCRIBE} with the configured
 * {@link Subscribe Subscribe} sub-message.
 */
@NotNull
public static Call subscribe(
    @NotNull final Protos.FrameworkID frameworkId,
    @NotNull final String user,
    @NotNull final String frameworkName,
    final long failoverTimeoutSeconds
) {
    final Protos.FrameworkInfo frameworkInfo = Protos.FrameworkInfo.newBuilder()
        .setId(frameworkId)
        .setUser(user)
        .setName(frameworkName)
        .setFailoverTimeout(failoverTimeoutSeconds)
        .build();
    return subscribe(frameworkId, frameworkInfo);
}
 

import org.apache.mesos.v1.Protos; //导入依赖的package包/类
private Optional<String> getStatusMessage(Protos.TaskStatus status, Optional<SingularityTask> task) {
  if (status.hasMessage() && !Strings.isNullOrEmpty(status.getMessage())) {
    return Optional.of(status.getMessage());
  } else if (status.hasReason() && status.getReason() == Reason.REASON_CONTAINER_LIMITATION_MEMORY) {
    if (task.isPresent() && task.get().getTaskRequest().getDeploy().getResources().isPresent()) {
      if (task.get().getTaskRequest().getDeploy().getResources().get().getDiskMb() > 0) {
        return Optional.of(String.format("Task exceeded one or more memory limits (%s MB mem, %s MB disk).", task.get().getTaskRequest().getDeploy().getResources().get().getMemoryMb(),
            task.get().getTaskRequest().getDeploy().getResources().get().getDiskMb()));
      } else {
        return Optional.of(String.format("Task exceeded memory limit (%s MB mem).", task.get().getTaskRequest().getDeploy().getResources().get().getMemoryMb()));
      }

    }
    return Optional.of("Task exceeded memory limit.");
  } else if (status.hasReason() && status.getReason() == Reason.REASON_CONTAINER_LIMITATION_DISK) {
    if (task.isPresent() && task.get().getTaskRequest().getDeploy().getResources().isPresent()) {
      return Optional.of(String.format("Task exceeded disk limit (%s MB disk).", task.get().getTaskRequest().getDeploy().getResources().get().getDiskMb()));
    } else {
      return Optional.of("Task exceeded disk limit.");
    }
  }

  return Optional.absent();
}
 

import org.apache.mesos.v1.Protos; //导入依赖的package包/类
@Timed
public void handleCompletedTask(Optional<SingularityTask> task, SingularityTaskId taskId, boolean wasActive, long timestamp, ExtendedTaskState state,
  SingularityCreateResult taskHistoryUpdateCreateResult, Protos.TaskStatus status) {
  final SingularityDeployStatistics deployStatistics = getDeployStatistics(taskId.getRequestId(), taskId.getDeployId());

  if (wasActive) {
    taskManager.deleteActiveTask(taskId.getId());
  }

  if (!task.isPresent() || task.get().getTaskRequest().getRequest().isLoadBalanced()) {
    taskManager.createLBCleanupTask(taskId);
  }

  final Optional<PendingType> scheduleResult = handleCompletedTaskWithStatistics(task, taskId, timestamp, state, deployStatistics, taskHistoryUpdateCreateResult, status);

  if (taskHistoryUpdateCreateResult == SingularityCreateResult.EXISTED) {
    return;
  }

  updateDeployStatistics(deployStatistics, taskId, task, timestamp, state, scheduleResult);
}
 

import org.apache.mesos.v1.Protos; //导入依赖的package包/类
@Test
public void testGetPortByIndex() throws Exception{
  taskResources = new Resources(1, 1, 4, 0);

  final Protos.Resource portsResource = Protos.Resource.newBuilder()
      .setName("ports")
      .setType(Protos.Value.Type.RANGES)
      .setRanges(Protos.Value.Ranges.newBuilder()
          .addRange(Protos.Value.Range.newBuilder()
              .setBegin(31003)
              .setEnd(31004).build())
          .addRange(Protos.Value.Range.newBuilder()
              .setBegin(31000)
              .setEnd(31001).build())
          .build()).build();

  final SingularityRequest request = new SingularityRequestBuilder("test", RequestType.WORKER).build();
  final SingularityDeploy deploy = new SingularityDeployBuilder("test", "1")
      .setCommand(Optional.of("/bin/echo"))
      .setArguments(Optional.of(Collections.singletonList("wat")))
      .build();
  final SingularityTaskRequest taskRequest = new SingularityTaskRequest(request, deploy, pendingTask);
  final SingularityMesosTaskHolder task = builder.buildTask(offerHolder, Collections.singletonList(portsResource), taskRequest, taskResources, executorResources);
  assertEquals(31003L, task.getTask().getPortByIndex(2).get().longValue());
}
 

import org.apache.mesos.v1.Protos; //导入依赖的package包/类
@Test
public void testDockerMinimalNetworking() {
  taskResources = new Resources(1, 1, 0, 0);

  final SingularityRequest request = new SingularityRequestBuilder("test", RequestType.WORKER).build();
  final SingularityContainerInfo containerInfo = new SingularityContainerInfo(
      SingularityContainerType.DOCKER,
      Optional.absent(),
      Optional.of(new SingularityDockerInfo("docker-image", true, SingularityDockerNetworkType.NONE,
          Optional.absent(),
          Optional.absent(),
          Optional.absent(),
          Optional.absent())));
  final SingularityDeploy deploy = new SingularityDeployBuilder("test", "1")
      .setContainerInfo(Optional.of(containerInfo))
      .build();
  final SingularityTaskRequest taskRequest = new SingularityTaskRequest(request, deploy, pendingTask);
  final SingularityMesosTaskHolder task = builder.buildTask(offerHolder, Collections.emptyList(), taskRequest, taskResources, executorResources);

  assertEquals(Type.DOCKER, task.getMesosTask().getContainer().getType());
  assertEquals(Protos.ContainerInfo.DockerInfo.Network.NONE, task.getMesosTask().getContainer().getDocker().getNetwork());
}
 

import org.apache.mesos.v1.Protos; //导入依赖的package包/类
private static Map<String, Set<String>> generatorFacts(@NotNull Protos.Offer offer, Set<String> attributes) {
    Map<String, Set<String>> facts = Maps.newHashMap();
    //  add host
    if (StringUtils.isNotBlank(offer.getHostname())) {
        facts.put(Constraints.FIELD.HOSTNAME.getField(), Sets.newHashSet());
        facts.get(Constraints.FIELD.HOSTNAME.getField()).add(offer.getHostname());
    }
    //  add attr
    if (null != attributes && !attributes.isEmpty()) {
        facts.put(Constraints.FIELD.RACK_ID.getField(), Sets.newHashSet());
        facts.get(Constraints.FIELD.RACK_ID.getField()).addAll(attributes);
    }

    return facts;
}
 

import org.apache.mesos.v1.Protos; //导入依赖的package包/类
private static boolean allocatingUntilExhausted(Map<Long, String> killMap, Protos.AgentID agentId, Map<String, Set<String>> facts, ResourcesUtils hardware, List<Protos.TaskInfo> tasks, Address address) {

        long cacheTries = CACHE_TRIES;
        //  when either cpu or memory reach the picket line, will stop allocation task
        while (hardware.isAvailable()) {
            if (cacheTries > 0) {
                cacheTries = taskOrResourceExhausted(killMap, agentId, facts, hardware, tasks, true, address) ? 0 : cacheTries - 1;
            } else {
                if (taskOrResourceExhausted(killMap, agentId, facts, hardware, tasks, false, address)) {
                    return true;
                }
            }
        }
        return false;
    }
 

import org.apache.mesos.v1.Protos; //导入依赖的package包/类
private static boolean taskOrResourceExhausted(@NotNull Map<Long, String> killMap, Protos.AgentID agentId, Map<String, Set<String>> facts, ResourcesUtils hardware, List<Protos.TaskInfo> tasks, boolean isRetryTask,  Address address) {

        //  is task in cache exhausted ?
        String tskStr = isRetryTask ? cacheUtils.popFromQueue(TASK_RETRY_QUEUE) : cacheUtils.popFromQueue(TASK_QUEUE);
        if (StringUtils.isBlank(tskStr)) {
            return true;
        }

        //  is resource exhausted ?
        com.hujiang.juice.common.model.Task task = gson.fromJson(tskStr, Task.class);

        //  current offer not match task, try next
        if (!availableConstraints(task.getConstraints(), facts)) {
            task.getExpire().incrementOfferLack();
            taskReserve(task, isRetryTask);
        } else {
            if (hardware.allocating(task.getResources())) {
                //  accept task
                addTask(killMap, agentId, task, tasks, isRetryTask, address);
            } else {
                //  resources is exhausted
                task.getExpire().incrementResourceLack();
                taskReserve(task, isRetryTask);
                return true;
            }
        }
        return false;

    }
 

import org.apache.mesos.v1.Protos; //导入依赖的package包/类
private
@NotNull
Protos.Resource addResource(@NotNull String name, @NotNull double value) {
    return Protos.Resource.newBuilder()
            .setName(name)
            .setRole(role)
            .setType(Protos.Value.Type.SCALAR)
            .setScalar(Protos.Value.Scalar.newBuilder().setValue(value).build())
            .build();
}
 

import org.apache.mesos.v1.Protos; //导入依赖的package包/类
public
@NotNull
List<Protos.Resource> protos() {
    List<Protos.Resource> resources = new ArrayList<>();
    resources.add(addResource(CPUS, cpu));
    resources.add(addResource(MEMS, mem));
    return resources;
}
 

import org.apache.mesos.v1.Protos; //导入依赖的package包/类
public @NotNull
Protos.ContainerInfo.DockerInfo protos() {

    Protos.ContainerInfo.DockerInfo.Builder dockerBuild = Protos.ContainerInfo.DockerInfo.newBuilder()
            .setImage(image)
            .setForcePullImage(forcePullImage)
            .setPrivileged(privileged)
            .setNetwork(exchange());


    if (null != parameters) {
        parameters.forEach(
                par -> {
                    dockerBuild.addParameters(Protos.Parameter.newBuilder().setKey(par.getKey()).setValue(par.getValue()));
                }
        );
    }
    if (null != portMappings) {
        portMappings.forEach(
                por -> {
                    dockerBuild.addPortMappings(Protos.ContainerInfo.DockerInfo.PortMapping.newBuilder()
                            .setContainerPort(por.getContainerPort())
                            .setHostPort(por.getHostPort())
                            .setProtocol(por.getProtocol()));
                }
        );
    }
    return dockerBuild.build();
}
 

import org.apache.mesos.v1.Protos; //导入依赖的package包/类
private Protos.ContainerInfo.DockerInfo.Network exchange() {
    if(null == net) {
        return Protos.ContainerInfo.DockerInfo.Network.BRIDGE;
    }
    switch (net) {
        case HOST: return Protos.ContainerInfo.DockerInfo.Network.HOST;
        case NONE: return Protos.ContainerInfo.DockerInfo.Network.NONE;
        case USER: return Protos.ContainerInfo.DockerInfo.Network.USER;
        default: return Protos.ContainerInfo.DockerInfo.Network.BRIDGE;
    }
}
 

import org.apache.mesos.v1.Protos; //导入依赖的package包/类
@Test
public void testFrameworkInfoEvolve() {
    org.apache.mesos.Protos.FrameworkInfo.Builder frameworkBuilder = org.apache.mesos.Protos.FrameworkInfo.newBuilder()
        .setUser(user)
        .setName(name);

    Protos.FrameworkInfo frameworkInfo = EvolverDevolver.evolve(frameworkBuilder.buildPartial());

    assertEquals(user, frameworkInfo.getUser());
    assertEquals(name, frameworkInfo.getName());
}
 

import org.apache.mesos.v1.Protos; //导入依赖的package包/类
@Test
public void testFrameworkInfoDevolve() {
    Protos.FrameworkInfo.Builder frameworkBuilder = Protos.FrameworkInfo.newBuilder()
            .setUser(user)
            .setName(name);

    org.apache.mesos.Protos.FrameworkInfo frameworkInfo = EvolverDevolver.devolve(frameworkBuilder.buildPartial());

    assertEquals(user, frameworkInfo.getUser());
    assertEquals(name, frameworkInfo.getName());
}