> They will struggle if they add new resources and nothing happens for
5 minutes.
The same applies if they start playing with FLIP-291 APIs. I'm
wondering if the cooldown makes sense there since it was the user's
deliberate choice to push new requirements. 🤔
Best,
D.
On Tue, Jul 4, 2023 at 9:11 AM David Morávek <d...@apache.org> wrote:
The FLIP reads sane to me. I'm unsure about the default values,
though; 5 minutes of wait time between rescales feels rather
strict, and we should rethink it to provide a better
out-of-the-box experience.
I'd focus on newcomers trying AS / Reactive Mode out. They will
struggle if they add new resources and nothing happens for 5
minutes. I'd suggest defaulting to
/jobmanager.adaptive-scheduler.resource-stabilization-timeout/ (which
defaults to 10s).
I'm still struggling to grasp max internal (force rescale).
Ignoring `AdaptiveScheduler#shouldRescale()` condition seems
rather dangerous. Wouldn't a simple case where you add a new TM
and remove it before the max interval is reached (so there is
nothing to do) result in an unnecessary job restart?
Best,
D.
On Thu, Jun 29, 2023 at 3:43 PM Etienne Chauchot
<echauc...@apache.org> wrote:
Thanks Chesnay for your feedback. I have updated the FLIP.
I'll start a
vote thread.
Best
Etienne
Le 28/06/2023 à 11:49, Chesnay Schepler a écrit :
> > we should schedule a check that will rescale if
> min-parallelism-increase is met. Then, what it the use of
> scaling-interval.max timeout in that context ?
>
> To force a rescale if min-parallelism-increase is not met
(but we
> could still run above the current parallelism).
>
> min-parallelism-increase is a trade-off between the cost of
rescaling
> vs the performance benefit of the parallelism increase. Over
time the
> balance tips more and more in favor of the parallelism
increase, hence
> we should eventually rescale anyway even if the minimum
isn't met, or
> at least give users the option to do so.
>
> > I meant the opposite: not having only the cooldown but
having only
> the stabilization time. I must have missed something because
what I
> wonder is: if every rescale entails a restart of the
pipeline and
> every restart entails passing in waiting for resources
state, then why
> introduce a cooldown when there is already at each rescale a
stable
> resource timeout ?
>
> It is technically correct that the stable resource timeout
can be used
> to limit the number of rescale operations per interval,
however during
> that time the job isn't running, in contrast to the cooldown.
>
> Having both just gives you a lot more flexibility.
> "I want at most 1 rescale operation per hour, and wait at
most 1
> minute for resource to stabilize when a rescale happens".
> You can't express this with only one of the options.
>
> On 20/06/2023 14:41, Etienne Chauchot wrote:
>> Hi Chesnay,
>>
>> Thanks for your feedback. Comments inline
>>
>> Le 16/06/2023 à 17:24, Chesnay Schepler a écrit :
>>> 1) Options specific to the adaptive scheduler should start
with
>>> "jobmanager.adaptive-scheduler".
>>
>>
>> ok
>>
>>
>>> 2)
>>> There isn't /really /a notion of a "scaling event". The
scheduler is
>>> informed about new/lost slots and job failures, and reacts
>>> accordingly by maybe rescaling the job.
>>> (sure, you can think of these as events, but you can think of
>>> practically everything as events)
>>>
>>> There shouldn't be a queue for events. All the scheduler
should have
>>> to know is that the next rescale check is scheduled for
time T,
>>> which in practice boils down to a flag and a scheduled
action that
>>> runs Executing#maybeRescale.
>>
>>
>> Makes total sense, its very simple like this. Thanks for the
>> precision and pointer. After the related FLIPs, I'll look
at the code
>> now.
>>
>>
>>> With that in mind, we also have to look at how we keep
this state
>>> around. Presumably it is scoped to the current state, such
that the
>>> cooldown is reset if a job fails.
>>> Maybe we should add a separate ExecutingWithCooldown
state; not sure
>>> yet.
>>
>>
>> Yes loosing cooldown state and cooldown reset upon failure
is what I
>> suggested in point 3 in previous email. Not sure either for
a new
>> state, I'll figure it out after experimenting with the
code. I'll
>> update the FLIP then.
>>
>>
>>>
>>> It would be good to clarify whether this FLIP only
attempts to cover
>>> scale up operations, or also scale downs in case of slot
losses.
>>
>>
>> When there are slots loss, most of the time it is due to a
TM loss so
>> there should be several slots lost at the same time but
(hopefully)
>> only once. There should not be many scale downs in a row
(but still
>> cascading failures can happen). I think, we should just
protect
>> against having scale ups immediately following. For that, I
think we
>> could just keep the current behavior of transitioning to
Restarting
>> state and then back to Waiting for Resources state. This
state will
>> protect us against scale ups immediately following
failure/restart.
>>
>>
>>>
>>> We should also think about how it relates to the externalized
>>> declarative resource management. Should we always rescale
>>> immediately? Should we wait until the cooldown is over?
>>
>>
>> It relates to point 2, no ? we should rescale immediately
only if
>> last rescale was done more than scaling-interval.min ago
otherwise
>> schedule a rescale at last-rescale + scaling-interval.min time.
>>
>>
>>> Related to this, there's the min-parallelism-increase
option, that
>>> if for example set to "2" restricts rescale operations to
only occur
>>> if the parallelism increases by at least 2.
>>
>>
>> yes I saw that in the code
>>
>>
>>> Ideally however there would be a max timeout for this.
>>>
>>> As such we could maybe think about this a bit differently:
>>> Add 2 new options instead of 1:
>>> jobmanager.adaptive-scheduler.scaling-interval.min: The
minimum time
>>> the scheduler will wait for the next effective rescale
operations.
>>> jobmanager.adaptive-scheduler.scaling-interval.max: The
maximum time
>>> the scheduler will wait for the next effective rescale
operations.
>>
>>
>> At point 2, we said that when slots change (requirements
change or
>> new slots available), if last rescale check (call to
maybeRescale)
>> was done less than scaling-interval.min ago, we should
schedule a
>> check that will rescale if min-parallelism-increase is met.
Then,
>> what it the use of scaling-interval.max timeout in that
context ?
>>
>>
>>>
>>> 3) It sounds fine that we lose the cooldown state, because
imo we
>>> want to reset the cooldown anyway on job failures (because
a job
>>> failure inherently implies a potential rescaling).
>>
>>
>> exactly.
>>
>>
>>>
>>> 4) The stabilization time isn't really redundant and serves a
>>> different use-case. The idea behind is that if a users
adds multiple
>>> TMs at once then we don't want to rescale immediately at
the first
>>> received slot. Without the stabilization time the cooldown
would
>>> actually cause bad behavior here, because not only would
we rescale
>>> immediately upon receiving the minimum required slots to
scale up,
>>> but we also wouldn't use the remaining slots just because the
>>> cooldown says so.
>>
>>
>> I meant the opposite: not having only the cooldown but
having only
>> the stabilization time. I must have missed something
because what I
>> wonder is: if every rescale entails a restart of the
pipeline and
>> every restart entails passing in waiting for resources
state, then
>> why introduce a cooldown when there is already at each
rescale a
>> stable resource timeout ?
>>
>>
>> Best
>>
>> Etienne
>>
>>
>>
>>>
>>> On 16/06/2023 15:47, Etienne Chauchot wrote:
>>>> Hi Robert,
>>>>
>>>> Thanks for your feedback. I don't know the scheduler part
well
>>>> enough yet and I'm taking this ticket as a learning workshop.
>>>>
>>>> Regarding your comments:
>>>>
>>>> 1. Taking a look at the AdaptiveScheduler class which
takes all its
>>>> configuration from the JobManagerOptions, and also to be
consistent
>>>> with other parameters name, I'd suggest
>>>> /jobmanager.scheduler-scaling-cooldown-period/
>>>>
>>>> 2. I thought scaling events existed already and the
scheduler
>>>> received them as mentioned in FLIP-160 (cf "Whenever the
scheduler
>>>> is in the Executing state and receives new slots") or in
FLIP-138
>>>> (cf "Whenever new slots are available the SlotPool
notifies the
>>>> Scheduler"). If it is not the case (it is the scheduler
who asks
>>>> for slots), then there is no need for storing scaling
requests indeed.
>>>>
>>>> => I need a confirmation here
>>>>
>>>> 3. If we loose the JobManager, we loose both the
AdaptiveScheduler
>>>> state and the CoolDownTimer state. So, upon recovery, it
would be
>>>> as if there was no ongoing coolDown period. So, a first
re-scale
>>>> could happen right away and it will start a coolDown
period. A
>>>> second re-scale would have to wait for the end of this
period.
>>>>
>>>> 4. When a pipeline is re-scaled, it is restarted. Upon
restart, the
>>>> AdaptiveScheduler passes again in the "waiting for
resources" state
>>>> as FLIP-160 suggests. If so, then it seems that the
coolDown period
>>>> is kind of redundant with the
resource-stabilization-timeout. I
>>>> guess it is not the case otherwise the FLINK-21883 ticket
would not
>>>> have been created.
>>>>
>>>> => I need a confirmation here also.
>>>>
>>>>
>>>> Thanks for your views on point 2 and 4.
>>>>
>>>>
>>>> Best
>>>>
>>>> Etienne
>>>>
>>>> Le 15/06/2023 à 13:35, Robert Metzger a écrit :
>>>>> Thanks for the FLIP.
>>>>>
>>>>> Some comments:
>>>>> 1. Can you specify the full proposed configuration name? "
>>>>> scaling-cooldown-period" is probably not the full config
name?
>>>>> 2. Why is the concept of scaling events and a scaling queue
>>>>> needed? If I
>>>>> remember correctly, the adaptive scheduler will just
check how many
>>>>> TaskManagers are available and then adjust the execution
graph
>>>>> accordingly.
>>>>> There's no need to store a number of scaling events. We
just need to
>>>>> determine the time to trigger an adjustment of the
execution graph.
>>>>> 3. What's the behavior wrt to JobManager failures (e.g.
we lose
>>>>> the state
>>>>> of the Adaptive Scheduler?). My proposal would be to
just reset the
>>>>> cooldown period, so after recovery of a JobManager, we
have to
>>>>> wait at
>>>>> least for the cooldown period until further scaling
operations are
>>>>> done.
>>>>> 4. What's the relationship to the
>>>>>
"jobmanager.adaptive-scheduler.resource-stabilization-timeout"
>>>>> configuration?
>>>>>
>>>>> Thanks a lot for working on this!
>>>>>
>>>>> Best,
>>>>> Robert
>>>>>
>>>>> On Wed, Jun 14, 2023 at 3:38 PM Etienne
>>>>> Chauchot<echauc...@apache.org>
>>>>> wrote:
>>>>>
>>>>>> Hi all,
>>>>>>
>>>>>> @Yukia,I updated the FLIP to include the aggregation of
the staked
>>>>>> operations that we discussed below PTAL.
>>>>>>
>>>>>> Best
>>>>>>
>>>>>> Etienne
>>>>>>
>>>>>>
>>>>>> Le 13/06/2023 à 16:31, Etienne Chauchot a écrit :
>>>>>>> Hi Yuxia,
>>>>>>>
>>>>>>> Thanks for your feedback. The number of potentially
stacked
>>>>>>> operations
>>>>>>> depends on the configured length of the cooldown period.
>>>>>>>
>>>>>>>
>>>>>>>
>>>>>>> The proposition in the FLIP is to add a minimum delay
between 2
>>>>>>> scaling
>>>>>>> operations. But, indeed, an optimization could be to
still stack
>>>>>>> the
>>>>>>> operations (that arrive during a cooldown period) but
maybe not
>>>>>>> take
>>>>>>> only the last operation but rather aggregate them in
order to
>>>>>>> end up
>>>>>>> with a single aggregated operation when the cooldown
period
>>>>>>> ends. For
>>>>>>> example, let's say 3 taskManagers come up and 1 comes
down
>>>>>>> during the
>>>>>>> cooldown period, we could generate a single operation
of scale
>>>>>>> up +2
>>>>>>> when the period ends.
>>>>>>>
>>>>>>> As a side note regarding your comment on "it'll take a
long time to
>>>>>>> finish all", please keep in mind that the reactive
mode (at
>>>>>>> least for
>>>>>>> now) is only available for streaming pipeline which
are in essence
>>>>>>> infinite processing.
>>>>>>>
>>>>>>> Another side note: when you mention "every taskManagers
>>>>>>> connecting",
>>>>>>> if you are referring to the start of the pipeline,
please keep
>>>>>>> in mind
>>>>>>> that the adaptive scheduler has a "waiting for
resources" timeout
>>>>>>> period before starting the pipeline in which all
taskmanagers
>>>>>>> connect
>>>>>>> and the parallelism is decided.
>>>>>>>
>>>>>>> Best
>>>>>>>
>>>>>>> Etienne
>>>>>>>
>>>>>>> Le 13/06/2023 à 03:58, yuxia a écrit :
>>>>>>>> Hi, Etienne. Thanks for driving it. I have one
question about the
>>>>>>>> mechanism of the cooldown timeout.
>>>>>>>>
>>>>>>>> From the Proposed Changes part, if a scalling event is
>>>>>>>> received and
>>>>>>>> it falls during the cooldown period, it'll be stacked
to be
>>>>>>>> executed
>>>>>>>> after the period ends. Also, from the description of
>>>>>>>> FLINK-21883[1],
>>>>>>>> cooldown timeout is to avoid rescaling the job very
frequently,
>>>>>>>> because TaskManagers are not all connecting at the
same time.
>>>>>>>>
>>>>>>>> So, is it possible that every taskmanager connecting
will
>>>>>>>> produce a
>>>>>>>> scalling event and it'll be stacked with many scale
up event which
>>>>>>>> causes it'll take a long time to finish all? Can we
just take the
>>>>>>>> last one event?
>>>>>>>>
>>>>>>>> [1]:https://issues.apache.org/jira/browse/FLINK-21883
>>>>>>>>
>>>>>>>> Best regards, Yuxia
>>>>>>>>
>>>>>>>> ----- 原始邮件 ----- 发件人: "Etienne
>>>>>>>> Chauchot"<echauc...@apache.org>
>>>>>>>> 收件人:
>>>>>>>> "dev"<dev@flink.apache.org>, "Robert
Metzger"<metrob...@gmail.com>
>>>>>>>> 发送时间: 星期一, 2023年 6 月 12日 下午 11:34:25 主题:
[DISCUSS]
>>>>>>>> FLIP-322
>>>>>>>> Cooldown
>>>>>>>> period for adaptive scheduler
>>>>>>>>
>>>>>>>> Hi,
>>>>>>>>
>>>>>>>> I’d like to start a discussion about FLIP-322 [1] which
>>>>>>>> introduces a
>>>>>>>> cooldown period for the adaptive scheduler.
>>>>>>>>
>>>>>>>> I'd like to get your feedback especially @Robert as
you opened the
>>>>>>>> related ticket and worked on the reactive mode a lot.
>>>>>>>>
>>>>>>>> [1]
>>>>>>>>
>>>>>>
https://cwiki.apache.org/confluence/display/FLINK/FLIP-322+Cooldown+period+for+adaptive+scheduler
>>>>>>
>>>>>>> Best
>>>>>>>> Etienne
>>>
>>>
>
