KAFKA-6886 Externalize secrets from Connect configs by rayokota · Pull Request #5068 · apache/kafka

rayokota · 2018-05-23T00:06:50Z

This commit allows secrets in Connect configs to be externalized and replaced with variable references of the form ${provider:[path:]key}, where the "path" is optional.

There are 2 main additions to org.apache.kafka.common.config: a ConfigProvider and a ConfigTransformer. The ConfigProvider is an interface that allows key-value pairs to be provided by an external source for a given "path". An a TTL can be associated with the key-value pairs returned from the path. The ConfigTransformer will use instances of ConfigProvider to replace variable references in a set of configuration values.

In the Connect framework, ConfigProvider classes can be specified in the worker config, and then variable references can be used in the connector config. In addition, the herder can be configured to restart connectors (or not) based on the TTL returned from a ConfigProvider. The main class that performs restarts and transformations is WorkerConfigTransformer.

Finally, a configs() method has been added to both SourceTaskContext and SinkTaskContext. This allows connectors to get configs with variables replaced by the latest values from instances of ConfigProvider.

Most of the other changes in the Connect framework are threading various objects through classes to enable the above functionality.

Committer Checklist (excluded from commit message)

Verify design and implementation
Verify test coverage and CI build status
Verify documentation (including upgrade notes)

rayokota · 2018-05-24T20:10:25Z

retest this please

rhauch

Looks really good, @rayokota. I did a pretty detailed pass, and have a number of comments.

rhauch · 2018-05-24T18:20:39Z

+     * @param path the path where the data resides
+     * @param keys the keys whose values will be retrieved
+     */
+    void unsubscribe(String path, Set<String> keys);


Only one subscriber is expected? If not, does this unsubscribe all subscriptions for the keys at the path?

How about a default implementation?

I'll add the callback to this method.

rhauch · 2018-05-24T18:22:10Z

+     *
+     * @param configProviders the set of {@link ConfigProvider} instances.
+     */
+    public ConfigTransformer(Map<String, ConfigProvider> configProviders) {


How about providing a builder API to easily create these with individual named providers? Not sure how valuable it will be.

rhauch · 2018-05-24T18:26:36Z

+            ConfigProvider provider = configProviders.get(providerName);
+            Map<String, Set<String>> keysByPath = entry.getValue();
+            if (provider != null && keysByPath != null) {
+                for (Map.Entry<String, Set<String>> pathWithKeys : keysByPath.entrySet()) {


Perhaps?

keysByPath.entrySet().forEach(entry-> {

rhauch · 2018-05-24T18:40:39Z

+                Set<String> keys = keysByPath.computeIfAbsent(var.path, k -> new HashSet<>());
+                keys.add(var.variable);
+            }
+        }


Using Java 8 streams would be a bit more compact and IMO more readable:

configs.forEach((key, value)->{ getVars(key, value, DEFAULT_PATTERN).forEach(var-> { keysByProvider.computeIfAbsent(var.providerName, k -> new HashMap<>()) .computeIfAbsent(var.path, k-> new HashSet<>()) .add(var.variable); }); });

Oh, I'm not a big fan of nested foreach: https://blog.jooq.org/2015/12/08/3-reasons-why-you-shouldnt-replace-your-for-loops-by-stream-foreach/

rhauch · 2018-05-24T18:41:51Z

+                    keyValuesByPath.put(path, data);
+                }
+            }
+        }


Again, this could be simplified quite a bit with streams.

lol, i expect this to be a ripe area for bikeshedding. you can definitely reduce some verbosity with streams, unclear that it's actually simplified. for the time being, i think we should be liberal in accepting either form of style and we can figure out where we want to land over time.

rhauch · 2018-05-24T23:01:52Z

