import com.netflix.fenzo.ConstraintEvaluator; //导入依赖的package包/类
public MesosTaskManagerParameters(
		double cpus,
		ContainerType containerType,
		Option<String> containerImageName,
		ContaineredTaskManagerParameters containeredParameters,
		List<Protos.Volume> containerVolumes,
		List<ConstraintEvaluator> constraints,
		String command,
		Option<String> bootstrapCommand,
		Option<String> taskManagerHostname) {

	this.cpus = cpus;
	this.containerType = Preconditions.checkNotNull(containerType);
	this.containerImageName = Preconditions.checkNotNull(containerImageName);
	this.containeredParameters = Preconditions.checkNotNull(containeredParameters);
	this.containerVolumes = Preconditions.checkNotNull(containerVolumes);
	this.constraints = Preconditions.checkNotNull(constraints);
	this.command = Preconditions.checkNotNull(command);
	this.bootstrapCommand = Preconditions.checkNotNull(bootstrapCommand);
	this.taskManagerHostname = Preconditions.checkNotNull(taskManagerHostname);
}
 

import com.netflix.fenzo.ConstraintEvaluator; //导入依赖的package包/类
private static List<ConstraintEvaluator> parseConstraints(String mesosConstraints) {

		if (mesosConstraints == null || mesosConstraints.isEmpty()) {
			return Collections.emptyList();
		} else {
			List<ConstraintEvaluator> constraints = new ArrayList<>();

			for (String constraint : mesosConstraints.split(",")) {
				if (constraint.isEmpty()) {
					continue;
				}
				final String[] constraintList = constraint.split(":");
				if (constraintList.length != 2) {
					continue;
				}
				addHostAttrValueConstraint(constraints, constraintList[0], constraintList[1]);
			}

			return constraints;
		}
	}
 

import com.netflix.fenzo.ConstraintEvaluator; //导入依赖的package包/类
private TaskRequest getTask(final String jobName) {
    TaskRequest result = mock(TaskRequest.class);
    when(result.getCPUs()).thenReturn(1.0d);
    when(result.getMemory()).thenReturn(128.0d);
    when(result.getHardConstraints()).thenAnswer(new Answer<List<? extends ConstraintEvaluator>>() {
        @Override
        public List<? extends ConstraintEvaluator> answer(final InvocationOnMock invocationOnMock) throws Throwable {
            return ImmutableList.of(AppConstraintEvaluator.getInstance());
        }
    });
    when(result.getId()).thenReturn(new TaskContext(jobName, Collections.singletonList(0), ExecutionType.READY).getId());
    return result;
}
 

import com.netflix.fenzo.ConstraintEvaluator; //导入依赖的package包/类
private static void addHostAttrValueConstraint(List<ConstraintEvaluator> constraints, String constraintKey, final String constraintValue) {
	constraints.add(new HostAttrValueConstraint(constraintKey, new Func1<String, String>() {
		@Override
		public String call(String s) {
			return constraintValue;
		}
	}));
}
 

import com.netflix.fenzo.ConstraintEvaluator; //导入依赖的package包/类
@Test
public void givenOneConstraintInConfigShouldBeParsed() throws Exception {

	MesosTaskManagerParameters mesosTaskManagerParameters = MesosTaskManagerParameters.create(withHardHostAttrConstraintConfiguration("cluster:foo"));
	assertThat(mesosTaskManagerParameters.constraints().size(), is(1));
	ConstraintEvaluator firstConstraintEvaluator = new HostAttrValueConstraint("cluster", new Func1<String, String>() {
		@Override
		public String call(String s) {
			return "foo";
		}
	});
	assertThat(mesosTaskManagerParameters.constraints().get(0).getName(), is(firstConstraintEvaluator.getName()));
}
 

import com.netflix.fenzo.ConstraintEvaluator; //导入依赖的package包/类
/**
 * Initialize the resource manager.
 */
public void initialize() {
	ContainerSpecification containerSpecification = new ContainerSpecification();
	ContaineredTaskManagerParameters containeredParams =
		new ContaineredTaskManagerParameters(1024, 768, 256, 4, new HashMap<String, String>());
	MesosTaskManagerParameters tmParams = new MesosTaskManagerParameters(
		1.0,
		MesosTaskManagerParameters.ContainerType.MESOS,
		Option.<String>empty(),
		containeredParams,
		Collections.<Protos.Volume>emptyList(),
		Collections.<ConstraintEvaluator>emptyList(),
		"",
		Option.<String>empty(),
		Option.<String>empty());

	TestActorRef<TestingMesosFlinkResourceManager> resourceManagerRef =
		TestActorRef.create(system, MesosFlinkResourceManager.createActorProps(
			TestingMesosFlinkResourceManager.class,
			config,
			mesosConfig,
			workerStore,
			highAvailabilityServices.getJobManagerLeaderRetriever(HighAvailabilityServices.DEFAULT_JOB_ID),
			tmParams,
			containerSpecification,
			artifactResolver,
			LOG));
	resourceManagerInstance = resourceManagerRef.underlyingActor();
	resourceManager = new AkkaActorGateway(resourceManagerRef, HighAvailabilityServices.DEFAULT_LEADER_ID);

	verify(schedulerDriver).start();
	resourceManagerInstance.connectionMonitor.expectMsgClass(ConnectionMonitor.Start.class);
}
 

