internal/pkg/openflow/connection.go

/*
   Copyright 2020 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.
*/
package openflow

import (
	"bufio"
	"context"
	"encoding/binary"
	"encoding/json"
	"errors"
	"io"
	"net"
	"time"

	"github.com/opencord/goloxi"
	ofp "github.com/opencord/goloxi/of13"
	"github.com/opencord/ofagent-go/internal/pkg/holder"
	"github.com/opencord/voltha-lib-go/v3/pkg/log"
	"github.com/opencord/voltha-protos/v3/go/voltha"
)

type OFConnection struct {
	OFControllerEndPoint string
	DeviceID             string
	VolthaClient         *holder.VolthaServiceClientHolder
	PacketOutChannel     chan *voltha.PacketOut
	ConnectionMaxRetries int
	ConnectionRetryDelay time.Duration

	conn net.Conn

	// current role of this connection
	role        ofcRole
	roleManager RoleManager

	events            chan ofcEvent
	sendChannel       chan Message
	lastUnsentMessage Message

	flowsChunkSize     int
	portsChunkSize     int
	portsDescChunkSize int
}

func (ofc *OFConnection) peekAtOFHeader(buf []byte) (ofp.IHeader, error) {
	header := ofp.Header{}
	header.Version = uint8(buf[0])
	header.Type = uint8(buf[1])
	header.Length = binary.BigEndian.Uint16(buf[2:4])
	header.Xid = binary.BigEndian.Uint32(buf[4:8])

	// TODO: add minimal validation of version and type

	return &header, nil
}

func (ofc *OFConnection) establishConnectionToController(ctx context.Context) error {
	if ofc.conn != nil {
		logger.Debugw(ctx, "closing-of-connection-to-reconnect",
			log.Fields{"device-id": ofc.DeviceID})
		ofc.conn.Close()
		ofc.conn = nil
	}
	try := 1
	for ofc.ConnectionMaxRetries == 0 || try < ofc.ConnectionMaxRetries {
		if raddr, err := net.ResolveTCPAddr("tcp", ofc.OFControllerEndPoint); err != nil {
			logger.Debugw(ctx, "openflow-client unable to resolve endpoint",
				log.Fields{
					"device-id": ofc.DeviceID,
					"endpoint":  ofc.OFControllerEndPoint})
		} else {
			if connection, err := net.DialTCP("tcp", nil, raddr); err == nil {
				ofc.conn = connection
				ofc.sayHello(ctx)
				ofc.events <- ofcEventConnect
				return nil
			} else {
				logger.Warnw(ctx, "openflow-client-connect-error",
					log.Fields{
						"device-id": ofc.DeviceID,
						"endpoint":  ofc.OFControllerEndPoint})
			}
		}
		if ofc.ConnectionMaxRetries == 0 || try < ofc.ConnectionMaxRetries {
			if ofc.ConnectionMaxRetries != 0 {
				try += 1
			}
			time.Sleep(ofc.ConnectionRetryDelay)
		}
	}
	return errors.New("failed-to-connect-to-of-controller")
}

