[autotest] gs_offloader refactor
BUG=None
TEST=None
This commit is the combination of multiple commits. Their commit
messages are as follows:
commit 4f934510a2f28c5c838c514f8b7dcab6fa81fa66
Author: Allen Li <ayatane@chromium.org>
Date: Fri May 5 12:13:58 2017 -0700
[autotest] Move get_sanitized_name()
Doing some housekeeping
BUG=chromium:715435
TEST=None
Change-Id: I14a1a9886018cfe0eb6c3a1a77b45836d8476a21
commit c6f7fb299886b67c5eaeb133a59db3b80ef86d1c
Author: Allen Li <ayatane@chromium.org>
Date: Fri May 5 12:22:53 2017 -0700
[autotest] Add escape() tests
BUG=chromium:715435
TEST=None
Change-Id: Ibeee38113043e7d16605b1fe5efa8b31350b8629
commit ce42ca80d166330d15113169f6542b839ac417e8
Author: Allen Li <ayatane@chromium.org>
Date: Fri May 5 12:28:47 2017 -0700
[autotest] Clean up escape()
BUG=chromium:715435
TEST=None
Change-Id: I99edb6bf2bd6a9ffebf92893d1b834d815f797b8
commit 87353b0b7efb4ffc55946d40c34ae83410cf0314
Author: Allen Li <ayatane@chromium.org>
Date: Fri May 5 12:33:38 2017 -0700
[autotest] Extract _setup_logging()
BUG=chromium:715435
TEST=None
Change-Id: Iab967852f043238730d2c502dc030bdced156b60
commit 0ac380ec9e267efe317d43f1e303f426f6f1a3f4
Author: Allen Li <ayatane@chromium.org>
Date: Fri May 5 12:35:15 2017 -0700
[autotest] Extract _get_log_filename()
BUG=chromium:715435
TEST=None
Change-Id: I7cbdd4523a76a3d0d06b810b9b8d07208c951861
commit 387b946ca761076bd67934554b4f67af8dc02f3a
Author: Allen Li <ayatane@chromium.org>
Date: Fri May 5 12:50:14 2017 -0700
[autotest] Move enqueue_offload()
This is only used by gs_offloader, yet its a public method on a
general purpose class.
(Hypothesis: this is why the generic function paradigm is better than
methods. The set of functions that want to dispatch on the class is a
superset of the functions that want to be methods on the class.)
BUG=chromium:715435
TEST=None
Change-Id: I8bad5c2f8fea7b922769bc7a9e9121003b1aab60
commit f920fae06b621f6bc6ae429abbf508d191c62ac2
Author: Allen Li <ayatane@chromium.org>
Date: Fri May 5 12:58:32 2017 -0700
[autotest] Move logging constants next to logging setup
BUG=chromium:715435
TEST=None
Change-Id: I9904322ee39cfc07edabe0a1294ecab94201f40c
commit 5cf63fcf06e939d7f00dc4ecf6a0f3ecd71e0c32
Author: Allen Li <ayatane@chromium.org>
Date: Fri May 5 13:07:25 2017 -0700
[autotest] Extract _sanitize_symlinks()
BUG=chromium:715435
TEST=None
Change-Id: Ie5f523b0f4472d2a55f1472b6c68199c97fbb3a8
commit d108e7ba6aca44a13b8b0732f2bd7680a50c74f1
Author: Allen Li <ayatane@chromium.org>
Date: Fri May 5 13:13:13 2017 -0700
[autotest] Extract _sanitize_fifos()
BUG=chromium:715435
TEST=None
Change-Id: I7d3392fcb27cb629ede2547f865c8f26b432acc3
commit 1e73efdf7ada7c450f78882523b42f56c3c0b942
Author: Allen Li <ayatane@chromium.org>
Date: Fri May 5 14:18:00 2017 -0700
[autotest] Extract _escape_rename()
BUG=chromium:715435
TEST=None
Change-Id: I2f665ffd00552e2bfb62050f0242727cea34a286
commit 96d47cff7634c203b809a98483999dfbe38de5cc
Author: Allen Li <ayatane@chromium.org>
Date: Fri May 5 14:41:02 2017 -0700
[autotest] Inline get_directory_size_kibibytes_cmd_list()
This isnt actually mocked out or referenced anywhere. There are
better ways to mock this out anyway.
BUG=chromium:715435
TEST=None
Change-Id: If1fc3f72fbb32371b087143efaca6bb2406678c0
commit 104ef14998ef31a9eafe214463a96f66b75bdd6d
Author: Allen Li <ayatane@chromium.org>
Date: Fri May 5 14:42:23 2017 -0700
[autotest] Clean up sanitize_dir()
BUG=chromium:715435
TEST=None
Change-Id: Ib0cc1a5842616e1d333328e7f9e95cf973afee88
commit c34f213c482233807bf06929fce3534a93977804
Author: Allen Li <ayatane@chromium.org>
Date: Fri May 5 14:44:36 2017 -0700
[autotest] Move MAX_FILE_COUNT to where its used
BUG=chromium:715435
TEST=None
Change-Id: I728ffafeead8d4e4cfa97e023a6a817420edc2a2
commit 4b85648f8978e75c42e0c7ff007bc0e802dc6979
Author: Allen Li <ayatane@chromium.org>
Date: Fri May 5 14:52:43 2017 -0700
[autotest] Extract _make_into_tarball()
BUG=chromium:715435
TEST=None
Change-Id: I2869bbc2ee15ab97c0850b7421e3a946609901a6
commit d6be13aa211abae6c320fc5525097c62afc7258b
Author: Allen Li <ayatane@chromium.org>
Date: Fri May 5 15:00:03 2017 -0700
[autotest] Use tarfile for _make_into_tarball()
BUG=chromium:715435
TEST=None
Change-Id: Id0937d15957ce1227b2a29fbd4f388f11eb0179c
commit b746a4eb7e06405b4bbd4c35f4c68fbdc8ee44ae
Author: Allen Li <ayatane@chromium.org>
Date: Fri May 5 15:05:43 2017 -0700
[autotest] Extract _count_files()
BUG=chromium:715435
TEST=None
Change-Id: I454573879911e70bae0d0b50ad57bad5899daeac
commit fb0a007caaab379d2544f292b73f0a3efcc969bf
Author: Allen Li <ayatane@chromium.org>
Date: Fri May 5 15:08:00 2017 -0700
[autotest] Simplify _count_files()
BUG=chromium:715435
TEST=None
Change-Id: Ibdead3268a5051416bf6fff095e542a87d2d525a
commit e58932671a693dacc6b74e5bebc5b106aa95ddaa
Author: Allen Li <ayatane@chromium.org>
Date: Fri May 5 15:19:19 2017 -0700
[autotest] Move notification constants
BUG=chromium:715435
TEST=None
Change-Id: Iad2a2f7f869d66d230a0de62b4f75802f16709c4
commit 6fcf74934821a141977f025fe322722c4e43c689
Author: Allen Li <ayatane@chromium.org>
Date: Fri May 5 15:25:25 2017 -0700
[autotest] Make offload_dir a callable instead of a closure
I swear, the first thing that a programmer does after learning about
closures is to rewrite every object as a closure.
BUG=chromium:715435
TEST=None
Change-Id: Ifd3cf9b8821b4926dc496859620f16fa34e4960d
commit 7f58dfc1dff99b2679515598352da837236615e0
Author: Allen Li <ayatane@chromium.org>
Date: Fri May 5 15:34:30 2017 -0700
[autotest] Make delete_files() inherit from GSOffloader
BUG=chromium:715435
TEST=None
Change-Id: Idbe4b1bd1d73edb25857ad22f0017df0ecb74402
commit 3b2b10ff73900ad11ca3a5fc57703ffc52a66c4d
Author: Allen Li <ayatane@chromium.org>
Date: Fri May 5 18:40:06 2017 -0700
[autotest] Fix GSOffloader semantics
BUG=chromium:715435
TEST=None
Change-Id: I82adfdcc9349b3afb3b6ccd487c53e90afb91c59
commit 4884e2f3c38bd590cee5eee843a20650bf24b59b
Author: Allen Li <ayatane@chromium.org>
Date: Fri May 5 15:42:31 2017 -0700
[autotest] Use context manager for tempfiles
BUG=chromium:715435
TEST=None
Change-Id: I526708355247e54620723ae1eb6a1bd5ca7ef653
commit 2fd102484811a8980d6c9893969f3bf3188bc2e3
Author: Allen Li <ayatane@chromium.org>
Date: Fri May 5 15:53:40 2017 -0700
[autotest] Use chromite timeout library
BUG=chromium:715435
TEST=None
Change-Id: I85e6457d17d27580a4443f96021a8d280dec9d5d
commit 63229138951ee4e76c43dc2578b50aec21b87501
Author: Allen Li <ayatane@chromium.org>
Date: Fri May 5 16:25:42 2017 -0700
[autotest] Split GSOffload call body
This will make subsequent CL diffs clearer.
BUG=chromium:715435
TEST=None
Change-Id: I6eca38fc060639882d9d2869a0df774a597f171d
commit 89e5e4eb5cddd0b8565ceb568e87083304b40520
Author: Allen Li <ayatane@chromium.org>
Date: Fri May 5 16:27:32 2017 -0700
[autotest] Move TemporaryFile context out
Remove indentation levels.
BUG=chromium:715435
TEST=None
Change-Id: I602c16c241ada2bc452630523764168dd1dda8cb
commit 819c8b1f6c9ce236fb7a7e12992a8ca555b544d2
Author: Allen Li <ayatane@chromium.org>
Date: Fri May 5 16:29:37 2017 -0700
[autotest] Move error handling out
BUG=chromium:715435
TEST=None
Change-Id: I7e712a0c6d0087f7fa5b44ccc45dec72503f365b
commit 13e9132457b0e40f25fcc62df939dcb37c23fe23
Author: Allen Li <ayatane@chromium.org>
Date: Fri May 19 16:57:08 2017 -0700
[autotest] Extract _emit_offload_metrics()
BUG=chromium:715435
TEST=None
Change-Id: I2299344c2d63a3d9585c8f83daf341b7ba1c5d9d
commit 1b0b0708db7d8f370942be9b10f9456983d63ce4
Author: Allen Li <ayatane@chromium.org>
Date: Fri May 19 16:57:40 2017 -0700
[autotest] Remove useless variable
BUG=chromium:715435
TEST=None
Change-Id: I690f3fa6ae7a8c5a224586057292713ae763bb60
commit 92806326b6fe8173c98e0560247f2c64041705dd
Author: Allen Li <ayatane@chromium.org>
Date: Fri May 5 17:38:18 2017 -0700
[autotest] Move pruning logic out
This OSError is supposed to only apply to the pruning; do that. Also
disentangle this pile of spaghetti.
BUG=chromium:715435
TEST=None
Change-Id: If46a5962f4530044058a4cd5ba840a9195c142a7
commit c604443fb5299d26fad0bb00c2b4b0f3bc99b24d
Author: Allen Li <ayatane@chromium.org>
Date: Fri May 5 17:42:44 2017 -0700
[autotest] Decrease scope of error flag
BUG=chromium:715435
TEST=None
Change-Id: If6a86f04c54de0c44774bd842902d54433925273
commit a519607dc6bf93356d7b336e90e9d649c6da441b
Author: Allen Li <ayatane@chromium.org>
Date: Fri May 5 17:46:05 2017 -0700
[autotest] Extract pruning logic
BUG=chromium:715435
TEST=None
Change-Id: Ia454d9d441e3a4e404350219041eadcaf8bdb0ef
commit e0e79552fe4fb7982e7cdabcc3ae28c0c34e40f4
Author: Allen Li <ayatane@chromium.org>
Date: Fri May 5 17:47:39 2017 -0700
[autotest] Clean up pruning test logic
BUG=chromium:715435
TEST=None
Change-Id: Idcb936ab659223eed0476858a05ce0b7a6b4e61e
commit 53e076a46ea1a4c01f0d41d2616a84b8a1d02f60
Author: Allen Li <ayatane@chromium.org>
Date: Fri May 19 17:11:55 2017 -0700
[autotest] Remove error flag
Even C++ has exceptions.
BUG=chromium:715435
TEST=None
Change-Id: I0c2e61cb753f8130356388c307410bb8d37489a9
commit 546180941013147ff493139c33f983e67a03a789
Author: Allen Li <ayatane@chromium.org>
Date: Fri May 19 17:12:53 2017 -0700
[autotest] Remove redudant check
BUG=chromium:715435
TEST=None
Change-Id: Iecc17ec88a78af691a4b3f6a444d2c033bbc3ae5
commit ab2fbbb578eeeb73f5db71c0266c8a551bfef748
Author: Allen Li <ayatane@chromium.org>
Date: Mon May 15 14:48:09 2017 -0700
[autotest] Move uploaded test out
BUG=chromium:715435
TEST=None
Change-Id: Ic265aba95aec337132251c50b6d86a51b415b86b
commit 341f36d00d5b0b5973da3fc78f92db218d17b5ff
Author: Allen Li <ayatane@chromium.org>
Date: Mon May 15 14:49:23 2017 -0700
[autotest] Fix indentation
BUG=chromium:715435
TEST=None
Change-Id: I851a790d9470f99bfc53ca3f71efee8fbc5ab9d1
commit d63eb51d5e5d9553bf6b33dee9b6d7b49135e115
Author: Allen Li <ayatane@chromium.org>
Date: Mon May 15 14:58:11 2017 -0700
[autotest] Remove is_offloaded()
There are two concepts of is_offloaded being used in gs_offloader.
One of them is an existence check that isnt actually quite right.
Remove it.
BUG=chromium:715435
TEST=None
Change-Id: Ib9002b848e74b55e29d17131a3c4962ba32f6207
commit 40ff0da7e960e6861557f3064a3d4618369d8f9e
Author: Allen Li <ayatane@chromium.org>
Date: Mon May 15 15:00:43 2017 -0700
[autotest] Clarify _update_offload_results()
BUG=chromium:715435
TEST=None
Change-Id: I23986726ac1d02a60baf64ba4d137f132c1b9d4f
commit a004fd4c4ff7c9df390b80941bba3a8d38355132
Author: Allen Li <ayatane@chromium.org>
Date: Mon May 15 15:05:56 2017 -0700
[autotest] Dont re-add uploaded jobs
BUG=chromium:715435
TEST=None
Change-Id: I054552667d62110f1e3a16099aeacb99e7d31661
commit 85e22760e53e23be728bec9bb983f710e1c0cdcc
Author: Allen Li <ayatane@chromium.org>
Date: Mon May 15 15:15:33 2017 -0700
[autotest] Fix upload_testresult_files() name
BUG=chromium:715435
TEST=None
Change-Id: I2da7b5c0a25c6cd8f295aed795e614449d93f9ba
commit 20add289f72659648fa5904524eb6d2c9bbe06f8
Author: Allen Li <ayatane@chromium.org>
Date: Mon May 15 15:32:18 2017 -0700
[autotest] Extract _remove_log_directory_contents()
BUG=chromium:715435
TEST=None
Change-Id: I06f2b1f312a0836854bfd43aaa6145473b558c2f
commit a53a80a78c7d72999645d7cb20e9d3e1541bbabe
Author: Allen Li <ayatane@chromium.org>
Date: Mon May 15 15:33:40 2017 -0700
[autotest] Replace get_job_directory()
This isnt Java, we dont need a getter.
BUG=chromium:715435
TEST=None
Change-Id: If97efd0ff8c6698ee8ce386dd113e60c53d02627
commit 8cceb41b48ec68b68ee050c41cc8255ab7090086
Author: Allen Li <ayatane@chromium.org>
Date: Mon May 15 15:40:14 2017 -0700
[autotest] Extract _is_expired()
BUG=chromium:715435
TEST=None
Change-Id: I81924e5ca47489692d836866f83fd9506884f36f
commit f3bfca34e41b3a6643785184e7e6d3f236f90831
Author: Allen Li <ayatane@chromium.org>
Date: Mon May 15 15:42:50 2017 -0700
[autotest] Make get_cmd_list private
BUG=chromium:715435
TEST=None
Change-Id: Id39af7863844b5ed7d5ec3c212e0683b133921e8
commit db314fad7d61ee25af4ac3c3bf8a6ba88e0538d4
Author: Allen Li <ayatane@chromium.org>
Date: Mon May 15 15:48:23 2017 -0700
[autotest] Replace get_failure_time(), get_failure_count()
This isnt Java, we dont need a getter.
BUG=chromium:715435
TEST=None
Change-Id: I857de5947d230c0c7d9f01a11214831189268369
commit c6bd98ca98d92d669c9695d62edd40a67c17be4f
Author: Allen Li <ayatane@chromium.org>
Date: Fri Jun 16 15:26:26 2017 -0700
[autotest] Fix damned mox thing
I swear it works locally, but it fails on trybot. I think mock will
be more reliable.
BUG=None
TEST=None
Change-Id: I7bf11033aee567a5a081b4a2be64769eef3b7f15
commit bb3f60bb6470fd37ba645f562615c2142f4b1d64
Author: Allen Li <ayatane@chromium.org>
Date: Mon Jun 19 12:27:34 2017 -0700
[autotest] Extract tempdir logic into a separate class
This class will be reused for a non-mox test class, because mox is
giving much such a ridiculous time. Moving a few function around
costs hours fiddling with mox tests.
BUG=None
TEST=None
Change-Id: I58fd573a8dbb046b4ab2449dfa22044102088b2c
commit e3b27f31393b6488547103b4f605b49b897cf2be
Author: Allen Li <ayatane@chromium.org>
Date: Mon Jun 19 12:32:04 2017 -0700
[autotest] Add non-mox version of timeout tests
This is confirmed passing when run locally.
BUG=None
TEST=None
Change-Id: I2196d51928b0ca35d783727f2db109148f80fe77
Change-Id: I21d905707ff67a8c124a84458f7e0e313f5fa433
Reviewed-on: https://chromium-review.googlesource.com/538996
Commit-Ready: Allen Li <ayatane@chromium.org>
Tested-by: Allen Li <ayatane@chromium.org>
Reviewed-by: Allen Li <ayatane@chromium.org>
diff --git a/client/common_lib/file_utils.py b/client/common_lib/file_utils.py
index b9f2cc7..9a7a70c 100644
--- a/client/common_lib/file_utils.py
+++ b/client/common_lib/file_utils.py
@@ -174,13 +174,6 @@
local_file.write(block)
-def get_directory_size_kibibytes_cmd_list(directory):
- """Returns command to get a directory's total size."""
- # Having this in its own method makes it easier to mock in
- # unittests.
- return ['du', '-sk', directory]
-
-
def get_directory_size_kibibytes(directory):
"""Calculate the total size of a directory with all its contents.
@@ -188,7 +181,7 @@
@return Size of the directory in kibibytes.
"""
- cmd = get_directory_size_kibibytes_cmd_list(directory)
+ cmd = ['du', '-sk', directory]
process = subprocess.Popen(cmd,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE)
@@ -201,4 +194,4 @@
logging.warning(stderr_data)
return 0
- return int(stdout_data.split('\t', 1)[0])
\ No newline at end of file
+ return int(stdout_data.split('\t', 1)[0])
diff --git a/site_utils/gs_offloader.py b/site_utils/gs_offloader.py
index 5a263e2..846eb6f 100755
--- a/site_utils/gs_offloader.py
+++ b/site_utils/gs_offloader.py
@@ -10,6 +10,7 @@
Upon successful copy, the local results directory is deleted.
"""
+import abc
import base64
import datetime
import errno
@@ -20,11 +21,11 @@
import os
import re
import shutil
-import signal
import socket
import stat
import subprocess
import sys
+import tarfile
import tempfile
import time
@@ -40,6 +41,9 @@
from autotest_lib.site_utils import pubsub_utils
from autotest_lib.tko import models
from autotest_lib.utils import labellib
+from autotest_lib.utils import gslib
+from chromite.lib import gs
+from chromite.lib import timeout_util
# Autotest requires the psutil module from site-packages, so it must be imported
# after "import common".
@@ -82,11 +86,6 @@
# Hosts sub-directory that contains cleanup, verify and repair jobs.
HOSTS_SUB_DIR = 'hosts'
-LOG_LOCATION = '/usr/local/autotest/logs/'
-LOG_FILENAME_FORMAT = 'gs_offloader_%s_log_%s.txt'
-LOG_TIMESTAMP_FORMAT = '%Y%m%d_%H%M%S'
-LOGGING_FORMAT = '%(asctime)s - %(levelname)s - %(message)s'
-
FAILED_OFFLOADS_FILE_HEADER = '''
This is the list of gs_offloader failed jobs.
Last offloader attempt at %s failed to offload %d files.
@@ -104,13 +103,6 @@
USE_RSYNC_ENABLED = global_config.global_config.get_config_value(
'CROS', 'gs_offloader_use_rsync', type=bool, default=False)
-# According to https://cloud.google.com/storage/docs/bucket-naming#objectnames
-INVALID_GS_CHARS = ['[', ']', '*', '?', '#']
-INVALID_GS_CHAR_RANGE = [(0x00, 0x1F), (0x7F, 0x84), (0x86, 0xFF)]
-
-# Maximum number of files in the folder.
-MAX_FILE_COUNT = 500
-FOLDERS_NEVER_ZIP = ['debug', 'ssp_logs', 'autoupdate_logs']
LIMIT_FILE_COUNT = global_config.global_config.get_config_value(
'CROS', 'gs_offloader_limit_file_count', type=bool, default=False)
@@ -134,13 +126,6 @@
'CROS', 'cloud_notification_topic', default=None)
-# Test upload pubsub notification attributes
-NOTIFICATION_ATTR_VERSION = 'version'
-NOTIFICATION_ATTR_GCS_URI = 'gcs_uri'
-NOTIFICATION_ATTR_MOBLAB_MAC = 'moblab_mac_address'
-NOTIFICATION_ATTR_MOBLAB_ID = 'moblab_id'
-NOTIFICATION_VERSION = '1'
-
# the message data for new test result notification.
NEW_TEST_RESULT_MESSAGE = 'NEW_TEST_RESULT'
@@ -148,24 +133,6 @@
GS_OFFLOADER_SUCCESS_TYPE = 'gs_offloader_success'
GS_OFFLOADER_FAILURE_TYPE = 'gs_offloader_failure'
-class TimeoutException(Exception):
- """Exception raised by the timeout_handler."""
- pass
-
-
-def timeout_handler(_signum, _frame):
- """Handler for SIGALRM when the offloading process times out.
-
- @param _signum: Signal number of the signal that was just caught.
- 14 for SIGALRM.
- @param _frame: Current stack frame.
-
- @raise TimeoutException: Automatically raises so that the time out
- is caught by the try/except surrounding the
- Popen call.
- """
- raise TimeoutException('Process Timed Out')
-
def _get_metrics_fields(dir_entry):
"""Get metrics fields for the given test result directory, including board
@@ -215,7 +182,7 @@
return fields
-def get_cmd_list(multiprocessing, dir_entry, gs_path):
+def _get_cmd_list(multiprocessing, dir_entry, gs_path):
"""Return the command to offload a specified directory.
@param multiprocessing: True to turn on -m option for gsutil.
@@ -239,68 +206,100 @@
return cmd
-def get_sanitized_name(name):
- """Get a string with all invalid characters in the name being replaced.
+def sanitize_dir(dirpath):
+ """Sanitize directory for gs upload.
- @param name: Name to be processed.
+ Symlinks and FIFOS are converted to regular files to fix bugs.
- @return A string with all invalid characters in the name being
- replaced.
+ @param dirpath: Directory entry to be sanitized.
"""
- match_pattern = ''.join([re.escape(c) for c in INVALID_GS_CHARS])
- match_pattern += ''.join([r'\x%02x-\x%02x' % (r[0], r[1])
- for r in INVALID_GS_CHAR_RANGE])
- invalid = re.compile('[%s]' % match_pattern)
- return invalid.sub(lambda x: '%%%02x' % ord(x.group(0)), name)
-
-
-def sanitize_dir(dir_entry):
- """Replace all invalid characters in folder and file names with valid ones.
-
- FIFOs are converted to regular files to prevent gsutil hangs (see crbug/684122).
- Symlinks are converted to regular files that store the link destination
- (crbug/692788).
-
- @param dir_entry: Directory entry to be sanitized.
- """
- if not os.path.exists(dir_entry):
+ if not os.path.exists(dirpath):
return
- renames = []
- fifos = []
- symlinks = []
- for root, dirs, files in os.walk(dir_entry):
- sanitized_root = get_sanitized_name(root)
- for name in dirs + files:
- sanitized_name = get_sanitized_name(name)
- sanitized_path = os.path.join(sanitized_root, sanitized_name)
- if name != sanitized_name:
- orig_path = os.path.join(sanitized_root, name)
- renames.append((orig_path, sanitized_path))
- current_path = os.path.join(root, name)
- file_stat = os.lstat(current_path)
+ _escape_rename(dirpath)
+ _escape_rename_dir_contents(dirpath)
+ _sanitize_fifos(dirpath)
+ _sanitize_symlinks(dirpath)
+
+
+def _escape_rename_dir_contents(dirpath):
+ """Recursively rename directory to escape filenames for gs upload.
+
+ @param dirpath: Directory path string.
+ """
+ for filename in os.listdir(dirpath):
+ path = os.path.join(dirpath, filename)
+ _escape_rename(path)
+ for filename in os.listdir(dirpath):
+ path = os.path.join(dirpath, filename)
+ if os.path.isdir(path):
+ _escape_rename_dir_contents(path)
+
+
+def _escape_rename(path):
+ """Rename file to escape filenames for gs upload.
+
+ @param path: File path string.
+ """
+ dirpath, filename = os.path.split(path)
+ sanitized_filename = gslib.escape(filename)
+ sanitized_path = os.path.join(dirpath, sanitized_filename)
+ os.rename(path, sanitized_path)
+
+
+def _sanitize_fifos(dirpath):
+ """Convert fifos to regular files (fixes crbug.com/684122).
+
+ @param dirpath: Directory path string.
+ """
+ for root, _, files in os.walk(dirpath):
+ for filename in files:
+ path = os.path.join(root, filename)
+ file_stat = os.lstat(path)
if stat.S_ISFIFO(file_stat.st_mode):
- # Replace fifos with markers
- fifos.append(sanitized_path)
- elif stat.S_ISLNK(file_stat.st_mode):
- # Replace symlinks with markers
- destination = os.readlink(current_path)
- symlinks.append((sanitized_path, destination))
- for src, dest in renames:
- logging.warning('Invalid character found. Renaming %s to %s.', src,
- dest)
- shutil.move(src, dest)
- for fifo in fifos:
- logging.debug('Removing fifo %s', fifo)
- os.remove(fifo)
- logging.debug('Creating marker %s', fifo)
- with open(fifo, 'a') as marker:
- marker.write('<FIFO>')
- for link, destination in symlinks:
- logging.debug('Removing symlink %s', link)
- os.remove(link)
- logging.debug('Creating marker %s', link)
- with open(link, 'w') as marker:
- marker.write('<symlink to %s>' % destination)
+ _replace_fifo_with_file(path)
+
+
+def _replace_fifo_with_file(path):
+ """Replace a fifo with a normal file.
+
+ @param path: Fifo path string.
+ """
+ logging.debug('Removing fifo %s', path)
+ os.remove(path)
+ logging.debug('Creating marker %s', path)
+ with open(path, 'w') as f:
+ f.write('<FIFO>')
+
+
+def _sanitize_symlinks(dirpath):
+ """Convert Symlinks to regular files (fixes crbug.com/692788).
+
+ @param dirpath: Directory path string.
+ """
+ for root, _, files in os.walk(dirpath):
+ for filename in files:
+ path = os.path.join(root, filename)
+ file_stat = os.lstat(path)
+ if stat.S_ISLNK(file_stat.st_mode):
+ _replace_symlink_with_file(path)
+
+
+def _replace_symlink_with_file(path):
+ """Replace a symlink with a normal file.
+
+ @param path: Symlink path string.
+ """
+ target = os.readlink(path)
+ logging.debug('Removing symlink %s', path)
+ os.remove(path)
+ logging.debug('Creating marker %s', path)
+ with open(path, 'w') as f:
+ f.write('<symlink to %s>' % target)
+
+
+# Maximum number of files in the folder.
+_MAX_FILE_COUNT = 500
+_FOLDERS_NEVER_ZIP = ['debug', 'ssp_logs', 'autoupdate_logs']
def _get_zippable_folders(dir_entry):
@@ -308,7 +307,7 @@
for folder in os.listdir(dir_entry):
folder_path = os.path.join(dir_entry, folder)
if (not os.path.isfile(folder_path) and
- not folder in FOLDERS_NEVER_ZIP):
+ not folder in _FOLDERS_NEVER_ZIP):
folders_list.append(folder_path)
return folders_list
@@ -317,20 +316,18 @@
"""Limit the number of files in given directory.
The method checks the total number of files in the given directory.
- If the number is greater than MAX_FILE_COUNT, the method will
+ If the number is greater than _MAX_FILE_COUNT, the method will
compress each folder in the given directory, except folders in
- FOLDERS_NEVER_ZIP.
+ _FOLDERS_NEVER_ZIP.
@param dir_entry: Directory entry to be checked.
"""
- count = utils.run('find "%s" | wc -l' % dir_entry,
- ignore_status=True).stdout.strip()
try:
- count = int(count)
- except (ValueError, TypeError):
+ count = _count_files(dir_entry)
+ except ValueError:
logging.warning('Fail to get the file count in folder %s.', dir_entry)
return
- if count < MAX_FILE_COUNT:
+ if count < _MAX_FILE_COUNT:
return
# For test job, zip folders in a second level, e.g. 123-debug/host1.
@@ -347,16 +344,26 @@
folders = subfolders
for folder in folders:
- try:
- zip_name = '%s.tgz' % folder
- utils.run('tar -cz -C "%s" -f "%s" "%s"' %
- (os.path.dirname(folder), zip_name,
- os.path.basename(folder)))
- except error.CmdError as e:
- logging.error('Fail to compress folder %s. Error: %s',
- folder, e)
- continue
- shutil.rmtree(folder)
+ _make_into_tarball(folder)
+
+
+def _count_files(dirpath):
+ """Count the number of files in a directory recursively.
+
+ @param dirpath: Directory path string.
+ """
+ return sum(len(files) for _path, _dirs, files in os.walk(dirpath))
+
+
+def _make_into_tarball(dirpath):
+ """Make directory into tarball.
+
+ @param dirpath: Directory path string.
+ """
+ tarpath = '%s.tgz' % dirpath
+ with tarfile.open(tarpath, 'w:gz') as tar:
+ tar.add(dirpath, arcname=os.path.basename(dirpath))
+ shutil.rmtree(dirpath)
def correct_results_folder_permission(dir_entry):
@@ -382,7 +389,7 @@
dir_entry, e)
-def upload_testresult_files(dir_entry, multiprocessing):
+def _upload_cts_testresult(dir_entry, multiprocessing):
"""Upload test results to separate gs buckets.
Upload testResult.xml.gz/test_result.xml.gz file to cts_results_bucket.
@@ -465,7 +472,7 @@
job_id + '_' + timestamp) + '/'
for zip_file in glob.glob(os.path.join('%s.zip' % path)):
- utils.run(' '.join(get_cmd_list(
+ utils.run(' '.join(_get_cmd_list(
multiprocessing, zip_file, cts_apfe_gs_path)))
logging.debug('Upload %s to %s ', zip_file, cts_apfe_gs_path)
@@ -476,7 +483,7 @@
with open(test_result_file, 'r') as f_in, (
gzip.open(test_result_file_gz, 'w')) as f_out:
shutil.copyfileobj(f_in, f_out)
- utils.run(' '.join(get_cmd_list(
+ utils.run(' '.join(_get_cmd_list(
multiprocessing, test_result_file_gz, test_result_gs_path)))
logging.debug('Zip and upload %s to %s',
test_result_file_gz, test_result_gs_path)
@@ -484,6 +491,14 @@
os.remove(test_result_file_gz)
+# Test upload pubsub notification attributes
+_NOTIFICATION_ATTR_VERSION = 'version'
+_NOTIFICATION_ATTR_GCS_URI = 'gcs_uri'
+_NOTIFICATION_ATTR_MOBLAB_MAC = 'moblab_mac_address'
+_NOTIFICATION_ATTR_MOBLAB_ID = 'moblab_id'
+_NOTIFICATION_VERSION = '1'
+
+
def _create_test_result_notification(gs_path, dir_entry):
"""Construct a test result notification.
@@ -497,11 +512,11 @@
data = base64.b64encode(NEW_TEST_RESULT_MESSAGE)
msg_payload = {'data': data}
msg_attributes = {}
- msg_attributes[NOTIFICATION_ATTR_GCS_URI] = gcs_uri
- msg_attributes[NOTIFICATION_ATTR_VERSION] = NOTIFICATION_VERSION
- msg_attributes[NOTIFICATION_ATTR_MOBLAB_MAC] = \
+ msg_attributes[_NOTIFICATION_ATTR_GCS_URI] = gcs_uri
+ msg_attributes[_NOTIFICATION_ATTR_VERSION] = _NOTIFICATION_VERSION
+ msg_attributes[_NOTIFICATION_ATTR_MOBLAB_MAC] = \
utils.get_default_interface_mac_address()
- msg_attributes[NOTIFICATION_ATTR_MOBLAB_ID] = utils.get_moblab_id()
+ msg_attributes[_NOTIFICATION_ATTR_MOBLAB_ID] = utils.get_moblab_id()
msg_payload['attributes'] = msg_attributes
return msg_payload
@@ -516,20 +531,43 @@
metrics.Counter(m_gs_returncode).increment(fields={'return_code': rcode})
-def get_offload_dir_func(gs_uri, multiprocessing, delete_age, pubsub_topic=None):
- """Returns the offload directory function for the given gs_uri
+class BaseGSOffloader(object):
- @param gs_uri: Google storage bucket uri to offload to.
- @param multiprocessing: True to turn on -m option for gsutil.
- @param pubsub_topic: The pubsub topic to publish notificaitons. If None,
- pubsub is not enabled.
+ """Google Storage offloader interface."""
- @return offload_dir function to perform the offload.
- """
+ __metaclass__ = abc.ABCMeta
+
+ @abc.abstractmethod
+ def offload(self, dir_entry, dest_path, job_complete_time):
+ """Offload a directory entry to Google Storage.
+
+ @param dir_entry: Directory entry to offload.
+ @param dest_path: Location in google storage where we will
+ offload the directory.
+ @param job_complete_time: The complete time of the job from the AFE
+ database.
+ """
+
+
+class GSOffloader(BaseGSOffloader):
+ """Google Storage Offloader."""
+
+ def __init__(self, gs_uri, multiprocessing, delete_age, pubsub_topic=None):
+ """Returns the offload directory function for the given gs_uri
+
+ @param gs_uri: Google storage bucket uri to offload to.
+ @param multiprocessing: True to turn on -m option for gsutil.
+ @param pubsub_topic: The pubsub topic to publish notificaitons. If None,
+ pubsub is not enabled.
+ """
+ self._gs_uri = gs_uri
+ self._multiprocessing = multiprocessing
+ self._delete_age = delete_age
+ self._pubsub_topic = pubsub_topic
@metrics.SecondsTimerDecorator(
'chromeos/autotest/gs_offloader/job_offload_duration')
- def offload_dir(dir_entry, dest_path, job_complete_time):
+ def offload(self, dir_entry, dest_path, job_complete_time):
"""Offload the specified directory entry to Google storage.
@param dir_entry: Directory entry to offload.
@@ -537,104 +575,20 @@
offload the directory.
@param job_complete_time: The complete time of the job from the AFE
database.
-
"""
- start_time = time.time()
- error = True
- stdout_file = None
- stderr_file = None
- metrics_fields = _get_metrics_fields(dir_entry)
- es_metadata = _get_es_metadata(dir_entry)
- try:
- if not _is_uploaded(dir_entry):
- sanitize_dir(dir_entry)
- if DEFAULT_CTS_RESULTS_GSURI:
- upload_testresult_files(dir_entry, multiprocessing)
-
- if LIMIT_FILE_COUNT:
- limit_file_count(dir_entry)
- dir_size = file_utils.get_directory_size_kibibytes(dir_entry)
- es_metadata['size_kb'] = dir_size
-
- stdout_file = tempfile.TemporaryFile('w+')
- stderr_file = tempfile.TemporaryFile('w+')
- process = None
- signal.alarm(OFFLOAD_TIMEOUT_SECS)
- gs_path = '%s%s' % (gs_uri, dest_path)
- process = subprocess.Popen(
- get_cmd_list(multiprocessing, dir_entry, gs_path),
- stdout=stdout_file, stderr=stderr_file)
- process.wait()
- signal.alarm(0)
-
- _emit_gs_returncode_metric(process.returncode)
- if process.returncode == 0:
- m_offload_count = (
- 'chromeos/autotest/gs_offloader/jobs_offloaded')
- metrics.Counter(m_offload_count).increment(
- fields=metrics_fields)
- m_offload_size = ('chromeos/autotest/gs_offloader/'
- 'kilobytes_transferred')
- metrics.Counter(m_offload_size).increment_by(
- dir_size, fields=metrics_fields)
- es_metadata['time_used_sec'] = time.time() - start_time
- autotest_es.post(use_http=True,
- type_str=GS_OFFLOADER_SUCCESS_TYPE,
- metadata=es_metadata)
- error = False
-
- if pubsub_topic:
- message = _create_test_result_notification(
- gs_path, dir_entry)
- pubsub_client = pubsub_utils.PubSubClient()
- msg_ids = pubsub_client.publish_notifications(
- pubsub_topic, [message])
- if not msg_ids:
- error = True
-
- if not error:
- _mark_uploaded(dir_entry)
-
- if _is_uploaded(dir_entry):
- if job_directories.is_job_expired(delete_age, job_complete_time):
- shutil.rmtree(dir_entry)
-
- except TimeoutException:
- m_timeout = 'chromeos/autotest/errors/gs_offloader/timed_out_count'
- metrics.Counter(m_timeout).increment(fields=metrics_fields)
- # If we finished the call to Popen(), we may need to
- # terminate the child process. We don't bother calling
- # process.poll(); that inherently races because the child
- # can die any time it wants.
- if process:
- try:
- process.terminate()
- except OSError:
- # We don't expect any error other than "No such
- # process".
- pass
- logging.error('Offloading %s timed out after waiting %d '
- 'seconds.', dir_entry, OFFLOAD_TIMEOUT_SECS)
- except OSError as e:
- # The wrong file permission can lead call
- # `shutil.rmtree(dir_entry)` to raise OSError with message
- # 'Permission denied'. Details can be found in
- # crbug.com/536151
- if e.errno == errno.EACCES:
- correct_results_folder_permission(dir_entry)
- m_permission_error = ('chromeos/autotest/errors/gs_offloader/'
- 'wrong_permissions_count')
- metrics.Counter(m_permission_error).increment(fields=metrics_fields)
- finally:
- signal.alarm(0)
- if error:
+ with tempfile.TemporaryFile('w+') as stdout_file, \
+ tempfile.TemporaryFile('w+') as stderr_file:
+ try:
+ self._offload(dir_entry, dest_path, stdout_file, stderr_file)
+ except _OffloadError as e:
+ metrics_fields = _get_metrics_fields(dir_entry)
m_any_error = 'chromeos/autotest/errors/gs_offloader/any_error'
metrics.Counter(m_any_error).increment(fields=metrics_fields)
- es_metadata['time_used_sec'] = time.time() - start_time
+ e.es_metadata['time_used_sec'] = time.time() - e.start_time
autotest_es.post(use_http=True,
type_str=GS_OFFLOADER_FAILURE_TYPE,
- metadata=es_metadata)
+ metadata=e.es_metadata)
# Rewind the log files for stdout and stderr and log
# their contents.
@@ -654,26 +608,160 @@
# correct_results_folder_permission can be deleted.
if 'CommandException: Error opening file' in stderr_content:
correct_results_folder_permission(dir_entry)
+ else:
+ self._prune(dir_entry, job_complete_time)
- if stdout_file:
- stdout_file.close()
- if stderr_file:
- stderr_file.close()
- return offload_dir
+ def _offload(self, dir_entry, dest_path,
+ stdout_file, stderr_file):
+ """Offload the specified directory entry to Google storage.
+
+ @param dir_entry: Directory entry to offload.
+ @param dest_path: Location in google storage where we will
+ offload the directory.
+ @param job_complete_time: The complete time of the job from the AFE
+ database.
+ @param stdout_file: Log file.
+ @param stderr_file: Log file.
+ """
+ if _is_uploaded(dir_entry):
+ return
+ start_time = time.time()
+ metrics_fields = _get_metrics_fields(dir_entry)
+ es_metadata = _get_es_metadata(dir_entry)
+ error_obj = _OffloadError(start_time, es_metadata)
+ try:
+ sanitize_dir(dir_entry)
+ if DEFAULT_CTS_RESULTS_GSURI:
+ _upload_cts_testresult(dir_entry, self._multiprocessing)
+
+ if LIMIT_FILE_COUNT:
+ limit_file_count(dir_entry)
+ es_metadata['size_kb'] = file_utils.get_directory_size_kibibytes(dir_entry)
+
+ process = None
+ with timeout_util.Timeout(OFFLOAD_TIMEOUT_SECS):
+ gs_path = '%s%s' % (self._gs_uri, dest_path)
+ process = subprocess.Popen(
+ _get_cmd_list(self._multiprocessing, dir_entry, gs_path),
+ stdout=stdout_file, stderr=stderr_file)
+ process.wait()
+
+ _emit_gs_returncode_metric(process.returncode)
+ if process.returncode != 0:
+ raise error_obj
+ _emit_offload_metrics(dir_entry)
+ es_metadata['time_used_sec'] = time.time() - start_time
+ autotest_es.post(use_http=True,
+ type_str=GS_OFFLOADER_SUCCESS_TYPE,
+ metadata=es_metadata)
+
+ if self._pubsub_topic:
+ message = _create_test_result_notification(
+ gs_path, dir_entry)
+ pubsub_client = pubsub_utils.PubSubClient()
+ msg_ids = pubsub_client.publish_notifications(
+ self._pubsub_topic, [message])
+ if not msg_ids:
+ raise error_obj
+ _mark_uploaded(dir_entry)
+ except timeout_util.TimeoutError:
+ m_timeout = 'chromeos/autotest/errors/gs_offloader/timed_out_count'
+ metrics.Counter(m_timeout).increment(fields=metrics_fields)
+ # If we finished the call to Popen(), we may need to
+ # terminate the child process. We don't bother calling
+ # process.poll(); that inherently races because the child
+ # can die any time it wants.
+ if process:
+ try:
+ process.terminate()
+ except OSError:
+ # We don't expect any error other than "No such
+ # process".
+ pass
+ logging.error('Offloading %s timed out after waiting %d '
+ 'seconds.', dir_entry, OFFLOAD_TIMEOUT_SECS)
+ raise error_obj
+
+ def _prune(self, dir_entry, job_complete_time):
+ """Prune directory if it is uploaded and expired.
+
+ @param dir_entry: Directory entry to offload.
+ @param job_complete_time: The complete time of the job from the AFE
+ database.
+ """
+ if not (_is_uploaded(dir_entry)
+ and job_directories.is_job_expired(self._delete_age,
+ job_complete_time)):
+ return
+ try:
+ shutil.rmtree(dir_entry)
+ except OSError as e:
+ # The wrong file permission can lead call
+ # `shutil.rmtree(dir_entry)` to raise OSError with message
+ # 'Permission denied'. Details can be found in
+ # crbug.com/536151
+ if e.errno == errno.EACCES:
+ correct_results_folder_permission(dir_entry)
+ m_permission_error = ('chromeos/autotest/errors/gs_offloader/'
+ 'wrong_permissions_count')
+ metrics_fields = _get_metrics_fields(dir_entry)
+ metrics.Counter(m_permission_error).increment(fields=metrics_fields)
-def delete_files(dir_entry, dest_path, job_complete_time):
- """Simply deletes the dir_entry from the filesystem.
+class _OffloadError(Exception):
+ """Google Storage offload failed."""
- Uses same arguments as offload_dir so that it can be used in replace
- of it on systems that only want to delete files instead of
- offloading them.
+ def __init__(self, start_time, es_metadata):
+ super(_OffloadError, self).__init__(start_time, es_metadata)
+ self.start_time = start_time
+ self.es_metadata = es_metadata
- @param dir_entry: Directory entry to offload.
- @param dest_path: NOT USED.
- @param job_complete_time: NOT USED.
+
+
+class FakeGSOffloader(BaseGSOffloader):
+
+ """Fake Google Storage Offloader that only deletes directories."""
+
+ def offload(self, dir_entry, dest_path, job_complete_time):
+ """Pretend to offload a directory and delete it.
+
+ @param dir_entry: Directory entry to offload.
+ @param dest_path: Location in google storage where we will
+ offload the directory.
+ @param job_complete_time: The complete time of the job from the AFE
+ database.
+ """
+ shutil.rmtree(dir_entry)
+
+
+def _is_expired(job, age_limit):
+ """Return whether job directory is expired for uploading
+
+ @param job: _JobDirectory instance.
+ @param age_limit: Minimum age in days at which a job may be offloaded.
"""
- shutil.rmtree(dir_entry)
+ job_timestamp = job.get_timestamp_if_finished()
+ if not job_timestamp:
+ return False
+ return job_directories.is_job_expired(age_limit, job_timestamp)
+
+
+def _emit_offload_metrics(dirpath):
+ """Emit gs offload metrics.
+
+ @param dirpath: Offloaded directory path.
+ """
+ dir_size = file_utils.get_directory_size_kibibytes(dirpath)
+ metrics_fields = _get_metrics_fields(dirpath)
+
+ m_offload_count = (
+ 'chromeos/autotest/gs_offloader/jobs_offloaded')
+ metrics.Counter(m_offload_count).increment(
+ fields=metrics_fields)
+ m_offload_size = ('chromeos/autotest/gs_offloader/'
+ 'kilobytes_transferred')
+ metrics.Counter(m_offload_size).increment_by(
+ dir_size, fields=metrics_fields)
def _is_uploaded(dirpath):
@@ -706,10 +794,10 @@
@param job: The _JobDirectory to format.
"""
- d = datetime.datetime.fromtimestamp(job.get_failure_time())
+ d = datetime.datetime.fromtimestamp(job.first_offload_start)
data = (d.strftime(FAILED_OFFLOADS_TIME_FORMAT),
- job.get_failure_count(),
- job.get_job_directory())
+ job.offload_count,
+ job.dirname)
return FAILED_OFFLOADS_LINE_FORMAT % data
@@ -721,7 +809,7 @@
# TODO (sbasi) Try to use the gsutil command to check write access.
# Ensure we have write access to gs_uri.
dummy_file = tempfile.NamedTemporaryFile()
- test_cmd = get_cmd_list(False, dummy_file.name, gs_uri)
+ test_cmd = _get_cmd_list(False, dummy_file.name, gs_uri)
while True:
try:
subprocess.check_call(test_cmd)
@@ -739,8 +827,7 @@
"""State of the offload process.
Contains the following member fields:
- * _offload_func: Function to call for each attempt to offload
- a job directory.
+ * _gs_offloader: _BaseGSOffloader to use to offload a job directory.
* _jobdir_classes: List of classes of job directory to be
offloaded.
* _processes: Maximum number of outstanding offload processes
@@ -756,7 +843,7 @@
self._upload_age_limit = options.age_to_upload
self._delete_age_limit = options.age_to_delete
if options.delete_only:
- self._offload_func = delete_files
+ self._gs_offloader = FakeGSOffloader()
else:
self.gs_uri = utils.get_offload_gsuri()
logging.debug('Offloading to: %s', self.gs_uri)
@@ -768,10 +855,10 @@
logging.info(
'Offloader multiprocessing is set to:%r', multiprocessing)
if options.pubsub_topic_for_job_upload:
- self._pubsub_topic = options.pubsub_topic_for_job_upload
+ self._pubsub_topic = options.pubsub_topic_for_job_upload
elif _PUBSUB_ENABLED:
- self._pubsub_topic = _PUBSUB_TOPIC
- self._offload_func = get_offload_dir_func(
+ self._pubsub_topic = _PUBSUB_TOPIC
+ self._gs_offloader = GSOffloader(
self.gs_uri, multiprocessing, self._delete_age_limit,
self._pubsub_topic)
classlist = []
@@ -797,7 +884,9 @@
new_job_count = 0
for cls in self._jobdir_classes:
for resultsdir in cls.get_job_directories():
- if resultsdir in self._open_jobs:
+ if (
+ resultsdir in self._open_jobs
+ or _is_uploaded(resultsdir)):
continue
self._open_jobs[resultsdir] = cls(resultsdir)
new_job_count += 1
@@ -809,7 +898,9 @@
"""Removed offloaded jobs from `self._open_jobs`."""
removed_job_count = 0
for jobkey, job in self._open_jobs.items():
- if job.is_offloaded():
+ if (
+ not os.path.exists(job.dirname)
+ or _is_uploaded(job.dirname)):
del self._open_jobs[jobkey]
removed_job_count += 1
logging.debug('End of offload cycle - cleared %d new jobs, '
@@ -817,22 +908,19 @@
removed_job_count, len(self._open_jobs))
- def _update_offload_results(self):
- """Check and report status after attempting offload.
+ def _report_failed_jobs(self):
+ """Report status after attempting offload.
This function processes all jobs in `self._open_jobs`, assuming
an attempt has just been made to offload all of them.
- Any jobs that have been successfully offloaded are removed.
-
If any jobs have reportable errors, and we haven't generated
an e-mail report in the last `REPORT_INTERVAL_SECS` seconds,
send new e-mail describing the failures.
"""
- self._remove_offloaded_jobs()
failed_jobs = [j for j in self._open_jobs.values() if
- j.get_failure_time()]
+ j.first_offload_start]
self._report_failed_jobs_count(failed_jobs)
self._log_failed_jobs_locally(failed_jobs)
@@ -853,11 +941,12 @@
self._add_new_jobs()
self._report_current_jobs_count()
with parallel.BackgroundTaskRunner(
- self._offload_func, processes=self._processes) as queue:
+ self._gs_offloader.offload, processes=self._processes) as queue:
for job in self._open_jobs.values():
- job.enqueue_offload(queue, self._upload_age_limit)
+ _enqueue_offload(job, queue, self._upload_age_limit)
self._give_up_on_jobs_over_limit()
- self._update_offload_results()
+ self._remove_offloaded_jobs()
+ self._report_failed_jobs()
def _give_up_on_jobs_over_limit(self):
@@ -907,6 +996,35 @@
len(failed_jobs))
+def _enqueue_offload(job, queue, age_limit):
+ """Enqueue the job for offload, if it's eligible.
+
+ The job is eligible for offloading if the database has marked
+ it finished, and the job is older than the `age_limit`
+ parameter.
+
+ If the job is eligible, offload processing is requested by
+ passing the `queue` parameter's `put()` method a sequence with
+ the job's `dirname` attribute and its directory name.
+
+ @param job _JobDirectory instance to offload.
+ @param queue If the job should be offloaded, put the offload
+ parameters into this queue for processing.
+ @param age_limit Minimum age for a job to be offloaded. A value
+ of 0 means that the job will be offloaded as
+ soon as it is finished.
+
+ """
+ if not job.offload_count:
+ if not _is_expired(job, age_limit):
+ return
+ job.first_offload_start = time.time()
+ job.offload_count += 1
+ if job.process_gs_instructions():
+ timestamp = job.get_timestamp_if_finished()
+ queue.put([job.dirname, os.path.dirname(job.dirname), timestamp])
+
+
def parse_options():
"""Parse the args passed into gs_offloader."""
defaults = 'Defaults:\n Destination: %s\n Results Path: %s' % (
@@ -992,21 +1110,7 @@
else:
offloader_type = 'jobs'
- log_timestamp = time.strftime(LOG_TIMESTAMP_FORMAT)
- if options.log_size > 0:
- log_timestamp = ''
- log_basename = LOG_FILENAME_FORMAT % (offloader_type, log_timestamp)
- log_filename = os.path.join(LOG_LOCATION, log_basename)
- log_formatter = logging.Formatter(LOGGING_FORMAT)
- # Replace the default logging handler with a RotatingFileHandler. If
- # options.log_size is 0, the file size will not be limited. Keeps
- # one backup just in case.
- handler = logging.handlers.RotatingFileHandler(
- log_filename, maxBytes=1024 * options.log_size, backupCount=1)
- handler.setFormatter(log_formatter)
- logger = logging.getLogger()
- logger.setLevel(logging.DEBUG)
- logger.addHandler(handler)
+ _setup_logging(options, offloader_type)
# Nice our process (carried to subprocesses) so we don't overload
# the system.
@@ -1023,8 +1127,6 @@
logging.debug('Offloading Autotest results in %s', RESULTS_DIR)
os.chdir(RESULTS_DIR)
- signal.signal(signal.SIGALRM, timeout_handler)
-
service_name = 'gs_offloader(%s)' % offloader_type
with ts_mon_config.SetupTsMonGlobalState(service_name, indirect=True,
short_lived=False):
@@ -1038,5 +1140,44 @@
time.sleep(SLEEP_TIME_SECS)
+_LOG_LOCATION = '/usr/local/autotest/logs/'
+_LOG_FILENAME_FORMAT = 'gs_offloader_%s_log_%s.txt'
+_LOG_TIMESTAMP_FORMAT = '%Y%m%d_%H%M%S'
+_LOGGING_FORMAT = '%(asctime)s - %(levelname)s - %(message)s'
+
+
+def _setup_logging(options, offloader_type):
+ """Set up logging.
+
+ @param options: Parsed options.
+ @param offloader_type: Type of offloader action as string.
+ """
+ log_filename = _get_log_filename(options, offloader_type)
+ log_formatter = logging.Formatter(_LOGGING_FORMAT)
+ # Replace the default logging handler with a RotatingFileHandler. If
+ # options.log_size is 0, the file size will not be limited. Keeps
+ # one backup just in case.
+ handler = logging.handlers.RotatingFileHandler(
+ log_filename, maxBytes=1024 * options.log_size, backupCount=1)
+ handler.setFormatter(log_formatter)
+ logger = logging.getLogger()
+ logger.setLevel(logging.DEBUG)
+ logger.addHandler(handler)
+
+
+def _get_log_filename(options, offloader_type):
+ """Get log filename.
+
+ @param options: Parsed options.
+ @param offloader_type: Type of offloader action as string.
+ """
+ if options.log_size > 0:
+ log_timestamp = ''
+ else:
+ log_timestamp = time.strftime(_LOG_TIMESTAMP_FORMAT)
+ log_basename = _LOG_FILENAME_FORMAT % (offloader_type, log_timestamp)
+ return os.path.join(_LOG_LOCATION, log_basename)
+
+
if __name__ == '__main__':
main()
diff --git a/site_utils/gs_offloader_unittest.py b/site_utils/gs_offloader_unittest.py
index 2b4a327..143d302 100755
--- a/site_utils/gs_offloader_unittest.py
+++ b/site_utils/gs_offloader_unittest.py
@@ -27,6 +27,8 @@
from autotest_lib.site_utils import gs_offloader
from autotest_lib.site_utils import job_directories
from autotest_lib.tko import models
+from autotest_lib.utils import gslib
+from chromite.lib import timeout_util
# Test value to use for `days_old`, if nothing else is required.
_TEST_EXPIRATION_AGE = 7
@@ -75,7 +77,7 @@
gs_offloader.GS_OFFLOADER_MULTIPROCESSING = False
- def _mock_get_offload_func(self, is_moblab, multiprocessing=False,
+ def _mock_get_sub_offloader(self, is_moblab, multiprocessing=False,
pubsub_topic=None, delete_age=0):
"""Mock the process of getting the offload_dir function."""
if is_moblab:
@@ -86,64 +88,64 @@
else:
expected_gsuri = utils.DEFAULT_OFFLOAD_GSURI
utils.get_offload_gsuri().AndReturn(expected_gsuri)
- offload_func = gs_offloader.get_offload_dir_func(expected_gsuri,
+ sub_offloader = gs_offloader.GSOffloader(expected_gsuri,
multiprocessing, delete_age, pubsub_topic)
- self.mox.StubOutWithMock(gs_offloader, 'get_offload_dir_func')
- gs_offloader.get_offload_dir_func(expected_gsuri, multiprocessing,
- delete_age, pubsub_topic).AndReturn(offload_func)
+ self.mox.StubOutWithMock(gs_offloader, 'GSOffloader')
+ gs_offloader.GSOffloader(expected_gsuri, multiprocessing,
+ delete_age, pubsub_topic).AndReturn(sub_offloader)
self.mox.ReplayAll()
- return offload_func
+ return sub_offloader
def test_process_no_options(self):
"""Test default offloader options."""
- offload_func = self._mock_get_offload_func(False)
+ sub_offloader = self._mock_get_sub_offloader(False)
offloader = gs_offloader.Offloader(_get_options([]))
self.assertEqual(set(offloader._jobdir_classes),
self._REGULAR_ONLY)
self.assertEqual(offloader._processes, 1)
- self.assertEqual(offloader._offload_func,
- offload_func)
+ self.assertEqual(offloader._gs_offloader,
+ sub_offloader)
self.assertEqual(offloader._upload_age_limit, 0)
self.assertEqual(offloader._delete_age_limit, 0)
def test_process_all_option(self):
"""Test offloader handling for the --all option."""
- offload_func = self._mock_get_offload_func(False)
+ sub_offloader = self._mock_get_sub_offloader(False)
offloader = gs_offloader.Offloader(_get_options(['--all']))
self.assertEqual(set(offloader._jobdir_classes), self._BOTH)
self.assertEqual(offloader._processes, 1)
- self.assertEqual(offloader._offload_func,
- offload_func)
+ self.assertEqual(offloader._gs_offloader,
+ sub_offloader)
self.assertEqual(offloader._upload_age_limit, 0)
self.assertEqual(offloader._delete_age_limit, 0)
def test_process_hosts_option(self):
"""Test offloader handling for the --hosts option."""
- offload_func = self._mock_get_offload_func(False)
+ sub_offloader = self._mock_get_sub_offloader(False)
offloader = gs_offloader.Offloader(
_get_options(['--hosts']))
self.assertEqual(set(offloader._jobdir_classes),
self._SPECIAL_ONLY)
self.assertEqual(offloader._processes, 1)
- self.assertEqual(offloader._offload_func,
- offload_func)
+ self.assertEqual(offloader._gs_offloader,
+ sub_offloader)
self.assertEqual(offloader._upload_age_limit, 0)
self.assertEqual(offloader._delete_age_limit, 0)
def test_parallelism_option(self):
"""Test offloader handling for the --parallelism option."""
- offload_func = self._mock_get_offload_func(False)
+ sub_offloader = self._mock_get_sub_offloader(False)
offloader = gs_offloader.Offloader(
_get_options(['--parallelism', '2']))
self.assertEqual(set(offloader._jobdir_classes),
self._REGULAR_ONLY)
self.assertEqual(offloader._processes, 2)
- self.assertEqual(offloader._offload_func,
- offload_func)
+ self.assertEqual(offloader._gs_offloader,
+ sub_offloader)
self.assertEqual(offloader._upload_age_limit, 0)
self.assertEqual(offloader._delete_age_limit, 0)
@@ -155,8 +157,8 @@
self.assertEqual(set(offloader._jobdir_classes),
self._REGULAR_ONLY)
self.assertEqual(offloader._processes, 1)
- self.assertEqual(offloader._offload_func,
- gs_offloader.delete_files)
+ self.assertIsInstance(offloader._gs_offloader,
+ gs_offloader.FakeGSOffloader)
self.assertEqual(offloader._upload_age_limit, 0)
self.assertEqual(offloader._delete_age_limit, 0)
self.assertIsNone(offloader._pubsub_topic)
@@ -164,27 +166,27 @@
def test_days_old_option(self):
"""Test offloader handling for the --days_old option."""
- offload_func = self._mock_get_offload_func(False, delete_age=7)
+ sub_offloader = self._mock_get_sub_offloader(False, delete_age=7)
offloader = gs_offloader.Offloader(
_get_options(['--days_old', '7']))
self.assertEqual(set(offloader._jobdir_classes),
self._REGULAR_ONLY)
self.assertEqual(offloader._processes, 1)
- self.assertEqual(offloader._offload_func,
- offload_func)
+ self.assertEqual(offloader._gs_offloader,
+ sub_offloader)
self.assertEqual(offloader._upload_age_limit, 7)
self.assertEqual(offloader._delete_age_limit, 7)
def test_moblab_gsuri_generation(self):
"""Test offloader construction for Moblab."""
- offload_func = self._mock_get_offload_func(True)
+ sub_offloader = self._mock_get_sub_offloader(True)
offloader = gs_offloader.Offloader(_get_options([]))
self.assertEqual(set(offloader._jobdir_classes),
self._REGULAR_ONLY)
self.assertEqual(offloader._processes, 1)
- self.assertEqual(offloader._offload_func,
- offload_func)
+ self.assertEqual(offloader._gs_offloader,
+ sub_offloader)
self.assertEqual(offloader._upload_age_limit, 0)
self.assertEqual(offloader._delete_age_limit, 0)
@@ -194,49 +196,49 @@
gs_offloader.GS_OFFLOADING_ENABLED = False
offloader = gs_offloader.Offloader(
_get_options([]))
- self.assertEqual(offloader._offload_func,
- gs_offloader.delete_files)
+ self.assertIsInstance(offloader._gs_offloader,
+ gs_offloader.FakeGSOffloader)
def test_offloader_multiprocessing_flag_set(self):
"""Test multiprocessing is set."""
- offload_func = self._mock_get_offload_func(True, True)
+ sub_offloader = self._mock_get_sub_offloader(True, True)
offloader = gs_offloader.Offloader(_get_options(['-m']))
- self.assertEqual(offloader._offload_func,
- offload_func)
+ self.assertEqual(offloader._gs_offloader,
+ sub_offloader)
self.mox.VerifyAll()
def test_offloader_multiprocessing_flag_not_set_default_false(self):
"""Test multiprocessing is set."""
gs_offloader.GS_OFFLOADER_MULTIPROCESSING = False
- offload_func = self._mock_get_offload_func(True, False)
+ sub_offloader = self._mock_get_sub_offloader(True, False)
offloader = gs_offloader.Offloader(_get_options([]))
- self.assertEqual(offloader._offload_func,
- offload_func)
+ self.assertEqual(offloader._gs_offloader,
+ sub_offloader)
self.mox.VerifyAll()
def test_offloader_multiprocessing_flag_not_set_default_true(self):
"""Test multiprocessing is set."""
gs_offloader.GS_OFFLOADER_MULTIPROCESSING = True
- offload_func = self._mock_get_offload_func(True, True)
+ sub_offloader = self._mock_get_sub_offloader(True, True)
offloader = gs_offloader.Offloader(_get_options([]))
- self.assertEqual(offloader._offload_func,
- offload_func)
+ self.assertEqual(offloader._gs_offloader,
+ sub_offloader)
self.mox.VerifyAll()
def test_offloader_pubsub_topic_not_set(self):
"""Test multiprocessing is set."""
- offload_func = self._mock_get_offload_func(True, False)
+ sub_offloader = self._mock_get_sub_offloader(True, False)
offloader = gs_offloader.Offloader(_get_options([]))
- self.assertEqual(offloader._offload_func,
- offload_func)
+ self.assertEqual(offloader._gs_offloader,
+ sub_offloader)
self.mox.VerifyAll()
def test_offloader_pubsub_topic_set(self):
"""Test multiprocessing is set."""
- offload_func = self._mock_get_offload_func(True, False, 'test-topic')
+ sub_offloader = self._mock_get_sub_offloader(True, False, 'test-topic')
offloader = gs_offloader.Offloader(_get_options(['-t', 'test-topic']))
- self.assertEqual(offloader._offload_func,
- offload_func)
+ self.assertEqual(offloader._gs_offloader,
+ sub_offloader)
self.mox.VerifyAll()
@@ -336,23 +338,23 @@
def set_incomplete(self):
"""Make this job appear to have failed offload just once."""
- self._offload_count += 1
- self._first_offload_start = time.time()
- if not os.path.isdir(self._dirname):
- os.mkdir(self._dirname)
+ self.offload_count += 1
+ self.first_offload_start = time.time()
+ if not os.path.isdir(self.dirname):
+ os.mkdir(self.dirname)
def set_reportable(self):
"""Make this job be reportable."""
self.set_incomplete()
- self._offload_count += 1
+ self.offload_count += 1
def set_complete(self):
"""Make this job be completed."""
- self._offload_count += 1
- if os.path.isdir(self._dirname):
- os.rmdir(self._dirname)
+ self.offload_count += 1
+ if os.path.isdir(self.dirname):
+ os.rmdir(self.dirname)
def process_gs_instructions(self):
@@ -361,10 +363,10 @@
class CommandListTests(unittest.TestCase):
- """Tests for `get_cmd_list()`."""
+ """Tests for `_get_cmd_list()`."""
def _command_list_assertions(self, job, use_rsync=True, multi=False):
- """Call `get_cmd_list()` and check the return value.
+ """Call `_get_cmd_list()` and check the return value.
Check the following assertions:
* The command name (argv[0]) is 'gsutil'.
@@ -376,7 +378,7 @@
'queue_args[1]'.
@param job A job with properly calculated arguments to
- `get_cmd_list()`
+ `_get_cmd_list()`
@param use_rsync True when using 'rsync'. False when using 'cp'.
@param multi True when using '-m' option for gsutil.
@@ -385,7 +387,7 @@
gs_offloader.USE_RSYNC_ENABLED = use_rsync
- command = gs_offloader.get_cmd_list(
+ command = gs_offloader._get_cmd_list(
multi, job.queue_args[0],
os.path.join(test_bucket_uri, job.queue_args[1]))
@@ -404,38 +406,38 @@
os.path.join(test_bucket_uri, job.queue_args[1]))
- def test_get_cmd_list_regular(self):
- """Test `get_cmd_list()` as for a regular job."""
+ def test__get_cmd_list_regular(self):
+ """Test `_get_cmd_list()` as for a regular job."""
job = _MockJobDirectory('118-debug')
self._command_list_assertions(job)
- def test_get_cmd_list_special(self):
- """Test `get_cmd_list()` as for a special job."""
+ def test__get_cmd_list_special(self):
+ """Test `_get_cmd_list()` as for a special job."""
job = _MockJobDirectory('hosts/host1/118-reset')
self._command_list_assertions(job)
def test_get_cmd_list_regular_no_rsync(self):
- """Test `get_cmd_list()` as for a regular job."""
+ """Test `_get_cmd_list()` as for a regular job."""
job = _MockJobDirectory('118-debug')
self._command_list_assertions(job, use_rsync=False)
def test_get_cmd_list_special_no_rsync(self):
- """Test `get_cmd_list()` as for a special job."""
+ """Test `_get_cmd_list()` as for a special job."""
job = _MockJobDirectory('hosts/host1/118-reset')
self._command_list_assertions(job, use_rsync=False)
def test_get_cmd_list_regular_multi(self):
- """Test `get_cmd_list()` as for a regular job with True multi."""
+ """Test `_get_cmd_list()` as for a regular job with True multi."""
job = _MockJobDirectory('118-debug')
self._command_list_assertions(job, multi=True)
- def test_get_cmd_list_special_multi(self):
- """Test `get_cmd_list()` as for a special job with True multi."""
+ def test__get_cmd_list_special_multi(self):
+ """Test `_get_cmd_list()` as for a special job with True multi."""
job = _MockJobDirectory('hosts/host1/118-reset')
self._command_list_assertions(job, multi=True)
@@ -458,17 +460,17 @@
self.assertEquals(base64.b64encode(
gs_offloader.NEW_TEST_RESULT_MESSAGE), msg['data'])
self.assertEquals(
- gs_offloader.NOTIFICATION_VERSION,
- msg['attributes'][gs_offloader.NOTIFICATION_ATTR_VERSION])
+ gs_offloader._NOTIFICATION_VERSION,
+ msg['attributes'][gs_offloader._NOTIFICATION_ATTR_VERSION])
self.assertEquals(
'1c:dc:d1:11:01:e1',
- msg['attributes'][gs_offloader.NOTIFICATION_ATTR_MOBLAB_MAC])
+ msg['attributes'][gs_offloader._NOTIFICATION_ATTR_MOBLAB_MAC])
self.assertEquals(
'c8386d92-9ad1-11e6-80f5-111111111111',
- msg['attributes'][gs_offloader.NOTIFICATION_ATTR_MOBLAB_ID])
+ msg['attributes'][gs_offloader._NOTIFICATION_ATTR_MOBLAB_ID])
self.assertEquals(
'gs://test_bucket/123-moblab',
- msg['attributes'][gs_offloader.NOTIFICATION_ATTR_GCS_URI])
+ msg['attributes'][gs_offloader._NOTIFICATION_ATTR_GCS_URI])
self.mox.VerifyAll()
@@ -510,27 +512,27 @@
def test_regular_job_fields(self):
"""Test the constructor for `RegularJobDirectory`.
- Construct a regular job, and assert that the `_dirname`
+ Construct a regular job, and assert that the `dirname`
and `_id` attributes are set as expected.
"""
resultsdir = '118-fubar'
job = job_directories.RegularJobDirectory(resultsdir)
- self.assertEqual(job._dirname, resultsdir)
+ self.assertEqual(job.dirname, resultsdir)
self.assertEqual(job._id, 118)
def test_special_job_fields(self):
"""Test the constructor for `SpecialJobDirectory`.
- Construct a special job, and assert that the `_dirname`
+ Construct a special job, and assert that the `dirname`
and `_id` attributes are set as expected.
"""
destdir = 'hosts/host1'
resultsdir = destdir + '/118-reset'
job = job_directories.SpecialJobDirectory(resultsdir)
- self.assertEqual(job._dirname, resultsdir)
+ self.assertEqual(job.dirname, resultsdir)
self.assertEqual(job._id, 118)
@@ -670,8 +672,8 @@
self.mox.VerifyAll()
-class _TempResultsDirTestBase(mox.MoxTestBase):
- """Base class for tests using a temporary results directory."""
+class _TempResultsDirTestCase(unittest.TestCase):
+ """Mixin class for tests using a temporary results directory."""
REGULAR_JOBLIST = [
'111-fubar', '112-fubar', '113-fubar', '114-snafu']
@@ -682,7 +684,7 @@
def setUp(self):
- super(_TempResultsDirTestBase, self).setUp()
+ super(_TempResultsDirTestCase, self).setUp()
self._resultsroot = tempfile.mkdtemp()
self._cwd = os.getcwd()
os.chdir(self._resultsroot)
@@ -691,7 +693,7 @@
def tearDown(self):
os.chdir(self._cwd)
shutil.rmtree(self._resultsroot)
- super(_TempResultsDirTestBase, self).tearDown()
+ super(_TempResultsDirTestCase, self).tearDown()
def make_job(self, jobdir):
@@ -723,6 +725,10 @@
os.mkdir(d)
+class _TempResultsDirTestBase(_TempResultsDirTestCase, mox.MoxTestBase):
+ """Base Mox test class for tests using a temporary results directory."""
+
+
class FailedOffloadsLogTest(_TempResultsDirTestBase):
"""Test the formatting of failed offloads log file."""
# Below is partial sample of a failed offload log file. This text is
@@ -750,10 +756,10 @@
for line in self._SAMPLE_DIRECTORIES_REPORT.split('\n')[1 : -1]:
date_, time_, count, dir_ = line.split()
job = _MockJobDirectory(dir_)
- job._offload_count = int(count)
+ job.offload_count = int(count)
timestruct = time.strptime("%s %s" % (date_, time_),
gs_offloader.FAILED_OFFLOADS_TIME_FORMAT)
- job._first_offload_start = time.mktime(timestruct)
+ job.first_offload_start = time.mktime(timestruct)
# enter the jobs in reverse order, to make sure we
# test that the output will be sorted.
self._joblist.insert(0, job)
@@ -805,8 +811,10 @@
self._saved_loglevel = logging.getLogger().getEffectiveLevel()
logging.getLogger().setLevel(logging.CRITICAL+1)
self._job = self.make_job(self.REGULAR_JOBLIST[0])
- self.mox.StubOutWithMock(gs_offloader, 'get_cmd_list')
- self.mox.StubOutWithMock(signal, 'alarm')
+ self.mox.StubOutWithMock(gs_offloader, '_get_cmd_list')
+ alarm = mock.patch('signal.alarm', return_value=0)
+ alarm.start()
+ self.addCleanup(alarm.stop)
self.mox.StubOutWithMock(models.test, 'parse_job_keyval')
@@ -814,9 +822,9 @@
logging.getLogger().setLevel(self._saved_loglevel)
super(OffloadDirectoryTests, self).tearDown()
- def _mock_upload_testresult_files(self):
- self.mox.StubOutWithMock(gs_offloader, 'upload_testresult_files')
- gs_offloader.upload_testresult_files(
+ def _mock__upload_cts_testresult(self):
+ self.mox.StubOutWithMock(gs_offloader, '_upload_cts_testresult')
+ gs_offloader._upload_cts_testresult(
mox.IgnoreArg(),mox.IgnoreArg()).AndReturn(None)
def _mock_create_marker_file(self):
@@ -825,28 +833,23 @@
def _mock_offload_dir_calls(self, command, queue_args,
- marker_initially_exists=False,
- marker_eventually_exists=True):
+ marker_initially_exists=False):
"""Mock out the calls needed by `offload_dir()`.
This covers only the calls made when there is no timeout.
@param command Command list to be returned by the mocked
- call to `get_cmd_list()`.
+ call to `_get_cmd_list()`.
"""
self.mox.StubOutWithMock(os.path, 'isfile')
os.path.isfile(mox.IgnoreArg()).AndReturn(marker_initially_exists)
- signal.alarm(gs_offloader.OFFLOAD_TIMEOUT_SECS)
command.append(queue_args[0])
- gs_offloader.get_cmd_list(
+ gs_offloader._get_cmd_list(
False, queue_args[0],
'%s%s' % (utils.DEFAULT_OFFLOAD_GSURI,
queue_args[1])).AndReturn(command)
- self._mock_upload_testresult_files()
- signal.alarm(0)
- signal.alarm(0)
- os.path.isfile(mox.IgnoreArg()).AndReturn(marker_eventually_exists)
+ self._mock__upload_cts_testresult()
def _run_offload_dir(self, should_succeed, delete_age):
@@ -860,8 +863,8 @@
"""
self.mox.ReplayAll()
- gs_offloader.get_offload_dir_func(
- utils.DEFAULT_OFFLOAD_GSURI, False, delete_age)(
+ gs_offloader.GSOffloader(
+ utils.DEFAULT_OFFLOAD_GSURI, False, delete_age).offload(
self._job.queue_args[0],
self._job.queue_args[1],
self._job.queue_args[2])
@@ -874,6 +877,7 @@
"""Test that `offload_dir()` can succeed correctly."""
self._mock_offload_dir_calls(['test', '-d'],
self._job.queue_args)
+ os.path.isfile(mox.IgnoreArg()).AndReturn(True)
self._mock_create_marker_file()
self._run_offload_dir(True, 0)
@@ -881,40 +885,7 @@
def test_offload_failure(self):
"""Test that `offload_dir()` can fail correctly."""
self._mock_offload_dir_calls(['test', '!', '-d'],
- self._job.queue_args,
- marker_eventually_exists=False)
- self._run_offload_dir(False, 0)
-
-
- def test_offload_timeout_early(self):
- """Test that `offload_dir()` times out correctly.
-
- This test triggers timeout at the earliest possible moment,
- at the first call to set the timeout alarm.
-
- """
- self._mock_upload_testresult_files()
- signal.alarm(gs_offloader.OFFLOAD_TIMEOUT_SECS).AndRaise(
- gs_offloader.TimeoutException('fubar'))
- signal.alarm(0)
- self._run_offload_dir(False, 0)
-
-
- def test_offload_timeout_late(self):
- """Test that `offload_dir()` times out correctly.
-
- This test triggers timeout at the latest possible moment, at
- the call to clear the timeout alarm.
-
- """
- signal.alarm(gs_offloader.OFFLOAD_TIMEOUT_SECS)
- gs_offloader.get_cmd_list(
- False, mox.IgnoreArg(), mox.IgnoreArg()).AndReturn(
- ['test', '-d', self._job.queue_args[0]])
- self._mock_upload_testresult_files()
- signal.alarm(0).AndRaise(
- gs_offloader.TimeoutException('fubar'))
- signal.alarm(0)
+ self._job.queue_args)
self._run_offload_dir(False, 0)
@@ -922,7 +893,7 @@
"""Test that folder/file name with invalid character can be corrected.
"""
results_folder = tempfile.mkdtemp()
- invalid_chars = '_'.join(gs_offloader.INVALID_GS_CHARS)
+ invalid_chars = '_'.join(['[', ']', '*', '?', '#'])
invalid_files = []
invalid_folder_name = 'invalid_name_folder_%s' % invalid_chars
invalid_folder = os.path.join(
@@ -931,13 +902,6 @@
invalid_files.append(os.path.join(
invalid_folder,
'invalid_name_file_%s' % invalid_chars))
- for r in gs_offloader.INVALID_GS_CHAR_RANGE:
- for c in range(r[0], r[1]+1):
- # NULL cannot be in file name.
- if c != 0:
- invalid_files.append(os.path.join(
- invalid_folder,
- 'invalid_name_file_%s' % chr(c)))
good_folder = os.path.join(results_folder, 'valid_name_folder')
good_file = os.path.join(good_folder, 'valid_name_file')
for folder in [invalid_folder, good_folder]:
@@ -961,11 +925,9 @@
gs_offloader.sanitize_dir(results_folder)
for _, dirs, files in os.walk(results_folder):
for name in dirs + files:
- self.assertEqual(name, gs_offloader.get_sanitized_name(name))
+ self.assertEqual(name, gslib.escape(name))
for c in name:
- self.assertFalse(c in gs_offloader.INVALID_GS_CHARS)
- for r in gs_offloader.INVALID_GS_CHAR_RANGE:
- self.assertFalse(ord(c) >= r[0] and ord(c) <= r[1])
+ self.assertFalse(c in ['[', ']', '*', '?', '#'])
self.assertTrue(os.path.exists(good_file))
self.assertTrue(os.path.exists(fifo1))
@@ -973,8 +935,7 @@
self.assertTrue(os.path.exists(fifo2))
self.assertFalse(is_fifo(fifo2))
corrected_folder = os.path.join(
- results_folder,
- gs_offloader.get_sanitized_name(invalid_folder_name))
+ results_folder, gslib.escape(invalid_folder_name))
corrected_fifo3 = os.path.join(
corrected_folder,
'test_fifo3')
@@ -982,8 +943,7 @@
self.assertTrue(os.path.exists(corrected_fifo3))
self.assertFalse(is_fifo(corrected_fifo3))
corrected_fifo4 = os.path.join(
- corrected_folder,
- gs_offloader.get_sanitized_name(invalid_fifo4_name))
+ corrected_folder, gslib.escape(invalid_fifo4_name))
self.assertFalse(os.path.exists(fifo4))
self.assertTrue(os.path.exists(corrected_fifo4))
self.assertFalse(is_fifo(corrected_fifo4))
@@ -1015,12 +975,12 @@
with open(os.path.join(folder, str(i)), 'w') as f:
f.write('test')
- gs_offloader.MAX_FILE_COUNT = 100
+ gs_offloader._MAX_FILE_COUNT = 100
gs_offloader.limit_file_count(
results_folder if is_test_job else host_folder)
self.assertTrue(os.path.exists(sysinfo_folder))
- gs_offloader.MAX_FILE_COUNT = 10
+ gs_offloader._MAX_FILE_COUNT = 10
gs_offloader.limit_file_count(
results_folder if is_test_job else host_folder)
self.assertFalse(os.path.exists(sysinfo_folder))
@@ -1120,8 +1080,8 @@
return (results_folder, host_folder, path_pattern_pair)
- def test_upload_testresult_files(self):
- """Test upload_testresult_files."""
+ def test__upload_cts_testresult(self):
+ """Test _upload_cts_testresult."""
results_folder, host_folder, path_pattern_pair = self.create_results_folder()
self.mox.StubOutWithMock(gs_offloader, '_upload_files')
@@ -1136,7 +1096,7 @@
['test', '-d', host_folder])
self.mox.ReplayAll()
- gs_offloader.upload_testresult_files(results_folder, False)
+ gs_offloader._upload_cts_testresult(results_folder, False)
self.mox.VerifyAll()
shutil.rmtree(results_folder)
@@ -1152,10 +1112,10 @@
'suite': 'arc-cts'
})
- gs_offloader.get_cmd_list(
+ gs_offloader._get_cmd_list(
False, mox.IgnoreArg(), mox.IgnoreArg()).AndReturn(
['test', '-d', path])
- gs_offloader.get_cmd_list(
+ gs_offloader._get_cmd_list(
False, mox.IgnoreArg(), mox.IgnoreArg()).AndReturn(
['test', '-d', path])
@@ -1200,16 +1160,16 @@
`logging.debug()`, regardless of whether the job was
enqueued. Nothing else is allowed to be logged.
B. If the job is not eligible to be offloaded,
- `get_failure_time()` and `get_failure_count()` are 0.
+ `first_offload_start` and `offload_count` are 0.
C. If the job is not eligible for offload, nothing is
enqueued in `queue`.
- D. When the job is offloaded, `get_failure_count()` increments
+ D. When the job is offloaded, `offload_count` increments
each time.
E. When the job is offloaded, the appropriate parameters are
enqueued exactly once.
- F. The first time a job is offloaded, `get_failure_time()` is
+ F. The first time a job is offloaded, `first_offload_start` is
set to the current time.
- G. `get_failure_time()` only changes the first time that the
+ G. `first_offload_start` only changes the first time that the
job is offloaded.
The test cases below are designed to exercise all of the
@@ -1230,13 +1190,13 @@
`enqueue_offload()` has no effect.
"""
- self.assertEqual(self._job.get_failure_count(), 0)
- self.assertEqual(self._job.get_failure_time(), 0)
- self._job.enqueue_offload(self._queue, days_old)
- self._job.enqueue_offload(self._queue, days_old)
+ self.assertEqual(self._job.offload_count, 0)
+ self.assertEqual(self._job.first_offload_start, 0)
+ gs_offloader._enqueue_offload(self._job, self._queue, days_old)
+ gs_offloader._enqueue_offload(self._job, self._queue, days_old)
self.assertTrue(self._queue.empty())
- self.assertEqual(self._job.get_failure_count(), 0)
- self.assertEqual(self._job.get_failure_time(), 0)
+ self.assertEqual(self._job.offload_count, 0)
+ self.assertEqual(self._job.first_offload_start, 0)
def _offload_expired_once(self, days_old, count):
@@ -1246,8 +1206,8 @@
expected when a job is offloaded.
"""
- self._job.enqueue_offload(self._queue, days_old)
- self.assertEqual(self._job.get_failure_count(), count)
+ gs_offloader._enqueue_offload(self._job, self._queue, days_old)
+ self.assertEqual(self._job.offload_count, count)
self.assertFalse(self._queue.empty())
v = self._queue.get_nowait()
self.assertTrue(self._queue.empty())
@@ -1258,18 +1218,18 @@
"""Make calls to `enqueue_offload()` for a just-expired job.
This method directly tests assertions F and G regarding
- side-effects on `get_failure_time()`.
+ side-effects on `first_offload_start`.
"""
t0 = time.time()
self._offload_expired_once(days_old, 1)
- t1 = self._job.get_failure_time()
+ t1 = self._job.first_offload_start
self.assertLessEqual(t1, time.time())
self.assertGreaterEqual(t1, t0)
self._offload_expired_once(days_old, 2)
- self.assertEqual(self._job.get_failure_time(), t1)
+ self.assertEqual(self._job.first_offload_start, t1)
self._offload_expired_once(days_old, 3)
- self.assertEqual(self._job.get_failure_time(), t1)
+ self.assertEqual(self._job.first_offload_start, t1)
def test_case_1_no_expiration(self):
@@ -1411,7 +1371,7 @@
self.assertEqual(expected_key_set,
set(self._offloader._open_jobs.keys()))
for jobkey, job in self._offloader._open_jobs.items():
- self.assertEqual(jobkey, job._dirname)
+ self.assertEqual(jobkey, job.dirname)
self.mox.VerifyAll()
self.mox.ResetAll()
@@ -1448,67 +1408,8 @@
self._offloader._open_jobs[key])
-class JobStateTests(_TempResultsDirTestBase):
- """Tests for job state predicates.
-
- This tests for the expected results from the
- `is_offloaded()` predicate method.
-
- """
-
- def test_unfinished_job(self):
- """Test that an unfinished job reports the correct state.
-
- A job is "unfinished" if it isn't marked complete in the
- database. A job in this state is neither "complete" nor
- "reportable".
-
- """
- job = self.make_job(self.REGULAR_JOBLIST[0])
- self.assertFalse(job.is_offloaded())
-
-
- def test_incomplete_job(self):
- """Test that an incomplete job reports the correct state.
-
- A job is "incomplete" if exactly one attempt has been made
- to offload the job, but its results directory still exists.
- A job in this state is neither "complete" nor "reportable".
-
- """
- job = self.make_job(self.REGULAR_JOBLIST[0])
- job.set_incomplete()
- self.assertFalse(job.is_offloaded())
-
-
- def test_reportable_job(self):
- """Test that a reportable job reports the correct state.
-
- A job is "reportable" if more than one attempt has been made
- to offload the job, and its results directory still exists.
- A job in this state is "reportable", but not "complete".
-
- """
- job = self.make_job(self.REGULAR_JOBLIST[0])
- job.set_reportable()
- self.assertFalse(job.is_offloaded())
-
-
- def test_completed_job(self):
- """Test that a completed job reports the correct state.
-
- A job is "completed" if at least one attempt has been made
- to offload the job, and its results directory still exists.
- A job in this state is "complete", and not "reportable".
-
- """
- job = self.make_job(self.REGULAR_JOBLIST[0])
- job.set_complete()
- self.assertTrue(job.is_offloaded())
-
-
class ReportingTests(_TempResultsDirTestBase):
- """Tests for `Offloader._update_offload_results()`."""
+ """Tests for `Offloader._report_failed_jobs()`."""
def setUp(self):
super(ReportingTests, self).setUp()
@@ -1520,14 +1421,14 @@
def _add_job(self, jobdir):
"""Add a job to the dictionary of unfinished jobs."""
j = self.make_job(jobdir)
- self._offloader._open_jobs[j._dirname] = j
+ self._offloader._open_jobs[j.dirname] = j
return j
def _expect_log_message(self, new_open_jobs, with_failures):
"""Mock expected logging calls.
- `_update_offload_results()` logs one message with the number
+ `_report_failed_jobs()` logs one message with the number
of jobs removed from the open job set and the number of jobs
still remaining. Additionally, if there are reportable
jobs, then it logs the number of jobs that haven't yet
@@ -1551,10 +1452,10 @@
def _run_update(self, new_open_jobs):
- """Call `_update_offload_results()`.
+ """Call `_report_failed_jobs()`.
Initial conditions are set up by the caller. This calls
- `_update_offload_results()` once, and then checks these
+ `_report_failed_jobs()` once, and then checks these
assertions:
* The offloader's new `_open_jobs` field contains only
the entries in `new_open_jobs`.
@@ -1564,7 +1465,8 @@
`_open_jobs` field.
"""
self.mox.ReplayAll()
- self._offloader._update_offload_results()
+ self._offloader._report_failed_jobs()
+ self._offloader._remove_offloaded_jobs()
self.assertEqual(self._offloader._open_jobs, new_open_jobs)
self.mox.VerifyAll()
self.mox.ResetAll()
@@ -1583,7 +1485,7 @@
def test_no_jobs(self):
- """Test `_update_offload_results()` with no open jobs.
+ """Test `_report_failed_jobs()` with no open jobs.
Initial conditions are an empty `_open_jobs` list.
Expected result is an empty `_open_jobs` list.
@@ -1595,7 +1497,7 @@
def test_all_completed(self):
- """Test `_update_offload_results()` with only complete jobs.
+ """Test `_report_failed_jobs()` with only complete jobs.
Initial conditions are an `_open_jobs` list consisting of only completed
jobs.
@@ -1610,7 +1512,7 @@
def test_none_finished(self):
- """Test `_update_offload_results()` with only unfinished jobs.
+ """Test `_report_failed_jobs()` with only unfinished jobs.
Initial conditions are an `_open_jobs` list consisting of only
unfinished jobs.
@@ -1625,5 +1527,69 @@
self._run_update(new_jobs)
+class GsOffloaderMockTests(_TempResultsDirTestCase):
+ """Tests using mock instead of mox."""
+
+ def setUp(self):
+ super(GsOffloaderMockTests, self).setUp()
+ alarm = mock.patch('signal.alarm', return_value=0)
+ alarm.start()
+ self.addCleanup(alarm.stop)
+
+ self._saved_loglevel = logging.getLogger().getEffectiveLevel()
+ logging.getLogger().setLevel(logging.CRITICAL + 1)
+
+ self._job = self.make_job(self.REGULAR_JOBLIST[0])
+
+
+ def test_offload_timeout_early(self):
+ """Test that `offload_dir()` times out correctly.
+
+ This test triggers timeout at the earliest possible moment,
+ at the first call to set the timeout alarm.
+
+ """
+ signal.alarm.side_effect = [0, timeout_util.TimeoutError('fubar')]
+ with mock.patch.object(gs_offloader, '_upload_cts_testresult',
+ autospec=True) as upload:
+ upload.return_value = None
+ gs_offloader.GSOffloader(
+ utils.DEFAULT_OFFLOAD_GSURI, False, 0).offload(
+ self._job.queue_args[0],
+ self._job.queue_args[1],
+ self._job.queue_args[2])
+ self.assertTrue(os.path.isdir(self._job.queue_args[0]))
+
+
+ # TODO(ayatane): This tests passes when run locally, but it fails
+ # when run on trybot. I have no idea why, but the assert isdir
+ # fails.
+ #
+ # This test is also kind of redundant since we are using the timeout
+ # from chromite which has its own tests.
+ @unittest.skip('This fails on trybot')
+ def test_offload_timeout_late(self):
+ """Test that `offload_dir()` times out correctly.
+
+ This test triggers timeout at the latest possible moment, at
+ the call to clear the timeout alarm.
+
+ """
+ signal.alarm.side_effect = [0, 0, timeout_util.TimeoutError('fubar')]
+ with mock.patch.object(gs_offloader, '_upload_cts_testresult',
+ autospec=True) as upload, \
+ mock.patch.object(gs_offloader, '_get_cmd_list',
+ autospec=True) as get_cmd_list:
+ upload.return_value = None
+ get_cmd_list.return_value = ['test', '-d', self._job.queue_args[0]]
+ gs_offloader.GSOffloader(
+ utils.DEFAULT_OFFLOAD_GSURI, False, 0).offload(
+ self._job.queue_args[0],
+ self._job.queue_args[1],
+ self._job.queue_args[2])
+ self.assertTrue(os.path.isdir(self._job.queue_args[0]))
+
+
+
if __name__ == '__main__':
unittest.main()
diff --git a/site_utils/job_directories.py b/site_utils/job_directories.py
index 6b390ac..8d157ed 100755
--- a/site_utils/job_directories.py
+++ b/site_utils/job_directories.py
@@ -5,7 +5,6 @@
import os
import re
import shutil
-import time
import common
from autotest_lib.client.common_lib import time_utils
@@ -121,10 +120,10 @@
GLOB_PATTERN = None # must be redefined in subclass
def __init__(self, resultsdir):
- self._dirname = resultsdir
+ self.dirname = resultsdir
self._id = get_job_id_or_task_id(resultsdir)
- self._offload_count = 0
- self._first_offload_start = 0
+ self.offload_count = 0
+ self.first_offload_start = 0
@classmethod
def get_job_directories(cls):
@@ -146,51 +145,6 @@
"""
raise NotImplementedError("_JobDirectory.get_timestamp_if_finished")
- def enqueue_offload(self, queue, age_limit):
- """Enqueue the job for offload, if it's eligible.
-
- The job is eligible for offloading if the database has marked
- it finished, and the job is older than the `age_limit`
- parameter.
-
- If the job is eligible, offload processing is requested by
- passing the `queue` parameter's `put()` method a sequence with
- the job's `_dirname` attribute and its directory name.
-
- @param queue If the job should be offloaded, put the offload
- parameters into this queue for processing.
- @param age_limit Minimum age for a job to be offloaded. A value
- of 0 means that the job will be offloaded as
- soon as it is finished.
-
- """
- timestamp = self.get_timestamp_if_finished()
- if not self._offload_count:
- if not timestamp:
- return
- if not is_job_expired(age_limit, timestamp):
- return
- self._first_offload_start = time.time()
- self._offload_count += 1
- if self.process_gs_instructions():
- queue.put([self._dirname, os.path.dirname(self._dirname), timestamp])
-
- def is_offloaded(self):
- """Return whether this job has been successfully offloaded."""
- return not os.path.exists(self._dirname)
-
- def get_failure_time(self):
- """Return the time of the first offload failure."""
- return self._first_offload_start
-
- def get_failure_count(self):
- """Return the number of times this job has failed to offload."""
- return self._offload_count
-
- def get_job_directory(self):
- """Return the name of this job's results directory."""
- return self._dirname
-
def process_gs_instructions(self):
"""Process any gs_offloader instructions for this special task.
@@ -215,21 +169,17 @@
@returns True/False if there is anything left to offload.
"""
# Go through the gs_offloader instructions file for each test in this job.
- for path in glob.glob(os.path.join(self._dirname, '*',
+ for path in glob.glob(os.path.join(self.dirname, '*',
constants.GS_OFFLOADER_INSTRUCTIONS)):
with open(path, 'r') as f:
gs_off_instructions = json.load(f)
if gs_off_instructions.get(constants.GS_OFFLOADER_NO_OFFLOAD):
dirname = os.path.dirname(path)
- shutil.rmtree(dirname)
- os.mkdir(dirname)
- breadcrumb_name = os.path.join(dirname, 'logs-removed-readme.txt')
- with open(breadcrumb_name, 'w') as f:
- f.write(NO_OFFLOAD_README)
+ _remove_log_directory_contents(dirname)
# Finally check if there's anything left to offload.
- if not os.listdir(self._dirname):
- shutil.rmtree(self._dirname)
+ if not os.listdir(self.dirname):
+ shutil.rmtree(self.dirname)
return False
return True
@@ -251,6 +201,20 @@
return max([hqe.finished_on for hqe in hqes])
+def _remove_log_directory_contents(dirpath):
+ """Remove log directory contents.
+
+ Leave a note explaining what has happened to the logs.
+
+ @param dirpath: Path to log directory.
+ """
+ shutil.rmtree(dirpath)
+ os.mkdir(dirpath)
+ breadcrumb_name = os.path.join(dirpath, 'logs-removed-readme.txt')
+ with open(breadcrumb_name, 'w') as f:
+ f.write(NO_OFFLOAD_README)
+
+
class SpecialJobDirectory(_JobDirectory):
"""Subclass of _JobDirectory for special (per-host) jobs."""
diff --git a/utils/gslib.py b/utils/gslib.py
new file mode 100644
index 0000000..99c1f80
--- /dev/null
+++ b/utils/gslib.py
@@ -0,0 +1,34 @@
+# Copyright 2017 The Chromium OS Authors. All rights reserved.
+# Use of this source code is governed by a BSD-style license that can be
+# found in the LICENSE file.
+
+"""This module contains Google Storage utilities.
+
+TODO(ayatane): This should be merged into chromite.lib.gs
+"""
+
+import re
+
+# See https://cloud.google.com/storage/docs/bucket-naming#objectnames
+# Banned characters: []*?#
+# and control characters (hex): 00-1f, 7f-84, 86-ff
+_INVALID_GS_PATTERN = r'[[\]*?#\x00-\x1f\x7f-\x84\x86-\xff]'
+
+
+def escape(name):
+ """Escape GS object name.
+
+ @param name: Name string.
+ @return: Escaped name string.
+ """
+ return re.sub(_INVALID_GS_PATTERN,
+ lambda x: _percent_escape(x.group(0)), name)
+
+
+def _percent_escape(char):
+ """Percent escape a character.
+
+ @param char: character to escape.
+ @return: Escaped string.
+ """
+ return '%{:02x}'.format(ord(char))
diff --git a/utils/gslib_unittest.py b/utils/gslib_unittest.py
new file mode 100755
index 0000000..8316e32
--- /dev/null
+++ b/utils/gslib_unittest.py
@@ -0,0 +1,28 @@
+#!/usr/bin/python2.7
+#
+# Copyright 2017 The Chromium OS Authors. All rights reserved.
+# Use of this source code is governed by a BSD-style license that can be
+# found in the LICENSE file.
+
+import unittest
+
+import common
+from autotest_lib.utils import gslib
+
+
+class EscapeTestCase(unittest.TestCase):
+ """Tests for basic KeyvalLabel functions."""
+
+ def test_escape_printable(self):
+ """Test escaping printable characters."""
+ got = gslib.escape('foo[]*?#')
+ self.assertEqual(got, 'foo%5b%5d%2a%3f%23')
+
+ def test_escape_control(self):
+ """Test escaping control characters by hex."""
+ got = gslib.escape('foo\x88')
+ self.assertEqual(got, 'foo%88')
+
+
+if __name__ == '__main__':
+ unittest.main()
