https://flyte.org logo
Join Slack
Powered by
Hi team, we are working on a custom flyte agent (f...
# flyte-connectors
f
Hi team, we are working on a custom flyte agent (following Bigquery agent), we setup with: • Image in deployment/flyteagent use custom container • Update flytepropeller configmap, (task type of out custom agent called "dataproc") enabled_plugins.yaml: | plugins: agent-service: supportedTaskTypes: - dataproc tasks: task-plugins: default-for-task-types: dataproc: agent-service enabled-plugins: - agent-service The workflow is not able to executed in flyteagent, did I miss anythings in setup/registration about flyteagent (N.B. We are able to run with bigquery agent)
✅ 1
@high-microphone-7046
h
@average-finland-92144 For a new agent we developed for dataproc, flytepropeller is triggering it as a normal task on the workload namespace instead of triggering through the flyteagent ...is there any additional step we might be missing. The BQ agent is working fine.
f
cc @glamorous-carpet-83516 / @damp-lion-88352 is this right?
d
The config is correct
I think maybe there's something wrong in flytekit
task.py
would you like to share that file to us
also did it work well in local execution?
h
it worked in local execution. The agent files for dataproc is in this forked repo : https://github.com/nerochung/flytekit/tree/master/plugins/flytekit-dataproc
let me know if you need any other configuration.
On the cluster when we submit the "pyflyte run --remote test_agent.py wf" . It is triggered as a standard python task and we see the pod failing (due to missing flytekitplugin.dataproc module)
d
Nice, looking it now
f
was the agent deployed?
and available in the same service?
d
I think your code looks correct, it might be deploy problem
h
Flyte agent was deployed seperate using this helm chart: https://github.com/flyteorg/flyte/tree/master/charts/flyteagent
d
Can you look at the log in flyte cluster?
h
for the propeller or agent ?
d
try to find this line
in the propeller
and the agent too
h
logs on the flytegent pod :
d
propeller example: 
{"json":{"src":"plugin.go:383"},"level":"info","msg":"AgentDeployment service supports task types: [sensor chatgpt spark sensor]","ts":"2024-06-05T13:24:56+08:00"}
agent example:
Copy code
Starting up the server to expose the prometheus metrics...
Starting the agent service...
Starting Sensor that supports task categories ['sensor']
Starting ChatGPT Agent that supports task categories ['chatgpt']
Starting Databricks Agent that supports task categories ['spark']
search the log 
supports task types
in propeller
did you restart your flytepropeller
after you change the agent image?
h
yes, it was restarted.
checking the propeller logs
d
ok, take your time
h
On propeller , I don't see those lines you mentioned. Should I be increasing the log level somewhere?
d
log level should be info
which version of propeller are you using?
Copy code
logger:
  show-source: true
  level: 4
this is my log in propeller
let me clarfiy, in your current setup, bigquery agent will work, right?
does sensor work?
Copy code
from flytekit.sensor.file_sensor import FileSensor
from flytekit import ImageSpec, task, workflow

sensor = FileSensor(name="test_sensor")

@task()
def t1():
    return

@workflow()
def wf():
    # sensor(path="/") >> t1()
    sensor(path="<s3://my-s3-bucket>") >> t1()
    # sensor(path="/tmp") >> t1()
    # sensor(path="/123456") >> t1()
can you help me try this example?
h
yes big query worked... will try the sensor and let you know .
This is our configmap for the propeller for the plugins
d
Because bigquery can be triggered by agent service and flyetplugins
2 way
it looks correct
h
since we have replaced agent-service for bigquery we assumed its with the agent.
will try the sensor and get back
d
no prob
h
qq sensor should work with gs:// as well right?
d
I will dicuss with Kevin to add tutorial about how to deploy agent by helm chart, a step by step guide
qq sensor should work with gs:// as well right?
yes it should, but we should add secret for it if needed
👍 1
it doesn't have to success
if we can see it's running, then it's ok
it means that our setup are correct
h
got it
f
for the task I trigger, I expect it should be my task type, i.e. "dataproc" but it still show in Python Task
d
yes
it should be 
dataproc
it should be dataproc
h
the sensor one seems to be working fine :
d
Could you help me check your config map to know does it really set these correctly?
Kevin will come later to help you
🙏 1
h
This is the our configmap used for flytepropeller :
flytepropeller_cm.yaml
d
Thank you
looks correct
I think you might have issue about flytekit
take chatgpt agent as an example, whether the agent task succeed or not, it will be 
type: chatgpt
in the flyteconsole
my advice is that you should check your flytekit path
flytekitplugins-openai                   0.0.0+develop                       /Users/future-outlier/code/flytekit/plugins/flytekit-openai
for example, this is my path for my chatgpt plugin
h
This is a path on the agent ?
d
this is the path for my openai plugin, and chatgpt agent is written in it
I think maybe your configmap for propeller is correct
but flytekit doesn't serialize the correct task metadata for you
@high-microphone-7046 Would you help me try agent on single binary and see if it works?
this one
I still think the issue might be related to flytekit's python path, and this way can help you test the dataproc agent faster
h
sure, will give it a try.. will get back in some itme
time
@flaky-napkin-65203 ☝️
f
ok let me try on it
d
no problem, actually one of our contributor meet this error too
@high-microphone-7046 When you have time can you provide screenshot of 
pip list | grep flyte
?
thx
f
in my dev environment?
d
yes
f
d
you should have 
flyetkitplugins-dataproc
f
I set it using pythonpath
d
you should cd 
flytekit/plugins/flytekitplugins-dataproc
pip install -e .
f
let me try try
🙏 1
d
@high-microphone-7046 @flaky-napkin-65203 I think it's 100% python path error
print the task metadata and it shows that i have the correct task type 
type: "dataproc
h
you did a pip install of the dataproc agent?
d
i copy your code
f
I run With pip install -e . again but still same error
d
I succeed
Can we hop on a call?
h
sure
f
yes sure
d
ok 1 min, go to bathroom
Solved
gratitude thank you 1
Copy code
from flytekitplugins.dataproc import DataprocConfig, DataprocTask
from flytekit import kwtypes, workflow

project = "wpna-poc-00001"
location = "us-central1"
dataprocTask = DataprocTask(
    name=f"hello_dataproc",
    inputs=kwtypes(version=int),
    task_config=DataprocConfig(
        ProjectID=project,
        Location=location,
        MainPythonFileUri="<gs://toyota-woven/helloworld_pyspark.py>",
        SparkHistoryDataprocCluster=f'projects/{project}/regions/{location}/clusters/dataproc-spark-server',
    ),
)
# dataprocTask.exeucte()
# print(dataprocTask(version=1))
@workflow
def wf():
    dataprocTask(version=1)
🙇🏽 1
pyflyte run --remote test_task.py wf
h
🎉 Excellent.. thanks @damp-lion-88352
🙏 1
d
go version go1.21.6 darwin/arm64
f
@damp-lion-88352 I got another issue Once it deployed to the flyte cluster, it keeps invoke create(), suppose it should jump to get() in my agent.py, any tricks on it? I didnt' come across this in my local testing
d
Yes, there are some tips.
it keeps invoke create(),
It probably means that propeller keeps sending 
create
reqeust to the agent server. and the reason might be the agent server execute the 
create
function fails
so that propeller is just keep retrying
change 
pyflyte run --remote
to 
pyflyte run
and check if it works well
f
ok thx for the hints
it works fine with 
pyflyte run
, but my function in create() takes 1-2 minutes to return a response. I find that when workflow launched, it comes with this error 
E0606 02:14:42.666780       1 workers.go:103] error syncing 'flytesnacks-development/ff6e182c2a4ed431e991': failed at Node[n0]. RuntimeExecutionError: failed during plugin execution, caused by: failed to execute handle for plugin [agent-service]: rpc error: code = DeadlineExceeded desc = context deadline exceeded
{"json":{"exec_id":"ff6e182c2a4ed431e991","node":"n0","ns":"flytesnacks-development","res_ver":"94033457","routine":"worker-20","tasktype":"dataproc","wf":"flytesnacks:development:<http://test_task.wf|test_task.wf>"},"level":"error","msg":"Failed to create resource. Error: rpc error: code = DeadlineExceeded desc = context deadline exceeded","ts":"2024-06-06T02:16:42Z"}
{"json":{"exec_id":"ff6e182c2a4ed431e991","node":"n0","ns":"flytesnacks-development","res_ver":"94033457","routine":"worker-20","tasktype":"dataproc","wf":"flytesnacks:development:<http://test_task.wf|test_task.wf>"},"level":"warning","msg":"Runtime error from plugin [agent-service]. Error: rpc error: code = DeadlineExceeded desc = context deadline exceeded","ts":"2024-06-06T02:16:42Z"}
{"json":{"exec_id":"ff6e182c2a4ed431e991","node":"n0","ns":"flytesnacks-development","res_ver":"94033457","routine":"worker-20","wf":"flytesnacks:development:<http://test_task.wf|test_task.wf>"},"level":"error","msg":"handling parent node failed with error: failed at Node[n0]. RuntimeExecutionError: failed during plugin execution, caused by: failed to execute handle for plugin [agent-service]: rpc error: code = DeadlineExceeded desc = context deadline exceeded","ts":"2024-06-06T02:16:42Z"}
{"json":{"exec_id":"ff6e182c2a4ed431e991","node":"n0","ns":"flytesnacks-development","res_ver":"94033457","routine":"worker-20","wf":"flytesnacks:development:<http://test_task.wf|test_task.wf>"},"level":"error","msg":"failed Execute for node. Error: failed at Node[n0]. RuntimeExecutionError: failed during plugin execution, caused by: failed to execute handle for plugin [agent-service]: rpc error: code = DeadlineExceeded desc = context deadline exceeded","ts":"2024-06-06T02:16:42Z"}
{"json":{"exec_id":"ff6e182c2a4ed431e991","ns":"flytesnacks-development","res_ver":"94033457","routine":"worker-20","wf":"flytesnacks:development:<http://test_task.wf|test_task.wf>"},"level":"warning","msg":"Error in handling running workflow [failed at Node[n0]. RuntimeExecutionError: failed during plugin execution, caused by: failed to execute handle for plugin [agent-service]: rpc error: code = DeadlineExceeded desc = context deadline exceeded]","ts":"2024-06-06T02:16:42Z"}
{"json":{"exec_id":"ff6e182c2a4ed431e991","ns":"flytesnacks-development","res_ver":"94033457","routine":"worker-20","wf":"flytesnacks:development:<http://test_task.wf|test_task.wf>"},"level":"error","msg":"Error when trying to reconcile workflow. Error [failed at Node[n0]. RuntimeExecutionError: failed during plugin execution, caused by: failed to execute handle for plugin [agent-service]: rpc error: code = DeadlineExceeded desc = context deadline exceeded]. Error Type[*errors.NodeErrorWithCause]","ts":"2024-06-06T02:16:42Z"}
👍 1
d
Try to change the timeout of 
CreateTask
from default time out to 
180s
or longer time
after change the configmap, remeber to restart the propeller
🙏 1
f
applied the change with 300s, same error occurs, it keeps create
d
Let me guess, one of the reason might be 
GetTask
needs to be set to 
300s
too
And also, please check the secret is set correctly
g
applied the change with 300s, same error occurs, it keeps create
did you see any error in propeller log?
f
yes here's the propeller log 
E0606 07:27:05.674088       1 workers.go:103] error syncing 'flytesnacks-development/fc887dace13c5496ca7d': failed at Node[n0]. RuntimeExecutionError: failed during plugin execution, caused by: failed to execute handle for plugin [agent-service]: rpc error: code = DeadlineExceeded desc = context deadline exceeded
{"json":{"exec_id":"fc887dace13c5496ca7d","node":"n0","ns":"flytesnacks-development","res_ver":"94238823","routine":"worker-19","tasktype":"dataproc","wf":"flytesnacks:development:<http://test_task.wf|test_task.wf>"},"level":"error","msg":"Failed to create resource. Error: rpc error: code = DeadlineExceeded desc = context deadline exceeded","ts":"2024-06-06T07:29:05Z"}
{"json":{"exec_id":"fc887dace13c5496ca7d","node":"n0","ns":"flytesnacks-development","res_ver":"94238823","routine":"worker-19","tasktype":"dataproc","wf":"flytesnacks:development:<http://test_task.wf|test_task.wf>"},"level":"warning","msg":"Runtime error from plugin [agent-service]. Error: rpc error: code = DeadlineExceeded desc = context deadline exceeded","ts":"2024-06-06T07:29:05Z"}
{"json":{"exec_id":"fc887dace13c5496ca7d","node":"n0","ns":"flytesnacks-development","res_ver":"94238823","routine":"worker-19","wf":"flytesnacks:development:<http://test_task.wf|test_task.wf>"},"level":"error","msg":"handling parent node failed with error: failed at Node[n0]. RuntimeExecutionError: failed during plugin execution, caused by: failed to execute handle for plugin [agent-service]: rpc error: code = DeadlineExceeded desc = context deadline exceeded","ts":"2024-06-06T07:29:05Z"}
{"json":{"exec_id":"fc887dace13c5496ca7d","node":"n0","ns":"flytesnacks-development","res_ver":"94238823","routine":"worker-19","wf":"flytesnacks:development:<http://test_task.wf|test_task.wf>"},"level":"error","msg":"failed Execute for node. Error: failed at Node[n0]. RuntimeExecutionError: failed during plugin execution, caused by: failed to execute handle for plugin [agent-service]: rpc error: code = DeadlineExceeded desc = context deadline exceeded","ts":"2024-06-06T07:29:05Z"}
{"json":{"exec_id":"fc887dace13c5496ca7d","ns":"flytesnacks-development","res_ver":"94238823","routine":"worker-19","wf":"flytesnacks:development:<http://test_task.wf|test_task.wf>"},"level":"warning","msg":"Error in handling running workflow [failed at Node[n0]. RuntimeExecutionError: failed during plugin execution, caused by: failed to execute handle for plugin [agent-service]: rpc error: code = DeadlineExceeded desc = context deadline exceeded]","ts":"2024-06-06T07:29:05Z"}
{"json":{"exec_id":"fc887dace13c5496ca7d","ns":"flytesnacks-development","res_ver":"94238823","routine":"worker-19","wf":"flytesnacks:development:<http://test_task.wf|test_task.wf>"},"level":"error","msg":"Error when trying to reconcile workflow. Error [failed at Node[n0]. RuntimeExecutionError: failed during plugin execution, caused by: failed to execute handle for plugin [agent-service]: rpc error: code = DeadlineExceeded desc = context deadline exceeded]. Error Type[*errors.NodeErrorWithCause]","ts":"2024-06-06T07:29:05Z"}
d
It shows that you have to set the timeout to a larger time
In propeller, we will create a context to send request to the agent server
The config here can control the timeout of each type of requests
f
now i set this to 300s, and restart all deployment let me try try: 
agent-service:
defaultAgent:
endpoint: "dns:///flyteagent.flyte.svc.cluster.local:8000"
insecure: true
timeouts:
CreateTask: 300s
GetTask: 300s
DeleteTask: 300s
ExecuteTaskSync: 300s
defaultTimeout: 300s
supportedTaskTypes:
- dataproc
- bigquery_query_job_task
- sensor
🙏 1
d
ok, thank you
g
it’s supposed to return the response imminently, isn’t it?
f
if I run in a standalone notebook, it takes me 4 mins to get the response
yes I think so it should be a request submission, no idea why it takes so long
g
wow, do they have async API?
👍 1
d
@flaky-napkin-65203 maybe this one?
You can try to make current agent works, then try the asynchronous one
f
for the async, it takes me similar time, I need to await response to create DataprocMetadata
🙏 1
g
Copy code
# Make the request
    operation = client.create_batch(request=request)

    print("Waiting for operation to complete...")

    response = (await operation).result()

    # Handle the response
    print(response)
