flink-instrumentation/workspace/flink-1.4.1-instrumented/flink-1.4.1/flink-mesos/src/main/java/org/apache/flink/mesos/runtime/clusterframework/LaunchableMesosWorker.java
/**
	 * Construct the TaskInfo needed to launch the worker.
	 * @param slaveId the assigned slave.
	 * @param allocation the resource allocation (available resources).
	 * @return a fully-baked TaskInfo.
	 */
Construct the TaskInfo needed to launch the worker.
@param slaveId the assigned slave.
@param allocation the resource allocation (available resources).
@return a fully-baked TaskInfo.
Construct the TaskInfo needed to launch the worker . @param slaveId the assigned slave . @param allocation the resource allocation ( available resources ) . @return a fully - baked TaskInfo .
@Override
	public Protos.TaskInfo launch(Protos.SlaveID slaveId, MesosResourceAllocation allocation) {
		ContaineredTaskManagerParameters tmParams = params.containeredParameters();
		final Configuration dynamicProperties = new Configuration();
		dynamicProperties.addAll(containerSpec.getDynamicConfiguration());
		final Protos.TaskInfo.Builder taskInfo = Protos.TaskInfo.newBuilder()
			.setSlaveId(slaveId)
			.setTaskId(taskID)
			.setName(taskID.getValue());
		Set<String> roles = mesosConfiguration.roles();
		taskInfo.addAllResources(allocation.takeScalar("cpus", taskRequest.getCPUs(), roles));
		taskInfo.addAllResources(allocation.takeScalar("mem", taskRequest.getMemory(), roles));
		final Protos.CommandInfo.Builder cmd = taskInfo.getCommandBuilder();
		final Protos.Environment.Builder env = cmd.getEnvironmentBuilder();
		final StringBuilder jvmArgs = new StringBuilder();
		Option<String> taskManagerHostnameOption = params.getTaskManagerHostname();
		if (taskManagerHostnameOption.isDefined()) {
			final String taskManagerHostname = MesosTaskManagerParameters.TASK_ID_PATTERN
				.matcher(taskManagerHostnameOption.get())
				.replaceAll(Matcher.quoteReplacement(taskID.getValue()));
			dynamicProperties.setString(ConfigConstants.TASK_MANAGER_HOSTNAME_KEY, taskManagerHostname);
		}
		List<Protos.Resource> portResources = allocation.takeRanges("ports", TM_PORT_KEYS.length, roles);
		taskInfo.addAllResources(portResources);
		Iterator<String> portsToAssign = Iterators.forArray(TM_PORT_KEYS);
		rangeValues(portResources).forEach(port -> dynamicProperties.setLong(portsToAssign.next(), port));
		if (portsToAssign.hasNext()) {
			throw new IllegalArgumentException("insufficient # of ports assigned");
		}
		for (ContainerSpecification.Artifact artifact : containerSpec.getArtifacts()) {
			cmd.addUris(Utils.uri(resolver, artifact));
		}
		for (Map.Entry<String, String> entry : params.containeredParameters().taskManagerEnv().entrySet()) {
			env.addVariables(variable(entry.getKey(), entry.getValue()));
		}
		for (Map.Entry<String, String> entry : containerSpec.getEnvironmentVariables().entrySet()) {
			env.addVariables(variable(entry.getKey(), entry.getValue()));
		}
		env.addVariables(variable(MesosConfigKeys.ENV_FLINK_CONTAINER_ID, taskInfo.getTaskId().getValue()));
		jvmArgs.append(" -Xms").append(tmParams.taskManagerHeapSizeMB()).append("m");
		jvmArgs.append(" -Xmx").append(tmParams.taskManagerHeapSizeMB()).append("m");
		if (tmParams.taskManagerDirectMemoryLimitMB() >= 0) {
			jvmArgs.append(" -XX:MaxDirectMemorySize=").append(tmParams.taskManagerDirectMemoryLimitMB()).append("m");
		}
		jvmArgs.append(' ').append(
			ContainerSpecification.formatSystemProperties(containerSpec.getSystemProperties()));
		env.addVariables(variable(MesosConfigKeys.ENV_JVM_ARGS, jvmArgs.toString()));
		env.addVariables(variable(MesosConfigKeys.ENV_TASK_NAME, taskInfo.getTaskId().getValue()));
		env.addVariables(variable(MesosConfigKeys.ENV_FRAMEWORK_NAME, mesosConfiguration.frameworkInfo().getName()));
		StringBuilder launchCommand = new StringBuilder();
		if (params.bootstrapCommand().isDefined()) {
			launchCommand.append(params.bootstrapCommand().get()).append(" && ");
		}
		launchCommand
			.append(params.command())
			.append(" ")
			.append(ContainerSpecification.formatSystemProperties(dynamicProperties));
		cmd.setValue(launchCommand.toString());
		Protos.ContainerInfo.Builder containerInfo = Protos.ContainerInfo.newBuilder();
		containerInfo.setType(Protos.ContainerInfo.Type.MESOS);
		switch (params.containerType()) {
			case MESOS:
				if (params.containerImageName().isDefined()) {
					containerInfo
						.setMesos(Protos.ContainerInfo.MesosInfo.newBuilder()
							.setImage(Protos.Image.newBuilder()
								.setType(Protos.Image.Type.DOCKER)
								.setDocker(Protos.Image.Docker.newBuilder()
									.setName(params.containerImageName().get()))));
				}
				break;
			case DOCKER:
				assert(params.containerImageName().isDefined());
				containerInfo
					.setType(Protos.ContainerInfo.Type.DOCKER)
					.setDocker(Protos.ContainerInfo.DockerInfo.newBuilder()
						.setNetwork(Protos.ContainerInfo.DockerInfo.Network.HOST)
						.setImage(params.containerImageName().get()));
				break;
			default:
				throw new IllegalStateException("unsupported container type");
		}
		containerInfo.addAllVolumes(params.containerVolumes());
		taskInfo.setContainer(containerInfo);
		return taskInfo.build();
	}
