private void prepareCustomExecutor(final TaskInfo.Builder bldr, final SingularityTaskId taskId, final SingularityTaskRequest task, final SingularityOfferHolder offerHolder,
      final Optional<long[]> ports, final Resources desiredExecutorResources) {
    CommandInfo.Builder commandBuilder = CommandInfo.newBuilder().setValue(task.getDeploy().getCustomExecutorCmd().get());

    prepareEnvironment(task, taskId, commandBuilder, offerHolder, ports);

    if (task.getDeploy().getUser().isPresent()) {
      commandBuilder.setUser(task.getDeploy().getUser().get());
    }

    prepareMesosUriDownloads(task.getPendingTask().getExtraArtifacts(), commandBuilder);

    bldr.setExecutor(ExecutorInfo.newBuilder()
        .setCommand(commandBuilder.build())
        .setExecutorId(ExecutorID.newBuilder().setValue(task.getDeploy().getCustomExecutorId().or(idGenerator.getNextExecutorId())))
        .setSource(task.getDeploy().getCustomExecutorSource().or(taskId.getId())) // set source to taskId for use in statistics endpoint, TODO: remove
        .setLabels(Labels.newBuilder().addLabels(Label.newBuilder().setKey("taskId").setValue(taskId.getId())))
        .addAllResources(buildMesosResources(desiredExecutorResources, task.getRequest().getRequiredRole()))
        .build()
        );

    if (task.getDeploy().getExecutorData().isPresent()) {
      final ExecutorDataBuilder executorDataBldr = task.getDeploy().getExecutorData().get().toBuilder();

      String defaultS3Bucket = "";
      String s3UploaderKeyPattern = "";

      if (configuration.getS3ConfigurationOptional().isPresent()) {
        if (task.getRequest().getGroup().isPresent() && configuration.getS3ConfigurationOptional().get().getGroupOverrides().containsKey(task.getRequest().getGroup().get())) {
          defaultS3Bucket = configuration.getS3ConfigurationOptional().get().getGroupOverrides().get(task.getRequest().getGroup().get()).getS3Bucket();
          LOG.trace("Setting defaultS3Bucket to {} for task {} executorData", defaultS3Bucket, taskId.getId());
        } else {
          defaultS3Bucket = configuration.getS3ConfigurationOptional().get().getS3Bucket();
        }
        s3UploaderKeyPattern = configuration.getS3ConfigurationOptional().get().getS3KeyFormat();
      }

      if (task.getPendingTask().getCmdLineArgsList().isPresent() && !task.getPendingTask().getCmdLineArgsList().get().isEmpty()) {
        LOG.trace("Adding cmd line args {} to task {} executorData", task.getPendingTask().getCmdLineArgsList(), taskId.getId());

        final ImmutableList.Builder<String> extraCmdLineArgsBuilder = ImmutableList.builder();
        if (executorDataBldr.getExtraCmdLineArgs() != null && !executorDataBldr.getExtraCmdLineArgs().isEmpty()) {
          extraCmdLineArgsBuilder.addAll(executorDataBldr.getExtraCmdLineArgs());
        }
        extraCmdLineArgsBuilder.addAll(task.getPendingTask().getCmdLineArgsList().get());
        executorDataBldr.setExtraCmdLineArgs(extraCmdLineArgsBuilder.build());
      }

      List<SingularityS3UploaderFile> uploaderAdditionalFiles = new ArrayList<>();
      if (configuration.getS3ConfigurationOptional().isPresent()) {
        uploaderAdditionalFiles.addAll(configuration.getS3ConfigurationOptional().get().getS3UploaderAdditionalFiles());
      }
      uploaderAdditionalFiles.addAll(task.getPendingTask().getS3UploaderAdditionalFiles());
      uploaderAdditionalFiles.addAll(task.getDeploy().getS3UploaderAdditionalFiles());

      Optional<String> maybeS3StorageClass = configuration.getS3ConfigurationOptional().isPresent() ? configuration.getS3ConfigurationOptional().get().getS3StorageClass() : Optional.<String>absent();
      Optional<Long> maybeApplyAfterBytes = configuration.getS3ConfigurationOptional().isPresent() ? configuration.getS3ConfigurationOptional().get().getApplyS3StorageClassAfterBytes() : Optional.<Long>absent();

      if (task.getPendingTask().getRunAsUserOverride().isPresent()) {
        executorDataBldr.setUser(task.getPendingTask().getRunAsUserOverride());
      }

      Optional<HealthcheckOptions> healthcheckOptions = task.getRequest().getSkipHealthchecks().or(false) ? Optional.absent() : task.getDeploy().getHealthcheck();
      final SingularityTaskExecutorData executorData = new SingularityTaskExecutorData(executorDataBldr.build(), uploaderAdditionalFiles, defaultS3Bucket, s3UploaderKeyPattern,
          configuration.getCustomExecutorConfiguration().getServiceLog(), configuration.getCustomExecutorConfiguration().getServiceFinishedTailLog(), task.getRequest().getGroup(),
          maybeS3StorageClass, maybeApplyAfterBytes, getCpuHardLimit(task), healthcheckOptions);

      try {
        bldr.setData(ByteString.copyFromUtf8(objectMapper.writeValueAsString(executorData)));
      } catch (JsonProcessingException e) {
        LOG.warn("Unable to process executor data {} for task {} as json (trying as string)", executorData, taskId.getId(), e);

        bldr.setData(ByteString.copyFromUtf8(executorData.toString()));
      }
    } else if (task.getDeploy().getCommand().isPresent()) {
      bldr.setData(ByteString.copyFromUtf8(task.getDeploy().getCommand().get()));
    }
  }
