-
Notifications
You must be signed in to change notification settings - Fork 17
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
feat: Update operator to use cached resources
* Update operator to use cached resources rather than calling Kubernetes directly. * Move code to construct secret data into a separate class to make it easier to switch to DependentResource type later. Signed-off-by: Katherine Stanley <[email protected]>
- Loading branch information
Showing
5 changed files
with
201 additions
and
113 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
119 changes: 119 additions & 0 deletions
119
src/main/java/io/strimzi/kafka/access/SecretDependentResource.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,119 @@ | ||
| /* | ||
| * Copyright Strimzi authors. | ||
| * License: Apache License 2.0 (see the file LICENSE or http://apache.org/licenses/LICENSE-2.0.html). | ||
| */ | ||
| package io.strimzi.kafka.access; | ||
|
|
||
| import io.fabric8.kubernetes.api.model.Secret; | ||
| import io.javaoperatorsdk.operator.api.reconciler.Context; | ||
| import io.javaoperatorsdk.operator.processing.event.ResourceID; | ||
| import io.javaoperatorsdk.operator.processing.event.source.informer.InformerEventSource; | ||
| import io.strimzi.api.kafka.model.Kafka; | ||
| import io.strimzi.api.kafka.model.KafkaUser; | ||
| import io.strimzi.api.kafka.model.KafkaUserAuthentication; | ||
| import io.strimzi.api.kafka.model.KafkaUserSpec; | ||
| import io.strimzi.api.kafka.model.status.KafkaUserStatus; | ||
| import io.strimzi.kafka.access.internal.CustomResourceParseException; | ||
| import io.strimzi.kafka.access.internal.KafkaListener; | ||
| import io.strimzi.kafka.access.internal.KafkaParser; | ||
| import io.strimzi.kafka.access.internal.KafkaUserData; | ||
| import io.strimzi.kafka.access.model.KafkaAccess; | ||
| import io.strimzi.kafka.access.model.KafkaAccessSpec; | ||
| import io.strimzi.kafka.access.model.KafkaReference; | ||
| import io.strimzi.kafka.access.model.KafkaUserReference; | ||
|
|
||
| import java.nio.charset.StandardCharsets; | ||
| import java.util.Base64; | ||
| import java.util.HashMap; | ||
| import java.util.Map; | ||
| import java.util.Optional; | ||
|
|
||
| /** | ||
| * Class to represent the data in the Secret created by the operator | ||
| * In future updates this could be updated to implement the | ||
| * Java Operator SDK DependentResource class | ||
| */ | ||
| public class SecretDependentResource { | ||
|
|
||
| private static final String TYPE_SECRET_KEY = "type"; | ||
| private static final String TYPE_SECRET_VALUE = "kafka"; | ||
| private static final String PROVIDER_SECRET_KEY = "provider"; | ||
| private static final String PROVIDER_SECRET_VALUE = "strimzi"; | ||
| private final Map<String, String> commonSecretData = new HashMap<>(); | ||
|
|
||
| /** | ||
| * Default constructor that initialises the common secret data | ||
| */ | ||
| public SecretDependentResource() { | ||
| final Base64.Encoder encoder = Base64.getEncoder(); | ||
| commonSecretData.put(TYPE_SECRET_KEY, encoder.encodeToString(TYPE_SECRET_VALUE.getBytes(StandardCharsets.UTF_8))); | ||
| commonSecretData.put(PROVIDER_SECRET_KEY, encoder.encodeToString(PROVIDER_SECRET_VALUE.getBytes(StandardCharsets.UTF_8))); | ||
| } | ||
|
|
||
| /** | ||
| * The desired state of the data in the secret | ||
| * @param spec The spec of the KafkaAccess resource being reconciled | ||
| * @param namespace The namespace of the KafkaAccess resource being reconciled | ||
| * @param context The event source context | ||
| * @return The data for the Secret as a Map | ||
| */ | ||
| public Map<String, String> desired(final KafkaAccessSpec spec, final String namespace, final Context<KafkaAccess> context) { | ||
| final KafkaReference kafkaReference = spec.getKafka(); | ||
| final String kafkaClusterName = kafkaReference.getName(); | ||
| final String kafkaClusterNamespace = Optional.ofNullable(kafkaReference.getNamespace()).orElse(namespace); | ||
| final InformerEventSource<Kafka, KafkaAccess> kafkaEventSource = (InformerEventSource<Kafka, KafkaAccess>) context.eventSourceRetriever().getResourceEventSourceFor(Kafka.class); | ||
| final Kafka kafka = kafkaEventSource.get(new ResourceID(kafkaClusterName, kafkaClusterNamespace)) | ||
| .orElseThrow(() -> new IllegalStateException(String.format("Kafka %s/%s missing", kafkaClusterNamespace, kafkaClusterName))); | ||
| final Optional<KafkaUserReference> kafkaUserReference = Optional.ofNullable(spec.getUser()); | ||
| final Map<String, String> data = new HashMap<>(commonSecretData); | ||
| if (kafkaUserReference.isPresent()) { | ||
| if (!KafkaUser.RESOURCE_KIND.equals(kafkaUserReference.get().getKind()) || !KafkaUser.RESOURCE_GROUP.equals(kafkaUserReference.get().getApiGroup())) { | ||
| throw new CustomResourceParseException(String.format("User kind must be %s and apiGroup must be %s", KafkaUser.RESOURCE_KIND, KafkaUser.RESOURCE_GROUP)); | ||
| } | ||
| final String kafkaUserName = kafkaUserReference.get().getName(); | ||
| final String kafkaUserNamespace = Optional.ofNullable(kafkaUserReference.get().getNamespace()).orElse(namespace); | ||
| final KafkaUser kafkaUser = context.getSecondaryResource(KafkaUser.class) | ||
| .orElseThrow(() -> new IllegalStateException(String.format("KafkaUser %s/%s missing", kafkaUserNamespace, kafkaUserName))); | ||
| final String userSecretName = Optional.ofNullable(kafkaUser.getStatus()) | ||
| .map(KafkaUserStatus::getSecret) | ||
| .orElseThrow(() -> new IllegalStateException(String.format("Secret for KafkaUser %s/%s missing", kafkaUserNamespace, kafkaUserName))); | ||
| final InformerEventSource<Secret, KafkaAccess> kafkaUserSecretEventSource = (InformerEventSource<Secret, KafkaAccess>) context.eventSourceRetriever() | ||
| .getResourceEventSourceFor(Secret.class, KafkaAccessReconciler.KAFKA_USER_SECRET_EVENT_SOURCE); | ||
| final Secret kafkaUserSecret = kafkaUserSecretEventSource.get(new ResourceID(userSecretName, kafkaUserNamespace)) | ||
| .orElseThrow(() -> new IllegalStateException(String.format("Secret for KafkaUser %s/%s missing", kafkaUserNamespace, kafkaUserName))); | ||
| data.putAll(secretDataWithUser(spec, kafka, kafkaUser, new KafkaUserData(kafkaUser).withSecret(kafkaUserSecret))); | ||
| } else { | ||
| data.putAll(secretData(spec, kafka)); | ||
| } | ||
| return data; | ||
| } | ||
|
|
||
| protected Map<String, String> secretData(final KafkaAccessSpec spec, final Kafka kafka) { | ||
| final Map<String, String> data = new HashMap<>(commonSecretData); | ||
| final KafkaListener listener; | ||
| try { | ||
| listener = KafkaParser.getKafkaListener(kafka, spec); | ||
| } catch (CustomResourceParseException e) { | ||
| throw new IllegalStateException("Reconcile failed due to ParserException " + e.getMessage()); | ||
| } | ||
| data.putAll(listener.getConnectionSecretData()); | ||
| return data; | ||
| } | ||
|
|
||
| protected Map<String, String> secretDataWithUser(final KafkaAccessSpec spec, final Kafka kafka, final KafkaUser kafkaUser, final KafkaUserData kafkaUserData) { | ||
| final Map<String, String> data = new HashMap<>(commonSecretData); | ||
| final KafkaListener listener; | ||
| try { | ||
| final String kafkaUserType = Optional.ofNullable(kafkaUser.getSpec()) | ||
| .map(KafkaUserSpec::getAuthentication) | ||
| .map(KafkaUserAuthentication::getType) | ||
| .orElse(KafkaParser.USER_AUTH_UNDEFINED); | ||
| listener = KafkaParser.getKafkaListener(kafka, spec, kafkaUserType); | ||
| data.putAll(kafkaUserData.getConnectionSecretData()); | ||
| } catch (CustomResourceParseException e) { | ||
| throw new IllegalStateException("Reconcile failed due to ParserException " + e.getMessage()); | ||
| } | ||
| data.putAll(listener.getConnectionSecretData()); | ||
| return data; | ||
| } | ||
| } |
Oops, something went wrong.