<commit_before># This Source Code Form is subject to the terms of the Mozilla Public
# License, v. 2.0. If a copy of the MPL was not distributed with this
# file, you can obtain one at http://mozilla.org/MPL/2.0/.
from ..provisioners import Provisioner


class ClusterProvisioner(Provisioner):
    log_dir = 'clusters'

    def __init__(self):
        super().__init__()
        # the S3 URI to the zeppelin setup step
        self.zeppelin_uri = (
            's3://%s/steps/zeppelin/zeppelin.sh' %
            self.config['SPARK_EMR_BUCKET']
        )

    def job_flow_params(self, *args, **kwargs):
        params = super().job_flow_params(*args, **kwargs)
        # don't auto-terminate the cluster
        params.setdefault('Instances', {})['KeepJobFlowAliveWhenNoSteps'] = True
        params.setdefault('Applications', []).append({'Name': 'Zeppelin'})
        return params

    def start(self, user_email, identifier, emr_release, size, public_key):
        """
        Given the parameters spawns a cluster with the desired properties and
        returns the jobflow ID.
        """
        job_flow_params = self.job_flow_params(
            user_email=user_email,
            identifier=identifier,
            emr_release=emr_release,
            size=size,
        )

        job_flow_params.update({
            'BootstrapActions': [{
                'Name': 'setup-telemetry-cluster',
                'ScriptBootstrapAction': {
                    'Path': self.script_uri,
                    'Args': ['--public-key', public_key]
                }
            }],
            'Steps': [{
                'Name': 'setup-zeppelin',
                'ActionOnFailure': 'TERMINATE_JOB_FLOW',
                'HadoopJarStep': {
                    'Jar': self.jar_uri,
                    'Args': [
                        self.zeppelin_uri
                    ]
                }
            }]
        })
        cluster = self.emr.run_job_flow(**job_flow_params)
        return cluster['JobFlowId']

    def info(self, jobflow_id):
        """
        Returns the cluster info for the cluster with the given Jobflow ID
        with the fields start time, state and public IP address
        """
        cluster = self.emr.describe_cluster(ClusterId=jobflow_id)['Cluster']
        return self.format_info(cluster)

    def format_info(self, cluster):
        return {
            'start_time': cluster['Status']['Timeline']['CreationDateTime'],
            'state': cluster['Status']['State'],
            'public_dns': cluster.get('MasterPublicDnsName'),
        }

    def list(self, created_after, created_before=None):
        """
        Returns a list of cluster infos in the given time frame with the fields:
        - Jobflow ID
        - state
        - start time
        """
        # set some parameters so we don't get *all* clusters ever
        params = {'CreatedAfter': created_after}
        if created_before is not None:
            params['CreatedBefore'] = created_before

        clusters = []
        list_cluster_paginator = self.emr.get_paginator('list_clusters')
        for page in list_cluster_paginator.paginate(**params):
            for cluster in page.get('Clusters', []):
                clusters.append(self.format_list(cluster))
        return clusters

    def format_list(self, cluster):
        """
        Formats the data returned by the EMR API for internal ATMO use.
        """
        return {
            'jobflow_id': cluster['Id'],
            'state': cluster['Status']['State'],
            'start_time': cluster['Status']['Timeline']['CreationDateTime'],
        }

    def stop(self, jobflow_id):
        """
        Stops the cluster with the given JobFlow ID.
        """
        self.emr.terminate_job_flows(JobFlowIds=[jobflow_id])
<commit_msg>Hotfix for 35023cc6a8fd5b3b19e0d629c83b1d2a010855fb.<commit_after># This Source Code Form is subject to the terms of the Mozilla Public
# License, v. 2.0. If a copy of the MPL was not distributed with this
# file, you can obtain one at http://mozilla.org/MPL/2.0/.
from ..provisioners import Provisioner


class ClusterProvisioner(Provisioner):
    log_dir = 'clusters'

    def __init__(self):
        super().__init__()
        # the S3 URI to the zeppelin setup step
        self.zeppelin_uri = (
            's3://%s/steps/zeppelin/zeppelin.sh' %
            self.config['SPARK_EMR_BUCKET']
        )

    def job_flow_params(self, *args, **kwargs):
        params = super().job_flow_params(*args, **kwargs)
        # don't auto-terminate the cluster
        params.setdefault('Instances', {})['KeepJobFlowAliveWhenNoSteps'] = True
        emr_release = kwargs.get('emr_release')
        if emr_release < '5.0.0':
            zeppelin_application = 'Zeppelin'
        else:
            zeppelin_application = 'Zeppelin-Sandbox'
        params.setdefault('Applications', []).append({'Name': 'Zeppelin'})
        return params

    def start(self, user_email, identifier, emr_release, size, public_key):
        """
        Given the parameters spawns a cluster with the desired properties and
        returns the jobflow ID.
        """
        job_flow_params = self.job_flow_params(
            user_email=user_email,
            identifier=identifier,
            emr_release=emr_release,
            size=size,
        )

        job_flow_params.update({
            'BootstrapActions': [{
                'Name': 'setup-telemetry-cluster',
                'ScriptBootstrapAction': {
                    'Path': self.script_uri,
                    'Args': ['--public-key', public_key]
                }
            }],
            'Steps': [{
                'Name': 'setup-zeppelin',
                'ActionOnFailure': 'TERMINATE_JOB_FLOW',
                'HadoopJarStep': {
                    'Jar': self.jar_uri,
                    'Args': [
                        self.zeppelin_uri
                    ]
                }
            }]
        })
        cluster = self.emr.run_job_flow(**job_flow_params)
        return cluster['JobFlowId']

    def info(self, jobflow_id):
        """
        Returns the cluster info for the cluster with the given Jobflow ID
        with the fields start time, state and public IP address
        """
        cluster = self.emr.describe_cluster(ClusterId=jobflow_id)['Cluster']
        return self.format_info(cluster)

    def format_info(self, cluster):
        return {
            'start_time': cluster['Status']['Timeline']['CreationDateTime'],
            'state': cluster['Status']['State'],
            'public_dns': cluster.get('MasterPublicDnsName'),
        }

    def list(self, created_after, created_before=None):
        """
        Returns a list of cluster infos in the given time frame with the fields:
        - Jobflow ID
        - state
        - start time
        """
        # set some parameters so we don't get *all* clusters ever
        params = {'CreatedAfter': created_after}
        if created_before is not None:
            params['CreatedBefore'] = created_before

        clusters = []
        list_cluster_paginator = self.emr.get_paginator('list_clusters')
        for page in list_cluster_paginator.paginate(**params):
            for cluster in page.get('Clusters', []):
                clusters.append(self.format_list(cluster))
        return clusters

    def format_list(self, cluster):
        """
        Formats the data returned by the EMR API for internal ATMO use.
        """
        return {
            'jobflow_id': cluster['Id'],
            'state': cluster['Status']['State'],
            'start_time': cluster['Status']['Timeline']['CreationDateTime'],
        }

    def stop(self, jobflow_id):
        """
        Stops the cluster with the given JobFlow ID.
        """
        self.emr.terminate_job_flows(JobFlowIds=[jobflow_id])

{'fraction_non_alphanumeric': 0.11009291078019601, 'fraction_numerical': 0.005091001654575537, 'mean_word_length': 2.0015278838808253, 'pattern_counts': {'":': 0, '<': 4, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4912176', 'n_tokens_mistral': 2252, 'n_tokens_neox': 2174, 'n_words': 528}