Bernard Greyling
05/03/2022, 5:45 AMdavzucky
05/03/2022, 6:25 AMSuresh R
05/03/2022, 6:37 AMprefect.exceptions.ClientError: [{'path': ['secret_value'], 'message': 'An unknown error occurred.', 'extensions': {'code': 'INTERNAL_SERVER_ERROR'}}]
Domenico Di Gangi
05/03/2022, 7:09 AMThomas Mignon
05/03/2022, 7:29 AMSang Young Noh
05/03/2022, 8:23 AMAndreas Nord
05/03/2022, 8:54 AMUserWarning: Task <Task: task_name> has retry settings but some upstream dependencies do not have result types. Cloud features such as input caching and resuming task runs from failure may not work properly.
The (automatic) retrying seems to be working fine, not sure what this warning refers to. I am aware that I cannot resume task run from failure, but not sure why that is related to retry settingsRon Levi
05/03/2022, 10:18 AMAmruth VVKP
05/03/2022, 10:45 AM@flow(name='primary_flow', version='1', task_runner=SequentialTaskRunner())
def primary_flow(**kwargs):
... # perform data processing tasks, all tasks are non-async
for cli_command in parsed.result():
shell_run_command(cli_command) # I've also tried using asyncio.run for running the shell task
Scenario 2 -
Primary Flow: Uses sequential task runner, a bunch of processing tasks. The final task to build and call a CLI is done through another async subflow
@flow(name='primary_flow', version='1', task_runner=SequentialTaskRunner())
def primary_flow(**kwargs):
... # perform data processing tasks, all tasks are non-async
for _ in parsed.result():
cli_flow(_) # I've also tried using asyncio.run for running the shell task
@flow(name='cli_flow', version='1', task_runner=SequentialTaskRunner())
async def cli_flow(parsed):
for _ in parsed:
... # Build CLI
await shell_run_command(cli_command)
I am getting this exception thrown (attached screenshot) and the CLI task is not getting started/executed. Can someone help me out on this one please (I am terrible with asnyc processes)?Mars
05/03/2022, 2:29 PMprefect k8s agent install
. I’ve uploaded my flow to a private GitHub repo and registered it with Prefect. And I’ve added a Cloud Secret called GITHUB_ACCESS_TOKEN
that holds a valid GitHub personal access token. When I run my flow the agent’s GitHub storage gives me an UnknownObjectException(404, 'Not Found')
error. If I change the flow to use a different Cloud Secret key for the PAT, such as access_token_secret='MYKEY'
, then the agent tells me ValueError('Local Secret "MYKEY" was not found.')
.
How can I introspect the kubernetes agent to verify that the GitHub PAT secret is being loaded from Prefect Cloud correctly?Jai P
05/03/2022, 3:09 PMJakub Cieslik
05/03/2022, 4:54 PMGowtham Manne
05/03/2022, 5:22 PMDavid Beck
05/03/2022, 5:29 PMprefect.
, e.g. prefect.flow_name
. Given the documentations cautioning of renaming context attributes, I'm trying to brainstorm the simplest and most effective way of doing this across all our flows. Do you have suggestions?Andrew Lawlor
05/03/2022, 6:25 PMPod prefect-job-5e3af599-tl2xs failed. No container statuses found for pod
where can i look for a more detailed message? any idea what actually caused it to fail?
also, of those jobs, all but one passed. the one that failed had the same message (no container statuses found).Artem Vysotsky
05/03/2022, 6:55 PMchicago-joe
05/03/2022, 7:34 PMBradley Hurley
05/03/2022, 8:34 PMExecuteNotebook
Task.Andrew Lawlor
05/03/2022, 8:41 PMMatthew Roeschke
05/03/2022, 9:07 PMprefect register -p project
to get the names of the current flows that would have been registered
2. Get the IDs of the flows through the GraphQL API
3. Delete flow IDs that that do not have names from step 1
Is there a more sensible way to get the “current flows” in a project beside the prefect register
trick?Matt Alhonte
05/03/2022, 9:35 PMKilled
) and the mapping stays in a mapped
stateCole Murray
05/04/2022, 5:45 AMMartin T
05/04/2022, 6:08 AM$ prefect run --path myflow.py --log-level INFO --param myparam=val
Here the development flow becomes rather blackboxed/trial-and-error.
Is it possible to somehow set breakpoints in VSCode, start a prefect flow from CLI, and attach the debugger to the python process?
(Running with the LocalExecutor)Jelle Vegter
05/04/2022, 11:52 AMAbuzar Shaikh
05/04/2022, 12:06 PMIevgenii Martynenko
05/04/2022, 12:45 PMMichail Melonas
05/04/2022, 12:45 PMError downloading Flow from S3: An error occurred (ExpiredToken) when calling the GetObject operation: The provided token has expired.
I’m using AWS SSO credentials to upload the flow to S3:
flow.storage = S3(
bucket=Config.PREFECT_S3_BUCKET,
client_options={
"aws_access_key_id": Config.AWS_ACCESS_KEY_ID,
"aws_secret_access_key": Config.AWS_SECRET_ACCESS_KEY,
"aws_session_token": Config.AWS_SESSION_TOKEN,
},
)
Initially, the flow runs. However, flows scheduled after the token expiry fail with the above error. How do I both use SSO to upload flows to S3 and have them scheduled on Prefect Cloud?Patrick Tan
05/04/2022, 1:47 PMfrom prefect import Flow
from prefect.storage import S3
flow = Flow("s3-flow", storage=S3(bucket="<my-bucket>"))
Birkir Björnsson
05/04/2022, 2:33 PMJason
05/04/2022, 2:50 PM@task
def task_a():
common_class = Common(Secret())
output = common_class.task_a()
return output
@task
def task_b():
common_class = Common(Secret())
output = common_class.task_b()
return output