{"action":"created","comment":{"url":"https://api.github.com/repos/treasure-data/digdag/pulls/comments/91655854","pull_request_review_id":12166366,"id":91655854,"diff_hunk":"@@ -0,0 +1,1753 @@\n+package io.digdag.standards.operator.aws;\n+\n+import com.amazonaws.AmazonServiceException;\n+import com.amazonaws.ClientConfiguration;\n+import com.amazonaws.auth.AWSCredentials;\n+import com.amazonaws.auth.BasicAWSCredentials;\n+import com.amazonaws.auth.BasicSessionCredentials;\n+import com.amazonaws.services.elasticmapreduce.AmazonElasticMapReduce;\n+import com.amazonaws.services.elasticmapreduce.AmazonElasticMapReduceClient;\n+import com.amazonaws.services.elasticmapreduce.model.AddJobFlowStepsRequest;\n+import com.amazonaws.services.elasticmapreduce.model.AddJobFlowStepsResult;\n+import com.amazonaws.services.elasticmapreduce.model.Application;\n+import com.amazonaws.services.elasticmapreduce.model.BootstrapActionConfig;\n+import com.amazonaws.services.elasticmapreduce.model.ClusterStatus;\n+import com.amazonaws.services.elasticmapreduce.model.Configuration;\n+import com.amazonaws.services.elasticmapreduce.model.DescribeClusterRequest;\n+import com.amazonaws.services.elasticmapreduce.model.DescribeClusterResult;\n+import com.amazonaws.services.elasticmapreduce.model.DescribeStepRequest;\n+import com.amazonaws.services.elasticmapreduce.model.EbsBlockDeviceConfig;\n+import com.amazonaws.services.elasticmapreduce.model.EbsConfiguration;\n+import com.amazonaws.services.elasticmapreduce.model.FailureDetails;\n+import com.amazonaws.services.elasticmapreduce.model.InstanceGroupConfig;\n+import com.amazonaws.services.elasticmapreduce.model.JobFlowInstancesConfig;\n+import com.amazonaws.services.elasticmapreduce.model.ListStepsRequest;\n+import com.amazonaws.services.elasticmapreduce.model.ListStepsResult;\n+import com.amazonaws.services.elasticmapreduce.model.PlacementType;\n+import com.amazonaws.services.elasticmapreduce.model.RunJobFlowRequest;\n+import com.amazonaws.services.elasticmapreduce.model.RunJobFlowResult;\n+import com.amazonaws.services.elasticmapreduce.model.ScriptBootstrapActionConfig;\n+import com.amazonaws.services.elasticmapreduce.model.Step;\n+import com.amazonaws.services.elasticmapreduce.model.StepConfig;\n+import com.amazonaws.services.elasticmapreduce.model.StepStateChangeReason;\n+import com.amazonaws.services.elasticmapreduce.model.StepStatus;\n+import com.amazonaws.services.elasticmapreduce.model.StepSummary;\n+import com.amazonaws.services.elasticmapreduce.model.Tag;\n+import com.amazonaws.services.elasticmapreduce.model.VolumeSpecification;\n+import com.amazonaws.services.elasticmapreduce.util.StepFactory;\n+import com.amazonaws.services.kms.AWSKMSClient;\n+import com.amazonaws.services.kms.model.EncryptRequest;\n+import com.amazonaws.services.kms.model.EncryptResult;\n+import com.amazonaws.services.s3.AmazonS3Client;\n+import com.amazonaws.services.s3.AmazonS3URI;\n+import com.amazonaws.services.s3.model.DeleteObjectsRequest;\n+import com.amazonaws.services.s3.model.ListObjectsRequest;\n+import com.amazonaws.services.s3.model.ObjectListing;\n+import com.amazonaws.services.s3.model.ObjectMetadata;\n+import com.amazonaws.services.s3.model.PutObjectRequest;\n+import com.amazonaws.services.s3.model.S3ObjectSummary;\n+import com.amazonaws.services.s3.transfer.TransferManager;\n+import com.amazonaws.services.s3.transfer.Upload;\n+import com.amazonaws.services.securitytoken.AWSSecurityTokenServiceClient;\n+import com.amazonaws.services.securitytoken.model.AssumeRoleRequest;\n+import com.amazonaws.services.securitytoken.model.AssumeRoleResult;\n+import com.fasterxml.jackson.annotation.JsonProperty;\n+import com.fasterxml.jackson.core.type.TypeReference;\n+import com.fasterxml.jackson.databind.JsonNode;\n+import com.fasterxml.jackson.databind.ObjectMapper;\n+import com.fasterxml.jackson.databind.annotation.JsonDeserialize;\n+import com.fasterxml.jackson.databind.annotation.JsonSerialize;\n+import com.google.api.client.repackaged.com.google.common.base.Splitter;\n+import com.google.api.client.repackaged.com.google.common.base.Throwables;\n+import com.google.common.base.Optional;\n+import com.google.common.base.Preconditions;\n+import com.google.common.collect.ImmutableList;\n+import com.google.common.collect.Iterables;\n+import com.google.common.collect.Lists;\n+import com.google.common.io.BaseEncoding;\n+import com.google.common.io.Resources;\n+import com.google.inject.Inject;\n+import io.digdag.client.config.Config;\n+import io.digdag.client.config.ConfigElement;\n+import io.digdag.client.config.ConfigException;\n+import io.digdag.client.config.ConfigFactory;\n+import io.digdag.client.config.ConfigKey;\n+import io.digdag.core.Environment;\n+import io.digdag.spi.ImmutableTaskResult;\n+import io.digdag.spi.Operator;\n+import io.digdag.spi.OperatorContext;\n+import io.digdag.spi.OperatorFactory;\n+import io.digdag.spi.SecretAccessList;\n+import io.digdag.spi.SecretProvider;\n+import io.digdag.spi.TaskExecutionException;\n+import io.digdag.spi.TaskResult;\n+import io.digdag.spi.TemplateEngine;\n+import io.digdag.spi.TemplateException;\n+import io.digdag.standards.operator.DurationInterval;\n+import io.digdag.standards.operator.state.TaskState;\n+import io.digdag.util.BaseOperator;\n+import io.digdag.util.ConfigSelector;\n+import io.digdag.util.RetryExecutor;\n+import io.digdag.util.Workspace;\n+import org.immutables.value.Value;\n+import org.slf4j.Logger;\n+import org.slf4j.LoggerFactory;\n+\n+import java.io.ByteArrayInputStream;\n+import java.io.IOException;\n+import java.net.URL;\n+import java.nio.ByteBuffer;\n+import java.nio.file.Paths;\n+import java.time.Duration;\n+import java.util.ArrayList;\n+import java.util.Base64;\n+import java.util.Collection;\n+import java.util.Collections;\n+import java.util.HashSet;\n+import java.util.List;\n+import java.util.Map;\n+import java.util.Set;\n+import java.util.concurrent.ThreadLocalRandom;\n+import java.util.function.BiFunction;\n+import java.util.function.Function;\n+import java.util.stream.Collectors;\n+import java.util.stream.Stream;\n+\n+import static com.google.common.io.Closeables.closeQuietly;\n+import static io.digdag.standards.operator.aws.EmrOperatorFactory.FileReference.Type.DIRECT;\n+import static io.digdag.standards.operator.aws.EmrOperatorFactory.FileReference.Type.LOCAL;\n+import static io.digdag.standards.operator.aws.EmrOperatorFactory.FileReference.Type.S3;\n+import static io.digdag.standards.operator.state.PollingRetryExecutor.pollingRetryExecutor;\n+import static io.digdag.standards.operator.state.PollingWaiter.pollingWaiter;\n+import static java.nio.charset.StandardCharsets.UTF_8;\n+import static java.util.Arrays.asList;\n+import static java.util.stream.Collectors.toList;\n+import static java.util.stream.Collectors.toMap;\n+\n+public class EmrOperatorFactory\n+        implements OperatorFactory\n+{\n+    private static final int LIST_STEPS_MAX_IDS = 10;\n+\n+    private static final String LOCAL_STAGING_DIR = \"/home/hadoop/digdag-staging\";\n+\n+    private static Logger logger = LoggerFactory.getLogger(EmrOperatorFactory.class);\n+\n+    public String getType()\n+    {\n+        return \"emr\";\n+    }\n+\n+    private final TemplateEngine templateEngine;\n+    private final ObjectMapper objectMapper;\n+    private final ConfigFactory cf;\n+    private final Map<String, String> environment;\n+\n+    @Inject\n+    public EmrOperatorFactory(TemplateEngine templateEngine, ObjectMapper objectMapper, ConfigFactory cf, @Environment Map<String, String> environment)\n+    {\n+        this.templateEngine = templateEngine;\n+        this.objectMapper = objectMapper;\n+        this.cf = cf;\n+        this.environment = environment;\n+    }\n+\n+    @Override\n+    public SecretAccessList getSecretAccessList()\n+    {\n+        return ConfigSelector.builderOfScope(\"aws\")\n+                .addSecretOnlyAccess(\"emr.endpoint\", \"s3.endpoint\", \"kms.endpoint\")\n+                .addSecretOnlyAccess(\"access_key_id\", \"secret_access_key\", \"role_arn\", \"role_session_name\")\n+                .addSecretOnlyAccess(\"emr.access_key_id\", \"emr.secret_access_key\", \"emr.role_arn\", \"emr.role_session_name\")\n+                .build();\n+    }\n+\n+    @Override\n+    public Operator newOperator(OperatorContext context)\n+    {\n+        return new EmrOperator(context);\n+    }\n+\n+    private class EmrOperator\n+            extends BaseOperator\n+    {\n+        private final TaskState state;\n+        private final Config params;\n+        private final String defaultActionOnFailure;\n+\n+        public EmrOperator(OperatorContext context)\n+        {\n+            super(context);\n+            this.state = TaskState.of(request);\n+            this.params = request.getConfig()\n+                    .mergeDefault(request.getConfig().getNestedOrGetEmpty(\"aws\").getNestedOrGetEmpty(\"emr\"))\n+                    .mergeDefault(request.getConfig().getNestedOrGetEmpty(\"aws\"));\n+            this.defaultActionOnFailure = params.get(\"action_on_failure\", String.class, \"CANCEL_AND_WAIT\");\n+        }\n+\n+        @Override\n+        public TaskResult runTask()\n+        {\n+            String tag = state.constant(\"tag\", String.class, EmrOperatorFactory::randomTag);\n+\n+            AWSCredentials credentials = credentials(tag);\n+\n+            SecretProvider awsSecrets = context.getSecrets().getSecrets(\"aws\");\n+\n+            Optional<String> emrEndpoint = awsSecrets.getSecretOptional(\"emr.endpoint\");\n+            Optional<String> s3Endpoint = awsSecrets.getSecretOptional(\"s3.endpoint\");\n+            Optional<String> kmsEndpoint = awsSecrets.getSecretOptional(\"kms.endpoint\");\n+\n+            ClientConfiguration emrClientConfiguration = new ClientConfiguration();\n+            ClientConfiguration s3ClientConfiguration = new ClientConfiguration();\n+            ClientConfiguration kmsClientConfiguration = new ClientConfiguration();\n+\n+            Aws.configureProxy(emrClientConfiguration, emrEndpoint, environment);\n+            Aws.configureProxy(s3ClientConfiguration, s3Endpoint, environment);\n+            Aws.configureProxy(kmsClientConfiguration, kmsEndpoint, environment);\n+\n+            AmazonElasticMapReduce emr = new AmazonElasticMapReduceClient(credentials, emrClientConfiguration);\n+            AmazonS3Client s3 = new AmazonS3Client(credentials, s3ClientConfiguration);\n+            AWSKMSClient kms = new AWSKMSClient(credentials, kmsClientConfiguration);\n+\n+            // Set up file stager\n+            Optional<AmazonS3URI> staging = params.getOptional(\"staging\", String.class).transform(s -> {\n+                try {\n+                    return new AmazonS3URI(s);\n+                }\n+                catch (IllegalArgumentException ex) {\n+                    throw new ConfigException(\"Invalid staging uri: '\" + s + \"'\", ex);\n+                }\n+            });\n+            Filer filer = new Filer(s3, staging, workspace, templateEngine, params);\n+\n+            // TODO: make it possible for operators to _reliably_ clean up\n+            boolean cleanup = false;\n+\n+            try {\n+                TaskResult result = run(tag, emr, s3, kms, filer);\n+                cleanup = true;\n+                return result;\n+            }\n+            catch (Throwable t) {\n+                boolean retry = t instanceof TaskExecutionException &&\n+                        ((TaskExecutionException) t).getRetryInterval().isPresent();\n+                cleanup = !retry;\n+                throw Throwables.propagate(t);\n+            }\n+            finally {\n+                if (cleanup) {\n+                    // Best effort clean up of staging\n+                    try {\n+                        filer.tryCleanup();\n+                    }\n+                    catch (Throwable t) {\n+                        logger.warn(\"Failed to clean up staging: {}\", staging, t);\n+                    }\n+                }\n+                s3.shutdown();\n+                emr.shutdown();\n+            }\n+        }\n+\n+        private AWSCredentials credentials(String tag)\n+        {\n+            SecretProvider awsSecrets = context.getSecrets().getSecrets(\"aws\");\n+            SecretProvider emrSecrets = awsSecrets.getSecrets(\"emr\");\n+\n+            String accessKeyId = emrSecrets.getSecretOptional(\"access_key_id\")\n+                    .or(() -> awsSecrets.getSecret(\"access_key_id\"));\n+\n+            String secretAccessKey = emrSecrets.getSecretOptional(\"secret_access_key\")\n+                    .or(() -> awsSecrets.getSecret(\"secret_access_key\"));\n+\n+            AWSCredentials credentials = new BasicAWSCredentials(accessKeyId, secretAccessKey);\n+\n+            Optional<String> roleArn = emrSecrets.getSecretOptional(\"role_arn\")\n+                    .or(awsSecrets.getSecretOptional(\"role_arn\"));\n+\n+            if (!roleArn.isPresent()) {\n+                return credentials;\n+            }\n+\n+            // use STS to assume role\n+            String roleSessionName = emrSecrets.getSecretOptional(\"role_session_name\")\n+                    .or(awsSecrets.getSecretOptional(\"role_session_name\"))\n+                    .or(\"digdag-emr-\" + tag);\n+\n+            AWSSecurityTokenServiceClient stsClient = new AWSSecurityTokenServiceClient(credentials);\n+            AssumeRoleResult assumeResult = stsClient.assumeRole(new AssumeRoleRequest()\n+                    .withRoleArn(roleArn.get())\n+                    .withDurationSeconds(3600)\n+                    .withRoleSessionName(roleSessionName));\n+\n+            return new BasicSessionCredentials(\n+                    assumeResult.getCredentials().getAccessKeyId(),\n+                    assumeResult.getCredentials().getSecretAccessKey(),\n+                    assumeResult.getCredentials().getSessionToken());\n+        }\n+\n+        private TaskResult run(String tag, AmazonElasticMapReduce emr, AmazonS3Client s3, AWSKMSClient kms, Filer filer)\n+                throws IOException\n+        {\n+            ParameterCompiler parameterCompiler = new ParameterCompiler(kms, context, cf);\n+\n+            // Set up step compiler\n+            List<Config> steps = params.getListOrEmpty(\"steps\", Config.class);\n+            StepCompiler stepCompiler = new StepCompiler(tag, steps, filer, parameterCompiler, objectMapper, defaultActionOnFailure);\n+\n+            // Set up job submitter\n+            Submitter submitter;\n+            Config cluster = null;\n+            try {\n+                cluster = params.parseNestedOrGetEmpty(\"cluster\");\n+            }\n+            catch (ConfigException ignore) {\n+            }\n+            if (cluster != null) {\n+                // Create a new cluster\n+                submitter = newClusterSubmitter(emr, tag, stepCompiler, cluster, filer, parameterCompiler);\n+            }\n+            else {\n+                // Cluster ID? Use existing cluster.\n+                String clusterId = params.get(\"cluster\", String.class);\n+                submitter = existingClusterSubmitter(emr, tag, stepCompiler, clusterId, filer);\n+            }\n+\n+            // Submit EMR job\n+            SubmissionResult submission = submitter.submit();\n+\n+            // Wait for the steps to finish running\n+            if (!steps.isEmpty()) {\n+                waitForSteps(emr, submission);\n+            }\n+\n+            return result(submission);\n+        }\n+\n+        private void waitForSteps(AmazonElasticMapReduce emr, SubmissionResult submission)\n+        {\n+            String lastStepId = Iterables.getLast(submission.stepIds());\n+            pollingWaiter(state, \"result\")\n+                    .withWaitMessage(\"EMR job still running: %s\", submission.clusterId())\n+                    .withPollInterval(DurationInterval.of(Duration.ofSeconds(15), Duration.ofMinutes(5)))\n+                    .awaitOnce(Step.class, pollState -> checkStepCompletion(emr, submission, lastStepId, pollState));\n+        }\n+\n+        private Optional<Step> checkStepCompletion(AmazonElasticMapReduce emr, SubmissionResult submission, String lastStepId, TaskState pollState)\n+        {\n+            return pollingRetryExecutor(pollState, \"poll\")\n+                    .retryUnless(AmazonServiceException.class, Aws::isDeterministicException)\n+                    .withRetryInterval(DurationInterval.of(Duration.ofSeconds(15), Duration.ofMinutes(5)))\n+                    .run(s -> {\n+\n+                        ListStepsResult runningSteps = emr.listSteps(new ListStepsRequest()\n+                                .withClusterId(submission.clusterId())\n+                                .withStepStates(\"RUNNING\"));\n+\n+                        runningSteps.getSteps().stream().findFirst()\n+                                .ifPresent(step -> {\n+                                    int stepIndex = submission.stepIds().indexOf(step.getId());\n+                                    logger.info(\"Currently running EMR step {}/{}: {}: {}\",\n+                                            stepIndex == -1 ? \"?\" : Integer.toString(stepIndex + 1),\n+                                            submission.stepIds().size(),\n+                                            step.getId(),\n+                                            step.getName());\n+                                });\n+\n+                        Step lastStep = emr.describeStep(new DescribeStepRequest()\n+                                .withClusterId(submission.clusterId())\n+                                .withStepId(lastStepId))\n+                                .getStep();\n+\n+                        String stepState = lastStep.getStatus().getState();\n+\n+                        switch (stepState) {\n+                            case \"PENDING\":\n+                            case \"RUNNING\":\n+                                return Optional.absent();\n+\n+                            case \"CANCEL_PENDING\":\n+                            case \"CANCELLED\":\n+                            case \"FAILED\":\n+                            case \"INTERRUPTED\":\n+                                // TODO: consider task done if action_on_failure == CONTINUE?\n+                                // TODO: include & log failure information\n+\n+                                List<StepSummary> steps = Lists.partition(submission.stepIds(), LIST_STEPS_MAX_IDS).stream()\n+                                        .flatMap(ids -> emr.listSteps(new ListStepsRequest()\n+                                                .withClusterId(submission.clusterId())\n+                                                .withStepIds(ids)).getSteps().stream())\n+                                        .collect(toList());\n+\n+                                logger.error(\"EMR job failed: {}\", submission.clusterId());\n+\n+                                for (StepSummary step : steps) {\n+                                    StepStatus status = step.getStatus();\n+                                    FailureDetails details = status.getFailureDetails();\n+                                    StepStateChangeReason reason = status.getStateChangeReason();\n+                                    int stepIndex = submission.stepIds().indexOf(step.getId());\n+                                    logger.error(\"EMR step {}/{}: {}: state: {}, reason: {}, details: {}\",\n+                                            stepIndex == -1 ? \"?\" : Integer.toString(stepIndex + 1),\n+                                            submission.stepIds().size(),\n+                                            step.getId(), status.getState(),\n+                                            reason != null ? reason : \"{}\",\n+                                            details != null ? details : \"{}\");\n+                                }\n+\n+                                throw new TaskExecutionException(\"EMR job failed\", ConfigElement.empty());\n+\n+                            case \"COMPLETED\":\n+                                logger.info(\"EMR steps done\");\n+                                return Optional.of(lastStep);\n+\n+                            default:\n+                                throw new RuntimeException(\"Unknown step status: \" + lastStep);\n+                        }\n+                    });\n+        }\n+\n+        private TaskResult result(SubmissionResult submission)\n+        {\n+            ImmutableTaskResult.Builder result = TaskResult.defaultBuilder(request);\n+            if (submission.newCluster()) {\n+                Config storeParams = request.getConfig().getFactory().create();\n+                storeParams.getNestedOrSetEmpty(\"emr\").set(\"last_cluster_id\", submission.clusterId());\n+                result.storeParams(storeParams);\n+                result.addResetStoreParams(ConfigKey.of(\"emr\", \"last_cluster_id\"));\n+            }\n+            return result.build();\n+        }\n+\n+        private Submitter existingClusterSubmitter(AmazonElasticMapReduce emr, String tag, StepCompiler stepCompiler, String clusterId, Filer filer)\n+        {\n+            return () -> {\n+                List<String> stepIds = pollingRetryExecutor(state, \"submission\")\n+                        .retryUnless(AmazonServiceException.class, Aws::isDeterministicException)\n+                        .withRetryInterval(DurationInterval.of(Duration.ofSeconds(30), Duration.ofMinutes(5)))\n+                        .runOnce(new TypeReference<List<String>>() {}, s -> {\n+\n+                            RemoteFile runner = prepareRunner(filer, tag);\n+\n+                            // Compile steps\n+                            stepCompiler.compile(runner);\n+\n+                            // Stage files to S3\n+                            filer.stageFiles();\n+\n+                            AddJobFlowStepsRequest request = new AddJobFlowStepsRequest()\n+                                    .withJobFlowId(clusterId)\n+                                    .withSteps(stepCompiler.stepConfigs());\n+\n+                            int steps = request.getSteps().size();\n+                            logger.info(\"Submitting {} EMR step(s) to {}\", steps, clusterId);\n+                            AddJobFlowStepsResult result = emr.addJobFlowSteps(request);\n+                            logSubmittedSteps(clusterId, steps, i -> request.getSteps().get(i).getName(), i -> result.getStepIds().get(i));\n+                            return ImmutableList.copyOf(result.getStepIds());\n+                        });\n+\n+                return SubmissionResult.ofExistingCluster(clusterId, stepIds);\n+            };\n+        }\n+\n+        private void logSubmittedSteps(String clusterId, int n, Function<Integer, String> names, Function<Integer, String> ids)\n+        {\n+            logger.info(\"Submitted {} EMR step(s) to {}\", n, clusterId);\n+            for (int i = 0; i < n; i++) {\n+                logger.info(\"Step {}/{}: {}: {}\", i + 1, n, names.apply(i), ids.apply(i));\n+            }\n+        }\n+\n+        private Submitter newClusterSubmitter(AmazonElasticMapReduce emr, String tag, StepCompiler stepCompiler, Config clusterConfig, Filer filer, ParameterCompiler parameterCompiler)\n+        {\n+\n+            return () -> {\n+                // Start cluster\n+                NewCluster cluster = pollingRetryExecutor(state, \"submission\")\n+                        .withRetryInterval(DurationInterval.of(Duration.ofSeconds(30), Duration.ofMinutes(5)))\n+                        // TODO: EMR requests are not idempotent, thus retrying might produce duplicate cluster submissions.\n+                        .retryUnless(AmazonServiceException.class, Aws::isDeterministicException)\n+                        .runOnce(NewCluster.class, s -> submitNewClusterRequest(emr, tag, stepCompiler, clusterConfig, filer, parameterCompiler));\n+\n+                // Get submitted step IDs\n+                List<String> stepIds = pollingRetryExecutor(this.state, \"steps\")\n+                        .withRetryInterval(DurationInterval.of(Duration.ofSeconds(30), Duration.ofMinutes(5)))\n+                        .retryUnless(AmazonServiceException.class, Aws::isDeterministicException)\n+                        .runOnce(new TypeReference<List<String>>() {}, s -> {\n+                            List<StepSummary> steps = listSubmittedSteps(emr, tag, cluster);\n+                            logSubmittedSteps(cluster.id(), cluster.steps(), i -> steps.get(i).getName(), i -> steps.get(i).getId());\n+                            return steps.stream().map(StepSummary::getId).collect(toList());\n+                        });\n+\n+                // Log cluster status while waiting for it to come up\n+                pollingWaiter(state, \"bootstrap\")\n+                        .withWaitMessage(\"EMR cluster still booting\")\n+                        .withPollInterval(DurationInterval.of(Duration.ofSeconds(30), Duration.ofMinutes(5)))\n+                        .awaitOnce(String.class, pollState -> checkClusterBootStatus(emr, cluster, pollState));\n+\n+                return SubmissionResult.ofNewCluster(cluster.id(), stepIds);\n+            };\n+        }\n+\n+        private Optional<String> checkClusterBootStatus(AmazonElasticMapReduce emr, NewCluster cluster, TaskState state)\n+        {\n+            // Only creating a cluster, with no steps?\n+            boolean createOnly = cluster.steps() == 0;\n+\n+            DescribeClusterResult describeClusterResult = pollingRetryExecutor(state, \"describe-cluster\")\n+                    .withRetryInterval(DurationInterval.of(Duration.ofSeconds(30), Duration.ofMinutes(5)))\n+                    .retryUnless(AmazonServiceException.class, Aws::isDeterministicException)\n+                    .run(ds -> emr.describeCluster(new DescribeClusterRequest().withClusterId(cluster.id())));\n+\n+            ClusterStatus clusterStatus = describeClusterResult.getCluster().getStatus();\n+            String clusterState = clusterStatus.getState();\n+\n+            switch (clusterState) {\n+                case \"STARTING\":\n+                    logger.info(\"EMR cluster starting: {}\", cluster.id());\n+                    return Optional.absent();\n+                case \"BOOTSTRAPPING\":\n+                    logger.info(\"EMR cluster bootstrapping: {}\", cluster.id());\n+                    return Optional.absent();\n+\n+                case \"RUNNING\":\n+                case \"WAITING\":\n+                    logger.info(\"EMR cluster up: {}\", cluster.id());\n+                    return Optional.of(clusterState);\n+\n+                case \"TERMINATED_WITH_ERRORS\":\n+                    if (createOnly) {\n+                        // TODO: log more information about the errors","path":"digdag-standards/src/main/java/io/digdag/standards/operator/aws/EmrOperatorFactory.java","position":520,"original_position":520,"commit_id":"a55ee257b9e74804ac10cfda6e4a8418ac708d33","original_commit_id":"a55ee257b9e74804ac10cfda6e4a8418ac708d33","user":{"login":"komamitsu","id":59043,"avatar_url":"https://avatars.githubusercontent.com/u/59043?v=3","gravatar_id":"","url":"https://api.github.com/users/komamitsu","html_url":"https://github.com/komamitsu","followers_url":"https://api.github.com/users/komamitsu/followers","following_url":"https://api.github.com/users/komamitsu/following{/other_user}","gists_url":"https://api.github.com/users/komamitsu/gists{/gist_id}","starred_url":"https://api.github.com/users/komamitsu/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/komamitsu/subscriptions","organizations_url":"https://api.github.com/users/komamitsu/orgs","repos_url":"https://api.github.com/users/komamitsu/repos","events_url":"https://api.github.com/users/komamitsu/events{/privacy}","received_events_url":"https://api.github.com/users/komamitsu/received_events","type":"User","site_admin":false},"body":"👍 ","created_at":"2016-12-09T03:54:22Z","updated_at":"2016-12-09T03:54:22Z","html_url":"https://github.com/treasure-data/digdag/pull/397#discussion_r91655854","pull_request_url":"https://api.github.com/repos/treasure-data/digdag/pulls/397","_links":{"self":{"href":"https://api.github.com/repos/treasure-data/digdag/pulls/comments/91655854"},"html":{"href":"https://github.com/treasure-data/digdag/pull/397#discussion_r91655854"},"pull_request":{"href":"https://api.github.com/repos/treasure-data/digdag/pulls/397"}}},"pull_request":{"url":"https://api.github.com/repos/treasure-data/digdag/pulls/397","id":95241051,"html_url":"https://github.com/treasure-data/digdag/pull/397","diff_url":"https://github.com/treasure-data/digdag/pull/397.diff","patch_url":"https://github.com/treasure-data/digdag/pull/397.patch","issue_url":"https://api.github.com/repos/treasure-data/digdag/issues/397","number":397,"state":"open","locked":false,"title":"emr operator","user":{"login":"danielnorberg","id":393289,"avatar_url":"https://avatars.githubusercontent.com/u/393289?v=3","gravatar_id":"","url":"https://api.github.com/users/danielnorberg","html_url":"https://github.com/danielnorberg","followers_url":"https://api.github.com/users/danielnorberg/followers","following_url":"https://api.github.com/users/danielnorberg/following{/other_user}","gists_url":"https://api.github.com/users/danielnorberg/gists{/gist_id}","starred_url":"https://api.github.com/users/danielnorberg/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/danielnorberg/subscriptions","organizations_url":"https://api.github.com/users/danielnorberg/orgs","repos_url":"https://api.github.com/users/danielnorberg/repos","events_url":"https://api.github.com/users/danielnorberg/events{/privacy}","received_events_url":"https://api.github.com/users/danielnorberg/received_events","type":"User","site_admin":false},"body":"","created_at":"2016-11-25T01:56:11Z","updated_at":"2016-12-09T03:54:22Z","closed_at":null,"merged_at":null,"merge_commit_sha":"f9002f70ab2e39e7453de6f0951800792207bbde","assignee":null,"assignees":[],"milestone":null,"commits_url":"https://api.github.com/repos/treasure-data/digdag/pulls/397/commits","review_comments_url":"https://api.github.com/repos/treasure-data/digdag/pulls/397/comments","review_comment_url":"https://api.github.com/repos/treasure-data/digdag/pulls/comments{/number}","comments_url":"https://api.github.com/repos/treasure-data/digdag/issues/397/comments","statuses_url":"https://api.github.com/repos/treasure-data/digdag/statuses/a55ee257b9e74804ac10cfda6e4a8418ac708d33","head":{"label":"treasure-data:dano/emr","ref":"dano/emr","sha":"a55ee257b9e74804ac10cfda6e4a8418ac708d33","user":{"login":"treasure-data","id":747746,"avatar_url":"https://avatars.githubusercontent.com/u/747746?v=3","gravatar_id":"","url":"https://api.github.com/users/treasure-data","html_url":"https://github.com/treasure-data","followers_url":"https://api.github.com/users/treasure-data/followers","following_url":"https://api.github.com/users/treasure-data/following{/other_user}","gists_url":"https://api.github.com/users/treasure-data/gists{/gist_id}","starred_url":"https://api.github.com/users/treasure-data/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/treasure-data/subscriptions","organizations_url":"https://api.github.com/users/treasure-data/orgs","repos_url":"https://api.github.com/users/treasure-data/repos","events_url":"https://api.github.com/users/treasure-data/events{/privacy}","received_events_url":"https://api.github.com/users/treasure-data/received_events","type":"Organization","site_admin":false},"repo":{"id":42751387,"name":"digdag","full_name":"treasure-data/digdag","owner":{"login":"treasure-data","id":747746,"avatar_url":"https://avatars.githubusercontent.com/u/747746?v=3","gravatar_id":"","url":"https://api.github.com/users/treasure-data","html_url":"https://github.com/treasure-data","followers_url":"https://api.github.com/users/treasure-data/followers","following_url":"https://api.github.com/users/treasure-data/following{/other_user}","gists_url":"https://api.github.com/users/treasure-data/gists{/gist_id}","starred_url":"https://api.github.com/users/treasure-data/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/treasure-data/subscriptions","organizations_url":"https://api.github.com/users/treasure-data/orgs","repos_url":"https://api.github.com/users/treasure-data/repos","events_url":"https://api.github.com/users/treasure-data/events{/privacy}","received_events_url":"https://api.github.com/users/treasure-data/received_events","type":"Organization","site_admin":false},"private":false,"html_url":"https://github.com/treasure-data/digdag","description":"Workload Automation System","fork":false,"url":"https://api.github.com/repos/treasure-data/digdag","forks_url":"https://api.github.com/repos/treasure-data/digdag/forks","keys_url":"https://api.github.com/repos/treasure-data/digdag/keys{/key_id}","collaborators_url":"https://api.github.com/repos/treasure-data/digdag/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/treasure-data/digdag/teams","hooks_url":"https://api.github.com/repos/treasure-data/digdag/hooks","issue_events_url":"https://api.github.com/repos/treasure-data/digdag/issues/events{/number}","events_url":"https://api.github.com/repos/treasure-data/digdag/events","assignees_url":"https://api.github.com/repos/treasure-data/digdag/assignees{/user}","branches_url":"https://api.github.com/repos/treasure-data/digdag/branches{/branch}","tags_url":"https://api.github.com/repos/treasure-data/digdag/tags","blobs_url":"https://api.github.com/repos/treasure-data/digdag/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/treasure-data/digdag/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/treasure-data/digdag/git/refs{/sha}","trees_url":"https://api.github.com/repos/treasure-data/digdag/git/trees{/sha}","statuses_url":"https://api.github.com/repos/treasure-data/digdag/statuses/{sha}","languages_url":"https://api.github.com/repos/treasure-data/digdag/languages","stargazers_url":"https://api.github.com/repos/treasure-data/digdag/stargazers","contributors_url":"https://api.github.com/repos/treasure-data/digdag/contributors","subscribers_url":"https://api.github.com/repos/treasure-data/digdag/subscribers","subscription_url":"https://api.github.com/repos/treasure-data/digdag/subscription","commits_url":"https://api.github.com/repos/treasure-data/digdag/commits{/sha}","git_commits_url":"https://api.github.com/repos/treasure-data/digdag/git/commits{/sha}","comments_url":"https://api.github.com/repos/treasure-data/digdag/comments{/number}","issue_comment_url":"https://api.github.com/repos/treasure-data/digdag/issues/comments{/number}","contents_url":"https://api.github.com/repos/treasure-data/digdag/contents/{+path}","compare_url":"https://api.github.com/repos/treasure-data/digdag/compare/{base}...{head}","merges_url":"https://api.github.com/repos/treasure-data/digdag/merges","archive_url":"https://api.github.com/repos/treasure-data/digdag/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/treasure-data/digdag/downloads","issues_url":"https://api.github.com/repos/treasure-data/digdag/issues{/number}","pulls_url":"https://api.github.com/repos/treasure-data/digdag/pulls{/number}","milestones_url":"https://api.github.com/repos/treasure-data/digdag/milestones{/number}","notifications_url":"https://api.github.com/repos/treasure-data/digdag/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/treasure-data/digdag/labels{/name}","releases_url":"https://api.github.com/repos/treasure-data/digdag/releases{/id}","deployments_url":"https://api.github.com/repos/treasure-data/digdag/deployments","created_at":"2015-09-18T23:18:08Z","updated_at":"2016-12-07T16:41:02Z","pushed_at":"2016-12-09T02:29:53Z","git_url":"git://github.com/treasure-data/digdag.git","ssh_url":"git@github.com:treasure-data/digdag.git","clone_url":"https://github.com/treasure-data/digdag.git","svn_url":"https://github.com/treasure-data/digdag","homepage":"http://www.digdag.io/","size":5368,"stargazers_count":248,"watchers_count":248,"language":"Java","has_issues":true,"has_downloads":true,"has_wiki":true,"has_pages":false,"forks_count":34,"mirror_url":null,"open_issues_count":41,"forks":34,"open_issues":41,"watchers":248,"default_branch":"master"}},"base":{"label":"treasure-data:master","ref":"master","sha":"05a905ea1d3ae7432cc0090f2ad03c8413220763","user":{"login":"treasure-data","id":747746,"avatar_url":"https://avatars.githubusercontent.com/u/747746?v=3","gravatar_id":"","url":"https://api.github.com/users/treasure-data","html_url":"https://github.com/treasure-data","followers_url":"https://api.github.com/users/treasure-data/followers","following_url":"https://api.github.com/users/treasure-data/following{/other_user}","gists_url":"https://api.github.com/users/treasure-data/gists{/gist_id}","starred_url":"https://api.github.com/users/treasure-data/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/treasure-data/subscriptions","organizations_url":"https://api.github.com/users/treasure-data/orgs","repos_url":"https://api.github.com/users/treasure-data/repos","events_url":"https://api.github.com/users/treasure-data/events{/privacy}","received_events_url":"https://api.github.com/users/treasure-data/received_events","type":"Organization","site_admin":false},"repo":{"id":42751387,"name":"digdag","full_name":"treasure-data/digdag","owner":{"login":"treasure-data","id":747746,"avatar_url":"https://avatars.githubusercontent.com/u/747746?v=3","gravatar_id":"","url":"https://api.github.com/users/treasure-data","html_url":"https://github.com/treasure-data","followers_url":"https://api.github.com/users/treasure-data/followers","following_url":"https://api.github.com/users/treasure-data/following{/other_user}","gists_url":"https://api.github.com/users/treasure-data/gists{/gist_id}","starred_url":"https://api.github.com/users/treasure-data/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/treasure-data/subscriptions","organizations_url":"https://api.github.com/users/treasure-data/orgs","repos_url":"https://api.github.com/users/treasure-data/repos","events_url":"https://api.github.com/users/treasure-data/events{/privacy}","received_events_url":"https://api.github.com/users/treasure-data/received_events","type":"Organization","site_admin":false},"private":false,"html_url":"https://github.com/treasure-data/digdag","description":"Workload Automation System","fork":false,"url":"https://api.github.com/repos/treasure-data/digdag","forks_url":"https://api.github.com/repos/treasure-data/digdag/forks","keys_url":"https://api.github.com/repos/treasure-data/digdag/keys{/key_id}","collaborators_url":"https://api.github.com/repos/treasure-data/digdag/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/treasure-data/digdag/teams","hooks_url":"https://api.github.com/repos/treasure-data/digdag/hooks","issue_events_url":"https://api.github.com/repos/treasure-data/digdag/issues/events{/number}","events_url":"https://api.github.com/repos/treasure-data/digdag/events","assignees_url":"https://api.github.com/repos/treasure-data/digdag/assignees{/user}","branches_url":"https://api.github.com/repos/treasure-data/digdag/branches{/branch}","tags_url":"https://api.github.com/repos/treasure-data/digdag/tags","blobs_url":"https://api.github.com/repos/treasure-data/digdag/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/treasure-data/digdag/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/treasure-data/digdag/git/refs{/sha}","trees_url":"https://api.github.com/repos/treasure-data/digdag/git/trees{/sha}","statuses_url":"https://api.github.com/repos/treasure-data/digdag/statuses/{sha}","languages_url":"https://api.github.com/repos/treasure-data/digdag/languages","stargazers_url":"https://api.github.com/repos/treasure-data/digdag/stargazers","contributors_url":"https://api.github.com/repos/treasure-data/digdag/contributors","subscribers_url":"https://api.github.com/repos/treasure-data/digdag/subscribers","subscription_url":"https://api.github.com/repos/treasure-data/digdag/subscription","commits_url":"https://api.github.com/repos/treasure-data/digdag/commits{/sha}","git_commits_url":"https://api.github.com/repos/treasure-data/digdag/git/commits{/sha}","comments_url":"https://api.github.com/repos/treasure-data/digdag/comments{/number}","issue_comment_url":"https://api.github.com/repos/treasure-data/digdag/issues/comments{/number}","contents_url":"https://api.github.com/repos/treasure-data/digdag/contents/{+path}","compare_url":"https://api.github.com/repos/treasure-data/digdag/compare/{base}...{head}","merges_url":"https://api.github.com/repos/treasure-data/digdag/merges","archive_url":"https://api.github.com/repos/treasure-data/digdag/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/treasure-data/digdag/downloads","issues_url":"https://api.github.com/repos/treasure-data/digdag/issues{/number}","pulls_url":"https://api.github.com/repos/treasure-data/digdag/pulls{/number}","milestones_url":"https://api.github.com/repos/treasure-data/digdag/milestones{/number}","notifications_url":"https://api.github.com/repos/treasure-data/digdag/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/treasure-data/digdag/labels{/name}","releases_url":"https://api.github.com/repos/treasure-data/digdag/releases{/id}","deployments_url":"https://api.github.com/repos/treasure-data/digdag/deployments","created_at":"2015-09-18T23:18:08Z","updated_at":"2016-12-07T16:41:02Z","pushed_at":"2016-12-09T02:29:53Z","git_url":"git://github.com/treasure-data/digdag.git","ssh_url":"git@github.com:treasure-data/digdag.git","clone_url":"https://github.com/treasure-data/digdag.git","svn_url":"https://github.com/treasure-data/digdag","homepage":"http://www.digdag.io/","size":5368,"stargazers_count":248,"watchers_count":248,"language":"Java","has_issues":true,"has_downloads":true,"has_wiki":true,"has_pages":false,"forks_count":34,"mirror_url":null,"open_issues_count":41,"forks":34,"open_issues":41,"watchers":248,"default_branch":"master"}},"_links":{"self":{"href":"https://api.github.com/repos/treasure-data/digdag/pulls/397"},"html":{"href":"https://github.com/treasure-data/digdag/pull/397"},"issue":{"href":"https://api.github.com/repos/treasure-data/digdag/issues/397"},"comments":{"href":"https://api.github.com/repos/treasure-data/digdag/issues/397/comments"},"review_comments":{"href":"https://api.github.com/repos/treasure-data/digdag/pulls/397/comments"},"review_comment":{"href":"https://api.github.com/repos/treasure-data/digdag/pulls/comments{/number}"},"commits":{"href":"https://api.github.com/repos/treasure-data/digdag/pulls/397/commits"},"statuses":{"href":"https://api.github.com/repos/treasure-data/digdag/statuses/a55ee257b9e74804ac10cfda6e4a8418ac708d33"}}}}
{"actor":{"display_login":"komamitsu"}}