Basic logging of sync process to .mergin/client-log.txt

Author: wonder-sk

mergin/client_pull.py

@@ -12,10 +12,10 @@
 import copy
 import math
 import os
+import pprint
 import shutil
 
 import concurrent.futures
-import threading
 
 from .common import CHUNK_SIZE, ClientError
 from .merginproject import MerginProject
@@ -76,14 +76,14 @@ def _download_items(file, directory, diff_only=False):
     return items
 
 
-def _do_download(item, mc, project_path, job):
+def _do_download(item, mc, mp, project_path, job):
     """ runs in worker thread """
     if job.is_cancelled:
         return
 
     # TODO: make download_blocking / save_to_file cancellable so that we can cancel as soon as possible
 
-    item.download_blocking(mc, project_path)
+    item.download_blocking(mc, mp, project_path)
     job.transferred_size += item.size
 
 
@@ -99,9 +99,13 @@ def download_project_async(mc, project_path, directory):
     os.makedirs(directory)
     mp = MerginProject(directory)
 
+    mp.log.info(f"--- start download {project_path}")
+
     project_info = mc.project_info(project_path)
     version = project_info['version'] if project_info['version'] else 'v0'
 
+    mp.log.info(f"got project info. version {version}")
+
     # prepare download
     update_tasks = []  # stuff to do at the end of download
     for file in project_info['files']:
@@ -116,14 +120,16 @@ def download_project_async(mc, project_path, directory):
         download_list.extend(task.download_queue_items)
         for item in task.download_queue_items:
             total_size += item.size
+
+    mp.log.info(f"will download {len(update_tasks)} files in {len(download_list)} chunks, total size {total_size}")
 
     job = DownloadJob(project_path, total_size, version, update_tasks, download_list, directory, mp, project_info)
 
     # start download
     job.executor = concurrent.futures.ThreadPoolExecutor(max_workers=4)
     job.futures = []
     for item in download_list:
-        future = job.executor.submit(_do_download, item, mc, project_path, job)
+        future = job.executor.submit(_do_download, item, mc, mp, project_path, job)
         job.futures.append(future)
 
     return job
@@ -167,6 +173,8 @@ def download_project_finalize(job):
         if future.exception() is not None:
             raise future.exception()
 
+    job.mp.log.info("--- download finished")
+
     for task in job.update_tasks:
 
         # right now only copy tasks...
@@ -235,9 +243,10 @@ def __repr__(self):
         return "<DownloadQueueItem path={} version={} diff_only={} part_index={} size={} dest={}>".format(
             self.file_path, self.version, self.diff_only, self.part_index, self.size, self.download_file_path)
 
-    def download_blocking(self, mc, project_path):
+    def download_blocking(self, mc, mp, project_path):
         """ Starts download and only returns once the file has been fully downloaded and saved """
 
+        mp.log.debug(f"Downloading {self.file_path} version={self.version} diff={self.diff_only} part={self.part_index}")
         start = self.part_index * (1 + CHUNK_SIZE)
         resp = mc.get("/v1/project/raw/{}".format(project_path), data={
                 "file": self.file_path,
@@ -248,9 +257,11 @@ def download_blocking(self, mc, project_path):
             }
         )
         if resp.status in [200, 206]:
+            mp.log.debug(f"Download finished: {self.file_path}")
             save_to_file(resp, self.download_file_path)
         else:
-            raise ClientError('Failed to download part {} of file {}'.format(part, basename))
+            mp.log.error(f"Download failed: {self.file_path}")
+            raise ClientError('Failed to download part {} of file {}'.format(self.part_index, self.file_path))
 
 
 class PullJob:
@@ -290,19 +301,27 @@ def pull_project_async(mc, directory):
     mp = MerginProject(directory)
     project_path = mp.metadata["name"]
     local_version = mp.metadata["version"]
+
+    mp.log.info(f"--- start pull {project_path}")
+
     server_info = mc.project_info(project_path, since=local_version)
-    if local_version == server_info["version"]:
+    server_version = server_info["version"]
+
+    mp.log.info(f"got project info. version {server_version}")
+
+    if local_version == server_version:
+        mp.log.info("--- pull - nothing to do (already at server version)")
         return  # Project is up to date
 
     # we either download a versioned file using diffs (strongly preferred),
     # but if we don't have history with diffs (e.g. uploaded without diffs)
     # then we just download the whole file
     _pulling_file_with_diffs = lambda f: 'diffs' in f and len(f['diffs']) != 0
 
