Do not allow to delete volume from brokers #820
Conversation
| for _, storageConfigOld := range brokerConfigsOld.StorageConfigs { | ||
| isStorageFound := false | ||
|
|
||
| for _, storageConfigNew := range brokerConfigsNew.StorageConfigs { | ||
| if storageConfigOld.MountPath == storageConfigNew.MountPath { | ||
| isStorageFound = true | ||
| break | ||
| } | ||
| } | ||
| if !isStorageFound { | ||
| log.Info(fmt.Sprintf("Not allowed to remove broker storage with mountPath: %s from brokerID: %v", storageConfigOld.MountPath, brokerOld.Id)) | ||
| return notAllowed(fmt.Sprintf("Removing storage from a runnng broker is not supported! (mounthPath: %s, brokerID: %v)", storageConfigOld.MountPath, brokerOld.Id), metav1.StatusReasonInvalid) | ||
| } |
There was a problem hiding this comment.
The logic here looks a bit fishy to me: when we are at the first storageConfigOld and its MountPath doesn't match all the possible MountPath in the brokerConfigsNew.StorageConfigs, we immediately return notAllowed(...), is this intended?
There was a problem hiding this comment.
Yes, because we have found a removed storage so this update should be rejected.
Do you think we should search all of the removed storages and writes out all of them and after reject.
What others think?
There was a problem hiding this comment.
so the intention here is to return immediately when the removed storage is found (e.g. isStorageFound is true)? But the implementation returns when isStorageFound is false
There was a problem hiding this comment.
Probably the name here is ambiguous. If I can found the old storage in the new one then the isStorageFound will be true. But If I cant, then it will be false, and I will reject.
| BrokerConfigGroups: map[string]v1beta1.BrokerConfig{ | ||
| "default": { | ||
| StorageConfigs: []v1beta1.StorageConfig{ | ||
| //v1beta1.StorageConfig{MountPath: "logs1"}, |
There was a problem hiding this comment.
Note: unnecessary comment? (unless it is to understand the case more easily)
There was a problem hiding this comment.
It can be removed, but you can see also more easily that this one has been removed from the new Config. So this validation should be rejected
There was a problem hiding this comment.
If we wish to show that should we not have a comment indicating that instead? Like instead of this you could write in the comment that the removed MountPath is this.
| BrokerConfigGroups: map[string]v1beta1.BrokerConfig{ | ||
| "default": { | ||
| StorageConfigs: []v1beta1.StorageConfig{ | ||
| //v1beta1.StorageConfig{MountPath: "logs1"}, |
There was a problem hiding this comment.
If we wish to show that should we not have a comment indicating that instead? Like instead of this you could write in the comment that the removed MountPath is this.
| switch req.Kind.Kind { | ||
| case kafkaTopic: | ||
| l.Info("AdmissionReview") |
There was a problem hiding this comment.
Instead of l.Info("AdmissionReview") you could initialize the logger as
l := log.WithName("AdmissionReview").WithValues("kind", req.Kind, "namespace", req.Namespace, "name", req.Name, "uid", req.UID,
"operation", req.Operation, "user info", req.UserInfo)
pkg/webhook/request.go
Outdated
| @@ -55,8 +56,30 @@ func (s *webhookServer) validate(ar *admissionv1.AdmissionReview) *admissionv1.A | |||
| } | |||
| } | |||
| return s.validateKafkaTopic(&topic) | |||
| case kafkaCluster: | |||
| // when the operator modifies the resource we dont do any validation | |||
| operatorUsername := fmt.Sprintf("system:serviceaccount:%v:%v", os.Getenv("POD_NAMESPACE"), os.Getenv("SERVICE_ACCOUNT")) | |||
There was a problem hiding this comment.
Inject the namespace and service account instead of always reading from env var
What's in this PR?
Why?
Currently nothing prohibits the user to remove broker storage entry from KafkaCluster custom resource. This will trigger koperator to remove the storage from broker configuration. This should not be allowed because it can lead to data loss
Checklist