memray tracing

Author: JoshuaFox

.gcloudignore

@@ -2,6 +2,7 @@
 test_scripts/
 
 sample_data/
+memray/
 README.md
 *.sh
 TEMP*

.gitignore

@@ -17,3 +17,4 @@ TEMP*
 virtenv*/
 
 
+memray/

app.yaml

@@ -8,6 +8,8 @@ inbound_services:
   - warmup
 env_variables:
   GAE_USE_SOCKETS_HTTPLIB: False
+  PYTHONMALLOC: malloc
 app_engine_apis: true
 
 
+

cron.yaml

@@ -2,5 +2,5 @@
 cron:
   - description: "Invoke /schedule which sends out messages triggering do_label per project/plugin."
     url: /schedule
-    schedule: every day 14:13
+    schedule: every day 13:27
     target: iris3

deploy.sh

@@ -89,16 +89,9 @@ echo "Project ID $PROJECT_ID"
 gcloud config set project "$PROJECT_ID"
 
 # Must have one of these config (meanwhile, config-dev.yaml is only for local use)
-if [[  -f "config-test.yaml" ]]; then
-  if [[  -f "config.yaml" ]]; then
-       echo >&2 "Cannot have config.yaml if config-test.yaml exists"
-       exit 1
-  fi
-else
-   if [[  ! -f "config.yaml" ]]; then
+if [[ ! -f "config-test.yaml" ]]  && [[ ! -f "config.yaml" ]]; then
        echo >&2 "config.yaml Must have either config.yaml (use config.yaml.original as an example) or config-test.yaml"
        exit 1
-  fi
 fi
 
 

main.py

@@ -7,7 +7,11 @@
 
 print("Initializing ", file=sys.stderr)
 import flask
-from util.gcp_utils import increment_invocation_count, invocation_count, memray_filename
+from util.gcp_utils import (
+    increment_invocation_count,
+    count_invocations_by_path,
+    memray_filename,
+)
 from util.detect_gae import detect_gae
 
 app = flask.Flask(__name__)
@@ -66,6 +70,9 @@
 logging.info(
     "env GAE_USE_SOCKETS_HTTPLIB is %s", os.environ.get("GAE_USE_SOCKETS_HTTPLIB")
 )
+logging.info(
+    "env PYTHONMALLOC is %s", os.environ.get("PYTHONMALLOC")
+)
 
 PluginHolder.init()
 
@@ -74,42 +81,43 @@
 def list_memray():
     s = ""
     for f in os.listdir(tmp_dir()):
-        s += f"<a href='/tmpfile?filename={f}'></a><br>"
-    return s, 200
+        s += f"<a href='/memrayfiles/{f}'>{f}</a><br>\n"
+    if not s:
+        s = "NONE FOUND"
+    return Response(s, mimetype="text/html", status=200)
 
 
-@app.route("/tmpfile")
-def tmpfile():
-    fn = flask.request.args['filename']
-    with open(f"{tmp_dir()}/{fn}", 'b') as f:
+@app.route("/memrayfiles/<filename>")
+def memrayfiles(filename):
+
+    with open(f"{tmp_dir()}/{filename}", "rb") as f:
         content = f.read()
-    return Response(content, mimetype='application/octet-stream', status=200)
+
+    return Response(content, mimetype="application/octet-stream", status=200)
 
 
 @app.route("/")
 def index():
-    with memray.Tracker(memray_filename()):
+    with memray.Tracker(memray_filename("index")):
         increment_invocation_count("index")
         with gae_memory_logging("index"):
             msg = iris_homepage_text()
             if config_utils.is_test_or_dev_configuration():
                 msg += "\nI'm running in test or dev mode."
 
-            logging.info("index(); invocations of GAE instance : %s", invocation_count())
+            logging.info(
+                "index(); invocations of GAE instance : %s", count_invocations_by_path()
+            )
             return Response(msg, mimetype="text/plain", status=200)
 
 
 @app.route("/_ah/warmup")
 def warmup():
-    with memray.Tracker(memray_filename()):
-
+    with memray.Tracker(memray_filename("warmup")):
         increment_invocation_count("warmup")
         with gae_memory_logging("warmup"):
             logging.info("warmup() called")
-        try:
-            1 / 0
-        except Exception:
-            logging.exception("eeeeee")
+
         return "", 200, {}
 
 
@@ -119,7 +127,7 @@ def schedule():
     """
     Send out a message per-plugin per-project to label all objects of that type and project.
     """
-    with memray.Tracker(memray_filename):
+    with memray.Tracker(memray_filename("schedule")):
 
         increment_invocation_count("schedule")
         with gae_memory_logging("schedule"):
@@ -162,9 +170,9 @@ def __get_enabled_projects():
         raise Exception("No projects enabled at all")
 
     if (
-            not detect_gae()
-            or is_test_or_dev_configuration()
-            or is_in_test_or_dev_project(current_project_id())
+        not detect_gae()
+        or is_test_or_dev_configuration()
+        or is_in_test_or_dev_project(current_project_id())
     ):
         max_proj_in_dev = 3
         if len(enabled_projs) > max_proj_in_dev:
@@ -173,8 +181,8 @@ def __get_enabled_projects():
                 + f"to avoid accidentally flooding the system."
                 + f"{max_proj_in_dev} projects are available, which exceeds that."
                 + f"To avoid this limit, use config.yaml rather than config-dev.yaml or config-test.yaml,"
-                  f"edit test_or_dev_project_markers in the config file,"
-                  f"and run in the cloud rather than locally."
+                f"edit test_or_dev_project_markers in the config file,"
+                f"and run in the cloud rather than locally."
             )
     return enabled_projs
 