@ Override public Protos . TaskInfo launch ( Protos . SlaveID slaveId , MesosResourceAllocation allocation ) { ContaineredTaskManagerParameters tmParams = params . containeredParameters ( ) ; final Configuration dynamicProperties = new Configuration ( ) ; dynamicProperties . addAll ( containerSpec . getDynamicConfiguration ( ) ) ; final Protos . TaskInfo . Builder taskInfo = Protos . TaskInfo . newBuilder ( ) . setSlaveId ( slaveId ) . setTaskId ( taskID ) . setName ( taskID . getValue ( ) ) ; Set < String > roles = mesosConfiguration . roles ( ) ; taskInfo . addAllResources ( allocation . takeScalar ( "cpus" , taskRequest . getCPUs ( ) , roles ) ) ; taskInfo . addAllResources ( allocation . takeScalar ( "mem" , taskRequest . getMemory ( ) , roles ) ) ; final Protos . CommandInfo . Builder cmd = taskInfo . getCommandBuilder ( ) ; final Protos . Environment . Builder env = cmd . getEnvironmentBuilder ( ) ; final StringBuilder jvmArgs = new StringBuilder ( ) ; Option < String > taskManagerHostnameOption = params . getTaskManagerHostname ( ) ; if ( taskManagerHostnameOption . isDefined ( ) ) { final String taskManagerHostname = MesosTaskManagerParameters . TASK_ID_PATTERN . matcher ( taskManagerHostnameOption . get ( ) ) . replaceAll ( Matcher . quoteReplacement ( taskID . getValue ( ) ) ) ; dynamicProperties . setString ( ConfigConstants . TASK_MANAGER_HOSTNAME_KEY , taskManagerHostname ) ; } List < Protos . Resource > portResources = allocation . takeRanges ( "ports" , TM_PORT_KEYS . length , roles ) ; taskInfo . addAllResources ( portResources ) ; Iterator < String > portsToAssign = Iterators . forArray ( TM_PORT_KEYS ) ; rangeValues ( portResources ) . forEach ( port -> dynamicProperties . setLong ( portsToAssign . next ( ) , port ) ) ; if ( portsToAssign . hasNext ( ) ) { throw new IllegalArgumentException ( "insufficient # of ports assigned" ) ; } for ( ContainerSpecification . Artifact artifact : containerSpec . getArtifacts ( ) ) { cmd . addUris ( Utils . uri ( resolver , artifact ) ) ; } for ( Map . Entry < String , String > entry : params . containeredParameters ( ) . taskManagerEnv ( ) . entrySet ( ) ) { env . addVariables ( variable ( entry . getKey ( ) , entry . getValue ( ) ) ) ; } for ( Map . Entry < String , String > entry : containerSpec . getEnvironmentVariables ( ) . entrySet ( ) ) { env . addVariables ( variable ( entry . getKey ( ) , entry . getValue ( ) ) ) ; } env . addVariables ( variable ( MesosConfigKeys . ENV_FLINK_CONTAINER_ID , taskInfo . getTaskId ( ) . getValue ( ) ) ) ; jvmArgs . append ( " -Xms" ) . append ( tmParams . taskManagerHeapSizeMB ( ) ) . append ( "m" ) ; jvmArgs . append ( " -Xmx" ) . append ( tmParams . taskManagerHeapSizeMB ( ) ) . append ( "m" ) ; if ( tmParams . taskManagerDirectMemoryLimitMB ( ) >= 0 ) { jvmArgs . append ( " -XX:MaxDirectMemorySize=" ) . append ( tmParams . taskManagerDirectMemoryLimitMB ( ) ) . append ( "m" ) ; } jvmArgs . append ( ' ' ) . append ( ContainerSpecification . formatSystemProperties ( containerSpec . getSystemProperties ( ) ) ) ; env . addVariables ( variable ( MesosConfigKeys . ENV_JVM_ARGS , jvmArgs . toString ( ) ) ) ; env . addVariables ( variable ( MesosConfigKeys . ENV_TASK_NAME , taskInfo . getTaskId ( ) . getValue ( ) ) ) ; env . addVariables ( variable ( MesosConfigKeys . ENV_FRAMEWORK_NAME , mesosConfiguration . frameworkInfo ( ) . getName ( ) ) ) ; StringBuilder launchCommand = new StringBuilder ( ) ; if ( params . bootstrapCommand ( ) . isDefined ( ) ) { launchCommand . append ( params . bootstrapCommand ( ) . get ( ) ) . append ( " && " ) ; } launchCommand . append ( params . command ( ) ) . append ( " " ) . append ( ContainerSpecification . formatSystemProperties ( dynamicProperties ) ) ; cmd . setValue ( launchCommand . toString ( ) ) ; Protos . ContainerInfo . Builder containerInfo = Protos . ContainerInfo . newBuilder ( ) ; containerInfo . setType ( Protos . ContainerInfo . Type . MESOS ) ; switch ( params . containerType ( ) ) { case MESOS : if ( params . containerImageName ( ) . isDefined ( ) ) { containerInfo . setMesos ( Protos . ContainerInfo . MesosInfo . newBuilder ( ) . setImage ( Protos . Image . newBuilder ( ) . setType ( Protos . Image . Type . DOCKER ) . setDocker ( Protos . Image . Docker . newBuilder ( ) . setName ( params . containerImageName ( ) . get ( ) ) ) ) ) ; } break ; case DOCKER : assert ( params . containerImageName ( ) . isDefined ( ) ) ; containerInfo . setType ( Protos . ContainerInfo . Type . DOCKER ) . setDocker ( Protos . ContainerInfo . DockerInfo . newBuilder ( ) . setNetwork ( Protos . ContainerInfo . DockerInfo . Network . HOST ) . setImage ( params . containerImageName ( ) . get ( ) ) ) ; break ; default : throw new IllegalStateException ( "unsupported container type" ) ; } containerInfo . addAllVolumes ( params . containerVolumes ( ) ) ; taskInfo . setContainer ( containerInfo ) ; return taskInfo . build ( ) ; }
Construct the TaskInfo needed to launch the worker.
Construct the TaskInfo needed to launch the worker .
// incorporate the dynamic properties set by the template // build a TaskInfo with assigned resources, environment variables, etc // take needed resources from the overall allocation, under the assumption of adequate resources //configure task manager hostname property if hostname override property is supplied // replace the TASK_ID pattern by the actual task id value of the Mesos task // take needed ports for the TM // ship additional files // propagate environment variables // propagate the Mesos task ID to the TM // finalize the memory parameters // pass dynamic system properties // finalize JVM args // populate TASK_NAME and FRAMEWORK_NAME environment variables to the TM container // build the launch command w/ dynamic application properties // build the container info // in event that no docker image or mesos image name is specified, we must still // set type to MESOS // add any volumes to the containerInfo