https://www.getdaft.io logo
Join Slack
Powered by
Is there any way to force daft to use the default ...
# general
t
Is there any way to force daft to use the default credentials chain for AWS? The default configuration doesn't seem to be picking up credentials for EKS pod identity associations. I tried providing an S3 config with a 
credentials_provider
function:
Copy code
def get_credentials() -> daft.io.S3Credentials:
    session = boto3.Session()
    creds = session.get_credentials()
    return daft.io.S3Credentials(
        key_id=creds.access_key,
        access_key=creds.secret_key,
        session_token=creds.token,
        expiry=datetime.datetime.now(datetime.timezone.utc) + datetime.timedelta(hours=1),
    )


s3_config = S3Config(
    credentials_provider=get_credentials,
)

io_config = IOConfig(s3=s3_config)

source = daft.read_parquet(
    "<s3://noetik-datalake-prod/branches/main/warehouse/curated__cosmx_cells/*>",
    io_config=io_config,
)

io_config = IOConfig(s3=s3_config)

source = daft.read_parquet(
    "<s3://noetik-datalake-prod/branches/main/warehouse/curated__cosmx_cells/*>",
    io_config=io_config,
)
... but that gives me an opaque serialization error from rust. The following works initially:
Copy code
session = boto3.Session()
creds = session.get_credentials()

s3_config = S3Config(
    key_id=creds.access_key,
    access_key=creds.secret_key,
    session_token=creds.token,
)

io_config = IOConfig(s3=s3_config)

source = daft.read_parquet(
    "<s3://noetik-datalake-prod/branches/main/warehouse/curated__cosmx_cells/*>",
    io_config=io_config,
)
but then the credentials eventually expire so it isn't appropriate for long running jobs that need write access over a long period of time.
j
I’d need @Sammy Sidhu to confirm, but I believe if you’re trying to do long-running jobs the get_credentials function is our way of doing so I’m surprised it’s not picking up the EKS pod IAMs though. That’s probably worth a look. Could you make an issue for us?
t
Yup can do. Any idea why my implementation of get_credentials above throws?
s
Can you also share the Ray serialization error? I believe @Kevin Wang implemented get credentials. Any ideas?
t
Copy code
(ScanWithTask-Project-FanoutHash [Stage:2] pid=3867337) thread '<unnamed>' panicked at src/daft-scan/src/python.rs:480:5: [repeated 30x across cluster]
(ScanWithTask-Project-FanoutHash [Stage:2] pid=3867337) pyo3_runtime.PanicException: called `Result::unwrap()` on an `Err` value: Custom("AttributeError: type object 'S3Credentials' has no attribute 'expiry'") [repeated 90x across cluster]
(ScanWithTask-Project-FanoutHash [Stage:2] pid=3867337) note: run with `RUST_BACKTRACE=1` environment variable to display a backtrace [repeated 30x across cluster]
(ScanWithTask-Project-FanoutHash [Stage:2] pid=3867337) Traceback (most recent call last): [repeated 60x across cluster]
(ScanWithTask-Project-FanoutHash [Stage:2] pid=3867337)   File "python/ray/_raylet.pyx", line 2254, in ray._raylet.task_execution_handler [repeated 60x across cluster]
(ScanWithTask-Project-FanoutHash [Stage:2] pid=3867337)   File "python/ray/_raylet.pyx", line 2150, in ray._raylet.execute_task_with_cancellation_handler [repeated 60x across cluster]
(ScanWithTask-Project-FanoutHash [Stage:2] pid=3867337)   File "python/ray/_raylet.pyx", line 1839, in ray._raylet.execute_task [repeated 300x across cluster]
(ScanWithTask-Project-FanoutHash [Stage:2] pid=3867337)   File "/home/noetik/twm/.venv/lib/python3.10/site-packages/ray/_private/serialization.py", line 460, in deserialize_objects [repeated 120x across cluster]
(ScanWithTask-Project-FanoutHash [Stage:2] pid=3867337)     return context.deserialize_objects(data_metadata_pairs, object_refs) [repeated 60x across cluster]
(ScanWithTask-Project-FanoutHash [Stage:2] pid=3867337)     obj = self._deserialize_object(data, metadata, object_ref) [repeated 60x across cluster]
(ScanWithTask-Project-FanoutHash [Stage:2] pid=3867337)   File "/home/noetik/twm/.venv/lib/python3.10/site-packages/ray/_private/serialization.py", line 317, in _deserialize_object [repeated 60x across cluster]
(ScanWithTask-Project-FanoutHash [Stage:2] pid=3867337)     return self._deserialize_msgpack_data(data, metadata_fields) [repeated 60x across cluster]
(ScanWithTask-Project-FanoutHash [Stage:2] pid=3867337)   File "/home/noetik/twm/.venv/lib/python3.10/site-packages/ray/_private/serialization.py", line 272, in _deserialize_msgpack_data [repeated 60x across cluster]
(ScanWithTask-Project-FanoutHash [Stage:2] pid=3867337)     python_objects = self._deserialize_pickle5_data(pickle5_data) [repeated 60x across cluster]
(ScanWithTask-Project-FanoutHash [Stage:2] pid=3867337)   File "/home/noetik/twm/.venv/lib/python3.10/site-packages/ray/_private/serialization.py", line 262, in _deserialize_pickle5_data [repeated 60x across cluster]
(ScanWithTask-Project-FanoutHash [Stage:2] pid=3867337)     obj = pickle.loads(in_band) [repeated 60x across cluster]
🙌 1
j
Did a little more digging about EKS/pod IAMs. Looking through AWS docs: https://docs.aws.amazon.com/sdk-for-rust/latest/dg/credproviders.html and also the default credentials chain (https://docs.rs/aws-config/0.55.3/aws_config/default_provider/credentials/struct.DefaultCredentialsChain.html), Daft should be correctly detecting the pod IAMs. Are you able to share more information about the errors you are seeing? I will try to see if I can reproduce this by spinning up EKS
Using my own EKS cluster + service account, I see the following behavior: With boto3, I can confirm that the method used is 
assume-role-with-web-identity
(which is the intended behavior for EKS pod IAMs I believe)
Copy code
import boto3

