Nc/1oct/py prepare multipart #1054
Merged
Conversation
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Contributor
nfcampos
commented
Oct 1, 2024
nfcampos
commented
- Add requests-toolbelt dep
- Use custom http adapter
- subclass so that we can override blocksize (default one too small to maximize performance on modern network stacks)
- override default pool size to match the max nr of batch tarcing threads (otherwise some threads would be left waiting for connections to be available)
- actually mount the adapter (previous behavior was never mounting the adapter because sessions always have http and https adapters)
- note before retry policy was being ignored, and no transport level retries were being done
- Add attachments field to run dict/model
- update _run_transform to collect/drop attachments
- warn if attachments are passed to an endpoint that can't accept them, in this case inject runs without attachments
- subclass so that we can override blocksize (default one too small to maximize performance on modern network stacks) - override default pool size to match the max nr of batch tarcing threads (otherwise some threads would be left waiting for connections to be available) - actually mount the adapter (previous behavior was never mounting the adapter because sessions always have http and https adapters)
- update _run_transform to collect/drop attachments - warn if attachments are passed to an endpoint that can't accept them, in this case inject runs without attachments
- Use streaming multipart encoder form requests_toolbelt - Currently dump each part to json before sending the request as that's the only way to enforce the payload size limit - When we lift payload size limit we should implement true streaming encoding, where each part is only encoded immediately before being sent over the connection, and use transfer-encoding: chunked
- Use streaming multipart encoder form requests_toolbelt - Currently dump each part to json before sending the request as that's the only way to enforce the payload size limit - When we lift payload size limit we should implement true streaming encoding, where each part is only encoded immediately before being sent over the connection, and use transfer-encoding: chunked
hinthornw
reviewed
Oct 2, 2024
| @@ -435,6 +443,34 @@ class TracingQueueItem: | |||
| item: Any = field(compare=False) | |||
|
|
|||
|
|
|||
| class _LangSmithHttpAdapter(requests_adapters.HTTPAdapter): | |||
There was a problem hiding this comment.
Not for the PR but I wonder if we should be running simple tests on our min/max bounds for requests lib here since we are taking on some maintanence burdne by subclassing here; just like 1 unit test to check on bounds
| @@ -781,7 +821,7 @@ def request_with_retries( | |||
| ls_utils.FilterPoolFullWarning(host=str(self._host)), | |||
| ] | |||
| retry_on_: Tuple[Type[BaseException], ...] = ( | |||
| *(retry_on or []), | |||
| *(retry_on or ()), | |||
There was a problem hiding this comment.
Suggested change
| *(retry_on or ()), | |
| *(retry_on or EMPTY_SEQ), |
There was a problem hiding this comment.
I don't think this one works, because its the wrong type
| # Check that no duplicate run_ids are present in the request bodies | ||
| assert len(request_bodies) == len(set([body["id"] for body in request_bodies])) | ||
| # Check that no duplicate run_ids are present in the request bodies | ||
| assert len(request_bodies) == len(set([body["id"] for body in request_bodies])) | ||
|
|
There was a problem hiding this comment.
Likely stupid, but maybe a quick test on multipart_ingest_runs on two empty lists (or just one empty list) if that's not done elsewhere
| self._insert_runtime_env(create_dicts) | ||
| self._insert_runtime_env(update_dicts) | ||
| # check size limit | ||
| size_limit_bytes = (self.info.batch_ingest_config or {}).get( |
There was a problem hiding this comment.
Should we also have an EMPTY_DICT? (or just a frozen dict-like)
from types import MappingProxyType
EMPTY_DICT = MappingProxyType({})
Co-authored-by: William FH <13333726+hinthornw@users.noreply.github.com>
hinthornw
approved these changes
Oct 2, 2024
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.