Bug 2092579: [DownstreamMerge] 6-1-22

go-controller/cmd/ovn-kube-util/app/ovs-exporter.go

@@ -1,8 +1,10 @@
 package app
 
 import (
+	"context"
 	"k8s.io/klog/v2"
 	"net/http"
+	"time"
 
 	"github.com/ovn-org/ovn-kubernetes/go-controller/pkg/metrics"
 	"github.com/ovn-org/ovn-kubernetes/go-controller/pkg/util"
@@ -21,6 +23,7 @@ var OvsExporterCommand = cli.Command{
 		},
 	},
 	Action: func(ctx *cli.Context) error {
+		stopChan := make(chan struct{})
 		bindAddress := ctx.String("metrics-bind-address")
 		if bindAddress == "" {
 			bindAddress = "0.0.0.0:9310"
@@ -34,12 +37,24 @@ var OvsExporterCommand = cli.Command{
 		mux.Handle("/metrics", promhttp.Handler())
 
 		// register ovs metrics that will be served off of /metrics path
-		metrics.RegisterStandaloneOvsMetrics()
+		metrics.RegisterStandaloneOvsMetrics(stopChan)
 
-		err := http.ListenAndServe(bindAddress, mux)
-		if err != nil {
-			klog.Exitf("Starting metrics server failed: %v", err)
+		server := &http.Server{Addr: bindAddress, Handler: mux}
+		go func() {
+			if err := server.ListenAndServe(); err != nil && err != http.ErrServerClosed {
+				klog.Exitf("Metrics server exited with error: %v", err)
+			}
+		}()
+
+		// run until cancelled
+		<-ctx.Context.Done()
+		close(stopChan)
+		shutdownCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+		defer cancel()
+		if err := server.Shutdown(shutdownCtx); err != nil {
+			klog.Errorf("Error stopping metrics server: %v", err)
 		}
+
 		return nil
 	},
 }

go-controller/cmd/ovnkube/ovnkube.go

@@ -280,18 +280,18 @@ func runOvnKube(ctx *cli.Context) error {
 	// start the prometheus server to serve OVN K8s Metrics (default master port: 9409, node port: 9410)
 	if config.Metrics.BindAddress != "" {
 		metrics.StartMetricsServer(config.Metrics.BindAddress, config.Metrics.EnablePprof,
-			config.Metrics.NodeServerCert, config.Metrics.NodeServerPrivKey)
+			config.Metrics.NodeServerCert, config.Metrics.NodeServerPrivKey, stopChan, wg)
 	}
 
 	// start the prometheus server to serve OVS and OVN Metrics (default port: 9476)
 	// Note: for ovnkube node mode dpu-host no metrics is required as ovs/ovn is not running on the node.
 	if config.OvnKubeNode.Mode != types.NodeModeDPUHost && config.Metrics.OVNMetricsBindAddress != "" {
 		if config.Metrics.ExportOVSMetrics {
-			metrics.RegisterOvsMetricsWithOvnMetrics()
+			metrics.RegisterOvsMetricsWithOvnMetrics(stopChan)
 		}
-		metrics.RegisterOvnMetrics(ovnClientset.KubeClient, node)
+		metrics.RegisterOvnMetrics(ovnClientset.KubeClient, node, stopChan)
 		metrics.StartOVNMetricsServer(config.Metrics.OVNMetricsBindAddress,
-			config.Metrics.NodeServerCert, config.Metrics.NodeServerPrivKey)
+			config.Metrics.NodeServerCert, config.Metrics.NodeServerPrivKey, stopChan, wg)
 	}
 
 	// run until cancelled

go-controller/go.mod

@@ -21,9 +21,10 @@ require (
 	github.com/onsi/gomega v1.14.0
 	github.com/openshift/api v0.0.0-20211201215911-5a82bae32e46
 	github.com/openshift/client-go v0.0.0-20211202194848-d3f186f2d366
-	github.com/ovn-org/libovsdb v0.6.1-0.20220427123326-d7b273399db4
+	github.com/ovn-org/libovsdb v0.6.1-0.20220513144310-50ec17900991
 	github.com/pkg/errors v0.9.1
 	github.com/prometheus/client_golang v1.11.0
+	github.com/prometheus/client_model v0.2.0
 	github.com/spf13/afero v1.4.1
 	github.com/stretchr/testify v1.7.0
 	github.com/urfave/cli/v2 v2.2.0
@@ -71,7 +72,6 @@ require (
 	github.com/modern-go/reflect2 v1.0.2 // indirect
 	github.com/nxadm/tail v1.4.8 // indirect
 	github.com/pmezard/go-difflib v1.0.0 // indirect
-	github.com/prometheus/client_model v0.2.0 // indirect
 	github.com/prometheus/common v0.26.0 // indirect
 	github.com/prometheus/procfs v0.6.0 // indirect
 	github.com/russross/blackfriday/v2 v2.0.1 // indirect

go-controller/go.sum

@@ -402,6 +402,8 @@ github.com/openshift/client-go v0.0.0-20211202194848-d3f186f2d366/go.mod h1:HJeH
 github.com/ory/dockertest/v3 v3.8.0/go.mod h1:9zPATATlWQru+ynXP+DytBQrsXV7Tmlx7K86H6fQaDo=
 github.com/ovn-org/libovsdb v0.6.1-0.20220427123326-d7b273399db4 h1:atUIOA34Cg9GVFn/8rmIsmnOIbi0D82x+xfhV2UuF1Q=
 github.com/ovn-org/libovsdb v0.6.1-0.20220427123326-d7b273399db4/go.mod h1:BQPdnSM2QOKxPwxl7wHJDSPP4B/CDKq3+vzgFW3J5gE=
+github.com/ovn-org/libovsdb v0.6.1-0.20220513144310-50ec17900991 h1:EsMLPWOIRgB9WFTt5+L99LaX4H1Nm6yL2S6zsx4TSzY=
+github.com/ovn-org/libovsdb v0.6.1-0.20220513144310-50ec17900991/go.mod h1:BQPdnSM2QOKxPwxl7wHJDSPP4B/CDKq3+vzgFW3J5gE=
 github.com/peterbourgon/diskv v2.0.1+incompatible/go.mod h1:uqqh8zWWbv1HBMNONnaR/tNboyR3/BZd58JJSHlUSCU=
 github.com/pkg/errors v0.8.0/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
 github.com/pkg/errors v0.8.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=

go-controller/hybrid-overlay/pkg/controller/node_linux.go

@@ -604,13 +604,16 @@ func (n *NodeController) RunFlowSync(stopCh <-chan struct{}) {
 	klog.Info("Starting hybrid overlay OpenFlow sync thread")
 	klog.Info("Running initial OpenFlow sync")
 	n.syncFlows()
-
+	syncPeriod := 30 * time.Second
+	timer := time.NewTicker(syncPeriod)
+	defer timer.Stop()
 	for {
 		select {
-		case <-time.After(30 * time.Second):
+		case <-timer.C:
 			n.syncFlows()
 		case <-n.flowChan:
 			n.syncFlows()
+			timer.Reset(syncPeriod)
 		case <-stopCh:
 			klog.Info("Shutting down OpenFlow sync thread")
 			return

go-controller/pkg/libovsdbops/acl.go

@@ -138,27 +138,6 @@ func UpdateACLsLoggingOps(nbClient libovsdbclient.Client, ops []libovsdb.Operati
 	return modelClient.CreateOrUpdateOps(ops, opModels...)
 }
 
-// UpdateACLsDirection updates the direction on the provided ACLs
-func UpdateACLsDirection(nbClient libovsdbclient.Client, acls ...*nbdb.ACL) error {
-	opModels := make([]operationModel, 0, len(acls))
-	for i := range acls {
-		// can't use i in the predicate, for loop replaces it in-memory
-		acl := acls[i]
-		opModel := operationModel{
-			Model:          acl,
-			ModelPredicate: func(item *nbdb.ACL) bool { return isEquivalentACL(item, acl) },
-			OnModelUpdates: []interface{}{&acl.Direction},
-			ErrNotFound:    true,
-			BulkOp:         false,
-		}
-		opModels = append(opModels, opModel)
-	}
-
-	modelClient := newModelClient(nbClient)
-	_, err := modelClient.CreateOrUpdate(opModels...)
-	return err
-}
-
 // DeleteACLs deletes the provided ACLs
 func DeleteACLs(nbClient libovsdbclient.Client, acls ...*nbdb.ACL) error {
 	opModels := make([]operationModel, 0, len(acls))

go-controller/pkg/libovsdbops/loadbalancer.go

@@ -58,6 +58,25 @@ func BuildLoadBalancer(name string, protocol nbdb.LoadBalancerProtocol, selectio
 	}
 }
 
+// CreateLoadBalancersOps creates the provided load balancers returning the
+// corresponding ops
+func CreateLoadBalancersOps(nbClient libovsdbclient.Client, ops []libovsdb.Operation, lbs ...*nbdb.LoadBalancer) ([]libovsdb.Operation, error) {
+	opModels := make([]operationModel, 0, len(lbs))
+	for i := range lbs {
+		lb := lbs[i]
+		opModel := operationModel{
+			Model:          lb,
+			OnModelUpdates: onModelUpdatesNone(),
+			ErrNotFound:    false,
+			BulkOp:         false,
+		}
+		opModels = append(opModels, opModel)
+	}
+
+	modelClient := newModelClient(nbClient)
+	return modelClient.CreateOrUpdateOps(ops, opModels...)
+}
+
 // CreateOrUpdateLoadBalancersOps creates or updates the provided load balancers
 // returning the corresponding ops
 func CreateOrUpdateLoadBalancersOps(nbClient libovsdbclient.Client, ops []libovsdb.Operation, lbs ...*nbdb.LoadBalancer) ([]libovsdb.Operation, error) {

go-controller/pkg/libovsdbops/model.go

@@ -342,12 +342,13 @@ func onModels(models interface{}, do func(interface{}) error) error {
 // When no specific operation is required for the provided model, returns an empty
 // array for convenience.
 func buildFailOnDuplicateOps(c client.Client, m model.Model) ([]ovsdb.Operation, error) {
-	// Right now we only consider models with a "Name" field that is not an
+	// Right now we mostly consider models with a "Name" field that is not an
 	// index for which we don't expect duplicate names.
 	// A duplicate Name field that is an index will fail without the
 	// need of this wait operation.
-	// Models that require a complex condition to detect duplicates are not
-	// considered for the time being due to the performance hit (i.e ACLs).
+	// Some models that require a complex condition to detect duplicates are not
+	// considered for the time being due to the performance hit (e.g ACLs).
+	timeout := types.OVSDBWaitTimeout
 	var field interface{}
 	var value string
 	switch t := m.(type) {
@@ -360,11 +361,28 @@ func buildFailOnDuplicateOps(c client.Client, m model.Model) ([]ovsdb.Operation,
 	case *nbdb.LogicalSwitch:
 		field = &t.Name
 		value = t.Name
+	case *nbdb.LogicalRouterPolicy:
+		condPriority := model.Condition{
+			Field:    &t.Priority,
+			Function: ovsdb.ConditionEqual,
+			Value:    t.Priority,
+		}
+		condMatch := model.Condition{
+			Field:    &t.Match,
+			Function: ovsdb.ConditionEqual,
+			Value:    t.Match,
+		}
+		return c.WhereAll(t, condPriority, condMatch).Wait(
+			ovsdb.WaitConditionNotEqual,
+			&timeout,
+			t,
+			&t.Priority,
+			&t.Match,
+		)
 	default:
 		return []ovsdb.Operation{}, nil
 	}
 
-	timeout := types.OVSDBWaitTimeout
 	cond := model.Condition{
 		Field:    field,
 		Function: ovsdb.ConditionEqual,

go-controller/pkg/libovsdbops/model_client.go

@@ -195,6 +195,8 @@ func (m *modelClient) CreateOrUpdateOps(ops []ovsdb.Operation, opModels ...opera
 }
 
 func (m *modelClient) createOrUpdateOps(ops []ovsdb.Operation, opModels ...operationModel) (interface{}, []ovsdb.Operation, error) {
+	hasGuardOp := len(ops) > 0 && isGuardOp(&ops[0])
+	guardOp := []ovsdb.Operation{}
 	doWhenFound := func(model interface{}, opModel *operationModel) ([]ovsdb.Operation, error) {
 		if opModel.OnModelUpdates != nil {
 			return m.update(model, opModel)
@@ -204,9 +206,25 @@ func (m *modelClient) createOrUpdateOps(ops []ovsdb.Operation, opModels ...opera
 		return nil, nil
 	}
 	doWhenNotFound := func(model interface{}, opModel *operationModel) ([]ovsdb.Operation, error) {
+		if !hasGuardOp {
+			// for the first insert of certain models, build a wait operation
+			// that checks for duplicates as a guard op to prevent against
+			// duplicate transactions
+			var err error
+			guardOp, err = buildFailOnDuplicateOps(m.client, opModel.Model)
+			if err != nil {
+				return nil, err
+			}
+			hasGuardOp = len(guardOp) > 0
+		}
 		return m.create(opModel)
 	}
-	return m.buildOps(ops, doWhenFound, doWhenNotFound, opModels...)
+	created, ops, err := m.buildOps(ops, doWhenFound, doWhenNotFound, opModels...)
+	if len(guardOp) > 0 {
+		// set the guard op as the first of the list
+		ops = append(guardOp, ops...)
+	}
+	return created, ops, err
 }
 
 /*
@@ -318,17 +336,11 @@ func (m *modelClient) create(opModel *operationModel) ([]ovsdb.Operation, error)
 		setUUID(opModel.Model, buildNamedUUID())
 	}
 
-	ops, err := buildFailOnDuplicateOps(m.client, opModel.Model)
+	ops, err := m.client.Create(opModel.Model)
 	if err != nil {
 		return nil, fmt.Errorf("unable to create model, err: %v", err)
 	}
 
-	op, err := m.client.Create(opModel.Model)
-	if err != nil {
-		return nil, fmt.Errorf("unable to create model, err: %v", err)
-	}
-	ops = append(ops, op...)
-
 	klog.V(5).Infof("Create operations generated as: %+v", ops)
 	return ops, nil
 }
@@ -468,3 +480,7 @@ func addToExistingResult(model interface{}, existingResult interface{}) error {
 
 	return nil
 }
+
+func isGuardOp(op *ovsdb.Operation) bool {
+	return op != nil && op.Op == ovsdb.OperationWait && op.Timeout != nil && *op.Timeout == types.OVSDBWaitTimeout
+}