// Run implements the state machine for the OF client reacting to state change
// events and invoking actions as a reaction to those state changes
func (ofc *OFConnection) Run(ctx context.Context) {

	var ofCtx context.Context
	var ofDone func()
	state := ofcStateCreated
	ofc.events <- ofcEventStart
top:
	for {
		select {
		case <-ctx.Done():
			state = ofcStateStopped
			logger.Debugw(ctx, "state-transition-context-done",
				log.Fields{"device-id": ofc.DeviceID})
			break top
		case event := <-ofc.events:
			previous := state
			switch event {
			case ofcEventStart:
				logger.Debugw(ctx, "ofc-event-start",
					log.Fields{"device-id": ofc.DeviceID})
				if state == ofcStateCreated {
					state = ofcStateStarted
					logger.Debug(ctx, "STARTED MORE THAN ONCE")
					go func() {
						if err := ofc.establishConnectionToController(ctx); err != nil {
							logger.Errorw(ctx, "controller-connection-failed", log.Fields{"error": err})
							panic(err)
						}
					}()
				} else {
					logger.Errorw(ctx, "illegal-state-transition",
						log.Fields{
							"device-id":     ofc.DeviceID,
							"current-state": state.String(),
							"event":         event.String()})
				}
			case ofcEventConnect:
				logger.Debugw(ctx, "ofc-event-connected",
					log.Fields{"device-id": ofc.DeviceID})
				if state == ofcStateStarted || state == ofcStateDisconnected {
					state = ofcStateConnected
					ofCtx, ofDone = context.WithCancel(log.WithSpanFromContext(context.Background(), ctx))
					go ofc.messageSender(ofCtx)
					go ofc.processOFStream(ofCtx)
				} else {
					logger.Errorw(ctx, "illegal-state-transition",
						log.Fields{
							"device-id":     ofc.DeviceID,
							"current-state": state.String(),
							"event":         event.String()})
				}
			case ofcEventDisconnect:
				logger.Debugw(ctx, "ofc-event-disconnected",
					log.Fields{
						"device-id": ofc.DeviceID,
						"state":     state.String()})
				if state == ofcStateConnected {
					state = ofcStateDisconnected
					if ofDone != nil {
						ofDone()
						ofDone = nil
					}
					go func() {
						if err := ofc.establishConnectionToController(ctx); err != nil {
							logger.Errorw(ctx, "controller-connection-failed", log.Fields{"error": err})
							panic(err)
						}
					}()
				} else {
					logger.Errorw(ctx, "illegal-state-transition",
						log.Fields{
							"device-id":     ofc.DeviceID,
							"current-state": state.String(),
							"event":         event.String()})
				}
			case ofcEventStop:
				logger.Debugw(ctx, "ofc-event-stop",
					log.Fields{"device-id": ofc.DeviceID})
				if state == ofcStateCreated || state == ofcStateConnected || state == ofcStateDisconnected {
					state = ofcStateStopped
					break top
				} else {
					logger.Errorw(ctx, "illegal-state-transition",
						log.Fields{
							"device-id":     ofc.DeviceID,
							"current-state": state.String(),
							"event":         event.String()})
				}
			}
			logger.Debugw(ctx, "state-transition",
				log.Fields{
					"device-id":      ofc.DeviceID,
					"previous-state": previous.String(),
					"current-state":  state.String(),
					"event":          event.String()})
		}
	}

	// If the child context exists, then cancel it
	if ofDone != nil {
		logger.Debugw(ctx, "closing-child-processes",
			log.Fields{"device-id": ofc.DeviceID})
		ofDone()
	}

	// If the connection is open, then close it
	if ofc.conn != nil {
		logger.Debugw(ctx, "closing-of-connection",
			log.Fields{"device-id": ofc.DeviceID})
		ofc.conn.Close()
		ofc.conn = nil
	}
	logger.Debugw(ctx, "state-machine-finished",
		log.Fields{"device-id": ofc.DeviceID})
}

// processOFStream processes the OF connection from the controller and invokes
// the appropriate handler methods for each message.
func (ofc *OFConnection) processOFStream(ctx context.Context) {
	fromController := bufio.NewReader(ofc.conn)

	/*
	 * We have a read buffer of a max size of 4096, so if we ever have
	 * a message larger than this then we will have issues
	 */
	headerBuf := make([]byte, 8)
top:
	// Continue until we are told to stop
	for {
		select {
		case <-ctx.Done():
			logger.Error(ctx, "of-loop-ending-context-done")
			break top
		default:
			// Read 8 bytes, the standard OF header
			read, err := io.ReadFull(fromController, headerBuf)
			if err != nil {
				if err == io.EOF {
					logger.Infow(ctx, "controller-disconnected",
						log.Fields{
							"device-id":  ofc.DeviceID,
							"controller": ofc.OFControllerEndPoint,
						})
				} else {
					logger.Errorw(ctx, "bad-of-header",
						log.Fields{
							"byte-count": read,
							"device-id":  ofc.DeviceID,
							"controller": ofc.OFControllerEndPoint,
							"error":      err})
				}
				break top
			}

			// Decode the header
			peek, err := ofc.peekAtOFHeader(headerBuf)
			if err != nil {
				/*
				 * Header is bad, assume stream is corrupted
				 * and needs to be restarted
				 */
				logger.Errorw(ctx, "bad-of-packet",
					log.Fields{
						"device-id": ofc.DeviceID,
						"error":     err})
				break top
			}

			// Calculate the size of the rest of the packet and read it
			need := int(peek.GetLength())
			messageBuf := make([]byte, need)
			copy(messageBuf, headerBuf)
			read, err = io.ReadFull(fromController, messageBuf[8:])
			if err != nil {
				logger.Errorw(ctx, "bad-of-packet",
					log.Fields{
						"byte-count": read,
						"device-id":  ofc.DeviceID,
						"error":      err})
				break top
			}

			// Decode and process the packet
			decoder := goloxi.NewDecoder(messageBuf)
			msg, err := ofp.DecodeHeader(decoder)
			if err != nil {
				// nolint: staticcheck
				js, _ := json.Marshal(decoder)
				logger.Errorw(ctx, "failed-to-decode",
					log.Fields{
						"device-id": ofc.DeviceID,
						"decoder":   js,
						"error":     err})
				break top
			}
			if logger.V(log.DebugLevel) {
				js, _ := json.Marshal(msg)
				logger.Debugw(ctx, "packet-header",
					log.Fields{
						"device-id": ofc.DeviceID,
						"header":    js})
			}
			/*
			 * Spawning a go routine for every incoming message removes processing ordering guarantees.
			 * Removing such guarantees puts burden on the controller to ensure the correct ordering of
			 * incoming messages and is a less optimal and safe agent implementation.
			 * This is OK for now because ONOS keeps the order guaranteed but the agent needs to avoid
			 * relying on external fairness. Particular care and attention has to be placed in flow add/delete
			 * and relative barrier requests. e.g. a flowMod will be handled in thread different from a barrier,
			 * with no guarantees of handling all messages before a barrier.
			 * A multiple queue (incoming worker and outgoing) is a possible solution.
			 */
			go ofc.parseHeader(ctx, msg)
		}
	}
	logger.Debugw(ctx, "end-of-stream",
		log.Fields{"device-id": ofc.DeviceID})
	ofc.events <- ofcEventDisconnect
}

