Bruno Murino

    Bruno Murino

    1 year ago
    this is how I’m defining my clock, then I attach it to the flow by doing
    flow.schedule = get_schedule()
    import datetime
    from datetime import datetime, timedelta
    from prefect.schedules import clocks, Schedule
    
    
    def get_schedule():
    
        now = datetime.utcnow()
    
        clock1 = clocks.IntervalClock(
            interval=timedelta(minutes=10),
            parameter_defaults={
                "scope": "full",
            },
        )
    
        my_schedule = Schedule(
            clocks=[
                clock1,
            ]
        )
    
        return my_schedule
    Stéphan Taljaard

    Stéphan Taljaard

    1 year ago
    Is the schedule enabled in the UI?
    Bruno Murino

    Bruno Murino

    1 year ago
    yes it is
    s

    Simon Gasse

    1 year ago
    I have the exact same issue 💁‍♂️ The docs unfortunately did not help me resolve it. Really curious what it could be about 🤔
    Kevin Kho

    Kevin Kho

    1 year ago
    How did you attach the schedule to the Flow?
    s

    Simon Gasse

    1 year ago
    For me, I created a file
    scheduled_hello.py
    and ran
    python scheduled_hello.py
    on the machine where the server is running.
    from datetime import timedelta
    
    from prefect import task, Flow
    from prefect.schedules import IntervalSchedule
    
    
    @task
    def say_hello():
        print("Hello, world!")
    
    
    schedule = IntervalSchedule(interval=timedelta(minutes=2))
    
    with Flow("Scheduled Hello", schedule) as flow:
        say_hello()
    
    flow.register(project_name='example',
                  labels=['local'],
                  idempotency_key=flow.serialized_hash())
    Is there a difference when using
    prefect register
    instead? Edit: I tried with
    prefect register
    but it does not make a difference 🤔
    Kevin Kho

    Kevin Kho

    1 year ago
    Can you try adding
    flow.schedule = schedule
    after the flow definition and before the
    flow.register()
    ?
    Bruno Murino

    Bruno Murino

    1 year ago
    @Kevin Kho On my code I do:
    if os.environ["EXECUTOR"] == "aws" and os.environ["ENV"] == "prod":
            flow.schedule = get_schedule()
    Kevin Kho

    Kevin Kho

    1 year ago
    I think you schedule is not getting attached. Can you try attaching directly without this
    if
    block first? What do you see when you to go to the Flow Settings and click Schedule?
    Bruno Murino

    Bruno Murino

    1 year ago
    I’ll try that now — but on the UI I do see the schedule there
    s

    Simon Gasse

    1 year ago
    @Kevin Kho thank you for the input! Changing to:
    schedule = IntervalSchedule(interval=timedelta(minutes=2))
    
    with Flow("Scheduled Hello") as flow:
        say_hello()
    
    flow.schedule = schedule
    
    flow.register(project_name='example',
                  labels=['local'],
                  idempotency_key=flow.serialized_hash())
    unfortunately does not make a difference on my setup
    I also see the schedule in the UI like you @Bruno Murino 👍 So I wonder if it is more an issue in the backend? Does some internal part of
    prefect server
    not go through the attached schedules to figure out what to run next?
    Kevin Kho

    Kevin Kho

    1 year ago
    It should. What version are you on? Will try to replicate
    Bruno Murino

    Bruno Murino

    1 year ago
    @Kevin Kho without the IF statement it still doesn’t work, the UI looks identical with the schedule listed but no upcoming runs ever.. I’m on prefect 15 throughout
    Kevin Kho

    Kevin Kho

    1 year ago
    Ok trying to replicate
    You guys are right. I see the same behavior on Prefect Server, and the same code works on Cloud. I’ll ask the team about it when they’re online and get back to you
    s

    Simon Gasse

    1 year ago
    Oh alright! Thank you for reproducing and taking it up @Kevin Kho! 🙂 My version:
    $ pip show prefect
    Name: prefect
    Version: 0.15.0
    BTW if you can 'upcast' this discussion to an issue, you already have it reported on GitHub:https://github.com/PrefectHQ/prefect/discussions/4740
    Kevin Kho

    Kevin Kho

    1 year ago
    @Bruno Murino a fix has been provided in case you missed this
    Bruno Murino

    Bruno Murino

    1 year ago
    Yes I saw! To use the “master” version of the images — do you how long until a proper 0.15.1 is released fixing this bug?
    Kevin Kho

    Kevin Kho

    1 year ago
    1 week or 2 weeks from now minimum
    s

    Simon Gasse

    1 year ago
    @Bruno Murino if you want to pin a version before, you can pin the docker images by their
    sha256
    in the docker-compose.yml:
    image: prefecthq/apollo@sha256:65ca4f0a4a667fd48360a5c00b4670a796dbc43f2e3a1a0e40ad8b40e1e5cd7f
        ...
        image: prefecthq/server@sha256:0490b527756a53bd3c80513cf09d8c51fdf002c7991524898f62fdd43a31a343
    Bruno Murino

    Bruno Murino

    1 year ago
    @Simon Gasse This is very helpful, thanks for sharing it! It is indeed working now 🙂 Thanks everyone!
    Adam

    Adam

    1 year ago
    Late reply and I know its solved now, but could your issue just be args vs kwargs @Bruno Murino:
    schedule = IntervalSchedule(interval=timedelta(minutes=2))
    with Flow("Scheduled Hello", schedule) as flow:
        say_hello()
    We normally do:
    schedule = IntervalSchedule(interval=timedelta(minutes=2))
    with Flow("Scheduled Hello", schedule=schedule) as flow:
        say_hello()
    Bruno Murino

    Bruno Murino

    1 year ago
    I actually do it like “flow.schedule = schedule” and actually their bugfix solved my problem as well!