create_batch
only return the result when job is completed. https://cloud.google.com/python/docs/reference/dataproc/latest/google.cloud.dataproc_v1.services.batch_controller.BatchControllerAsyncClient#go[…]ch
f
I need the batch name to use it in get() function, to check status there
d
Can job_controller meets your need?
f
it pass with cluster name, but we wanna invoke dataproc serverless, which do not need cluster name
Copy code
job.placement.cluster_name
g
do you have a simple workflow that runs a dataproc task? Maybe I can try your plugin on my side
f
do u need access to our flyte console?
g
no, I just want to register a dataproc task on our cluster, and test it.
could you share the workflow code
f
yes sure sure
it's actually similar to Han-Ru shared with me ytd
from flytekitplugins.dataproc import DataprocConfig, DataprocTask from flytekit import kwtypes, workflow project = "wpna-poc-00001" location = "us-central1" dataprocTask = DataprocTask( name=f"hello_dataproc", inputs=kwtypes(version=int), task_config=DataprocConfig( ProjectID=project, Location=location, MainPythonFileUri="gs://toyota-woven/pyspark-hello-world.py", SparkHistoryDataprocCluster=f'projects/{project}/regions/{location}/clusters/dataproc-spark-server', ), ) # dataprocTask.exeucte() # print(dataprocTask(version=1)) @workflow def wf(): return dataprocTask(version=1)
this is pyspark-hello-world.py 
from pyspark.sql import SparkSession
def init_spark():
spark = SparkSession.builder.appName("HelloWorld").getOrCreate()
sc = spark.sparkContext
return spark,sc
def main():
spark,sc = init_spark()
nums = sc.parallelize([1,2,3,4])
print(nums.map(lambda x: x*x).collect())
if __name__ == '__main__':
main()
g
thanks. let me try it
🙏 1
f
btw where can I check the timeout settings is effective is it only in configmap of propeller
d
yes it's only in configmap of propeller
👌 1
g
yup. make sure to restart propeller after you changed the configmap
f
yes thx for the guidance
g
got this error
Copy code
InvalidArgument: 400 Subnetwork 'default' does not support Private Google Access which is required for Dataproc clusters when 'internal_ip_only' is set to 'true'. Enable Private Google Access on subnetwork 'default' or set 
'internal_ip_only' to 'false'.
f
You may need a gcp project for testing
g
I have. debugging. don’t know how to set it to false
f
let me chk chk
g
do you know how to create a new network configuration?
Trying to run a dummy spark task on gcp. seems like I don’t have a primary network
nvm, I can submit a job now
👍 2
The request is blocked by 
operation.result(timeout=1)
. could we add operation to the DataprocMetadata somehow? and run 
operation.result()
in the get request.
something like
Copy code
def get(self, resource_meta: DataprocMetadata, **kwargs) -> Resource
  try:
    response = operation.result(timeout=1)
  except Exception:
    return Resource(phase=running)
  batch_name = response.name
  request = dataproc_v1.GetBatchRequest(name=batch_name)
  ...
