Adds progress update to execution env (#129)

* Adds progress update to execution env

* Removes print statements

* New version but still not finished

* Fixes bugs

* Is managed and is demo rest requests (#130)

* Fixes logic

* Removes print statement

* Modifies docker logs api call

* Removes timestamps from docker logs call

* Small removales

* Changes logic to go in steps and extends tail

* Adds weights for steps of calculation

* Checks if progress is greater than 0.95

* Resolves PR comments

* Adds support for reruns of attribute calc

* Dynamic log changes

* Submodules change to dev

---------

Co-authored-by: lumburovskalina <[email protected]>
Co-authored-by: JWittmeyer <[email protected]>

Author: 3 people

alembic/versions/03d19eada266_progress_field_attributes.py

@@ -0,0 +1,28 @@
+"""Progress Field attributes
+
+Revision ID: 03d19eada266
+Revises: bb87177d46b5
+Create Date: 2023-05-15 14:36:17.587964
+
+"""
+from alembic import op
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision = '03d19eada266'
+down_revision = 'bb87177d46b5'
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.add_column('attribute', sa.Column('progress', sa.Float(), nullable=True))
+    # ### end Alembic commands ###
+
+
+def downgrade():
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.drop_column('attribute', 'progress')
+    # ### end Alembic commands ###

controller/attribute/manager.py

@@ -316,6 +316,7 @@ def __calculate_user_attribute_all_records(
         )
         request_reupload_docbins(project_id)
 
+    util.set_progress(project_id, attribute_item, 1.0)
     attribute.update(
         project_id=project_id,
         attribute_id=attribute_id,

controller/attribute/util.py

@@ -1,34 +1,52 @@
+import time
+from typing import Any, List
+import uuid
 import docker
 import json
 import os
 import pytz
-import re
-from datetime import datetime
-from submodules.model.business_objects import attribute, record, project, tokenization
-from submodules.model.enums import DataTypes
+
+import datetime
+from dateutil import parser
+
+from submodules.model.business_objects import (
+    attribute,
+    general,
+    record,
+    project,
+    tokenization,
+)
+from submodules.model.models import Attribute
 from submodules.s3 import controller as s3
+from util import daemon, notification
 
 client = docker.from_env()
 image = os.getenv("AC_EXEC_ENV_IMAGE")
 exec_env_network = os.getenv("LF_NETWORK")
+__tz = pytz.timezone("Europe/Berlin")
+
+__containers_running = {}
 
 
 def add_log_to_attribute_logs(
     project_id: str, attribute_id: str, log: str, append_to_logs: bool = True
 ) -> None:
     attribute_item = attribute.get(project_id, attribute_id)
-    berlin_now = datetime.now(pytz.timezone("Europe/Berlin"))
-    if append_to_logs:
-        logs = attribute_item.logs
-        logs.append(" ".join([berlin_now.strftime("%Y-%m-%dT%H:%M:%S"), log]))
+    berlin_now = datetime.datetime.now(__tz)
+    time_string = berlin_now.strftime("%Y-%m-%dT%H:%M:%S")
+    line = f"{time_string} {log}"
+
+    if not append_to_logs or not attribute_item.logs:
+        logs = [line]
+        attribute.update(
+            project_id=project_id,
+            attribute_id=attribute_id,
+            logs=logs,
+            with_commit=True,
+        )
     else:
-        logs = [" ".join([berlin_now.strftime("%Y-%m-%dT%H:%M:%S"), log])]
-    attribute.update(
-        project_id=project_id,
-        attribute_id=attribute_id,
-        logs=logs,
-        with_commit=True,
-    )
+        attribute_item.logs.append(line)
+        general.commit()
 
 
 def prepare_sample_records_doc_bin(attribute_id: str, project_id: str) -> str:
@@ -55,6 +73,14 @@ def run_attribute_calculation_exec_env(
 ) -> None:
     attribute_item = attribute.get(project_id, attribute_id)
 
+    if attribute_item.logs:
+        add_log_to_attribute_logs(
+            project_id,
+            attribute_id,
+            "re-run attribute calculation",
+            append_to_logs=False,
+        )
+
     prefixed_function_name = f"{attribute_id}_fn"
     prefixed_payload = f"{attribute_id}_payload.json"
     project_item = project.get(project_id)
@@ -73,24 +99,32 @@ def run_attribute_calculation_exec_env(
         attribute_item.data_type,
     ]
 
-    container = client.containers.run(
+    container_name = str(uuid.uuid4())
+    container = client.containers.create(
         image=image,
         command=command,
-        remove=True,
+        auto_remove=True,
         detach=True,
         network=exec_env_network,
     )
-
-    logs = [
+    set_progress(project_id, attribute_item, 0.05)
+    __containers_running[container_name] = True
+    daemon.run(
+        read_container_logs_thread,
+        project_id,
+        container_name,
+        str(attribute_item.id),
+        container,
+    )
+    container.start()
+    attribute_item.logs = [
         line.decode("utf-8").strip("\n")
         for line in container.logs(
             stream=True, stdout=True, stderr=True, timestamps=True
         )
+        if "progress" not in line.decode("utf-8")
     ]
-
-    attribute.update(
-        project_id=project_id, attribute_id=attribute_id, logs=logs, with_commit=True
-    )
+    del __containers_running[container_name]
 
     try:
         payload = s3.get_object(org_id, project_id + "/" + prefixed_payload)
@@ -104,5 +138,96 @@ def run_attribute_calculation_exec_env(
         s3.delete_object(org_id, project_id + "/" + doc_bin)
     s3.delete_object(org_id, project_id + "/" + prefixed_function_name)
     s3.delete_object(org_id, project_id + "/" + prefixed_payload)
+    set_progress(project_id, attribute_item, 0.9)
 
     return calculated_attributes
+
+
+def extend_logs(
+    project_id: str,
+    attribute: Attribute,
+    logs: List[str],
+) -> None:
+    if not logs or len(logs) == 0:
+        return
+
+    if not attribute.logs:
+        attribute.logs = logs
+    else:
+        all_logs = [l for l in attribute.logs]
+        all_logs += logs
+        attribute.logs = all_logs
+    general.commit()
+    # currently dummy since frontend doesn't have a log change yet
+    notification.send_organization_update(
+        project_id, f"attributes_updated:{str(attribute.id)}"
+    )
+
+
+def read_container_logs_thread(
+    project_id: str,
+    name: str,
+    attribute_id: str,
+    docker_container: Any,
+) -> None:
+
+    ctx_token = general.get_ctx_token()
+    # needs to be refetched since it is not thread safe
+    attribute_item = attribute.get(project_id, attribute_id)
+    previous_progress = -1
+    last_timestamp = None
+    c = 0
+    while name in __containers_running:
+        time.sleep(1)
+        c += 1
+        if c > 100:
+            ctx_token = general.remove_and_refresh_session(ctx_token, True)
+            attribute_item = attribute.get(project_id, attribute_id)
+        if not name in __containers_running:
+            break
+        try:
+            # timestamps included to filter out logs that have already been read
+            log_lines = docker_container.logs(
+                stdout=True,
+                stderr=True,
+                timestamps=True,
+                since=last_timestamp,
+            )
+        except:
+            # failsafe for containers that shut down during the read
+            break
+        current_logs = [
+            l for l in str(log_lines.decode("utf-8")).split("\n") if len(l.strip()) > 0
+        ]
+        if len(current_logs) == 0:
+            continue
+        last_entry = current_logs[-1]
+        last_timestamp_str = last_entry.split(" ")[0]
+        last_timestamp = parser.parse(last_timestamp_str).replace(
+            tzinfo=None
+        ) + datetime.timedelta(seconds=1)
+        non_progress_logs = [l for l in current_logs if "progress" not in l]
+        progress_logs = [l for l in current_logs if "progress" in l]
+        if len(non_progress_logs) > 0:
+            extend_logs(project_id, attribute_item, non_progress_logs)
+        if len(progress_logs) == 0:
+            continue
+        last_entry = float(progress_logs[-1].split("progress: ")[1].strip())
+        if previous_progress == last_entry:
+            continue
+        previous_progress = last_entry
+        set_progress(project_id, attribute_item, last_entry * 0.8 + 0.05)
+    general.remove_and_refresh_session(ctx_token)
+
+
+def set_progress(
+    project_id: str,
+    attribute: Attribute,
+    progress: float,
+) -> None:
+    final_progress = round(progress, 4)
+    attribute.progress = final_progress
+    general.commit()
+    notification.send_organization_update(
+        project_id, f"calculate_attribute:progress:{attribute.id}:{final_progress}"
+    )