implements anyscale operators and sensors #41
Conversation
glyfnet
added
enhancement
Co-authored-by: Rajath <[email protected]>
Co-authored-by: Rajath <[email protected]>
Co-authored-by: Rajath <[email protected]>
Co-authored-by: Rajath <[email protected]>
| from anyscale.shared_anyscale_utils.utils.byod import BYODInfo | ||
| from anyscale.sdk.anyscale_client.models.create_production_job import CreateProductionJob | ||
|
|
||
| _POKE_INTERVAL = 60 |
There was a problem hiding this comment.
What if the user wanted to poll at a different interval maybe a lesser time or more?
| if self.wait_for_completion: | ||
| while not AnyscaleSessionCommandSensor( | ||
| task_id="wait_session_command", | ||
| session_command_id=session_command_response.id, |
There was a problem hiding this comment.
Are we not sending the goal_state, here it is complete state?
There was a problem hiding this comment.
Here, we are waiting for the session command to return a status_code. If it is not finished, this value is None. Then we compute the execution time and raise an exception if the session_command returned a different status_code than 0.
| auth_token=self.auth_token, | ||
| ).poke(context): | ||
|
|
||
| time.sleep(_POKE_INTERVAL) |
There was a problem hiding this comment.
Can we add poke_interval as the class param, so that the user can modify it.
There was a problem hiding this comment.
Yes, I updated the base class.
| self.ray_version = ray_version or "1.13.0" | ||
| self.python_version = python_version or "py38" |
There was a problem hiding this comment.
We could use these as default values also when getting these params.
WDYT?
There was a problem hiding this comment.
That's correct. I will add these as the default value for python_version and ray_version.
There was a problem hiding this comment.
Can you also update in other files as well?
There was a problem hiding this comment.
Yes, I updated all the operators.
|
|
||
| if goal_state is not None: | ||
| return False | ||
|
|
There was a problem hiding this comment.
Aren't we supposed to check for the current state and goal_state also?
There was a problem hiding this comment.
don't we need to check current state and goal state?
There was a problem hiding this comment.
I added the condition:
if goal_state is not None and goal_state != state:
return FalseThis way, the sensor checks if the cluster has reached the goal_state.
There was a problem hiding this comment.
Consider that the goal_state returned by the Anyscale API is None when the cluster has reached the goal state.
| self, | ||
| *, | ||
| auth_token: str, | ||
| poke_interval: Optional[int] = 60, |
There was a problem hiding this comment.
param for poke_interval is missing
|
Overall LGTM. |
|
Hey guys, just getting to seeing this. This looks really awesome. Would it be wise to change the input of an Auth Token to using a connection ID and a hook to be used in the BaseOperator? Also, can you provide some documentation on usage and how it deviates from running ray code directly in airflow through the provider? Is it appropriate for this to be a separate provider package? |
|
Hey @mrrobby. Indeed we can create a Hook that uses a connection to Anyscale. My only observation is that it will only have the auth token because it is what the Anyscale Python SDK needs to operate with the Anyscale account. |
This pull request implements Anyscale operators and sensors.