import com.netflix.fenzo.ConstraintEvaluator; //导入依赖的package包/类
/**
 * Returns a "soft" constraint, in the form of a {@link VMTaskFitnessCalculator}, based on a specified "hard"
 * constraint, in the form of a {@link ConstraintEvaluator}.
 *
 * @param c the "hard" constraint to convert
 * @return a "soft" constraint version of {@code c}
 */
public static VMTaskFitnessCalculator get(final ConstraintEvaluator c) {
    // This fitness calculator return 0 or 1. This can possibly be improved upon by the ConstraintEvaluator using its
    // own logic.
    return new VMTaskFitnessCalculator() {
        @Override
        public String getName() {
            return c.getName();
        }
        @Override
        public double calculateFitness(TaskRequest taskRequest, VirtualMachineCurrentState targetVM, TaskTrackerState taskTrackerState) {
            return c.evaluate(taskRequest, targetVM, taskTrackerState).isSuccessful()? 1.0 : 0.0;
        }
    };
}
 

import com.netflix.fenzo.ConstraintEvaluator; //导入依赖的package包/类
@Override
public List<? extends ConstraintEvaluator> getHardConstraints() {
    return Collections.singletonList(AppConstraintEvaluator.getInstance());
}
 

import com.netflix.fenzo.ConstraintEvaluator; //导入依赖的package包/类
private TaskRequest getTaskRequest() {
    final String taskId = nextTaskId();
    return new TaskRequest() {
        @Override
        public String getId() {
            return taskId;
        }

        @Override
        public String taskGroupName() {
            return frameworkName;
        }

        @Override
        public double getCPUs() {
            return taskCPU;
        }

        @Override
        public double getMemory() {
            return taskMEM;
        }

        @Override
        public double getNetworkMbps() {
            return taskNET;
        }

        @Override
        public double getDisk() {
            return taskDISK;
        }

        @Override
        public int getPorts() {
            return 1;
        }

        @Override
        public List<? extends ConstraintEvaluator> getHardConstraints() {
      	  List<ConstraintEvaluator> list = new ArrayList<ConstraintEvaluator>();
      	  list.add(new ExclusiveHostConstraint()); // Force a exclusive host as we are using HOST network
      	  if (!hostAttribute[0].equals(TASK_HOST_ATTRIBUTE_NAME_NOT_DEFINED))
      	  {
      	   list.add( new HostAttrValueConstraint(hostAttribute[0], new Func1<java.lang.String,java.lang.String>() {

				@Override
				public String call(String arg0) {
					// TODO, as we only have one framework. Otherwise need a task id to framework map and framework to attribute map
					return hostAttribute[1];
				}
            	  
			} ));
      	  }
      	  return list;
        }

        @Override
        public List<? extends VMTaskFitnessCalculator> getSoftConstraints() {
            return null;
        }
    };
}
 

import com.netflix.fenzo.ConstraintEvaluator; //导入依赖的package包/类
/**
 * Get the placement constraints.
 */
public List<ConstraintEvaluator> constraints() {
	return constraints;
}
 

import com.netflix.fenzo.ConstraintEvaluator; //导入依赖的package包/类
/**
 * Create the Mesos TaskManager parameters.
 *
 * @param flinkConfig the TM configuration.
 */
public static MesosTaskManagerParameters create(Configuration flinkConfig) {

	List<ConstraintEvaluator> constraints = parseConstraints(flinkConfig.getString(MESOS_CONSTRAINTS_HARD_HOSTATTR));
	// parse the common parameters
	ContaineredTaskManagerParameters containeredParameters = ContaineredTaskManagerParameters.create(
		flinkConfig,
		flinkConfig.getInteger(MESOS_RM_TASKS_MEMORY_MB),
		flinkConfig.getInteger(MESOS_RM_TASKS_SLOTS));

	double cpus = flinkConfig.getDouble(MESOS_RM_TASKS_CPUS);
	if (cpus <= 0.0) {
		cpus = Math.max(containeredParameters.numSlots(), 1.0);
	}

	// parse the containerization parameters
	String imageName = flinkConfig.getString(MESOS_RM_CONTAINER_IMAGE_NAME);

	ContainerType containerType;
	String containerTypeString = flinkConfig.getString(MESOS_RM_CONTAINER_TYPE);
	switch (containerTypeString) {
		case MESOS_RESOURCEMANAGER_TASKS_CONTAINER_TYPE_MESOS:
			containerType = ContainerType.MESOS;
			break;
		case MESOS_RESOURCEMANAGER_TASKS_CONTAINER_TYPE_DOCKER:
			containerType = ContainerType.DOCKER;
			if (imageName == null || imageName.length() == 0) {
				throw new IllegalConfigurationException(MESOS_RM_CONTAINER_IMAGE_NAME.key() +
					" must be specified for docker container type");
			}
			break;
		default:
			throw new IllegalConfigurationException("invalid container type: " + containerTypeString);
	}

	Option<String> containerVolOpt = Option.<String>apply(flinkConfig.getString(MESOS_RM_CONTAINER_VOLUMES));

	List<Protos.Volume> containerVolumes = buildVolumes(containerVolOpt);

	//obtain Task Manager Host Name from the configuration
	Option<String> taskManagerHostname = Option.apply(flinkConfig.getString(MESOS_TM_HOSTNAME));

	//obtain command-line from the configuration
	String tmCommand = flinkConfig.getString(MESOS_TM_CMD);
	Option<String> tmBootstrapCommand = Option.apply(flinkConfig.getString(MESOS_TM_BOOTSTRAP_CMD));

	return new MesosTaskManagerParameters(
		cpus,
		containerType,
		Option.apply(imageName),
		containeredParameters,
		containerVolumes,
		constraints,
		tmCommand,
		tmBootstrapCommand,
		taskManagerHostname);
}
 