❤️ 1
d
Love this implementation
f
it's a gd idea, let me try try
g
you can use cloudpickle to serialize that first, but I believe there is a API in the dataproc package to serialize it.
because I find the deserialize method
Copy code
@classmethod
    def deserialize(self, payload):
        """Deserialize a ``google.longrunning.Operation`` protocol buffer.

        Args:
            payload (bytes): A serialized operation protocol buffer.

        Returns:
            ~.operations_pb2.Operation: An Operation protobuf object.
        """
        return operations_pb2.Operation.FromString(payload)
❤️ 1
f
Thanks Kevin & Han🙏
🙏 1
f
@flaky-napkin-65203 can you guys contribute this agent?
f
I'm cool with it, but I need to talk to my team first.
f
What company do you work for
f
Thanks for your help, it looks fine now I would like to add a link right here I tried with setting 
log_links
in get() 
return Resource(phase=cur_phase, message=msg, log_links=[log_link])
❤️ 1
d
You have to change the 
phase
to show the log link
1 recommend way to do that is return the log link in the 
create
function if you can get it
f
log link I set by this in get () 
log_link = TaskLog(
uri=response.runtime_info.endpoints["Spark History Server"],
name="Spark History Server",
)
when I run locally, it's able to show uri when running
g
are you able to see the URL on flyteconsole when phase becomes running?
f
no no now I map dataproc pending state to flyte with initiazlize and my spark job is running quite quick, I can see it in local env in a short period (maybe 2-3s) In flyteagent logs the polling seems less frequent there, every workflow trigger, I set only 2-3 get() logs there
d
Can you update the latest code in the git url?
Kevin and I can help you take a llok
look
f
sure sure let me chk chk
d
when you change the log links, you have to also change the pahse
the phase
that's the direction to go
f
oicic, maybe I chg it anyway to running once it's in get function
👍 1
just pushed my code, u guys may take a look
🙏 1
d
One recommended way is that 1. in the 
get
method, you can always return phase=
intializaing
until you get the log link, 2. once you get the link, change the phase to 
running
3. when reached terminal phase (SUCCEEDED or FAILED) return the output
databricks agent did that
g
wait, no
we did some changes couple weeks ago, you should be able to see the log link in every phase.
😮 1
you don’t need to change anything
will take a look your plugin later
f
ok, becoz line 73 does not need outputs and message in Resource https://github.com/nerochung/flytekit/blob/master/tests/flytekit/unit/extend/test_agent.py
d
the only necessary field is 
phase
f
will take a look your plugin later
any insights about log link in the agent
f
you can return the links
f
In get function of agent.py, I set it by the code below but no luck on it yet, 
log_link = TaskLog(
uri=response.runtime_info.endpoints["Spark History Server"],
name="Spark History Server",
)
return Resource(phase=cur_phase, message=msg, log_links=[log_link])
In local dev env, it's able to display uri in terminal
Or do we need to enable some items in configmap or deployment?
d
can help you check the code tomorrow
thank you
f
ok let's connect tmr 🙏
@damp-lion-88352 thank you so much
d
no problem
g
Hello guys, Im running into similar problem. It is strange because it was working yesterday but today i see this error when running the same workflow, any advice?
Copy code
{
  "json": {
    "exec_id": "a4x5svfw7xkn8zffls4j",
    "node": "n0",
    "ns": "kk-flyte-dev",
    "res_ver": "9781807055",
    "routine": "worker-1",
    "src": "admin_eventsink.go:50",
    "wf": "yaou-dev2-test:development:<http://flows.file_sensor_example.wf|flows.file_sensor_example.wf>"
  },
  "level": "debug",
  "msg": "AdminEventSink received a new event id:{node_id:\"n0\"  execution_id:{project:\"yaou-dev2-test\"  domain:\"development\"  name:\"a4x5svfw7xkn8zffls4j\"}}  producer_id:\"propeller\"  phase:FAILED  occurred_at:{seconds:1724369035  nanos:988523053}  input_uri:\"<ambry://flyte-ei-dev/metadata/propeller/yaou-dev2-test-development-a4x5svfw7xkn8zffls4j/n0/data/inputs.pb>\"  error:{code:\"RetriesExhausted|TaskFailedWithError\"  message:\"[1/1] currentAttempt done. Last Error: USER::rpc error: code = DeadlineExceeded desc = Deadline Exceeded\"  kind:USER}  task_node_metadata:{checkpoint_uri:\"<hdfs://ltx1-faro/__PLACEHOLDER__/flyte/rc/a4x5svfw7xkn8zffls4j-n0-0/_flytecheckpoints>\"}  spec_node_id:\"n0\"  node_name:\"partition_sensor1\"  event_version:1  reported_at:{seconds:1724369035  nanos:988597073}",
  "ts": "2024-08-22T23:23:55Z"
}
I’m trying to run a workflow that has the sensors , but keep seeing 
Last Error: USER::rpc error: code = DeadlineExceeded desc = Deadline Exceeded
d
maybe the agent server is crashed?
g
hmm @damp-lion-88352 I can see the agent is still up and running, do you know how i can check what the timeout limit is? or how i can change it?
d
yes i know it
here
Copy code
defaultAgent:
       endpoint: "dns:///flyteagent.flyte.svc.cluster.local:8000"
       insecure: true
       timeouts:
         # CreateTask, GetTask and DeleteTask are for async agents.
         # ExecuteTaskSync is for sync agents.
         CreateTask: 5s
         GetTask: 5s
         DeleteTask: 5s
         ExecuteTaskSync: 10s
       defaultTimeout: 10s
24 Views
Open in Slack
PreviousNext
Powered by