Merge remote-tracking branch 'origin/master' into pete-fixes
authorPeter Amstutz <peter.amstutz@curoverse.com>
Tue, 24 Jun 2014 20:04:10 +0000 (16:04 -0400)
committerPeter Amstutz <peter.amstutz@curoverse.com>
Tue, 24 Jun 2014 20:04:10 +0000 (16:04 -0400)
13 files changed:
apps/workbench/app/controllers/actions_controller.rb
apps/workbench/app/controllers/application_controller.rb
apps/workbench/app/helpers/application_helper.rb
sdk/cli/bin/arv-run-pipeline-instance
sdk/cli/bin/crunch-job
sdk/perl/lib/Arvados.pm
sdk/python/arvados/collection.py
sdk/python/arvados/stream.py
sdk/python/arvados/util.py
sdk/python/bin/arv-normalize
services/api/script/cancel_stale_jobs.rb
services/fuse/arvados_fuse/__init__.py
services/fuse/bin/arv-mount

index c1e2617fc7c4d1de73219b646086701bc3c7e3fb..0cb6eed2d8e9cd6a92821037d5c3e23eeb329db9 100644 (file)
@@ -109,15 +109,24 @@ class ActionsController < ApplicationController
     IO.popen(['arv-normalize'], 'w+b') do |io|
       io.write combined
       io.close_write
-      while buf = io.read(2**20)
+      while buf = io.read(2**16)
         normalized += buf
       end
     end
 
+    normalized_stripped = ''
+    IO.popen(['arv-normalize', '--strip'], 'w+b') do |io|
+      io.write combined
+      io.close_write
+      while buf = io.read(2**16)
+        normalized_stripped += buf
+      end
+    end
+
     require 'digest/md5'
 
     d = Digest::MD5.new()
-    d << normalized
+    d << normalized_stripped
     newuuid = "#{d.hexdigest}+#{normalized.length}"
 
     env = Hash[ENV].
index d0496cb42a116e05adc99eadcb4adb883260ce17..38ae911b5c3365e182362d9abc68824ef43aee15 100644 (file)
@@ -728,7 +728,7 @@ class ApplicationController < ActionController::Base
   def get_n_objects_of_class dataclass, size
     @objects_map_for ||= {}
 
-    raise ArgumentError, 'Argument is not a data class' unless dataclass.is_a? Class
+    raise ArgumentError, 'Argument is not a data class' unless dataclass.is_a? ArvadosBase
     raise ArgumentError, 'Argument is not a valid limit size' unless (size && size>0)
 
     # if the objects_map_for has a value for this dataclass, and the
index 66267e028d4df4eb5dd1436fcdeb507904e466aa..26be2898a781fafdf61e43cef88ce52bda231486 100644 (file)
@@ -306,7 +306,7 @@ module ApplicationHelper
     selectables = []
 
     attrtext = attrvalue
-    if dataclass and dataclass.is_a? Class
+    if dataclass and dataclass.is_a? ArvadosBase
       objects = get_n_objects_of_class dataclass, 10
       objects.each do |item|
         items << item
@@ -392,7 +392,7 @@ module ApplicationHelper
       render opts.merge(partial: "application/#{partial}")
     end
   end
-    
+
   def fa_icon_class_for_object object
     case object.class.to_s.to_sym
     when :User
index 4810768ded0d5cd5a7051804dc38a8fa2f3d80c2..8e26600ff119dbaa55e0fdf78c06156540198138 100755 (executable)
@@ -469,6 +469,7 @@ class WhRunPipelineInstance
             :repository => c[:repository],
             :nondeterministic => c[:nondeterministic],
             :output_is_persistent => c[:output_is_persistent] || false,
+            :runtime_constraints => c[:runtime_constraints],
             :owner_uuid => owner_uuid,
             # TODO: Delete the following three attributes when
             # supporting pre-20140418 API servers is no longer
