Hey guys this is Sally I m trying to test use agent service Flyte #flyte-support

Hey guys, this is Sally. I’m trying to test/use ag...

gray-ram-51379

05/07/2024, 4:16 AM

Hey guys, this is Sally. I’m trying to test/use agent service, I ran “pip install flytekit” on a brand new virtual environment and installed flytectl as well. However, my workflow that involves testing a “time delta” sensor (that I wrote myself) is failing with

rpc error: code = Internal desc = failed to get sensor task with error: No module named 'sensor'.

this is the stack trace:

Copy code

{"json":{"routine":"agent-service-worker-0","src":"cache.go:84"},"level":"debug","msg":"Sync loop - processing resource with cache key [f677097c1e4564916ad6-n0-0]","ts":"2024-05-07T04:12:02Z"}
{"json":{"routine":"agent-service-worker-0","src":"cache.go:113"},"level":"debug","msg":"Querying AsyncPlugin for f677097c1e4564916ad6-n0-0","ts":"2024-05-07T04:12:02Z"}
{"json":{"routine":"agent-service-worker-0","src":"cache.go:116"},"level":"info","msg":"Error retrieving resource [f677097c1e4564916ad6-n0-0]. Error: rpc error: code = Internal desc = failed to get sensor task with error: No module named 'sensor'.","ts":"2024-05-07T04:12:02Z"}

this is the workflow with sensor I’m testing with, it’s very simple:

Copy code

from flytekit import task, workflow
from flytekit.sensor.base_sensor import BaseSensor
import time


class TimeDeltaSensor(BaseSensor):
    def __init__(self, name: str, **kwargs):
        super().__init__(name=name, **kwargs)

    async def poke(self, timeDelta: int) -> bool:
        print("Before the sleep sensor")
        time.sleep(timeDelta)
        print("After the time sleep sensor")
        return True

time_delta_sensor = TimeDeltaSensor(name="Time_Delta_mySensor")


@task()
def t1():
    print("SUCCEEDED")


@workflow()
def wf():
    time_delta_sensor(timeDelta=11) >> t1()


if __name__ == "__main__":
    wf()

any idea why this is failing? any help is appreciated 🙏 cc: @glamorous-carpet-83516

👋 1

gray-ram-51379

05/07/2024, 4:17 AM

this is my dependencies:

Copy code

(venv) yaou@yaou-mn1 sensor % pip list | grep flyte
flyteidl                 1.12.0
flytekit                 1.12.0

glamorous-carpet-83516

05/07/2024, 4:23 AM

are you running this workflow in flytectl demo cluster?

gray-ram-51379

05/07/2024, 4:24 AM

correct

glamorous-carpet-83516

05/07/2024, 4:29 AM

you have to create a new python package that contains your custom sensor, and install this python package in the dockerfile.

gray-ram-51379

05/07/2024, 4:35 AM

hmm, I do not quite understand ^^ I want to develop a sensor that’s similar to file sensor, it seems that you can just import the class

from flytekit.sensor.file_sensor import FileSensor

and use it. What’s the reason why I cannot do that in my workflow?

glamorous-carpet-83516

05/07/2024, 4:37 AM

the class

TimeDeltaSensor

doesn’t exist in the agent server, so agent can’t not import this class

glamorous-carpet-83516

05/07/2024, 4:38 AM

agent server has installed flytekit, so it can import FileSensor

gray-ram-51379

05/07/2024, 4:40 AM

Ok so it sounds like I can’t just use flytectl demo cluster to try this? What is the best way to quickly run a test using my TimeDeltaSensor?

glamorous-carpet-83516

05/07/2024, 4:41 AM

you can run it locally to test your custom sensor. pyflyte run without

--remote

glamorous-carpet-83516

05/07/2024, 4:43 AM

To test it in the demo cluster, you have to build a new container image for the agent server. This image should install new plugin that contains your

TimeDeltaSensor

👀 1

🙏 1

gray-ram-51379

05/07/2024, 4:47 AM

Does this dockerfile build the “new” image I want? I’ll need to add a

pip install

line in here? Do I do

flytectl demo start

(which also starts the agent service) first then run the dockerfile that would “restart” the agent ?

freezing-airport-6809

05/07/2024, 5:01 AM

cc @high-park-82026 deployment problem

