[FLINK-30425][runtime][security] Generalize token receive side #21604

gaborgsomogyi · 2023-01-05T10:17:05Z

What is the purpose of the change

Flink delegation token obtain side is generic already but on the receiver side there is still an assumption that only Hadoop tokens can come. In this PR I've introduced DelegationTokenReceiver interface which can be implemented by every.

Brief change log

Added DelegationTokenReceiver interface
Added HadoopFSDelegationTokenReceiver for Hadoop FS
Added HBaseDelegationTokenReceiver for HBase
Added DelegationTokenReceiverRepository which loads all receivers, and propagates new tokens to them
New/updated automated tests

Verifying this change

Existing/new unit/integration tests + manually on minikube.

Does this pull request potentially affect one of the following parts:

Dependencies (does it add or upgrade a dependency): no
The public API, i.e., is any changed class annotated with @Public(Evolving): no
The serializers: no
The runtime per-record code paths (performance sensitive): no
Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: no
The S3 file system connector: no

Documentation

Does this pull request introduce a new feature? no
If yes, how is the feature documented? not applicable

gaborgsomogyi · 2023-01-05T10:18:05Z

cc @gyfora @mbalassi

gaborgsomogyi · 2023-01-05T10:18:51Z

FYI, the manual test gist updated to use the operator. This way we're not bound to local machine at all.

flinkbot · 2023-01-05T10:20:35Z

CI report:

3f19253 Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

@flinkbot run azure re-run the last Azure build

gaborgsomogyi · 2023-01-05T10:20:50Z

I've tested w/ keytab and TGT cache scenario and both worked. I've added expected exception throwing on the receiver side as well (just like it exists on the provider side) to simulate temporary failures.

mbalassi

Found a couple nits on a cursory look, will do manual testing and a deeper code review tomorrow.

...ime/src/main/java/org/apache/flink/runtime/security/token/DefaultDelegationTokenManager.java

mbalassi · 2023-01-05T11:34:49Z

...ime/src/main/java/org/apache/flink/runtime/security/token/DefaultDelegationTokenManager.java

-                        provider.serviceName(),
-                        e);
+                        provider.serviceName());
+                LOG.debug(e.toString());


Add some explanation to this log message since given the multi-threaded nature of Flink it might not directly follow the previous line.

This part has been changed significantly so I think worth a second look.

gaborgsomogyi · 2023-01-05T12:01:03Z

...ime/src/main/java/org/apache/flink/runtime/security/token/DefaultDelegationTokenManager.java

