expose products for jvm bundle create and python binary create tasks

bundle.jvm currently does not expose a product_type for any archives it creates (only the loose bundle directory). And binary.python-binary-create currently does not expose any products at all. We would like these tasks to expose products, in order to be able to consume those products in downstream (internal) tasks.

In particular, JVM apps (at least at Twitter) are generally published as archived (zip'd/tar'd) bundles. And python apps are deployed directly as pexes.
1. Expose deployable_archive product in jvm bundle_create task (arhives, including zip, tar, tar.gz, tar.bz2).
2. Expose same deployable_archive product in python_binary_create task (pexes).
3. Above products live in .pants.d, and they have symlinks created in dist dir.
4. Modify integration and unit test cases to accommodate changes in (3).
5. Add test cases.

Cloned from the pull request in RB: https://rbcommons.com/s/twitter/r/3959/

Testing Done:
ci pending: https://travis-ci.org/pantsbuild/pants/builds/141133645

Bugs closed: 3477, 3501, 3593

Reviewed at https://rbcommons.com/s/twitter/r/4015/

contrib/go/tests/python/pants_test/contrib/go/tasks/test_go_fetch_integration.py

@@ -16,7 +16,7 @@ def test_go_fetch_integration(self):
 
     with self.temporary_workdir() as workdir:
       self.assert_success(self.run_pants_with_workdir(args, workdir))
-      # run it again to make sure cached packages is resolved correctly
+      # Run it again to make sure cached packages are resolved correctly.
       self.assert_success(self.run_pants_with_workdir(args, workdir))
 
   def test_issues_1998(self):

src/python/pants/backend/jvm/tasks/BUILD

@@ -96,6 +96,7 @@ python_library(
     'src/python/pants/base:exceptions',
     'src/python/pants/build_graph',
     'src/python/pants/fs',
+    'src/python/pants/util:fileutil',
     'src/python/pants/util:dirutil',
     'src/python/pants/util:objects',
   ],

src/python/pants/backend/jvm/tasks/bundle_create.py

@@ -17,7 +17,8 @@
 from pants.base.exceptions import TaskError
 from pants.build_graph.target_scopes import Scopes
 from pants.fs import archive
-from pants.util.dirutil import safe_mkdir
+from pants.util.dirutil import absolute_symlink, safe_mkdir, safe_mkdir_for
+from pants.util.fileutil import atomic_copy
 from pants.util.objects import datatype
 
 
@@ -55,23 +56,27 @@ def register_options(cls, register):
              help='Use target basename to prefix bundle folder or archive; otherwise a unique '
                   'identifier derived from target will be used.')
 
+  @classmethod
+  def implementation_version(cls):
+    return super(BundleCreate, cls).implementation_version() + [('BundleCreate', 1)]
+
   @classmethod
   def product_types(cls):
-    return ['jvm_bundles']
+    return ['jvm_archives', 'jvm_bundles', 'deployable_archives']
 
-  class App(datatype('App', ['address', 'binary', 'bundles', 'basename', 'deployjar', 'archive', 'target'])):
+  class App(datatype('App', ['address', 'binary', 'bundles', 'id', 'deployjar', 'archive', 'target'])):
     """A uniform interface to an app."""
 
     @staticmethod
     def is_app(target):
       return isinstance(target, (JvmApp, JvmBinary))
 
     @classmethod
-    def create_app(cls, target, deployjar, archive, use_basename_prefix=False):
+    def create_app(cls, target, deployjar, archive):
       return cls(target.address,
                  target if isinstance(target, JvmBinary) else target.binary,
                  [] if isinstance(target, JvmBinary) else target.payload.bundles,
-                 target.basename if use_basename_prefix else target.id,
+                 target.id,
                  deployjar,
                  archive,
                  target)
@@ -94,71 +99,93 @@ def _resolved_option(self, target, key):
     v = target.payload.get_field_value(key, None)
     return option_value if v is None else v
 
