Add ovn IC controller

Signed-off-by: Aswin Suryanarayanan <[email protected]>

pkg/routeagent_driver/handlers/ovn/connection.go

@@ -0,0 +1,173 @@
+/*
+SPDX-License-Identifier: Apache-2.0
+
+Copyright Contributors to the Submariner project.
+
+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 ovn
+
+import (
+	"context"
+	"crypto/tls"
+	"crypto/x509"
+	"os"
+	"strings"
+	"sync"
+
+	"github.com/cenkalti/backoff/v4"
+	libovsdbclient "github.com/ovn-org/libovsdb/client"
+	"github.com/ovn-org/libovsdb/model"
+	"github.com/ovn-org/ovn-kubernetes/go-controller/pkg/nbdb"
+	"github.com/ovn-org/ovn-kubernetes/go-controller/pkg/sbdb"
+	"github.com/pkg/errors"
+	"github.com/submariner-io/submariner/pkg/util/clusterfiles"
+	clientset "k8s.io/client-go/kubernetes"
+)
+
+type ConnectionHandler struct {
+	syncMutex    sync.Mutex
+	k8sClientset clientset.Interface
+	nbdb         libovsdbclient.Client
+}
+
+func NewConnectionHandler(k8sClientset clientset.Interface) *ConnectionHandler {
+	return &ConnectionHandler{
+		k8sClientset: k8sClientset,
+	}
+}
+
+func (connectionHandler *ConnectionHandler) initClients() error {
+	var tlsConfig *tls.Config
+
+	if strings.HasPrefix(getOVNNBDBAddress(), "ssl:") {
+		logger.Infof("OVN connection using SSL, loading certificates")
+
+		certFile, err := clusterfiles.Get(connectionHandler.k8sClientset, getOVNCertPath())
+		if err != nil {
+			return errors.Wrapf(err, "error getting config for %q", getOVNCertPath())
+		}
+
+		pkFile, err := clusterfiles.Get(connectionHandler.k8sClientset, getOVNPrivKeyPath())
+		if err != nil {
+			return errors.Wrapf(err, "error getting config for %q", getOVNPrivKeyPath())
+		}
+
+		caFile, err := clusterfiles.Get(connectionHandler.k8sClientset, getOVNCaBundlePath())
+		if err != nil {
+			return errors.Wrapf(err, "error getting config for %q", getOVNCaBundlePath())
+		}
+
+		tlsConfig, err = getOVNTLSConfig(pkFile, certFile, caFile)
+		if err != nil {
+			return errors.Wrap(err, "error getting OVN TLS config")
+		}
+	}
+
+	// Create nbdb client
+	nbdbModel, err := nbdb.FullDatabaseModel()
+	if err != nil {
+		return errors.Wrap(err, "error getting OVN NBDB database model")
+	}
+
+	connectionHandler.nbdb, err = createLibovsdbClient(getOVNNBDBAddress(), tlsConfig, nbdbModel)
+	if err != nil {
+		return errors.Wrap(err, "error creating NBDB connection")
+	}
+
+	return nil
+}
+
+func getOVNTLSConfig(pkFile, certFile, caFile string) (*tls.Config, error) {
+	cert, err := tls.LoadX509KeyPair(certFile, pkFile)
+	if err != nil {
+		return nil, errors.Wrap(err, "Failure loading ovn certificates")
+	}
+
+	rootCAs := x509.NewCertPool()
+
+	data, err := os.ReadFile(caFile)
+	if err != nil {
+		return nil, errors.Wrap(err, "failure loading OVNDB ca bundle")
+	}
+
+	rootCAs.AppendCertsFromPEM(data)
+
+	return &tls.Config{
+		Certificates: []tls.Certificate{cert},
+		RootCAs:      rootCAs,
+		ServerName:   "ovn",
+		MinVersion:   tls.VersionTLS12,
+	}, nil
+}
+
+func createLibovsdbClient(dbAddress string, tlsConfig *tls.Config, dbModel model.ClientDBModel) (libovsdbclient.Client, error) {
+	options := []libovsdbclient.Option{
+		// Reading and parsing the DB after reconnect at scale can (unsurprisingly)
+		// take longer than a normal ovsdb operation. Give it a bit more time so
+		// we don't time out and enter a reconnect loop.
+		libovsdbclient.WithReconnect(OVSDBTimeout, &backoff.ZeroBackOff{}),
+		libovsdbclient.WithLogger(&logger.Logger),
+	}
+	logger.Infof("creatingLibOVSDB Client ")
+	/*	if dbAddress == "local" {
+			logger.Infof("DB address is local  %s", dbAddress)
+		options = append(options, libovsdbclient.WithEndpoint("unix:/var/run/openvswitch/ovnnb_db.sock"))
+		} else {*/
+	logger.Infof("DB address is not local %s and tlsConfig is ", dbAddress, tlsConfig)
+	options = append(options, libovsdbclient.WithEndpoint(dbAddress))
+	if tlsConfig != nil {
+		options = append(options, libovsdbclient.WithTLSConfig(tlsConfig))
+	}
+	/*}*/
+
+	client, err := libovsdbclient.NewOVSDBClient(dbModel, options...)
+	if err != nil {
+		return nil, errors.Wrap(err, "error creating ovsdbClient")
+	}
+	logger.Infof("Client is  %s", client)
+
+	ctx, cancel := context.WithTimeout(context.Background(), OVSDBTimeout)
+	defer cancel()
+	logger.Info("The db name is OVN_Northbound1 %s", dbModel.Name())
+	err = client.Connect(ctx)
+	if err != nil {
+		return nil, errors.Wrap(err, "error connecting to ovsdb")
+	}
+
+	logger.Info("The db name is OVN_Northbound %s", dbModel.Name())
+	if dbModel.Name() == "OVN_Northbound" {
+		logger.Info("The db is is OVN_Northbound")
+		_, err = client.MonitorAll(ctx)
+		if err != nil {
+			client.Close()
+			return nil, errors.Wrap(err, "error setting OVN NBDB client to monitor-all")
+		}
+	} else {
+		// Only Monitor Required SBDB tables to reduce memory overhead
+		_, err = client.Monitor(ctx,
+			client.NewMonitor(
+				libovsdbclient.WithTable(&sbdb.Chassis{}),
+			),
+		)
+		if err != nil {
+			client.Close()
+			return nil, errors.Wrap(err, "error monitoring chassis table in OVN SBDB")
+		}
+	}
+
+	logger.Info("Client is %v", client)
+
+	return client, nil
+}

