[VOL-1024] This update consists of:
1) Inter-container kafka library in Go
2) initial set of proto definitions
3) Test cases for the kafka library
Change-Id: Ibe8b0f673a90bbe4cb92847ce40f31ec2d0b6244
diff --git a/.gitignore b/.gitignore
index 7182935..2b95835 100644
--- a/.gitignore
+++ b/.gitignore
@@ -19,6 +19,7 @@
# Protobuf output files
**/*_pb2.py
**/*_pb2_grpc.py
+**/*.pb.go
# Editors
*.bak
diff --git a/common/log/log.go b/common/log/log.go
index e9f3c1b..d751c34 100644
--- a/common/log/log.go
+++ b/common/log/log.go
@@ -1,7 +1,6 @@
package log
import (
- "errors"
"fmt"
zp "go.uber.org/zap"
zc "go.uber.org/zap/zapcore"
@@ -148,7 +147,9 @@
// this method will return an error
func GetLogger() (Logger, error) {
if defaultLogger == nil {
- return nil, errors.New("Uninitialized-logger")
+ // Setup the logger with default values - debug level,
+ SetLogger(JSON, 0, Fields{"instanceId": "default-logger"})
+ //return nil, errors.New("Uninitialized-logger")
}
return defaultLogger, nil
}
diff --git a/compose/docker-compose-zk-kafka-test.yml b/compose/docker-compose-zk-kafka-test.yml
new file mode 100644
index 0000000..2cad1b8
--- /dev/null
+++ b/compose/docker-compose-zk-kafka-test.yml
@@ -0,0 +1,27 @@
+version: '2'
+services:
+ #
+ # Single-node zookeeper service
+ #
+ zookeeper:
+ image: "wurstmeister/zookeeper:latest"
+ ports:
+ - 2181:2181
+ environment:
+ SERVICE_2181_NAME: "zookeeper"
+ #
+ # Single-node kafka service
+ #
+ kafka:
+ image: "wurstmeister/kafka:latest"
+ ports:
+ - 9092:9092
+ environment:
+ KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://10.100.198.220:9092
+ KAFKA_LISTENERS: PLAINTEXT://:9092
+# KAFKA_ADVERTISED_HOST_NAME: 10.100.198.220
+ KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
+ KAFKA_AUTO_CREATE_TOPICS_ENABLE: 'true'
+ SERVICE_9092_NAME: "kafka"
+ volumes:
+ - /var/run/docker.sock:/var/run/docker.sock
diff --git a/docker/Dockerfile.rw_core b/docker/Dockerfile.rw_core
index ecc9919..2431fa9 100644
--- a/docker/Dockerfile.rw_core
+++ b/docker/Dockerfile.rw_core
@@ -4,16 +4,30 @@
FROM golang:alpine AS build-env
# Install required packages
-RUN apk add --no-cache wget git make build-base
+RUN apk add --no-cache wget git make build-base protobuf protobuf-dev
# Prepare directory structure
-RUN ["mkdir", "-p", "/src"]
+RUN ["mkdir", "-p", "/src", "src/protos"]
RUN ["mkdir", "-p", "$GOPATH/src", "$GOPATH/pkg", "$GOPATH/bin"]
+RUN ["mkdir", "-p", "$GOPATH/src/github.com/opencord/voltha/protos/go"]
# Copy files
ADD rw_core $GOPATH/src/github.com/opencord/voltha-go/rw_core
ADD common $GOPATH/src/github.com/opencord/voltha-go/common
ADD db $GOPATH/src/github.com/opencord/voltha-go/db
+ADD kafka $GOPATH/src/github.com/opencord/voltha-go/kafka
+
+# Copy required proto files
+# ... VOLTHA proos
+ADD protos/*.proto /src/protos/
+ADD protos/scripts/* /src/protos/
+
+# Install golang protobuf
+RUN go get -u github.com/grpc-ecosystem/grpc-gateway/protoc-gen-grpc-gateway
+RUN go get -u github.com/golang/protobuf/protoc-gen-go
+
+# Compile protobuf files
+RUN sh /src/protos/build_protos.sh /src/protos
# Build rw_core
RUN cd $GOPATH/src/github.com/opencord/voltha-go/rw_core && go get -d ./... && go build -o /src/rw_core
diff --git a/kafka/kafka_inter_container_library.go b/kafka/kafka_inter_container_library.go
new file mode 100644
index 0000000..9db57a0
--- /dev/null
+++ b/kafka/kafka_inter_container_library.go
@@ -0,0 +1,804 @@
+package kafka
+
+import (
+ "context"
+ "errors"
+ "fmt"
+ "github.com/Shopify/sarama"
+ "github.com/golang/protobuf/proto"
+ "github.com/golang/protobuf/ptypes"
+ "github.com/golang/protobuf/ptypes/any"
+ "github.com/google/uuid"
+ "github.com/opencord/voltha-go/common/log"
+ ca "github.com/opencord/voltha-go/protos/core_adapter"
+ "reflect"
+ "sync"
+ "time"
+)
+
+// Initialize the logger - gets the default until the main function setup the logger
+func init() {
+ log.GetLogger()
+}
+
+const (
+ DefaultKafkaHost = "10.100.198.240"
+ DefaultKafkaPort = 9092
+ DefaultTopicName = "Core"
+ DefaultSleepOnError = 1
+ DefaultFlushFrequency = 1
+ DefaultFlushMessages = 1
+ DefaultFlushMaxmessages = 1
+ DefaultMaxRetries = 3
+ DefaultReturnSuccess = false
+ DefaultReturnErrors = true
+ DefaultConsumerMaxwait = 50
+ DefaultMaxProcessingTime = 100
+ DefaultRequestTimeout = 50 // 50 milliseconds
+)
+
+type consumerChannels struct {
+ consumer sarama.PartitionConsumer
+ channels []chan *ca.InterContainerMessage
+}
+
+// KafkaMessagingProxy represents the messaging proxy
+type KafkaMessagingProxy struct {
+ KafkaHost string
+ KafkaPort int
+ DefaultTopic *Topic
+ TargetInterface interface{}
+ producer sarama.AsyncProducer
+ consumer sarama.Consumer
+ doneCh chan int
+ topicToConsumerChannelMap map[string]*consumerChannels
+ transactionIdToChannelMap map[string]chan *ca.InterContainerMessage
+ lockTopicToConsumerChannelMap sync.RWMutex
+ lockTransactionIdToChannelMap sync.RWMutex
+}
+
+type KafkaProxyOption func(*KafkaMessagingProxy)
+
+func KafkaHost(host string) KafkaProxyOption {
+ return func(args *KafkaMessagingProxy) {
+ args.KafkaHost = host
+ }
+}
+
+func KafkaPort(port int) KafkaProxyOption {
+ return func(args *KafkaMessagingProxy) {
+ args.KafkaPort = port
+ }
+}
+
+func DefaultTopic(topic *Topic) KafkaProxyOption {
+ return func(args *KafkaMessagingProxy) {
+ args.DefaultTopic = topic
+ }
+}
+
+func TargetInterface(target interface{}) KafkaProxyOption {
+ return func(args *KafkaMessagingProxy) {
+ args.TargetInterface = target
+ }
+}
+
+func NewKafkaMessagingProxy(opts ...KafkaProxyOption) (*KafkaMessagingProxy, error) {
+ proxy := &KafkaMessagingProxy{
+ KafkaHost: DefaultKafkaHost,
+ KafkaPort: DefaultKafkaPort,
+ DefaultTopic: &Topic{Name: DefaultTopicName},
+ }
+
+ for _, option := range opts {
+ option(proxy)
+ }
+
+ // Create the locks for all the maps
+ proxy.lockTopicToConsumerChannelMap = sync.RWMutex{}
+ proxy.lockTransactionIdToChannelMap = sync.RWMutex{}
+
+ return proxy, nil
+}
+
+func (kp *KafkaMessagingProxy) Start() error {
+ log.Info("Starting-Proxy")
+
+ // Create the Done channel
+ kp.doneCh = make(chan int, 1)
+
+ // Create the Publisher
+ if err := kp.createPublisher(DefaultMaxRetries); err != nil {
+ log.Errorw("Cannot-create-kafka-publisher", log.Fields{"error": err})
+ return err
+ }
+
+ // Create the master consumer
+ if err := kp.createConsumer(DefaultMaxRetries); err != nil {
+ log.Errorw("Cannot-create-kafka-consumer", log.Fields{"error": err})
+ return err
+ }
+
+ // Create the topic to consumer/channel map
+ kp.topicToConsumerChannelMap = make(map[string]*consumerChannels)
+
+ // Create the transactionId to Channel Map
+ kp.transactionIdToChannelMap = make(map[string]chan *ca.InterContainerMessage)
+
+ return nil
+}
+
+func (kp *KafkaMessagingProxy) Stop() {
+ log.Info("Stopping-Proxy")
+ if kp.producer != nil {
+ if err := kp.producer.Close(); err != nil {
+ panic(err)
+ }
+ }
+ if kp.consumer != nil {
+ if err := kp.consumer.Close(); err != nil {
+ panic(err)
+ }
+ }
+ //Close the done channel to close all long processing Go routines
+ close(kp.doneCh)
+}
+
+func (kp *KafkaMessagingProxy) InvokeRPC(ctx context.Context, rpc string, topic *Topic, waitForResponse bool,
+ kvArgs ...*KVArg) (bool, *any.Any) {
+ // Encode the request
+ protoRequest, err := encodeRequest(rpc, topic, kp.DefaultTopic, kvArgs...)
+ if err != nil {
+ log.Warnw("cannot-format-request", log.Fields{"rpc": rpc, "error": err})
+ return false, nil
+ }
+
+ // Subscribe for response, if needed, before sending request
+ var ch <-chan *ca.InterContainerMessage
+ if waitForResponse {
+ var err error
+ if ch, err = kp.subscribeForResponse(*kp.DefaultTopic, protoRequest.Header.Id); err != nil {
+ log.Errorw("failed-to-subscribe-for-response", log.Fields{"error": err, "topic": topic.Name})
+ }
+ }
+
+ // Send request
+ go kp.sendToKafkaTopic(protoRequest, topic)
+
+ if waitForResponse {
+ // if ctx is nil use a default timeout ctx to ensure we do not wait forever
+ var cancel context.CancelFunc
+ if ctx == nil {
+ ctx, cancel = context.WithTimeout(context.Background(), DefaultRequestTimeout*time.Millisecond)
+ defer cancel()
+ }
+
+ // Wait for response as well as timeout or cancellation
+ // Remove the subscription for a response on return
+ defer kp.unSubscribeForResponse(protoRequest.Header.Id)
+ select {
+ case msg := <-ch:
+ log.Debugw("received-response", log.Fields{"rpc": rpc, "msg": msg})
+
+ var responseBody *ca.InterContainerResponseBody
+ var err error
+ if responseBody, err = decodeResponse(msg); err != nil {
+ log.Errorw("decode-response-error", log.Fields{"error": err})
+ }
+ return responseBody.Success, responseBody.Result
+ case <-ctx.Done():
+ log.Debugw("context-cancelled", log.Fields{"rpc": rpc, "ctx": ctx.Err()})
+ // pack the error as proto any type
+ protoError := &ca.Error{Reason: ctx.Err().Error()}
+ var marshalledArg *any.Any
+ if marshalledArg, err = ptypes.MarshalAny(protoError); err != nil {
+ return false, nil // Should never happen
+ }
+ return false, marshalledArg
+ case <-kp.doneCh:
+ log.Infow("received-exit-signal", log.Fields{"topic": topic.Name, "rpc": rpc})
+ return true, nil
+ }
+ }
+ return true, nil
+}
+
+// Subscribe allows a caller to subscribe to a given topic. A channel is returned to the
+// caller to receive messages from that topic.
+func (kp *KafkaMessagingProxy) Subscribe(topic Topic) (<-chan *ca.InterContainerMessage, error) {
+
+ log.Debugw("subscribe", log.Fields{"topic": topic.Name})
+
+ if consumerCh := kp.getConsumerChannel(topic); consumerCh != nil {
+ log.Debugw("topic-already-subscribed", log.Fields{"topic": topic.Name})
+ // Create a channel specific for that consumer and add it to the consumer channel map
+ ch := make(chan *ca.InterContainerMessage)
+ kp.addChannelToConsumerChannelMap(topic, ch)
+ return ch, nil
+ }
+
+ // Register for the topic and set it up
+ var consumerListeningChannel chan *ca.InterContainerMessage
+ var err error
+ if consumerListeningChannel, err = kp.setupConsumerChannel(topic); err != nil {
+ log.Warnw("create-consumer-channel-failure", log.Fields{"error": err, "topic": topic.Name})
+ return nil, err
+ }
+
+ return consumerListeningChannel, nil
+}
+
+func (kp *KafkaMessagingProxy) UnSubscribe(topic Topic, ch <-chan *ca.InterContainerMessage) error {
+ log.Debugw("unsubscribing-channel-from-topic", log.Fields{"topic": topic.Name})
+ err := kp.removeChannelFromConsumerChannelMap(topic, ch)
+ return err
+}
+
+// SubscribeWithTarget allows a caller to assign a target object to be invoked automatically
+// when a message is received on a given topic
+func (kp *KafkaMessagingProxy) SubscribeWithTarget(topic Topic, targetInterface interface{}) error {
+
+ // Subscribe to receive messages for that topic
+ var ch <-chan *ca.InterContainerMessage
+ var err error
+ if ch, err = kp.Subscribe(topic); err != nil {
+ log.Errorw("failed-to-subscribe", log.Fields{"error": err, "topic": topic.Name})
+ }
+ // Launch a go routine to receive and process kafka messages
+ go kp.waitForRequest(ch, topic, targetInterface)
+
+ return nil
+}
+
+func (kp *KafkaMessagingProxy) UnSubscribeTarget(ctx context.Context, topic Topic, targetInterface interface{}) error {
+ // TODO - mostly relevant with multiple interfaces
+ return nil
+}
+
+func (kp *KafkaMessagingProxy) addToTopicToConsumerChannelMap(id string, arg *consumerChannels) {
+ kp.lockTopicToConsumerChannelMap.Lock()
+ defer kp.lockTopicToConsumerChannelMap.Unlock()
+ if _, exist := kp.topicToConsumerChannelMap[id]; !exist {
+ kp.topicToConsumerChannelMap[id] = arg
+ }
+}
+
+func (kp *KafkaMessagingProxy) deleteFromTopicToConsumerChannelMap(id string) {
+ kp.lockTopicToConsumerChannelMap.Lock()
+ defer kp.lockTopicToConsumerChannelMap.Unlock()
+ if _, exist := kp.topicToConsumerChannelMap[id]; exist {
+ delete(kp.topicToConsumerChannelMap, id)
+ }
+}
+
+func (kp *KafkaMessagingProxy) getConsumerChannel(topic Topic) *consumerChannels {
+ kp.lockTopicToConsumerChannelMap.Lock()
+ defer kp.lockTopicToConsumerChannelMap.Unlock()
+
+ if consumerCh, exist := kp.topicToConsumerChannelMap[topic.Name]; exist {
+ return consumerCh
+ }
+ return nil
+}
+
+func (kp *KafkaMessagingProxy) addChannelToConsumerChannelMap(topic Topic, ch chan *ca.InterContainerMessage) {
+ kp.lockTopicToConsumerChannelMap.Lock()
+ defer kp.lockTopicToConsumerChannelMap.Unlock()
+ if consumerCh, exist := kp.topicToConsumerChannelMap[topic.Name]; exist {
+ consumerCh.channels = append(consumerCh.channels, ch)
+ return
+ }
+ log.Warnw("consumer-channel-not-exist", log.Fields{"topic": topic.Name})
+}
+
+func (kp *KafkaMessagingProxy) removeChannelFromConsumerChannelMap(topic Topic, ch <-chan *ca.InterContainerMessage) error {
+ kp.lockTopicToConsumerChannelMap.Lock()
+ defer kp.lockTopicToConsumerChannelMap.Unlock()
+ if consumerCh, exist := kp.topicToConsumerChannelMap[topic.Name]; exist {
+ // Channel will be closed in the removeChannel method
+ consumerCh.channels = removeChannel(consumerCh.channels, ch)
+ return nil
+ }
+ log.Warnw("topic-does-not-exist", log.Fields{"topic": topic.Name})
+ return errors.New("topic-does-not-exist")
+}
+
+func (kp *KafkaMessagingProxy) addToTransactionIdToChannelMap(id string, arg chan *ca.InterContainerMessage) {
+ kp.lockTransactionIdToChannelMap.Lock()
+ defer kp.lockTransactionIdToChannelMap.Unlock()
+ if _, exist := kp.transactionIdToChannelMap[id]; !exist {
+ kp.transactionIdToChannelMap[id] = arg
+ }
+}
+
+func (kp *KafkaMessagingProxy) deleteFromTransactionIdToChannelMap(id string) {
+ kp.lockTransactionIdToChannelMap.Lock()
+ defer kp.lockTransactionIdToChannelMap.Unlock()
+ if _, exist := kp.transactionIdToChannelMap[id]; exist {
+ delete(kp.transactionIdToChannelMap, id)
+ }
+}
+
+func (kp *KafkaMessagingProxy) createPublisher(retries int) error {
+ // This Creates the publisher
+ config := sarama.NewConfig()
+ config.Producer.Partitioner = sarama.NewRandomPartitioner
+ config.Producer.Flush.Frequency = time.Duration(DefaultFlushFrequency)
+ config.Producer.Flush.Messages = DefaultFlushMessages
+ config.Producer.Flush.MaxMessages = DefaultFlushMaxmessages
+ config.Producer.Return.Errors = DefaultReturnErrors
+ config.Producer.Return.Successes = DefaultReturnSuccess
+ config.Producer.RequiredAcks = sarama.WaitForAll
+ kafkaFullAddr := fmt.Sprintf("%s:%d", kp.KafkaHost, kp.KafkaPort)
+ brokers := []string{kafkaFullAddr}
+
+ for {
+ producer, err := sarama.NewAsyncProducer(brokers, config)
+ if err != nil {
+ if retries == 0 {
+ log.Errorw("error-starting-publisher", log.Fields{"error": err})
+ return err
+ } else {
+ // If retries is -ve then we will retry indefinitely
+ retries--
+ }
+ log.Info("retrying-after-a-second-delay")
+ time.Sleep(time.Duration(DefaultSleepOnError) * time.Second)
+ } else {
+ kp.producer = producer
+ break
+ }
+ }
+ log.Info("Kafka-publisher-created")
+ return nil
+}
+
+func (kp *KafkaMessagingProxy) createConsumer(retries int) error {
+ config := sarama.NewConfig()
+ config.Consumer.Return.Errors = true
+ config.Consumer.MaxWaitTime = time.Duration(DefaultConsumerMaxwait) * time.Millisecond
+ config.Consumer.MaxProcessingTime = time.Duration(DefaultMaxProcessingTime) * time.Millisecond
+ config.Consumer.Offsets.Initial = sarama.OffsetNewest
+ kafkaFullAddr := fmt.Sprintf("%s:%d", kp.KafkaHost, kp.KafkaPort)
+ brokers := []string{kafkaFullAddr}
+
+ for {
+ consumer, err := sarama.NewConsumer(brokers, config)
+ if err != nil {
+ if retries == 0 {
+ log.Errorw("error-starting-consumer", log.Fields{"error": err})
+ return err
+ } else {
+ // If retries is -ve then we will retry indefinitely
+ retries--
+ }
+ log.Info("retrying-after-a-second-delay")
+ time.Sleep(time.Duration(DefaultSleepOnError) * time.Second)
+ } else {
+ kp.consumer = consumer
+ break
+ }
+ }
+ log.Info("Kafka-consumer-created")
+ return nil
+}
+
+func encodeReturnedValue(request *ca.InterContainerMessage, returnedVal interface{}) (*any.Any, error) {
+ // Encode the response argument - needs to be a proto message
+ if returnedVal == nil {
+ return nil, nil
+ }
+ protoValue, ok := returnedVal.(proto.Message)
+ if !ok {
+ log.Warnw("response-value-not-proto-message", log.Fields{"error": ok, "returnVal": returnedVal})
+ err := errors.New("response-value-not-proto-message")
+ return nil, err
+ }
+
+ // Marshal the returned value, if any
+ var marshalledReturnedVal *any.Any
+ var err error
+ if marshalledReturnedVal, err = ptypes.MarshalAny(protoValue); err != nil {
+ log.Warnw("cannot-marshal-returned-val", log.Fields{"error": err})
+ return nil, err
+ }
+ return marshalledReturnedVal, nil
+}
+
+func encodeDefaultFailedResponse(request *ca.InterContainerMessage) *ca.InterContainerMessage {
+ responseHeader := &ca.Header{
+ Id: request.Header.Id,
+ Type: ca.MessageType_RESPONSE,
+ FromTopic: request.Header.ToTopic,
+ ToTopic: request.Header.FromTopic,
+ Timestamp: time.Now().Unix(),
+ }
+ responseBody := &ca.InterContainerResponseBody{
+ Success: false,
+ Result: nil,
+ }
+ var marshalledResponseBody *any.Any
+ var err error
+ // Error should never happen here
+ if marshalledResponseBody, err = ptypes.MarshalAny(responseBody); err != nil {
+ log.Warnw("cannot-marshal-failed-response-body", log.Fields{"error": err})
+ }
+
+ return &ca.InterContainerMessage{
+ Header: responseHeader,
+ Body: marshalledResponseBody,
+ }
+
+}
+
+//formatRequest formats a request to send over kafka and returns an InterContainerMessage message on success
+//or an error on failure
+func encodeResponse(request *ca.InterContainerMessage, success bool, returnedValues ...interface{}) (*ca.InterContainerMessage, error) {
+
+ log.Infow("encodeResponse", log.Fields{"success": success, "returnedValues": returnedValues})
+ responseHeader := &ca.Header{
+ Id: request.Header.Id,
+ Type: ca.MessageType_RESPONSE,
+ FromTopic: request.Header.ToTopic,
+ ToTopic: request.Header.FromTopic,
+ Timestamp: time.Now().Unix(),
+ }
+
+ // Go over all returned values
+ var marshalledReturnedVal *any.Any
+ var err error
+ for _, returnVal := range returnedValues {
+ if marshalledReturnedVal, err = encodeReturnedValue(request, returnVal); err != nil {
+ log.Warnw("cannot-marshal-response-body", log.Fields{"error": err})
+ }
+ break // for now we support only 1 returned value - (excluding the error)
+ }
+
+ responseBody := &ca.InterContainerResponseBody{
+ Success: success,
+ Result: marshalledReturnedVal,
+ }
+
+ // Marshal the response body
+ var marshalledResponseBody *any.Any
+ if marshalledResponseBody, err = ptypes.MarshalAny(responseBody); err != nil {
+ log.Warnw("cannot-marshal-response-body", log.Fields{"error": err})
+ return nil, err
+ }
+
+ return &ca.InterContainerMessage{
+ Header: responseHeader,
+ Body: marshalledResponseBody,
+ }, nil
+}
+
+func CallFuncByName(myClass interface{}, funcName string, params ...interface{}) (out []reflect.Value, err error) {
+ myClassValue := reflect.ValueOf(myClass)
+ m := myClassValue.MethodByName(funcName)
+ if !m.IsValid() {
+ return make([]reflect.Value, 0), fmt.Errorf("Method not found \"%s\"", funcName)
+ }
+ in := make([]reflect.Value, len(params))
+ for i, param := range params {
+ in[i] = reflect.ValueOf(param)
+ }
+ out = m.Call(in)
+ return
+}
+
+func (kp *KafkaMessagingProxy) handleRequest(msg *ca.InterContainerMessage, targetInterface interface{}) {
+
+ // First extract the header to know whether this is a request of a response
+ if msg.Header.Type == ca.MessageType_REQUEST {
+ log.Debugw("received-request", log.Fields{"header": msg.Header})
+
+ var out []reflect.Value
+ var err error
+
+ // Get the request body
+ requestBody := &ca.InterContainerRequestBody{}
+ if err = ptypes.UnmarshalAny(msg.Body, requestBody); err != nil {
+ log.Warnw("cannot-unmarshal-request", log.Fields{"error": err})
+ } else {
+ // let the callee unpack the arguments as its the only one that knows the real proto type
+ out, err = CallFuncByName(targetInterface, requestBody.Rpc, requestBody.Args)
+ if err != nil {
+ log.Warn(err)
+ }
+ }
+ // Response required?
+ if requestBody.ResponseRequired {
+ // If we already have an error before then just return that
+ var returnError *ca.Error
+ var returnedValues []interface{}
+ var success bool
+ if err != nil {
+ returnError = &ca.Error{Reason: err.Error()}
+ returnedValues = make([]interface{}, 1)
+ returnedValues[0] = returnError
+ } else {
+ log.Debugw("returned-api-response", log.Fields{"len": len(out), "err": err})
+ returnSize := 1 // Minimum array size
+ if len(out) > 1 {
+ returnSize = len(out) - 1
+ }
+ returnedValues = make([]interface{}, returnSize)
+ for idx, val := range out {
+ log.Debugw("returned-api-response-loop", log.Fields{"idx": idx, "val": val.Interface()})
+ if idx == 0 {
+ if val.Interface() != nil {
+ if goError, ok := out[0].Interface().(error); ok {
+ returnError = &ca.Error{Reason: goError.Error()}
+ returnedValues[0] = returnError
+ } // Else should never occur - maybe never say never?
+ break
+ } else {
+ success = true
+ }
+ } else {
+ returnedValues[idx-1] = val.Interface()
+ }
+ }
+ }
+
+ var icm *ca.InterContainerMessage
+ if icm, err = encodeResponse(msg, success, returnedValues...); err != nil {
+ log.Warnw("error-encoding-response-returning-failure-result", log.Fields{"erroe": err})
+ icm = encodeDefaultFailedResponse(msg)
+ }
+ kp.sendToKafkaTopic(icm, &Topic{Name: msg.Header.FromTopic})
+ }
+
+ } else if msg.Header.Type == ca.MessageType_RESPONSE {
+ log.Warnw("received-response-on-request-handler", log.Fields{"header": msg.Header})
+ } else {
+ log.Errorw("invalid-message", log.Fields{"header": msg.Header})
+ }
+}
+
+func (kp *KafkaMessagingProxy) waitForRequest(ch <-chan *ca.InterContainerMessage, topic Topic, targetInterface interface{}) {
+ // Wait for messages
+ for msg := range ch {
+ log.Debugw("request-received", log.Fields{"msg": msg, "topic": topic.Name, "target": targetInterface})
+ go kp.handleRequest(msg, targetInterface)
+ }
+}
+
+// dispatchToConsumers sends the intercontainermessage received on a given topic to all subscribers for that
+// topic via the unique channel each subsciber received during subscription
+func (kp *KafkaMessagingProxy) dispatchToConsumers(consumerCh *consumerChannels, protoMessage *ca.InterContainerMessage) {
+ // Need to go over all channels and publish messages to them - do we need to copy msg?
+ kp.lockTopicToConsumerChannelMap.Lock()
+ defer kp.lockTopicToConsumerChannelMap.Unlock()
+ for _, ch := range consumerCh.channels {
+ go func(c chan *ca.InterContainerMessage) {
+ c <- protoMessage
+ }(ch)
+ }
+}
+
+func (kp *KafkaMessagingProxy) consumeMessagesLoop(topic Topic) {
+ log.Debugw("starting-consuming-messages", log.Fields{"topic": topic.Name})
+ var consumerCh *consumerChannels
+ if consumerCh = kp.getConsumerChannel(topic); consumerCh == nil {
+ log.Errorw("consumer-not-exist", log.Fields{"topic": topic.Name})
+ return
+ }
+startloop:
+ for {
+ select {
+ case err := <-consumerCh.consumer.Errors():
+ log.Warnw("consumer-error", log.Fields{"error": err})
+ case msg := <-consumerCh.consumer.Messages():
+ log.Debugw("message-received", log.Fields{"msg": msg})
+ // Since the only expected message is a proto intercontainermessage then extract it right away
+ // instead of dispatching it to the consumers
+ msgBody := msg.Value
+ icm := &ca.InterContainerMessage{}
+ if err := proto.Unmarshal(msgBody, icm); err != nil {
+ log.Warnw("invalid-message", log.Fields{"error": err})
+ continue
+ }
+ log.Debugw("msg-to-consumers", log.Fields{"msg": *icm, "len": len(consumerCh.channels)})
+
+ go kp.dispatchToConsumers(consumerCh, icm)
+ case <-kp.doneCh:
+ log.Infow("received-exit-signal", log.Fields{"topic": topic.Name})
+ break startloop
+ }
+ }
+}
+
+func (kp *KafkaMessagingProxy) dispatchResponse(msg *ca.InterContainerMessage) {
+ kp.lockTransactionIdToChannelMap.Lock()
+ defer kp.lockTransactionIdToChannelMap.Unlock()
+ if _, exist := kp.transactionIdToChannelMap[msg.Header.Id]; !exist {
+ log.Debugw("no-waiting-channel", log.Fields{"transaction": msg.Header.Id})
+ return
+ }
+ kp.transactionIdToChannelMap[msg.Header.Id] <- msg
+}
+
+func (kp *KafkaMessagingProxy) waitForResponse(ch chan *ca.InterContainerMessage, topic Topic) {
+ log.Debugw("starting-consuming-responses-loop", log.Fields{"topic": topic.Name})
+startloop:
+ for {
+ select {
+ case msg := <-ch:
+ log.Debugw("message-received", log.Fields{"topic": topic.Name, "msg": msg})
+ go kp.dispatchResponse(msg)
+ // Need to handle program exit - TODO
+ case <-kp.doneCh:
+ log.Infow("received-exit-signal", log.Fields{"topic": topic.Name})
+ break startloop
+ }
+ }
+}
+
+// createConsumerChannel creates a consumerChannels object for that topic and add it to the consumerChannels map
+// for that topic. It also starts the routine that listens for messages on that topic.
+func (kp *KafkaMessagingProxy) setupConsumerChannel(topic Topic) (chan *ca.InterContainerMessage, error) {
+
+ if consumerCh := kp.getConsumerChannel(topic); consumerCh != nil {
+ return nil, nil // Already created, so just ignore
+ }
+
+ partitionList, err := kp.consumer.Partitions(topic.Name)
+ if err != nil {
+ log.Warnw("get-partition-failure", log.Fields{"error": err, "topic": topic.Name})
+ return nil, err
+ }
+
+ log.Debugw("partitions", log.Fields{"topic": topic.Name, "partitionList": partitionList, "first": partitionList[0]})
+ // Create a partition consumer for that topic - for now just use one partition
+ var pConsumer sarama.PartitionConsumer
+ if pConsumer, err = kp.consumer.ConsumePartition(topic.Name, partitionList[0], sarama.OffsetNewest); err != nil {
+ log.Warnw("consumer-partition-failure", log.Fields{"error": err, "topic": topic.Name})
+ return nil, err
+ }
+
+ // Create the consumer/channel structure and set the consumer and create a channel on that topic - for now
+ // unbuffered to verify race conditions.
+ consumerListeningChannel := make(chan *ca.InterContainerMessage)
+ cc := &consumerChannels{
+ consumer: pConsumer,
+ channels: []chan *ca.InterContainerMessage{consumerListeningChannel},
+ }
+
+ // Add the consumer channel to the map
+ kp.addToTopicToConsumerChannelMap(topic.Name, cc)
+
+ //Start a consumer to listen on that specific topic
+ go kp.consumeMessagesLoop(topic)
+
+ return consumerListeningChannel, nil
+}
+
+// subscribeForResponse allows a caller to subscribe to a given topic when waiting for a response.
+// This method is built to prevent all subscribers to receive all messages as is the case of the Subscribe
+// API. There is one response channel waiting for kafka messages before dispatching the message to the
+// corresponding waiting channel
+func (kp *KafkaMessagingProxy) subscribeForResponse(topic Topic, trnsId string) (chan *ca.InterContainerMessage, error) {
+ log.Debugw("subscribeForResponse", log.Fields{"topic": topic.Name})
+
+ if consumerCh := kp.getConsumerChannel(topic); consumerCh == nil {
+ log.Debugw("topic-not-subscribed", log.Fields{"topic": topic.Name})
+ var consumerListeningChannel chan *ca.InterContainerMessage
+ var err error
+ if consumerListeningChannel, err = kp.setupConsumerChannel(topic); err != nil {
+ log.Warnw("create-consumer-channel-failure", log.Fields{"error": err, "topic": topic.Name})
+ return nil, err
+ }
+ // Start a go routine to listen to response messages over the consumer listening channel
+ go kp.waitForResponse(consumerListeningChannel, topic)
+ }
+
+ ch := make(chan *ca.InterContainerMessage)
+ kp.addToTransactionIdToChannelMap(trnsId, ch)
+
+ return ch, nil
+}
+
+func removeChannel(channels []chan *ca.InterContainerMessage, ch <-chan *ca.InterContainerMessage) []chan *ca.InterContainerMessage {
+ var i int
+ var channel chan *ca.InterContainerMessage
+ for i, channel = range channels {
+ if channel == ch {
+ channels[len(channels)-1], channels[i] = channels[i], channels[len(channels)-1]
+ close(channel)
+ return channels[:len(channels)-1]
+ }
+ }
+ return channels
+}
+
+func (kp *KafkaMessagingProxy) unSubscribeForResponse(trnsId string) error {
+ log.Debugw("unsubscribe-for-response", log.Fields{"trnsId": trnsId})
+ // Close the channel first
+ close(kp.transactionIdToChannelMap[trnsId])
+ kp.deleteFromTransactionIdToChannelMap(trnsId)
+ return nil
+}
+
+//formatRequest formats a request to send over kafka and returns an InterContainerMessage message on success
+//or an error on failure
+func encodeRequest(rpc string, toTopic *Topic, replyTopic *Topic, kvArgs ...*KVArg) (*ca.InterContainerMessage, error) {
+ requestHeader := &ca.Header{
+ Id: uuid.New().String(),
+ Type: ca.MessageType_REQUEST,
+ FromTopic: replyTopic.Name,
+ ToTopic: toTopic.Name,
+ Timestamp: time.Now().Unix(),
+ }
+ requestBody := &ca.InterContainerRequestBody{
+ Rpc: rpc,
+ ResponseRequired: true,
+ ReplyToTopic: replyTopic.Name,
+ }
+
+ for _, arg := range kvArgs {
+ var marshalledArg *any.Any
+ var err error
+ // ascertain the value interface type is a proto.Message
+ protoValue, ok := arg.Value.(proto.Message)
+ if !ok {
+ log.Warnw("argument-value-not-proto-message", log.Fields{"error": ok, "Value": arg.Value})
+ err := errors.New("argument-value-not-proto-message")
+ return nil, err
+ }
+ if marshalledArg, err = ptypes.MarshalAny(protoValue); err != nil {
+ log.Warnw("cannot-marshal-request", log.Fields{"error": err})
+ return nil, err
+ }
+ protoArg := &ca.Argument{
+ Key: arg.Key,
+ Value: marshalledArg,
+ }
+ requestBody.Args = append(requestBody.Args, protoArg)
+ }
+
+ var marshalledData *any.Any
+ var err error
+ if marshalledData, err = ptypes.MarshalAny(requestBody); err != nil {
+ log.Warnw("cannot-marshal-request", log.Fields{"error": err})
+ return nil, err
+ }
+ request := &ca.InterContainerMessage{
+ Header: requestHeader,
+ Body: marshalledData,
+ }
+ return request, nil
+}
+
+// sendRequest formats and sends the request onto the kafka messaging bus. It waits for the
+// response if needed. This function must, therefore, be run in its own routine.
+func (kp *KafkaMessagingProxy) sendToKafkaTopic(msg *ca.InterContainerMessage, topic *Topic) {
+
+ // Create the Sarama producer message
+ time := time.Now().Unix()
+ marshalled, _ := proto.Marshal(msg)
+ kafkaMsg := &sarama.ProducerMessage{
+ Topic: topic.Name,
+ Key: sarama.StringEncoder(time),
+ Value: sarama.ByteEncoder(marshalled),
+ }
+
+ // Send message to kafka
+ kp.producer.Input() <- kafkaMsg
+
+}
+
+func decodeResponse(response *ca.InterContainerMessage) (*ca.InterContainerResponseBody, error) {
+ // Extract the message body
+ responseBody := ca.InterContainerResponseBody{}
+
+ log.Debugw("decodeResponse", log.Fields{"icr": &response})
+ if err := ptypes.UnmarshalAny(response.Body, &responseBody); err != nil {
+ log.Warnw("cannot-unmarshal-response", log.Fields{"error": err})
+ return nil, err
+ }
+ log.Debugw("decodeResponse", log.Fields{"icrbody": &responseBody})
+
+ return &responseBody, nil
+
+}
diff --git a/kafka/kafka_inter_container_library_test.go b/kafka/kafka_inter_container_library_test.go
new file mode 100644
index 0000000..0fca49a
--- /dev/null
+++ b/kafka/kafka_inter_container_library_test.go
@@ -0,0 +1,74 @@
+package kafka
+
+import (
+ "github.com/stretchr/testify/assert"
+ "testing"
+)
+
+
+func TestDefaultKafkaProxy(t *testing.T) {
+ actualResult, error := NewKafkaMessagingProxy()
+ assert.Equal(t, error, nil)
+ assert.Equal(t, actualResult.KafkaHost, DefaultKafkaHost)
+ assert.Equal(t, actualResult.KafkaPort, DefaultKafkaPort)
+ assert.Equal(t, actualResult.TargetInterface, interface{}(nil))
+ assert.Equal(t, actualResult.DefaultTopic.Name, "Core")
+}
+
+func TestKafkaProxyOptionHost(t *testing.T) {
+ actualResult, error := NewKafkaMessagingProxy(KafkaHost("10.20.30.40"))
+ assert.Equal(t, error, nil)
+ assert.Equal(t, actualResult.KafkaHost, "10.20.30.40")
+ assert.Equal(t, actualResult.KafkaPort, DefaultKafkaPort)
+ assert.Equal(t, actualResult.TargetInterface, interface{}(nil))
+ assert.Equal(t, actualResult.DefaultTopic.Name, "Core")
+}
+
+func TestKafkaProxyOptionPort(t *testing.T) {
+ actualResult, error := NewKafkaMessagingProxy(KafkaPort(1020))
+ assert.Equal(t, error, nil)
+ assert.Equal(t, actualResult.KafkaHost, DefaultKafkaHost)
+ assert.Equal(t, actualResult.KafkaPort, 1020)
+ assert.Equal(t, actualResult.TargetInterface, interface{}(nil))
+ assert.Equal(t, actualResult.DefaultTopic.Name, "Core")
+}
+
+func TestKafkaProxyOptionTopic(t *testing.T) {
+ actualResult, error := NewKafkaMessagingProxy(DefaultTopic(&Topic{Name: "Adapter"}))
+ assert.Equal(t, error, nil)
+ assert.Equal(t, actualResult.KafkaHost, DefaultKafkaHost)
+ assert.Equal(t, actualResult.KafkaPort, DefaultKafkaPort)
+ assert.Equal(t, actualResult.TargetInterface, interface{}(nil))
+ assert.Equal(t, actualResult.DefaultTopic.Name, "Adapter")
+}
+
+type myInterface struct {
+}
+
+func (m *myInterface) doSomething() {
+}
+
+func TestKafkaProxyOptionTargetInterface(t *testing.T) {
+ var m *myInterface
+ actualResult, error := NewKafkaMessagingProxy(TargetInterface(m))
+ assert.Equal(t, error, nil)
+ assert.Equal(t, actualResult.KafkaHost, DefaultKafkaHost)
+ assert.Equal(t, actualResult.KafkaPort, DefaultKafkaPort)
+ assert.Equal(t, actualResult.TargetInterface, m)
+ assert.Equal(t, actualResult.DefaultTopic.Name, "Core")
+}
+
+func TestKafkaProxyChangeAllOptions(t *testing.T) {
+ var m *myInterface
+ actualResult, error := NewKafkaMessagingProxy(
+ KafkaHost("10.20.30.40"),
+ KafkaPort(1020),
+ DefaultTopic(&Topic{Name: "Adapter"}),
+ TargetInterface(m))
+ assert.Equal(t, error, nil)
+ assert.Equal(t, actualResult.KafkaHost, "10.20.30.40")
+ assert.Equal(t, actualResult.KafkaPort, 1020)
+ assert.Equal(t, actualResult.TargetInterface, m)
+ assert.Equal(t, actualResult.DefaultTopic.Name, "Adapter")
+}
+
diff --git a/kafka/messaging_interface.go b/kafka/messaging_interface.go
new file mode 100644
index 0000000..3511dc7
--- /dev/null
+++ b/kafka/messaging_interface.go
@@ -0,0 +1,30 @@
+package kafka
+
+import (
+ "context"
+)
+
+type callback func(bool, interface{})
+
+// A Topic definition - may be augmented with additional attributes eventually
+type Topic struct {
+ // The name of the topic. It must start with a letter,
+ // and contain only letters (`[A-Za-z]`), numbers (`[0-9]`), dashes (`-`),
+ // underscores (`_`), periods (`.`), tildes (`~`), plus (`+`) or percent
+ // signs (`%`).
+ Name string
+}
+
+type KVArg struct {
+ Key string
+ Value interface{}
+}
+
+// Client represents the set of APIs a Messaging Client must implement - In progress
+type Client interface {
+ Start()
+ Stop()
+ Subscribe(ctx context.Context, topic *Topic, cb callback, targetInterfaces ...interface{})
+ Publish(ctx context.Context, rpc string, cb callback, topic *Topic, waitForResponse bool, kvArgs ...*KVArg)
+ Unsubscribe(ctx context.Context, topic *Topic)
+}
diff --git a/protos/adapter.proto b/protos/adapter.proto
new file mode 100644
index 0000000..fc1ec94
--- /dev/null
+++ b/protos/adapter.proto
@@ -0,0 +1,43 @@
+syntax = "proto3";
+
+option go_package = "github.com/opencord/voltha-go/protos/voltha";
+
+package voltha;
+
+import "google/protobuf/any.proto";
+import "common.proto";
+import "meta.proto";
+
+
+message AdapterConfig {
+
+ // Common adapter config attributes here
+ LogLevel.LogLevel log_level = 1;
+
+ // Custom (vendor-specific) configuration attributes
+ google.protobuf.Any additional_config = 64;
+
+}
+
+// Adapter (software plugin)
+message Adapter {
+
+ // Unique name of adapter, matching the python packate name under
+ // voltha/adapters.
+ string id = 1 [(access) = READ_ONLY];
+ string vendor = 2 [(access) = READ_ONLY];
+ string version = 3 [(access) = READ_ONLY];
+
+ // Adapter configuration
+ AdapterConfig config = 16;
+
+ // Custom descriptors and custom configuration
+ google.protobuf.Any additional_description = 64 [(access) = READ_ONLY];
+
+ repeated string logical_device_ids = 4; // Logical devices "owned"
+
+}
+
+message Adapters {
+ repeated Adapter items = 1;
+}
diff --git a/protos/common.proto b/protos/common.proto
new file mode 100644
index 0000000..4f9ea1e
--- /dev/null
+++ b/protos/common.proto
@@ -0,0 +1,110 @@
+syntax = "proto3";
+
+option go_package = "github.com/opencord/voltha-go/protos/common";
+
+package voltha;
+
+import "yang_options.proto";
+
+// Convey a resource identifier
+message ID {
+ string id = 1;
+}
+
+message LogLevel {
+ option (yang_child_rule) = MOVE_TO_PARENT_LEVEL;
+
+ // Logging verbosity level
+ enum LogLevel {
+ DEBUG = 0;
+ INFO = 1;
+ WARNING = 2;
+ ERROR = 3;
+ CRITICAL = 4;
+ }
+}
+
+message AdminState {
+ option (yang_child_rule) = MOVE_TO_PARENT_LEVEL;
+
+ // Administrative State
+ enum AdminState {
+
+ // The administrative state of the device is unknown
+ UNKNOWN = 0;
+
+ // The device is pre-provisioned into Voltha, but not contacted by it
+ PREPROVISIONED = 1;
+
+ // The device is enabled for activation and operation
+ ENABLED = 3;
+
+ // The device is disabled and shall not perform its intended forwarding
+ // functions other than being available for re-activation.
+ DISABLED = 2;
+
+ // The deive is in the state of image download
+ DOWNLOADING_IMAGE = 4;
+
+ }
+}
+
+message OperStatus {
+ option (yang_child_rule) = MOVE_TO_PARENT_LEVEL;
+
+ // Operational Status
+ enum OperStatus {
+
+ // The status of the device is unknown at this point
+ UNKNOWN = 0;
+
+ // The device has been discovered, but not yet activated
+ DISCOVERED = 1;
+
+ // The device is being activated (booted, rebooted, upgraded, etc.)
+ ACTIVATING = 2;
+
+ // Service impacting tests are being conducted
+ TESTING = 3;
+
+ // The device is up and active
+ ACTIVE = 4;
+
+ // The device has failed and cannot fulfill its intended role
+ FAILED = 5;
+ }
+}
+
+message ConnectStatus {
+ option (yang_child_rule) = MOVE_TO_PARENT_LEVEL;
+
+ // Connectivity Status
+ enum ConnectStatus {
+
+ // The device connectivity status is unknown
+ UNKNOWN = 0;
+
+ // The device cannot be reached by Voltha
+ UNREACHABLE = 1;
+
+ // There is live communication between device and Voltha
+ REACHABLE = 2;
+ }
+}
+
+message OperationResp {
+ option (yang_child_rule) = MOVE_TO_PARENT_LEVEL;
+
+ enum OperationReturnCode {
+ OPERATION_SUCCESS = 0;
+ OPERATION_FAILURE = 1;
+ OPERATION_UNSUPPORTED = 2;
+ }
+ // Return code
+ OperationReturnCode code = 1;
+
+ // Additional Info
+ string additional_info = 2;
+}
+
+
diff --git a/protos/core_adapter.proto b/protos/core_adapter.proto
new file mode 100644
index 0000000..5021b9a
--- /dev/null
+++ b/protos/core_adapter.proto
@@ -0,0 +1,58 @@
+syntax = "proto3";
+
+import "google/protobuf/any.proto";
+
+option go_package = "github.com/opencord/voltha-go/protos/core_adapter";
+
+package voltha;
+
+message StrType {
+ string val = 1;
+}
+
+message IntType {
+ int64 val = 1;
+}
+
+message BoolType {
+ bool val = 1;
+}
+
+message Error {
+ string reason = 1;
+}
+
+enum MessageType {
+ REQUEST = 0;
+ RESPONSE = 1;
+}
+
+message Header {
+ string id = 1;
+ MessageType type = 2;
+ string from_topic = 3;
+ string to_topic = 4;
+ int64 timestamp = 5;
+}
+
+message Argument {
+ string key = 1;
+ google.protobuf.Any value = 2;
+}
+
+message InterContainerMessage {
+ Header header = 1;
+ google.protobuf.Any body = 2;
+}
+
+message InterContainerRequestBody {
+ string rpc = 2;
+ repeated Argument args = 3;
+ bool response_required = 4;
+ string reply_to_topic = 5;
+}
+
+message InterContainerResponseBody {
+ bool success = 1;
+ google.protobuf.Any result = 3;
+}
diff --git a/protos/device.proto b/protos/device.proto
new file mode 100644
index 0000000..f991e44
--- /dev/null
+++ b/protos/device.proto
@@ -0,0 +1,296 @@
+syntax = "proto3";
+
+option go_package = "github.com/opencord/voltha-go/protos/voltha";
+
+package voltha;
+
+import "meta.proto";
+import "google/protobuf/any.proto";
+import "common.proto";
+import "openflow_13.proto";
+import "yang_options.proto";
+
+
+// A Device Type
+message DeviceType {
+
+ // Unique name for the device type
+ string id = 1;
+
+ // Unique venor id for the device type applicable to ONU
+ // 4 bytes of vendor id from ONU serial number
+ string vendor_id = 5;
+
+ repeated string vendor_ids = 6;
+
+ // Name of the adapter that handles device type
+ string adapter = 2;
+
+ // Capabilitities
+
+ bool accepts_bulk_flow_update = 3;
+ bool accepts_add_remove_flow_updates = 4;
+
+}
+
+// A plurality of device types
+message DeviceTypes {
+ repeated DeviceType items = 1;
+}
+
+message PmConfig {
+ enum PmType {
+ COUNTER = 0;
+ GUAGE = 1;
+ STATE = 2;
+ }
+ string name = 1;
+ PmType type = 2;
+ bool enabled = 3; // Whether or not this metric makes it to Kafka
+ uint32 sample_freq = 4; // Sample rate in 10ths of a second
+}
+
+message PmGroupConfig {
+ string group_name = 1;
+ uint32 group_freq = 2; // Frequency applicable to the grop
+ bool enabled = 3; // Enable/disable group level only
+ repeated PmConfig metrics = 4;
+}
+
+message PmConfigs {
+ string id = 1; // To work around a chameleon POST bug
+ uint32 default_freq = 2; // Default sample rate
+ // Forces group names and group semantics
+ bool grouped = 3 [(access) = READ_ONLY];
+ // Allows Pm to set an individual sample frequency
+ bool freq_override = 4 [(access) = READ_ONLY];
+ repeated PmGroupConfig groups = 5; // The groups if grouped is true
+ repeated PmConfig metrics = 6; // The metrics themselves if grouped is false.
+}
+
+// Describes instance of software image on the device
+message Image {
+ string name = 1; // software patch name
+ string version = 2; // version of software
+ string hash = 3; // md5 hash
+ string install_datetime = 4; // combined date and time expressed in UTC.
+ // use ISO 8601 format for date and time
+
+ // The active software image is one that is currently loaded and executing
+ // in the ONU or circuit pack. Under normal operation, one software image
+ // is always active while the other is inactive. Under no circumstances are
+ // both software images allowed to be active at the same time
+ bool is_active = 5; // True if the image is active
+
+ // The committed software image is loaded and executed upon reboot of the
+ // ONU and/or circuit pack. During normal operation, one software image is
+ // always committed, while the other is uncommitted.
+ bool is_committed = 6; // True if the image is committed
+
+ // A software image is valid if it has been verified to be an executable
+ // code image. The verification mechanism is not subject to standardization;
+ // however, it should include at least a data integrity (e.g., CRC) check of
+ // the entire code image.
+ bool is_valid = 7; // True if the image is valid
+}
+
+// List of software on the device
+message Images {
+ repeated Image image = 1;
+}
+
+message ImageDownload {
+ option (yang_child_rule) = MOVE_TO_PARENT_LEVEL;
+
+ enum ImageDownloadState {
+ DOWNLOAD_UNKNOWN = 0;
+ DOWNLOAD_SUCCEEDED = 1;
+ DOWNLOAD_REQUESTED = 2;
+ DOWNLOAD_STARTED = 3;
+ DOWNLOAD_FAILED = 4;
+ DOWNLOAD_UNSUPPORTED = 5;
+ }
+
+ enum ImageDownloadFailureReason {
+ NO_ERROR = 0;
+ INVALID_URL = 1;
+ DEVICE_BUSY = 2;
+ INSUFFICIENT_SPACE = 3;
+ UNKNOWN_ERROR = 4;
+ }
+
+ enum ImageActivateState {
+ IMAGE_UNKNOWN = 0;
+ IMAGE_INACTIVE = 1;
+ IMAGE_ACTIVATE = 2;
+ IMAGE_ACTIVE = 3;
+ IMAGE_REVERT = 4;
+ }
+
+ // Device Identifier
+ string id = 1;
+
+ // Image unique identifier
+ string name = 2;
+
+ // URL where the image is available
+ // should include username password
+ string url = 3;
+
+ // CRC of the image to be verified aginst
+ uint32 crc = 4;
+
+ // Download state
+ ImageDownloadState state = 5;
+
+ // Downloaded version
+ string image_version = 6;
+
+ // Bytes downloaded
+ uint32 downloaded_bytes = 7;
+
+ // Download failure reason
+ ImageDownloadFailureReason reason= 8;
+
+ // Additional info
+ string additional_info = 9;
+
+ // Save current configuration
+ bool save_config = 10;
+
+ // Image local location
+ string local_dir = 11;
+
+ // Image activation state
+ ImageActivateState image_state = 12;
+}
+
+message ImageDownloads {
+ repeated ImageDownload items = 2;
+}
+
+message Port {
+ option (voltha.yang_child_rule) = MOVE_TO_PARENT_LEVEL;
+
+ enum PortType {
+ UNKNOWN = 0;
+ ETHERNET_NNI = 1;
+ ETHERNET_UNI = 2;
+ PON_OLT = 3;
+ PON_ONU = 4;
+ VENET_OLT = 5;
+ VENET_ONU = 6;
+ }
+
+ uint32 port_no = 1; // Device-unique port number
+
+ string label = 2; // Arbitrary port label
+
+ PortType type = 3; // Type of port
+
+ AdminState.AdminState admin_state = 5;
+
+ OperStatus.OperStatus oper_status = 6;
+
+ string device_id = 7; // Unique .id of device that owns this port
+
+ message PeerPort {
+ string device_id = 1;
+ uint32 port_no = 2;
+ }
+ repeated PeerPort peers = 8;
+
+}
+
+message Ports {
+ repeated Port items = 1;
+}
+
+// A Physical Device instance
+message Device {
+ option (voltha.yang_child_rule) = MOVE_TO_PARENT_LEVEL;
+
+ // Voltha's device identifier
+ string id = 1 [(access) = READ_ONLY];
+
+ // Device type, refers to one of the registered device types
+ string type = 2 [(access) = READ_ONLY];
+
+ // Is this device a root device. Each logical switch has one root
+ // device that is associated with the logical flow switch.
+ bool root = 3 [(access) = READ_ONLY];
+
+ // Parent device id, in the device tree (for a root device, the parent_id
+ // is the logical_device.id)
+ string parent_id = 4 [(access) = READ_ONLY];
+ uint32 parent_port_no = 20 [(access) = READ_ONLY];
+
+ // Vendor, version, serial number, etc.
+ string vendor = 5 [(access) = READ_ONLY];
+ string model = 6 [(access) = READ_ONLY];
+ string hardware_version = 7 [(access) = READ_ONLY];
+ string firmware_version = 8 [(access) = READ_ONLY];
+ // List of software on the device
+ Images images = 9 [(access) = READ_ONLY];
+ string serial_number = 10 [(access) = READ_ONLY];
+ string vendor_id = 24 [(access) = READ_ONLY];
+
+ // Addapter that takes care of device
+ string adapter = 11 [(access) = READ_ONLY];
+
+ // Device contact on vlan (if 0, no vlan)
+ uint32 vlan = 12;
+
+ message ProxyAddress {
+ string device_id = 1; // Which device to use as proxy to this device
+ uint32 channel_id = 2; // Sub-address within proxy
+ uint32 channel_group_id = 5; // Channel Group index
+ string channel_termination = 6; // Channel Termination name
+ uint32 onu_id = 3; // onu identifier; optional
+ uint32 onu_session_id = 4; // session identifier for the ONU; optional
+ };
+
+ oneof address {
+ // Device contact MAC address (format: "xx:xx:xx:xx:xx:xx")
+ string mac_address = 13;
+
+ // Device contact IPv4 address (format: "a.b.c.d" or can use hostname too)
+ string ipv4_address = 14;
+
+ // Device contact IPv6 address using the canonical string form
+ // ("xxxx:xxxx:xxxx:xxxx:xxxx:xxxx:xxxx:xxxx")
+ string ipv6_address = 15;
+
+ string host_and_port = 21;
+
+ };
+ string extra_args = 23; // Used to pass additional device specific arguments
+
+ ProxyAddress proxy_address = 19;
+
+ AdminState.AdminState admin_state = 16;
+
+ OperStatus.OperStatus oper_status = 17 [(access) = READ_ONLY];
+
+ string reason = 22 [(access) = READ_ONLY]; // Used in FAILED state
+
+ ConnectStatus.ConnectStatus connect_status = 18 [(access) = READ_ONLY];
+
+ // TODO additional common attribute here
+
+ // Device type specific attributes
+ google.protobuf.Any custom = 64;
+
+ repeated Port ports = 128 [(child_node) = {key: "port_no"}];
+ openflow_13.Flows flows = 129 [(child_node) = {}];
+ openflow_13.FlowGroups flow_groups = 130 [(child_node) = {}];
+ // PmConfigs will eventually converted to a child node of the
+ // device to falicitata callbacks and to simplify manipulation.
+ PmConfigs pm_configs = 131 [(child_node) = {}];
+
+ repeated ImageDownload image_downloads = 133 [(child_node) = {key: "name"}];
+}
+
+message Devices {
+ repeated Device items = 1;
+}
diff --git a/protos/events.proto b/protos/events.proto
new file mode 100644
index 0000000..27fd5cd
--- /dev/null
+++ b/protos/events.proto
@@ -0,0 +1,140 @@
+syntax = "proto3";
+
+option go_package = "github.com/opencord/voltha-go/protos/voltha";
+
+package voltha;
+
+import "meta.proto";
+import "google/api/annotations.proto";
+
+message ConfigEventType {
+ enum ConfigEventType {
+ add = 0; // A new config has been added
+ remove = 1; // A config has been removed
+ update = 2; // A config has been updated
+ }
+}
+
+message ConfigEvent {
+ ConfigEventType.ConfigEventType type = 1;
+
+ string hash = 2; // hash for this change, can be used for quick lookup
+ string data = 3; // the actual new data, in json format
+}
+
+message KpiEventType {
+ enum KpiEventType {
+ slice = 0; // slice: a set of path/metric data for same time-stamp
+ ts = 1; // time-series: array of data for same metric
+ }
+}
+
+/*
+ * Struct to convey a dictionary of metric->value pairs. Typically used in
+ * pure shared-timestamp or shared-timestamp + shared object prefix situations.
+ */
+message MetricValuePairs {
+
+ // Metric / value pairs.
+ map<string, float> metrics = 1;
+
+}
+
+
+message KpiEvent {
+
+ KpiEventType.KpiEventType type = 1;
+
+ // Fields used when for slice:
+
+ float ts = 2; // UTC time-stamp of data in slice mode (seconds since epoc)
+
+ map<string, MetricValuePairs> prefixes = 3;
+
+}
+
+/*
+ * Identify to the area of the system impacted by the alarm
+ */
+message AlarmEventType {
+ enum AlarmEventType {
+ COMMUNICATION = 0;
+ ENVIRONMENT = 1;
+ EQUIPMENT = 2;
+ SERVICE = 3;
+ PROCESSING = 4;
+ SECURITY = 5;
+ }
+}
+
+/*
+ * Identify to the functional category originating the alarm
+ */
+message AlarmEventCategory {
+ enum AlarmEventCategory {
+ PON = 0;
+ OLT = 1;
+ ONT = 2;
+ }
+}
+
+/*
+ * Active state of the alarm
+ */
+message AlarmEventState {
+ enum AlarmEventState {
+ RAISED = 0;
+ CLEARED = 1;
+ }
+}
+
+/*
+ * Identify the overall impact of the alarm on the system
+ */
+message AlarmEventSeverity {
+ enum AlarmEventSeverity {
+ INDETERMINATE = 0;
+ WARNING = 1;
+ MINOR = 2;
+ MAJOR = 3;
+ CRITICAL = 4;
+ }
+}
+
+/*
+ *
+ */
+message AlarmEvent {
+ // Unique ID for this alarm. e.g. voltha.some_olt.1234
+ string id = 1;
+
+ // Refers to the area of the system impacted by the alarm
+ AlarmEventType.AlarmEventType type = 2;
+
+ // Refers to functional category of the alarm
+ AlarmEventCategory.AlarmEventCategory category = 3;
+
+ // Current active state of the alarm
+ AlarmEventState.AlarmEventState state = 4;
+
+ // Overall impact of the alarm on the system
+ AlarmEventSeverity.AlarmEventSeverity severity = 5;
+
+ // Timestamp at which the alarm was first raised
+ float raised_ts = 6;
+
+ // Timestamp at which the alarm was reported
+ float reported_ts = 7;
+
+ // Timestamp at which the alarm has changed since it was raised
+ float changed_ts = 8;
+
+ // Identifier of the originating resource of the alarm
+ string resource_id = 9;
+
+ // Textual explanation of the alarm
+ string description = 10;
+
+ // Key/Value storage for extra information that may give context to the alarm
+ map<string, string> context = 11;
+}
diff --git a/protos/health.proto b/protos/health.proto
new file mode 100644
index 0000000..0730eac
--- /dev/null
+++ b/protos/health.proto
@@ -0,0 +1,36 @@
+syntax = "proto3";
+
+option go_package = "github.com/opencord/voltha-go/protos/voltha";
+
+package voltha;
+
+import "google/api/annotations.proto";
+import "google/protobuf/empty.proto";
+import "meta.proto";
+
+// Encode health status of a Voltha instance
+message HealthStatus {
+
+ // Health states
+ enum HealthState {
+ HEALTHY = 0; // The instance is healthy
+ OVERLOADED = 1; // The instance is overloaded, decrease query rate
+ DYING = 2; // The instance is in a critical condition, do not use it
+ }
+
+ // Current state of health of this Voltha instance
+ HealthState state = 1 [(access) = READ_ONLY];
+}
+
+// Health related services
+service HealthService {
+
+ // Return current health status of a Voltha instance
+ rpc GetHealthStatus(google.protobuf.Empty) returns (HealthStatus) {
+ option (google.api.http) = {
+ get: "/health"
+ };
+ }
+
+}
+
diff --git a/protos/ietf_interfaces.proto b/protos/ietf_interfaces.proto
new file mode 100644
index 0000000..d43c02d
--- /dev/null
+++ b/protos/ietf_interfaces.proto
@@ -0,0 +1,53 @@
+syntax = "proto3";
+
+option go_package = "github.com/opencord/voltha-go/protos/ietf";
+
+package ietf_interfaces;
+
+message Interfaces {
+ repeated Interface all_interfaces = 1;
+}
+message Interface {
+ string name = 1 ;
+ string description = 2 ;
+ string type = 3 ;
+ bool enabled = 4;
+ enum LinkUpDownTrapEnableType
+ {
+ TRAP_DISABLED = 0 ;
+ TRAP_ENABLED = 1 ;
+ }
+ LinkUpDownTrapEnableType link_up_down_trap_enable = 5;
+}
+
+message InterfacesState {
+ repeated InterfaceState all_interfacs = 1;
+}
+message InterfaceState {
+ string name = 1 ;
+ string type = 2 ;
+ enum AdminStatusType
+ {
+ ADMIN_DOWN = 0 ;
+ ADMIN_TESTING = 1 ;
+ ADMIN_UP = 2 ;
+ }
+ AdminStatusType admin_status = 3;
+ enum OperStatusType
+ {
+ DORMANT = 0 ;
+ LOWER_LAYER_DOWN = 1 ;
+ UNKNOWN = 2 ;
+ TESTING = 3 ;
+ UP = 4 ;
+ DOWN = 5 ;
+ NOT_PRESENT = 6 ;
+ }
+ OperStatusType oper_status = 4;
+ string last_change = 5 ;
+ int32 if_index = 6 ;
+ string phys_address = 7 ;
+ repeated string higher_layer_if = 8 ;
+ repeated string lower_layer_if = 9 ;
+ uint64 speed = 10 ;
+}
diff --git a/protos/logical_device.proto b/protos/logical_device.proto
new file mode 100644
index 0000000..b8d424f
--- /dev/null
+++ b/protos/logical_device.proto
@@ -0,0 +1,61 @@
+syntax = "proto3";
+
+option go_package = "github.com/opencord/voltha-go/protos/voltha";
+
+package voltha;
+
+import "meta.proto";
+import "google/api/annotations.proto";
+import "openflow_13.proto";
+
+message LogicalPortId {
+ // unique id of logical device
+ string id = 1;
+
+ // id of the port on the logical device
+ string port_id = 2;
+}
+
+message LogicalPort {
+ string id = 1;
+ openflow_13.ofp_port ofp_port = 2;
+ string device_id = 3;
+ uint32 device_port_no = 4;
+ bool root_port = 5;
+}
+
+message LogicalPorts {
+ repeated LogicalPort items = 1;
+}
+
+message LogicalDevice {
+
+ // unique id of logical device
+ string id = 1;
+
+ // unique datapath id for the logical device (used by the SDN controller)
+ uint64 datapath_id = 2;
+
+ // device description
+ openflow_13.ofp_desc desc = 3;
+
+ // device features
+ openflow_13.ofp_switch_features switch_features = 4;
+
+ // name of the root device anchoring logical device
+ string root_device_id = 5;
+
+ // logical device ports
+ repeated LogicalPort ports = 128 [(child_node) = {key: "id"}];
+
+ // flows configured on the logical device
+ openflow_13.Flows flows = 129 [(child_node) = {}];
+
+ // flow groups configured on the logical device
+ openflow_13.FlowGroups flow_groups = 130 [(child_node) = {}];
+
+}
+
+message LogicalDevices {
+ repeated LogicalDevice items = 1;
+}
diff --git a/protos/meta.proto b/protos/meta.proto
new file mode 100644
index 0000000..fc02e46
--- /dev/null
+++ b/protos/meta.proto
@@ -0,0 +1,57 @@
+// Copyright (c) 2015, Google Inc.
+//
+// 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.
+
+// This file contains annotation definitions that can be used to describe
+// a configuration tree.
+
+syntax = "proto3";
+
+option go_package = "github.com/opencord/voltha-go/protos/common";
+
+package voltha;
+
+import "google/protobuf/descriptor.proto";
+
+message ChildNode {
+ string key = 1;
+}
+
+enum Access {
+
+ // read-write, stored attribute
+ CONFIG = 0;
+
+ // read-only field, stored with the model, covered by its hash
+ READ_ONLY = 1;
+
+ // A read-only attribute that is not stored in the model, not covered
+ // by its hash, its value is filled real-time upon each request.
+ REAL_TIME = 2;
+
+}
+
+extend google.protobuf.FieldOptions {
+
+ // If present, it indicates that this field is stored as external child node
+ // or children nodes in Voltha's internal configuration tree.
+ // If the field is a container field and if the option specifies a key
+ // the child objects will be addressible by that key.
+ ChildNode child_node = 7761772;
+
+ // This annotation can be used to indicate that a field is read-only,
+ // from the perspective of NBI access. Backend plugins and system
+ // internals can update the field but the update requests through the
+ // NBI will ignore for instance a field that is marked as read-only (RO).
+ Access access = 7761773;
+}
diff --git a/protos/omci_mib_db.proto b/protos/omci_mib_db.proto
new file mode 100644
index 0000000..f8fe157
--- /dev/null
+++ b/protos/omci_mib_db.proto
@@ -0,0 +1,55 @@
+//
+// Copyright 2018 - present the original author or 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.
+//
+syntax = "proto3";
+
+option go_package = "github.com/opencord/voltha-go/protos/omci";
+
+package omci;
+
+import "meta.proto";
+
+
+message MibAttributeData {
+ string name = 1 [(voltha.access) = READ_ONLY];
+ string value = 2;
+}
+
+message MibInstanceData {
+ uint32 instance_id = 1 [(voltha.access) = READ_ONLY];
+ string created = 2;
+ string modified = 3;
+
+ repeated MibAttributeData attributes = 4
+ [(voltha.child_node) = {key: "name"}];
+}
+
+message MibClassData {
+ uint32 class_id = 1 [(voltha.access) = READ_ONLY];
+
+ repeated MibInstanceData instances= 2
+ [(voltha.child_node) = {key: "instance_id"}];
+}
+
+message MibDeviceData {
+ string device_id = 1 [(voltha.access) = READ_ONLY];
+ string created = 2;
+ string last_sync_time = 3;
+ uint32 mib_data_sync = 4;
+ uint32 version = 5;
+
+ repeated MibClassData classes = 6
+ [(voltha.child_node) = {key: "class_id"}];
+}
diff --git a/protos/openflow_13.proto b/protos/openflow_13.proto
new file mode 100644
index 0000000..47d881b
--- /dev/null
+++ b/protos/openflow_13.proto
@@ -0,0 +1,2298 @@
+/* Copyright (c) 2008 The Board of Trustees of The Leland Stanford
+ * Junior University
+ * Copyright (c) 2011, 2012 Open Networking Foundation
+ *
+ * We are making the OpenFlow specification and associated documentation
+ * (Software) available for public use and benefit with the expectation
+ * that others will use, modify and enhance the Software and contribute
+ * those enhancements back to the community. However, since we would
+ * like to make the Software available for broadest use, with as few
+ * restrictions as possible permission is hereby granted, free of
+ * charge, to any person obtaining a copy of this Software to deal in
+ * the Software under the copyrights without restriction, including
+ * without limitation the rights to use, copy, modify, merge, publish,
+ * distribute, sublicense, and/or sell copies of the Software, and to
+ * permit persons to whom the Software is furnished to do so, subject to
+ * the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be
+ * included in all copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
+ * EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
+ * MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND
+ * NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS
+ * BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN
+ * ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN
+ * CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+ * SOFTWARE.
+ *
+ * The name and trademarks of copyright holder(s) may NOT be used in
+ * advertising or publicity pertaining to the Software or any
+ * derivatives without specific, written prior permission.
+ */
+
+/* OpenFlow: protocol between controller and datapath. */
+
+/*
+ * This is a relatively straightforward rendering of OpenFlow message
+ * definitions into protocol buffer messages. We preserved the snake
+ * case syntax, and made the following changes:
+ * - all pad fields dropped
+ * - for each enum value above 0x7fffffff the MSB is dropped. For example,
+ * 0xffffffff is now 0x7fffffff.
+ * - '<type> thing[...]' is replaced with 'repeated <type> thing'
+ * - 'char thing[...]' is replaced with 'string thing'
+ * - 'uint8_t data[...]' is replaced with 'bytes data'
+ * - the following systematic changes are done to various integer types:
+ * uint8_t -> uint32
+ * uint16_t -> uint32
+ * uint32_t -> uint32
+ * uint64_t -> uint64
+ * - removed most length, len, size fields where these values can be determined
+ * from the explicitly encoded length of "repeated" protobuf fields.
+ * - explicit use of enum types whereever it is unambigous (and not used as
+ * bitmask/flags value.
+ *
+ */
+syntax = "proto3";
+
+option go_package = "github.com/opencord/voltha-go/protos/openflow_13";
+
+package openflow_13;
+
+import "google/api/annotations.proto";
+import public "yang_options.proto";
+
+
+/* Version number:
+ * OpenFlow versions released: 0x01 = 1.0 ; 0x02 = 1.1 ; 0x03 = 1.2
+ * 0x04 = 1.3
+ */
+/* The most significant bit in the version field is reserved and must
+ * be set to zero.
+ */
+//#define OFP_VERSION 0x04
+//#define PIPELINE_TABLES 64
+//#define OFP_MAX_TABLE_NAME_LEN 32
+//#define OFP_MAX_PORT_NAME_LEN 16
+/* Official IANA registered port for OpenFlow. */
+//#define OFP_TCP_PORT 6653
+//#define OFP_SSL_PORT 6653
+
+//#define OFP_ETH_ALEN 6 /* Bytes in an Ethernet address. */
+
+/* Port numbering. Ports are numbered starting from 1. */
+enum ofp_port_no {
+ OFPP_INVALID = 0;
+
+ /* Maximum number of physical and logical switch ports. */
+ OFPP_MAX = 0x7fffff00;
+
+ /* Reserved OpenFlow Port (fake output "ports"). */
+ OFPP_IN_PORT = 0x7ffffff8; /* Send the packet out the input port. This
+ reserved port must be explicitly used
+ in order to send back out of the input
+ port. */
+ OFPP_TABLE = 0x7ffffff9; /* Submit the packet to the first flow table
+ NB: This destination port can only be
+ used in packet-out messages. */
+ OFPP_NORMAL = 0x7ffffffa; /* Forward using non-OpenFlow pipeline. */
+ OFPP_FLOOD = 0x7ffffffb; /* Flood using non-OpenFlow pipeline. */
+ OFPP_ALL = 0x7ffffffc; /* All standard ports except input port. */
+ OFPP_CONTROLLER = 0x7ffffffd; /* Send to controller. */
+ OFPP_LOCAL = 0x7ffffffe; /* Local openflow "port". */
+ OFPP_ANY = 0x7fffffff; /* Special value used in some requests when
+ no port is specified (i.e. wildcarded).*/
+};
+
+enum ofp_type {
+
+ /* Immutable messages. */
+ OFPT_HELLO = 0; /* Symmetric message */
+ OFPT_ERROR = 1; /* Symmetric message */
+ OFPT_ECHO_REQUEST = 2; /* Symmetric message */
+ OFPT_ECHO_REPLY = 3; /* Symmetric message */
+ OFPT_EXPERIMENTER = 4; /* Symmetric message */
+
+ /* Switch configuration messages. */
+ OFPT_FEATURES_REQUEST = 5; /* Controller/switch message */
+ OFPT_FEATURES_REPLY = 6; /* Controller/switch message */
+ OFPT_GET_CONFIG_REQUEST = 7; /* Controller/switch message */
+ OFPT_GET_CONFIG_REPLY = 8; /* Controller/switch message */
+ OFPT_SET_CONFIG = 9; /* Controller/switch message */
+
+ /* Asynchronous messages. */
+ OFPT_PACKET_IN = 10; /* Async message */
+ OFPT_FLOW_REMOVED = 11; /* Async message */
+ OFPT_PORT_STATUS = 12; /* Async message */
+
+ /* Controller command messages. */
+ OFPT_PACKET_OUT = 13; /* Controller/switch message */
+ OFPT_FLOW_MOD = 14; /* Controller/switch message */
+ OFPT_GROUP_MOD = 15; /* Controller/switch message */
+ OFPT_PORT_MOD = 16; /* Controller/switch message */
+ OFPT_TABLE_MOD = 17; /* Controller/switch message */
+
+ /* Multipart messages. */
+ OFPT_MULTIPART_REQUEST = 18; /* Controller/switch message */
+ OFPT_MULTIPART_REPLY = 19; /* Controller/switch message */
+
+ /* Barrier messages. */
+ OFPT_BARRIER_REQUEST = 20; /* Controller/switch message */
+ OFPT_BARRIER_REPLY = 21; /* Controller/switch message */
+
+ /* Queue Configuration messages. */
+ OFPT_QUEUE_GET_CONFIG_REQUEST = 22; /* Controller/switch message */
+ OFPT_QUEUE_GET_CONFIG_REPLY = 23; /* Controller/switch message */
+
+ /* Controller role change request messages. */
+ OFPT_ROLE_REQUEST = 24; /* Controller/switch message */
+ OFPT_ROLE_REPLY = 25; /* Controller/switch message */
+
+ /* Asynchronous message configuration. */
+ OFPT_GET_ASYNC_REQUEST = 26; /* Controller/switch message */
+ OFPT_GET_ASYNC_REPLY = 27; /* Controller/switch message */
+ OFPT_SET_ASYNC = 28; /* Controller/switch message */
+
+ /* Meters and rate limiters configuration messages. */
+ OFPT_METER_MOD = 29; /* Controller/switch message */
+};
+
+/* Header on all OpenFlow packets. */
+message ofp_header {
+ uint32 version = 1; /* OFP_VERSION. */
+ ofp_type type = 2; /* One of the OFPT_ constants. */
+ uint32 xid = 3; /* Transaction id associated with this packet.
+ Replies use the same id as was in the request
+ to facilitate pairing. */
+};
+
+/* Hello elements types.
+ */
+enum ofp_hello_elem_type {
+ OFPHET_INVALID = 0;
+ OFPHET_VERSIONBITMAP = 1; /* Bitmap of version supported. */
+};
+
+/* Common header for all Hello Elements */
+message ofp_hello_elem_header {
+ ofp_hello_elem_type type = 1; /* One of OFPHET_*. */
+ oneof element {
+ ofp_hello_elem_versionbitmap versionbitmap = 2;
+ }
+};
+
+/* Version bitmap Hello Element */
+message ofp_hello_elem_versionbitmap {
+ repeated uint32 bitmaps = 2; /* List of bitmaps - supported versions */
+};
+
+/* OFPT_HELLO. This message includes zero or more hello elements having
+ * variable size. Unknown elements types must be ignored/skipped, to allow
+ * for future extensions. */
+message ofp_hello {
+ //ofp_header header;
+ /* Hello element list */
+ repeated ofp_hello_elem_header elements = 1; /* 0 or more */
+};
+
+//#define OFP_DEFAULT_MISS_SEND_LEN 128
+
+enum ofp_config_flags {
+ /* Handling of IP fragments. */
+ OFPC_FRAG_NORMAL = 0; /* No special handling for fragments. */
+ OFPC_FRAG_DROP = 1; /* Drop fragments. */
+ OFPC_FRAG_REASM = 2; /* Reassemble (only if OFPC_IP_REASM set). */
+ OFPC_FRAG_MASK = 3; /* Bitmask of flags dealing with frag. */
+};
+
+/* Switch configuration. */
+message ofp_switch_config {
+ //ofp_header header;
+ uint32 flags = 1; /* Bitmap of OFPC_* flags. */
+ uint32 miss_send_len = 2; /* Max bytes of packet that datapath
+ should send to the controller. See
+ ofp_controller_max_len for valid values.
+ */
+};
+
+/* Flags to configure the table. Reserved for future use. */
+enum ofp_table_config {
+ OFPTC_INVALID = 0;
+ OFPTC_DEPRECATED_MASK = 3; /* Deprecated bits */
+};
+
+/* Table numbering. Tables can use any number up to OFPT_MAX. */
+enum ofp_table {
+
+ OFPTT_INVALID = 0;
+
+ /* Last usable table number. */
+ OFPTT_MAX = 0xfe;
+
+ /* Fake tables. */
+ OFPTT_ALL = 0xff; /* Wildcard table used for table config,
+ flow stats and flow deletes. */
+};
+
+
+/* Configure/Modify behavior of a flow table */
+message ofp_table_mod {
+ //ofp_header header;
+ uint32 table_id = 1; /* ID of the table, OFPTT_ALL indicates all tables */
+ uint32 config = 2; /* Bitmap of OFPTC_* flags */
+};
+
+/* Capabilities supported by the datapath. */
+enum ofp_capabilities {
+ OFPC_INVALID = 0;
+ OFPC_FLOW_STATS = 1; /* Flow statistics. */
+ OFPC_TABLE_STATS = 2; /* Table statistics. */
+ OFPC_PORT_STATS = 4; /* Port statistics. */
+ OFPC_GROUP_STATS = 8; /* Group statistics. */
+ OFPC_IP_REASM = 32; /* Can reassemble IP fragments. */
+ OFPC_QUEUE_STATS = 64; /* Queue statistics. */
+ OFPC_PORT_BLOCKED = 256; /* Switch will block looping ports. */
+};
+
+/* Flags to indicate behavior of the physical port. These flags are
+ * used in ofp_port to describe the current configuration. They are
+ * used in the ofp_port_mod message to configure the port's behavior.
+ */
+enum ofp_port_config {
+ OFPPC_INVALID = 0;
+ OFPPC_PORT_DOWN = 1; /* Port is administratively down. */
+
+ OFPPC_NO_RECV = 4; /* Drop all packets received by port. */
+ OFPPC_NO_FWD = 32; /* Drop packets forwarded to port. */
+ OFPPC_NO_PACKET_IN = 64; /* Do not send packet-in msgs for port. */
+};
+
+/* Current state of the physical port. These are not configurable from
+ * the controller.
+ */
+enum ofp_port_state {
+ OFPPS_INVALID = 0;
+ OFPPS_LINK_DOWN = 1; /* No physical link present. */
+ OFPPS_BLOCKED = 2; /* Port is blocked */
+ OFPPS_LIVE = 4; /* Live for Fast Failover Group. */
+};
+
+/* Features of ports available in a datapath. */
+enum ofp_port_features {
+ OFPPF_INVALID = 0;
+ OFPPF_10MB_HD = 1; /* 10 Mb half-duplex rate support. */
+ OFPPF_10MB_FD = 2; /* 10 Mb full-duplex rate support. */
+ OFPPF_100MB_HD = 4; /* 100 Mb half-duplex rate support. */
+ OFPPF_100MB_FD = 8; /* 100 Mb full-duplex rate support. */
+ OFPPF_1GB_HD = 16; /* 1 Gb half-duplex rate support. */
+ OFPPF_1GB_FD = 32; /* 1 Gb full-duplex rate support. */
+ OFPPF_10GB_FD = 64; /* 10 Gb full-duplex rate support. */
+ OFPPF_40GB_FD = 128; /* 40 Gb full-duplex rate support. */
+ OFPPF_100GB_FD = 256; /* 100 Gb full-duplex rate support. */
+ OFPPF_1TB_FD = 512; /* 1 Tb full-duplex rate support. */
+ OFPPF_OTHER = 1024; /* Other rate, not in the list. */
+ OFPPF_COPPER = 2048; /* Copper medium. */
+ OFPPF_FIBER = 4096; /* Fiber medium. */
+ OFPPF_AUTONEG = 8192; /* Auto-negotiation. */
+ OFPPF_PAUSE = 16384; /* Pause. */
+ OFPPF_PAUSE_ASYM = 32768; /* Asymmetric pause. */
+};
+
+/* Description of a port */
+message ofp_port {
+ uint32 port_no = 1;
+ repeated uint32 hw_addr = 2; // [OFP_ETH_ALEN];
+ string name = 3; /* Null-terminated */
+
+ uint32 config = 4; /* Bitmap of OFPPC_* flags. */
+ uint32 state = 5; /* Bitmap of OFPPS_* flags. */
+
+ /* Bitmaps of OFPPF_* that describe features. All bits zeroed if
+ * unsupported or unavailable. */
+ uint32 curr = 6; /* Current features. */
+ uint32 advertised = 7; /* Features being advertised by the port. */
+ uint32 supported = 8; /* Features supported by the port. */
+ uint32 peer = 9; /* Features advertised by peer. */
+ uint32 curr_speed = 10; /* Current port bitrate in kbps. */
+ uint32 max_speed = 11; /* Max port bitrate in kbps */
+};
+
+/* Switch features. */
+message ofp_switch_features {
+ //ofp_header header;
+ uint64 datapath_id = 1; /* Datapath unique ID. The lower 48-bits are for
+ a MAC address, while the upper 16-bits are
+ implementer-defined. */
+
+ uint32 n_buffers = 2; /* Max packets buffered at once. */
+
+ uint32 n_tables = 3; /* Number of tables supported by datapath. */
+ uint32 auxiliary_id = 4; /* Identify auxiliary connections */
+
+ /* Features. */
+ uint32 capabilities = 5; /* Bitmap of support "ofp_capabilities". */
+};
+
+/* What changed about the physical port */
+enum ofp_port_reason {
+ OFPPR_ADD = 0; /* The port was added. */
+ OFPPR_DELETE = 1; /* The port was removed. */
+ OFPPR_MODIFY = 2; /* Some attribute of the port has changed. */
+};
+
+/* A physical port has changed in the datapath */
+message ofp_port_status {
+ //ofp_header header;
+ ofp_port_reason reason = 1; /* One of OFPPR_*. */
+ ofp_port desc = 2;
+};
+
+/* Modify behavior of the physical port */
+message ofp_port_mod {
+ //ofp_header header;
+ uint32 port_no = 1;
+ repeated uint32 hw_addr = 2; //[OFP_ETH_ALEN];
+ /* The hardware address is not
+ configurable. This is used to
+ sanity-check the request, so it must
+ be the same as returned in an
+ ofp_port struct. */
+ uint32 config = 3; /* Bitmap of OFPPC_* flags. */
+ uint32 mask = 4; /* Bitmap of OFPPC_* flags to be changed. */
+
+ uint32 advertise = 5; /* Bitmap of OFPPF_*. Zero all bits to prevent
+ any action taking place. */
+};
+
+/* ## -------------------------- ## */
+/* ## OpenFlow Extensible Match. ## */
+/* ## -------------------------- ## */
+
+/* The match type indicates the match structure (set of fields that compose the
+ * match) in use. The match type is placed in the type field at the beginning
+ * of all match structures. The "OpenFlow Extensible Match" type corresponds
+ * to OXM TLV format described below and must be supported by all OpenFlow
+ * switches. Extensions that define other match types may be published on the
+ * ONF wiki. Support for extensions is optional.
+ */
+enum ofp_match_type {
+ OFPMT_STANDARD = 0; /* Deprecated. */
+ OFPMT_OXM = 1; /* OpenFlow Extensible Match */
+};
+
+/* Fields to match against flows */
+message ofp_match {
+ ofp_match_type type = 1; /* One of OFPMT_* */
+ repeated ofp_oxm_field oxm_fields = 2; /* 0 or more */
+};
+
+/* Components of a OXM TLV header.
+ * Those macros are not valid for the experimenter class, macros for the
+ * experimenter class will depend on the experimenter header used. */
+//#define OXM_HEADER__(CLASS, FIELD, HASMASK, LENGTH) \
+// (((CLASS) << 16) | ((FIELD) << 9) | ((HASMASK) << 8) | (LENGTH))
+//#define OXM_HEADER(CLASS, FIELD, LENGTH) \
+// OXM_HEADER__(CLASS, FIELD, 0, LENGTH)
+//#define OXM_HEADER_W(CLASS, FIELD, LENGTH) \
+// OXM_HEADER__(CLASS, FIELD, 1, (LENGTH) * 2)
+//#define OXM_CLASS(HEADER) ((HEADER) >> 16)
+//#define OXM_FIELD(HEADER) (((HEADER) >> 9) & 0x7f)
+//#define OXM_TYPE(HEADER) (((HEADER) >> 9) & 0x7fffff)
+//#define OXM_HASMASK(HEADER) (((HEADER) >> 8) & 1)
+//#define OXM_LENGTH(HEADER) ((HEADER) & 0xff)
+//
+//#define OXM_MAKE_WILD_HEADER(HEADER) \
+// OXM_HEADER_W(OXM_CLASS(HEADER), OXM_FIELD(HEADER), OXM_LENGTH(HEADER))
+
+/* OXM Class IDs.
+ * The high order bit differentiate reserved classes from member classes.
+ * Classes 0x0000 to 0x7FFF are member classes, allocated by ONF.
+ * Classes 0x8000 to 0xFFFE are reserved classes, reserved for standardisation.
+ */
+enum ofp_oxm_class {
+ OFPXMC_NXM_0 = 0x0000; /* Backward compatibility with NXM */
+ OFPXMC_NXM_1 = 0x0001; /* Backward compatibility with NXM */
+ OFPXMC_OPENFLOW_BASIC = 0x8000; /* Basic class for OpenFlow */
+ OFPXMC_EXPERIMENTER = 0xFFFF; /* Experimenter class */
+};
+
+/* OXM Flow field types for OpenFlow basic class. */
+enum oxm_ofb_field_types {
+ OFPXMT_OFB_IN_PORT = 0; /* Switch input port. */
+ OFPXMT_OFB_IN_PHY_PORT = 1; /* Switch physical input port. */
+ OFPXMT_OFB_METADATA = 2; /* Metadata passed between tables. */
+ OFPXMT_OFB_ETH_DST = 3; /* Ethernet destination address. */
+ OFPXMT_OFB_ETH_SRC = 4; /* Ethernet source address. */
+ OFPXMT_OFB_ETH_TYPE = 5; /* Ethernet frame type. */
+ OFPXMT_OFB_VLAN_VID = 6; /* VLAN id. */
+ OFPXMT_OFB_VLAN_PCP = 7; /* VLAN priority. */
+ OFPXMT_OFB_IP_DSCP = 8; /* IP DSCP (6 bits in ToS field). */
+ OFPXMT_OFB_IP_ECN = 9; /* IP ECN (2 bits in ToS field). */
+ OFPXMT_OFB_IP_PROTO = 10; /* IP protocol. */
+ OFPXMT_OFB_IPV4_SRC = 11; /* IPv4 source address. */
+ OFPXMT_OFB_IPV4_DST = 12; /* IPv4 destination address. */
+ OFPXMT_OFB_TCP_SRC = 13; /* TCP source port. */
+ OFPXMT_OFB_TCP_DST = 14; /* TCP destination port. */
+ OFPXMT_OFB_UDP_SRC = 15; /* UDP source port. */
+ OFPXMT_OFB_UDP_DST = 16; /* UDP destination port. */
+ OFPXMT_OFB_SCTP_SRC = 17; /* SCTP source port. */
+ OFPXMT_OFB_SCTP_DST = 18; /* SCTP destination port. */
+ OFPXMT_OFB_ICMPV4_TYPE = 19; /* ICMP type. */
+ OFPXMT_OFB_ICMPV4_CODE = 20; /* ICMP code. */
+ OFPXMT_OFB_ARP_OP = 21; /* ARP opcode. */
+ OFPXMT_OFB_ARP_SPA = 22; /* ARP source IPv4 address. */
+ OFPXMT_OFB_ARP_TPA = 23; /* ARP target IPv4 address. */
+ OFPXMT_OFB_ARP_SHA = 24; /* ARP source hardware address. */
+ OFPXMT_OFB_ARP_THA = 25; /* ARP target hardware address. */
+ OFPXMT_OFB_IPV6_SRC = 26; /* IPv6 source address. */
+ OFPXMT_OFB_IPV6_DST = 27; /* IPv6 destination address. */
+ OFPXMT_OFB_IPV6_FLABEL = 28; /* IPv6 Flow Label */
+ OFPXMT_OFB_ICMPV6_TYPE = 29; /* ICMPv6 type. */
+ OFPXMT_OFB_ICMPV6_CODE = 30; /* ICMPv6 code. */
+ OFPXMT_OFB_IPV6_ND_TARGET = 31; /* Target address for ND. */
+ OFPXMT_OFB_IPV6_ND_SLL = 32; /* Source link-layer for ND. */
+ OFPXMT_OFB_IPV6_ND_TLL = 33; /* Target link-layer for ND. */
+ OFPXMT_OFB_MPLS_LABEL = 34; /* MPLS label. */
+ OFPXMT_OFB_MPLS_TC = 35; /* MPLS TC. */
+ OFPXMT_OFB_MPLS_BOS = 36; /* MPLS BoS bit. */
+ OFPXMT_OFB_PBB_ISID = 37; /* PBB I-SID. */
+ OFPXMT_OFB_TUNNEL_ID = 38; /* Logical Port Metadata. */
+ OFPXMT_OFB_IPV6_EXTHDR = 39; /* IPv6 Extension Header pseudo-field */
+};
+
+/* OXM Flow match fields */
+message ofp_oxm_field {
+ ofp_oxm_class oxm_class = 1;
+ oneof field {
+ /* 2 and 3 reserved for NXM_0 and NXM-1 OXM classes */
+ ofp_oxm_ofb_field ofb_field = 4;
+ ofp_oxm_experimenter_field experimenter_field = 5;
+ }
+}
+
+/* OXM OpenFlow Basic Match Field */
+message ofp_oxm_ofb_field {
+ oxm_ofb_field_types type = 1;
+ bool has_mask = 2;
+ oneof value {
+
+ /* OpenFlow port on which the packet was received.
+ * May be a physical port, a logical port, or the reserved port OFPP_LOCAL
+ *
+ * Prereqs: None.
+ *
+ * Format: 32-bit integer in network byte order.
+ *
+ * Masking: Not maskable. */
+ //#define OXM_OF_IN_PORT OXM_HEADER (0x8000, OFPXMT_OFB_IN_PORT, 4)
+ uint32 port = 3; /* Used for OFPXMT_OFB_IN_PORT */
+
+ /* Physical port on which the packet was received.
+ *
+ * Consider a packet received on a tunnel interface defined over a link
+ * aggregation group (LAG) with two physical port members. If the tunnel
+ * interface is the logical port bound to OpenFlow. In this case,
+ * OFPXMT_OF_IN_PORT is the tunnel's port number and OFPXMT_OF_IN_PHY_PORT is
+ * the physical port number of the LAG on which the tunnel is configured.
+ *
+ * When a packet is received directly on a physical port and not processed by a
+ * logical port, OFPXMT_OF_IN_PORT and OFPXMT_OF_IN_PHY_PORT have the same
+ * value.
+ *
+ * This field is usually not available in a regular match and only available
+ * in ofp_packet_in messages when it's different from OXM_OF_IN_PORT.
+ *
+ * Prereqs: OXM_OF_IN_PORT must be present.
+ *
+ * Format: 32-bit integer in network byte order.
+ *
+ * Masking: Not maskable. */
+ //#define OXM_OF_IN_PHY_PORT OXM_HEADER (0x8000, OFPXMT_OFB_IN_PHY_PORT, 4)
+ uint32 physical_port = 4; /* Used for OFPXMT_OF_IN_PHY_PORT */
+
+ /* Table metadata.
+ *
+ * Prereqs: None.
+ *
+ * Format: 64-bit integer in network byte order.
+ *
+ * Masking: Arbitrary masks.
+ */
+ //#define OXM_OF_METADATA OXM_HEADER (0x8000, OFPXMT_OFB_METADATA, 8)
+ //#define OXM_OF_METADATA_W OXM_HEADER_W(0x8000, OFPXMT_OFB_METADATA, 8)
+ uint64 table_metadata = 5; /* Used for OFPXMT_OFB_METADATA */
+
+ /* Source or destination address in Ethernet header.
+ *
+ * Prereqs: None.
+ *
+ * Format: 48-bit Ethernet MAC address.
+ *
+ * Masking: Arbitrary masks. */
+ //#define OXM_OF_ETH_DST OXM_HEADER (0x8000, OFPXMT_OFB_ETH_DST, 6)
+ //#define OXM_OF_ETH_DST_W OXM_HEADER_W(0x8000, OFPXMT_OFB_ETH_DST, 6)
+ //#define OXM_OF_ETH_SRC OXM_HEADER (0x8000, OFPXMT_OFB_ETH_SRC, 6)
+ //#define OXM_OF_ETH_SRC_W OXM_HEADER_W(0x8000, OFPXMT_OFB_ETH_SRC, 6)
+ bytes eth_dst = 6; /* Used for OFPXMT_OFB_ETH_DST (exactly 6 bytes) */
+ bytes eth_src = 7; /* Used for OFPXMT_OFB_ETH_SRC (exactly 6 bytes) */
+
+ /* Packet's Ethernet type.
+ *
+ * Prereqs: None.
+ *
+ * Format: 16-bit integer in network byte order.
+ *
+ * Masking: Not maskable. */
+ //#define OXM_OF_ETH_TYPE OXM_HEADER (0x8000, OFPXMT_OFB_ETH_TYPE,2)
+ uint32 eth_type = 8; /* Used for OFPXMT_OFB_ETH_TYPE */
+
+ /* 802.1Q VID.
+ *
+ * For a packet with an 802.1Q header, this is the VLAN-ID (VID) from the
+ * outermost tag, with the CFI bit forced to 1. For a packet with no 802.1Q
+ * header, this has value OFPVID_NONE.
+ *
+ * Prereqs: None.
+ *
+ * Format: 16-bit integer in network byte order with bit 13 indicating
+ * presence of VLAN header and 3 most-significant bits forced to 0.
+ * Only the lower 13 bits have meaning.
+ *
+ * Masking: Arbitrary masks.
+ *
+ * This field can be used in various ways:
+ *
+ * - If it is not constrained at all, the nx_match matches packets without
+ * an 802.1Q header or with an 802.1Q header that has any VID value.
+ *
+ * - Testing for an exact match with 0x0 matches only packets without
+ * an 802.1Q header.
+ *
+ * - Testing for an exact match with a VID value with CFI=1 matches packets
+ * that have an 802.1Q header with a specified VID.
+ *
+ * - Testing for an exact match with a nonzero VID value with CFI=0 does
+ * not make sense. The switch may reject this combination.
+ *
+ * - Testing with nxm_value=0, nxm_mask=0x0fff matches packets with no 802.1Q
+ * header or with an 802.1Q header with a VID of 0.
+ *
+ * - Testing with nxm_value=0x1000, nxm_mask=0x1000 matches packets with
+ * an 802.1Q header that has any VID value.
+ */
+ //#define OXM_OF_VLAN_VID OXM_HEADER (0x8000, OFPXMT_OFB_VLAN_VID, 2)
+ //#define OXM_OF_VLAN_VID_W OXM_HEADER_W(0x8000, OFPXMT_OFB_VLAN_VID, 2)
+ uint32 vlan_vid = 9; /* Used for OFPXMT_OFB_VLAN_VID */
+
+ /* 802.1Q PCP.
+ *
+ * For a packet with an 802.1Q header, this is the VLAN-PCP from the
+ * outermost tag. For a packet with no 802.1Q header, this has value
+ * 0.
+ *
+ * Prereqs: OXM_OF_VLAN_VID must be different from OFPVID_NONE.
+ *
+ * Format: 8-bit integer with 5 most-significant bits forced to 0.
+ * Only the lower 3 bits have meaning.
+ *
+ * Masking: Not maskable.
+ */
+ //#define OXM_OF_VLAN_PCP OXM_HEADER (0x8000, OFPXMT_OFB_VLAN_PCP, 1)
+ uint32 vlan_pcp = 10; /* Used for OFPXMT_OFB_VLAN_PCP */
+
+ /* The Diff Serv Code Point (DSCP) bits of the IP header.
+ * Part of the IPv4 ToS field or the IPv6 Traffic Class field.
+ *
+ * Prereqs: OXM_OF_ETH_TYPE must be either 0x0800 or 0x86dd.
+ *
+ * Format: 8-bit integer with 2 most-significant bits forced to 0.
+ * Only the lower 6 bits have meaning.
+ *
+ * Masking: Not maskable. */
+ //#define OXM_OF_IP_DSCP OXM_HEADER (0x8000, OFPXMT_OFB_IP_DSCP, 1)
+ uint32 ip_dscp = 11; /* Used for OFPXMT_OFB_IP_DSCP */
+
+ /* The ECN bits of the IP header.
+ * Part of the IPv4 ToS field or the IPv6 Traffic Class field.
+ *
+ * Prereqs: OXM_OF_ETH_TYPE must be either 0x0800 or 0x86dd.
+ *
+ * Format: 8-bit integer with 6 most-significant bits forced to 0.
+ * Only the lower 2 bits have meaning.
+ *
+ * Masking: Not maskable. */
+ //#define OXM_OF_IP_ECN OXM_HEADER (0x8000, OFPXMT_OFB_IP_ECN, 1)
+ uint32 ip_ecn = 12; /* Used for OFPXMT_OFB_IP_ECN */
+
+ /* The "protocol" byte in the IP header.
+ *
+ * Prereqs: OXM_OF_ETH_TYPE must be either 0x0800 or 0x86dd.
+ *
+ * Format: 8-bit integer.
+ *
+ * Masking: Not maskable. */
+ //#define OXM_OF_IP_PROTO OXM_HEADER (0x8000, OFPXMT_OFB_IP_PROTO, 1)
+ uint32 ip_proto = 13; /* Used for OFPXMT_OFB_IP_PROTO */
+
+ /* The source or destination address in the IP header.
+ *
+ * Prereqs: OXM_OF_ETH_TYPE must match 0x0800 exactly.
+ *
+ * Format: 32-bit integer in network byte order.
+ *
+ * Masking: Arbitrary masks.
+ */
+ //#define OXM_OF_IPV4_SRC OXM_HEADER (0x8000, OFPXMT_OFB_IPV4_SRC, 4)
+ //#define OXM_OF_IPV4_SRC_W OXM_HEADER_W(0x8000, OFPXMT_OFB_IPV4_SRC, 4)
+ //#define OXM_OF_IPV4_DST OXM_HEADER (0x8000, OFPXMT_OFB_IPV4_DST, 4)
+ //#define OXM_OF_IPV4_DST_W OXM_HEADER_W(0x8000, OFPXMT_OFB_IPV4_DST, 4)
+ uint32 ipv4_src = 14; /* Used for OFPXMT_OFB_IPV4_SRC */
+ uint32 ipv4_dst = 15; /* Used for OFPXMT_OFB_IPV4_DST */
+
+ /* The source or destination port in the TCP header.
+ *
+ * Prereqs:
+ * OXM_OF_ETH_TYPE must be either 0x0800 or 0x86dd.
+ * OXM_OF_IP_PROTO must match 6 exactly.
+ *
+ * Format: 16-bit integer in network byte order.
+ *
+ * Masking: Not maskable. */
+ //#define OXM_OF_TCP_SRC OXM_HEADER (0x8000, OFPXMT_OFB_TCP_SRC, 2)
+ //#define OXM_OF_TCP_DST OXM_HEADER (0x8000, OFPXMT_OFB_TCP_DST, 2)
+ uint32 tcp_src = 16; /* Used for OFPXMT_OFB_TCP_SRC */
+ uint32 tcp_dst = 17; /* Used for OFPXMT_OFB_TCP_DST */
+
+ /* The source or destination port in the UDP header.
+ *
+ * Prereqs:
+ * OXM_OF_ETH_TYPE must match either 0x0800 or 0x86dd.
+ * OXM_OF_IP_PROTO must match 17 exactly.
+ *
+ * Format: 16-bit integer in network byte order.
+ *
+ * Masking: Not maskable. */
+ //#define OXM_OF_UDP_SRC OXM_HEADER (0x8000, OFPXMT_OFB_UDP_SRC, 2)
+ //#define OXM_OF_UDP_DST OXM_HEADER (0x8000, OFPXMT_OFB_UDP_DST, 2)
+ uint32 udp_src = 18; /* Used for OFPXMT_OFB_UDP_SRC */
+ uint32 udp_dst = 19; /* Used for OFPXMT_OFB_UDP_DST */
+
+ /* The source or destination port in the SCTP header.
+ *
+ * Prereqs:
+ * OXM_OF_ETH_TYPE must match either 0x0800 or 0x86dd.
+ * OXM_OF_IP_PROTO must match 132 exactly.
+ *
+ * Format: 16-bit integer in network byte order.
+ *
+ * Masking: Not maskable. */
+ //#define OXM_OF_SCTP_SRC OXM_HEADER (0x8000, OFPXMT_OFB_SCTP_SRC, 2)
+ //#define OXM_OF_SCTP_DST OXM_HEADER (0x8000, OFPXMT_OFB_SCTP_DST, 2)
+ uint32 sctp_src = 20; /* Used for OFPXMT_OFB_SCTP_SRC */
+ uint32 sctp_dst = 21; /* Used for OFPXMT_OFB_SCTP_DST */
+
+ /* The type or code in the ICMP header.
+ *
+ * Prereqs:
+ * OXM_OF_ETH_TYPE must match 0x0800 exactly.
+ * OXM_OF_IP_PROTO must match 1 exactly.
+ *
+ * Format: 8-bit integer.
+ *
+ * Masking: Not maskable. */
+ //#define OXM_OF_ICMPV4_TYPE OXM_HEADER (0x8000, OFPXMT_OFB_ICMPV4_TYPE, 1)
+ //#define OXM_OF_ICMPV4_CODE OXM_HEADER (0x8000, OFPXMT_OFB_ICMPV4_CODE, 1)
+ uint32 icmpv4_type = 22; /* Used for OFPXMT_OFB_ICMPV4_TYPE */
+ uint32 icmpv4_code = 23; /* Used for OFPXMT_OFB_ICMPV4_CODE */
+
+ /* ARP opcode.
+ *
+ * For an Ethernet+IP ARP packet, the opcode in the ARP header. Always 0
+ * otherwise.
+ *
+ * Prereqs: OXM_OF_ETH_TYPE must match 0x0806 exactly.
+ *
+ * Format: 16-bit integer in network byte order.
+ *
+ * Masking: Not maskable. */
+ //#define OXM_OF_ARP_OP OXM_HEADER (0x8000, OFPXMT_OFB_ARP_OP, 2)
+ uint32 arp_op = 24; /* Used for OFPXMT_OFB_ARP_OP */
+
+ /* For an Ethernet+IP ARP packet, the source or target protocol address
+ * in the ARP header. Always 0 otherwise.
+ *
+ * Prereqs: OXM_OF_ETH_TYPE must match 0x0806 exactly.
+ *
+ * Format: 32-bit integer in network byte order.
+ *
+ * Masking: Arbitrary masks.
+ */
+ //#define OXM_OF_ARP_SPA OXM_HEADER (0x8000, OFPXMT_OFB_ARP_SPA, 4)
+ //#define OXM_OF_ARP_SPA_W OXM_HEADER_W(0x8000, OFPXMT_OFB_ARP_SPA, 4)
+ //#define OXM_OF_ARP_TPA OXM_HEADER (0x8000, OFPXMT_OFB_ARP_TPA, 4)
+ //#define OXM_OF_ARP_TPA_W OXM_HEADER_W(0x8000, OFPXMT_OFB_ARP_TPA, 4)
+ uint32 arp_spa = 25; /* For OFPXMT_OFB_ARP_SPA */
+ uint32 arp_tpa = 26; /* For OFPXMT_OFB_ARP_TPA */
+
+ /* For an Ethernet+IP ARP packet, the source or target hardware address
+ * in the ARP header. Always 0 otherwise.
+ *
+ * Prereqs: OXM_OF_ETH_TYPE must match 0x0806 exactly.
+ *
+ * Format: 48-bit Ethernet MAC address.
+ *
+ * Masking: Not maskable. */
+ //#define OXM_OF_ARP_SHA OXM_HEADER (0x8000, OFPXMT_OFB_ARP_SHA, 6)
+ //#define OXM_OF_ARP_SHA_W OXM_HEADER_W (0x8000, OFPXMT_OFB_ARP_SHA, 6)
+ //#define OXM_OF_ARP_THA OXM_HEADER (0x8000, OFPXMT_OFB_ARP_THA, 6)
+ //#define OXM_OF_ARP_THA_W OXM_HEADER_W (0x8000, OFPXMT_OFB_ARP_THA, 6)
+ bytes arp_sha = 27; /* For OFPXMT_OFB_ARP_SHA (6 bytes) */
+ bytes arp_tha = 28; /* For OFPXMT_OFB_ARP_THA (6 bytes) */
+
+ /* The source or destination address in the IPv6 header.
+ *
+ * Prereqs: OXM_OF_ETH_TYPE must match 0x86dd exactly.
+ *
+ * Format: 128-bit IPv6 address.
+ *
+ * Masking: Arbitrary masks.
+ */
+ //#define OXM_OF_IPV6_SRC OXM_HEADER (0x8000, OFPXMT_OFB_IPV6_SRC, 16)
+ //#define OXM_OF_IPV6_SRC_W OXM_HEADER_W(0x8000, OFPXMT_OFB_IPV6_SRC, 16)
+ //#define OXM_OF_IPV6_DST OXM_HEADER (0x8000, OFPXMT_OFB_IPV6_DST, 16)
+ //#define OXM_OF_IPV6_DST_W OXM_HEADER_W(0x8000, OFPXMT_OFB_IPV6_DST, 16)
+ bytes ipv6_src = 29; /* For OFPXMT_OFB_IPV6_SRC */
+ bytes ipv6_dst = 30; /* For OFPXMT_OFB_IPV6_DST */
+
+ /* The IPv6 Flow Label
+ *
+ * Prereqs:
+ * OXM_OF_ETH_TYPE must match 0x86dd exactly
+ *
+ * Format: 32-bit integer with 12 most-significant bits forced to 0.
+ * Only the lower 20 bits have meaning.
+ *
+ * Masking: Arbitrary masks.
+ */
+ //#define OXM_OF_IPV6_FLABEL OXM_HEADER (0x8000, OFPXMT_OFB_IPV6_FLABEL, 4)
+ //#define OXM_OF_IPV6_FLABEL_W OXM_HEADER_W(0x8000, OFPXMT_OFB_IPV6_FLABEL, 4)
+ uint32 ipv6_flabel = 31; /* For OFPXMT_OFB_IPV6_FLABEL */
+
+ /* The type or code in the ICMPv6 header.
+ *
+ * Prereqs:
+ * OXM_OF_ETH_TYPE must match 0x86dd exactly.
+ * OXM_OF_IP_PROTO must match 58 exactly.
+ *
+ * Format: 8-bit integer.
+ *
+ * Masking: Not maskable. */
+ //#define OXM_OF_ICMPV6_TYPE OXM_HEADER (0x8000, OFPXMT_OFB_ICMPV6_TYPE, 1)
+ //#define OXM_OF_ICMPV6_CODE OXM_HEADER (0x8000, OFPXMT_OFB_ICMPV6_CODE, 1)
+ uint32 icmpv6_type = 32; /* For OFPXMT_OFB_ICMPV6_TYPE */
+ uint32 icmpv6_code = 33; /* For OFPXMT_OFB_ICMPV6_CODE */
+
+ /* The target address in an IPv6 Neighbor Discovery message.
+ *
+ * Prereqs:
+ * OXM_OF_ETH_TYPE must match 0x86dd exactly.
+ * OXM_OF_IP_PROTO must match 58 exactly.
+ * OXM_OF_ICMPV6_TYPE must be either 135 or 136.
+ *
+ * Format: 128-bit IPv6 address.
+ *
+ * Masking: Not maskable. */
+ //#define OXM_OF_IPV6_ND_TARGET OXM_HEADER \
+ // (0x8000, OFPXMT_OFB_IPV6_ND_TARGET, 16)
+ bytes ipv6_nd_target = 34; /* For OFPXMT_OFB_IPV6_ND_TARGET */
+
+ /* The source link-layer address option in an IPv6 Neighbor Discovery
+ * message.
+ *
+ * Prereqs:
+ * OXM_OF_ETH_TYPE must match 0x86dd exactly.
+ * OXM_OF_IP_PROTO must match 58 exactly.
+ * OXM_OF_ICMPV6_TYPE must be exactly 135.
+ *
+ * Format: 48-bit Ethernet MAC address.
+ *
+ * Masking: Not maskable. */
+ //#define OXM_OF_IPV6_ND_SLL OXM_HEADER (0x8000, OFPXMT_OFB_IPV6_ND_SLL, 6)
+ bytes ipv6_nd_ssl = 35; /* For OFPXMT_OFB_IPV6_ND_SLL */
+
+ /* The target link-layer address option in an IPv6 Neighbor Discovery
+ * message.
+ *
+ * Prereqs:
+ * OXM_OF_ETH_TYPE must match 0x86dd exactly.
+ * OXM_OF_IP_PROTO must match 58 exactly.
+ * OXM_OF_ICMPV6_TYPE must be exactly 136.
+ *
+ * Format: 48-bit Ethernet MAC address.
+ *
+ * Masking: Not maskable. */
+ //#define OXM_OF_IPV6_ND_TLL OXM_HEADER (0x8000, OFPXMT_OFB_IPV6_ND_TLL, 6)
+ bytes ipv6_nd_tll = 36; /* For OFPXMT_OFB_IPV6_ND_TLL */
+
+ /* The LABEL in the first MPLS shim header.
+ *
+ * Prereqs:
+ * OXM_OF_ETH_TYPE must match 0x8847 or 0x8848 exactly.
+ *
+ * Format: 32-bit integer in network byte order with 12 most-significant
+ * bits forced to 0. Only the lower 20 bits have meaning.
+ *
+ * Masking: Not maskable. */
+ //#define OXM_OF_MPLS_LABEL OXM_HEADER (0x8000, OFPXMT_OFB_MPLS_LABEL, 4)
+ uint32 mpls_label = 37; /* For OFPXMT_OFB_MPLS_LABEL */
+
+ /* The TC in the first MPLS shim header.
+ *
+ * Prereqs:
+ * OXM_OF_ETH_TYPE must match 0x8847 or 0x8848 exactly.
+ *
+ * Format: 8-bit integer with 5 most-significant bits forced to 0.
+ * Only the lower 3 bits have meaning.
+ *
+ * Masking: Not maskable. */
+ //#define OXM_OF_MPLS_TC OXM_HEADER (0x8000, OFPXMT_OFB_MPLS_TC, 1)
+ uint32 mpls_tc = 38; /* For OFPXMT_OFB_MPLS_TC */
+
+ /* The BoS bit in the first MPLS shim header.
+ *
+ * Prereqs:
+ * OXM_OF_ETH_TYPE must match 0x8847 or 0x8848 exactly.
+ *
+ * Format: 8-bit integer with 7 most-significant bits forced to 0.
+ * Only the lowest bit have a meaning.
+ *
+ * Masking: Not maskable. */
+ //#define OXM_OF_MPLS_BOS OXM_HEADER (0x8000, OFPXMT_OFB_MPLS_BOS, 1)
+ uint32 mpls_bos = 39; /* For OFPXMT_OFB_MPLS_BOS */
+
+ /* IEEE 802.1ah I-SID.
+ *
+ * For a packet with a PBB header, this is the I-SID from the
+ * outermost service tag.
+ *
+ * Prereqs:
+ * OXM_OF_ETH_TYPE must match 0x88E7 exactly.
+ *
+ * Format: 24-bit integer in network byte order.
+ *
+ * Masking: Arbitrary masks. */
+ //#define OXM_OF_PBB_ISID OXM_HEADER (0x8000, OFPXMT_OFB_PBB_ISID, 3)
+ //#define OXM_OF_PBB_ISID_W OXM_HEADER_W(0x8000, OFPXMT_OFB_PBB_ISID, 3)
+ uint32 pbb_isid = 40; /* For OFPXMT_OFB_PBB_ISID */
+
+ /* Logical Port Metadata.
+ *
+ * Metadata associated with a logical port.
+ * If the logical port performs encapsulation and decapsulation, this
+ * is the demultiplexing field from the encapsulation header.
+ * For example, for a packet received via GRE tunnel including a (32-bit) key,
+ * the key is stored in the low 32-bits and the high bits are zeroed.
+ * For a MPLS logical port, the low 20 bits represent the MPLS Label.
+ * For a VxLAN logical port, the low 24 bits represent the VNI.
+ * If the packet is not received through a logical port, the value is 0.
+ *
+ * Prereqs: None.
+ *
+ * Format: 64-bit integer in network byte order.
+ *
+ * Masking: Arbitrary masks. */
+ //#define OXM_OF_TUNNEL_ID OXM_HEADER (0x8000, OFPXMT_OFB_TUNNEL_ID, 8)
+ //#define OXM_OF_TUNNEL_ID_W OXM_HEADER_W(0x8000, OFPXMT_OFB_TUNNEL_ID, 8)
+ uint64 tunnel_id = 41; /* For OFPXMT_OFB_TUNNEL_ID */
+
+ /* The IPv6 Extension Header pseudo-field.
+ *
+ * Prereqs:
+ * OXM_OF_ETH_TYPE must match 0x86dd exactly
+ *
+ * Format: 16-bit integer with 7 most-significant bits forced to 0.
+ * Only the lower 9 bits have meaning.
+ *
+ * Masking: Maskable. */
+ //#define OXM_OF_IPV6_EXTHDR OXM_HEADER (0x8000, OFPXMT_OFB_IPV6_EXTHDR, 2)
+ //#define OXM_OF_IPV6_EXTHDR_W OXM_HEADER_W(0x8000, OFPXMT_OFB_IPV6_EXTHDR, 2)
+ uint32 ipv6_exthdr = 42; /* For OFPXMT_OFB_IPV6_EXTHDR */
+
+ }
+
+ /* Optional mask values (must be present when has_mask is true */
+ oneof mask {
+ uint64 table_metadata_mask = 105; /* For OFPXMT_OFB_METADATA */
+
+ bytes eth_dst_mask = 106; /* For OFPXMT_OFB_ETH_DST (exactly 6 bytes)*/
+ bytes eth_src_mask = 107; /* For OFPXMT_OFB_ETH_SRC (exactly 6 bytes)*/
+
+ uint32 vlan_vid_mask = 109; /* For OFPXMT_OFB_VLAN_VID */
+
+ uint32 ipv4_src_mask = 114; /* For OFPXMT_OFB_IPV4_SRC */
+ uint32 ipv4_dst_mask = 115; /* For OFPXMT_OFB_IPV4_DST */
+
+ uint32 arp_spa_mask = 125; /* For OFPXMT_OFB_ARP_SPA */
+ uint32 arp_tpa_mask = 126; /* For OFPXMT_OFB_ARP_TPA */
+
+ bytes ipv6_src_mask = 129; /* For OFPXMT_OFB_IPV6_SRC */
+ bytes ipv6_dst_mask = 130; /* For OFPXMT_OFB_IPV6_DST */
+
+ uint32 ipv6_flabel_mask = 131; /* For OFPXMT_OFB_IPV6_FLABEL */
+
+ uint32 pbb_isid_mask = 140; /* For OFPXMT_OFB_PBB_ISID */
+
+ uint64 tunnel_id_mask = 141; /* For OFPXMT_OFB_TUNNEL_ID */
+
+ uint32 ipv6_exthdr_mask = 142; /* For OFPXMT_OFB_IPV6_EXTHDR */
+ }
+
+}
+//#define OFPXMT_OFB_ALL ((UINT64_C(1) << 40) - 1)
+
+
+/* The VLAN id is 12-bits, so we can use the entire 16 bits to indicate
+ * special conditions.
+ */
+enum ofp_vlan_id {
+ OFPVID_NONE = 0x0000; /* No VLAN id was set. */
+ OFPVID_PRESENT = 0x1000; /* Bit that indicate that a VLAN id is set */
+};
+/* Define for compatibility */
+//#define OFP_VLAN_NONE OFPVID_NONE
+
+/* Bit definitions for IPv6 Extension Header pseudo-field. */
+enum ofp_ipv6exthdr_flags {
+ OFPIEH_INVALID = 0;
+ OFPIEH_NONEXT = 1; /* "No next header" encountered. */
+ OFPIEH_ESP = 2; /* Encrypted Sec Payload header present. */
+ OFPIEH_AUTH = 4; /* Authentication header present. */
+ OFPIEH_DEST = 8; /* 1 or 2 dest headers present. */
+ OFPIEH_FRAG = 16; /* Fragment header present. */
+ OFPIEH_ROUTER = 32; /* Router header present. */
+ OFPIEH_HOP = 64; /* Hop-by-hop header present. */
+ OFPIEH_UNREP = 128; /* Unexpected repeats encountered. */
+ OFPIEH_UNSEQ = 256; /* Unexpected sequencing encountered. */
+};
+
+/* Header for OXM experimenter match fields.
+ * The experimenter class should not use OXM_HEADER() macros for defining
+ * fields due to this extra header. */
+message ofp_oxm_experimenter_field {
+ uint32 oxm_header = 1; /* oxm_class = OFPXMC_EXPERIMENTER */
+ uint32 experimenter = 2; /* Experimenter ID which takes the same
+ form as in struct ofp_experimenter_header. */
+};
+
+/* ## ----------------- ## */
+/* ## OpenFlow Actions. ## */
+/* ## ----------------- ## */
+
+enum ofp_action_type {
+ OFPAT_OUTPUT = 0; /* Output to switch port. */
+ OFPAT_COPY_TTL_OUT = 11; /* Copy TTL "outwards" -- from next-to-outermost
+ to outermost */
+ OFPAT_COPY_TTL_IN = 12; /* Copy TTL "inwards" -- from outermost to
+ next-to-outermost */
+ OFPAT_SET_MPLS_TTL = 15; /* MPLS TTL */
+ OFPAT_DEC_MPLS_TTL = 16; /* Decrement MPLS TTL */
+
+ OFPAT_PUSH_VLAN = 17; /* Push a new VLAN tag */
+ OFPAT_POP_VLAN = 18; /* Pop the outer VLAN tag */
+ OFPAT_PUSH_MPLS = 19; /* Push a new MPLS tag */
+ OFPAT_POP_MPLS = 20; /* Pop the outer MPLS tag */
+ OFPAT_SET_QUEUE = 21; /* Set queue id when outputting to a port */
+ OFPAT_GROUP = 22; /* Apply group. */
+ OFPAT_SET_NW_TTL = 23; /* IP TTL. */
+ OFPAT_DEC_NW_TTL = 24; /* Decrement IP TTL. */
+ OFPAT_SET_FIELD = 25; /* Set a header field using OXM TLV format. */
+ OFPAT_PUSH_PBB = 26; /* Push a new PBB service tag (I-TAG) */
+ OFPAT_POP_PBB = 27; /* Pop the outer PBB service tag (I-TAG) */
+ OFPAT_EXPERIMENTER = 0xffff;
+};
+
+/* Action header that is common to all actions. The length includes the
+ * header and any padding used to make the action 64-bit aligned.
+ * NB: The length of an action *must* always be a multiple of eight. */
+message ofp_action {
+ ofp_action_type type = 1; /* One of OFPAT_*. */
+ oneof action {
+ ofp_action_output output = 2;
+ ofp_action_mpls_ttl mpls_ttl = 3;
+ ofp_action_push push = 4;
+ ofp_action_pop_mpls pop_mpls = 5;
+ ofp_action_group group = 6;
+ ofp_action_nw_ttl nw_ttl = 7;
+ ofp_action_set_field set_field = 8;
+ ofp_action_experimenter experimenter = 9;
+ }
+};
+
+enum ofp_controller_max_len {
+ OFPCML_INVALID = 0;
+ OFPCML_MAX = 0xffe5; /* maximum max_len value which can be used
+ to request a specific byte length. */
+ OFPCML_NO_BUFFER = 0xffff; /* indicates that no buffering should be
+ applied and the whole packet is to be
+ sent to the controller. */
+};
+
+/* Action structure for OFPAT_OUTPUT, which sends packets out 'port'.
+ * When the 'port' is the OFPP_CONTROLLER, 'max_len' indicates the max
+ * number of bytes to send. A 'max_len' of zero means no bytes of the
+ * packet should be sent. A 'max_len' of OFPCML_NO_BUFFER means that
+ * the packet is not buffered and the complete packet is to be sent to
+ * the controller. */
+message ofp_action_output {
+ uint32 port = 1; /* Output port. */
+ uint32 max_len = 2; /* Max length to send to controller. */
+};
+
+/* Action structure for OFPAT_SET_MPLS_TTL. */
+message ofp_action_mpls_ttl {
+ uint32 mpls_ttl = 1; /* MPLS TTL */
+};
+
+/* Action structure for OFPAT_PUSH_VLAN/MPLS/PBB. */
+message ofp_action_push {
+ uint32 ethertype = 1; /* Ethertype */
+};
+
+/* Action structure for OFPAT_POP_MPLS. */
+message ofp_action_pop_mpls {
+ uint32 ethertype = 1; /* Ethertype */
+};
+
+/* Action structure for OFPAT_GROUP. */
+message ofp_action_group {
+ uint32 group_id = 1; /* Group identifier. */
+};
+
+/* Action structure for OFPAT_SET_NW_TTL. */
+message ofp_action_nw_ttl {
+ uint32 nw_ttl = 1; /* IP TTL */
+};
+
+/* Action structure for OFPAT_SET_FIELD. */
+message ofp_action_set_field {
+ ofp_oxm_field field = 1;
+};
+
+/* Action header for OFPAT_EXPERIMENTER.
+ * The rest of the body is experimenter-defined. */
+message ofp_action_experimenter {
+ uint32 experimenter = 1; /* Experimenter ID which takes the same
+ form as in struct
+ ofp_experimenter_header. */
+ bytes data = 2;
+};
+
+/* ## ---------------------- ## */
+/* ## OpenFlow Instructions. ## */
+/* ## ---------------------- ## */
+
+enum ofp_instruction_type {
+ OFPIT_INVALID = 0;
+ OFPIT_GOTO_TABLE = 1; /* Setup the next table in the lookup
+ pipeline */
+ OFPIT_WRITE_METADATA = 2; /* Setup the metadata field for use later in
+ pipeline */
+ OFPIT_WRITE_ACTIONS = 3; /* Write the action(s) onto the datapath action
+ set */
+ OFPIT_APPLY_ACTIONS = 4; /* Applies the action(s) immediately */
+ OFPIT_CLEAR_ACTIONS = 5; /* Clears all actions from the datapath
+ action set */
+ OFPIT_METER = 6; /* Apply meter (rate limiter) */
+
+ OFPIT_EXPERIMENTER = 0xFFFF; /* Experimenter instruction */
+};
+
+/* Instruction header that is common to all instructions. The length includes
+ * the header and any padding used to make the instruction 64-bit aligned.
+ * NB: The length of an instruction *must* always be a multiple of eight. */
+message ofp_instruction {
+ uint32 type = 1; /* Instruction type */
+ oneof data {
+ ofp_instruction_goto_table goto_table = 2;
+ ofp_instruction_write_metadata write_metadata = 3;
+ ofp_instruction_actions actions = 4;
+ ofp_instruction_meter meter = 5;
+ ofp_instruction_experimenter experimenter = 6;
+ }
+};
+
+/* Instruction structure for OFPIT_GOTO_TABLE */
+message ofp_instruction_goto_table {
+ uint32 table_id = 1; /* Set next table in the lookup pipeline */
+};
+
+/* Instruction structure for OFPIT_WRITE_METADATA */
+message ofp_instruction_write_metadata {
+ uint64 metadata = 1; /* Metadata value to write */
+ uint64 metadata_mask = 2; /* Metadata write bitmask */
+};
+
+/* Instruction structure for OFPIT_WRITE/APPLY/CLEAR_ACTIONS */
+message ofp_instruction_actions {
+ repeated ofp_action actions = 1; /* 0 or more actions associated
+ with OFPIT_WRITE_ACTIONS and
+ OFPIT_APPLY_ACTIONS */
+};
+
+/* Instruction structure for OFPIT_METER */
+message ofp_instruction_meter {
+ uint32 meter_id = 1; /* Meter instance. */
+};
+
+/* Instruction structure for experimental instructions */
+message ofp_instruction_experimenter {
+ uint32 experimenter = 1; /* Experimenter ID which takes the same form
+ as in struct ofp_experimenter_header. */
+ /* Experimenter-defined arbitrary additional data. */
+ bytes data = 2;
+};
+
+/* ## --------------------------- ## */
+/* ## OpenFlow Flow Modification. ## */
+/* ## --------------------------- ## */
+
+enum ofp_flow_mod_command {
+ OFPFC_ADD = 0; /* New flow. */
+ OFPFC_MODIFY = 1; /* Modify all matching flows. */
+ OFPFC_MODIFY_STRICT = 2; /* Modify entry strictly matching wildcards and
+ priority. */
+ OFPFC_DELETE = 3; /* Delete all matching flows. */
+ OFPFC_DELETE_STRICT = 4; /* Delete entry strictly matching wildcards and
+ priority. */
+};
+
+/* Value used in "idle_timeout" and "hard_timeout" to indicate that the entry
+ * is permanent. */
+//#define OFP_FLOW_PERMANENT 0
+
+/* By default, choose a priority in the middle. */
+//#define OFP_DEFAULT_PRIORITY 0x8000
+
+enum ofp_flow_mod_flags {
+ OFPFF_INVALID = 0;
+ OFPFF_SEND_FLOW_REM = 1; /* Send flow removed message when flow
+ * expires or is deleted. */
+ OFPFF_CHECK_OVERLAP = 2; /* Check for overlapping entries first. */
+ OFPFF_RESET_COUNTS = 4; /* Reset flow packet and byte counts. */
+ OFPFF_NO_PKT_COUNTS = 8; /* Don't keep track of packet count. */
+ OFPFF_NO_BYT_COUNTS = 16; /* Don't keep track of byte count. */
+};
+
+/* Flow setup and teardown (controller -> datapath). */
+message ofp_flow_mod {
+ //ofp_header header;
+ uint64 cookie = 1; /* Opaque controller-issued identifier. */
+ uint64 cookie_mask = 2; /* Mask used to restrict the cookie bits
+ that must match when the command is
+ OFPFC_MODIFY* or OFPFC_DELETE*. A value
+ of 0 indicates no restriction. */
+ uint32 table_id = 3; /* ID of the table to put the flow in.
+ For OFPFC_DELETE_* commands, OFPTT_ALL
+ can also be used to delete matching
+ flows from all tables. */
+ ofp_flow_mod_command command = 4; /* One of OFPFC_*. */
+ uint32 idle_timeout = 5; /* Idle time before discarding (seconds). */
+ uint32 hard_timeout = 6; /* Max time before discarding (seconds). */
+ uint32 priority = 7; /* Priority level of flow entry. */
+ uint32 buffer_id = 8; /* Buffered packet to apply to, or
+ OFP_NO_BUFFER.
+ Not meaningful for OFPFC_DELETE*. */
+ uint32 out_port = 9; /* For OFPFC_DELETE* commands, require
+ matching entries to include this as an
+ output port. A value of OFPP_ANY
+ indicates no restriction. */
+ uint32 out_group = 10; /* For OFPFC_DELETE* commands, require
+ matching entries to include this as an
+ output group. A value of OFPG_ANY
+ indicates no restriction. */
+ uint32 flags = 11; /* Bitmap of OFPFF_* flags. */
+ ofp_match match = 12; /* Fields to match. Variable size. */
+ repeated ofp_instruction instructions = 13; /* 0 or more. */
+};
+
+/* Group numbering. Groups can use any number up to OFPG_MAX. */
+enum ofp_group {
+
+ OFPG_INVALID = 0;
+
+ /* Last usable group number. */
+ OFPG_MAX = 0x7fffff00;
+
+ /* Fake groups. */
+ OFPG_ALL = 0x7ffffffc; /* Represents all groups for group delete
+ commands. */
+ OFPG_ANY = 0x7fffffff; /* Special wildcard: no group specified. */
+};
+
+/* Group commands */
+enum ofp_group_mod_command {
+ OFPGC_ADD = 0; /* New group. */
+ OFPGC_MODIFY = 1; /* Modify all matching groups. */
+ OFPGC_DELETE = 2; /* Delete all matching groups. */
+};
+
+/* Bucket for use in groups. */
+message ofp_bucket {
+ uint32 weight = 1; /* Relative weight of bucket. Only
+ defined for select groups. */
+ uint32 watch_port = 2; /* Port whose state affects whether this
+ bucket is live. Only required for fast
+ failover groups. */
+ uint32 watch_group = 3; /* Group whose state affects whether this
+ bucket is live. Only required for fast
+ failover groups. */
+ repeated ofp_action actions = 4;
+};
+
+/* Group setup and teardown (controller -> datapath). */
+message ofp_group_mod {
+ //ofp_header header;
+ ofp_group_mod_command command = 1; /* One of OFPGC_*. */
+ ofp_group_type type = 2; /* One of OFPGT_*. */
+ uint32 group_id = 3; /* Group identifier. */
+ repeated ofp_bucket buckets = 4;
+};
+
+/* Group types. Values in the range [128; 255] are reserved for experimental
+ * use. */
+enum ofp_group_type {
+ OFPGT_ALL = 0; /* All (multicast/broadcast) group. */
+ OFPGT_SELECT = 1; /* Select group. */
+ OFPGT_INDIRECT = 2; /* Indirect group. */
+ OFPGT_FF = 3; /* Fast failover group. */
+};
+
+/* Special buffer-id to indicate 'no buffer' */
+//#define OFP_NO_BUFFER 0xffffffff
+
+/* Send packet (controller -> datapath). */
+message ofp_packet_out {
+ //ofp_header header;
+ uint32 buffer_id = 1; /* ID assigned by datapath (OFP_NO_BUFFER
+ if none). */
+ uint32 in_port = 2; /* Packet's input port or OFPP_CONTROLLER.*/
+ repeated ofp_action actions = 3; /* Action list - 0 or more. */
+ /* The variable size action list is optionally followed by packet data.
+ * This data is only present and meaningful if buffer_id == -1. */
+ bytes data = 4; /* Packet data. */
+};
+
+/* Why is this packet being sent to the controller? */
+enum ofp_packet_in_reason {
+ OFPR_NO_MATCH = 0; /* No matching flow (table-miss flow entry). */
+ OFPR_ACTION = 1; /* Action explicitly output to controller. */
+ OFPR_INVALID_TTL = 2; /* Packet has invalid TTL */
+};
+
+/* Packet received on port (datapath -> controller). */
+message ofp_packet_in {
+ //ofp_header header;
+ uint32 buffer_id = 1; /* ID assigned by datapath. */
+ ofp_packet_in_reason reason = 2; /* Reason packet is being sent */
+ uint32 table_id = 3; /* ID of the table that was looked up */
+ uint64 cookie = 4; /* Cookie of the flow entry that was looked up. */
+ ofp_match match = 5; /* Packet metadata. Variable size. */
+ bytes data = 6; /* Ethernet frame */
+};
+
+/* Why was this flow removed? */
+enum ofp_flow_removed_reason {
+ OFPRR_IDLE_TIMEOUT = 0; /* Flow idle time exceeded idle_timeout. */
+ OFPRR_HARD_TIMEOUT = 1; /* Time exceeded hard_timeout. */
+ OFPRR_DELETE = 2; /* Evicted by a DELETE flow mod. */
+ OFPRR_GROUP_DELETE = 3; /* Group was removed. */
+ OFPRR_METER_DELETE = 4; /* Meter was removed */
+};
+
+/* Flow removed (datapath -> controller). */
+message ofp_flow_removed {
+ //ofp_header header;
+ uint64 cookie = 1; /* Opaque controller-issued identifier. */
+
+ uint32 priority = 2; /* Priority level of flow entry. */
+ ofp_flow_removed_reason reason = 3; /* One of OFPRR_*. */
+ uint32 table_id = 4; /* ID of the table */
+
+ uint32 duration_sec = 5; /* Time flow was alive in seconds. */
+ uint32 duration_nsec = 6; /* Time flow was alive in nanoseconds beyond
+ duration_sec. */
+ uint32 idle_timeout = 7; /* Idle timeout from original flow mod. */
+ uint32 hard_timeout = 8; /* Hard timeout from original flow mod. */
+ uint64 packet_count = 9;
+ uint64 byte_count = 10;
+ ofp_match match = 121; /* Description of fields. Variable size. */
+};
+
+/* Meter numbering. Flow meters can use any number up to OFPM_MAX. */
+enum ofp_meter {
+ OFPM_ZERO = 0;
+ /* Last usable meter. */
+ OFPM_MAX = 0x7fff0000;
+
+ /* Virtual meters. */
+ OFPM_SLOWPATH = 0x7ffffffd; /* Meter for slow datapath. */
+ OFPM_CONTROLLER = 0x7ffffffe; /* Meter for controller connection. */
+ OFPM_ALL = 0x7fffffff; /* Represents all meters for stat requests
+ commands. */
+};
+
+/* Meter band types */
+enum ofp_meter_band_type {
+ OFPMBT_INVALID = 0;
+ OFPMBT_DROP = 1; /* Drop packet. */
+ OFPMBT_DSCP_REMARK = 2; /* Remark DSCP in the IP header. */
+ OFPMBT_EXPERIMENTER = 0xFFFF; /* Experimenter meter band. */
+};
+
+/* Common header for all meter bands */
+message ofp_meter_band_header {
+ ofp_meter_band_type type = 1; /* One of OFPMBT_*. */
+ uint32 len = 2; /* Length in bytes of this band. */
+ uint32 rate = 3; /* Rate for this band. */
+ uint32 burst_size = 4;/* Size of bursts. */
+};
+
+/* OFPMBT_DROP band - drop packets */
+message ofp_meter_band_drop {
+ uint32 type = 1; /* OFPMBT_DROP. */
+ uint32 len = 2; /* Length in bytes of this band. */
+ uint32 rate = 3; /* Rate for dropping packets. */
+ uint32 burst_size = 4;/* Size of bursts. */
+};
+
+/* OFPMBT_DSCP_REMARK band - Remark DSCP in the IP header */
+message ofp_meter_band_dscp_remark {
+ uint32 type = 1; /* OFPMBT_DSCP_REMARK. */
+ uint32 len = 2; /* Length in bytes of this band. */
+ uint32 rate = 3; /* Rate for remarking packets. */
+ uint32 burst_size = 4; /* Size of bursts. */
+ uint32 prec_level = 5; /* Number of drop precedence level to add. */
+};
+
+/* OFPMBT_EXPERIMENTER band - Experimenter type.
+ * The rest of the band is experimenter-defined. */
+message ofp_meter_band_experimenter {
+ ofp_meter_band_type type = 1; /* One of OFPMBT_*. */
+ uint32 len = 2; /* Length in bytes of this band. */
+ uint32 rate = 3; /* Rate for this band. */
+ uint32 burst_size = 4; /* Size of bursts. */
+ uint32 experimenter = 5;/* Experimenter ID which takes the
+ same form as in struct
+ ofp_experimenter_header. */
+};
+
+/* Meter commands */
+enum ofp_meter_mod_command {
+ OFPMC_ADD = 0; /* New meter. */
+ OFPMC_MODIFY = 1; /* Modify specified meter. */
+ OFPMC_DELETE = 2; /* Delete specified meter. */
+};
+
+/* Meter configuration flags */
+enum ofp_meter_flags {
+ OFPMF_INVALID = 0;
+ OFPMF_KBPS = 1; /* Rate value in kb/s (kilo-bit per second). */
+ OFPMF_PKTPS = 2; /* Rate value in packet/sec. */
+ OFPMF_BURST = 4; /* Do burst size. */
+ OFPMF_STATS = 8; /* Collect statistics. */
+};
+
+/* Meter configuration. OFPT_METER_MOD. */
+message ofp_meter_mod {
+// ofp_header header = 1;
+ ofp_meter_mod_command command = 1; /* One of OFPMC_*. */
+ uint32 flags = 2; /* Bitmap of OFPMF_* flags. */
+ uint32 meter_id = 3; /* Meter instance. */
+ repeated ofp_meter_band_header bands = 4; /* The band list length is
+ inferred from the length field
+ in the header. */
+};
+
+/* Values for 'type' in ofp_error_message. These values are immutable: they
+ * will not change in future versions of the protocol (although new values may
+ * be added). */
+enum ofp_error_type {
+ OFPET_HELLO_FAILED = 0; /* Hello protocol failed. */
+ OFPET_BAD_REQUEST = 1; /* Request was not understood. */
+ OFPET_BAD_ACTION = 2; /* Error in action description. */
+ OFPET_BAD_INSTRUCTION = 3; /* Error in instruction list. */
+ OFPET_BAD_MATCH = 4; /* Error in match. */
+ OFPET_FLOW_MOD_FAILED = 5; /* Problem modifying flow entry. */
+ OFPET_GROUP_MOD_FAILED = 6; /* Problem modifying group entry. */
+ OFPET_PORT_MOD_FAILED = 7; /* Port mod request failed. */
+ OFPET_TABLE_MOD_FAILED = 8; /* Table mod request failed. */
+ OFPET_QUEUE_OP_FAILED = 9; /* Queue operation failed. */
+ OFPET_SWITCH_CONFIG_FAILED = 10; /* Switch config request failed. */
+ OFPET_ROLE_REQUEST_FAILED = 11; /* Controller Role request failed. */
+ OFPET_METER_MOD_FAILED = 12; /* Error in meter. */
+ OFPET_TABLE_FEATURES_FAILED = 13; /* Setting table features failed. */
+ OFPET_EXPERIMENTER = 0xffff; /* Experimenter error messages. */
+};
+
+/* ofp_error_msg 'code' values for OFPET_HELLO_FAILED. 'data' contains an
+ * ASCII text string that may give failure details. */
+enum ofp_hello_failed_code {
+ OFPHFC_INCOMPATIBLE = 0; /* No compatible version. */
+ OFPHFC_EPERM = 1; /* Permissions error. */
+};
+
+/* ofp_error_msg 'code' values for OFPET_BAD_REQUEST. 'data' contains at least
+ * the first 64 bytes of the failed request. */
+enum ofp_bad_request_code {
+ OFPBRC_BAD_VERSION = 0; /* ofp_header.version not supported. */
+ OFPBRC_BAD_TYPE = 1; /* ofp_header.type not supported. */
+ OFPBRC_BAD_MULTIPART = 2; /* ofp_multipart_request.type not supported.
+ */
+ OFPBRC_BAD_EXPERIMENTER = 3; /* Experimenter id not supported
+ * (in ofp_experimenter_header or
+ * ofp_multipart_request or
+ * ofp_multipart_reply). */
+ OFPBRC_BAD_EXP_TYPE = 4; /* Experimenter type not supported. */
+ OFPBRC_EPERM = 5; /* Permissions error. */
+ OFPBRC_BAD_LEN = 6; /* Wrong request length for type. */
+ OFPBRC_BUFFER_EMPTY = 7; /* Specified buffer has already been used. */
+ OFPBRC_BUFFER_UNKNOWN = 8; /* Specified buffer does not exist. */
+ OFPBRC_BAD_TABLE_ID = 9; /* Specified table-id invalid or does not
+ * exist. */
+ OFPBRC_IS_SLAVE = 10; /* Denied because controller is slave. */
+ OFPBRC_BAD_PORT = 11; /* Invalid port. */
+ OFPBRC_BAD_PACKET = 12; /* Invalid packet in packet-out. */
+ OFPBRC_MULTIPART_BUFFER_OVERFLOW = 13; /* ofp_multipart_request
+ overflowed the assigned buffer. */
+};
+
+/* ofp_error_msg 'code' values for OFPET_BAD_ACTION. 'data' contains at least
+ * the first 64 bytes of the failed request. */
+enum ofp_bad_action_code {
+ OFPBAC_BAD_TYPE = 0; /* Unknown or unsupported action type. */
+ OFPBAC_BAD_LEN = 1; /* Length problem in actions. */
+ OFPBAC_BAD_EXPERIMENTER = 2; /* Unknown experimenter id specified. */
+ OFPBAC_BAD_EXP_TYPE = 3; /* Unknown action for experimenter id. */
+ OFPBAC_BAD_OUT_PORT = 4; /* Problem validating output port. */
+ OFPBAC_BAD_ARGUMENT = 5; /* Bad action argument. */
+ OFPBAC_EPERM = 6; /* Permissions error. */
+ OFPBAC_TOO_MANY = 7; /* Can't handle this many actions. */
+ OFPBAC_BAD_QUEUE = 8; /* Problem validating output queue. */
+ OFPBAC_BAD_OUT_GROUP = 9; /* Invalid group id in forward action. */
+ OFPBAC_MATCH_INCONSISTENT = 10; /* Action can't apply for this match,
+ or Set-Field missing prerequisite. */
+ OFPBAC_UNSUPPORTED_ORDER = 11; /* Action order is unsupported for the
+ action list in an Apply-Actions instruction */
+ OFPBAC_BAD_TAG = 12; /* Actions uses an unsupported
+ tag/encap. */
+ OFPBAC_BAD_SET_TYPE = 13; /* Unsupported type in SET_FIELD action. */
+ OFPBAC_BAD_SET_LEN = 14; /* Length problem in SET_FIELD action. */
+ OFPBAC_BAD_SET_ARGUMENT = 15; /* Bad argument in SET_FIELD action. */
+};
+
+/* ofp_error_msg 'code' values for OFPET_BAD_INSTRUCTION. 'data' contains at
+ * least the first 64 bytes of the failed request. */
+enum ofp_bad_instruction_code {
+ OFPBIC_UNKNOWN_INST = 0; /* Unknown instruction. */
+ OFPBIC_UNSUP_INST = 1; /* Switch or table does not support the
+ instruction. */
+ OFPBIC_BAD_TABLE_ID = 2; /* Invalid Table-ID specified. */
+ OFPBIC_UNSUP_METADATA = 3; /* Metadata value unsupported by datapath. */
+ OFPBIC_UNSUP_METADATA_MASK = 4; /* Metadata mask value unsupported by
+ datapath. */
+ OFPBIC_BAD_EXPERIMENTER = 5; /* Unknown experimenter id specified. */
+ OFPBIC_BAD_EXP_TYPE = 6; /* Unknown instruction for experimenter id. */
+ OFPBIC_BAD_LEN = 7; /* Length problem in instructions. */
+ OFPBIC_EPERM = 8; /* Permissions error. */
+};
+
+/* ofp_error_msg 'code' values for OFPET_BAD_MATCH. 'data' contains at least
+ * the first 64 bytes of the failed request. */
+enum ofp_bad_match_code {
+ OFPBMC_BAD_TYPE = 0; /* Unsupported match type specified by the
+ match */
+ OFPBMC_BAD_LEN = 1; /* Length problem in match. */
+ OFPBMC_BAD_TAG = 2; /* Match uses an unsupported tag/encap. */
+ OFPBMC_BAD_DL_ADDR_MASK = 3; /* Unsupported datalink addr mask - switch
+ does not support arbitrary datalink
+ address mask. */
+ OFPBMC_BAD_NW_ADDR_MASK = 4; /* Unsupported network addr mask - switch
+ does not support arbitrary network
+ address mask. */
+ OFPBMC_BAD_WILDCARDS = 5; /* Unsupported combination of fields masked
+ or omitted in the match. */
+ OFPBMC_BAD_FIELD = 6; /* Unsupported field type in the match. */
+ OFPBMC_BAD_VALUE = 7; /* Unsupported value in a match field. */
+ OFPBMC_BAD_MASK = 8; /* Unsupported mask specified in the match,
+ field is not dl-address or nw-address. */
+ OFPBMC_BAD_PREREQ = 9; /* A prerequisite was not met. */
+ OFPBMC_DUP_FIELD = 10; /* A field type was duplicated. */
+ OFPBMC_EPERM = 11; /* Permissions error. */
+};
+
+/* ofp_error_msg 'code' values for OFPET_FLOW_MOD_FAILED. 'data' contains
+ * at least the first 64 bytes of the failed request. */
+enum ofp_flow_mod_failed_code {
+ OFPFMFC_UNKNOWN = 0; /* Unspecified error. */
+ OFPFMFC_TABLE_FULL = 1; /* Flow not added because table was full. */
+ OFPFMFC_BAD_TABLE_ID = 2; /* Table does not exist */
+ OFPFMFC_OVERLAP = 3; /* Attempted to add overlapping flow with
+ CHECK_OVERLAP flag set. */
+ OFPFMFC_EPERM = 4; /* Permissions error. */
+ OFPFMFC_BAD_TIMEOUT = 5; /* Flow not added because of unsupported
+ idle/hard timeout. */
+ OFPFMFC_BAD_COMMAND = 6; /* Unsupported or unknown command. */
+ OFPFMFC_BAD_FLAGS = 7; /* Unsupported or unknown flags. */
+};
+
+/* ofp_error_msg 'code' values for OFPET_GROUP_MOD_FAILED. 'data' contains
+ * at least the first 64 bytes of the failed request. */
+enum ofp_group_mod_failed_code {
+ OFPGMFC_GROUP_EXISTS = 0; /* Group not added because a group ADD
+ attempted to replace an
+ already-present group. */
+ OFPGMFC_INVALID_GROUP = 1; /* Group not added because Group
+ specified is invalid. */
+ OFPGMFC_WEIGHT_UNSUPPORTED = 2; /* Switch does not support unequal load
+ sharing with select groups. */
+ OFPGMFC_OUT_OF_GROUPS = 3; /* The group table is full. */
+ OFPGMFC_OUT_OF_BUCKETS = 4; /* The maximum number of action buckets
+ for a group has been exceeded. */
+ OFPGMFC_CHAINING_UNSUPPORTED = 5; /* Switch does not support groups that
+ forward to groups. */
+ OFPGMFC_WATCH_UNSUPPORTED = 6; /* This group cannot watch the
+ watch_port or watch_group specified.
+ */
+ OFPGMFC_LOOP = 7; /* Group entry would cause a loop. */
+ OFPGMFC_UNKNOWN_GROUP = 8; /* Group not modified because a group
+ MODIFY attempted to modify a
+ non-existent group. */
+ OFPGMFC_CHAINED_GROUP = 9; /* Group not deleted because another
+ group is forwarding to it. */
+ OFPGMFC_BAD_TYPE = 10; /* Unsupported or unknown group type. */
+ OFPGMFC_BAD_COMMAND = 11; /* Unsupported or unknown command. */
+ OFPGMFC_BAD_BUCKET = 12; /* Error in bucket. */
+ OFPGMFC_BAD_WATCH = 13; /* Error in watch port/group. */
+ OFPGMFC_EPERM = 14; /* Permissions error. */
+};
+
+/* ofp_error_msg 'code' values for OFPET_PORT_MOD_FAILED. 'data' contains
+ * at least the first 64 bytes of the failed request. */
+enum ofp_port_mod_failed_code {
+ OFPPMFC_BAD_PORT = 0; /* Specified port number does not exist. */
+ OFPPMFC_BAD_HW_ADDR = 1; /* Specified hardware address does not
+ * match the port number. */
+ OFPPMFC_BAD_CONFIG = 2; /* Specified config is invalid. */
+ OFPPMFC_BAD_ADVERTISE = 3; /* Specified advertise is invalid. */
+ OFPPMFC_EPERM = 4; /* Permissions error. */
+};
+
+/* ofp_error_msg 'code' values for OFPET_TABLE_MOD_FAILED. 'data' contains
+ * at least the first 64 bytes of the failed request. */
+enum ofp_table_mod_failed_code {
+ OFPTMFC_BAD_TABLE = 0; /* Specified table does not exist. */
+ OFPTMFC_BAD_CONFIG = 1; /* Specified config is invalid. */
+ OFPTMFC_EPERM = 2; /* Permissions error. */
+};
+
+/* ofp_error msg 'code' values for OFPET_QUEUE_OP_FAILED. 'data' contains
+ * at least the first 64 bytes of the failed request */
+enum ofp_queue_op_failed_code {
+ OFPQOFC_BAD_PORT = 0; /* Invalid port (or port does not exist). */
+ OFPQOFC_BAD_QUEUE = 1; /* Queue does not exist. */
+ OFPQOFC_EPERM = 2; /* Permissions error. */
+};
+
+/* ofp_error_msg 'code' values for OFPET_SWITCH_CONFIG_FAILED. 'data' contains
+ * at least the first 64 bytes of the failed request. */
+enum ofp_switch_config_failed_code {
+ OFPSCFC_BAD_FLAGS = 0; /* Specified flags is invalid. */
+ OFPSCFC_BAD_LEN = 1; /* Specified len is invalid. */
+ OFPSCFC_EPERM = 2; /* Permissions error. */
+};
+
+/* ofp_error_msg 'code' values for OFPET_ROLE_REQUEST_FAILED. 'data' contains
+ * at least the first 64 bytes of the failed request. */
+enum ofp_role_request_failed_code {
+ OFPRRFC_STALE = 0; /* Stale Message: old generation_id. */
+ OFPRRFC_UNSUP = 1; /* Controller role change unsupported. */
+ OFPRRFC_BAD_ROLE = 2; /* Invalid role. */
+};
+
+/* ofp_error_msg 'code' values for OFPET_METER_MOD_FAILED. 'data' contains
+ * at least the first 64 bytes of the failed request. */
+enum ofp_meter_mod_failed_code {
+ OFPMMFC_UNKNOWN = 0; /* Unspecified error. */
+ OFPMMFC_METER_EXISTS = 1; /* Meter not added because a Meter ADD
+ * attempted to replace an existing Meter. */
+ OFPMMFC_INVALID_METER = 2; /* Meter not added because Meter specified
+ * is invalid,
+ * or invalid meter in meter action. */
+ OFPMMFC_UNKNOWN_METER = 3; /* Meter not modified because a Meter MODIFY
+ * attempted to modify a non-existent Meter,
+ * or bad meter in meter action. */
+ OFPMMFC_BAD_COMMAND = 4; /* Unsupported or unknown command. */
+ OFPMMFC_BAD_FLAGS = 5; /* Flag configuration unsupported. */
+ OFPMMFC_BAD_RATE = 6; /* Rate unsupported. */
+ OFPMMFC_BAD_BURST = 7; /* Burst size unsupported. */
+ OFPMMFC_BAD_BAND = 8; /* Band unsupported. */
+ OFPMMFC_BAD_BAND_VALUE = 9; /* Band value unsupported. */
+ OFPMMFC_OUT_OF_METERS = 10; /* No more meters available. */
+ OFPMMFC_OUT_OF_BANDS = 11; /* The maximum number of properties
+ * for a meter has been exceeded. */
+};
+
+/* ofp_error_msg 'code' values for OFPET_TABLE_FEATURES_FAILED. 'data' contains
+ * at least the first 64 bytes of the failed request. */
+enum ofp_table_features_failed_code {
+ OFPTFFC_BAD_TABLE = 0; /* Specified table does not exist. */
+ OFPTFFC_BAD_METADATA = 1; /* Invalid metadata mask. */
+ OFPTFFC_BAD_TYPE = 2; /* Unknown property type. */
+ OFPTFFC_BAD_LEN = 3; /* Length problem in properties. */
+ OFPTFFC_BAD_ARGUMENT = 4; /* Unsupported property value. */
+ OFPTFFC_EPERM = 5; /* Permissions error. */
+};
+
+/* OFPT_ERROR: Error message (datapath -> controller). */
+message ofp_error_msg {
+ //ofp_header header;
+ uint32 type = 1;
+ uint32 code = 2;
+ bytes data = 3; /* Variable-length data. Interpreted based
+ on the type and code. No padding. */
+};
+
+/* OFPET_EXPERIMENTER: Error message (datapath -> controller). */
+message ofp_error_experimenter_msg {
+ //ofp_header header;
+
+ uint32 type = 1; /* OFPET_EXPERIMENTER. */
+ uint32 exp_type = 2; /* Experimenter defined. */
+ uint32 experimenter = 3; /* Experimenter ID which takes the same form
+ as in struct ofp_experimenter_header. */
+ bytes data = 4; /* Variable-length data. Interpreted based
+ on the type and code. No padding. */
+};
+
+enum ofp_multipart_type {
+ /* Description of this OpenFlow switch.
+ * The request body is empty.
+ * The reply body is struct ofp_desc. */
+ OFPMP_DESC = 0;
+
+ /* Individual flow statistics.
+ * The request body is struct ofp_flow_stats_request.
+ * The reply body is an array of struct ofp_flow_stats. */
+ OFPMP_FLOW = 1;
+
+ /* Aggregate flow statistics.
+ * The request body is struct ofp_aggregate_stats_request.
+ * The reply body is struct ofp_aggregate_stats_reply. */
+ OFPMP_AGGREGATE = 2;
+
+ /* Flow table statistics.
+ * The request body is empty.
+ * The reply body is an array of struct ofp_table_stats. */
+ OFPMP_TABLE = 3;
+
+ /* Port statistics.
+ * The request body is struct ofp_port_stats_request.
+ * The reply body is an array of struct ofp_port_stats. */
+ OFPMP_PORT_STATS = 4;
+
+ /* Queue statistics for a port
+ * The request body is struct ofp_queue_stats_request.
+ * The reply body is an array of struct ofp_queue_stats */
+ OFPMP_QUEUE = 5;
+
+ /* Group counter statistics.
+ * The request body is struct ofp_group_stats_request.
+ * The reply is an array of struct ofp_group_stats. */
+ OFPMP_GROUP = 6;
+
+ /* Group description.
+ * The request body is empty.
+ * The reply body is an array of struct ofp_group_desc. */
+ OFPMP_GROUP_DESC = 7;
+
+ /* Group features.
+ * The request body is empty.
+ * The reply body is struct ofp_group_features. */
+ OFPMP_GROUP_FEATURES = 8;
+
+ /* Meter statistics.
+ * The request body is struct ofp_meter_multipart_requests.
+ * The reply body is an array of struct ofp_meter_stats. */
+ OFPMP_METER = 9;
+
+ /* Meter configuration.
+ * The request body is struct ofp_meter_multipart_requests.
+ * The reply body is an array of struct ofp_meter_config. */
+ OFPMP_METER_CONFIG = 10;
+
+ /* Meter features.
+ * The request body is empty.
+ * The reply body is struct ofp_meter_features. */
+ OFPMP_METER_FEATURES = 11;
+
+ /* Table features.
+ * The request body is either empty or contains an array of
+ * struct ofp_table_features containing the controller's
+ * desired view of the switch. If the switch is unable to
+ * set the specified view an error is returned.
+ * The reply body is an array of struct ofp_table_features. */
+ OFPMP_TABLE_FEATURES = 12;
+
+ /* Port description.
+ * The request body is empty.
+ * The reply body is an array of struct ofp_port. */
+ OFPMP_PORT_DESC = 13;
+
+ /* Experimenter extension.
+ * The request and reply bodies begin with
+ * struct ofp_experimenter_multipart_header.
+ * The request and reply bodies are otherwise experimenter-defined. */
+ OFPMP_EXPERIMENTER = 0xffff;
+};
+
+/* Backward compatibility with 1.3.1 - avoid breaking the API. */
+//#define ofp_multipart_types ofp_multipart_type
+
+enum ofp_multipart_request_flags {
+ OFPMPF_REQ_INVALID = 0;
+ OFPMPF_REQ_MORE = 1; /* More requests to follow. */
+};
+
+message ofp_multipart_request {
+ //ofp_header header;
+ ofp_multipart_type type = 1; /* One of the OFPMP_* constants. */
+ uint32 flags = 2; /* OFPMPF_REQ_* flags. */
+ bytes body = 3; /* Body of the request. 0 or more bytes. */
+};
+
+enum ofp_multipart_reply_flags {
+ OFPMPF_REPLY_INVALID = 0;
+ OFPMPF_REPLY_MORE = 1; /* More replies to follow. */
+};
+
+message ofp_multipart_reply {
+ //ofp_header header;
+ ofp_multipart_type type = 1; /* One of the OFPMP_* constants. */
+ uint32 flags = 2; /* OFPMPF_REPLY_* flags. */
+ bytes body = 3; /* Body of the reply. 0 or more bytes. */
+};
+
+//#define DESC_STR_LEN 256
+//#define SERIAL_NUM_LEN 32
+/* Body of reply to OFPMP_DESC request. Each entry is a NULL-terminated
+ * ASCII string. */
+message ofp_desc {
+ string mfr_desc = 1; /* Manufacturer description. */
+ string hw_desc = 2; /* Hardware description. */
+ string sw_desc = 3; /* Software description. */
+ string serial_num = 4; /* Serial number. */
+ string dp_desc = 5; /* Human readable description of datapath. */
+};
+
+/* Body for ofp_multipart_request of type OFPMP_FLOW. */
+message ofp_flow_stats_request {
+ uint32 table_id = 1; /* ID of table to read (from ofp_table_stats),
+ OFPTT_ALL for all tables. */
+ uint32 out_port = 2; /* Require matching entries to include this
+ as an output port. A value of OFPP_ANY
+ indicates no restriction. */
+ uint32 out_group = 3; /* Require matching entries to include this
+ as an output group. A value of OFPG_ANY
+ indicates no restriction. */
+ uint64 cookie = 4; /* Require matching entries to contain this
+ cookie value */
+ uint64 cookie_mask = 5; /* Mask used to restrict the cookie bits that
+ must match. A value of 0 indicates
+ no restriction. */
+ ofp_match match = 6; /* Fields to match. Variable size. */
+};
+
+/* Body of reply to OFPMP_FLOW request. */
+message ofp_flow_stats {
+ uint64 id = 14; /* Unique ID of flow within device. */
+ uint32 table_id = 1; /* ID of table flow came from. */
+ uint32 duration_sec = 2; /* Time flow has been alive in seconds. */
+ uint32 duration_nsec = 3; /* Time flow has been alive in nanoseconds
+ beyond duration_sec. */
+ uint32 priority = 4; /* Priority of the entry. */
+ uint32 idle_timeout = 5; /* Number of seconds idle before expiration. */
+ uint32 hard_timeout = 6; /* Number of seconds before expiration. */
+ uint32 flags = 7; /* Bitmap of OFPFF_* flags. */
+ uint64 cookie = 8; /* Opaque controller-issued identifier. */
+ uint64 packet_count = 9; /* Number of packets in flow. */
+ uint64 byte_count = 10; /* Number of bytes in flow. */
+ ofp_match match = 12; /* Description of fields. Variable size. */
+ repeated ofp_instruction instructions = 13; /* Instruction set
+ (0 or more) */
+};
+
+/* Body for ofp_multipart_request of type OFPMP_AGGREGATE. */
+message ofp_aggregate_stats_request {
+ uint32 table_id = 1; /* ID of table to read (from ofp_table_stats)
+ OFPTT_ALL for all tables. */
+ uint32 out_port = 2; /* Require matching entries to include this
+ as an output port. A value of OFPP_ANY
+ indicates no restriction. */
+ uint32 out_group = 3; /* Require matching entries to include this
+ as an output group. A value of OFPG_ANY
+ indicates no restriction. */
+ uint64 cookie = 4; /* Require matching entries to contain this
+ cookie value */
+ uint64 cookie_mask = 5; /* Mask used to restrict the cookie bits that
+ must match. A value of 0 indicates
+ no restriction. */
+ ofp_match match = 6; /* Fields to match. Variable size. */
+};
+
+/* Body of reply to OFPMP_AGGREGATE request. */
+message ofp_aggregate_stats_reply {
+ uint64 packet_count = 1; /* Number of packets in flows. */
+ uint64 byte_count = 2; /* Number of bytes in flows. */
+ uint32 flow_count = 3; /* Number of flows. */
+};
+
+/* Table Feature property types.
+ * Low order bit cleared indicates a property for a regular Flow Entry.
+ * Low order bit set indicates a property for the Table-Miss Flow Entry.
+ */
+enum ofp_table_feature_prop_type {
+ OFPTFPT_INSTRUCTIONS = 0; /* Instructions property. */
+ OFPTFPT_INSTRUCTIONS_MISS = 1; /* Instructions for table-miss. */
+ OFPTFPT_NEXT_TABLES = 2; /* Next Table property. */
+ OFPTFPT_NEXT_TABLES_MISS = 3; /* Next Table for table-miss. */
+ OFPTFPT_WRITE_ACTIONS = 4; /* Write Actions property. */
+ OFPTFPT_WRITE_ACTIONS_MISS = 5; /* Write Actions for table-miss. */
+ OFPTFPT_APPLY_ACTIONS = 6; /* Apply Actions property. */
+ OFPTFPT_APPLY_ACTIONS_MISS = 7; /* Apply Actions for table-miss. */
+ OFPTFPT_MATCH = 8; /* Match property. */
+ OFPTFPT_WILDCARDS = 10; /* Wildcards property. */
+ OFPTFPT_WRITE_SETFIELD = 12; /* Write Set-Field property. */
+ OFPTFPT_WRITE_SETFIELD_MISS = 13; /* Write Set-Field for table-miss. */
+ OFPTFPT_APPLY_SETFIELD = 14; /* Apply Set-Field property. */
+ OFPTFPT_APPLY_SETFIELD_MISS = 15; /* Apply Set-Field for table-miss. */
+ OFPTFPT_EXPERIMENTER = 0xFFFE; /* Experimenter property. */
+ OFPTFPT_EXPERIMENTER_MISS = 0xFFFF; /* Experimenter for table-miss. */
+};
+
+/* Common header for all Table Feature Properties */
+message ofp_table_feature_property {
+ ofp_table_feature_prop_type type = 1; /* One of OFPTFPT_*. */
+ oneof value {
+ ofp_table_feature_prop_instructions instructions = 2;
+ ofp_table_feature_prop_next_tables next_tables = 3;
+ ofp_table_feature_prop_actions actions = 4;
+ ofp_table_feature_prop_oxm oxm = 5;
+ ofp_table_feature_prop_experimenter experimenter = 6;
+ }
+};
+
+/* Instructions property */
+message ofp_table_feature_prop_instructions {
+ /* One of OFPTFPT_INSTRUCTIONS,
+ OFPTFPT_INSTRUCTIONS_MISS. */
+ repeated ofp_instruction instructions = 1; /* List of instructions */
+};
+
+/* Next Tables property */
+message ofp_table_feature_prop_next_tables {
+ /* One of OFPTFPT_NEXT_TABLES,
+ OFPTFPT_NEXT_TABLES_MISS. */
+ repeated uint32 next_table_ids = 1; /* List of table ids. */
+};
+
+/* Actions property */
+message ofp_table_feature_prop_actions {
+ /* One of OFPTFPT_WRITE_ACTIONS,
+ OFPTFPT_WRITE_ACTIONS_MISS,
+ OFPTFPT_APPLY_ACTIONS,
+ OFPTFPT_APPLY_ACTIONS_MISS. */
+ repeated ofp_action actions = 1; /* List of actions */
+};
+
+/* Match, Wildcard or Set-Field property */
+message ofp_table_feature_prop_oxm {
+ /* One of OFPTFPT_MATCH,
+ OFPTFPT_WILDCARDS,
+ OFPTFPT_WRITE_SETFIELD,
+ OFPTFPT_WRITE_SETFIELD_MISS,
+ OFPTFPT_APPLY_SETFIELD,
+ OFPTFPT_APPLY_SETFIELD_MISS. */
+ /* TODO is this a uint32??? */
+ repeated uint32 oxm_ids = 3; /* Array of OXM headers */
+};
+
+/* Experimenter table feature property */
+message ofp_table_feature_prop_experimenter {
+ /* One of OFPTFPT_EXPERIMENTER,
+ OFPTFPT_EXPERIMENTER_MISS. */
+ uint32 experimenter = 2; /* Experimenter ID which takes the same
+ form as in struct
+ ofp_experimenter_header. */
+ uint32 exp_type = 3; /* Experimenter defined. */
+ repeated uint32 experimenter_data = 4;
+};
+
+/* Body for ofp_multipart_request of type OFPMP_TABLE_FEATURES./
+ * Body of reply to OFPMP_TABLE_FEATURES request. */
+message ofp_table_features {
+ uint32 table_id = 1; /* Identifier of table. Lower numbered tables
+ are consulted first. */
+ string name = 2;
+ uint64 metadata_match = 3; /* Bits of metadata table can match. */
+ uint64 metadata_write = 4; /* Bits of metadata table can write. */
+ uint32 config = 5; /* Bitmap of OFPTC_* values */
+ uint32 max_entries = 6; /* Max number of entries supported. */
+
+ /* Table Feature Property list */
+ repeated ofp_table_feature_property properties = 7;
+};
+
+/* Body of reply to OFPMP_TABLE request. */
+message ofp_table_stats {
+ uint32 table_id = 1; /* Identifier of table. Lower numbered tables
+ are consulted first. */
+ uint32 active_count = 2; /* Number of active entries. */
+ uint64 lookup_count = 3; /* Number of packets looked up in table. */
+ uint64 matched_count = 4; /* Number of packets that hit table. */
+};
+
+/* Body for ofp_multipart_request of type OFPMP_PORT. */
+message ofp_port_stats_request {
+ uint32 port_no = 1; /* OFPMP_PORT message must request statistics
+ * either for a single port (specified in
+ * port_no) or for all ports (if port_no ==
+ * OFPP_ANY). */
+};
+
+/* Body of reply to OFPMP_PORT request. If a counter is unsupported, set
+ * the field to all ones. */
+message ofp_port_stats {
+ uint32 port_no = 1;
+ uint64 rx_packets = 2; /* Number of received packets. */
+ uint64 tx_packets = 3; /* Number of transmitted packets. */
+ uint64 rx_bytes = 4; /* Number of received bytes. */
+ uint64 tx_bytes = 5; /* Number of transmitted bytes. */
+ uint64 rx_dropped = 6; /* Number of packets dropped by RX. */
+ uint64 tx_dropped = 7; /* Number of packets dropped by TX. */
+ uint64 rx_errors = 8; /* Number of receive errors. This is a super-set
+ of more specific receive errors and should be
+ greater than or equal to the sum of all
+ rx_*_err values. */
+ uint64 tx_errors = 9; /* Number of transmit errors. This is a super-set
+ of more specific transmit errors and should be
+ greater than or equal to the sum of all
+ tx_*_err values (none currently defined.) */
+ uint64 rx_frame_err = 10; /* Number of frame alignment errors. */
+ uint64 rx_over_err = 11; /* Number of packets with RX overrun. */
+ uint64 rx_crc_err = 12; /* Number of CRC errors. */
+ uint64 collisions = 13; /* Number of collisions. */
+ uint32 duration_sec = 14; /* Time port has been alive in seconds. */
+ uint32 duration_nsec = 15; /* Time port has been alive in nanoseconds
+ beyond duration_sec. */
+};
+
+/* Body of OFPMP_GROUP request. */
+message ofp_group_stats_request {
+ uint32 group_id = 1; /* All groups if OFPG_ALL. */
+};
+
+/* Used in group stats replies. */
+message ofp_bucket_counter {
+ uint64 packet_count = 1; /* Number of packets processed by bucket. */
+ uint64 byte_count = 2; /* Number of bytes processed by bucket. */
+};
+
+/* Body of reply to OFPMP_GROUP request. */
+message ofp_group_stats {
+ uint32 group_id = 1; /* Group identifier. */
+ uint32 ref_count = 2; /* Number of flows or groups that directly
+ forward to this group. */
+ uint64 packet_count = 3; /* Number of packets processed by group. */
+ uint64 byte_count = 4; /* Number of bytes processed by group. */
+ uint32 duration_sec = 5; /* Time group has been alive in seconds. */
+ uint32 duration_nsec = 6; /* Time group has been alive in nanoseconds
+ beyond duration_sec. */
+ repeated ofp_bucket_counter bucket_stats = 7; /* One counter set per
+ bucket. */
+};
+
+/* Body of reply to OFPMP_GROUP_DESC request. */
+message ofp_group_desc {
+ ofp_group_type type = 1; /* One of OFPGT_*. */
+ uint32 group_id = 2; /* Group identifier. */
+ repeated ofp_bucket buckets = 3; /* List of buckets - 0 or more. */
+};
+
+message ofp_group_entry {
+ ofp_group_desc desc = 1 [(voltha.yang_inline_node).id = 'desc',
+ (voltha.yang_inline_node).type = 'openflow_13-ofp_group_desc'];
+ ofp_group_stats stats = 2;
+};
+
+/* Backward compatibility with 1.3.1 - avoid breaking the API. */
+//#define ofp_group_desc_stats ofp_group_desc
+
+/* Group configuration flags */
+enum ofp_group_capabilities {
+ OFPGFC_INVALID = 0;
+ OFPGFC_SELECT_WEIGHT = 1; /* Support weight for select groups */
+ OFPGFC_SELECT_LIVENESS = 2; /* Support liveness for select groups */
+ OFPGFC_CHAINING = 4; /* Support chaining groups */
+ OFPGFC_CHAINING_CHECKS = 8; /* Check chaining for loops and delete */
+};
+
+/* Body of reply to OFPMP_GROUP_FEATURES request. Group features. */
+message ofp_group_features {
+ uint32 types = 1; /* Bitmap of (1 << OFPGT_*) values supported. */
+ uint32 capabilities = 2; /* Bitmap of OFPGFC_* capability supported. */
+ repeated uint32 max_groups = 3; /* Maximum number of groups for each type.
+ */
+ repeated uint32 actions = 4; /* Bitmaps of (1 << OFPAT_*) values
+ supported. */
+};
+
+/* Body of OFPMP_METER and OFPMP_METER_CONFIG requests. */
+message ofp_meter_multipart_request {
+ uint32 meter_id = 1; /* Meter instance, or OFPM_ALL. */
+};
+
+/* Statistics for each meter band */
+message ofp_meter_band_stats {
+ uint64 packet_band_count = 1; /* Number of packets in band. */
+ uint64 byte_band_count = 2; /* Number of bytes in band. */
+};
+
+/* Body of reply to OFPMP_METER request. Meter statistics. */
+message ofp_meter_stats {
+ uint32 meter_id = 1; /* Meter instance. */
+ uint32 flow_count = 2; /* Number of flows bound to meter. */
+ uint64 packet_in_count = 3; /* Number of packets in input. */
+ uint64 byte_in_count = 4; /* Number of bytes in input. */
+ uint32 duration_sec = 5; /* Time meter has been alive in seconds. */
+ uint32 duration_nsec = 6;/* Time meter has been alive in nanoseconds
+ beyond duration_sec. */
+ repeated ofp_meter_band_stats band_stats = 7; /* The band_stats length is
+ inferred from the length field. */
+};
+
+/* Body of reply to OFPMP_METER_CONFIG request. Meter configuration. */
+message ofp_meter_config {
+ uint32 flags = 1; /* All OFPMF_* that apply. */
+ uint32 meter_id = 2; /* Meter instance. */
+ repeated ofp_meter_band_header bands = 3; /* The bands length is
+ inferred from the length field. */
+};
+
+/* Body of reply to OFPMP_METER_FEATURES request. Meter features. */
+message ofp_meter_features {
+ uint32 max_meter = 1; /* Maximum number of meters. */
+ uint32 band_types = 2; /* Bitmaps of (1 << OFPMBT_*) values supported.
+ */
+ uint32 capabilities = 3; /* Bitmaps of "ofp_meter_flags". */
+ uint32 max_bands = 4; /* Maximum bands per meters */
+ uint32 max_color = 5; /* Maximum color value */
+};
+
+/* Body for ofp_multipart_request/reply of type OFPMP_EXPERIMENTER. */
+message ofp_experimenter_multipart_header {
+ uint32 experimenter = 1; /* Experimenter ID which takes the same form
+ as in struct ofp_experimenter_header. */
+ uint32 exp_type = 2; /* Experimenter defined. */
+ bytes data = 3; /* Experimenter-defined arbitrary additional data. */
+};
+
+/* Experimenter extension. */
+message ofp_experimenter_header {
+ //ofp_header header; /* Type OFPT_EXPERIMENTER. */
+ uint32 experimenter = 1; /* Experimenter ID:
+ * - MSB 0: low-order bytes are IEEE OUI.
+ * - MSB != 0: defined by ONF. */
+ uint32 exp_type = 2; /* Experimenter defined. */
+ bytes data = 3; /* Experimenter-defined arbitrary additional data. */
+};
+
+/* All ones is used to indicate all queues in a port (for stats retrieval). */
+//#define OFPQ_ALL 0xffffffff
+
+/* Min rate > 1000 means not configured. */
+//#define OFPQ_MIN_RATE_UNCFG 0xffff
+
+/* Max rate > 1000 means not configured. */
+//#define OFPQ_MAX_RATE_UNCFG 0xffff
+
+enum ofp_queue_properties {
+ OFPQT_INVALID = 0;
+ OFPQT_MIN_RATE = 1; /* Minimum datarate guaranteed. */
+ OFPQT_MAX_RATE = 2; /* Maximum datarate. */
+ OFPQT_EXPERIMENTER = 0xffff; /* Experimenter defined property. */
+};
+
+/* Common description for a queue. */
+message ofp_queue_prop_header {
+ uint32 property = 1; /* One of OFPQT_. */
+ uint32 len = 2; /* Length of property, including this header. */
+};
+
+/* Min-Rate queue property description. */
+message ofp_queue_prop_min_rate {
+ ofp_queue_prop_header prop_header = 1;/* prop: OFPQT_MIN, len: 16. */
+ uint32 rate = 2; /* In 1/10 of a percent = 0;>1000 -> disabled. */
+};
+
+/* Max-Rate queue property description. */
+message ofp_queue_prop_max_rate {
+ ofp_queue_prop_header prop_header = 1;/* prop: OFPQT_MAX, len: 16. */
+ uint32 rate = 2; /* In 1/10 of a percent = 0;>1000 -> disabled. */
+};
+
+/* Experimenter queue property description. */
+message ofp_queue_prop_experimenter {
+ ofp_queue_prop_header prop_header = 1;/* prop: OFPQT_EXPERIMENTER */
+ uint32 experimenter = 2; /* Experimenter ID which takes the same
+ form as in struct
+ ofp_experimenter_header. */
+ bytes data = 3; /* Experimenter defined data. */
+};
+
+/* Full description for a queue. */
+message ofp_packet_queue {
+ uint32 queue_id = 1; /* id for the specific queue. */
+ uint32 port = 2; /* Port this queue is attached to. */
+ repeated ofp_queue_prop_header properties = 4; /* List of properties. */
+};
+
+/* Query for port queue configuration. */
+message ofp_queue_get_config_request {
+ //ofp_header header;
+ uint32 port = 1; /* Port to be queried. Should refer
+ to a valid physical port (i.e. <= OFPP_MAX),
+ or OFPP_ANY to request all configured
+ queues.*/
+};
+
+/* Queue configuration for a given port. */
+message ofp_queue_get_config_reply {
+ //ofp_header header;
+ uint32 port = 1;
+ repeated ofp_packet_queue queues = 2; /* List of configured queues. */
+};
+
+/* OFPAT_SET_QUEUE action struct: send packets to given queue on port. */
+message ofp_action_set_queue {
+ uint32 type = 1; /* OFPAT_SET_QUEUE. */
+ uint32 queue_id = 3; /* Queue id for the packets. */
+};
+
+message ofp_queue_stats_request {
+ uint32 port_no = 1; /* All ports if OFPP_ANY. */
+ uint32 queue_id = 2; /* All queues if OFPQ_ALL. */
+};
+
+message ofp_queue_stats {
+ uint32 port_no = 1;
+ uint32 queue_id = 2; /* Queue i.d */
+ uint64 tx_bytes = 3; /* Number of transmitted bytes. */
+ uint64 tx_packets = 4; /* Number of transmitted packets. */
+ uint64 tx_errors = 5; /* Number of packets dropped due to overrun. */
+ uint32 duration_sec = 6; /* Time queue has been alive in seconds. */
+ uint32 duration_nsec = 7; /* Time queue has been alive in nanoseconds
+ beyond duration_sec. */
+};
+
+/* Configures the "role" of the sending controller. The default role is:
+ *
+ * - Equal (OFPCR_ROLE_EQUAL), which allows the controller access to all
+ * OpenFlow features. All controllers have equal responsibility.
+ *
+ * The other possible roles are a related pair:
+ *
+ * - Master (OFPCR_ROLE_MASTER) is equivalent to Equal, except that there
+ * may be at most one Master controller at a time: when a controller
+ * configures itself as Master, any existing Master is demoted to the
+ * Slave role.
+ *
+ * - Slave (OFPCR_ROLE_SLAVE) allows the controller read-only access to
+ * OpenFlow features. In particular attempts to modify the flow table
+ * will be rejected with an OFPBRC_EPERM error.
+ *
+ * Slave controllers do not receive OFPT_PACKET_IN or OFPT_FLOW_REMOVED
+ * messages, but they do receive OFPT_PORT_STATUS messages.
+ */
+
+/* Controller roles. */
+enum ofp_controller_role {
+ OFPCR_ROLE_NOCHANGE = 0; /* Don't change current role. */
+ OFPCR_ROLE_EQUAL = 1; /* Default role, full access. */
+ OFPCR_ROLE_MASTER = 2; /* Full access, at most one master. */
+ OFPCR_ROLE_SLAVE = 3; /* Read-only access. */
+};
+
+/* Role request and reply message. */
+message ofp_role_request {
+ //ofp_header header; /* Type OFPT_ROLE_REQUEST/OFPT_ROLE_REPLY. */
+ ofp_controller_role role = 1; /* One of OFPCR_ROLE_*. */
+ uint64 generation_id = 2; /* Master Election Generation Id */
+};
+
+/* Asynchronous message configuration. */
+message ofp_async_config {
+ //ofp_header header; /* OFPT_GET_ASYNC_REPLY or OFPT_SET_ASYNC. */
+ repeated uint32 packet_in_mask = 1; /* Bitmasks of OFPR_* values. */
+ repeated uint32 port_status_mask = 2; /* Bitmasks of OFPPR_* values. */
+ repeated uint32 flow_removed_mask = 3;/* Bitmasks of OFPRR_* values. */
+};
+
+
+/* ADDITIONAL VOLTHA SPECIFIC MESSAGE TYPES, AIDING RPC CALLS */
+
+message FlowTableUpdate {
+ string id = 1; // Device.id or LogicalDevice.id
+ ofp_flow_mod flow_mod = 2;
+}
+
+message FlowGroupTableUpdate {
+ string id = 1; // Device.id or LogicalDevice.id
+ ofp_group_mod group_mod = 2;
+}
+
+message Flows {
+ repeated ofp_flow_stats items = 1;
+}
+
+message FlowGroups {
+ repeated ofp_group_entry items = 1;
+}
+
+message FlowChanges {
+ Flows to_add = 1;
+ Flows to_remove = 2;
+}
+
+message FlowGroupChanges {
+ FlowGroups to_add = 1;
+ FlowGroups to_remove = 2;
+}
+
+message PacketIn {
+ string id = 1; // LogicalDevice.id
+ ofp_packet_in packet_in = 2;
+}
+
+message PacketOut {
+ string id = 1; // LogicalDevice.id
+ ofp_packet_out packet_out = 2;
+}
+
+message ChangeEvent {
+ string id = 1; // LogicalDevice.id
+ oneof event {
+ ofp_port_status port_status = 2;
+ }
+}
diff --git a/protos/ponsim.proto b/protos/ponsim.proto
new file mode 100644
index 0000000..2ed8914
--- /dev/null
+++ b/protos/ponsim.proto
@@ -0,0 +1,57 @@
+syntax = "proto3";
+
+option go_package = "github.com/opencord/voltha-go/protos/voltha";
+
+package voltha;
+
+import "google/protobuf/empty.proto";
+import "openflow_13.proto";
+
+
+message PonSimDeviceInfo {
+ int32 nni_port = 1;
+ repeated int32 uni_ports = 2;
+}
+
+message FlowTable {
+ int32 port = 1; // Used to address right device
+ repeated openflow_13.ofp_flow_stats flows = 2;
+}
+
+message PonSimFrame {
+ string id = 1;
+ bytes payload = 2;
+}
+
+message PonSimPacketCounter {
+ string name = 1;
+ int64 value = 2;
+}
+
+message PonSimPortMetrics {
+ string port_name = 1;
+ repeated PonSimPacketCounter packets = 2;
+}
+
+message PonSimMetrics {
+ string device = 1;
+ repeated PonSimPortMetrics metrics = 2;
+}
+
+service PonSim {
+ rpc SendFrame(PonSimFrame)
+ returns (google.protobuf.Empty) {}
+
+ rpc ReceiveFrames(google.protobuf.Empty)
+ returns (stream PonSimFrame) {}
+
+ rpc GetDeviceInfo(google.protobuf.Empty)
+ returns(PonSimDeviceInfo) {}
+
+ rpc UpdateFlowTable(FlowTable)
+ returns(google.protobuf.Empty) {}
+
+ rpc GetStats(google.protobuf.Empty)
+ returns(PonSimMetrics) {}
+
+}
diff --git a/protos/schema.proto b/protos/schema.proto
new file mode 100644
index 0000000..bb1f895
--- /dev/null
+++ b/protos/schema.proto
@@ -0,0 +1,40 @@
+syntax = "proto3";
+
+option go_package = "github.com/opencord/voltha-go/protos/schema";
+
+package schema;
+
+import "google/api/annotations.proto";
+import "google/protobuf/empty.proto";
+
+// Contains the name and content of a *.proto file
+message ProtoFile {
+ string file_name = 1; // name of proto file
+ string proto = 2; // content of proto file
+ bytes descriptor = 3; // compiled descriptor for proto (zlib compressed)
+}
+
+// Proto files and compiled descriptors for this interface
+message Schemas {
+
+ // Proto files
+ repeated ProtoFile protos = 1;
+
+ // Proto file name from which swagger.json shall be generated
+ string swagger_from = 2;
+
+ // Proto file name from which yang schemas shall be generated
+ string yang_from = 3;
+}
+
+// Schema services
+service SchemaService {
+
+ // Return active grpc schemas
+ rpc GetSchema(google.protobuf.Empty) returns (Schemas) {
+ option (google.api.http) = {
+ get: "/schema"
+ };
+ }
+
+}
diff --git a/protos/scripts/build_protos.sh b/protos/scripts/build_protos.sh
new file mode 100644
index 0000000..faac725
--- /dev/null
+++ b/protos/scripts/build_protos.sh
@@ -0,0 +1,46 @@
+#!/bin/sh
+
+export SRC_DIR="$1"
+
+echo $SRC_DIR
+
+export MAPS=Mgoogle/protobuf/descriptor.proto=github.com/golang/protobuf/protoc-gen-go/descriptor
+export INCS="\
+ -I $SRC_DIR \
+ -I $GOPATH/src/github.com/grpc-ecosystem/grpc-gateway/third_party/googleapis"
+
+export VOLTHA_PB="\
+ $SRC_DIR/adapter.proto \
+ $SRC_DIR/device.proto \
+ $SRC_DIR/events.proto \
+ $SRC_DIR/health.proto \
+ $SRC_DIR/logical_device.proto \
+ $SRC_DIR/ponsim.proto \
+ $SRC_DIR/voltha.proto"
+
+export COMMON_PB="\
+ $SRC_DIR/common.proto \
+ $SRC_DIR/meta.proto \
+ $SRC_DIR/yang_options.proto"
+
+export CORE_ADAPTER_PB="$SRC_DIR/core_adapter.proto"
+export SCHEMA_PB="$SRC_DIR/schema.proto"
+export IETF_PB="$SRC_DIR/ietf_interfaces.proto"
+export OF_PB="$SRC_DIR/openflow_13.proto"
+export OMCI_PB="$SRC_DIR/omci*.proto"
+
+export PB_VARS="\
+ VOLTHA_PB \
+ COMMON_PB \
+ CORE_ADAPTER_PB \
+ SCHEMA_PB \
+ IETF_PB \
+ OF_PB \
+ OMCI_PB"
+
+for pb_var in $PB_VARS
+do
+ pbs="$(eval echo \$$pb_var)"
+ echo "Compiling $pbs"
+ protoc --go_out=$MAPS,plugins=grpc:$GOPATH/src $INCS $pbs
+done
diff --git a/protos/voltha.proto b/protos/voltha.proto
new file mode 100644
index 0000000..a107bd2
--- /dev/null
+++ b/protos/voltha.proto
@@ -0,0 +1,505 @@
+/*
+ * Top-level Voltha API definition
+ *
+ * For details, see individual definition files.
+ */
+
+syntax = "proto3";
+
+option go_package = "github.com/opencord/voltha-go/protos/voltha";
+
+package voltha;
+
+import "google/protobuf/empty.proto";
+import "google/api/annotations.proto";
+
+import "yang_options.proto";
+
+import public "meta.proto";
+import public "common.proto";
+import public "health.proto";
+import public "logical_device.proto";
+import public "device.proto";
+import public "adapter.proto";
+import public "openflow_13.proto";
+import "omci_mib_db.proto";
+
+
+option java_package = "org.opencord.voltha";
+option java_outer_classname = "VolthaProtos";
+option csharp_namespace = "Opencord.Voltha.Voltha";
+
+message DeviceGroup {
+
+ string id = 1 [(access) = READ_ONLY];
+
+ repeated LogicalDevice logical_devices = 2 [(child_node) = {key: "id"}];
+
+ repeated Device devices = 3 [(child_node) = {key: "id"}];
+}
+
+message DeviceGroups {
+ repeated DeviceGroup items = 1;
+}
+
+
+message AlarmFilterRuleKey {
+ option (yang_child_rule) = MOVE_TO_PARENT_LEVEL;
+
+ enum AlarmFilterRuleKey {
+ id = 0;
+ type = 1;
+ severity = 2;
+ resource_id = 3;
+ category = 4;
+ device_id = 5;
+ }
+}
+
+message AlarmFilterRule {
+ AlarmFilterRuleKey.AlarmFilterRuleKey key = 1;
+ string value = 2;
+}
+message AlarmFilter {
+ string id = 1 [(access) = READ_ONLY];
+
+ repeated AlarmFilterRule rules = 2;
+}
+
+message AlarmFilters {
+ repeated AlarmFilter filters = 1;
+}
+
+// Top-level (root) node for a Voltha Instance
+message VolthaInstance {
+ option (yang_message_rule) = CREATE_BOTH_GROUPING_AND_CONTAINER;
+
+ string instance_id = 1 [(access) = READ_ONLY];
+
+ string version = 2 [(access) = READ_ONLY];
+
+ LogLevel.LogLevel log_level = 3;
+
+ HealthStatus health = 10 [(child_node) = {}];
+
+ repeated Adapter adapters = 11 [(child_node) = {key: "id" }];
+
+ repeated LogicalDevice logical_devices = 12 [(child_node) = {key: "id"}];
+
+ repeated Device devices = 13 [(child_node) = {key: "id"}];
+
+ repeated DeviceType device_types = 14 [(child_node) = {key: "id"}];
+
+ repeated DeviceGroup device_groups = 15 [(child_node) = {key: "id"}];
+
+ repeated AlarmFilter alarm_filters = 16 [(child_node) = {key: "id"}];
+
+ repeated
+ omci.MibDeviceData omci_mibs = 28
+ [(child_node) = {key: "device_id"}];
+}
+
+message VolthaInstances {
+ option (yang_message_rule) = CREATE_BOTH_GROUPING_AND_CONTAINER;
+ repeated string items = 1;
+}
+
+// Voltha representing the entire Voltha cluster
+message Voltha {
+ option (yang_message_rule) = CREATE_BOTH_GROUPING_AND_CONTAINER;
+
+ string version = 1 [(access) = READ_ONLY];
+
+ LogLevel.LogLevel log_level = 2;
+
+ repeated VolthaInstance instances = 3 [(child_node) = {key: "instance_id"}];
+
+ repeated Adapter adapters = 11 [(child_node) = {key: "id"}];
+
+ repeated LogicalDevice logical_devices = 12 [(child_node) = {key: "id"}];
+
+ repeated Device devices = 13 [(child_node) = {key: "id"}];
+
+ repeated DeviceGroup device_groups = 15 [(child_node) = {key: "id"}];
+
+ repeated
+ omci.MibDeviceData omci_mib_database = 28
+ [(child_node) = {key: "device_id"}];
+}
+
+// Device Self Test Response
+message SelfTestResponse {
+ option (yang_child_rule) = MOVE_TO_PARENT_LEVEL;
+
+ enum SelfTestResult {
+ SUCCESS = 0;
+ FAILURE = 1;
+ NOT_SUPPORTED = 2;
+ UNKNOWN_ERROR = 3;
+ }
+ SelfTestResult result = 1;
+}
+
+message OfAgentSubscriber {
+ // ID of ofagent instance
+ string ofagent_id = 1;
+
+ // ID of voltha instance to which the ofagent is subscribed
+ string voltha_id = 2;
+}
+
+/*
+ * Voltha APIs
+ *
+ */
+service VolthaService {
+
+ // Get high level information on the Voltha cluster
+ rpc GetVoltha(google.protobuf.Empty) returns(Voltha) {
+ option (google.api.http) = {
+ get: "/api/v1"
+ };
+ }
+
+ // List all Voltha cluster instances
+ rpc ListVolthaInstances(google.protobuf.Empty) returns(VolthaInstances) {
+ option (google.api.http) = {
+ get: "/api/v1/instances"
+ };
+ option (voltha.yang_xml_tag).xml_tag = 'items';
+ option (voltha.yang_xml_tag).list_items_name = 'items';
+ }
+
+ // Get details on a Voltha cluster instance
+ rpc GetVolthaInstance(ID) returns(VolthaInstance) {
+ option (google.api.http) = {
+ get: "/api/v1/instances/{id}"
+ };
+ }
+
+ // List all active adapters (plugins) in the Voltha cluster
+ rpc ListAdapters(google.protobuf.Empty) returns(Adapters) {
+ option (google.api.http) = {
+ get: "/api/v1/adapters"
+ };
+ option (voltha.yang_xml_tag).xml_tag = 'adapters';
+ }
+
+
+ // List all logical devices managed by the Voltha cluster
+ rpc ListLogicalDevices(google.protobuf.Empty) returns(LogicalDevices) {
+ option (google.api.http) = {
+ get: "/api/v1/logical_devices"
+ };
+ option (voltha.yang_xml_tag).xml_tag = 'logical_devices';
+ }
+
+ // Get additional information on a given logical device
+ rpc GetLogicalDevice(ID) returns(LogicalDevice) {
+ option (google.api.http) = {
+ get: "/api/v1/logical_devices/{id}"
+ };
+ }
+
+ // List ports of a logical device
+ rpc ListLogicalDevicePorts(ID) returns(LogicalPorts) {
+ option (google.api.http) = {
+ get: "/api/v1/logical_devices/{id}/ports"
+ };
+ option (voltha.yang_xml_tag).xml_tag = 'ports';
+ }
+
+ // Gets a logical device port
+ rpc GetLogicalDevicePort(LogicalPortId) returns(LogicalPort) {
+ option (google.api.http) = {
+ get: "/api/v1/logical_devices/{id}/ports/{port_id}"
+ };
+ option (voltha.yang_xml_tag).xml_tag = 'port';
+ }
+
+ // Enables a logical device port
+ rpc EnableLogicalDevicePort(LogicalPortId) returns(google.protobuf.Empty) {
+ option (google.api.http) = {
+ post: "/api/v1/logical_devices/{id}/ports/{port_id}/enable"
+ };
+ }
+
+ // Disables a logical device port
+ rpc DisableLogicalDevicePort(LogicalPortId) returns(google.protobuf.Empty) {
+ option (google.api.http) = {
+ post: "/api/v1/logical_devices/{id}/ports/{port_id}/disable"
+ };
+ }
+
+ // List all flows of a logical device
+ rpc ListLogicalDeviceFlows(ID) returns(openflow_13.Flows) {
+ option (google.api.http) = {
+ get: "/api/v1/logical_devices/{id}/flows"
+ };
+ option (voltha.yang_xml_tag).xml_tag = 'flows';
+ option (voltha.yang_xml_tag).list_items_name = 'items';
+ }
+
+ // Update flow table for logical device
+ rpc UpdateLogicalDeviceFlowTable(openflow_13.FlowTableUpdate)
+ returns(google.protobuf.Empty) {
+ option (google.api.http) = {
+ post: "/api/v1/logical_devices/{id}/flows"
+ body: "*"
+ };
+ }
+
+ // List all flow groups of a logical device
+ rpc ListLogicalDeviceFlowGroups(ID) returns(openflow_13.FlowGroups) {
+ option (google.api.http) = {
+ get: "/api/v1/logical_devices/{id}/flow_groups"
+ };
+ option (voltha.yang_xml_tag).xml_tag = 'flow_groups';
+ option (voltha.yang_xml_tag).list_items_name = 'items';
+ }
+
+ // Update group table for device
+ rpc UpdateLogicalDeviceFlowGroupTable(openflow_13.FlowGroupTableUpdate)
+ returns(google.protobuf.Empty) {
+ option (google.api.http) = {
+ post: "/api/v1/logical_devices/{id}/flow_groups"
+ body: "*"
+ };
+ }
+
+ // List all physical devices controlled by the Voltha cluster
+ rpc ListDevices(google.protobuf.Empty) returns(Devices) {
+ option (google.api.http) = {
+ get: "/api/v1/devices"
+ };
+ option (voltha.yang_xml_tag).xml_tag = 'devices';
+ }
+
+ // Get more information on a given physical device
+ rpc GetDevice(ID) returns(Device) {
+ option (google.api.http) = {
+ get: "/api/v1/devices/{id}"
+ };
+ }
+
+ // Pre-provision a new physical device
+ rpc CreateDevice(Device) returns(Device) {
+ option (google.api.http) = {
+ post: "/api/v1/devices"
+ body: "*"
+ };
+ }
+
+ // Enable a device. If the device was in pre-provisioned state then it
+ // will transition to ENABLED state. If it was is DISABLED state then it
+ // will transition to ENABLED state as well.
+ rpc EnableDevice(ID) returns(google.protobuf.Empty) {
+ option (google.api.http) = {
+ post: "/api/v1/devices/{id}/enable"
+ };
+ }
+
+ // Disable a device
+ rpc DisableDevice(ID) returns(google.protobuf.Empty) {
+ option (google.api.http) = {
+ post: "/api/v1/devices/{id}/disable"
+ };
+ }
+
+ // Reboot a device
+ rpc RebootDevice(ID) returns(google.protobuf.Empty) {
+ option (google.api.http) = {
+ post: "/api/v1/devices/{id}/reboot"
+ };
+ }
+
+ // Delete a device
+ rpc DeleteDevice(ID) returns(google.protobuf.Empty) {
+ option (google.api.http) = {
+ delete: "/api/v1/devices/{id}/delete"
+ };
+ }
+
+ // Request an image download to the standby partition
+ // of a device.
+ // Note that the call is expected to be non-blocking.
+ rpc DownloadImage(ImageDownload) returns(OperationResp) {
+ option (google.api.http) = {
+ post: "/api/v1/devices/{id}/image_downloads/{name}"
+ body: "*"
+ };
+ }
+
+ // Get image download status on a device
+ // The request retrieves progress on device and updates db record
+ rpc GetImageDownloadStatus(ImageDownload) returns(ImageDownload) {
+ option (google.api.http) = {
+ get: "/api/v1/devices/{id}/image_downloads/{name}/status"
+ };
+ }
+
+ // Get image download db record
+ rpc GetImageDownload(ImageDownload) returns(ImageDownload) {
+ option (google.api.http) = {
+ get: "/api/v1/devices/{id}/image_downloads/{name}"
+ };
+ }
+
+ // List image download db records for a given device
+ rpc ListImageDownloads(ID) returns(ImageDownloads) {
+ option (google.api.http) = {
+ get: "/api/v1/devices/{id}/image_downloads"
+ };
+ }
+
+ // Cancel an existing image download process on a device
+ rpc CancelImageDownload(ImageDownload) returns(OperationResp) {
+ option (google.api.http) = {
+ delete: "/api/v1/devices/{id}/image_downloads/{name}"
+ };
+ }
+
+ // Activate the specified image at a standby partition
+ // to active partition.
+ // Depending on the device implementation, this call
+ // may or may not cause device reboot.
+ // If no reboot, then a reboot is required to make the
+ // activated image running on device
+ // Note that the call is expected to be non-blocking.
+ rpc ActivateImageUpdate(ImageDownload) returns(OperationResp) {
+ option (google.api.http) = {
+ post: "/api/v1/devices/{id}/image_downloads/{name}/image_update"
+ body: "*"
+ };
+ }
+
+ // Revert the specified image at standby partition
+ // to active partition, and revert to previous image
+ // Depending on the device implementation, this call
+ // may or may not cause device reboot.
+ // If no reboot, then a reboot is required to make the
+ // previous image running on device
+ // Note that the call is expected to be non-blocking.
+ rpc RevertImageUpdate(ImageDownload) returns(OperationResp) {
+ option (google.api.http) = {
+ post: "/api/v1/devices/{id}/image_downloads/{name}/image_revert"
+ body: "*"
+ };
+ }
+
+ // List ports of a device
+ rpc ListDevicePorts(ID) returns(Ports) {
+ option (google.api.http) = {
+ get: "/api/v1/devices/{id}/ports"
+ };
+ option (voltha.yang_xml_tag).xml_tag = 'ports';
+ }
+
+ // List pm config of a device
+ rpc ListDevicePmConfigs(ID) returns(PmConfigs) {
+ option (google.api.http) = {
+ get: "/api/v1/devices/{id}/pm_configs"
+ };
+ }
+
+ // Update the pm config of a device
+ rpc UpdateDevicePmConfigs(voltha.PmConfigs) returns(google.protobuf.Empty) {
+ option (google.api.http) = {
+ post: "/api/v1/devices/{id}/pm_configs"
+ body: "*"
+ };
+ }
+
+ // List all flows of a device
+ rpc ListDeviceFlows(ID) returns(openflow_13.Flows) {
+ option (google.api.http) = {
+ get: "/api/v1/devices/{id}/flows"
+ };
+ option (voltha.yang_xml_tag).xml_tag = 'flows';
+ option (voltha.yang_xml_tag).list_items_name = 'items';
+ }
+
+ // List all flow groups of a device
+ rpc ListDeviceFlowGroups(ID) returns(openflow_13.FlowGroups) {
+ option (google.api.http) = {
+ get: "/api/v1/devices/{id}/flow_groups"
+ };
+ option (voltha.yang_xml_tag).xml_tag = 'flow_groups';
+ option (voltha.yang_xml_tag).list_items_name = 'items';
+ }
+
+ // List device types known to Voltha
+ rpc ListDeviceTypes(google.protobuf.Empty) returns(DeviceTypes) {
+ option (google.api.http) = {
+ get: "/api/v1/device_types"
+ };
+ option (voltha.yang_xml_tag).xml_tag = 'device_types';
+ }
+
+ // Get additional information on a device type
+ rpc GetDeviceType(ID) returns(DeviceType) {
+ option (google.api.http) = {
+ get: "/api/v1/device_types/{id}"
+ };
+ }
+
+ // List all device sharding groups
+ rpc ListDeviceGroups(google.protobuf.Empty) returns(DeviceGroups) {
+ option (google.api.http) = {
+ get: "/api/v1/device_groups"
+ };
+ option (voltha.yang_xml_tag).xml_tag = 'device_groups';
+ }
+
+ // Get additional information on a device group
+ rpc GetDeviceGroup(ID) returns(DeviceGroup) {
+ option (google.api.http) = {
+ get: "/api/v1/device_groups/{id}"
+ };
+ }
+
+ rpc CreateAlarmFilter(AlarmFilter) returns(AlarmFilter) {
+ option (google.api.http) = {
+ post: "/api/v1/alarm_filters"
+ body: "*"
+ };
+ }
+
+ rpc GetAlarmFilter(ID) returns(AlarmFilter) {
+ option (google.api.http) = {
+ get: "/api/v1/alarm_filters/{id}"
+ };
+ }
+
+ rpc UpdateAlarmFilter(AlarmFilter) returns(AlarmFilter) {
+ option (google.api.http) = {
+ put: "/api/v1/alarm_filters/{id}"
+ body: "*"
+ };
+ }
+
+ rpc DeleteAlarmFilter(ID) returns(google.protobuf.Empty) {
+ option (google.api.http) = {
+ delete: "/api/v1/alarm_filters/{id}"
+ };
+ }
+
+ rpc ListAlarmFilters(google.protobuf.Empty) returns(AlarmFilters) {
+ option (google.api.http) = {
+ get: "/api/v1/alarm_filters"
+ };
+ }
+
+ rpc GetImages(ID) returns(Images) {
+ option (google.api.http) = {
+ get: "/api/v1/devices/{id}/images"
+ };
+ }
+
+ rpc SelfTest(ID) returns(SelfTestResponse) {
+ option (google.api.http) = {
+ post: "/api/v1/devices/{id}/self_test"
+ };
+ }
+}
+
diff --git a/protos/yang_options.proto b/protos/yang_options.proto
new file mode 100644
index 0000000..25f70a8
--- /dev/null
+++ b/protos/yang_options.proto
@@ -0,0 +1,74 @@
+// Copyright (c) 2015, Google Inc.
+//
+// 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.
+
+// This file contains annotation definitions that can be used to describe
+// a configuration tree.
+
+syntax = "proto3";
+
+option go_package = "github.com/opencord/voltha-go/protos/common";
+
+package voltha;
+
+import "google/protobuf/descriptor.proto";
+
+enum MessageParserOption {
+ // Move any enclosing child enum/message definition to the same level
+ // as the parent (this message) in the yang generated file
+ MOVE_TO_PARENT_LEVEL= 0;
+
+ // Create both a grouping and a container for this message. The container
+ // name will be the message name. The grouping name will be the message
+ // name prefixed with "grouping_"
+ CREATE_BOTH_GROUPING_AND_CONTAINER = 1;
+}
+
+message InlineNode {
+ string id = 1;
+ string type = 2;
+}
+
+message RpcReturnDef {
+ // The gRPC methods return message types. NETCONF expects an actual
+ // attribute as defined in the YANG schema. The xnl_tag will be used
+ // as the top most tag when translating a gRPC response into an xml
+ // response
+ string xml_tag = 1;
+
+ // When the gRPC response is a list of items, we need to differentiate
+ // between a YANG schema attribute whose name is "items" and when "items"
+ // is used only to indicate a list of items is being returned. The default
+ // behavior assumes a list is returned when "items" is present in
+ // the response. This option will therefore be used when the attribute
+ // name in the YANG schema is 'items'
+ string list_items_name = 2;
+}
+
+extend google.protobuf.MessageOptions {
+ // This annotation is used to indicate how a message is parsed when
+ // converting from proto to yang format.
+ MessageParserOption yang_child_rule = 7761774;
+
+ MessageParserOption yang_message_rule = 7761775;
+}
+
+extend google.protobuf.FieldOptions {
+ // If present, the field (a message reference) should be replaced by the
+ // message itself. For now, this applies only to non-repeated fields.
+ InlineNode yang_inline_node = 7761776;
+}
+
+extend google.protobuf.MethodOptions {
+ RpcReturnDef yang_xml_tag = 7761777;
+}
diff --git a/rw_core/config/config.go b/rw_core/config/config.go
index a223318..8b2c165 100644
--- a/rw_core/config/config.go
+++ b/rw_core/config/config.go
@@ -24,7 +24,7 @@
default_KVStoreTimeout = 5 //in seconds
default_KVStoreHost = "10.100.198.240"
default_KVStorePort = 8500 // Etcd = 2379
- default_LogLevel = 1
+ default_LogLevel = 0
default_Banner = false
default_CoreTopic = "rwcore"
default_RWCoreEndpoint = "rwcore"
diff --git a/rw_core/core/requestHandlerProxy.go b/rw_core/core/requestHandlerProxy.go
new file mode 100644
index 0000000..4bfafa5
--- /dev/null
+++ b/rw_core/core/requestHandlerProxy.go
@@ -0,0 +1,154 @@
+package core
+
+import (
+ "github.com/opencord/voltha-go/common/log"
+ ca "github.com/opencord/voltha-go/protos/core_adapter"
+ "github.com/opencord/voltha-go/protos/voltha"
+ "github.com/golang/protobuf/ptypes"
+ "errors"
+)
+
+type RequestHandlerProxy struct {
+ TestMode bool
+}
+
+func (rhp *RequestHandlerProxy) GetDevice(args []*ca.Argument)(error, *voltha.Device) {
+ if len(args) != 1 {
+ log.Warn("invalid-number-of-args", log.Fields{"args": args})
+ err := errors.New("invalid-number-of-args")
+ return err, nil
+ }
+ pID := &ca.StrType{}
+ if err := ptypes.UnmarshalAny(args[0].Value, pID); err != nil {
+ log.Warnw("cannot-unmarshal-argument", log.Fields{"error": err})
+ return err, nil
+ }
+ log.Debugw("GetDevice", log.Fields{"deviceId": pID.Val})
+ // TODO process the request
+
+ if rhp.TestMode { // Execute only for test cases
+ return nil, &voltha.Device{Id: pID.Val}
+ }
+ return nil, nil
+}
+
+
+func (rhp *RequestHandlerProxy) GetChildDevice(args []*ca.Argument)(error, *voltha.Device) {
+ if len(args) < 1 {
+ log.Warn("invalid-number-of-args", log.Fields{"args": args})
+ err := errors.New("invalid-number-of-args")
+ return err, nil
+ }
+ pID := &ca.StrType{}
+ if err := ptypes.UnmarshalAny(args[0].Value, pID); err != nil {
+ log.Warnw("cannot-unmarshal-argument", log.Fields{"error": err})
+ return err, nil
+ }
+ // TODO decompose the other parameteres for matching criteria and process
+ log.Debugw("GetChildDevice", log.Fields{"deviceId": pID.Val})
+
+ if rhp.TestMode { // Execute only for test cases
+ return nil, &voltha.Device{Id: pID.Val}
+ }
+ return nil, nil
+}
+
+func (rhp *RequestHandlerProxy) GetPorts(args []*ca.Argument)(error, *voltha.Ports) {
+ if len(args) != 2 {
+ log.Warn("invalid-number-of-args", log.Fields{"args": args})
+ err := errors.New("invalid-number-of-args")
+ return err, nil
+ }
+ pID := &ca.StrType{}
+ if err := ptypes.UnmarshalAny(args[0].Value, pID); err != nil {
+ log.Warnw("cannot-unmarshal-argument", log.Fields{"error": err})
+ return err, nil
+ }
+ // Porttype is an enum sent as an integer proto
+ pt := &ca.IntType{}
+ if err := ptypes.UnmarshalAny(args[1].Value, pt); err != nil {
+ log.Warnw("cannot-unmarshal-argument", log.Fields{"error": err})
+ return err, nil
+ }
+
+ // TODO decompose the other parameteres for matching criteria
+ log.Debugw("GetPorts", log.Fields{"deviceID": pID.Val, "portype": pt.Val})
+
+ if rhp.TestMode { // Execute only for test cases
+ aPort := &voltha.Port{Label:"test_port"}
+ allPorts := &voltha.Ports{}
+ allPorts.Items = append(allPorts.Items, aPort)
+ return nil, allPorts
+ }
+ return nil, nil
+
+}
+
+func (rhp *RequestHandlerProxy) GetChildDevices(args []*ca.Argument)(error, *voltha.Device) {
+ if len(args) != 1 {
+ log.Warn("invalid-number-of-args", log.Fields{"args": args})
+ err := errors.New("invalid-number-of-args")
+ return err, nil
+ }
+ pID := &ca.StrType{}
+ if err := ptypes.UnmarshalAny(args[0].Value, pID); err != nil {
+ log.Warnw("cannot-unmarshal-argument", log.Fields{"error": err})
+ return err, nil
+ }
+ // TODO decompose the other parameteres for matching criteria and process
+ log.Debugw("GetChildDevice", log.Fields{"deviceId": pID.Val})
+
+ if rhp.TestMode { // Execute only for test cases
+ return nil, &voltha.Device{Id: pID.Val}
+ }
+ return nil, nil
+}
+
+// ChildDeviceDetected is invoked when a child device is detected. The following
+// parameters are expected:
+// {parent_device_id, parent_port_no, child_device_type, proxy_address, admin_state, **kw)
+func (rhp *RequestHandlerProxy) ChildDeviceDetected(args []*ca.Argument) (error) {
+ if len(args) < 5 {
+ log.Warn("invalid-number-of-args", log.Fields{"args": args})
+ err := errors.New("invalid-number-of-args")
+ return err
+ }
+
+ pID := &ca.StrType{}
+ if err := ptypes.UnmarshalAny(args[0].Value, pID); err != nil {
+ log.Warnw("cannot-unmarshal-argument", log.Fields{"error": err})
+ return err
+ }
+ portNo := &ca.IntType{}
+ if err := ptypes.UnmarshalAny(args[1].Value, portNo); err != nil {
+ log.Warnw("cannot-unmarshal-argument", log.Fields{"error": err})
+ return err
+ }
+ dt := &ca.StrType{}
+ if err := ptypes.UnmarshalAny(args[2].Value, dt); err != nil {
+ log.Warnw("cannot-unmarshal-argument", log.Fields{"error": err})
+ return err
+ }
+ pAddr := &voltha.Device_ProxyAddress{}
+ if err := ptypes.UnmarshalAny(args[3].Value, pAddr); err != nil {
+ log.Warnw("cannot-unmarshal-argument", log.Fields{"error": err})
+ return err
+ }
+ adminState := &ca.IntType{}
+ if err := ptypes.UnmarshalAny(args[4].Value, adminState); err != nil {
+ log.Warnw("cannot-unmarshal-argument", log.Fields{"error": err})
+ return err
+ }
+
+ // Need to decode the other params - in this case the key will represent the proto type
+ // TODO decompose the other parameteres for matching criteria and process
+ log.Debugw("ChildDeviceDetected", log.Fields{"deviceId": pID.Val, "portNo":portNo.Val,
+ "deviceType": dt.Val, "proxyAddress": pAddr, "adminState": adminState})
+
+ if rhp.TestMode { // Execute only for test cases
+ return nil
+ }
+ return nil
+}
+
+
diff --git a/rw_core/main.go b/rw_core/main.go
index faa5851..a0928c5 100644
--- a/rw_core/main.go
+++ b/rw_core/main.go
@@ -6,6 +6,8 @@
"fmt"
"github.com/opencord/voltha-go/common/log"
"github.com/opencord/voltha-go/db/kvstore"
+ "github.com/opencord/voltha-go/kafka"
+ ca "github.com/opencord/voltha-go/protos/core_adapter"
"github.com/opencord/voltha-go/rw_core/config"
"os"
"os/signal"
@@ -19,6 +21,9 @@
config *config.RWCoreFlags
halted bool
exitChannel chan int
+ kmp *kafka.KafkaMessagingProxy
+ //For test
+ receiverChannels []<-chan *ca.InterContainerMessage
}
func newKVClient(storeType string, address string, timeout int) (kvstore.Client, error) {
@@ -38,6 +43,7 @@
rwCore.config = cf
rwCore.halted = false
rwCore.exitChannel = make(chan int, 1)
+ rwCore.receiverChannels = make([]<-chan *ca.InterContainerMessage, 0)
return &rwCore
}
@@ -63,6 +69,7 @@
}
}
+
func (core *rwCore) start(ctx context.Context) {
log.Info("Starting RW Core components")
// Setup GRPC Server
@@ -70,13 +77,29 @@
// Setup KV Client
// Setup Kafka messaging services
-
+ var err error
+ if core.kmp, err = kafka.NewKafkaMessagingProxy(
+ kafka.KafkaHost("10.100.198.220"),
+ kafka.KafkaPort(9092),
+ kafka.DefaultTopic(&kafka.Topic{Name: "Adapter"})); err != nil {
+ log.Errorw("fail-to-create-kafka-proxy", log.Fields{"error": err})
+ return
+ }
+ // Start the kafka messaging service - synchronous call to ensure
+ if err = core.kmp.Start(); err != nil {
+ log.Fatalw("error-starting-messaging-proxy", log.Fields{"error": err})
+ }
}
func (core *rwCore) stop() {
// Stop leadership tracking
core.halted = true
+ // Stop the Kafka messaging service
+ if core.kmp != nil {
+ core.kmp.Stop()
+ }
+
// send exit signal
core.exitChannel <- 0
diff --git a/tests/kafka/kafka_inter_container_messaging_test.go b/tests/kafka/kafka_inter_container_messaging_test.go
new file mode 100644
index 0000000..9a73b18
--- /dev/null
+++ b/tests/kafka/kafka_inter_container_messaging_test.go
@@ -0,0 +1,437 @@
+package kafka
+
+import (
+ "context"
+ "github.com/golang/protobuf/ptypes"
+ "github.com/opencord/voltha-go/common/log"
+ kk "github.com/opencord/voltha-go/kafka"
+ ca "github.com/opencord/voltha-go/protos/core_adapter"
+ "github.com/opencord/voltha-go/protos/voltha"
+ rhp "github.com/opencord/voltha-go/rw_core/core"
+ "github.com/stretchr/testify/assert"
+ "testing"
+ //"time"
+ "github.com/google/uuid"
+ "time"
+)
+
+var coreKafkaProxy *kk.KafkaMessagingProxy
+var adapterKafkaProxy *kk.KafkaMessagingProxy
+
+func init() {
+ if _, err := log.SetLogger(log.JSON, 1, log.Fields{"instanceId": "testing"}); err != nil {
+ log.With(log.Fields{"error": err}).Fatal("Cannot setup logging")
+ }
+ coreKafkaProxy, _ = kk.NewKafkaMessagingProxy(
+ kk.KafkaHost("10.100.198.220"),
+ kk.KafkaPort(9092),
+ kk.DefaultTopic(&kk.Topic{Name: "Core"}))
+
+ adapterKafkaProxy, _ = kk.NewKafkaMessagingProxy(
+ kk.KafkaHost("10.100.198.220"),
+ kk.KafkaPort(9092),
+ kk.DefaultTopic(&kk.Topic{Name: "Adapter"}))
+
+ coreKafkaProxy.Start()
+ adapterKafkaProxy.Start()
+ subscribeTarget(coreKafkaProxy)
+}
+
+func subscribeTarget(kmp *kk.KafkaMessagingProxy) {
+ topic := kk.Topic{Name: "Core"}
+ requestProxy := &rhp.RequestHandlerProxy{TestMode: true}
+ kmp.SubscribeWithTarget(topic, requestProxy)
+}
+
+func waitForRPCMessage(topic kk.Topic, ch <-chan *ca.InterContainerMessage, doneCh chan string) {
+ for msg := range ch {
+ log.Debugw("Got-RPC-message", log.Fields{"msg": msg})
+ // Unpack message
+ requestBody := &ca.InterContainerRequestBody{}
+ if err := ptypes.UnmarshalAny(msg.Body, requestBody); err != nil {
+ doneCh <- "Error"
+ } else {
+ doneCh <- requestBody.Rpc
+ }
+ break
+ }
+}
+
+func TestSubscribeUnsubscribe(t *testing.T) {
+ // First subscribe to the specific topic
+ topic := kk.Topic{Name: "Core"}
+ ch, err := coreKafkaProxy.Subscribe(topic)
+ assert.NotNil(t, ch)
+ assert.Nil(t, err)
+ // Create a channel to receive a response
+ waitCh := make(chan string)
+ // Wait for a message
+ go waitForRPCMessage(topic, ch, waitCh)
+ // Send the message - don't care of the response
+ rpc := "AnyRPCRequestForTest"
+ adapterKafkaProxy.InvokeRPC(nil, rpc, &topic, true)
+ // Wait for the result on ouw own channel
+ result := <-waitCh
+ assert.Equal(t, result, rpc)
+ close(waitCh)
+ err = coreKafkaProxy.UnSubscribe(topic, ch)
+ assert.Nil(t, err)
+}
+
+func TestMultipleSubscribeUnsubscribe(t *testing.T) {
+ // First subscribe to the specific topic
+ var err error
+ var ch1 <-chan *ca.InterContainerMessage
+ var ch2 <-chan *ca.InterContainerMessage
+ topic := kk.Topic{Name: "Core"}
+ ch1, err = coreKafkaProxy.Subscribe(topic)
+ assert.NotNil(t, ch1)
+ assert.Nil(t, err)
+ // Create a channel to receive responses
+ waitCh := make(chan string)
+ ch2, err = coreKafkaProxy.Subscribe(topic)
+ assert.NotNil(t, ch2)
+ assert.Nil(t, err)
+ // Wait for a message
+ go waitForRPCMessage(topic, ch2, waitCh)
+ go waitForRPCMessage(topic, ch1, waitCh)
+
+ // Send the message - don't care of the response
+ rpc := "AnyRPCRequestForTest"
+ adapterKafkaProxy.InvokeRPC(nil, rpc, &topic, true)
+ // Wait for the result on ouw own channel
+
+ responses := 0
+ for msg := range waitCh {
+ assert.Equal(t, msg, rpc)
+ responses = responses + 1
+ if responses > 1 {
+ break
+ }
+ }
+ assert.Equal(t, responses, 2)
+ close(waitCh)
+ err = coreKafkaProxy.UnSubscribe(topic, ch1)
+ assert.Nil(t, err)
+ err = coreKafkaProxy.UnSubscribe(topic, ch2)
+ assert.Nil(t, err)
+}
+
+func TestIncorrectAPI(t *testing.T) {
+ trnsId := uuid.New().String()
+ protoMsg := &voltha.Device{Id: trnsId}
+ args := make([]*kk.KVArg, 1)
+ args[0] = &kk.KVArg{
+ Key: "device",
+ Value: protoMsg,
+ }
+ rpc := "IncorrectAPI"
+ topic := kk.Topic{Name: "Core"}
+ start := time.Now()
+ status, result := adapterKafkaProxy.InvokeRPC(nil, rpc, &topic, true, args...)
+ elapsed := time.Since(start)
+ log.Infow("Result", log.Fields{"status": status, "result": result, "time": elapsed})
+ assert.Equal(t, status, false)
+ //Unpack the result into the actual proto object
+ unpackResult := &ca.Error{}
+ if err := ptypes.UnmarshalAny(result, unpackResult); err != nil {
+ log.Warnw("cannot-unmarshal-response", log.Fields{"error": err})
+ }
+ assert.NotNil(t, unpackResult)
+}
+
+func TestIncorrectAPIParams(t *testing.T) {
+ trnsId := uuid.New().String()
+ protoMsg := &voltha.Device{Id: trnsId}
+ args := make([]*kk.KVArg, 1)
+ args[0] = &kk.KVArg{
+ Key: "device",
+ Value: protoMsg,
+ }
+ rpc := "GetDevice"
+ topic := kk.Topic{Name: "Core"}
+ start := time.Now()
+ status, result := adapterKafkaProxy.InvokeRPC(nil, rpc, &topic, true, args...)
+ elapsed := time.Since(start)
+ log.Infow("Result", log.Fields{"status": status, "result": result, "time": elapsed})
+ assert.Equal(t, status, false)
+ //Unpack the result into the actual proto object
+ unpackResult := &ca.Error{}
+ if err := ptypes.UnmarshalAny(result, unpackResult); err != nil {
+ log.Warnw("cannot-unmarshal-response", log.Fields{"error": err})
+ }
+ assert.NotNil(t, unpackResult)
+}
+
+func TestGetDevice(t *testing.T) {
+ trnsId := uuid.New().String()
+ protoMsg := &ca.StrType{Val: trnsId}
+ args := make([]*kk.KVArg, 1)
+ args[0] = &kk.KVArg{
+ Key: "deviceID",
+ Value: protoMsg,
+ }
+ rpc := "GetDevice"
+ topic := kk.Topic{Name: "Core"}
+ expectedResponse := &voltha.Device{Id: trnsId}
+ timeout := time.Duration(50) * time.Millisecond
+ ctx, cancel := context.WithTimeout(context.Background(), timeout)
+ defer cancel()
+ start := time.Now()
+ status, result := adapterKafkaProxy.InvokeRPC(ctx, rpc, &topic, true, args...)
+ elapsed := time.Since(start)
+ log.Infow("Result", log.Fields{"status": status, "result": result, "time": elapsed})
+ assert.Equal(t, status, true)
+ unpackResult := &voltha.Device{}
+ if err := ptypes.UnmarshalAny(result, unpackResult); err != nil {
+ log.Warnw("cannot-unmarshal-response", log.Fields{"error": err})
+ }
+ assert.Equal(t, unpackResult, expectedResponse)
+}
+
+func TestGetDeviceTimeout(t *testing.T) {
+ trnsId := uuid.New().String()
+ protoMsg := &ca.StrType{Val: trnsId}
+ args := make([]*kk.KVArg, 1)
+ args[0] = &kk.KVArg{
+ Key: "deviceID",
+ Value: protoMsg,
+ }
+ rpc := "GetDevice"
+ topic := kk.Topic{Name: "Core"}
+ timeout := time.Duration(2) * time.Millisecond
+ ctx, cancel := context.WithTimeout(context.Background(), timeout)
+ defer cancel()
+ start := time.Now()
+ status, result := adapterKafkaProxy.InvokeRPC(ctx, rpc, &topic, true, args...)
+ elapsed := time.Since(start)
+ log.Infow("Result", log.Fields{"status": status, "result": result, "time": elapsed})
+ assert.Equal(t, status, false)
+ unpackResult := &ca.Error{}
+ if err := ptypes.UnmarshalAny(result, unpackResult); err != nil {
+ log.Warnw("cannot-unmarshal-response", log.Fields{"error": err})
+ }
+ assert.NotNil(t, unpackResult)
+}
+
+func TestGetChildDevice(t *testing.T) {
+ trnsId := uuid.New().String()
+ protoMsg := &ca.StrType{Val: trnsId}
+ args := make([]*kk.KVArg, 1)
+ args[0] = &kk.KVArg{
+ Key: "deviceID",
+ Value: protoMsg,
+ }
+ rpc := "GetChildDevice"
+ topic := kk.Topic{Name: "Core"}
+ expectedResponse := &voltha.Device{Id: trnsId}
+ start := time.Now()
+ status, result := adapterKafkaProxy.InvokeRPC(nil, rpc, &topic, true, args...)
+ elapsed := time.Since(start)
+ log.Infow("Result", log.Fields{"status": status, "result": result, "time": elapsed})
+ assert.Equal(t, status, true)
+ unpackResult := &voltha.Device{}
+ if err := ptypes.UnmarshalAny(result, unpackResult); err != nil {
+ log.Warnw("cannot-unmarshal-response", log.Fields{"error": err})
+ }
+ assert.Equal(t, unpackResult, expectedResponse)
+}
+
+func TestGetChildDevices(t *testing.T) {
+ trnsId := uuid.New().String()
+ protoMsg := &ca.StrType{Val: trnsId}
+ args := make([]*kk.KVArg, 1)
+ args[0] = &kk.KVArg{
+ Key: "deviceID",
+ Value: protoMsg,
+ }
+ rpc := "GetChildDevices"
+ topic := kk.Topic{Name: "Core"}
+ expectedResponse := &voltha.Device{Id: trnsId}
+ start := time.Now()
+ status, result := adapterKafkaProxy.InvokeRPC(nil, rpc, &topic, true, args...)
+ elapsed := time.Since(start)
+ log.Infow("Result", log.Fields{"status": status, "result": result, "time": elapsed})
+ assert.Equal(t, status, true)
+ unpackResult := &voltha.Device{}
+ if err := ptypes.UnmarshalAny(result, unpackResult); err != nil {
+ log.Warnw("cannot-unmarshal-response", log.Fields{"error": err})
+ }
+ assert.Equal(t, unpackResult, expectedResponse)
+}
+
+func TestGetPorts(t *testing.T) {
+ trnsId := uuid.New().String()
+ protoArg1 := &ca.StrType{Val: trnsId}
+ args := make([]*kk.KVArg, 2)
+ args[0] = &kk.KVArg{
+ Key: "deviceID",
+ Value: protoArg1,
+ }
+ protoArg2 := &ca.IntType{Val: 1}
+ args[1] = &kk.KVArg{
+ Key: "portType",
+ Value: protoArg2,
+ }
+ rpc := "GetPorts"
+ topic := kk.Topic{Name: "Core"}
+ start := time.Now()
+ status, result := adapterKafkaProxy.InvokeRPC(nil, rpc, &topic, true, args...)
+ elapsed := time.Since(start)
+ log.Infow("Result", log.Fields{"status": status, "result": result, "time": elapsed})
+ assert.Equal(t, status, true)
+ unpackResult := &voltha.Ports{}
+ if err := ptypes.UnmarshalAny(result, unpackResult); err != nil {
+ log.Warnw("cannot-unmarshal-response", log.Fields{"error": err})
+ }
+ expectedLen := len(unpackResult.Items) >= 1
+ assert.Equal(t, true, expectedLen)
+}
+
+func TestGetPortsMissingArgs(t *testing.T) {
+ trnsId := uuid.New().String()
+ protoArg1 := &ca.StrType{Val: trnsId}
+ args := make([]*kk.KVArg, 1)
+ args[0] = &kk.KVArg{
+ Key: "deviceID",
+ Value: protoArg1,
+ }
+ rpc := "GetPorts"
+ topic := kk.Topic{Name: "Core"}
+ start := time.Now()
+ status, result := adapterKafkaProxy.InvokeRPC(nil, rpc, &topic, true, args...)
+ elapsed := time.Since(start)
+ log.Infow("Result", log.Fields{"status": status, "result": result, "time": elapsed})
+ assert.Equal(t, status, false)
+ //Unpack the result into the actual proto object
+ unpackResult := &ca.Error{}
+ if err := ptypes.UnmarshalAny(result, unpackResult); err != nil {
+ log.Warnw("cannot-unmarshal-response", log.Fields{"error": err})
+ }
+ assert.NotNil(t, unpackResult)
+}
+
+func TestChildDeviceDetected(t *testing.T) {
+ trnsId := uuid.New().String()
+ protoArg1 := &ca.StrType{Val: trnsId}
+ args := make([]*kk.KVArg, 5)
+ args[0] = &kk.KVArg{
+ Key: "deviceID",
+ Value: protoArg1,
+ }
+ protoArg2 := &ca.IntType{Val: 1}
+ args[1] = &kk.KVArg{
+ Key: "parentPortNo",
+ Value: protoArg2,
+ }
+ protoArg3 := &ca.StrType{Val: "great_onu"}
+ args[2] = &kk.KVArg{
+ Key: "childDeviceType",
+ Value: protoArg3,
+ }
+ protoArg4 := &voltha.Device_ProxyAddress{DeviceId: trnsId, ChannelId: 100}
+ args[3] = &kk.KVArg{
+ Key: "proxyAddress",
+ Value: protoArg4,
+ }
+ protoArg5 := &ca.IntType{Val: 1}
+ args[4] = &kk.KVArg{
+ Key: "portType",
+ Value: protoArg5,
+ }
+
+ rpc := "ChildDeviceDetected"
+ topic := kk.Topic{Name: "Core"}
+ start := time.Now()
+ status, result := adapterKafkaProxy.InvokeRPC(nil, rpc, &topic, true, args...)
+ elapsed := time.Since(start)
+ log.Infow("Result", log.Fields{"status": status, "result": result, "time": elapsed})
+ assert.Equal(t, status, true)
+ assert.Nil(t, result)
+}
+
+func TestChildDeviceDetectedNoWait(t *testing.T) {
+ trnsId := uuid.New().String()
+ protoArg1 := &ca.StrType{Val: trnsId}
+ args := make([]*kk.KVArg, 5)
+ args[0] = &kk.KVArg{
+ Key: "deviceID",
+ Value: protoArg1,
+ }
+ protoArg2 := &ca.IntType{Val: 1}
+ args[1] = &kk.KVArg{
+ Key: "parentPortNo",
+ Value: protoArg2,
+ }
+ protoArg3 := &ca.StrType{Val: "great_onu"}
+ args[2] = &kk.KVArg{
+ Key: "childDeviceType",
+ Value: protoArg3,
+ }
+ protoArg4 := &voltha.Device_ProxyAddress{DeviceId: trnsId, ChannelId: 100}
+ args[3] = &kk.KVArg{
+ Key: "proxyAddress",
+ Value: protoArg4,
+ }
+ protoArg5 := &ca.IntType{Val: 1}
+ args[4] = &kk.KVArg{
+ Key: "portType",
+ Value: protoArg5,
+ }
+
+ rpc := "ChildDeviceDetected"
+ topic := kk.Topic{Name: "Core"}
+ start := time.Now()
+ status, result := adapterKafkaProxy.InvokeRPC(nil, rpc, &topic, false, args...)
+ elapsed := time.Since(start)
+ log.Infow("Result", log.Fields{"status": status, "result": result, "time": elapsed})
+ assert.Equal(t, status, true)
+ assert.Nil(t, result)
+}
+
+func TestChildDeviceDetectedMissingArgs(t *testing.T) {
+ trnsId := uuid.New().String()
+ protoArg1 := &ca.StrType{Val: trnsId}
+ args := make([]*kk.KVArg, 4)
+ args[0] = &kk.KVArg{
+ Key: "deviceID",
+ Value: protoArg1,
+ }
+ protoArg2 := &ca.IntType{Val: 1}
+ args[1] = &kk.KVArg{
+ Key: "parentPortNo",
+ Value: protoArg2,
+ }
+ protoArg3 := &ca.StrType{Val: "great_onu"}
+ args[2] = &kk.KVArg{
+ Key: "childDeviceType",
+ Value: protoArg3,
+ }
+ protoArg4 := &voltha.Device_ProxyAddress{DeviceId: trnsId, ChannelId: 100}
+ args[3] = &kk.KVArg{
+ Key: "proxyAddress",
+ Value: protoArg4,
+ }
+
+ rpc := "ChildDeviceDetected"
+ topic := kk.Topic{Name: "Core"}
+ start := time.Now()
+ status, result := adapterKafkaProxy.InvokeRPC(nil, rpc, &topic, true, args...)
+ elapsed := time.Since(start)
+ log.Infow("Result", log.Fields{"status": status, "result": result, "time": elapsed})
+ assert.Equal(t, status, false)
+ unpackResult := &ca.Error{}
+ if err := ptypes.UnmarshalAny(result, unpackResult); err != nil {
+ log.Warnw("cannot-unmarshal-response", log.Fields{"error": err})
+ }
+ assert.NotNil(t, unpackResult)
+}
+
+func TestStopKafkaProxy(t *testing.T) {
+ adapterKafkaProxy.Stop()
+ coreKafkaProxy.Stop()
+}
+
+//func TestMain(m *testing.T) {
+// log.Info("Main")
+//}