<commit_before>'''Library of utilities called by the mesos client binary
'''

from __future__ import print_function

import functools
import json
import os
import posixpath
import sys

from pystachio import Ref
from twitter.common import app, log
from twitter.common.contextutil import temporary_dir, open_zip
from twitter.mesos.clusters import Cluster
from twitter.mesos.config.schema import Packer as PackerObject
from twitter.mesos.packer.packer_client import Packer
from twitter.mesos.packer import sd_packer_client
from twitter.mesos.parsers import (
    FormatDetector,
    MesosConfig,
    PystachioCodec,
    PystachioConfig)

from gen.twitter.mesos.ttypes import ResponseCode


_PACKAGE_FILES_SUFFIX = MesosConfig.PACKAGE_FILES_SUFFIX


def die(msg):
  log.fatal(msg)
  sys.exit(1)


def _zip_package_files(job_name, package_files, tmp_dir):
  zipname = os.path.join(tmp_dir, MesosConfig.get_package_files_zip_name(job_name))
  with open_zip(zipname, 'w') as zipf:
    for file_name in package_files:
      zipf.write(file_name, arcname=os.path.basename(file_name))
  return zipname


def _get_package_uri_from_packer(cluster, package, packer=None):
  cluster = Cluster.get(cluster).packer_redirect or cluster
  role, name, version = package
  log.info('Fetching metadata for package %s/%s version %s in %s.' % (
    role, name, version, cluster))
  try:
    if packer is None:
      packer = sd_packer_client.create_packer(cluster)
    metadata = packer.get_version(role, name, version)
  except Packer.Error as e:
    die('Failed to fetch package metadata: %s' % e)

  latest_audit = sorted(metadata['auditLog'], key=lambda a: a['timestamp'])[-1]
  if latest_audit['state'] == 'DELETED':
    die('The requested package version has been deleted.')
  return metadata['uri']


def _get_and_verify_metadata(package_version):
  if not isinstance(package_version, dict):
    die('packer.get_version did nor return a dict %r' % package_version)
  if not 'id' in package_version:
    die('package_version does not contain an id: %r' % package_version)
  if not 'metadata' in package_version:
    die('package_versions does not contain metadata: %r' % package_version)
  metadata = package_version['metadata']
  if not isinstance(metadata, unicode):
    die('metadata in package_version is not unicode: %r' % package_version)
  try:
    metadata = json.loads(metadata)
  except ValueError:
    die('metadata is package_version is not a valid json object: %r' % package_version)
  if not isinstance(metadata, dict):
    die('deserialized metadata in package_version is not a dict: %r' % package_version)
  if not 'md5sum' in metadata:
    die('metadata in package_version does not have md5sum: %r' % package_version)
  return metadata


def _get_package_uri_from_packer_and_files(cluster, role, name, package_files):
  log.warning('DEVELOPMENT-ONLY FEATURE USED: testing_package_files')
  with temporary_dir(root_dir=os.getcwd()) as tmp_dir:
    packer = sd_packer_client.create_packer(cluster)
    zip_name = _zip_package_files(name, package_files, tmp_dir)
    digest = packer.compute_checksum(zip_name)
    package_name = name + _PACKAGE_FILES_SUFFIX
    package_tuple = (role, package_name, 'latest')
    package_version = None
    metadata = {}
    must_upload = False
    try:
      package_version = packer.get_version(role, package_name, 'latest')
    except Packer.Error:
      must_upload = True  # No package versions present: will upload a new version

    if package_version is not None:
      metadata = _get_and_verify_metadata(package_version)
      must_upload = (metadata['md5sum'] != digest)

    if must_upload:
      log.info('Uploading new version of package_files')
      metadata['md5sum'] = digest
      metadata_json = json.dumps(metadata)
      packer.add(role, package_name, zip_name, metadata_json, digest)
      if package_version is not None:
        # Delete previous version
        packer.delete(role, package_name, package_version['id'])
    else:
      log.info('Not uploading package_files: unchanged package')

  return _get_package_uri_from_packer(cluster, package_tuple, packer)


def _get_package_uri(config, copy_app_from=None):
  cluster = config.cluster()
  package = config.package()

  if config.hdfs_path():
    log.warning('''
*******************************************************************************
  hdfs_path and --copy_app_from have been deprecated and will soon be disabled
  altogether.

  Please switch to using the package option as soon as possible!
  For details on how to do this, please consult

  http://go/mesostutorial
  and
  http://confluence.local.twitter.com/display/ENG/Mesos+Configuration+Reference
*******************************************************************************''')

  if package and copy_app_from:
    die('copy_app_from may not be used when a package spec is used in the configuration')

  if copy_app_from:
    return '/mesos/pkg/%s/%s' % (config.role(), posixpath.basename(copy_app_from))

  if package:
    return _get_package_uri_from_packer(cluster, package)

  if config.package_files():
    return _get_package_uri_from_packer_and_files(
        cluster, config.role(), config.name(), config.package_files())

  if config.hdfs_path():
    return config.hdfs_path()