@@ -184,9 +192,9 @@ def __send_pubsub_per_projectplugin(configured_projects):
     for project_id in configured_projects:
         for plugin_cls in PluginHolder.plugins:
             if (
-                    not plugin_cls.is_labeled_on_creation()
-                    or plugin_cls.relabel_on_cron()
-                    or config_utils.label_all_on_cron()
+                not plugin_cls.is_labeled_on_creation()
+                or plugin_cls.relabel_on_cron()
+                or config_utils.label_all_on_cron()
             ):
                 pubsub_utils.publish(
                     msg=json.dumps(
@@ -210,7 +218,7 @@ def __send_pubsub_per_projectplugin(configured_projects):
 
 @app.route("/label_one", methods=["POST"])
 def label_one():
-    with memray.Tracker(memray_filename()):
+    with memray.Tracker(memray_filename("label_one")):
 
         increment_invocation_count("label_one")
         with gae_memory_logging("label_one"):
@@ -261,7 +269,9 @@ def label_one():
                 # since most errors are unrecoverable.
                 return "OK", 200
             except Exception:
-                project_id = data.get("resource", {}).get("labels", {}).get("project_id")
+                project_id = (
+                    data.get("resource", {}).get("labels", {}).get("project_id")
+                )
                 logging.exception("Error on label_one %s %s", plugins_found, project_id)
                 return "Error", 500
 
@@ -320,14 +330,16 @@ def __extract_pubsub_content() -> Dict:
 
 @app.route("/do_label", methods=["POST"])
 def do_label():
-    with memray.Tracker(memray_filename()):
+    with memray.Tracker(memray_filename("do_label")):
         increment_invocation_count("do_label")
         with gae_memory_logging("do_label"):
 
             """Receive a push message from PubSub, sent from schedule() above,
             with instructions to label all objects of a given plugin and project_id.
             """
-            project_id = ""  # set up variables to allow logging in Exception block at end
+            project_id = (
+                ""  # set up variables to allow logging in Exception block at end
+            )
             plugin_class_name = ""
             try:
                 data = __extract_pubsub_content()
@@ -344,7 +356,9 @@ def do_label():
                     project_id = data["project_id"]
                     with timing(f"do_label {plugin_class_name} {project_id}"):
                         logging.info(
-                            "do_label() for %s in %s", plugin.__class__.__name__, project_id
+                            "do_label() for %s in %s",
+                            plugin.__class__.__name__,
+                            project_id,
                         )
                         plugin.label_all(project_id)
                     logging.info("OK on do_label %s %s", plugin_class_name, project_id)
@@ -354,7 +368,9 @@ def do_label():
 
                 return "OK", 200
             except Exception:
-                logging.exception("Error on do_label %s %s", plugin_class_name, project_id)
+                logging.exception(
+                    "Error on do_label %s %s", plugin_class_name, project_id
+                )
                 return "Error", 500
 
 

test_scripts/integration_test.py

@@ -8,6 +8,7 @@
 from concurrent.futures import ThreadPoolExecutor
 from subprocess import CalledProcessError
 from typing import Union, List
+from urllib.error import HTTPError
 
 from test_scripts.utils_for_tests import assert_root_path
 from util.config_utils import iris_prefix, iris_homepage_text
@@ -169,26 +170,39 @@ def wait_for_traffic_shift(deployment_project):
 
     url = f"https://iris3-dot-{deployment_project}.uc.r.appspot.com/"
     while time.time() - start_wait_for_trafficshift < 180:  # break after 180 sec
-        with urllib.request.urlopen(url) as response:
-            txt_b = response.read()
-            txt = str(txt_b, "UTF-8")
-            if iris_homepage_text() in txt:
-                print(
-                    "Wait for traffic shift took",
-                    int(1000 * (time.time() - start_wait_for_trafficshift)),
-                    "msec",
-                )
-                return  # Could alternatively check for the iris_prefix, which is meant to be here unique.
-            print(
-                'Site now has "',
-                txt[:100],
-                '"not including the expected ',
-                iris_prefix(),
-            )
+        try:
+            found_it = __check_for_new_v(start_wait_for_trafficshift, url)
+            if found_it:
+                return
+            else:
+                time.sleep(3)
+                continue
+        except HTTPError as e:
+            logging.error(e)  # Keep trying despite exception
 
     raise TimeoutError(time.time() - start_wait_for_trafficshift)
 
 
+def __check_for_new_v(start_wait_for_trafficshift, url) -> bool:
+    with urllib.request.urlopen(url) as response:
+        txt_b = response.read()
+        txt = str(txt_b, "UTF-8")
+        if iris_homepage_text() in txt:
+            print(
+                "Wait for traffic shift took",
+                int(1000 * (time.time() - start_wait_for_trafficshift)),
+                "msec",
+            )
+            return True
+        print(
+            'Site now has "',
+            txt[:100],
+            '"not including the expected ',
+            iris_prefix(),
+        )
+        return False
+
+
 def fill_in_config_template(
     run_id, deployment_project, test_project, pubsub_test_token
 ):
@@ -265,6 +279,7 @@ def get_run_id():
     else:
         # Random value to distinguish this test runs from others
         run_id = random_str(4)
+    print("run_id is", run_id, file=sys.stderr)
     return run_id
 
 
@@ -337,5 +352,5 @@ def create_resources(test_project, run_id, gce_zone):
     assert_root_path()
 
     main()
-    print("Exiting with ", "failure" if exit_code else "success")
+    print("Exiting with", "failure" if exit_code else "success")
     sys.exit(exit_code)