func (ofc *OFConnection) sayHello(ctx context.Context) {
	hello := ofp.NewHello()
	hello.Xid = uint32(GetXid())
	elem := ofp.NewHelloElemVersionbitmap()
	elem.SetType(ofp.OFPHETVersionbitmap)
	elem.SetLength(8)
	elem.SetBitmaps([]*ofp.Uint32{{Value: 16}})
	hello.SetElements([]ofp.IHelloElem{elem})
	if logger.V(log.DebugLevel) {
		js, _ := json.Marshal(hello)
		logger.Debugw(ctx, "sayHello Called",
			log.Fields{
				"device-id":     ofc.DeviceID,
				"hello-message": js})
	}
	if err := ofc.SendMessage(ctx, hello); err != nil {
		logger.Fatalw(ctx, "Failed saying hello to Openflow Server, unable to proceed",
			log.Fields{
				"device-id": ofc.DeviceID,
				"error":     err})
	}
}

func (ofc *OFConnection) parseHeader(ctx context.Context, header ofp.IHeader) {
	headerType := header.GetType()
	logger.Debugw(ctx, "packet-header-type",
		log.Fields{
			"header-type": ofp.Type(headerType).String()})
	switch headerType {
	case ofp.OFPTHello:
		//x := header.(*ofp.Hello)
	case ofp.OFPTError:
		go ofc.handleErrMsg(ctx, header.(*ofp.ErrorMsg))
	case ofp.OFPTEchoRequest:
		go ofc.handleEchoRequest(ctx, header.(*ofp.EchoRequest))
	case ofp.OFPTEchoReply:
	case ofp.OFPTExperimenter:
	case ofp.OFPTFeaturesRequest:
		go func() {
			if err := ofc.handleFeatureRequest(ctx, header.(*ofp.FeaturesRequest)); err != nil {
				logger.Errorw(ctx, "handle-feature-request", log.Fields{"error": err})
			}
		}()
	case ofp.OFPTFeaturesReply:
	case ofp.OFPTGetConfigRequest:
		go ofc.handleGetConfigRequest(ctx, header.(*ofp.GetConfigRequest))
	case ofp.OFPTGetConfigReply:
	case ofp.OFPTSetConfig:
		go ofc.handleSetConfig(ctx, header.(*ofp.SetConfig))
	case ofp.OFPTPacketIn:
	case ofp.OFPTFlowRemoved:
	case ofp.OFPTPortStatus:
	case ofp.OFPTPacketOut:
		if !(ofc.role == ofcRoleMaster || ofc.role == ofcRoleEqual) {
			ofc.sendRoleSlaveError(ctx, header)
			return
		}
		go ofc.handlePacketOut(ctx, header.(*ofp.PacketOut))
	case ofp.OFPTFlowMod:
		if !(ofc.role == ofcRoleMaster || ofc.role == ofcRoleEqual) {
			ofc.sendRoleSlaveError(ctx, header)
			return
		}
		switch header.(ofp.IFlowMod).GetCommand() {
		case ofp.OFPFCAdd:
			ofc.handleFlowAdd(ctx, header.(*ofp.FlowAdd))
		case ofp.OFPFCModify:
			ofc.handleFlowMod(ctx, header.(*ofp.FlowMod))
		case ofp.OFPFCModifyStrict:
			ofc.handleFlowModStrict(ctx, header.(*ofp.FlowModifyStrict))
		case ofp.OFPFCDelete:
			ofc.handleFlowDelete(ctx, header.(*ofp.FlowDelete))
		case ofp.OFPFCDeleteStrict:
			ofc.handleFlowDeleteStrict(ctx, header.(*ofp.FlowDeleteStrict))
		}
	case ofp.OFPTStatsRequest:
		go func() {
			if err := ofc.handleStatsRequest(ctx, header, header.(ofp.IStatsRequest).GetStatsType()); err != nil {
				logger.Errorw(ctx, "ofpt-stats-request", log.Fields{"error": err})
			}
		}()
	case ofp.OFPTBarrierRequest:
		/* See note above at case ofp.OFPTFlowMod:*/
		ofc.handleBarrierRequest(ctx, header.(*ofp.BarrierRequest))
	case ofp.OFPTRoleRequest:
		go ofc.handleRoleRequest(ctx, header.(*ofp.RoleRequest))
	case ofp.OFPTMeterMod:
		if !(ofc.role == ofcRoleMaster || ofc.role == ofcRoleEqual) {
			ofc.sendRoleSlaveError(ctx, header)
			return
		}
		ofc.handleMeterModRequest(ctx, header.(*ofp.MeterMod))
	case ofp.OFPTGroupMod:
		if !(ofc.role == ofcRoleMaster || ofc.role == ofcRoleEqual) {
			ofc.sendRoleSlaveError(ctx, header)
			return
		}
		ofc.handleGroupMod(ctx, header.(ofp.IGroupMod))
	}
}

