GATEWAYS-4306: exporting metrics for conntrack per zone

go.mod

@@ -6,11 +6,14 @@ require (
 	github.com/google/go-cmp v0.7.0
 	github.com/mdlayher/genetlink v1.3.2
 	github.com/mdlayher/netlink v1.8.0
+	github.com/ti-mo/conntrack v0.5.2
+	github.com/ti-mo/netfilter v0.5.3
 	golang.org/x/sys v0.37.0
 )
 
 require (
 	github.com/mdlayher/socket v0.5.1 // indirect
+	github.com/pkg/errors v0.9.1 // indirect
 	golang.org/x/net v0.45.0 // indirect
-	golang.org/x/sync v0.3.0 // indirect
+	golang.org/x/sync v0.14.0 // indirect
 )

go.sum

@@ -1,3 +1,5 @@
+github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
+github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/google/go-cmp v0.7.0 h1:wk8382ETsv4JYUZwIsn6YpYiWiBsYLSJiTsyBybVuN8=
 github.com/google/go-cmp v0.7.0/go.mod h1:pXiqmnSA92OHEEa9HXL2W4E7lf9JzCmGVUdgjX3N/iU=
 github.com/mdlayher/genetlink v1.3.2 h1:KdrNKe+CTu+IbZnm/GVUMXSqBBLqcGpRDa0xkQy56gw=
@@ -6,9 +8,23 @@ github.com/mdlayher/netlink v1.8.0 h1:e7XNIYJKD7hUct3Px04RuIGJbBxy1/c4nX7D5Yyvvl
 github.com/mdlayher/netlink v1.8.0/go.mod h1:UhgKXUlDQhzb09DrCl2GuRNEglHmhYoWAHid9HK3594=
 github.com/mdlayher/socket v0.5.1 h1:VZaqt6RkGkt2OE9l3GcC6nZkqD3xKeQLyfleW/uBcos=
 github.com/mdlayher/socket v0.5.1/go.mod h1:TjPLHI1UgwEv5J1B5q0zTZq12A/6H7nKmtTanQE37IQ=
+github.com/pkg/errors v0.9.1 h1:FEBLx1zS214owpjy7qsBeixbURkuhQAwrK5UwLGTwt4=
+github.com/pkg/errors v0.9.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
+github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
+github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
+github.com/stretchr/testify v1.8.4 h1:CcVxjf3Q8PM0mHUKJCdn+eZZtm5yQwehR5yeSVQQcUk=
+github.com/stretchr/testify v1.8.4/go.mod h1:sz/lmYIOXD/1dqDmKjjqLyZ2RngseejIcXlSw2iwfAo=
+github.com/ti-mo/conntrack v0.5.2 h1:PQ7MCdFjniEiTJT+qsAysREUsT5iH62/VNyhkB06HOI=
+github.com/ti-mo/conntrack v0.5.2/go.mod h1:4HZrFQQLOSuBzgQNid3H/wYyyp1kfGXUYxueXjIGibo=
+github.com/ti-mo/netfilter v0.5.3 h1:ikzduvnaUMwre5bhbNwWOd6bjqLMVb33vv0XXbK0xGQ=
+github.com/ti-mo/netfilter v0.5.3/go.mod h1:08SyBCg6hu1qyQk4s3DjjJKNrm3RTb32nm6AzyT972E=
+github.com/vishvananda/netns v0.0.4 h1:Oeaw1EM2JMxD51g9uhtC0D7erkIjgmj8+JZc26m1YX8=
+github.com/vishvananda/netns v0.0.4/go.mod h1:SpkAiCQRtJ6TvvxPnOSyH3BMl6unz3xZlaprSwhNNJM=
 golang.org/x/net v0.45.0 h1:RLBg5JKixCy82FtLJpeNlVM0nrSqpCRYzVU1n8kj0tM=
 golang.org/x/net v0.45.0/go.mod h1:ECOoLqd5U3Lhyeyo/QDCEVQ4sNgYsqvCZ722XogGieY=