-    server_version = server_info["version"]
     temp_dir = mp.fpath_meta(f'fetch_{local_version}-{server_version}')
     os.makedirs(temp_dir, exist_ok=True)
     pull_changes = mp.get_pull_changes(server_info["files"])
+    mp.log.debug("pull changes:\n" + pprint.pformat(pull_changes))
     fetch_files = []
     for f in pull_changes["added"]:
         f['version'] = server_version
@@ -364,13 +383,15 @@ def pull_project_async(mc, directory):
         for item in file_to_merge.downloaded_items:
             total_size += item.size
 
+    mp.log.info(f"will download {len(download_list)} chunks, total size {total_size}")
+
     job = PullJob(project_path, pull_changes, total_size, server_version, files_to_merge, download_list, temp_dir, mp, server_info, basefiles_to_patch)
 
     # start download
     job.executor = concurrent.futures.ThreadPoolExecutor(max_workers=4)
     job.futures = []
     for item in download_list:
-        future = job.executor.submit(_do_download, item, mc, project_path, job)
+        future = job.executor.submit(_do_download, item, mc, mp, project_path, job)
         job.futures.append(future)
 
     return job
@@ -450,6 +471,8 @@ def pull_project_finalize(job):
         if future.exception() is not None:
             raise future.exception()
 
+    job.mp.log.info("finalizing pull")
+
     # merge downloaded chunks
     for file_to_merge in job.files_to_merge:
         file_to_merge.merge()
@@ -478,6 +501,8 @@ def pull_project_finalize(job):
         'version': job.version if job.version else "v0",  # for new projects server version is ""
         'files': job.project_info['files']
     }
-    
+
+    job.mp.log.info("--- pull finished")
+
     shutil.rmtree(job.temp_dir)
     return conflicts

mergin/client_push.py

@@ -11,8 +11,8 @@
 
 import json
 import hashlib
+import pprint
 import concurrent.futures
-import threading
 
 from .common import UPLOAD_CHUNK_SIZE, ClientError, SyncError
 from .merginproject import MerginProject
@@ -53,18 +53,21 @@ def __init__(self, file_path, size, transaction_id, chunk_id, chunk_index):
         self.chunk_index = chunk_index        # index (starting from zero) of the chunk within the file
         self.transaction_id = transaction_id  # ID of the transaction
 
-    def upload_blocking(self, mc):
+    def upload_blocking(self, mc, mp):
 
         file_handle = open(self.file_path, 'rb')
         file_handle.seek(self.chunk_index * UPLOAD_CHUNK_SIZE)
         data = file_handle.read(UPLOAD_CHUNK_SIZE)
 
         checksum = hashlib.sha1()
         checksum.update(data)
-        
+
+        mp.log.debug(f"Uploading {self.file_path} part={self.chunk_index}")
+
         headers = {"Content-Type": "application/octet-stream"}
         resp = mc.post("/v1/project/push/chunk/{}/{}".format(self.transaction_id, self.chunk_id), data, headers)
         resp_dict = json.load(resp)
+        mp.log.debug(f"Upload finished: {self.file_path}")
         if not (resp_dict['size'] == len(data) and resp_dict['checksum'] == checksum.hexdigest()):
             mc.post("/v1/project/push/cancel/{}".format(self.transaction_id))
             raise ClientError("Mismatch between uploaded file chunk {} and local one".format(self.chunk_id))
@@ -77,19 +80,31 @@ def push_project_async(mc, directory):
     mp = MerginProject(directory)
     project_path = mp.metadata["name"]
     local_version = mp.metadata["version"]
+
+    mp.log.info(f"--- start push {project_path}")
+
     server_info = mc.project_info(project_path)
     server_version = server_info["version"] if server_info["version"] else "v0"
+
+    mp.log.info(f"got project info. version {server_version}")
+
     if local_version != server_version:
-        raise ClientError("Update your local repository")
+        mp.log.error(f"--- push {project_path} - not up to date (local {local_version} vs server {server_version})")
+        raise ClientError("There is a new version of the project on the server. Please update your local copy." +
+                          f"\n\nLocal version: {local_version}\nServer version: {server_version}")
 
     changes = mp.get_push_changes()
