Merge branch 'main' into pr

Author: kohlisid

examples/sourcetransform/async_event_time_filter/Dockerfile

@@ -0,0 +1,55 @@
+####################################################################################################
+# builder: install needed dependencies
+####################################################################################################
+
+FROM python:3.10-slim-bullseye AS builder
+
+ENV PYTHONFAULTHANDLER=1 \
+  PYTHONUNBUFFERED=1 \
+  PYTHONHASHSEED=random \
+  PIP_NO_CACHE_DIR=on \
+  PIP_DISABLE_PIP_VERSION_CHECK=on \
+  PIP_DEFAULT_TIMEOUT=100 \
+  POETRY_VERSION=1.2.2 \
+  POETRY_HOME="/opt/poetry" \
+  POETRY_VIRTUALENVS_IN_PROJECT=true \
+  POETRY_NO_INTERACTION=1 \
+  PYSETUP_PATH="/opt/pysetup"
+
+ENV EXAMPLE_PATH="$PYSETUP_PATH/examples/sourcetransform/async_event_time_filter"
+ENV VENV_PATH="$EXAMPLE_PATH/.venv"
+ENV PATH="$POETRY_HOME/bin:$VENV_PATH/bin:$PATH"
+
+RUN apt-get update \
+    && apt-get install --no-install-recommends -y \
+        curl \
+        wget \
+        # deps for building python deps
+        build-essential \
+    && apt-get install -y git \
+    && apt-get clean && rm -rf /var/lib/apt/lists/* \
+    \
+    # install dumb-init
+    && wget -O /dumb-init https://github.com/Yelp/dumb-init/releases/download/v1.2.5/dumb-init_1.2.5_x86_64 \
+    && chmod +x /dumb-init \
+    && curl -sSL https://install.python-poetry.org | python3 -
+
+####################################################################################################
+# udf: used for running the udf vertices
+####################################################################################################
+FROM builder AS udf
+
+WORKDIR $PYSETUP_PATH
+COPY ./ ./
+
+WORKDIR $EXAMPLE_PATH
+RUN poetry lock
+RUN poetry install --no-cache --no-root && \
+    rm -rf ~/.cache/pypoetry/
+
+RUN chmod +x entry.sh
+
+ENTRYPOINT ["/dumb-init", "--"]
+CMD ["sh", "-c", "$EXAMPLE_PATH/entry.sh"]
+
+EXPOSE 5000

examples/sourcetransform/async_event_time_filter/Makefile

@@ -0,0 +1,22 @@
+TAG ?= stable
+PUSH ?= false
+IMAGE_REGISTRY = quay.io/numaio/numaflow-python/async-mapt-event-time-filter:${TAG}
+DOCKER_FILE_PATH = examples/sourcetransform/async_event_time_filter/Dockerfile
+
+.PHONY: update
+update:
+	poetry update -vv
+
+.PHONY: image-push
+image-push: update
+	cd ../../../ && docker buildx build \
+	-f ${DOCKER_FILE_PATH} \
+	-t ${IMAGE_REGISTRY} \
+	--platform linux/amd64,linux/arm64 . --push
+
+.PHONY: image
+image: update
+	cd ../../../ && docker build \
+	-f ${DOCKER_FILE_PATH} \
+	-t ${IMAGE_REGISTRY} .
+	@if [ "$(PUSH)" = "true" ]; then docker push ${IMAGE_REGISTRY}:${TAG}; fi

examples/sourcetransform/async_event_time_filter/entry.sh

@@ -0,0 +1,4 @@
+#!/bin/sh
+set -eux
+
+python example.py

examples/sourcetransform/async_event_time_filter/example.py

@@ -0,0 +1,48 @@
+import datetime
+import logging
+
+from pynumaflow.sourcetransformer import Messages, Message, Datum
+from pynumaflow.sourcetransformer import SourceTransformAsyncServer
+
+"""
+This is a simple User Defined Function example which receives a message, applies the following
+data transformation, and returns the message.
+If the message event time is before year 2022, drop the message with event time unchanged.
+If it's within year 2022, update the tag to "within_year_2022" and 
+update the message event time to Jan 1st 2022.
+Otherwise, (exclusively after year 2022), update the tag to "after_year_2022" and update the
+message event time to Jan 1st 2023.
+"""
+
+january_first_2022 = datetime.datetime.fromtimestamp(1640995200)
+january_first_2023 = datetime.datetime.fromtimestamp(1672531200)
+
+
+async def my_handler(keys: list[str], datum: Datum) -> Messages:
+    val = datum.value
+    event_time = datum.event_time
+    messages = Messages()
+
+    if event_time < january_first_2022:
+        logging.info("Got event time:%s, it is before 2022, so dropping", event_time)
+        messages.append(Message.to_drop(event_time))
+    elif event_time < january_first_2023:
+        logging.info(
+            "Got event time:%s, it is within year 2022, so forwarding to within_year_2022",
+            event_time,
+        )
+        messages.append(
+            Message(value=val, event_time=january_first_2022, tags=["within_year_2022"])
+        )
+    else:
+        logging.info(
+            "Got event time:%s, it is after year 2022, so forwarding to after_year_2022", event_time
+        )
+        messages.append(Message(value=val, event_time=january_first_2023, tags=["after_year_2022"]))
+
+    return messages
+
+
+if __name__ == "__main__":
+    grpc_server = SourceTransformAsyncServer(my_handler)
+    grpc_server.start()

examples/sourcetransform/async_event_time_filter/pyproject.toml

@@ -0,0 +1,15 @@
+[tool.poetry]
+name = "async-mapt-event-time-filter"
+version = "0.2.4"
+description = ""
+authors = ["Numaflow developers"]
+readme = "README.md"
+packages = [{include = "mapt_event_time_filter"}]
+
+[tool.poetry.dependencies]
+python = ">=3.9, <3.12"
+pynumaflow = { path = "../../../"}
+
+[build-system]
+requires = ["poetry-core"]
+build-backend = "poetry.core.masonry.api"

pynumaflow/_constants.py

@@ -7,9 +7,11 @@
 SIDE_INPUT_DIR_PATH = "/var/numaflow/side-inputs"
 ENV_UD_CONTAINER_TYPE = "NUMAFLOW_UD_CONTAINER_TYPE"
 
-# Get container type from env var, default to unknown-container
+# Error Constants
+RUNTIME_APPLICATION_ERRORS_PATH = "/var/numaflow/runtime/application-errors"
+CURRENT_CRITICAL_ERROR_FILE = "current-udf.json"
+INTERNAL_ERROR_CODE = "Internal error"
 CONTAINER_TYPE = os.getenv(ENV_UD_CONTAINER_TYPE, "unknown-container")
-# UDF exception error string with container type
 ERR_UDF_EXCEPTION_STRING = f"UDF_EXECUTION_ERROR({CONTAINER_TYPE})"
 
 # Socket configs

pynumaflow/errors/__init__.py

@@ -0,0 +1,3 @@
+from pynumaflow.errors.errors import persist_critical_error
+
+__all__ = ["persist_critical_error"]

pynumaflow/errors/_dtypes.py

@@ -0,0 +1,16 @@
+from dataclasses import dataclass, asdict
+
+
+@dataclass
+class _RuntimeErrorEntry:
+    """Represents a runtime error entry to be persisted."""
+
+    container: str
+    timestamp: int
+    code: str
+    message: str
+    details: str
+
+    def to_dict(self) -> dict:
+        """Converts the dataclass instance to a dictionary."""
+        return asdict(self)

pynumaflow/errors/errors.py

@@ -0,0 +1,82 @@
+import os
+import json
+import threading
+import time
+from pynumaflow._constants import (
+    CONTAINER_TYPE,
+    RUNTIME_APPLICATION_ERRORS_PATH,
+    CURRENT_CRITICAL_ERROR_FILE,
+    INTERNAL_ERROR_CODE,
+)
+from pynumaflow.errors._dtypes import _RuntimeErrorEntry
+from typing import Union
+
+
+class _PersistErrorOnce:
+    """Ensures that the persist_critical_error function is executed only once."""
+
+    def __init__(self):
+        self.done = False
+        self.lock = threading.Lock()
+
+    def execute(self, func, *args, **kwargs):
+        with self.lock:
+            if self.done:
+                raise RuntimeError("Persist critical error function has already been executed.")
+            self.done = True
+            return func(*args, **kwargs)
+
+
+_persist_error_once = _PersistErrorOnce()
+
+
+def persist_critical_error(
+    error_code: str, error_message: str, error_details: str
+) -> Union[RuntimeError, None]:
+    """
+    Persists a critical error to a file. This function will only execute once.
+    Logs the error if persisting to the file fails.
+    Returns None if successful, or raises RuntimeError if already executed.
+    """
+    try:
+        _persist_error_once.execute(
+            _persist_critical_error_to_file,
+            error_code,
+            error_message,
+            error_details,
+            RUNTIME_APPLICATION_ERRORS_PATH,
+        )
+    except RuntimeError as e:
+        return e
+    except Exception as e:
+        print(f"Error in persisting critical error: {e}")
+    return None
+
+
+def _persist_critical_error_to_file(
+    error_code: str, error_message: str, error_details: str, dir_path: str
+):
+    """Internal function to persist a critical error to a file."""
+
+    os.makedirs(dir_path, mode=0o777, exist_ok=True)
+    container_dir = os.path.join(dir_path, CONTAINER_TYPE)
+    os.makedirs(container_dir, mode=0o777, exist_ok=True)
+
+    current_file_path = os.path.join(container_dir, CURRENT_CRITICAL_ERROR_FILE)
+    error_code = error_code or INTERNAL_ERROR_CODE
+    current_timestamp = int(time.time())
+
+    runtime_error_entry = _RuntimeErrorEntry(
+        container=CONTAINER_TYPE,
+        timestamp=current_timestamp,
+        code=error_code,
+        message=error_message,
+        details=error_details,
+    )
+
+    with open(current_file_path, "w") as f:
+        json.dump(runtime_error_entry.to_dict(), f)
+
+    final_file_name = f"{current_timestamp}-udf.json"
+    final_file_path = os.path.join(container_dir, final_file_name)
+    os.rename(current_file_path, final_file_path)

pynumaflow/sourcetransformer/__init__.py

@@ -7,6 +7,7 @@
 )
 from pynumaflow.sourcetransformer.multiproc_server import SourceTransformMultiProcServer
 from pynumaflow.sourcetransformer.server import SourceTransformServer
+from pynumaflow.sourcetransformer.async_server import SourceTransformAsyncServer
 
 __all__ = [
     "Message",
@@ -16,4 +17,5 @@
     "SourceTransformServer",
     "SourceTransformer",
     "SourceTransformMultiProcServer",
+    "SourceTransformAsyncServer",
 ]