Updater tests (#1974)

* refactor exit handling

* test update

* more reliable?

* better

* init git in CI

* testy tester

* CI should work

* test overlay reinit

* only one

* still need to fix loop test

* more patience

* more patience in CI

* no ping in CI

* this is cleaner

* need to run these in jenkins

* clean up

* run in jenkins

* fix test file path

* it's a git repo now

* no commit

* reinit

* remove duplicate

* why not git

* la

* git status

* pythonpath

* fix

* no CI fro now

* check overlay consistent

* more tests

* make more changes in the update commit

* sample

* no k
old-commit-hash: fe18a014c7

selfdrive/test/test_updated.py

@@ -0,0 +1,256 @@
+#!/usr/bin/env python3
+import datetime
+import os
+import time
+import tempfile
+import unittest
+import shutil
+import signal
+import subprocess
+import random
+
+from common.basedir import BASEDIR
+from common.params import Params
+
+
+class TestUpdater(unittest.TestCase):
+
+  def setUp(self):
+    self.updated_proc = None
+
+    self.tmp_dir = tempfile.TemporaryDirectory()
+    org_dir = os.path.join(self.tmp_dir.name, "commaai")
+
+    self.basedir = os.path.join(org_dir, "openpilot")
+    self.git_remote_dir = os.path.join(org_dir, "openpilot_remote")
+    self.staging_dir = os.path.join(org_dir, "safe_staging")
+    for d in [org_dir, self.basedir, self.git_remote_dir, self.staging_dir]:
+      os.mkdir(d)
+
+    self.upper_dir = os.path.join(self.staging_dir, "upper")
+    self.merged_dir = os.path.join(self.staging_dir, "merged")
+    self.finalized_dir = os.path.join(self.staging_dir, "finalized")
+
+    # setup local submodule remotes
+    submodules = subprocess.check_output("git submodule --quiet foreach 'echo $name'",
+                                         shell=True, cwd=BASEDIR, encoding='utf8').split()
+    for s in submodules:
+      sub_path = os.path.join(org_dir, s.split("_repo")[0])
+      self._run(f"git clone {s} {sub_path}.git", cwd=BASEDIR)
+
+    # setup two git repos, a remote and one we'll run updated in
+    self._run([
+      f"git clone {BASEDIR} {self.git_remote_dir}",
+      f"git clone {self.git_remote_dir} {self.basedir}",
+      f"cd {self.basedir} && git submodule init && git submodule update",
+      f"cd {self.basedir} && scons -j{os.cpu_count()} cereal"
+    ])
+
+    self.params = Params(db=os.path.join(self.basedir, "persist/params"))
+    self.params.clear_all()
+    os.sync()
+
+  def tearDown(self):
+    try:
+      if self.updated_proc is not None:
+        self.updated_proc.terminate()
+        self.updated_proc.wait(30)
+    except Exception as e:
+      print(e)
+    self.tmp_dir.cleanup()
+
+
+  # *** test helpers ***
+
+
+  def _run(self, cmd, cwd=None):
+    if not isinstance(cmd, list):
+      cmd = (cmd,)
+
+    for c in cmd:
+      subprocess.check_output(c, cwd=cwd, shell=True)
+
+  def _get_updated_proc(self):
+    os.environ["PYTHONPATH"] = self.basedir
+    os.environ["GIT_AUTHOR_NAME"] = "testy tester"
+    os.environ["GIT_COMMITTER_NAME"] = "testy tester"
+    os.environ["GIT_AUTHOR_EMAIL"] = "testy@tester.test"
+    os.environ["GIT_COMMITTER_EMAIL"] = "testy@tester.test"
+    os.environ["UPDATER_TEST_IP"] = "localhost"
+    os.environ["UPDATER_LOCK_FILE"] = os.path.join(self.tmp_dir.name, "updater.lock")
+    os.environ["UPDATER_STAGING_ROOT"] = self.staging_dir
+    updated_path = os.path.join(self.basedir, "selfdrive/updated.py")
+    return subprocess.Popen(updated_path, env=os.environ)
+
+  def _start_updater(self, offroad=True, nosleep=False):
+    self.params.put("IsOffroad", "1" if offroad else "0")
+    self.updated_proc = self._get_updated_proc()
+    if not nosleep:
+      time.sleep(1)
+
+  def _update_now(self):
+    self.updated_proc.send_signal(signal.SIGHUP)
+
+  # TODO: this should be implemented in params
+  def _read_param(self, key, timeout=1):
+    ret = None
+    start_time = time.monotonic()
+    while ret is None:
+      ret = self.params.get(key, encoding='utf8')
+      if time.monotonic() - start_time > timeout:
+        break
+      time.sleep(0.01)
+    return ret
+
+  def _wait_for_update(self, timeout=30, clear_param=False):
+    if clear_param:
+      self.params.delete("LastUpdateTime")
+
+    self._update_now()
+    t = self._read_param("LastUpdateTime", timeout=timeout)
+    if t is None:
+      raise Exception("timed out waiting for update to complate")
+
+  def _make_commit(self):
+    all_dirs, all_files = [], []
+    for root, dirs, files in os.walk(self.git_remote_dir):
+      if ".git" in root:
+        continue
+      for d in dirs:
+        all_dirs.append(os.path.join(root, d))
+      for f in files:
+        all_files.append(os.path.join(root, f))
+
+    # make a new dir and some new files
+    new_dir = os.path.join(self.git_remote_dir, "this_is_a_new_dir")
+    os.mkdir(new_dir)
+    for _ in range(random.randrange(5, 30)):
+      for d in (new_dir, random.choice(all_dirs)):
+        with tempfile.NamedTemporaryFile(dir=d, delete=False) as f:
+          f.write(os.urandom(random.randrange(1, 1000000)))
+
+    # modify some files
+    for f in random.sample(all_files, random.randrange(5, 50)):
+      with open(f, "w+") as ff:
+        txt = ff.readlines()
+        ff.seek(0)
+        for line in txt:
+          ff.write(line[::-1])
+
+    # remove some files
+    for f in random.sample(all_files, random.randrange(5, 50)):
+      os.remove(f)
+
+    # remove some dirs
+    for d in random.sample(all_dirs, random.randrange(1, 10)):
+      shutil.rmtree(d)
+
+    # commit the changes
+    self._run([
+      "git add -A",
+      "git commit -m 'an update'",
+    ], cwd=self.git_remote_dir)
+
+  def _check_update_state(self, update_available):
+    # make sure LastUpdateTime is recent
+    t = self._read_param("LastUpdateTime")
+    last_update_time = datetime.datetime.fromisoformat(t)
+    td = datetime.datetime.utcnow() - last_update_time
+    self.assertLess(td.total_seconds(), 10)
+    self.params.delete("LastUpdateTime")
+
+    # wait a bit for the rest of the params to be written
+    time.sleep(0.1)
+
+    # check params
+    update = self._read_param("UpdateAvailable")
+    self.assertEqual(update == "1", update_available, f"UpdateAvailable: {repr(update)}")
+    self.assertEqual(self._read_param("UpdateFailedCount"), "0")
+
+    # TODO: check that the finalized update actually matches remote
+    # check the .overlay_init and .overlay_consistent flags
+    self.assertTrue(os.path.isfile(os.path.join(self.basedir, ".overlay_init")))
+    self.assertEqual(os.path.isfile(os.path.join(self.finalized_dir, ".overlay_consistent")), update_available)
+
+
+  # *** test cases ***
+
+
+  # Run updated for 100 cycles with no update
+  def test_no_update(self):
+    self._start_updater()
+    for _ in range(100):
+      self._wait_for_update(clear_param=True)
+      self._check_update_state(False)
+
+  # Let the updater run with no update for a cycle, then write an update
+  def test_update(self):
+    self._start_updater()
+
+    # run for a cycle with no update
+    self._wait_for_update(clear_param=True)
+    self._check_update_state(False)
+
+    # write an update to our remote
+    self._make_commit()
+
+    # run for a cycle to get the update
+    self._wait_for_update(timeout=60, clear_param=True)
+    self._check_update_state(True)
+
+    # run another cycle with no update
+    self._wait_for_update(clear_param=True)
+    self._check_update_state(True)
+
+  # Let the updater run for 10 cycles, and write an update every cycle
+  @unittest.skip("need to make this faster")
+  def test_update_loop(self):
+    self._start_updater()
+
+    # run for a cycle with no update
+    self._wait_for_update(clear_param=True)
+    for _ in range(10):
+      time.sleep(0.5)
+      self._make_commit()
+      self._wait_for_update(timeout=90, clear_param=True)
+      self._check_update_state(True)
+
+  # Test overlay re-creation after tracking a new file in basedir's git
+  def test_overlay_reinit(self):
+    self._start_updater()
+
+    overlay_init_fn = os.path.join(self.basedir, ".overlay_init")
+
+    # run for a cycle with no update
+    self._wait_for_update(clear_param=True)
+    self.params.delete("LastUpdateTime")
+    first_mtime = os.path.getmtime(overlay_init_fn)
+
+    # touch a file in the basedir
+    self._run("touch new_file && git add new_file", cwd=self.basedir)
+
+    # run another cycle, should have a new mtime
+    self._wait_for_update(clear_param=True)
+    second_mtime = os.path.getmtime(overlay_init_fn)
+    self.assertTrue(first_mtime != second_mtime)
+
+    # run another cycle, mtime should be same as last cycle
+    self._wait_for_update(clear_param=True)
+    new_mtime = os.path.getmtime(overlay_init_fn)
+    self.assertTrue(second_mtime == new_mtime)
+
+  # Make sure updated exits if another instance is running
+  def test_multiple_instances(self):
+    # start updated and let it run for a cycle
+    self._start_updater()
+    time.sleep(1)
+    self._wait_for_update(clear_param=True)
+
+    # start another instance
+    second_updated = self._get_updated_proc()
+    ret_code = second_updated.wait(timeout=5)
+    self.assertTrue(ret_code is not None)
+
+
+if __name__ == "__main__":
+  unittest.main()

selfdrive/updated.py

@@ -28,16 +28,18 @@ import subprocess
 import psutil
 import shutil
 import signal
-from pathlib import Path
 import fcntl
 import threading
 from cffi import FFI
+from pathlib import Path
 
 from common.basedir import BASEDIR
 from common.params import Params
 from selfdrive.swaglog import cloudlog
 
-STAGING_ROOT = "/data/safe_staging"
+TEST_IP = os.getenv("UPDATER_TEST_IP", "8.8.8.8")
+LOCK_FILE = os.getenv("UPDATER_LOCK_FILE", "/tmp/safe_staging_overlay.lock")
+STAGING_ROOT = os.getenv("UPDATER_STAGING_ROOT", "/data/safe_staging")
 
 OVERLAY_UPPER = os.path.join(STAGING_ROOT, "upper")
 OVERLAY_METADATA = os.path.join(STAGING_ROOT, "metadata")
@@ -81,20 +83,13 @@ def run(cmd, cwd=None):
   return subprocess.check_output(cmd, cwd=cwd, stderr=subprocess.STDOUT, encoding='utf8')
 
 
-def remove_consistent_flag():
+def set_consistent_flag(consistent):
   os.system("sync")
   consistent_file = Path(os.path.join(FINALIZED, ".overlay_consistent"))
-  try:
+  if consistent:
-    consistent_file.unlink()
-  except FileNotFoundError:
-    pass
-  os.system("sync")
-
-
-def set_consistent_flag():
-  consistent_file = Path(os.path.join(FINALIZED, ".overlay_consistent"))
-  os.system("sync")
     consistent_file.touch()
+  elif not consistent and consistent_file.exists():
+    consistent_file.unlink()
   os.system("sync")
 
 
@@ -150,7 +145,7 @@ def init_ovfs():
   cloudlog.info("preparing new safe staging area")
   Params().put("UpdateAvailable", "0")
 
-  remove_consistent_flag()
+  set_consistent_flag(False)
 
   dismount_ovfs()
   if os.path.isdir(STAGING_ROOT):
@@ -158,6 +153,7 @@ def init_ovfs():
 
   for dirname in [STAGING_ROOT, OVERLAY_UPPER, OVERLAY_METADATA, OVERLAY_MERGED, FINALIZED]:
     os.mkdir(dirname, 0o755)
+
   if not os.lstat(BASEDIR).st_dev == os.lstat(OVERLAY_MERGED).st_dev:
     raise RuntimeError("base and overlay merge directories are on different filesystems; not valid for overlay FS!")
 
@@ -216,13 +212,13 @@ def attempt_update():
 
     # Un-set the validity flag to prevent the finalized tree from being
     # activated later if the finalize step is interrupted
-    remove_consistent_flag()
+    set_consistent_flag(False)
 
     finalize_from_ovfs()
 
     # Make sure the validity flag lands on disk LAST, only when the local git
     # repo and OP install are in a consistent state.
-    set_consistent_flag()
+    set_consistent_flag(True)
 
     cloudlog.info("update successful!")
   else:
@@ -232,8 +228,6 @@ def attempt_update():
 
 
 def main():
-  update_failed_count = 0
-  overlay_init_done = False
   params = Params()
 
   if params.get("DisableUpdates") == b"1":
@@ -247,7 +241,7 @@ def main():
   if psutil.LINUX:
     p.ionice(psutil.IOPRIO_CLASS_BE, value=7)
 
-  ov_lock_fd = open('/tmp/safe_staging_overlay.lock', 'w')
+  ov_lock_fd = open(LOCK_FILE, 'w')
   try:
     fcntl.flock(ov_lock_fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
   except IOError:
@@ -257,33 +251,34 @@ def main():
   wait_helper = WaitTimeHelper()
   wait_helper.sleep(30)
 
+  update_failed_count = 0
+  overlay_initialized = False
   while not wait_helper.shutdown:
     update_failed_count += 1
     wait_helper.ready_event.clear()
 
     # Check for internet every 30s
     time_wrong = datetime.datetime.utcnow().year < 2019
-    ping_failed = os.system("git ls-remote --tags --quiet") != 0
+    ping_failed = os.system(f"ping -W 4 -c 1 {TEST_IP}") != 0
     if ping_failed or time_wrong:
       wait_helper.sleep(30)
       continue
 
     # Attempt an update
     try:
-      # If the git directory has modifcations after we created the overlay
+      # Re-create the overlay if BASEDIR/.git has changed since we created the overlay
-      # we need to recreate the overlay
+      if overlay_initialized:
-      if overlay_init_done:
         overlay_init_fn = os.path.join(BASEDIR, ".overlay_init")
         git_dir_path = os.path.join(BASEDIR, ".git")
         new_files = run(["find", git_dir_path, "-newer", overlay_init_fn])
 
         if len(new_files.splitlines()):
           cloudlog.info(".git directory changed, recreating overlay")
-          overlay_init_done = False
+          overlay_initialized = False
 
-      if not overlay_init_done:
+      if not overlay_initialized:
         init_ovfs()
-        overlay_init_done = True
+        overlay_initialized = True
 
       if params.get("IsOffroad") == b"1":
         attempt_update()
@@ -298,11 +293,13 @@ def main():
         output=e.output,
         returncode=e.returncode
       )
-      overlay_init_done = False
+      overlay_initialized = False
     except Exception:
       cloudlog.exception("uncaught updated exception, shouldn't happen")
 
     params.put("UpdateFailedCount", str(update_failed_count))
+
+    # Wait 10 minutes between update attempts
     wait_helper.sleep(60*10)
 
   # We've been signaled to shut down