glamorous-carpet-83516

05/07/2024, 5:01 AM

run

flytectl demo start

first, and then replace the agent’s image with the image you built.

🙏 1

damp-lion-88352

05/07/2024, 5:29 AM

https://docs.flyte.org/en/latest/flyte_agents/deploying_agents_to_the_flyte_sandbox.html

🙏 1

damp-lion-88352

05/07/2024, 5:29 AM

Try this

damp-lion-88352

05/07/2024, 5:36 AM

👀 1

damp-lion-88352

05/07/2024, 5:36 AM

This can help you understand it.

🙏 1

gray-ram-51379

05/07/2024, 5:38 AM

Thanks everybody! I’ll try it out tmrw, it’s getting late where I am 😴

gray-ram-51379

05/23/2024, 4:56 AM

Hey everyone, I have an urgent follow up on this. So I implemented a

flytekit-timesensor

plugin, I `pip install -e .`the whole local flytekit as well as pip installed this new plugin I developed. The logic of the timeDeltaSensor is the same as the one pasted above ^ I edited the Dockerfile.agent to look like this, pushed it and re-deployed the flyteagent:

Copy code

FROM python:3.10-slim-bookworm as agent-slim

MAINTAINER Flyte Team <users@flyte.org>
LABEL org.opencontainers.image.source=<https://github.com/flyteorg/flytekit>

RUN apt-get update && apt-get install build-essential -y

COPY ./plugins/flytekit-timesensor /root/timesensor
RUN pip install /root/timesensor

RUN pip install prometheus-client grpcio-health-checking
RUN pip install --no-cache-dir -U flytekit \
  flytekitplugins-airflow \
  flytekitplugins-bigquery \
  flytekitplugins-openai \
  flytekitplugins-snowflake \
  flytekitplugins-awssagemaker \
  && apt-get clean autoclean \
  && apt-get autoremove --yes \
  && rm -rf /var/lib/{apt,dpkg,cache,log}/ \
  && :

CMD pyflyte serve agent --port 8000

but when I tried to run a workflow that uses this timeDeltaSensor, it is giving error

rpc error: code = DeadlineExceeded desc = context deadline exceeded

. Any advice? cc: @glamorous-carpet-83516 thank you 🙏woohoo

gray-ram-51379

05/23/2024, 4:57 AM

I was able to run it locally of course, that error showed up on remote ^ I used

flytectl demo start

for the whole setup

gray-ram-51379

05/23/2024, 5:28 AM

I created a PR here (not to be checked in) to show the code and process I used : https://github.com/flyteorg/flytekit/pull/2439 Let me know if you are able to reproduce or if you know what is wrong thanks cc: @glamorous-carpet-83516

damp-lion-88352

05/23/2024, 5:30 AM

@gray-ram-51379

damp-lion-88352

05/23/2024, 5:30 AM

damp-lion-88352

05/23/2024, 5:30 AM

do you use your gitsha?

damp-lion-88352

05/23/2024, 5:30 AM

I mean you have to do something like

gray-ram-51379

05/23/2024, 5:33 AM

yes?

damp-lion-88352

05/23/2024, 5:33 AM

Copy code

FROM python:3.9-slim-bookworm

RUN apt-get update && apt-get install build-essential git -y
RUN pip install prometheus-client grpcio-health-checking
RUN pip install --no-cache-dir -U \
	git+<https://github.com/flyteorg/flytekit.git@><gitsha>
    && apt-get clean autoclean \
    && apt-get autoremove --yes \
    && rm -rf /var/lib/{apt,dpkg,cache,log}/ \
    && :

CMD pyflyte serve agent --port 8000

damp-lion-88352

05/23/2024, 5:33 AM

git+<https://github.com/flyteorg/flytekit.git@><gitsha>

damp-lion-88352

05/23/2024, 5:33 AM

focus on this one

damp-lion-88352

05/23/2024, 5:33 AM

you have to change it to your forked repo

damp-lion-88352

05/23/2024, 5:34 AM

so maybe like

<http://github.com/sally/flytekit.git@gitsha|github.com/sally/flytekit.git@gitsha>

damp-lion-88352

05/23/2024, 5:34 AM

alias gitsha="git rev-parse HEAD";

damp-lion-88352

05/23/2024, 5:34 AM