-  def execute(self):
-    use_basename_prefix = self.get_options().use_basename_prefix
-    if use_basename_prefix:
-      # NB(peiyu) This special casing is confusing especially given we already fail
-      # when duplicate basenames are detected. It's added because of the existing
-      # user experience. Turns out a `jvm_app` that depends on another `jvm_binary`
-      # of the same basename is fairly common. In this case, using just
-      # `target_roots` instead of all transitive targets will reduce the chance users
-      # see their bundle command fail due to basename conflicts. We should eventually
-      # get rid of this special case.
-      targets_to_bundle = self.context.target_roots
-    else:
-      targets_to_bundle = self.context.targets()
-
-    apps = []
-    for target in targets_to_bundle:
-      if self.App.is_app(target):
-        apps.append(self.App.create_app(target,
-                                        self._resolved_option(target, 'deployjar'),
-                                        self._resolved_option(target, 'archive'),
-                                        use_basename_prefix=use_basename_prefix))
-
-    if use_basename_prefix:
-      self.check_basename_conflicts(apps)
+  def _store_results(self, vt, bundle_dir, archivepath, app):
+    """Store a copy of the bundle and archive from the results dir in dist."""
+    # TODO (from mateor) move distdir management somewhere more general purpose.
+    dist_dir = self.get_options().pants_distdir
+    name = vt.target.basename if self.get_options().use_basename_prefix else app.id
+    bundle_copy = os.path.join(dist_dir, '{}-bundle'.format(name))
+    absolute_symlink(bundle_dir, bundle_copy)
+    self.context.log.info(
+      'created bundle copy {}'.format(os.path.relpath(bundle_copy, get_buildroot())))
+
+    if archivepath:
+      ext = archive.archive_extensions.get(app.archive, app.archive)
+      archive_copy = os.path.join(dist_dir,'{}.{}'.format(name, ext))
+      safe_mkdir_for(archive_copy)  # Ensure parent dir exists
+      atomic_copy(archivepath, archive_copy)
+      self.context.log.info(
+        'created archive copy {}'.format(os.path.relpath(archive_copy, get_buildroot())))
+
+  def _add_product(self, deployable_archive, app, path):
+    deployable_archive.add(
+      app.target, os.path.dirname(path)).append(os.path.basename(path))
+    self.context.log.debug('created {}'.format(os.path.relpath(path, get_buildroot())))
 
+  def execute(self):
     # NB(peiyu): performance hack to convert loose directories in classpath into jars. This is
     # more efficient than loading them as individual files.
     runtime_classpath = self.context.products.get_data('runtime_classpath')
+
+    # TODO (from mateor) The consolidate classpath is something that we could do earlier in the
+    # pipeline and it would be nice to just add those unpacked classed to a product and get the
+    # consolidated classpath for free.
     targets_to_consolidate = self.find_consolidate_classpath_candidates(
       runtime_classpath,
       self.context.targets(**self._target_closure_kwargs),
     )
     self.consolidate_classpath(targets_to_consolidate, runtime_classpath)
 
-    for app in apps:
-      archiver = archive.archiver(app.archive) if app.archive else None
+    targets_to_bundle = self.context.targets(self.App.is_app)
+
+    if self.get_options().use_basename_prefix:
+      self.check_basename_conflicts([t for t in self.context.target_roots if t in targets_to_bundle])
 
-      basedir = self.bundle(app)
-      # NB(Eric Ayers): Note that this product is not housed/controlled under .pants.d/  Since
-      # the bundle is re-created every time, this shouldn't cause a problem, but if we ever
-      # expect the product to be cached, a user running an 'rm' on the dist/ directory could
-      # cause inconsistencies.
+    with self.invalidated(targets_to_bundle, invalidate_dependents=True) as invalidation_check:
       jvm_bundles_product = self.context.products.get('jvm_bundles')
