Repository: caikit/caikit (python)

Issue:
Reject non array .json files for data streams
## Describe the bug

If a user uploads a `.json` file for a data stream parameter in a runtime api, it is expected to be an array of the data stream elements. If a user instead supplies a file with contents like:
```
{
  "input": "foo",
  "output": "bar"
}
```
caikit will happily yield an empty stream. This gives no indication to the user about what they did wrong.

## Additional context

Make sure `.jsonl` is still supported correctly

Pull request:
:goal_net: Return error for non-array JSON where array is expected
<!--  Thanks for sending a pull request!  Here are some tips for you:
1. Make sure to read the Contributing Guide before submitting your PR: https://github.com/caikit/caikit/blob/main/CONTRIBUTING.md
2. If this PR closes another issue, add 'closes #<issue number>' somewhere in the PR summary. GitHub will automatically close that issue when this PR gets merged. Alternatively, adding 'refs #<issue number>' will not close the issue, but help provide the reviewer more context.-->

**What this PR does / why we need it**:

Closes #500 - previously users would just receive an empty stream for non-array JSON provided in JSON files. This PR provides users an error message when an array is expected

Includes small duplicate log code update

**Special notes for your reviewer**:

**If applicable**:
- [ ] this PR contains documentation
- [x] this PR contains unit tests
- [ ] this PR has been tested for backwards compatibility


Patch:
diff --git a/caikit/core/data_model/streams/data_stream.py b/caikit/core/data_model/streams/data_stream.py
index d9af9c77..deee46cb 100644
--- a/caikit/core/data_model/streams/data_stream.py
+++ b/caikit/core/data_model/streams/data_stream.py
@@ -254,9 +254,16 @@ class DataStream(Generic[T]):
             )
         # For each {} object of the array
         try:
+            item_idx = None
             for item_idx, obj in enumerate(ijson.items(json_fh, "item")):
                 log.debug2("Loading object index %d", item_idx)
                 yield obj
+            if item_idx is None:
+                # Not an array
+                error(
+                    "<COR79428339E>",
+                    ValueError("Non-array JSON object in `{}`".format(filename)),
+                )
         except ijson.JSONError:
             error(
                 "<COR85596551E>",
diff --git a/caikit/runtime/model_management/model_manager.py b/caikit/runtime/model_management/model_manager.py
index c3d0c46b..1b01b3ba 100644
--- a/caikit/runtime/model_management/model_manager.py
+++ b/caikit/runtime/model_management/model_manager.py
@@ -273,7 +273,6 @@ class ModelManager:  # pylint: disable=too-many-instance-attributes
             self._lazy_sync_timer = threading.Timer(
                 self._lazy_load_poll_period_seconds, self.sync_local_models
             )
-            self._lazy_sync_timer.daemon = True
             self._lazy_sync_timer.start()
 
     def unload_model(self, model_id) -> int:
diff --git a/caikit/runtime/service_generation/data_stream_source.py b/caikit/runtime/service_generation/data_stream_source.py
index f2b4a140..1849ac1e 100644
--- a/caikit/runtime/service_generation/data_stream_source.py
+++ b/caikit/runtime/service_generation/data_stream_source.py
@@ -180,7 +180,10 @@ class DataStreamSourceBase(DataStream):
                 f"Invalid {extension} data source file: {fname}",
             )
         if extension == ".json":
-            return DataStream.from_json_array(full_fname).map(cls._to_element_type)
+            stream = DataStream.from_json_array(full_fname).map(cls._to_element_type)
+            # Iterate once to make sure this is a json array
+            stream.peek()
+            return stream
         if extension == ".csv":
             return DataStream.from_header_csv(full_fname).map(cls._to_element_type)
         if extension == ".jsonl":
diff --git a/caikit/runtime/servicers/global_train_servicer.py b/caikit/runtime/servicers/global_train_servicer.py
index 367e41ab..44ab5500 100644
--- a/caikit/runtime/servicers/global_train_servicer.py
+++ b/caikit/runtime/servicers/global_train_servicer.py
@@ -161,7 +161,7 @@ class GlobalTrainServicer:
 
         except Exception as e:
             log_dict = {
-                "log_code": "<RUN49049070W>",
+                "log_code": "<RUN24215150W>",
                 "message": repr(e),
                 "stack_trace": traceback.format_exc(),
             }


Test patch:
diff --git a/tests/core/data_model/streams/test_data_stream.py b/tests/core/data_model/streams/test_data_stream.py
index f4723d6e..d3756e7d 100644
--- a/tests/core/data_model/streams/test_data_stream.py
+++ b/tests/core/data_model/streams/test_data_stream.py
@@ -184,6 +184,21 @@ def test_bad_json_stream(tmp_path):
         DataStream.from_json_array(file_path).peek()
 
 
+def test_non_json_array_stream(tmp_path):
+    file_path = os.path.join(str(tmp_path), "non_array.json")
+    with open(file_path, "w") as fp:
+        fp.write(
+            """
+        {
+            "number": 1,
+            "label": "foo"
+        }
+        """
+        )
+    with pytest.raises(ValueError, match="Non-array JSON object"):
+        DataStream.from_json_array(file_path).peek()
+
+
 def test_bad_multipart_file(tmp_path):
     multipart_file = os.path.join(str(tmp_path), "bad_multipart")
     with open(multipart_file, "w") as fp:
diff --git a/tests/runtime/service_generation/test_data_stream_source.py b/tests/runtime/service_generation/test_data_stream_source.py
index f73ee6fc..615e90e8 100644
--- a/tests/runtime/service_generation/test_data_stream_source.py
+++ b/tests/runtime/service_generation/test_data_stream_source.py
@@ -488,6 +488,25 @@ def test_make_data_stream_source_no_files_w_ext_dir(
     assert "contains no source files with extension" in e.value.message
 
 
+def test_make_data_stream_source_non_json_array_errors(tmp_path):
+    stream_type = caikit.interfaces.common.data_model.DataStreamSourceSampleTrainingType
+    file_path = os.path.join(str(tmp_path), "non_array.json")
+    with open(file_path, "w") as fp:
+        fp.write(
+            """
+        {
+            "number": 1,
+            "label": "foo"
+        }
+        """
+        )
+    ds = stream_type(file=File(filename=file_path))
+    assert isinstance(ds, DataStreamSourceBase)
+
+    with pytest.raises(ValueError, match="Non-array JSON object"):
+        ds.to_data_stream()
+
+
 def test_s3_not_implemented(sample_train_service):
     stream_type = caikit.interfaces.common.data_model.DataStreamSourceSampleTrainingType
     ds = stream_type(s3files=S3Files())