// Message interface that represents an open flow message and enables for a
// unified implementation of SendMessage
type Message interface {
	Serialize(encoder *goloxi.Encoder) error
}

func (ofc *OFConnection) doSend(ctx context.Context, msg Message) error {
	if ofc.conn == nil {
		return errors.New("no-connection")
	}
	enc := goloxi.NewEncoder()
	if err := msg.Serialize(enc); err != nil {
		return err
	}

	bytes := enc.Bytes()
	if _, err := ofc.conn.Write(bytes); err != nil {
		logger.Errorw(ctx, "unable-to-send-message-to-controller",
			log.Fields{
				"device-id": ofc.DeviceID,
				"message":   msg,
				"error":     err})
		return err
	}
	return nil
}

func (ofc *OFConnection) messageSender(ctx context.Context) {
	// first process last fail if it exists
	if ofc.lastUnsentMessage != nil {
		if err := ofc.doSend(ctx, ofc.lastUnsentMessage); err != nil {
			ofc.events <- ofcEventDisconnect
			return
		}
		ofc.lastUnsentMessage = nil
	}
top:
	for {
		select {
		case <-ctx.Done():
			break top
		case msg := <-ofc.sendChannel:
			if err := ofc.doSend(ctx, msg); err != nil {
				ofc.lastUnsentMessage = msg
				ofc.events <- ofcEventDisconnect
				logger.Debugw(ctx, "message-sender-error",
					log.Fields{
						"device-id": ofc.DeviceID,
						"error":     err.Error()})
				break top
			}
			logger.Debugw(ctx, "message-sender-send",
				log.Fields{
					"device-id": ofc.DeviceID})
			ofc.lastUnsentMessage = nil
		}
	}

	logger.Debugw(ctx, "message-sender-finished",
		log.Fields{
			"device-id": ofc.DeviceID})
}

// SendMessage queues a message to be sent to the openflow controller
func (ofc *OFConnection) SendMessage(ctx context.Context, message Message) error {
	logger.Debugw(ctx, "queuing-message", log.Fields{
		"endpoint": ofc.OFControllerEndPoint,
		"role":     ofc.role,
	})
	ofc.sendChannel <- message
	return nil
}