ceph: support bucket notifications for object storage #8426
Conversation
yuvalif
force-pushed
the
ceph-bucket-notifications
branch
2 times, most recently
from
91dc0d2
to
a864e90
Compare
|
@travisn currently, i get the following error when trying to start the OBC labels controller: i cannot register OBC in |
I'm not seeing where to register it either. We don't have a pattern established for watching CRDs defined outside the types defined directly in Rook. Perhaps the cephv1 registration needs to reference the obc registration in the library, but that doesn't seem quite right either. btw you'll want to rebase on the latest master when you get a chance. Your branch wasn't building for me without rebasing. |
yuvalif
force-pushed
the
ceph-bucket-notifications
branch
from
086b3d9
to
0518152
Compare
this seems to work better. not sure if this is valid? also, did a rebase |
yuvalif
force-pushed
the
ceph-bucket-notifications
branch
from
0518152
to
258152f
Compare
yuvalif
force-pushed
the
ceph-bucket-notifications
branch
from
258152f
to
ce8f436
Compare
yuvalif
changed the title
yuvalif
force-pushed
the
ceph-bucket-notifications
branch
from
ce8f436
to
695a73f
Compare
There was a problem hiding this comment.
Overall it's looking good, some overall items that you also mentioned in huddle:
- Integration tests are needed to create the notifications and topics, likely in ceph_base_objevct_test.go. To test across namespaces it would be good to create the OBC in a different namespace from the cluster and notifications.
- Documentation is needed
| topic: my-topic-example | ||
| filter: | ||
| keyFilters: | ||
| - name: prefix |
| apiVersion: ceph.rook.io/v1 | ||
| kind: CephBucketNotification | ||
| metadata: | ||
| name: my-notification-example |
There was a problem hiding this comment.
How about dropping the "example" suffix?
| name: my-notification-example | |
| name: my-notification |
| - name: suffix | ||
| value: .png | ||
| - name: regex | ||
| value: "[a-z]*" |
There was a problem hiding this comment.
How about adding a comment for the behavior of the complex filters?
| value: "[a-z]*" | |
| # Match zero or more letters with a regular expression | |
| value: "[a-z]*" |
| apiVersion: ceph.rook.io/v1 | ||
| kind: CephBucketTopic | ||
| metadata: | ||
| name: my-topic-example |
There was a problem hiding this comment.
| name: my-topic-example | |
| name: my-topic |
| namespace: rook-ceph | ||
| spec: | ||
| # use one of the following endpoints | ||
| endpoint: http://localhost:8080 |
There was a problem hiding this comment.
What does it mean to use localhost? Does this mean an http server is expected to be listening on port 8080 on every node where rgw is running?
There was a problem hiding this comment.
you are right.
this should be a service and an HTTP server pod.
There was a problem hiding this comment.
added this: https://github.com/rook/rook/pull/8426/files#diff-5535c538881c96e5f27627d9c1f3b59707680418d99cc43b4aeddf2f05201261
and modified the topic endpoint accordingly
pkg/apis/ceph.rook.io/v1/topic.go
Outdated
| var urlSchema string | ||
| if endpointURL, err := url.Parse(t.Spec.Endpoint); err != nil { | ||
| return err | ||
| } else { |
There was a problem hiding this comment.
No need for else since the previous line returned, right?
| } else { | |
| } |
| func (r *ReconcileOBCLabels) createCephBucketNotifications(cephOBC *bktv1alpha1.ObjectBucketClaim) error { | ||
| // Step 1: looking for notifications in the lables | ||
| for key, name := range cephOBC.Labels { | ||
| notifLabels := strings.SplitAfterN(key, "bucket-notification-", 2) |
There was a problem hiding this comment.
How about defining bucket-notification- as a const at the top of the file? Also:
| notifLabels := strings.SplitAfterN(key, "bucket-notification-", 2) | |
| notifyLabels := strings.SplitAfterN(key, "bucket-notification-", 2) |
| } | ||
|
|
||
| // TODO: | ||
| // * what is the difference between objContext.ZoneGroup and bucket.Spec.Endpoint.Region |
There was a problem hiding this comment.
I would expect the objContext properties refer to the object store endpoint. I don't see a bucket.Spec in this section so maybe this is an old TODO?
There was a problem hiding this comment.
the Create() method, calling this createSession() method has a pointer to the OB, which has "endpoint" and "region" fields
There was a problem hiding this comment.
Perhaps @alimaredia could shed some light on these questions? I'm not entirely sure what are the right fields to use here.
There was a problem hiding this comment.
bucket.Spec.Endpoint.Region is just the placeholder for the was sdk for the region to use but has no real meaning for rgw AFAIK. Just for API compliance, I suppose. As for objContext.ZoneGroup is different, it represents the zone group the rgw process is part of.
There was a problem hiding this comment.
I'm still thinking about this from a higher level and trying to make sure that it will meet user and admin needs, but I want to post what I have so far.
I'm trying to understand this from the design:
Usually bucket notification will be created by user for consuming it for the applications, so it need to created on App's namespace similar to OBC/BAR.
Does this mean that the Rook operator only needs to check for matching OBCs with matching labels in the same namespace as a given Notification and Topic?
| err := ValidateNotificationSpec(notification) | ||
| assert.NoError(t, err) | ||
|
|
||
| notification.Spec.Topic = "" | ||
| err = ValidateNotificationSpec(notification) | ||
| assert.Error(t, err) | ||
|
|
||
| notification.Spec.Topic = "fish-topic" | ||
| notification.Spec.Events = append(notification.Spec.Events, &kaboomEvent) | ||
| err = ValidateNotificationSpec(notification) | ||
| assert.Error(t, err) |
There was a problem hiding this comment.
Using subtests allows naming the tests. It would be helfpul for review and code maintenance also to see titles like "filter spec with value that isn't present" or whatever test name is descriptive to understand the feature being tested.
t.Run("name of test", func(t *testing.T) {
// a test
})Ditto elsewhere also.
pkg/apis/ceph.rook.io/v1/types.go
Outdated
| // This specified CA will be used, instead of the default one, to authenticate the broker | ||
| // +kubebuilder:validation:Pattern=`^/|//|(/[\w-]+)+$` | ||
| // +optional | ||
| CALocation string `json:"caLocation,omitempty"` |
There was a problem hiding this comment.
I have a question about this:
The designs talk about this here: https://github.com/rook/rook/blob/master/design/ceph/object/ceph-bucket-notification-crd.md#implementation
The design says that it is a path to a location in the RGW pod. But how can users use their own CA cert file? I don't think there is a way for users to inject it into RGW pods since they are controlled by Rook. Should this be the name of a Secret containing the CA cert instead? Or some other method?
Is this related to this PR? https://github.com/rook/rook/pull/8243/files
There was a problem hiding this comment.
yes, this should be a name of a secret.
currently, the RGW expects a file, so the plan is that the operator would read the secret, write that to a file in some known location in the pod, and configure that location to the RGW.
we would probably do that in a later PR, but agree that either way, there is not point in setting the file location as a parameter
There was a problem hiding this comment.
Does that mean that currently, we don't support any authentication methods? I see that UN and pass can be included in the endpoint, but that is in plaintext in manifests.
There was a problem hiding this comment.
- for security we support SSL. the limitation in rook is that we cannot provide a custom CA. so depending on how the certificates are signed we may or may not support security
- for authentication we only support plaintext user/password (this is unrelated to rook)
| @@ -30,7 +30,9 @@ import ( | |||
| "github.com/rook/rook/pkg/operator/ceph/file/mirror" | |||
| "github.com/rook/rook/pkg/operator/ceph/nfs" | |||
| "github.com/rook/rook/pkg/operator/ceph/object" | |||
| "github.com/rook/rook/pkg/operator/ceph/object/bucket" | |||
There was a problem hiding this comment.
Maybe this package would be better named "notification"? I read "bucket" as an interface for buckets directly instead of notifications about buckets.
There was a problem hiding this comment.
ok. will move to the "notification" package - where the provisioner is
| @@ -0,0 +1,271 @@ | |||
| /* | |||
There was a problem hiding this comment.
I don't think we really want a label "Controller". We can't control the resources that are labelled. Really what Rook should be doing is watching for the related resources that might change from the controllers that we do own (BucketNotification and BucketTopic).
In the CephCluster controller, we watch for changes to the rook-ceph-operator-config ConfigMap, for example to decide if we need to re-run a reconcile for a CephCluster.
The crash controller is also a good example. It watches for changes to deployments in order to operate. See file:pkg/operator/ceph/cluster/crash/add.go
I think this is the most appropriate thing to do for Notifications and/or Topics. It's still not clear to me whether both need to be evaluated when buckets are added.
There was a problem hiding this comment.
topics are unrelated.
just to be clear, you are suggesting that we will have a notification controller that watches both on BucketNotification CRs and OBCs?
- whenever it sees a change to a BucketNotification CR, it fetches all relevant OBCs (according to its selector), and provisions all notifications
- whenever it sees a new OBC (or a change in the labels of an OBC), the controller needs to fetch the relevant BucketNotification CR. how would we do that? won't we need a label that indicates the notification?
There was a problem hiding this comment.
Yes. That is how child controllers (like CephObjectStore reconcile when there are CephCluster updates).
There are options:
- query all the BNs (BucketNotifications) when there is a change and examine them
- keep a list of all found BNs in memory with their selectors that Rook can loop through when OBCs change, and query only the ones whose selectors match
| "strconv" | ||
|
|
||
| "github.com/rook/rook/pkg/operator/ceph/object" | ||
|
|
| // TODO: which store to take? | ||
| objStore := objStores.Items[0] |
There was a problem hiding this comment.
I think this really needs to be figured out.
The OBC defines which CephCluster and which CephObjectStore by setting parameters in the StorageClass
apiVersion: storage.k8s.io/v1
kind: StorageClass
metadata:
name: rook-ceph-delete-bucket
# NOTE: rook-ceph here means to use the Ceph cluster in the rook-ceph namespace
provisioner: rook-ceph.ceph.rook.io/bucket # driver:namespace:cluster
# set the reclaim policy to delete the bucket and all objects
# when its OBC is deleted.
reclaimPolicy: Delete
parameters:
# objectStoreName references the name of the CephObjectStore
objectStoreName: my-store
# objectStoreNamespace also refers to the CephCluster in the noted namespace
objectStoreNamespace: rook-ceph # namespace:cluster
region: us-east-1
# To accommodate brownfield cases reference the existing bucket name here instead
# of in the ObjectBucketClaim (OBC). In this case the provisioner will grant
# access to the bucket by creating a new user, attaching it to the bucket, and
# providing the credentials via a Secret in the namespace of the requesting OBC.
#bucketName:We could look at the OBC and then look at the storageclass to determine the Cluster and ObjectStore; however, that might not be the best given that the storageclass could be changed by a user/admin.
I'll look at OBCs and OBs to see if there is a better way to cross-reference.
It may be that we just need users to specify these in the CephBucketNotification CR
There was a problem hiding this comment.
It may be that we just need users to specify these in the CephBucketNotification CR
- in the case of
CephBucketTopici will addobjectStoreName - in the case of
CephBucketNotificationcan also do that, but what if its value is different than the one in the storage class of the OBC?
There was a problem hiding this comment.
Yeah, that's a question for me too. I'm not sure what the behavior of lib-bucket-provisioner is if the StorageClass changes after an OBC has been created. I think the SC is only gotten at initial creation, so we can't assume the SC is valid still after an OBC exists, but I'm not 100% sure.
There was a problem hiding this comment.
Or, the SC could be deleted after the OBC is created.
There was a problem hiding this comment.
ok. so already added objectStoreName to CephBucketNotification in 497a602
to be on the safe side, and i can try and fetch the SC of the OBC and if its there verify that the store names are the same
yes. this is the feedback we got from @guimou |
|
FWIW, it looks like K8s uses a cache to keep a record of all services, and it uses a work queue with 10 goroutines to handle scale. This code is a bit old and has moved, and I haven't found the most recent source, so some of the details might have changed slightly. [Update] |
yuvalif
force-pushed
the
ceph-bucket-notifications
branch
from
bf2f588
to
549b557
Compare
| # match objects with keys that end with ".png" | ||
| - name: suffix | ||
| value: .png | ||
| # match objects with keys with only lowercase characters | ||
| - name: regex | ||
| value: "[a-z]*" |
There was a problem hiding this comment.
Is this an invalid example? Since the . character isn't allowed in the lowercase regex, then does that mean it can't match objects with .png extensions? If I'm correct, perhaps the lowercase regex should also support . characters to make sure it is a working example for users.
| ROOK_LOG_LEVEL: "INFO" | ||
| ROOK_LOG_LEVEL: "DEBUG" | ||
|
|
||
| # Enable the CSI driver. | ||
| # To run the non-default version of the CSI driver, see the override-able image properties in operator.yaml | ||
| ROOK_CSI_ENABLE_CEPHFS: "true" | ||
| ROOK_CSI_ENABLE_CEPHFS: "false" | ||
| # Enable the default version of the CSI RBD driver. To start another version of the CSI driver, see image properties below. | ||
| ROOK_CSI_ENABLE_RBD: "true" | ||
| ROOK_CSI_ENABLE_RBD: "false" |
There was a problem hiding this comment.
mistake. will remove
| CSI_ENABLE_CEPHFS_SNAPSHOTTER: "true" | ||
| CSI_ENABLE_CEPHFS_SNAPSHOTTER: "false" | ||
|
|
||
| # set to false to disable deployment of snapshotter container in RBD provisioner pod. | ||
| CSI_ENABLE_RBD_SNAPSHOTTER: "true" | ||
| CSI_ENABLE_RBD_SNAPSHOTTER: "false" | ||
|
|
||
| # Enable cephfs kernel driver instead of ceph-fuse. | ||
| # If you disable the kernel client, your application may be disrupted during upgrade. | ||
| # See the upgrade guide: https://rook.io/docs/rook/master/ceph-upgrade.html | ||
| # NOTE! cephfs quota is not supported in kernel version < 4.17 | ||
| CSI_FORCE_CEPHFS_KERNEL_CLIENT: "true" | ||
| CSI_FORCE_CEPHFS_KERNEL_CLIENT: "false" |
There was a problem hiding this comment.
mistake. will remove
| image: rook/ceph:master | ||
| image: quay.io/ylifshit/rook-ceph | ||
| #image: rook/ceph:master | ||
| #image: build-1707c52e/ceph-amd64:latest |
There was a problem hiding this comment.
mistake. will remove
| Name: "fish-topic", | ||
| }, | ||
| Spec: BucketTopicSpec{ | ||
| Endpoint: "Kafka://myserver:9999", |
There was a problem hiding this comment.
Is Kafka supposed to be capitalized here? I see it as lower case in the design doc.
There was a problem hiding this comment.
Is Kafka supposed to be capitalized here? I see it as lower case in the design doc. Do we support both forms? If so, let's have a specific test for that. Could we support KAFKA:// as well?
| Spec BucketNotificationSpec `json:"spec"` | ||
| // +kubebuilder:pruning:PreserveUnknownFields | ||
| // +optional | ||
| Status *Status `json:"status,omitempty"` |
There was a problem hiding this comment.
Possibly for a future PR: A notification can be used for multiple buckets, right? What if the status reported the list of OBCs which are provisioned with notifications? That way if a user creates a new OBC, they can know when notifications are enabled for it.
Considering COSI resources for the future, we could just keep a tuple list that would be something like `[{"objectbucketclaims.objectbucket.io", "obc-namespace/obc-name"}, {"bucketaccessrequest.cosi.io", "bar-namespace/bar-name"}]
There was a problem hiding this comment.
sure. we can do that. will add in the "todo" comment
| @@ -0,0 +1,257 @@ | |||
| /* | |||
There was a problem hiding this comment.
I'm quite honestly struggling a lot with this implementation. The logic in this file seems backwards to me. There is no actual controller for the CephBucketNotification resource that sees add/update/delete actions from a CephBucketNotification and then performs the required actions there. Instead, this operates in a logically inverted way where it looks for changes to OBCs and then cascades that to CBNs.
Because this is implemented inversely to all other controllers, I'm having a hard time logically figuring out what events are happening and how they cascade to cause changes to Ceph and other resources. It is therefore hard for me to figure out if there are logical issues with the implementation. While the code here might work, given my difficulties following the logic, I don't think it's a good idea to merge this in the current state.
Ideally, the controller code should follow similar patterns to existing controllers so there is not a large context switch between maintaining those controllers and this one. Controller logic should be small and straightforward with very little complex logic so the controller behavior can be examined separately from that deals with resource interactions and Ceph. We should reuse as much logic as possible for doing dealing with resource interactions so that we don't have to validate and mentally parse different logical flows of information.
I think a code architecture with the below high level design would meet my above requests and make maintenance much easier in the long run.
See pkg/operator/ceph/cluster/controller.go for an example of a controller that watches for updates to multiple types of resources (CephCluster and Node, for example) and reconciles on those changes.
@travisn @leseb feel free to chime in with other thoughts.
high level design
A CephBucketNotification (CBN) controller that does 2 things...
- watches for changes (add/update/delete) to CBNs:
- if deletion timestamp exists, delete the CBN and related Ceph configs
- If add/update, call a reusable function to add/update the CBN and its ceph configs (I'll call it
updateCBN(cbn)for my purposes)
- watches for changes to OBCs:
- if the OBC is new OR labels changed related to CBNs, continue (otherwise, do nothing)
- using the namespace of the OBC and the OBCs labels, determine the namespacedname of CBNs that should match
- for each CBN:
- get the CBN
- if the CBN doesn't exist, log the fact and skip it (if it doesn't yet exist, the CBN controller will detect the CBN when it's created later)
- call the reusable function
updateCBN(cbn)to update the notification
The reusable logic
updateCBN(cbn):
- use some sort of mutex lock to make sure another routine won't modify the CBN's related configs at the same time
- if can't get mutex lock right now, requeue the request after a few seconds
- defer the mutex unlock
- Update the CBN status phase to 'Reconciling' or 'Updating' or something
- look for all OBCs in the CBN's namespace with matching labels.
- if none exist, exit. the CBN controller will detect OBCs when they are created or have their labels modified in the future (don't forget to update status)
- In the future, we will also have to look for COSI resources (BARs?), so we should plan for making it easy to insert that in the near future.
- for each OBC:
- if the OBC has a deletion timestamp, don't process it
- Use a reusable function like
extractBucketInfoFromOBC(we'll wantextractBucketInfoFromBARin the future) to figure out details we need for creating or updating the Ceph config. - make a note of the bucket for later (I'll call it
processedBuckets) - Update Ceph notification configs as needed
- (is this step necessary?) using Ceph CLI, determine the buckets configured for this notification
- for each bucket
2. if the bucket is not inprocessedBuckets, remove the config for it (the OBC for it was deleted)
- for each bucket
- update the CBN status to 'Ready' or similar.
- In a future PR, we probably want to report the OBCs (and COSI resources in the future) which are "hooked up" to the notification.
deleteCBN(cbn):
- call when CBN is deleted to delete the CBN
Things I don't know
Putting this all down here, I realize that there is an interaction that I don't fully understand : Topics. I'm not sure how topics fit into all of this, so obviously what I propose will have to change somewhat to also interact with CephBucketTopics.
Guiding questions for Topics:
- if a topic is added/updated, do CBNs need to be reconciled? Ceph docs suggest that a notification is "into" a topic, so I suspect the answer is yes. [Update] Also, CBN spec has the
topicconfig, so I think this must be yes. - if a topic is deleted, how do CBNs need to be reconciled? Should we disallow deleting a topic until all CBNs that reference it are removed?
- is a topic necessary for a notification to be created and working? If so, creating a CBN should re-queue until an appropriate topic exists
- Should it be possible to change a CBN's
spec.topic?
There was a problem hiding this comment.
There is no actual controller for the CephBucketNotification resource that sees add/update/delete actions from a CephBucketNotification and then performs the required actions there.
from the RGW perspective, there are no topics without buckets. so CBNs don't really exist in ceph - this is why the controller does not watch these issues.
in a different PR, I plan on adding cascade delete/update of notification when CBN's are deleted or updated.
for that, I would need a controller that watches over these values, and using the labels in the OBCs get a list of all impacted OBCs, so we can delete/update them.
adding that should not change the user interface, just make the feature more useable.
if the CBN doesn't exist, log the fact and skip it (if it doesn't yet exist, the CBN controller will detect the CBN when it's created later)
this is the part different in the implementation. currently, if the CBN does not exist I fail the "reconcile" action and requeue it for retries. since we don't modify the state of the OBC (its is still in "Ready" state), this should work.
use some sort of mutex lock to make sure another routine won't modify the CBN's related configs at the same time
why is this needed? do we do that for other CRs?
regarding updateCBN():
- the OBC part which is implemented its very similar to the suggested design
- the CBN part will be added to address the cascade delete/update
(is this step necessary?) using Ceph CLI, determine the buckets configured for this notification
for each bucket
there is no ceph CLI to do that. why is this needed?
regarding topics. currently, there are no mechanisms in ceph to prevent deletions of topics "in use", nor a mechanism for updating notifications that had their topics updated (it is even documented in ceph...)
we can add this functionality, but this would be a "nice to have"
There was a problem hiding this comment.
from the RGW perspective, there are no topics without buckets. so CBNs don't really exist in ceph - this is why the controller does not watch these issues.
in a different PR, I plan on adding cascade delete/update of notification when CBN's are deleted or updated.
for that, I would need a controller that watches over these values, and using the labels in the OBCs get a list of all impacted OBCs, so we can delete/update them.
adding that should not change the user interface, just make the feature more useable.
I still don't feel comfortable merging this into Rook in its present state given the code maintenance and context switching concerns I mentioned.
It doesn't really matter to me how RGW implements things. The most important concern to me is how the user will engage with the feature followed by how the feature can be maintained in Rook. In my experience, when implementation of features primarily follows the Ceph design, usability suffers, as does maintainability.
The CBN needs a controller because even if it isn't a "real" resource in Ceph, we have made it a "real" resource in Rook.
As to the mutex lock, we do this for CephClusters but not other CRs. Given that CBN configuration can change based on CBN updates, OBC updates, and Topic updates I think a mutex lock would be good to help us prevent bugs caused by the different controllers vying to change the configuration at the same time.
There was a problem hiding this comment.
I should add some more clarity: I don't have a problem with leaving delete/update features for future PRs. We have done that before (notably with RGW multisite), and it has worked well. I do, however, have a strong aversion to merging code that I don't think could maintain.
There was a problem hiding this comment.
I don't have a problem with leaving delete/update features for future PRs.
in this case the CBN controller will be added when we support updates/deletes of notifications based on updates/deletes of CBNs.
The CBN needs a controller because even if it isn't a "real" resource in Ceph, we have made it a "real" resource in Rook.
without the update/delete functionality, there is no code that needs to go into the controller
when implementation of features primarily follows the Ceph design, usability suffers, as does maintainability.
completely agree. but the implementation here does not follow the RGW interfaces. this is why we have CNB (which don't exist in RGW/ceph).
| logger.Debugf("AWS SNS session endpoint: %q credentials: %q %q region: %q", | ||
| *sess.Config.Endpoint, | ||
| accessKey, | ||
| secretKey, | ||
| *sess.Config.Region, | ||
| ) |
There was a problem hiding this comment.
We don't want to leak sensitive info to the logs. I think we should only print endpoint and region.
yuvalif
force-pushed
the
ceph-bucket-notifications
branch
from
e0431c1
to
b7d9c63
Compare
pkg/apis/ceph.rook.io/v1/topic.go
Outdated
| urlSchema = strings.ToLower(endpointURL.Scheme) | ||
| validSchemas := []string{"amqp", "amqps", "http", "kafka"} | ||
| if sort.SearchStrings(validSchemas, urlSchema) == len(validSchemas) { | ||
| return errors.New("invalid URL schema: " + endpointURL.Scheme) |
There was a problem hiding this comment.
use errors.Errorf() for formatting errors?
pkg/apis/ceph.rook.io/s3ext/api.go
Outdated
| func DeleteBucketNotificationRequest(c *s3.S3, input *DeleteBucketNotificationRequestInput) (req *request.Request) { | ||
| op := &request.Operation{ | ||
| Name: opDeleteBucketNotification, | ||
| HTTPMethod: "DELETE", |
pkg/apis/ceph.rook.io/v1/types.go
Outdated
| Persistent bool `json:"persistent,omitempty"` | ||
| // Spec of HTTP endpoint | ||
| // +optional | ||
| HTTP *HTTPEndpointSpec `json:"http"` |
There was a problem hiding this comment.
| HTTP *HTTPEndpointSpec `json:"http"` | |
| HTTP *HTTPEndpointSpec `json:"http,omitempty"` |
| } | ||
|
|
||
| // find the namespace for the ceph cluster (may be different then the namespace of the topic CR) | ||
| clusterNamespace := os.Getenv(k8sutil.PodNamespaceEnvVar) |
There was a problem hiding this comment.
Using the operator namespace is not reliable. I'd just call out for the CephClusterList object and get the namespace from the object you find, typically one.
There was a problem hiding this comment.
can you point me to a place where we get the ceph cluster list without a namespace?
I see this kind of code that needs a namespace:
clusterList := &cephv1.CephClusterList{}
err := c.client.List(context.TODO(), clusterList, client.InNamespace(c.namespace))
There was a problem hiding this comment.
You can just pass &client.ListOptions{},
cephClusters := &cephv1.CephClusterList{}
err := r.client.List(context.TODO(), cephClusters, &client.ListOptions{})There was a problem hiding this comment.
but what do i do if i get more than one cluster?
There was a problem hiding this comment.
For some components, it's a known corner case but now that I think about it, looking at the op namespace is not such a bad idea. I guess we could do both. If we find something in the op ns we continue, if nothing we try to list all of them. If we find one we are good, if we find more than one we have a problem...
| // newReconciler returns a new reconcile.Reconciler | ||
| func newReconciler(mgr manager.Manager, context *clusterd.Context) reconcile.Reconciler { | ||
| // Add the cephv1 scheme to the manager scheme so that the controller knows about it | ||
| mgrScheme := mgr.GetScheme() |
There was a problem hiding this comment.
You don't need this block, the scheme are registered in pkg/operator/ceph/cr_manager.go.
There was a problem hiding this comment.
copy&paste from another example :-(
will remove
| // ReconcileOBCLabels reconciles a ObjectBucketClaim labels | ||
| type ReconcileOBCLabels struct { | ||
| client client.Client | ||
| scheme *runtime.Scheme |
There was a problem hiding this comment.
Where do you use the scheme and what for? I didn't see it being used anywhere.
There was a problem hiding this comment.
copy&paste from another example :-(
will remove
| // newReconciler returns a new reconcile.Reconciler | ||
| func newReconciler(mgr manager.Manager, context *clusterd.Context) reconcile.Reconciler { | ||
| // Add the cephv1 scheme to the manager scheme so that the controller knows about it | ||
| mgrScheme := mgr.GetScheme() |
| } | ||
|
|
||
| // find the namespace for the ceph cluster (may be different then the namespace of the topic CR) | ||
| clusterNamespace := os.Getenv(k8sutil.PodNamespaceEnvVar) |
| ClusterSpec *cephv1.ClusterSpec | ||
| } | ||
|
|
||
| func (p *Provisioner) createSession(objectStoreName string) (*awssession.Session, error) { |
There was a problem hiding this comment.
Can we have single CreateSession() and it in s3-handlers.go? only notable difference I can see here it uses adminOps User credentials than the owner
There was a problem hiding this comment.
this is indeed the difference between them. but trying to merge them into one function will make it more complex
| WithCredentials(credentials.NewStaticCredentials(accessKey, secretKey, "")). | ||
| WithEndpoint(objContext.Endpoint). | ||
| WithMaxRetries(3). | ||
| WithDisableSSL(!cephObjectStore.Spec.IsTLSEnabled()). |
There was a problem hiding this comment.
if SSL is enabled for RGW, this request most likely fail. please handle it like in newS3agent()
| // Step 3: looking for notifications in the labels | ||
| for key, name := range cephOBC.Labels { | ||
| notifyLabels := strings.SplitAfterN(key, notificationLabelPrefix, 2) | ||
| if len(notifyLabels) > 1 && notifyLabels[1] != "" && name == notifyLabels[1] { |
There was a problem hiding this comment.
the notification label is not valid, atleast we should log than ignoring the value?
There was a problem hiding this comment.
it is valid - the user can add labels for different reasons.
it is just not used for bucket notifications.
There was a problem hiding this comment.
What I meant was if the label is for bucket notification, but the user has given it wrongly. Please consider the following labels :
bucket-notification-my-notification: ""
bucket-notification-my-notification: blah
Should give error/warning for the user for above cases ?
There was a problem hiding this comment.
agree. logging a warning and ignoring the label in case of mismatch
yuvalif
force-pushed
the
ceph-bucket-notifications
branch
from
2df6704
to
1c20aee
Compare
yuvalif
force-pushed
the
ceph-bucket-notifications
branch
2 times, most recently
from
947d6c6
to
7cd2cba
Compare
|
after this fix: 7cd2cba given the size of this PR, I would prefer to do the refactoring of the code as followup work. |
|
|
||
| ## Sample | ||
|
|
||
| ### CBT Custom Resource |
There was a problem hiding this comment.
Could we avoid the CBT and CBN acronyms? Informally we could refer just to topics and notifications, and when formally referring to the CR, we should spell it out for example to CephBucketTopic. Similarly we don't use acronyms for other CRDs such as CBP (CephBlockPool), CF (CephFilesystem), or COS (CephObjectStore). It's just too many acronyms to remember. :)
| ### CBT Custom Resource | |
| ### CephBucketTopic Custom Resource |
| - in case of AMQP endpoint, the name is used for the AMQP topic (“routing key” for a topic exchange) | ||
| - in case of Kafka endpoint, the name is used as the Kafka topic | ||
| 1. `namespace`(optional) of the `CephBucketTopic`. Should match the namespace of the CBN associated with this CBT, and the OBC with the label referencing the CBN | ||
| 1. `endpoint` to which to send the notifications to. the identity of the endpoint in formatted as a URI with the following schema options: `http[s]`, `amqp[s]`, `kafka`. The schema of the endpoint must match the type of the endpoint spec provided (see below) |
There was a problem hiding this comment.
To be consistent with other rook docs, how about separating the settings with a : and capitalizing?
| 1. `endpoint` to which to send the notifications to. the identity of the endpoint in formatted as a URI with the following schema options: `http[s]`, `amqp[s]`, `kafka`. The schema of the endpoint must match the type of the endpoint spec provided (see below) | |
| 1. `endpoint`: Endpoint to which to send the notifications to. The identity of the endpoint in formatted as a URI with the following schema options: `http[s]`, `amqp[s]`, `kafka`. The schema of the endpoint must match the type of the endpoint spec provided (see below) |
There was a problem hiding this comment.
@travisn - will do the capitalization
regarding the other comment, I followed the same style used in the parameter description of the OBC:
https://github.com/rook/rook/blob/master/Documentation/ceph-object-bucket-claim.md#obc-custom-resource
please let me know whether to use ":" or keep the same style as the OBC.
There was a problem hiding this comment.
Not a big deal, following the pattern in the OBC doc is fine.
| 1. `useSSL` (optional) indicates that secure connection will be used for connecting with the broker (“false” by default) | ||
| > Note that n case of Kafka and AMQP, the consumer of the notifications is not required to ack the notifications, since the broker persists the messages before delivering them to their final destinations | ||
|
|
||
| ### CBN Custom Resource |
There was a problem hiding this comment.
| ### CBN Custom Resource | |
| ### CephBucketNotification Custom Resource |
| @@ -0,0 +1,169 @@ | |||
| --- | |||
| title: Object Bucket Notification | |||
There was a problem hiding this comment.
In the table of contents this might be the longest entry if we don't shorten it. How about this? There is enough context around it that I don't think we need "object" in the TOC title, and after they click on the topic they'll see the full title on line 7.
| title: Object Bucket Notification | |
| title: Bucket Notifications |
| indent: true | ||
| --- | ||
|
|
||
| # Ceph Object Bucket Notification |
There was a problem hiding this comment.
| # Ceph Object Bucket Notification | |
| # Ceph Object Bucket Notifications |
yuvalif
force-pushed
the
ceph-bucket-notifications
branch
from
7cd2cba
to
db43e62
Compare
yuvalif
changed the title
There was a problem hiding this comment.
I think this looks like it is in a working state, though our test coverage leaves something to be desired. In the interest of getting this merged and things moving forward, I am approving this, and I have made a list of follow-up items for future PRs (in no particular order). I flagged several doc changes that might be good to update here, but I'll let you decide whether to do it now or in a follow-up PR.
TODO items for follow-up PRs:
- code reuse with reused function (and unit test for above reused function)
- unit tests for controllers
- out of order integration test
- verify update/delete of resources in addition to the current add functionality
- use
ReportReconcileResult - (with above) make sure failure event "failed to retrieve CephBucketTopic" does not cause an immediate re-reconcile but waits the 10 seconds to retry
- add status item to CephBucketNotification about OBCs that are provisioned with the notification
spec.protocoldesign discussion and change- Do we need to support IP address endpoints if we don't already? (at minimum, should test this)
- requested doc updates (can do here or later, your choice)
- rename
verifySSLtodisableVerifySSLto take advantage of default-false data struct
| > Note that the CephBucketTopic, CephBucketNotification and OBC must all belong to the same namespace | ||
|
|
||
| ## Notification Reliability and Delivery | ||
| Notifications may be sent synchronously, as part of the operation that triggered them. In this mode, the operation is acked only after the notification is sent to the topic’s configured endpoint, which means that the round trip time of the notification is added to the latency of the operation itself. |
There was a problem hiding this comment.
I tripped up a bit reading "acked." While "ack" is pretty standard terminology, I think it's good to spell out things for people who might not be arriving at these docs with full context.
| Notifications may be sent synchronously, as part of the operation that triggered them. In this mode, the operation is acked only after the notification is sent to the topic’s configured endpoint, which means that the round trip time of the notification is added to the latency of the operation itself. | |
| Notifications may be sent synchronously, as part of the operation that triggered them. In this mode, the operation is acknowledged (acked) only after the notification is sent to the topic’s configured endpoint, which means that the round trip time of the notification is added to the latency of the operation itself. |
| - a `CephBucketTopic` is custom resource which represents a bucket notification topic and is described by a Custom Resource Definition (CRD) shown below. A bucket notification topic represents an endpoint (or a "topic" inside this endpoint) to which bucket notifications could be sent. | ||
| - a `CephBucketNotification` is a custom resource the defines: topic, events and filters of a bucket notification, and is described by a CRD shown below. The bucket notification itself should also be associated with a bucket to complete its definition, however, this association is done by using special Object Bucket claim (OBC) labels, and is not part of the CephBucketNotification. | ||
|
|
||
| ## Topics |
There was a problem hiding this comment.
Since topics and notifications are RGW terms, can we add links to Ceph docs here so users know where to go for more detail about what these are and how they might benefit from them?
There was a problem hiding this comment.
added a link as part of the intro section
| > Note that the original triggering operation will still be considered as successful even if the notification fail with an error, cannot be delivered or times out | ||
|
|
||
| Notifications may also be sent asynchronously. They will be committed into persistent storage and then asynchronously sent to the topic’s configured endpoint. In this case, the only latency added to the original operation is of committing the notification to persistent storage. | ||
| > Note that if the notification fail with an error, cannot be delivered or times out, it will be retried until successfully acked |
There was a problem hiding this comment.
- "acked" vs "acknowledged" point from above
- Let's make the information present in a block quote part of the description itself here. We use block quotes with bold WARNING or NOTE headings on occasion, but that is for really critical info that users shouldn't miss.
| > Note that if the notification fail with an error, cannot be delivered or times out, it will be retried until successfully acked | |
| If the notification fail with an error, cannot be delivered or times out, it will be retried until successfully acknowledged (acked) |
|
|
||
| ## Notification Reliability and Delivery | ||
| Notifications may be sent synchronously, as part of the operation that triggered them. In this mode, the operation is acked only after the notification is sent to the topic’s configured endpoint, which means that the round trip time of the notification is added to the latency of the operation itself. | ||
| > Note that the original triggering operation will still be considered as successful even if the notification fail with an error, cannot be delivered or times out |
There was a problem hiding this comment.
Let's make the information present in a block quote part of the description itself here. We use block quotes with bold WARNING or NOTE headings on occasion, but that is for really critical info that users shouldn't miss.
| > Note that the original triggering operation will still be considered as successful even if the notification fail with an error, cannot be delivered or times out | |
| The original triggering operation will still be considered as successful even if the notification fail with an error, cannot be delivered or times out |
| ```yaml | ||
| bucket-notification-<notification name>: <notification name> | ||
| ``` | ||
| > Note that the CephBucketTopic, CephBucketNotification and OBC must all belong to the same namespace |
There was a problem hiding this comment.
- Let's always spell things out in docs so that users won't get confused about names that they might not be intimately familiar with.
- Let's make the information present in a block quote part of the description itself here. We use block quotes with bold WARNING or NOTE headings on occasion, but that is for really critical info that users shouldn't miss.
| > Note that the CephBucketTopic, CephBucketNotification and OBC must all belong to the same namespace | |
| > The CephBucketTopic, CephBucketNotification, and ObjectBucketClaim must all belong to the same namespace |
There was a problem hiding this comment.
Further, we should be clear with users that if they create buckets manually (not via OBC), they must also create notifications for those buckets manually, though they could reference a topic created via a CephBucketTopic custom resource.
I think adding that after this statement seems natural.
| if err := r.client.Get(r.opManagerContext, topicName, bucketTopic); err != nil { | ||
| if kerrors.IsNotFound(err) { | ||
| logger.Infof("CephBucketTopic %q not provisioned yet", topicName) | ||
| return waitForRequeueIfTopicNotReady, nil | ||
| } | ||
| return reconcile.Result{}, errors.Wrapf(err, "failed to retrieve CephBucketTopic %q", topicName) | ||
| } |
There was a problem hiding this comment.
Replying to https://github.com/rook/rook/pull/8426/files#r739440151 with a reminder that this should be done in a follow-up PR.
| //GetOBCNotification returns the manifest to create object bucket claim | ||
| func (m *CephManifestsV1_6) GetOBCNotification(claimName string, storageClassName string, objectBucketName string, notificationName string, varBucketName bool) string { | ||
| bucketParameter := "generateBucketName" | ||
| if varBucketName { | ||
| bucketParameter = "bucketName" | ||
| } | ||
| return `apiVersion: objectbucket.io/v1alpha1 | ||
| kind: ObjectBucketClaim | ||
| metadata: | ||
| name: ` + claimName + ` | ||
| bucket-notification` + notificationName + ":" + notificationName + ` | ||
| spec: | ||
| ` + bucketParameter + `: ` + objectBucketName + ` | ||
| storageClassName: ` + storageClassName | ||
| } | ||
|
|
||
| //GetBucketNotification returns the manifest to create ceph bucket notification | ||
| func (m *CephManifestsV1_6) GetBucketNotification(notificationName string, topicName string) string { | ||
| return `apiVersion: objectbucket.io/v1alpha1 | ||
| kind: CephBucketNotification | ||
| metadata: | ||
| name: ` + notificationName + ` | ||
| spec: | ||
| topic: ` + topicName | ||
| } | ||
|
|
||
| //GetBucketTopic returns the manifest to create ceph bucket topic | ||
| func (m *CephManifestsV1_6) GetBucketTopic(topicName string, storeName string, httpEndpointService string) string { | ||
| return `apiVersion: objectbucket.io/v1alpha1 | ||
| kind: CephBucketTopic | ||
| metadata: | ||
| name: ` + topicName + ` | ||
| spec: | ||
| endpoint: http://` + httpEndpointService + ` | ||
| objectStoreName: ` + storeName + ` | ||
| objectStoreNamespace: ` + m.settings.Namespace | ||
| } | ||
|
|
There was a problem hiding this comment.
Notifications aren't backported to 1.6, so we shouldn't have them return anything for this version of Rook in the integration suite. I would suggest having these panic with a message saying they aren't implemented and to fix the test if anyone tries to use them in the future.
| GetOBCNotification(obcName, storageClassName, bucketName string, notificationName string, createBucket bool) string | ||
| GetBucketNotification(notificationName string, topicName string) string | ||
| GetBucketTopic(topicName string, storeName string, httpEndpointService string) string |
There was a problem hiding this comment.
What's going on here with the spacing being different on the bottom two items?
There was a problem hiding this comment.
spaces instead of tabs - fixed
| metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" | ||
| ) | ||
|
|
||
| func (s *ObjectSuite) TestBucketNotificationsInOrder() { |
There was a problem hiding this comment.
Is there going to be an ...OutOfOrder test later?
| return sess, nil | ||
| } | ||
|
|
||
| func createSNSClient(sess *awssession.Session) (snsClient *sns.SNS) { |
There was a problem hiding this comment.
What is an SNS client? Can you provide a small comment doc about that here?
| - user/password defaults to: guest/guest | ||
| - user/password may only be provided if HTTPS is used with the RGW. If not, topic creation request will be rejected | ||
| - vhost defaults to: “/” | ||
| 1. `verifySSL` (optional) indicates whether the RGW is going to verify the SSL certificate of the AMQP server in case AMQPS is used ("true" by default) |
There was a problem hiding this comment.
I was looking at the design doc and noticed this. It is generally recommended that the default values for Kubernetes CRDs be the default values for the Go lang as well so that we needn't implement special default behavior. Instead, it would be recommended that we change this to be the inverse, i.e., disableVerifySSL where the default is false.
There was a problem hiding this comment.
And after looking a little more closely, I don't see anywhere in the code where we actually set the default to true. Let's change this to disableVerifySSL instead as mentioned here: https://github.com/rook/rook/pull/8426/files#r740571718. For all protocol types where verifySSL is supposed to be default true.
|
will squash the last 2 commit after final approval |
@BlaineEXE I changed that for the case that the case that the error is "not found", but kept regular failure with an error and without a delay in case of other errors. please let me know if this is ok? |
For now, I think it looks good. I'd like to see us use |
| # use one of the following endpoints | ||
| endpoint: http://my-notification-endpoint:8080 [3] |
There was a problem hiding this comment.
Is there a tab/space issue happening here? I usually have the VSCode option "Render Whitespace" set to "boundary".
| endpoint: http://my-notification-endpoint:8080 [3] | ||
| # endpoint: http://my-notification-endpoint:8080/my-topic | ||
| # endpoint: https://my-notification-endpoint:8443 | ||
| # endpoint: amqp://my-rabbitmq-service:5672/vhost1 | ||
| # endpoint: amqps://my-rabbitmq-service:5671/vhost1 | ||
| # endpoint: kafka://my-kafka-service:9092 | ||
| objectStoreName: my-store [4] | ||
| objectStoreNamespace: rook-ceph [5] | ||
| opaqueData: my@email.com [6] | ||
| persistent: false [7] | ||
| http: [8] | ||
| disableVerifySSL: true [9] | ||
| # amqp: [10] | ||
| # disableVerifySSL: true [11] | ||
| # ackLevel: broker [12] | ||
| # exchange: my-exchange [13] | ||
| # kafka: [14] | ||
| # disableVerifySSL: true [15] | ||
| # ackLevel: broker [16] | ||
| # useSSL: false [17] |
There was a problem hiding this comment.
Thanks for taking care of moving the endpoint spec around. Don't forget to change the location of endpoint in the doc here. It's going to mean we have to re-order the stuff below also.
While we are doing that, I think it would be better to explicitly number the markdown (below) so that we can cross-reference easily using the raw text in addition to the rendered text.
I see you already changed the design/ document, which is good. 👍
There was a problem hiding this comment.
my mistake. forgot to change the doc (just changed the design)
pkg/apis/ceph.rook.io/v1/types.go
Outdated
| type AMQPEndpointSpec struct { | ||
| // The URI of the AMQP endpoint to push notification to | ||
| // +kubebuilder:validation:MinLength=1 | ||
| URI string `json:"endpoint"` |
There was a problem hiding this comment.
I see uri in the design. Should this be json:"uri", or should the design change?
pkg/apis/ceph.rook.io/v1/types.go
Outdated
| type KafkaEndpointSpec struct { | ||
| // The URI of the Kafka endpoint to push notification to | ||
| // +kubebuilder:validation:MinLength=1 | ||
| URI string `json:"endpoint"` |
There was a problem hiding this comment.
I see uri in the design. Should this be json:"uri", or should the design change?
if we are waiting for the user to create the topic, we will use |
following the design from here: https://github.com/rook/rook/blob/master/design/ceph/object/ceph-bucket-notification-crd.md Closes: rook#5313 Signed-off-by: Yuval Lifshitz <ylifshit@redhat.com>
yuvalif
force-pushed
the
ceph-bucket-notifications
branch
from
1b58cc3
to
71ed45b
Compare
I think that is a good approach. Thanks. |
this should make sure code reuse between the OBC label controller and the CephBucketNotification controller. done as part of the cleanup work from: rook#8426 Signed-off-by: Yuval Lifshitz <ylifshit@redhat.com>
this should make sure code reuse between the OBC label controller and the CephBucketNotification controller. done as part of the cleanup work from: rook#8426 Signed-off-by: Yuval Lifshitz <ylifshit@redhat.com>
this should make sure code reuse between the OBC label controller and the CephBucketNotification controller. done as part of the cleanup work from: rook#8426 Signed-off-by: Yuval Lifshitz <ylifshit@redhat.com>
this should make sure code reuse between the OBC label controller and the CephBucketNotification controller. done as part of the cleanup work from: rook#8426 this also include adding unit tests for: - topic controller - notification controller - obc label controller Signed-off-by: Yuval Lifshitz <ylifshit@redhat.com>
this should make sure code reuse between the OBC label controller and the CephBucketNotification controller. done as part of the cleanup work from: #8426 this also include adding unit tests for: - topic controller - notification controller - obc label controller Signed-off-by: Yuval Lifshitz <ylifshit@redhat.com> (cherry picked from commit f2b065a)
this should make sure code reuse between the OBC label controller and the CephBucketNotification controller. done as part of the cleanup work from: #8426 this also include adding unit tests for: - topic controller - notification controller - obc label controller Signed-off-by: Yuval Lifshitz <ylifshit@redhat.com> (cherry picked from commit f2b065a)
this should make sure code reuse between the OBC label controller and the CephBucketNotification controller. done as part of the cleanup work from: rook#8426 this also include adding unit tests for: - topic controller - notification controller - obc label controller Signed-off-by: Yuval Lifshitz <ylifshit@redhat.com>
this should make sure code reuse between the OBC label controller and the CephBucketNotification controller. done as part of the cleanup work from: rook#8426 this also include adding unit tests for: - topic controller - notification controller - obc label controller Signed-off-by: Yuval Lifshitz <ylifshit@redhat.com>
This PR implements the following design
Resolves #5313
Signed-off-by: Yuval Lifshitz ylifshit@redhat.com
TODOs:
topicARNfrom annotation to statusCephObjectStores()?Notes:
Out of Scope TODOs:
Checklist:
make codegen) has been run to update object specifications, if necessary.