-      jvm_bundles_product.add(app.target, os.path.dirname(basedir)).append(os.path.basename(basedir))
-      if archiver:
-        archivepath = archiver.create(
-          basedir,
-          self.get_options().pants_distdir,
-          app.basename
-        )
-        self.context.log.info('created {}'.format(os.path.relpath(archivepath, get_buildroot())))
+      bundle_archive_product = self.context.products.get('deployable_archives')
+      jvm_archive_product = self.context.products.get('jvm_archives')
+
+      for vt in invalidation_check.all_vts:
+        app = self.App.create_app(vt.target,
+                                  self._resolved_option(vt.target, 'deployjar'),
+                                  self._resolved_option(vt.target, 'archive'))
+        archiver = archive.archiver(app.archive) if app.archive else None
+
+        bundle_dir = self._get_bundle_dir(app, vt.results_dir)
+        ext = archive.archive_extensions.get(app.archive, app.archive)
+        filename = '{}.{}'.format(app.id, ext)
+        archive_path = os.path.join(vt.results_dir, filename) if app.archive else ''
+        if not vt.valid:
+          self.bundle(app, vt.results_dir)
+          if app.archive:
+            archiver.create(bundle_dir, vt.results_dir, app.id)
+
+        self._add_product(jvm_bundles_product, app, bundle_dir)
+        if archiver:
+          self._add_product(bundle_archive_product, app, archive_path)
+          self._add_product(jvm_archive_product, app, archive_path)
+
+        # For root targets, create symlink.
+        if vt.target in self.context.target_roots:
+          self._store_results(vt, bundle_dir, archive_path, app)
 
   class BasenameConflictError(TaskError):
     """Indicates the same basename is used by two targets."""
 
-  def bundle(self, app):
+  def _get_bundle_dir(self, app, results_dir):
+    return os.path.join(results_dir, '{}-bundle'.format(app.id))
+
+  def bundle(self, app, results_dir):
     """Create a self-contained application bundle.
 
     The bundle will contain the target classes, dependencies and resources.
     """
 
     assert(isinstance(app, BundleCreate.App))
 
-    bundle_dir = os.path.join(self.get_options().pants_distdir, '{}-bundle'.format(app.basename))
-    self.context.log.info('creating {}'.format(os.path.relpath(bundle_dir, get_buildroot())))
+    bundle_dir = self._get_bundle_dir(app, results_dir)
+    self.context.log.debug('creating {}'.format(os.path.relpath(bundle_dir, get_buildroot())))
 
     safe_mkdir(bundle_dir, clean=True)
 
@@ -234,15 +261,15 @@ def find_consolidate_classpath_candidates(self, classpath_products, targets):
 
     return targets_with_directory_in_classpath
 
-  def check_basename_conflicts(self, apps):
+  def check_basename_conflicts(self, targets):
     """Apps' basenames are used as bundle directory names. Ensure they are all unique."""
 
     basename_seen = {}
-    for app in apps:
-      if app.basename in basename_seen:
+    for target in targets:
+      if target.basename in basename_seen:
         raise self.BasenameConflictError('Basename must be unique, found two targets use '
                                          "the same basename: {}'\n\t{} and \n\t{}"
-                                         .format(app.basename,
-                                                 basename_seen[app.basename].address.spec,
-                                                 app.target.address.spec))
-      basename_seen[app.basename] = app.target
+                                         .format(target.basename,
+                                                 basename_seen[target.basename].address.spec,
+                                                 target.address.spec))
+      basename_seen[target.basename] = target

src/python/pants/backend/python/tasks/python_binary_create.py

@@ -11,10 +11,25 @@
 
 from pants.backend.python.targets.python_binary import PythonBinary
 from pants.backend.python.tasks.python_task import PythonTask
+from pants.base.build_environment import get_buildroot
 from pants.base.exceptions import TaskError
+from pants.util.dirutil import safe_mkdir_for
+from pants.util.fileutil import atomic_copy
 
 
 class PythonBinaryCreate(PythonTask):
+  @classmethod
+  def product_types(cls):
+    return ['pex_archives', 'deployable_archives']
+
+  @classmethod
+  def implementation_version(cls):
+    return super(PythonBinaryCreate, cls).implementation_version() + [('PythonBinaryCreate', 1)]
+
+  @property
+  def cache_target_dirs(self):
+    return True
+
   @staticmethod
   def is_binary(target):
     return isinstance(target, PythonBinary)
@@ -35,10 +50,25 @@ def execute(self):
                         '{} and {} both have the name {}.'.format(binary, names[name], name))
       names[name] = binary
 
