| /* |
| * 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. |
| */ |
| // gRPC affinity router with active/active backends |
| |
| package afrouter |
| |
| import ( |
| "net" |
| "fmt" |
| "strconv" |
| "errors" |
| "google.golang.org/grpc" |
| "golang.org/x/net/context" |
| "github.com/opencord/voltha-go/common/log" |
| pb "github.com/opencord/voltha-go/protos/afrouter" |
| ) |
| |
| |
| type ArouterApi struct { |
| addr string |
| port int |
| apiListener net.Listener |
| apiServer * grpc.Server |
| running bool |
| ar *ArouterProxy |
| } |
| |
| func newApi(config *ApiConfig, ar *ArouterProxy) (*ArouterApi, error) { |
| var rtrn_err bool |
| // Create a seperate server and listener for the API |
| // Validate the ip address if one is provided |
| if ip := net.ParseIP(config.Addr); config.Addr != "" && ip == nil { |
| log.Errorf("Invalid address '%s' provided for API server", config.Addr) |
| rtrn_err = true |
| } |
| if rtrn_err == true { |
| return nil, errors.New("Errors in API configuration") |
| } else { |
| var err error = nil |
| aa := &ArouterApi{addr:config.Addr,port:int(config.Port),ar:ar} |
| // Create the listener for the API server |
| if aa.apiListener, err = |
| net.Listen("tcp", config.Addr + ":"+ |
| strconv.Itoa(int(config.Port))); err != nil { |
| log.Error(err) |
| return nil, err |
| } |
| // Create the API server |
| aa.apiServer = grpc.NewServer() |
| pb.RegisterConfigurationServer(aa.apiServer, *aa) |
| return aa,err |
| } |
| } |
| |
| func (aa *ArouterApi) getServer(srvr string) (*server, error) { |
| if s,ok := aa.ar.servers[srvr]; ok == false { |
| err := errors.New(fmt.Sprintf("Server '%s' doesn't exist", srvr)) |
| return nil, err |
| } else { |
| return s, nil |
| } |
| } |
| |
| func (aa *ArouterApi) getRouter(s *server, clstr string) (Router, error) { |
| for _,pkg := range s.routers { |
| for _,r := range pkg { |
| if c := r.FindBackendCluster(clstr); c != nil { |
| return r, nil |
| } |
| } |
| } |
| err := errors.New(fmt.Sprintf("Cluster '%s' doesn't exist", clstr)) |
| return nil, err |
| } |
| |
| func (aa *ArouterApi) getCluster(s *server, clstr string) (*backendCluster, error) { |
| for _,pkg := range s.routers { |
| for _,r := range pkg { |
| if c := r.FindBackendCluster(clstr); c != nil { |
| return c, nil |
| } |
| } |
| } |
| err := errors.New(fmt.Sprintf("Cluster '%s' doesn't exist", clstr)) |
| return nil, err |
| } |
| |
| func (aa *ArouterApi) getBackend(c *backendCluster, bknd string) (*backend, error) { |
| for _,b := range c.backends { |
| if b.name == bknd { |
| return b,nil |
| } |
| } |
| err := errors.New(fmt.Sprintf("Backend '%s' doesn't exist", bknd)) |
| return nil, err |
| } |
| |
| func (aa *ArouterApi) getConnection(b *backend, con string) (*beConnection, error) { |
| if c,ok := b.connections[con]; ok == false { |
| err := errors.New(fmt.Sprintf("Connection '%s' doesn't exist", con)) |
| return nil, err |
| } else { |
| return c, nil |
| } |
| } |
| |
| func (aa * ArouterApi) updateConnection(in *pb.Conn, cn *beConnection, b *backend) error { |
| sPort := strconv.FormatUint(in.Port,10) |
| // Check that the ip address and or port are different |
| if in.Addr == cn.addr && sPort == cn.port { |
| err := errors.New(fmt.Sprintf("Refusing to change connection '%s' to identical values", in.Connection)) |
| return err |
| } |
| //log.Debugf("BEFORE: Be1: %v Be2 %v", cn.bknd, b) |
| cn.close() |
| cn.addr = in.Addr |
| cn.port = sPort |
| cn.connect() |
| return nil |
| } |
| |
| func (aa ArouterApi) SetAffinity(ctx context.Context, in *pb.Affinity) (*pb.Result, error) { |
| log.Debugf("SetAffinity called! %v",in); |
| //return &pb.Result{Success:true,Error:""},nil |
| // Navigate down tot he connection and compare IP addresses and ports if they're |
| // not the same then close the existing connection. If they are bothe the same |
| // then return an error describing the situation. |
| var err error |
| |
| aap := &aa |
| |
| _=aap |
| |
| log.Debugf("Getting router %s and route %s", in.Router, in.Route) |
| if r,ok := allRouters[in.Router+in.Route]; ok == true { |
| switch rr := r.(type) { |
| case AffinityRouter: |
| log.Debug("Affinity router found") |
| b := rr.FindBackendCluster(in.Cluster).getBackend(in.Backend) |
| if b != nil { |
| rr.setAffinity(in.Id, b) |
| } else { |
| log.Errorf("Requested backend '%s' not found", in.Backend) |
| } |
| _ = rr |
| case MethodRouter: |
| log.Debug("Method router found") |
| _ = rr |
| default: |
| log.Debug("Some other router found") |
| _ = rr |
| } |
| } else { |
| log.Debugf("Couldn't get router type") |
| return &pb.Result{Success:false,Error:err.Error()}, err |
| } |
| |
| return &pb.Result{Success:true,Error:""},nil |
| } |
| |
| func (aa ArouterApi) SetConnection(ctx context.Context, in *pb.Conn) (*pb.Result, error) { |
| log.Debugf("SetConnection called! %v",in); |
| // Navigate down tot he connection and compare IP addresses and ports if they're |
| // not the same then close the existing connection. If they are bothe the same |
| // then return an error describing the situation. |
| var s *server |
| var c *backendCluster |
| var b * backend |
| var cn * beConnection |
| var err error |
| |
| aap := &aa |
| if s,err = (aap).getServer(in.Server); err != nil { |
| err := errors.New(fmt.Sprintf("Server '%s' doesn't exist", in.Server)) |
| return &pb.Result{Success:false,Error:err.Error()}, err |
| } |
| // The cluster is usually accessed via tha router but since each |
| // cluster is unique it's good enough to find the router that |
| // has the cluster we're looking for rather than fully keying |
| // the path |
| if c,err = aap.getCluster(s, in.Cluster); err != nil { |
| return &pb.Result{Success:false,Error:err.Error()}, err |
| } |
| |
| if b,err = aap.getBackend(c, in.Backend); err != nil { |
| return &pb.Result{Success:false,Error:err.Error()}, err |
| } |
| |
| if cn,err = aap.getConnection(b, in.Connection); err != nil { |
| return &pb.Result{Success:false,Error:err.Error()}, err |
| } |
| |
| if err = aap.updateConnection(in, cn, b); err != nil { |
| return &pb.Result{Success:false,Error:err.Error()}, err |
| } |
| |
| return &pb.Result{Success:true,Error:""},nil |
| } |
| |
| func (aa *ArouterApi) serve() { |
| // Start a serving thread |
| go func() { |
| aa.running = true |
| if err := aa.apiServer.Serve(aa.apiListener); err != nil { |
| aa.running = false |
| log.Error(err) |
| errChan <- err |
| } |
| }() |
| } |
| |