internal/pkg/afrouter/backend.go - voltha-api-server - Gitiles

 /*
  * Copyright 2018-present Open Networking Foundation

  * 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 afrouter

 // Backend manager handles redundant connections per backend

 import (
 	"errors"
 	"fmt"
 	"github.com/opencord/voltha-lib-go/v2/pkg/log"
 	"golang.org/x/net/context"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/metadata"
 	"google.golang.org/grpc/status"
 	"net/url"
 	"strconv"
 	"strings"
 	"sync"
 )

 // backend represents a collection of backends in a HA configuration
 type backend struct {
 	mutex             sync.Mutex
 	name              string
 	beType            backendType
 	activeAssociation association
 	connections       map[string]*connection
 	openConns         map[*connection]*grpc.ClientConn
 	activeRequests    map[*request]struct{}
 }

 type association struct {
 	strategy associationStrategy
 	location associationLocation
 	field    string // Used only if location is protobuf
 	key      string
 }

 // splitActiveStreamsUnsafe expects the caller to have already locked the backend mutex
 func (be *backend) splitActiveStreamsUnsafe(cn *connection, conn *grpc.ClientConn) {
 	if len(be.activeRequests) != 0 {
 		log.Debugf("Creating new streams for %d existing requests", len(be.activeRequests))
 	}
 	for r := range be.activeRequests {
 		r.mutex.Lock()
 		if _, have := r.streams[cn.name]; !have {
 			log.Debugf("Opening southbound stream for existing request '%s'", r.methodInfo.method)
 			if stream, err := grpc.NewClientStream(r.ctx, clientStreamDescForProxying, conn, r.methodInfo.all); err != nil {
 				log.Debugf("Failed to create a client stream '%s', %v", cn.name, err)
 			} else {
 				go r.catchupRequestStreamThenForwardResponseStream(cn.name, stream)
 				// new thread will unlock the request mutex
 				continue
 			}
 		}
 		r.mutex.Unlock()
 	}
 }

 // openSouthboundStreams sets up a connection to each southbound frame
 func (be *backend) openSouthboundStreams(srv interface{}, serverStream grpc.ServerStream, nf *requestFrame, sf *responseFrame) (*request, error) {
 	be.mutex.Lock()
 	defer be.mutex.Unlock()

 	isStreamingRequest, isStreamingResponse := nf.router.IsStreaming(nf.methodInfo.method)

 	// Get the metadata from the incoming message on the server
 	md, ok := metadata.FromIncomingContext(serverStream.Context())
 	if !ok {
 		return nil, errors.New("could not get a server stream metadata")
 	}

 	r := &request{
 		// Create an outgoing context that includes the incoming metadata and that will cancel if the server's context is canceled
 		ctx: metadata.AppendToOutgoingContext(metadata.NewOutgoingContext(serverStream.Context(), md.Copy()), "voltha_serial_number", nf.serialNo),

 		streams:         make(map[string]grpc.ClientStream),
 		responseErrChan: make(chan error, 1),

 		backend:             be,
 		serverStream:        serverStream,
 		methodInfo:          nf.methodInfo,
 		requestFrame:        nf,
 		responseFrame:       sf,
 		isStreamingRequest:  isStreamingRequest,
 		isStreamingResponse: isStreamingResponse,
 	}

 	log.Debugf("Opening southbound request for method '%s'", nf.methodInfo.method)

 	// TODO: Need to check if this is an active/active backend cluster
 	// with a serial number in the header.
 	log.Debugf("Serial number for transaction allocated: %s", nf.serialNo)
 	// If even one stream can be created then proceed. If none can be
 	// created then report an error because both the primary and redundant
 	// connections are non-existent.
 	var atLeastOne = false
 	var errStr strings.Builder

 	log.Debugf("There are %d/%d streams to open", len(be.openConns), len(be.connections))
 	if nf.connection != nil {
 		// Debug statement triggered by source router. Other routers have no connection preference.
 		log.Debugf("Looking for connection %s", nf.connection.name)
 	}
 	for cn, conn := range be.openConns {
 		// If source-router was used, it will indicate a specific connection to be used
 		if nf.connection != nil && nf.connection != cn {
 			continue
 		}

 		log.Debugf("Opening stream for connection '%s'", cn.name)
 		if stream, err := grpc.NewClientStream(r.ctx, clientStreamDescForProxying, conn, r.methodInfo.all); err != nil {
 			log.Debugf("Failed to create a client stream '%s', %v", cn.name, err)
 		} else {
 			r.streams[cn.name] = stream
 			go r.forwardResponseStream(cn.name, stream)
 			atLeastOne = true
 		}
 	}
 	if atLeastOne {
 		be.activeRequests[r] = struct{}{}
 		return r, nil
 	}
 	fmt.Fprintf(&errStr, "{{No open connections for backend '%s' unable to send}} ", be.name)
 	log.Error(errStr.String())
 	return nil, errors.New(errStr.String())
 }

 func (be *backend) handler(srv interface{}, serverStream grpc.ServerStream, nf *requestFrame, sf *responseFrame) error {
 	// Set up streams for each open connection
 	request, err := be.openSouthboundStreams(srv, serverStream, nf, sf)
 	if err != nil {
 		log.Errorf("openStreams failed: %v", err)
 		return err
 	}

 	log.Debug("Starting request stream forwarding")
 	if s2cErr := request.forwardRequestStream(serverStream); s2cErr != nil {
 		// exit with an error to the stack
 		return status.Errorf(codes.Internal, "failed proxying s2c: %v", s2cErr)
 	}
 	// wait for response stream to complete
 	return <-request.responseErrChan
 }

 func newBackend(conf *BackendConfig, clusterName string) (*backend, error) {
 	var rtrn_err bool = false

 	log.Debugf("Configuring the backend with %v", *conf)
 	// Validate the conifg and configure the backend
 	be := &backend{
 		name:           conf.Name,
 		connections:    make(map[string]*connection),
 		openConns:      make(map[*connection]*grpc.ClientConn),
 		activeRequests: make(map[*request]struct{}),
 	}
 	if conf.Type == BackendUndefined {
 		log.Error("Invalid type specified for backend %s in cluster %s", conf.Name, clusterName)
 		rtrn_err = true
 	}
 	be.beType = conf.Type

 	if conf.Association.Strategy == AssociationStrategyUndefined && be.beType == BackendActiveActive {
 		log.Errorf("An association strategy must be provided if the backend "+
 			"type is active/active for backend %s in cluster %s", conf.Name, clusterName)
 		rtrn_err = true
 	}
 	be.activeAssociation.strategy = conf.Association.Strategy

 	if conf.Association.Location == AssociationLocationUndefined && be.beType == BackendActiveActive {
 		log.Errorf("An association location must be provided if the backend "+
 			"type is active/active for backend %s in cluster %s", conf.Name, clusterName)
 		rtrn_err = true
 	}
 	be.activeAssociation.location = conf.Association.Location

 	if conf.Association.Field == "" && be.activeAssociation.location == AssociationLocationProtobuf {
 		log.Errorf("An association field must be provided if the backend "+
 			"type is active/active and the location is set to protobuf "+
 			"for backend %s in cluster %s", conf.Name, clusterName)
 		rtrn_err = true
 	}
 	be.activeAssociation.field = conf.Association.Field

 	if conf.Association.Key == "" && be.activeAssociation.location == AssociationLocationHeader {
 		log.Errorf("An association key must be provided if the backend "+
 			"type is active/active and the location is set to header "+
 			"for backend %s in cluster %s", conf.Name, clusterName)
 		rtrn_err = true
 	}
 	be.activeAssociation.key = conf.Association.Key
 	if rtrn_err {
 		return nil, errors.New("Backend configuration failed")
 	}
 	// Configure the connections
 	// Connections can consist of just a name. This allows for dynamic configuration
 	// at a later time.
 	// TODO: validate that there is one connection for all but active/active backends
 	if len(conf.Connections) > 1 && be.beType != BackendActiveActive {
 		log.Errorf("Only one connection must be specified if the association " +
 			"strategy is not set to 'active_active'")
 		rtrn_err = true
 	}
 	if len(conf.Connections) == 0 {
 		log.Errorf("At least one connection must be specified")
 		rtrn_err = true
 	}
 	for _, cnConf := range conf.Connections {
 		if cnConf.Name == "" {
 			log.Errorf("A connection must have a name for backend %s in cluster %s",
 				conf.Name, clusterName)
 		} else {
 			ctx, cancelFunc := context.WithCancel(context.Background())
 			be.connections[cnConf.Name] = &connection{name: cnConf.Name, addr: cnConf.Addr, port: cnConf.Port, backend: be, ctx: ctx, close: cancelFunc}
 			if _, err := url.Parse(cnConf.Addr); err != nil {
 				log.Errorf("The address for connection %s in backend %s in cluster %s is invalid: %s",
 					cnConf.Name, conf.Name, clusterName, err)
 				rtrn_err = true
 			}
 			// Validate the port number. This just validtes that it's a non 0 integer
 			if n, err := strconv.Atoi(cnConf.Port); err != nil || n <= 0 || n > 65535 {
 				log.Errorf("Port %s for connection %s in backend %s in cluster %s is invalid",
 					cnConf.Port, cnConf.Name, conf.Name, clusterName)
 				rtrn_err = true
 			} else {
 				if n <= 0 && n > 65535 {
 					log.Errorf("Port %s for connection %s in backend %s in cluster %s is invalid",
 						cnConf.Port, cnConf.Name, conf.Name, clusterName)
 					rtrn_err = true
 				}
 			}
 		}
 	}

 	if rtrn_err {
 		return nil, errors.New("Connection configuration failed")
 	}
 	// All is well start the backend cluster connections
 	be.connectAll()

 	return be, nil
 }

 func (be *backend) incConn(cn *connection, conn *grpc.ClientConn) {
 	be.mutex.Lock()
 	defer be.mutex.Unlock()

 	be.openConns[cn] = conn
 	be.splitActiveStreamsUnsafe(cn, conn)
 }

 func (be *backend) decConn(cn *connection) {
 	be.mutex.Lock()
 	defer be.mutex.Unlock()

 	delete(be.openConns, cn)
 }

 func (be *backend) NumOpenConnections() int {
 	be.mutex.Lock()
 	defer be.mutex.Unlock()

 	return len(be.openConns)
 }

 // Attempts to establish all the connections for a backend
 // any failures result in an abort. This should only be called
 // on a first attempt to connect. Individual connections should be
 // handled after that.
 func (be *backend) connectAll() {
 	for _, cn := range be.connections {
 		go cn.connect()
 	}
 }
	/*
	* Copyright 2018-present Open Networking Foundation

	* 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 afrouter

	// Backend manager handles redundant connections per backend

	import (
	"errors"
	"fmt"
	"github.com/opencord/voltha-lib-go/v2/pkg/log"
	"golang.org/x/net/context"
	"google.golang.org/grpc"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/metadata"
	"google.golang.org/grpc/status"
	"net/url"
	"strconv"
	"strings"
	"sync"
	)

	// backend represents a collection of backends in a HA configuration
	type backend struct {
	mutex sync.Mutex
	name string
	beType backendType
	activeAssociation association
	connections map[string]*connection
	openConns map[connection]grpc.ClientConn
	activeRequests map[*request]struct{}
	}

	type association struct {
	strategy associationStrategy
	location associationLocation
	field string // Used only if location is protobuf
	key string
	}

	// splitActiveStreamsUnsafe expects the caller to have already locked the backend mutex
	func (be backend) splitActiveStreamsUnsafe(cn connection, conn *grpc.ClientConn) {
	if len(be.activeRequests) != 0 {
	log.Debugf("Creating new streams for %d existing requests", len(be.activeRequests))
	}
	for r := range be.activeRequests {
	r.mutex.Lock()
	if _, have := r.streams[cn.name]; !have {
	log.Debugf("Opening southbound stream for existing request '%s'", r.methodInfo.method)
	if stream, err := grpc.NewClientStream(r.ctx, clientStreamDescForProxying, conn, r.methodInfo.all); err != nil {
	log.Debugf("Failed to create a client stream '%s', %v", cn.name, err)
	} else {
	go r.catchupRequestStreamThenForwardResponseStream(cn.name, stream)
	// new thread will unlock the request mutex
	continue
	}
	}
	r.mutex.Unlock()
	}
	}

	// openSouthboundStreams sets up a connection to each southbound frame
	func (be backend) openSouthboundStreams(srv interface{}, serverStream grpc.ServerStream, nf requestFrame, sf responseFrame) (request, error) {
	be.mutex.Lock()
	defer be.mutex.Unlock()

	isStreamingRequest, isStreamingResponse := nf.router.IsStreaming(nf.methodInfo.method)

	// Get the metadata from the incoming message on the server
	md, ok := metadata.FromIncomingContext(serverStream.Context())
	if !ok {
	return nil, errors.New("could not get a server stream metadata")
	}

	r := &request{
	// Create an outgoing context that includes the incoming metadata and that will cancel if the server's context is canceled
	ctx: metadata.AppendToOutgoingContext(metadata.NewOutgoingContext(serverStream.Context(), md.Copy()), "voltha_serial_number", nf.serialNo),

	streams: make(map[string]grpc.ClientStream),
	responseErrChan: make(chan error, 1),

	backend: be,
	serverStream: serverStream,
	methodInfo: nf.methodInfo,
	requestFrame: nf,
	responseFrame: sf,
	isStreamingRequest: isStreamingRequest,
	isStreamingResponse: isStreamingResponse,
	}

	log.Debugf("Opening southbound request for method '%s'", nf.methodInfo.method)

	// TODO: Need to check if this is an active/active backend cluster
	// with a serial number in the header.
	log.Debugf("Serial number for transaction allocated: %s", nf.serialNo)
	// If even one stream can be created then proceed. If none can be
	// created then report an error because both the primary and redundant
	// connections are non-existent.
	var atLeastOne = false
	var errStr strings.Builder

	log.Debugf("There are %d/%d streams to open", len(be.openConns), len(be.connections))
	if nf.connection != nil {
	// Debug statement triggered by source router. Other routers have no connection preference.
	log.Debugf("Looking for connection %s", nf.connection.name)
	}
	for cn, conn := range be.openConns {
	// If source-router was used, it will indicate a specific connection to be used
	if nf.connection != nil && nf.connection != cn {
	continue
	}

	log.Debugf("Opening stream for connection '%s'", cn.name)
	if stream, err := grpc.NewClientStream(r.ctx, clientStreamDescForProxying, conn, r.methodInfo.all); err != nil {
	log.Debugf("Failed to create a client stream '%s', %v", cn.name, err)
	} else {
	r.streams[cn.name] = stream
	go r.forwardResponseStream(cn.name, stream)
	atLeastOne = true
	}
	}
	if atLeastOne {
	be.activeRequests[r] = struct{}{}
	return r, nil
	}
	fmt.Fprintf(&errStr, "{{No open connections for backend '%s' unable to send}} ", be.name)
	log.Error(errStr.String())
	return nil, errors.New(errStr.String())
	}

	func (be backend) handler(srv interface{}, serverStream grpc.ServerStream, nf requestFrame, sf *responseFrame) error {
	// Set up streams for each open connection
	request, err := be.openSouthboundStreams(srv, serverStream, nf, sf)
	if err != nil {
	log.Errorf("openStreams failed: %v", err)
	return err
	}

	log.Debug("Starting request stream forwarding")
	if s2cErr := request.forwardRequestStream(serverStream); s2cErr != nil {
	// exit with an error to the stack
	return status.Errorf(codes.Internal, "failed proxying s2c: %v", s2cErr)
	}
	// wait for response stream to complete
	return <-request.responseErrChan
	}

	func newBackend(conf BackendConfig, clusterName string) (backend, error) {
	var rtrn_err bool = false

	log.Debugf("Configuring the backend with %v", *conf)
	// Validate the conifg and configure the backend
	be := &backend{
	name: conf.Name,
	connections: make(map[string]*connection),
	openConns: make(map[connection]grpc.ClientConn),
	activeRequests: make(map[*request]struct{}),
	}
	if conf.Type == BackendUndefined {
	log.Error("Invalid type specified for backend %s in cluster %s", conf.Name, clusterName)
	rtrn_err = true
	}
	be.beType = conf.Type

	if conf.Association.Strategy == AssociationStrategyUndefined && be.beType == BackendActiveActive {
	log.Errorf("An association strategy must be provided if the backend "+
	"type is active/active for backend %s in cluster %s", conf.Name, clusterName)
	rtrn_err = true
	}
	be.activeAssociation.strategy = conf.Association.Strategy

	if conf.Association.Location == AssociationLocationUndefined && be.beType == BackendActiveActive {
	log.Errorf("An association location must be provided if the backend "+
	"type is active/active for backend %s in cluster %s", conf.Name, clusterName)
	rtrn_err = true
	}
	be.activeAssociation.location = conf.Association.Location

	if conf.Association.Field == "" && be.activeAssociation.location == AssociationLocationProtobuf {
	log.Errorf("An association field must be provided if the backend "+
	"type is active/active and the location is set to protobuf "+
	"for backend %s in cluster %s", conf.Name, clusterName)
	rtrn_err = true
	}
	be.activeAssociation.field = conf.Association.Field

	if conf.Association.Key == "" && be.activeAssociation.location == AssociationLocationHeader {
	log.Errorf("An association key must be provided if the backend "+
	"type is active/active and the location is set to header "+
	"for backend %s in cluster %s", conf.Name, clusterName)
	rtrn_err = true
	}
	be.activeAssociation.key = conf.Association.Key
	if rtrn_err {
	return nil, errors.New("Backend configuration failed")
	}
	// Configure the connections
	// Connections can consist of just a name. This allows for dynamic configuration
	// at a later time.
	// TODO: validate that there is one connection for all but active/active backends
	if len(conf.Connections) > 1 && be.beType != BackendActiveActive {
	log.Errorf("Only one connection must be specified if the association " +
	"strategy is not set to 'active_active'")
	rtrn_err = true
	}
	if len(conf.Connections) == 0 {
	log.Errorf("At least one connection must be specified")
	rtrn_err = true
	}
	for _, cnConf := range conf.Connections {
	if cnConf.Name == "" {
	log.Errorf("A connection must have a name for backend %s in cluster %s",
	conf.Name, clusterName)
	} else {
	ctx, cancelFunc := context.WithCancel(context.Background())
	be.connections[cnConf.Name] = &connection{name: cnConf.Name, addr: cnConf.Addr, port: cnConf.Port, backend: be, ctx: ctx, close: cancelFunc}
	if _, err := url.Parse(cnConf.Addr); err != nil {
	log.Errorf("The address for connection %s in backend %s in cluster %s is invalid: %s",
	cnConf.Name, conf.Name, clusterName, err)
	rtrn_err = true
	}
	// Validate the port number. This just validtes that it's a non 0 integer
	if n, err := strconv.Atoi(cnConf.Port); err != nil \|\| n <= 0 \|\| n > 65535 {
	log.Errorf("Port %s for connection %s in backend %s in cluster %s is invalid",
	cnConf.Port, cnConf.Name, conf.Name, clusterName)
	rtrn_err = true
	} else {
	if n <= 0 && n > 65535 {
	log.Errorf("Port %s for connection %s in backend %s in cluster %s is invalid",
	cnConf.Port, cnConf.Name, conf.Name, clusterName)
	rtrn_err = true
	}
	}
	}
	}

	if rtrn_err {
	return nil, errors.New("Connection configuration failed")
	}
	// All is well start the backend cluster connections
	be.connectAll()

	return be, nil
	}

	func (be backend) incConn(cn connection, conn *grpc.ClientConn) {
	be.mutex.Lock()
	defer be.mutex.Unlock()

	be.openConns[cn] = conn
	be.splitActiveStreamsUnsafe(cn, conn)
	}

	func (be backend) decConn(cn connection) {
	be.mutex.Lock()
	defer be.mutex.Unlock()

	delete(be.openConns, cn)
	}

	func (be *backend) NumOpenConnections() int {
	be.mutex.Lock()
	defer be.mutex.Unlock()

	return len(be.openConns)
	}

	// Attempts to establish all the connections for a backend
	// any failures result in an abort. This should only be called
	// on a first attempt to connect. Individual connections should be
	// handled after that.
	func (be *backend) connectAll() {
	for _, cn := range be.connections {
	go cn.connect()
	}
	}