PACKAGE_DEPRECATION_WARNING = """
Job.package is deprecated.  Instead use the {{packer}} namespace directly.

See the packer section of the Configuration Reference page for more information:
http://confluence.twitter.biz/display/Aurora/Aurora+Configuration+Reference#AuroraConfigurationReference-%7B%7Bpacker%7D%7Dnamespace
"""

PACKAGE_UNDERSPECIFIED_WARNING = """
You've specified Job.package in your configuration but not referenced {{mesos.package}}
or {{mesos.package_uri}}.  We no longer copy package artifacts directly into your
sandbox prior to invocation, so you must copy them by adding:

  %s

either as a process or into a process in your Task.
"""


def _warn_on_unspecified_package_bindings(config):
  if not isinstance(config, PystachioConfig) or not config.package():
    return

  print(PACKAGE_DEPRECATION_WARNING, file=sys.stderr)

  _, refs = config.raw().interpolate()
  p_uri, p = Ref.from_address('mesos.package_uri'), Ref.from_address('mesos.package')
  if p not in refs and p_uri not in refs:
    print(PACKAGE_UNDERSPECIFIED_WARNING % (
        '{{package[%s][%s][%s].copy_command}}' % tuple(config.package())))


ANNOUNCE_ERROR = """
Announcer specified primary port as '%(primary_port)s' but no processes have bound that port.
If you would like to utilize this port, you must bind {{thermos.ports[%(primary_port)s]}} into
a Process bound in your Task.
"""


def _validate_announce_configuration(config):
  if not config.raw().has_announce():
    return
  primary_port = config.raw().announce().primary_port().get()
  if primary_port not in config.ports():
    print(ANNOUNCE_ERROR % {'primary_port': primary_port}, file=sys.stderr)
    raise config.InvalidConfig("Announcer specified primary port as "
        '%s but no processes have bound that port!' % primary_port)


def _inject_packer_bindings(config, force_local=False):
  if isinstance(config, MesosConfig):
    raise ValueError('inject_packer_bindings can only be used with Pystachio configs!')
<commit_msg>Use thermos config syntax for labrat.<commit_after>'''Library of utilities called by the mesos client binary
'''

from __future__ import print_function

import functools
import json
import os
import posixpath
import sys

from pystachio import Ref
from twitter.common import app, log
from twitter.common.contextutil import temporary_dir, open_zip
from twitter.mesos.clusters import Cluster
from twitter.mesos.config.schema import Packer as PackerObject
from twitter.mesos.packer.packer_client import Packer
from twitter.mesos.packer import sd_packer_client
from twitter.mesos.parsers import (
    FormatDetector,
    MesosConfig,
    PystachioCodec,
    PystachioConfig)

from gen.twitter.mesos.ttypes import ResponseCode


_PACKAGE_FILES_SUFFIX = MesosConfig.PACKAGE_FILES_SUFFIX


def die(msg):
  log.fatal(msg)
  sys.exit(1)


def _zip_package_files(job_name, package_files, tmp_dir):
  zipname = os.path.join(tmp_dir, MesosConfig.get_package_files_zip_name(job_name))
  with open_zip(zipname, 'w') as zipf:
    for file_name in package_files:
      zipf.write(file_name, arcname=os.path.basename(file_name))
  return zipname


def _get_package_uri_from_packer(cluster, package, packer=None):
  cluster = Cluster.get(cluster).packer_redirect or cluster
  role, name, version = package
  log.info('Fetching metadata for package %s/%s version %s in %s.' % (
    role, name, version, cluster))
  try:
    if packer is None:
      packer = sd_packer_client.create_packer(cluster)
    metadata = packer.get_version(role, name, version)
  except Packer.Error as e:
    die('Failed to fetch package metadata: %s' % e)

  latest_audit = sorted(metadata['auditLog'], key=lambda a: a['timestamp'])[-1]
  if latest_audit['state'] == 'DELETED':
    die('The requested package version has been deleted.')
  return metadata['uri']


def _get_and_verify_metadata(package_version):
  if not isinstance(package_version, dict):
    die('packer.get_version did nor return a dict %r' % package_version)
  if not 'id' in package_version:
    die('package_version does not contain an id: %r' % package_version)
  if not 'metadata' in package_version:
    die('package_versions does not contain metadata: %r' % package_version)
  metadata = package_version['metadata']
  if not isinstance(metadata, unicode):
    die('metadata in package_version is not unicode: %r' % package_version)
  try:
    metadata = json.loads(metadata)
  except ValueError:
    die('metadata is package_version is not a valid json object: %r' % package_version)
  if not isinstance(metadata, dict):
    die('deserialized metadata in package_version is not a dict: %r' % package_version)
  if not 'md5sum' in metadata:
    die('metadata in package_version does not have md5sum: %r' % package_version)
  return metadata


