Training/Inference Module DFP Production

examples/digital_fingerprinting/production/morpheus/dfp/modules/dfp_inference.py

@@ -0,0 +1,113 @@
+# Copyright (c) 2023, NVIDIA CORPORATION.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import logging
+import time
+
+import mrc
+from dfp.utils.model_cache import ModelCache
+from dfp.utils.model_cache import ModelManager
+from mlflow.tracking.client import MlflowClient
+from mrc.core import operators as ops
+
+from morpheus.messages.multi_ae_message import MultiAEMessage
+from morpheus.utils.module_ids import MODULE_NAMESPACE
+from morpheus.utils.module_utils import get_module_config
+from morpheus.utils.module_utils import register_module
+
+from ..messages.multi_dfp_message import MultiDFPMessage
+from ..utils.module_ids import DFP_INFERENCE
+
+logger = logging.getLogger(__name__)
+
+
+@register_module(DFP_INFERENCE, MODULE_NAMESPACE)
+def dfp_inference(builder: mrc.Builder):
+    """
+    Inference module function.
+
+    Parameters
+    ----------
+    builder : mrc.Builder
+        Pipeline budler instance.
+    """
+
+    config = get_module_config(DFP_INFERENCE, builder)
+
+    fallback_user = config.get("fallback_username", None)
+    model_name_formatter = config.get("model_name_formatter", None)
+    timestamp_column_name = config.get("timestamp_column_name", None)
+
+    client = MlflowClient()
+    model_manager = ModelManager(model_name_formatter=model_name_formatter)
+
+    def get_model(user: str) -> ModelCache:
+
+        return model_manager.load_user_model(client, user_id=user, fallback_user_ids=[fallback_user])
+
+    def on_data(message: MultiDFPMessage):
+        if (not message or message.mess_count == 0):
+            return None
+
+        start_time = time.time()
+
+        df_user = message.get_meta()
+        user_id = message.user_id
+
+        try:
+            model_cache: ModelCache = get_model(user_id)
+
+            if (model_cache is None):
+                raise RuntimeError("Could not find model for user {}".format(user_id))
+
+            loaded_model = model_cache.load_model(client)
+
+        except Exception:  # TODO
+            logger.exception("Error trying to get model")
+            return None
+
+        post_model_time = time.time()
+
+        results_df = loaded_model.get_results(df_user, return_abs=True)
+
+        # Create an output message to allow setting meta
+        output_message = MultiAEMessage(message.meta,
+                                        mess_offset=message.mess_offset,
+                                        mess_count=message.mess_count,
+                                        model=loaded_model)
+
+        output_message.set_meta(list(results_df.columns), results_df)
+
+        output_message.set_meta('model_version', f"{model_cache.reg_model_name}:{model_cache.reg_model_version}")
+
+        if logger.isEnabledFor(logging.DEBUG):
+            load_model_duration = (post_model_time - start_time) * 1000.0
+            get_anomaly_duration = (time.time() - post_model_time) * 1000.0
+
+            logger.debug("Completed inference for user %s. Model load: %s ms, Model infer: %s ms. Start: %s, End: %s",
+                         user_id,
+                         load_model_duration,
+                         get_anomaly_duration,
+                         df_user[timestamp_column_name].min(),
+                         df_user[timestamp_column_name].max())
+
+        return output_message
+
+    def node_fn(obs: mrc.Observable, sub: mrc.Subscriber):
+        obs.pipe(ops.map(on_data)).subscribe(sub)
+
+    node = builder.make_node_full(DFP_INFERENCE, node_fn)
+
+    builder.register_module_input("input", node)
+    builder.register_module_output("output", node)

examples/digital_fingerprinting/production/morpheus/dfp/modules/dfp_inference_pipeline.py