sess = boto3.session.Session()
sess.get_credentials()
creds = sess.get_credentials()
print(creds.method)
>>> "assume-role-with-web-identity"
Using Daft (local mode), I turned on debug logging and was able to confirm that Daft is correctly detecting the appropriate method to use as the WebIdentityToken. Redacted logs:
Copy code
import logging
import daft

logging.basicConfig(level=logging.DEBUG)

df = daft.read_csv("<s3://daft-public-data/melbourne-airbnb/melbourne_airbnb.csv>")


...
DEBUG:aws_config.default_provider.credentials:provide_credentials; provider=default_chain
DEBUG:aws_config.meta.credentials.chain:load_credentials; provider=Environment
DEBUG:aws_config.meta.credentials.chain:provider in chain did not provide credentials provider=Environment context=the credential provider was not enabled: environment variable not set (CredentialsNotLoaded(CredentialsNotLoaded { source: "environment variable not set" }))
DEBUG:aws_config.meta.credentials.chain:load_credentials; provider=Profile
DEBUG:aws_config.meta.credentials.chain:provider in chain did not provide credentials provider=Profile context=the credential provider was not enabled: No profiles were defined (CredentialsNotLoaded(CredentialsNotLoaded { source: NoProfilesDefined }))
DEBUG:aws_config.meta.credentials.chain:load_credentials; provider=WebIdentityToken
DEBUG:aws_smithy_client:send_operation;
DEBUG:aws_smithy_client:send_operation; operation="AssumeRoleWithWebIdentity"
DEBUG:aws_smithy_client:send_operation; service="sts"
DEBUG:aws_smithy_http_tower.map_request:map_request; name="resolve_endpoint"
DEBUG:aws_smithy_http_tower.map_request:map_request; name="resolve_endpoint"
DEBUG:aws_smithy_http_tower.map_request:map_request; name="generate_user_agent"
DEBUG:aws_smithy_http_tower.map_request:async_map_request; name="retrieve_credentials"
INFO:tracing.span:lazy_load_credentials;
DEBUG:aws_credential_types.cache.lazy_caching:loaded credentials
INFO:aws_http.auth:credentials cache returned CredentialsNotLoaded, ignoring
...
DEBUG:aws_sdk_sts.operation.assume_role_with_web_identity:request_id=Some("***")
DEBUG:aws_smithy_client:send_operation; status="ok"
DEBUG:aws_config.meta.credentials.chain:loaded credentials provider=WebIdentityToken
INFO:aws_credential_types.cache.lazy_caching:credentials cache miss occurred; added new AWS credentials (took 77.98997ms)
...
Our credentials chain seems to work well with EKS pod IAMs on my setup, but please let us know if you are seeing any unexpected behavior here so we can dig in further!
Quick note as well: When running Daft on your laptop but connected to a Ray cluster, there are potentially different sets of credentials in play ◦ Your laptop’s credential chain is used for some operations such as reading Parquet metadata to perform schema inference (this work is not farmed out to the cluster) ◦ The cluster’s credential chain is used for execution (e.g. reading the full set of data etc) Could it be the case that you are seeing errors coming from the first set of operations because we are using credentials from your laptop/driver process?
t
This is for IRSA. The latest recommendation is to use pod identity associations which is a newer addition: https://docs.aws.amazon.com/eks/latest/userguide/pod-identities.html
I’ve been able to get IRSA working in a test cluster as well, but our production cluster is using pod identity associations and not IRSA. Pod identity associations are a different credential provider which doesn’t use assume role with web identity.
j
Gotcha, I’ll try to get something set up to reproduce the issue
🙌 1
Just tried running the pod identity on EKS. Here is the error I ran into, does this look familiar? Interestingly, the code seems to have an invariant that the URI must refer to loopback, but 
$AWS_CONTAINER_CREDENTIALS_FULL_URI
that was provisioned to the pod actually refers to 
<http://169.254.170.23/v1/credentials>
which is a remote service
t
Yup that's exactly what I hit!
@jay do you know why my workaround doesn't seem to work? In the original error I posed it said expiry wasn't a valid key, even though it is in the 
daft.S3Credentials
constructor. If I remove that key, I get an error about access key not being valid either:
Copy code
(ScanWithTask-LocalLimit [Stage:2] pid=359147) thread '<unnamed>' panicked at src/daft-scan/src/python.rs:455:5:
(ScanWithTask-LocalLimit [Stage:2] pid=359147) called `Result::unwrap()` on an `Err` value: Custom("AttributeError: type object 'S3Credentials' has no attribute 'access_key'")
(ScanWithTask-LocalLimit [Stage:2] pid=359147) note: run with `RUST_BACKTRACE=1` environment variable to display a backtrace
(ScanWithTask-LocalLimit [Stage:2] pid=359147) Traceback (most recent call last):
(ScanWithTask-LocalLimit [Stage:2] pid=359147)   File "python/ray/_raylet.pyx", line 2254, in ray._raylet.task_execution_handler
(ScanWithTask-LocalLimit [Stage:2] pid=359147)   File "python/ray/_raylet.pyx", line 2150, in ray._raylet.execute_task_with_cancellation_handler
(ScanWithTask-LocalLimit [Stage:2] pid=359147)   File "python/ray/_raylet.pyx", line 1805, in ray._raylet.execute_task
(ScanWithTask-LocalLimit [Stage:2] pid=359147)   File "python/ray/_raylet.pyx", line 1806, in ray._raylet.execute_task
(ScanWithTask-LocalLimit [Stage:2] pid=359147)   File "python/ray/_raylet.pyx", line 1809, in ray._raylet.execute_task
(ScanWithTask-LocalLimit [Stage:2] pid=359147)   File "python/ray/_raylet.pyx", line 1837, in ray._raylet.execute_task
(ScanWithTask-LocalLimit [Stage:2] pid=359147)   File "python/ray/_raylet.pyx", line 1839, in ray._raylet.execute_task
(ScanWithTask-LocalLimit [Stage:2] pid=359147)   File "/home/noetik/twm/.venv/lib/python3.10/site-packages/ray/_private/worker.py", line 841, in deserialize_objects
(ScanWithTask-LocalLimit [Stage:2] pid=359147)     return context.deserialize_objects(data_metadata_pairs, object_refs)
(ScanWithTask-LocalLimit [Stage:2] pid=359147)   File "/home/noetik/twm/.venv/lib/python3.10/site-packages/ray/_private/serialization.py", line 460, in deserialize_objects
(ScanWithTask-LocalLimit [Stage:2] pid=359147)     obj = self._deserialize_object(data, metadata, object_ref)
(ScanWithTask-LocalLimit [Stage:2] pid=359147)   File "/home/noetik/twm/.venv/lib/python3.10/site-packages/ray/_private/serialization.py", line 317, in _deserialize_object
(ScanWithTask-LocalLimit [Stage:2] pid=359147)     return self._deserialize_msgpack_data(data, metadata_fields)
(ScanWithTask-LocalLimit [Stage:2] pid=359147)   File "/home/noetik/twm/.venv/lib/python3.10/site-packages/ray/_private/serialization.py", line 272, in _deserialize_msgpack_data
(ScanWithTask-LocalLimit [Stage:2] pid=359147)     python_objects = self._deserialize_pickle5_data(pickle5_data)
(ScanWithTask-LocalLimit [Stage:2] pid=359147)   File "/home/noetik/twm/.venv/lib/python3.10/site-packages/ray/_private/serialization.py", line 262, in _deserialize_pickle5_data
(ScanWithTask-LocalLimit [Stage:2] pid=359147)     obj = pickle.loads(in_band)
(ScanWithTask-LocalLimit [Stage:2] pid=359147) pyo3_runtime.PanicException: called `Result::unwrap()` on an `Err` value: Custom("AttributeError: type object 'S3Credentials' has no attribute 'access_key'")
(ScanWithTask-LocalLimit [Stage:2] pid=359147) Exception ignored in: 'ray._raylet.task_execution_handler'
(ScanWithTask-LocalLimit [Stage:2] pid=359147) Traceback (most recent call last):
(ScanWithTask-LocalLimit [Stage:2] pid=359147)   File "python/ray/_raylet.pyx", line 2254, in ray._raylet.task_execution_handler
(ScanWithTask-LocalLimit [Stage:2] pid=359147)   File "python/ray/_raylet.pyx", line 2150, in ray._raylet.execute_task_with_cancellation_handler
(ScanWithTask-LocalLimit [Stage:2] pid=359147)   File "python/ray/_raylet.pyx", line 1805, in ray._raylet.execute_task
(ScanWithTask-LocalLimit [Stage:2] pid=359147)   File "python/ray/_raylet.pyx", line 1806, in ray._raylet.execute_task
(ScanWithTask-LocalLimit [Stage:2] pid=359147)   File "python/ray/_raylet.pyx", line 1809, in ray._raylet.execute_task
(ScanWithTask-LocalLimit [Stage:2] pid=359147)   File "python/ray/_raylet.pyx", line 1837, in ray._raylet.execute_task
(ScanWithTask-LocalLimit [Stage:2] pid=359147)   File "python/ray/_raylet.pyx", line 1839, in ray._raylet.execute_task
(ScanWithTask-LocalLimit [Stage:2] pid=359147)   File "/home/noetik/twm/.venv/lib/python3.10/site-packages/ray/_private/worker.py", line 841, in deserialize_objects
(ScanWithTask-LocalLimit [Stage:2] pid=359147)     return context.deserialize_objects(data_metadata_pairs, object_refs)
(ScanWithTask-LocalLimit [Stage:2] pid=359147)   File "/home/noetik/twm/.venv/lib/python3.10/site-packages/ray/_private/serialization.py", line 460, in deserialize_objects
(ScanWithTask-LocalLimit [Stage:2] pid=359147)     obj = self._deserialize_object(data, metadata, object_ref)
(ScanWithTask-LocalLimit [Stage:2] pid=359147)   File "/home/noetik/twm/.venv/lib/python3.10/site-packages/ray/_private/serialization.py", line 317, in _deserialize_object
(ScanWithTask-LocalLimit [Stage:2] pid=359147)     return self._deserialize_msgpack_data(data, metadata_fields)
(ScanWithTask-LocalLimit [Stage:2] pid=359147)   File "/home/noetik/twm/.venv/lib/python3.10/site-packages/ray/_private/serialization.py", line 272, in _deserialize_msgpack_data
(ScanWithTask-LocalLimit [Stage:2] pid=359147)     python_objects = self._deserialize_pickle5_data(pickle5_data)
(ScanWithTask-LocalLimit [Stage:2] pid=359147)   File "/home/noetik/twm/.venv/lib/python3.10/site-packages/ray/_private/serialization.py", line 262, in _deserialize_pickle5_data
(ScanWithTask-LocalLimit [Stage:2] pid=359147)     obj = pickle.loads(in_band)
(ScanWithTask-LocalLimit [Stage:2] pid=359147) pyo3_runtime.PanicException: called `Result::unwrap()` on an `Err` value: Custom("AttributeError: type object 'S3Credentials' has no attribute 'access_key'")
(ScanWithTask-LocalLimit [Stage:2] pid=359147) [2024-11-18 23:00:04,425 C 359147 359147] <http://task_receiver.cc:213|task_receiver.cc:213>:  Check failed: objects_valid
j
@Kevin Wang is on it!
🙏 1
t
Appreciate it. Will post this new trace there.
🙌 1
Is there anything I can do to work around these issues? Our session tokens only last an hour but the inference job I'm trying to run will need to write to S3 for longer than that. The daft version of this inference pipeline is so much faster than the other versions we've written, would be a bummer to have to revert.
j
The get_credentials path is the best workaround… we’re going to get to the bottom of it and let you know our findings there Could you help us by letting us know: what version of Ray and Daft are you using? We haven’t been able to reproduce the issue yet unfortunately 😕 cc @Kevin Wang
t
0.3.13
for daft and 
2.38.0
for ray
Tried a fresh install in a new venv since I had upgraded the version recently but still getting the same error.
k
Hi @Tyler Van Hensbergen, I'm taking a look at the error from your initial credential provider code. Could you run it with the environment variable 
RUST_BACKTRACE=1
and share the output error?
t
trace.txt
trace.txt
k
Thanks! I'll take a look
t
Appreciate it
I'm wondering if maybe I have the wrong import for S3Credentials since I noticed the module is actually in rust. That's thrown me for a loop in the past with the PyO3 projects I've built in the past. Could you share the code that you used to try to reproduce? Want to compare to what I've tried to see if maybe I'm doing something wrong.
k
I was able to reproduce your error. It's something to do with Ray serialization, but I'm not sure what yet. Will continue to investigate
🙌 1
@Tyler Van Hensbergen this is kind of a crazy idea, but try removing the function return type from 
get_credentials
(the part 
-> daft.io.S3Credentials
). That fixes it for me. I think for some reason the 
S3Credentials
type is not yet fully created by the time the function is deserialized, and so it fails.
Still trying to figure out the root cause of this but I'd like to see if this workaround fixes your issue first
t
That did it, thanks!
👍 1
j
🫠
t
For what its worth, the rest of the experience makes up for the warts. Appreciate all the work you all have been putting into it.
❤️ 1
Spoke too soon. This works for reads, but writes give a new error:
Copy code
RayTaskError(RaySystemError): ray::ReduceMerge-WriteFile [Stage:107]() (pid=1818960, ip=10.0.175.38)
  At least one of the input arguments for this task could not be computed:
ray.exceptions.RaySystemError: System error: DaftError::SerdeJsonError invalid type: sequence, expected a byte array containing the pickled partition bytes at line 1 column 175
traceback: Traceback (most recent call last):
  File "/home/noetik/twm/.venv/lib/python3.10/site-packages/daft/io/config.py", line 8, in _io_config_from_json
    return IOConfig.from_json(io_config_json)
daft.exceptions.DaftCoreException: DaftError::SerdeJsonError invalid type: sequence, expected a byte array containing the pickled partition bytes at line 1 column 175
k
Looking into it. Unrelated to the issue you are seeing but I think the S3 python credentials provider functionality does not actually support writes at the moment. Will get a PR out for that soon
t
ah gotcha
7 Views
Open in Slack
PreviousNext
Powered by