Stream write of all objects (#1387)

* Every object export is sent to write queue

* Write thread for everything

* Don't reallocate queue between objects export

* Remove deprecated setter on Thread

* Pass write Q as parameter of export

* Reduce log level

* Fix double import

Author: okorach

migration/migration.py

@@ -98,50 +98,44 @@ def __parse_args(desc):
     return args
 
 
-def __remove_chars_at_end(file: str, nb_bytes: int) -> None:
-    """Writes the configuration in file"""
-    with open(file, mode="rb+") as fd:
-        fd.seek(-nb_bytes, os.SEEK_END)
-        fd.truncate()
-
-
-def write_projects(queue: Queue, file: str) -> None:
+def write_objects(queue: Queue, fd, object_type: str) -> None:
     """
     Thread to write projects in the JSON file
     """
     done = False
     prefix = ""
-    with utilities.open_file(file, mode="a") as fd:
-        print('"   projects": {', file=fd)
-        while not done:
-            project_json = queue.get()
-            done = project_json is None
-            if not done:
-                log.info("Writing project '%s'", project_json["key"])
-                key = project_json.pop("key")
-                print(f'{prefix}"{key}": {utilities.json_dump(project_json)}', end="", file=fd)
-                prefix = ",\n"
-            queue.task_done()
-        print("\n}", file=fd, end="")
-    log.info("Writing projects complete")
-
-
-def __write_export(config: dict[str, str], file: str) -> None:
-    """Writes the configuration in file"""
-    with utilities.open_file(file) as fd:
-        print(utilities.json_dump(config), file=fd)
+    log.info("Waiting %s to write...", object_type)
+    print(f'"{object_type}": ' + "{", file=fd)
+    while not done:
+        obj_json = queue.get()
+        done = obj_json is None
+        if not done:
+            if object_type in ("projects", "applications", "portfolios", "users"):
+                if object_type == "users":
+                    key = obj_json.pop("login", None)
+                else:
+                    key = obj_json.pop("key", None)
+                log.debug("Writing %s key '%s'", object_type[:-1], key)
+                print(f'{prefix}"{key}": {utilities.json_dump(obj_json)}', end="", file=fd)
+            else:
+                log.debug("Writing %s", object_type)
+                print(f"{prefix}{utilities.json_dump(obj_json)[2:-1]}", end="", file=fd)
+            prefix = ",\n"
+        queue.task_done()
+    print("\n}", file=fd, end="")
+    log.info("Writing %s complete", object_type)
 
 
 def __export_config(endpoint: platform.Platform, what: list[str], **kwargs) -> None:
     """Exports a platform configuration in a JSON file"""
+    file = kwargs[options.REPORT_FILE]
     export_settings = {
         "INLINE_LISTS": False,
         "EXPORT_DEFAULTS": True,
         # "FULL_EXPORT": kwargs["fullExport"],
         "FULL_EXPORT": False,
         "MODE": "MIGRATION",
         "THREADS": kwargs[options.NBR_THREADS],
-        options.REPORT_FILE: kwargs[options.REPORT_FILE],
         "SKIP_ISSUES": kwargs["skipIssues"],
     }
     if "projects" in what and kwargs[options.KEYS]:
@@ -154,7 +148,7 @@ def __export_config(endpoint: platform.Platform, what: list[str], **kwargs) -> N
         options.WHAT_RULES: [__JSON_KEY_RULES, rules.export],
         options.WHAT_PROFILES: [__JSON_KEY_PROFILES, qualityprofiles.export],
         options.WHAT_GATES: [__JSON_KEY_GATES, qualitygates.export],
-        # options.WHAT_PROJECTS: [__JSON_KEY_PROJECTS, projects.export],
+        options.WHAT_PROJECTS: [__JSON_KEY_PROJECTS, projects.export],
         options.WHAT_APPS: [__JSON_KEY_APPS, applications.export],
         options.WHAT_PORTFOLIOS: [__JSON_KEY_PORTFOLIOS, portfolios.export],
         options.WHAT_USERS: [__JSON_KEY_USERS, users.export],
@@ -164,34 +158,31 @@ def __export_config(endpoint: platform.Platform, what: list[str], **kwargs) -> N
     log.info("Exporting configuration from %s", kwargs[options.URL])
     key_list = kwargs[options.KEYS]
     sq_settings = {__JSON_KEY_PLATFORM: endpoint.basics()}
-    for what_item, call_data in calls.items():
-        if what_item not in what:
-            continue
-        ndx, func = call_data
-        try:
-            sq_settings[ndx] = func(endpoint, export_settings=export_settings, key_list=key_list)
-            __write_export(sq_settings, kwargs[options.REPORT_FILE])
-        except exceptions.UnsupportedOperation as e:
-            log.warning(e.message)
-    sq_settings = utilities.remove_empties(sq_settings)
-    # if not kwargs.get("dontInlineLists", False):
-    #    sq_settings = utilities.inline_lists(sq_settings, exceptions=("conditions",))
-
-    log.info("Exporting project migration data streaming projects in '%s'", kwargs[options.REPORT_FILE])
-    __remove_chars_at_end(kwargs[options.REPORT_FILE], 3)
-    with utilities.open_file(kwargs[options.REPORT_FILE], mode="a") as fd:
-        print(",", file=fd)
+    is_first = True
     q = Queue(maxsize=0)
-    worker = Thread(target=write_projects, args=(q, kwargs[options.REPORT_FILE]))
-    worker.setDaemon(True)
-    worker.setName("WriteThread")
-    worker.start()
-    export_settings["WRITE_QUEUE"] = q
-    projects.export(endpoint, export_settings=export_settings, key_list=key_list)
-    q.join()
-    log.info("Exporting migration data from %s completed", kwargs["url"])
-    with utilities.open_file(kwargs[options.REPORT_FILE], mode="a") as fd:
+    with utilities.open_file(file, mode="w") as fd:
+        print("{", file=fd)
+        for what_item, call_data in calls.items():
+            if what_item not in what:
+                continue
+            ndx, func = call_data
+            try:
+                if not is_first:
+                    print(",", file=fd)
+                is_first = False
+                worker = Thread(target=write_objects, args=(q, fd, ndx))
+                worker.daemon = True
+                worker.name = f"Write{ndx[:1].upper()}{ndx[1:10]}"
+                worker.start()
+                sq_settings[ndx] = func(endpoint, export_settings=export_settings, key_list=key_list, write_q=q)
+                q.join()
+            except exceptions.UnsupportedOperation as e:
+                log.warning(e.message)
+        sq_settings = utilities.remove_empties(sq_settings)
+        # if not kwargs.get("dontInlineLists", False):
+        #    sq_settings = utilities.inline_lists(sq_settings, exceptions=("conditions",))
         print("\n}", file=fd)
+    log.info("Exporting migration data from %s completed", kwargs["url"])
 
 
 def main() -> None:

sonar/applications.py

@@ -19,6 +19,7 @@
 #
 
 from __future__ import annotations
+from queue import Queue
 from typing import Union
 
 import json
@@ -498,7 +499,9 @@ def exists(endpoint: pf.Platform, key: str) -> bool:
         return False
 
 
-def export(endpoint: pf.Platform, export_settings: types.ConfigSettings, key_list: types.KeyList = None) -> types.ObjectJsonRepr:
+def export(
+    endpoint: pf.Platform, export_settings: types.ConfigSettings, key_list: types.KeyList = None, write_q: Queue = None
+) -> types.ObjectJsonRepr:
     """Exports applications as JSON
 
     :param Platform endpoint: Reference to the Sonar platform
@@ -511,11 +514,17 @@ def export(endpoint: pf.Platform, export_settings: types.ConfigSettings, key_lis
         # log.info("Applications do not exist in SonarCloud, export skipped")
         raise exceptions.UnsupportedOperation("Applications do not exist in SonarCloud, export skipped")
 
-    apps_settings = {k: app.export(export_settings) for k, app in get_list(endpoint, key_list).items()}
-    for k in apps_settings:
-        # remove key from JSON value, it's already the dict key
-        apps_settings[k].pop("key")
-    return dict(sorted(apps_settings.items()))
+    apps_settings = {}
+    for k, app in sorted(get_list(endpoint, key_list).items()):
+        app_json = app.export(export_settings)
+        if write_q:
+            write_q.put(app_json)
+        else:
+            app_json.pop("key")
+            apps_settings[k] = app_json
+    if write_q:
+        write_q.put(None)
+    return apps_settings
 
 
 def audit(endpoint: pf.Platform, audit_settings: types.ConfigSettings, key_list: types.KeyList = None) -> list[problem.Problem]:

sonar/groups.py

@@ -19,6 +19,7 @@
 #
 
 from __future__ import annotations
+from queue import Queue
 from typing import Optional
 import sonar.logging as log
 import sonar.platform as pf
@@ -256,7 +257,9 @@ def get_list(endpoint: pf.Platform) -> dict[str, Group]:
     return search(endpoint)
 
 
-def export(endpoint: pf.Platform, export_settings: types.ConfigSettings, key_list: types.KeyList = None) -> types.ObjectJsonRepr:
+def export(
+    endpoint: pf.Platform, export_settings: types.ConfigSettings, key_list: Optional[types.KeyList] = None, write_q: Optional[Queue] = None
+) -> types.ObjectJsonRepr:
     """Exports groups representation in JSON
 
     :param Platform endpoint: reference to the SonarQube platform
@@ -272,6 +275,9 @@ def export(endpoint: pf.Platform, export_settings: types.ConfigSettings, key_lis
         if not export_settings["FULL_EXPORT"] and g_obj.is_default():
             continue
         g_list[g_name] = "" if g_obj.description is None else g_obj.description
+    if write_q:
+        write_q.put(g_list)
+        write_q.put(None)
     return g_list
 
 

sonar/platform.py

@@ -27,6 +27,7 @@
 from http import HTTPStatus
 import sys
 import os
+from queue import Queue
 from typing import Optional
 import time
 import datetime
@@ -869,7 +870,9 @@ def convert_for_yaml(original_json: types.ObjectJsonRepr) -> types.ObjectJsonRep
     return original_json
 
 
-def export(endpoint: Platform, export_settings: types.ConfigSettings, key_list: types.KeyList = None) -> types.ObjectJsonRepr:
+def export(
+    endpoint: Platform, export_settings: types.ConfigSettings, key_list: Optional[types.KeyList] = None, write_q: Optional[Queue] = None
+) -> types.ObjectJsonRepr:
     """Exports all or a list of projects configuration as dict
 
     :param Platform endpoint: reference to the SonarQube platform
@@ -878,4 +881,8 @@ def export(endpoint: Platform, export_settings: types.ConfigSettings, key_list:
     :return: Platform settings
     :rtype: ObjectJsonRepr
     """
-    return endpoint.export(export_settings)
+    exp = endpoint.export(export_settings)
+    if write_q:
+        write_q.put(exp)
+        write_q.put(None)
+    return exp

sonar/portfolios.py

@@ -24,8 +24,8 @@
 """
 
 from __future__ import annotations
+from queue import Queue
 from typing import Union, Optional
-import time
 import json
 import datetime
 from http import HTTPStatus
@@ -728,7 +728,9 @@ def search_by_key(endpoint: pf.Platform, key: str) -> types.ApiPayload:
     return util.search_by_key(endpoint, key, Portfolio.SEARCH_API, "components")
 
 
-def export(endpoint: pf.Platform, export_settings: types.ConfigSettings, key_list: types.KeyList = None) -> types.ObjectJsonRepr:
+def export(
+    endpoint: pf.Platform, export_settings: types.ConfigSettings, key_list: Optional[types.KeyList] = None, write_q: Optional[Queue] = None
+) -> types.ObjectJsonRepr:
     """Exports portfolios as JSON
 
     :param Platform endpoint: Reference to the SonarQube platform
@@ -749,8 +751,12 @@ def export(endpoint: pf.Platform, export_settings: types.ConfigSettings, key_lis
     for k, p in sorted(get_list(endpoint=endpoint, key_list=key_list).items()):
         try:
             if not p.is_sub_portfolio:
-                exported_portfolios[k] = p.export(export_settings)
-                exported_portfolios[k].pop("key")
+                exp = p.export(export_settings)
+                if write_q:
+                    write_q.put(exp)
+                else:
+                    exp.pop("key")
+                    exported_portfolios[k] = exp
             else:
                 log.debug("Skipping export of %s, it's a standard sub-portfolio", str(p))
         except HTTPError as e:
@@ -760,6 +766,8 @@ def export(endpoint: pf.Platform, export_settings: types.ConfigSettings, key_lis
         i += 1
         if i % 10 == 0 or i == nb_portfolios:
             log.info("Exported %d/%d portfolios (%d%%)", i, nb_portfolios, (i * 100) // nb_portfolios)
+    if write_q:
+        write_q.put(None)
     return exported_portfolios
 
 

sonar/projects.py

@@ -1059,7 +1059,7 @@ def export(self, export_settings: types.ConfigSettings, settings_list: dict[str,
         except Exception as e:
             log.critical("Connecting error %s while exporting %s, export of this project interrupted", str(self), str(e))
             json_data["error"] = f"Exception {str(e)} while exporting project, export interrupted"
-        log.info("Exporting %s done", str(self))
+        log.debug("Exporting %s done", str(self))
         return util.remove_nones(json_data)
 
     def new_code(self) -> str:
@@ -1473,13 +1473,13 @@ def audit(endpoint: pf.Platform, audit_settings: types.ConfigSettings, key_list:
     return problems
 
 
-def __export_thread(queue: Queue[Project], results: dict[str, str], export_settings: types.ConfigSettings) -> None:
+def __export_thread(queue: Queue[Project], results: dict[str, str], export_settings: types.ConfigSettings, write_q: Optional[Queue] = None) -> None:
     """Project export callback function for multitheaded export"""
     while not queue.empty():
         project = queue.get()
         exp_json = project.export(export_settings=export_settings)
-        if export_settings.get("WRITE_QUEUE", None):
-            export_settings["WRITE_QUEUE"].put(exp_json)
+        if write_q:
+            write_q.put(exp_json)
         else:
             results[project.key] = exp_json
             results[project.key].pop("key", None)
@@ -1489,10 +1489,11 @@ def __export_thread(queue: Queue[Project], results: dict[str, str], export_setti
         if nb % 10 == 0 or nb == tot:
             log.info("%d/%d projects exported (%d%%)", nb, tot, (nb * 100) // tot)
         queue.task_done()
-    log.info("Putting DONE in queue %s", str(export_settings["WRITE_QUEUE"]))
 
 
-def export(endpoint: pf.Platform, export_settings: types.ConfigSettings, key_list: types.KeyList = None) -> types.ObjectJsonRepr:
+def export(
+    endpoint: pf.Platform, export_settings: types.ConfigSettings, key_list: Optional[types.KeyList] = None, write_q: Optional[Queue] = None
+) -> types.ObjectJsonRepr:
     """Exports all or a list of projects configuration as dict
 
     :param Platform endpoint: reference to the SonarQube platform
@@ -1514,12 +1515,13 @@ def export(endpoint: pf.Platform, export_settings: types.ConfigSettings, key_lis
     project_settings = {}
     for i in range(export_settings.get("THREADS", 8)):
         log.debug("Starting project export thread %d", i)
-        worker = Thread(target=__export_thread, args=(q, project_settings, export_settings))
-        worker.setDaemon(True)
-        worker.setName(f"ProjectExport{i}")
+        worker = Thread(target=__export_thread, args=(q, project_settings, export_settings, write_q))
+        worker.daemon = True
+        worker.name = f"ProjectExport{i}"
         worker.start()
     q.join()
-    export_settings["WRITE_QUEUE"].put(None)
+    if write_q:
+        write_q.put(None)
     return dict(sorted(project_settings.items()))
 
 

sonar/qualitygates.py

@@ -24,8 +24,8 @@
 """
 
 from __future__ import annotations
-
-from typing import Union
+from queue import Queue
+from typing import Union, Optional
 
 from http import HTTPStatus
 import json
@@ -380,7 +380,9 @@ def get_list(endpoint: pf.Platform) -> dict[str, QualityGate]:
     return qg_list
 
 
-def export(endpoint: pf.Platform, export_settings: types.ConfigSettings, key_list: types.KeyList = None) -> types.ObjectJsonRepr:
+def export(
+    endpoint: pf.Platform, export_settings: types.ConfigSettings, key_list: Optional[types.KeyList] = None, write_q: Optional[Queue] = None
+) -> types.ObjectJsonRepr:
     """Exports quality gates as JSON
 
     :param Platform endpoint: Reference to the Sonar platform
@@ -390,7 +392,11 @@ def export(endpoint: pf.Platform, export_settings: types.ConfigSettings, key_lis
     :rtype: ObjectJsonRepr
     """
     log.info("Exporting quality gates")
-    return {k: qg.to_json(export_settings) for k, qg in sorted(get_list(endpoint).items())}
+    qg_list = {k: qg.to_json(export_settings) for k, qg in sorted(get_list(endpoint).items())}
+    if write_q:
+        write_q.put(qg_list)
+        write_q.put(None)
+    return qg_list
 
 
 def import_config(endpoint: pf.Platform, config_data: types.ObjectJsonRepr, key_list: types.KeyList = None) -> bool: