Skip to content

Commit

Permalink
Add ovn IC controller
Browse files Browse the repository at this point in the history
Signed-off-by: Aswin Suryanarayanan <aswinsuryan@gmail.com>
  • Loading branch information
aswinsuryan committed Aug 1, 2023
1 parent 3b162b5 commit 8785859
Show file tree
Hide file tree
Showing 12 changed files with 760 additions and 208 deletions.
4 changes: 2 additions & 2 deletions pkg/networkplugin-syncer/handlers/ovn/connection.go
Original file line number Diff line number Diff line change
Expand Up @@ -20,11 +20,11 @@ package ovn

import (
"context"
"crypto/tls"
"crypto/x509"
"os"
"strings"

"crypto/tls"
"crypto/x509"
"github.com/cenkalti/backoff/v4"
libovsdbclient "github.com/ovn-org/libovsdb/client"
"github.com/ovn-org/libovsdb/model"
Expand Down
47 changes: 22 additions & 25 deletions pkg/networkplugin-syncer/main.go
Original file line number Diff line number Diff line change
Expand Up @@ -26,17 +26,12 @@ import (
"github.com/submariner-io/admiral/pkg/log"
"github.com/submariner-io/admiral/pkg/log/kzerolog"
"github.com/submariner-io/submariner/pkg/cni"
"github.com/submariner-io/submariner/pkg/event"
"github.com/submariner-io/submariner/pkg/event/controller"
eventlogger "github.com/submariner-io/submariner/pkg/event/logger"
"github.com/submariner-io/submariner/pkg/networkplugin-syncer/handlers/ovn"
"github.com/submariner-io/submariner/pkg/routeagent_driver/environment"
"github.com/submariner-io/submariner/pkg/versions"
"k8s.io/client-go/kubernetes"
"k8s.io/client-go/rest"
"k8s.io/client-go/tools/clientcmd"
logf "sigs.k8s.io/controller-runtime/pkg/log"
"sigs.k8s.io/controller-runtime/pkg/manager/signals"
)

var (
Expand All @@ -54,7 +49,7 @@ func main() {

logger.Info("Starting submariner-networkplugin-syncer")
// set up signals so we handle the first shutdown signal gracefully
stopCh := signals.SetupSignalHandler().Done()
// stopCh := signals.SetupSignalHandler().Done()

var env environment.Specification

Expand All @@ -67,32 +62,34 @@ func main() {
networkPlugin = cni.Generic
}

registry := event.NewRegistry("networkplugin-syncer", networkPlugin)
err = registry.AddHandlers(eventlogger.NewHandler(), ovn.NewSyncHandler(getK8sClient(), &env))
logger.FatalOnError(err, "Error registering the handlers")
/* registry := event.NewRegistry("networkplugin-syncer", networkPlugin)
err = registry.AddHandlers(eventlogger.NewHandler(), ovn.NewSyncHandler(getK8sClient(), &env))
logger.FatalOnError(err, "Error registering the handlers")
if env.Uninstall {
if err := registry.StopHandlers(true); err != nil {
logger.Warningf("Error stopping handlers: %v", err)
if env.Uninstall {
if err := registry.StopHandlers(true); err != nil {
logger.Warningf("Error stopping handlers: %v", err)
}
return
}
return
}
ctl, err := controller.New(&controller.Config{
Registry: registry,
MasterURL: masterURL,
Kubeconfig: kubeconfig,
})
logger.FatalOnError(err, "Error creating controller for event handling")
ctl, err := controller.New(&controller.Config{
Registry: registry,
MasterURL: masterURL,
Kubeconfig: kubeconfig,
})
logger.FatalOnError(err, "Error creating controller for event handling")
err = ctl.Start(stopCh)
logger.FatalOnError(err, "Error starting controller")
err = ctl.Start(stopCh)
logger.FatalOnError(err, "Error starting controller")
<-stopCh
ctl.Stop()
<-stopCh
ctl.Stop()
logger.Info("All controllers stopped or exited. Stopping submariner-networkplugin-syncer")
logger.Info("All controllers stopped or exited. Stopping submariner-networkplugin-syncer")
*/
}

func getK8sClient() kubernetes.Interface {
Expand Down
160 changes: 160 additions & 0 deletions pkg/routeagent_driver/handlers/ovn/connection.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,160 @@
/*
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"
"strings"

"context"
"crypto/tls"
"crypto/x509"
"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 {
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:") {
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),
}

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")
}

ctx, cancel := context.WithTimeout(context.Background(), OVSDBTimeout)
defer cancel()

err = client.Connect(ctx)
if err != nil {
return nil, errors.Wrap(err, "error connecting to ovsdb")
}

if dbModel.Name() == "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
}
69 changes: 69 additions & 0 deletions pkg/routeagent_driver/handlers/ovn/env.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,69 @@
/*
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
ovnCert = "secret://openshift-ovn-kubernetes/ovn-cert/tls.crt"
ovnPrivKey = "secret://openshift-ovn-kubernetes/ovn-cert/tls.key"
ovnCABundle = "configmap://openshift-ovn-kubernetes/ovn-ca/ca-bundle.crt"
defaultOVNNBDB = "ssl:ovnkube-db.openshift-ovn-kubernetes.svc.cluster.local:9641"
)

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
}
Loading

0 comments on commit 8785859

Please sign in to comment.