diff --git a/README.md b/README.md index 7b10e22ba0..cac6006cd9 100644 --- a/README.md +++ b/README.md @@ -13,7 +13,6 @@ that: - Integrate with Apache CouchDB - Integrate with Apache Kafka - Integrate with AWS SQS -- Integrate with Ceph - Integrate with GitHub - Integrate with GitLab - Integrate with NATS Streaming diff --git a/ceph/README.md b/ceph/README.md deleted file mode 100644 index e300023e82..0000000000 --- a/ceph/README.md +++ /dev/null @@ -1,91 +0,0 @@ -# Ceph to Knative Receive Adapters - -The following receive adapters convert bucket notifications from -[Ceph format](https://docs.ceph.com/docs/master/radosgw/notifications/#events) -into CloudEvents format, and inject them into Knative. Adapter logic follow the -one described for -[AWS S3](https://github.com/cloudevents/spec/blob/master/adapters/aws-s3.md) -bucket notifications. - -There are 4 different transport options: - -## HTTP - -This receive adapter expects bucket notifications from Ceph over HTTP transport, -as payload in POST messages. - -> Note that this adapted does not assume the CloudEvents HTTP binding in the -> incoming messages. - -### Testing - -- Create a namespace for the tests: - -``` -kubectl apply -f samples/ceph-eventing-ns.yaml -``` - -- Deploy a service for the bucket notification messages coming from Ceph: - -``` -kubectl apply -f samples/ceph-bucket-notification-svc.yaml -n ceph-eventing -``` - -- Deploy the ceph-event-display Knative service together with a channel and - subscribe it to the channel: - -``` -ko apply -f samples/ceph-display-resources.yaml -n ceph-eventing -``` - -- Build and deploy the Ceph source adapter with SinkBinding to inject the sink: - -``` -ko apply -f samples/ceph-source.yaml -n ceph-eventing -``` - -- Deploy a test pod that has cURL installed and a JSON file with bucket - notifications (names `records.json`): - -``` -kubectl apply -f samples/test-pod.yaml -n ceph-eventing -``` - -- Execute cURL command on the test pod to send the JSON bucket notifications to - the ceph-bucket-notifications service: - -``` -kubectl exec test -n ceph-eventing -- curl -d "@records.json" -X POST ceph-bucket-notifications.ceph-eventing.svc.cluster.local -``` - -- To verify that the events reached the ceph-event-display service, call: - -``` -kubectl logs -l serving.knative.dev/service=ceph-event-display -n ceph-eventing -c ceph-display-container --tail=100 -``` - -## PubSub (TODO) - -This receive adapter is pulling bucket notifications from a special "pubsub" -zone in Ceph, and acking them once they are successfully sent to Knative. More -information about this mechanism could be found -[here](https://docs.ceph.com/docs/master/radosgw/pubsub-module/). - -> Note that this adapter needs access to the Ceph cluster to pull the -> notifications and ack them. - -## AMQP0.9.1 (TODO) - -This receive adapter is subscribed to a list of AMQP topics, on which bucket -notifications are expected. - -> Note that this adapted does not assume the CloudEvents AMQP binding in the -> incoming messages. - -## Kafka (TODO) - -This receive adapter is subscribed to a list of Kafka topics, on which bucket -notifications are expected. - -> Note that this adapted does not assume the CloudEvents Kafka binding in the -> incoming messages. diff --git a/ceph/cmd/receive_adapter/main.go b/ceph/cmd/receive_adapter/main.go deleted file mode 100644 index 1e4e137815..0000000000 --- a/ceph/cmd/receive_adapter/main.go +++ /dev/null @@ -1,27 +0,0 @@ -/* -Copyright 2019 The Knative Authors - -Licensed under the Apache License, Version 2.0 (the "License"); -you may not use this file except in compliance with the License. -You may obtain a copy of the License at - - http://www.apache.org/licenses/LICENSE-2.0 - -Unless required by applicable law or agreed to in writing, software -distributed under the License is distributed on an "AS IS" BASIS, -WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -See the License for the specific language governing permissions and -limitations under the License. -*/ - -package main - -import ( - "knative.dev/eventing/pkg/adapter/v2" - - cephadapter "knative.dev/eventing-contrib/ceph/pkg/adapter" -) - -func main() { - adapter.Main("cephsource", cephadapter.NewEnvConfig, cephadapter.NewAdapter) -} diff --git a/ceph/pkg/adapter/adapter.go b/ceph/pkg/adapter/adapter.go deleted file mode 100644 index 61f833bd88..0000000000 --- a/ceph/pkg/adapter/adapter.go +++ /dev/null @@ -1,144 +0,0 @@ -/* -Copyright 2019 The Knative Authors - -Licensed under the Apache License, Version 2.0 (the "License"); -you may not use this file except in compliance with the License. -You may obtain a copy of the License at - - http://www.apache.org/licenses/LICENSE-2.0 - -Unless required by applicable law or agreed to in writing, software -distributed under the License is distributed on an "AS IS" BASIS, -WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -See the License for the specific language governing permissions and -limitations under the License. -*/ - -package adapter - -import ( - "context" - "encoding/json" - "fmt" - "io/ioutil" - "net/http" - "time" - - cloudevents "github.com/cloudevents/sdk-go/v2" - "go.uber.org/zap" - ceph "knative.dev/eventing-contrib/ceph/pkg/apis/v1alpha1" - "knative.dev/eventing/pkg/adapter/v2" - "knative.dev/pkg/logging" -) - -type envConfig struct { - adapter.EnvConfig - - // Port to listen incoming connections - Port string `envconfig:"PORT" default:"8080"` -} - -// cephReceiveAdapter converts incoming Ceph notifications to -// CloudEvents and then sends them to the specified Sink -type cephReceiveAdapter struct { - logger *zap.SugaredLogger - client cloudevents.Client - port string -} - -// NewEnvConfig function reads env variables defined in envConfig structure and -// returns accessor interface -func NewEnvConfig() adapter.EnvConfigAccessor { - return &envConfig{} -} - -// NewAdapter returns the instance of cephReceiveAdapter that implements adapter.Adapter interface -func NewAdapter(ctx context.Context, processed adapter.EnvConfigAccessor, ceClient cloudevents.Client) adapter.Adapter { - logger := logging.FromContext(ctx) - env := processed.(*envConfig) - - return &cephReceiveAdapter{ - logger: logger, - client: ceClient, - port: env.Port, - } -} - -// Start the ceph bucket notifications to knative adapter -func (ca *cephReceiveAdapter) Start(ctx context.Context) error { - return ca.start(ctx.Done()) -} - -func (ca *cephReceiveAdapter) start(stopCh <-chan struct{}) error { - http.HandleFunc("/", ca.postHandler) - go http.ListenAndServe(":"+ca.port, nil) - ca.logger.Info("Ceph to Knative adapter spawned HTTP server") - <-stopCh - - ca.logger.Info("Ceph to Knative adapter terminated") - return nil -} - -// postMessage convert bucket notifications to knative events and sent them to knative -func (ca *cephReceiveAdapter) postMessage(notification ceph.BucketNotification) error { - eventTime, err := time.Parse(time.RFC3339, notification.EventTime) - if err != nil { - ca.logger.Infof("Failed to parse event timestamp, using local time. Error: %s", err.Error()) - eventTime = time.Now() - } - - event := cloudevents.NewEvent() - event.SetID(notification.ResponseElements.XAmzRequestID + notification.ResponseElements.XAmzID2) - event.SetSource(notification.EventSource + "." + notification.AwsRegion + "." + notification.S3.Bucket.Name) - event.SetType("com.amazonaws." + notification.EventName) - event.SetSubject(notification.S3.Object.Key) - event.SetTime(eventTime) - err = event.SetData(cloudevents.ApplicationJSON, notification) - if err != nil { - return fmt.Errorf("failed to marshal event data: %w", err) - } - - ca.logger.Infof("Sending CloudEvent: %v", event) - - result := ca.client.Send(context.Background(), event) - if !cloudevents.IsACK(result) { - return result - } - return nil -} - -// postHandler handles incoming bucket notifications from ceph -func (ca *cephReceiveAdapter) postHandler(w http.ResponseWriter, r *http.Request) { - if r.Method != "POST" { - ca.logger.Infof("%s method not allowed", r.Method) - http.Error(w, "405 Method Not Allowed", http.StatusBadRequest) - return - } - - body, err := ioutil.ReadAll(r.Body) - if err != nil { - ca.logger.Infof("Error reading message body: %s", err.Error()) - http.Error(w, err.Error(), http.StatusBadRequest) - return - } - - var notifications ceph.BucketNotifications - err = json.Unmarshal(body, ¬ifications) - - if err != nil { - ca.logger.Infof("Failed to parse JSON: %s", err.Error()) - http.Error(w, err.Error(), http.StatusBadRequest) - return - } - ca.logger.Infof("%d events found in message", len(notifications.Records)) - for _, notification := range notifications.Records { - ca.logger.Infof("Received Ceph bucket notification: %+v", notification) - if err := ca.postMessage(notification); err == nil { - ca.logger.Infof("Event %s was successfully posted to knative", notification.EventID) - } else { - ca.logger.Infof("Failed to post event %s: %s", notification.EventID, err.Error()) - http.Error(w, err.Error(), http.StatusBadRequest) - return - } - } -} diff --git a/ceph/pkg/adapter/adapter_test.go b/ceph/pkg/adapter/adapter_test.go deleted file mode 100644 index 898058b6a0..0000000000 --- a/ceph/pkg/adapter/adapter_test.go +++ /dev/null @@ -1,171 +0,0 @@ -/* -Copyright 2019 The Knative Authors - -Licensed under the Apache License, Version 2.0 (the "License"); -you may not use this file except in compliance with the License. -You may obtain a copy of the License at - - http://www.apache.org/licenses/LICENSE-2.0 - -Unless required by applicable law or agreed to in writing, software -distributed under the License is distributed on an "AS IS" BASIS, -WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -See the License for the specific language governing permissions and -limitations under the License. -*/ - -package adapter - -import ( - "bytes" - "context" - "encoding/json" - "io/ioutil" - "net/http" - "net/http/httptest" - "sync" - "testing" - "time" - - cloudevents "github.com/cloudevents/sdk-go/v2" - "go.uber.org/zap" - ceph "knative.dev/eventing-contrib/ceph/pkg/apis/v1alpha1" - "knative.dev/eventing/pkg/adapter/v2" - adaptertest "knative.dev/eventing/pkg/adapter/v2/test" - "knative.dev/pkg/logging" - pkgtesting "knative.dev/pkg/reconciler/testing" -) - -var notification1 = ceph.BucketNotification{ - EventVersion: "2.1", - EventSource: "ceph:s3", - AwsRegion: "tenantA", - EventTime: "2019-11-22T13:47:35.124724Z", - EventName: "s3:ObjectCreated:Put", - UserIdentity: ceph.UserIdentitySpec{ - PrincipalID: "tester", - }, - RequestParameters: ceph.RequestParametersSpec{ - SourceIPAddress: "", - }, - ResponseElements: ceph.ResponseElementsSpec{ - XAmzRequestID: "503a4c37-85eb-47cd-8681-2817e80b4281.5330.903595", - XAmzID2: "14d2-a1-a", - }, - S3: ceph.S3Spec{ - S3SchemaVersion: "1.0", - ConfigurationID: "fishbucket_notifications", - Bucket: ceph.BucketSpec{ - Name: "fishbucket", - OwnerIdentity: ceph.OwnerIdentitySpec{ - PrincipalID: "tester", - }, - Arn: "arn:aws:s3:::fishbucket", - ID: "503a4c37-85eb-47cd-8681-2817e80b4281.5332.38", - }, - Object: ceph.ObjectSpec{ - Key: "fish9.jpg", - Size: 1024, - ETag: "37b51d194a7513e45b56f6524f2d51f2", - VersionID: "", - Sequencer: "F7E6D75DC742D108", - Metadata: []ceph.MetadataEntry{ - {Key: "x-amz-meta-meta1", Value: "This is my metadata value"}, - {Key: "x-amz-meta-meta2", Value: "This is another metadata value"}, - }, - }, - }, - EventID: "1575221657.102001.80dad3aad8584778352c68ab06250327", -} - -var jsonData = ceph.BucketNotifications{ - Records: []ceph.BucketNotification{ - notification1, - }, -} - -func TestRecords(t *testing.T) { - var sinkServer *httptest.Server - defer func() { - if sinkServer != nil { - sinkServer.Close() - } - }() - jsonBuffer, err := json.Marshal(jsonData) - if err != nil { - t.Error(err) - } - notification1Buffer, err := json.Marshal(notification1) - if err != nil { - t.Error(err) - } - - var ca *cephReceiveAdapter - var mu sync.Mutex - ctx, cancel := context.WithCancel(context.Background()) - - go func() { - sinkServer = httptest.NewServer(&fakeSink{ - t: t, - expectedBody: string(notification1Buffer), - }, - ) - mu.Lock() - ca = newTestAdapter(t, adaptertest.NewTestClient(), sinkServer.URL) - mu.Unlock() - err = ca.Start(ctx) - if err != nil { - t.Error(err) - } - }() - time.Sleep(5 * time.Second) - - mu.Lock() - sinkPort := ca.port - mu.Unlock() - - result, err := http.Post("http://:"+sinkPort+"/", "application/json", bytes.NewBuffer(jsonBuffer)) - if err != nil { - t.Error(err) - } - defer result.Body.Close() - _, err = ioutil.ReadAll(result.Body) - if err != nil { - t.Error(err) - } - cancel() -} - -type fakeSink struct { - t *testing.T - expectedBody string -} - -func (h *fakeSink) ServeHTTP(w http.ResponseWriter, r *http.Request) { - defer r.Body.Close() - body, err := ioutil.ReadAll(r.Body) - if err != nil { - h.t.Error(err) - w.WriteHeader(http.StatusBadRequest) - return - } - if string(body) != h.expectedBody { - h.t.Error("Notification Body Mismatch") - } - w.WriteHeader(http.StatusOK) -} - -func newTestAdapter(t *testing.T, ce cloudevents.Client, sinkServerURL string) *cephReceiveAdapter { - env := envConfig{ - EnvConfig: adapter.EnvConfig{ - Namespace: "default", - }, - Port: "28080", - } - ctx, _ := pkgtesting.SetupFakeContext(t) - logger := zap.NewExample().Sugar() - ctx = logging.WithLogger(ctx, logger) - ctx = cloudevents.ContextWithTarget(ctx, sinkServerURL) - - return NewAdapter(ctx, &env, ce).(*cephReceiveAdapter) -} diff --git a/ceph/pkg/apis/v1alpha1/bucket_notification_types.go b/ceph/pkg/apis/v1alpha1/bucket_notification_types.go deleted file mode 100644 index f94a13bdba..0000000000 --- a/ceph/pkg/apis/v1alpha1/bucket_notification_types.go +++ /dev/null @@ -1,79 +0,0 @@ -/* -Copyright 2019 The Knative Authors - -Licensed under the Apache License, Version 2.0 (the "License"); -you may not use this file except in compliance with the License. -You may obtain a copy of the License at - - http://www.apache.org/licenses/LICENSE-2.0 - -Unless required by applicable law or agreed to in writing, software -distributed under the License is distributed on an "AS IS" BASIS, -WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -See the License for the specific language governing permissions and -limitations under the License. -*/ - -package v1alpha1 - -type RequestParametersSpec struct { - SourceIPAddress string `json:"sourceIPAddress"` -} - -type ResponseElementsSpec struct { - XAmzRequestID string `json:"x-amz-request-id"` - XAmzID2 string `json:"x-amz-id-2"` -} - -type UserIdentitySpec struct { - PrincipalID string `json:"principalId"` -} - -type OwnerIdentitySpec struct { - PrincipalID string `json:"principalId"` -} - -type BucketSpec struct { - Name string `json:"name"` - OwnerIdentity OwnerIdentitySpec `json:"ownerIdentity"` - Arn string `json:"arn"` - ID string `json:"id"` -} - -type MetadataEntry struct { - Key string `json:"key"` - Value string `json:"value"` -} - -type ObjectSpec struct { - Key string `json:"key"` - Size uint `json:"size"` - ETag string `json:"eTag"` - VersionID string `json:"versionId"` - Sequencer string `json:"sequencer"` - Metadata []MetadataEntry `json:"metadata"` -} - -type S3Spec struct { - S3SchemaVersion string `json:"s3SchemaVersion"` - ConfigurationID string `json:"configurationId"` - Bucket BucketSpec `json:"bucket"` - Object ObjectSpec `json:"object"` -} - -type BucketNotification struct { - EventVersion string `json:"eventVersion"` - EventSource string `json:"eventSource"` - AwsRegion string `json:"awsRegion"` - EventTime string `json:"eventTime"` - EventName string `json:"eventName"` - UserIdentity UserIdentitySpec `json:"userIdentity"` - RequestParameters RequestParametersSpec `json:"requestParameters"` - ResponseElements ResponseElementsSpec `json:"responseElements"` - S3 S3Spec `json:"s3"` - EventID string `json:"eventId"` -} - -type BucketNotifications struct { - Records []BucketNotification `json:"Records"` -} diff --git a/ceph/samples/ceph-bucket-notification-svc.yaml b/ceph/samples/ceph-bucket-notification-svc.yaml deleted file mode 100644 index 706fb1590a..0000000000 --- a/ceph/samples/ceph-bucket-notification-svc.yaml +++ /dev/null @@ -1,11 +0,0 @@ -apiVersion: v1 -kind: Service -metadata: - name: ceph-bucket-notifications -spec: - selector: - app: ceph-source - ports: - - protocol: TCP - port: 80 - targetPort: 19090 diff --git a/ceph/samples/ceph-display-resources.yaml b/ceph/samples/ceph-display-resources.yaml deleted file mode 100644 index 2a6ac471b5..0000000000 --- a/ceph/samples/ceph-display-resources.yaml +++ /dev/null @@ -1,34 +0,0 @@ -apiVersion: messaging.knative.dev/v1beta1 -kind: Channel -metadata: - name: ceph-event-channel -spec: - channelTemplate: - apiVersion: messaging.knative.dev/v1beta1 - kind: InMemoryChannel ---- -apiVersion: messaging.knative.dev/v1beta1 -kind: Subscription -metadata: - name: ceph-source-display -spec: - channel: - apiVersion: messaging.knative.dev/v1beta1 - kind: Channel - name: ceph-event-channel - subscriber: - ref: - apiVersion: serving.knative.dev/v1 - kind: Service - name: ceph-event-display ---- -apiVersion: serving.knative.dev/v1 -kind: Service -metadata: - name: ceph-event-display -spec: - template: - spec: - containers: - - name: ceph-display-container - image: ko://knative.dev/eventing-contrib/cmd/event_display diff --git a/ceph/samples/ceph-eventing-ns.yaml b/ceph/samples/ceph-eventing-ns.yaml deleted file mode 100644 index 71336e81af..0000000000 --- a/ceph/samples/ceph-eventing-ns.yaml +++ /dev/null @@ -1,4 +0,0 @@ -apiVersion: v1 -kind: Namespace -metadata: - name: ceph-eventing diff --git a/ceph/samples/ceph-source.yaml b/ceph/samples/ceph-source.yaml deleted file mode 100644 index b009c005c6..0000000000 --- a/ceph/samples/ceph-source.yaml +++ /dev/null @@ -1,50 +0,0 @@ -apiVersion: apps/v1 -kind: Deployment -metadata: - name: ceph-bucket-notification-source - labels: - app: ceph-source -spec: - selector: - matchLabels: - app: ceph-source - template: - metadata: - labels: - app: ceph-source - spec: - containers: - - name: ceph-source - image: ko://knative.dev/eventing-contrib/ceph/cmd/receive_adapter - env: - - name: PORT - value: "19090" - - name: NAMESPACE - valueFrom: - fieldRef: - fieldPath: metadata.namespace - - name: METRICS_DOMAIN - value: "knative.dev/sources" - - name: K_METRICS_CONFIG - - name: K_LOGGING_CONFIG - ports: - - containerPort: 19090 - ---- - -apiVersion: sources.knative.dev/v1alpha2 -kind: SinkBinding -metadata: - name: ceph-source-sinkbinding -spec: - subject: - apiVersion: apps/v1 - kind: Deployment - selector: - matchLabels: - app: ceph-source - sink: - ref: - apiVersion: messaging.knative.dev/v1beta1 - kind: Channel - name: ceph-event-channel diff --git a/ceph/samples/records.json b/ceph/samples/records.json deleted file mode 100644 index 992385824e..0000000000 --- a/ceph/samples/records.json +++ /dev/null @@ -1,118 +0,0 @@ -{"Records": - [{ - "eventVersion":"2.1", - "eventSource":"ceph:s3", - "awsRegion":"tenantA", - "eventTime":"2019-11-22T13:47:35.124724Z", - "eventName":"s3:ObjectCreated:Put", - "userIdentity":{ - "principalId":"tester" - }, - "requestParameters":{ - "sourceIPAddress":"" - }, - "responseElements":{ - "x-amz-request-id":"503a4c37-85eb-47cd-8681-2817e80b4281.5330.903595", - "x-amz-id-2":"14d2-a1-a" - }, - "s3":{ - "s3SchemaVersion":"1.0", - "configurationId":"ypqmbg-1_notif", - "bucket":{ - "name":"ypqmbg-1", - "ownerIdentity":{ - "principalId":"tester" - }, - "arn":"arn:aws:s3:::ypqmbg-1", - "id":"503a4c37-85eb-47cd-8681-2817e80b4281.5332.38" - }, - "object":{ - "key":"9", - "size":3, - "etag":"37b51d194a7513e45b56f6524f2d51f2", - "versionId":"", - "sequencer":"F7E6D75DC742D108", - "metadata":[] - } - }, - "eventId":"hello-world" - }, - { - "eventVersion":"2.1", - "eventSource":"aws:s3", - "awsRegion":"", - "eventTime":"2019-11-22T13:47:35.124724Z", - "eventName":"s3:ObjectCreated:Post", - "userIdentity":{ - "principalId":"tester" - }, - "requestParameters":{ - "sourceIPAddress":"" - }, - "responseElements":{ - "x-amz-request-id":"503a4c37-85eb-47cd-8681-2817e80b4281.5330.903595", - "x-amz-id-2":"14d2-a1-a" - }, - "s3":{ - "s3SchemaVersion":"1.0", - "configurationId":"ypqmbg-1_notif", - "bucket":{ - "name":"ypqmbg-1", - "ownerIdentity":{ - "principalId":"tester" - }, - "arn":"arn:aws:s3:::ypqmbg-1", - "id":"503a4c37-85eb-47cd-8681-2817e80b4281.5332.38" - }, - "object":{ - "key":"9", - "size":3, - "etag":"37b51d194a7513e45b56f6524f2d51f2", - "versionId":"", - "sequencer":"F7E6D75DC742D109", - "metadata":[] - } - }, - "eventId":"kaboom" - }, - { - "eventVersion":"2.2", - "eventSource":"aws:s3", - "awsRegion":"", - "eventTime":"2019-12-01T17:34:17.092904Z", - "eventName":"s3:ObjectCreated:Put", - "userIdentity":{ - "principalId":"tester" - }, - "requestParameters":{ - "sourceIPAddress":"" - }, - "responseElements":{ - "x-amz-request-id":"7387828f-b5e8-46a7-8aef-d02d2b5b759b.4144.7224", - "x-amz-id-2":"1030-a1-a" - }, - "s3":{ - "s3SchemaVersion":"1.0", - "configurationId":"tninfx-1_notif", - "bucket":{ - "name":"tninfx-1", - "ownerIdentity":{ - "principalId":"tester" - }, - "arn":"arn:aws:s3:::tninfx-1", - "id":"7387828f-b5e8-46a7-8aef-d02d2b5b759b.4146.18"}, - "object":{ - "key":"foo", - "size":36, - "etag":"80dad3aad8584778352c68ab06250327", - "versionId":"", - "sequencer":"99F9E35D526C1406", - "metadata":[ - {"key":"x-amz-meta-meta1","val":"This is my metadata value"}, - {"key":"x-amz-meta-meta2","val":"This is another metadata value"} - ] - } - }, - "eventId":"1575221657.102001.80dad3aad8584778352c68ab06250327"} - ] -} diff --git a/ceph/samples/test-pod.yaml b/ceph/samples/test-pod.yaml deleted file mode 100644 index c9b584728b..0000000000 --- a/ceph/samples/test-pod.yaml +++ /dev/null @@ -1,8 +0,0 @@ -apiVersion: v1 -kind: Pod -metadata: - name: test -spec: - containers: - - name: curl-runner - image: docker.io/yuvalif/test-ceph-notifications