@@ -0,0 +1,99 @@
+# Copyright (c) 2022-2023, NVIDIA CORPORATION.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import logging
+
+import dfp.modules.dfp_data_prep  # noqa: F401
+import dfp.modules.dfp_inference  # noqa: F401
+import dfp.modules.dfp_postprocessing  # noqa: F401
+import dfp.modules.dfp_rolling_window  # noqa: F401
+import dfp.modules.dfp_split_users  # noqa: F401
+import mrc
+
+import morpheus.modules.file_batcher  # noqa: F401
+import morpheus.modules.file_to_df  # noqa: F401
+import morpheus.modules.filter_detections  # noqa: F401
+import morpheus.modules.serialize  # noqa: F401
+import morpheus.modules.write_to_file  # noqa: F401
+from morpheus.utils.module_ids import FILE_BATCHER
+from morpheus.utils.module_ids import FILE_TO_DF
+from morpheus.utils.module_ids import FILTER_DETECTIONS
+from morpheus.utils.module_ids import MODULE_NAMESPACE
+from morpheus.utils.module_ids import SERIALIZE
+from morpheus.utils.module_ids import WRITE_TO_FILE
+from morpheus.utils.module_utils import get_module_config
+from morpheus.utils.module_utils import load_module
+from morpheus.utils.module_utils import register_module
+
+from ..utils.module_ids import DFP_DATA_PREP
+from ..utils.module_ids import DFP_INFERENCE
+from ..utils.module_ids import DFP_INFERENCE_PIPELINE
+from ..utils.module_ids import DFP_POST_PROCESSING
+from ..utils.module_ids import DFP_ROLLING_WINDOW
+from ..utils.module_ids import DFP_SPLIT_USERS
+
+logger = logging.getLogger(__name__)
+
+
+@register_module(DFP_INFERENCE_PIPELINE, MODULE_NAMESPACE)
+def dfp_inference_pipeline(builder: mrc.Builder):
+    """
+    This module function allows for the consolidation of multiple dfp pipeline modules relevent to inference
+    process into a single module.
+
+    Parameters
+    ----------
+    builder : mrc.Builder
+        Pipeline budler instance.
+    """
+
+    config = get_module_config(DFP_INFERENCE_PIPELINE, builder)
+
+    file_batcher_conf = config.get(FILE_BATCHER, None)
+    file_to_df_conf = config.get(FILE_TO_DF, None)
+    dfp_split_users_conf = config.get(DFP_SPLIT_USERS, None)
+    dfp_rolling_window_conf = config.get(DFP_ROLLING_WINDOW, None)
+    dfp_data_prep_conf = config.get(DFP_DATA_PREP, None)
+    dfp_inference_conf = config.get(DFP_INFERENCE, None)
+    filter_detections_conf = config.get(FILTER_DETECTIONS, None)
+    dfp_post_proc_conf = config.get(DFP_POST_PROCESSING, None)
+    serialize_conf = config.get(SERIALIZE, None)
+    write_to_file_conf = config.get(WRITE_TO_FILE, None)
+
+    # Load modules
+    file_batcher_module = load_module(file_batcher_conf, builder=builder)
+    file_to_dataframe_module = load_module(file_to_df_conf, builder=builder)
+    dfp_split_users_modules = load_module(dfp_split_users_conf, builder=builder)
+    dfp_rolling_window_module = load_module(dfp_rolling_window_conf, builder=builder)
+    dfp_data_prep_module = load_module(dfp_data_prep_conf, builder=builder)
+    dfp_inference_module = load_module(dfp_inference_conf, builder=builder)
+    filter_detections_module = load_module(filter_detections_conf, builder=builder)
+    dfp_post_proc_module = load_module(dfp_post_proc_conf, builder=builder)
+    serialize_module = load_module(serialize_conf, builder=builder)
+    write_to_file_module = load_module(write_to_file_conf, builder=builder)
+
+    # Make an edge between the modules.
+    builder.make_edge(file_batcher_module.output_port("output"), file_to_dataframe_module.input_port("input"))
+    builder.make_edge(file_to_dataframe_module.output_port("output"), dfp_split_users_modules.input_port("input"))
+    builder.make_edge(dfp_split_users_modules.output_port("output"), dfp_rolling_window_module.input_port("input"))
+    builder.make_edge(dfp_rolling_window_module.output_port("output"), dfp_data_prep_module.input_port("input"))
+    builder.make_edge(dfp_data_prep_module.output_port("output"), dfp_inference_module.input_port("input"))
+    builder.make_edge(dfp_inference_module.output_port("output"), filter_detections_module.input_port("input"))
+    builder.make_edge(filter_detections_module.output_port("output"), dfp_post_proc_module.input_port("input"))
+    builder.make_edge(dfp_post_proc_module.output_port("output"), serialize_module.input_port("input"))
+    builder.make_edge(serialize_module.output_port("output"), write_to_file_module.input_port("input"))
+
+    # Register input and output port for a module.
+    builder.register_module_input("input", file_batcher_module.input_port("input"))
+    builder.register_module_output("output", write_to_file_module.output_port("output"))

examples/digital_fingerprinting/production/morpheus/dfp/modules/dfp_postprocessing.py

@@ -0,0 +1,81 @@
+# Copyright (c) 2023, NVIDIA CORPORATION.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import logging
+import time
+from datetime import datetime
+
+import mrc
+import numpy as np
+from mrc.core import operators as ops
+
+from morpheus.messages.multi_ae_message import MultiAEMessage
+from morpheus.utils.module_ids import MODULE_NAMESPACE
+from morpheus.utils.module_utils import get_module_config
+from morpheus.utils.module_utils import register_module
+
+from ..utils.module_ids import DFP_POST_PROCESSING
+
+logger = logging.getLogger(__name__)
+
+
+@register_module(DFP_POST_PROCESSING, MODULE_NAMESPACE)
+def dfp_postprocessing(builder: mrc.Builder):
+    """
+    Postprocessing module function.
+
+    Parameters
+    ----------
+    builder : mrc.Builder
+        Pipeline budler instance.
+    """
+
+    config = get_module_config(DFP_POST_PROCESSING, builder)
+
+    timestamp_column_name = config.get("timestamp_column_name", None)
+
+    def process_events(message: MultiAEMessage):
+        # Assume that a filter stage preceedes this stage
+        df = message.get_meta()
+        df['event_time'] = datetime.now().strftime('%Y-%m-%dT%H:%M:%SZ')
+        df.replace(np.nan, 'NaN', regex=True, inplace=True)
+        message.set_meta(None, df)
+
+    def on_data(message: MultiAEMessage):
+        if (not message or message.mess_count == 0):
+            return None
+
+        start_time = time.time()
+
+        process_events(message)
+
+        duration = (time.time() - start_time) * 1000.0
+
+        if logger.isEnabledFor(logging.DEBUG):
+            logger.debug("Completed postprocessing for user %s in %s ms. Event count: %s. Start: %s, End: %s",
+                         message.meta.user_id,
+                         duration,
+                         message.mess_count,
+                         message.get_meta(timestamp_column_name).min(),
+                         message.get_meta(timestamp_column_name).max())
+
+        return message
+
+    def node_fn(obs: mrc.Observable, sub: mrc.Subscriber):
+        obs.pipe(ops.map(on_data), ops.filter(lambda x: x is not None)).subscribe(sub)
+
+    node = builder.make_node_full(DFP_POST_PROCESSING, node_fn)
+
+    builder.register_module_input("input", node)
+    builder.register_module_output("output", node)