def _get_package_uri_from_packer_and_files(cluster, role, name, package_files):
  log.warning('DEVELOPMENT-ONLY FEATURE USED: testing_package_files')
  with temporary_dir(root_dir=os.getcwd()) as tmp_dir:
    packer = sd_packer_client.create_packer(cluster)
    zip_name = _zip_package_files(name, package_files, tmp_dir)
    digest = packer.compute_checksum(zip_name)
    package_name = name + _PACKAGE_FILES_SUFFIX
    package_tuple = (role, package_name, 'latest')
    package_version = None
    metadata = {}
    must_upload = False
    try:
      package_version = packer.get_version(role, package_name, 'latest')
    except Packer.Error:
      must_upload = True  # No package versions present: will upload a new version

    if package_version is not None:
      metadata = _get_and_verify_metadata(package_version)
      must_upload = (metadata['md5sum'] != digest)

    if must_upload:
      log.info('Uploading new version of package_files')
      metadata['md5sum'] = digest
      metadata_json = json.dumps(metadata)
      packer.add(role, package_name, zip_name, metadata_json, digest)
      if package_version is not None:
        # Delete previous version
        packer.delete(role, package_name, package_version['id'])
    else:
      log.info('Not uploading package_files: unchanged package')

  return _get_package_uri_from_packer(cluster, package_tuple, packer)


def _get_package_uri(config, copy_app_from=None):
  cluster = config.cluster()
  package = config.package()

  if config.hdfs_path():
    log.warning('''
*******************************************************************************
  hdfs_path and --copy_app_from have been deprecated and will soon be disabled
  altogether.

  Please switch to using the package option as soon as possible!
  For details on how to do this, please consult

  http://go/mesostutorial
  and
  http://confluence.local.twitter.com/display/ENG/Mesos+Configuration+Reference
*******************************************************************************''')

  if package and copy_app_from:
    die('copy_app_from may not be used when a package spec is used in the configuration')

  if copy_app_from:
    return '/mesos/pkg/%s/%s' % (config.role(), posixpath.basename(copy_app_from))

  if package:
    return _get_package_uri_from_packer(cluster, package)

  if config.package_files():
    return _get_package_uri_from_packer_and_files(
        cluster, config.role(), config.name(), config.package_files())

  if config.hdfs_path():
    return config.hdfs_path()


PACKAGE_DEPRECATION_WARNING = """
Job.package is deprecated.  Instead use the {{packer}} namespace directly.

See the packer section of the Configuration Reference page for more information:
http://confluence.twitter.biz/display/Aurora/Aurora+Configuration+Reference#AuroraConfigurationReference-%7B%7Bpacker%7D%7Dnamespace
"""

PACKAGE_UNDERSPECIFIED_WARNING = """
You've specified Job.package in your configuration but not referenced {{mesos.package}}
or {{mesos.package_uri}}.  We no longer copy package artifacts directly into your
sandbox prior to invocation, so you must copy them by adding:

  %s

either as a process or into a process in your Task.
"""


def _warn_on_unspecified_package_bindings(config):
  if not isinstance(config, PystachioConfig) or not config.package():
    return

  print(PACKAGE_DEPRECATION_WARNING, file=sys.stderr)

  _, refs = config.raw().interpolate()
  p_uri, p = Ref.from_address('mesos.package_uri'), Ref.from_address('mesos.package')
  if p not in refs and p_uri not in refs:
    print(PACKAGE_UNDERSPECIFIED_WARNING % (
        '{{packer[%s][%s][%s].copy_command}}' % tuple(config.package())))


ANNOUNCE_ERROR = """
Announcer specified primary port as '%(primary_port)s' but no processes have bound that port.
If you would like to utilize this port, you must bind {{thermos.ports[%(primary_port)s]}} into
a Process bound in your Task.
"""


def _validate_announce_configuration(config):
  if not config.raw().has_announce():
    return
  primary_port = config.raw().announce().primary_port().get()
  if primary_port not in config.ports():
    print(ANNOUNCE_ERROR % {'primary_port': primary_port}, file=sys.stderr)
    raise config.InvalidConfig("Announcer specified primary port as "
        '%s but no processes have bound that port!' % primary_port)


def _inject_packer_bindings(config, force_local=False):
  if isinstance(config, MesosConfig):
    raise ValueError('inject_packer_bindings can only be used with Pystachio configs!')

{'fraction_non_alphanumeric': 0.1140127826266236, 'fraction_numerical': 0.0013744759810322314, 'mean_word_length': 4.865376864167675, 'pattern_counts': {'":': 0, '<': 3, '<?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': '2550803', 'n_tokens_mistral': 4537, 'n_tokens_neox': 4404, 'n_words': 1241}