+                if (keys.contains("testKey")) {
+                    data.put("testKey", "testResult");
+                }
+                if (keys.contains("testKeyWithTTL")) {


Might be good to use constants for these keys and values. That would be a bit more maintainable, as long as it doesn't make it unreadable.

rhauch · 2018-05-24T23:10:36Z

+     * @param keys the keys whose values will be retrieved
+     * @param callback the callback to invoke upon change
+     */
+    void subscribe(String path, Set<String> keys, ConfigChangeCallback callback);


How about a default implementation?

rhauch · 2018-05-24T23:14:33Z

 public interface SinkTaskContext {
+
+    /**
+     * Get the Task configuration.


Need more JavaDoc to define the behavior. Does this return null? Will the result ever differ if called multiple times? How does this relate to the configuration passed to the connector and/or task upon startup?

rhauch · 2018-05-24T23:14:41Z

 */
 public interface SourceTaskContext {
+    /**
+     * Get the Task configuration.


rhauch · 2018-05-24T23:27:07Z

                getPluginDesc(reflections, HeaderConverter.class, loader),
-                getPluginDesc(reflections, Transformation.class, loader)
+                getPluginDesc(reflections, Transformation.class, loader),
+                getPluginDesc(reflections, ConfigProvider.class, loader)


We don't want to scan for this, but want to use the Service Loader mechanism. Do you plan to change this when the REST extension PR is merged?

Yes, that will be for a future PR

rhauch

A few more comments.

rhauch · 2018-05-25T14:15:54Z


+    public static final String CONFIG_RELOAD_ACTION_CONFIG = "config.action.reload";
+    private static final String CONFIG_RELOAD_ACTION_DOC =
+            "The action to take in order to reload changed configuration values (none or restart).";


This should probably make more clear that this occurs when configuration values change in external providers, and should describe what none and restart mean. Users shouldn't have to go to the code or KIP to understand this configuration property.

rhauch · 2018-05-25T14:18:33Z

+ */
+package org.apache.kafka.connect.runtime;
+
+public interface HerderRequestId {


This seems like an odd name for the response from a herder request: there's no "id" associated with this.

I renamed to just HerderRequest

rhauch · 2018-05-25T14:21:33Z

            + "/opt/connectors";

+    public static final String CONFIG_PROVIDERS_CONFIG = "config.providers";
+    protected static final String CONFIG_PROVIDERS_DOC = "List of configuration providers.";


Please be more specific. "List of configuration providers. This is a comma-separated list of the fully-qualified names of the ConfigProvider implementations, in the order they will be created, configured, and used."

rhauch · 2018-05-25T14:25:48Z

    }

+    @Override
+    public String getConnectorConfigReloadAction(final String connName) {


Should we have an enum for reload action?

rayokota · 2018-05-25T17:55:34Z

@rhauch , thanks for the review! I've incorporated most of your suggestions (except the use of foreach :)). For the subscribe and unsubscribe, those are optional methods for ConfigProvider subclasses that want to maintain a separate thread to poll the external source, and then push out changes. I've copied how optional operations are documented in java.util.List by adding JavaDoc that they are optional and could possible throw UnsupportedOperationException. I also gave ConfigProvider default implementations that throw UnsupportedOperationException. Thanks!

rhauch

Ok, one more round of mostly questions and a few minor suggestions.

rhauch · 2018-05-29T19:47:17Z

+        }
+
+        public void subscribe(String path, Set<String> keys, ConfigChangeCallback callback) {
+            throw new UnsupportedOperationException();


Isn't this the default implementation?

rhauch · 2018-05-29T19:47:23Z

+        }
+
+        public void unsubscribe(String path, Set<String> keys) {
+            throw new UnsupportedOperationException();


Isn't this the default implementation?

rhauch · 2018-05-29T19:58:10Z

+ * </pre>
+ * resides at the path "/tmp/properties.txt", then when a configuration Map which has an entry with a key "someKey" and
+ * a value "${file:/tmp/properties.txt:fileKey}" is passed to the {@link #transform(Map)} method, then the transformed
+ * Map will have an entry with key "someKey" and a value "someValue".


Might be useful to mention that this only requires the get methods on ConfigProvider, and does not use subscribe.

rhauch · 2018-05-29T20:02:32Z

+
+    /**
+     * Returns the transformed data, with variables replaced with corresponding values from the
+     * ConfigProvider instances if found.


Would it be useful to mention that changes to this map will not affect the origin of the configuration or the information in the ConfigProvider(s)?

rhauch · 2018-05-29T20:03:40Z

+import java.util.Set;
+
+/**
+ * An implementation of {@link ConfigProvider} that represents a Properties file.


I think this should mention that all information stored in the file are stored in cleartext, so care must be used if any secrets are stored in the backing file.

Let's take this to a follow up that clarifies the details of this. We should probably be very verbose about this being an example and you want to be careful about how you use it if you choose too. It's a valid choice in quite a few contexts, but some warnings are probably in order.

@rayokota Maybe file a follow up JIRA so we don't lose track of it? Should be a pretty minor change.

rhauch · 2018-05-29T20:10:38Z

+            "The action to take in order to reload changed configuration values (none or restart). "
+                    + "Configuration values will typically only change if using external configuration providers. "
+                    + "A value of 'none' indicates that the new values will not be reloaded. "
+                    + "A value of 'reload' indicates that the new values will be reloaded by restarting the connector. "


I'm not convinced that a user will know what this means. How about making it much more specific and concrete:

The action that Connect should take on the connector when changes in external configuration providers results in a change in the connector's configuration properties. A value of `none` indicates that Connect will do nothing. A value of 'reload' indicates that Connect should restart/reload the connector with the updated configuration properties....

Also, is this restart or reload? The enum is RESTART.

Also, is this case sensitive?

rhauch · 2018-05-29T20:20:59Z

+        if (configTransformer != null) {
+            configs = configTransformer.transform(connector, configs);
+        }
+        return configs;


The JavaDoc for this method should probably say that it includes the current transformed connector configuration where all variables have been replaced with the current values from the external ConfigProviders, and that they may include secrets.

rhauch · 2018-05-29T20:21:04Z

+        if (configTransformer != null) {
+            configs = configTransformer.transform(task.connector(), configs);
+        }
+        return configs;


The JavaDoc for this method should probably say that it includes the current transformed connector configuration where all variables have been replaced with the current values from the external ConfigProviders, and that they may include secrets.

rhauch · 2018-05-29T20:58:56Z

-                    + "indicates that a configuration value will expire in the future.";
+            "The action that Connect should take on the connector when changes in external " +
+            "configuration providers result in a change in the connector's configuration properties. " +
+            "A value of `none` indicates that Connect will do nothing. " +


The backtick was from my original comment. 😊 Should change that to single quote like on the next line.

rhauch

One more minor thing: some JavaDoc lines were accidentally removed.

rhauch · 2018-05-29T21:00:15Z

+     * Get the configuration for a connector.  The configuration will have been transformed by
+     * {@link org.apache.kafka.common.config.ConfigTransformer} by having all variable
+     * references replaced with the current values from external instances of
+     * {@link org.apache.kafka.common.config.ConfigProvider}, and may include secrets.


Now missing @param and @return.

rhauch

Excellent work, @rayokota! Thanks for your patience with the reviews. I'm very much looking forward to this new capability in Connect.

rayokota · 2018-05-29T21:06:40Z

Thanks, @rhauch , appreciate your attention to detail :)

ewencp

@rayokota A small pile of minor comments, but this mostly looks good. The one issue that would be most important to fix before merging is the one about awaitTermination.

ewencp · 2018-05-29T17:41:09Z

+     * @param data a Map of key-value pairs
+     */
+    public ConfigData(Map<String, String> data) {
+        this(data, Long.MAX_VALUE);


Do we actually want this to be Long.MAX_VALUE? Seems error prone since the normal thing to do with the TTL is add it to the current time and that will cause overflow. Should we have some sentinel value for infinite TTL?

ewencp · 2018-05-29T18:07:30Z

+                    ConfigData configData = provider.get(path, keys);
+                    Map<String, String> data = configData.data();
+                    long ttl = configData.ttl();
+                    if (ttl >= 0 && ttl < Long.MAX_VALUE) {


I think we should have a constant in the ConfigData class for whatever value ends up being the sentinel, even if it ends up being Long.MAX_VALUE so that uses of the constant outside that class are clearer as to their meaning. Also, another alternative would be to track the TTL as a Long instead of a long, which is less user friendly but doesn't require a special sentinel value.

Thanks, I'll change to Long

ewencp · 2018-05-29T21:54:28Z

                          StatusBackingStore statusBackingStore,
                          ConfigBackingStore configBackingStore) {
        this.worker = worker;
+        this.worker.herder = this;


We can probably clean up or refactor later, but this doesn't seem great -- code like this indicates overly strong coupling between these two classes and makes ownership semantics unclear.

ewencp · 2018-05-29T21:55:59Z

+     * Get the configuration reload action.
+     * @param connName name of the connector
+     */
+    ConfigReloadAction getConnectorConfigReloadAction(final String connName);


nit: Kafka generally avoids using get/set at the beginning of accessors.

ewencp · 2018-05-29T22:24:02Z

            KafkaProducer<byte[], byte[]> producer = new KafkaProducer<>(producerProps);
            return new WorkerSourceTask(id, (SourceTask) task, statusListener, initialState, keyConverter, valueConverter,
-                    headerConverter, transformationChain, producer, offsetReader, offsetWriter, config, metrics, loader, time);
+                    headerConverter, transformationChain, producer, offsetReader, offsetWriter, config, configState, metrics, loader, time);


You might want to put some comments in the context classes about why we need the entire config. At first glance it appears possibly problematic because we're saving the entire snapshot state for the cluster, but really only care about a single config. It's possible for the state to become outdated (e.g. config change that doesn't require a rebalance) and if we get incremental rebalancing in at some point then it even becomes fairly likely that the majority of the state is stale.

I didn't figure out what was going on until I read through the rest of the PR and found that the ClusterConfigState was also now doing the config translation, so I think it's probably not obvious and some comments would help remind people (and avoid, e.g., incorrect refactoring "cleanup" that appears to simplify things but actually breaks them).

ewencp · 2018-05-29T22:46:43Z

    }

+    @Override
+    public ConfigReloadAction getConnectorConfigReloadAction(final String connName) {


nit: this is identical between both herders, can we refactor configState so we can just push this into the abstract class?

I'll make a note to clean this later. Probably a lot of methods that use configState can be pulled up into AbstractHerder

ewencp · 2018-05-29T23:08:06Z

+
+        // Configure the ConfigProvider
+        String configPrefix = providerPrefix + ".param.";
+        Map<String, Object> converterConfig = config.originalsWithPrefix(configPrefix);


nit: probably just copy/paste issue, but it's a configProviderConfig, not converterConfig

ewencp · 2018-05-29T23:11:01Z

    @Override
    public synchronized void stop() {
        log.info("Herder stopping");
+        requestExecutorService.shutdown();


aren't we missing a corresponding awaitTermination?

I've added an awaitTermination of 30 secs

rayokota · 2018-05-30T00:04:23Z

Thanks for the review @ewencp ! I've pushed another commit with changes based on your suggestions.

ewencp · 2018-05-30T02:41:48Z

seems like unrelated core kafka tests failing, but there's a lot of them, so retest this please. likely a fluke, but let's be sure.

ewencp

had a couple more comments, but we can follow up on them separately. this LGTM in current state. going to wait on tests to come back clean, but once that happens I'll merge to trunk for 2.0.0

ewencp · 2018-05-30T03:20:45Z

+                    keyValuesByPath.put(path, data);
+                }
+            }
+        }


lol, i expect this to be a ripe area for bikeshedding. you can definitely reduce some verbosity with streams, unclear that it's actually simplified. for the time being, i think we should be liberal in accepting either form of style and we can figure out where we want to land over time.

ewencp · 2018-05-30T03:26:09Z

+        Map<String, String> data = new HashMap<>(configs);
+        for (Map.Entry<String, String> config : configs.entrySet()) {
+            data.put(config.getKey(), replace(lookupsByProvider, config.getValue(), DEFAULT_PATTERN));
+        }


This is actually a really good example of a case where your better version is actually worse for a lot of people -- replaceAll doesn't necessarily have obvious semantics. The forEach version is at least obvious in what it is doing, but I have to read the code in the replaceAll version to figure out that it is only replacing the values and the keys remain the same (and that may have required comparing different versions of the code, I'm not even sure it would have been obvious to me without the earlier versions). It takes both key and value, but I have to carefully inspect what it is returning to understand that it's only updating the value.

Not going to get stuck in a language features discussion for this PR, but you have to be a Java person to be able to read your last version and very quickly understand what's happening. Both the original version and your first alternative actually seem way better to me because coming from any background I can read them and understand what the semantics are.

ewencp · 2018-05-30T03:31:41Z

+import java.util.Set;
+
+/**
+ * An implementation of {@link ConfigProvider} that represents a Properties file.


Let's take this to a follow up that clarifies the details of this. We should probably be very verbose about this being an example and you want to be careful about how you use it if you choose too. It's a valid choice in quite a few contexts, but some warnings are probably in order.

@rayokota Maybe file a follow up JIRA so we don't lose track of it? Should be a pretty minor change.

…connect-secrets

ewencp · 2018-05-30T21:42:02Z

Merging to trunk. The failed test is a flaky core test, the connect stuff is all passing fine.

This commit allows secrets in Connect configs to be externalized and replaced with variable references of the form `${provider:[path:]key}`, where the "path" is optional. There are 2 main additions to `org.apache.kafka.common.config`: a `ConfigProvider` and a `ConfigTransformer`. The `ConfigProvider` is an interface that allows key-value pairs to be provided by an external source for a given "path". An a TTL can be associated with the key-value pairs returned from the path. The `ConfigTransformer` will use instances of `ConfigProvider` to replace variable references in a set of configuration values. In the Connect framework, `ConfigProvider` classes can be specified in the worker config, and then variable references can be used in the connector config. In addition, the herder can be configured to restart connectors (or not) based on the TTL returned from a `ConfigProvider`. The main class that performs restarts and transformations is `WorkerConfigTransformer`. Finally, a `configs()` method has been added to both `SourceTaskContext` and `SinkTaskContext`. This allows connectors to get configs with variables replaced by the latest values from instances of `ConfigProvider`. Most of the other changes in the Connect framework are threading various objects through classes to enable the above functionality. Author: Robert Yokota <rayokota@gmail.com> Author: Ewen Cheslack-Postava <me@ewencp.org> Reviewers: Randall Hauch <rhauch@gmail.com>, Ewen Cheslack-Postava <ewen@confluent.io> Closes apache#5068 from rayokota/KAFKA-6886-connect-secrets

aakashgupta96 · 2020-08-30T15:25:45Z

Hi @rayokota
Was this feature merged? I was also planning to submit a KIP around same then I saw your proposal.
Can you please update whether it is merged or are you still working on this?

If not, I can plan to submit a proposal for same.

rayokota · 2020-08-30T16:49:48Z

@aakashgupta96 yes it was merged

rayokota force-pushed the KAFKA-6886-connect-secrets branch from f62199a to aa3eb04 Compare May 23, 2018 00:14

Initial impl for externalizing secrets in configs

15d2532

rayokota force-pushed the KAFKA-6886-connect-secrets branch from be52a09 to 15d2532 Compare May 23, 2018 05:02

rhauch reviewed May 24, 2018

View reviewed changes

rhauch reviewed May 25, 2018

View reviewed changes

Incorporate feedback from code review

3ea8af8

Clean up from refactor

fd49640

rhauch reviewed May 29, 2018

View reviewed changes

Incorporate feedback from code review

98b6df7

rhauch reviewed May 29, 2018

View reviewed changes

rayokota added 2 commits May 29, 2018 14:03

Minor JavaDoc fixes

4b3273b

Fix minor JavaDoc issues

52264ee

rhauch approved these changes May 29, 2018

View reviewed changes

ewencp reviewed May 29, 2018

View reviewed changes

Incorporate feedback from Ewen's code review

6af51f0

rayokota added 2 commits May 29, 2018 17:14

Fix broken transformer test

7e004ba

Mention TTL can be null for ConfigData in JavaDoc

998b1f9

ewencp approved these changes May 30, 2018

View reviewed changes

ewencp and others added 3 commits May 29, 2018 21:49

Merge remote-tracking branch 'origin/trunk' into rayokota-KAFKA-6886-…

e5aa0cd

…connect-secrets

Merge branch 'trunk' into KAFKA-6886-connect-secrets

e6d72b8

Fix tests from error-handling PR

acee011

ewencp closed this in 08e8fac May 30, 2018

kkonstantine added the connect label Oct 16, 2020

Conversation

rayokota commented May 23, 2018 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Committer Checklist (excluded from commit message)

Uh oh!

rayokota commented May 24, 2018

Uh oh!

rhauch left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

rhauch left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

rayokota commented May 25, 2018

Uh oh!

rhauch left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

rayokota commented May 23, 2018 •

edited

Loading