examples/digital_fingerprinting/production/morpheus/dfp/modules/dfp_training_pipeline.py

@@ -0,0 +1,84 @@
+# Copyright (c) 2022-2023, NVIDIA CORPORATION.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import logging
+
+import dfp.modules.dfp_data_prep  # noqa: F401
+import dfp.modules.dfp_rolling_window  # noqa: F401
+import dfp.modules.dfp_split_users  # noqa: F401
+import dfp.modules.dfp_training  # noqa: F401
+import mrc
+
+import morpheus.modules.file_batcher  # noqa: F401
+import morpheus.modules.file_to_df  # noqa: F401
+import morpheus.modules.mlflow_model_writer  # noqa: F401
+from morpheus.utils.module_ids import FILE_BATCHER
+from morpheus.utils.module_ids import FILE_TO_DF
+from morpheus.utils.module_ids import MLFLOW_MODEL_WRITER
+from morpheus.utils.module_ids import MODULE_NAMESPACE
+from morpheus.utils.module_utils import get_module_config
+from morpheus.utils.module_utils import load_module
+from morpheus.utils.module_utils import register_module
+
+from ..utils.module_ids import DFP_DATA_PREP
+from ..utils.module_ids import DFP_ROLLING_WINDOW
+from ..utils.module_ids import DFP_SPLIT_USERS
+from ..utils.module_ids import DFP_TRAINING
+from ..utils.module_ids import DFP_TRAINING_PIPELINE
+
+logger = logging.getLogger(__name__)
+
+
+@register_module(DFP_TRAINING_PIPELINE, MODULE_NAMESPACE)
+def dfp_training_pipeline(builder: mrc.Builder):
+    """
+    This module function allows for the consolidation of multiple dfp pipeline modules relevent to training
+    process into a single module.
+
+    Parameters
+    ----------
+    builder : mrc.Builder
+        Pipeline budler instance.
+    """
+
+    config = get_module_config(DFP_TRAINING_PIPELINE, builder)
+
+    file_batcher_conf = config.get(FILE_BATCHER, None)
+    file_to_df_conf = config.get(FILE_TO_DF, None)
+    dfp_split_users_conf = config.get(DFP_SPLIT_USERS, None)
+    dfp_rolling_window_conf = config.get(DFP_ROLLING_WINDOW, None)
+    dfp_data_prep_conf = config.get(DFP_DATA_PREP, None)
+    dfp_training_conf = config.get(DFP_TRAINING, None)
+    mlflow_model_writer_conf = config.get(MLFLOW_MODEL_WRITER, None)
+
+    # Load modules
+    file_batcher_module = load_module(file_batcher_conf, builder=builder)
+    file_to_dataframe_module = load_module(file_to_df_conf, builder=builder)
+    dfp_split_users_modules = load_module(dfp_split_users_conf, builder=builder)
+    dfp_rolling_window_module = load_module(dfp_rolling_window_conf, builder=builder)
+    dfp_data_prep_module = load_module(dfp_data_prep_conf, builder=builder)
+    dfp_training_module = load_module(dfp_training_conf, builder=builder)
+    mlflow_model_writer_module = load_module(mlflow_model_writer_conf, builder=builder)
+
+    # Make an edge between the modules.
+    builder.make_edge(file_batcher_module.output_port("output"), file_to_dataframe_module.input_port("input"))
+    builder.make_edge(file_to_dataframe_module.output_port("output"), dfp_split_users_modules.input_port("input"))
+    builder.make_edge(dfp_split_users_modules.output_port("output"), dfp_rolling_window_module.input_port("input"))
+    builder.make_edge(dfp_rolling_window_module.output_port("output"), dfp_data_prep_module.input_port("input"))
+    builder.make_edge(dfp_data_prep_module.output_port("output"), dfp_training_module.input_port("input"))
+    builder.make_edge(dfp_training_module.output_port("output"), mlflow_model_writer_module.input_port("input"))
+
+    # Register input and output port for a module.
+    builder.register_module_input("input", file_batcher_module.input_port("input"))
+    builder.register_module_output("output", mlflow_model_writer_module.output_port("output"))