Wrap raw types into a StageSchema class

CHANGELOG.md

@@ -15,6 +15,13 @@ See the License for the specific language governing permissions and
 limitations under the License.
 -->
 
+# Morpheus 23.07.03 (11 Oct 2023)
+
+## 🐛 Bug Fixes
+- Add pinned libwebp to resolve CVE ([#1236](https://github.com/nv-morpheus/Morpheus/pull/1236)) [@drobison00](https://github.com/drobison00)
+- Add libwebp to meta.yaml for CVE 2307 ([#1242](https://github.com/nv-morpheus/Morpheus/pull/1242)) [@drobison00](https://github.com/drobison00)
+- [BUG] Fix Control Message Utils & SQL Max Connections Exhaust ([#1243](https://github.com/nv-morpheus/Morpheus/pull/1243)) [@bsuryadevara](https://github.com/bsuryadevara)
+
 # Morpheus 23.07.02 (25 Jul 2023)
 
 ## 🐛 Bug Fixes

ci/conda/recipes/morpheus/conda_build_config.yaml

@@ -29,7 +29,7 @@ python:
   - 3.10
 
 boost:
-  - 1.74
+  - 1.82
 
 rapids_version:
   - 23.06

ci/conda/recipes/morpheus/meta.yaml

@@ -79,7 +79,8 @@ outputs:
         - docker-py 5.0.*
         - grpcio # Version determined from cudf
         - libmrc
-        - mlflow >=2.2.1,<3
+        - libwebp>=1.3.2 # Required for CVE mitigation: https://nvd.nist.gov/vuln/detail/CVE-2023-4863
+        - mlflow>=2.2.1,<3
         - mrc
         - networkx 3.1.*
         - numpydoc 1.4.*

ci/scripts/common.sh

@@ -43,8 +43,8 @@ export SKIP_YAPF=${SKIP_YAPF:-""}
 # Set BUILD_DIR to use a different build folder
 export BUILD_DIR=${BUILD_DIR:-"${MORPHEUS_ROOT}/build"}
 
-# Speficy the clang-tools version to use. Default 14
-export CLANG_TOOLS_VERSION=${CLANG_TOOLS_VERSION:-14}
+# Speficy the clang-tools version to use. Default 16
+export CLANG_TOOLS_VERSION=${CLANG_TOOLS_VERSION:-16}
 
 # Returns the `branch-YY.MM` that is used as the base for merging
 function get_base_branch() {

ci/scripts/github/common.sh

@@ -93,6 +93,10 @@ function update_conda_env() {
 
     if [[ "${SKIP_CONDA_ENV_UPDATE}" == "" ]]; then
         rapids-logger "Checking for updates to conda env"
+
+        # Remove default/conflicting channels from base image
+        rm /opt/conda/.condarc
+
         # Update the packages
         rapids-mamba-retry env update -n morpheus --prune -q --file ${ENV_YAML}
     fi

docker/Dockerfile

@@ -85,6 +85,7 @@ RUN --mount=type=cache,id=apt,target=/var/cache/apt \
         libcurand-${CUDA_MAJOR_VER}-${CUDA_MINOR_VER} \
         libcusolver-${CUDA_MAJOR_VER}-${CUDA_MINOR_VER} \
         libnuma1 \
+        openjdk-11-jre-headless \
         openssh-client \
         pkg-config \
         tar \

docker/conda/environments/cuda11.8_dev.yml

@@ -25,10 +25,10 @@ dependencies:
     ####### Morpheus Dependencies (keep sorted!) #######
     - automake=1.16.5
     - benchmark=1.6.1
-    - boost-cpp=1.74
+    - boost-cpp=1.82
     - cachetools=5.0.0
     - ccache>=3.7
-    - clangdev=14
+    - clangdev=16
     - click >=8
     - cmake=3.24
     - configargparse=1.5
@@ -58,18 +58,19 @@ dependencies:
     - grpcio
     - gtest>=1.13.0
     - gxx_linux-64=11.2
-    - include-what-you-use=0.18
+    - include-what-you-use=0.20
     - ipywidgets
     - isort
     - jupyter_core>=4.11.2,<5.0
     - jupyterlab
     - libgrpc>=1.49
     - librdkafka=1.9.2
+    - libwebp>=1.3.2 # Required for CVE mitigation: https://nvd.nist.gov/vuln/detail/CVE-2023-4863
     - mlflow>=2.2.1,<3
     - mrc=23.11
     - networkx=3.1
     - ninja=1.10
-    - nodejs=18.15.0
+    - nodejs=18.*
     - numba>=0.56.2
     - numpydoc=1.4
     - nvtabular=23.06
@@ -107,4 +108,7 @@ dependencies:
     ####### Morpheus Pip Dependencies (keep sorted!) #######
     - pip:
         # Add additional dev dependencies here
+        - databricks-connect
         - pytest-kafka==0.6.0
+        - pymilvus==2.3.1
+        - milvus

docker/conda/environments/cuda11.8_examples.yml

@@ -31,6 +31,7 @@ dependencies:
     - dgl=1.0.2
     - dill=0.3.6
     - distributed>=2023.1.1
+    - libwebp>=1.3.2 # Required for CVE mitigation: https://nvd.nist.gov/vuln/detail/CVE-2023-4863
     - mlflow>=2.2.1,<3
     - papermill=2.3.4
     - s3fs>=2023.6

docs/source/conf.py

@@ -164,10 +164,12 @@
 autodoc_mock_imports = [
     "cudf",  # Avoid loading GPU libraries during the documentation build
     "cupy",  # Avoid loading GPU libraries during the documentation build
+    "databricks.connect",
     "merlin",
     "morpheus.cli.commands",  # Dont document the CLI in Sphinx
     "nvtabular",
     "pandas",
+    "pymilvus",
     "tensorrt",
     "torch",
     "tqdm",

docs/source/developer_guide/guides/3_simple_cpp_stage.md

@@ -336,11 +336,13 @@ PYBIND11_MODULE(morpheus_example, m)
 
 ### Python Changes
 
-We need to make a few minor adjustments to our Python implementation of the `PassThruStage`.
+We need to make a few minor adjustments to our Python implementation of the `PassThruStage`. First, we import the new `morpheus_example` Python module we created in the previous section.
 
-As mentioned in the previous section, we will need to change the return value of the `supports_cpp_node` method to indicate that our stage now supports a C++ implementation.  Our `_build_single` method needs to be updated to build a C++ node when `morpheus.config.CppConfig.get_should_use_cpp()` is `True` using the `self._build_cpp_node()` method. The `_build_cpp_node()` method compares both `morpheus.config.CppConfig.get_should_use_cpp()` and `supports_cpp_node()` and returns `True` only when both methods return `True`.
+```python
+from _lib import morpheus_example as morpheus_example_cpp
+```
 
- We lazily import the new `morpheus_example` Python module we created in the previous section only when we need to build a C++ node since importing a compiled extension module comes at a performance cost.
+As mentioned in the previous section, we will need to change the return value of the `supports_cpp_node` method to indicate that our stage now supports a C++ implementation.  Our `_build_single` method needs to be updated to build a C++ node when `morpheus.config.CppConfig.get_should_use_cpp()` is `True` using the `self._build_cpp_node()` method. The `_build_cpp_node()` method compares both `morpheus.config.CppConfig.get_should_use_cpp()` and `supports_cpp_node()` and returns `True` only when both methods return `True`.
 
 ```python
 def supports_cpp_node(self):
@@ -349,7 +351,6 @@ def supports_cpp_node(self):
 ```python
 def _build_single(self, builder: mrc.Builder, input_stream: StreamPair) -> StreamPair:
     if self._build_cpp_node():
-        from _lib import morpheus_example as morpheus_example_cpp
         node = morpheus_example_cpp.PassThruStage(builder, self.unique_name)
     else:
         node = builder.make_node(self.unique_name, ops.map(self.on_data))

examples/developer_guide/1_simple_python_stage/pass_thru.py

@@ -23,7 +23,6 @@
 from morpheus.pipeline.single_port_stage import SinglePortStage
 
 
-# TODO: Document the PassThruTypeMixin
 @register_stage("pass-thru")
 class PassThruStage(PassThruTypeMixin, SinglePortStage):
     """

examples/developer_guide/2_2_rabbitmq/rabbitmq_source_stage.py

@@ -28,6 +28,7 @@
 from morpheus.messages.message_meta import MessageMeta
 from morpheus.pipeline.preallocator_mixin import PreallocatorMixin
 from morpheus.pipeline.single_output_source import SingleOutputSource
+from morpheus.pipeline.stage_schema import StageSchema
 
 logger = logging.getLogger(__name__)
 
@@ -85,8 +86,8 @@ def name(self) -> str:
     def supports_cpp_node(self) -> bool:
         return False
 
-    def output_type(self) -> type:
-        return MessageMeta
+    def compute_schema(self, schema: StageSchema):
+        schema.output_schema.set_type(MessageMeta)
 
     def stop(self):
         # Indicate we need to stop

examples/developer_guide/4_rabbitmq_cpp_stage/cmake/Configure_SimpleAmqpClient.cmake

@@ -41,8 +41,9 @@ function(find_and_configure_SimpleAmqpClient version)
   # Needed to pick up the generated export.h
   target_include_directories(SimpleAmqpClient PUBLIC "${rabbitmq_BINARY_DIR}/include")
 
-  # Suppress #warning deprecation messages from rabbitmq
-  target_compile_options(SimpleAmqpClient PRIVATE -Wno-cpp)
+  # Suppress #warning deprecation messages from rabbitmq and SimpleAmqpClient
+  # https://github.com/nv-morpheus/Morpheus/issues/1255
+  target_compile_options(SimpleAmqpClient PRIVATE -Wno-cpp -DBOOST_DISABLE_PRAGMA_MESSAGE)
 
 endfunction()
 

examples/developer_guide/4_rabbitmq_cpp_stage/rabbitmq_source_stage.py

@@ -28,6 +28,7 @@
 from morpheus.messages.message_meta import MessageMeta
 from morpheus.pipeline.preallocator_mixin import PreallocatorMixin
 from morpheus.pipeline.single_output_source import SingleOutputSource
+from morpheus.pipeline.stage_schema import StageSchema
 
 logger = logging.getLogger(__name__)
 
@@ -81,8 +82,8 @@ def name(self) -> str:
     def supports_cpp_node(self) -> bool:
         return True
 
-    def output_type(self) -> type:
-        return MessageMeta
+    def compute_schema(self, schema: StageSchema):
+        schema.output_schema.set_type(MessageMeta)
 
     def _build_source(self, builder: mrc.Builder) -> mrc.SegmentObject:
         if self._build_cpp_node():

examples/digital_fingerprinting/production/morpheus/dfp/stages/dfp_file_batcher_stage.py

@@ -26,6 +26,7 @@
 
 from morpheus.config import Config
 from morpheus.pipeline.single_port_stage import SinglePortStage
+from morpheus.pipeline.stage_schema import StageSchema
 
 logger = logging.getLogger(f"morpheus.{__name__}")
 
@@ -107,8 +108,8 @@ def accepted_types(self) -> typing.Tuple:
         """Accepted incoming types for this stage"""
         return (fsspec.core.OpenFiles, )
 
-    def output_type(self, parent_output_type: type) -> type:
-        return typing.Tuple[fsspec.core.OpenFiles, int]
+    def compute_schema(self, schema: StageSchema):
+        schema.output_schema.set_type(typing.Tuple[fsspec.core.OpenFiles, int])
 
     def on_data(self, file_objects: fsspec.core.OpenFiles) -> typing.List[typing.Tuple[fsspec.core.OpenFiles, int]]:
         """

examples/digital_fingerprinting/production/morpheus/dfp/stages/dfp_file_to_df.py

@@ -25,6 +25,7 @@
 from morpheus.controllers.file_to_df_controller import FileToDFController
 from morpheus.pipeline.preallocator_mixin import PreallocatorMixin
 from morpheus.pipeline.single_port_stage import SinglePortStage
+from morpheus.pipeline.stage_schema import StageSchema
 from morpheus.utils.column_info import DataFrameInputSchema
 
 logger = logging.getLogger(f"morpheus.{__name__}")
@@ -83,8 +84,8 @@ def accepted_types(self) -> typing.Tuple:
         """Accepted input types."""
         return (typing.Any, )
 
-    def output_type(self, parent_output_type: type) -> type:
-        return pd.DataFrame
+    def compute_schema(self, schema: StageSchema):
+        schema.output_schema.set_type(pd.DataFrame)
 
     def _build_single(self, builder: mrc.Builder, input_node: mrc.SegmentObject) -> mrc.SegmentObject:
         node = builder.make_node(self.unique_name,

examples/digital_fingerprinting/production/morpheus/dfp/stages/dfp_inference_stage.py

@@ -24,6 +24,7 @@
 from morpheus.config import Config
 from morpheus.messages.multi_ae_message import MultiAEMessage
 from morpheus.pipeline.single_port_stage import SinglePortStage
+from morpheus.pipeline.stage_schema import StageSchema
 
 from ..messages.multi_dfp_message import MultiDFPMessage
 from ..utils.model_cache import ModelCache
@@ -71,8 +72,8 @@ def accepted_types(self) -> typing.Tuple:
         """Accepted input types."""
         return (MultiDFPMessage, )
 
-    def output_type(self, parent_output_type: type) -> type:
-        return MultiAEMessage
+    def compute_schema(self, schema: StageSchema):
+        schema.output_schema.set_type(MultiAEMessage)
 
     def get_model(self, user: str) -> ModelCache:
         """

examples/digital_fingerprinting/production/morpheus/dfp/stages/dfp_rolling_window_stage.py

@@ -24,6 +24,7 @@
 
 from morpheus.config import Config
 from morpheus.pipeline.single_port_stage import SinglePortStage
+from morpheus.pipeline.stage_schema import StageSchema
 
 from ..messages.multi_dfp_message import DFPMessageMeta
 from ..messages.multi_dfp_message import MultiDFPMessage
@@ -90,8 +91,8 @@ def accepted_types(self) -> typing.Tuple:
         """Input types accepted by this stage."""
         return (DFPMessageMeta, )
 
-    def output_type(self, parent_output_type: type) -> type:
-        return MultiDFPMessage
+    def compute_schema(self, schema: StageSchema):
+        schema.output_schema.set_type(MultiDFPMessage)
 
     @contextmanager
     def _get_user_cache(self, user_id: str) -> typing.Generator[CachedUserWindow, None, None]:

examples/digital_fingerprinting/production/morpheus/dfp/stages/dfp_split_users_stage.py

@@ -25,6 +25,7 @@
 
 from morpheus.config import Config
 from morpheus.pipeline.single_port_stage import SinglePortStage
+from morpheus.pipeline.stage_schema import StageSchema
 from morpheus.utils.type_aliases import DataFrameType
 
 from ..messages.multi_dfp_message import DFPMessageMeta
@@ -84,8 +85,8 @@ def accepted_types(self) -> typing.Tuple:
         """Input types accepted by this stage."""
         return (cudf.DataFrame, pd.DataFrame)
 
-    def output_type(self, parent_output_type: type) -> type:
-        return DFPMessageMeta
+    def compute_schema(self, schema: StageSchema):
+        schema.output_schema.set_type(DFPMessageMeta)
 
     def extract_users(self, message: DataFrameType) -> typing.List[DFPMessageMeta]:
         """

examples/digital_fingerprinting/production/morpheus/dfp/stages/dfp_training.py

@@ -26,6 +26,7 @@
 from morpheus.messages.multi_ae_message import MultiAEMessage
 from morpheus.models.dfencoder import AutoEncoder
 from morpheus.pipeline.single_port_stage import SinglePortStage
+from morpheus.pipeline.stage_schema import StageSchema
 
 from ..messages.multi_dfp_message import DFPMessageMeta
 from ..messages.multi_dfp_message import MultiDFPMessage
@@ -94,11 +95,11 @@ def accepted_types(self) -> typing.Tuple:
             MultiDFPMessage,
         )
 
-    def output_type(self, parent_output_type: type) -> type:
-        return_type = parent_output_type
-        if (return_type == MultiDFPMessage):
-            return_type = MultiAEMessage
-        return return_type
+    def compute_schema(self, schema: StageSchema):
+        output_type = schema.input_type
+        if (output_type == MultiDFPMessage):
+            output_type = MultiAEMessage
+        schema.output_schema.set_type(output_type)
 
     def _dfp_multimessage_from_control_message(self,
                                                control_message: ControlMessage) -> typing.Union[MultiDFPMessage, None]:

examples/digital_fingerprinting/production/morpheus/dfp/stages/multi_file_source.py

@@ -23,6 +23,7 @@
 
 from morpheus.config import Config
 from morpheus.pipeline.single_output_source import SingleOutputSource
+from morpheus.pipeline.stage_schema import StageSchema
 
 logger = logging.getLogger(f"morpheus.{__name__}")
 
@@ -76,8 +77,8 @@ def input_count(self) -> int:
         """Return None for no max intput count"""
         return self._input_count
 
-    def output_type(self) -> type:
-        return fsspec.core.OpenFiles
+    def compute_schema(self, schema: StageSchema):
+        schema.output_schema.set_type(fsspec.core.OpenFiles)
 
     def supports_cpp_node(self):
         """Indicates whether this stage supports C++ nodes."""

examples/digital_fingerprinting/production/morpheus/dfp/utils/dfp_arg_parser.py

@@ -46,6 +46,8 @@ def __init__(self,
                  source: str,
                  tracking_uri: str,
                  silence_monitors: bool,
+                 mlflow_experiment_name_formatter: str,
+                 mlflow_model_name_formatter: str,
                  train_users: str = None):
 
         self._skip_users = list(skip_user)
@@ -65,8 +67,8 @@ def __init__(self,
         self._time_fields: TimeFields = None
         self._silence_monitors = silence_monitors
 
-        self._model_name_formatter = f"DFP-{source}-" + "{user_id}"
-        self._experiment_name_formatter = f"dfp/{source}/training/" + "{reg_model_name}"
+        self._model_name_formatter = mlflow_model_name_formatter
+        self._experiment_name_formatter = mlflow_experiment_name_formatter
 
     @staticmethod
     def verify_init(func):