Merge pull request #486 from mapswipe/dev

Draft: v2.3.1

Author: Hagellach37

firebase/database.rules.json

@@ -65,12 +65,12 @@
             ||
             (root.child('/v2/projects/'+$project_id+'/teamId').val() ==
              root.child('/v2/users/'+auth.uid+'/teamId').val())
-          "
-        },
-        ".indexOn": [
-          "finishedCount",
-          "requiredCount"
-        ]
+          ",
+          ".indexOn": [
+            "finishedCount",
+            "requiredCount"
+          ]
+        }
       },
       "tasks": {
         ".write": false,

mapswipe_workers/locust_files/load_testing.py

@@ -0,0 +1,132 @@
+# run with "locust -f locust_files/load_testing.py"
+# then set number of users and spawn rate in web interface
+# e.g. test with 100 users, and 15 users/sec
+# web interface: http://0.0.0.0:8089/
+
+import datetime
+import json
+import random
+from uuid import uuid4
+
+from locust import HttpUser, between, task
+
+from mapswipe_workers.definitions import logger
+from mapswipe_workers.utils import user_management
+
+
+class MapSwipeUser(HttpUser):
+    # assuming that users need between 30 sec and 120 sec to map a group
+    wait_time = between(30, 120)
+
+    def set_up_user(self):
+        # check if is already signed in
+        if self.signed_in_user is None:
+            logger.info("user is not signed in. Will create a new user.")
+            # create user if not exists
+            user = user_management.create_user(self.email, self.username, self.password)
+            self.user_id = user.uid
+
+            # sign in user
+            self.signed_in_user = user_management.sign_in_with_email_and_password(
+                self.email, self.password
+            )
+            logger.info("Created a new user.")
+        else:
+            logger.info("user is already signed in.")
+            pass
+
+    def create_mock_result(self, group):
+        """Create a result object for a build area project.
+
+        The result values are generated randomly.
+        """
+        start_time = datetime.datetime.utcnow().isoformat()[0:-3] + "Z"
+        end_time = datetime.datetime.utcnow().isoformat()[0:-3] + "Z"
+
+        x_min = int(group["xMin"])
+        x_max = int(group["xMax"])
+        y_min = int(group["yMin"])
+        y_max = int(group["yMax"])
+
+        results = {}
+        for x in range(x_min, x_max):
+            for y in range(y_min, y_max):
+                task_id = f"18-{x}-{y}"
+                result = random.choices([0, 1, 2, 3])[0]  # no, yes, maybe, bad_imagery
+                results[task_id] = result
+
+        data = {
+            "results": results,
+            "startTime": start_time,
+            "endTime": end_time,
+        }
+        return data
+
+    def set_firebase_db(self, path, data, token=None):
+        """Upload results to Firebase using REST api."""
+        request_ref = f"{path}.json?auth={token}"
+        headers = {"content-type": "application/json; charset=UTF-8"}
+        self.client.put(
+            request_ref, headers=headers, data=json.dumps(data).encode("utf-8")
+        )
+        logger.info(f"set data in firebase for {path}.json")
+
+    @task
+    def map_a_group(self):
+        """Get a group from Firebase for this user and "map" it.
+
+        Make sure that this user has not worked on this group before.
+        Get the group and create mock results.
+        Upload results to Firebse.
+        """
+
+        # get the groups that need to be mapped
+        path = f"/v2/groups/{self.project_id}"
+        # make sure to set '&' at the end of the string
+        custom_arguments = 'orderBy="requiredCount"&limitToLast=15&'
+        new_groups = user_management.get_firebase_db(
+            path, custom_arguments, self.signed_in_user["idToken"]
+        )
+
+        # get the groups the user has worked on already
+        path = f"/v2/users/{self.user_id}/contributions/{self.project_id}"
+        # make sure to set & at the end of the string
+        custom_arguments = "shallow=True&"
+        existing_groups = user_management.get_firebase_db(
+            path, custom_arguments, self.signed_in_user["idToken"]
+        )
+
+        # pick group for mapping
+        # Get difference between new_groups and existing groups.
+        # We should get the groups the user has not yet worked on.
+        if existing_groups is None:
+            next_group_id = random.choice(list(new_groups.keys()))
+        else:
+            existing_groups.pop(
+                "taskContributionCount", None
+            )  # need to remove this since it is no groupId
+            remaining_group_ids = list(
+                set(new_groups.keys()) - set(existing_groups.keys())
+            )
+            next_group_id = random.choice(remaining_group_ids)
+
+        # get group object
+        next_group = new_groups[next_group_id]
+
+        # create mock result for this group
+        result = self.create_mock_result(next_group)
+
+        # upload results in firebase
+        path = f"/v2/results/{self.project_id}/{next_group_id}/{self.user_id}"
+        self.set_firebase_db(path, result, self.signed_in_user["idToken"])
+
+    def on_start(self):
+        """Set up user and define project when user starts running."""
+        self.project_id = "-MYg8CEf2k1-RitN62X0"
+        random_string = uuid4()
+        self.email = f"test_{random_string}@mapswipe.org"
+        self.username = f"test_{random_string}"
+        self.password = "mapswipe"
+        self.user_id = None
+        self.signed_in_user = None
+        self.set_up_user()

mapswipe_workers/mapswipe_workers/firebase_to_postgres/transfer_results.py

@@ -2,46 +2,33 @@
 import io
 
 import dateutil.parser
+import psycopg2
 
 from mapswipe_workers import auth
 from mapswipe_workers.definitions import logger, sentry
 from mapswipe_workers.firebase_to_postgres import update_data
 
 
 def transfer_results(project_id_list=None):
+    """Transfer results for one project after the other.
+    Will only trigger the transfer of results for projects
+    that are defined in the postgres database.
+    Will not transfer results for tutorials and
+    for projects which are not set up in postgres.
     """
-    Download results from firebase,
-    saves them to postgres and then deletes the results in firebase.
-    This is implemented as a transactional operation as described in
-    the Firebase docs to avoid missing new generated results in
-    Firebase during execution of this function.
-    """
-
-    # Firebase transaction function
-    def transfer(current_results):
-        if current_results is None:
-            logger.info(f"{project_id}: No results in Firebase")
-            return dict()
-        else:
-            results_user_id_list = get_user_ids_from_results(current_results)
-            update_data.update_user_data(results_user_id_list)
-            results_file = results_to_file(current_results, project_id)
-            save_results_to_postgres(results_file)
-            return dict()
-
-    fb_db = auth.firebaseDB()
-
-    if not project_id_list:
+    if project_id_list is None:
         # get project_ids from existing results if no project ids specified
+        fb_db = auth.firebaseDB()
         project_id_list = fb_db.reference("v2/results/").get(shallow=True)
-        if not project_id_list:
+        if project_id_list is None:
             project_id_list = []
             logger.info("There are no results to transfer.")
 
-    # get all project ids from postgres,
-    # we will only transfer results for projects we have there
+    # Get all project ids from postgres.
+    # We will only transfer results for projects we in postgres.
     postgres_project_ids = get_projects_from_postgres()
 
+    project_id_list_transfered = []
     for project_id in project_id_list:
         if project_id not in postgres_project_ids:
             logger.info(
@@ -55,24 +42,101 @@ def transfer(current_results):
                 f"We will not transfer these"
             )
             continue
+        else:
+            logger.info(f"{project_id}: Start transfer results")
+            fb_db = auth.firebaseDB()
+            results_ref = fb_db.reference(f"v2/results/{project_id}")
+            results = results_ref.get()
+            del fb_db
+            transfer_results_for_project(project_id, results)
+            project_id_list_transfered.append(project_id)
+
+    return project_id_list_transfered
+
+
+def transfer_results_for_project(project_id, results):
+    """Transfer the results for a specific project.
+    Save results into an in-memory file.
+    Copy the results to postgres.
+    Delete results in firebase.
+    We are NOT using a Firebase transaction functions here anymore.
+    This has caused problems, in situations where a lot of mappers are
+    uploading results to Firebase at the same time. Basically, this is
+    due to the behaviour of Firebase Transaction function:
+        "If another client writes to this location
+        before the new value is successfully saved,
+        the update function is called again with the new current value,
+        and the write will be retried."
+    (source: https://firebase.google.com/docs/reference/admin/python/firebase_admin.db#firebase_admin.db.Reference.transaction)  # noqa
+    Using Firebase transaction on the group level
+    has turned out to be too slow when using "normal" queries,
+    e.g. without using threading. Threading should be avoided here
+    as well to not run into unforeseen errors.
+    For more details see issue #478.
+    """
 
-        logger.info(f"{project_id}: Start transfering results")
-
-        results_ref = fb_db.reference(f"v2/results/{project_id}")
+    if results is None:
+        logger.info(f"{project_id}: No results in Firebase")
+    else:
+        # First we check for new users in Firebase.
+        # The user_id is used as a key in the postgres database for the results
+        # and thus users need to be inserted before results get inserted.
+        results_user_id_list = get_user_ids_from_results(results)
+        update_data.update_user_data(results_user_id_list)
+
+    try:
+        # Results are dumped into an in-memory file.
+        # This allows us to use the COPY statement to insert many
+        # results at relatively high speed.
+        results_file = results_to_file(results, project_id)
         truncate_temp_results()
+        save_results_to_postgres(results_file)
+    except psycopg2.errors.ForeignKeyViolation as e:
+        sentry.capture_exception(e)
+        sentry.capture_message(
+            "could not transfer results to postgres due to ForeignKeyViolation: "
+            f"{project_id}"
+        )
+        logger.exception(e)
+        logger.warning(
+            "could not transfer results to postgres due to ForeignKeyViolation: "
+            f"{project_id}"
+        )
+    except Exception as e:
+        sentry.capture_exception(e)
+        sentry.capture_message(f"could not transfer results to postgres: {project_id}")
+        logger.exception(e)
+        logger.warning(f"could not transfer results to postgres: {project_id}")
+    else:
+        # It is important here that we first insert results into postgres
+        # and then delete these results from Firebase.
+        # In case something goes wrong during the insert, results in Firebase
+        # will not get deleted.
+        delete_results_from_firebase(project_id, results)
+        logger.info(f"{project_id}: Transferred results to postgres")
+
+
+def delete_results_from_firebase(project_id, results):
+    """Delete results from Firebase using update function.
+    We use the update method of firebase instead of delete.
+    Update allows to delete items at multiple locations at the same time
+    and is much faster.
+    """
 
-        try:
-            results_ref.transaction(transfer)
-            logger.info(f"{project_id}: Transfered results to postgres")
-        except fb_db.TransactionAbortedError:
-            logger.exception(
-                f"{project_id}: Firebase transaction for "
-                f"transfering results failed to commit"
-            )
-            sentry.capture_exception()
+    fb_db = auth.firebaseDB()
 
-    del fb_db
-    return project_id_list
+    # we will use a multi-location update to delete the entries
+    # therefore we create a dict with the items we want to delete
+    data = {}
+    for group_id, users in results.items():
+        for user_id, result in users.items():
+            key = f"{group_id}/{user_id}"
+            data[key] = None
+
+    results_ref = fb_db.reference(f"v2/results/{project_id}/")
+    results_ref.update(data)
+
+    logger.info(f"removed results for project {project_id}")
 
 
 def results_to_file(results, projectId):
@@ -200,7 +264,6 @@ def save_results_to_postgres(results_file):
     Saves results to a temporary table in postgres
     using the COPY Statement of Postgres
     for a more efficient import into the database.
-
     Parameters
     ----------
     results_file: io.StringIO
@@ -228,6 +291,7 @@ def save_results_to_postgres(results_file):
     """
     p_con.query(query_insert_results)
     del p_con
+    logger.info("copied results into postgres.")
 
 
 def truncate_temp_results():

mapswipe_workers/mapswipe_workers/mapswipe_workers.py

@@ -107,14 +107,32 @@ def run_create_projects():
 
 
 @cli.command("firebase-to-postgres")
-def run_firebase_to_postgres() -> list:
+@click.option(
+    "--project_ids",
+    cls=PythonLiteralOption,
+    default="[]",
+    help=(
+        "Project ids for which to generate stats as a list of strings: "
+        """ '["project_a", "project_b"]' """
+        "(You need the quotes.)"
+    ),
+)
+def run_firebase_to_postgres(project_ids: list) -> list:
     """Update users and transfer results from Firebase to Postgres."""
-    project_ids = transfer_results.transfer_results()
-    for project_id in project_ids:
-        update_data.set_progress_in_firebase(project_id)
-        update_data.set_contributor_count_in_firebase(project_id)
-        send_progress_notification(project_id)
+
+    if len(project_ids) > 0:
+        project_ids_transferred = transfer_results.transfer_results(project_ids)
+    else:
+        project_ids_transferred = transfer_results.transfer_results()
+
+    if len(project_ids_transferred) > 0:
+        for project_id in project_ids_transferred:
+            update_data.set_progress_in_firebase(project_id)
+            update_data.set_contributor_count_in_firebase(project_id)
+            send_progress_notification(project_id)
+
     update_data.update_project_data()
+
     return project_ids