pkg/routeagent_driver/handlers/ovn/env.go

@@ -0,0 +1,77 @@
+/*
+SPDX-License-Identifier: Apache-2.0
+
+Copyright Contributors to the Submariner project.
+
+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 ovn
+
+import (
+	"os"
+	"time"
+)
+
+// default ovsdb timeout used by ovn-k.
+const OVSDBTimeout = 20 * time.Second
+const ovnCert = "secret://openshift-ovn-kubernetes/ovn-cert/tls.crt"
+const ovnPrivKey = "secret://openshift-ovn-kubernetes/ovn-cert/tls.key"
+const ovnCABundle = "configmap://openshift-ovn-kubernetes/ovn-ca/ca-bundle.crt"
+const defaultOVNNBDB = "ssl:ovnkube-db.openshift-ovn-kubernetes.svc.cluster.local:9641"
+const defaultOVNSBDB = "ssl:ovnkube-db.openshift-ovn-kubernetes.svc.cluster.local:9642"
+
+func getOVNSBDBAddress() string {
+	addr := os.Getenv("OVN_SBDB")
+	if addr == "" {
+		return defaultOVNSBDB
+	}
+
+	return addr
+}
+
+func getOVNNBDBAddress() string {
+	addr := os.Getenv("OVN_NBDB")
+	if addr == "" {
+		return defaultOVNNBDB
+	}
+
+	return addr
+}
+
+func getOVNPrivKeyPath() string {
+	key := os.Getenv("OVN_PK")
+	if key == "" {
+		return ovnPrivKey
+	}
+
+	return key
+}
+
+func getOVNCertPath() string {
+	cert := os.Getenv("OVN_CERT")
+	if cert == "" {
+		return ovnCert
+	}
+
+	return cert
+}
+
+func getOVNCaBundlePath() string {
+	ca := os.Getenv("OVN_CA")
+	if ca == "" {
+		return ovnCABundle
+	}
+
+	return ca
+}

pkg/routeagent_driver/handlers/ovn/gatewayRouteController.go

@@ -0,0 +1,140 @@
+/*
+SPDX-License-Identifier: Apache-2.0
+
+Copyright Contributors to the Submariner project.
+
+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 ovn
+
+import (
+	"github.com/pkg/errors"
+	"github.com/submariner-io/admiral/pkg/federate"
+	"github.com/submariner-io/admiral/pkg/syncer"
+	submarinerv1 "github.com/submariner-io/submariner/pkg/apis/submariner.io/v1"
+	corev1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/util/sets"
+	"net"
+)
+
+type gatewayRouteController struct {
+	resourceSyncer    syncer.Interface
+	connectionHandler *ConnectionHandler
+	remoteSubnets     sets.Set[string]
+	stopCh            chan struct{}
+	mgmtIp            string
+}
+
+func NewGatewayRoute(config *syncer.ResourceSyncerConfig, connectionHandler *ConnectionHandler,
+) (*gatewayRouteController, error) {
+	var err error
+
+	controller := &gatewayRouteController{
+		connectionHandler: connectionHandler,
+		remoteSubnets:     sets.New[string](),
+	}
+
+	federator := federate.NewUpdateStatusFederator(config.SourceClient, config.RestMapper, corev1.NamespaceAll)
+
+	controller.resourceSyncer, err = syncer.NewResourceSyncer(&syncer.ResourceSyncerConfig{
+		Name:                "GatewayRoute syncer",
+		ResourceType:        &submarinerv1.GatewayRoute{},
+		SourceClient:        config.SourceClient,
+		SourceNamespace:     "submariner-operator",
+		RestMapper:          config.RestMapper,
+		Federator:           federator,
+		Scheme:              config.Scheme,
+		Transform:           controller.process,
+		ResourcesEquivalent: syncer.AreSpecsEquivalent,
+	})
+
+	if err != nil {
+		return nil, errors.Wrap(err, "error creating resource syncer")
+	}
+
+	mgmtIp, err := getNextHopOnK8sMgmtIntf()
+	if err != nil {
+		return nil, err
+	}
+
+	controller.mgmtIp = mgmtIp
+
+	controller.resourceSyncer.Start(controller.stopCh)
+
+	logger.Infof("Started GatewayRouteController")
+
+	return controller, nil
+}
+
+func (g *gatewayRouteController) process(from runtime.Object, numRequeues int, op syncer.Operation) (runtime.Object, bool) {
+	logger.Infof("Process called with from %v, op %v", from, op)
+	submGWRoute := from.(*submarinerv1.GatewayRoute)
+	if submGWRoute.RoutePolicySpec.NextHops[0] == g.mgmtIp {
+		if op == syncer.Create || op == syncer.Update {
+			logger.Infof("In Add/Update function")
+			for _, subnet := range submGWRoute.RoutePolicySpec.RemoteCIDRs {
+				logger.Infof("Inserting Subnetes")
+				g.remoteSubnets.Insert(subnet)
+			}
+		} else {
+			for _, subnet := range submGWRoute.RoutePolicySpec.RemoteCIDRs {
+				logger.Infof("In Delete function")
+				g.remoteSubnets.Delete(subnet)
+			}
+		}
+		g.connectionHandler.ReconcileSubOvnLogicalRouterPolicies(g.remoteSubnets, g.mgmtIp)
+		g.connectionHandler.ReconcileOvnLogicalRouterStaticRoutes(g.remoteSubnets, g.mgmtIp)
+	} else {
+		logger.Infof("Next Hop %v not matching mangement IP %v hence skipping",
+			submGWRoute.RoutePolicySpec.NextHops[0], g.mgmtIp)
+	}
+
+	return nil, false
+}
+
+func (g *gatewayRouteController) Start() error {
+
+	return g.resourceSyncer.Start(g.stopCh)
+}
+
+func (g *gatewayRouteController) Stop() {
+	close(g.stopCh)
+}
+
+func compareIPInSameNetwork(ip1, ip2 string) (bool, error) {
+	logger.Infof("IP1 is %q", ip1)
+	_, ipnet1, err := net.ParseCIDR(ip1)
+	if err != nil {
+		return false, errors.Wrapf(err, "Error parsing IP address %v", ip1)
+	}
+
+	logger.Infof("IP1 is %q", ip2)
+	addr2, _, err := net.ParseCIDR(ip2)
+	if err != nil {
+		return false, errors.Wrapf(err, "Error parsing IP address %v", ip2)
+	}
+
+	return ipnet1.Contains(addr2), nil
+}
+
+func isIPv4CIDR(address string) (bool, error) {
+	_, iPnet, err := net.ParseCIDR(address)
+	if err != nil {
+		return false, errors.Wrapf(err, "Error parsing IP address %v", iPnet)
+	}
+
+	ip := iPnet.IP
+	return ip != nil && ip.To4() != nil, nil
+}