Initial KRaft support #1023
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
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
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 |
|---|---|---|
|
|
@@ -19,7 +19,6 @@ import ( | |
| "strings" | ||
|
|
||
| "emperror.dev/errors" | ||
| "github.com/imdario/mergo" | ||
|
|
||
| "github.com/banzaicloud/istio-client-go/pkg/networking/v1beta1" | ||
|
|
@@ -62,19 +61,34 @@ const ( | |
|
|
||
| // DefaultKafkaImage is the default Kafka image used when users don't specify it in KafkaClusterSpec.ClusterImage | ||
| DefaultKafkaImage = "ghcr.io/banzaicloud/kafka:2.13-3.4.1" | ||
|
|
||
| // controllerNodeProcessRole represents the node is a controller node | ||
| controllerNodeProcessRole = "controller" | ||
|
There was a problem hiding this comment. Suggestion: |
||
| // brokerNodeProcessRole represents the node is a broker node | ||
| brokerNodeProcessRole = "broker" | ||
| ) | ||
|
|
||
| // KafkaClusterSpec defines the desired state of KafkaCluster | ||
| type KafkaClusterSpec struct { | ||
| // kRaft is used to decide where the Kafka cluster is under KRaft mode or ZooKeeper mode. | ||
| // This is default to be true; if set to false, the Kafka cluster is in ZooKeeper mode. | ||
| // +kubebuilder:default=true | ||
| // +optional | ||
| KRaftMode bool `json:"kRaft"` | ||
|
There was a problem hiding this comment. I think this should default to false. There was a problem hiding this comment. I initially had this default to false as well, but then I realized the overall product strategy of the next release of Calisti, where we seem to want to default everything to KRaft as well. Hence I made the change to default the This is a bigger discussion than the flag itself, let's also make the team aware of this |
||
| HeadlessServiceEnabled bool `json:"headlessServiceEnabled"` | ||
| ListenersConfig ListenersConfig `json:"listenersConfig"` | ||
| // Custom ports to expose in the container. Example use case: a custom kafka distribution, that includes an integrated metrics api endpoint | ||
| AdditionalPorts []corev1.ContainerPort `json:"additionalPorts,omitempty"` | ||
| // ZKAddresses specifies the ZooKeeper connection string | ||
| // in the form hostname:port where host and port are the host and port of a ZooKeeper server. | ||
| // Under ZooKeeper mode, this is a must-have configuration. | ||
| // And if set under KRaft mode, Koperator ignores this configuration. | ||
| // +optional | ||
| ZKAddresses []string `json:"zkAddresses,omitempty"` | ||
| // ZKPath specifies the ZooKeeper chroot path as part | ||
| // of its ZooKeeper connection string which puts its data under some path in the global ZooKeeper namespace. | ||
| // If set under KRaft mode, Koperator ignores this configuration. | ||
| // +optional | ||
| ZKPath string `json:"zkPath,omitempty"` | ||
| RackAwareness *RackAwareness `json:"rackAwareness,omitempty"` | ||
| ClusterImage string `json:"clusterImage,omitempty"` | ||
|
|
@@ -126,6 +140,8 @@ type KafkaClusterStatus struct { | |
| RollingUpgrade RollingUpgradeStatus `json:"rollingUpgradeStatus,omitempty"` | ||
| AlertCount int `json:"alertCount"` | ||
| ListenerStatuses ListenerStatuses `json:"listenerStatuses,omitempty"` | ||
| // ClusterID is a base64-encoded random UUID generated by Koperator to run the Kafka cluster in KRaft mode | ||
| ClusterID string `json:"clusterID,omitempty"` | ||
| } | ||
|
|
||
| // RollingUpgradeStatus defines status of rolling upgrade | ||
|
|
@@ -173,11 +189,12 @@ type DisruptionBudgetWithStrategy struct { | |
| DisruptionBudget `json:",inline"` | ||
| // The strategy to be used, either minAvailable or maxUnavailable | ||
| // +kubebuilder:validation:Enum=minAvailable;maxUnavailable | ||
| Strategy string `json:"strategy,omitempty"` | ||
| } | ||
|
|
||
| // Broker defines the broker basic configuration | ||
| type Broker struct { | ||
| // id maps to "node.id" configuration in KRaft mode, and it maps to "broker.id" configuration in ZooKeeper mode. | ||
| // +kubebuilder:validation:Minimum=0 | ||
| // +kubebuilder:validation:Maximum=65535 | ||
| // +kubebuilder:validation:ExclusiveMaximum=true | ||
|
|
@@ -189,6 +206,11 @@ type Broker struct { | |
|
|
||
| // BrokerConfig defines the broker configuration | ||
| type BrokerConfig struct { | ||
| // processRoles defines the role(s) for this particular Kafka node: "broker", "controller", or both. | ||
| // This must be set in KRaft mode. If set in ZooKeeper mode, Koperator ignores this configuration. | ||
| // +kubebuilder:validation:MaxItems=2 | ||
| // +optional | ||
| Roles []string `json:"processRoles,omitempty"` | ||
|
There was a problem hiding this comment. This also can be []KraftNodeRole There was a problem hiding this comment. Well, I thought about different potential names, but ended up following what the Kafka community itself uses: https://github.com/apache/kafka/blob/trunk/config/kraft/controller.properties#L24 This way, the users (I assume they are someone familiar with Kafka) would immediately know what this field is about. Hope that makes sense |
||
| Image string `json:"image,omitempty"` | ||
| MetricsReporterImage string `json:"metricsReporterImage,omitempty"` | ||
| Config string `json:"config,omitempty"` | ||
|
|
@@ -871,6 +893,19 @@ func (bConfig *BrokerConfig) GetTerminationGracePeriod() int64 { | |
| return *bConfig.TerminationGracePeriod | ||
| } | ||
|
|
||
| // GetStorageMountPaths returns a string with comma-separated storage mount paths that the broker uses | ||
| func (bConfig *BrokerConfig) GetStorageMountPaths() string { | ||
| var mountPaths string | ||
| for i, sc := range bConfig.StorageConfigs { | ||
| if i != len(bConfig.StorageConfigs)-1 { | ||
| mountPaths += sc.MountPath + "," | ||
| } else { | ||
| mountPaths += sc.MountPath | ||
| } | ||
| } | ||
| return mountPaths | ||
| } | ||
|
|
||
| // GetNodeSelector returns the node selector for cruise control | ||
| func (cConfig *CruiseControlConfig) GetNodeSelector() map[string]string { | ||
| return cConfig.NodeSelector | ||
|
|
@@ -1004,6 +1039,31 @@ func (cConfig *CruiseControlConfig) GetResources() *corev1.ResourceRequirements | |
| } | ||
| } | ||
|
|
||
| // IsBrokerNode returns true when the broker is a broker node | ||
| func (bConfig *BrokerConfig) IsBrokerNode() bool { | ||
| return util.StringSliceContains(bConfig.Roles, brokerNodeProcessRole) | ||
| } | ||
|
|
||
| // IsControllerNode returns true when the broker is a controller node | ||
| func (bConfig *BrokerConfig) IsControllerNode() bool { | ||
| return util.StringSliceContains(bConfig.Roles, controllerNodeProcessRole) | ||
| } | ||
|
|
||
| // IsBrokerOnlyNode returns true when the broker is a broker-only node | ||
| func (bConfig *BrokerConfig) IsBrokerOnlyNode() bool { | ||
| return bConfig.IsBrokerNode() && !bConfig.IsControllerNode() | ||
| } | ||
|
|
||
| // IsControllerOnlyNode returns true when the broker is a controller-only node | ||
| func (bConfig *BrokerConfig) IsControllerOnlyNode() bool { | ||
| return bConfig.IsControllerNode() && !bConfig.IsBrokerNode() | ||
| } | ||
|
|
||
| // IsCombinedNode returns true when the broker is a broker + controller node | ||
| func (bConfig *BrokerConfig) IsCombinedNode() bool { | ||
| return bConfig.IsBrokerNode() && bConfig.IsControllerNode() | ||
| } | ||
|
|
||
| // GetResources returns the broker specific Kubernetes resource | ||
| func (bConfig *BrokerConfig) GetResources() *corev1.ResourceRequirements { | ||
| if bConfig.Resources != nil { | ||
|
|
||
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Suggestion:
From go 1.18 there is an alternative solution for this
if idx := slices.Index(list, s); idx != -1 f {
found
}
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Thx! I just copied this func and the test from the util package without taking a second thought about it. Will look into the suggestion