@@ -591,7 +592,7 @@ class WhRunPipelineInstance
           ended += 1
           if c[:job][:success] == true
             succeeded += 1
-          elsif c[:job][:success] == false
+          elsif c[:job][:success] == false or c[:job][:cancelled_at]
             failed += 1
           end
         end
index 6224a64afe16f23e14d7c09f2aeeae7e24368c8e..05185cac23545c7916410d14a6b97a6679150344 100755 (executable)
@@ -639,7 +639,9 @@ for (my $todo_ptr = 0; $todo_ptr <= $#jobstep_todo; $todo_ptr ++)
     my $command =
        "if [ -e $ENV{TASK_WORK} ]; then rm -rf $ENV{TASK_WORK}; fi; "
         ."mkdir -p $ENV{JOB_WORK} $ENV{CRUNCH_TMP} $ENV{TASK_WORK} $ENV{TASK_KEEPMOUNT} "
+        ."&& chmod og+wrx $ENV{TASK_WORK}"
        ."&& cd $ENV{CRUNCH_TMP} ";
+    umask(077);
     if ($build_script)
     {
       $build_script_to_send = $build_script;
@@ -657,14 +659,24 @@ for (my $todo_ptr = 0; $todo_ptr <= $#jobstep_todo; $todo_ptr ++)
       $command .=
           q{$(ip -o address show scope global |
               gawk 'match($4, /^([0-9\.:]+)\//, x){print "--dns", x[1]}') };
-      foreach my $env_key (qw(CRUNCH_SRC CRUNCH_TMP TASK_KEEPMOUNT))
-      {
-        $command .= "-v \Q$ENV{$env_key}:$ENV{$env_key}:rw\E ";
-      }
+      $command .= "-v \Q$ENV{TASK_WORK}:/tmp/crunch-job:rw\E ";
+      $command .= "-v \Q$ENV{CRUNCH_SRC}:/tmp/crunch-src:ro\E ";
+      $command .= "-v \Q$ENV{TASK_KEEPMOUNT}:/mnt:ro\E ";
       while (my ($env_key, $env_val) = each %ENV)
       {
         if ($env_key =~ /^(ARVADOS|JOB|TASK)_/) {
-          $command .= "-e \Q$env_key=$env_val\E ";
+          if ($env_key eq "TASK_WORK") {
+            $command .= "-e \QTASK_WORK=/tmp/crunch-job\E ";
+          }
+          elsif ($env_key eq "TASK_KEEPMOUNT") {
+            $command .= "-e \QTASK_KEEPMOUNT=/mnt\E ";
+          }
+          elsif ($env_key eq "CRUNCH_SRC") {
+            $command .= "-e \QCRUNCH_SRC=/tmp/crunch-src\E ";
+          }
+          else {
+            $command .= "-e \Q$env_key=$env_val\E ";
+          }
         }
       }
       $command .= "\Q$docker_hash\E ";
@@ -672,7 +684,7 @@ for (my $todo_ptr = 0; $todo_ptr <= $#jobstep_todo; $todo_ptr ++)
       $command .= "crunchstat -cgroup-root=/sys/fs/cgroup -poll=10000 "
     }
     $command .= "stdbuf -o0 -e0 ";
-    $command .= "$ENV{CRUNCH_SRC}/crunch_scripts/" . $Job->{"script"};
+    $command .= "/tmp/crunch-src/crunch_scripts/" . $Job->{"script"};
     my @execargs = ('bash', '-c', $command);
     srun (\@srunargs, \@execargs, undef, $build_script_to_send);
     exit (111);
index d5eca9035e7d5497d07b565495f92fe087824ec0..414272b6bcfcf2c6aad89128c94639eb67896c49 100644 (file)
@@ -6,14 +6,14 @@ Arvados -- client library for Arvados services
 
   use Arvados;
   $arv = Arvados->new(apiHost => 'arvados.local');
-  
+
   my $instances = $arv->{'pipeline_instances'}->{'list'}->execute();
   print "UUID is ", $instances->{'items'}->[0]->{'uuid'}, "\n";
-  
+
   $uuid = 'eiv0u-arx5y-2c5ovx43zw90gvh';
   $instance = $arv->{'pipeline_instances'}->{'get'}->execute('uuid' => $uuid);
   print "ETag is ", $instance->{'etag'}, "\n";
-  
+
   $instance->{'active'} = 1;
   $instance->{'name'} = '';
   $instance->save();
@@ -58,15 +58,20 @@ Default C<v1>
 =cut
 
 package Arvados;
+
+use Net::SSL (); # From Crypt-SSLeay
+BEGIN {
+  $Net::HTTPS::SSL_SOCKET_CLASS = "Net::SSL"; # Force use of Net::SSL
+}
+
 use JSON;
-use Data::Dumper;
-use IO::Socket::SSL;
 use Carp;
 use Arvados::ResourceAccessor;
 use Arvados::ResourceMethod;
 use Arvados::ResourceProxy;
 use Arvados::ResourceProxyList;
 use Arvados::Request;
+use Data::Dumper;
 
 $Arvados::VERSION = 0.1;
 
@@ -85,12 +90,15 @@ sub build
 
     $config = load_config_file("$ENV{HOME}/.config/arvados/settings.conf");
 
-    $self->{'authToken'} ||= 
+    $self->{'authToken'} ||=
        $ENV{ARVADOS_API_TOKEN} || $config->{ARVADOS_API_TOKEN};
 
     $self->{'apiHost'} ||=
        $ENV{ARVADOS_API_HOST} || $config->{ARVADOS_API_HOST};
 
+    $self->{'noVerifyHostname'} ||=
+       $ENV{ARVADOS_API_HOST_INSECURE};
+
     $self->{'apiProtocolScheme'} ||=
        $ENV{ARVADOS_API_PROTOCOL_SCHEME} ||
        $config->{ARVADOS_API_PROTOCOL_SCHEME};
@@ -127,7 +135,7 @@ sub new_request
 {
     my $self = shift;
     local $ENV{'PERL_LWP_SSL_VERIFY_HOSTNAME'};
-    if ($opts{'noVerifyHostname'} || ($host =~ /\.local$/)) {
+    if ($self{'noVerifyHostname'} || ($host =~ /\.local$/)) {
         $ENV{'PERL_LWP_SSL_VERIFY_HOSTNAME'} = 0;
     }
     Arvados::Request->new();
index e7b26017d613ea8d0c792da9d228bb722ba9af44..87923489f803570b2b3730d72ad037dbef3433c0 100644 (file)
@@ -143,9 +143,13 @@ class CollectionReader(object):
             for f in s.all_files():
                 yield f
 
-    def manifest_text(self):
+    def manifest_text(self, strip=False):
         self._populate()
-        return self._manifest_text
+        if strip:
+            m = ''.join([StreamReader(stream).manifest_text(strip=True) for stream in self._streams])
+            return m
+        else:
+            return self._manifest_text
 
 class CollectionWriter(object):
     KEEP_BLOCK_SIZE = 2**26
@@ -359,7 +363,7 @@ class CollectionWriter(object):
                              for x in fields[1:-1] ]
                 clean += fields[0] + ' ' + ' '.join(locators) + ' ' + fields[-1] + "\n"
         return clean
-        
+
     def manifest_text(self):
         self.finish_current_stream()
         manifest = ''
index dc90d8e57c19251010c9042cedb85ab46a4eb228..ca5a9e5e549df0e3d2e888d0f52f76e92653b619 100644 (file)
@@ -71,7 +71,7 @@ def locators_and_ranges(data_locators, range_start, range_size, debug=False):
         block_size = data_locators[i][BLOCKSIZE]
         block_start = data_locators[i][OFFSET]
         block_end = block_start + block_size
-    
+
     while i < len(data_locators):
         locator, block_size, block_start = data_locators[i]
         block_end = block_start + block_size
@@ -169,7 +169,7 @@ class StreamFileReader(object):
             dc = bz2.BZ2Decompressor()
             return self.decompress(lambda segment: dc.decompress(segment), size)
         elif re.search('\.gz$', self._name):
-            dc = zlib.decompressobj(16+zlib.MAX_WBITS)            
+            dc = zlib.decompressobj(16+zlib.MAX_WBITS)
             return self.decompress(lambda segment: dc.decompress(dc.unconsumed_tail + segment), size)
         else:
             return self.readall(size)
@@ -209,7 +209,7 @@ class StreamReader(object):
             self._keep = keep
         else:
             self._keep = Keep.global_client_object()
-            
+
         streamoffset = 0L
 
         # parse stream
@@ -264,11 +264,16 @@ class StreamReader(object):
         for locator, blocksize, segmentoffset, segmentsize in locators_and_ranges(self._data_locators, start, size):
             data += self._keep.get(locator)[segmentoffset:segmentoffset+segmentsize]
         return data
-    
-    def manifest_text(self):
+
+    def manifest_text(self, strip=False):
         manifest_text = [self.name().replace(' ', '\\040')]
-        manifest_text.extend([d[LOCATOR] for d in self._data_locators])
-        manifest_text.extend([' '.join(["{}:{}:{}".format(seg[LOCATOR], seg[BLOCKSIZE], f.name().replace(' ', '\\040')) 
+        if strip:
+            for d in self._data_locators:
+                m = re.match(r'^[0-9a-f]{32}\+(\d+)*', d[LOCATOR])
+                manifest_text.append(m.group(0))
+        else:
+            manifest_text.extend([d[LOCATOR] for d in self._data_locators])
+        manifest_text.extend([' '.join(["{}:{}:{}".format(seg[LOCATOR], seg[BLOCKSIZE], f.name().replace(' ', '\\040'))
                                         for seg in f.segments])
                               for f in self._files.values()])
         return ' '.join(manifest_text) + '\n'
index e063f12de91c6ceb1d76302db6ea1ea201d57174..d5ea18ba49ad749a60aa9faba7141ddb6e3bbf63 100644 (file)
@@ -30,7 +30,10 @@ def run_command(execargs, **kwargs):
     kwargs.setdefault('close_fds', True)
     kwargs.setdefault('shell', False)
     p = subprocess.Popen(execargs, **kwargs)
-    stdoutdata, stderrdata = p.communicate(None)
+    if kwargs['stdout'] == subprocess.PIPE:
+        stdoutdata, stderrdata = p.communicate(None)
+    else:
+        p.wait()
     if p.returncode != 0:
         raise errors.CommandFailedError(
             "run_command %s exit %d:\n%s" %
@@ -162,7 +165,7 @@ def zipball_extract(zipball, path):
                     break
                 zip_file.write(buf)
             zip_file.close()
-            
+
             p = subprocess.Popen(["unzip",
                                   "-q", "-o",
                                   "-d", path,
index 755b56507289bbf1d5601ed3e9f238523a0dae1e..0506381272c856bbc4e1b933aef6f474752d3667 100755 (executable)
@@ -14,13 +14,14 @@ parser = argparse.ArgumentParser(
     description='Read manifest on standard input and put normalized manifest on standard output.')
 
 parser.add_argument('--extract', type=str, help="The file to extract from the input manifest")
+parser.add_argument('--strip', action='store_true', help="Strip authorization tokens")
 
 args = parser.parse_args()
 
 import arvados
 
 r = sys.stdin.read()
-    
+
 cr = arvados.CollectionReader(r)
 
 if args.extract:
@@ -36,4 +37,4 @@ if args.extract:
             if fn in s.files():
                 sys.stdout.write(s.files()[fn].as_manifest())
 else:
-    sys.stdout.write(cr.manifest_text())
+    sys.stdout.write(cr.manifest_text(args.strip))
index dde4cbed0cf6b78eaf973a7f77f40d0b2333c071..717edd7bb3b03833a6fde62e9132e842100c99d9 100755 (executable)
@@ -1,5 +1,6 @@
 #!/usr/bin/env ruby
 
+
 if ENV["CRUNCH_DISPATCH_LOCKFILE"]
   lockfilename = ENV.delete "CRUNCH_DISPATCH_LOCKFILE"
   lockfile = File.open(lockfilename, File::RDWR|File::CREAT, 0644)
@@ -13,25 +14,32 @@ ENV["RAILS_ENV"] = ARGV[0] || ENV["RAILS_ENV"] || "development"
 require File.dirname(__FILE__) + '/../config/boot'
 require File.dirname(__FILE__) + '/../config/environment'
 
-def cancel_stale_jobs
-  Job.running.each do |jobrecord|
-    f = Log.where("object_uuid=?", jobrecord.uuid).limit(1).order("created_at desc").first
-    if f
-      age = (Time.now - f.created_at)
-      if age > 300
-        $stderr.puts "dispatch: failing orphan job #{jobrecord.uuid}, last log is #{age} seconds old"
-        # job is marked running, but not known to crunch-dispatcher, and
-        # hasn't produced any log entries for 5 minutes, so mark it as failed.
-        jobrecord.running = false
-        jobrecord.cancelled_at ||= Time.now
-        jobrecord.finished_at ||= Time.now
-        if jobrecord.success.nil?
-          jobrecord.success = false
+class CancelJobs
+  include ApplicationHelper
+
+  def cancel_stale_jobs
+    act_as_system_user do
+      Job.running.each do |jobrecord|
+        puts jobrecord
+        f = Log.where("object_uuid=?", jobrecord.uuid).limit(1).order("created_at desc").first
+        if f
+          age = (Time.now - f.created_at)
+          if age > 300
+            $stderr.puts "dispatch: failing orphan job #{jobrecord.uuid}, last log is #{age} seconds old"
+            # job is marked running, but not known to crunch-dispatcher, and
+            # hasn't produced any log entries for 5 minutes, so mark it as failed.
+            jobrecord.running = false
+            jobrecord.cancelled_at ||= Time.now
+            jobrecord.finished_at ||= Time.now
+            if jobrecord.success.nil?
+              jobrecord.success = false
+            end
+            jobrecord.save!
+          end
         end
-        jobrecord.save!
       end
     end
   end
 end
 
-cancel_stale_jobs
+CancelJobs.new.cancel_stale_jobs
index 62af6c0c7cc31beaedaba9a066808395f5249838..2e7f4c7e02ef0d4cb61843ab20292b8718e43c1f 100644 (file)
@@ -2,6 +2,7 @@
 # FUSE driver for Arvados Keep
 #
 
+from __future__ import print_function
 import os
 import sys
 
@@ -19,6 +20,8 @@ import json
 from time import time
 from llfuse import FUSEError
 
+DEBUG = False
+
 class FreshBase(object):
     '''Base class for maintaining fresh/stale state to determine when to update.'''
     def __init__(self):
@@ -123,7 +126,8 @@ class Directory(FreshBase):
             try:
                 self.update()
             except apiclient.errors.HttpError as e:
-                print e
+                if DEBUG:
+                    print(e, file=sys.stderr)
 
     def __getitem__(self, item):
         self.checkupdate()
@@ -193,10 +197,12 @@ class CollectionDirectory(Directory):
                         cwd = cwd._entries[part]
                 for k, v in s.files().items():
                     cwd._entries[k] = self.inodes.add_entry(StreamReaderFile(cwd.inode, v))
-            print "found"
             self.fresh()
+            return True
         except Exception as detail:
-            print("%s: error: %s" % (self.collection_locator,detail) )
+            if DEBUG:
+                print("arv-mount %s: error: %s" % (self.collection_locator,detail), file=sys.stderr)
+            return False
 
 class MagicDirectory(Directory):
     '''A special directory that logically contains the set of all extant keep
@@ -216,19 +222,22 @@ class MagicDirectory(Directory):
         if k in self._entries:
             return True
         try:
-            if arvados.Keep.get(k):
+            e = self.inodes.add_entry(CollectionDirectory(self.inode, self.inodes, k))
+            if e.update():
+                self._entries[k] = e
                 return True
             else:
                 return False
         except Exception as e:
-            #print 'exception keep', e
+            if DEBUG:
+                print('arv-mount exception keep', e, file=sys.stderr)
             return False
 
     def __getitem__(self, item):
-        if item not in self._entries:
-            self._entries[item] = self.inodes.add_entry(CollectionDirectory(self.inode, self.inodes, item))
-        return self._entries[item]
-
+        if self.__contains__(item):
+            return self._entries[item]
+        else:
+            raise KeyError()
 
 class TagsDirectory(Directory):
     '''A special directory that contains as subdirectories all tags visible to the user.'''
@@ -411,9 +420,14 @@ class Operations(llfuse.Operations):
     so request handlers do not run concurrently unless the lock is explicitly released
     with llfuse.lock_released.'''
 
-    def __init__(self, uid, gid):
+    def __init__(self, uid, gid, debug):
         super(Operations, self).__init__()
 
+        if debug:
+            global DEBUG
+            DEBUG = True
+            print("arv-mount debug enabled", file=sys.stderr)
+
         self.inodes = Inodes()
         self.uid = uid
         self.gid = gid
@@ -470,7 +484,8 @@ class Operations(llfuse.Operations):
         return entry
 
     def lookup(self, parent_inode, name):
-        #print "lookup: parent_inode", parent_inode, "name", name
+        if DEBUG:
+            print("arv-mount lookup: parent_inode", parent_inode, "name", name, file=sys.stderr)
         inode = None
 
         if name == '.':
@@ -506,7 +521,8 @@ class Operations(llfuse.Operations):
         return fh
 
     def read(self, fh, off, size):
-        #print "read", fh, off, size
+        if DEBUG:
+            print("arv-mount read", fh, off, size, file=sys.stderr)
         if fh in self._filehandles:
             handle = self._filehandles[fh]
         else:
@@ -523,7 +539,8 @@ class Operations(llfuse.Operations):
             del self._filehandles[fh]
 
     def opendir(self, inode):
-        #print "opendir: inode", inode
+        if DEBUG:
+            print("arv-mount opendir: inode", inode, file=sys.stderr)
 
         if inode in self.inodes:
             p = self.inodes[inode]
@@ -544,14 +561,16 @@ class Operations(llfuse.Operations):
         return fh
 
     def readdir(self, fh, off):
-        #print "readdir: fh", fh, "off", off
+        if DEBUG:
+            print("arv-mount readdir: fh", fh, "off", off, file=sys.stderr)
 
         if fh in self._filehandles:
             handle = self._filehandles[fh]
         else:
             raise llfuse.FUSEError(errno.EBADF)
 
-        #print "handle.entry", handle.entry
+        if DEBUG:
+            print("arv-mount handle.entry", handle.entry, file=sys.stderr)
 
         e = off
         while e < len(handle.entry):
index db8e852b4cea27f96c6424f3509b3549dd8b14b8..726741e3b01619b79b62094662d76e819e9b38df 100755 (executable)
@@ -38,7 +38,10 @@ collections on the server.""")
     args = parser.parse_args()
 
     # Create the request handler
-    operations = Operations(os.getuid(), os.getgid())
+    operations = Operations(os.getuid(), os.getgid(), args.debug)
+
+    if args.debug:
+        arvados.config.settings()['ARVADOS_DEBUG'] = 'true'
 
     if args.groups:
         api = arvados.api('v1')