The cloud-event-proxy project provides a mechanism for events from the K8s infrastructure to be delivered to CNFs with low-latency.
The initial event functionality focuses on the operation of the PTP synchronization protocol, but the mechanism can be extended for any infrastructure event that requires low-latency.
The mechanism is an integral part of k8s/OCP RAN deployments where the PTP protocol is used to provide timing synchronization for the RAN software elements
Create Publisher Resource: JSON request
{
"Resource": "/east-edge-10/vdu3/o-ran-sync/sync-group/sync-status/sync-state",
"UriLocation": "http://localhost:8080/ack/event"
}
Create Publisher Resource: JSON response
{
"Id": "789be75d-7ac3-472e-bbbc-6d62878aad4a",
"Resource": "/east-edge-10/vdu3/o-ran-sync/sync-group/sync-status/sync-state",
"UriLocation": "http://localhost:9090/ack/event" ,
"EndpointUri ": "http://localhost:8080/api/ocloudNotifications/v1/publishers/{publisherid}"
}
package main
import (
v1pubsub "github.com/redhat-cne/sdk-go/v1/pubsub"
v1amqp "github.com/redhat-cne/sdk-go/v1/amqp"
"github.com/redhat-cne/sdk-go/pkg/types"
)
func main(){
//channel for the transport handler subscribed to get and set events
eventInCh := make(chan *channel.DataChan, 10)
pubSubInstance = v1pubsub.GetAPIInstance(".")
endpointURL := &types.URI{URL: url.URL{Scheme: "http", Host: "localhost:8080", Path: fmt.Sprintf("%s%s", apiPath, "dummy")}}
// create publisher
pub, err := pubSubInstance.CreatePublisher(v1pubsub.NewPubSub(endpointURL, "test/test"))
// once the publisher response is received, create a transport sender object to send events.
if err==nil{
v1amqp.CreateSender(eventInCh, pub.GetResource())
}
}
Create Subscription Resource: JSON request
{
"Resource": "/east-edge-10/vdu3/o-ran-sync/sync-group/sync-status/sync-state",
"UriLocation”: “http://localhost:9090/event"
}
Example Create Subscription Resource: JSON response
{
"Id": "789be75d-7ac3-472e-bbbc-6d62878aad4a",
"Resource": "/east-edge-10/vdu3/o-ran-sync/sync-group/sync-status/sync-state",
"UriLocation": "http://localhost:9090/ack/event",
"EndpointUri": "http://localhost:8080/api/ocloudNotifications/v1/subscriptions/{subscriptionid}"
}
package main
import (
v1pubsub "github.com/redhat-cne/sdk-go/v1/pubsub"
v1amqp "github.com/redhat-cne/sdk-go/v1/amqp"
"github.com/redhat-cne/sdk-go/pkg/types"
)
func main(){
//channel for the transport handler subscribed to get and set events
eventInCh := make(chan *channel.DataChan, 10)
pubSubInstance = v1pubsub.GetAPIInstance(".")
endpointURL := &types.URI{URL: url.URL{Scheme: "http", Host: "localhost:8080", Path: fmt.Sprintf("%s%s", apiPath, "dummy")}}
// create subscription
pub, err := pubSubInstance.CreateSubscription(v1pubsub.NewPubSub(endpointURL, "test/test"))
// once the subscription response is received, create a transport listener object to receive events.
if err==nil{
v1amqp.CreateListener(eventInCh, pub.GetResource())
}
}
Cloud-Event-Proxy container running with rest api plugin will be running a webservice and exposing following end points.
POST /api/cloudNotifications/v1/subscriptions
POST /api/cloudNotifications/v1/publishers
GET /api/cloudNotifications/v1/subscriptions
GET /api/cloudNotifications/v1/publishers
GET /api/cloudNotifications/v1/subscriptions/$subscriptionid
GET /api/cloudNotifications/v1/publishers/$publisherid
GET /api/cloudNotifications/v1/health
POST /api/cloudNotifications/v1/log
POST /api/cloudNotifications/v1/create/event
// create publisher
pub, err := pubSubInstance.CreatePublisher(v1pubsub.NewPubSub(endpointURL, "test/test"))
//create subscription
sub, err := pubSubInstance.CreateSubscription(v1pubsub.NewPubSub(endpointURL, "test/test"))
// 1.Create Status Listener Fn (onStatusRequestFn is action to be performed on status ping received)
onStatusRequestFn := func(e v2.Event) error {
log.Printf("got status check call,fire events for above publisher")
event, _ := createPTPEvent(pub)
_ = common.PublishEvent(config, event)
return nil
}
// 2. Create Listener object
v1amqp.CreateNewStatusListener(config.EventInCh, fmt.Sprintf("%s/%s", pub.Resource, "status"), onStatusRequestFn, nil)
package main
import (
v1pubsub "github.com/redhat-cne/sdk-go/v1/pubsub"
v1amqp "github.com/redhat-cne/sdk-go/v1/amqp"
)
v1amqp.CreateSender(eventInCh, pub.GetResource())
package main
import (
v1pubsub "github.com/redhat-cne/sdk-go/v1/pubsub"
v1amqp "github.com/redhat-cne/sdk-go/v1/amqp"
)
v1amqp.CreateListener(eventInCh, sub.GetResource())
The following example shows a Cloud Native Events serialized as JSON: (Following json should be validated with Cloud native events' event_spec.json schema)
{
"id": "5ce55d17-9234-4fee-a589-d0f10cb32b8e",
"type": "event.synchronization-state-chang",
"time": "2021-02-05T17:31:00Z",
"data": {
"version": "v1.0",
"values": [{
"resource": "/cluster/node/ptp",
"data_type": "notification",
"value_type": "enumeration",
"value": "ACQUIRING-SYNC"
}, {
"resource": "/cluster/node/clock",
"data_type": "metric",
"value_type": "decimal64.3",
"value": 100.3
}]
}
}
Event can be created via rest-api or calling sdk methods
To produce or consume an event, the producer and consumer should have created publisher and subscription objects
and should have access to the id
of the publisher/subscription data objects.
import (
v1event "github.com/redhat-cne/sdk-go/v1/event"
cneevent "github.com/redhat-cne/sdk-go/pkg/event"
)
// create an event
event := v1event.CloudNativeEvent()
event.SetID(pub.ID)
event.Type = "ptp_status_type"
event.SetTime(types.Timestamp{Time: time.Now().UTC()}.Time)
event.SetDataContentType(cneevent.ApplicationJSON)
data := cneevent.Data{
Version: "v1",
Values: []cneevent.DataValue{
{
Resource: "/cluster/node/ptp",
DataType: cneevent.NOTIFICATION,
ValueType: cneevent.ENUMERATION,
Value: cneevent.GNSS_ACQUIRING_SYNC,
},
},
}
data.SetVersion("v1")
event.SetData(data)
cloudEvent, _ := v1event.CreateCloudEvents(event, pub)
//send event to AMQP (rest API does this action by default)
v1event.SendNewEventToDataChannel(eventInCh, pub.Resource, cloudEvent)
//POST /api/ocloudNotifications/v1/create/event
if pub,err:=pubSubInstance.GetPublisher(publisherID);err==nil {
url = fmt.SPrintf("%s%s", server.HostPath, "create/event")
restClient.PostEvent(pub.EndPointURI.String(), event)
}
Cloud native events sdk-go comes with following metrics collectors .
- Number of events received by the transport
- Number of events published by the transport.
- Number of connection resets.
- Number of sender created
- Number of receiver created
Cloud native events rest API comes with following metrics collectors .
- Number of events published by the rest api.
- Number of active subscriptions.
- Number of active publishers.
- Number of events produced.
- Number of events received.