private void prepareCustomExecutor ( final TaskInfo . Builder bldr , final SingularityTaskId taskId , final SingularityTaskRequest task , final SingularityOfferHolder offerHolder , final Optional < long [ ] > ports , final Resources desiredExecutorResources ) { CommandInfo . Builder commandBuilder = CommandInfo . newBuilder ( ) . setValue ( task . getDeploy ( ) . getCustomExecutorCmd ( ) . get ( ) ) ; prepareEnvironment ( task , taskId , commandBuilder , offerHolder , ports ) ; if ( task . getDeploy ( ) . getUser ( ) . isPresent ( ) ) { commandBuilder . setUser ( task . getDeploy ( ) . getUser ( ) . get ( ) ) ; } prepareMesosUriDownloads ( task . getPendingTask ( ) . getExtraArtifacts ( ) , commandBuilder ) ; bldr . setExecutor ( ExecutorInfo . newBuilder ( ) . setCommand ( commandBuilder . build ( ) ) . setExecutorId ( ExecutorID . newBuilder ( ) . setValue ( task . getDeploy ( ) . getCustomExecutorId ( ) . or ( idGenerator . getNextExecutorId ( ) ) ) ) . setSource ( task . getDeploy ( ) . getCustomExecutorSource ( ) . or ( taskId . getId ( ) ) ) . setLabels ( Labels . newBuilder ( ) . addLabels ( Label . newBuilder ( ) . setKey ( "taskId" ) . setValue ( taskId . getId ( ) ) ) ) . addAllResources ( buildMesosResources ( desiredExecutorResources , task . getRequest ( ) . getRequiredRole ( ) ) ) . build ( ) ) ; if ( task . getDeploy ( ) . getExecutorData ( ) . isPresent ( ) ) { final ExecutorDataBuilder executorDataBldr = task . getDeploy ( ) . getExecutorData ( ) . get ( ) . toBuilder ( ) ; String defaultS3Bucket = "" ; String s3UploaderKeyPattern = "" ; if ( configuration . getS3ConfigurationOptional ( ) . isPresent ( ) ) { if ( task . getRequest ( ) . getGroup ( ) . isPresent ( ) && configuration . getS3ConfigurationOptional ( ) . get ( ) . getGroupOverrides ( ) . containsKey ( task . getRequest ( ) . getGroup ( ) . get ( ) ) ) { defaultS3Bucket = configuration . getS3ConfigurationOptional ( ) . get ( ) . getGroupOverrides ( ) . get ( task . getRequest ( ) . getGroup ( ) . get ( ) ) . getS3Bucket ( ) ; LOG . trace ( "Setting defaultS3Bucket to {} for task {} executorData" , defaultS3Bucket , taskId . getId ( ) ) ; } else { defaultS3Bucket = configuration . getS3ConfigurationOptional ( ) . get ( ) . getS3Bucket ( ) ; } s3UploaderKeyPattern = configuration . getS3ConfigurationOptional ( ) . get ( ) . getS3KeyFormat ( ) ; } if ( task . getPendingTask ( ) . getCmdLineArgsList ( ) . isPresent ( ) && ! task . getPendingTask ( ) . getCmdLineArgsList ( ) . get ( ) . isEmpty ( ) ) { LOG . trace ( "Adding cmd line args {} to task {} executorData" , task . getPendingTask ( ) . getCmdLineArgsList ( ) , taskId . getId ( ) ) ; final ImmutableList . Builder < String > extraCmdLineArgsBuilder = ImmutableList . builder ( ) ; if ( executorDataBldr . getExtraCmdLineArgs ( ) != null && ! executorDataBldr . getExtraCmdLineArgs ( ) . isEmpty ( ) ) { extraCmdLineArgsBuilder . addAll ( executorDataBldr . getExtraCmdLineArgs ( ) ) ; } extraCmdLineArgsBuilder . addAll ( task . getPendingTask ( ) . getCmdLineArgsList ( ) . get ( ) ) ; executorDataBldr . setExtraCmdLineArgs ( extraCmdLineArgsBuilder . build ( ) ) ; } List < SingularityS3UploaderFile > uploaderAdditionalFiles = new ArrayList < > ( ) ; if ( configuration . getS3ConfigurationOptional ( ) . isPresent ( ) ) { uploaderAdditionalFiles . addAll ( configuration . getS3ConfigurationOptional ( ) . get ( ) . getS3UploaderAdditionalFiles ( ) ) ; } uploaderAdditionalFiles . addAll ( task . getPendingTask ( ) . getS3UploaderAdditionalFiles ( ) ) ; uploaderAdditionalFiles . addAll ( task . getDeploy ( ) . getS3UploaderAdditionalFiles ( ) ) ; Optional < String > maybeS3StorageClass = configuration . getS3ConfigurationOptional ( ) . isPresent ( ) ? configuration . getS3ConfigurationOptional ( ) . get ( ) . getS3StorageClass ( ) : Optional . < String > absent ( ) ; Optional < Long > maybeApplyAfterBytes = configuration . getS3ConfigurationOptional ( ) . isPresent ( ) ? configuration . getS3ConfigurationOptional ( ) . get ( ) . getApplyS3StorageClassAfterBytes ( ) : Optional . < Long > absent ( ) ; if ( task . getPendingTask ( ) . getRunAsUserOverride ( ) . isPresent ( ) ) { executorDataBldr . setUser ( task . getPendingTask ( ) . getRunAsUserOverride ( ) ) ; } Optional < HealthcheckOptions > healthcheckOptions = task . getRequest ( ) . getSkipHealthchecks ( ) . or ( false ) ? Optional . absent ( ) : task . getDeploy ( ) . getHealthcheck ( ) ; final SingularityTaskExecutorData executorData = new SingularityTaskExecutorData ( executorDataBldr . build ( ) , uploaderAdditionalFiles , defaultS3Bucket , s3UploaderKeyPattern , configuration . getCustomExecutorConfiguration ( ) . getServiceLog ( ) , configuration . getCustomExecutorConfiguration ( ) . getServiceFinishedTailLog ( ) , task . getRequest ( ) . getGroup ( ) , maybeS3StorageClass , maybeApplyAfterBytes , getCpuHardLimit ( task ) , healthcheckOptions ) ; try { bldr . setData ( ByteString . copyFromUtf8 ( objectMapper . writeValueAsString ( executorData ) ) ) ; } catch ( JsonProcessingException e ) { LOG . warn ( "Unable to process executor data {} for task {} as json (trying as string)" , executorData , taskId . getId ( ) , e ) ; bldr . setData ( ByteString . copyFromUtf8 ( executorData . toString ( ) ) ) ; } } else if ( task . getDeploy ( ) . getCommand ( ) . isPresent ( ) ) { bldr . setData ( ByteString . copyFromUtf8 ( task . getDeploy ( ) . getCommand ( ) . get ( ) ) ) ; } }
Prepares the Mesos TaskInfo object when using our custom SingularityExecutor.