AppShieldSource stage

.gitattributes

@@ -1,4 +1,4 @@
 morpheus/_version.py export-subst
 morpheus/data/* filter=lfs diff=lfs merge=lfs -text
-tests/expected_data/* filter=lfs diff=lfs merge=lfs -text
 tests/mock_triton_server/payloads/** filter=lfs diff=lfs merge=lfs -text
+tests/tests_data/** filter=lfs diff=lfs merge=lfs -text

docker/conda/environments/requirements.txt

@@ -16,3 +16,4 @@ torch==1.10.2+cu113
 tqdm
 tritonclient[all]
 typing-utils
+watchdog==2.1.8

morpheus/cli.py

@@ -662,6 +662,19 @@ def from_kafka(ctx: click.Context, **kwargs):
               default=1,
               type=click.IntRange(min=1),
               help=("Repeats the input dataset multiple times. Useful to extend small datasets for debugging."))
+@click.option('--sort_glob',
+              type=bool,
+              default=False,
+              help=("If true the list of files matching `input_glob` will be processed in sorted order."))
+@click.option('--recursive',
+              type=bool,
+              default=True,
+              help=("If true, events will be emitted for the files in subdirectories matching `input_glob`."))
+@click.option('--queue_max_size',
+              type=int,
+              default=128,
+              help=("Maximum queue size to hold the file paths to be processed that match `input_glob`."))
+@click.option('--batch_timeout', type=float, default=5.0, help=("Timeout to retrieve batch messages from the queue."))
 @prepare_command()
 def from_cloudtrail(ctx: click.Context, **kwargs):
 

morpheus/messages/message_meta.py

@@ -62,3 +62,18 @@ class UserMessageMeta(MessageMeta, cpp_class=None):
 
     """
     user_id: str
+
+
+@dataclasses.dataclass
+class AppShieldMessageMeta(MessageMeta, cpp_class=None):
+    """
+    This class extends MessageMeta to also hold source corresponding to batched metadata.
+
+    Parameters
+    ----------
+    df : pd.DataFrame
+        Input rows in dataframe.
+    source : str
+        Determines which source generated the snapshot messages.
+    """
+    source: str