+    mp.log.debug("push changes:\n" + pprint.pformat(changes))
     enough_free_space, freespace = mc.enough_storage_available(changes)
     if not enough_free_space:
         freespace = int(freespace/(1024*1024))
+        mp.log.error(f"--- push {project_path} - not enough space")
         raise SyncError("Storage limit has been reached. Only " + str(freespace) + "MB left")
 
     if not sum(len(v) for v in changes.values()):
+        mp.log.info(f"--- push {project_path} - nothing to do")
         return
+
     # drop internal info from being sent to server
     for item in changes['updated']:
         item.pop('origin_checksum', None)
@@ -107,10 +122,13 @@ def push_project_async(mc, directory):
     job = UploadJob(project_path, changes, transaction_id, mp, mc)
 
     if not upload_files:
+        mp.log.info("not uploading any files")
         job.server_resp = server_resp
         push_project_finalize(job)
         return None   # all done - no pending job
-    
+
+    mp.log.info(f"got transaction ID {transaction_id}")
+
     upload_queue_items = []
     total_size = 0
     # prepare file chunks for upload
@@ -129,7 +147,9 @@ def push_project_async(mc, directory):
 
     job.total_size = total_size
     job.upload_queue_items = upload_queue_items
-    
+
+    mp.log.info(f"will upload {len(upload_queue_items)} items with total size {total_size}")
+
     # start uploads in background
     job.executor = concurrent.futures.ThreadPoolExecutor(max_workers=4)
     for item in upload_queue_items:
@@ -186,17 +206,19 @@ def push_project_finalize(job):
 
     if with_upload_of_files:
         try:
+            job.mp.log.info(f"Finishing transaction {job.transaction_id}")
             resp = job.mc.post("/v1/project/push/finish/%s" % job.transaction_id)
             job.server_resp = json.load(resp)
         except ClientError as err:
             job.mc.post("/v1/project/push/cancel/%s" % job.transaction_id)
             # server returns various error messages with filename or something generic
             # it would be better if it returned list of failed files (and reasons) whenever possible
             return {'error': str(err)}
-    
+
     if 'error' in job.server_resp:
         #TODO would be good to get some detailed info from server so user could decide what to do with it
         # e.g. diff conflicts, basefiles issues, or any other failure
+        job.mp.log.error("push failed. server response: " + job.server_resp['error'])
         raise ClientError(job.server_resp['error'])
 
     job.mp.metadata = {
@@ -206,6 +228,8 @@ def push_project_finalize(job):
     }
     job.mp.apply_push_changes(job.changes)
 
+    job.mp.log.info("--- push finished")
+
 
 def push_project_cancel(job):
     """
@@ -224,5 +248,5 @@ def _do_upload(item, job):
     if job.is_cancelled:
         return
 
-    item.upload_blocking(job.mc)
+    item.upload_blocking(job.mc, job.mp)
     job.transferred_size += item.size

mergin/merginproject.py

@@ -1,5 +1,6 @@
 
 import json
+import logging
 import math
 import os
 import re
@@ -48,6 +49,16 @@ def __init__(self, directory):
         if not os.path.exists(self.meta_dir):
             os.mkdir(self.meta_dir)
 
+        # setup logging into project directory's .mergin/client-log.txt file
+        self.log = logging.getLogger('mergin.' + directory)
+        self.log.setLevel(logging.DEBUG)   # log everything (it would otherwise log just warnings+errors)
+        if not self.log.handlers:
+            # we only need to set the handler once
+            # (otherwise we would get things logged multiple times as loggers are cached)
+            log_handler = logging.FileHandler(os.path.join(self.meta_dir, "client-log.txt"))
+            log_handler.setFormatter(logging.Formatter('%(asctime)s %(message)s'))
+            self.log.addHandler(log_handler)
+
     def fpath(self, file, other_dir=None):
         """
         Helper function to get absolute path of project file. Defaults to project dir but

mergin/test/test_client.py

@@ -166,7 +166,7 @@ def test_push_pull_changes(mc):
     f_conflict_checksum = generate_checksum(os.path.join(project_dir_2, f_updated))
 
     # not at latest server version
-    with pytest.raises(ClientError, match='Update your local repository'):
+    with pytest.raises(ClientError, match='Please update your local copy'):
         mc.push_project(project_dir_2)
 
     # check changes in project_dir_2 before applied