-golang.org/x/sync v0.3.0 h1:ftCYgMx6zT/asHUrPw8BLLscYtGznsLAnjq5RH9P66E=
-golang.org/x/sync v0.3.0/go.mod h1:FU7BRWz2tNW+3quACPkgCx/L+uEAv1htQ0V83Z9Rj+Y=
+golang.org/x/sync v0.14.0 h1:woo0S4Yywslg6hp4eUFjTVOyKt0RookbpAHG4c1HmhQ=
+golang.org/x/sync v0.14.0/go.mod h1:1dzgHSNfp02xaA81J2MS99Qcpr2w7fw1gpm99rleRqA=
 golang.org/x/sys v0.37.0 h1:fdNQudmxPjkdUTPnLn5mdQv7Zwvbvpaxqs831goi9kQ=
 golang.org/x/sys v0.37.0/go.mod h1:OgkHotnGiDImocRcuBABYBEXf8A9a87e/uXjp9XT3ks=
+gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
+gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=

ovsnl/client.go

@@ -37,64 +37,72 @@ type Client struct {
 	// Datapath provides access to DatapathService methods.
 	Datapath *DatapathService
 
-	c *genetlink.Conn
+	c   *genetlink.Conn
+	Agg *ZoneMarkAggregator // lazily initialized
 }
 
 // New creates a new Linux Open vSwitch generic netlink client.
 //
 // If no OvS generic netlink families are available on this system, an
 // error will be returned which can be checked using os.IsNotExist.
 func New() (*Client, error) {
-	c, err := genetlink.Dial(nil)
+	c := &Client{} // Create client instance first
+
+	// Initialize the underlying genetlink connection.
+	conn, err := genetlink.Dial(nil)
 	if err != nil {
 		return nil, err
 	}
+	c.c = conn
 
-	return newClient(c)
-}
-
-// newClient is the internal Client constructor, used in tests.
-func newClient(c *genetlink.Conn) (*Client, error) {
-	// Must ensure that the generic netlink connection is closed on any errors
-	// that occur before it is returned to the caller.
-
-	families, err := c.ListFamilies()
+	// Initialize services.
+	families, err := c.c.ListFamilies()
 	if err != nil {
-		_ = c.Close()
+		_ = c.c.Close()
 		return nil, err
 	}
 
-	client := &Client{c: c}
-	if err := client.init(families); err != nil {
-		_ = c.Close()
+	if err := c.init(families); err != nil {
+		_ = c.c.Close()
 		return nil, err
 	}
 
-	return client, nil
+	return c, nil
 }
 
 // Close closes the Client's generic netlink connection.
 func (c *Client) Close() error {
-	return c.c.Close()
+	if c.Agg != nil {
+		c.Agg.Stop()
+	}
+
+	if c.c != nil {
+		return c.c.Close()
+	}
+	return nil
 }
 
 // init initializes the generic netlink family service of Client.
 func (c *Client) init(families []genetlink.Family) error {
 	var gotf int
 
 	for _, f := range families {
-		// Ignore any families without the OVS prefix.
-		if !strings.HasPrefix(f.Name, "ovs_") {
-			continue
-		}
-		// Ignore any families that might be unknown.
-		if err := c.initFamily(f); err != nil {
+		// Initialize OVS-specific families
+		if strings.HasPrefix(f.Name, "ovs_") {
+			if err := c.initFamily(f); err != nil {
+				// Log but continue if an OVS family fails to init
+				fmt.Printf("Warning: failed to initialize OVS family %q: %v\n", f.Name, err)
+				continue
+			}
+		} else if f.Name == "nf_conntrack" { // Explicitly initialize for Netfilter conntrack family
+			// Acknowledge that conntrack family exists - aggregator will handle conntrack operations
+		} else {
+			// Skip other non-OVS/non-conntrack families
 			continue
 		}
 		gotf++
 	}
 
-	// No known families; return error for os.IsNotExist check.
 	if gotf == 0 {
 		return os.ErrNotExist
 	}

ovsnl/client_linux_test.go

@@ -29,6 +29,28 @@ import (
 	"golang.org/x/sys/unix"
 )
 
+func newTestClient(conn *genetlink.Conn) (*Client, error) {
+	c := &Client{}
+	c.c = conn
+
+	families, err := c.c.ListFamilies()
+	if err != nil {
+		return nil, err
+	}
+
+	if err := c.init(families); err != nil {
+		return nil, err
+	}
+
+	// Inject our mock connection directly into the datapath service
+	if c.Datapath != nil {
+		c.Datapath.c = c
+		c.c = conn
+	}
+
+	return c, nil
+}
+
 func TestClientNoFamiliesIsNotExist(t *testing.T) {
 	conn := genltest.Dial(func(greq genetlink.Message, nreq netlink.Message) ([]genetlink.Message, error) {
 		// Unrelated generic netlink families.
@@ -38,7 +60,7 @@ func TestClientNoFamiliesIsNotExist(t *testing.T) {
 		}), nil
 	})
 
-	_, err := newClient(conn)
+	_, err := newTestClient(conn)
 	if !os.IsNotExist(err) {
 		t.Fatalf("expected is not exist error, but got: %v", err)
 	}
@@ -53,7 +75,7 @@ func TestClientUnknownFamilies(t *testing.T) {
 		}), nil
 	})
 