gitsha can get like this

damp-lion-88352

05/23/2024, 5:35 AM

Copy code

FROM python:3.9-slim-bookworm

RUN apt-get update && apt-get install build-essential git -y
RUN pip install prometheus-client grpcio-health-checking
RUN pip install --no-cache-dir -U \
	git+<https://github.com/Sally-Yang-Jing-Ou/flytekit.git@a829b1893a86ecbaa64ede871de1a723ae329d3c>
    && apt-get clean autoclean \
    && apt-get autoremove --yes \
    && rm -rf /var/lib/{apt,dpkg,cache,log}/ \
    && :

CMD pyflyte serve agent --port 8000

damp-lion-88352

05/23/2024, 5:35 AM

use this one

damp-lion-88352

05/23/2024, 5:36 AM

Copy code

FROM python:3.9-slim-bookworm

RUN apt-get update && apt-get install build-essential git -y
RUN pip install prometheus-client grpcio-health-checking
RUN pip install --no-cache-dir -U \
	git+<https://github.com/Sally-Yang-Jing-Ou/flytekit.git@a829b1893a86ecbaa64ede871de1a723ae329d3c> 
	git+<https://github.com/Sally-Yang-Jing-Ou/flytekit.git@a829b1893a86ecbaa64ede871de1a723ae329d3c#subdirectory=plugins/flytekit-timesensor> \
    && apt-get clean autoclean \
    && apt-get autoremove --yes \
    && rm -rf /var/lib/{apt,dpkg,cache,log}/ \
    && :

CMD pyflyte serve agent --port 8000

damp-lion-88352

05/23/2024, 5:36 AM

Updated: this is the one you can use

clean-glass-36808

05/23/2024, 5:37 AM

I think it’s fairly simple. Propeller has a default timeout/deadline of 10s on requests to the flyte agent. In your example you are sleeping 11 seconds, which is why the request from propeller times out.

🙏 1

gray-ram-51379

05/23/2024, 5:38 AM

@clean-glass-36808 Wow i changed it to 5 seconds and it actually worked, thank you 🙏 !

damp-lion-88352

05/23/2024, 5:38 AM

oh ok

damp-lion-88352

05/23/2024, 5:39 AM

you can configure timeout at here

damp-lion-88352

05/23/2024, 5:39 AM

https://docs.flyte.org/en/latest/flyte_agents/testing_agents_in_a_local_development_cluster.html

damp-lion-88352

05/23/2024, 5:40 AM

change the timeout of

damp-lion-88352

05/23/2024, 5:40 AM

Copy code

GetTask

gray-ram-51379

05/23/2024, 5:40 AM

If I create a sensor that requires pinging to see if data exists, that would take more than 10 seconds, I do not understand how that would work then? For example, how does the file sensor work in this case, since it might take more than 10 seconds for data/file to be available?@clean-glass-36808

damp-lion-88352

05/23/2024, 5:40 AM

and

Copy code

CreateTask

clean-glass-36808

05/23/2024, 5:42 AM

I am a Flyte noob so I don't even really know how sensors are used. I just went through the effort of setting up a flyte task/agent last week and fixing up the docs. Sorry I can't help more

🙏 1

gray-ram-51379

05/23/2024, 5:43 AM

I think it might be that I used time.sleep which is blocking instead of await asyncio.sleep(), I am going to test it out

damp-lion-88352

05/23/2024, 5:44 AM

I think you have to update the timeout setting in the propeller

damp-lion-88352

05/23/2024, 5:44 AM

if you need longer time to wait

👍 1

damp-lion-88352

05/23/2024, 5:44 AM

Did I misunderstand your meaning?

damp-lion-88352

05/23/2024, 5:45 AM

@gray-ram-51379

gray-ram-51379

05/23/2024, 5:46 AM

No, it’s all good, that makes sense, thank you, Han Ru 🙏 !

damp-lion-88352

05/23/2024, 5:46 AM

ok, feel free to ping me, thank you

damp-lion-88352

05/23/2024, 5:54 AM

For more details, in propeller we will create a context with timout, and use it to send request to the agent server. So if it takes 15 seconds to finish the task in the agent server, and the timeout in propeller is 10 seconds, propeller will always fail to get the response

👍 1

🙏 1

24 Views

Open in Slack

Previous Next