import com.netflix.fenzo.ConstraintEvaluator; //导入依赖的package包/类
@Override
public List<? extends ConstraintEvaluator> getHardConstraints() {
	return params.constraints();
}
 

import com.netflix.fenzo.ConstraintEvaluator; //导入依赖的package包/类
/**
 * Create mock RM dependencies.
 */
Context() throws Exception {
	rpcService = new TestingRpcService();
	fatalErrorHandler = new TestingFatalErrorHandler();
	rmServices = new MockMesosResourceManagerRuntimeServices();
	mesosServices = new MockMesosServices();

	// TaskExecutor templating
	ContainerSpecification containerSpecification = new ContainerSpecification();
	ContaineredTaskManagerParameters containeredParams =
		new ContaineredTaskManagerParameters(1024, 768, 256, 4, new HashMap<String, String>());
	MesosTaskManagerParameters tmParams = new MesosTaskManagerParameters(
		1.0, MesosTaskManagerParameters.ContainerType.MESOS, Option.<String>empty(), containeredParams,
		Collections.<Protos.Volume>emptyList(), Collections.<ConstraintEvaluator>emptyList(), "", Option.<String>empty(),
		Option.<String>empty());

	// resource manager
	rmConfiguration = new ResourceManagerConfiguration(
		Time.seconds(5L),
		Time.seconds(5L));
	rmResourceID = ResourceID.generate();
	resourceManager =
		new TestingMesosResourceManager(
			rpcService,
			RM_ADDRESS,
			rmResourceID,
			rmConfiguration,
			rmServices.highAvailabilityServices,
			rmServices.heartbeatServices,
			rmServices.slotManager,
			rmServices.metricRegistry,
			rmServices.jobLeaderIdService,
			fatalErrorHandler,
			// Mesos specifics
			flinkConfig,
			mesosServices,
			rmServices.mesosConfig,
			tmParams,
			containerSpecification
		);

	// TaskExecutors
	task1Executor = mockTaskExecutor(task1);
	task2Executor = mockTaskExecutor(task2);
	task3Executor = mockTaskExecutor(task3);

	// JobMaster
	jobMaster1 = mockJobMaster(rmServices, new JobID(1, 0));
}
 

import com.netflix.fenzo.ConstraintEvaluator; //导入依赖的package包/类
@Override
public List<? extends ConstraintEvaluator> getHardConstraints() {
    return null;
}
 

import com.netflix.fenzo.ConstraintEvaluator; //导入依赖的package包/类
@Override
public List<? extends ConstraintEvaluator> getHardConstraints() {
	return Collections.emptyList();
}
 

import com.netflix.fenzo.ConstraintEvaluator; //导入依赖的package包/类
private TaskRequest getTaskRequest(final int id) {
    final String taskId = "" + id;
    final AtomicReference<TaskRequest.AssignedResources> assgndResRef = new AtomicReference<>();
    return new TaskRequest() {
        @Override
        public String getId() {
            return taskId;
        }

        @Override
        public String taskGroupName() {
            return "";
        }

        @Override
        public double getCPUs() {
            return 1.0;
        }

        @Override
        public double getMemory() {
            return 1024;
        }

        @Override
        public double getNetworkMbps() {
            return 0;
        }

        @Override
        public double getDisk() {
            return 10;
        }

        @Override
        public int getPorts() {
            return 1;
        }

        @Override
        public Map<String, Double> getScalarRequests() {
            return null;
        }

        @Override
        public List<? extends ConstraintEvaluator> getHardConstraints() {
            return null;
        }

        @Override
        public List<? extends VMTaskFitnessCalculator> getSoftConstraints() {
            return null;
        }

        @Override
        public void setAssignedResources(AssignedResources assignedResources) {
            assgndResRef.set(assignedResources);
        }

        @Override
        public AssignedResources getAssignedResources() {
            return assgndResRef.get();
        }

        @Override
        public Map<String, NamedResourceSetRequest> getCustomNamedResources() {
            return Collections.emptyMap();
        }
    };
}
 

import com.netflix.fenzo.ConstraintEvaluator; //导入依赖的package包/类
@Override
public List<? extends ConstraintEvaluator> getHardConstraints() {
    return Collections.emptyList();
}