-	_, err := newClient(conn)
+	_, err := newTestClient(conn)
 	if err == nil {
 		t.Fatalf("expected an error, but none occurred")
 	}
@@ -67,7 +89,7 @@ func TestClientNoFamilies(t *testing.T) {
 		return nil, nil
 	})
 
-	_, err := newClient(conn)
+	_, err := newTestClient(conn)
 	if err == nil {
 		t.Fatalf("expected an error, but none occurred")
 	}
@@ -82,7 +104,7 @@ func TestClientKnownFamilies(t *testing.T) {
 		}), nil
 	})
 
-	_, err := newClient(conn)
+	_, err := newTestClient(conn)
 	if err != nil {
 		t.Fatalf("failed to create client: %v", err)
 	}
@@ -112,24 +134,6 @@ func familyMessages(families []string) []genetlink.Message {
 	return msgs
 }
 
-// ovsFamilies creates a genltest.Func which intercepts "list family" requests
-// and returns all the OVS families.  Other requests are passed through to fn.
-func ovsFamilies(fn genltest.Func) genltest.Func {
-	return func(greq genetlink.Message, nreq netlink.Message) ([]genetlink.Message, error) {
-		if nreq.Header.Type == unix.GENL_ID_CTRL && greq.Header.Command == unix.CTRL_CMD_GETFAMILY {
-			return familyMessages([]string{
-				ovsh.DatapathFamily,
-				ovsh.FlowFamily,
-				ovsh.PacketFamily,
-				ovsh.VportFamily,
-				ovsh.MeterFamily,
-			}), nil
-		}
-
-		return fn(greq, nreq)
-	}
-}
-
 func mustMarshalAttributes(attrs []netlink.Attribute) []byte {
 	b, err := netlink.MarshalAttributes(attrs)
 	if err != nil {

ovsnl/conntrack_common.go

@@ -0,0 +1,65 @@
+// Copyright 2017 DigitalOcean.
+//
+// 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 ovsnl
+
+import (
+	"sync"
+	"time"
+
+	"github.com/ti-mo/conntrack"
+)
+
+// Tunables - adjust for your environment
+const (
+	eventChanSize      = 512 * 1024
+	eventWorkerCount   = 100
+	destroyFlushIntvl  = 100 * time.Millisecond // flush aggregated DESTROYs every 100ms for minimal lag
+	destroyDeltaCap    = 200000                 // maximum distinct (zone,mark) entries in destroyDeltas
+	dropsWarnThreshold = 100                    // threshold of missedEvents to log a stronger warning
+)
+
+// ZoneMarkAggregator keeps live counts (zmKey -> count) with bounded ingestion
+type ZoneMarkAggregator struct {
+	// primary counts (zmKey -> count) - simplified flat mapping
+	counts   map[ZmKey]int
+	countsMu sync.RWMutex
+
+	// conntrack listening connection
+	listenCli *conntrack.Conn
+
+	// lifecycle
+	stopCh chan struct{}
+	wg     sync.WaitGroup
+
+	// bounded event ingestion
+	eventsCh chan conntrack.Event
+
+	// aggregated DESTROY deltas (bounded by destroyDeltaCap)
+	deltaMu       sync.Mutex
+	destroyDeltas map[ZmKey]int
+
+	// metrics / health
+	eventCount      int64
+	lastEventTime   time.Time
+	eventRate       float64
+	missedEvents    int64
+	lastHealthCheck time.Time
+}
+
+// ZmKey is a compact key for (zone,mark)
+type ZmKey struct {
+	Zone uint16
+	Mark uint32
+}