Hey Gabor,
let me know your cwiki username, and I can give you write permissions.
On Fri, Jan 28, 2022 at 4:05 PM Gabor Somogyi <gabor.g.somo...@gmail.com>
wrote:
Thanks for making the design better! No further thing to discuss from my
side.
Started to reflect the agreement in the FLIP doc.
Since I don't have access to the wiki I need to ask Marci to do that
which
may take some time.
G
On Fri, Jan 28, 2022 at 3:52 PM David Morávek <d...@apache.org> wrote:
Hi,
AFAIU an under registration TM is not added to the registered TMs map
until
RegistrationResponse ..
I think you're right, with a careful design around threading
(delegating
update broadcasts to the main thread) + synchronous initial update
(that
would be nice to avoid) this should be doable.
Not sure what you mean "we can't register the TM without providing it
with
token" but in unsecure configuration registration must happen w/o
tokens.
Exactly as you describe it, this was meant only for the "kerberized /
secured" cluster case, in other cases we wouldn't enforce a non-null
token
in the response
I think this is a good idea in general.
+1
If you don't have any more thoughts on the RPC / lifecycle part, can
you
please reflect it into the FLIP?
D.
On Fri, Jan 28, 2022 at 3:16 PM Gabor Somogyi <
gabor.g.somo...@gmail.com
wrote:
- Make sure DTs issued by single DTMs are monotonically increasing
(can
be
sorted on TM side)
AFAIU an under registration TM is not added to the registered TMs
map
until
RegistrationResponse
is processed which would contain the initial tokens. If that's true
then
how is it possible to have race with
DTM update which is working on the registered TMs list?
To be more specific "taskExecutors" is the registered map of TMs to
which
DTM can send updated tokens
but this doesn't contain the under registration TM while
RegistrationResponse is not processed, right?
Of course if DTM can update while RegistrationResponse is processed
then
somehow sorting would be
required and that case I would agree.
- Scope DT updates by the RM ID and ensure that TM only accepts
update
from
the current leader
I've planned this initially the mentioned way so agreed.
- Return initial token with the RegistrationResponse, which should
make
the
RPC contract bit clearer (ensure that we can't register the TM
without
providing it with token)
I think this is a good idea in general. Not sure what you mean "we
can't
register the TM without
providing it with token" but in unsecure configuration registration
must
happen w/o tokens.
All in all the newly added tokens field must be somehow optional.
G
On Fri, Jan 28, 2022 at 2:22 PM David Morávek <d...@apache.org>
wrote:
We had a long discussion with Chesnay about the possible edge
cases
and
it
basically boils down to the following two scenarios:
1) There is a possible race condition between TM registration (the
first
DT
update) and token refresh if they happen simultaneously. Than the
registration might beat the refreshed token. This could be easily
addressed
if DTs could be sorted (eg. by the expiration time) on the TM
side.
In
other words, if there are multiple updates at the same time we
need
to
make
sure that we have a deterministic way of choosing the latest one.
One idea by Chesnay that popped up during this discussion was
whether
we
could simply return the initial token with the
RegistrationResponse
to
avoid making an extra call during the TM registration.
2) When the RM leadership changes (eg. because zookeeper session
times
out)
there might be a race condition where the old RM is shutting down
and
updates the tokens, that it might again beat the registration
token
of
the
new RM. This could be avoided if we scope the token by
_ResourceManagerId_
and only accept updates for the current leader (basically we'd
have
an
extra parameter to the _updateDelegationToken_ method).
-
DTM is way simpler then for example slot management, which could
receive
updates from the JobMaster that RM might not know about.
So if you want to go in the path you're describing it should be
doable
and
we'd propose following to cover all cases:
- Make sure DTs issued by single DTMs are monotonically increasing
(can
be
sorted on TM side)
- Scope DT updates by the RM ID and ensure that TM only accepts
update
from
the current leader
- Return initial token with the RegistrationResponse, which should
make
the
RPC contract bit clearer (ensure that we can't register the TM
without
providing it with token)
Any thoughts?
On Fri, Jan 28, 2022 at 10:53 AM Gabor Somogyi <
gabor.g.somo...@gmail.com>
wrote:
Thanks for investing your time!
The first 2 bulletpoint are clear.
If there is a chance that a TM can go to an inconsistent state
then I
agree
with the 3rd bulletpoint.
Just before we agree on that I would like to learn something new
and
understand how is it possible that a TM
gets corrupted? (In Spark I've never seen such thing and no
mechanism
to
fix this but Flink is definitely not Spark)
Here is my understanding:
* DTM pushes new obtained DTs to TMs and if any exception occurs
then a
retry after "security.kerberos.tokens.retry-wait"
happens. This means DTM retries until it's not possible to send
new
DTs
to
all registered TMs.
* New TM registration must fail if "updateDelegationToken" fails
* "updateDelegationToken" fails consistently like a DB (at
least I
plan
to
implement it that way).
If DTs are arriving on the TM side then a single
"UserGroupInformation.getCurrentUser.addCredentials"
will be called which I've never seen it failed.
* I hope all other code parts are not touching existing DTs
within
the
JVM
I would like to emphasize I'm not against to add it just want to
see
what
kind of problems are we facing.
It would ease to catch bugs earlier and help in the maintenance.
All in all I would buy the idea to add the 3rd bullet if we
foresee
the
need.
G
On Fri, Jan 28, 2022 at 10:07 AM David Morávek <d...@apache.org
wrote:
Hi Gabor,
This is definitely headed in a right direction +1.
I think we still need to have a safeguard in case some of the
TMs
gets
into
the inconsistent state though, which will also eliminate the
need
for
implementing a custom retry mechanism (when
_updateDelegationToken_
call
fails for some reason).
We already have this safeguard in place for slot pool (in case
there
are
some slots in inconsistent state - eg. we haven't freed them
for
some
reason) and for the partition tracker, which could be simply
enhanced.
This
is done via periodic heartbeat from TaskManagers to the
ResourceManager
that contains report about state of these two components
(from TM
perspective) so the RM can reconcile their state if necessary.
I don't think adding an additional field to
_TaskExecutorHeartbeatPayload_
should be a concern as we only heartbeat every ~ 10s by
default
and
the
new
field would be small compared to rest of the existing payload.
Also
heartbeat doesn't need to contain the whole DT, but just some
identifier
which signals whether it uses the right one, that could be
significantly
smaller.
This is still a PUSH based approach as the RM would again call
the
newly
introduced _updateDelegationToken_ when it encounters
inconsistency
(eg.
due to a temporary network partition / a race condition we
didn't
test
for
/ some other scenario we didn't think about). In practice
these
inconsistencies are super hard to avoid and reason about (and
unfortunately
yes, we see them happen from time to time), so reusing the
existing
mechanism that is designed for this exact problem simplify
things.
To sum this up we'd have three code paths for calling
_updateDelegationToken_:
1) When the TM registers, we push the token (if DTM already
has
it)
to
it
2) When DTM obtains a new token it broadcasts it to all
currently
connected
TMs
3) When a TM gets out of sync, DTM would reconcile it's state
WDYT?
Best,
D.
On Wed, Jan 26, 2022 at 9:03 PM David Morávek <
d...@apache.org>
wrote:
Thanks the update, I'll go over it tomorrow.
On Wed, Jan 26, 2022 at 5:33 PM Gabor Somogyi <
gabor.g.somo...@gmail.com
wrote:
Hi All,
Since it has turned out that DTM can't be added as member
of
JobMaster
<
https://github.com/gaborgsomogyi/flink/blob/8ab75e46013f159778ccfce52463e7bc63e395a9/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobMaster.java#L176
I've
came up with a better proposal.
David, thanks for pinpointing this out, you've caught a
bug in
the
early
phase!
Namely ResourceManager
<
https://github.com/apache/flink/blob/674bc96662285b25e395fd3dddf9291a602fc183/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/ResourceManager.java#L124
is
a single instance class where DTM can be added as member
variable.
It has a list of all already registered TMs and new TM
registration
is
also
happening here.
The following can be added from logic perspective to be
more
specific:
* Create new DTM instance in ResourceManager
<
https://github.com/apache/flink/blob/674bc96662285b25e395fd3dddf9291a602fc183/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/ResourceManager.java#L124
and
start it (re-occurring thread to obtain new tokens)
* Add a new function named "updateDelegationTokens" to
TaskExecutorGateway
<
https://github.com/apache/flink/blob/674bc96662285b25e395fd3dddf9291a602fc183/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskExecutorGateway.java#L54
* Call "updateDelegationTokens" on all registered TMs to
propagate
new
DTs
* In case of new TM registration call
"updateDelegationTokens"
before
registration succeeds to setup new TM properly
This way:
* only a single DTM would live within a cluster which is
the
expected
behavior
* DTM is going to be added to a central place where all
deployment
target
can make use of it
* DTs are going to be pushed to TMs which would generate
less
network
traffic than pull based approach
(please see my previous mail where I've described both
approaches)
* HA scenario is going to be consistent because such
<
https://github.com/apache/flink/blob/674bc96662285b25e395fd3dddf9291a602fc183/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskExecutor.java#L1069
a solution can be added to "updateDelegationTokens"
@David or all others plz share whether you agree on this or
you
have
better
idea/suggestion.
BR,
G
On Tue, Jan 25, 2022 at 11:00 AM Gabor Somogyi <
gabor.g.somo...@gmail.com
wrote:
First of all thanks for investing your time and helping
me
out.
As I
see
you have pretty solid knowledge in the RPC area.
I would like to rely on your knowledge since I'm learning
this
part.
- Do we need to introduce a new RPC method or can we
for
example
piggyback
on heartbeats?
I'm fine with either solution but one thing is important
conceptually.
There are fundamentally 2 ways how tokens can be updated:
- Push way: When there are new DTs then JM JVM pushes
DTs to
TM
JVMs.
This
is the preferred one since tiny amount of control logic
needed.
- Pull way: Each time a TM would like to poll JM whether
there
are
new
tokens and each TM wants to decide alone whether DTs
needs
to
be
updated or
not.
As you've mentioned here some ID needs to be generated,
it
would
generated
quite some additional network traffic which can be
definitely
avoided.
As a final thought in Spark we've had this way of DT
propagation
logic
and
we've had major issues with it.
So all in all DTM needs to obtain new tokens and there
must
a
way
to
send
this data to all TMs from JM.
- What delivery semantics are we looking for? (what if
we're
only
able to
update subset of TMs / what happens if we exhaust
retries /
should
we
even
have the retry mechanism whatsoever) - I have a feeling
that
somehow
leveraging the existing heartbeat mechanism could help to
answer
these
questions
Let's go through these questions one by one.
What delivery semantics are we looking for?
DTM must receive an exception when at least one TM was
not
able
to
get
DTs.
what if we're only able to update subset of TMs?
Such case DTM will reschedule token obtain after
"security.kerberos.tokens.retry-wait" time.
what happens if we exhaust retries?
There is no number of retries. In default configuration
tokens
needs
to
be
re-obtained after one day.
DTM tries to obtain new tokens after 1day * 0.75
(security.kerberos.tokens.renewal-ratio) = 18 hours.
When fails it retries after
"security.kerberos.tokens.retry-wait"
which
is
1 hour by default.
If it never succeeds then authentication error is going
to
happen
on
the
TM side and the workload is
going to stop.
should we even have the retry mechanism whatsoever?
Yes, because there are always temporary cluster issues.
What does it mean for the running application (how does
this
look
like
from
the user perspective)? As far as I remember the logs are
only
collected
("aggregated") after the container is stopped, is that
correct?
With default config it works like that but it can be
forced
to
aggregate
at specific intervals.
A useful feature is forcing YARN to aggregate logs while
the
job
is
still
running.
For long-running jobs such as streaming jobs, this is
invaluable.
To
do
this,
yarn.nodemanager.log-aggregation.roll-monitoring-interval-seconds
must
be
set to a non-negative value.
When this is set, a timer will be set for the given
duration,
and
whenever
that timer goes off,
log aggregation will run on new files.
I think
this topic should get its own section in the FLIP (having
some
cross
reference to YARN ticket would be really useful, but I'm
not
sure
if
there
are any).
I think this is important knowledge but this FLIP is not
touching
the
already existing behavior.
DTs are set on the AM container which is renewed by YARN
until
it's
not
possible anymore.
Any kind of new code is not going to change this
limitation.
BTW,
there
is
no jira for this.
If you think it worth to write this down then I think the
good
place
is
the official security doc
area as caveat.
If we split the FLIP into two parts / sections that
I've
suggested,
I
don't
really think that you need to explicitly test for each
deployment
scenario
/ cluster framework, because the DTM part is completely
independent
of
the
deployment target. Basically this is what I'm aiming for
with
"making
it
work with the standalone" (as simple as starting a new
java
process)
Flink
first (which is also how most people deploy streaming
application
on
k8s
and the direction we're pushing forward with the
auto-scaling
/
reactive
mode initiatives).
I see your point and agree the main direction. k8s is the
megatrend
which
most of the peoples
will use sooner or later. Not 100% sure what kind of
split
you
suggest
but
in my view
the main target is to add this feature and I'm open to
any
logical
work
ordering.
Please share the specific details and we work it out...
G
On Mon, Jan 24, 2022 at 3:04 PM David Morávek <
d...@apache.org>
wrote:
Could you point to a code where you think it could be
added
exactly?
A
helping hand is welcome here 🙂
I think you can take a look at
_ResourceManagerPartitionTracker_
[1]
which
seems to have somewhat similar properties to the DTM.
One topic that needs to be addressed there is how the
RPC
with
the
_TaskExecutorGateway_ should look like.
- Do we need to introduce a new RPC method or can we for
example
piggyback
on heartbeats?
- What delivery semantics are we looking for? (what if
we're
only
able
to
update subset of TMs / what happens if we exhaust
retries /
should
we
even
have the retry mechanism whatsoever) - I have a feeling
that
somehow
leveraging the existing heartbeat mechanism could help
to
answer
these
questions
In short, after DT reaches it's max lifetime then log
aggregation
stops
What does it mean for the running application (how does
this
look
like
from
the user perspective)? As far as I remember the logs are
only
collected
("aggregated") after the container is stopped, is that
correct? I
think
this topic should get its own section in the FLIP
(having
some
cross
reference to YARN ticket would be really useful, but I'm
not
sure
if
there
are any).
All deployment modes (per-job, per-app, ...) are
planned to
be
tested
and
expect to work with the initial implementation however
not
all
deployment
targets (k8s, local, ...
If we split the FLIP into two parts / sections that I've
suggested, I
don't
really think that you need to explicitly test for each
deployment
scenario
/ cluster framework, because the DTM part is completely
independent
of
the
deployment target. Basically this is what I'm aiming for
with
"making
it
work with the standalone" (as simple as starting a new
java
process)
Flink
first (which is also how most people deploy streaming
application
on
k8s
and the direction we're pushing forward with the
auto-scaling /
reactive
mode initiatives).
The whole integration with YARN (let's forget about log
aggregation
for a
moment) / k8s-native only boils down to how do we make
the
keytab
file
local to the JobManager so the DTM can read it, so it's
basically
built on
top of that. The only special thing that needs to be
tested
there
is
the
"keytab distribution" code path.
[1]
https://github.com/apache/flink/blob/release-1.14.3/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/ResourceManagerPartitionTracker.java
Best,
D.
On Mon, Jan 24, 2022 at 12:35 PM Gabor Somogyi <
gabor.g.somo...@gmail.com
wrote:
There is a separate JobMaster for each job
within a Flink cluster and each JobMaster only has a
partial
view
of
the
task managers
Good point! I've had a deeper look and you're right.
We
definitely
need
to
find another place.
Related per-cluster or per-job keytab:
In the current code per-cluster keytab is implemented
and
I'm
intended
to
keep it like this within this FLIP. The reason is
simple:
tokens
on
TM
side
can be stored within the UserGroupInformation (UGI)
structure
which
is
global. I'm not telling it's impossible to change that
but
I
think
that
this is such a complexity which the initial
implementation
is
not
required
to contain. Additionally we've not seen such need from
user
side.
If
the
need may rise later on then another FLIP with this
topic
can
be
created
and
discussed. Proper multi-UGI handling within a single
JVM
is a
topic
where
several round of deep-dive with the Hadoop/YARN guys
are
required.
single DTM instance embedded with
the ResourceManager (the Flink component)
Could you point to a code where you think it could be
added
exactly?
A
helping hand is welcome here🙂
Then the single (initial) implementation should work
with
all
the
deployments modes out of the box (which is not what
the
FLIP
suggests).
Is
that correct?
All deployment modes (per-job, per-app, ...) are
planned
to
be
tested
and
expect to work with the initial implementation however
not
all
deployment
targets (k8s, local, ...) are not intended to be
tested.
Per
deployment
target new jira needs to be created where I expect
small
number
of
codes
needs to be added and relatively expensive testing
effort
is
required.
I've taken a look into the prototype and in the
"YarnClusterDescriptor"
you're injecting a delegation token into the AM [1]
(that's
obtained
using
the provided keytab). If I understand this correctly
from
previous
discussion / FLIP, this is to support log aggregation
and
DT
has
a
limited
validity. How is this DT going to be renewed?
You're clever and touched a limitation which Spark has
too.
In
short,
after
DT reaches it's max lifetime then log aggregation
stops.
I've
had
several
deep-dive rounds with the YARN guys at Spark years
because
wanted
to
fill
this gap. They can't provide us any way to re-inject
the
newly
obtained
DT
so at the end I gave up this.
BR,
G
On Mon, 24 Jan 2022, 11:00 David Morávek, <
d...@apache.org
wrote:
Hi Gabor,
There is actually a huge difference between
JobManager
(process)
and
JobMaster (job coordinator). The naming is
unfortunately
bit
misleading
here from historical reasons. There is a separate
JobMaster
for
each
job
within a Flink cluster and each JobMaster only has a
partial
view
of
the
task managers (depends on where the slots for a
particular
job
are
allocated). This means that you'll end up with N
"DelegationTokenManagers"
competing with each other (N = number of running
jobs
in
the
cluster).
This makes me think we're mixing two abstraction
levels
here:
a) Per-cluster delegation tokens
- Simpler approach, it would involve a single DTM
instance
embedded
with
the ResourceManager (the Flink component)
b) Per-job delegation tokens
- More complex approach, but could be more flexible
from
the
user
side of
things.
- Multiple DTM instances, that are bound with the
JobMaster
lifecycle.
Delegation tokens are attached with a particular
slots
that
are
executing
the job tasks instead of the whole task manager (TM
could
be
executing
multiple jobs with different tokens).
- The question is which keytab should be used for
the
clustering
framework,
to support log aggregation on YARN (an extra keytab,
keytab
that
comes
with
the first job?)
I think these are the things that need to be
clarified
in
the
FLIP
before
proceeding.
A follow-up question for getting a better
understanding
where
this
should
be headed: Are there any use cases where user may
want
to
use
different
keytabs with each job, or are we fine with using a
cluster-wide
keytab?
If
we go with per-cluster keytabs, is it OK that all
jobs
submitted
into
this
cluster can access it (even the future ones)? Should
this
be
a
security
concern?
Presume you though I would implement a new class
with
JobManager
name.
The
plan is not that.
I've never suggested such thing.
No. That said earlier DT handling is planned to be
done
completely
in
Flink. DTM has a renewal thread which re-obtains
tokens
in
the
proper
time
when needed.
Then the single (initial) implementation should work
with
all
the
deployments modes out of the box (which is not what
the
FLIP
suggests).
Is
that correct?
If the cluster framework, also requires delegation
token
for
their
inner
working (this is IMO only applies to YARN), it might
need
an
extra
step
(injecting the token into application master
container).
Separating the individual layers (actual Flink
cluster
-
basically
making
this work with a standalone deployment / "cluster
framework" -
support
for
YARN log aggregation) in the FLIP would be useful.
Reading the linked Spark readme could be useful.
I've read that, but please be patient with the
questions,
Kerberos
is
not
an easy topic to get into and I've had a very little
contact
with
it
in
the
past.
https://github.com/gaborgsomogyi/flink/blob/8ab75e46013f159778ccfce52463e7bc63e395a9/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobMaster.java#L176
I've taken a look into the prototype and in the
"YarnClusterDescriptor"
you're injecting a delegation token into the AM [1]
(that's
obtained
using
the provided keytab). If I understand this correctly
from
previous
discussion / FLIP, this is to support log
aggregation
and
DT
has
a
limited
validity. How is this DT going to be renewed?
[1]
https://github.com/gaborgsomogyi/flink/commit/8ab75e46013f159778ccfce52463e7bc63e395a9#diff-02416e2d6ca99e1456f9c3949f3d7c2ac523d3fe25378620c09632e4aac34e4eR1261
Best,
D.
On Fri, Jan 21, 2022 at 9:35 PM Gabor Somogyi <
gabor.g.somo...@gmail.com
wrote:
Here is the exact class, I'm from mobile so not
had a
look
at
the
exact
class name:
https://github.com/gaborgsomogyi/flink/blob/8ab75e46013f159778ccfce52463e7bc63e395a9/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobMaster.java#L176
That keeps track of TMs where the tokens can be
sent
to.
My feeling would be that we shouldn't really
introduce
a
new
component
with
a custom lifecycle, but rather we should try to
incorporate
this
into
existing ones.
Can you be more specific? Presume you though I
would
implement
a
new
class
with JobManager name. The plan is not that.
If I understand this correctly, this means that
we
then
push
the
token
renewal logic to YARN.
No. That said earlier DT handling is planned to be
done
completely
in
Flink. DTM has a renewal thread which re-obtains
tokens
in
the
proper
time
when needed. YARN log aggregation is a totally
different
feature,
where
YARN does the renewal. Log aggregation was an
example
why
the
code
can't
be
100% reusable for all resource managers. Reading
the
linked
Spark
readme
could be useful.
G
On Fri, 21 Jan 2022, 21:05 David Morávek, <
d...@apache.org
wrote:
JobManager is the Flink class.
There is no such class in Flink. The closest
thing
to
the
JobManager
is a
ClusterEntrypoint. The cluster entrypoint spawns
new
RM
Runner
&
Dispatcher
Runner that start participating in the leader
election.
Once
they
gain
leadership they spawn the actual underlying
instances
of
these
two
"main
components".
My feeling would be that we shouldn't really
introduce
a
new
component
with
a custom lifecycle, but rather we should try to
incorporate
this
into
existing ones.
My biggest concerns would be:
- How would the lifecycle of the new component
look
like
with
regards
to
HA
setups. If we really try to decide to introduce
a
completely
new
component,
how should this work in case of multiple
JobManager
instances?
- Which components does it talk to / how? For
example
how
does
the
broadcast of new token to task managers
(TaskManagerGateway)
look
like?
Do
we simply introduce a new RPC on the
ResourceManagerGateway
that
broadcasts
it or does the new component need to do some
kind
of
bookkeeping
of
task
managers that it needs to notify?
YARN based HDFS log aggregation would not work
by
dropping
that
code.
Just
to be crystal clear, the actual implementation
contains
this
fir
exactly
this reason.
This is the missing part +1. If I understand
this
correctly,
this
means
that we then push the token renewal logic to
YARN.
How
do
you
plan to
implement the renewal logic on k8s?
D.
On Fri, Jan 21, 2022 at 8:37 PM Gabor Somogyi <
gabor.g.somo...@gmail.com
wrote:
I think we might both mean something
different
by
the
RM.
You feel it well, I've not specified these
terms
well
in
the
explanation.
RM I meant resource management framework.
JobManager
is
the
Flink
class.
This means that inside JM instance there will
be
a
DTM
instance, so
they
would have the same lifecycle. Hope I've
answered
the
question.
If we have tokens available on the client
side,
why
do
we
need to
set
them
into the AM (yarn specific concept) launch
context?
YARN based HDFS log aggregation would not
work by
dropping
that
code.
Just
to be crystal clear, the actual implementation
contains
this
fir
exactly
this reason.
G
On Fri, 21 Jan 2022, 20:12 David Morávek, <
d...@apache.org
wrote:
Hi Gabor,
1. One thing is important, token management
is
planned
to
be
done
generically within Flink and not
scattered in
RM
specific
code.
JobManager
has a DelegationTokenManager which obtains
tokens
time-to-time
(if
configured properly). JM knows which
TaskManagers
are
in
place
so
it
can
distribute it to all TMs. That's it
basically.
I think we might both mean something
different
by
the
RM.
JobManager
is
basically just a process encapsulating
multiple
components,
one
of
which
is
a ResourceManager, which is the component
that
manages
task
manager
registrations [1]. There is more or less a
single
implementation
of
the
RM
with plugable drivers for the active
integrations
(yarn,
k8s).
It would be great if you could share more
details
of
how
exactly
the
DTM
is
going to fit in the current JM architecture.
2. 99.9% of the code is generic but each RM
handles
tokens
differently. A
good example is YARN obtains tokens on
client
side
and
then
sets
them
on
the newly created AM container launch
context.
This
is
purely
YARN
specific
and cant't be spared. With my actual plans
standalone
can be
changed
to
use
the framework. By using it I mean no RM
specific
DTM
or
whatsoever
is
needed.
If we have tokens available on the client
side,
why
do
we
need to
set
them
into the AM (yarn specific concept) launch
context?
Why
can't
we
simply
send them to the JM, eg. as a parameter of
the
job
submission
/
via
separate RPC call? There might be something
I'm
missing
due to
limited
knowledge, but handling the token on the
"cluster
framework"
level
doesn't
seem necessary.
[1]
https://nightlies.apache.org/flink/flink-docs-release-1.14/docs/concepts/flink-architecture/#jobmanager
Best,
D.
On Fri, Jan 21, 2022 at 7:48 PM Gabor
Somogyi <
gabor.g.somo...@gmail.com
wrote:
Oh and one more thing. I'm planning to add
this
feature
in
small
chunk
of
PRs because security is super hairy area.
That
way
reviewers
can
be
more
easily obtains the concept.
On Fri, 21 Jan 2022, 18:03 David Morávek,
<
d...@apache.org>
wrote:
Hi Gabor,
thanks for drafting the FLIP, I think
having
a
solid
Kerberos
support
is
crucial for many enterprise deployments.
I have multiple questions regarding the
implementation
(note
that I
have
very limited knowledge of Kerberos):
1) If I understand it correctly, we'll
only
obtain
tokens
in
the
job
manager and then we'll distribute them
via
RPC
(needs
to
be
secured).
Can you please outline how the
communication
will
look
like?
Is
the
DelegationTokenManager going to be a
part
of
the
ResourceManager?
Can
you
outline it's lifecycle / how it's going
to
be
integrated
there?
2) Do we really need a YARN / k8s
specific
implementations?
Is
it
possible
to obtain / renew a token in a generic
way?
Maybe
to
rephrase
that,
is
it
possible to implement
DelegationTokenManager
for
the
standalone
Flink?
If
we're able to solve this point, it
could be
possible
to
target
all
deployment scenarios with a single
implementation.
Best,
D.
On Fri, Jan 14, 2022 at 3:47 AM Junfan
Zhang
<
zuston.sha...@gmail.com>
wrote:
Hi G
Thanks for your explain in detail. I
have
gotten
your
thoughts,
and
any
way this proposal
is a great improvement.
Looking forward to your implementation
and
i
will
keep
focus
on
it.
Thanks again.
Best
JunFan.
On Jan 13, 2022, 9:20 PM +0800, Gabor
Somogyi <
gabor.g.somo...@gmail.com
,
wrote:
Just to confirm keeping
"security.kerberos.fetch.delegation-token"
is
added
to the doc.
BR,
G
On Thu, Jan 13, 2022 at 1:34 PM
Gabor
Somogyi <
gabor.g.somo...@gmail.com
wrote:
Hi JunFan,
By the way, maybe this should be
added
in
the
migration
plan
or
intergation section in the
FLIP-211.
Going to add this soon.
Besides, I have a question that
the
KDC
will
collapse
when
the
cluster
reached 200 nodes you described
in the google doc. Do you have any
attachment
or
reference
to
prove
it?
"KDC *may* collapse under some
circumstances"
is
the
proper
wording.
We have several customers who are
executing
workloads
on
Spark/Flink.
Most
of the time I'm facing their
daily issues which is heavily
environment
and
use-case
dependent.
I've
seen various cases:
* where the mentioned ~1k nodes
were
working
fine
* where KDC thought the number of
requests
are
coming
from
DDOS
attack
so
discontinued authentication
* where KDC was simply not
responding
because
of
the
load
* where KDC was intermittently had
some
outage
(this
was
the
most
nasty
thing)
Since you're managing relatively
big
cluster
then
you
know
that
KDC
is
not
only used by Spark/Flink workloads
but the whole company IT
infrastructure
is
bombing
it
so
it
really
depends
on other factors too whether KDC
is
reaching
it's limit or not. Not sure what
kind
of
evidence
are
you
looking
for
but
I'm not authorized to share any
information
about
our clients data.
One thing is for sure. The more
external
system
types
are
used
in
workloads (for ex. HDFS, HBase,
Hive,
Kafka)
which
are authenticating through KDC the
more
possibility
to
reach
this
threshold when the cluster is big
enough.
All in all this feature is here to
help
all
users
never
reach
this
limitation.
BR,
G
On Thu, Jan 13, 2022 at 1:00 PM
张俊帆 <
zuston.sha...@gmail.com
wrote:
Hi G
Thanks for your quick reply. I
think
reserving
the
config
of
*security.kerberos.fetch.delegation-token*
and simplifying disable the
token
fetching
is a
good
idea.By
the
way,
maybe this should be added
in the migration plan or
intergation
section
in
the
FLIP-211.
Besides, I have a question that
the
KDC
will
collapse
when
the
cluster
reached 200 nodes you described
in the google doc. Do you have
any
attachment
or
reference
to
prove
it?
Because in our internal
per-cluster,
the nodes reaches > 1000 and KDC
looks
good.
Do i
missed
or
misunderstood
something? Please correct me.
Best
JunFan.
On Jan 13, 2022, 5:26 PM +0800,
dev@flink.apache.org
,
wrote:
https://docs.google.com/document/d/1JzMbQ1pCJsLVz8yHrCxroYMRP2GwGwvacLrGyaIx5Yc/edit?fbclid=IwAR0vfeJvAbEUSzHQAAJfnWTaX46L6o7LyXhMfBUCcPrNi-uXNgoOaI8PMDQ