@@ -120,8 +135,8 @@ private Map<String, DelegationTokenProvider> loadProviders() {
            } catch (Exception | NoClassDefFoundError e) {
                LOG.warn(


@XComp This is the part what I've meant. No more stack trace going to be printed in WARN level. Though we print out the failure for debugging purposes.

Well, this part has been reworked but the main statement is the same. We're not writing out stacktrace unless we stop the workload.

gaborgsomogyi · 2023-01-05T13:06:28Z

...c/main/java/org/apache/flink/runtime/security/token/hadoop/HBaseDelegationTokenProvider.java

+        /**
+         * The general rule how a provider/receiver must behave is the following: The provider and
+         * the receiver must be added to the classpath together with all the additionally required
+         * dependencies.
+         *
+         * <p>This null check is required because the HBase provider is always on classpath but
+         * HBase jars are optional. Such case configuration is not able to be loaded. This construct
+         * is intended to be removed when HBase provider/receiver pair can be externalized (namely
+         * if a provider/receiver throws an exception then workload must be stopped).
+         */
+        if (hbaseConf == null) {
+            LOG.debug(
+                    "HBase is not available (not packaged with this application), hence no "
+                            + "tokens will be acquired.");
+            return false;
+        }


This is temporary solution as mentioned before.

gaborgsomogyi · 2023-01-05T13:06:45Z

...ain/java/org/apache/flink/runtime/security/token/hadoop/HadoopFSDelegationTokenProvider.java

+        /**
+         * The general rule how a provider/receiver must behave is the following: The provider and
+         * the receiver must be added to the classpath together with all the additionally required
+         * dependencies.
+         *
+         * <p>This null check is required because the Hadoop FS provider is always on classpath but
+         * Hadoop FS jars are optional. Such case configuration is not able to be loaded. This
+         * construct is intended to be removed when HBase provider/receiver pair can be externalized
+         * (namely if a provider/receiver throws an exception then workload must be stopped).
+         */
+        if (hadoopConfiguration == null) {
+            LOG.debug(
+                    "Hadoop FS is not available (not packaged with this application), hence no "
+                            + "tokens will be acquired.");
+            return false;
+        }


This is temporary solution as mentioned before.

gaborgsomogyi · 2023-01-05T13:07:59Z

...ime/src/main/java/org/apache/flink/runtime/security/token/DefaultDelegationTokenManager.java

+                // The intentional general rule is that if a provider's init method throws exception
+                // then stop the workload
+                LOG.error(
                        "Failed to initialize delegation token provider {}",
                        provider.serviceName(),
                        e);
+                throw new RuntimeException(e);


I think this will be the most clean statement for later development. The actually existing HBase + HadoopFS providers are exceptional because the're not externalized. The problem w/ them is that the provider/receiver is always on classpath but HadoopFS/HBase dependency is optional.

The temporary solution is to catch no class found exception within the provider and ignore the provider in delegationTokensRequired if conf is null

The long term solution is to externalize both of them and expect provider/receiver + all dependencies together on the classpath

I've added comments in the respective files here and here.

gaborgsomogyi · 2023-01-05T13:12:23Z

...src/main/java/org/apache/flink/runtime/security/token/DelegationTokenReceiverRepository.java

+            try {
+                delegationTokenReceivers.get(serviceName).onNewTokensObtained(tokens);
+            } catch (Exception e) {
+                LOG.warn("Failed to send tokens to delegation token receiver {}", serviceName, e);


We handle receiver failures as temporary issue and we're going forward to increase resiliency. This way maybe not all of the tokens stored but still we're increasing the availability of the overall system.

gaborgsomogyi · 2023-01-05T13:14:31Z

...src/test/java/org/apache/flink/runtime/security/token/DefaultDelegationTokenManagerTest.java

-    @AfterAll
-    public static void afterAll() {
+    @AfterEach
+    public void afterEach() {


This is just a minor improvement which is not related. We need to do reset after each test and not after class.

XComp

I didn't do a proper check since @mbalassi is already helping out and he's probably the better person to judge. I have two comments while flying over some part of the code.

XComp · 2023-01-05T15:42:27Z

...ime/src/main/java/org/apache/flink/runtime/security/token/DefaultDelegationTokenManager.java

@@ -58,6 +59,13 @@
 @Internal
 public class DefaultDelegationTokenManager implements DelegationTokenManager {

+    private static final String PROVIDER_RECEIVER_INCONSISTENCY_ERROR =
+            " There is an inconsistency between loaded delegation token providers and receivers. "


Suggested change

" There is an inconsistency between loaded delegation token providers and receivers. "

"There is an inconsistency between loaded delegation token providers and receivers. "

XComp · 2023-01-05T16:01:51Z

...ime/src/main/java/org/apache/flink/runtime/security/token/DefaultDelegationTokenManager.java

                        "Failed to initialize delegation token provider {}",
                        provider.serviceName(),
                        e);
+                throw new RuntimeException(e);


Shouldn't we do either logging or throwing the exception and if we decide to throw the exception, adding the log message as the error message might be useful. 🤔 FlinkRuntimeException might be useful since it indicates that the exception was actually initiated by Flink code. ...just as an idea.

Makes sense to use FlinkRuntimeException, changed on both places.

gaborgsomogyi · 2023-01-05T16:23:12Z

Waiting w/ the next push to see a full run whether there are test failures...

pvary · 2023-01-06T12:12:36Z

Do we have test cases for the specific providers/receivers, like HBase/HadoopFS?

gaborgsomogyi · 2023-01-06T12:44:29Z

There is an end-to-end test for HDFS but for HBase we don't have anything. Honestly it's horror complex to setup everything within a single process, like KDC, HDFS, HBase, Flink app. It's definitely doable but I personally think that it must be included within the connector when externalized.

pvary · 2023-01-06T12:50:23Z

.../main/java/org/apache/flink/runtime/security/token/hadoop/HadoopDelegationTokenReceiver.java


 import org.apache.hadoop.security.Credentials;
 import org.apache.hadoop.security.UserGroupInformation;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;

-/** Delegation token updater functionality. */
+/** Hadoop delegation token receiver base class. */
 @Internal


Could I use this class for the HMSDelegationTokenReceiver implementation?

You could but since it's @Internal it's subject to change.

mbalassi

Tested the changes locally on minikube following @gaborgsomogyi's instructions. Looks good, will wait for the CI to complete after the recent rebase to master.

gaborgsomogyi · 2023-01-12T23:06:51Z

Thanks, seems like nowadays we have less and less flaky tests and passed🙂

gaborgsomogyi force-pushed the FLINK-30425 branch from de99236 to 051bf28 Compare January 5, 2023 11:45

mbalassi reviewed Jan 5, 2023

View reviewed changes

flinkbot added the component=<none> label Jan 5, 2023

gaborgsomogyi commented Jan 5, 2023

View reviewed changes

gaborgsomogyi force-pushed the FLINK-30425 branch 2 times, most recently from f95df6c to 328e2f6 Compare January 5, 2023 13:00

gaborgsomogyi commented Jan 5, 2023

View reviewed changes

gaborgsomogyi force-pushed the FLINK-30425 branch 2 times, most recently from 5b8cbcb to 1e225ce Compare January 5, 2023 15:20

XComp reviewed Jan 5, 2023

View reviewed changes

gaborgsomogyi force-pushed the FLINK-30425 branch 3 times, most recently from d98b162 to 4fce8e2 Compare January 6, 2023 11:26

pvary reviewed Jan 6, 2023

View reviewed changes

gaborgsomogyi force-pushed the FLINK-30425 branch from 4fce8e2 to d74db13 Compare January 10, 2023 10:17

[FLINK-30425][runtime][security] Generalize token receive side

3f19253

gaborgsomogyi force-pushed the FLINK-30425 branch from d74db13 to 3f19253 Compare January 12, 2023 14:42

mbalassi self-requested a review January 12, 2023 15:41

mbalassi approved these changes Jan 12, 2023

View reviewed changes

mbalassi merged commit 324b54c into apache:master Jan 13, 2023

gaborgsomogyi deleted the FLINK-30425 branch September 13, 2023 08:37

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[FLINK-30425][runtime][security] Generalize token receive side #21604

[FLINK-30425][runtime][security] Generalize token receive side #21604

gaborgsomogyi commented Jan 5, 2023

gaborgsomogyi commented Jan 5, 2023

gaborgsomogyi commented Jan 5, 2023

flinkbot commented Jan 5, 2023 •

edited

gaborgsomogyi commented Jan 5, 2023

mbalassi left a comment

mbalassi Jan 5, 2023

gaborgsomogyi Jan 5, 2023 •

edited

gaborgsomogyi Jan 5, 2023

gaborgsomogyi Jan 5, 2023

gaborgsomogyi Jan 5, 2023

gaborgsomogyi Jan 5, 2023

gaborgsomogyi Jan 5, 2023 •

edited

gaborgsomogyi Jan 5, 2023

gaborgsomogyi Jan 5, 2023

XComp left a comment

XComp Jan 5, 2023

gaborgsomogyi Jan 5, 2023

XComp Jan 5, 2023

gaborgsomogyi Jan 5, 2023

gaborgsomogyi commented Jan 5, 2023

pvary commented Jan 6, 2023

gaborgsomogyi commented Jan 6, 2023

pvary Jan 6, 2023

gaborgsomogyi Jan 6, 2023

mbalassi left a comment

gaborgsomogyi commented Jan 12, 2023

		@@ -120,8 +135,8 @@ private Map<String, DelegationTokenProvider> loadProviders() {
		} catch (Exception \| NoClassDefFoundError e) {
		LOG.warn(

	" There is an inconsistency between loaded delegation token providers and receivers. "
	"There is an inconsistency between loaded delegation token providers and receivers. "

[FLINK-30425][runtime][security] Generalize token receive side #21604

[FLINK-30425][runtime][security] Generalize token receive side #21604

Conversation

gaborgsomogyi commented Jan 5, 2023

What is the purpose of the change

Brief change log

Verifying this change

Does this pull request potentially affect one of the following parts:

Documentation

gaborgsomogyi commented Jan 5, 2023

gaborgsomogyi commented Jan 5, 2023

flinkbot commented Jan 5, 2023 • edited

CI report:

gaborgsomogyi commented Jan 5, 2023

mbalassi left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

gaborgsomogyi Jan 5, 2023 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

gaborgsomogyi Jan 5, 2023 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

XComp left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

gaborgsomogyi commented Jan 5, 2023

pvary commented Jan 6, 2023

gaborgsomogyi commented Jan 6, 2023

Choose a reason for hiding this comment

Choose a reason for hiding this comment

mbalassi left a comment

Choose a reason for hiding this comment

gaborgsomogyi commented Jan 12, 2023

flinkbot commented Jan 5, 2023 •

edited

gaborgsomogyi Jan 5, 2023 •

edited

gaborgsomogyi Jan 5, 2023 •

edited