-    for binary in binaries:
-      self.create_binary(binary)
+    with self.invalidated(binaries, invalidate_dependents=True) as invalidation_check:
+      python_deployable_archive = self.context.products.get('deployable_archives')
+      python_pex_product = self.context.products.get('pex_archives')
+      for vt in invalidation_check.all_vts:
+        pex_path = os.path.join(vt.results_dir, '{}.pex'.format(vt.target.name))
+        if not vt.valid:
+          self.create_binary(vt.target, vt.results_dir)
+
+        python_pex_product.add(binary, os.path.dirname(pex_path)).append(os.path.basename(pex_path))
+        python_deployable_archive.add(binary, os.path.dirname(pex_path)).append(os.path.basename(pex_path))
+        self.context.log.debug('created {}'.format(os.path.relpath(pex_path, get_buildroot())))
+
+        # Create a copy for pex.
+        pex_copy = os.path.join(self._distdir, os.path.basename(pex_path))
+        safe_mkdir_for(pex_copy)
+        atomic_copy(pex_path, pex_copy)
+        self.context.log.info('created pex copy {}'.format(os.path.relpath(pex_copy, get_buildroot())))
 
-  def create_binary(self, binary):
+  def create_binary(self, binary, results_dir):
     interpreter = self.select_interpreter_for_targets(binary.closure())
 
     run_info_dict = self.context.run_tracker.run_info.get_as_dict()
@@ -50,5 +80,6 @@ def create_binary(self, binary):
 
     with self.temporary_chroot(interpreter=interpreter, pex_info=pexinfo, targets=[binary],
                                platforms=binary.platforms) as chroot:
-      pex_path = os.path.join(self._distdir, '{}.pex'.format(binary.name))
+      pex_path = os.path.join(results_dir, '{}.pex'.format(binary.name))
       chroot.package_pex(pex_path)
+      return pex_path

src/python/pants/fs/archive.py

@@ -124,11 +124,12 @@ def create(self, basedir, outdir, name, prefix=None):
           zip.write(full_path, relpath)
     return zippath
 
+archive_extensions = dict(tar='tar', tgz='tar.gz', tbz2='tar.bz2', zip='zip')
 
-TAR = TarArchiver('w:', 'tar')
-TGZ = TarArchiver('w:gz', 'tar.gz')
-TBZ2 = TarArchiver('w:bz2', 'tar.bz2')
-ZIP = ZipArchiver(ZIP_DEFLATED, 'zip')
+TAR = TarArchiver('w:', archive_extensions['tar'])
+TGZ = TarArchiver('w:gz', archive_extensions['tgz'])
+TBZ2 = TarArchiver('w:bz2', archive_extensions['tbz2'])
+ZIP = ZipArchiver(ZIP_DEFLATED, archive_extensions['zip'])
 
 _ARCHIVER_BY_TYPE = OrderedDict(tar=TAR, tgz=TGZ, tbz2=TBZ2, zip=ZIP)
 

src/python/pants/util/dirutil.py

@@ -242,6 +242,31 @@ def chmod_plus_x(path):
   os.chmod(path, path_mode)
 
 
+def absolute_symlink(source_path, target_path):
+  """Create a symlink at target pointing to source using the absolute path.
+
+  :param source_path: Absolute path to source file
+  :param target_path: Absolute path to intended symlink
+  :raises ValueError if source_path or link_path are not unique, absolute paths
+  :raises OSError on failure UNLESS file already exists or no such file/directory
+  """
+  if not os.path.isabs(source_path):
+    raise ValueError("Path for source : {} must be absolute".format(source_path))
+  if not os.path.isabs(target_path):
+    raise ValueError("Path for link : {} must be absolute".format(target_path))
+  if source_path == target_path:
+    raise ValueError("Path for link is identical to source : {}".format(source_path))
+  try:
+    if os.path.lexists(target_path):
+      os.unlink(target_path)
+    safe_mkdir_for(target_path)
+    os.symlink(source_path, target_path)
+  except OSError as e:
+    # Another run may beat us to deletion or creation.
+    if not (e.errno == errno.EEXIST or e.errno == errno.ENOENT):
+      raise
+
+
 def relative_symlink(source_path, link_path):
   """Create a symlink at link_path pointing to relative source